Ported changes from master and updated tests

Author: chillaq

splitio/client/factory.py

@@ -66,7 +66,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):
     """Factory Status."""
@@ -355,9 +355,6 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata, telemetry_runtime_producer),
     }
 
-    if not input_validator.validate_apikey_type(apis['segments']):
-        return None
-
     storages = {
         'splits': InMemorySplitStorage(),
         'segments': InMemorySegmentStorage(),
@@ -424,7 +421,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=_MAX_RETRY_SYNC_ALL)
         synchronizer._split_synchronizers._segment_sync.shutdown()
         return SplitFactory(api_key, storages, cfg['labelsEnabled'],
                             recorder, manager, None, telemetry_producer, telemetry_consumer.get_telemetry_init_consumer(), apis['telemetry'], preforked_initialization=preforked_initialization)

splitio/client/input_validator.py

@@ -448,31 +448,6 @@ def filter(self, record):
         return record.name not in ('SegmentsAPI', 'HttpClient')
 
 
-def validate_apikey_type(segment_api):
-    """
-    Try to guess if the apikey is of browser type and let the user know.
-
-    :param segment_api: Segments API client.
-    :type segment_api: splitio.api.segments.SegmentsAPI
-    """
-    api_messages_filter = _ApiLogFilter()
-    _logger = logging.getLogger('splitio.api.segments')
-    try:
-        _logger.addFilter(api_messages_filter)  # pylint: disable=protected-access
-        segment_api.fetch_segment('__SOME_INVALID_SEGMENT__', -1, FetchOptions())
-    except APIException as exc:
-        if exc.status_code == 403:
-            _LOGGER.error('factory instantiation: you passed a browser type '
-                          + 'api_key, please grab an api key from the Split '
-                          + 'console that is of type sdk')
-            return False
-    finally:
-        _logger.removeFilter(api_messages_filter)  # pylint: disable=protected-access
-
-    # True doesn't mean that the APIKEY is right, only that it's not of type "browser"
-    return True
-
-
 def validate_factory_instantiation(apikey):
     """
     Check if the factory if being instantiated with the appropriate arguments.

splitio/engine/impressions/adapters.py

@@ -91,10 +91,16 @@ def flush_counters(self, to_send):
         :param uniques: unique keys disctionary
         :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
         """
-        bulk_counts = self._build_counters(to_send)
         try:
-            inserted = self._redis_client.rpush(self.IMP_COUNT_QUEUE_KEY, *bulk_counts)
-            self._expire_keys(self.IMP_COUNT_QUEUE_KEY, self.IMP_COUNT_KEY_DEFAULT_TTL, inserted, len(bulk_counts))
+            resulted = 0
+            counted = 0
+            pipe = self._redis_client.pipeline()
+            for pf_count in to_send:
+                pipe.hincrby(self.IMP_COUNT_QUEUE_KEY, pf_count.feature + "::" + str(pf_count.timeframe), pf_count.count)
+                counted += pf_count.count
+            resulted = sum(pipe.execute())
+            self._expire_keys(self.IMP_COUNT_QUEUE_KEY,
+                              self.IMP_COUNT_KEY_DEFAULT_TTL, resulted, counted)
             return True
         except RedisAdapterException:
             _LOGGER.error('Something went wrong when trying to add counters to redis')
@@ -124,23 +130,3 @@ def _uniques_formatter(self, uniques):
         :rtype: json
         """
         return [json.dumps({'f': feature, 'ks': list(keys)}) for feature, keys in uniques.items()]
-
-    def _build_counters(self, counters):
-        """
-        Build an impression bulk formatted as the API expects it.
-
-        :param counters: List of impression counters per feature.
-        :type counters: list[splitio.engine.impressions.Counter.CountPerFeature]
-
-        :return: dict with list of impression count dtos
-        :rtype: dict
-        """
-        return json.dumps({
-            'pf': [
-                {
-                    'f': pf_count.feature,
-                    'm': pf_count.timeframe,
-                    'rc': pf_count.count
-                } for pf_count in counters
-            ]
-        })

splitio/storage/inmemmory.py

@@ -205,7 +205,7 @@ def get(self, segment_name):
         with self._lock:
             fetched = self._segments.get(segment_name)
             if fetched is None:
-                _LOGGER.warning(
+                _LOGGER.debug(
                     "Tried to retrieve nonexistant segment %s. Skipping",
                     segment_name
                 )

splitio/sync/manager.py

@@ -10,6 +10,7 @@
 from splitio.util.backoff import Backoff
 from splitio.util.time import get_current_epoch_time_ms
 from splitio.models.telemetry import SSESyncMode, StreamingEventTypes
+from splitio.sync.synchronizer import _SYNC_ALL_NO_RETRIES
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -61,10 +62,10 @@ 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."""
         try:
