Merge pull request #61 from launchdarkly/eb/ch18901/4xx-errors

fail permanently on most 4xx errors

Author: eli-darkly

ldclient/event_processor.py

@@ -26,7 +26,7 @@
 from ldclient.util import _headers
 from ldclient.util import create_http_pool_manager
 from ldclient.util import log
-from ldclient.util import throw_if_unsuccessful_response
+from ldclient.util import http_error_message, is_http_error_recoverable, throw_if_unsuccessful_response
 
 
 __MAX_FLUSH_THREADS__ = 5
@@ -174,7 +174,6 @@ def _do_send(self, output_events):
                                    body=json_body,
                                    retries=1)
             self._response_fn(r)
-            throw_if_unsuccessful_response(r)
             return r
         except Exception as e:
             log.warning(
@@ -323,10 +322,11 @@ def _handle_response(self, r):
             if server_date is not None:
                 timestamp = int(time.mktime(server_date) * 1000)
                 self._last_known_past_time = timestamp
-        if r.status == 401:
-            log.error('Received 401 error, no further events will be posted since SDK key is invalid')
-            self._disabled = True
-            return
+        if r.status > 299:
+            log.error(http_error_message(r.status, "event delivery", "some events were dropped"))
+            if not is_http_error_recoverable(r.status):
+                self._disabled = True
+                return
 
     def _do_shutdown(self):
         self._flush_workers.stop()

ldclient/polling.py

@@ -2,7 +2,7 @@
 
 from ldclient.interfaces import UpdateProcessor
 from ldclient.util import log
-from ldclient.util import UnsuccessfulResponseException
+from ldclient.util import UnsuccessfulResponseException, http_error_message, is_http_error_recoverable
 
 import time
 
@@ -30,15 +30,14 @@ def run(self):
                         log.info("PollingUpdateProcessor initialized ok")
                         self._ready.set()
                 except UnsuccessfulResponseException as e:
-                    log.error('Received unexpected status code %d from polling request' % e.status)
-                    if e.status == 401:
-                        log.error('Received 401 error, no further polling requests will be made since SDK key is invalid')
+                    log.error(http_error_message(e.status, "polling request"))
+                    if not is_http_error_recoverable(e.status):
                         self._ready.set() # if client is initializing, make it stop waiting; has no effect if already inited
                         self.stop()
                     break
-                except Exception:
+                except Exception as e:
                     log.exception(
-                        'Error: Exception encountered when updating flags.')
+                        'Error: Exception encountered when updating flags. %s' % e)
 
                 elapsed = time.time() - start_time
                 if elapsed < self._config.poll_interval:

ldclient/streaming.py

@@ -9,7 +9,7 @@
 
 from ldclient.interfaces import UpdateProcessor
 from ldclient.sse_client import SSEClient
-from ldclient.util import _stream_headers, log, UnsuccessfulResponseException
+from ldclient.util import _stream_headers, log, UnsuccessfulResponseException, http_error_message, is_http_error_recoverable
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
 # allows for up to 5 minutes to elapse without any data sent across the stream. The heartbeats sent as comments on the
@@ -49,14 +49,11 @@ def run(self):
                         log.info("StreamingUpdateProcessor initialized ok.")
                         self._ready.set()
             except UnsuccessfulResponseException as e:
-                log.error("Received unexpected status code %d for stream connection" % e.status)
-                if e.status == 401:
-                    log.error("Received 401 error, no further streaming connection will be made since SDK key is invalid")
+                log.error(http_error_message(e.status, "stream connection"))
+                if not is_http_error_recoverable(e.status):
                     self._ready.set()  # if client is initializing, make it stop waiting; has no effect if already inited
                     self.stop()
                     break
-                else:
-                    log.warning("Restarting stream connection after one second.")
             except Exception as e:
                 log.warning("Caught exception. Restarting stream connection after one second. %s" % e)
                 # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of urllib3 internals
@@ -66,7 +63,7 @@ def _backoff_expo():
         return backoff.expo(max_value=30)
 
     def should_not_retry(e):
