automatic metrics for gRPC client interceptor (#917)

Author: cnnradams

Diff for: ext/opentelemetry-ext-grpc/CHANGELOG.md

@@ -11,6 +11,8 @@ Released 2020-07-28
 - Add gRPC client and server instrumentors
   ([788](https://github.com/open-telemetry/opentelemetry-python/pull/788))
 
+- Add metric recording (bytes in/out, errors, latency) to gRPC client
+
 ## 0.8b0
 
 Released 2020-05-27

Diff for: ext/opentelemetry-ext-grpc/setup.cfg

@@ -41,6 +41,7 @@ package_dir=
 packages=find_namespace:
 install_requires =
     opentelemetry-api == 0.12.dev0
+    opentelemetry-sdk == 0.12.dev0
     grpcio ~= 1.27
 
 [options.extras_require]

Diff for: ext/opentelemetry-ext-grpc/src/opentelemetry/ext/grpc/__init__.py

@@ -33,6 +33,10 @@
         SimpleExportSpanProcessor,
     )
 
+    from opentelemetry import metrics
+    from opentelemetry.sdk.metrics import MeterProvider
+    from opentelemetry.sdk.metrics.export import ConsoleMetricsExporter
+
     try:
         from .gen import helloworld_pb2, helloworld_pb2_grpc
     except ImportError:
@@ -42,7 +46,12 @@
     trace.get_tracer_provider().add_span_processor(
         SimpleExportSpanProcessor(ConsoleSpanExporter())
     )
-    instrumentor = GrpcInstrumentorClient()
+
+    # Set meter provider to opentelemetry-sdk's MeterProvider
+    metrics.set_meter_provider(MeterProvider())
+
+    # Optional - export GRPC specific metrics (latency, bytes in/out, errors) by passing an exporter
+    instrumentor = GrpcInstrumentorClient(exporter=ConsoleMetricsExporter(), interval=10)
     instrumentor.instrument()
 
     def run():
@@ -109,6 +118,7 @@ def serve():
         serve()
 """
 from contextlib import contextmanager
+from functools import partial
 
 import grpc
 from wrapt import wrap_function_wrapper as _wrap
@@ -139,11 +149,21 @@ def wrapper_fn(self, original_func, instance, args, kwargs):
 
 class GrpcInstrumentorClient(BaseInstrumentor):
     def _instrument(self, **kwargs):
+        exporter = kwargs.get("exporter", None)
+        interval = kwargs.get("interval", 30)
         if kwargs.get("channel_type") == "secure":
-            _wrap("grpc", "secure_channel", self.wrapper_fn)
+            _wrap(
+                "grpc",
+                "secure_channel",
+                partial(self.wrapper_fn, exporter, interval),
+            )
 
         else:
-            _wrap("grpc", "insecure_channel", self.wrapper_fn)
+            _wrap(
+                "grpc",
+                "insecure_channel",
+                partial(self.wrapper_fn, exporter, interval),
+            )
 
     def _uninstrument(self, **kwargs):
         if kwargs.get("channel_type") == "secure":
@@ -152,17 +172,28 @@ def _uninstrument(self, **kwargs):
         else:
             unwrap(grpc, "insecure_channel")
 
-    @contextmanager
-    def wrapper_fn(self, original_func, instance, args, kwargs):
-        with original_func(*args, **kwargs) as channel:
-            yield intercept_channel(channel, client_interceptor())
-
-
-def client_interceptor(tracer_provider=None):
+    def wrapper_fn(
+        self, exporter, interval, original_func, instance, args, kwargs
+    ):
+        channel = original_func(*args, **kwargs)
+        tracer_provider = kwargs.get("tracer_provider")
+        return intercept_channel(
+            channel,
+            client_interceptor(
+                tracer_provider=tracer_provider,
+                exporter=exporter,
+                interval=interval,
+            ),
+        )
+
+
+def client_interceptor(tracer_provider=None, exporter=None, interval=30):
     """Create a gRPC client channel interceptor.
 
     Args:
         tracer: The tracer to use to create client-side spans.
+        exporter: The exporter that will receive client metrics
+        interval: Time between every export call
 
     Returns:
         An invocation-side interceptor object.
@@ -171,7 +202,7 @@ def client_interceptor(tracer_provider=None):
 
     tracer = trace.get_tracer(__name__, __version__, tracer_provider)
 
-    return _client.OpenTelemetryClientInterceptor(tracer)
+    return _client.OpenTelemetryClientInterceptor(tracer, exporter, interval)
 
 
 def server_interceptor(tracer_provider=None):

Diff for: ext/opentelemetry-ext-grpc/src/opentelemetry/ext/grpc/_client.py

@@ -24,11 +24,12 @@
 
 import grpc
 
-from opentelemetry import propagators, trace
+from opentelemetry import metrics, propagators, trace
+from opentelemetry.sdk.metrics.export.controller import PushController
 from opentelemetry.trace.status import Status, StatusCanonicalCode
 
 from . import grpcext
-from ._utilities import RpcInfo
+from ._utilities import RpcInfo, TimedMetricRecorder
 
 
 class _GuardedSpan:
@@ -63,7 +64,7 @@ def append_metadata(
     propagators.inject(append_metadata, metadata)
 
 
-def _make_future_done_callback(span, rpc_info):
+def _make_future_done_callback(span, rpc_info, client_info, metrics_recorder):
     def callback(response_future):
         with span:
             code = response_future.code()
@@ -72,28 +73,45 @@ def callback(response_future):
                 return
             response = response_future.result()
             rpc_info.response = response
+            if "ByteSize" in dir(response):
+                metrics_recorder.record_bytes_in(
+                    response.ByteSize(), client_info.full_method
+                )
 
     return callback
 
 
 class OpenTelemetryClientInterceptor(
     grpcext.UnaryClientInterceptor, grpcext.StreamClientInterceptor
 ):
-    def __init__(self, tracer):
+    def __init__(self, tracer, exporter, interval):
         self._tracer = tracer
 
+        self._meter = None
+        if exporter and interval:
+            self._meter = metrics.get_meter(__name__)
+            self.controller = PushController(
+                meter=self._meter, exporter=exporter, interval=interval
+            )
+        self._metrics_recorder = TimedMetricRecorder(self._meter, "client")
+
     def _start_span(self, method):
         return self._tracer.start_as_current_span(
             name=method, kind=trace.SpanKind.CLIENT
         )
 
     # pylint:disable=no-self-use
-    def _trace_result(self, guarded_span, rpc_info, result):
+    def _trace_result(self, guarded_span, rpc_info, result, client_info):
         # If the RPC is called asynchronously, release the guard and add a
         # callback so that the span can be finished once the future is done.
         if isinstance(result, grpc.Future):
             result.add_done_callback(
-                _make_future_done_callback(guarded_span.release(), rpc_info)
+                _make_future_done_callback(
+                    guarded_span.release(),
+                    rpc_info,
+                    client_info,
+                    self._metrics_recorder,
+                )
             )
             return result
         response = result
@@ -104,37 +122,62 @@ def _trace_result(self, guarded_span, rpc_info, result):
         if isinstance(result, tuple):
             response = result[0]
         rpc_info.response = response
+
+        if "ByteSize" in dir(response):
+            self._metrics_recorder.record_bytes_in(
+                response.ByteSize(), client_info.full_method
+            )
         return result
 
     def _start_guarded_span(self, *args, **kwargs):
         return _GuardedSpan(self._start_span(*args, **kwargs))
 
+    def _bytes_out_iterator_wrapper(self, iterator, client_info):
+        for request in iterator:
+            if "ByteSize" in dir(request):
+                self._metrics_recorder.record_bytes_out(
+                    request.ByteSize(), client_info.full_method
+                )
+            yield request
+
     def intercept_unary(self, request, metadata, client_info, invoker):
         if not metadata:
             mutable_metadata = OrderedDict()
         else:
             mutable_metadata = OrderedDict(metadata)
 
         with self._start_guarded_span(client_info.full_method) as guarded_span:
-            _inject_span_context(mutable_metadata)
-            metadata = tuple(mutable_metadata.items())
-
-            rpc_info = RpcInfo(
-                full_method=client_info.full_method,
-                metadata=metadata,
-                timeout=client_info.timeout,
-                request=request,
-            )
-
-            try:
-                result = invoker(request, metadata)
-            except grpc.RpcError as exc:
-                guarded_span.generated_span.set_status(
-                    Status(StatusCanonicalCode(exc.code().value[0]))
+            with self._metrics_recorder.record_latency(
+                client_info.full_method
+            ):
+                _inject_span_context(mutable_metadata)
+                metadata = tuple(mutable_metadata.items())
+
+                # If protobuf is used, we can record the bytes in/out. Otherwise, we have no way
+                # to get the size of the request/response properly, so don't record anything
+                if "ByteSize" in dir(request):
+                    self._metrics_recorder.record_bytes_out(
+                        request.ByteSize(), client_info.full_method
+                    )
+
+                rpc_info = RpcInfo(
+                    full_method=client_info.full_method,
+                    metadata=metadata,
+                    timeout=client_info.timeout,
+                    request=request,
                 )
-                raise
 
-            return self._trace_result(guarded_span, rpc_info, result)
+                try:
+                    result = invoker(request, metadata)
+                except grpc.RpcError as exc:
+                    guarded_span.generated_span.set_status(
+                        Status(StatusCanonicalCode(exc.code().value[0]))
+                    )
+                    raise
+
+                return self._trace_result(
+                    guarded_span, rpc_info, result, client_info
+                )
 
     # For RPCs that stream responses, the result can be a generator. To record
     # the span across the generated responses and detect any errors, we wrap
@@ -148,25 +191,44 @@ def _intercept_server_stream(
             mutable_metadata = OrderedDict(metadata)
 
         with self._start_span(client_info.full_method) as span:
-            _inject_span_context(mutable_metadata)
-            metadata = tuple(mutable_metadata.items())
-            rpc_info = RpcInfo(
-                full_method=client_info.full_method,
-                metadata=metadata,
-                timeout=client_info.timeout,
-            )
-            if client_info.is_client_stream:
-                rpc_info.request = request_or_iterator
-
-            try:
-                result = invoker(request_or_iterator, metadata)
-                for response in result:
-                    yield response
-            except grpc.RpcError as exc:
-                span.set_status(
-                    Status(StatusCanonicalCode(exc.code().value[0]))
+            with self._metrics_recorder.record_latency(
+                client_info.full_method
+            ):
+                _inject_span_context(mutable_metadata)
+                metadata = tuple(mutable_metadata.items())
+                rpc_info = RpcInfo(
+                    full_method=client_info.full_method,
+                    metadata=metadata,
+                    timeout=client_info.timeout,
                 )
-                raise
+
+                if client_info.is_client_stream:
+                    rpc_info.request = request_or_iterator
+                    request_or_iterator = self._bytes_out_iterator_wrapper(
+                        request_or_iterator, client_info
+                    )
+                else:
+                    if "ByteSize" in dir(request_or_iterator):
+                        self._metrics_recorder.record_bytes_out(
+                            request_or_iterator.ByteSize(),
+                            client_info.full_method,
+                        )
+
+                try:
+                    result = invoker(request_or_iterator, metadata)
+
+                    # Rewrap the result stream into a generator, and record the bytes received
+                    for response in result:
+                        if "ByteSize" in dir(response):
+                            self._metrics_recorder.record_bytes_in(
+                                response.ByteSize(), client_info.full_method
+                            )
+                        yield response
+                except grpc.RpcError as exc:
+                    span.set_status(
+                        Status(StatusCanonicalCode(exc.code().value[0]))
+                    )
+                    raise
 
     def intercept_stream(
         self, request_or_iterator, metadata, client_info, invoker
@@ -182,21 +244,32 @@ def intercept_stream(
             mutable_metadata = OrderedDict(metadata)
 
         with self._start_guarded_span(client_info.full_method) as guarded_span:
-            _inject_span_context(mutable_metadata)
-            metadata = tuple(mutable_metadata.items())
-            rpc_info = RpcInfo(
-                full_method=client_info.full_method,
-                metadata=metadata,
-                timeout=client_info.timeout,
-                request=request_or_iterator,
-            )
+            with self._metrics_recorder.record_latency(
+                client_info.full_method
+            ):
+                _inject_span_context(mutable_metadata)
+                metadata = tuple(mutable_metadata.items())
+                rpc_info = RpcInfo(
+                    full_method=client_info.full_method,
+                    metadata=metadata,
+                    timeout=client_info.timeout,
+                    request=request_or_iterator,
+                )
+
+                rpc_info.request = request_or_iterator
 
-            try:
-                result = invoker(request_or_iterator, metadata)
-            except grpc.RpcError as exc:
-                guarded_span.generated_span.set_status(
-                    Status(StatusCanonicalCode(exc.code().value[0]))
+                request_or_iterator = self._bytes_out_iterator_wrapper(
+                    request_or_iterator, client_info
                 )
-                raise
 
-            return self._trace_result(guarded_span, rpc_info, result)
+                try:
+                    result = invoker(request_or_iterator, metadata)
+                except grpc.RpcError as exc:
+                    guarded_span.generated_span.set_status(
+                        Status(StatusCanonicalCode(exc.code().value[0]))
+                    )
+                    raise
+
+                return self._trace_result(
+                    guarded_span, rpc_info, result, client_info
+                )