Implemented redis none mode

Author: chillaq

splitio/client/factory.py

@@ -15,7 +15,7 @@
 from splitio.engine.impressions import ImpressionsMode
 from splitio.engine.manager import Counter as ImpressionsCounter
 from splitio.engine.strategies import StrategyNoneMode, StrategyDebugMode, StrategyOptimizedMode
-from splitio.engine.adapters import InMemorySenderAdapter
+from splitio.engine.adapters import InMemorySenderAdapter, RedisSenderAdapter
 
 # Storage
 from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
@@ -43,7 +43,7 @@
 # Synchronizer
 from splitio.sync.synchronizer import SplitTasks, SplitSynchronizers, Synchronizer, \
     LocalhostSynchronizer
-from splitio.sync.manager import Manager
+from splitio.sync.manager import Manager, RedisManager
 from splitio.sync.split import SplitSynchronizer, LocalSplitSynchronizer
 from splitio.sync.segment import SegmentSynchronizer
 from splitio.sync.impression import ImpressionSynchronizer, ImpressionsCountSynchronizer
@@ -215,6 +215,7 @@ def destroy(self, destroyed_event=None):
             return
 
         try:
+            _LOGGER.info('Factory destroy called, stopping tasks.')
             if self._sync_manager is not None:
                 if destroyed_event is not None:
 
@@ -424,7 +425,23 @@ def _build_redis_factory(api_key, cfg):
                         _MIN_DEFAULT_DATA_SAMPLING_ALLOWED)
         data_sampling = _MIN_DEFAULT_DATA_SAMPLING_ALLOWED
 
-    imp_strategy = StrategyDebugMode() if cfg['impressionsMode'] == ImpressionsMode.DEBUG else StrategyOptimizedMode(ImpressionsCounter())
+    imp_counter = ImpressionsCounter() if cfg['impressionsMode'] != ImpressionsMode.DEBUG else None
+    unique_keys_synchronizer = None
+    clear_filter_sync = None
+    unique_keys_task = None
+    clear_filter_task = None
+    if cfg['impressionsMode'] == ImpressionsMode.NONE:
+        imp_strategy = StrategyNoneMode(imp_counter)
+        clear_filter_sync = ClearFilterSynchronizer(imp_strategy.get_unique_keys_tracker())
+        unique_keys_synchronizer = UniqueKeysSynchronizer(RedisSenderAdapter(redis_adapter), imp_strategy.get_unique_keys_tracker())
+        unique_keys_task = UniqueKeysSyncTask(unique_keys_synchronizer.send_all)
+        clear_filter_task = ClearFilterSyncTask(clear_filter_sync.clear_all)
+        imp_strategy.get_unique_keys_tracker().set_queue_full_hook(unique_keys_task.flush)
+    elif cfg['impressionsMode'] == ImpressionsMode.DEBUG:
+        imp_strategy = StrategyDebugMode()
+    else:
+        imp_strategy = StrategyOptimizedMode(imp_counter)
+
     imp_manager = ImpressionsManager(
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
         imp_strategy)
@@ -436,11 +453,17 @@ def _build_redis_factory(api_key, cfg):
         storages['impressions'],
         data_sampling,
     )
+    manager = RedisManager(unique_keys_task, clear_filter_task)
+    initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer")
+    initialization_thread.setDaemon(True)
+    initialization_thread.start()
+
     return SplitFactory(
         api_key,
         storages,
         cfg['labelsEnabled'],
         recorder,
+        manager,
     )
 
 

splitio/engine/adapters.py

@@ -1,6 +1,10 @@
 import abc
+import logging
 import json
 
+from splitio.storage.adapters.redis import RedisAdapterException
+
+_LOGGER = logging.getLogger(__name__)
 
 class ImpressionsSenderAdapter(object, metaclass=abc.ABCMeta):
     """Impressions Sender Adapter interface."""
