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

CHANGELOG.md

@@ -9,7 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- `opentelemetry-instrumentation-redis` Add `sanitize_query` config option to allow query sanitization.  ([#1572](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1572))
+- grpc instrumentation uses official grpc.*ClientInterceptor-interfaces
+  ([#1583](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1583))
+- `opentelemetry-instrumentation-redis` Add `sanitize_query` config option to allow query sanitization.
+  ([#1572](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1572))
 - `opentelemetry-instrumentation-celery` Record exceptions as events on the span.
   ([#1573](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1573))
 - Add metric instrumentation for urllib

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

@@ -59,6 +59,18 @@ def run():
         logging.basicConfig()
         run()
 
+You can also add the interceptors manually, rather than using
+:py:class:`~opentelemetry.instrumentation.grpc.GrpcInstrumentorClient`:
+
+.. code-block:: python
+
+    from opentelemetry.instrumentation.grpc import client_interceptors
+
+    channel = grpc.intercept_channel(
+        grpc.insecure_channel("localhost:50051"),
+        *client_interceptors()
+    )
+
 Usage Server
 ------------
 .. code-block:: python
@@ -285,7 +297,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
@@ -469,11 +480,10 @@ def _uninstrument(self, **kwargs):
     def wrapper_fn(self, original_func, instance, args, kwargs):
         channel = original_func(*args, **kwargs)
         tracer_provider = kwargs.get("tracer_provider")
-        return intercept_channel(
+        return grpc.intercept_channel(
             channel,
-            client_interceptor(
-                tracer_provider=tracer_provider,
-                filter_=self._filter,
+            *client_interceptors(
+                tracer_provider=tracer_provider, filter_=self._filter
             ),
         )
 
@@ -541,8 +551,8 @@ def _uninstrument(self, **kwargs):
         grpc.aio.secure_channel = self._original_secure
 
 
-def client_interceptor(tracer_provider=None, filter_=None):
-    """Create a gRPC client channel interceptor.
+def client_interceptors(tracer_provider=None, filter_=None):
+    """Create gRPC client channel interceptors.
 
     Args:
         tracer: The tracer to use to create client-side spans.
@@ -552,13 +562,18 @@ def client_interceptor(tracer_provider=None, filter_=None):
                  all requests.
 
     Returns:
-        An invocation-side interceptor object.
+        A list of invocation-side interceptor objects.
     """
     from . import _client
 
     tracer = trace.get_tracer(__name__, __version__, tracer_provider)
 
-    return _client.OpenTelemetryClientInterceptor(tracer, filter_=filter_)
+    return [
+        _client.UnaryUnaryClientInterceptor(tracer, filter_=filter_),
+        _client.UnaryStreamClientInterceptor(tracer, filter_=filter_),
+        _client.StreamUnaryClientInterceptor(tracer, filter_=filter_),
+        _client.StreamStreamClientInterceptor(tracer, filter_=filter_),
+    ]
 
 
 def server_interceptor(tracer_provider=None, filter_=None):

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

@@ -18,14 +18,11 @@
 import grpc
 from grpc.aio import ClientCallDetails
 
-from opentelemetry import context
-from opentelemetry.instrumentation.grpc._client import (
-    OpenTelemetryClientInterceptor,
-    _carrier_setter,
-)
+from opentelemetry import context, trace
+from opentelemetry.instrumentation.grpc._client import _carrier_setter
 from opentelemetry.instrumentation.utils import _SUPPRESS_INSTRUMENTATION_KEY
 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
 
 
@@ -49,7 +46,12 @@ def callback(call):
     return callback
 
 
-class _BaseAioClientInterceptor(OpenTelemetryClientInterceptor):
+class _BaseAioClientInterceptor:
+
+    def __init__(self, tracer, filter_=None):
+        self._tracer = tracer
+        self._filter = filter_
+
     @staticmethod
     def propagate_trace_in_details(client_call_details):
         metadata = client_call_details.metadata
@@ -99,6 +101,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()