Adding propagators API and b3 SDK implementation (#51, #52) (#78)

Specification: https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/api-propagators.md.

opentelemetry-api/src/opentelemetry/context/propagation/__init__.py

@@ -0,0 +1,4 @@
+from .binaryformat import BinaryFormat
+from .httptextformat import HTTPTextFormat
+
+__all__ = ["BinaryFormat", "HTTPTextFormat"]

opentelemetry-api/src/opentelemetry/context/propagation/binaryformat.py

@@ -0,0 +1,58 @@
+# Copyright 2019, OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import abc
+import typing
+
+from opentelemetry.trace import SpanContext
+
+
+class BinaryFormat(abc.ABC):
+    """API for serialization of span context into binary formats.
+
+    This class provides an interface that enables converting span contexts
+    to and from a binary format.
+    """
+    @staticmethod
+    @abc.abstractmethod
+    def to_bytes(context: SpanContext) -> bytes:
+        """Creates a byte representation of a SpanContext.
+
+        to_bytes should read values from a SpanContext and return a data
+        format to represent it, in bytes.
+
+        Args:
+            context: the SpanContext to serialize
+
+        Returns:
+            A bytes representation of the SpanContext.
+
+        """
+    @staticmethod
+    @abc.abstractmethod
+    def from_bytes(byte_representation: bytes) -> typing.Optional[SpanContext]:
+        """Return a SpanContext that was represented by bytes.
+
+        from_bytes should return back a SpanContext that was constructed from
+        the data serialized in the byte_representation passed. If it is not
+        possible to read in a proper SpanContext, return None.
+
+        Args:
+            byte_representation: the bytes to deserialize
+
+        Returns:
+            A bytes representation of the SpanContext if it is valid.
+            Otherwise return None.
+
+        """

opentelemetry-api/src/opentelemetry/context/propagation/httptextformat.py

@@ -0,0 +1,109 @@
+# Copyright 2019, OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import abc
+import typing
+
+from opentelemetry.trace import SpanContext
+
+Setter = typing.Callable[[object, str, str], None]
+Getter = typing.Callable[[object, str], typing.List[str]]
+
+
+class HTTPTextFormat(abc.ABC):
+    """API for propagation of span context via headers.
+
+    This class provides an interface that enables extracting and injecting
+    span context into headers of HTTP requests. HTTP frameworks and clients
+    can integrate with HTTPTextFormat by providing the object containing the
+    headers, and a getter and setter function for the extraction and
+    injection of values, respectively.
+
+    Example::
+
+        import flask
+        import requests
+        from opentelemetry.context.propagation import HTTPTextFormat
+
+        PROPAGATOR = HTTPTextFormat()
+
+
+
+        def get_header_from_flask_request(request, key):
+            return request.headers.get_all(key)
+
+        def set_header_into_requests_request(request: requests.Request,
+                                             key: str, value: str):
+            request.headers[key] = value
+
+        def example_route():
+            span_context = PROPAGATOR.extract(
+                get_header_from_flask_request,
+                flask.request
+            )
+            request_to_downstream = requests.Request(
+                "GET", "http://httpbin.org/get"
+            )
+            PROPAGATOR.inject(
+                span_context,
+                set_header_into_requests_request,
+                request_to_downstream
+            )
+            session = requests.Session()
+            session.send(request_to_downstream.prepare())
+
+
+    .. _Propagation API Specification:
+       https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/api-propagators.md
+    """
+    @abc.abstractmethod
+    def extract(self, get_from_carrier: Getter,
+                carrier: object) -> SpanContext:
+        """Create a SpanContext from values in the carrier.
+
+        The extract function should retrieve values from the carrier
+        object using get_from_carrier, and use values to populate a
+        SpanContext value and return it.
+
+        Args:
+            get_from_carrier: a function that can retrieve zero
+                or more values from the carrier. In the case that
+                the value does not exist, return an empty list.
+            carrier: and object which contains values that are
+                used to construct a SpanContext. This object
+                must be paired with an appropriate get_from_carrier
+                which understands how to extract a value from it.
+        Returns:
+            A SpanContext with configuration found in the carrier.
+
+        """
+    @abc.abstractmethod
+    def inject(self, context: SpanContext, set_in_carrier: Setter,
+               carrier: object) -> None:
+        """Inject values from a SpanContext into a carrier.
+
+        inject enables the propagation of values into HTTP clients or
+        other objects which perform an HTTP request. Implementations
+        should use the set_in_carrier method to set values on the
+        carrier.
+
+        Args:
+            context: The SpanContext to read values from.
+            set_in_carrier: A setter function that can set values
+                on the carrier.
+            carrier: An object that a place to define HTTP headers.
+                Should be paired with set_in_carrier, which should
+                know how to set header values on the carrier.
+
+        """

opentelemetry-sdk/src/opentelemetry/sdk/context/propagation/b3_format.py

@@ -0,0 +1,109 @@
+# Copyright 2019, OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import typing
+
+from opentelemetry.context.propagation.httptextformat import HTTPTextFormat
+import opentelemetry.trace as trace
+
+
+class B3Format(HTTPTextFormat):
+    """Propagator for the B3 HTTP header format.
+
+    See: https://github.com/openzipkin/b3-propagation
+    """
+
+    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"])
+
+    @classmethod
+    def extract(cls, get_from_carrier, carrier):
+        trace_id = format_trace_id(trace.INVALID_TRACE_ID)
+        span_id = format_span_id(trace.INVALID_SPAN_ID)
+        sampled = 0
+        flags = None
+
+        single_header = _extract_first_element(
+            get_from_carrier(carrier, cls.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 trace.INVALID_SPAN_CONTEXT
+        else:
+            trace_id = _extract_first_element(
+                get_from_carrier(carrier, cls.TRACE_ID_KEY)) or trace_id
+            span_id = _extract_first_element(
+                get_from_carrier(carrier, cls.SPAN_ID_KEY)) or span_id
+            sampled = _extract_first_element(
+                get_from_carrier(carrier, cls.SAMPLED_KEY)) or sampled
+            flags = _extract_first_element(
+                get_from_carrier(carrier, cls.FLAGS_KEY)) or flags
+
+        options = 0
+        # The b3 spec provides no defined behavior for both sample and
+        # flag values set. Since the setting of at least one implies
+        # the desire for some form of sampling, propagate if either
+        # header is set to allow.
+        if sampled in cls._SAMPLE_PROPAGATE_VALUES or flags == "1":
+            options |= trace.TraceOptions.RECORDED
+
+        return trace.SpanContext(
+            # trace an span ids are encoded in hex, so must be converted
+            trace_id=int(trace_id, 16),
+            span_id=int(span_id, 16),
+            trace_options=options,
+            trace_state={},
+        )
+
+    @classmethod
+    def inject(cls, context, set_in_carrier, carrier):
+        sampled = (trace.TraceOptions.RECORDED & context.trace_options) != 0
+        set_in_carrier(carrier, cls.TRACE_ID_KEY,
+                       format_trace_id(context.trace_id))
+        set_in_carrier(carrier, cls.SPAN_ID_KEY,
+                       format_span_id(context.span_id))
+        set_in_carrier(carrier, cls.SAMPLED_KEY, "1" if sampled else "0")
+
+
+def format_trace_id(trace_id: int):
+    """Format the trace id according to b3 specification."""
+    return format(trace_id, "032x")
+
+
+def format_span_id(span_id: int):
+    """Format the span id according to b3 specification."""
+    return format(span_id, "016x")
+
+
+def _extract_first_element(list_object: list) -> typing.Optional[object]:
+    if list_object:
+        return list_object[0]
+    return None