Merge pull request #535 from splitio/async-pr

Refactored Recorder classes

Author: chillaq

splitio/client/factory.py

@@ -16,7 +16,7 @@
 from splitio.engine.impressions.strategies import StrategyDebugMode
 from splitio.engine.telemetry import TelemetryStorageProducer, TelemetryStorageConsumer, \
     TelemetryStorageProducerAsync, TelemetryStorageConsumerAsync
-from splitio.engine.impressions.manager import Counter as ImpressionsCounter, CounterAsync as ImpressionsCounterAsync
+from splitio.engine.impressions.manager import Counter as ImpressionsCounter
 from splitio.engine.impressions.unique_keys_tracker import UniqueKeysTracker, UniqueKeysTrackerAsync
 
 # Storage
@@ -663,7 +663,7 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
 
     telemetry_submitter = InMemoryTelemetrySubmitterAsync(telemetry_consumer, storages['splits'], storages['segments'], apis['telemetry'])
 
-    imp_counter = ImpressionsCounterAsync()
+    imp_counter = ImpressionsCounter()
     unique_keys_tracker = UniqueKeysTrackerAsync(_UNIQUE_KEYS_CACHE_SIZE)
     unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
     clear_filter_task, impressions_count_sync, impressions_count_task, \
@@ -840,7 +840,7 @@ async def _build_redis_factory_async(api_key, cfg):
                         _MIN_DEFAULT_DATA_SAMPLING_ALLOWED)
         data_sampling = _MIN_DEFAULT_DATA_SAMPLING_ALLOWED
 
-    imp_counter = ImpressionsCounterAsync()
+    imp_counter = ImpressionsCounter()
     unique_keys_tracker = UniqueKeysTrackerAsync(_UNIQUE_KEYS_CACHE_SIZE)
     unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
     clear_filter_task, impressions_count_sync, impressions_count_task, \
@@ -999,7 +999,7 @@ async def _build_pluggable_factory_async(api_key, cfg):
     # Using same class as redis
     telemetry_submitter = RedisTelemetrySubmitterAsync(storages['telemetry'])
 
-    imp_counter = ImpressionsCounterAsync()
+    imp_counter = ImpressionsCounter()
     unique_keys_tracker = UniqueKeysTrackerAsync(_UNIQUE_KEYS_CACHE_SIZE)
     unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
     clear_filter_task, impressions_count_sync, impressions_count_task, \

splitio/engine/impressions/__init__.py

