Release 0.20.2 (#804)

brettlangdon · web-flow · commit 639decbb4f60 · 2019-01-29T16:03:36.000-05:00
* [tests] limit grpcio version to >=1.8.0,<1.18.0 * [tools] Add confirmation to 'rake pypi:release' task (#791) * [tools] Add confirmation to 'rake pypi:release' task * check if we are on tagged version instead * [core] Call HTTPResponse.read() before HTTPConnection.close() (#800) * [core] Call HTTPResponse.read() before HTTPConnection.close() * engrish * add reason and msg as well * use API.Response in integration tests * limit version of grpc * ensure we have regression test * move API.Response to just Response * result_traces/services => traces/services_response * fix logging error message tests * fix test for python 2.7 * Update ddtrace/writer.py Co-Authored-By: brettlangdon <me@brett.is> * Update ddtrace/api.py Co-Authored-By: brettlangdon <me@brett.is> * fix integration tests * Bump version to 0.20.2 (#803)
diff --git a/Rakefile b/Rakefile
@@ -117,6 +117,31 @@ end
 namespace :pypi do
   RELEASE_DIR = '/tmp/dd-trace-py-release'
 
+  def get_version()
+    return `python setup.py --version`.strip
+  end
+
+  def get_branch()
+    return `git name-rev --name-only HEAD`.strip
+  end
+
+  task :confirm do
+    ddtrace_version = get_version
+
+    if get_branch.downcase != 'tags/v#{ddtrace_version}'
+      print "WARNING: Expected current commit to be tagged as 'tags/v#{ddtrace_version}, instead we are on '#{get_branch}', proceed anyways [y|N]? "
+      $stdout.flush
+
+      abort if $stdin.gets.to_s.strip.downcase != 'y'
+    end
+
+    puts "WARNING: This task will build and release a new wheel to https://pypi.org/project/ddtrace/, this action cannot be undone"
+    print "         To proceed please type the version '#{ddtrace_version}': "
+    $stdout.flush
+
+    abort if $stdin.gets.to_s.strip.downcase != ddtrace_version
+  end
+
   task :clean do
     FileUtils.rm_rf(RELEASE_DIR)
   end
@@ -130,7 +155,7 @@ namespace :pypi do
     sh "python setup.py -q sdist -d #{RELEASE_DIR}"
   end
 
-  task :release => [:install, :build] do
+  task :release => [:confirm, :install, :build] do
     builds = Dir.entries(RELEASE_DIR).reject {|f| f == '.' || f == '..'}
     if builds.length == 0
         fail "no build found in #{RELEASE_DIR}"
diff --git a/ddtrace/__init__.py b/ddtrace/__init__.py
@@ -4,7 +4,7 @@
 from .tracer import Tracer
 from .settings import config
 
-__version__ = '0.20.1'
+__version__ = '0.20.2'
 
 # a global tracer instance with integration settings
 tracer = Tracer()
diff --git a/ddtrace/api.py b/ddtrace/api.py
@@ -27,27 +27,71 @@
                       'fallback': None}}
 
 
-def _parse_response_json(response):
+class Response(object):
     """
-    Parse the content of a response object, and return the right type,
-    can be a string if the output was plain text, or a dictionnary if
-    the output was a JSON.
+    Custom API Response object to represent a response from calling the API.
+
+    We do this to ensure we know expected properties will exist, and so we
+    can call `resp.read()` and load the body once into an instance before we
+    close the HTTPConnection used for the request.
     """
