Merge pull request #740 from DataDog/0.17-dev

v0.17.0

Author: Kyle-Verhoog

.circleci/config.yml

@@ -4,7 +4,7 @@
 from .tracer import Tracer
 from .settings import config
 
-__version__ = '0.16.0'
+__version__ = '0.17.0'
 
 # a global tracer instance with integration settings
 tracer = Tracer()

ddtrace/__init__.py

@@ -43,6 +43,18 @@ def update_patched_modules():
         EXTRA_PATCHED_MODULES.update({module: should_patch.lower() == 'true'})
 
 
+def add_global_tags(tracer):
+    tags = {}
+    for tag in os.environ.get('DD_TRACE_GLOBAL_TAGS', '').split(','):
+        tag_name, _, tag_value = tag.partition(':')
+        if not tag_name or not tag_value:
+            log.debug('skipping malformed tracer tag')
+            continue
+
+        tags[tag_name] = tag_value
+    tracer.set_tags(tags)
+
+
 try:
     from ddtrace import tracer
     patch = True
@@ -51,7 +63,7 @@ def update_patched_modules():
     # TODO: these variables are deprecated; use utils method and update our documentation
     # correct prefix should be DD_*
     enabled = os.environ.get("DATADOG_TRACE_ENABLED")
-    hostname = os.environ.get("DATADOG_TRACE_AGENT_HOSTNAME")
+    hostname = os.environ.get('DD_AGENT_HOST', os.environ.get('DATADOG_TRACE_AGENT_HOSTNAME'))
     port = os.environ.get("DATADOG_TRACE_AGENT_PORT")
     priority_sampling = os.environ.get("DATADOG_PRIORITY_SAMPLING")
 
@@ -81,6 +93,9 @@ def update_patched_modules():
     if 'DATADOG_ENV' in os.environ:
         tracer.set_tags({"env": os.environ["DATADOG_ENV"]})
 
+    if 'DD_TRACE_GLOBAL_TAGS' in os.environ:
+        add_global_tags(tracer)
+
     # Ensure sitecustomize.py is properly called if available in application directories:
     # * exclude `bootstrap_dir` from the search
     # * find a user `sitecustomize.py` module

ddtrace/bootstrap/sitecustomize.py

