update configs

Author: mredolatti

splitio/client/config.py

@@ -1,15 +1,20 @@
 """Default settings for the Split.IO SDK Python client."""
 from __future__ import absolute_import, division, print_function, unicode_literals
+
 import os.path
 
+from splitio.engine.impressions import ImpressionsMode
+
+
 DEFAULT_CONFIG = {
+    'operationMode': 'in-memory',
     'connectionTimeout': 1500,
     'splitSdkMachineName': None,
     'splitSdkMachineIp': None,
     'featuresRefreshRate': 5,
     'segmentsRefreshRate': 60,
     'metricsRefreshRate': 60,
-    'impressionsRefreshRate': 10,
+    'impressionsRefreshRate': 5 * 60,
     'impressionsBulkSize': 5000,
     'impressionsQueueSize': 10000,
     'eventsPushRate': 10,
@@ -47,3 +52,69 @@
     'machineIp': None,
     'splitFile': os.path.join(os.path.expanduser('~'), '.split')
 }
+
+
+def _parse_operation_mode(apikey, config):
+    """
+    Process incoming config to determine operation mode.
+
+    :param config: user supplied config
+    :type config: dict
+
+    :returns: operation mode
+    :rtype: str
+    """
+    if apikey == 'localhost':
+        return 'localhost-standalone'
+
+    if 'redisHost' in config or 'redisSentinels' in config:
+        return 'redis-consumer'
+
+    if 'uwsgiClient' in config:
+        return 'uwsgi-consumer'
+
+    return 'inmemory-standalone'
+
+def _sanitize_impressions_mode(mode, refresh_rate=None):
+    """
+    Check supplied impressions mode and adjust refresh rate.
+
+    :param config: default + supplied config
+    :type config: dict
+
+    :returns: config with sanitized impressions mode & refresh rate
+    :rtype: config
+    """
+    if not isinstance(mode, ImpressionsMode):
+        try:
+            mode = ImpressionsMode(mode)
+        except ValueError:
+            mode = ImpressionsMode.OPTIMIZED
+
+    if mode == ImpressionsMode.DEBUG:
+        refresh_rate = max(1, refresh_rate) if refresh_rate is not None else 60
+    else:
+        refresh_rate = max(60, refresh_rate) if refresh_rate is not None else 5 * 60
+
+    return mode, refresh_rate
+
+def sanitize(apikey, config):
+    """
+    Look for inconsistencies or ill-formed configs and tune it accordingly.
+
+    :param apikey: customer's apikey
+    :type apikey: str
+
+    :param config: DEFAULT + user supplied config
+    :type config: dict
+
+    :returns: sanitized config
+    :rtype: dict
+    """
+    config['operationMode'] = _parse_operation_mode(apikey, config)
+    processed = DEFAULT_CONFIG.copy()
+    processed.update(config)
+    imp_mode, imp_rate = _sanitize_impressions_mode(config.get('impressionsMode'), config.get('impressionsRefreshRate'))
+    processed['impressionsMode'] = imp_mode
+    processed['impressionsRefreshRate'] = imp_rate
+    return processed

splitio/client/factory.py

@@ -12,7 +12,7 @@
 from splitio.client.client import Client
 from splitio.client import input_validator
 from splitio.client.manager import SplitManager
-from splitio.client.config import DEFAULT_CONFIG
+from splitio.client.config import sanitize as sanitize_config
 from splitio.client import util
 from splitio.client.listener import ImpressionListenerWrapper
 from splitio.engine.impressions import Manager as ImpressionsManager
@@ -38,7 +38,7 @@
 # Tasks
 from splitio.tasks.split_sync import SplitSynchronizationTask
 from splitio.tasks.segment_sync import SegmentSynchronizationTask
-from splitio.tasks.impressions_sync import ImpressionsSyncTask
+from splitio.tasks.impressions_sync import ImpressionsSyncTask, ImpressionsCountSyncTask
 from splitio.tasks.events_sync import EventsSyncTask
 from splitio.tasks.telemetry_sync import TelemetrySynchronizationTask
 
@@ -234,13 +234,11 @@ def _wrap_impression_listener(listener, metadata):
     return None
 
 
-def _build_in_memory_factory(api_key, config, sdk_url=None, events_url=None):  # pylint: disable=too-many-locals
+def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None):  # pylint: disable=too-many-locals
     """Build and return a split factory tailored to the supplied config."""
     if not input_validator.validate_factory_instantiation(api_key):
         return None
 
