updated factory and client classes

Author: Bilal Al

splitio/client/client.py

@@ -3,7 +3,7 @@
 
 from splitio.engine.evaluator import Evaluator, CONTROL, EvaluationDataFactory, AsyncEvaluationDataFactory
 from splitio.engine.splitters import Splitter
-from splitio.models.impressions import Impression, Label
+from splitio.models.impressions import Impression, Label, ImpressionDecorated
 from splitio.models.events import Event, EventWrapper
 from splitio.models.telemetry import get_latency_bucket_index, MethodExceptionsAndLatencies
 from splitio.client import input_validator
@@ -22,7 +22,8 @@ class ClientBase(object):  # pylint: disable=too-many-instance-attributes
         'impression': {
             'label': Label.EXCEPTION,
             'change_number': None,
-        }
+        },
+        'track': True
     }
 
     _NON_READY_EVAL_RESULT = {
@@ -31,7 +32,8 @@ class ClientBase(object):  # pylint: disable=too-many-instance-attributes
         'impression': {
             'label': Label.NOT_READY,
             'change_number': None
-        }
+        },
+        'track': True
     }
 
     def __init__(self, factory, recorder, labels_enabled=True):
@@ -116,14 +118,15 @@ def _validate_treatments_input(key, features, attributes, method):
 
     def _build_impression(self, key, bucketing, feature, result):
         """Build an impression based on evaluation data & it's result."""
-        return Impression(
-                matching_key=key,
+        return ImpressionDecorated(
+                Impression(matching_key=key,
                 feature_name=feature,
                 treatment=result['treatment'],
                 label=result['impression']['label'] if self._labels_enabled else None,
                 change_number=result['impression']['change_number'],
                 bucketing_key=bucketing,
-                time=utctime_ms())
+                time=utctime_ms()),
+                track=result['track'])
 
     def _build_impressions(self, key, bucketing, results):
         """Build an impression based on evaluation data & it's result."""
@@ -228,7 +231,7 @@ def get_treatment(self, key, feature_flag_name, attributes=None):
             treatment, _ = self._get_treatment(MethodExceptionsAndLatencies.TREATMENT, key, feature_flag_name, attributes)
             return treatment
 
-        except:
+        except Exception as e:
             _LOGGER.error('get_treatment failed')
             return CONTROL
 
@@ -296,8 +299,8 @@ def _get_treatment(self, method, key, feature, attributes=None):
                 result = self._FAILED_EVAL_RESULT
 
         if result['impression']['label'] != Label.SPLIT_NOT_FOUND:
-            impression = self._build_impression(key, bucketing, feature, result)
-            self._record_stats([(impression, attributes)], start, method)
+            impression_decorated = self._build_impression(key, bucketing, feature, result)
+            self._record_stats([(impression_decorated, attributes)], start, method)
 
         return result['treatment'], result['configurations']
 
@@ -571,23 +574,23 @@ def _get_treatments(self, key, features, method, attributes=None):
                 self._telemetry_evaluation_producer.record_exception(method)
                 results = {n: self._FAILED_EVAL_RESULT for n in features}
 
