polishing

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 = 3
 
 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,10 +314,11 @@ 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
+
+            retry_attempts = self._retry_block(max_retry_attempts, retry_attempts)
+            if max_retry_attempts != -1 and retry_attempts == -1:
+                break
+            continue
 
         _LOGGER.error("Could not correctly synchronize splits and segments after %d attempts.", retry_attempts)