Implemented impression count for redis and refactoring of synchronizer class for redis.

chillaq · chillaq · commit 2818dc6871f7 · 2022-09-02T10:42:04.000-07:00
diff --git a/splitio/client/factory.py b/splitio/client/factory.py
@@ -42,7 +42,7 @@
 
 # Synchronizer
 from splitio.sync.synchronizer import SplitTasks, SplitSynchronizers, Synchronizer, \
-    LocalhostSynchronizer
+    LocalhostSynchronizer, RedisSynchronizer
 from splitio.sync.manager import Manager, RedisManager
 from splitio.sync.split import SplitSynchronizer, LocalSplitSynchronizer
 from splitio.sync.segment import SegmentSynchronizer
@@ -430,10 +430,11 @@ def _build_redis_factory(api_key, cfg):
     clear_filter_sync = None
     unique_keys_task = None
     clear_filter_task = None
+    redis_sender_adapter = RedisSenderAdapter(redis_adapter)
     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_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)
@@ -446,14 +447,28 @@ def _build_redis_factory(api_key, cfg):
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
         imp_strategy)
 
+    synchronizers = SplitSynchronizers(None, None, None, None,
+        ImpressionsCountSynchronizer(redis_sender_adapter, imp_counter),
+        unique_keys_synchronizer,
+        clear_filter_sync
+    )
+
+    tasks = SplitTasks(None, None, None, None,
+        ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters),
+        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(unique_keys_task, clear_filter_task)
+
+    manager = RedisManager(synchronizer)
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer")
     initialization_thread.setDaemon(True)
     initialization_thread.start()
diff --git a/splitio/engine/adapters.py b/splitio/engine/adapters.py
@@ -55,6 +55,8 @@ class RedisSenderAdapter(ImpressionsSenderAdapter):
 
     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):
         """
@@ -67,21 +69,50 @@ def __init__(self, redis_client):
 
     def record_unique_keys(self, uniques):
         """
-        post the unique keys to split back end.
+        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:
-            self._redis_client.rpush(self.MTK_QUEUE_KEY, *bulk_mtks)
-            self._redis_client.expire(self.MTK_QUEUE_KEY, self.MTK_KEY_DEFAULT_TTL)
+            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 array to a JSON body
@@ -93,3 +124,23 @@ def _uniques_formatter(self, uniques):
         :rtype: json
         """
         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
+            ]
+        })]
diff --git a/splitio/sync/manager.py b/splitio/sync/manager.py
@@ -134,7 +134,7 @@ class RedisManager(object):  # pylint:disable=too-many-instance-attributes
 
     _CENTINEL_EVENT = object()
 
-    def __init__(self, unique_keys_task, clear_filter_task):  # pylint:disable=too-many-arguments
+    def __init__(self, synchronizer):  # pylint:disable=too-many-arguments
         """
         Construct Manager.
 
@@ -145,9 +145,8 @@ def __init__(self, unique_keys_task, clear_filter_task):  # pylint:disable=too-m
         :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
+        self._synchronizer = synchronizer
 
     def recreate(self):
         """Not implemented"""
@@ -156,8 +155,7 @@ def recreate(self):
     def start(self):
         """Start the SDK synchronization tasks."""
         try:
-            self._unique_keys_task.start()
-            self._clear_filter_task.start()
+            self._synchronizer.start_periodic_data_recording()
 
         except (APIException, RuntimeError):
             _LOGGER.error('Exception raised starting Split Manager')
@@ -172,16 +170,4 @@ def stop(self, blocking):
         :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()
+        self._synchronizer.shutdown(blocking)
diff --git a/splitio/sync/synchronizer.py b/splitio/sync/synchronizer.py
@@ -377,6 +377,94 @@ def kill_split(self, split_name, default_treatment, change_number):
         self._split_synchronizers.split_sync.kill_split(split_name, default_treatment,
                                                         change_number)
 
+class RedisSynchronizer(BaseSynchronizer):
+    """Redis Synchronizer."""
+
+    def __init__(self, split_synchronizers, split_tasks):
+        """
+        Class constructor.
+
+        :param split_synchronizers: syncs for performing synchronization of segments and splits
+        :type split_synchronizers: splitio.sync.synchronizer.SplitSynchronizers
+        :param split_tasks: tasks for starting/stopping tasks
+        :type split_tasks: splitio.sync.synchronizer.SplitTasks
+        """
+        self._split_synchronizers = split_synchronizers
+        self._split_tasks = split_tasks
+
+    def sync_all(self):
+        """
+        Not implemented
+        """
+        pass
+
+    def shutdown(self, blocking):
+        """
+        Stop tasks.
+
+        :param blocking:flag to wait until tasks are stopped
+        :type blocking: bool
+        """
+        _LOGGER.debug('Shutting down tasks.')
+        self.stop_periodic_data_recording(blocking)
+
+    def start_periodic_data_recording(self):
+        """Start recorders."""
+        _LOGGER.debug('Starting periodic data recording')
+        self._split_tasks.impressions_count_task.start()
+        if self._split_tasks.unique_keys_task is not None:
+            self._split_tasks.unique_keys_task.start()
+        if self._split_tasks.clear_filter_task is not None:
+            self._split_tasks.clear_filter_task.start()
+
+    def stop_periodic_data_recording(self, blocking):
+        """
+        Stop recorders.
+
+        :param blocking: flag to wait until tasks are stopped
+        :type blocking: bool
+        """
+        _LOGGER.debug('Stopping periodic data recording')
+        if blocking:
+            events = []
+            tasks = [self._split_tasks.impressions_count_task]
+            if self._split_tasks.unique_keys_task is not None:
+                tasks.append(self._split_tasks.unique_keys_task)
+            if self._split_tasks.clear_filter_task is not None:
+                tasks.append(self._split_tasks.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._split_tasks.impressions_count_task.stop()
+            if self._split_tasks.unique_keys_task is not None:
+                self._split_tasks.unique_keys_task.stop()
+            if self._split_tasks.clear_filter_task is not None:
+                self._split_tasks.clear_filter_task.stop()
+
+    def kill_split(self, split_name, default_treatment, change_number):
+        """Kill a split locally."""
+        raise NotImplementedError()
+
+    def synchronize_splits(self, till):
+        """Synchronize all splits."""
+        raise NotImplementedError()
+
+    def synchronize_segment(self, segment_name, till):
+        """Synchronize particular segment."""
+        raise NotImplementedError()
+
+    def start_periodic_fetching(self):
+        """Start fetchers for splits and segments."""
+        raise NotImplementedError()
+
+    def stop_periodic_fetching(self):
+        """Stop fetchers for splits and segments."""
+        raise NotImplementedError()
+
 class LocalhostSynchronizer(BaseSynchronizer):
     """LocalhostSynchronizer."""
 
diff --git a/splitio/tasks/unique_keys_sync.py b/splitio/tasks/unique_keys_sync.py
@@ -73,13 +73,11 @@ def __init__(self, clear_filter, period = None):
     def start(self):
         """Start executing the unique keys synchronization task."""
 
-        _LOGGER.debug('Starting periodic clear filter')
         self._task.start()
 
     def stop(self, event=None):
         """Stop executing the unique keys synchronization task."""
 
-        _LOGGER.debug('Stopping periodic clear filter')
         self._task.stop(event)
 
     def is_running(self):
