Removed counter and impression mode from ImpressionsManager and created strategy from factory instead

Author: chillaq

splitio/client/factory.py

@@ -15,7 +15,6 @@
 from splitio.engine.strategies.strategy_debug_mode import StrategyDebugMode
 from splitio.engine.strategies.strategy_optimized_mode import StrategyOptimizedMode
 
-
 # Storage
 from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
     InMemoryImpressionStorage, InMemoryEventStorage
@@ -317,11 +316,10 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'events': InMemoryEventStorage(cfg['eventsQueueSize']),
     }
 
-    imp_strategy = StrategyOptimizedMode(True) if cfg['ImpressionsMode'] == 'OPTIMIZED' else StrategyDebugMode(True)
+    imp_counter = Counter() if cfg['impressionsMode'] == 'OPTIMIZED' else None
+    imp_strategy = StrategyOptimizedMode(imp_counter) if cfg['impressionsMode'] == 'OPTIMIZED' else StrategyDebugMode()
 
     imp_manager = ImpressionsManager(
-        cfg['impressionsMode'],
-        True,
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
         imp_strategy)
 
@@ -331,8 +329,9 @@ 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_manager),
+        ImpressionsCountSynchronizer(apis['impressions'], imp_counter),
     )
+    imp_count_sync_task = ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters) if cfg['impressionsMode'] == 'OPTIMIZED' else None
 
     tasks = SplitTasks(
         SplitSynchronizationTask(
@@ -348,7 +347,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)
+        imp_count_sync_task
     )
 
     synchronizer = Synchronizer(synchronizers, tasks)
@@ -399,10 +398,15 @@ def _build_redis_factory(api_key, cfg):
         _LOGGER.warning("dataSampling cannot be less than %.2f, defaulting to minimum",
                         _MIN_DEFAULT_DATA_SAMPLING_ALLOWED)
         data_sampling = _MIN_DEFAULT_DATA_SAMPLING_ALLOWED
+
+    imp_strategy = StrategyOptimizedMode(Counter()) if cfg['impressionsMode'] == 'OPTIMIZED' else StrategyDebugMode()
+    imp_manager = ImpressionsManager(
+        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
+        imp_strategy)
+
     recorder = PipelinedRecorder(
         redis_adapter.pipeline,
-        ImpressionsManager(cfg['impressionsMode'], False,
-                           _wrap_impression_listener(cfg['impressionListener'], sdk_metadata)),
+        imp_manager,
         storages['events'],
         storages['impressions'],
         data_sampling,
@@ -414,7 +418,6 @@ def _build_redis_factory(api_key, cfg):
         recorder,
     )
 
