Added retry attempts to sync_all

Author: chillaq

splitio/client/factory.py

@@ -379,7 +379,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     )
 
     if preforked_initialization:
-        synchronizer.sync_all()
+        synchronizer.sync_all(max_retry_attempts=3)
         synchronizer._split_synchronizers._segment_sync.shutdown()
         return SplitFactory(api_key, storages, cfg['labelsEnabled'],
                             recorder, manager, preforked_initialization=preforked_initialization)
@@ -391,7 +391,6 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     return SplitFactory(api_key, storages, cfg['labelsEnabled'],
                         recorder, manager, sdk_ready_flag)
 
-
 def _build_redis_factory(api_key, cfg):
     """Build and return a split factory with redis-based storage."""
     sdk_metadata = util.get_metadata(cfg)

splitio/sync/manager.py

@@ -58,10 +58,15 @@ def recreate(self):
         """Recreate poolers for forked processes."""
         self._synchronizer._split_synchronizers._segment_sync.recreate()
 
-    def start(self):
-        """Start the SDK synchronization tasks."""
+    def start(self, retry_attempts=-1):
+        """
+        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._synchronizer.sync_all(retry_attempts)
             self._ready_flag.set()
             self._synchronizer.start_periodic_data_recording()
             if self._streaming_enabled:

splitio/sync/synchronizer.py

@@ -292,15 +292,20 @@ def synchronize_splits(self, till, sync_segments=True):
             _LOGGER.debug('Error: ', exc_info=True)
             return False
 
-    def sync_all(self):
-        """Synchronize all split data."""
-        attempts = 3
-        while attempts > 0:
+    def sync_all(self, max_retry_attempts=-1):
+        """
+        Synchronize all splits.
+
+        :param max_retry_attempts: apply max attempts if it set to absilute integer.
+        :type max_retry_attempts: int
+        """
+        retry_attempts = 0
+        while True:
             try:
                 if not self.synchronize_splits(None, False):
-                    attempts -= 1
-                    how_long = self._backoff.get()
-                    time.sleep(how_long)
+                    retry_attempts = self._retry_block(max_retry_attempts, retry_attempts)
+                    if not max_retry_attempts == -1 and retry_attempts == -1:
+                        break
                     continue
 
                 # Only retrying splits, since segments may trigger too many calls.
@@ -310,11 +315,23 @@ def sync_all(self):
                 # All is good
                 return
             except Exception as exc:  # pylint:disable=broad-except
-                attempts -= 1
                 _LOGGER.error("Exception caught when trying to sync all data: %s", str(exc))
                 _LOGGER.debug('Error: ', exc_info=True)
-
-        _LOGGER.error("Could not correctly synchronize splits and segments after 3 attempts.")
+                retry_attempts = self._retry_block(max_retry_attempts, retry_attempts)
+                if not max_retry_attempts == -1 and retry_attempts == -1:
+                    break
+                continue
+
+        _LOGGER.error("Could not correctly synchronize splits and segments after %d attempts.", retry_attempts)
+
+    def _retry_block(self, max_retry_attempts, retry_attempts):
+        if not 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):
         """
@@ -478,8 +495,12 @@ def __init__(self, split_synchronizers, split_tasks):
         self._split_synchronizers = split_synchronizers
         self._split_tasks = split_tasks
 
-    def sync_all(self):
-        """Synchronize all split data."""
+    def sync_all(self, max_retry_attempts=-1):
+        """
+        Synchronize all splits.
+
+        :param max_retry_attempts: Not used, added for compatibility
+        """
         try:
             self._split_synchronizers.split_sync.synchronize_splits(None)
         except APIException as exc:

tests/client/test_factory.py

@@ -5,6 +5,7 @@
 import os
 import time
 import threading
+import pytest
 from splitio.client.factory import get_factory, SplitFactory, _INSTANTIATED_FACTORIES, Status,\
     _LOGGER as _logger
 from splitio.client.config import DEFAULT_CONFIG
@@ -56,7 +57,10 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk
         assert isinstance(factory._recorder._impression_storage, inmemmory.ImpressionStorage)
 
         assert factory._labels_enabled is True
-        factory.block_until_ready()
+        try:
+            factory.block_until_ready(1)
+        except:
+            pass
         assert factory.ready
         factory.destroy()
 
@@ -129,12 +133,16 @@ def test_redis_client_creation(self, mocker):
         assert isinstance(factory._recorder._make_pipe(), RedisPipelineAdapter)
         assert isinstance(factory._recorder._event_sotrage, redis.RedisEventsStorage)
         assert isinstance(factory._recorder._impression_storage, redis.RedisImpressionsStorage)
-        factory.block_until_ready()
+        try:
+            factory.block_until_ready(1)
+        except:
+            pass
         assert factory.ready
         factory.destroy()
 
     def test_uwsgi_forked_client_creation(self):
         """Test client with preforked initialization."""