@@ -18,7 +18,7 @@ def set_classes(storage_mode, impressions_mode, api_adapter, imp_counter, unique
     :param api_adapter: api adapter instance(s)
     :type impressions_mode: dict or splitio.storage.adapters.redis.RedisAdapter/splitio.storage.adapters.redis.RedisAdapterAsync
     :param imp_counter: Impressions Counter instance
-    :type imp_counter: splitio.engine.impressions.Counter/splitio.engine.impressions.CounterAsync
+    :type imp_counter: splitio.engine.impressions.Counter/splitio.engine.impressions.Counter
     :param unique_keys_tracker: Unique Keys Tracker instance
     :type unique_keys_tracker: splitio.engine.unique_keys_tracker.UniqueKeysTracker/splitio.engine.unique_keys_tracker.UniqueKeysTrackerAsync
     :param prefix: Prefix used for redis or pluggable adapters
@@ -83,7 +83,7 @@ def set_classes_async(storage_mode, impressions_mode, api_adapter, imp_counter,
     :param api_adapter: api adapter instance(s)
     :type impressions_mode: dict or splitio.storage.adapters.redis.RedisAdapter/splitio.storage.adapters.redis.RedisAdapterAsync
     :param imp_counter: Impressions Counter instance
-    :type imp_counter: splitio.engine.impressions.Counter/splitio.engine.impressions.CounterAsync
+    :type imp_counter: splitio.engine.impressions.Counter/splitio.engine.impressions.Counter
     :param unique_keys_tracker: Unique Keys Tracker instance
     :type unique_keys_tracker: splitio.engine.unique_keys_tracker.UniqueKeysTracker/splitio.engine.unique_keys_tracker.UniqueKeysTrackerAsync
     :param prefix: Prefix used for redis or pluggable adapters

splitio/engine/impressions/manager.py

@@ -153,40 +153,3 @@ def pop_all(self):
 
         return [Counter.CountPerFeature(k.feature, k.timeframe, v)
                 for (k, v) in old.items()]
-
-class CounterAsync(object):
-    """Class that counts impressions per timeframe."""
-
-    def __init__(self):
-        """Class constructor."""
-        self._data = defaultdict(lambda: 0)
-        self._lock = asyncio.Lock()
-
-    async def track(self, impressions, inc=1):
-        """
-        Register N new impressions for a feature in a specific timeframe.
-
-        :param impressions: generated impressions
-        :type impressions: list[splitio.models.impressions.Impression]
-
-        :param inc: amount to increment (defaults to 1)
-        :type inc: int
-        """
-        keys = [Counter.CounterKey(i.feature_name, truncate_time(i.time)) for i in impressions]
-        async with self._lock:
-            for key in keys:
-                self._data[key] += inc
-
-    async def pop_all(self):
-        """
-        Clear and return all the counters currently stored.
-
-        :returns: List of count per feature/timeframe objects
-        :rtype: list[ImpressionCounter.CountPerFeature]
-        """
-        async with self._lock:
-            old = self._data
-            self._data = defaultdict(lambda: 0)
-
-        return [Counter.CountPerFeature(k.feature, k.timeframe, v)
-                for (k, v) in old.items()]

splitio/engine/telemetry.py

@@ -668,7 +668,7 @@ async def pop_formatted_stats(self):
         last_synchronization = await self.get_last_synchronization()
         http_errors = await self.pop_http_errors()
         http_latencies = await self.pop_http_latencies()
-
+        # TODO: if ufs value is too large, use gather to fetch events instead of serial style.
         return {
             'iQ': await self.get_impressions_stats(CounterConstants.IMPRESSIONS_QUEUED),
             'iDe': await self.get_impressions_stats(CounterConstants.IMPRESSIONS_DEDUPED),

splitio/recorder/recorder.py

@@ -7,13 +7,36 @@
 from splitio.client.listener import ImpressionListenerException
 from splitio.models.telemetry import MethodExceptionsAndLatencies
 from splitio.models import telemetry
+from splitio.optional.loaders import asyncio
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class StatsRecorder(object, metaclass=abc.ABCMeta):
     """StatsRecorder interface."""
 
+    def __init__(self, impressions_manager, event_storage, impression_storage, listener=None, unique_keys_tracker=None, imp_counter=None):
+        """
+        Class constructor.
+
+        :param impressions_manager: impression manager instance
+        :type impressions_manager: splitio.engine.impressions.Manager
+        :param event_storage: event storage instance
+        :type event_storage: splitio.storage.EventStorage
+        :param impression_storage: impression storage instance
+        :type impression_storage: splitio.storage.ImpressionStorage
+        :param unique_keys_tracker: Unique Keys Tracker instance
+        :type unique_keys_tracker: splitio.engine.unique_keys_tracker.UniqueKeysTracker
+        :param imp_counter: Impressions Counter instance
+        :type imp_counter: splitio.engine.impressions.Counter
+        """
+        self._impressions_manager = impressions_manager
+        self._event_sotrage = event_storage
+        self._impression_storage = impression_storage
+        self._listener = listener
+        self._unique_keys_tracker = unique_keys_tracker
+        self._imp_counter = imp_counter
+
     @abc.abstractmethod
     def record_treatment_stats(self, impressions, latency, operation):
         """
@@ -38,7 +61,27 @@ def record_track_stats(self, events):
         """
         pass
 
-    async def _send_impressions_to_listener_async(self, impressions):
+class StatsRecorderThreadingBase(StatsRecorder):
+    """StandardRecorder class."""
+
+    def __init__(self, impressions_manager, event_storage, impression_storage, listener=None, unique_keys_tracker=None, imp_counter=None):
+        """
+        Class constructor.
+
+        :param impressions_manager: impression manager instance
+        :type impressions_manager: splitio.engine.impressions.Manager
+        :param event_storage: event storage instance
+        :type event_storage: splitio.storage.EventStorage
+        :param impression_storage: impression storage instance
+        :type impression_storage: splitio.storage.ImpressionStorage
+        :param unique_keys_tracker: Unique Keys Tracker instance
+        :type unique_keys_tracker: splitio.engine.unique_keys_tracker.UniqueKeysTracker
+        :param imp_counter: Impressions Counter instance
+        :type imp_counter: splitio.engine.impressions.Counter
+        """
+        StatsRecorder.__init__(self, impressions_manager, event_storage, impression_storage, listener, unique_keys_tracker, imp_counter)
+
+    def _send_impressions_to_listener(self, impressions):
         """
         Send impression result to custom listener.
 
@@ -48,11 +91,31 @@ async def _send_impressions_to_listener_async(self, impressions):
         if self._listener is not None:
             try:
                 for impression, attributes in impressions:
-                    await self._listener.log_impression(impression, attributes)
+                    self._listener.log_impression(impression, attributes)
             except ImpressionListenerException:
                 pass
 
-    def _send_impressions_to_listener(self, impressions):
+class StatsRecorderAsyncBase(StatsRecorder):
+    """StandardRecorder class."""
+
+    def __init__(self, impressions_manager, event_storage, impression_storage, listener=None, unique_keys_tracker=None, imp_counter=None):
+        """
+        Class constructor.
+
+        :param impressions_manager: impression manager instance
+        :type impressions_manager: splitio.engine.impressions.Manager
+        :param event_storage: event storage instance
+        :type event_storage: splitio.storage.EventStorage
+        :param impression_storage: impression storage instance
+        :type impression_storage: splitio.storage.ImpressionStorage
+        :param unique_keys_tracker: Unique Keys Tracker instance
+        :type unique_keys_tracker: splitio.engine.unique_keys_tracker.UniqueKeysTracker
+        :param imp_counter: Impressions Counter instance
+        :type imp_counter: splitio.engine.impressions.Counter
+        """
+        StatsRecorder.__init__(self, impressions_manager, event_storage, impression_storage, listener, unique_keys_tracker, imp_counter)
+
+    async def _send_impressions_to_listener_async(self, impressions):
         """
         Send impression result to custom listener.
 
@@ -62,11 +125,11 @@ def _send_impressions_to_listener(self, impressions):
         if self._listener is not None:
             try:
                 for impression, attributes in impressions:
-                    self._listener.log_impression(impression, attributes)
+                    await self._listener.log_impression(impression, attributes)
             except ImpressionListenerException:
                 pass
 
-class StandardRecorder(StatsRecorder):
+class StandardRecorder(StatsRecorderThreadingBase):
     """StandardRecorder class."""
 
     def __init__(self, impressions_manager, event_storage, impression_storage, telemetry_evaluation_producer, telemetry_runtime_producer, listener=None, unique_keys_tracker=None, imp_counter=None):
@@ -84,14 +147,9 @@ def __init__(self, impressions_manager, event_storage, impression_storage, telem
         :param imp_counter: Impressions Counter instance
         :type imp_counter: splitio.engine.impressions.Counter
         """
-        self._impressions_manager = impressions_manager
-        self._event_sotrage = event_storage
-        self._impression_storage = impression_storage
+        StatsRecorderThreadingBase.__init__(self, impressions_manager, event_storage, impression_storage, listener, unique_keys_tracker, imp_counter)
         self._telemetry_evaluation_producer = telemetry_evaluation_producer
         self._telemetry_runtime_producer = telemetry_runtime_producer
-        self._listener = listener
-        self._unique_keys_tracker = unique_keys_tracker
-        self._imp_counter = imp_counter
 
     def record_treatment_stats(self, impressions, latency, operation, method_name):
         """
@@ -130,8 +188,7 @@ def record_track_stats(self, event, latency):
         self._telemetry_evaluation_producer.record_latency(MethodExceptionsAndLatencies.TRACK, latency)
         return self._event_sotrage.put(event)
 
-
-class StandardRecorderAsync(StatsRecorder):
+class StandardRecorderAsync(StatsRecorderAsyncBase):
     """StandardRecorder async class."""
 
     def __init__(self, impressions_manager, event_storage, impression_storage, telemetry_evaluation_producer, telemetry_runtime_producer, listener=None, unique_keys_tracker=None, imp_counter=None):
@@ -147,16 +204,11 @@ def __init__(self, impressions_manager, event_storage, impression_storage, telem
         :param unique_keys_tracker: Unique Keys Tracker instance
         :type unique_keys_tracker: splitio.engine.unique_keys_tracker.UniqueKeysTrackerAsync
         :param imp_counter: Impressions Counter instance
-        :type imp_counter: splitio.engine.impressions.CounterAsync
+        :type imp_counter: splitio.engine.impressions.Counter
         """
-        self._impressions_manager = impressions_manager
-        self._event_sotrage = event_storage
-        self._impression_storage = impression_storage
+        StatsRecorderAsyncBase.__init__(self, impressions_manager, event_storage, impression_storage, listener, unique_keys_tracker, imp_counter)
         self._telemetry_evaluation_producer = telemetry_evaluation_producer
         self._telemetry_runtime_producer = telemetry_runtime_producer
-        self._listener = listener
-        self._unique_keys_tracker = unique_keys_tracker
-        self._imp_counter = imp_counter
 
     async def record_treatment_stats(self, impressions, latency, operation, method_name):
         """
@@ -179,9 +231,10 @@ async def record_treatment_stats(self, impressions, latency, operation, method_n
             await self._impression_storage.put(impressions)
             await self._send_impressions_to_listener_async(for_listener)
             if len(for_counter) > 0:
-                await self._imp_counter.track(for_counter)
+                self._imp_counter.track(for_counter)
             if len(for_unique_keys_tracker) > 0:
-                [await self._unique_keys_tracker.track(item[0], item[1]) for item in for_unique_keys_tracker]
+                unique_keys_coros = [self._unique_keys_tracker.track(item[0], item[1]) for item in for_unique_keys_tracker]
+                await asyncio.gather(*unique_keys_coros)
         except Exception:  # pylint: disable=broad-except
             _LOGGER.error('Error recording impressions')
             _LOGGER.debug('Error: ', exc_info=True)
@@ -196,8 +249,7 @@ async def record_track_stats(self, event, latency):
         await self._telemetry_evaluation_producer.record_latency(MethodExceptionsAndLatencies.TRACK, latency)
         return await self._event_sotrage.put(event)
 
-
-class PipelinedRecorder(StatsRecorder):
+class PipelinedRecorder(StatsRecorderThreadingBase):
     """PipelinedRecorder class."""
 
     def __init__(self, pipe, impressions_manager, event_storage,
@@ -220,15 +272,10 @@ def __init__(self, pipe, impressions_manager, event_storage,
         :param imp_counter: Impressions Counter instance
         :type imp_counter: splitio.engine.impressions.Counter
         """
+        StatsRecorderThreadingBase.__init__(self, impressions_manager, event_storage, impression_storage, listener, unique_keys_tracker, imp_counter)
         self._make_pipe = pipe
-        self._impressions_manager = impressions_manager
-        self._event_sotrage = event_storage
-        self._impression_storage = impression_storage
         self._data_sampling = data_sampling
         self._telemetry_redis_storage = telemetry_redis_storage
-        self._listener = listener
-        self._unique_keys_tracker = unique_keys_tracker
-        self._imp_counter = imp_counter
 
     def record_treatment_stats(self, impressions, latency, operation, method_name):
         """
@@ -246,6 +293,7 @@ def record_treatment_stats(self, impressions, latency, operation, method_name):
                 rnumber = random.uniform(0, 1)
                 if self._data_sampling < rnumber:
                     return
+
             impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions)
             if impressions:
                 pipe = self._make_pipe()
@@ -291,7 +339,7 @@ def record_track_stats(self, event, latency):
             _LOGGER.debug('Error: ', exc_info=True)
             return False
 
-class PipelinedRecorderAsync(StatsRecorder):
+class PipelinedRecorderAsync(StatsRecorderAsyncBase):
     """PipelinedRecorder async class."""
 
     def __init__(self, pipe, impressions_manager, event_storage,
@@ -312,17 +360,12 @@ def __init__(self, pipe, impressions_manager, event_storage,
         :param unique_keys_tracker: Unique Keys Tracker instance
         :type unique_keys_tracker: splitio.engine.unique_keys_tracker.UniqueKeysTrackerAsync
         :param imp_counter: Impressions Counter instance
-        :type imp_counter: splitio.engine.impressions.CounterAsync
+        :type imp_counter: splitio.engine.impressions.Counter
         """
+        StatsRecorderAsyncBase.__init__(self, impressions_manager, event_storage, impression_storage, listener, unique_keys_tracker, imp_counter)
         self._make_pipe = pipe
-        self._impressions_manager = impressions_manager
-        self._event_sotrage = event_storage
-        self._impression_storage = impression_storage
         self._data_sampling = data_sampling
         self._telemetry_redis_storage = telemetry_redis_storage
-        self._listener = listener
-        self._unique_keys_tracker = unique_keys_tracker
-        self._imp_counter = imp_counter
 
     async def record_treatment_stats(self, impressions, latency, operation, method_name):
         """
@@ -340,6 +383,7 @@ async def record_treatment_stats(self, impressions, latency, operation, method_n
                 rnumber = random.uniform(0, 1)
                 if self._data_sampling < rnumber:
                     return
+
             impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions)
             if impressions:
                 pipe = self._make_pipe()
@@ -353,9 +397,10 @@ async def record_treatment_stats(self, impressions, latency, operation, method_n
                 await self._send_impressions_to_listener_async(for_listener)
 
             if len(for_counter) > 0:
-                await self._imp_counter.track(for_counter)
+                self._imp_counter.track(for_counter)
             if len(for_unique_keys_tracker) > 0:
-                [await self._unique_keys_tracker.track(item[0], item[1]) for item in for_unique_keys_tracker]
+                unique_keys_coros = [self._unique_keys_tracker.track(item[0], item[1]) for item in for_unique_keys_tracker]
+                await asyncio.gather(*unique_keys_coros)
         except Exception:  # pylint: disable=broad-except
             _LOGGER.error('Error recording impressions')
             _LOGGER.debug('Error: ', exc_info=True)