Merge pull request #123 from palazzem/msgpack-api

[core] provide support for trace-agent API v0.3

Author: Emanuele Palazzetti

ddtrace/api.py

@@ -3,7 +3,7 @@
 import time
 
 # project
-from .encoding import get_encoder
+from .encoding import get_encoder, JSONEncoder
 from .compat import httplib
 
 
@@ -14,43 +14,66 @@ class API(object):
     """
     Send data to the trace agent using the HTTP protocol and JSON format
     """
-    def __init__(self, hostname, port, wait_response=False, headers=None, encoder=None):
+    def __init__(self, hostname, port, headers=None, encoder=None):
         self.hostname = hostname
         self.port = port
+        self._traces = '/v0.3/traces'
+        self._services = '/v0.3/services'
+        self._compatibility_mode = False
         self._encoder = encoder or get_encoder()
-        self._wait_response = wait_response
 
         # overwrite the Content-type with the one chosen in the Encoder
         self._headers = headers or {}
         self._headers.update({'Content-Type': self._encoder.content_type})
 
+    def _downgrade(self):
+        """
+        Downgrades the used encoder and API level. This method must fallback to a safe
+        encoder and API, so that it will success despite users' configurations. This action
+        ensures that the compatibility mode is activated so that the downgrade will be
+        executed only once.
+        """
+        self._compatibility_mode = True
+        self._traces = '/v0.2/traces'
+        self._services = '/v0.2/services'
+        self._encoder = JSONEncoder()
+        self._headers.update({'Content-Type': self._encoder.content_type})
+
     def send_traces(self, traces):
         if not traces:
             return
         start = time.time()
         data = self._encoder.encode_traces(traces)
-        response = self._send_span_data(data)
+        response = self._put(self._traces, data)
+
+        # the API endpoint is not available so we should downgrade the connection and re-try the call
+        if response.status in [404, 415] and self._compatibility_mode is False:
+            log.debug('calling the endpoint "%s" but received %s; downgrading the API', self._traces, response.status)
+            self._downgrade()
+            return self.send_traces(traces)
+
         log.debug("reported %d spans in %.5fs", len(traces), time.time() - start)
         return response
 
     def send_services(self, services):
         if not services:
             return
-        log.debug("Reporting %d services", len(services))
         s = {}
         for service in services:
             s.update(service)
         data = self._encoder.encode_services(s)
-        return self._put("/v0.2/services", data)
+        response = self._put(self._services, data)
 
-    def _send_span_data(self, data):
-        return self._put("/v0.2/traces", data)
+        # the API endpoint is not available so we should downgrade the connection and re-try the call
+        if response.status in [404, 415] and self._compatibility_mode is False:
+            log.debug('calling the endpoint "%s" but received 404; downgrading the API', self._services)
+            self._downgrade()
+            return self.send_services(services)
+
+        log.debug("reported %d services", len(services))
+        return response
 
     def _put(self, endpoint, data):
         conn = httplib.HTTPConnection(self.hostname, self.port)
         conn.request("PUT", endpoint, data, self._headers)
-
-        # read the server response only if the
-        # API object is configured to do so
-        if self._wait_response:
-            return conn.getresponse()
+        return conn.getresponse()

ddtrace/encoding.py

@@ -1,18 +1,15 @@
-import os
 import json
 import logging
 
 
 # check msgpack CPP implementation; if the import fails, we're using the
 # pure Python implementation that is really slow, so the ``Encoder`` should use
-# a different encoding format. To enable msgpack encoding, you should set
-# the ``DD_MSGPACK_ENCODING=1`` environment variable otherwise, the ``JSONEncoder``
-# will be used as a default.
+# a different encoding format.
 try:
     import msgpack
     from msgpack._packer import Packer  # noqa
     from msgpack._unpacker import unpack, unpackb, Unpacker  # noqa
-    MSGPACK_ENCODING = os.getenv('DD_MSGPACK_ENCODING') == '1'  # shortcut to accept only '1'
+    MSGPACK_ENCODING = True
 except ImportError:
     MSGPACK_ENCODING = False
 

setup.py