@@ -2,17 +2,12 @@
 Generic dbapi tracing code.
 """
 
-# stdlib
 import logging
 
-# 3p
 import wrapt
 
-# project
 from ddtrace import Pin
-from ddtrace.ext import sql
-
-from ...ext import AppTypes
+from ddtrace.ext import AppTypes, sql
 
 log = logging.getLogger(__name__)
 
@@ -24,38 +19,92 @@ def __init__(self, cursor, pin):
         super(TracedCursor, self).__init__(cursor)
         pin.onto(self)
         name = pin.app or 'sql'
-        self._self_datadog_name = '%s.query' % name
-
-    def _trace_method(self, method, resource, extra_tags, *args, **kwargs):
+        self._self_datadog_name = '{}.query'.format(name)
+        self._self_last_execute_operation = None
+
+    def _trace_method(self, method, name, resource, extra_tags, *args, **kwargs):
+        """
+        Internal function to trace the call to the underlying cursor method
+        :param method: The callable to be wrapped
+        :param name: The name of the resulting span.
+        :param resource: The sql query. Sql queries are obfuscated on the agent side.
+        :param extra_tags: A dict of tags to store into the span's meta
+        :param args: The args that will be passed as positional args to the wrapped method
+        :param kwargs: The args that will be passed as kwargs to the wrapped method
+        :return: The result of the wrapped method invocation
+        """
         pin = Pin.get_from(self)
         if not pin or not pin.enabled():
             return method(*args, **kwargs)
         service = pin.service
-
-        with pin.tracer.trace(self._self_datadog_name, service=service, resource=resource) as s:
+        with pin.tracer.trace(name, service=service, resource=resource) as s:
             s.span_type = sql.TYPE
+            # No reason to tag the query since it is set as the resource by the agent. See:
+            # https://github.com/DataDog/datadog-trace-agent/blob/bda1ebbf170dd8c5879be993bdd4dbae70d10fda/obfuscate/sql.go#L232
             s.set_tags(pin.tags)
             s.set_tags(extra_tags)
 
             try:
                 return method(*args, **kwargs)
             finally:
-                s.set_metric("db.rowcount", self.rowcount)
+                row_count = self.__wrapped__.rowcount
+                s.set_metric('db.rowcount', row_count)
+                # Necessary for django integration backward compatibility. Django integration used to provide its own
+                # implementation of the TracedCursor, which used to store the row count into a tag instead of
+                # as a metric. Such custom implementation has been replaced by this generic dbapi implementation and
+                # this tag has been added since.
+                if row_count and row_count >= 0:
+                    s.set_tag(sql.ROWS, row_count)
 
     def executemany(self, query, *args, **kwargs):
+        """ Wraps the cursor.executemany method"""
+        self._self_last_execute_operation = query
         # FIXME[matt] properly handle kwargs here. arg names can be different
         # with different libs.
-        return self._trace_method(
-            self.__wrapped__.executemany, query, {'sql.executemany': 'true'},
+        self._trace_method(
+            self.__wrapped__.executemany, self._self_datadog_name, query, {'sql.executemany': 'true'},
             query, *args, **kwargs)
+        return self
 
     def execute(self, query, *args, **kwargs):
-        return self._trace_method(
-            self.__wrapped__.execute, query, {}, query, *args, **kwargs)
+        """ Wraps the cursor.execute method"""
+        self._self_last_execute_operation = query
+        self._trace_method(self.__wrapped__.execute, self._self_datadog_name, query, {}, query, *args, **kwargs)
+        return self
+
+    def fetchone(self, *args, **kwargs):
+        """ Wraps the cursor.fetchone method"""
+        span_name = '{}.{}'.format(self._self_datadog_name, 'fetchone')
+        return self._trace_method(self.__wrapped__.fetchone, span_name, self._self_last_execute_operation, {},
+                                  *args, **kwargs)
+
+    def fetchall(self, *args, **kwargs):
+        """ Wraps the cursor.fetchall method"""
+        span_name = '{}.{}'.format(self._self_datadog_name, 'fetchall')
+        return self._trace_method(self.__wrapped__.fetchall, span_name, self._self_last_execute_operation, {},
+                                  *args, **kwargs)
+
+    def fetchmany(self, *args, **kwargs):
+        """ Wraps the cursor.fetchmany method"""
+        span_name = '{}.{}'.format(self._self_datadog_name, 'fetchmany')
+        # We want to trace the information about how many rows were requested. Note that this number may be larger
+        # the number of rows actually returned if less then requested are available from the query.
+        size_tag_key = 'db.fetch.size'
+        if 'size' in kwargs:
+            extra_tags = {size_tag_key: kwargs.get('size')}
+        elif len(args) == 1 and isinstance(args[0], int):
+            extra_tags = {size_tag_key: args[0]}
+        else:
+            default_array_size = getattr(self.__wrapped__, 'arraysize', None)
+            extra_tags = {size_tag_key: default_array_size} if default_array_size else {}
+
+        return self._trace_method(self.__wrapped__.fetchmany, span_name, self._self_last_execute_operation, extra_tags,
+                                  *args, **kwargs)
 
     def callproc(self, proc, args):
-        return self._trace_method(self.__wrapped__.callproc, proc, {}, proc,
-                                  args)
+        """ Wraps the cursor.callproc method"""
+        self._self_last_execute_operation = proc
+        return self._trace_method(self.__wrapped__.callproc, self._self_datadog_name, proc, {}, proc, args)
 
     def __enter__(self):
         # previous versions of the dbapi didn't support context managers. let's
@@ -73,16 +122,36 @@ class TracedConnection(wrapt.ObjectProxy):
     def __init__(self, conn, pin=None):
         super(TracedConnection, self).__init__(conn)
         name = _get_vendor(conn)
+        self._self_datadog_name = '{}.connection'.format(name)
         db_pin = pin or Pin(service=name, app=name, app_type=AppTypes.db)
         db_pin.onto(self)
 
+    def _trace_method(self, method, name, extra_tags, *args, **kwargs):
+        pin = Pin.get_from(self)
+        if not pin or not pin.enabled():
+            return method(*args, **kwargs)
+        service = pin.service
+
+        with pin.tracer.trace(name, service=service) as s:
+            s.set_tags(pin.tags)
+            s.set_tags(extra_tags)
+
+            return method(*args, **kwargs)
+
     def cursor(self, *args, **kwargs):
         cursor = self.__wrapped__.cursor(*args, **kwargs)
         pin = Pin.get_from(self)
         if not pin:
             return cursor
         return TracedCursor(cursor, pin)
 
+    def commit(self, *args, **kwargs):
+        span_name = '{}.{}'.format(self._self_datadog_name, 'commit')
+        return self._trace_method(self.__wrapped__.commit, span_name, {}, *args, **kwargs)
+
+    def rollback(self, *args, **kwargs):
+        span_name = '{}.{}'.format(self._self_datadog_name, 'rollback')
+        return self._trace_method(self.__wrapped__.rollback, span_name, {}, *args, **kwargs)
 
 def _get_vendor(conn):
     """ Return the vendor (e.g postgres, mysql) of the given

