Added telemetry to iff update

Author: chillaq

splitio/engine/telemetry.py

@@ -139,6 +139,10 @@ def record_session_length(self, session):
         """Record session length."""
         self._telemetry_storage.record_session_length(session)
 
+    def record_update_from_sse(self):
+        """Record session length."""
+        self._telemetry_storage.record_update_from_sse()
+
 class TelemetryStorageConsumer(object):
     """Telemetry storage consumer class."""
 
@@ -271,6 +275,10 @@ def pop_streaming_events(self):
         """Get and reset streaming events."""
         return self._telemetry_storage.pop_streaming_events()
 
+    def pop_update_from_sse(self):
+        """Get and reset update from sse."""
+        return self._telemetry_storage.pop_update_from_sse()
+
     def get_session_length(self):
         """Get session length"""
         return self._telemetry_storage.get_session_length()

splitio/models/telemetry.py

@@ -482,6 +482,7 @@ def _reset_all(self):
             self._auth_rejections = 0
             self._token_refreshes = 0
             self._session_length = 0
+            self._update_from_sse = 0
 
     def record_impressions_value(self, resource, value):
         """
@@ -519,6 +520,14 @@ def record_events_value(self, resource, value):
             else:
                 return
 
+    def record_update_from_sse(self):
+        """
+        Increament the update from sse resource by one.
+
+        """
+        with self._lock:
+            self._update_from_sse += 1
+
     def record_auth_rejections(self):
         """
         Increament the auth rejection resource by one.
@@ -604,6 +613,18 @@ def pop_token_refreshes(self):
             self._token_refreshes = 0
             return token_refreshes
 
+    def pop_update_from_sse(self):
+        """
+        Pop update from sse
+
+        :return: token refreshes value
+        :rtype: int
+        """
+        with self._lock:
+            update_from_sse = self._update_from_sse
+            self._update_from_sse = 0
+            return update_from_sse
+
 class StreamingEvent(object):
     """
     Streaming event class

splitio/push/manager.py

@@ -44,7 +44,7 @@ def __init__(self, auth_api, synchronizer, feedback_loop, sdk_metadata, telemetr
         """
         self._auth_api = auth_api
         self._feedback_loop = feedback_loop