@@ -58,6 +58,7 @@ def run_tests(self):
     packages=find_packages(exclude=['tests*']),
     install_requires=[
         "wrapt",
+        "msgpack-python",
     ],
     # plugin tox
     tests_require=['tox', 'flake8'],

tests/test_integration.py

@@ -10,7 +10,7 @@
 from ddtrace.api import API
 from ddtrace.span import Span
 from ddtrace.tracer import Tracer
-from ddtrace.encoding import JSONEncoder, MsgpackEncoder
+from ddtrace.encoding import JSONEncoder, MsgpackEncoder, get_encoder
 from tests.test_tracer import get_dummy_tracer
 
 
@@ -67,7 +67,7 @@ def test_worker_single_trace(self):
         # check arguments
         endpoint = self.api._put.call_args[0][0]
         payload = self._decode(self.api._put.call_args[0][1])
-        eq_(endpoint, '/v0.2/traces')
+        eq_(endpoint, '/v0.3/traces')
         eq_(len(payload), 1)
         eq_(len(payload[0]), 1)
         eq_(payload[0][0]['name'], 'client.testing')
@@ -84,7 +84,7 @@ def test_worker_multiple_traces(self):
         # check arguments
         endpoint = self.api._put.call_args[0][0]
         payload = self._decode(self.api._put.call_args[0][1])
-        eq_(endpoint, '/v0.2/traces')
+        eq_(endpoint, '/v0.3/traces')
         eq_(len(payload), 2)
         eq_(len(payload[0]), 1)
         eq_(len(payload[1]), 1)
@@ -104,7 +104,7 @@ def test_worker_single_trace_multiple_spans(self):
         # check arguments
         endpoint = self.api._put.call_args[0][0]
         payload = self._decode(self.api._put.call_args[0][1])
-        eq_(endpoint, '/v0.2/traces')
+        eq_(endpoint, '/v0.3/traces')
         eq_(len(payload), 1)
         eq_(len(payload[0]), 2)
         eq_(payload[0][0]['name'], 'client.testing')
@@ -122,7 +122,7 @@ def test_worker_single_service(self):
         # check arguments
         endpoint = self.api._put.call_args[0][0]
         payload = self._decode(self.api._put.call_args[0][1])
-        eq_(endpoint, '/v0.2/services')
+        eq_(endpoint, '/v0.3/services')
         eq_(len(payload.keys()), 1)
         eq_(payload['client.service'], {'app': 'django', 'app_type': 'web'})
 
@@ -139,7 +139,7 @@ def test_worker_service_called_multiple_times(self):
         # check arguments
         endpoint = self.api._put.call_args[0][0]
         payload = self._decode(self.api._put.call_args[0][1])
-        eq_(endpoint, '/v0.2/services')
+        eq_(endpoint, '/v0.3/services')
         eq_(len(payload.keys()), 2)
         eq_(payload['backend'], {'app': 'django', 'app_type': 'web'})
         eq_(payload['database'], {'app': 'postgres', 'app_type': 'db'})