-    if hasattr(response, 'read'):
-        body = response.read()
+    __slots__ = ['status', 'body', 'reason', 'msg']
+
+    def __init__(self, status=None, body=None, reason=None, msg=None):
+        self.status = status
+        self.body = body
+        self.reason = reason
+        self.msg = msg
+
+    @classmethod
+    def from_http_response(cls, resp):
+        """
+        Build a ``Response`` from the provided ``HTTPResponse`` object.
+
+        This function will call `.read()` to consume the body of the ``HTTPResponse`` object.
+
+        :param resp: ``HTTPResponse`` object to build the ``Response`` from
+        :type resp: ``HTTPResponse``
+        :rtype: ``Response``
+        :returns: A new ``Response``
+        """
+        return cls(
+            status=resp.status,
+            body=resp.read(),
+            reason=getattr(resp, 'reason', None),
+            msg=getattr(resp, 'msg', None),
+        )
+
+    def get_json(self):
+        """Helper to parse the body of this request as JSON"""
         try:
+            body = self.body
+            if not body:
+                log.debug('Empty reply from Datadog Agent, %r', self)
+                return
+
             if not isinstance(body, str) and hasattr(body, 'decode'):
                 body = body.decode('utf-8')
+
             if hasattr(body, 'startswith') and body.startswith('OK'):
                 # This typically happens when using a priority-sampling enabled
                 # library with an outdated agent. It still works, but priority sampling
                 # will probably send too many traces, so the next step is to upgrade agent.
-                log.debug("'OK' is not a valid JSON, please make sure trace-agent is up to date")
+                log.debug('Cannot parse Datadog Agent response, please make sure your Datadog Agent is up to date')
                 return
-            content = loads(body)
-            return content
+
+            return loads(body)
         except (ValueError, TypeError) as err:
-            log.debug("unable to load JSON '%s': %s" % (body, err))
+            log.debug('Unable to parse Datadog Agent JSON response: %s %r', err, body)
+
+    def __repr__(self):
+        return '{0}(status={1!r}, body={2!r}, reason={3!r}, msg={4!r})'.format(
+            self.__class__.__name__,
+            self.status,
+            self.body,
+            self.reason,
+            self.msg,
+        )
 
 
 class API(object):
@@ -142,6 +186,11 @@ def _put(self, endpoint, data, count=0):
                 headers[TRACE_COUNT_HEADER] = str(count)
 
             conn.request("PUT", endpoint, data, headers)
-            return get_connection_response(conn)
+
+            # Parse the HTTPResponse into an API.Response
+            # DEV: This will call `resp.read()` which must happen before the `conn.close()` below,
+            #      if we call `.close()` then all future `.read()` calls will return `b''`
+            resp = get_connection_response(conn)
+            return Response.from_http_response(resp)
         finally:
             conn.close()
diff --git a/ddtrace/writer.py b/ddtrace/writer.py
@@ -8,8 +8,6 @@
 
 from ddtrace import api
 
-from .api import _parse_response_json
-
 log = logging.getLogger(__name__)
 
 
@@ -126,8 +124,8 @@ def _on_shutdown(self):
                     time.sleep(0.05)
 
     def _target(self):
-        result_traces = None
-        result_services = None
+        traces_response = None
+        services_response = None
 
         while True:
             traces = self._trace_queue.pop()
@@ -141,43 +139,50 @@ def _target(self):
             if traces:
                 # If we have data, let's try to send it.
                 try:
-                    result_traces = self.api.send_traces(traces)
+                    traces_response = self.api.send_traces(traces)
                 except Exception as err:
                     log.error("cannot send spans to {1}:{2}: {0}".format(err, self.api.hostname, self.api.port))
 
             services = self._service_queue.pop()
             if services:
                 try:
-                    result_services = self.api.send_services(services)
+                    services_response = self.api.send_services(services)
                 except Exception as err:
                     log.error("cannot send services to {1}:{2}: {0}".format(err, self.api.hostname, self.api.port))
 
             if self._trace_queue.closed() and self._trace_queue.size() == 0:
                 # no traces and the queue is closed. our work is done
                 return
 
-            if self._priority_sampler:
-                result_traces_json = _parse_response_json(result_traces)
+            if self._priority_sampler and traces_response:
+                result_traces_json = traces_response.get_json()
                 if result_traces_json and 'rate_by_service' in result_traces_json:
                     self._priority_sampler.set_sample_rate_by_service(result_traces_json['rate_by_service'])
 
