Merge branch 'Feature/Async' into async-flagsets-sync

Author: chillaq

splitio/api/commons.py

@@ -6,7 +6,7 @@
 class FetchOptions(object):
     """Fetch Options object."""
 
-    def __init__(self, cache_control_headers=False, change_number=None):
+    def __init__(self, cache_control_headers=False, change_number=None, sets=None):
         """
         Class constructor.
 
@@ -15,9 +15,13 @@ def __init__(self, cache_control_headers=False, change_number=None):
 
         :param change_number: ChangeNumber to use for bypassing CDN in request.
         :type change_number: int
+
+        :param sets: list of flag sets
+        :type sets: list
         """
         self._cache_control_headers = cache_control_headers
         self._change_number = change_number
+        self._sets = sets
 
     @property
     def cache_control_headers(self):
@@ -29,12 +33,19 @@ def change_number(self):
         """Return change number."""
         return self._change_number
 
+    @property
+    def sets(self):
+        """Return sets."""
+        return self._sets
+
     def __eq__(self, other):
         """Match between other options."""
         if self._cache_control_headers != other._cache_control_headers:
             return False
         if self._change_number != other._change_number:
             return False
+        if self._sets != other._sets:
+            return False
         return True
 
 
@@ -62,4 +73,6 @@ def build_fetch(change_number, fetch_options, metadata):
         extra_headers[_CACHE_CONTROL] = _CACHE_CONTROL_NO_CACHE
     if fetch_options.change_number is not None:
         query['till'] = fetch_options.change_number
+    if fetch_options.sets is not None:
+        query['sets'] = fetch_options.sets
     return query, extra_headers

splitio/api/splits.py

@@ -57,6 +57,8 @@ def fetch_splits(self, change_number, fetch_options):
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)
             else:
+                if response.status_code == 414:
+                    _LOGGER.error('Error fetching feature flags; the amount of flag sets provided are too big, causing uri length error.')
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:
             _LOGGER.error('Error fetching feature flags because an exception was raised by the HTTPClient')
@@ -109,6 +111,8 @@ async def fetch_splits(self, change_number, fetch_options):
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)
             else:
+                if response.status_code == 414:
+                    _LOGGER.error('Error fetching feature flags; the amount of flag sets provided are too big, causing uri length error.')
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:
             _LOGGER.error('Error fetching feature flags because an exception was raised by the HTTPClient')

splitio/api/telemetry.py

@@ -71,7 +71,6 @@ def record_init(self, configs):
                 'Error posting init config because an exception was raised by the HTTPClient'
             )
             _LOGGER.debug('Error: ', exc_info=True)
-            raise APIException('Init config data not flushed properly.') from exc
 
     def record_stats(self, stats):
         """
@@ -162,7 +161,6 @@ async def record_init(self, configs):
                 'Error posting init config because an exception was raised by the HTTPClient'
             )
             _LOGGER.debug('Error: ', exc_info=True)
-            raise APIException('Init config data not flushed properly.') from exc
 
     async def record_stats(self, stats):
         """

splitio/client/client.py

@@ -3,7 +3,7 @@
 import logging
 
 from splitio.engine.impressions import ImpressionsMode
-
+from splitio.client.input_validator import validate_flag_sets
 
 _LOGGER = logging.getLogger(__name__)
 DEFAULT_DATA_SAMPLING = 1
@@ -58,7 +58,8 @@
     'dataSampling': DEFAULT_DATA_SAMPLING,
     'storageWrapper': None,
     'storagePrefix': None,
-    'storageType': None
+    'storageType': None,
+    'flagSetsFilter': None
 }
 
 
@@ -143,4 +144,10 @@ def sanitize(sdk_key, config):
         _LOGGER.warning('metricRefreshRate parameter minimum value is 60 seconds, defaulting to 3600 seconds.')
         processed['metricsRefreshRate'] = 3600
 