-        return isinstance(e, UnsuccessfulResponseException) and (e.response.status_code == 401)
+        return isinstance(e, UnsuccessfulResponseException) and (not is_http_error_recoverable(e.status))
 
     @backoff.on_exception(_backoff_expo, BaseException, max_tries=None, jitter=backoff.full_jitter,
                           giveup=should_not_retry)

ldclient/util.py

@@ -93,3 +93,18 @@ def create_http_pool_manager(num_pools=1, verify_ssl=False):
 def throw_if_unsuccessful_response(resp):
     if resp.status >= 400:
         raise UnsuccessfulResponseException(resp.status)
+
+
+def is_http_error_recoverable(status):
+    if status >= 400 and status < 500:
+        return (status == 400) or (status == 408) or (status == 429)  # all other 4xx besides these are unrecoverable
+    return True  # all other errors are recoverable
+
+
+def http_error_message(status, context, retryable_message = "will retry"):
+    return "Received HTTP error %d%s for %s - %s" % (
+        status,
+        " (invalid SDK key)" if (status == 401 or status == 403) else "",
+        context,
+        retryable_message if is_http_error_recoverable(status) else "giving up permanently"
+        )

testing/test_event_processor.py

@@ -362,9 +362,24 @@ def test_sdk_key_is_sent():
     assert mock_http.request_headers.get('Authorization') is 'SDK_KEY'
 
 def test_no_more_payloads_are_sent_after_401_error():
+    verify_unrecoverable_http_error(401)
+
+def test_no_more_payloads_are_sent_after_403_error():
+    verify_unrecoverable_http_error(403)
+
+def test_will_still_send_after_408_error():
+    verify_recoverable_http_error(408)
+
+def test_will_still_send_after_429_error():
+    verify_recoverable_http_error(429)
+
+def test_will_still_send_after_500_error():
+    verify_recoverable_http_error(500)
+
+def verify_unrecoverable_http_error(status):
     setup_processor(Config(sdk_key = 'SDK_KEY'))
 
-    mock_http.set_response_status(401)
+    mock_http.set_response_status(status)
     ep.send_event({ 'kind': 'identify', 'user': user })
     ep.flush()
     ep._wait_until_inactive()
@@ -375,6 +390,19 @@ def test_no_more_payloads_are_sent_after_401_error():
     ep._wait_until_inactive()
     assert mock_http.request_data is None
 
+def verify_recoverable_http_error(status):
+    setup_processor(Config(sdk_key = 'SDK_KEY'))
+
+    mock_http.set_response_status(status)
+    ep.send_event({ 'kind': 'identify', 'user': user })
+    ep.flush()
+    ep._wait_until_inactive()
+    mock_http.reset()
+
+    ep.send_event({ 'kind': 'identify', 'user': user })
+    ep.flush()
+    ep._wait_until_inactive()
+    assert mock_http.request_data is not None
 
 def flush_and_get_events():
     ep.flush()

testing/test_polling_processor.py

@@ -64,10 +64,30 @@ def test_general_connection_error_does_not_cause_immediate_failure():
     assert not pp.initialized()
 
 def test_http_401_error_causes_immediate_failure():
-    mock_requester.exception = UnsuccessfulResponseException(401)
-    start_time = time.time()
+    verify_unrecoverable_http_error(401)
+
+def test_http_403_error_causes_immediate_failure():
+    verify_unrecoverable_http_error(401)
+
+def test_http_408_error_does_not_cause_immediate_failure():
+    verify_recoverable_http_error(408)
+
+def test_http_429_error_does_not_cause_immediate_failure():
+    verify_recoverable_http_error(429)
+
+def test_http_500_error_does_not_cause_immediate_failure():
+    verify_recoverable_http_error(500)
+
+def verify_unrecoverable_http_error(status):
+    mock_requester.exception = UnsuccessfulResponseException(status)
     setup_processor(config)
-    ready.wait(5.0)
-    elapsed_time = time.time() - start_time
-    assert elapsed_time < 5.0
+    finished = ready.wait(5.0)
+    assert finished
+    assert not pp.initialized()
+
+def verify_recoverable_http_error(status):
+    mock_requester.exception = UnsuccessfulResponseException(status)
+    setup_processor(config)
+    finished = ready.wait(0.2)
+    assert not finished
     assert not pp.initialized()