Merge pull request #45 from davejohnston/FFM-5352

[FFM-5352]: Stream errors cause the SDK to spam ff server requests

Author: Bryan Jen

featureflags/client.py

@@ -75,6 +75,7 @@ def run(self):
                 token=self._auth_token,
                 config=self._config,
                 ready=streaming_event,
+                poller=polling_event,
                 cluster=self._cluster,
             )
             self._stream.start()

featureflags/polling.py

@@ -29,8 +29,8 @@ def __init__(self, client: AuthenticatedClient, config: Config,
     def run(self):
         if not self.__running:
             if self.__config.pull_interval < 60:
-                log.warning("Pull Interval must be greater than or equal "
-                            "to 60 seconds, was: " +
+                log.warning("Pull Interval must be greater than or equal to "
+                            "60 seconds, was: " +
                             str(self.__config.pull_interval) +
                             " setting to 60")
                 self.__config.pull_interval = 60
@@ -46,14 +46,16 @@ def run(self):
                     t2.start()
                     t1.join()
                     t2.join()
-                    if not self.__ready.is_set() is True:
-                        log.info("PollingProcessor initialized ok")
-                        if self.__config.enable_stream and \
-                                not self.__stream_ready.is_set():
-                            log.debug('Poller is in pause mode...')
-                            self.__ready.wait()
-                        else:
-                            self.__ready.set()
+
+                    if self.__config.enable_stream and \
+                            self.__stream_ready.is_set():
+                        log.debug('Poller will be paused because' +
+                                  ' streaming mode is active')
+                        #  Block until ready.set() is called
+                        self.__ready.wait()
+                        log.debug('Poller resuming ')
+                    else:
+                        self.__ready.set()
                 except Exception as e:
                     log.exception(
                         'Error: Exception encountered when polling flags. %s',
@@ -62,6 +64,9 @@ def run(self):
 
                 elapsed = time.time() - start_time
                 if elapsed < self.__config.pull_interval:
+                    log.info("Poller sleeping for " +
+                             (self.__config.pull_interval - elapsed).__str__())
+                    " seconds"
                     time.sleep(self.__config.pull_interval - elapsed)
 
     def stop(self):

featureflags/streaming.py

@@ -1,10 +1,10 @@
-
+import random
 import threading
+import time
 from threading import Thread
 from typing import List, Union
 
 from featureflags.repository import DataProviderInterface
-
 from .api.client import AuthenticatedClient
 from .api.default.get_feature_config_by_identifier import \
     sync as get_feature_config
@@ -14,18 +14,23 @@
 from .sse_client import SSEClient
 from .util import log
 
+BACK_OFF_IN_SECONDS = 5
+
 
 class StreamProcessor(Thread):
     def __init__(self, repository: DataProviderInterface,
                  client: AuthenticatedClient,
                  environment_id: str, api_key: str, token: str,
-                 config: Config, ready: threading.Event,
+                 config: Config,
+                 ready: threading.Event,
+                 poller: threading.Event,
                  cluster: str):
 
         Thread.__init__(self)
         self.daemon = True
         self._running = False
         self._ready = ready
+        self.poller = poller
         self._client = client
         self._environment_id = environment_id
         self._api_key = api_key
@@ -39,10 +44,14 @@ def run(self):
         log.info("Starting StreamingProcessor connecting to uri: " +
                  self._stream_url)
         self._running = True
-
+        retries = 0
         while self._running:
             try:
                 messages = self._connect()
+                self.poller.clear()  # were streaming now, so tell any poller
+                # threads calling wait to wait...
+                self._ready.set()
+                retries = 0  # reset the retry counter
                 for msg in messages:
                     if not self._running:
                         break
@@ -53,7 +62,17 @@ def run(self):
                     if self._ready.is_set() is False:
                         self._ready.set()
             except Exception as e:
-                log.warning("Unexpected error on stream connection: %s", e)
+                log.error("Unexpected error on stream connection: %s", e)
+                # Signal the poller than it should start due to stream error.
+                if self.poller.is_set() is False:
+                    self.poller.set()
+
+                # Calculate back of sleep
+                sleep = (BACK_OFF_IN_SECONDS * 2 ** retries +
+                         random.uniform(0, 1))
+                log.info(f"retrying stream connection in {sleep.__str__()}s")
+                time.sleep(sleep)
+                retries += 1
 
     def _connect(self) -> SSEClient:
         return SSEClient(self._stream_url, headers={

tests/unit/test_evaluator.py

@@ -2,7 +2,8 @@
 
 from featureflags.evaluations.auth_target import Target
 from featureflags.evaluations.clause import Clause
-from featureflags.evaluations.constants import EQUAL_OPERATOR, STARTS_WITH_OPERATOR
+from featureflags.evaluations.constants import EQUAL_OPERATOR, \
+    STARTS_WITH_OPERATOR
 from featureflags.evaluations.distribution import Distribution
 from featureflags.evaluations.enum import FeatureState
 from featureflags.evaluations.evaluator import Evaluator
@@ -193,10 +194,14 @@ def test_evaluate_clauses(data_provider, target):
     )
 
     testcases = [
-        {"scenario": "Evaluate clauses with no clauses", "input": [], "expected": False},
-        {"scenario": "Evaluate clauses with 2 Clauses both will match", "input": [clause1, clause2], "expected": True},
-        {"scenario": "Evaluate clauses with 2 Clauses only 1 match", "input": [clause1, clause3], "expected": True},
-        {"scenario": "Evaluate clauses with 2 Clauses but no match", "input": [clause3, clause4], "expected": False}
+        {"scenario": "Evaluate clauses with no clauses",
+         "input": [], "expected": False},
+        {"scenario": "Evaluate clauses with 2 Clauses both will match",
+         "input": [clause1, clause2], "expected": True},
+        {"scenario": "Evaluate clauses with 2 Clauses only 1 match",
+         "input": [clause1, clause3], "expected": True},
+        {"scenario": "Evaluate clauses with 2 Clauses but no match",
+         "input": [clause3, clause4], "expected": False}
     ]
 
     for tc in testcases: