updated sync.split, sync.synchronizer and tasks.util.asynctask classes

Author: chillaq

splitio/sync/split.py

@@ -10,9 +10,11 @@
 
 from splitio.api import APIException
 from splitio.api.commons import FetchOptions
+from splitio.client.input_validator import validate_flag_sets
 from splitio.models import splits
 from splitio.util.backoff import Backoff
 from splitio.util.time import get_current_epoch_time_ms
+from splitio.util.storage_helper import update_feature_flag_storage, update_feature_flag_storage_async
 from splitio.sync import util
 from splitio.optional.loaders import asyncio, aiofiles
 
@@ -28,7 +30,7 @@
 _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES = 10
 
 
-class SplitSynchronizer(object):
+class SplitSynchronizerBase(object):
     """Feature Flag changes synchronizer."""
 
     def __init__(self, feature_flag_api, feature_flag_storage):
@@ -52,6 +54,31 @@ def feature_flag_storage(self):
         """Return Feature_flag storage object"""
         return self._feature_flag_storage
 
+    def _get_config_sets(self):
+        """
+        Get all filter flag sets cnverrted to string, if no filter flagsets exist return None
+        :return: string with flagsets
+        :rtype: str
+        """
+        if self._feature_flag_storage.flag_set_filter.flag_sets == set({}):
+            return None
+        return ','.join(self._feature_flag_storage.flag_set_filter.sorted_flag_sets)
+
+class SplitSynchronizer(SplitSynchronizerBase):
+    """Feature Flag changes synchronizer."""
+
+    def __init__(self, feature_flag_api, feature_flag_storage):
+        """
+        Class constructor.
+
+        :param feature_flag_api: Feature Flag API Client.
+        :type feature_flag_api: splitio.api.splits.SplitsAPI
+
+        :param feature_flag_storage: Feature Flag Storage.
+        :type feature_flag_storage: splitio.storage.InMemorySplitStorage
+        """
+        super().__init__(feature_flag_api, feature_flag_storage)
+
     def _fetch_until(self, fetch_options, till=None):
         """
         Hit endpoint, update storage and return when since==till.
@@ -81,14 +108,9 @@ def _fetch_until(self, fetch_options, till=None):
                 _LOGGER.debug('Exception information: ', exc_info=True)
                 raise exc
 
-            for feature_flag in feature_flag_changes.get('splits', []):
-                if feature_flag['status'] == splits.Status.ACTIVE.value:
-                    parsed = splits.from_raw(feature_flag)
-                    self._feature_flag_storage.put(parsed)
-                    segment_list.update(set(parsed.get_segment_names()))
-                else:
-                    self._feature_flag_storage.remove(feature_flag['name'])
-            self._feature_flag_storage.set_change_number(feature_flag_changes['till'])
+            fetched_feature_flags = []
+            [fetched_feature_flags.append(splits.from_raw(feature_flag)) for feature_flag in feature_flag_changes.get('splits', [])]
+            segment_list = update_feature_flag_storage(self._feature_flag_storage, fetched_feature_flags, feature_flag_changes['till'])
             if feature_flag_changes['till'] == feature_flag_changes['since']:
                 return feature_flag_changes['till'], segment_list
 
@@ -127,15 +149,15 @@ def synchronize_splits(self, till=None):
         :type till: int
         """
         final_segment_list = set()
-        fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
+        fetch_options = FetchOptions(True, sets=self._get_config_sets())  # Set Cache-Control to no-cache
         successful_sync, remaining_attempts, change_number, segment_list = self._attempt_feature_flag_sync(fetch_options,
                                                                                       till)
         final_segment_list.update(segment_list)
         attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if successful_sync:  # succedeed sync
             _LOGGER.debug('Refresh completed in %d attempts.', attempts)
             return final_segment_list
-        with_cdn_bypass = FetchOptions(True, change_number)  # Set flag for bypassing CDN
+        with_cdn_bypass = FetchOptions(True, change_number, sets=self._get_config_sets())  # Set flag for bypassing CDN
         without_cdn_successful_sync, remaining_attempts, change_number, segment_list = self._attempt_feature_flag_sync(with_cdn_bypass, till)
         final_segment_list.update(segment_list)
         without_cdn_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
