new redis contrib integration

circle.yml

@@ -6,9 +6,11 @@ machine:
 dependencies:
   pre:
     - sudo service postgresql stop
+    - sudo service redis-server stop
 test:
   override:
     - docker run -d -p 9200:9200 elasticsearch:2.3
     - docker run -d -p 5432:5432 -e POSTGRES_PASSWORD=test -e POSTGRES_USER=test -e POSTGRES_DB=test postgres:9.5
+    - docker run -d -p 6379:6379 redis:3.2
     - python2.7 setup.py test
     - python3.4 setup.py test

ddtrace/contrib/redis/__init__.py

@@ -0,0 +1,9 @@
+from ..util import require_modules
+
+required_modules = ['redis', 'redis.client']
+
+with require_modules(required_modules) as missing_modules:
+    if not missing_modules:
+        from .tracers import get_traced_redis, get_traced_redis_from
+
+        __all__ = ['get_traced_redis', 'get_traced_redis_from']

ddtrace/contrib/redis/tracers.py

@@ -0,0 +1,113 @@
+"""
+tracers exposed publicly
+"""
+# stdlib
+import time
+
+from redis import StrictRedis
+from redis.client import StrictPipeline
+
+# dogtrace
+from .util import format_command_args, _extract_conn_tags
+from ...ext import redis as redisx
+
+
+DEFAULT_SERVICE = 'redis'
+
+
+def get_traced_redis(ddtracer, service=DEFAULT_SERVICE, meta=None):
+    return _get_traced_redis(ddtracer, StrictRedis, service, meta)
+
+
+def get_traced_redis_from(ddtracer, baseclass, service=DEFAULT_SERVICE, meta=None):
+    return _get_traced_redis(ddtracer, baseclass, service, meta)
+
+# pylint: disable=protected-access
+def _get_traced_redis(ddtracer, baseclass, service, meta):
+    class TracedPipeline(StrictPipeline):
+        _datadog_tracer = ddtracer
+        _datadog_service = service
+        _datadog_meta = meta
+
+        def __init__(self, *args, **kwargs):
+            self._datadog_pipeline_creation = time.time()
+            super(TracedPipeline, self).__init__(*args, **kwargs)
+
+        def execute(self, *args, **kwargs):
+            commands, queries = [], []
+            with self._datadog_tracer.trace('redis.pipeline') as s:
+                if s.sampled:
+                    s.service = self._datadog_service
+                    s.span_type = redisx.TYPE
+
+                    for cargs, _ in self.command_stack:
+                        commands.append(cargs[0])
+                        queries.append(format_command_args(cargs))
+
+                    s.set_tag(redisx.CMD, ', '.join(commands))
+                    query = '\n'.join(queries)
+                    s.resource = query
+
+                    s.set_tags(_extract_conn_tags(self.connection_pool.connection_kwargs))
+                    s.set_tags(self._datadog_meta)
+                    # FIXME[leo]: convert to metric?
+                    s.set_tag(redisx.PIPELINE_LEN, len(self.command_stack))
+                    s.set_tag(redisx.PIPELINE_AGE, time.time()-self._datadog_pipeline_creation)
+
+                return super(TracedPipeline, self).execute(self, *args, **kwargs)
+
+        def immediate_execute_command(self, *args, **kwargs):
+            command_name = args[0]
+
+            with self._datadog_tracer.trace('redis.command') as s:
+                if s.sampled:
+                    s.service = self._datadog_service
+                    s.span_type = redisx.TYPE
+                    # currently no quantization on the client side
+                    s.resource = format_command_args(args)
+                    s.set_tag(redisx.CMD, (args or [None])[0])
+                    s.set_tags(_extract_conn_tags(self.connection_pool.connection_kwargs))
+                    s.set_tags(self._datadog_meta)
+                    # FIXME[leo]: convert to metric?
+                    s.set_tag(redisx.ARGS_LEN, len(args))
+
+                    s.set_tag(redisx.IMMEDIATE_PIPELINE, True)
+
+                return super(TracedPipeline, self).immediate_execute_command(*args, **options)
+
+    class TracedRedis(baseclass):
+        _datadog_tracer = ddtracer
+        _datadog_service = service
+        _datadog_meta = meta
+
+        @classmethod
+        def set_datadog_meta(cls, meta):
+            cls._datadog_meta = meta
+
+        def execute_command(self, *args, **options):
+            with self._datadog_tracer.trace('redis.command') as s:
+                if s.sampled:
+                    command_name = args[0]
+                    s.service = self._datadog_service
+                    s.span_type = redisx.TYPE
+                    # currently no quantization on the client side
+                    s.resource = format_command_args(args)
+                    s.set_tag(redisx.CMD, (args or [None])[0])
+                    s.set_tags(_extract_conn_tags(self.connection_pool.connection_kwargs))
+                    s.set_tags(self._datadog_meta)
+                    # FIXME[leo]: convert to metric?
+                    s.set_tag(redisx.ARGS_LEN, len(args))
+
+                return super(TracedRedis, self).execute_command(*args, **options)
+
+        def pipeline(self, transaction=True, shard_hint=None):
+            tp = TracedPipeline(
+                self.connection_pool,
+                self.response_callbacks,
+                transaction,
+                shard_hint
+            )
+            tp._datadog_meta = meta
+            return tp
+
+    return TracedRedis

