[WIP] Context Prop

examples/opentelemetry-example-app/src/opentelemetry_example_app/context_propagation_example.py

@@ -0,0 +1,101 @@
+# 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.
+#
+"""
+This module serves as an example for baggage, which exists
+to pass application-defined key-value pairs from service to service.
+"""
+# import opentelemetry.ext.http_requests
+# from opentelemetry.ext.wsgi import OpenTelemetryMiddleware
+
+import flask
+from flask import request
+import requests
+
+from opentelemetry import propagation, trace
+from opentelemetry.distributedcontext import CorrelationContextManager
+from opentelemetry.sdk.context.propagation import b3_format
+from opentelemetry.sdk.trace import Tracer
+from opentelemetry.sdk.trace.export import (
+    BatchExportSpanProcessor,
+    ConsoleSpanExporter,
+)
+from opentelemetry.baggage import BaggageManager
+
+
+def configure_opentelemetry(flask_app: flask.Flask):
+    trace.set_preferred_tracer_implementation(lambda T: Tracer())
+
+    # Global initialization
+    (baggage_extractor, baggage_injector) = BaggageManager.http_propagator()
+    (b3_extractor, b3_injector) = b3_format.http_propagator()
+    # propagation.set_http_extractors([b3_extractor, baggage_extractor])
+    # propagation.set_http_injectors([b3_injector, baggage_injector])
+    propagation.set_http_extractors([b3_extractor])
+    propagation.set_http_injectors([b3_injector])
+
+    # opentelemetry.ext.http_requests.enable(trace.tracer())
+    # flask_app.wsgi_app = OpenTelemetryMiddleware(flask_app.wsgi_app)
+
+
+def fetch_from_service_b() -> str:
+    # Inject the contexts to be propagated. Note that there is no direct
+    # reference to tracing or baggage.
+    headers = {"Accept": "application/json"}
+    propagation.inject(headers)
+    print(headers)
+    resp = requests.get("https://opentelemetry.io", headers=headers)
+    return resp.text
+
+
+def fetch_from_service_c() -> str:
+    # Inject the contexts to be propagated. Note that there is no direct
+    # reference to tracing or baggage.
+    headers = {"Accept": "application/json"}
+    propagation.inject(headers)
+    print(headers)
+    resp = requests.get("https://opentelemetry.io", headers=headers)
+    return resp.text
+
+
+app = flask.Flask(__name__)
+
+
+@app.route("/")
+def hello():
+    tracer = trace.tracer()
+    tracer.add_span_processor(BatchExportSpanProcessor(ConsoleSpanExporter()))
+    with propagation.extract(request.headers):
+        # extract a baggage header
+        with tracer.start_as_current_span("service-span"):
+            with tracer.start_as_current_span("external-req-span"):
+                headers = {"Accept": "application/json"}
+                propagation.inject(headers)
+                version = CorrelationContextManager.value("version")
+                if version == "2.0":
+                    return fetch_from_service_c()
+
+                return fetch_from_service_b()
+
+
+request_headers = {
+    "Accept": "application/json",
+    "x-b3-traceid": "038c3fb613811e30898424c863eeae5a",
+    "x-b3-spanid": "6c7f9e56212a6ffa",
+    "x-b3-sampled": "0",
+}
+
+if __name__ == "__main__":
+    configure_opentelemetry(app)
+    app.run(debug=True)

examples/opentelemetry-example-app/src/opentelemetry_example_app/flask_example.py

@@ -20,9 +20,12 @@
 import requests
 
 import opentelemetry.ext.http_requests
-from opentelemetry import propagators, trace
+from opentelemetry import propagation, trace
 from opentelemetry.ext.flask import instrument_app
-from opentelemetry.sdk.context.propagation.b3_format import B3Format
+from opentelemetry.sdk.context.propagation.b3_format import (
+    B3Extractor,
+    B3Injector,
+)
 from opentelemetry.sdk.trace import Tracer
 
 
@@ -51,7 +54,8 @@ def configure_opentelemetry(flask_app: flask.Flask):
     # carry this value).
 
     # TBD: can remove once default TraceContext propagators are installed.
-    propagators.set_global_httptextformat(B3Format())
+    propagation.set_http_extractors([B3Extractor])
+    propagation.set_http_injectors([B3Injector])
 
     # Integrations are the glue that binds the OpenTelemetry API
     # and the frameworks and libraries that are used together, automatically

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

@@ -6,7 +6,8 @@
 from flask import request as flask_request
 
 import opentelemetry.ext.wsgi as otel_wsgi
-from opentelemetry import propagators, trace
+from opentelemetry import propagation, trace
+from opentelemetry.context import Context
 from opentelemetry.util import time_ns
 
 logger = logging.getLogger(__name__)
@@ -56,8 +57,8 @@ def _before_flask_request():
     span_name = flask_request.endpoint or otel_wsgi.get_default_span_name(
         environ
     )