+    if config['operationMode'] == 'consumer' and config.get('flagSetsFilter') is not None:
+        processed['flagSetsFilter'] = None
+        _LOGGER.warning('config: FlagSets filter is not applicable for Consumer modes where the SDK does keep rollout data in sync. FlagSet filter was discarded.')
+    else:
+        processed['flagSetsFilter'] = sorted(validate_flag_sets(processed['flagSetsFilter'], 'SDK Config')) if processed['flagSetsFilter'] is not None else None
+
     return processed

splitio/client/config.py

@@ -410,7 +410,7 @@ async def block_until_ready(self, timeout=None):
             await asyncio.wait_for(asyncio.shield(self._sdk_ready_flag.wait()), timeout)
         except asyncio.TimeoutError as e:
             _LOGGER.error("Exception initializing SDK")
-            _LOGGER.error(str(e))
+            _LOGGER.debug(str(e))
             await self._telemetry_init_producer.record_bur_time_out()
             raise TimeoutException('SDK Initialization: time of %d exceeded' % timeout)
 
@@ -498,7 +498,8 @@ def _wrap_impression_listener_async(listener, metadata):
     return None
 
 def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pylint:disable=too-many-arguments,too-many-locals
-                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None):
+                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None,
+                             total_flag_sets=0, invalid_flag_sets=0):
     """Build and return a split factory tailored to the supplied config."""
     if not input_validator.validate_factory_instantiation(api_key):
         return None
@@ -536,7 +537,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     }
 
     storages = {
-        'splits': InMemorySplitStorage(),
+        'splits': InMemorySplitStorage(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorage(),
         'impressions': InMemoryImpressionStorage(cfg['impressionsQueueSize'], telemetry_runtime_producer),
         'events': InMemoryEventStorage(cfg['eventsQueueSize'], telemetry_runtime_producer),
@@ -607,7 +608,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         unique_keys_tracker=unique_keys_tracker
     )
 
-    telemetry_init_producer.record_config(cfg, extra_cfg)
+    telemetry_init_producer.record_config(cfg, extra_cfg, total_flag_sets, invalid_flag_sets)
 
     if preforked_initialization:
         synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
@@ -625,7 +626,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
                         telemetry_submitter)
 
 async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=None,  # pylint:disable=too-many-arguments,too-many-localsa
-                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None):
+                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None,
+                             total_flag_sets=0, invalid_flag_sets=0):
     """Build and return a split factory tailored to the supplied config in async mode."""
     if not input_validator.validate_factory_instantiation(api_key):
         return None
@@ -663,7 +665,7 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
     }
 
     storages = {
-        'splits': InMemorySplitStorageAsync(),
+        'splits': InMemorySplitStorageAsync(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorageAsync(),
         'impressions': InMemoryImpressionStorageAsync(cfg['impressionsQueueSize'], telemetry_runtime_producer),
         'events': InMemoryEventStorageAsync(cfg['eventsQueueSize'], telemetry_runtime_producer),
@@ -733,7 +735,7 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
         unique_keys_tracker=unique_keys_tracker
     )
 
-    await telemetry_init_producer.record_config(cfg, extra_cfg)
+    await telemetry_init_producer.record_config(cfg, extra_cfg, total_flag_sets, invalid_flag_sets)
 
     if preforked_initialization:
         await synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
@@ -814,7 +816,7 @@ def _build_redis_factory(api_key, cfg):
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True)
     initialization_thread.start()
 
-    telemetry_init_producer.record_config(cfg, {})
+    telemetry_init_producer.record_config(cfg, {}, 0, 0)
 
     split_factory = SplitFactory(
         api_key,
@@ -894,7 +896,7 @@ async def _build_redis_factory_async(api_key, cfg):
     )
 
     manager = RedisManagerAsync(synchronizer)
-    await telemetry_init_producer.record_config(cfg, {})
+    await telemetry_init_producer.record_config(cfg, {}, 0, 0)
     manager.start()
 
     split_factory = SplitFactoryAsync(
@@ -977,7 +979,7 @@ def _build_pluggable_factory(api_key, cfg):
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True)
     initialization_thread.start()
 
