Fix GRPC context propagation (#392)

CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased](https://github.com/open-telemetry/opentelemetry-python-contrib/compare/v0.19b0...HEAD)
 
 ### Changed
+- GRPC instrumentation now correctly injects trace context into outgoing requests.
+  ([#392](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/39))
 - Publish `opentelemetry-propagator-ot-trace` package as a part of the release process
   ([#387](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/387))
 

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

@@ -28,6 +28,7 @@ from opentelemetry import trace
 from opentelemetry.instrumentation.grpc import grpcext
 from opentelemetry.instrumentation.grpc._utilities import RpcInfo
 from opentelemetry.propagate import inject
+from opentelemetry.propagators.textmap import Setter
 from opentelemetry.trace.status import Status, StatusCode
 
 
@@ -52,15 +53,16 @@ class _GuardedSpan:
         return self.span
 
 
-def _inject_span_context(metadata: MutableMapping[str, str]) -> None:
-    # pylint:disable=unused-argument
-    def append_metadata(
-        carrier: MutableMapping[str, str], key: str, value: str
-    ):
-        metadata[key] = value
+class _CarrierSetter(Setter):
+    """We use a custom setter in order to be able to lower case
+    keys as is required by grpc.
+    """
 
-    # Inject current active span from the context
-    inject(append_metadata, metadata)
+    def set(self, carrier: MutableMapping[str, str], key: str, value: str):
+        carrier[key.lower()] = value
+
+
+_carrier_setter = _CarrierSetter()
 
 
 def _make_future_done_callback(span, rpc_info):
@@ -125,7 +127,7 @@ class OpenTelemetryClientInterceptor(
             mutable_metadata = OrderedDict(metadata)
 
         with self._start_guarded_span(client_info.full_method) as guarded_span:
-            _inject_span_context(mutable_metadata)
+            inject(mutable_metadata, setter=_carrier_setter)
             metadata = tuple(mutable_metadata.items())
 
             rpc_info = RpcInfo(
@@ -160,7 +162,7 @@ class OpenTelemetryClientInterceptor(
             mutable_metadata = OrderedDict(metadata)
 
         with self._start_span(client_info.full_method) as span:
-            _inject_span_context(mutable_metadata)
+            inject(mutable_metadata, setter=_carrier_setter)
             metadata = tuple(mutable_metadata.items())
             rpc_info = RpcInfo(
                 full_method=client_info.full_method,
@@ -195,7 +197,7 @@ class OpenTelemetryClientInterceptor(
             mutable_metadata = OrderedDict(metadata)
 
         with self._start_guarded_span(client_info.full_method) as guarded_span:
-            _inject_span_context(mutable_metadata)
+            inject(mutable_metadata, setter=_carrier_setter)
             metadata = tuple(mutable_metadata.items())
             rpc_info = RpcInfo(
                 full_method=client_info.full_method,

instrumentation/opentelemetry-instrumentation-grpc/tests/test_client_interceptor.py

@@ -20,6 +20,14 @@ from tests.protobuf import (  # pylint: disable=no-name-in-module
 import opentelemetry.instrumentation.grpc
 from opentelemetry import trace
 from opentelemetry.instrumentation.grpc import GrpcInstrumentorClient
+from opentelemetry.instrumentation.grpc._client import (
+    OpenTelemetryClientInterceptor,
+)
+from opentelemetry.instrumentation.grpc.grpcext._interceptor import (
+    _UnaryClientInfo,
+)
+from opentelemetry.propagate import get_global_textmap, set_global_textmap
+from opentelemetry.test.mock_textmap import MockTextMapPropagator
 from opentelemetry.test.test_base import TestBase
 
 from ._client import (
@@ -29,6 +37,7 @@ from ._client import (
     simple_method,
 )
 from ._server import create_test_server
+from .protobuf.test_server_pb2 import Request
 
 
 class TestClientProto(TestBase):
@@ -187,3 +196,40 @@ class TestClientProto(TestBase):
         self.assertIs(
             span.status.status_code, trace.StatusCode.ERROR,
         )
+
+    def test_client_interceptor_trace_context_propagation(
+        self,
+    ):  # pylint: disable=no-self-use
+        """ensure that client interceptor correctly inject trace context into all outgoing requests."""
+        previous_propagator = get_global_textmap()
+        try:
+            set_global_textmap(MockTextMapPropagator())
+            interceptor = OpenTelemetryClientInterceptor(
+                trace._DefaultTracer()
+            )
+
+            carrier = tuple()
+
+            def invoker(request, metadata):
+                nonlocal carrier
+                carrier = metadata
+                return {}
+
+            request = Request(client_id=1, request_data="data")
+            interceptor.intercept_unary(
+                request,
+                {},
+                _UnaryClientInfo(
+                    full_method="/GRPCTestServer/SimpleMethod", timeout=None
+                ),
+                invoker=invoker,
+            )
+
+            assert len(carrier) == 2
+            assert carrier[0][0] == "mock-traceid"
+            assert carrier[0][1] == "0"
+            assert carrier[1][0] == "mock-spanid"
+            assert carrier[1][1] == "0"
+
+        finally:
+            set_global_textmap(previous_propagator)