Merge pull request #271 from splitio/redis-none

Redis none

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, \
@@ -42,8 +42,8 @@
 
 # Synchronizer
 from splitio.sync.synchronizer import SplitTasks, SplitSynchronizers, Synchronizer, \
-    LocalhostSynchronizer
-from splitio.sync.manager import Manager
+    LocalhostSynchronizer, RedisSynchronizer
+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:
 
@@ -329,17 +330,24 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     clear_filter_sync = None
     unique_keys_task = None
     clear_filter_task = None
+    impressions_count_sync = None
+    impressions_count_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(InMemorySenderAdapter(apis['telemetry']), 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)
+        impressions_count_sync = ImpressionsCountSynchronizer(apis['impressions'], imp_counter)
+        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
     elif cfg['impressionsMode'] == ImpressionsMode.DEBUG:
         imp_strategy = StrategyDebugMode()
     else:
         imp_strategy = StrategyOptimizedMode(imp_counter)
+        impressions_count_sync = ImpressionsCountSynchronizer(apis['impressions'], imp_counter)
+        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
 
     imp_manager = ImpressionsManager(
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
@@ -351,7 +359,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         ImpressionSynchronizer(apis['impressions'], storages['impressions'],
                                cfg['impressionsBulkSize']),
         EventSynchronizer(apis['events'], storages['events'], cfg['eventsBulkSize']),
-        ImpressionsCountSynchronizer(apis['impressions'], imp_counter),
+        impressions_count_sync,
         unique_keys_synchronizer,
         clear_filter_sync
     )
@@ -370,7 +378,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
             cfg['impressionsRefreshRate'],
         ),
         EventsSyncTask(synchronizers.events_sync.synchronize_events, cfg['eventsPushRate']),
-        ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters),
+        impressions_count_task,
         unique_keys_task,
         clear_filter_task
     )
@@ -424,23 +432,68 @@ 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())
+    unique_keys_synchronizer = None
+    clear_filter_sync = None
+    unique_keys_task = None
+    clear_filter_task = None
+    impressions_count_sync = None
+    impressions_count_task = None
+    redis_sender_adapter = RedisSenderAdapter(redis_adapter)
+
+    if cfg['impressionsMode'] == ImpressionsMode.NONE:
+        imp_counter = ImpressionsCounter()
+        imp_strategy = StrategyNoneMode(imp_counter)
+        clear_filter_sync = ClearFilterSynchronizer(imp_strategy.get_unique_keys_tracker())
+        unique_keys_synchronizer = UniqueKeysSynchronizer(redis_sender_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)
+        impressions_count_sync = ImpressionsCountSynchronizer(redis_sender_adapter, imp_counter)
+        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
+    elif cfg['impressionsMode'] == ImpressionsMode.DEBUG:
+        imp_strategy = StrategyDebugMode()
+    else:
+        imp_counter = ImpressionsCounter()
+        imp_strategy = StrategyOptimizedMode(imp_counter)
+        impressions_count_sync = ImpressionsCountSynchronizer(redis_sender_adapter, imp_counter)
+        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
+
     imp_manager = ImpressionsManager(
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
         imp_strategy)
 
+    synchronizers = SplitSynchronizers(None, None, None, None,
+        impressions_count_sync,
+        unique_keys_synchronizer,
+        clear_filter_sync
+    )
+
+    tasks = SplitTasks(None, None, None, None,
+        impressions_count_task,
+        unique_keys_task,
+        clear_filter_task
+    )
+
+    synchronizer = RedisSynchronizer(synchronizers, tasks)
     recorder = PipelinedRecorder(
         redis_adapter.pipeline,
         imp_manager,
         storages['events'],
         storages['impressions'],
         data_sampling,
     )
+
+    manager = RedisManager(synchronizer)
+    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,111 @@ 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
+    IMP_COUNT_QUEUE_KEY = 'SPLITIO.impressions.count'
+    IMP_COUNT_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 redis.
+
+        :param uniques: unique keys disctionary
+        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+        """
+        bulk_mtks = self._uniques_formatter(uniques)
+        try:
+            inserted = self._redis_client.rpush(self.MTK_QUEUE_KEY, *bulk_mtks)
+            self._expire_keys(self.MTK_QUEUE_KEY, self.MTK_KEY_DEFAULT_TTL, inserted, len(bulk_mtks))
+            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 flush_counters(self, to_send):
+        """
+        post the impression counters to redis.
+
+        :param uniques: unique keys disctionary
+        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+        """
+        bulk_counts = self._build_counters(to_send)
+        try:
+            inserted = self._redis_client.rpush(self.IMP_COUNT_QUEUE_KEY, *bulk_counts)
+            self._expire_keys(self.IMP_COUNT_QUEUE_KEY, self.IMP_COUNT_KEY_DEFAULT_TTL, inserted, len(bulk_counts))
+            return True
+        except RedisAdapterException:
+            _LOGGER.error('Something went wrong when trying to add counters to redis')
+            _LOGGER.error('Error: ', exc_info=True)
+            return False
+
+    def _expire_keys(self, queue_key, key_default_ttl, total_keys, inserted):
+        """
+        Set expire
+
+        :param total_keys: length of keys.
+        :type total_keys: int
+        :param inserted: added keys.
+        :type inserted: int
+        """
+        if total_keys == inserted:
+            self._redis_client.expire(queue_key, key_default_ttl)
 
     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()]
+
+    def _build_counters(self, counters):
+        """
+        Build an impression bulk formatted as the API expects it.
+
+        :param counters: List of impression counters per feature.
+        :type counters: list[splitio.engine.impressions.Counter.CountPerFeature]
+
+        :return: dict with list of impression count dtos
+        :rtype: dict
+        """
+        return [json.dumps({
+            'pf': [
+                {
+                    'f': pf_count.feature,
+                    'm': pf_count.timeframe,
+                    'rc': pf_count.count
+                } for pf_count in counters
+            ]
+        })]

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,44 @@ 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."""
+
+    def __init__(self, synchronizer):  # 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._ready_flag = True
+        self._synchronizer = synchronizer
+
+    def recreate(self):
+        """Not implemented"""
+        return
+
+    def start(self):
+        """Start the SDK synchronization tasks."""
+        try:
+            self._synchronizer.start_periodic_data_recording()
+
+        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')
+        self._synchronizer.shutdown(blocking)