New aiohttp instrumentation (#133)

* Add aiohttp to test requirements

* aiohttp instrumentation & tests

* Add aiohttp to exlude list on Python versions earlier than 3.5.3

* Update min py version for test packages

* Additional way to inject headers

* Fix asynqp version specifier

* Exception and redirect handling, reporting & tests

* Fix instrumentation message

* aiohttp server instrumentation & piu (in omaggio):

* aiohttp server instrumentation
* aiohttp instrumented server that runs in background thread
* Extend HTTP propagator to support CIMultiDict
* First test for basic aiohttp client <--> server requests

* Add support for custom headers and param scrubbing

* Safer exception msg extraction

* Client safeties

* More test app routes

* Moar tests

* Loosen test reqs for platform variations

* Do not load asyncio app on unsupported versions

* Add Py 3.7 to tests

* Do not run spyne/sudsjurko on 3.7 until support is added

* Clean up version limiters

* Travis work-around for Py 3.7 tests

See travis-ci/travis-ci#9069 (comment)

* Better msg extraction for inconsisten exception usage across libraries

* Remove py3.7 because no rabbitmq support

Time is coming to eventually migrate to CircleCI

* Major/minor version specifiers only

* Update reported span names

* Update tests to follow changes.

* In case of webapp handler issue, act accordingly

* Include example application used for testing.

* Change port number so as not to conflict with test suite

Author: pglombardo

example/asyncio/README.md

@@ -0,0 +1,40 @@
+# Asyncio Examples
+
+This directory includes an example asyncio application and client with aiohttp and asynqp used for testing.
+
+# Requirements
+
+* Python 3.5 or greater
+* instana, aiohttp and asynqp Python packages installed
+* A RabbitMQ server with it's location specified in the `RABBITMQ_HOST` environment variable
+
+
+# Run
+
+* Make sure the Instana Python package is installed or you have this git repository checked out.
+
+* Set the environment variable `AUTOWRAPT_BOOTSTRAP=instana` for immediate instrumentation.
+
+* Boot the aiohttpserver.py file as follows.  It will launch an aiohttp server that listens on port localhost:5102.  See the source code for published endpoints.
+
+```bash
+python aiohttpserver.py
+```
+
+* Boot the `aiohttpclient.py` file to generate a request (every 1 second) to the aiohttp server.
+
+```bash
+python aiohttpclient.py
+```
+
+From here, you can modify the `aiohttpclient.py` file as needed to change requested paths and so on.
+
+# Results
+
+Some example traces from local tests.
+
+aiohttp client calling aiohttp server:
+![screen shot 2019-02-25 at 19 12 28](https://user-images.githubusercontent.com/395132/53401921-0f49cc00-39b1-11e9-8606-24844925a478.png)
+
+aiohttp server making multiple asynqp calls (publish & consume)
+![screen shot 2019-02-26 at 10 21 50](https://user-images.githubusercontent.com/395132/53401997-2e485e00-39b1-11e9-97fd-460b136cf92a.png)

example/asyncio/aioclient.py

@@ -0,0 +1,20 @@
+from __future__ import absolute_import
+
+import aiohttp
+import asyncio
+
+from instana.singletons import async_tracer, agent
+
+async def test():
+    while True:
+        await asyncio.sleep(1)
+        with async_tracer.start_active_span('JobRunner'):
+            async with aiohttp.ClientSession() as session:
+                    async with session.get("http://localhost:5102/?secret=iloveyou") as response:
+                        print(response.status)
+
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(test())
+loop.run_forever()
+

example/asyncio/aioserver.py

@@ -0,0 +1,77 @@
+import os
+import asyncio
+import asynqp
+from aiohttp import web
+
+RABBITMQ_HOST = ""
+if "RABBITMQ_HOST" in os.environ:
+    RABBITMQ_HOST = os.environ["RABBITMQ_HOST"]
+else:
+    RABBITMQ_HOST = "localhost"
+
+class RabbitUtil():
+
+    def __init__(self, loop):
+        self.loop = loop
+        self.loop.run_until_complete(self.connect())
+
+    @asyncio.coroutine
+    def connect(self):
+        # connect to the RabbitMQ broker
+        self.connection = yield from asynqp.connect(RABBITMQ_HOST, 5672, username='guest', password='guest')
+
+        # Open a communications channel
+        self.channel = yield from self.connection.open_channel()
+
+        # Create a queue and an exchange on the broker
+        self.exchange = yield from self.channel.declare_exchange('test.exchange', 'direct')
+        self.queue = yield from self.channel.declare_queue('test.queue')
+
+        # Bind the queue to the exchange, so the queue will get messages published to the exchange
+        yield from self.queue.bind(self.exchange, 'routing.key')
+        yield from self.queue.purge()
+
+
+@asyncio.coroutine
+def publish_msg(request):
+    msg = asynqp.Message({'hello': 'world'})
+    rabbit_util.exchange.publish(msg, 'routing.key')
+    rabbit_util.exchange.publish(msg, 'routing.key')
+    rabbit_util.exchange.publish(msg, 'routing.key')
+    rabbit_util.exchange.publish(msg, 'routing.key')
+
+    msg = yield from rabbit_util.queue.get()
+
+    return web.Response(text='Published 4 messages. Got 1. %s' % str(msg))
+
+
+async def say_hello(request):
+    return web.Response(text='Hello, world')
+
+
+async def four_hundred_one(request):
+    return web.HTTPUnauthorized(reason="I must simulate errors.", text="Simulated server error.")
+
+
+async def five_hundred(request):
+    return web.HTTPInternalServerError(reason="I must simulate errors.", text="Simulated server error.")
+
+
+loop = asyncio.new_event_loop()
+asyncio.set_event_loop(loop)
+
+rabbit_util = RabbitUtil(loop)
+
+app = web.Application(debug=False)
+app.add_routes([web.get('/', say_hello)])
+app.add_routes([web.get('/401', four_hundred_one)])
+app.add_routes([web.get('/500', five_hundred)])
+app.add_routes([web.get('/publish', publish_msg)])
+
+runner = web.AppRunner(app)
+loop.run_until_complete(runner.setup())
+site = web.TCPSite(runner, 'localhost', 5102)
+
+loop.run_until_complete(site.start())
+loop.run_forever()
+

instana/__init__.py

@@ -61,7 +61,9 @@ def boot_agent():
 
     if "INSTANA_DISABLE_AUTO_INSTR" not in os.environ:
         # Import & initialize instrumentation
-        if (sys.version_info >= (3, 4)) and (sys.version_info < (3, 7)):
+        if sys.version_info >= (3, 5, 3):
+            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

instana/http_propagator.py

@@ -55,19 +55,26 @@ def inject(self, span_context, carrier):
                 carrier.append((self.HEADER_KEY_S, span_id))
                 carrier.append((self.HEADER_KEY_L, "1"))
                 carrier.append((self.HEADER_KEY_ST, "intid;desc=%s" % trace_id))
+            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")
+                carrier.__setitem__(self.HEADER_KEY_ST, "intid;desc=%s" % trace_id)
             else:
                 raise Exception("Unsupported carrier type", type(carrier))
 
-        except Exception as e:
-            logger.debug("inject error: ", str(e))
+        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:
@@ -97,4 +104,4 @@ def extract(self, carrier):  # noqa
             return ctx
 
         except Exception as e:
-            logger.debug("extract error: ", str(e))
+            logger.debug("extract error:", exc_info=True)

instana/instrumentation/aiohttp/__init__.py

@@ -0,0 +1,79 @@
+from __future__ import absolute_import
+
+import opentracing
+import wrapt
+
+from ...log import logger
+from ...singletons import agent, async_tracer
+from ...util import strip_secrets
+
+
+try:
+    import aiohttp
+    import asyncio
+
+    async def stan_request_start(session, trace_config_ctx, params):
+        try:
+            parent_span = async_tracer.active_span
+
+            # If we're not tracing, just return
+            if parent_span is None:
+                return
+
+            scope = async_tracer.start_active_span("aiohttp-client", child_of=parent_span)
+            trace_config_ctx.scope = scope
+
+            async_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, params.headers)
+
+            parts = str(params.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', params.method)
+        except:
+            logger.debug("stan_request_start", exc_info=True)
+
+    async def stan_request_end(session, trace_config_ctx, params):
+        try:
+            scope = trace_config_ctx.scope
+            if scope is not None:
+                scope.span.set_tag('http.status_code', params.response.status)
+
+                if 400 <= params.response.status <= 599:
+                    scope.span.set_tag("http.error", params.response.reason)
+                    scope.span.set_tag("error", True)
+                    ec = scope.span.tags.get('ec', 0)
+                    scope.span.set_tag("ec", ec + 1)
+
+                scope.close()
+        except:
+            logger.debug("stan_request_end", exc_info=True)
+
+    async def stan_request_exception(session, trace_config_ctx, params):
+        try:
+            scope = trace_config_ctx.scope
+            if scope is not None:
+                scope.span.log_exception(params.exception)
+                scope.span.set_tag("http.error", str(params.exception))
+                scope.close()
+        except:
+            logger.debug("stan_request_exception", exc_info=True)
+
+    @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)
+        instana_trace_config.on_request_end.append(stan_request_end)
+        instana_trace_config.on_request_exception.append(stan_request_exception)
+        if 'trace_configs' in kwargs:
+            kwargs['trace_configs'].append(instana_trace_config)
+        else:
+            kwargs['trace_configs'] = [instana_trace_config]
+
+        return wrapped(*argv, **kwargs)
+
+    logger.debug("Instrumenting aiohttp client")
+except ImportError:
+    pass
+

instana/instrumentation/aiohttp/client.py

@@ -0,0 +1,73 @@
+from __future__ import absolute_import
+
+import opentracing
+import wrapt
+
+from ...log import logger
+from ...singletons import agent, async_tracer
+from ...util import strip_secrets
+
+
+try:
+    import aiohttp
+    import asyncio
+
+    from aiohttp.web import middleware
+
+    @middleware
+    async def stan_middleware(request, handler):
+        try:
+            ctx = async_tracer.extract(opentracing.Format.HTTP_HEADERS, request.headers)
+            request['scope'] = async_tracer.start_active_span('aiohttp-server', child_of=ctx)
+            scope = request['scope']
+
+            # Query param scrubbing
+            url = str(request.url)
+            parts = 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)
+
+            # Custom header tracking support
+            if agent.extra_headers is not None:
+                for custom_header in agent.extra_headers:
+                    if custom_header in request.headers:
+                        scope.span.set_tag("http.%s" % custom_header, request.headers[custom_header])
+
+            response = await handler(request)
+
+            if response is not None:
+                # Mark 500 responses as errored
+                if 500 <= response.status <= 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", response.status)
+                async_tracer.inject(scope.span.context, opentracing.Format.HTTP_HEADERS, response.headers)
+
+            return response
+        except:
+            logger.debug("aiohttp stan_middleware", exc_info=True)
+        finally:
+            if scope is not None:
+                scope.close()
+
+
+    @wrapt.patch_function_wrapper('aiohttp.web','Application.__init__')
+    def init_with_instana(wrapped, instance, argv, kwargs):
+        if "middlewares" in kwargs:
+            kwargs["middlewares"].append(stan_middleware)
+        else:
+            kwargs["middlewares"] = [stan_middleware]
+
+        return wrapped(*argv, **kwargs)
+
+    logger.debug("Instrumenting aiohttp server")
+except ImportError:
+    pass
+

instana/instrumentation/aiohttp/server.py

@@ -23,13 +23,14 @@
 
 
 class InstanaRecorder(SpanRecorder):
-    registered_spans = ("django", "memcache", "mysql", "rabbitmq", "redis",
-                        "rpc-client", "rpc-server", "sqlalchemy", "soap", "urllib3", "wsgi", "log")
-    http_spans = ("django", "wsgi", "urllib3", "soap")
-
-    exit_spans = ("memcache", "mysql", "rabbitmq", "redis", "rpc-client",
-                  "sqlalchemy", "soap", "urllib3", "log")
-    entry_spans = ("django", "wsgi", "rabbitmq", "rpc-server")
+    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")
+
+    exit_spans = ("aiohttp-client", "log", "memcache", "mysql", "rabbitmq", "redis", "rpc-client",
+                  "sqlalchemy", "soap", "urllib3")
+    entry_spans = ("aiohttp-server", "django", "wsgi", "rabbitmq", "rpc-server")
 
     entry_kind = ["entry", "server", "consumer"]
     exit_kind = ["exit", "client", "producer"]

instana/recorder.py

@@ -8,12 +8,10 @@ def finish(self, finish_time=None):
         super(InstanaSpan, self).finish(finish_time)
 
     def log_exception(self, e):
-        if hasattr(e, 'message') and len(e.message):
-            self.log_kv({'message': e.message})
-        elif hasattr(e, '__str__'):
-            self.log_kv({'message': e.__str__()})
-        else:
+        if hasattr(e, '__str__'):
             self.log_kv({'message': str(e)})
+        elif hasattr(e, 'message') and e.message is not None:
+            self.log_kv({'message': e.message})
 
         self.set_tag("error", True)
         ec = self.tags.get('ec', 0)