LS-5330: Add connection timeout to HTTP

- Also streamlined handling of exceptions to ensure we're capturing
the appropriate ones for HTTP requests

Author: frenchfrywpepper

lightstep/http_connection.py

@@ -6,19 +6,16 @@
 
 from lightstep.collector_pb2 import ReportResponse
 
-CONSECUTIVE_ERRORS_BEFORE_RECONNECT = 200
-
 
 class _HTTPConnection(object):
     """Instances of _Connection are used to establish a connection to the
     server via HTTP protocol.
     """
-    def __init__(self, collector_url):
+    def __init__(self, collector_url, timeout_seconds):
         self._collector_url = collector_url
         self._lock = threading.Lock()
         self.ready = True
-        self._report_eof_count = 0
-        self._report_consecutive_errors = 0
+        self._timeout_seconds = timeout_seconds
 
     def open(self):
         """Establish HTTP connection to the server.
@@ -28,9 +25,6 @@ def open(self):
     # May throw an Exception on failure.
     def report(self, *args, **kwargs):
         """Report to the server."""
-        # Notice the annoying case change on the method name. I chose to stay
-        # consistent with casing in this class vs staying consistent with the
-        # casing of the pass-through method.
         auth = args[0]
         report = args[1]
         with self._lock:
@@ -39,22 +33,16 @@ def report(self, *args, **kwargs):
                 headers = {"Content-Type": "application/octet-stream",
                            "Accept": "application/octet-stream"}
 
-                r = requests.post(self._collector_url, headers=headers, data=report.SerializeToString())
+                r = requests.post(
+                    self._collector_url,
+                    headers=headers,
+                    data=report.SerializeToString(),
+                    timeout=self._timeout_seconds)
                 resp = ReportResponse()
                 resp.ParseFromString(r.content)
-                self._report_consecutive_errors = 0
                 return resp
-            except EOFError:
-                self._report_consecutive_errors += 1
-                self._report_eof_count += 1
-                raise Exception('EOFError')
-            finally:
-                # In case the client has fallen into an unrecoverable state,
-                # recreate the data structure if there are continued report
-                # failures
-                if self._report_consecutive_errors == CONSECUTIVE_ERRORS_BEFORE_RECONNECT:
-                    self._report_consecutive_errors = 0
-                    self.ready = False
+            except requests.exceptions.RequestException as err:
+                raise err
 
     def close(self):
         """Close HTTP connection to the server."""

lightstep/recorder.py

@@ -45,7 +45,8 @@ def __init__(self,
                  verbosity=0,
                  certificate_verification=True,
                  use_thrift=False,
-                 use_http=True):
+                 use_http=True,
+                 timeout_seconds=30):
         self.verbosity = verbosity
         # Fail fast on a bad access token
         if not isinstance(access_token, str):
@@ -73,6 +74,7 @@ def __init__(self,
                 collector_host,
                 collector_port,
                 self.use_thrift)
+        self._timeout_seconds = timeout_seconds
         self._auth = self.converter.create_auth(access_token)
         self._mutex = threading.Lock()
         self._span_records = []
@@ -107,7 +109,7 @@ def _maybe_init_flush_thread(self):
             if self.use_thrift:
                 self._flush_connection = _ThriftConnection(self._collector_url)
             else:
-                self._flush_connection = _HTTPConnection(self._collector_url)
+                self._flush_connection = _HTTPConnection(self._collector_url, self._timeout_seconds)
             self._flush_connection.open()
             self._flush_thread = threading.Thread(target=self._flush_periodically,
                                                   name=constants.FLUSH_THREAD_NAME)

lightstep/tracer.py

@@ -49,6 +49,7 @@ def Tracer(**kwargs):
         configuration). Defaults to False (i.e., binary format is enabled).
     :param bool use_thrift: Forces the use of Thrift as the transport protocol.
     :param bool use_http: Forces the use of Proto over http.
+    :param float timeout_seconds: Number of seconds allowed for the HTTP report transaction (fractions are permitted)
     """
     enable_binary_format = True
     if 'disable_binary_format' in kwargs: