implement w3c and b3 context propagators

This commit implements new http propagators in other to be compatible with
OpenTelemetry instrumented apps:
- w3c: default context propagator in opentelemetry
- b3: another choice
tracer.configure() was extended to accept a new parameter, "http_propagator",
this is a factory function that will be used to get new instances of the
propagator to be used.

The following is an example of how to set b3 as the propagator:

...
from oteltrace.propagation.b3 import B3HTTPPropagator

oteltrace.tracer.configure(
    http_propagator=B3HTTPPropagator
)
...

Users of oteltrace-run can choose the propagator to be used by setting the
OTEL_TRACER_PROPAGATOR env variable to one of the  current supported values:
w3c, b3 and datadog.

Author: mauriciovasquezbernal

docs/advanced_usage.rst

@@ -26,7 +26,7 @@ To trace requests across hosts, the spans on the secondary hosts must be linked
 - On the server side, it means to read propagated attributes and set them to the active tracing context.
 - On the client side, it means to propagate the attributes, commonly as a header/metadata.
 
-`oteltrace` already provides default propagators but you can also implement your own.
+`oteltrace` already provides default propagators (``w3c``, ``b3`` and ``datadog``) but you can also implement your own.
 
 Web Frameworks
 ^^^^^^^^^^^^^^
@@ -69,8 +69,8 @@ on the other side, the metadata is retrieved and the trace can continue.
 To propagate the tracing information, HTTP headers are used to transmit the
 required metadata to piece together the trace.
 
-.. autoclass:: oteltrace.propagation.http.HTTPPropagator
-    :members:
+:func:`oteltrace.propagation.http.HTTPPropagator` returns an instance of the configured
+propagator.
 
 Custom
 ^^^^^^
@@ -477,6 +477,13 @@ The specific configuration for each type of exporter is defined by using the
 The text after ``OTEL_EXPORTER_OPTIONS_`` will be passed to
 ``OTEL_EXPORTER_FACTORY`` as kwargs.
 
+Propagator Configuration
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+``oteltrace-run`` supports different formats to distribute the trace context.
+The propagator used is defined by the ``OTEL_TRACER_PROPAGATOR`` env variable.
+Currently ``w3c`` (default), ``b3`` and ``datadog`` are supported.
+
 ``oteltrace-run`` respects a variety of common entrypoints for web applications:
 
 - ``oteltrace-run python my_app.py``

oteltrace/bootstrap/sitecustomize.py

@@ -9,6 +9,10 @@
 import logging
 import importlib
 
+from oteltrace.propagation.datadog import DatadogHTTPPropagator
+from oteltrace.propagation.w3c import W3CHTTPPropagator
+from oteltrace.propagation.b3 import B3HTTPPropagator
+
 from oteltrace import api_otel_exporter
 
 from oteltrace.utils.formats import asbool, get_env
@@ -116,6 +120,25 @@ def load_otel_exporter():
         return None
 
 
+OTEL_TRACER_PROPAGATOR = 'OTEL_TRACER_PROPAGATOR'
+OTEL_TRACER_PROPAGATOR_W3C = 'w3c'
+OTEL_TRACER_PROPAGATOR_B3 = 'b3'
+OTEL_TRACER_PROPAGATOR_DATADOG = 'datadog'
+OTEL_TRACER_PROPAGATOR_DEFAULT = OTEL_TRACER_PROPAGATOR_W3C
+
+OTEL_TRACER_PROPAGATOR_MAP = {
+    OTEL_TRACER_PROPAGATOR_W3C: W3CHTTPPropagator,
+    OTEL_TRACER_PROPAGATOR_B3: B3HTTPPropagator,
+    OTEL_TRACER_PROPAGATOR_DATADOG: DatadogHTTPPropagator,
+}
+
+
+def get_http_propagator_factory():
+    """Returns an http propagator factory based on set env variables"""
+    prop = os.getenv(OTEL_TRACER_PROPAGATOR, OTEL_TRACER_PROPAGATOR_DEFAULT)
+    return OTEL_TRACER_PROPAGATOR_MAP[prop.lower()]
+
+
 try:
     from oteltrace import tracer
     patch = True
@@ -137,6 +160,8 @@ def load_otel_exporter():
 
     opts['api'] = api_otel_exporter.APIOtel(exporter=load_otel_exporter())
 
+    opts['http_propagator'] = get_http_propagator_factory()
+
     if opts:
         tracer.configure(**opts)
 

oteltrace/propagation/b3.py

