Asynqp: Update and use the TEXT_MAP propagator (#144)

* Update and use the TEXT_MAP propagator

* Update tests to follow changes.

Author: pglombardo

instana/instrumentation/asynqp.py

@@ -24,7 +24,7 @@ def publish_with_instana(wrapped, instance, argv, kwargs):
             msg = argv[0]
             if msg.headers is None:
                 msg.headers = {}
-            async_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, msg.headers)
+            async_tracer.inject(scope.span.context, opentracing.Format.TEXT_MAP, msg.headers)
 
             try:
                 scope.span.set_tag("exchange", instance.name)
@@ -77,7 +77,7 @@ def callback_with_instana(*argv, **kwargs):
                 ctx = None
                 msg = argv[0]
                 if msg.headers is not None:
-                    ctx = async_tracer.extract(opentracing.Format.HTTP_HEADERS, dict(msg.headers))
+                    ctx = async_tracer.extract(opentracing.Format.TEXT_MAP, dict(msg.headers))
 
                 with async_tracer.start_active_span("rabbitmq", child_of=ctx) as scope:
                     host, port = msg.sender.protocol.transport._sock.getsockname()

instana/text_propagator.py

@@ -6,45 +6,66 @@
 from .log import logger
 from .util import header_to_id
 
-prefix_tracer_state = 'X-INSTANA-'
-prefix_baggage = 'X-INSTANA-BAGGAGE-'
-field_name_trace_id = prefix_tracer_state + 'T'
-field_name_span_id = prefix_tracer_state + 'S'
-
 
 class TextPropagator():
     """
     A Propagator for TEXT_MAP.
     """
 
+    HEADER_KEY_T = 'X-INSTANA-T'
+    HEADER_KEY_S = 'X-INSTANA-S'
+    HEADER_KEY_L = 'X-INSTANA-L'
+
     def inject(self, span_context, carrier):
         try:
-            carrier[field_name_trace_id] = span_context.trace_id
-            carrier[field_name_span_id] = span_context.span_id
-            if span_context.baggage is not None:
-                for k in span_context.baggage:
-                    carrier[prefix_baggage+k] = span_context.baggage[k]
-        except Exception as e:
-            logger.debug("inject error: ", str(e))
+            trace_id = span_context.trace_id
+            span_id = span_context.span_id
+
+            if type(carrier) is dict or hasattr(carrier, "__dict__"):
+                carrier[self.HEADER_KEY_T] = trace_id
+                carrier[self.HEADER_KEY_S] = span_id
+                carrier[self.HEADER_KEY_L] = "1"
+            elif type(carrier) is list:
+                carrier.append((self.HEADER_KEY_T, trace_id))
+                carrier.append((self.HEADER_KEY_S, span_id))
+                carrier.append((self.HEADER_KEY_L, "1"))
+            elif hasattr(carrier, '__setitem__'):
+                carrier.__setitem__(self.HEADER_KEY_T, trace_id)
+                carrier.__setitem__(self.HEADER_KEY_S, span_id)
+                carrier.__setitem__(self.HEADER_KEY_L, "1")
+            else:
+                raise Exception("Unsupported carrier type", type(carrier))
+
+        except:
+            logger.debug("inject error:", exc_info=True)
 
     def extract(self, carrier):  # noqa
+        trace_id = None
+        span_id = None
+
         try:
-            if type(carrier) is dict or hasattr(carrier, "__dict__"):
+            if type(carrier) is dict or hasattr(carrier, "__getitem__"):
                 dc = carrier
+            elif hasattr(carrier, "__dict__"):
+                dc = carrier.__dict__
             elif type(carrier) is list:
                 dc = dict(carrier)
             else:
                 raise ot.SpanContextCorruptedException()
 
-            if field_name_trace_id in dc and field_name_span_id in dc:
-                trace_id = header_to_id(dc[field_name_trace_id])
-                span_id = header_to_id(dc[field_name_span_id])
+            for key in dc.keys():
+                if self.HEADER_KEY_T == key:
+                    trace_id = header_to_id(dc[key])
+                elif self.HEADER_KEY_S == key:
+                    span_id = header_to_id(dc[key])
 
-            return SpanContext(span_id=span_id,
-                               trace_id=trace_id,
-                               baggage={},
-                               sampled=True)
+            ctx = None
+            if trace_id is not None and span_id is not None:
+                ctx = SpanContext(span_id=span_id,
+                                  trace_id=trace_id,
+                                  baggage={},
+                                  sampled=True)
+            return ctx
 
         except Exception as e:
-            logger.debug("extract error: ", str(e))
-            return SpanContext()
+            logger.debug("extract error:", exc_info=True)

tests/test_asynqp.py

@@ -367,7 +367,7 @@ def handle_message(msg):
             # print("")
             # print("handle_message active scope: %s" % async_tracer.scope_manager.active)
             # print("")
-            async_tracer.inject(async_tracer.active_span.context, opentracing.Format.HTTP_HEADERS, msg.headers)
+            async_tracer.inject(async_tracer.active_span.context, opentracing.Format.TEXT_MAP, msg.headers)
             asyncio.ensure_future(run_later(msg))
             msg.ack()
 

tests/test_ot_propagators.py

@@ -5,11 +5,12 @@
 from nose.tools import assert_equals
 
 import instana.http_propagator as ihp