@@ -160,8 +182,7 @@ def kill_split(self, feature_flag_name, default_treatment, change_number):
         """
         self._feature_flag_storage.kill_locally(feature_flag_name, default_treatment, change_number)
 
-
-class SplitSynchronizerAsync(object):
+class SplitSynchronizerAsync(SplitSynchronizerBase):
     """Feature Flag changes synchronizer async."""
 
     def __init__(self, feature_flag_api, feature_flag_storage):
@@ -174,16 +195,7 @@ def __init__(self, feature_flag_api, feature_flag_storage):
         :param feature_flag_storage: Feature Flag Storage.
         :type feature_flag_storage: splitio.storage.InMemorySplitStorage
         """
-        self._api = feature_flag_api
-        self._feature_flag_storage = feature_flag_storage
-        self._backoff = Backoff(
-                                _ON_DEMAND_FETCH_BACKOFF_BASE,
-                                _ON_DEMAND_FETCH_BACKOFF_MAX_WAIT)
-
-    @property
-    def feature_flag_storage(self):
-        """Return Feature_flag storage object"""
-        return self._feature_flag_storage
+        super().__init__(feature_flag_api, feature_flag_storage)
 
     async def _fetch_until(self, fetch_options, till=None):
         """
@@ -214,13 +226,9 @@ async def _fetch_until(self, fetch_options, till=None):
                 _LOGGER.debug('Exception information: ', exc_info=True)
                 raise exc
 
-            for feature_flag in feature_flag_changes.get('splits', []):
-                if feature_flag['status'] == splits.Status.ACTIVE.value:
-                    parsed = splits.from_raw(feature_flag)
-                    await self._feature_flag_storage.put(parsed)
-                    segment_list.update(set(parsed.get_segment_names()))
-                else:
-                    await self._feature_flag_storage.remove(feature_flag['name'])
+            fetched_feature_flags = []
+            [fetched_feature_flags.append(splits.from_raw(feature_flag)) for feature_flag in feature_flag_changes.get('splits', [])]
+            segment_list = await update_feature_flag_storage_async(self._feature_flag_storage, fetched_feature_flags, feature_flag_changes['till'])
             await self._feature_flag_storage.set_change_number(feature_flag_changes['till'])
             if feature_flag_changes['till'] == feature_flag_changes['since']:
                 return feature_flag_changes['till'], segment_list
@@ -260,15 +268,15 @@ async def synchronize_splits(self, till=None):
         :type till: int
         """
         final_segment_list = set()
-        fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
+        fetch_options = FetchOptions(True, sets=self._get_config_sets())  # Set Cache-Control to no-cache
         successful_sync, remaining_attempts, change_number, segment_list = await self._attempt_feature_flag_sync(fetch_options,
                                                                                       till)
         final_segment_list.update(segment_list)
         attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if successful_sync:  # succedeed sync
             _LOGGER.debug('Refresh completed in %d attempts.', attempts)
             return final_segment_list
-        with_cdn_bypass = FetchOptions(True, change_number)  # Set flag for bypassing CDN
+        with_cdn_bypass = FetchOptions(True, change_number, sets=self._get_config_sets())  # Set flag for bypassing CDN
         without_cdn_successful_sync, remaining_attempts, change_number, segment_list = await self._attempt_feature_flag_sync(with_cdn_bypass, till)
         final_segment_list.update(segment_list)
         without_cdn_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
@@ -430,6 +438,9 @@ def _sanitize_feature_flag_elements(self, parsed_feature_flags):
                             ('algo', 2, 2, 2, None, None)]:
                 feature_flag = util._sanitize_object_element(feature_flag, 'split', element[0], element[1], lower_value=element[2], upper_value=element[3], in_list=element[4], not_in_list=element[5])
             feature_flag = self._sanitize_condition(feature_flag)
