W3c trace context (#331)

* adding traceparent,tracestate forwarding

* added validation of traceparent/tracestate headers and update methods

* send updated w3c context headers

* adding span ctx calculation based on w3c trace context

* adding the w3c trace context relevant span properties

* apply several fixes discovered through testing

* performing some optimisations

* adding unittests for traceparent and tarcestate

* add fixes for adjusting behaviour to the testing tracer suite

* moving the logic of the extra span attributes relative to w3c to only be applied for entry spans

* adding asgi in the registered spans

* Update instana/version.py

* changes based on the review, propagators got merged, a parameter is defining the w3c trace context propagation

Author: pdimitra

instana/instrumentation/aiohttp/client.py

@@ -10,11 +10,11 @@
 from ...singletons import agent, async_tracer
 from ...util.secrets import strip_secrets_from_query
 
-
 try:
     import aiohttp
     import asyncio
 
+
     async def stan_request_start(session, trace_config_ctx, params):
         try:
             parent_span = async_tracer.active_span
@@ -31,13 +31,15 @@ async def stan_request_start(session, trace_config_ctx, params):
 
             parts = str(params.url).split('?')
             if len(parts) > 1:
-                cleaned_qp = strip_secrets_from_query(parts[1], agent.options.secrets_matcher, agent.options.secrets_list)
+                cleaned_qp = strip_secrets_from_query(parts[1], agent.options.secrets_matcher,
+                                                      agent.options.secrets_list)
                 scope.span.set_tag("http.params", cleaned_qp)
             scope.span.set_tag("http.url", parts[0])
             scope.span.set_tag('http.method', params.method)
         except Exception:
             logger.debug("stan_request_start", exc_info=True)
 
+
     async def stan_request_end(session, trace_config_ctx, params):
         try:
             scope = trace_config_ctx.scope
@@ -56,6 +58,7 @@ async def stan_request_end(session, trace_config_ctx, params):
         except Exception:
             logger.debug("stan_request_end", exc_info=True)
 
+
     async def stan_request_exception(session, trace_config_ctx, params):
         try:
             scope = trace_config_ctx.scope
@@ -66,7 +69,8 @@ async def stan_request_exception(session, trace_config_ctx, params):
         except Exception:
             logger.debug("stan_request_exception", exc_info=True)
 
-    @wrapt.patch_function_wrapper('aiohttp.client','ClientSession.__init__')
+
+    @wrapt.patch_function_wrapper('aiohttp.client', 'ClientSession.__init__')
     def init_with_instana(wrapped, instance, argv, kwargs):
         instana_trace_config = aiohttp.TraceConfig()
         instana_trace_config.on_request_start.append(stan_request_start)
@@ -79,7 +83,7 @@ def init_with_instana(wrapped, instance, argv, kwargs):
 
         return wrapped(*argv, **kwargs)
 
+
     logger.debug("Instrumenting aiohttp client")
 except ImportError:
     pass
-

instana/instrumentation/aiohttp/server.py

@@ -10,13 +10,13 @@
 from ...singletons import agent, async_tracer
 from ...util.secrets import strip_secrets_from_query
 
-
 try:
     import aiohttp
     import asyncio
 
     from aiohttp.web import middleware
 
+
     @middleware
     async def stan_middleware(request, handler):
         try:
@@ -80,6 +80,7 @@ def init_with_instana(wrapped, instance, argv, kwargs):
 
         return wrapped(*argv, **kwargs)
 
+
     logger.debug("Instrumenting aiohttp server")
 except ImportError:
     pass

instana/instrumentation/asgi.py

@@ -10,10 +10,12 @@
 from ..singletons import async_tracer, agent
 from ..util.secrets import strip_secrets_from_query
 
+
 class InstanaASGIMiddleware:
     """
     Instana ASGI Middleware
     """
+
     def __init__(self, app):
         self.app = app
 
@@ -41,7 +43,8 @@ def _collect_kvs(self, scope, span):
             if isinstance(query, (str, bytes)) and len(query):
                 if isinstance(query, bytes):
                     query = query.decode('utf-8')
-                scrubbed_params = strip_secrets_from_query(query, agent.options.secrets_matcher, agent.options.secrets_list)
+                scrubbed_params = strip_secrets_from_query(query, agent.options.secrets_matcher,
+                                                           agent.options.secrets_list)
                 span.set_tag("http.params", scrubbed_params)
 
             app = scope.get('app')
@@ -55,7 +58,6 @@ def _collect_kvs(self, scope, span):
         except Exception:
             logger.debug("ASGI collect_kvs: ", exc_info=True)
 
-
     async def __call__(self, scope, receive, send):
         request_context = None
 

instana/instrumentation/asynqp.py

@@ -13,7 +13,8 @@
     import asynqp
     import asyncio
 
-    @wrapt.patch_function_wrapper('asynqp.exchange','Exchange.publish')
+
+    @wrapt.patch_function_wrapper('asynqp.exchange', 'Exchange.publish')
     def publish_with_instana(wrapped, instance, argv, kwargs):
         parent_span = async_tracer.active_span
 
@@ -27,12 +28,13 @@ 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.HTTP_HEADERS, msg.headers,
+                                disable_w3c_trace_context=True)
 
             try:
                 scope.span.set_tag("exchange", instance.name)
                 scope.span.set_tag("sort", "publish")
