feat: add client logging support for sync gRPC

Author: ohmayr

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2

@@ -2,6 +2,7 @@
 
 {% block content %}
 
+import logging as std_logging
 import warnings
 from typing import Callable, Dict, Optional, Sequence, Tuple, Union
 
@@ -13,6 +14,7 @@ from google.api_core import gapic_v1
 import google.auth                         # type: ignore
 from google.auth import credentials as ga_credentials  # type: ignore
 from google.auth.transport.grpc import SslCredentials  # type: ignore
+from google.protobuf.json_format import MessageToJson
 
 import grpc  # type: ignore
 
@@ -42,6 +44,66 @@ from google.longrunning import operations_pb2 # type: ignore
 {% endfilter %}
 from .base import {{ service.name }}Transport, DEFAULT_CLIENT_INFO
 
+try:
+    from google.api_core import client_logging  # type: ignore
+    CLIENT_LOGGING_SUPPORTED = True  # pragma: NO COVER
+except ImportError:  # pragma: NO COVER
+    CLIENT_LOGGING_SUPPORTED = False
+
+_LOGGER = std_logging.getLogger(__name__)
+
+
+class _LoggingClientInterceptor(grpc.UnaryUnaryClientInterceptor):  # pragma: NO COVER
+    def intercept_unary_unary(self, continuation, client_call_details, request):
+        request_metadata = client_call_details.metadata
+        if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(std_logging.DEBUG):
+            try:
+                request_payload = type(request).to_json(request)
+            except:
+                request_payload = MessageToJson(request)
+            grpc_request = {
+                "payload": request_payload,
+                "requestMethod": "grpc",
+                "metadata": dict(request_metadata),
+            }
+            _LOGGER.debug(
+                f"Sending request for {client_call_details.method}",
+                extra = {
+                    "serviceName": "{{ service.meta.address.proto }}",
+                    "rpcName": client_call_details.method,
+                    "request": grpc_request,
+                    {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2275): logging `metadata` seems repetitive and may need to be cleaned up. We're including it within "request" for consistency with REST transport.' #}
+                    "metadata": grpc_request["metadata"],
+                },
+            )
+
+        response = continuation(client_call_details, request)
+        response_metadata = response.trailing_metadata()
+        # Convert gRPC metadata `<class 'grpc.aio._metadata.Metadata'>` to list of tuples
+        metadata = dict([(k, v) for k, v in response_metadata]) if response_metadata else None
+        result = response.result()
+        if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(std_logging.DEBUG):  # pragma: NO COVER
+            try:
+                response_payload = type(result).to_json(result)
+            except:
+                response_payload = MessageToJson(result)
+            grpc_response = {
+                "payload": response_payload,
+                "metadata": metadata,
+                "status": "OK", 
+            }
+            _LOGGER.debug(
+                f"Received response for {client_call_details.method}.",
+                extra = {
+                    "serviceName": "{{ service.meta.address.proto }}",
+                    "rpcName": client_call_details.method,
+                    "response": grpc_response,
+                    {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2275): logging `metadata` seems repetitive and may need to be cleaned up. We're including it within "request" for consistency with REST transport.' #}
+                    "metadata": grpc_response["metadata"],
+                },
+            )
+        return response
+
 
 class {{ service.name }}GrpcTransport({{ service.name }}Transport):
     """gRPC backend transport for {{ service.name }}.
@@ -123,7 +185,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
           google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
               and ``credentials_file`` are passed.
         """
-        self._grpc_channel = None
+        self._base_channel = None
         self._ssl_channel_credentials = ssl_channel_credentials
         self._stubs: Dict[str, Callable] = {}
         {% if service.has_lro %}
@@ -140,7 +202,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             credentials = None
             self._ignore_credentials = True
             # If a channel was explicitly provided, set it.
-            self._grpc_channel = channel
+            self._base_channel = channel
             self._ssl_channel_credentials = None
 
         else:
@@ -176,10 +238,10 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             api_audience=api_audience,
         )
 
-        if not self._grpc_channel:
+        if not self._base_channel:
             # initialize with the provided callable or the default channel
             channel_init = channel or type(self).create_channel