@@ -0,0 +1,82 @@
+from ..context import Context
+from ..ext import priority
+
+
+class B3HTTPPropagator:
+    """b3 compatible propagator"""
+
+    SINGLE_HEADER_KEY = 'b3'
+    TRACE_ID_KEY = 'x-b3-traceid'
+    SPAN_ID_KEY = 'x-b3-spanid'
+    SAMPLED_KEY = 'x-b3-sampled'
+    FLAGS_KEY = 'x-b3-flags'
+    _SAMPLE_PROPAGATE_VALUES = set(['1', 'True', 'true', 'd'])
+
+    _SAMPLING_PRIORITY_MAP = {
+        priority.USER_REJECT: '0',
+        priority.AUTO_REJECT: '0',
+        priority.AUTO_KEEP: '1',
+        priority.USER_KEEP: '1',
+    }
+
+    def inject(self, span_context, headers):
+        # TODO: what should be a default value?
+        sampled = '0'
+        if span_context.sampling_priority is not None:
+            sampled = self._SAMPLING_PRIORITY_MAP[span_context.sampling_priority]
+
+        headers[self.TRACE_ID_KEY] = format_trace_id(span_context.trace_id)
+        headers[self.SPAN_ID_KEY] = format_span_id(span_context.span_id)
+        headers[self.SAMPLED_KEY] = sampled
+
+    def extract(self, headers):
+        trace_id = '0'
+        span_id = '0'
+        sampled = '0'
+        flags = None
+
+        single_header = headers.get(self.SINGLE_HEADER_KEY)
+
+        if single_header:
+            # The b3 spec calls for the sampling state to be
+            # "deferred", which is unspecified. This concept does not
+            # translate to SpanContext, so we set it as recorded.
+            sampled = '1'
+            fields = single_header.split('-', 4)
+
+            if len(fields) == 1:
+                sampled = fields[0]
+            elif len(fields) == 2:
+                trace_id, span_id = fields
+            elif len(fields) == 3:
+                trace_id, span_id, sampled = fields
+            elif len(fields) == 4:
+                trace_id, span_id, sampled, _parent_span_id = fields
+            else:
+                return Context()
+        else:
+            trace_id = headers.get(self.TRACE_ID_KEY) or trace_id
+            span_id = headers.get(self.SPAN_ID_KEY) or span_id
+            sampled = headers.get(self.SAMPLED_KEY) or sampled
+            flags = headers.get(self.FLAGS_KEY) or flags
+
+        if sampled in self._SAMPLE_PROPAGATE_VALUES or flags == '1':
+            sampling_priority = priority.AUTO_KEEP
+        else:
+            sampling_priority = priority.AUTO_REJECT
+
+        return Context(
+            trace_id=int(trace_id, 16),
+            span_id=int(span_id, 16),
+            sampling_priority=sampling_priority,
+        )
+
+
+def format_trace_id(trace_id: int) -> str:
+    """Format the trace id according to b3 specification."""
+    return format(trace_id, '032x')
+
+
+def format_span_id(span_id: int) -> str:
+    """Format the span id according to b3 specification."""
+    return format(span_id, '016x')

oteltrace/propagation/datadog.py