-                scope.span.set_tag("address", host + ":" + str(port) )
+                scope.span.set_tag("address", host + ":" + str(port))
 
                 if 'routing_key' in kwargs:
                     scope.span.set_tag("key", kwargs['routing_key'])
@@ -46,8 +48,9 @@ def publish_with_instana(wrapped, instance, argv, kwargs):
             else:
                 return rv
 
+
     @asyncio.coroutine
-    @wrapt.patch_function_wrapper('asynqp.queue','Queue.get')
+    @wrapt.patch_function_wrapper('asynqp.queue', 'Queue.get')
     def get_with_instana(wrapped, instance, argv, kwargs):
         parent_span = async_tracer.active_span
 
@@ -59,7 +62,7 @@ def get_with_instana(wrapped, instance, argv, kwargs):
             host, port = instance.sender.protocol.transport._sock.getsockname()
 
             scope.span.set_tag("sort", "consume")
-            scope.span.set_tag("address", host + ":" + str(port) )
+            scope.span.set_tag("address", host + ":" + str(port))
 
             msg = yield from wrapped(*argv, **kwargs)
 
@@ -69,23 +72,25 @@ def get_with_instana(wrapped, instance, argv, kwargs):
 
             return msg
 
+
     @asyncio.coroutine
-    @wrapt.patch_function_wrapper('asynqp.queue','Queue.consume')
+    @wrapt.patch_function_wrapper('asynqp.queue', 'Queue.consume')
     def consume_with_instana(wrapped, instance, argv, kwargs):
         def callback_generator(original_callback):
             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.HTTP_HEADERS, dict(msg.headers),
+                                               disable_w3c_trace_context=True)
 
                 with async_tracer.start_active_span("rabbitmq", child_of=ctx) as scope:
                     host, port = msg.sender.protocol.transport._sock.getsockname()
 
                     try:
                         scope.span.set_tag("exchange", msg.exchange_name)
                         scope.span.set_tag("sort", "consume")
-                        scope.span.set_tag("address", host + ":" + str(port) )
+                        scope.span.set_tag("address", host + ":" + str(port))
                         scope.span.set_tag("key", msg.routing_key)
 
                         original_callback(*argv, **kwargs)
@@ -99,6 +104,7 @@ def callback_with_instana(*argv, **kwargs):
         argv = (callback_generator(cb),)
         return wrapped(*argv, **kwargs)
 
+
     logger.debug("Instrumenting asynqp")
 except ImportError:
     pass

instana/instrumentation/aws/triggers.py

@@ -8,6 +8,7 @@
 import json
 import base64
 from io import BytesIO
