removed old, urllib2-based transport class, and renamed urllib3 transport (#107)

closes #107

Author: beniwohli

CHANGELOG.asciidoc

@@ -16,6 +16,7 @@ https://github.com/elastic/apm-agent-python/compare/v1.0.0.dev3\...master[Check
  * changed name that the agent reports itself with to the APM server from `elasticapm-python` to `python`. This aligns the Python agent with other languages. ({pull}104[#104])
  * changed Celery integration to store the task state (e.g. `SUCCESS` or `FAILURE`) in `transaction.result` ({pull}100[#100])
  * BREAKING: renamed `server` configuration variable to `server_url` to better align with other language agents ({pull}105[#105]) 
+ * BREAKING: removed the old and unused urllib2-based HTTP transport, and renamed the urllib3 transport ({pull}107[#107])
 
 [[release-v1.0.0.dev3]]
 [float]

docs/configuration.asciidoc

@@ -77,14 +77,14 @@ The URL must be fully qualified, including protocol (`http` or `https`) and port
 
 |============
 | Environment                   | Django/Flask      | Default
-| `ELASTIC_APM_TRANSPORT_CLASS` | `TRANSPORT_CLASS` | `elasticapm.transport.http_urllib3.AsyncUrllib3Transport`
+| `ELASTIC_APM_TRANSPORT_CLASS` | `TRANSPORT_CLASS` | `elasticapm.transport.http.AsyncTransport`
 |============
 
 
 The transport class to use when sending events to the APM server.
-The default `AsyncUrllib3Transport` uses a background thread to send data.
+The default `AsyncTransport` uses a background thread to send data.
 If your environment doesn't allow background threads, you can use
-`elasticapm.transport.http_urllib3.Urllib3Transport` instead.
+`elasticapm.transport.http.Transport` instead.
 Note however that this can have adverse effects on performance.
 
 

elasticapm/base.py

@@ -300,12 +300,12 @@ def _get_log_message(self, data):
         return message
 
     def _get_transport(self, parsed_url):
-        if self._transport_class.async_mode and is_master_process():
+        if hasattr(self._transport_class, 'sync_transport') and is_master_process():
             # when in the master process, always use SYNC mode. This avoids
             # the danger of being forked into an inconsistent threading state
             self.logger.info('Sending message synchronously while in master '
                              'process. PID: %s', os.getpid())
-            return import_string(defaults.SYNC_TRANSPORT_CLASS)(parsed_url)
+            return self._transport_class.sync_transport(parsed_url)
         if parsed_url not in self._transports:
             self._transports[parsed_url] = self._transport_class(parsed_url)
         return self._transports[parsed_url]

elasticapm/conf/__init__.py

@@ -143,7 +143,7 @@ class Config(_ConfigBase):
     hostname = _ConfigValue('HOSTNAME', default=socket.gethostname())
     auto_log_stacks = _BoolConfigValue('AUTO_LOG_STACKS', default=True)
     keyword_max_length = _ConfigValue('KEYWORD_MAX_LENGTH', type=int, default=1024)
-    transport_class = _ConfigValue('TRANSPORT_CLASS', default='elasticapm.transport.http_urllib3.AsyncUrllib3Transport',
+    transport_class = _ConfigValue('TRANSPORT_CLASS', default='elasticapm.transport.http.AsyncTransport',
                                    required=True)
     processors = _ListConfigValue('PROCESSORS', default=[
         'elasticapm.processors.sanitize_stacktrace_locals',

elasticapm/conf/defaults.py

@@ -79,8 +79,4 @@
 # Should elasticapm wrap middleware for better metrics detection
 INSTRUMENT_DJANGO_MIDDLEWARE = True
 
-SYNC_TRANSPORT_CLASS = 'elasticapm.transport.http_urllib3.Urllib3Transport'
-
-ASYNC_TRANSPORT_CLASS = 'elasticapm.transport.http_urllib3.AsyncUrllib3Transport'
-
 TIMESTAMP_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'

elasticapm/transport/__init__.py

@@ -1,6 +1 @@
 # -*- coding: utf-8 -*-
-
-from elasticapm.transport.http import AsyncHTTPTransport, HTTPTransport
-
-
-default = [HTTPTransport, AsyncHTTPTransport]

elasticapm/transport/asyncio.py

@@ -5,19 +5,16 @@
 from elasticapm.conf import defaults
 
 from .base import TransportException
-from .http import HTTPTransport
+from .http_base import HTTPTransportBase
 
 
-class AsyncioHTTPTransport(HTTPTransport):
+class AsyncioHTTPTransport(HTTPTransportBase):
     """
     HTTP Transport ready for asyncio
     """
 
     def __init__(self, parsed_url):
-        self.check_scheme(parsed_url)
-
-        self._parsed_url = parsed_url
-        self._url = parsed_url.geturl()
+        super(AsyncioHTTPTransport, self).__init__(parsed_url)
         loop = asyncio.get_event_loop()
         self.client = aiohttp.ClientSession(loop=loop)
 

elasticapm/transport/http.py

@@ -1,110 +1,86 @@
 # -*- coding: utf-8 -*-
 import logging
-import socket
+import os
 
-from elasticapm.conf import defaults
-from elasticapm.contrib.async_worker import AsyncWorker
-from elasticapm.transport.base import (AsyncTransport, Transport,
-                                       TransportException)
-from elasticapm.utils.compat import HTTPError
-
-try:
-    from urllib2 import Request, urlopen
-except ImportError:
-    from urllib.request import Request, urlopen
+import certifi
+import urllib3
+from urllib3.exceptions import MaxRetryError, TimeoutError
 
+from elasticapm.conf import defaults
+from elasticapm.transport.base import TransportException
+from elasticapm.transport.http_base import (AsyncHTTPTransportBase,
+                                            HTTPTransportBase)
+from elasticapm.utils import compat
 
-logger = logging.getLogger('elasticapm')
+logger = logging.getLogger(__name__)
 
 
-class HTTPTransport(Transport):
+class Transport(HTTPTransportBase):
 
     scheme = ['http', 'https']
 
     def __init__(self, parsed_url):
-        self.check_scheme(parsed_url)
-
-        self._parsed_url = parsed_url
-        self._url = parsed_url.geturl()
+        kwargs = {
+            'cert_reqs': 'CERT_REQUIRED',
+            'ca_certs': certifi.where(),
+            'block': True,
+        }
+        proxy_url = os.environ.get('HTTPS_PROXY', os.environ.get('HTTP_PROXY'))
+        if proxy_url:
+            self.http = urllib3.ProxyManager(proxy_url, **kwargs)
+        else:
+            self.http = urllib3.PoolManager(**kwargs)
+        super(Transport, self).__init__(parsed_url)
 
     def send(self, data, headers, timeout=None):
-        """
-        Sends a request to a remote webserver using HTTP POST.
-
-        Returns the shortcut URL of the recorded error on Elastic APM
-        """
-        req = Request(self._url, headers=headers)
         if timeout is None:
             timeout = defaults.TIMEOUT
         response = None
+
+        # ensure headers are byte strings
+        headers = {k.encode('ascii') if isinstance(k, compat.text_type) else k:
+                   v.encode('ascii') if isinstance(v, compat.text_type) else v
+                   for k, v in headers.items()}
+        if compat.PY2 and isinstance(self._url, compat.text_type):
+            url = self._url.encode('utf-8')
+        else:
+            url = self._url
         try:
             try:
-                response = urlopen(req, data, timeout)
-            except TypeError:
-                response = urlopen(req, data)
-        except Exception as e:
-            print_trace = True
-            if isinstance(e, socket.timeout):
-                message = (
-                    "Connection to APM Server timed out "
-                    "(url: %s, timeout: %d seconds)" % (self._url, timeout)
+                response = self.http.urlopen(
+                    'POST', url, body=data, headers=headers, timeout=timeout, preload_content=False
                 )
-            elif isinstance(e, HTTPError):
-                body = e.read()
-                if e.code == 429:  # rate-limited
+                logger.info('Sent request, url=%s size=%.2fkb status=%s', url, len(data) / 1024.0, response.status)
+            except Exception as e:
+                print_trace = True
+                if isinstance(e, MaxRetryError) and isinstance(e.reason, TimeoutError):
+                    message = (
+                        "Connection to APM Server timed out "
+                        "(url: %s, timeout: %d seconds)" % (self._url, timeout)
+                    )
+                    print_trace = False
+                else:
+                    message = 'Unable to reach APM Server: %s (url: %s)' % (
+                        e, self._url
+                    )
+                raise TransportException(message, data, print_trace=print_trace)
+            body = response.read()
+            if response.status >= 400:
+                if response.status == 429:  # rate-limited
                     message = 'Temporarily rate limited: '
                     print_trace = False
                 else:
-                    message = 'Unable to reach APM Server: '
-                message += '%s (url: %s, body: %s)' % (e, self._url, body)
-            else:
-                message = 'Unable to reach APM Server: %s (url: %s)' % (
-                    e, self._url
-                )
-            raise TransportException(message, data, print_trace=print_trace)
+                    message = 'HTTP %s: ' % response.status
+                    print_trace = True
+                message += body.decode('utf8')
+                raise TransportException(message, data, print_trace=print_trace)
+            return response.getheader('Location')
         finally:
             if response:
                 response.close()
 
-        return response.info().get('Location')
 
-
-class AsyncHTTPTransport(AsyncTransport, HTTPTransport):
+class AsyncTransport(AsyncHTTPTransportBase, Transport):
     scheme = ['http', 'https']
     async_mode = True
-
-    def __init__(self, parsed_url):
-        super(AsyncHTTPTransport, self).__init__(parsed_url)
-        if self._url.startswith('async+'):
-            self._url = self._url[6:]
-        self._worker = None
-
-    @property
-    def worker(self):
-        if not self._worker or not self._worker.is_alive():
-            self._worker = AsyncWorker()
-        return self._worker
-
-    def send_sync(self, data=None, headers=None, success_callback=None,
-                  fail_callback=None):
-        try:
-            url = HTTPTransport.send(self, data, headers)
-            if callable(success_callback):
-                success_callback(url=url)
-        except Exception as e:
-            if callable(fail_callback):
-                fail_callback(exception=e)
-
-    def send_async(self, data, headers, success_callback=None,
-                   fail_callback=None):
-        kwargs = {
-            'data': data,
-            'headers': headers,
-            'success_callback': success_callback,
-            'fail_callback': fail_callback,
-        }
-        self.worker.queue(self.send_sync, kwargs)
-
-    def close(self):
-        if self._worker:
-            self._worker.main_thread_terminated()
+    sync_transport = Transport

elasticapm/transport/http_base.py

@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+from elasticapm.contrib.async_worker import AsyncWorker
+from elasticapm.transport.base import AsyncTransport, Transport
+
+
+class HTTPTransportBase(Transport):
+    scheme = ['http', 'https']
+
+    def __init__(self, parsed_url):
+        self.check_scheme(parsed_url)
+
+        self._parsed_url = parsed_url
+        self._url = parsed_url.geturl()
+
+    def send(self, data, headers, timeout=None):
+        """
+        Sends a request to a remote webserver using HTTP POST.
+
+        Returns the shortcut URL of the recorded error on Elastic APM
+        """
+        raise NotImplementedError()
+
+
+class AsyncHTTPTransportBase(AsyncTransport, HTTPTransportBase):
+    scheme = ['http', 'https']
+    async_mode = True
+    sync_transport = HTTPTransportBase
+
+    def __init__(self, parsed_url):
+        super(AsyncHTTPTransportBase, self).__init__(parsed_url)
+        if self._url.startswith('async+'):
+            self._url = self._url[6:]
+        self._worker = None
+
+    @property
+    def worker(self):
+        if not self._worker or not self._worker.is_alive():
+            self._worker = AsyncWorker()
+        return self._worker
+
+    def send_sync(self, data=None, headers=None, success_callback=None,
+                  fail_callback=None):
+        try:
+            url = self.sync_transport.send(self, data, headers)
+            if callable(success_callback):
+                success_callback(url=url)
+        except Exception as e:
+            if callable(fail_callback):
+                fail_callback(exception=e)
+
+    def send_async(self, data, headers, success_callback=None,
+                   fail_callback=None):
+        kwargs = {
+            'data': data,
+            'headers': headers,
+            'success_callback': success_callback,
+            'fail_callback': fail_callback,
+        }
+        self.worker.queue(self.send_sync, kwargs)
+
+    def close(self):
+        if self._worker:
+            self._worker.main_thread_terminated()