Redis Instrumentation & Tests (#104)

* Initial Redis instrumentation, tests and infra

* Pipeline support; subCommands; More Tests

* Fix error logging

Author: pglombardo

instana/__init__.py

@@ -61,6 +61,7 @@ def load_instrumentation():
         # Import & initialize instrumentation
         from .instrumentation import asynqp  # 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

instana/instrumentation/redis.py

@@ -0,0 +1,79 @@
+from __future__ import absolute_import
+
+import opentracing
+import opentracing.ext.tags as ext
+import wrapt
+
+from ..log import logger
+from ..singletons import tracer
+
+try:
+    import redis
+
+    @wrapt.patch_function_wrapper('redis.client','StrictRedis.execute_command')
+    def execute_command_with_instana(wrapped, instance, args, kwargs):
+        parent_span = tracer.active_span
+
+        # If we're not tracing, just return
+        if parent_span is None:
+            return wrapped(*args, **kwargs)
+
+        with tracer.start_active_span("redis", child_of=parent_span) as scope:
+
+            try:
+                ckw = instance.connection_pool.connection_kwargs
+                url = "redis://%s:%d/%d" % (ckw['host'], ckw['port'], ckw['db'])
+                scope.span.set_tag("connection", url)
+                scope.span.set_tag("driver", "redis-py")
+                scope.span.set_tag("command", args[0])
+
+                rv = wrapped(*args, **kwargs)
+            except Exception as e:
+                scope.span.set_tag("redis.error", str(e))
+                scope.span.set_tag("error", True)
+                ec = scope.span.tags.get('ec', 0)
+                scope.span.set_tag("ec", ec+1)
+                raise
+            else:
+                return rv
+
+    @wrapt.patch_function_wrapper('redis.client','BasePipeline.execute')
+    def execute_with_instana(wrapped, instance, args, kwargs):
+        parent_span = tracer.active_span
+
+        # If we're not tracing, just return
+        if parent_span is None:
+            return wrapped(*args, **kwargs)
+
+        with tracer.start_active_span("redis", child_of=parent_span) as scope:
+
+            try:
+                ckw = instance.connection_pool.connection_kwargs
+                url = "redis://%s:%d/%d" % (ckw['host'], ckw['port'], ckw['db'])
+                scope.span.set_tag("connection", url)
+                scope.span.set_tag("driver", "redis-py")
+                scope.span.set_tag("command", 'PIPELINE')
+
+                try:
+                    pipe_cmds = []
+                    for e in instance.command_stack:
+                        pipe_cmds.append(e[0][0])
+                    scope.span.set_tag("subCommands", pipe_cmds)
+                except Exception as e:
+                    # If anything breaks during cmd collection, just log a
+                    # debug message
+                    logger.debug("Error collecting pipeline commands")
+
+                rv = wrapped(*args, **kwargs)
+            except Exception as e:
+                scope.span.set_tag("redis.error", str(e))
+                scope.span.set_tag("error", True)
+                ec = scope.span.tags.get('ec', 0)
+                scope.span.set_tag("ec", ec+1)
+                raise
+            else:
+                return rv
+
+    logger.debug("Instrumenting redis")
+except ImportError:
+    pass

instana/json_span.py

@@ -30,6 +30,7 @@ class Data(object):
     custom = None
     http = None
     rabbitmq = None
+    redis = None
     sdk = None
     service = None
     sqlalchemy = None
@@ -72,6 +73,18 @@ class RabbitmqData(object):
     def __init__(self, **kwds):
         self.__dict__.update(kwds)
 
+
+class RedisData(object):
+    connection = None
+    driver = None
+    command = None
+    error = None
+    subCommands = None
+
+    def __init__(self, **kwds):
+        self.__dict__.update(kwds)
+
+
 class SQLAlchemyData(object):
     sql = None
     url = None

instana/recorder.py

@@ -12,7 +12,8 @@
 import instana.singletons
 
 from .json_span import (CustomData, Data, HttpData, JsonSpan, MySQLData,
-                        RabbitmqData, SDKData, SoapData, SQLAlchemyData)
+                        RabbitmqData, RedisData, SDKData, SoapData,
+                        SQLAlchemyData)
 from .log import logger
 
 if sys.version_info.major is 2:
@@ -22,12 +23,12 @@
 
 
 class InstanaRecorder(SpanRecorder):
-    registered_spans = ("django", "memcache", "mysql", "rabbitmq", "rpc-client",
-                        "rpc-server", "sqlalchemy", "soap", "urllib3", "wsgi")
+    registered_spans = ("django", "memcache", "mysql", "rabbitmq", "redis",
+                        "rpc-client", "rpc-server", "sqlalchemy", "soap", "urllib3", "wsgi")
     http_spans = ("django", "wsgi", "urllib3", "soap")
 
-    exit_spans = ("memcache", "mysql", "rabbitmq", "rpc-client", "sqlalchemy",
-                  "soap", "urllib3")
+    exit_spans = ("memcache", "mysql", "rabbitmq", "redis", "rpc-client",
+                  "sqlalchemy", "soap", "urllib3")
     entry_spans = ("django", "wsgi", "rabbitmq", "rpc-server")
 
     entry_kind = ["entry", "server", "consumer"]
@@ -115,6 +116,13 @@ def build_registered_span(self, span):
                                          address=span.tags.pop('address', None),
                                          key=span.tags.pop('key', None))
 
+        if span.operation_name == "redis":
+            data.redis = RedisData(connection=span.tags.pop('connection', None),
+                                   driver=span.tags.pop('driver', None),
+                                   command=span.tags.pop('command', None),
+                                   error=span.tags.pop('redis.error', None),
+                                   subCommands=span.tags.pop('subCommands', None))
+
         if span.operation_name == "sqlalchemy":
             data.sqlalchemy = SQLAlchemyData(sql=span.tags.pop('sqlalchemy.sql', None),
                                              eng=span.tags.pop('sqlalchemy.eng', None),

setup.py

@@ -57,6 +57,7 @@ def check_setuptools():
             'psycopg2>=2.7.1',
             'pyOpenSSL>=16.1.0;python_version<="2.7"',
             'pytest>=3.0.1',
+            'redis>=2.10.6',
             'requests>=2.17.1',
             'sqlalchemy>=1.1.15',
             'spyne>=2.9',

tests/helpers.py

@@ -43,3 +43,11 @@
     testenv['postgresql_pw'] = os.environ['TRAVIS_POSTGRESQL_PASS']
 else:
     testenv['postgresql_pw'] = ''
+
+"""
+Redis Environment
+"""
+if 'REDIS' in os.environ:
+    testenv['redis_url']= os.environ['REDIS']
+else:
+    testenv['redis_url'] = '127.0.0.1:6379'