Merge pull request #289 from splitio/development

Patch version 9.2.1

Author: chillaq

CHANGES.txt

@@ -1,3 +1,8 @@
+9.2.1 (Dec 2, 2022)
+- Changed redis record type for impressions counts from list using rpush to hashed key using hincrby.
+- Apply Timeout Exception when incorrect SDK API Key is used.
+- Changed potential initial fetching segment Warning to Debug in logging.
+
 9.2.0 (Oct 14, 2022)
 - Added a new impressions mode for the SDK called NONE , to be used in factory when there is no desire to capture impressions on an SDK factory to feed Split's analytics engine. Running NONE mode, the SDK will only capture unique keys evaluated for a particular feature flag instead of full blown impressions
 

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):
@@ -316,9 +317,6 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata),
     }
 
-    if not input_validator.validate_apikey_type(apis['segments']):
-        return None
-
     storages = {
         'splits': InMemorySplitStorage(),
         'segments': InMemorySegmentStorage(),
@@ -382,7 +380,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, preforked_initialization=preforked_initialization)
@@ -394,7 +392,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/client/input_validator.py

@@ -446,32 +446,7 @@ def validate_attributes(attributes, method_name):
 class _ApiLogFilter(logging.Filter):  # pylint: disable=too-few-public-methods
     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):
     """

splitio/engine/impressions/adapters.py

@@ -88,13 +88,18 @@ def flush_counters(self, to_send):
         """
         post the impression counters to redis.
 
-        :param uniques: unique keys disctionary
-        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+        :param to_send: unique keys disctionary
+        :type to_send: 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 +129,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/adapters/redis.py

@@ -241,6 +241,13 @@ def hget(self, name, key):
         except RedisError as exc:
             raise RedisAdapterException('Error executing hget operation') from exc
 
+    def hincrby(self, name, key, amount=1):
+        """Mimic original redis function but using user custom prefix."""
+        try:
+            return self._decorated.hincrby(self._prefix_helper.add_prefix(name), key, amount)
+        except RedisError as exc:
+            raise RedisAdapterException('Error executing hincrby operation') from exc
+
     def incr(self, name, amount=1):
         """Mimic original redis function but using user custom prefix."""
         try:
@@ -323,6 +330,10 @@ def incr(self, name, amount=1):
         """Mimic original redis function but using user custom prefix."""
         self._pipe.incr(self._prefix_helper.add_prefix(name), amount)
 
+    def hincrby(self, name, key, amount=1):
+        """Mimic original redis function but using user custom prefix."""
+        self._pipe.hincrby(self._prefix_helper.add_prefix(name), key, amount)
+
     def execute(self):
         """Mimic original redis function but using user custom prefix."""
         try:

splitio/storage/inmemmory.py

@@ -193,7 +193,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

@@ -7,7 +7,7 @@
 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__)
 
@@ -58,10 +58,12 @@ 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, 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,14 @@
 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."""
@@ -212,6 +214,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.
@@ -221,6 +226,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 = [
@@ -284,14 +292,18 @@ 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.
                 if not self._synchronize_segments():
@@ -300,11 +312,19 @@ 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 _retry_block(self, max_retry_attempts, retry_attempts):
+        return retry_attempts
 
     def shutdown(self, blocking):
         """
@@ -468,8 +488,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:

splitio/version.py

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