+import opentracing as ot
 
 from ...log import logger
 
@@ -21,9 +22,9 @@ def get_context(tracer, event):
         is_application_load_balancer_trigger(event)
 
     if is_proxy_event:
-        return tracer.extract('http_headers', event.get('headers', {}))
+        return tracer.extract(ot.Format.HTTP_HEADERS, event.get('headers', {}), disable_w3c_trace_context=True)
 
-    return tracer.extract('http_headers', event)
+    return tracer.extract(ot.Format.HTTP_HEADERS, event, disable_w3c_trace_context=True)
 
 
 def is_api_gateway_proxy_trigger(event):

instana/instrumentation/boto3_inst.py

@@ -12,6 +12,7 @@
 from ..util.traceutils import get_active_tracer
 
 try:
+    import opentracing as ot
     import boto3
     from boto3.s3 import inject
 
@@ -28,7 +29,7 @@ def lambda_inject_context(payload, scope):
             if not isinstance(invoke_payload, dict):
                 invoke_payload = json.loads(invoke_payload)
 
-            tracer.inject(scope.span.context, 'http_headers', invoke_payload)
+            tracer.inject(scope.span.context, ot.Format.HTTP_HEADERS, invoke_payload)
             payload['Payload'] = json.dumps(invoke_payload)
         except Exception:
             logger.debug("non-fatal lambda_inject_context: ", exc_info=True)

instana/instrumentation/celery/hooks.py

@@ -19,6 +19,7 @@
         import urlparse as parse
         import urllib
 
+
     def add_broker_tags(span, broker_url):
         try:
             url = parse.urlparse(broker_url)
@@ -45,6 +46,7 @@ def add_broker_tags(span, broker_url):
         except Exception:
             logger.debug("Error parsing broker URL: %s" % broker_url, exc_info=True)
 
+
     @signals.task_prerun.connect
     def task_prerun(*args, **kwargs):
         try:
@@ -55,7 +57,7 @@ def task_prerun(*args, **kwargs):
 
             headers = task.request.get('headers', {})
             if headers is not None:
-                ctx = tracer.extract(opentracing.Format.HTTP_HEADERS, headers)
+                ctx = tracer.extract(opentracing.Format.HTTP_HEADERS, headers, disable_w3c_trace_context=True)
 
             scope = tracer.start_active_span("celery-worker", child_of=ctx)
             scope.span.set_tag("task", task.name)
@@ -67,6 +69,7 @@ def task_prerun(*args, **kwargs):
         except:
             logger.debug("task_prerun: ", exc_info=True)
 
+
     @signals.task_postrun.connect
     def task_postrun(*args, **kwargs):
         try:
@@ -78,6 +81,7 @@ def task_postrun(*args, **kwargs):
         except:
             logger.debug("after_task_publish: ", exc_info=True)
 
+
     @signals.task_failure.connect
     def task_failure(*args, **kwargs):
         try:
@@ -95,6 +99,7 @@ def task_failure(*args, **kwargs):
         except:
             logger.debug("task_failure: ", exc_info=True)
 
+
     @signals.task_retry.connect
     def task_retry(*args, **kwargs):
         try:
@@ -109,6 +114,7 @@ def task_retry(*args, **kwargs):
         except:
             logger.debug("task_failure: ", exc_info=True)
 
+
     @signals.before_task_publish.connect
     def before_task_publish(*args, **kwargs):
         try:
@@ -127,7 +133,8 @@ def before_task_publish(*args, **kwargs):
 
                 # Context propagation
                 context_headers = {}
-                active_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, context_headers)
+                active_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, context_headers,
+                                     disable_w3c_trace_context=True)
 
                 # Fix for broken header propagation
                 # https://github.com/celery/celery/issues/4875
@@ -141,6 +148,7 @@ def before_task_publish(*args, **kwargs):
         except:
             logger.debug("before_task_publish: ", exc_info=True)
 
