Merge pull request #499 from splitio/async-flagsets-sync

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

Author: chillaq

splitio/api/__init__.py

@@ -14,6 +14,13 @@ def status_code(self):
         """Return HTTP status code."""
         return self._status_code
 
+class APIUriException(APIException):
+    """Exception to raise when an API call fails due to 414  http error."""
+
+    def __init__(self, custom_message, status_code=None):
+        """Constructor."""
+        APIException.__init__(self, custom_message)
+
 def headers_from_metadata(sdk_metadata, client_key=None):
     """
     Generate a dict with headers required by data-recording API endpoints.

splitio/push/workers.py

@@ -232,7 +232,13 @@ def _run(self):
                         _LOGGER.error('Exception raised in updating feature flag')
                         _LOGGER.debug('Exception information: ', exc_info=True)
                         pass
-                self._handler(event.change_number)
+                sync_result = self._handler(event.change_number)
+                if not sync_result.success and sync_result.error_code is not None and sync_result.error_code == 414:
+                    _LOGGER.error("URI too long exception caught, sync failed")
+
+                if not sync_result.success:
+                    _LOGGER.error("feature flags sync failed")
+
             except Exception as e:  # pylint: disable=broad-except
                 _LOGGER.error('Exception raised in feature flag synchronization')
                 _LOGGER.debug('Exception information: ', exc_info=True)

splitio/sync/manager.py

@@ -285,7 +285,7 @@ def __init__(self, synchronizer):  # pylint:disable=too-many-arguments
         :param synchronizer: synchronizers for performing start/stop logic
         :type synchronizer: splitio.sync.synchronizer.Synchronizer
         """
-        super().__init__(synchronizer)
+        RedisManagerBase.__init__(self, synchronizer)
 
     def stop(self, blocking):
         """
@@ -308,7 +308,7 @@ def __init__(self, synchronizer):  # pylint:disable=too-many-arguments
         :param synchronizer: synchronizers for performing start/stop logic
         :type synchronizer: splitio.sync.synchronizer.Synchronizer
         """
-        super().__init__(synchronizer)
+        RedisManagerBase.__init__(self, synchronizer)
 
     async def stop(self, blocking):
         """

splitio/sync/split.py

@@ -8,11 +8,13 @@
 import hashlib
 from enum import Enum
 
-from splitio.api import APIException
+from splitio.api import APIException, APIUriException
 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
+        """
+        SplitSynchronizerBase.__init__(self, feature_flag_api, feature_flag_storage)
+
     def _fetch_until(self, fetch_options, till=None):
         """
         Hit endpoint, update storage and return when since==till.
@@ -77,18 +104,17 @@ def _fetch_until(self, fetch_options, till=None):
             try:
                 feature_flag_changes = self._api.fetch_splits(change_number, fetch_options)
             except APIException as exc:
+                if exc._status_code is not None and exc._status_code == 414:
+                    _LOGGER.error('Exception caught: the amount of flag sets provided are big causing uri length error.')
+                    _LOGGER.debug('Exception information: ', exc_info=True)
+                    raise APIUriException("URI is too long due to FlagSets count", exc._status_code)
+
                 _LOGGER.error('Exception raised while fetching feature flags')
                 _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 = [(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 +153,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 +186,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 +199,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
+        SplitSynchronizerBase.__init__(self, feature_flag_api, feature_flag_storage)
 
     async def _fetch_until(self, fetch_options, till=None):
         """
@@ -210,18 +226,17 @@ async def _fetch_until(self, fetch_options, till=None):
             try:
                 feature_flag_changes = await self._api.fetch_splits(change_number, fetch_options)
             except APIException as exc:
+                if exc._status_code is not None and exc._status_code == 414:
+                    _LOGGER.error('Exception caught: the amount of flag sets provided are big causing uri length error.')
+                    _LOGGER.debug('Exception information: ', exc_info=True)
+                    raise APIUriException("URI is too long due to FlagSets count", exc._status_code)
+
                 _LOGGER.error('Exception raised while fetching feature flags')
                 _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'])
-            await self._feature_flag_storage.set_change_number(feature_flag_changes['till'])
+            fetched_feature_flags = [(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'])
             if feature_flag_changes['till'] == feature_flag_changes['since']:
                 return feature_flag_changes['till'], segment_list
 
@@ -260,15 +275,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 +445,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
 
@@ -587,7 +605,7 @@ def synchronize_splits(self, till=None):  # pylint:disable=unused-argument
         try:
             return self._synchronize_json() if self._localhost_mode == LocalhostMode.JSON else self._synchronize_legacy()
         except Exception as exc:
-            _LOGGER.error(str(exc))
+            _LOGGER.debug('Exception: ', exc_info=True)
             raise APIException("Error fetching feature flags information") from exc
 
     def _synchronize_legacy(self):
@@ -604,12 +622,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 +642,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 = [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('Exception: ', exc_info=True)
             raise ValueError("Error reading feature flags from json.") from exc
 
     def _read_feature_flags_from_json_file(self, filename):
@@ -658,7 +666,7 @@ def _read_feature_flags_from_json_file(self, filename):
             santitized = self._sanitize_feature_flag(parsed)
             return santitized['splits'], santitized['till']
         except Exception as exc:
-            _LOGGER.error(str(exc))
+            _LOGGER.debug('Exception: ', exc_info=True)
             raise ValueError("Error parsing file %s. Make sure it's readable." % filename) from exc
 
 
@@ -741,7 +749,7 @@ async def synchronize_splits(self, till=None):  # pylint:disable=unused-argument
         try:
             return await self._synchronize_json() if self._localhost_mode == LocalhostMode.JSON else await self._synchronize_legacy()
         except Exception as exc:
-            _LOGGER.error(str(exc))
+            _LOGGER.debug('Exception: ', exc_info=True)
             raise APIException("Error fetching feature flags information") from exc
 
     async def _synchronize_legacy(self):
@@ -758,11 +766,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 +787,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 = [splits.from_raw(feature_flag) for feature_flag in fetched]
+            segment_list = await update_feature_flag_storage_async(self._feature_flag_storage, fetched_feature_flags, till)
             return segment_list
         except Exception as exc:
+            _LOGGER.debug('Exception: ', exc_info=True)
             raise ValueError("Error reading feature flags from json.") from exc
 
     async def _read_feature_flags_from_json_file(self, filename):
@@ -812,5 +810,5 @@ async def _read_feature_flags_from_json_file(self, filename):
             santitized = self._sanitize_feature_flag(parsed)
             return santitized['splits'], santitized['till']
         except Exception as exc:
-            _LOGGER.error(str(exc))
+            _LOGGER.debug('Exception: ', exc_info=True)
             raise ValueError("Error parsing file %s. Make sure it's readable." % filename) from exc