+#        pytest.set_trace()
         factory = get_factory('some_api_key', config={'preforkedInitialization': True})
         assert isinstance(factory._storages['splits'], inmemmory.InMemorySplitStorage)
         assert isinstance(factory._storages['segments'], inmemmory.InMemorySegmentStorage)
@@ -221,8 +229,11 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk
 
         # Start factory and make assertions
         factory = get_factory('some_api_key')
-        factory.block_until_ready()
-        assert factory.ready
+        try:
+            factory.block_until_ready(1) 
+        except:
+            pass
+        assert factory.ready is False
         assert factory.destroyed is False
 
         factory.destroy()
@@ -304,8 +315,11 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sdk
 
         # Start factory and make assertions
         factory = get_factory('some_api_key')
-        factory.block_until_ready()
-        assert factory.ready
+        try:
+            factory.block_until_ready(1) 
+        except:
+            pass
+        assert factory.ready is False
         assert factory.destroyed is False
 
         event = threading.Event()
@@ -470,7 +484,10 @@ def _get_storage_mock(self, name):
             'preforkedInitialization': True,
         }
         factory = get_factory("none", config=config)
-        factory.block_until_ready(10)
+        try:
+            factory.block_until_ready(10)
+        except:
+            pass
         assert factory._status == Status.WAITING_FORK
         assert len(sync_all_mock.mock_calls) == 1
         assert len(start_mock.mock_calls) == 0
@@ -481,6 +498,7 @@ def _get_storage_mock(self, name):
 
         assert clear_impressions._called == 1
         assert clear_events._called == 1
+        factory.destroy()
 
     def test_error_prefork(self, mocker):
         """Test not handling fork."""
@@ -490,9 +508,12 @@ def test_error_prefork(self, mocker):
 
         filename = os.path.join(os.path.dirname(__file__), '../integration/files', 'file2.yaml')
         factory = get_factory('localhost', config={'splitFile': filename})
-        factory.block_until_ready(1)
-
+        try:
+            factory.block_until_ready(1)
+        except:
+            pass
         _logger = mocker.Mock()
         mocker.patch('splitio.client.factory._LOGGER', new=_logger)
         factory.resume()
         assert _logger.warning.mock_calls == expected_msg
+        factory.destroy()

tests/sync/test_manager.py

@@ -2,7 +2,6 @@
 
 import threading
 import unittest.mock as mock
-
 from splitio.tasks.split_sync import SplitSynchronizationTask
 from splitio.tasks.segment_sync import SegmentSynchronizationTask
 from splitio.tasks.impressions_sync import ImpressionsSyncTask, ImpressionsCountSyncTask
@@ -46,14 +45,16 @@ def run(x):
         synchronizer = Synchronizer(synchronizers, split_tasks)
         manager = Manager(threading.Event(), synchronizer,  mocker.Mock(), False, SdkMetadata('1.0', 'some', '1.2.3.4'))
 
-        manager.start()  # should not throw!
+        manager.start(1)  # should not throw!
 
     def test_start_streaming_false(self, mocker):
         splits_ready_event = threading.Event()
         synchronizer = mocker.Mock(spec=Synchronizer)
         manager = Manager(splits_ready_event, synchronizer, mocker.Mock(), False, SdkMetadata('1.0', 'some', '1.2.3.4'))
-        manager.start()
-
+        try:
+            manager.start()
+        except:
+            pass
         splits_ready_event.wait(2)
         assert splits_ready_event.is_set()
         assert len(synchronizer.sync_all.mock_calls) == 1

tests/sync/test_synchronizer.py

@@ -34,7 +34,7 @@ def run(x, c):
 
         sychronizer.synchronize_splits(None)  # APIExceptions are handled locally and should not be propagated!
 
-        sychronizer.sync_all()  # sync_all should not throw!
+        sychronizer.sync_all(1)  # sync_all should not throw!
 
     def test_sync_all_failed_segments(self, mocker):
         api = mocker.Mock()
@@ -53,7 +53,7 @@ def run(x, y):
                                                  mocker.Mock(), mocker.Mock())
         sychronizer = Synchronizer(split_synchronizers, mocker.Mock(spec=SplitTasks))
 
-        sychronizer.sync_all()  # SyncAll should not throw!
+        sychronizer.sync_all(1)  # SyncAll should not throw!
         assert not sychronizer._synchronize_segments()
 
     splits = [{
@@ -319,7 +319,7 @@ def sync_splits(*_):
         split_tasks = mocker.Mock(spec=SplitTasks)
         synchronizer = Synchronizer(split_synchronizers, split_tasks)
 
-        synchronizer.sync_all()
+        synchronizer.sync_all(2)
         assert counts['splits'] == 3
 
     def test_sync_all_segment_attempts(self, mocker):