@@ -0,0 +1,151 @@
+from ..context import Context
+from ..internal.logger import get_logger
+
+from .utils import get_wsgi_header
+
+log = get_logger(__name__)
+
+# HTTP headers one should set for distributed tracing.
+# These are cross-language (eg: Python, Go and other implementations should honor these)
+HTTP_HEADER_TRACE_ID = 'x-datadog-trace-id'
+HTTP_HEADER_PARENT_ID = 'x-datadog-parent-id'
+HTTP_HEADER_SAMPLING_PRIORITY = 'x-datadog-sampling-priority'
+HTTP_HEADER_ORIGIN = 'x-datadog-origin'
+
+
+# Note that due to WSGI spec we have to also check for uppercased and prefixed
+# versions of these headers
+POSSIBLE_HTTP_HEADER_TRACE_IDS = frozenset(
+    [HTTP_HEADER_TRACE_ID, get_wsgi_header(HTTP_HEADER_TRACE_ID)]
+)
+POSSIBLE_HTTP_HEADER_PARENT_IDS = frozenset(
+    [HTTP_HEADER_PARENT_ID, get_wsgi_header(HTTP_HEADER_PARENT_ID)]
+)
+POSSIBLE_HTTP_HEADER_SAMPLING_PRIORITIES = frozenset(
+    [HTTP_HEADER_SAMPLING_PRIORITY, get_wsgi_header(HTTP_HEADER_SAMPLING_PRIORITY)]
+)
+POSSIBLE_HTTP_HEADER_ORIGIN = frozenset(
+    [HTTP_HEADER_ORIGIN, get_wsgi_header(HTTP_HEADER_ORIGIN)]
+)
+
+
+class DatadogHTTPPropagator(object):
+    """A HTTP Propagator using HTTP headers as carrier."""
+
+    def inject(self, span_context, headers):
+        """Inject Context attributes that have to be propagated as HTTP headers.
+
+        Here is an example using `requests`::
+
+            import requests
+            from oteltrace.propagation.http import DatadogHTTPPropagator
+
+            def parent_call():
+                with tracer.trace('parent_span') as span:
+                    headers = {}
+                    propagator = DatadogHTTPPropagator()
+                    propagator.inject(span.context, headers)
+                    url = '<some RPC endpoint>'
+                    r = requests.get(url, headers=headers)
+
+        :param Context span_context: Span context to propagate.
+        :param dict headers: HTTP headers to extend with tracing attributes.
+        """
+        headers[HTTP_HEADER_TRACE_ID] = str(span_context.trace_id)
+        headers[HTTP_HEADER_PARENT_ID] = str(span_context.span_id)
+        sampling_priority = span_context.sampling_priority
+        # Propagate priority only if defined
+        if sampling_priority is not None:
+            headers[HTTP_HEADER_SAMPLING_PRIORITY] = str(span_context.sampling_priority)
+        # Propagate origin only if defined
+        if span_context._otel_origin is not None:
+            headers[HTTP_HEADER_ORIGIN] = str(span_context._otel_origin)
+
+    @staticmethod
+    def extract_header_value(possible_header_names, headers, default=None):
+        for header, value in headers.items():
+            for header_name in possible_header_names:
+                if header.lower() == header_name.lower():
+                    return value
+
+        return default
+
+    @staticmethod
+    def extract_trace_id(headers):
+        return int(
+            DatadogHTTPPropagator.extract_header_value(
+                POSSIBLE_HTTP_HEADER_TRACE_IDS, headers, default=0,
+            )
+        )
+
+    @staticmethod
+    def extract_parent_span_id(headers):
+        return int(
+            DatadogHTTPPropagator.extract_header_value(
+                POSSIBLE_HTTP_HEADER_PARENT_IDS, headers, default=0,
+            )
+        )
+
+    @staticmethod
+    def extract_sampling_priority(headers):
+        return DatadogHTTPPropagator.extract_header_value(
+            POSSIBLE_HTTP_HEADER_SAMPLING_PRIORITIES, headers,
+        )
+
+    @staticmethod
+    def extract_origin(headers):
+        return DatadogHTTPPropagator.extract_header_value(
+            POSSIBLE_HTTP_HEADER_ORIGIN, headers,
+        )
+
+    def extract(self, headers):
+        """Extract a Context from HTTP headers into a new Context.
+
+        Here is an example from a web endpoint::
+
+            from oteltrace.propagation.http import DatadogHTTPPropagator
+
+            def my_controller(url, headers):
+                propagator = DatadogHTTPPropagator()
+                context = propagator.extract(headers)
+                tracer.context_provider.activate(context)
+
+                with tracer.trace('my_controller') as span:
+                    span.set_meta('http.url', url)
+
+        :param dict headers: HTTP headers to extract tracing attributes.
+        :return: New `Context` with propagated attributes.
+        """
+        if not headers:
+            return Context()
+
+        try:
+            trace_id = DatadogHTTPPropagator.extract_trace_id(headers)
+            parent_span_id = DatadogHTTPPropagator.extract_parent_span_id(headers)
+            sampling_priority = DatadogHTTPPropagator.extract_sampling_priority(headers)
+            origin = DatadogHTTPPropagator.extract_origin(headers)
+
+            if sampling_priority is not None:
+                sampling_priority = int(sampling_priority)
+
+            return Context(
+                trace_id=trace_id,
+                span_id=parent_span_id,
+                sampling_priority=sampling_priority,
+                _otel_origin=origin,
+            )
+        # If headers are invalid and cannot be parsed, return a new context and log the issue.
+        except Exception as error:
+            try:
+                log.debug(
+                    'invalid x-datadog-* headers, trace-id: %s, parent-id: %s, priority: %s, origin: %s, error: %s',
+                    headers.get(HTTP_HEADER_TRACE_ID, 0),
+                    headers.get(HTTP_HEADER_PARENT_ID, 0),
+                    headers.get(HTTP_HEADER_SAMPLING_PRIORITY),
+                    headers.get(HTTP_HEADER_ORIGIN, ''),
+                    error,
+                )
+            # We might fail on string formatting errors ; in that case only format the first error
+            except Exception:
+                log.debug(error)
+            return Context()