-            self._grpc_channel = channel_init(
+            self._base_channel = channel_init(
                 self._host,
                 # use the credentials which are saved
                 credentials=self._credentials,
@@ -195,6 +257,11 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 ],
             )
 
+        self._grpc_channel = self._base_channel
+        if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(std_logging.DEBUG):  # pragma: NO COVER
+            self._interceptor = _LoggingClientInterceptor()
+            self._grpc_channel =  grpc.intercept_channel(self._base_channel, self._interceptor)
+
         # Wrap messages. This must be done after self._grpc_channel exists
         self._prep_wrapped_messages(client_info)
 
@@ -248,7 +315,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
     def grpc_channel(self) -> grpc.Channel:
         """Return the channel designed to connect to this service.
         """
-        return self._grpc_channel
+        return self._base_channel
 
     {% if service.has_lro %}
 
@@ -262,7 +329,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         # Quick check: Only create a new client if we do not already have one.
         if self._operations_client is None:
             self._operations_client = operations_v1.OperationsClient(
-                self.grpc_channel
+                self._grpc_channel
             )
 
         # Return the client from cache.
@@ -292,7 +359,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         # gRPC handles serialization and deserialization, so we just need
         # to pass in the functions for each.
         if '{{ method.transport_safe_name|snake_case }}' not in self._stubs:
-            self._stubs['{{ method.transport_safe_name|snake_case }}'] = self.grpc_channel.{{ method.grpc_stub_type }}(
+            self._stubs['{{ method.transport_safe_name|snake_case }}'] = self._grpc_channel.{{ method.grpc_stub_type }}(
                 '/{{ '.'.join(method.meta.address.package) }}.{{ service.name }}/{{ method.name }}',
                 request_serializer={{ method.input.ident }}.{% if method.input.ident.python_import.module.endswith('_pb2') %}SerializeToString{% else %}serialize{% endif %},
                 response_deserializer={{ method.output.ident }}.{% if method.output.ident.python_import.module.endswith('_pb2') %}FromString{% else %}deserialize{% endif %},
@@ -320,7 +387,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         # gRPC handles serialization and deserialization, so we just need
         # to pass in the functions for each.
         if "set_iam_policy" not in self._stubs:
-            self._stubs["set_iam_policy"] = self.grpc_channel.unary_unary(
+            self._stubs["set_iam_policy"] = self._grpc_channel.unary_unary(
                 "/google.iam.v1.IAMPolicy/SetIamPolicy",
                 request_serializer=iam_policy_pb2.SetIamPolicyRequest.SerializeToString,
                 response_deserializer=policy_pb2.Policy.FromString,
@@ -346,7 +413,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         # gRPC handles serialization and deserialization, so we just need
         # to pass in the functions for each.
         if "get_iam_policy" not in self._stubs:
-            self._stubs["get_iam_policy"] = self.grpc_channel.unary_unary(
+            self._stubs["get_iam_policy"] = self._grpc_channel.unary_unary(
                 "/google.iam.v1.IAMPolicy/GetIamPolicy",
                 request_serializer=iam_policy_pb2.GetIamPolicyRequest.SerializeToString,
                 response_deserializer=policy_pb2.Policy.FromString,
@@ -374,7 +441,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         # gRPC handles serialization and deserialization, so we just need
         # to pass in the functions for each.
         if "test_iam_permissions" not in self._stubs:
-            self._stubs["test_iam_permissions"] = self.grpc_channel.unary_unary(
+            self._stubs["test_iam_permissions"] = self._grpc_channel.unary_unary(
                 "/google.iam.v1.IAMPolicy/TestIamPermissions",
                 request_serializer=iam_policy_pb2.TestIamPermissionsRequest.SerializeToString,
                 response_deserializer=iam_policy_pb2.TestIamPermissionsResponse.FromString,
@@ -383,7 +450,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
     {% endif %}
 
     def close(self):
-        self.grpc_channel.close()
+        self._grpc_channel.close()
 
     {% include '%namespace/%name_%version/%sub/services/%service/transports/_mixins.py.j2' %}