+            if 'sets' not in feature_flag:
+                feature_flag['sets'] = []
+            feature_flag['sets'] = validate_flag_sets(feature_flag['sets'], 'Localhost Validator')
             sanitized_feature_flags.append(feature_flag)
         return sanitized_feature_flags
 
@@ -604,12 +615,8 @@ def _synchronize_legacy(self):
             fetched = self._read_feature_flags_from_legacy_file(self._filename)
         to_delete = [name for name in self._feature_flag_storage.get_split_names()
                      if name not in fetched.keys()]
-        for feature_flag in fetched.values():
-            self._feature_flag_storage.put(feature_flag)
-
-        for feature_flag in to_delete:
-            self._feature_flag_storage.remove(feature_flag)
-
+        to_add = [feature_flag for feature_flag in fetched.values()]
+        self._feature_flag_storage.update(to_add, to_delete, 0)
         return []
 
     def _synchronize_json(self):
@@ -628,18 +635,12 @@ def _synchronize_json(self):
             self._current_json_sha = fecthed_sha
             if self._feature_flag_storage.get_change_number() > till and till != self._DEFAULT_FEATURE_FLAG_TILL:
                 return []
-            for feature_flag in fetched:
-                if feature_flag['status'] == splits.Status.ACTIVE.value:
-                    parsed = splits.from_raw(feature_flag)
-                    self._feature_flag_storage.put(parsed)
-                    _LOGGER.debug("feature flag %s is updated", parsed.name)
-                    segment_list.update(set(parsed.get_segment_names()))
-                else:
-                    self._feature_flag_storage.remove(feature_flag['name'])
 
-                self._feature_flag_storage.set_change_number(till)
+            fetched_feature_flags = [fetched_feature_flags.append(splits.from_raw(feature_flag)) for feature_flag in fetched]
+            segment_list = update_feature_flag_storage(self._feature_flag_storage, fetched_feature_flags, till)
             return segment_list
         except Exception as exc:
+            _LOGGER.debug(exc)
             raise ValueError("Error reading feature flags from json.") from exc
 
     def _read_feature_flags_from_json_file(self, filename):
@@ -758,11 +759,8 @@ async def _synchronize_legacy(self):
             fetched = await self._read_feature_flags_from_legacy_file(self._filename)
         to_delete = [name for name in await self._feature_flag_storage.get_split_names()
                      if name not in fetched.keys()]
-        for feature_flag in fetched.values():
-            await self._feature_flag_storage.put(feature_flag)
-
-        for feature_flag in to_delete:
-            await self._feature_flag_storage.remove(feature_flag)
+        to_add = [feature_flag for feature_flag in fetched.values()]
+        await self._feature_flag_storage.update(to_add, to_delete, 0)
 
         return []
 
@@ -782,18 +780,11 @@ async def _synchronize_json(self):
             self._current_json_sha = fecthed_sha
             if await self._feature_flag_storage.get_change_number() > till and till != self._DEFAULT_FEATURE_FLAG_TILL:
                 return []
-            for feature_flag in fetched:
-                if feature_flag['status'] == splits.Status.ACTIVE.value:
-                    parsed = splits.from_raw(feature_flag)
-                    await self._feature_flag_storage.put(parsed)
-                    _LOGGER.debug("feature flag %s is updated", parsed.name)
-                    segment_list.update(set(parsed.get_segment_names()))
-                else:
-                    await self._feature_flag_storage.remove(feature_flag['name'])
-
-                await self._feature_flag_storage.set_change_number(till)
+            fetched_feature_flags = [fetched_feature_flags.append(splits.from_raw(feature_flag)) for feature_flag in fetched]
+            segment_list = await update_feature_flag_storage(self._feature_flag_storage, fetched_feature_flags, till)
             return segment_list
         except Exception as exc:
+            _LOGGER.debug(exc)
             raise ValueError("Error reading feature flags from json.") from exc
 
     async def _read_feature_flags_from_json_file(self, filename):

splitio/sync/synchronizer.py