+
     @signals.after_task_publish.connect
     def after_task_publish(*args, **kwargs):
         try:
@@ -152,6 +160,7 @@ def after_task_publish(*args, **kwargs):
         except:
             logger.debug("after_task_publish: ", exc_info=True)
 
+
     logger.debug("Instrumenting celery")
 except ImportError:
     pass

instana/instrumentation/flask/vanilla.py

@@ -22,8 +22,7 @@ def before_request_with_instana(*argv, **kwargs):
         env = flask.request.environ
         ctx = None
 
-        if 'HTTP_X_INSTANA_T' in env and 'HTTP_X_INSTANA_S' in env:
-            ctx = tracer.extract(opentracing.Format.HTTP_HEADERS, env)
+        ctx = tracer.extract(opentracing.Format.HTTP_HEADERS, env)
 
         flask.g.scope = tracer.start_active_span('wsgi', child_of=ctx)
         span = flask.g.scope.span
@@ -39,7 +38,8 @@ def before_request_with_instana(*argv, **kwargs):
         if 'PATH_INFO' in env:
             span.set_tag(ext.HTTP_URL, env['PATH_INFO'])
         if 'QUERY_STRING' in env and len(env['QUERY_STRING']):
-            scrubbed_params = strip_secrets_from_query(env['QUERY_STRING'], agent.options.secrets_matcher, agent.options.secrets_list)
+            scrubbed_params = strip_secrets_from_query(env['QUERY_STRING'], agent.options.secrets_matcher,
+                                                       agent.options.secrets_list)
             span.set_tag("http.params", scrubbed_params)
         if 'HTTP_HOST' in env:
             span.set_tag("http.host", env['HTTP_HOST'])

instana/instrumentation/flask/with_blinker.py

@@ -23,8 +23,7 @@ def request_started_with_instana(sender, **extra):
         env = flask.request.environ
         ctx = None
 
-        if 'HTTP_X_INSTANA_T' in env and 'HTTP_X_INSTANA_S' in env:
-            ctx = tracer.extract(opentracing.Format.HTTP_HEADERS, env)
+        ctx = tracer.extract(opentracing.Format.HTTP_HEADERS, env)
 
         flask.g.scope = tracer.start_active_span('wsgi', child_of=ctx)
         span = flask.g.scope.span
@@ -40,7 +39,8 @@ def request_started_with_instana(sender, **extra):
         if 'PATH_INFO' in env:
             span.set_tag(ext.HTTP_URL, env['PATH_INFO'])
         if 'QUERY_STRING' in env and len(env['QUERY_STRING']):
-            scrubbed_params = strip_secrets_from_query(env['QUERY_STRING'], agent.options.secrets_matcher, agent.options.secrets_list)
+            scrubbed_params = strip_secrets_from_query(env['QUERY_STRING'], agent.options.secrets_matcher,
+                                                       agent.options.secrets_list)
             span.set_tag("http.params", scrubbed_params)
         if 'HTTP_HOST' in env:
             span.set_tag("http.host", env['HTTP_HOST'])

instana/instrumentation/google/cloud/pubsub.py

@@ -47,7 +47,7 @@ def publish_with_instana(wrapped, instance, args, kwargs):
         with tracer.start_active_span('gcps-producer', child_of=parent_span) as scope:
             # trace continuity, inject to the span context
             headers = dict()
-            tracer.inject(scope.span.context, Format.TEXT_MAP, headers)
+            tracer.inject(scope.span.context, Format.TEXT_MAP, headers, disable_w3c_trace_context=True)
 
             # update the metadata dict with instana trace attributes
             kwargs.update(headers)
@@ -73,7 +73,7 @@ def subscribe_with_instana(wrapped, instance, args, kwargs):
 
         def callback_with_instana(message):
             if message.attributes:
-                parent_span = tracer.extract(Format.TEXT_MAP, message.attributes)
+                parent_span = tracer.extract(Format.TEXT_MAP, message.attributes, disable_w3c_trace_context=True)
             else:
                 parent_span = None