Merge pull request #392 from splitio/iff-telemetry-update

Added telemetry to iff update

Author: chillaq

splitio/engine/telemetry.py

@@ -6,7 +6,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 from  splitio.storage.inmemmory import InMemoryTelemetryStorage
-from splitio.models.telemetry import CounterConstants
+from splitio.models.telemetry import CounterConstants, UpdateFromSSE
 
 class TelemetryStorageProducer(object):
     """Telemetry storage producer class."""
@@ -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, event):
+        """Record update from sse."""
+        self._telemetry_storage.record_update_from_sse(event)
+
 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, event):
+        """Get and reset update from sse."""
+        return self._telemetry_storage.pop_update_from_sse(event)
+
     def get_session_length(self):
         """Get session length"""
         return self._telemetry_storage.get_session_length()
@@ -292,6 +300,7 @@ def pop_formatted_stats(self):
             'iDr': self.get_impressions_stats(CounterConstants.IMPRESSIONS_DROPPED),
             'eQ': self.get_events_stats(CounterConstants.EVENTS_QUEUED),
             'eD': self.get_events_stats(CounterConstants.EVENTS_DROPPED),
+            'ufs': {event.value: self.pop_update_from_sse(event) for event in UpdateFromSSE},
             'lS': {'sp': last_synchronization['split'],
                       'se': last_synchronization['segment'],
                       'im': last_synchronization['impression'],

splitio/models/telemetry.py

@@ -131,6 +131,10 @@ class OperationMode(Enum):
     CONSUMER = 'consumer'
     PARTIAL_CONSUMER = 'partial_consumer'
 
+class UpdateFromSSE(Enum):
+    """Update from sse constants"""
+    SPLIT_UPDATE = 'sp'
+
 def get_latency_bucket_index(micros):
     """
     Find the bucket index for a measured latency.
@@ -482,6 +486,7 @@ def _reset_all(self):
             self._auth_rejections = 0
             self._token_refreshes = 0
             self._session_length = 0
+            self._update_from_sse = {}
 
     def record_impressions_value(self, resource, value):
         """
@@ -519,17 +524,27 @@ def record_events_value(self, resource, value):
             else:
                 return
 
+    def record_update_from_sse(self, event):
+        """
+        Increment the update from sse resource by one.
+
+        """
+        with self._lock:
+            if event.value not in self._update_from_sse:
+                self._update_from_sse[event.value] = 0
+            self._update_from_sse[event.value] += 1
+
     def record_auth_rejections(self):
         """
-        Increament the auth rejection resource by one.
+        Increment the auth rejection resource by one.
 
         """
         with self._lock:
             self._auth_rejections += 1
 
     def record_token_refreshes(self):
         """
-        Increament the token refreshes resource by one.
+        Increment the token refreshes resource by one.
 
         """
         with self._lock:
@@ -604,6 +619,18 @@ def pop_token_refreshes(self):
             self._token_refreshes = 0
             return token_refreshes
 
+    def pop_update_from_sse(self, event):
+        """
+        Pop update from sse
+
+        :return: update from sse value
+        :rtype: int
+        """
+        with self._lock:
+            update_from_sse = self._update_from_sse[event.value]
+            self._update_from_sse[event.value] = 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,7 +19,7 @@ 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._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)
         self._handlers = {
             UpdateType.SPLIT_UPDATE: self._handle_feature_flag_update,

splitio/push/splitworker.py

@@ -8,8 +8,10 @@
 from enum import Enum
 
 from splitio.models.splits import from_raw, Status
+from splitio.models.telemetry import UpdateFromSSE
 from splitio.push.parser import UpdateType
 
+
 _LOGGER = logging.getLogger(__name__)
 
 class CompressionMode(Enum):
@@ -24,7 +26,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 +46,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,9 +78,11 @@ 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)
+                            _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)
+                        self._telemetry_runtime_producer.record_update_from_sse(UpdateFromSSE.SPLIT_UPDATE)
                         continue
                     except Exception as e:
                         _LOGGER.error('Exception raised in updating feature flag')

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, event):
+        """Record update from sse."""
+        self._counters.record_update_from_sse(event)
+
     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, event):
+        """Get and reset update from sse."""
+        return self._counters.pop_update_from_sse(event)
+
 class LocalhostTelemetryStorage():
     """Localhost telemetry storage."""
     def do_nothing(*_, **__):

splitio/sync/telemetry.py

@@ -3,6 +3,7 @@
 
 from splitio.api.telemetry import TelemetryAPI
 from splitio.engine.telemetry import TelemetryStorageConsumer
+from splitio.models.telemetry import UpdateFromSSE
 
 class TelemetrySynchronizer(object):
     """Telemetry synchronizer class."""

tests/engine/test_telemetry.py

@@ -163,6 +163,13 @@ def test_record_token_refreshes(self, mocker):
         telemetry_runtime_producer.record_token_refreshes()
         assert(mocker.called)
 
+    @mock.patch('splitio.storage.inmemmory.InMemoryTelemetryStorage.record_update_from_sse')
+    def test_record_update_from_sse(self, mocker):
+        telemetry_storage = InMemoryTelemetryStorage()
+        telemetry_runtime_producer = TelemetryRuntimeProducer(telemetry_storage)
+        telemetry_runtime_producer.record_update_from_sse('sp')
+        assert(mocker.called)
+
     def test_record_streaming_event(self, mocker):
         telemetry_storage = mocker.Mock()
         telemetry_runtime_producer = TelemetryRuntimeProducer(telemetry_storage)