@@ -161,8 +161,8 @@ def setUp(self):
         """
         # create a new API object to test the transport using synchronous calls
         self.tracer = get_dummy_tracer()
-        self.api_json = API('localhost', 7777, wait_response=True, encoder=JSONEncoder())
-        self.api_msgpack = API('localhost', 7777, wait_response=True, encoder=MsgpackEncoder())
+        self.api_json = API('localhost', 7777, encoder=JSONEncoder())
+        self.api_msgpack = API('localhost', 7777, encoder=MsgpackEncoder())
 
     def test_send_single_trace(self):
         # register a single trace with a span and send them to the trace agent
@@ -278,3 +278,44 @@ def test_send_service_called_multiple_times(self):
         response = self.api_msgpack.send_services(services)
         ok_(response)
         eq_(response.status, 200)
+
+@skipUnless(
+    os.environ.get('TEST_DATADOG_INTEGRATION', False),
+    'You should have a running trace agent and set TEST_DATADOG_INTEGRATION=1 env variable'
+)
+class TestAPIDowngrade(TestCase):
+    """
+    Ensures that if the tracing client found an earlier trace agent,
+    it will downgrade the current connection to a stable API version
+    """
+    def test_get_encoder_default(self):
+        # get_encoder should return MsgpackEncoder instance if
+        # msgpack and the CPP implementaiton are available
+        encoder = get_encoder()
+        ok_(isinstance(encoder, MsgpackEncoder))
+
+    @mock.patch('ddtrace.encoding.MSGPACK_ENCODING', False)
+    def test_get_encoder_fallback(self):
+        # get_encoder should return JSONEncoder instance if
+        # msgpack or the CPP implementaiton, are not available
+        encoder = get_encoder()
+        ok_(isinstance(encoder, JSONEncoder))
+
+    def test_downgrade_api(self):
+        # make a call to a not existing endpoint, downgrades
+        # the current API to a stable one
+        tracer = get_dummy_tracer()
+        tracer.trace('client.testing').finish()
+        trace = tracer.writer.pop()
+
+        # the encoder is right but we're targeting an API
+        # endpoint that is not available
+        api = API('localhost', 7777)
+        api._traces = '/v0.0/traces'
+        ok_(isinstance(api._encoder, MsgpackEncoder))
+
+        # after the call, we downgrade to a working endpoint
+        response = api.send_traces([trace])
+        ok_(response)
+        eq_(response.status, 200)
+        ok_(isinstance(api._encoder, JSONEncoder))

tox.ini

@@ -9,6 +9,10 @@
 envlist =
     flake8
     wait
+
+    {py27,py34}-tracer
+    {py27,py34}-integration
+    {py27,py34}-contrib
     {py27,py34}-bottle{12}-webtest
     {py27,py34}-cassandra
     {py27,py34}-elasticsearch{23}
@@ -25,7 +29,6 @@ envlist =
     {py27,py34}-sqlalchemy{10,11}-psycopg2
     {py27,py34}-redis
     {py27,py34}-sqlite3
-    {py27,py34}-all
 
 [testenv]
 basepython =
@@ -36,25 +39,25 @@ deps =
 # test dependencies installed in all envs
     mock
     nose
-    msgpack-python<0.4.9
+    msgpack-python
 # integrations
-    all: blinker
-    all: bottle
-    all: cassandra-driver
-    all: elasticsearch
-    all: falcon
-    all: flask
-    all: flask_cache
-    all: mongoengine
-    all: mysql-connector
-    all: psycopg2
-    all: pylibmc
-    all: pymongo
-    all: python-memcached
-    all: redis
-    all: requests
-    all: sqlalchemy
-    all: WebTest
+    contrib: blinker
+    contrib: bottle
+    contrib: cassandra-driver
+    contrib: elasticsearch
+    contrib: falcon
+    contrib: flask
+    contrib: flask_cache
+    contrib: mongoengine
+    contrib: mysql-connector
+    contrib: psycopg2
+    contrib: pylibmc
+    contrib: pymongo
+    contrib: python-memcached
+    contrib: redis
+    contrib: requests
+    contrib: sqlalchemy
+    contrib: WebTest
     blinker: blinker
     bottle12: bottle>=0.12
     cassandra: cassandra-driver
@@ -98,8 +101,12 @@ passenv=TEST_*
 commands =
 # wait for services script
     {py34}-wait: python tests/wait-for-services.py
+# run only essential tests related to the tracing client
+    {py27,py34}-tracer: nosetests {posargs} --exclude=".*(contrib|integration).*" tests/
+# integration tests
+    {py27,py34}-integration: nosetests {posargs} tests/test_integration.py
 # run all tests for the release jobs except the ones with a different test runner
-    {py27,py34}-all: nosetests {posargs} --exclude=".*(django).*"
+    {py27,py34}-contrib: nosetests {posargs} --exclude=".*(django).*" tests/contrib/
 # run subsets of the tests for particular library versions
     {py27,py34}-bottle{12}: nosetests {posargs} tests/contrib/bottle/
     {py27,py34}-cassandra: nosetests {posargs} tests/contrib/cassandra
@@ -137,5 +144,5 @@ basepython=python
 
 [flake8]
 ignore=W391,E231,E201,E202,E203,E261,E302,E128,E126,E124
-max-line-length=100
+max-line-length=120
 exclude = tests