-            self._log_error_status(result_traces, "traces")
-            result_traces = None
-            self._log_error_status(result_services, "services")
-            result_services = None
+            self._log_error_status(traces_response, "traces")
+            traces_response = None
+            self._log_error_status(services_response, "services")
+            services_response = None
 
             time.sleep(1)  # replace with a blocking pop.
 
-    def _log_error_status(self, result, result_name):
+    def _log_error_status(self, response, response_name):
+        if not isinstance(response, api.Response):
+            return
+
         log_level = log.debug
-        if result and getattr(result, "status", None) >= 400:
+        if response.status >= 400:
             now = time.time()
             if now > self._last_error_ts + LOG_ERR_INTERVAL:
                 log_level = log.error
                 self._last_error_ts = now
-            log_level("failed_to_send %s to Agent: HTTP error status %s, reason %s, message %s", result_name,
-                      getattr(result, "status", None), getattr(result, "reason", None),
-                      getattr(result, "msg", None))
+            log_level(
+                'failed_to_send %s to Datadog Agent: HTTP error status %s, reason %s, message %s',
+                response_name,
+                response.status,
+                response.reason,
+                response.msg,
+            )
 
     def _apply_filters(self, traces):
         """
diff --git a/tests/test_api.py b/tests/test_api.py
@@ -1,16 +1,18 @@
 import mock
+import re
 import warnings
 
 from unittest import TestCase
 from nose.tools import eq_, ok_
 
 from tests.test_tracer import get_dummy_tracer
-from ddtrace.api import _parse_response_json, API
+from ddtrace.api import API, Response
 from ddtrace.compat import iteritems, httplib
 
 
 class ResponseMock:
-    def __init__(self, content):
+    def __init__(self, content, status=200):
+        self.status = status
         self.content = content
 
     def read(self):
@@ -34,36 +36,49 @@ def test_parse_response_json(self, log):
         tracer.debug_logging = True
 
         test_cases = {
-            'OK': {'js': None, 'log': "please make sure trace-agent is up to date"},
-            'OK\n': {'js': None, 'log': "please make sure trace-agent is up to date"},
-            'error:unsupported-endpoint': {'js': None, 'log': "unable to load JSON 'error:unsupported-endpoint'"},
-            42: {'js': None, 'log': "unable to load JSON '42'"},  # int as key to trigger TypeError
-            '{}': {'js': {}},
-            '[]': {'js': []},
-            '{"rate_by_service": {"service:,env:":0.5, "service:mcnulty,env:test":0.9, "service:postgres,env:test":0.6}}': {  # noqa
-                'js': {
-                    'rate_by_service': {
+            'OK': dict(
+                js=None,
+                log='Cannot parse Datadog Agent response, please make sure your Datadog Agent is up to date',
+            ),
+            'OK\n': dict(
+                js=None,
+                log='Cannot parse Datadog Agent response, please make sure your Datadog Agent is up to date',
+            ),
+            'error:unsupported-endpoint': dict(
+                js=None,
+                log='Unable to parse Datadog Agent JSON response: .*? \'error:unsupported-endpoint\'',
+            ),
+            42: dict(  # int as key to trigger TypeError
+                js=None,
+                log='Unable to parse Datadog Agent JSON response: .*? 42',
+            ),
+            '{}': dict(js={}),
+            '[]': dict(js=[]),
+
+            # Priority sampling "rate_by_service" response
+            ('{"rate_by_service": '
+             '{"service:,env:":0.5, "service:mcnulty,env:test":0.9, "service:postgres,env:test":0.6}}'): dict(
+                js=dict(
+                    rate_by_service={
                         'service:,env:': 0.5,
                         'service:mcnulty,env:test': 0.9,
                         'service:postgres,env:test': 0.6,
                     },
-                },
-            },
-            ' [4,2,1] ': {'js': [4, 2, 1]},
+                ),
+            ),
+            ' [4,2,1] ': dict(js=[4, 2, 1]),
         }
 
         for k, v in iteritems(test_cases):
-            r = ResponseMock(k)
-            js = _parse_response_json(r)
+            log.reset_mock()
+
+            r = Response.from_http_response(ResponseMock(k))
+            js = r.get_json()
             eq_(v['js'], js)
             if 'log' in v:
-                ok_(
-                    1 <= len(log.call_args_list),
-                    'not enough elements in call_args_list: {}'.format(log.call_args_list),
-                )
-                print(log.call_args_list)
-                args = log.call_args_list[-1][0][0]
-                ok_(v['log'] in args, 'unable to find {} in {}'.format(v['log'], args))
+                log.assert_called_once()
+                msg = log.call_args[0][0] % log.call_args[0][1:]
+                ok_(re.match(v['log'], msg), msg)
 
     @mock.patch('ddtrace.compat.httplib.HTTPConnection')
     def test_put_connection_close(self, HTTPConnection):
diff --git a/tests/test_integration.py b/tests/test_integration.py
@@ -9,7 +9,7 @@
 from unittest import TestCase, skip, skipUnless
 from nose.tools import eq_, ok_
 
-from ddtrace.api import API
+from ddtrace.api import API, Response
 from ddtrace.ext import http
 from ddtrace.filters import FilterRequestsOnUrl
 from ddtrace.constants import FILTERS_KEY
@@ -41,7 +41,7 @@ class FlawedAPI(API):
     def _put(self, endpoint, data, count=0):
         conn = httplib.HTTPConnection(self.hostname, self.port)
         conn.request('HEAD', endpoint, data, self._headers)
-        return conn.getresponse()
+        return Response.from_http_response(conn.getresponse())
 
 
 @skipUnless(
@@ -188,7 +188,7 @@ def test_worker_http_error_logging(self):
 
         logged_errors = log_handler.messages['error']
         eq_(len(logged_errors), 1)
-        ok_('failed_to_send traces to Agent: HTTP error status 400, reason Bad Request, message Content-Type:'
+        ok_('failed_to_send traces to Datadog Agent: HTTP error status 400, reason Bad Request, message Content-Type:'
             in logged_errors[0])
 
     def test_worker_filter_request(self):
@@ -488,8 +488,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', 8126, encoder=JSONEncoder())
-        self.api_msgpack = API('localhost', 8126, encoder=MsgpackEncoder())
+        self.api_json = API('localhost', 8126, encoder=JSONEncoder(), priority_sampling=True)
+        self.api_msgpack = API('localhost', 8126, encoder=MsgpackEncoder(), priority_sampling=True)
 
     def test_send_single_trace(self):
         # register a single trace with a span and send them to the trace agent
@@ -506,11 +506,13 @@ def test_send_single_trace(self):
         response = self.api_json.send_traces(traces)
         ok_(response)
         eq_(response.status, 200)
+        eq_(response.get_json(), dict(rate_by_service={'service:,env:': 1}))
 
         # test Msgpack encoder
         response = self.api_msgpack.send_traces(traces)
         ok_(response)
         eq_(response.status, 200)
+        eq_(response.get_json(), dict(rate_by_service={'service:,env:': 1}))
 
 
 @skipUnless(
diff --git a/tox.ini b/tox.ini
@@ -202,7 +202,7 @@ deps =
     gevent11: gevent>=1.1,<1.2
     gevent12: gevent>=1.2,<1.3
     gevent13: gevent>=1.3,<1.4
-    grpc: grpcio>=1.8.0
+    grpc: grpcio>=1.8.0,<1.18.0
     grpc: googleapis-common-protos
     jinja27: jinja2>=2.7,<2.8
     jinja28: jinja2>=2.8,<2.9
