Enable client async and mixed tracing (#329)

* added retrieving of any active tracer (tracer, async_tracer or tornado) for each client implementation

* isolate the asynqp testing for avoiding infecting other random tests

* update the version

Author: pdimitra

.circleci/config.yml

@@ -235,6 +235,26 @@ jobs:
             . venv/bin/activate
             pytest -v tests/clients/test_cassandra-driver.py
 
+  py37asynqp:
+    docker:
+      - image: circleci/python:3.7.9
+      - image: rabbitmq:3.5.4
+    working_directory: ~/repo
+    steps:
+      - checkout
+      - pip-install-deps:
+          requirements: "tests/requirements-asynqp.txt"
+      - run:
+          name: run tests
+          environment:
+            INSTANA_TEST: "true"
+            ASYNQP_TEST: "true"
+          command: |
+            . venv/bin/activate
+            # We uninstall uvloop as it interferes with asyncio changing the event loop policy
+            pip uninstall -y uvloop
+            pytest -v tests/clients/test_asynqp.py
+
   gevent38:
     docker:
       - image: circleci/python:3.8.5
@@ -262,4 +282,5 @@ workflows:
       - python39
       - py27cassandra
       - py36cassandra
+      - py37asynqp
       - gevent38

instana/autoprofile/runtime.py

@@ -3,6 +3,7 @@
 
 import sys
 import signal
+import os
 
 
 class runtime_info(object):

instana/instrumentation/boto3_inst.py

@@ -9,12 +9,13 @@
 
 from ..log import logger
 from ..singletons import tracer
-
+from ..util.traceutils import get_active_tracer
 
 try:
     import boto3
     from boto3.s3 import inject
 
+
     def lambda_inject_context(payload, scope):
         """
         When boto3 lambda client 'Invoke' is called, we want to inject the tracing context.
@@ -36,13 +37,13 @@ def lambda_inject_context(payload, scope):
     @wrapt.patch_function_wrapper('botocore.client', 'BaseClient._make_api_call')
     def make_api_call_with_instana(wrapped, instance, arg_list, kwargs):
         # pylint: disable=protected-access
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
         # If we're not tracing, just return
-        if parent_span is None:
+        if active_tracer is None:
             return wrapped(*arg_list, **kwargs)
 
-        with tracer.start_active_span("boto3", child_of=parent_span) as scope:
+        with active_tracer.start_active_span("boto3", child_of=active_tracer.active_span) as scope:
             try:
                 operation = arg_list[0]
                 payload = arg_list[1]
@@ -62,7 +63,6 @@ def make_api_call_with_instana(wrapped, instance, arg_list, kwargs):
                 if 'lambda' in instance._endpoint.host and operation == 'Invoke':
                     lambda_inject_context(payload, scope)
 
-
             except Exception as exc:
                 logger.debug("make_api_call_with_instana: collect error", exc_info=True)
 
@@ -81,19 +81,20 @@ def make_api_call_with_instana(wrapped, instance, arg_list, kwargs):
                 scope.span.mark_as_errored({'error': exc})
                 raise
 
+
     def s3_inject_method_with_instana(wrapped, instance, arg_list, kwargs):
         fas = inspect.getfullargspec(wrapped)
         fas_args = fas.args
         fas_args.remove('self')
 
         # pylint: disable=protected-access
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
         # If we're not tracing, just return
-        if parent_span is None:
+        if active_tracer is None:
             return wrapped(*arg_list, **kwargs)
 
-        with tracer.start_active_span("boto3", child_of=parent_span) as scope:
+        with active_tracer.start_active_span("boto3", child_of=active_tracer.active_span) as scope:
             try:
                 operation = wrapped.__name__
                 scope.span.set_tag('op', operation)
@@ -119,6 +120,7 @@ def s3_inject_method_with_instana(wrapped, instance, arg_list, kwargs):
                 scope.span.mark_as_errored({'error': exc})
                 raise
 
+
     for method in ['upload_file', 'upload_fileobj', 'download_file', 'download_fileobj']:
         wrapt.wrap_function_wrapper('boto3.s3.inject', method, s3_inject_method_with_instana)
 

instana/instrumentation/cassandra_inst.py

@@ -7,12 +7,9 @@
 https://github.com/datastax/python-driver
 """
 from __future__ import absolute_import
-
-from distutils.version import LooseVersion
 import wrapt
-
 from ..log import logger
-from ..singletons import tracer
+from ..util.traceutils import get_active_tracer
 
 try:
     import cassandra
@@ -29,6 +26,7 @@
                                9: "LOCAL_SERIAL",
                                10: "LOCAL_ONE"})
 
+
     def collect_response(span, fn):
         tried_hosts = list()
         for host in fn.attempted_hosts:
