added pipelined op for redis

Author: Matias Melograno

splitio/client/client.py

@@ -12,6 +12,7 @@
 from splitio.models.telemetry import get_latency_bucket_index
 from splitio.client import input_validator
 from splitio.util import utctime_ms
+from splitio.recorder.recorder import StandardRecorder
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -25,7 +26,7 @@ class Client(object):  # pylint: disable=too-many-instance-attributes
     _METRIC_GET_TREATMENT_WITH_CONFIG = 'sdk.getTreatmentWithConfig'
     _METRIC_GET_TREATMENTS_WITH_CONFIG = 'sdk.getTreatmentsWithConfig'
 
-    def __init__(self, factory, impressions_manager, labels_enabled=True):
+    def __init__(self, factory, recorder, labels_enabled=True):
         """
         Construct a Client instance.
 
@@ -35,21 +36,18 @@ def __init__(self, factory, impressions_manager, labels_enabled=True):
         :param labels_enabled: Whether to store labels on impressions
         :type labels_enabled: bool
 
-        :param impressions_manager: impression manager instance
-        :type impressions_manager: splitio.engine.impressions.Manager
+        :param recorder: recorder instance
+        :type recorder: splitio.recorder.StatsRecorder
 
         :rtype: Client
         """
         self._factory = factory
         self._labels_enabled = labels_enabled
-        self._impressions_manager = impressions_manager
-
+        self._recorder = recorder
         self._splitter = Splitter()
         self._split_storage = factory._get_storage('splits')  # pylint: disable=protected-access
         self._segment_storage = factory._get_storage('segments')  # pylint: disable=protected-access
-        self._impressions_storage = factory._get_storage('impressions')  # pylint: disable=protected-access
         self._events_storage = factory._get_storage('events')  # pylint: disable=protected-access
-        self._telemetry_storage = factory._get_storage('telemetry')  # pylint: disable=protected-access
         self._evaluator = Evaluator(self._split_storage, self._segment_storage, self._splitter)
 
     def destroy(self):
@@ -341,13 +339,9 @@ def _record_stats(self, impressions, start, operation):
         :param operation: operation performed.
         :type operation: str
         """
-        try:
-            end = int(round(time.time() * 1000))
-            self._impressions_manager.track(impressions)
-            self._telemetry_storage.inc_latency(operation, get_latency_bucket_index(end - start))
-        except Exception:  # pylint: disable=broad-except
-            _LOGGER.error('Error recording impressions and metrics')
-            _LOGGER.debug('Error: ', exc_info=True)
+        end = int(round(time.time() * 1000))
+        self._recorder.record_treatment_stats(impressions, get_latency_bucket_index(end - start),
+                                              operation)
 
     def track(self, key, traffic_type, event_type, value=None, properties=None):
         """
@@ -395,7 +389,7 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
             timestamp=utctime_ms(),
             properties=properties,
         )