@@ -32,18 +36,60 @@ def record_unique_keys(self, uniques):
         :param uniques: unique keys disctionary
         :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
         """
-        self._telemtry_http_client.record_unique_keys(self._uniques_formatter(uniques))
+        self._telemtry_http_client.record_unique_keys({'keys': self._uniques_formatter(uniques)})
+
+    def _uniques_formatter(self, uniques):
+        """
+        Format the unique keys dictionary array to a JSON body
+
+        :param uniques: unique keys disctionary
+        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+
+        :return: unique keys JSON array
+        :rtype: json
+        """
+        return [{'f': feature, 'ks': list(keys)} for feature, keys in uniques.items()]
+
+class RedisSenderAdapter(ImpressionsSenderAdapter):
+    """In Memory Impressions Sender Adapter class."""
+
+    MTK_QUEUE_KEY = 'SPLITIO.uniquekeys'
+    MTK_KEY_DEFAULT_TTL = 3600
+
+    def __init__(self, redis_client):
+        """
+        Initialize In memory sender adapter instance
+
+        :param telemtry_http_client: instance of telemetry http api
+        :type telemtry_http_client: splitio.api.telemetry.TelemetryAPI
+        """
+        self._redis_client = redis_client
+
+    def record_unique_keys(self, uniques):
+        """
+        post the unique keys to split back end.
+
+        :param uniques: unique keys disctionary
+        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+        """
+        bulk_mtks = self._uniques_formatter(uniques)
+        try:
+            self._redis_client.rpush(self.MTK_QUEUE_KEY, *bulk_mtks)
+            self._redis_client.expire(self.MTK_QUEUE_KEY, self.MTK_KEY_DEFAULT_TTL)
+            return True
+        except RedisAdapterException:
+            _LOGGER.error('Something went wrong when trying to add mtks to redis')
+            _LOGGER.error('Error: ', exc_info=True)
+            return False
 
     def _uniques_formatter(self, uniques):
         """
-        Format the unique keys dictionary to a JSON body
+        Format the unique keys dictionary array to a JSON body
 
         :param uniques: unique keys disctionary
         :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
 
-        :return: unique keys JSON
+        :return: unique keys JSON array
         :rtype: json
         """
-        return {
-            'keys':  [{'f': feature, 'ks': list(keys)} for feature, keys in uniques.items()]
-        }
+        return [json.dumps({'f': feature, 'ks': list(keys)}) for feature, keys in uniques.items()]

splitio/engine/unique_keys_tracker.py

@@ -59,6 +59,7 @@ def track(self, key, feature_name):
                 'Unique Keys queue is full, flushing the current queue now.'
             )
             if self._queue_full_hook is not None and callable(self._queue_full_hook):
+                _LOGGER.info('Calling hook.')
                 self._queue_full_hook()
         return True
 

splitio/sync/manager.py

@@ -1,6 +1,7 @@
 """Synchronization manager module."""
 import logging
 import time
+import threading
 from threading import Thread
 from queue import Queue
 from splitio.push.manager import PushManager, Status
@@ -127,3 +128,60 @@ def _streaming_feedback_handler(self):
                 self._synchronizer.start_periodic_fetching()
                 _LOGGER.info('non-recoverable error in streaming. switching to polling.')
                 return
+
+class RedisManager(object):  # pylint:disable=too-many-instance-attributes
+    """Manager Class."""
+
+    _CENTINEL_EVENT = object()
+
+    def __init__(self, unique_keys_task, clear_filter_task):  # pylint:disable=too-many-arguments
+        """
+        Construct Manager.
+
+        :param unique_keys_task: unique keys task instance
+        :type unique_keys_task: splitio.tasks.unique_keys_sync.UniqueKeysSyncTask
+
+        :param clear_filter_task: clear filter task instance
+        :type clear_filter_task: splitio.tasks.clear_filter_task.ClearFilterSynchronizer
+
+        """
+        self._unique_keys_task = unique_keys_task
+        self._clear_filter_task = clear_filter_task
+        self._ready_flag = True
+
+    def recreate(self):
+        """Not implemented"""
+        return
+
+    def start(self):
+        """Start the SDK synchronization tasks."""
+        try:
+            self._unique_keys_task.start()
+            self._clear_filter_task.start()
+
+        except (APIException, RuntimeError):
+            _LOGGER.error('Exception raised starting Split Manager')
+            _LOGGER.debug('Exception information: ', exc_info=True)
+            raise
+
+    def stop(self, blocking):
+        """
+        Stop manager logic.
+
+        :param blocking: flag to wait until tasks are stopped
+        :type blocking: bool
+        """
+        _LOGGER.info('Stopping manager tasks')
+        if blocking:
+            events = []
+            tasks = [self._unique_keys_task,
+                self._clear_filter_task]
+            for task in tasks:
+                stop_event = threading.Event()
+                task.stop(stop_event)
+                events.append(stop_event)
+            if all(event.wait() for event in events):
+                _LOGGER.debug('all tasks finished successfully.')
+        else:
+            self._unique_keys_task.stop()
+            self._clear_filter_task.stop()

splitio/sync/unique_keys.py

@@ -1,6 +1,3 @@
-from splitio.engine.filters import BloomFilter
-from splitio.engine.adapters import InMemorySenderAdapter
-
 _UNIQUE_KEYS_MAX_BULK_SIZE = 5000
 
 class UniqueKeysSynchronizer(object):