FFM-8544 Reorganise streaming logs (#77)

* FFM-8300 Log status codes correctly

* FFM-8544 Only log error in SSE if a reconnect has breached threshold

* FFM-8544 Change logging

* FFM-8544 Make streaming, not sse client, responsible for network warnings

* FFM-8544 1.2.1 release prep

Author: erdirowlands

featureflags/__init__.py

@@ -2,4 +2,4 @@
 
 __author__ = """Enver Bisevac"""
 __email__ = "enver.bisevac@harness.io"
-__version__ = '1.2.0'
+__version__ = '1.2.1'

featureflags/analytics.py

@@ -206,7 +206,6 @@ def _send_data(self) -> None:
                         unique_target)
                 target_data_batch_index += 1
 
-
         finally:
             self._data = {}
             self._target_data_batches = [{}]
@@ -256,18 +255,17 @@ def _send_data(self) -> None:
 
                 # Log any error codes
                 for unique_code, count in unique_responses_codes.items():
-                    if response.status_code >= 400:
+                    if unique_code >= 400:
                         warn_post_metrics_target_batch_failed(
                             f'{count} batches received code {unique_code}')
+                        continue
                     info_metrics_target_batch_success(
                         f'{count} batches successful')
 
-
             info_metrics_success()
         except httpx.RequestError as ex:
             warn_post_metrics_failed(ex)
 
-
     def process_target_data_batch(self, target_data_batch):
         batch_request_body: Metrics = Metrics(
             target_data=target_data_batch, metrics_data=[]

featureflags/polling.py

@@ -91,7 +91,6 @@ def stop(self):
         self.__running = False
         info_polling_stopped("Client was closed")
 
-
     def retrieve_flags_and_segments(self):
         t1 = Thread(target=self.__retrieve_segments)
         t2 = Thread(target=self.__retrieve_flags)

featureflags/sdk_logging_codes.py

@@ -27,6 +27,7 @@ def get_sdk_code_message(key):
         5002: "SSE event received: ",
         5003: "SSE retrying to connect in",
         5004: "SSE stopped",
+        5005: "Stream is still retrying after 5 retries: ",
         # SDK_EVAL_6xxx - these are hardcoded in `variation.py` as they
         # are more complex
         # SDK_METRICS_7xxx
@@ -145,6 +146,10 @@ def warn_stream_retrying(seconds):
     log.warning(sdk_err_msg(5003, seconds))
 
 
+def warn_stream_retrying_long_duration(seconds):
+    log.warning(sdk_err_msg(5005, f'{seconds} seconds'))
+
+
 def warn_post_metrics_failed(reason):
     log.warning(sdk_err_msg(7002, reason))
 

featureflags/sse_client.py

@@ -108,18 +108,13 @@ def __next__(self):
                     raise EOFError()
                 self.buf += self.decoder.decode(next_chunk)
 
-            except (StopIteration, requests.RequestException, EOFError) as e:
-                if isinstance(e, StopIteration):
-                    log.error("Error when iterating through stream messages, "
-                              "attempting to resume")
+            except (StopIteration, EOFError) as e:
 
                 if isinstance(e, EOFError):
-                    log.error("Received unexpected EOF from stream, "
-                              "attempting to reconnect")
+                    log.debug("Recieved empty chunk of data from stream, "
+                              "reconnecting")
 
-                if isinstance(e, requests.RequestException):
-                    log.error("Error encountered in stream, "
-                              "attempting to reconnect: %s", e)
+                log.debug("Streaming stopped abruptly, reconnecting")
 
                 time.sleep(self.retry / 1000.0)
                 self._connect()

featureflags/streaming.py

@@ -13,7 +13,8 @@
 from .models.message import Message
 from .sdk_logging_codes import info_stream_connected, \
     info_stream_event_received, warn_stream_disconnected, \
-    warn_stream_retrying, info_stream_stopped
+    warn_stream_retrying, info_stream_stopped, \
+    warn_stream_retrying_long_duration
 from .sse_client import SSEClient
 from .util import log
 
@@ -40,6 +41,7 @@ def __init__(self, repository: DataProviderInterface,
         self._token = token
         self._stream_url = f'{config.base_url}/stream?cluster={cluster}'
         self._repository = repository
+        self.reconnect_timer = 0
 
     def run(self):
         log.info("Starting StreamingProcessor connecting to uri: " +
@@ -69,18 +71,26 @@ def run(self):
                 if self.poller.is_set() is False:
                     self.poller.set()
 
+                # Check if retry attempts have reached a threshold before
+                # we log an error to the user
+                retry_error_log_threshold = 4
+                if retries >= retry_error_log_threshold:
+                    warn_stream_retrying_long_duration(
+                        retry_error_log_threshold)
+
                 # Calculate back of sleep
                 sleep = (BACK_OFF_IN_SECONDS * 2 ** retries +
                          random.uniform(0, 1))
+
                 warn_stream_retrying(f'{sleep.__str__()}s')
                 time.sleep(sleep)
                 retries += 1
 
     def _connect(self) -> SSEClient:
-        return SSEClient(self._stream_url, headers={
+        return SSEClient(url=self._stream_url, headers={
             'Authorization': f'Bearer {self._token}',
             'API-Key': self._api_key
-        })
+        }, retry=BACK_OFF_IN_SECONDS)
 
     def process_message(self, msg: Message) -> None:
         processor: Union[FlagMsgProcessor, SegmentMsgProcessor, None] = None

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 1.2.0
+current_version = 1.2.1
 commit = True
 tag = True
 

setup.py

@@ -57,6 +57,6 @@
     test_suite="tests",
     tests_require=test_requirements,
     url="https://github.com/harness/ff-python-server-sdk",
-    version='1.2.0',
+    version='1.2.1',
     zip_safe=False,
 )

tests/unit/test_sdk_logging_codes.py

@@ -25,6 +25,7 @@ def test_logs_dont_raise_exception():
     sdk_codes.warn_auth_retying(1, "some error")
     sdk_codes.warn_stream_disconnected("example reason")
     sdk_codes.warn_stream_retrying(5)
+    sdk_codes.warn_stream_retrying_long_duration(5)
     sdk_codes.warn_post_metrics_failed("example reason")
     sdk_codes.warn_post_metrics_target_batch_failed("example reason")
     sdk_codes.warn_default_variation_served("identifier", target, "default")