Conformity of client intereptor(s) with the official grpc.*ClientInterceptor-interfaces

Signed-off-by: Jason Parraga <[email protected]>

Author: Sovietaced

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/__init__.py

@@ -286,7 +286,6 @@ async def serve():
     negate,
     service_name,
 )
-from opentelemetry.instrumentation.grpc.grpcext import intercept_channel
 from opentelemetry.instrumentation.grpc.package import _instruments
 from opentelemetry.instrumentation.grpc.version import __version__
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
@@ -339,7 +338,8 @@ def server(*args, **kwargs):
                 kwargs["interceptors"].insert(
                     0,
                     server_interceptor(
-                        tracer_provider=tracer_provider, filter_=self._filter
+                        tracer_provider=tracer_provider,
+                        filter_=self._filter,
                     ),
                 )
             else:
@@ -476,9 +476,10 @@ def wrapper_fn(self, original_func, instance, args, kwargs):
         tracer_provider = kwargs.get("tracer_provider")
         request_hook = self._request_hook
         response_hook = self._response_hook
-        return intercept_channel(
+
+        return grpc.intercept_channel(
             channel,
-            client_interceptor(
+            *client_interceptors(
                 tracer_provider=tracer_provider,
                 filter_=self._filter,
                 request_hook=request_hook,
@@ -560,10 +561,10 @@ def _uninstrument(self, **kwargs):
         grpc.aio.secure_channel = self._original_secure
 
 
-def client_interceptor(
+def client_interceptors(
     tracer_provider=None, filter_=None, request_hook=None, response_hook=None
 ):
-    """Create a gRPC client channel interceptor.
+    """Create gRPC client channel interceptors.
 
     Args:
         tracer: The tracer to use to create client-side spans.
@@ -573,7 +574,7 @@ def client_interceptor(
                  all requests.
 
     Returns:
-        An invocation-side interceptor object.
+        A list of invocation-side interceptor objects.
     """
     from . import _client
 
@@ -584,12 +585,32 @@ def client_interceptor(
         schema_url="https://opentelemetry.io/schemas/1.11.0",
     )
 
-    return _client.OpenTelemetryClientInterceptor(
-        tracer,
-        filter_=filter_,
-        request_hook=request_hook,
-        response_hook=response_hook,
-    )
+    return [
+        _client.UnaryUnaryClientInterceptor(
+            tracer,
+            filter_=filter_,
+            request_hook=request_hook,
+            response_hook=response_hook,
+        ),
+        _client.UnaryStreamClientInterceptor(
+            tracer,
+            filter_=filter_,
+            request_hook=request_hook,
+            response_hook=response_hook,
+        ),
+        _client.StreamUnaryClientInterceptor(
+            tracer,
+            filter_=filter_,
+            request_hook=request_hook,
+            response_hook=response_hook,
+        ),
+        _client.StreamStreamClientInterceptor(
+            tracer,
+            filter_=filter_,
+            request_hook=request_hook,
+            response_hook=response_hook,
+        ),
+    ]
 
 
 def server_interceptor(tracer_provider=None, filter_=None):

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_client.py

@@ -18,15 +18,15 @@
 import grpc
 from grpc.aio import ClientCallDetails, Metadata
 
-from opentelemetry.instrumentation.grpc._client import (
-    OpenTelemetryClientInterceptor,
-    _carrier_setter,
-)
+from opentelemetry import trace
+from opentelemetry.instrumentation.grpc._client import _carrier_setter
 from opentelemetry.instrumentation.utils import is_instrumentation_enabled
 from opentelemetry.propagate import inject
-from opentelemetry.semconv.trace import SpanAttributes
+from opentelemetry.semconv.trace import RpcSystemValues, SpanAttributes
 from opentelemetry.trace.status import Status, StatusCode
 
+from ._client import _safe_invoke
+
 logger = logging.getLogger(__name__)
 
 
@@ -52,7 +52,25 @@ def callback(call):
     return callback
 
 
-class _BaseAioClientInterceptor(OpenTelemetryClientInterceptor):
+class _BaseAioClientInterceptor:
+    def __init__(
+        self, tracer, filter_=None, request_hook=None, response_hook=None
+    ):
+        self._tracer = tracer
+        self._filter = filter_
+        self._request_hook = request_hook
+        self._response_hook = response_hook
+
+    def _call_request_hook(self, span, request):
+        if not callable(self._request_hook):
+            return
+        _safe_invoke(self._request_hook, span, request)
+
+    def _call_response_hook(self, span, response):
+        if not callable(self._response_hook):
+            return
+        _safe_invoke(self._response_hook, span, response)
+
     @staticmethod
     def propagate_trace_in_details(client_call_details: ClientCallDetails):
         metadata = client_call_details.metadata
@@ -101,6 +119,22 @@ def _start_interceptor_span(self, method):
             set_status_on_exception=False,
         )
 
+    def _start_span(self, method, **kwargs):
+        service, meth = method.lstrip("/").split("/", 1)
+        attributes = {
+            SpanAttributes.RPC_SYSTEM: RpcSystemValues.GRPC.value,
+            SpanAttributes.RPC_SERVICE: service,
+            SpanAttributes.RPC_METHOD: meth,
+            SpanAttributes.RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[0],
+        }
+
+        return self._tracer.start_as_current_span(
+            name=method,
+            kind=trace.SpanKind.CLIENT,
+            attributes=attributes,
+            **kwargs,
+        )
+
     async def _wrap_unary_response(self, continuation, span):
         try:
             call = await continuation()