@@ -252,6 +252,7 @@ def __init__(self, split_synchronizers, split_tasks):
             self._periodic_data_recording_tasks.append(self._split_tasks.unique_keys_task)
         if self._split_tasks.clear_filter_task:
             self._periodic_data_recording_tasks.append(self._split_tasks.clear_filter_task)
+        self._break_sync_all = False
 
     @property
     def split_sync(self):
@@ -384,6 +385,7 @@ def synchronize_splits(self, till, sync_segments=True):
         :returns: whether the synchronization was successful or not.
         :rtype: bool
         """
+        self._break_sync_all = False
         _LOGGER.debug('Starting splits synchronization')
         try:
             new_segments = []
@@ -399,7 +401,9 @@ def synchronize_splits(self, till, sync_segments=True):
                 else:
                     _LOGGER.debug('Segment sync scheduled.')
             return True
-        except APIException:
+        except APIException as exc:
+            if exc._status_code is not None and exc._status_code == 414:
+                self._break_sync_all = True
             _LOGGER.error('Failed syncing feature flags')
             _LOGGER.debug('Error: ', exc_info=True)
             return False
@@ -429,7 +433,7 @@ def sync_all(self, max_retry_attempts=_SYNC_ALL_NO_RETRIES):
                 _LOGGER.debug('Error: ', exc_info=True)
                 if max_retry_attempts != _SYNC_ALL_NO_RETRIES:
                     retry_attempts += 1
-                    if retry_attempts > max_retry_attempts:
+                    if retry_attempts > max_retry_attempts or self._break_sync_all:
                         break
                 how_long = self._backoff.get()
                 time.sleep(how_long)
@@ -536,6 +540,7 @@ async def synchronize_splits(self, till, sync_segments=True):
         :returns: whether the synchronization was successful or not.
         :rtype: bool
         """
+        self._break_sync_all = False
         _LOGGER.debug('Starting feature flags synchronization')
         try:
             new_segments = []
@@ -551,7 +556,9 @@ async def synchronize_splits(self, till, sync_segments=True):
                 else:
                     _LOGGER.debug('Segment sync scheduled.')
             return True
-        except APIException:
+        except APIException as exc:
+            if exc._status_code is not None and exc._status_code == 414:
+                self._break_sync_all = True
             _LOGGER.error('Failed syncing feature flags')
             _LOGGER.debug('Error: ', exc_info=True)
             return False
@@ -581,7 +588,7 @@ async def sync_all(self, max_retry_attempts=_SYNC_ALL_NO_RETRIES):
                 _LOGGER.debug('Error: ', exc_info=True)
                 if max_retry_attempts != _SYNC_ALL_NO_RETRIES:
                     retry_attempts += 1
-                    if retry_attempts > max_retry_attempts:
+                    if retry_attempts > max_retry_attempts or self._break_sync_all:
                         break
                 how_long = self._backoff.get()
                 time.sleep(how_long)

splitio/tasks/util/asynctask.py

@@ -113,7 +113,7 @@ def _execution_wrapper(self):
                         _LOGGER.debug("Force execution signal received. Running now")
                         if not _safe_run(self._main):
                             _LOGGER.error("An error occurred when executing the task. "
-                                          "Retrying after perio expires")
+                                          "Retrying after period expires")
                         continue
                 except queue.Empty:
                     # If no message was received, the timeout has expired
@@ -123,7 +123,7 @@ def _execution_wrapper(self):
                 if not _safe_run(self._main):
                     _LOGGER.error(
                         "An error occurred when executing the task. "
-                        "Retrying after perio expires"
+                        "Retrying after period expires"
                     )
         finally:
             self._cleanup()
@@ -252,7 +252,7 @@ async def _execution_wrapper(self):
                         _LOGGER.debug("Force execution signal received. Running now")
                         if not await _safe_run_async(self._main):
                             _LOGGER.error("An error occurred when executing the task. "
-                                          "Retrying after perio expires")
+                                          "Retrying after period expires")
                         continue
                 except asyncio.QueueEmpty:
                     # If no message was received, the timeout has expired