ddtrace/contrib/redis/util.py

@@ -0,0 +1,51 @@
+"""
+Some utils used by the dogtrace redis integration
+"""
+from ...ext import redis as redisx, net
+
+VALUE_PLACEHOLDER = "?"
+VALUE_MAX_LENGTH = 100
+VALUE_TOO_LONG_MARK = "..."
+COMMAND_MAX_LENGTH = 1000
+
+
+def _extract_conn_tags(conn_kwargs):
+    """ Transform redis conn info into dogtrace metas """
+    try:
+        return {
+            net.TARGET_HOST: conn_kwargs['host'],
+            net.TARGET_PORT: conn_kwargs['port'],
+            redisx.DB: conn_kwargs['db'] or 0,
+        }
+    except Exception:
+        return {}
+
+
+def format_command_args(args):
+    """Format a command by removing unwanted values
+
+    Restrict what we keep from the values sent (with a SET, HGET, LPUSH, ...):
+      - Skip binary content
+      - Truncate
+    """
+    formatted_length = 0
+    formatted_args = []
+    for arg in args:
+        try:
+            command = unicode(arg)
+            if len(command) > VALUE_MAX_LENGTH:
+                command = command[:VALUE_MAX_LENGTH] + VALUE_TOO_LONG_MARK
+            if formatted_length + len(command) > COMMAND_MAX_LENGTH:
+                formatted_args.append(
+                    command[:COMMAND_MAX_LENGTH-formatted_length]
+                    + VALUE_TOO_LONG_MARK
+                )
+                break
+
+            formatted_args.append(command)
+            formatted_length += len(command)
+        except Exception:
+            formatted_args.append(VALUE_PLACEHOLDER)
+            break
+
+    return " ".join(formatted_args)

ddtrace/ext/redis.py

@@ -0,0 +1,12 @@
+# type of the spans
+TYPE = 'redis'
+
+# net extension
+DB = 'out.redis_db'
+
+# standard tags
+CMD = 'redis.command'
+ARGS_LEN = 'redis.args_length'
+PIPELINE_LEN = 'redis.pipeline_length'
+PIPELINE_AGE = 'redis.pipeline_age'
+IMMEDIATE_PIPELINE = 'redis.pipeline_immediate_command'