+import instana.text_propagator as itp
 from instana import options, util
 from instana.tracer import InstanaTracer
 
 
-def test_basics():
+def test_http_basics():
     inspect.isclass(ihp.HTTPPropagator)
 
     inject_func = getattr(ihp.HTTPPropagator, "inject", None)
@@ -21,7 +22,7 @@ def test_basics():
     assert callable(extract_func)
 
 
-def test_inject_with_dict():
+def test_http_inject_with_dict():
     opts = options.Options()
     ot.tracer = InstanaTracer(opts)
 
@@ -37,7 +38,7 @@ def test_inject_with_dict():
     assert_equals(carrier['X-Instana-L'], "1")
 
 
-def test_inject_with_list():
+def test_http_inject_with_list():
     opts = options.Options()
     ot.tracer = InstanaTracer(opts)
 
@@ -50,7 +51,7 @@ def test_inject_with_list():
     assert ('X-Instana-L', "1") in carrier
 
 
-def test_basic_extract():
+def test_http_basic_extract():
     opts = options.Options()
     ot.tracer = InstanaTracer(opts)
 
@@ -62,7 +63,7 @@ def test_basic_extract():
     assert_equals('0000000000000001', ctx.span_id)
 
 
-def test_mixed_case_extract():
+def test_http_mixed_case_extract():
     opts = options.Options()
     ot.tracer = InstanaTracer(opts)
 
@@ -74,7 +75,7 @@ def test_mixed_case_extract():
     assert_equals('0000000000000001', ctx.span_id)
 
 
-def test_no_context_extract():
+def test_http_no_context_extract():
     opts = options.Options()
     ot.tracer = InstanaTracer(opts)
 
@@ -84,7 +85,7 @@ def test_no_context_extract():
     assert ctx is None
 
 
-def test_128bit_headers():
+def test_http_128bit_headers():
     opts = options.Options()
     ot.tracer = InstanaTracer(opts)
 
@@ -96,3 +97,88 @@ def test_128bit_headers():
     assert_equals('b0789916ff8f319f', ctx.trace_id)
     assert_equals('b0789916ff8f319f', ctx.span_id)
 
+
+def test_text_basics():
+    inspect.isclass(itp.TextPropagator)
+
+    inject_func = getattr(itp.TextPropagator, "inject", None)
+    assert inject_func
+    assert callable(inject_func)
+
+    extract_func = getattr(itp.TextPropagator, "extract", None)
+    assert extract_func
+    assert callable(extract_func)
+
+
+def test_text_inject_with_dict():
+    opts = options.Options()
+    ot.tracer = InstanaTracer(opts)
+
+    carrier = {}
+    span = ot.tracer.start_span("nosetests")
+    ot.tracer.inject(span.context, ot.Format.TEXT_MAP, carrier)
+
+    assert 'X-INSTANA-T' in carrier
+    assert_equals(carrier['X-INSTANA-T'], span.context.trace_id)
+    assert 'X-INSTANA-S' in carrier
+    assert_equals(carrier['X-INSTANA-S'], span.context.span_id)
+    assert 'X-INSTANA-L' in carrier
+    assert_equals(carrier['X-INSTANA-L'], "1")
+
+
+def test_text_inject_with_list():
+    opts = options.Options()
+    ot.tracer = InstanaTracer(opts)
+
+    carrier = []
+    span = ot.tracer.start_span("nosetests")
+    ot.tracer.inject(span.context, ot.Format.TEXT_MAP, carrier)
+
+    assert ('X-INSTANA-T', span.context.trace_id) in carrier
+    assert ('X-INSTANA-S', span.context.span_id) in carrier
+    assert ('X-INSTANA-L', "1") in carrier
+
+
+def test_text_basic_extract():
+    opts = options.Options()
+    ot.tracer = InstanaTracer(opts)
+
+    carrier = {'X-INSTANA-T': '1', 'X-INSTANA-S': '1', 'X-INSTANA-L': '1'}
+    ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier)
+
+    assert type(ctx) is basictracer.context.SpanContext
+    assert_equals('0000000000000001', ctx.trace_id)
+    assert_equals('0000000000000001', ctx.span_id)
+
+
+def test_text_mixed_case_extract():
+    opts = options.Options()
+    ot.tracer = InstanaTracer(opts)
+
+    carrier = {'x-insTana-T': '1', 'X-inSTANa-S': '1', 'X-INstana-l': '1'}
+    ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier)
+
+    assert(ctx is None)
+
+
+def test_text_no_context_extract():
+    opts = options.Options()
+    ot.tracer = InstanaTracer(opts)
+
+    carrier = {}
+    ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier)
+
+    assert ctx is None
+
+
+def test_text_128bit_headers():
+    opts = options.Options()
+    ot.tracer = InstanaTracer(opts)
+
+    carrier = {'X-INSTANA-T': '0000000000000000b0789916ff8f319f',
+               'X-INSTANA-S': ' 0000000000000000b0789916ff8f319f', 'X-INSTANA-L': '1'}
+    ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier)
+
+    assert type(ctx) is basictracer.context.SpanContext
+    assert_equals('b0789916ff8f319f', ctx.trace_id)
+    assert_equals('b0789916ff8f319f', ctx.span_id)