@@ -46,15 +44,18 @@ def cb_request_finish(results, span, fn):
         collect_response(span, fn)
         span.finish()
 
+
     def cb_request_error(results, span, fn):
         collect_response(span, fn)
         span.mark_as_errored({"cassandra.error": results.message})
         span.finish()
 
+
     def request_init_with_instana(fn):
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
-        if parent_span is not None:
+        if active_tracer is not None:
+            parent_span = active_tracer.active_span
             ctags = dict()
             if isinstance(fn.query, cassandra.query.SimpleStatement):
                 ctags["cassandra.query"] = fn.query.query_string
@@ -64,21 +65,23 @@ def request_init_with_instana(fn):
             ctags["cassandra.keyspace"] = fn.session.keyspace
             ctags["cassandra.cluster"] = fn.session.cluster.metadata.cluster_name
 
-            span = tracer.start_span(
+            span = active_tracer.start_span(
                 operation_name="cassandra",
                 child_of=parent_span,
                 tags=ctags)
 
             fn.add_callback(cb_request_finish, span, fn)
             fn.add_errback(cb_request_error, span, fn)
 
+
     @wrapt.patch_function_wrapper('cassandra.cluster', 'Session.__init__')
     def init_with_instana(wrapped, instance, args, kwargs):
-        session =  wrapped(*args, **kwargs)
+        session = wrapped(*args, **kwargs)
         instance.add_request_init_listener(request_init_with_instana)
         return session
 
+
     logger.debug("Instrumenting cassandra")
 
 except ImportError:
-    pass
+    pass

instana/instrumentation/celery/hooks.py

@@ -6,6 +6,7 @@
 import opentracing
 from ...log import logger
 from ...singletons import tracer
+from ...util.traceutils import get_active_tracer
 
 try:
     import celery
@@ -111,22 +112,22 @@ def task_retry(*args, **kwargs):
     @signals.before_task_publish.connect
     def before_task_publish(*args, **kwargs):
         try:
-            parent_span = tracer.active_span
-            if parent_span is not None:
+            active_tracer = get_active_tracer()
+            if active_tracer is not None:
                 body = kwargs['body']
                 headers = kwargs['headers']
                 task_name = kwargs['sender']
                 task = registry.tasks.get(task_name)
                 task_id = get_task_id(headers, body)
 
-                scope = tracer.start_active_span("celery-client", child_of=parent_span)
+                scope = active_tracer.start_active_span("celery-client", child_of=active_tracer.active_span)
                 scope.span.set_tag("task", task_name)
                 scope.span.set_tag("task_id", task_id)
                 add_broker_tags(scope.span, task.app.conf['broker_url'])
 
                 # Context propagation
                 context_headers = {}
-                tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, context_headers)
+                active_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, context_headers)
 
                 # Fix for broken header propagation
                 # https://github.com/celery/celery/issues/4875

instana/instrumentation/couchbase_inst.py

@@ -11,7 +11,7 @@
 import wrapt
 
 from ..log import logger
-from ..singletons import tracer
+from ..util.traceutils import get_active_tracer
 
 try:
     import couchbase
@@ -47,13 +47,13 @@ def capture_kvs(scope, instance, query_arg, op):
 
     def make_wrapper(op):
         def wrapper(wrapped, instance, args, kwargs):
-            parent_span = tracer.active_span
+            active_tracer = get_active_tracer()
 
             # If we're not tracing, just return
-            if parent_span is None:
+            if active_tracer is None:
                 return wrapped(*args, **kwargs)
 
-            with tracer.start_active_span("couchbase", child_of=parent_span) as scope:
+            with active_tracer.start_active_span("couchbase", child_of=active_tracer.active_span) as scope:
                 capture_kvs(scope, instance, None, op)
                 try:
                     return wrapped(*args, **kwargs)
@@ -64,13 +64,13 @@ def wrapper(wrapped, instance, args, kwargs):
         return wrapper
 
     def query_with_instana(wrapped, instance, args, kwargs):
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
         # If we're not tracing, just return
-        if parent_span is None:
+        if active_tracer is None:
             return wrapped(*args, **kwargs)
 
-        with tracer.start_active_span("couchbase", child_of=parent_span) as scope:
+        with active_tracer.start_active_span("couchbase", child_of=active_tracer.active_span) as scope:
             capture_kvs(scope, instance, args[0], 'n1ql_query')
             try:
                 return wrapped(*args, **kwargs)

instana/instrumentation/logging.py

@@ -9,7 +9,7 @@
 import collections
 
 from ..log import logger
-from ..singletons import tracer
+from ..util.traceutils import get_active_tracer
 
 
 @wrapt.patch_function_wrapper('logging', 'Logger._log')