ddtrace/contrib/dbapi/__init__.py

@@ -41,7 +41,7 @@ def patch_cache(tracer):
           Django supported cache servers (Redis, Memcached, Database, Custom)
     """
     # discover used cache backends
-    cache_backends = [cache['BACKEND'] for cache in django_settings.CACHES.values()]
+    cache_backends = set([cache['BACKEND'] for cache in django_settings.CACHES.values()])
 
     def _trace_operation(fn, method_name):
         """
@@ -102,7 +102,7 @@ def unpatch_method(cls, method_name):
     delattr(cls, DATADOG_NAMESPACE.format(method=method_name))
 
 def unpatch_cache():
-    cache_backends = [cache['BACKEND'] for cache in django_settings.CACHES.values()]
+    cache_backends = set([cache['BACKEND'] for cache in django_settings.CACHES.values()])
     for cache_module in cache_backends:
         cache = import_from_string(cache_module, cache_module)
 

ddtrace/contrib/django/cache.py

@@ -87,15 +87,19 @@ def __init__(self, user_settings=None, defaults=None, import_strings=None):
             self.defaults['TAGS'].update({'env': os.environ.get('DATADOG_ENV')})
         if os.environ.get('DATADOG_SERVICE_NAME'):
             self.defaults['DEFAULT_SERVICE'] = os.environ.get('DATADOG_SERVICE_NAME')
-        if os.environ.get('DATADOG_TRACE_AGENT_HOSTNAME'):
-            self.defaults['AGENT_HOSTNAME'] = os.environ.get('DATADOG_TRACE_AGENT_HOSTNAME')
-        if os.environ.get('DATADOG_TRACE_AGENT_PORT'):
+
+        host = os.environ.get('DD_AGENT_HOST', os.environ.get('DATADOG_TRACE_AGENT_HOSTNAME'))
+        if host:
+            self.defaults['AGENT_HOSTNAME'] = host
+
+        port = os.environ.get('DD_TRACE_AGENT_PORT', os.environ.get('DATADOG_TRACE_AGENT_PORT'))
+        if port:
             # if the agent port is a string, the underlying library that creates the socket
             # stops working
             try:
-                port = int(os.environ.get('DATADOG_TRACE_AGENT_PORT'))
+                port = int(port)
             except ValueError:
-                log.warning('DATADOG_TRACE_AGENT_PORT is not an integer value; default to 8126')
+                log.warning('DD_TRACE_AGENT_PORT is not an integer value; default to 8126')
             else:
                 self.defaults['AGENT_PORT'] = port
 

ddtrace/contrib/django/conf.py

@@ -8,7 +8,8 @@
 from ...ext import AppTypes
 
 from .conf import settings
-
+from ..dbapi import TracedCursor as DbApiTracedCursor
+from ddtrace import Pin
 
 log = logging.getLogger(__name__)
 
@@ -30,6 +31,7 @@ def all_connections(self):
 
     connections.all = all_connections.__get__(connections, type(connections))
 