-        return self._events_storage.put([EventWrapper(
+        return self._recorder.record_track_stats([EventWrapper(
             event=event,
             size=size,
         )])

splitio/client/factory.py

@@ -35,7 +35,6 @@
 from splitio.api.telemetry import TelemetryAPI
 from splitio.api.auth import AuthAPI
 
-
 # Tasks
 from splitio.tasks.split_sync import SplitSynchronizationTask
 from splitio.tasks.segment_sync import SegmentSynchronizationTask
@@ -53,6 +52,9 @@
 from splitio.sync.event import EventSynchronizer
 from splitio.sync.telemetry import TelemetrySynchronizer
 
+# Recorder
+from splitio.recorder.recorder import StandardRecorder, PipelinedRecorder
+
 # Localhost stuff
 from splitio.client.localhost import LocalhostEventsStorage, LocalhostImpressionsStorage, \
     LocalhostTelemetryStorage
@@ -85,7 +87,7 @@ def __init__(  # pylint: disable=too-many-arguments
             apikey,
             storages,
             labels_enabled,
-            impressions_manager,
+            recorder,
             sync_manager=None,
             sdk_ready_flag=None,
     ):
@@ -102,16 +104,16 @@ def __init__(  # pylint: disable=too-many-arguments
         :type sync_manager: splitio.sync.manager.Manager
         :param sdk_ready_flag: Event to set when the sdk is ready.
         :type sdk_ready_flag: threading.Event
-        :param impression_manager: Impressions manager instance
-        :type impression_listener: ImpressionsManager
+        :param recorder: StatsRecorder instance
+        :type recorder: StatsRecorder
         """
         self._apikey = apikey
         self._storages = storages
         self._labels_enabled = labels_enabled
         self._sync_manager = sync_manager
         self._sdk_internal_ready_flag = sdk_ready_flag
         self._sdk_ready_flag = threading.Event()
-        self._impressions_manager = impressions_manager
+        self._recorder = recorder
 
         # If we have a ready flag, it means we have sync tasks that need to finish
         # before the SDK client becomes ready.
@@ -150,7 +152,7 @@ def client(self):
         This client is only a set of references to structures hold by the factory.
         Creating one a fast operation and safe to be used anywhere.
         """
-        return Client(self, self._impressions_manager, self._labels_enabled)
+        return Client(self, self._recorder, self._labels_enabled)
 
     def manager(self):
         """
@@ -280,7 +282,6 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     }
 
     imp_manager = ImpressionsManager(
-        storages['impressions'].put,
         cfg['impressionsMode'],
         True,
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata))
@@ -329,8 +330,14 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     storages['events'].set_queue_full_hook(tasks.events_task.flush)
     storages['impressions'].set_queue_full_hook(tasks.impressions_task.flush)
 
+    recorder = StandardRecorder(
+        imp_manager,
+        storages['telemetry'],
+        storages['events'],
+        storages['impressions'],
+    )
     return SplitFactory(api_key, storages, cfg['labelsEnabled'],
-                        imp_manager, manager, sdk_ready_flag)
+                        recorder, manager, sdk_ready_flag)
 
 
 def _build_redis_factory(api_key, cfg):
@@ -346,12 +353,19 @@ def _build_redis_factory(api_key, cfg):
         'events': RedisEventsStorage(redis_adapter, sdk_metadata),
         'telemetry': RedisTelemetryStorage(redis_adapter, sdk_metadata)
     }
+    recorder = PipelinedRecorder(
+        redis_adapter,
+        ImpressionsManager(cfg['impressionsMode'], False,
+                           _wrap_impression_listener(cfg['impressionListener'], sdk_metadata)),
+        storages['telemetry'],
+        storages['events'],
+        storages['impressions'],
+    )
     return SplitFactory(
         api_key,
         storages,
         cfg['labelsEnabled'],
-        ImpressionsManager(storages['impressions'].put, cfg['impressionsMode'], False,
-                           _wrap_impression_listener(cfg['impressionListener'], sdk_metadata))
+        recorder,
     )
 
 
@@ -366,12 +380,18 @@ def _build_uwsgi_factory(api_key, cfg):
         'events': UWSGIEventStorage(uwsgi_adapter),
         'telemetry': UWSGITelemetryStorage(uwsgi_adapter)
     }
+    recorder = StandardRecorder(
+        ImpressionsManager(cfg['impressionsMode'], True,
+                           _wrap_impression_listener(cfg['impressionListener'], sdk_metadata)),
+        storages['telemetry'],
+        storages['events'],
+        storages['impressions'],
+    )
     return SplitFactory(
         api_key,
         storages,
         cfg['labelsEnabled'],
-        ImpressionsManager(storages['impressions'].put, cfg['impressionsMode'], True,
-                           _wrap_impression_listener(cfg['impressionListener'], sdk_metadata))
+        recorder,
     )
 
 
@@ -401,12 +421,17 @@ def _build_localhost_factory(cfg):
     synchronizer = LocalhostSynchronizer(synchronizers, tasks)
     manager = Manager(ready_event, synchronizer, None, False)
     manager.start()
-
+    recorder = StandardRecorder(
+        ImpressionsManager(cfg['impressionsMode'], True, None),
+        storages['telemetry'],
+        storages['events'],
+        storages['impressions'],
+    )
     return SplitFactory(
         'localhost',
         storages,
         False,
-        ImpressionsManager(storages['impressions'].put, cfg['impressionsMode'], True, None),
+        recorder,
         manager,
         ready_event
     )

splitio/engine/impressions.py

@@ -157,13 +157,10 @@ def pop_all(self):
 class Manager(object):  # pylint:disable=too-few-public-methods
     """Impression manager."""
 
-    def __init__(self, forwarder, mode=ImpressionsMode.OPTIMIZED, standalone=True, listener=None):
+    def __init__(self, mode=ImpressionsMode.OPTIMIZED, standalone=True, listener=None):
         """
         Construct a manger to track and forward impressions to the queue.
 
-        :param forwarder: function accepting a list of impressions to be added to the queue.
-        :type forwarder: callable[list[splitio.models.impressions.Impression]]
-
         :param mode: Impressions capturing mode.
         :type mode: ImpressionsMode
 
@@ -173,31 +170,30 @@ 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
         """
-        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
         self._listener = listener
 
-    def track(self, impressions):
+    def process_impressions(self, impressions):
         """
-        Track impressions.
+        Process impressions.
 
         Impressions are analyzed to see if they've been seen before and counted.
 
         :param impressions: List of impression objects with attributes
         :type impressions: 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
+        imps = [(self._observer.test_and_set(imp), attrs) for imp, attrs in impressions] \
+            if self._observer else impressions
 
         if self._counter:
             self._counter.track([imp for imp, _ in imps])
 
         self._send_impressions_to_listener(imps)
 
         this_hour = truncate_time(util.utctime_ms())
-        self._forwarder([imp for imp, _ in imps] if self._counter is None
-                        else [i for i, _ in imps if i.previous_time is None or i.previous_time < this_hour])
+        return [imp for imp, _ in imps] if self._counter is None \
+            else [i for i, _ in imps if i.previous_time is None or i.previous_time < this_hour]
 
     def get_counts(self):
         """