-    telemetry_init_producer.record_config(cfg, {})
+    telemetry_init_producer.record_config(cfg, {}, 0, 0)
 
     split_factory = SplitFactory(
         api_key,
@@ -1056,7 +1058,7 @@ async def _build_pluggable_factory_async(api_key, cfg):
     # Using same class as redis for consumer mode only
     manager = RedisManagerAsync(synchronizer)
     manager.start()
-    await telemetry_init_producer.record_config(cfg, {})
+    await telemetry_init_producer.record_config(cfg, {}, 0, 0)
 
     split_factory = SplitFactoryAsync(
         api_key,
@@ -1083,7 +1085,7 @@ def _build_localhost_factory(cfg):
     telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
 
     storages = {
-        'splits': InMemorySplitStorage(),
+        'splits': InMemorySplitStorage(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorage(),  # not used, just to avoid possible future errors.
         'impressions': LocalhostImpressionsStorage(),
         'events': LocalhostEventsStorage(),
@@ -1238,7 +1240,10 @@ def get_factory(api_key, **kwargs):
     _INSTANTIATED_FACTORIES.update([api_key])
     _INSTANTIATED_FACTORIES_LOCK.release()
 
-    config = sanitize_config(api_key, kwargs.get('config', {}))
+    config_raw = kwargs.get('config', {})
+    total_flag_sets, invalid_flag_sets = _get_total_and_invalid_flag_sets(config_raw)
+
+    config = sanitize_config(api_key, config_raw)
 
     if config['operationMode'] == 'localhost':
         split_factory =  _build_localhost_factory(config)
@@ -1254,7 +1259,9 @@ def get_factory(api_key, **kwargs):
         kwargs.get('events_api_base_url'),
         kwargs.get('auth_api_base_url'),
         kwargs.get('streaming_api_base_url'),
-        kwargs.get('telemetry_api_base_url'))
+        kwargs.get('telemetry_api_base_url'),
+        total_flag_sets,
+        invalid_flag_sets)
 
     return split_factory
 
@@ -1282,8 +1289,10 @@ async def get_factory_async(api_key, **kwargs):
     _INSTANTIATED_FACTORIES.update([api_key])
     _INSTANTIATED_FACTORIES_LOCK.release()
 
-    config = sanitize_config(api_key, kwargs.get('config', {}))
+    config_raw = kwargs.get('config', {})
+    total_flag_sets, invalid_flag_sets = _get_total_and_invalid_flag_sets(config_raw)
 
+    config = sanitize_config(api_key, config_raw)
     if config['operationMode'] == 'localhost':
         split_factory =  await _build_localhost_factory_async(config)
     elif config['storageType'] == 'redis':
@@ -1298,8 +1307,9 @@ async def get_factory_async(api_key, **kwargs):
         kwargs.get('events_api_base_url'),
         kwargs.get('auth_api_base_url'),
         kwargs.get('streaming_api_base_url'),
-        kwargs.get('telemetry_api_base_url'))
-
+        kwargs.get('telemetry_api_base_url'),
+        total_flag_sets,
+        invalid_flag_sets)
     return split_factory
 
 def _get_active_and_redundant_count():
@@ -1310,4 +1320,13 @@ def _get_active_and_redundant_count():
         redundant_factory_count += _INSTANTIATED_FACTORIES[item] - 1
         active_factory_count += _INSTANTIATED_FACTORIES[item]
     _INSTANTIATED_FACTORIES_LOCK.release()
-    return redundant_factory_count, active_factory_count
+    return redundant_factory_count, active_factory_count
+
+def _get_total_and_invalid_flag_sets(config_raw):
+    total_flag_sets = 0
+    invalid_flag_sets = 0
+    if config_raw.get('flagSetsFilter') is not None and isinstance(config_raw.get('flagSetsFilter'), list):
+        total_flag_sets = len(config_raw.get('flagSetsFilter'))
+        invalid_flag_sets = total_flag_sets - len(input_validator.validate_flag_sets(config_raw.get('flagSetsFilter'), 'Telemetry Init'))
+
+    return total_flag_sets, invalid_flag_sets