@@ -290,6 +297,12 @@ def test_pop_auth_rejections(self, mocker):
         telemetry_runtime_consumer = TelemetryRuntimeConsumer(telemetry_storage)
         telemetry_runtime_consumer.pop_auth_rejections()
 
+    @mock.patch('splitio.storage.inmemmory.InMemoryTelemetryStorage.pop_update_from_sse')
+    def test_pop_auth_rejections(self, mocker):
+        telemetry_storage = InMemoryTelemetryStorage()
+        telemetry_runtime_consumer = TelemetryRuntimeConsumer(telemetry_storage)
+        telemetry_runtime_consumer.pop_update_from_sse('sp')
+
     @mock.patch('splitio.storage.inmemmory.InMemoryTelemetryStorage.pop_token_refreshes')
     def test_pop_token_refreshes(self, mocker):
         telemetry_storage = InMemoryTelemetryStorage()

tests/models/test_telemetry_model.py

@@ -5,7 +5,7 @@
 
 from splitio.models.telemetry import StorageType, OperationMode, MethodLatencies, MethodExceptions, \
     HTTPLatencies, HTTPErrors, LastSynchronization, TelemetryCounters, TelemetryConfig, \
-    StreamingEvent, StreamingEvents, get_latency_bucket_index
+    StreamingEvent, StreamingEvents, UpdateFromSSE
 
 import splitio.models.telemetry as ModelTelemetry
 
@@ -195,6 +195,7 @@ def test_telemetry_counters(self):
         assert(telemetry_counter._events_queued == 0)
         assert(telemetry_counter._auth_rejections == 0)
         assert(telemetry_counter._token_refreshes == 0)
+        assert(telemetry_counter._update_from_sse == {})
 
         telemetry_counter.record_session_length(20)
         assert(telemetry_counter.get_session_length() == 20)
@@ -219,6 +220,11 @@ def test_telemetry_counters(self):
         assert(telemetry_counter._events_queued == 30)
         telemetry_counter.record_events_value(ModelTelemetry.CounterConstants.EVENTS_DROPPED, 1)
         assert(telemetry_counter._events_dropped == 1)
+        telemetry_counter.record_update_from_sse(UpdateFromSSE.SPLIT_UPDATE)
+        assert(telemetry_counter._update_from_sse[UpdateFromSSE.SPLIT_UPDATE.value] == 1)
+        updates = telemetry_counter.pop_update_from_sse(UpdateFromSSE.SPLIT_UPDATE)
+        assert(telemetry_counter._update_from_sse[UpdateFromSSE.SPLIT_UPDATE.value] == 0)
+        assert(updates == 1)
 
     def test_streaming_event(self, mocker):
         streaming_event = StreamingEvent((ModelTelemetry.StreamingEventTypes.CONNECTION_ESTABLISHED, 'split', 1234))

tests/push/test_manager.py

@@ -145,14 +145,13 @@ def test_split_change(self, mocker):
         processor_mock = mocker.Mock(spec=MessageProcessor)
         mocker.patch('splitio.push.manager.MessageProcessor', new=processor_mock)
 
-        telemetry_storage = InMemoryTelemetryStorage()
-        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
-        manager = PushManager(mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock(), telemetry_runtime_producer)
+        telemetry_runtime_producer = mocker.Mock()
+        synchronizer = mocker.Mock()
+        manager = PushManager(mocker.Mock(), synchronizer, mocker.Mock(), mocker.Mock(), telemetry_runtime_producer)
         manager._event_handler(sse_event)
         assert parse_event_mock.mock_calls == [mocker.call(sse_event)]
         assert processor_mock.mock_calls == [
-            mocker.call(Any()),
+            mocker.call(synchronizer, telemetry_runtime_producer),
             mocker.call().handle(update_message)
         ]
 
@@ -167,11 +166,13 @@ def test_split_kill(self, mocker):
         processor_mock = mocker.Mock(spec=MessageProcessor)
         mocker.patch('splitio.push.manager.MessageProcessor', new=processor_mock)
 
-        manager = PushManager(mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock())
+        telemetry_runtime_producer = mocker.Mock()
+        synchronizer = mocker.Mock()
+        manager = PushManager(mocker.Mock(), synchronizer, mocker.Mock(), mocker.Mock(), telemetry_runtime_producer)
         manager._event_handler(sse_event)
         assert parse_event_mock.mock_calls == [mocker.call(sse_event)]
         assert processor_mock.mock_calls == [
-            mocker.call(Any()),
+            mocker.call(synchronizer, telemetry_runtime_producer),
             mocker.call().handle(update_message)
         ]
 
@@ -186,11 +187,13 @@ def test_segment_change(self, mocker):
         processor_mock = mocker.Mock(spec=MessageProcessor)
         mocker.patch('splitio.push.manager.MessageProcessor', new=processor_mock)
 
-        manager = PushManager(mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock())
+        telemetry_runtime_producer = mocker.Mock()
+        synchronizer = mocker.Mock()
+        manager = PushManager(mocker.Mock(), synchronizer, mocker.Mock(), mocker.Mock(), telemetry_runtime_producer)
         manager._event_handler(sse_event)
         assert parse_event_mock.mock_calls == [mocker.call(sse_event)]
         assert processor_mock.mock_calls == [
-            mocker.call(Any()),
+            mocker.call(synchronizer, telemetry_runtime_producer),
             mocker.call().handle(update_message)
         ]