Merge pull request #286 from splitio/patch-fix-timeout

patch fix for timeout and redis imp count

Author: chillaq

splitio/client/factory.py

@@ -63,6 +63,7 @@
 _INSTANTIATED_FACTORIES = Counter()
 _INSTANTIATED_FACTORIES_LOCK = threading.RLock()
 _MIN_DEFAULT_DATA_SAMPLING_ALLOWED = 0.1  # 10%
+_MAX_RETRY_SYNC_ALL = 3
 
 
 class Status(Enum):
@@ -379,7 +380,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     )
 
     if preforked_initialization:
-        synchronizer.sync_all(max_retry_attempts=3)
+        synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
         synchronizer._split_synchronizers._segment_sync.shutdown()
         return SplitFactory(api_key, storages, cfg['labelsEnabled'],
                             recorder, manager, preforked_initialization=preforked_initialization)

splitio/engine/impressions/adapters.py

@@ -66,8 +66,6 @@ def __init__(self, redis_client):
         :type telemtry_http_client: splitio.api.telemetry.TelemetryAPI
         """
         self._redis_client = redis_client
-        self.pipe = self._redis_client.pipeline()
-
 
     def record_unique_keys(self, uniques):
         """
@@ -96,10 +94,11 @@ def flush_counters(self, to_send):
         try:
             resulted = 0
             counted = 0
+            pipe = self._redis_client.pipeline()
             for pf_count in to_send:
-                self.pipe.hincrby(self.IMP_COUNT_QUEUE_KEY, pf_count.feature + "::" + str(pf_count.timeframe), pf_count.count)
+                pipe.hincrby(self.IMP_COUNT_QUEUE_KEY, pf_count.feature + "::" + str(pf_count.timeframe), pf_count.count)
                 counted += pf_count.count
-            resulted = sum(self.pipe.execute())
+            resulted = sum(pipe.execute())
             self._expire_keys(self.IMP_COUNT_QUEUE_KEY, self.IMP_COUNT_KEY_DEFAULT_TTL, resulted, counted)
             return True
         except RedisAdapterException:

splitio/sync/manager.py

@@ -7,15 +7,14 @@
 from splitio.push.manager import PushManager, Status
 from splitio.api import APIException
 from splitio.util.backoff import Backoff
-
+from splitio.sync.synchronizer import _SYNC_ALL_NO_RETRIES
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class Manager(object):  # pylint:disable=too-many-instance-attributes
     """Manager Class."""
 
-    _SYNC_ALL_ATTEMPTS = -1
     _CENTINEL_EVENT = object()
 
     def __init__(self, ready_flag, synchronizer, auth_api, streaming_enabled, sdk_metadata, sse_url=None, client_key=None):  # pylint:disable=too-many-arguments
@@ -59,15 +58,12 @@ def recreate(self):
         """Recreate poolers for forked processes."""
         self._synchronizer._split_synchronizers._segment_sync.recreate()
 
-    def start(self):
+    def start(self, max_retry_attempts=_SYNC_ALL_NO_RETRIES):
         """
         Start the SDK synchronization tasks.
-
-        :param max_retry_attempts: apply max attempts if it set to absilute integer.
-        :type max_retry_attempts: int
         """
         try:
-            self._synchronizer.sync_all(self._SYNC_ALL_ATTEMPTS)
+            self._synchronizer.sync_all(max_retry_attempts)
             self._ready_flag.set()
             self._synchronizer.start_periodic_data_recording()
             if self._streaming_enabled:

splitio/sync/synchronizer.py

@@ -10,7 +10,7 @@
 
 
 _LOGGER = logging.getLogger(__name__)
-
+_SYNC_ALL_NO_RETRIES = -1
 
 class SplitSynchronizers(object):
     """SplitSynchronizers."""
@@ -292,7 +292,7 @@ def synchronize_splits(self, till, sync_segments=True):
             _LOGGER.debug('Error: ', exc_info=True)
             return False
 