@@ -18,10 +18,10 @@ def log_with_instana(wrapped, instance, argv, kwargs):
     # argv[1] = message
     # argv[2] = args for message
     try:
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
         # Only needed if we're tracing and serious log
-        if parent_span and argv[0] >= logging.WARN:
+        if active_tracer and argv[0] >= logging.WARN:
 
             msg = str(argv[1])
             args = argv[2]
@@ -38,7 +38,7 @@ def log_with_instana(wrapped, instance, argv, kwargs):
                 parameters = '{} {}'.format(t , v)
 
             # create logging span
-            with tracer.start_active_span('log', child_of=parent_span) as scope:
+            with active_tracer.start_active_span('log', child_of=active_tracer.active_span) as scope:
                 scope.span.log_kv({ 'message': msg })
                 if parameters is not None:
                     scope.span.log_kv({ 'parameters': parameters })

instana/instrumentation/pep0249.py

@@ -6,7 +6,7 @@
 import wrapt
 
 from ..log import logger
-from ..singletons import tracer
+from ..util.traceutils import get_active_tracer
 from ..util.sql import sql_sanitizer
 
 
@@ -39,13 +39,13 @@ def _collect_kvs(self, span, sql):
             return span
 
     def execute(self, sql, params=None):
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
         # If not tracing or we're being called from sqlalchemy, just pass through
-        if (parent_span is None) or (parent_span.operation_name == "sqlalchemy"):
+        if (active_tracer is None) or (active_tracer.active_span.operation_name == "sqlalchemy"):
             return self.__wrapped__.execute(sql, params)
 
-        with tracer.start_active_span(self._module_name, child_of=parent_span) as scope:
+        with active_tracer.start_active_span(self._module_name, child_of=active_tracer.active_span) as scope:
             try:
                 self._collect_kvs(scope.span, sql)
 
@@ -58,13 +58,13 @@ def execute(self, sql, params=None):
                 return result
 
     def executemany(self, sql, seq_of_parameters):
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
         # If not tracing or we're being called from sqlalchemy, just pass through
-        if (parent_span is None) or (parent_span.operation_name == "sqlalchemy"):
+        if (active_tracer is None) or (active_tracer.active_span.operation_name == "sqlalchemy"):
             return self.__wrapped__.executemany(sql, seq_of_parameters)
 
-        with tracer.start_active_span(self._module_name, child_of=parent_span) as scope:
+        with active_tracer.start_active_span(self._module_name, child_of=active_tracer.active_span) as scope:
             try:
                 self._collect_kvs(scope.span, sql)
 
@@ -77,13 +77,13 @@ def executemany(self, sql, seq_of_parameters):
                 return result
 
     def callproc(self, proc_name, params):
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
         # If not tracing or we're being called from sqlalchemy, just pass through
-        if (parent_span is None) or (parent_span.operation_name == "sqlalchemy"):
+        if (active_tracer is None) or (active_tracer.active_span.operation_name == "sqlalchemy"):
             return self.__wrapped__.execute(proc_name, params)
 
-        with tracer.start_active_span(self._module_name, child_of=parent_span) as scope:
+        with active_tracer.start_active_span(self._module_name, child_of=active_tracer.active_span) as scope:
             try:
                 self._collect_kvs(scope.span, proc_name)
 

instana/instrumentation/pika.py

@@ -11,6 +11,7 @@
 
 from ..log import logger
 from ..singletons import tracer
+from ..util.traceutils import get_active_tracer
 
 try:
     import pika
@@ -41,14 +42,14 @@ def basic_publish_with_instana(wrapped, instance, args, kwargs):
         def _bind_args(exchange, routing_key, body, properties=None, *args, **kwargs):
             return (exchange, routing_key, body, properties, args, kwargs)
 
-        parent_span = tracer.active_span
+        active_tracer = get_active_tracer()
 
-        if parent_span is None:
+        if active_tracer is None:
             return wrapped(*args, **kwargs)
 
         (exchange, routing_key, body, properties, args, kwargs) = (_bind_args(*args, **kwargs))
 
-        with tracer.start_active_span("rabbitmq", child_of=parent_span) as scope:
+        with tracer.start_active_span("rabbitmq", child_of=active_tracer.active_span) as scope:
             try:
                 _extract_publisher_tags(scope.span,
                                         conn=instance.connection,

instana/instrumentation/pymongo.py

@@ -22,9 +22,7 @@ def started(self, event):
             if active_tracer is None:
                 return
 
-            parent_span = active_tracer.active_span
-
-            with active_tracer.start_active_span("mongo", child_of=parent_span) as scope:
+            with active_tracer.start_active_span("mongo", child_of=active_tracer.active_span) as scope:
                 self._collect_connection_tags(scope.span, event)
                 self._collect_command_tags(scope.span, event)