Merge branch '0.13-dev' into 'master'

Author: Emanuele Palazzetti

.circleci/config.yml

@@ -521,6 +521,27 @@ jobs:
           paths:
             - .tox
 
+  pymemcache:
+    docker:
+      - image: datadog/docker-library:dd_trace_py_1_0_0
+      - image: memcached:1.4
+    steps:
+      - checkout
+      - restore_cache:
+          keys:
+              - tox-cache-pymemcache-{{ checksum "tox.ini" }}
+      - run: tox -e '{py27,py34,py35,py36}-pymemcache{130,140}' --result-json /tmp/pymemcache.1.results
+      - run: tox -e '{py27,py34,py35,py36}-pymemcache-autopatch{130,140}' --result-json /tmp/pymemcache.2.results
+      - persist_to_workspace:
+          root: /tmp
+          paths:
+            - pymemcache.1.results
+            - pymemcache.2.results
+      - save_cache:
+          key: tox-cache-pymemcache-{{ checksum "tox.ini" }}
+          paths:
+            - .tox
+
   pymongo:
     docker:
       - image: datadog/docker-library:dd_trace_py_1_0_0
@@ -806,6 +827,7 @@ workflows:
       - mysqlconnector
       - mysqlpython
       - mysqldb
+      - pymemcache
       - pymysql
       - pylibmc
       - pymongo
@@ -844,6 +866,7 @@ workflows:
             - mysqldb
             - pymysql
             - pylibmc
+            - pymemcache
             - pymongo
             - pyramid
             - requests

ddtrace/api.py

@@ -6,7 +6,7 @@
 
 # project
 from .encoding import get_encoder, JSONEncoder
-from .compat import httplib, PYTHON_VERSION, PYTHON_INTERPRETER
+from .compat import httplib, PYTHON_VERSION, PYTHON_INTERPRETER, get_connection_response
 
 
 log = logging.getLogger(__name__)
@@ -140,4 +140,4 @@ def _put(self, endpoint, data, count=0):
             headers[TRACE_COUNT_HEADER] = str(count)
 
         conn.request("PUT", endpoint, data, headers)
-        return conn.getresponse()
+        return get_connection_response(conn)

ddtrace/compat.py

@@ -70,6 +70,24 @@ def to_unicode(s):
     return stringify(s)
 
 
+def get_connection_response(conn):
+    """Returns the response for a connection.
+
+    If using Python 2 enable buffering.
+
+    Python 2 does not enable buffering by default resulting in many recv
+    syscalls.
+
+    See:
+    https://bugs.python.org/issue4879
+    https://github.com/python/cpython/commit/3c43fcba8b67ea0cec4a443c755ce5f25990a6cf
+    """
+    if PY2:
+        return conn.getresponse(buffering=True)
+    else:
+        return conn.getresponse()
+
+
 if PY2:
     string_type = basestring
     msgpack_type = basestring

ddtrace/contrib/pymemcache/__init__.py

@@ -0,0 +1,30 @@
+"""Instrument pymemcache to report memcached queries.
+
+``patch_all`` will automatically patch the pymemcache ``Client``::
+
+    from ddtrace import Pin, patch
+
+    # If not patched yet, patch pymemcache specifically
+    patch(pymemcache=True)
+
+    # Import reference to Client AFTER patching
+    import pymemcache
+    from pymemcache.client.base import Client
+
+    # Use a pin to specify metadata related all clients
+    Pin.override(pymemcache, service='my-memcached-service')
+
+    # This will report a span with the default settings
+    client = Client(('localhost', 11211))
+    client.set("my-key", "my-val")
+
+    # Use a pin to specify metadata related to this particular client
+    Pin.override(client, service='my-memcached-service')
+
+Pymemcache's ``HashClient`` will also be indirectly patched as it uses
+``Client``s under the hood.
+"""
+
+from .patch import patch, unpatch
+
+__all__ = [patch, unpatch]

ddtrace/contrib/pymemcache/client.py

