Converted TextMap propagator getter to a class and added keys method

docs/conf.py

@@ -102,6 +102,7 @@
     ("py:class", "ObjectProxy"),
     # TODO: Understand why sphinx is not able to find this local class
     ("py:class", "opentelemetry.trace.propagation.textmap.TextMapPropagator",),
+    ("py:class", "opentelemetry.trace.propagation.textmap.DictGetter",),
     (
         "any",
         "opentelemetry.trace.propagation.textmap.TextMapPropagator.extract",

exporter/opentelemetry-exporter-datadog/src/opentelemetry/exporter/datadog/propagator.py

@@ -39,25 +39,23 @@ class DatadogFormat(TextMapPropagator):
 
     def extract(
         self,
-        get_from_carrier: Getter[TextMapPropagatorT],
+        getter: Getter[TextMapPropagatorT],
         carrier: TextMapPropagatorT,
         context: typing.Optional[Context] = None,
     ) -> Context:
         trace_id = extract_first_element(
-            get_from_carrier(carrier, self.TRACE_ID_KEY)
+            getter.get(carrier, self.TRACE_ID_KEY)
         )
 
         span_id = extract_first_element(
-            get_from_carrier(carrier, self.PARENT_ID_KEY)
+            getter.get(carrier, self.PARENT_ID_KEY)
         )
 
         sampled = extract_first_element(
-            get_from_carrier(carrier, self.SAMPLING_PRIORITY_KEY)
+            getter.get(carrier, self.SAMPLING_PRIORITY_KEY)
         )
 
-        origin = extract_first_element(
-            get_from_carrier(carrier, self.ORIGIN_KEY)
-        )
+        origin = extract_first_element(getter.get(carrier, self.ORIGIN_KEY))
 
         trace_flags = trace.TraceFlags()
         if sampled and int(sampled) in (

exporter/opentelemetry-exporter-datadog/tests/test_datadog_format.py

@@ -18,13 +18,11 @@
 from opentelemetry.exporter.datadog import constants, propagator
 from opentelemetry.sdk import trace
 from opentelemetry.trace import get_current_span, set_span_in_context
+from opentelemetry.trace.propagation.textmap import DictGetter
 
 FORMAT = propagator.DatadogFormat()
 
-
-def get_as_list(dict_object, key):
-    value = dict_object.get(key)
-    return [value] if value is not None else []
+carrier_getter = DictGetter()
 
 
 class TestDatadogFormat(unittest.TestCase):
@@ -45,7 +43,7 @@ def test_malformed_headers(self):
         malformed_parent_id_key = FORMAT.PARENT_ID_KEY + "-x"
         context = get_current_span(
             FORMAT.extract(
-                get_as_list,
+                carrier_getter,
                 {
                     malformed_trace_id_key: self.serialized_trace_id,
                     malformed_parent_id_key: self.serialized_parent_id,
@@ -63,7 +61,7 @@ def test_missing_trace_id(self):
             FORMAT.PARENT_ID_KEY: self.serialized_parent_id,
         }
 
-        ctx = FORMAT.extract(get_as_list, carrier)
+        ctx = FORMAT.extract(carrier_getter, carrier)
         span_context = get_current_span(ctx).get_span_context()
         self.assertEqual(span_context.trace_id, trace_api.INVALID_TRACE_ID)
 
@@ -73,15 +71,15 @@ def test_missing_parent_id(self):
             FORMAT.TRACE_ID_KEY: self.serialized_trace_id,
         }
 
-        ctx = FORMAT.extract(get_as_list, carrier)
+        ctx = FORMAT.extract(carrier_getter, carrier)
         span_context = get_current_span(ctx).get_span_context()
         self.assertEqual(span_context.span_id, trace_api.INVALID_SPAN_ID)
 
     def test_context_propagation(self):
         """Test the propagation of Datadog headers."""
         parent_span_context = get_current_span(
             FORMAT.extract(
-                get_as_list,
+                carrier_getter,
                 {
                     FORMAT.TRACE_ID_KEY: self.serialized_trace_id,
                     FORMAT.PARENT_ID_KEY: self.serialized_parent_id,
@@ -138,7 +136,7 @@ def test_sampling_priority_auto_reject(self):
         """Test sampling priority rejected."""
         parent_span_context = get_current_span(
             FORMAT.extract(
-                get_as_list,
+                carrier_getter,
                 {
                     FORMAT.TRACE_ID_KEY: self.serialized_trace_id,
                     FORMAT.PARENT_ID_KEY: self.serialized_parent_id,

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

@@ -29,21 +29,31 @@
 from opentelemetry import context, propagators, trace
 from opentelemetry.instrumentation.asgi.version import __version__  # noqa
 from opentelemetry.instrumentation.utils import http_status_to_canonical_code
+from opentelemetry.trace.propagation.textmap import DictGetter
 from opentelemetry.trace.status import Status, StatusCanonicalCode
 
 
-def get_header_from_scope(scope: dict, header_name: str) -> typing.List[str]:
-    """Retrieve a HTTP header value from the ASGI scope.
+class CarrierGetter(DictGetter):
+    def get(self, carrier: dict, key: str) -> typing.List[str]:
+        """Getter implementation to retrieve a HTTP header value from the ASGI
+        scope.
 
-    Returns:
-        A list with a single string with the header value if it exists, else an empty list.
-    """
-    headers = scope.get("headers")
-    return [
-        value.decode("utf8")
-        for (key, value) in headers
-        if key.decode("utf8") == header_name
-    ]
+        Args:
+            carrier: ASGI scope object
+            key: header name in scope
+        Returns:
+            A list with a single string with the header value if it exists,
+             else an empty list.
+        """
+        headers = carrier.get("headers")
+        return [
+            _value.decode("utf8")
+            for (_key, _value) in headers
+            if _key.decode("utf8") == key
+        ]
+
+
+carrier_getter = CarrierGetter()
 
 
 def collect_request_attributes(scope):
@@ -72,10 +82,10 @@ def collect_request_attributes(scope):
     http_method = scope.get("method")
     if http_method:
         result["http.method"] = http_method
-    http_host_value = ",".join(get_header_from_scope(scope, "host"))
+    http_host_value = ",".join(carrier_getter.get(scope, "host"))
     if http_host_value:
         result["http.server_name"] = http_host_value
-    http_user_agent = get_header_from_scope(scope, "user-agent")
+    http_user_agent = carrier_getter.get(scope, "user-agent")
     if len(http_user_agent) > 0:
         result["http.user_agent"] = http_user_agent[0]
 
@@ -154,9 +164,7 @@ async def __call__(self, scope, receive, send):
         if scope["type"] not in ("http", "websocket"):
             return await self.app(scope, receive, send)
 
-        token = context.attach(
-            propagators.extract(get_header_from_scope, scope)
-        )
+        token = context.attach(propagators.extract(carrier_getter, scope))
         span_name, additional_attributes = self.span_details_callback(scope)
 
         try:

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

@@ -67,7 +67,7 @@ def add(x, y):
 from opentelemetry.instrumentation.celery import utils
 from opentelemetry.instrumentation.celery.version import __version__
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
-from opentelemetry.trace.propagation import get_current_span
+from opentelemetry.trace.propagation.textmap import DictGetter
 from opentelemetry.trace.status import Status, StatusCanonicalCode
 
 logger = logging.getLogger(__name__)
@@ -84,6 +84,20 @@ def add(x, y):
 _MESSAGE_ID_ATTRIBUTE_NAME = "messaging.message_id"
 
 
+class CarrierGetter(DictGetter):
+    def get(self, carrier, key):
+        value = getattr(carrier, key, [])
+        if isinstance(value, str) or not isinstance(value, Iterable):
+            value = (value,)
+        return value
+
+    def keys(self, carrier):
+        return []
+
+
+carrier_getter = CarrierGetter()
+
+
 class CeleryInstrumentor(BaseInstrumentor):
     def _instrument(self, **kwargs):
         tracer_provider = kwargs.get("tracer_provider")
@@ -118,7 +132,7 @@ def _trace_prerun(self, *args, **kwargs):
             return
 
         request = task.request
-        tracectx = propagators.extract(carrier_extractor, request) or None
+        tracectx = propagators.extract(carrier_getter, request) or None
 
         logger.debug("prerun signal start task_id=%s", task_id)
 
@@ -247,10 +261,3 @@ def _trace_retry(*args, **kwargs):
         # Use `str(reason)` instead of `reason.message` in case we get
         # something that isn't an `Exception`
         span.set_attribute(_TASK_RETRY_REASON_KEY, str(reason))
-
-
-def carrier_extractor(carrier, key):
-    value = getattr(carrier, key, [])
-    if isinstance(value, str) or not isinstance(value, Iterable):
-        value = (value,)
-    return value

instrumentation/opentelemetry-instrumentation-django/src/opentelemetry/instrumentation/django/middleware.py

@@ -23,8 +23,8 @@
 from opentelemetry.instrumentation.utils import extract_attributes_from_object
 from opentelemetry.instrumentation.wsgi import (
     add_response_attributes,
+    carrier_getter,
     collect_request_attributes,
-    get_header_from_environ,
 )
 from opentelemetry.propagators import extract
 from opentelemetry.trace import SpanKind, get_tracer
@@ -125,7 +125,7 @@ def process_request(self, request):
 
         environ = request.META
 
-        token = attach(extract(get_header_from_environ, environ))
+        token = attach(extract(carrier_getter, environ))
 
         tracer = get_tracer(__name__, __version__)
 

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

@@ -115,7 +115,7 @@ def __call__(self, env, start_response):
         start_time = time_ns()
 
         token = context.attach(
-            propagators.extract(otel_wsgi.get_header_from_environ, env)
+            propagators.extract(otel_wsgi.carrier_getter, env)
         )
         span = self._tracer.start_span(
             otel_wsgi.get_default_span_name(env),

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

@@ -118,7 +118,7 @@ def _before_request():
     if span_name is None:
         span_name = otel_wsgi.get_default_span_name(environ)
     token = context.attach(
-        propagators.extract(otel_wsgi.get_header_from_environ, environ)
+        propagators.extract(otel_wsgi.carrier_getter, environ)
     )
 
     tracer = trace.get_tracer(__name__, __version__)

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

@@ -24,12 +24,12 @@
 """
 
 from contextlib import contextmanager
-from typing import List
 
 import grpc
 
 from opentelemetry import propagators, trace
 from opentelemetry.context import attach, detach
+from opentelemetry.trace.propagation.textmap import DictGetter
 
 from . import grpcext
 from ._utilities import RpcInfo
@@ -115,19 +115,15 @@ class OpenTelemetryServerInterceptor(
 ):
     def __init__(self, tracer):
         self._tracer = tracer
+        self._carrier_getter = DictGetter()
 
     @contextmanager
     # pylint:disable=no-self-use
     def _set_remote_context(self, servicer_context):
         metadata = servicer_context.invocation_metadata()
         if metadata:
             md_dict = {md.key: md.value for md in metadata}
-
-            def get_from_grpc_metadata(metadata, key) -> List[str]:
-                return [md_dict[key]] if key in md_dict else []
-
-            # Update the context with the traceparent from the RPC metadata.
-            ctx = propagators.extract(get_from_grpc_metadata, metadata)
+            ctx = propagators.extract(self._carrier_getter, md_dict)
             token = attach(ctx)
             try:
                 yield

instrumentation/opentelemetry-instrumentation-opentracing-shim/src/opentelemetry/instrumentation/opentracing_shim/__init__.py

@@ -112,6 +112,7 @@
     get_current_span,
     set_span_in_context,
 )
+from opentelemetry.trace.propagation.textmap import DictGetter
 from opentelemetry.util.types import Attributes
 
 ValueT = TypeVar("ValueT", int, float, bool, str)
@@ -527,6 +528,7 @@ def __init__(self, tracer: OtelTracer):
             Format.TEXT_MAP,
             Format.HTTP_HEADERS,
         )
+        self._carrier_getter = DictGetter()
 
     def unwrap(self):
         """Returns the :class:`opentelemetry.trace.Tracer` object that is
@@ -710,12 +712,8 @@ def extract(self, format: object, carrier: object):
         if format not in self._supported_formats:
             raise UnsupportedFormatException
 
-        def get_as_list(dict_object, key):
-            value = dict_object.get(key)
-            return [value] if value is not None else []
-
         propagator = propagators.get_global_textmap()
-        ctx = propagator.extract(get_as_list, carrier)
+        ctx = propagator.extract(self._carrier_getter, carrier)
         span = get_current_span(ctx)
         if span is not None:
             otel_context = span.get_span_context()

instrumentation/opentelemetry-instrumentation-pyramid/src/opentelemetry/instrumentation/pyramid/callbacks.py

@@ -70,7 +70,7 @@ def _before_traversal(event):
     start_time = environ.get(_ENVIRON_STARTTIME_KEY)
 
     token = context.attach(
-        propagators.extract(otel_wsgi.get_header_from_environ, environ)
+        propagators.extract(otel_wsgi.carrier_getter, environ)
     )
     tracer = trace.get_tracer(__name__, __version__)
 

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

@@ -54,6 +54,7 @@ def get(self):
     http_status_to_canonical_code,
     unwrap,
 )
+from opentelemetry.trace.propagation.textmap import DictGetter
 from opentelemetry.trace.status import Status
 from opentelemetry.util import ExcludeList, time_ns
 
@@ -84,6 +85,8 @@ def get_traced_request_attrs():
 _excluded_urls = get_excluded_urls()
 _traced_attrs = get_traced_request_attrs()
 
+carrier_getter = DictGetter()
+
 
 class TornadoInstrumentor(BaseInstrumentor):
     patched_handlers = []
@@ -185,13 +188,6 @@ def _log_exception(tracer, func, handler, args, kwargs):
     return func(*args, **kwargs)
 
 
-def _get_header_from_request_headers(
-    headers: dict, header_name: str
-) -> typing.List[str]:
-    header = headers.get(header_name)
-    return [header] if header else []
-
-
 def _get_attributes_from_request(request):
     attrs = {
         "component": "tornado",
@@ -218,9 +214,7 @@ def _get_operation_name(handler, request):
 
 def _start_span(tracer, handler, start_time) -> _TraceContext:
     token = context.attach(
-        propagators.extract(
-            _get_header_from_request_headers, handler.request.headers,
-        )
+        propagators.extract(carrier_getter, handler.request.headers,)
     )
 
     span = tracer.start_span(