-    cfg = DEFAULT_CONFIG.copy()
-    cfg.update(config)
     http_client = HttpClient(
         sdk_url=sdk_url,
         events_url=events_url,
@@ -272,6 +270,12 @@ def _build_in_memory_factory(api_key, config, sdk_url=None, events_url=None):  #
     segments_ready_flag = threading.Event()
     sdk_ready_flag = threading.Event()
 
+    imp_manager = ImpressionsManager(
+        storages['impressions'].put,
+        cfg['impressionsMode'],
+        True,
+        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata))
+
     tasks = {
         'splits': SplitSynchronizationTask(
             apis['splits'],
@@ -295,6 +299,11 @@ def _build_in_memory_factory(api_key, config, sdk_url=None, events_url=None):  #
             cfg['impressionsBulkSize']
         ),
 
+        'impressions_count': ImpressionsCountSyncTask(
+            apis['impressions'],
+            imp_manager
+        ),
+
         'events': EventsSyncTask(
             apis['events'],
             storages['events'],
@@ -312,6 +321,7 @@ def _build_in_memory_factory(api_key, config, sdk_url=None, events_url=None):  #
     # Start tasks that have no dependencies
     tasks['splits'].start()
     tasks['impressions'].start()
+    tasks['impressions_count'].start()
     tasks['events'].start()
     tasks['telemetry'].start()
 
@@ -334,22 +344,13 @@ def segment_ready_task():
     segment_completion_thread = threading.Thread(target=segment_ready_task)
     segment_completion_thread.setDaemon(True)
     segment_completion_thread.start()
-    return SplitFactory(
-        api_key,
-        storages,
-        cfg['labelsEnabled'],
-        ImpressionsManager(storages['impressions'].put, cfg['impressionsMode'], True,
-                           _wrap_impression_listener(cfg['impressionListener'], sdk_metadata)),
-        apis,
-        tasks,
-        sdk_ready_flag,
-    )
 
+    return SplitFactory(api_key, storages, cfg['labelsEnabled'],
+                        imp_manager, apis, tasks, sdk_ready_flag)
 
-def _build_redis_factory(api_key, config):
+
+def _build_redis_factory(api_key, cfg):
     """Build and return a split factory with redis-based storage."""
-    cfg = DEFAULT_CONFIG.copy()
-    cfg.update(config)
     sdk_metadata = util.get_metadata(cfg)
     redis_adapter = redis.build(cfg)
     cache_enabled = cfg.get('redisLocalCacheEnabled', False)
@@ -370,10 +371,8 @@ def _build_redis_factory(api_key, config):
     )
 
 
-def _build_uwsgi_factory(api_key, config):
+def _build_uwsgi_factory(api_key, cfg):
     """Build and return a split factory with redis-based storage."""
-    cfg = DEFAULT_CONFIG.copy()
-    cfg.update(config)
     sdk_metadata = util.get_metadata(cfg)
     uwsgi_adapter = get_uwsgi()
     storages = {
@@ -392,10 +391,8 @@ def _build_uwsgi_factory(api_key, config):
     )
 
 
-def _build_localhost_factory(config):
+def _build_localhost_factory(cfg):
     """Build and return a localhost factory for testing/development purposes."""
-    cfg = DEFAULT_CONFIG.copy()
-    cfg.update(config)
     storages = {
         'splits': InMemorySplitStorage(),
         'segments': InMemorySegmentStorage(),  # not used, just to avoid possible future errors.
@@ -413,12 +410,12 @@ def _build_localhost_factory(config):
     )}
     tasks['splits'].start()
     return SplitFactory(
-        'localhost', 
-        storages, 
-        False, 
+        'localhost',
+        storages,
+        False,
         ImpressionsManager(storages['impressions'].put, cfg['impressionsMode'], True, None),
-        None, 
-        tasks, 
+        None,
+        tasks,
         ready_event
     )
 
@@ -444,15 +441,15 @@ def get_factory(api_key, **kwargs):
                     "(Singleton pattern) and reusing it throughout your application."
                 )
 
-        config = kwargs.get('config', {})
+        config = sanitize_config(api_key, kwargs.get('config', {}))
 
-        if api_key == 'localhost':
+        if config['operationMode'] == 'localhost-standalone':
             return _build_localhost_factory(config)
 
-        if 'redisHost' in config or 'redisSentinels' in config:
+        if config['operationMode'] == 'redis-consumer':
             return _build_redis_factory(api_key, config)
 