-    parent_span = propagators.extract(
-        otel_wsgi.get_header_from_environ, environ
+    parent_span = propagation.extract(
+        Context.current(), otel_wsgi.get_header_from_environ, environ,
     )
 
     tracer = trace.tracer()

ext/opentelemetry-ext-http-requests/src/opentelemetry/ext/http_requests/__init__.py

@@ -22,7 +22,7 @@
 
 from requests.sessions import Session
 
-from opentelemetry import propagators
+from opentelemetry import propagation
 from opentelemetry.context import Context
 from opentelemetry.trace import SpanKind
 
@@ -74,7 +74,9 @@ def instrumented_request(self, method, url, *args, **kwargs):
             # to access propagators.
 
             headers = kwargs.setdefault("headers", {})
-            propagators.inject(tracer, type(headers).__setitem__, headers)
+            propagation.inject(
+                Context.current(), type(headers).__setitem__, headers,
+            )
             result = wrapped(self, method, url, *args, **kwargs)  # *** PROCEED
 
             span.set_attribute("http.status_code", result.status_code)

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

@@ -84,8 +84,13 @@
 from deprecated import deprecated
 
 import opentelemetry.trace as trace_api
-from opentelemetry import propagators
+from opentelemetry import propagation
+from opentelemetry.context import Context
 from opentelemetry.ext.opentracing_shim import util
+from opentelemetry.trace.propagation.context import (
+    from_context,
+    with_span_context,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -657,24 +662,22 @@ def inject(self, span_context, format, carrier):
         # TODO: Finish documentation.
         # pylint: disable=redefined-builtin
         # This implementation does not perform the injecting by itself but
-        # uses the configured propagators in opentelemetry.propagators.
+        # uses the configured propagation in opentelemetry.propagation.
         # TODO: Support Format.BINARY once it is supported in
         # opentelemetry-python.
         if format not in self._supported_formats:
             raise opentracing.UnsupportedFormatException
 
-        propagator = propagators.get_global_httptextformat()
-        propagator.inject(
-            span_context.unwrap(), type(carrier).__setitem__, carrier
-        )
+        ctx = with_span_context(Context.current(), span_context.unwrap())
+        propagation.inject(ctx, type(carrier).__setitem__, carrier)
 
     def extract(self, format, carrier):
         """Implements the ``extract`` method from the base class."""
 
         # TODO: Finish documentation.
         # pylint: disable=redefined-builtin
         # This implementation does not perform the extracing by itself but
-        # uses the configured propagators in opentelemetry.propagators.
+        # uses the configured propagation in opentelemetry.propagation.
         # TODO: Support Format.BINARY once it is supported in
         # opentelemetry-python.
         if format not in self._supported_formats:
@@ -684,7 +687,10 @@ def get_as_list(dict_object, key):
             value = dict_object.get(key)
             return [value] if value is not None else []
 
-        propagator = propagators.get_global_httptextformat()
-        otel_context = propagator.extract(get_as_list, carrier)
+        # propagator = propagation.get_global_httptextformat()
+
+        otel_context = from_context(
+            propagation.extract(Context.current(), get_as_list, carrier)
+        )
 
         return SpanContextShim(otel_context)

ext/opentelemetry-ext-opentracing-shim/tests/test_shim.py

@@ -18,8 +18,11 @@
 import opentracing
 
 import opentelemetry.ext.opentracing_shim as opentracingshim
-from opentelemetry import propagators, trace
-from opentelemetry.context.propagation.httptextformat import HTTPTextFormat
+from opentelemetry import propagation, trace
+from opentelemetry.context.propagation.httptextformat import (
+    HTTPExtractor,
+    HTTPInjector,
+)
 from opentelemetry.ext.opentracing_shim import util
 from opentelemetry.sdk.trace import Tracer
 
@@ -42,15 +45,18 @@ def setUpClass(cls):
         trace.set_preferred_tracer_implementation(lambda T: Tracer())
 
         # Save current propagator to be restored on teardown.
-        cls._previous_propagator = propagators.get_global_httptextformat()
+        cls._previous_injectors = propagation.get_http_injectors()
+        cls._previous_extractors = propagation.get_http_extractors()
 
         # Set mock propagator for testing.
-        propagators.set_global_httptextformat(MockHTTPTextFormat)
+        propagation.set_http_extractors([MockHTTPExtractor])
+        propagation.set_http_injectors([MockHTTPInjector])
 
     @classmethod
     def tearDownClass(cls):
         # Restore previous propagator.
-        propagators.set_global_httptextformat(cls._previous_propagator)
+        propagation.set_http_extractors(cls._previous_extractors)
+        propagation.set_http_injectors(cls._previous_injectors)
 
     def test_shim_type(self):
         # Verify shim is an OpenTracing tracer.
@@ -474,8 +480,8 @@ def test_inject_http_headers(self):
 
         headers = {}
         self.shim.inject(context, opentracing.Format.HTTP_HEADERS, headers)
-        self.assertEqual(headers[MockHTTPTextFormat.TRACE_ID_KEY], str(1220))
-        self.assertEqual(headers[MockHTTPTextFormat.SPAN_ID_KEY], str(7478))
+        self.assertEqual(headers[_TRACE_ID_KEY], str(1220))
+        self.assertEqual(headers[_SPAN_ID_KEY], str(7478))
 
     def test_inject_text_map(self):
         """Test `inject()` method for Format.TEXT_MAP."""
@@ -486,8 +492,8 @@ def test_inject_text_map(self):
         # Verify Format.TEXT_MAP
         text_map = {}
         self.shim.inject(context, opentracing.Format.TEXT_MAP, text_map)
-        self.assertEqual(text_map[MockHTTPTextFormat.TRACE_ID_KEY], str(1220))
-        self.assertEqual(text_map[MockHTTPTextFormat.SPAN_ID_KEY], str(7478))
+        self.assertEqual(text_map[_TRACE_ID_KEY], str(1220))
+        self.assertEqual(text_map[_SPAN_ID_KEY], str(7478))
 
     def test_inject_binary(self):
         """Test `inject()` method for Format.BINARY."""
@@ -503,8 +509,8 @@ def test_extract_http_headers(self):
         """Test `extract()` method for Format.HTTP_HEADERS."""
 
         carrier = {
-            MockHTTPTextFormat.TRACE_ID_KEY: 1220,
-            MockHTTPTextFormat.SPAN_ID_KEY: 7478,
+            _TRACE_ID_KEY: 1220,
+            _SPAN_ID_KEY: 7478,
         }
 
         ctx = self.shim.extract(opentracing.Format.HTTP_HEADERS, carrier)
@@ -515,8 +521,8 @@ def test_extract_text_map(self):
         """Test `extract()` method for Format.TEXT_MAP."""
 
         carrier = {
-            MockHTTPTextFormat.TRACE_ID_KEY: 1220,
-            MockHTTPTextFormat.SPAN_ID_KEY: 7478,
+            _TRACE_ID_KEY: 1220,
+            _SPAN_ID_KEY: 7478,
         }
 
         ctx = self.shim.extract(opentracing.Format.TEXT_MAP, carrier)
@@ -531,16 +537,17 @@ def test_extract_binary(self):
             self.shim.extract(opentracing.Format.BINARY, bytearray())
 
 
-class MockHTTPTextFormat(HTTPTextFormat):
-    """Mock propagator for testing purposes."""
+_TRACE_ID_KEY = "mock-traceid"
+_SPAN_ID_KEY = "mock-spanid"
 
-    TRACE_ID_KEY = "mock-traceid"
-    SPAN_ID_KEY = "mock-spanid"
+
+class MockHTTPExtractor(HTTPExtractor):
+    """Mock extractor for testing purposes."""
 
     @classmethod
-    def extract(cls, get_from_carrier, carrier):
-        trace_id_list = get_from_carrier(carrier, cls.TRACE_ID_KEY)
-        span_id_list = get_from_carrier(carrier, cls.SPAN_ID_KEY)
+    def extract(cls, context, get_from_carrier, carrier):
+        trace_id_list = get_from_carrier(carrier, _TRACE_ID_KEY)
+        span_id_list = get_from_carrier(carrier, _SPAN_ID_KEY)
 
         if not trace_id_list or not span_id_list:
             return trace.INVALID_SPAN_CONTEXT
@@ -549,7 +556,11 @@ def extract(cls, get_from_carrier, carrier):
             trace_id=int(trace_id_list[0]), span_id=int(span_id_list[0])
         )
 
+
+class MockHTTPInjector(HTTPInjector):
+    """Mock injector for testing purposes."""
+
     @classmethod
     def inject(cls, context, set_in_carrier, carrier):
-        set_in_carrier(carrier, cls.TRACE_ID_KEY, str(context.trace_id))
-        set_in_carrier(carrier, cls.SPAN_ID_KEY, str(context.span_id))
+        set_in_carrier(carrier, _TRACE_ID_KEY, str(context.trace_id))
+        set_in_carrier(carrier, _SPAN_ID_KEY, str(context.span_id))

ext/opentelemetry-ext-wsgi/src/opentelemetry/ext/wsgi/__init__.py

@@ -22,7 +22,8 @@
 import typing
 import wsgiref.util as wsgiref_util
 
-from opentelemetry import propagators, trace
+from opentelemetry import propagation, trace
+from opentelemetry.context import Context
 from opentelemetry.ext.wsgi.version import __version__  # noqa
 from opentelemetry.trace.status import Status, StatusCanonicalCode
 
@@ -183,7 +184,10 @@ def __call__(self, environ, start_response):
         """
 
         tracer = trace.tracer()
-        parent_span = propagators.extract(get_header_from_environ, environ)
+        # TODO: fix the return value here, expect a context
+        parent_span = propagation.extract(
+            Context.current(), get_header_from_environ, environ
+        )
         span_name = get_default_span_name(environ)
 
         span = tracer.start_span(