+
 def unpatch_db():
     for c in connections.all():
         unpatch_conn(c)
@@ -41,95 +43,42 @@ def unpatch_db():
     connections.all = all_connections
     delattr(connections, ALL_CONNS_ATTR)
 
+
 def patch_conn(tracer, conn):
     if hasattr(conn, CURSOR_ATTR):
         return
 
     setattr(conn, CURSOR_ATTR, conn.cursor)
 
     def cursor():
-        return TracedCursor(tracer, conn, conn._datadog_original_cursor())
-
-    conn.cursor = cursor
-
-def unpatch_conn(conn):
-    cursor = getattr(conn, CURSOR_ATTR, None)
-    if cursor is None:
-        log.debug('nothing to do, the connection is not patched')
-        return
-    conn.cursor = cursor
-    delattr(conn, CURSOR_ATTR)
-
-class TracedCursor(object):
-
-    def __init__(self, tracer, conn, cursor):
-        self.tracer = tracer
-        self.conn = conn
-        self.cursor = cursor
-
-        self._vendor = getattr(conn, 'vendor', 'db')     # e.g sqlite, postgres
-        self._alias = getattr(conn, 'alias', 'default')  # e.g. default, users
-
-        prefix = sqlx.normalize_vendor(self._vendor)
-        self._name = "%s.%s" % (prefix, "query")  # e.g sqlite.query
-
         database_prefix = (
             '{}-'.format(settings.DEFAULT_DATABASE_PREFIX)
             if settings.DEFAULT_DATABASE_PREFIX else ''
         )
-
-        self._service = "%s%s%s" % (
-            database_prefix,
-            self._alias,
-            "db"
-        )  # e.g. service-defaultdb or service-postgresdb
-
-        self.tracer.set_service_info(
-            service=self._service,
+        alias = getattr(conn, 'alias', 'default')
+        service = '{}{}{}'.format(database_prefix, alias, 'db')
+        vendor = getattr(conn, 'vendor', 'db')
+        prefix = sqlx.normalize_vendor(vendor)
+        tags = {
+            'django.db.vendor': vendor,
+            'django.db.alias': alias,
+        }
+        tracer.set_service_info(
+            service=service,
             app=prefix,
             app_type=AppTypes.db,
         )
 
-    def _trace(self, func, sql, params):
-        span = self.tracer.trace(
-            self._name,
-            resource=sql,
-            service=self._service,
-            span_type=sqlx.TYPE
-        )
-
-        with span:
-            # No reason to tag the query since it is set as the resource by the agent. See:
-            # https://github.com/DataDog/datadog-trace-agent/blob/bda1ebbf170dd8c5879be993bdd4dbae70d10fda/obfuscate/sql.go#L232
-            span.set_tag("django.db.vendor", self._vendor)
-            span.set_tag("django.db.alias", self._alias)
-            try:
-                return func(sql, params)
-            finally:
-                rows = self.cursor.cursor.rowcount
-                if rows and 0 <= rows:
-                    span.set_tag(sqlx.ROWS, self.cursor.cursor.rowcount)
-
-    def callproc(self, procname, params=None):
-        return self._trace(self.cursor.callproc, procname, params)
-
-    def execute(self, sql, params=None):
-        return self._trace(self.cursor.execute, sql, params)
-
-    def executemany(self, sql, param_list):
-        return self._trace(self.cursor.executemany, sql, param_list)
-
-    def close(self):
-        return self.cursor.close()
-
-    def __getattr__(self, attr):
-        return getattr(self.cursor, attr)
+        pin = Pin(service, tags=tags, tracer=tracer, app=prefix)
+        return DbApiTracedCursor(conn._datadog_original_cursor(), pin)
 
-    def __iter__(self):
-        return iter(self.cursor)
+    conn.cursor = cursor
 
-    def __enter__(self):
-        return self
 
-    def __exit__(self, type, value, traceback):
-        self.close()
+def unpatch_conn(conn):
+    cursor = getattr(conn, CURSOR_ATTR, None)
+    if cursor is None:
+        log.debug('nothing to do, the connection is not patched')
+        return
+    conn.cursor = cursor
+    delattr(conn, CURSOR_ATTR)