New Tornado instrumentation (#146)

* Tornado Server instrumentation, tests and support

* Don't log HTTPError as a general exception

* Tornado server tests

* Tornado client instrumentation and tests

* Updated 3xx handling

* Init AsyncHTTPClient after loop

* Limit tornado version to <6.0.0

* pylint improvements

* Better tracing of Tornado internal 301 handling

Author: pglombardo

instana/__init__.py

@@ -1,11 +1,3 @@
-from __future__ import absolute_import
-
-import os
-import sys
-import pkg_resources
-from threading import Timer
-
-
 """
 The Instana package has two core components: the agent and the tracer.
 
@@ -25,6 +17,14 @@
   Recorder
 """
 
+from __future__ import absolute_import
+
+import os
+import sys
+from threading import Timer
+import pkg_resources
+
+
 if "INSTANA_MAGIC" in os.environ:
     pkg_resources.working_set.add_entry("/tmp/instana/python")
 
@@ -41,7 +41,7 @@
     __version__ = 'unknown'
 
 
-def load(module):
+def load(_):
     """
     Method used to activate the Instana sensor via AUTOWRAPT_BOOTSTRAP
     environment variable.
@@ -53,26 +53,33 @@ def load(module):
 
 
 # User configurable EUM API key for instana.helpers.eum_snippet()
+# pylint: disable=invalid-name
 eum_api_key = ''
 
 
 def boot_agent():
-    import instana.singletons  # noqa
+    """Initialize the Instana agent and conditionally load auto-instrumentation."""
+    # Disable all the unused-import violations in this function
+    # pylint: disable=unused-import
+
+    import instana.singletons
 
     if "INSTANA_DISABLE_AUTO_INSTR" not in os.environ:
         # Import & initialize instrumentation
         if sys.version_info >= (3, 5, 3):
-            from .instrumentation import asyncio  # noqa
-            from .instrumentation.aiohttp import client  # noqa
-            from .instrumentation.aiohttp import server  # noqa
-            from .instrumentation import asynqp  # noqa
-        from .instrumentation import logging  # noqa
-        from .instrumentation import mysqlpython  # noqa
-        from .instrumentation import redis  # noqa
-        from .instrumentation import sqlalchemy  # noqa
-        from .instrumentation import sudsjurko  # noqa
-        from .instrumentation import urllib3  # noqa
-        from .instrumentation.django import middleware  # noqa
+            from .instrumentation import asyncio
+            from .instrumentation.aiohttp import client
+            from .instrumentation.aiohttp import server
+            from .instrumentation import asynqp
+            from .instrumentation.tornado import client
+            from .instrumentation.tornado import server
+        from .instrumentation import logging
+        from .instrumentation import mysqlpython
+        from .instrumentation import redis
+        from .instrumentation import sqlalchemy
+        from .instrumentation import sudsjurko
+        from .instrumentation import urllib3
+        from .instrumentation.django import middleware
 
 
 if "INSTANA_MAGIC" in os.environ:

instana/instrumentation/tornado/client.py

@@ -0,0 +1,86 @@
+from __future__ import absolute_import
+
+import opentracing
+import wrapt
+import functools
+import basictracer
+import sys
+
+from ...log import logger
+from ...singletons import agent, tornado_tracer
+from ...util import strip_secrets
+
+from distutils.version import LooseVersion
+
+
+# Tornado >=6.0 switched to contextvars for context management.  This requires changes to the opentracing
+# scope managers which we will tackle soon.
+# Limit Tornado version for the time being.
+if (('tornado' in sys.modules) and
+        hasattr(sys.modules['tornado'], 'version') and
+        (LooseVersion(sys.modules['tornado'].version) < LooseVersion('6.0.0'))):
+    try:
+        import tornado
+
+        @wrapt.patch_function_wrapper('tornado.httpclient', 'AsyncHTTPClient.fetch')
+        def fetch_with_instana(wrapped, instance, argv, kwargs):
+            try:
+                parent_span = tornado_tracer.active_span
+
+                # If we're not tracing, just return
+                if (parent_span is None) or (parent_span.operation_name == "tornado-client"):
+                    return wrapped(*argv, **kwargs)
+
+                request = argv[0]
+
+                # To modify request headers, we have to preemptively create an HTTPRequest object if a
+                # URL string was passed.
+                if not isinstance(request, tornado.httpclient.HTTPRequest):
+                    request = tornado.httpclient.HTTPRequest(url=request, **kwargs)
+
+                    new_kwargs = {}
+                    for param in ('callback', 'raise_error'):
+                        # if not in instead and pop
+                        if param in kwargs:
+                            new_kwargs[param] = kwargs.pop(param)
+                    kwargs = new_kwargs
+
+                scope = tornado_tracer.start_active_span('tornado-client', child_of=parent_span)
+                tornado_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, request.headers)
+
+                # Query param scrubbing
+                parts = request.url.split('?')
+                if len(parts) > 1:
+                    cleaned_qp = strip_secrets(parts[1], agent.secrets_matcher, agent.secrets_list)
+                    scope.span.set_tag("http.params", cleaned_qp)
+
+                scope.span.set_tag("http.url", parts[0])
+                scope.span.set_tag("http.method", request.method)
+
+                future = wrapped(request, **kwargs)
+
+                if future is not None:
+                    cb = functools.partial(finish_tracing, scope=scope)
+                    future.add_done_callback(cb)
+
+                return future
+            except Exception:
+                logger.debug("tornado fetch", exc_info=True)
+                raise
+
+        def finish_tracing(future, scope):
+            try:
+                response = future.result()
+                scope.span.set_tag("http.status_code", response.code)
+            except tornado.httpclient.HTTPClientError as e:
+                scope.span.set_tag("http.status_code", e.code)
+                scope.span.log_exception(e)
+                raise
+            finally:
+                scope.close()
+
+
+        logger.debug("Instrumenting tornado client")
+    except ImportError:
+        pass
+

instana/instrumentation/tornado/server.py

@@ -0,0 +1,109 @@
+from __future__ import absolute_import
+
+import opentracing
+from opentracing.scope_managers.tornado import tracer_stack_context
+import wrapt
+import sys
+
+from ...log import logger
+from ...singletons import agent, tornado_tracer
+from ...util import strip_secrets
+
+from distutils.version import LooseVersion
+
+# Tornado >=6.0 switched to contextvars for context management.  This requires changes to the opentracing
+# scope managers which we will tackle soon.
+# Limit Tornado version for the time being.
+if (('tornado' in sys.modules) and
+        hasattr(sys.modules['tornado'], 'version') and
+        (LooseVersion(sys.modules['tornado'].version) < LooseVersion('6.0.0'))):
+
+    try:
+        import tornado
+
+        @wrapt.patch_function_wrapper('tornado.web', 'RequestHandler._execute')
+        def execute_with_instana(wrapped, instance, argv, kwargs):
+            try:
+                ctx = tornado_tracer.extract(opentracing.Format.HTTP_HEADERS, instance.request.headers)
+                scope = tornado_tracer.start_active_span('tornado-server', child_of=ctx)
+
+                # Query param scrubbing
+                if instance.request.query is not None and len(instance.request.query) > 0:
+                    cleaned_qp = strip_secrets(instance.request.query, agent.secrets_matcher, agent.secrets_list)
+                    scope.span.set_tag("http.params", cleaned_qp)
+
+                scope.span.set_tag("http.host", instance.request.host)
+                scope.span.set_tag("http.method", instance.request.method)
+                scope.span.set_tag("http.path", instance.request.path)
+
+                # Custom header tracking support
+                if agent.extra_headers is not None:
+                    for custom_header in agent.extra_headers:
+                        if custom_header in instance.request.headers:
+                            scope.span.set_tag("http.%s" % custom_header, instance.request.headers[custom_header])
+
+                with tracer_stack_context():
+                    setattr(instance.request, "_instana", scope)
+
+                    # Set the context response headers now because tornado doesn't give us a better option to do so
+                    # later for this request.
+                    tornado_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, instance._headers)
+                    instance.set_header(name='Server-Timing', value="intid;desc=%s" % scope.span.context.trace_id)
+
+                    return wrapped(*argv, **kwargs)
+            except Exception:
+                logger.debug("tornado execute", exc_info=True)
+
+
+        @wrapt.patch_function_wrapper('tornado.web', 'RequestHandler.set_default_headers')
+        def set_default_headers_with_instana(wrapped, instance, argv, kwargs):
+            if not hasattr(instance.request, '_instana'):
+                return wrapped(*argv, **kwargs)
+
+            scope = instance.request._instana
+            tornado_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, instance._headers)
+            instance.set_header(name='Server-Timing', value="intid;desc=%s" % scope.span.context.trace_id)
+
+
+        @wrapt.patch_function_wrapper('tornado.web', 'RequestHandler.on_finish')
+        def on_finish_with_instana(wrapped, instance, argv, kwargs):
+            try:
+                if not hasattr(instance.request, '_instana'):
+                    return wrapped(*argv, **kwargs)
+
+                scope = instance.request._instana
+                status_code = instance.get_status()
+
+                # Mark 500 responses as errored
+                if 500 <= status_code <= 511:
+                    scope.span.set_tag("error", True)
+                    ec = scope.span.tags.get('ec', 0)
+                    if ec is 0:
+                        scope.span.set_tag("ec", ec + 1)
+
+                scope.span.set_tag("http.status_code", status_code)
+                scope.span.finish()
+                scope.close()
+
+                return wrapped(*argv, **kwargs)
+            except Exception:
+                logger.debug("tornado on_finish", exc_info=True)
+
+        @wrapt.patch_function_wrapper('tornado.web', 'RequestHandler.log_exception')
+        def log_exception_with_instana(wrapped, instance, argv, kwargs):
+            try:
+                if not hasattr(instance.request, '_instana'):
+                    return wrapped(*argv, **kwargs)
+
+                if not isinstance(argv[1], tornado.web.HTTPError):
+                    scope = instance.request._instana
+                    scope.span.log_exception(argv[0])
+
+                return wrapped(*argv, **kwargs)
+            except Exception:
+                logger.debug("tornado log_exception", exc_info=True)
+
+        logger.debug("Instrumenting tornado server")
+    except ImportError:
+        pass
+

instana/json_span.py

@@ -51,6 +51,7 @@ class HttpData(BaseSpan):
     params = None
     status = 0
     method = None
+    path = None
     path_tpl = None
     error = None
 

instana/recorder.py

@@ -1,7 +1,6 @@
 from __future__ import absolute_import
 
 import os
-import socket
 import sys
 import threading as t
 
@@ -25,12 +24,13 @@
 class InstanaRecorder(SpanRecorder):
     registered_spans = ("aiohttp-client", "aiohttp-server", "django", "log", "memcache", "mysql",
                         "rabbitmq", "redis", "rpc-client", "rpc-server", "sqlalchemy", "soap",
-                        "urllib3", "wsgi")
-    http_spans = ("aiohttp-client", "aiohttp-server", "django", "wsgi", "urllib3", "soap")
+                        "tornado-server", "tornado-client", "urllib3", "wsgi")
+    http_spans = ("aiohttp-client", "aiohttp-server", "django", "http", "soap", "tornado-server",
+                  "tornado-client", "urllib3", "wsgi")
 
     exit_spans = ("aiohttp-client", "log", "memcache", "mysql", "rabbitmq", "redis", "rpc-client",
-                  "sqlalchemy", "soap", "urllib3")
-    entry_spans = ("aiohttp-server", "django", "wsgi", "rabbitmq", "rpc-server")
+                  "sqlalchemy", "soap", "tornado-client", "urllib3")
+    entry_spans = ("aiohttp-server", "django", "wsgi", "rabbitmq", "rpc-server", "tornado-server")
 
     entry_kind = ["entry", "server", "consumer"]
     exit_kind = ["exit", "client", "producer"]
@@ -118,6 +118,7 @@ def build_registered_span(self, span):
         if span.operation_name in self.http_spans:
             data.http = HttpData(host=span.tags.pop("http.host", None),
                                  url=span.tags.pop(ext.HTTP_URL, None),
+                                 path=span.tags.pop("http.path", None),
                                  params=span.tags.pop('http.params', None),
                                  method=span.tags.pop(ext.HTTP_METHOD, None),
                                  status=span.tags.pop(ext.HTTP_STATUS_CODE, None),

instana/singletons.py

@@ -20,7 +20,10 @@
 
 if sys.version_info >= (3,4):
     from opentracing.scope_managers.asyncio import AsyncioScopeManager
+    from opentracing.scope_managers.tornado import TornadoScopeManager
+
     async_tracer = InstanaTracer(scope_manager=AsyncioScopeManager())
+    tornado_tracer = InstanaTracer(scope_manager=TornadoScopeManager())
 
 # Set ourselves as the tracer.
 opentracing.tracer = tracer

runtests.py

@@ -5,7 +5,7 @@
 command_line = [__file__, '--verbose']
 
 if (LooseVersion(sys.version) < LooseVersion('3.5.3')):
-    command_line.extend(['-e', 'asynqp', '-e', 'aiohttp', '-e', 'async'])
+    command_line.extend(['-e', 'asynqp', '-e', 'aiohttp', '-e', 'async', '-e', 'tornado'])
 
 if (LooseVersion(sys.version) >= LooseVersion('3.7.0')):
     command_line.extend(['-e', 'sudsjurko'])

setup.py

@@ -69,6 +69,7 @@ def check_setuptools():
             'sqlalchemy>=1.1.15',
             'spyne>=2.9,<=2.12.14',
             'suds-jurko>=0.6',
+            'tornado>=4.5.3,<6.0',
             'urllib3[secure]>=1.15'
         ],
       },

tests/__init__.py

@@ -45,4 +45,17 @@
     print("Starting background aiohttp server...")
     aio_server.start()
 
+
+if sys.version_info >= (3, 5, 3):
+    # Background Tornado application
+    from .apps.tornado import run_server
+
+    # Spawn our background Tornado app that the tests will throw
+    # requests at.
+    tornado_server = threading.Thread(target=run_server)
+    tornado_server.daemon = True
+    tornado_server.name = "Background Tornado server"
+    print("Starting background Tornado server...")
+    tornado_server.start()
+
 time.sleep(1)