-        if 'uwsgiClient' in config:
+        if config['operationMode'] == 'uwsgi-consumer':
             return _build_uwsgi_factory(api_key, config)
 
         return _build_in_memory_factory(

splitio/engine/impressions.py

@@ -173,12 +173,6 @@ def __init__(self, forwarder, mode=ImpressionsMode.OPTIMIZED, standalone=True, l
         :param listener: Optional impressions listener that will capture all seen impressions.
         :type listener: splitio.client.listener.ImpressionListenerWrapper
         """
-        if not isinstance(mode, ImpressionsMode):
-            try:
-                mode = ImpressionsMode(mode)
-            except ValueError:
-                mode = ImpressionsMode.OPTIMIZED
-
         self._forwarder = forwarder
         self._observer = Observer(_IMPRESSION_OBSERVER_CACHE_SIZE) if standalone else None
         self._counter = Counter() if standalone and mode == ImpressionsMode.OPTIMIZED else None

splitio/tasks/impressions_sync.py

@@ -9,7 +9,6 @@
 from splitio.api import APIException
 from splitio.tasks import BaseSynchronizationTask
 from splitio.tasks.util.asynctask import AsyncTask
-from splitio.engine.impressions import Manager as ImpressionsManager
 
 
 class ImpressionsSyncTask(BaseSynchronizationTask):
@@ -105,60 +104,34 @@ def flush(self):
 class ImpressionsCountSyncTask(BaseSynchronizationTask):
     """Impressions synchronization task uses an asynctask.AsyncTask to send impressions."""
 
-    def __init__(self, impressions_manager):
+    _PERIOD = 30 * 60 # 30 minutes
+
+    def __init__(self, impressions_api, impressions_manager):
         """
         Class constructor.
 
+        :param impressions_api: Impressions Api object to send data to the backend
+        :type impressions_api: splitio.api.impressions.ImpressionsAPI
+
         :param impressions_manager: Impressions manager instance
-        :type impressions_manager: ImpressionsManager
+        :type impressions_manager: splitio.engine.impressions.Manager
         """
         self._logger = logging.getLogger(self.__class__.__name__)
+        self._impressions_api = impressions_api
         self._impressions_manager = impressions_manager
-        self._task = AsyncTask(self._send_impressions, self._period, on_stop=self._send_impressions)
+        self._task = AsyncTask(self._send_counters, self._PERIOD, on_stop=self._send_counters)
 
-    def _get_failed(self):
-        """Return up to <BULK_SIZE> impressions stored in the failed impressions queue."""
-        imps = []
-        count = 0
-        while count < self._bulk_size:
-            try:
-                imps.append(self._failed.get(False))
-                count += 1
-            except queue.Empty:
-                # If no more items in queue, break the loop
-                break
-        return imps
-
-    def _add_to_failed_queue(self, imps):
-        """
-        Add impressions that were about to be sent to a secondary queue for failed sends.
-
-        :param imps: List of impressions that failed to be pushed.
-        :type imps: list
-        """
-        for impression in imps:
-            self._failed.put(impression, False)
-
-    def _send_impressions(self):
+    def _send_counters(self):
         """Send impressions from both the failed and new queues."""
-        to_send = self._get_failed()
-        if len(to_send) < self._bulk_size:
-            # If the amount of previously failed items is less than the bulk
-            # size, try to complete with new impressions from storage
-            to_send.extend(self._storage.pop_many(self._bulk_size - len(to_send)))
-
-        if not to_send:
-            return
-
+        to_send = self._impressions_manager.get_counts()
         try:
-            self._impressions_api.flush_impressions(to_send)
+            self._impressions_api.flush_counters(to_send)
         except APIException as exc:
             self._logger.error(
                 'Exception raised while reporting impressions: %s -- %d',
                 exc.message,
                 exc.status_code
             )
-            self._add_to_failed_queue(to_send)
 
     def start(self):
         """Start executing the impressions synchronization task."""

splitio/version.py

@@ -1 +1 @@
-__version__ = '8.2.1'
+__version__ = '8.3.0-rc1'

tests/client/test_config.py

@@ -0,0 +1,42 @@
+"""Configuration unit tests."""
+#pylint: disable=protected-access,no-self-use,line-too-long
+
+from splitio.client import config
+from splitio.engine.impressions import ImpressionsMode
+
+
+class ConfigSanitizationTests(object):
+    """Inmemory storage-based integration tests."""
+
+    def test_parse_operation_mode(self):
+        """Make sure operation mode is correctly captured."""
+        assert config._parse_operation_mode('some', {}) == 'inmemory-standalone'
+        assert config._parse_operation_mode('localhost', {}) == 'localhost-standalone'
+        assert config._parse_operation_mode('some', {'redisHost': 'x'}) == 'redis-consumer'
+        assert config._parse_operation_mode('some', {'uwsgiClient': True}) == 'uwsgi-consumer'
+
+    def test_sanitize_imp_mode(self):
+        """Test sanitization of impressions mode."""
+        mode, rate = config._sanitize_impressions_mode('OPTIMIZED', 1)
+        assert mode == ImpressionsMode.OPTIMIZED
+        assert rate == 60
+
+        mode, rate = config._sanitize_impressions_mode('DEBUG', 1)
+        assert mode == ImpressionsMode.DEBUG
+        assert rate == 1
+
+        mode, rate = config._sanitize_impressions_mode('ANYTHING', 200)
+        assert mode == ImpressionsMode.OPTIMIZED
+        assert rate == 200
+
+        mode, rate = config._sanitize_impressions_mode(43, -1)
+        assert mode == ImpressionsMode.OPTIMIZED
+        assert rate == 60
+
+        mode, rate = config._sanitize_impressions_mode('OPTIMIZED')
+        assert mode == ImpressionsMode.OPTIMIZED
+        assert rate == 300
+
+        mode, rate = config._sanitize_impressions_mode('DEBUG')
+        assert mode == ImpressionsMode.DEBUG
+        assert rate == 60