updated models and recorder

splitio · chillaq · Jan 17, 2025 · Dec 18, 2024 · Dec 18, 2024 · Dec 18, 2024
commit ec814ebff957fc25c3f1affdf480931d1238c372
@@ -16,6 +16,14 @@
     ]
 )
 
+ImpressionDecorated = namedtuple(
+    'ImpressionDecorated',
+    [
+        'Impression',
+        'track'
+    ]
+)
+
 # pre-python3.7 hack to make previous_time optional
 Impression.__new__.__defaults__ = (None,)
 

@@ -10,7 +10,7 @@
 
 SplitView = namedtuple(
     'SplitView',
-    ['name', 'traffic_type', 'killed', 'treatments', 'change_number', 'configs', 'default_treatment', 'sets']
+    ['name', 'traffic_type', 'killed', 'treatments', 'change_number', 'configs', 'default_treatment', 'sets', 'trackImpressions']
 )
 
 _DEFAULT_CONDITIONS_TEMPLATE =   {
@@ -73,7 +73,8 @@ def __init__(  # pylint: disable=too-many-arguments
             traffic_allocation=None,
             traffic_allocation_seed=None,
             configurations=None,
-            sets=None
+            sets=None,
+            trackImpressions=None
     ):
         """
         Class constructor.
@@ -96,6 +97,8 @@ def __init__(  # pylint: disable=too-many-arguments
         :type traffic_allocation_seed: int
         :pram sets: list of flag sets
         :type sets: list
+        :pram trackImpressions: track impressions flag
+        :type trackImpressions: boolean
         """
         self._name = name
         self._seed = seed
@@ -125,6 +128,7 @@ def __init__(  # pylint: disable=too-many-arguments
 
         self._configurations = configurations
         self._sets = set(sets) if sets is not None else set()
+        self._trackImpressions = trackImpressions if trackImpressions is not None else True
 
     @property
     def name(self):
@@ -186,6 +190,11 @@ def sets(self):
         """Return the flag sets of the split."""
         return self._sets
 
+    @property
+    def trackImpressions(self):
+        """Return trackImpressions of the split."""
+        return self._trackImpressions
+
     def get_configurations_for(self, treatment):
         """Return the mapping of treatments to configurations."""
         return self._configurations.get(treatment) if self._configurations else None
@@ -214,7 +223,8 @@ def to_json(self):
             'algo': self.algo.value,
             'conditions': [c.to_json() for c in self.conditions],
             'configurations': self._configurations,
-            'sets': list(self._sets)
+            'sets': list(self._sets),
+            'trackImpressions': self._trackImpressions
         }
 
     def to_split_view(self):
@@ -232,7 +242,8 @@ def to_split_view(self):
             self.change_number,
             self._configurations if self._configurations is not None else {},
             self._default_treatment,
-            list(self._sets) if self._sets is not None else []
+            list(self._sets) if self._sets is not None else [],
+            self._trackImpressions
         )
 
     def local_kill(self, default_treatment, change_number):
@@ -288,5 +299,6 @@ def from_raw(raw_split):
         traffic_allocation=raw_split.get('trafficAllocation'),
         traffic_allocation_seed=raw_split.get('trafficAllocationSeed'),
         configurations=raw_split.get('configurations'),
-        sets=set(raw_split.get('sets')) if raw_split.get('sets') is not None else []
+        sets=set(raw_split.get('sets')) if raw_split.get('sets') is not None else [],
+        trackImpressions=raw_split.get('trackImpressions') if raw_split.get('trackImpressions') is not None else True
     )