-            self._synchronizer.sync_all()
+            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

@@ -3,12 +3,13 @@
 import abc
 import logging
 import threading
+import time
 
 from splitio.api import APIException
-
+from splitio.util.backoff import Backoff
 
 _LOGGER = logging.getLogger(__name__)
-
+_SYNC_ALL_NO_RETRIES = -1
 
 class SplitSynchronizers(object):
     """SplitSynchronizers."""
@@ -224,6 +225,9 @@ def shutdown(self, blocking):
 class Synchronizer(BaseSynchronizer):
     """Synchronizer."""
 
+    _ON_DEMAND_FETCH_BACKOFF_BASE = 10  # backoff base starting at 10 seconds
+    _ON_DEMAND_FETCH_BACKOFF_MAX_WAIT = 30  # don't sleep for more than 1 minute
+
     def __init__(self, split_synchronizers, split_tasks):
         """
         Class constructor.
@@ -233,6 +237,9 @@ def __init__(self, split_synchronizers, split_tasks):
         :param split_tasks: tasks for starting/stopping tasks
         :type split_tasks: splitio.sync.synchronizer.SplitTasks
         """
+        self._backoff = Backoff(
+                                self._ON_DEMAND_FETCH_BACKOFF_BASE,
+                                self._ON_DEMAND_FETCH_BACKOFF_MAX_WAIT)
         self._split_synchronizers = split_synchronizers
         self._split_tasks = split_tasks
         self._periodic_data_recording_tasks = [
@@ -296,14 +303,17 @@ 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=_SYNC_ALL_NO_RETRIES):
+        """
+        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
-                    continue
+                    raise Exception("split sync failed")
 
                 # Only retrying splits, since segments may trigger too many calls.
 
@@ -313,11 +323,16 @@ 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)
+                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 3 attempts.")
+            _LOGGER.error("Could not correctly synchronize splits and segments after %d attempts.", retry_attempts)
 
     def shutdown(self, blocking):
         """
@@ -486,8 +501,11 @@ 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:

splitio/version.py

@@ -1 +1 @@
-__version__ = '9.1.3'
+__version__ = '9.2.1'

tests/client/test_client.py

@@ -39,10 +39,10 @@ def test_get_treatment(self, mocker):
         mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5)
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -108,10 +108,10 @@ def test_get_treatment_with_config(self, mocker):
         destroyed_property.return_value = False
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -185,10 +185,10 @@ def test_get_treatments(self, mocker):
         destroyed_property.return_value = False
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -258,10 +258,10 @@ def test_get_treatments_with_config(self, mocker):
         destroyed_property = mocker.PropertyMock()
         destroyed_property.return_value = False
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -335,10 +335,10 @@ def test_destroy(self, mocker):
         event_storage = mocker.Mock(spec=EventStorage)
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -367,10 +367,10 @@ def test_track(self, mocker):
         event_storage.put.return_value = True
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -404,10 +404,10 @@ def test_evaluations_before_running_post_fork(self, mocker):
         destroyed_property.return_value = False
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, mocker.Mock(), mocker.Mock())
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': mocker.Mock(),
             'segments': mocker.Mock(),
@@ -454,10 +454,10 @@ def test_evaluations_before_running_post_fork(self, mocker):
     @mock.patch('splitio.client.client.Client.ready', side_effect=None)
     def test_telemetry_not_ready(self, mocker):
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, mocker.Mock(), mocker.Mock())
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory('localhost',
             {'splits': mocker.Mock(),
             'segments': mocker.Mock(),
@@ -492,10 +492,10 @@ def test_telemetry_record_treatment_exception(self, mocker):
         mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5)
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -536,10 +536,10 @@ def test_telemetry_record_treatments_exception(self, mocker):
         mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5)
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -579,10 +579,10 @@ def test_telemetry_method_latency(self, mocker):
         mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5)
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,
@@ -621,10 +621,10 @@ def test_telemetry_track_exception(self, mocker):
         mocker.patch('splitio.client.client.get_latency_bucket_index', new=lambda x: 5)
 
         impmanager = mocker.Mock(spec=ImpressionManager)
-        recorder = StandardRecorder(impmanager, event_storage, impression_storage)
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
+        recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer())
         factory = SplitFactory(mocker.Mock(),
             {'splits': split_storage,
             'segments': segment_storage,