-    def sync_all(self, max_retry_attempts=-1):
+    def sync_all(self, max_retry_attempts=_SYNC_ALL_NO_RETRIES):
         """
         Synchronize all splits.
 
@@ -303,10 +303,7 @@ def sync_all(self, max_retry_attempts=-1):
         while True:
             try:
                 if not self.synchronize_splits(None, False):
-                    retry_attempts = self._retry_block(max_retry_attempts, retry_attempts)
-                    if max_retry_attempts != -1 and retry_attempts == -1:
-                        break
-                    continue
+                    raise Exception("split sync failed")
 
                 # Only retrying splits, since segments may trigger too many calls.
                 if not self._synchronize_segments():
@@ -317,20 +314,16 @@ def sync_all(self, max_retry_attempts=-1):
             except Exception as exc:  # pylint:disable=broad-except
                 _LOGGER.error("Exception caught when trying to sync all data: %s", str(exc))
                 _LOGGER.debug('Error: ', exc_info=True)
-                retry_attempts = self._retry_block(max_retry_attempts, retry_attempts)
-                if max_retry_attempts != -1 and retry_attempts == -1:
-                    break
-                continue
+                if max_retry_attempts != _SYNC_ALL_NO_RETRIES:
+                    retry_attempts += 1
+                    if retry_attempts > max_retry_attempts:
+                        break
+                how_long = self._backoff.get()
+                time.sleep(how_long)
 
         _LOGGER.error("Could not correctly synchronize splits and segments after %d attempts.", retry_attempts)
 
     def _retry_block(self, max_retry_attempts, retry_attempts):
-        if max_retry_attempts != -1:
-            retry_attempts += 1
-            if retry_attempts > max_retry_attempts:
-                return -1
-        how_long = self._backoff.get()
-        time.sleep(how_long)
         return retry_attempts
 
     def shutdown(self, blocking):

tests/client/test_factory.py

@@ -142,7 +142,7 @@ def test_redis_client_creation(self, mocker):
 
     def test_uwsgi_forked_client_creation(self):
         """Test client with preforked initialization."""
-#        pytest.set_trace()
+        # Invalid API Key with preforked should exit after 3 attempts.
         factory = get_factory('some_api_key', config={'preforkedInitialization': True})
         assert isinstance(factory._storages['splits'], inmemmory.InMemorySplitStorage)
         assert isinstance(factory._storages['segments'], inmemmory.InMemorySegmentStorage)
@@ -228,9 +228,10 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk
         mocker.patch('splitio.sync.manager.Manager.__init__', new=_split_synchronizer)
 
         # Start factory and make assertions
+        # Using invalid key should result in a timeout exception
         factory = get_factory('some_api_key')
         try:
-            factory.block_until_ready(1) 
+            factory.block_until_ready(1)
         except:
             pass
         assert factory.ready is False
@@ -316,7 +317,7 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk
         # Start factory and make assertions
         factory = get_factory('some_api_key')
         try:
-            factory.block_until_ready(1) 
+            factory.block_until_ready(1)
         except:
             pass
         assert factory.ready is False

tests/sync/test_manager.py

@@ -46,7 +46,7 @@ def run(x):
         manager = Manager(threading.Event(), synchronizer,  mocker.Mock(), False, SdkMetadata('1.0', 'some', '1.2.3.4'))
 
         manager._SYNC_ALL_ATTEMPTS = 1
-        manager.start()  # should not throw!
+        manager.start(2)  # should not throw!
 
     def test_start_streaming_false(self, mocker):
         splits_ready_event = threading.Event()

tests/sync/test_synchronizer.py

@@ -34,6 +34,7 @@ def run(x, c):
 
         sychronizer.synchronize_splits(None)  # APIExceptions are handled locally and should not be propagated!
 
+        # test forcing to have only one retry attempt and then exit
         sychronizer.sync_all(1)  # sync_all should not throw!
 
     def test_sync_all_failed_segments(self, mocker):