-        imp_attrs = [
+        imp_decorated_attrs = [
             (i, attributes) for i in self._build_impressions(key, bucketing, results)
-            if i.label != Label.SPLIT_NOT_FOUND
+            if i.Impression.label != Label.SPLIT_NOT_FOUND
         ]
-        self._record_stats(imp_attrs, start, method)
+        self._record_stats(imp_decorated_attrs, start, method)
 
         return {
             feature: (results[feature]['treatment'], results[feature]['configurations'])
             for feature in results
         }
 
-    def _record_stats(self, impressions, start, operation):
+    def _record_stats(self, impressions_decorated, start, operation):
         """
         Record impressions.
 
-        :param impressions: Generated impressions
-        :type impressions: list[tuple[splitio.models.impression.Impression, dict]]
+        :param impressions_decorated: Generated impressions
+        :type impressions_decorated: list[tuple[splitio.models.impression.ImpressionDecorated, dict]]
 
         :param start: timestamp when get_treatment or get_treatments was called
         :type start: int
@@ -596,7 +599,7 @@ def _record_stats(self, impressions, start, operation):
         :type operation: str
         """
         end = get_current_epoch_time_ms()
-        self._recorder.record_treatment_stats(impressions, get_latency_bucket_index(end - start),
+        self._recorder.record_treatment_stats(impressions_decorated, get_latency_bucket_index(end - start),
                                               operation, 'get_' + operation.value)
 
     def track(self, key, traffic_type, event_type, value=None, properties=None):
@@ -695,7 +698,7 @@ async def get_treatment(self, key, feature_flag_name, attributes=None):
             treatment, _ = await self._get_treatment(MethodExceptionsAndLatencies.TREATMENT, key, feature_flag_name, attributes)
             return treatment
 
-        except:
+        except Exception as e:
             _LOGGER.error('get_treatment failed')
             return CONTROL
 
@@ -763,8 +766,8 @@ async def _get_treatment(self, method, key, feature, attributes=None):
                 result = self._FAILED_EVAL_RESULT
 
         if result['impression']['label'] != Label.SPLIT_NOT_FOUND:
-            impression = self._build_impression(key, bucketing, feature, result)
-            await self._record_stats([(impression, attributes)], start, method)
+            impression_decorated = self._build_impression(key, bucketing, feature, result)
+            await self._record_stats([(impression_decorated, attributes)], start, method)
         return result['treatment'], result['configurations']
 
     async def get_treatments(self, key, feature_flag_names, attributes=None):
@@ -960,23 +963,23 @@ async def _get_treatments(self, key, features, method, attributes=None):
                 await self._telemetry_evaluation_producer.record_exception(method)
                 results = {n: self._FAILED_EVAL_RESULT for n in features}
 
-        imp_attrs = [
+        imp_decorated_attrs = [
             (i, attributes) for i in self._build_impressions(key, bucketing, results)
-            if i.label != Label.SPLIT_NOT_FOUND
+            if i.Impression.label != Label.SPLIT_NOT_FOUND
         ]
-        await self._record_stats(imp_attrs, start, method)
+        await self._record_stats(imp_decorated_attrs, start, method)
 
         return {
             feature: (res['treatment'], res['configurations'])
             for feature, res in results.items()
         }
 
-    async def _record_stats(self, impressions, start, operation):
+    async def _record_stats(self, impressions_decorated, start, operation):
         """
         Record impressions for async calls
 
-        :param impressions: Generated impressions
-        :type impressions: list[tuple[splitio.models.impression.Impression, dict]]
+        :param impressions_decorated: Generated impressions decorated
+        :type impressions_decorated: list[tuple[splitio.models.impression.Impression, dict]]
 
         :param start: timestamp when get_treatment or get_treatments was called
         :type start: int
@@ -985,7 +988,7 @@ async def _record_stats(self, impressions, start, operation):
         :type operation: str
         """
         end = get_current_epoch_time_ms()
-        await self._recorder.record_treatment_stats(impressions, get_latency_bucket_index(end - start),
+        await self._recorder.record_treatment_stats(impressions_decorated, get_latency_bucket_index(end - start),
                                               operation, 'get_' + operation.value)
 
     async def track(self, key, traffic_type, event_type, value=None, properties=None):

splitio/client/factory.py

@@ -13,7 +13,7 @@
 from splitio.client.listener import ImpressionListenerWrapper, ImpressionListenerWrapperAsync
 from splitio.engine.impressions.impressions import Manager as ImpressionsManager
 from splitio.engine.impressions import set_classes, set_classes_async
-from splitio.engine.impressions.strategies import StrategyDebugMode
+from splitio.engine.impressions.strategies import StrategyDebugMode, StrategyNoneMode
 from splitio.engine.telemetry import TelemetryStorageProducer, TelemetryStorageConsumer, \
     TelemetryStorageProducerAsync, TelemetryStorageConsumerAsync
 from splitio.engine.impressions.manager import Counter as ImpressionsCounter
@@ -553,10 +553,10 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     unique_keys_tracker = UniqueKeysTracker(_UNIQUE_KEYS_CACHE_SIZE)
     unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
     clear_filter_task, impressions_count_sync, impressions_count_task, \
-    imp_strategy = set_classes('MEMORY', cfg['impressionsMode'], apis, imp_counter, unique_keys_tracker)
+    imp_strategy, none_strategy = set_classes('MEMORY', cfg['impressionsMode'], apis, imp_counter, unique_keys_tracker)
 
     imp_manager = ImpressionsManager(
-        imp_strategy, telemetry_runtime_producer)
+        imp_strategy, none_strategy, telemetry_runtime_producer)
 
     synchronizers = SplitSynchronizers(
         SplitSynchronizer(apis['splits'], storages['splits']),
@@ -681,10 +681,10 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
     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, \
-    imp_strategy = set_classes_async('MEMORY', cfg['impressionsMode'], apis, imp_counter, unique_keys_tracker)
+    imp_strategy, none_strategy = set_classes_async('MEMORY', cfg['impressionsMode'], apis, imp_counter, unique_keys_tracker)
 
     imp_manager = ImpressionsManager(
-        imp_strategy, telemetry_runtime_producer)
+        imp_strategy, none_strategy, telemetry_runtime_producer)
 
     synchronizers = SplitSynchronizers(
         SplitSynchronizerAsync(apis['splits'], storages['splits']),
@@ -775,10 +775,10 @@ def _build_redis_factory(api_key, cfg):
     unique_keys_tracker = UniqueKeysTracker(_UNIQUE_KEYS_CACHE_SIZE)
     unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
     clear_filter_task, impressions_count_sync, impressions_count_task, \
-    imp_strategy = set_classes('REDIS', cfg['impressionsMode'], redis_adapter, imp_counter, unique_keys_tracker)
+    imp_strategy, none_strategy = set_classes('REDIS', cfg['impressionsMode'], redis_adapter, imp_counter, unique_keys_tracker)
 
     imp_manager = ImpressionsManager(
-        imp_strategy,
+        imp_strategy, none_strategy,
         telemetry_runtime_producer)
 
     synchronizers = SplitSynchronizers(None, None, None, None,
@@ -858,10 +858,10 @@ async def _build_redis_factory_async(api_key, cfg):
     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, \
-    imp_strategy = set_classes_async('REDIS', cfg['impressionsMode'], redis_adapter, imp_counter, unique_keys_tracker)
+    imp_strategy, none_strategy = set_classes_async('REDIS', cfg['impressionsMode'], redis_adapter, imp_counter, unique_keys_tracker)
 
     imp_manager = ImpressionsManager(
-        imp_strategy,
+        imp_strategy, none_strategy,
         telemetry_runtime_producer)
 
     synchronizers = SplitSynchronizers(None, None, None, None,
@@ -936,10 +936,10 @@ def _build_pluggable_factory(api_key, cfg):
     unique_keys_tracker = UniqueKeysTracker(_UNIQUE_KEYS_CACHE_SIZE)
     unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
     clear_filter_task, impressions_count_sync, impressions_count_task, \
-    imp_strategy = set_classes('PLUGGABLE', cfg['impressionsMode'], pluggable_adapter, imp_counter, unique_keys_tracker, storage_prefix)
+    imp_strategy, none_strategy = set_classes('PLUGGABLE', cfg['impressionsMode'], pluggable_adapter, imp_counter, unique_keys_tracker, storage_prefix)
 
     imp_manager = ImpressionsManager(
-        imp_strategy,
+        imp_strategy, none_strategy,
         telemetry_runtime_producer)
 
     synchronizers = SplitSynchronizers(None, None, None, None,
@@ -1017,10 +1017,10 @@ async def _build_pluggable_factory_async(api_key, cfg):
     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, \
-    imp_strategy = set_classes_async('PLUGGABLE', cfg['impressionsMode'], pluggable_adapter, imp_counter, unique_keys_tracker, storage_prefix)
+    imp_strategy, none_strategy = set_classes_async('PLUGGABLE', cfg['impressionsMode'], pluggable_adapter, imp_counter, unique_keys_tracker, storage_prefix)
 
     imp_manager = ImpressionsManager(
-        imp_strategy,
+        imp_strategy, none_strategy,
         telemetry_runtime_producer)
 
     synchronizers = SplitSynchronizers(None, None, None, None,
@@ -1123,7 +1123,7 @@ def _build_localhost_factory(cfg):
         manager.start()
 
     recorder = StandardRecorder(
-        ImpressionsManager(StrategyDebugMode(), telemetry_runtime_producer),
+        ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer),
         storages['events'],
         storages['impressions'],
         telemetry_evaluation_producer,
@@ -1192,7 +1192,7 @@ async def _build_localhost_factory_async(cfg):
         await manager.start()
 
     recorder = StandardRecorderAsync(
-        ImpressionsManager(StrategyDebugMode(), telemetry_runtime_producer),
+        ImpressionsManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer),
         storages['events'],
         storages['impressions'],
         telemetry_evaluation_producer,

splitio/engine/impressions/__init__.py

@@ -53,24 +53,24 @@ def set_classes(storage_mode, impressions_mode, api_adapter, imp_counter, unique
         api_impressions_adapter = api_adapter['impressions']
         sender_adapter = InMemorySenderAdapter(api_telemetry_adapter)
 
+    none_strategy = StrategyNoneMode()
+    unique_keys_synchronizer = UniqueKeysSynchronizer(sender_adapter, unique_keys_tracker)
+    unique_keys_task = UniqueKeysSyncTask(unique_keys_synchronizer.send_all)
+    clear_filter_sync = ClearFilterSynchronizer(unique_keys_tracker)
+    impressions_count_sync = ImpressionsCountSynchronizer(api_impressions_adapter, imp_counter)
+    impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
+    clear_filter_task = ClearFilterSyncTask(clear_filter_sync.clear_all)
+    unique_keys_tracker.set_queue_full_hook(unique_keys_task.flush)
+
     if impressions_mode == ImpressionsMode.NONE:
         imp_strategy = StrategyNoneMode()
-        unique_keys_synchronizer = UniqueKeysSynchronizer(sender_adapter, unique_keys_tracker)
-        unique_keys_task = UniqueKeysSyncTask(unique_keys_synchronizer.send_all)
-        clear_filter_sync = ClearFilterSynchronizer(unique_keys_tracker)
-        impressions_count_sync = ImpressionsCountSynchronizer(api_impressions_adapter, imp_counter)
-        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
-        clear_filter_task = ClearFilterSyncTask(clear_filter_sync.clear_all)
-        unique_keys_tracker.set_queue_full_hook(unique_keys_task.flush)
     elif impressions_mode == ImpressionsMode.DEBUG:
         imp_strategy = StrategyDebugMode()
     else:
         imp_strategy = StrategyOptimizedMode()
-        impressions_count_sync = ImpressionsCountSynchronizer(api_impressions_adapter, imp_counter)
-        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
 
     return unique_keys_synchronizer, clear_filter_sync, unique_keys_task, clear_filter_task, \
-            impressions_count_sync, impressions_count_task, imp_strategy
+            impressions_count_sync, impressions_count_task, imp_strategy, none_strategy
 
 def set_classes_async(storage_mode, impressions_mode, api_adapter, imp_counter, unique_keys_tracker, prefix=None):
     """
@@ -118,21 +118,20 @@ def set_classes_async(storage_mode, impressions_mode, api_adapter, imp_counter,
         api_impressions_adapter = api_adapter['impressions']
         sender_adapter = InMemorySenderAdapterAsync(api_telemetry_adapter)
 
+    unique_keys_synchronizer = UniqueKeysSynchronizerAsync(sender_adapter, unique_keys_tracker)
+    unique_keys_task = UniqueKeysSyncTaskAsync(unique_keys_synchronizer.send_all)
+    clear_filter_sync = ClearFilterSynchronizerAsync(unique_keys_tracker)
+    impressions_count_sync = ImpressionsCountSynchronizerAsync(api_impressions_adapter, imp_counter)
+    impressions_count_task = ImpressionsCountSyncTaskAsync(impressions_count_sync.synchronize_counters)
+    clear_filter_task = ClearFilterSyncTaskAsync(clear_filter_sync.clear_all)
+    unique_keys_tracker.set_queue_full_hook(unique_keys_task.flush)
+
     if impressions_mode == ImpressionsMode.NONE:
         imp_strategy = StrategyNoneMode()
-        unique_keys_synchronizer = UniqueKeysSynchronizerAsync(sender_adapter, unique_keys_tracker)
-        unique_keys_task = UniqueKeysSyncTaskAsync(unique_keys_synchronizer.send_all)
-        clear_filter_sync = ClearFilterSynchronizerAsync(unique_keys_tracker)
-        impressions_count_sync = ImpressionsCountSynchronizerAsync(api_impressions_adapter, imp_counter)
-        impressions_count_task = ImpressionsCountSyncTaskAsync(impressions_count_sync.synchronize_counters)
-        clear_filter_task = ClearFilterSyncTaskAsync(clear_filter_sync.clear_all)
-        unique_keys_tracker.set_queue_full_hook(unique_keys_task.flush)
     elif impressions_mode == ImpressionsMode.DEBUG:
         imp_strategy = StrategyDebugMode()
     else:
         imp_strategy = StrategyOptimizedMode()
-        impressions_count_sync = ImpressionsCountSynchronizerAsync(api_impressions_adapter, imp_counter)
-        impressions_count_task = ImpressionsCountSyncTaskAsync(impressions_count_sync.synchronize_counters)
 
     return unique_keys_synchronizer, clear_filter_sync, unique_keys_task, clear_filter_task, \
             impressions_count_sync, impressions_count_task, imp_strategy