tests/contrib/redis/test.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import unittest
+
+from ddtrace.contrib.redis import missing_modules
+
+if missing_modules:
+    raise unittest.SkipTest("Missing dependencies %s" % missing_modules)
+
+import redis
+from nose.tools import eq_, ok_
+
+from ddtrace.tracer import Tracer
+from ddtrace.contrib.redis import get_traced_redis, get_traced_redis_from
+
+from ...test_tracer import DummyWriter
+
+
+class RedisTest(unittest.TestCase):
+    SERVICE = 'test-cache'
+
+    def setUp(self):
+        """ purge redis """
+        r = redis.Redis()
+        r.flushall()
+
+    def tearDown(self):
+        r = redis.Redis()
+        r.flushall()
+
+    def test_basic_class(self):
+        writer = DummyWriter()
+        tracer = Tracer(writer=writer)
+
+        TracedRedisCache = get_traced_redis(tracer, service=self.SERVICE)
+        r = TracedRedisCache()
+
+        us = r.get('cheese')
+        eq_(us, None)
+        spans = writer.pop()
+        eq_(len(spans), 1)
+        span = spans[0]
+        eq_(span.service, self.SERVICE)
+        eq_(span.name, 'redis.command')
+        eq_(span.span_type, 'redis')
+        eq_(span.error, 0)
+        eq_(span.meta, {'out.host': u'localhost', 'redis.command': u'GET', 'out.port': u'6379', 'redis.args_length': u'2', 'out.redis_db': u'0'})
+        eq_(span.resource, 'GET cheese')
+
+    def test_meta_override(self):
+        writer = DummyWriter()
+        tracer = Tracer(writer=writer)
+
+        TracedRedisCache = get_traced_redis(tracer, service=self.SERVICE, meta={'cheese': 'camembert'})
+        r = TracedRedisCache()
+
+        r.get('cheese')
+        spans = writer.pop()
+        eq_(len(spans), 1)
+        span = spans[0]
+        eq_(span.service, self.SERVICE)
+        ok_('cheese' in span.meta and span.meta['cheese'] == 'camembert')
+
+    def test_basic_class_pipeline(self):
+        writer = DummyWriter()
+        tracer = Tracer(writer=writer)
+
+        TracedRedisCache = get_traced_redis(tracer, service=self.SERVICE)
+        r = TracedRedisCache()
+
+        with r.pipeline() as p:
+            p.set('blah', 32)
+            p.rpush('foo', u'éé')
+            p.hgetall('xxx')
+
+            p.execute()
+
+        spans = writer.pop()
+        eq_(len(spans), 1)
+        span = spans[0]
+        eq_(span.service, self.SERVICE)
+        eq_(span.name, 'redis.pipeline')
+        eq_(span.span_type, 'redis')
+        eq_(span.error, 0)
+        eq_(span.get_tag('out.redis_db'), '0')
+        eq_(span.get_tag('out.host'), 'localhost')
+        ok_(float(span.get_tag('redis.pipeline_age')) > 0)
+        eq_(span.get_tag('redis.pipeline_length'), '3')
+        eq_(span.get_tag('redis.command'), 'SET, RPUSH, HGETALL')
+        eq_(span.get_tag('out.port'), '6379')
+        eq_(span.resource, u'SET blah 32\nRPUSH foo éé\nHGETALL xxx')
+
+    def test_custom_class(self):
+        class MyCustomRedis(redis.Redis):
+            def execute_command(self, *args, **kwargs):
+                response = super(MyCustomRedis, self).execute_command(*args, **kwargs)
+                return 'YO%sYO' % response
+
+
+        writer = DummyWriter()
+        tracer = Tracer(writer=writer)
+
+        TracedRedisCache = get_traced_redis_from(tracer, MyCustomRedis, service=self.SERVICE)
+        r = TracedRedisCache()
+
+        r.set('foo', 42)
+        resp = r.get('foo')
+        eq_(resp, 'YO42YO')
+
+        spans = writer.pop()
+        eq_(len(spans), 2)
+        eq_(spans[0].name, 'redis.command')
+        eq_(spans[0].resource, 'SET foo 42')
+        eq_(spans[1].name, 'redis.command')
+        eq_(spans[1].resource, 'GET foo')