@@ -151,7 +151,7 @@ def __init__(self, impressions_manager, event_storage, impression_storage, telem
         self._telemetry_evaluation_producer = telemetry_evaluation_producer
         self._telemetry_runtime_producer = telemetry_runtime_producer
 
-    def record_treatment_stats(self, impressions, latency, operation, method_name):
+    def record_treatment_stats(self, impressions_decorated, latency, operation, method_name):
         """
         Record stats for treatment evaluation.
 
@@ -165,7 +165,7 @@ def record_treatment_stats(self, impressions, latency, operation, method_name):
         try:
             if method_name is not None:
                 self._telemetry_evaluation_producer.record_latency(operation, latency)
-            impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions)
+            impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions_decorated)
             if deduped > 0:
                 self._telemetry_runtime_producer.record_impression_stats(telemetry.CounterConstants.IMPRESSIONS_DEDUPED, deduped)
             self._impression_storage.put(impressions)
@@ -210,7 +210,7 @@ def __init__(self, impressions_manager, event_storage, impression_storage, telem
         self._telemetry_evaluation_producer = telemetry_evaluation_producer
         self._telemetry_runtime_producer = telemetry_runtime_producer
 
-    async def record_treatment_stats(self, impressions, latency, operation, method_name):
+    async def record_treatment_stats(self, impressions_decorated, latency, operation, method_name):
         """
         Record stats for treatment evaluation.
 
@@ -224,7 +224,7 @@ async def record_treatment_stats(self, impressions, latency, operation, method_n
         try:
             if method_name is not None:
                 await self._telemetry_evaluation_producer.record_latency(operation, latency)
-            impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions)
+            impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions_decorated)
             if deduped > 0:
                 await self._telemetry_runtime_producer.record_impression_stats(telemetry.CounterConstants.IMPRESSIONS_DEDUPED, deduped)
 
@@ -277,7 +277,7 @@ def __init__(self, pipe, impressions_manager, event_storage,
         self._data_sampling = data_sampling
         self._telemetry_redis_storage = telemetry_redis_storage
 
-    def record_treatment_stats(self, impressions, latency, operation, method_name):
+    def record_treatment_stats(self, impressions_decorated, latency, operation, method_name):
         """
         Record stats for treatment evaluation.
 
@@ -294,7 +294,7 @@ def record_treatment_stats(self, impressions, latency, operation, method_name):
                 if self._data_sampling < rnumber:
                     return
 
-            impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions)
+            impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions_decorated)
             if impressions:
                 pipe = self._make_pipe()
                 self._impression_storage.add_impressions_to_pipe(impressions, pipe)
@@ -367,7 +367,7 @@ def __init__(self, pipe, impressions_manager, event_storage,
         self._data_sampling = data_sampling
         self._telemetry_redis_storage = telemetry_redis_storage
 
-    async def record_treatment_stats(self, impressions, latency, operation, method_name):
+    async def record_treatment_stats(self, impressions_decorated, latency, operation, method_name):
         """
         Record stats for treatment evaluation.
 
@@ -384,7 +384,7 @@ async def record_treatment_stats(self, impressions, latency, operation, method_n
                 if self._data_sampling < rnumber:
                     return
 
-            impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions)
+            impressions, deduped, for_listener, for_counter, for_unique_keys_tracker = self._impressions_manager.process_impressions(impressions_decorated)
             if impressions:
                 pipe = self._make_pipe()
                 self._impression_storage.add_impressions_to_pipe(impressions, pipe)

@@ -60,7 +60,8 @@ class SplitTests(object):
         'configurations': {
             'on': '{"color": "blue", "size": 13}'
         },
-        'sets': ['set1', 'set2']
+        'sets': ['set1', 'set2'],
+        'trackImpressions': True
     }
 
     def test_from_raw(self):
@@ -81,6 +82,7 @@ def test_from_raw(self):
         assert parsed.get_configurations_for('on') == '{"color": "blue", "size": 13}'
         assert parsed._configurations == {'on': '{"color": "blue", "size": 13}'}
         assert parsed.sets == {'set1', 'set2'}
+        assert parsed.trackImpressions == True
 
     def test_get_segment_names(self, mocker):
         """Test fetching segment names."""
@@ -107,6 +109,7 @@ def test_to_json(self):
         assert as_json['algo'] == 2
         assert len(as_json['conditions']) == 2
         assert sorted(as_json['sets']) == ['set1', 'set2']
+        assert as_json['trackImpressions'] is True
 
     def test_to_split_view(self):
         """Test SplitView creation."""
@@ -118,6 +121,7 @@ def test_to_split_view(self):
         assert as_split_view.traffic_type == self.raw['trafficTypeName']
         assert set(as_split_view.treatments) == set(['on', 'off'])
         assert sorted(as_split_view.sets) == sorted(list(self.raw['sets']))
+        assert as_split_view.trackImpressions == self.raw['trackImpressions']
 
     def test_incorrect_matcher(self):
         """Test incorrect matcher in split model parsing."""