-        self._processor = MessageProcessor(synchronizer)
+        self._processor = MessageProcessor(synchronizer, telemetry_runtime_producer)
         self._status_tracker = PushStatusTracker(telemetry_runtime_producer)
         self._event_handlers = {
             EventType.MESSAGE: self._handle_message,

splitio/push/processor.py

@@ -9,7 +9,7 @@
 class MessageProcessor(object):
     """Message processor class."""
 
-    def __init__(self, synchronizer):
+    def __init__(self, synchronizer, telemetry_runtime_producer):
         """
         Class constructor.
 
@@ -19,8 +19,8 @@ def __init__(self, synchronizer):
         self._feature_flag_queue = Queue()
         self._segments_queue = Queue()
         self._synchronizer = synchronizer
-        self._feature_flag_worker = SplitWorker(synchronizer.synchronize_splits, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage)
-        self._segments_worker = SegmentWorker(synchronizer.synchronize_segment, self._segments_queue)
+        self._feature_flag_worker = SplitWorker(synchronizer.synchronize_splits, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage, telemetry_runtime_producer)
+        self._segments_worker = SegmentWorker(synchronizer.synchronize_segment, self._segments_queue, telemetry_runtime_producer)
         self._handlers = {
             UpdateType.SPLIT_UPDATE: self._handle_feature_flag_update,
             UpdateType.SPLIT_KILL: self._handle_feature_flag_kill,

splitio/push/segmentworker.py

@@ -11,7 +11,7 @@ class SegmentWorker(object):
 
     _centinel = object()
 
-    def __init__(self, synchronize_segment, segment_queue):
+    def __init__(self, synchronize_segment, segment_queue, telemetry_runtime_producer):
         """
         Class constructor.
 
@@ -25,6 +25,7 @@ def __init__(self, synchronize_segment, segment_queue):
         self._handler = synchronize_segment
         self._running = False
         self._worker = None
+        self._telemetry_runtime_producer = telemetry_runtime_producer
 
     def is_running(self):
         """Return whether the working is running."""

splitio/push/splitworker.py

@@ -24,7 +24,7 @@ class SplitWorker(object):
 
     _centinel = object()
 
-    def __init__(self, synchronize_feature_flag, feature_flag_queue, feature_flag_storage):
+    def __init__(self, synchronize_feature_flag, feature_flag_queue, feature_flag_storage, telemetry_runtime_producer):
         """
         Class constructor.
 
@@ -44,6 +44,7 @@ def __init__(self, synchronize_feature_flag, feature_flag_queue, feature_flag_st
             CompressionMode.GZIP_COMPRESSION: lambda event: gzip.decompress(base64.b64decode(event.feature_flag_definition)).decode('utf-8'),
             CompressionMode.ZLIB_COMPRESSION: lambda event: zlib.decompress(base64.b64decode(event.feature_flag_definition)).decode('utf-8'),
         }
+        self._telemetry_runtime_producer = telemetry_runtime_producer
 
     def is_running(self):
         """Return whether the working is running."""
@@ -75,6 +76,8 @@ def _run(self):
                         new_split = from_raw(json.loads(self._get_feature_flag_definition(event)))
                         if new_split.status == Status.ACTIVE:
                             self._feature_flag_storage.put(new_split)
+                            self._telemetry_runtime_producer.record_update_from_sse()
+                            _LOGGER.debug('Feature flag %s is updated', new_split.name)
                         else:
                             self._feature_flag_storage.remove(new_split.name)
                         self._feature_flag_storage.set_change_number(event.change_number)

splitio/storage/inmemmory.py

@@ -563,6 +563,10 @@ def record_session_length(self, session):
         """Record session length."""
         self._counters.record_session_length(session)
 
+    def record_update_from_sse(self):
+        """Record update from sse."""
+        self._counters.record_update_from_sse()
+
     def get_bur_time_outs(self):
         """Get block until ready timeout."""
         return self._tel_config.get_bur_time_outs()
@@ -632,6 +636,10 @@ def get_session_length(self):
         """Get session length"""
         return self._counters.get_session_length()
 
+    def pop_update_from_sse(self):
+        """Get and reset update from sse."""
+        return self._counters.pop_update_from_sse()
+
 class LocalhostTelemetryStorage():
     """Localhost telemetry storage."""
     def do_nothing(*_, **__):

splitio/sync/telemetry.py

@@ -60,7 +60,8 @@ def _build_stats(self):
         merged_dict = {
             'spC': self._feature_flag_storage.get_splits_count(),
             'seC': self._segment_storage.get_segments_count(),
-            'skC': self._segment_storage.get_segments_keys_count()
+            'skC': self._segment_storage.get_segments_keys_count(),
+            'ufs': {'sp': self._telemetry_runtime_consumer.pop_update_from_sse()}
         }
         merged_dict.update(self._telemetry_runtime_consumer.pop_formatted_stats())
         merged_dict.update(self._telemetry_evaluation_consumer.pop_formatted_stats())

tests/push/test_split_worker.py

@@ -6,6 +6,8 @@
 from splitio.api import APIException
 from splitio.push.splitworker import SplitWorker
 from splitio.push.parser import SplitChangeUpdate
+from splitio.engine.telemetry import TelemetryStorageProducer
+from splitio.storage.inmemmory import InMemoryTelemetryStorage
 
 change_number_received = None
 
@@ -20,11 +22,14 @@ class SplitWorkerTests(object):
 
     def test_on_error(self, mocker):
         q = queue.Queue()
+        telemetry_storage = InMemoryTelemetryStorage()
+        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
 
         def handler_sync(change_number):
             raise APIException('some')
 
-        split_worker = SplitWorker(handler_sync, q, mocker.Mock())
+        split_worker = SplitWorker(handler_sync, q, mocker.Mock(), telemetry_runtime_producer)
         split_worker.start()
         assert split_worker.is_running()
 
@@ -41,7 +46,10 @@ def handler_sync(change_number):
 
     def test_handler(self, mocker):
         q = queue.Queue()
-        split_worker = SplitWorker(handler_sync, q, mocker.Mock())
+        telemetry_storage = InMemoryTelemetryStorage()
+        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+        split_worker = SplitWorker(handler_sync, q, mocker.Mock(), telemetry_runtime_producer)
 
         global change_number_received
         assert not split_worker.is_running()
@@ -90,7 +98,10 @@ def set_change_number(new_change_number):
 
     def test_compression(self, mocker):
         q = queue.Queue()
-        split_worker = SplitWorker(handler_sync, q, mocker.Mock())
+        telemetry_storage = InMemoryTelemetryStorage()
+        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+        split_worker = SplitWorker(handler_sync, q, mocker.Mock(), telemetry_runtime_producer)
         global change_number_received
         split_worker.start()
         def get_change_number():
@@ -111,18 +122,21 @@ def remove(feature_flag):
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 2345,  'eyJ0cmFmZmljVHlwZU5hbWUiOiJ1c2VyIiwiaWQiOiIzM2VhZmE1MC0xYTY1LTExZWQtOTBkZi1mYTMwZDk2OTA0NDUiLCJuYW1lIjoiYmlsYWxfc3BsaXQiLCJ0cmFmZmljQWxsb2NhdGlvbiI6MTAwLCJ0cmFmZmljQWxsb2NhdGlvblNlZWQiOi0xMzY0MTE5MjgyLCJzZWVkIjotNjA1OTM4ODQzLCJzdGF0dXMiOiJBQ1RJVkUiLCJraWxsZWQiOmZhbHNlLCJkZWZhdWx0VHJlYXRtZW50Ijoib2ZmIiwiY2hhbmdlTnVtYmVyIjoxNjg0MzQwOTA4NDc1LCJhbGdvIjoyLCJjb25maWd1cmF0aW9ucyI6e30sImNvbmRpdGlvbnMiOlt7ImNvbmRpdGlvblR5cGUiOiJST0xMT1VUIiwibWF0Y2hlckdyb3VwIjp7ImNvbWJpbmVyIjoiQU5EIiwibWF0Y2hlcnMiOlt7ImtleVNlbGVjdG9yIjp7InRyYWZmaWNUeXBlIjoidXNlciJ9LCJtYXRjaGVyVHlwZSI6IklOX1NFR01FTlQiLCJuZWdhdGUiOmZhbHNlLCJ1c2VyRGVmaW5lZFNlZ21lbnRNYXRjaGVyRGF0YSI6eyJzZWdtZW50TmFtZSI6ImJpbGFsX3NlZ21lbnQifX1dfSwicGFydGl0aW9ucyI6W3sidHJlYXRtZW50Ijoib24iLCJzaXplIjowfSx7InRyZWF0bWVudCI6Im9mZiIsInNpemUiOjEwMH1dLCJsYWJlbCI6ImluIHNlZ21lbnQgYmlsYWxfc2VnbWVudCJ9LHsiY29uZGl0aW9uVHlwZSI6IlJPTExPVVQiLCJtYXRjaGVyR3JvdXAiOnsiY29tYmluZXIiOiJBTkQiLCJtYXRjaGVycyI6W3sia2V5U2VsZWN0b3IiOnsidHJhZmZpY1R5cGUiOiJ1c2VyIn0sIm1hdGNoZXJUeXBlIjoiQUxMX0tFWVMiLCJuZWdhdGUiOmZhbHNlfV19LCJwYXJ0aXRpb25zIjpbeyJ0cmVhdG1lbnQiOiJvbiIsInNpemUiOjB9LHsidHJlYXRtZW50Ijoib2ZmIiwic2l6ZSI6MTAwfV0sImxhYmVsIjoiZGVmYXVsdCBydWxlIn1dfQ==', 0))
         time.sleep(0.1)
         assert self._feature_flag.name == 'bilal_split'
+        assert telemetry_storage._counters._update_from_sse == 1
 
         # compression 2
         self._feature_flag = None
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 2345,  'eJzEUtFq20AQ/JUwz2c4WZZr3ZupTQh1FKjcQinGrKU95cjpZE6nh9To34ssJ3FNX0sfd3Zm53b2TgietDbF9vXIGdUMha5lDwFTQiGOmTQlchLRPJlEEZeTVJZ6oimWZTpP5WyWQMCNyoOxZPft0ZoA8TZ5aW1TUDCNg4qk/AueM5dQkyiez6IonS6mAu0IzWWSxovFLBZoA4WuhcLy8/bh+xoCL8bagaXJtixQsqbOhq1nCjW7AIVGawgUz+Qqzrr6wB4qmi9m00/JIk7TZCpAtmqgpgJF47SpOn9+UQt16s9YaS71z9NHOYQFha9Pm83Tty0EagrFM/t733RHqIFZH4wb7LDMVh+Ecc4Lv+ZsuQiNH8hXF3hLv39XXNCHbJ+v7x/X2eDmuKLA74sPihVr47jMuRpWfxy1Kwo0GLQjmv1xpBFD3+96gSP5cLVouM7QQaA1vxhK9uKmd853bEZS9jsBSwe2UDDu7mJxd2Mo/muQy81m/2X9I7+N8R/FcPmUd76zjH7X/w4AAP//90glTw==', 2))
         time.sleep(0.1)
         assert self._feature_flag.name == 'bilal_split'
+        assert telemetry_storage._counters._update_from_sse == 2
 
         # compression 1
         self._feature_flag = None
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 2345,  'H4sIAAkVZWQC/8WST0+DQBDFv0qzZ0ig/BF6a2xjGismUk2MaZopzOKmy9Isy0EbvrtDwbY2Xo233Tdv5se85cCMBs5FtvrYYwIlsglratTMYiKns+chcAgc24UwsF0Xczt2cm5z8Jw8DmPH9wPyqr5zKyTITb2XwpA4TJ5KWWVgRKXYxHWcX/QUkVi264W+68bjaGyxupdCJ4i9KPI9UgyYpibI9Ha1eJnT/J2QsnNxkDVaLEcOjTQrjWBKVIasFefky95BFZg05Zb2mrhh5I9vgsiL44BAIIuKTeiQVYqLotHHLyLOoT1quRjub4fztQuLxj89LpePzytClGCyd9R3umr21ErOcitUh2PTZHY29HN2+JGixMxUujNfvMB3+u2pY1AXySad3z3Mk46msACDp8W7jhly4uUpFt3qD33vDAx0gLpXkx+P1GusbdcE24M2F4uaywwVEWvxSa1Oa13Vjvn2RXradm0xCVuUVBJqNCBGV0DrX4OcLpeb+/lreh3jH8Uw/JQj3UhkxPgCCurdEnADAAA=', 1))
         time.sleep(0.1)
         assert self._feature_flag.name == 'bilal_split'
+        assert telemetry_storage._counters._update_from_sse == 3
 
         # should call delete split
         self._feature_flag = None
@@ -134,7 +148,10 @@ def remove(feature_flag):
 
     def test_edge_cases(self, mocker):
         q = queue.Queue()
-        split_worker = SplitWorker(handler_sync, q, mocker.Mock())
+        telemetry_storage = InMemoryTelemetryStorage()
+        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+        split_worker = SplitWorker(handler_sync, q, mocker.Mock(), telemetry_runtime_producer)
         global change_number_received
         split_worker.start()