-
 def _build_localhost_factory(cfg):
     """Build and return a localhost factory for testing/development purposes."""
     storages = {
@@ -441,8 +444,9 @@ def _build_localhost_factory(cfg):
     synchronizer = LocalhostSynchronizer(synchronizers, tasks)
     manager = Manager(ready_event, synchronizer, None, False, sdk_metadata)
     manager.start()
+
     recorder = StandardRecorder(
-        ImpressionsManager(cfg['impressionsMode'], True, None),
+        ImpressionsManager(cfg['impressionsMode'], StrategyDebugMode()),
         storages['events'],
         storages['impressions'],
     )

splitio/engine/impressions.py

@@ -4,7 +4,7 @@
 from splitio.client.listener import ImpressionListenerException
 from splitio.engine.strategies.strategy_debug_mode import StrategyDebugMode
 from splitio.engine.strategies.strategy_optimized_mode import StrategyOptimizedMode
-from splitio.engine.strategies import Observer, Counter
+#from splitio.engine.strategies import Observer, Counter
 
 class ImpressionsMode(Enum):
     """Impressions tracking mode."""
@@ -15,7 +15,7 @@ class ImpressionsMode(Enum):
 class Manager(object):  # pylint:disable=too-few-public-methods
     """Impression manager."""
 
-    def __init__(self, mode=ImpressionsMode.OPTIMIZED, standalone=True, listener=None, strategy=None):
+    def __init__(self, listener=None, strategy=None):
         """
         Construct a manger to track and forward impressions to the queue.
 
@@ -29,7 +29,6 @@ def __init__(self, mode=ImpressionsMode.OPTIMIZED, standalone=True, listener=Non
         :type listener: splitio.client.listener.ImpressionListenerWrapper
         """
 
-        self._counter = Counter() if standalone else None
         self._strategy = strategy
         self._listener = listener
 
@@ -42,19 +41,10 @@ def process_impressions(self, impressions):
         :param impressions: List of impression objects with attributes
         :type impressions: list[tuple[splitio.models.impression.Impression, dict]]
         """
-        for_log, for_listener = self._strategy.process_impressions(impressions, self._counter)
+        for_log, for_listener = self._strategy.process_impressions(impressions)
         self._send_impressions_to_listener(for_listener)
         return for_log
 
-    def get_counts(self):
-        """
-        Return counts of impressions per features.
-
-        :returns: A list of counter objects.
-        :rtype: list[Counter.CountPerFeature]
-        """
-        return self._counter.pop_all() if self._counter is not None else []
-
     def _send_impressions_to_listener(self, impressions):
         """
         Send impression result to custom listener.

splitio/engine/strategies/strategy_debug_mode.py

@@ -6,14 +6,14 @@
 class StrategyDebugMode(BaseStrategy):
     """Debug mode strategy."""
 
-    def __init__(self, standalone=True):
+    def __init__(self):
         """
         Construct a strategy instance for debug mode.
 
         """
-        self._observer = Observer(_IMPRESSION_OBSERVER_CACHE_SIZE) if standalone else None
+        self._observer = Observer(_IMPRESSION_OBSERVER_CACHE_SIZE)
 
-    def process_impressions(self, impressions, counter=None):
+    def process_impressions(self, impressions):
         """
         Process impressions.
 

splitio/engine/strategies/strategy_optimized_mode.py

@@ -7,14 +7,15 @@
 class StrategyOptimizedMode(BaseStrategy):
     """Optimized mode strategy."""
 
-    def __init__(self, standalone=True):
+    def __init__(self, counter=None):
         """
         Construct a strategy instance for optimized mode.
 
         """
-        self._observer = Observer(_IMPRESSION_OBSERVER_CACHE_SIZE) if standalone else None
+        self._observer = Observer(_IMPRESSION_OBSERVER_CACHE_SIZE)
+        self._counter = counter
 
-    def process_impressions(self, impressions, counter):
+    def process_impressions(self, impressions):
         """
         Process impressions.
 
@@ -27,6 +28,6 @@ def process_impressions(self, impressions, counter):
         :rtype: list[tuple[splitio.models.impression.Impression, dict]]
         """
         imps = [(self._observer.test_and_set(imp), attrs) for imp, attrs in impressions] if self._observer else impressions
-        counter.track([imp for imp, _ in imps])
+        self._counter.track([imp for imp, _ in imps])
         this_hour = truncate_time(util.utctime_ms())
         return [i for i, _ in imps if i.previous_time is None or i.previous_time < this_hour], imps

splitio/sync/impression.py

@@ -2,6 +2,7 @@
 import queue
 
 from splitio.api import APIException
+from splitio.engine.strategies import Counter
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -68,22 +69,22 @@ def synchronize_impressions(self):
 
 
 class ImpressionsCountSynchronizer(object):
-    def __init__(self, impressions_api, impressions_manager):
+    def __init__(self, impressions_api, impressions_counter):
         """
         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: splitio.engine.impressions.Manager
+        :type impressions_counter: splitio.engine.strategies
 
         """
         self._impressions_api = impressions_api
-        self._impressions_manager = impressions_manager
+        self._impressions_counter = impressions_counter
 
     def synchronize_counters(self):
         """Send impressions from both the failed and new queues."""
-        to_send = self._impressions_manager.get_counts()
+        to_send = self._impressions_counter.pop_all()
         if not to_send:
             return
 

splitio/sync/synchronizer.py

@@ -291,7 +291,8 @@ def start_periodic_data_recording(self):
         _LOGGER.debug('Starting periodic data recording')
         self._split_tasks.impressions_task.start()
         self._split_tasks.events_task.start()
-        self._split_tasks.impressions_count_task.start()
+        if self._split_tasks.impressions_count_task is not None:
+            self._split_tasks.impressions_count_task.start()
 
     def stop_periodic_data_recording(self, blocking):
         """
@@ -303,9 +304,11 @@ def stop_periodic_data_recording(self, blocking):
         _LOGGER.debug('Stopping periodic data recording')
         if blocking:
             events = []
-            for task in [self._split_tasks.impressions_task,
-                         self._split_tasks.events_task,
-                         self._split_tasks.impressions_count_task]:
+            tasks = [self._split_tasks.impressions_task,
+                        self._split_tasks.events_task]
+            if self._split_tasks.impressions_count_task is not None:
+                tasks.append(self._split_tasks.impressions_count_task)
+            for task in tasks:
                 stop_event = threading.Event()
                 task.stop(stop_event)
                 events.append(stop_event)
@@ -314,7 +317,8 @@ def stop_periodic_data_recording(self, blocking):
         else:
             self._split_tasks.impressions_task.stop()
             self._split_tasks.events_task.stop()
-            self._split_tasks.impressions_count_task.stop()
+            if self._split_tasks.impressions_count_task is not None:
+                self._split_tasks.impressions_count_task.stop()
 
     def kill_split(self, split_name, default_treatment, change_number):
         """

tests/integration/test_client_e2e.py

@@ -15,6 +15,9 @@
 from splitio.storage.adapters.redis import build, RedisAdapter
 from splitio.models import splits, segments
 from splitio.engine.impressions import Manager as ImpressionsManager, ImpressionsMode
+from splitio.engine.strategies.strategy_debug_mode import StrategyDebugMode
+from splitio.engine.strategies.strategy_optimized_mode import StrategyOptimizedMode
+from splitio.engine.strategies import Counter
 from splitio.recorder.recorder import StandardRecorder, PipelinedRecorder
 from splitio.client.config import DEFAULT_CONFIG
 
@@ -48,7 +51,8 @@ def setup_method(self):
             'impressions': InMemoryImpressionStorage(5000),
             'events': InMemoryEventStorage(5000),
         }
-        impmanager = ImpressionsManager(storages['impressions'].put, ImpressionsMode.DEBUG)
+#        impmanager = ImpressionsManager(storages['impressions'].put, StrategyDebugMode())
+        impmanager = ImpressionsManager(None, StrategyDebugMode()) # no listener
         recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'])
         self.factory = SplitFactory('some_api_key', storages, True, recorder)  # pylint:disable=attribute-defined-outside-init
 
@@ -296,7 +300,8 @@ def setup_method(self):
             'impressions': InMemoryImpressionStorage(5000),
             'events': InMemoryEventStorage(5000),
         }
-        impmanager = ImpressionsManager(ImpressionsMode.OPTIMIZED, True)
+#        impmanager = ImpressionsManager(ImpressionsMode.OPTIMIZED, True)
+        impmanager = ImpressionsManager(None, StrategyOptimizedMode(Counter()))
         recorder = StandardRecorder(impmanager, storages['events'], storages['impressions'])
         self.factory = SplitFactory('some_api_key', storages, True, recorder)  # pylint:disable=attribute-defined-outside-init
 
@@ -514,7 +519,7 @@ def setup_method(self):
             'impressions': RedisImpressionsStorage(redis_client, metadata),
             'events': RedisEventsStorage(redis_client, metadata),
         }
-        impmanager = ImpressionsManager(ImpressionsMode.DEBUG, False)
+        impmanager = ImpressionsManager(None, StrategyDebugMode())
         recorder = PipelinedRecorder(redis_client.pipeline, impmanager,
                                      storages['events'], storages['impressions'])
         self.factory = SplitFactory('some_api_key', storages, True, recorder)  # pylint:disable=attribute-defined-outside-init
@@ -791,7 +796,8 @@ def setup_method(self):
             'impressions': RedisImpressionsStorage(redis_client, metadata),
             'events': RedisEventsStorage(redis_client, metadata),
         }
-        impmanager = ImpressionsManager(storages['impressions'].put, ImpressionsMode.DEBUG)
+#        impmanager = ImpressionsManager(storages['impressions'].put, ImpressionsMode.DEBUG)
+        impmanager = ImpressionsManager(None, StrategyDebugMode())
         recorder = PipelinedRecorder(redis_client.pipeline, impmanager,
                                      storages['events'], storages['impressions'])
         self.factory = SplitFactory('some_api_key', storages, True, recorder)  # pylint:disable=attribute-defined-outside-init

tests/sync/test_impressions_count_synchronizer.py

@@ -16,7 +16,7 @@ class ImpressionsCountSynchronizerTests(object):
     """ImpressionsCount synchronizer test cases."""
 
     def test_synchronize_impressions_counts(self, mocker):
-        manager = mocker.Mock(spec=ImpressionsManager)
+        counter = mocker.Mock(spec=Counter)
 
         counters = [
             Counter.CountPerFeature('f1', 123, 2),
@@ -25,13 +25,13 @@ def test_synchronize_impressions_counts(self, mocker):
             Counter.CountPerFeature('f2', 456, 222)
         ]
 
-        manager.get_counts.return_value = counters
+        counter.pop_all.return_value = counters
         api = mocker.Mock(spec=ImpressionsAPI)
         api.flush_counters.return_value = HttpResponse(200, '')
-        impression_count_synchronizer = ImpressionsCountSynchronizer(api, manager)
+        impression_count_synchronizer = ImpressionsCountSynchronizer(api, counter)
         impression_count_synchronizer.synchronize_counters()
 
-        assert manager.get_counts.mock_calls[0] == mocker.call()
+        assert counter.pop_all.mock_calls[0] == mocker.call()
         assert api.flush_counters.mock_calls[0] == mocker.call(counters)
 
         assert len(api.flush_counters.mock_calls) == 1

tests/tasks/test_impressions_sync.py

@@ -51,7 +51,7 @@ class ImpressionsCountSyncTests(object):
 
     def test_normal_operation(self, mocker):
         """Test that the task works properly under normal circumstances."""
-        manager = mocker.Mock(spec=ImpressionsManager)
+        counter = mocker.Mock(spec=Counter)
 
         counters = [
             Counter.CountPerFeature('f1', 123, 2),
@@ -60,18 +60,18 @@ def test_normal_operation(self, mocker):
             Counter.CountPerFeature('f2', 456, 222)
         ]
 
-        manager.get_counts.return_value = counters
+        counter.pop_all.return_value = counters
         api = mocker.Mock(spec=ImpressionsAPI)
         api.flush_counters.return_value = HttpResponse(200, '')
         impressions_sync.ImpressionsCountSyncTask._PERIOD = 1
-        impression_synchronizer = ImpressionsCountSynchronizer(api, manager)
+        impression_synchronizer = ImpressionsCountSynchronizer(api, counter)
         task = impressions_sync.ImpressionsCountSyncTask(
             impression_synchronizer.synchronize_counters
         )
         task.start()
         time.sleep(2)
         assert task.is_running()
-        assert manager.get_counts.mock_calls[0] == mocker.call()
+        assert counter.pop_all.mock_calls[0] == mocker.call()
         assert api.flush_counters.mock_calls[0] == mocker.call(counters)
         stop_event = threading.Event()
         calls_now = len(api.flush_counters.mock_calls)