@@ -0,0 +1,212 @@
+# stdlib
+import logging
+import sys
+
+# 3p
+import wrapt
+import pymemcache
+from pymemcache.client.base import Client
+from pymemcache.exceptions import (
+    MemcacheClientError,
+    MemcacheServerError,
+    MemcacheUnknownCommandError,
+    MemcacheUnknownError,
+    MemcacheIllegalInputError,
+)
+
+# project
+from ddtrace import Pin
+from ddtrace.compat import reraise
+from ddtrace.ext import net, memcached as memcachedx
+
+log = logging.getLogger(__name__)
+
+
+# keep a reference to the original unpatched clients
+_Client = Client
+
+
+class WrappedClient(wrapt.ObjectProxy):
+    """Wrapper providing patched methods of a pymemcache Client.
+
+    Relevant connection information is obtained during initialization and
+    attached to each span.
+
+    Keys are tagged in spans for methods that act upon a key.
+    """
+
+    def __init__(self, *args, **kwargs):
+        c = _Client(*args, **kwargs)
+        super(WrappedClient, self).__init__(c)
+
+        # tags to apply to each span generated by this client
+        tags = _get_address_tags(*args, **kwargs)
+
+        parent_pin = Pin.get_from(pymemcache)
+
+        if parent_pin:
+            pin = parent_pin.clone(tags=tags)
+        else:
+            pin = Pin(tags=tags)
+
+        # attach the pin onto this instance
+        pin.onto(self)
+
+    def set(self, *args, **kwargs):
+        return self._traced_cmd("set", *args, **kwargs)
+
+    def set_many(self, *args, **kwargs):
+        return self._traced_cmd("set_many", *args, **kwargs)
+
+    def add(self, *args, **kwargs):
+        return self._traced_cmd("add", *args, **kwargs)
+
+    def replace(self, *args, **kwargs):
+        return self._traced_cmd("replace", *args, **kwargs)
+
+    def append(self, *args, **kwargs):
+        return self._traced_cmd("append", *args, **kwargs)
+
+    def prepend(self, *args, **kwargs):
+        return self._traced_cmd("prepend", *args, **kwargs)
+
+    def cas(self, *args, **kwargs):
+        return self._traced_cmd("cas", *args, **kwargs)
+
+    def get(self, *args, **kwargs):
+        return self._traced_cmd("get", *args, **kwargs)
+
+    def get_many(self, *args, **kwargs):
+        return self._traced_cmd("get_many", *args, **kwargs)
+
+    def gets(self, *args, **kwargs):
+        return self._traced_cmd("gets", *args, **kwargs)
+
+    def gets_many(self, *args, **kwargs):
+        return self._traced_cmd("gets_many", *args, **kwargs)
+
+    def delete(self, *args, **kwargs):
+        return self._traced_cmd("delete", *args, **kwargs)
+
+    def delete_many(self, *args, **kwargs):
+        return self._traced_cmd("delete_many", *args, **kwargs)
+
+    def incr(self, *args, **kwargs):
+        return self._traced_cmd("incr", *args, **kwargs)
+
+    def decr(self, *args, **kwargs):
+        return self._traced_cmd("decr", *args, **kwargs)
+
+    def touch(self, *args, **kwargs):
+        return self._traced_cmd("touch", *args, **kwargs)
+
+    def stats(self, *args, **kwargs):
+        return self._traced_cmd("stats", *args, **kwargs)
+
+    def version(self, *args, **kwargs):
+        return self._traced_cmd("version", *args, **kwargs)
+
+    def flush_all(self, *args, **kwargs):
+        return self._traced_cmd("flush_all", *args, **kwargs)
+
+    def quit(self, *args, **kwargs):
+        return self._traced_cmd("quit", *args, **kwargs)
+
+    def set_multi(self, *args, **kwargs):
+        """set_multi is an alias for set_many"""
+        return self._traced_cmd("set_many", *args, **kwargs)
+
+    def get_multi(self, *args, **kwargs):
+        """set_multi is an alias for set_many"""
+        return self._traced_cmd("get_many", *args, **kwargs)
+
+    def _traced_cmd(self, method_name, *args, **kwargs):
+        """Run and trace the given command.
+
+        Any pymemcache exception is caught and span error information is
+        set. The exception is then reraised for the application to handle
+        appropriately.
+
+        Relevant tags are set in the span.
+        """
+        method = getattr(self.__wrapped__, method_name)
+        p = Pin.get_from(self)
+
+        # if the pin does not exist or is not enabled, shortcut
+        if not p or not p.enabled():
+            return method(*args, **kwargs)
+
+        with p.tracer.trace(
+            memcachedx.CMD,
+            service=p.service,
+            resource=method_name,
+            span_type=memcachedx.TYPE,
+        ) as span:
+            # try to set relevant tags, catch any exceptions so we don't mess
+            # with the application
+            try:
+                span.set_tags(p.tags)
+                vals = _get_query_string(args)
+                query = "{}{}{}".format(method_name, " " if vals else "", vals)
+                span.set_tag(memcachedx.QUERY, query)
+            except Exception:
+                log.debug("Error setting relevant pymemcache tags")
+
+            try:
+                return method(*args, **kwargs)
+            except (
+                MemcacheClientError,
+                MemcacheServerError,
+                MemcacheUnknownCommandError,
+                MemcacheUnknownError,
+                MemcacheIllegalInputError,
+            ):
+                (typ, val, tb) = sys.exc_info()
+                span.set_exc_info(typ, val, tb)
+                reraise(typ, val, tb)
+
+
+def _get_address_tags(*args, **kwargs):
+    """Attempt to get host and port from args passed to Client initializer."""
+    tags = {}
+    try:
+        if len(args):
+            host, port = args[0]
+            tags[net.TARGET_HOST] = host
+            tags[net.TARGET_PORT] = port
+    except Exception:
+        log.debug("Error collecting client address tags")
+
+    return tags
+
+
+def _get_query_string(args):
+    """Return the query values given the arguments to a pymemcache command.
+
+    If there are multiple query values, they are joined together
+    space-separated.
+    """
+    keys = ""
+
+    # shortcut if no args
+    if not args:
+        return keys
+
+    # pull out the first arg which will contain any key
+    arg = args[0]
+
+    # if we get a dict, convert to list of keys
+    if type(arg) is dict:
+        arg = list(arg)
+
+    if type(arg) is str:
+        keys = arg
+    elif type(arg) is bytes:
+        keys = arg.decode()
+    elif type(arg) is list and len(arg):
+        if type(arg[0]) is str:
+            keys = " ".join(arg)
+        elif type(arg[0]) is bytes:
+            keys = b" ".join(arg).decode()
+
+    return keys

ddtrace/contrib/pymemcache/patch.py

@@ -0,0 +1,32 @@
+import pymemcache
+
+from ddtrace.ext import memcached as memcachedx
+from ddtrace.pin import Pin, _DD_PIN_NAME, _DD_PIN_PROXY_NAME
+from .client import WrappedClient
+
+_Client = pymemcache.client.base.Client
+
+
+def patch():
+    if getattr(pymemcache.client, "_datadog_patch", False):
+        return
+
+    setattr(pymemcache.client, "_datadog_patch", True)
+    setattr(pymemcache.client.base, "Client", WrappedClient)
+
+    # Create a global pin with default configuration for our pymemcache clients
+    Pin(
+        app=memcachedx.SERVICE, service=memcachedx.SERVICE, app_type=memcachedx.TYPE
+    ).onto(pymemcache)
+
+
+def unpatch():
+    """Remove pymemcache tracing"""
+    if not getattr(pymemcache.client, "_datadog_patch", False):
+        return
+    setattr(pymemcache.client, "_datadog_patch", False)
+    setattr(pymemcache.client.base, "Client", _Client)
+
+    # Remove any pins that may exist on the pymemcache reference
+    setattr(pymemcache, _DD_PIN_NAME, None)
+    setattr(pymemcache, _DD_PIN_PROXY_NAME, None)

ddtrace/ext/memcached.py

@@ -1,7 +1,4 @@
-
-from ddtrace.ext import AppTypes
-
+CMD = "memcached.command"
 SERVICE = "memcached"
-TYPE = AppTypes.cache
-
+TYPE = "memcached"
 QUERY = "memcached.query"

ddtrace/monkey.py

@@ -29,6 +29,7 @@
     'pymysql': True,
     'psycopg': True,
     'pylibmc': True,
+    'pymemcache': True,
     'pymongo': True,
     'redis': True,
     'requests': False,  # Not ready yet

ddtrace/tracer.py

@@ -1,6 +1,6 @@
 import functools
 import logging
-from os import getpid
+from os import environ, getpid
 
 from .ext import system
 from .provider import DefaultContextProvider
@@ -27,7 +27,7 @@ class Tracer(object):
         from ddtrace import tracer
         trace = tracer.trace("app.request", "web-server").finish()
     """
-    DEFAULT_HOSTNAME = 'localhost'
+    DEFAULT_HOSTNAME = environ.get('DATADOG_TRACE_AGENT_HOSTNAME', 'localhost')
     DEFAULT_PORT = 8126
 
     def __init__(self):