Merge pull request #432 from splitio/async-sync-telemetry

added sync.telemetry async classes with a fix in engine.telemetry

Author: chillaq

splitio/engine/telemetry.py

@@ -345,7 +345,7 @@ async def get_not_ready_usage(self):
     async def get_config_stats(self):
         """Get config stats."""
         config_stats = await self._telemetry_storage.get_config_stats()
-        config_stats.update({'t': self.pop_config_tags()})
+        config_stats.update({'t': await self.pop_config_tags()})
         return config_stats
 
     async def get_config_stats_to_json(self):
@@ -427,9 +427,9 @@ async def pop_formatted_stats(self):
         :returns: formatted stats
         :rtype: Dict
         """
-        exceptions = await self.pop_exceptions()['methodExceptions']
-        latencies = await self.pop_latencies()['methodLatencies']
-        return self._to_json(exceptions, latencies)
+        exceptions = await self.pop_exceptions()
+        latencies = await self.pop_latencies()
+        return self._to_json(exceptions['methodExceptions'], latencies['methodLatencies'])
 
 
 class TelemetryRuntimeConsumerBase(object):
@@ -627,8 +627,8 @@ async def pop_formatted_stats(self):
         :rtype: Dict
         """
         last_synchronization = await self.get_last_synchronization()
-        http_errors = await self.pop_http_errors()['httpErrors']
-        http_latencies = await self.pop_http_latencies()['httpLatencies']
+        http_errors = await self.pop_http_errors()
+        http_latencies = await self.pop_http_latencies()
 
         return {
             'iQ': await self.get_impressions_stats(CounterConstants.IMPRESSIONS_QUEUED),
@@ -638,8 +638,8 @@ async def pop_formatted_stats(self):
             'eD': await self.get_events_stats(CounterConstants.EVENTS_DROPPED),
             'lS': self._last_synchronization_to_json(last_synchronization),
             't': await self.pop_tags(),
-            'hE': self._http_errors_to_json(http_errors),
-            'hL': self._http_latencies_to_json(http_latencies),
+            'hE': self._http_errors_to_json(http_errors['httpErrors']),
+            'hL': self._http_latencies_to_json(http_latencies['httpLatencies']),
             'aR': await self.pop_auth_rejections(),
             'tR': await self.pop_token_refreshes(),
             'sE': self._streaming_events_to_json(await self.pop_streaming_events()),

splitio/sync/telemetry.py

@@ -19,6 +19,23 @@ def synchronize_stats(self):
         """synchronize runtime stats class."""
         self._telemetry_submitter.synchronize_stats()
 
+
+class TelemetrySynchronizerAsync(object):
+    """Telemetry synchronizer class."""
+
+    def __init__(self, telemetry_submitter):
+        """Initialize Telemetry sync class."""
+        self._telemetry_submitter = telemetry_submitter
+
+    async def synchronize_config(self):
+        """synchronize initial config data class."""
+        await self._telemetry_submitter.synchronize_config()
+
+    async def synchronize_stats(self):
+        """synchronize runtime stats class."""
+        await self._telemetry_submitter.synchronize_stats()
+
+
 class TelemetrySubmitter(object, metaclass=abc.ABCMeta):
     """Telemetry sumbitter interface."""
 
@@ -30,7 +47,8 @@ def synchronize_config(self):
     def synchronize_stats(self):
         """synchronize runtime stats class."""
 
-class InMemoryTelemetrySubmitter(object):
+
+class InMemoryTelemetrySubmitter(TelemetrySubmitter):
     """Telemetry sumbitter class."""
 
     def __init__(self, telemetry_consumer, feature_flag_storage, segment_storage, telemetry_api):
@@ -66,6 +84,43 @@ def _build_stats(self):
         merged_dict.update(self._telemetry_evaluation_consumer.pop_formatted_stats())
         return merged_dict
 
+
+class InMemoryTelemetrySubmitterAsync(TelemetrySubmitter):
+    """Telemetry sumbitter async class."""
+
+    def __init__(self, telemetry_consumer, feature_flag_storage, segment_storage, telemetry_api):
+        """Initialize all producer classes."""
+        self._telemetry_init_consumer = telemetry_consumer.get_telemetry_init_consumer()
+        self._telemetry_evaluation_consumer = telemetry_consumer.get_telemetry_evaluation_consumer()
+        self._telemetry_runtime_consumer = telemetry_consumer.get_telemetry_runtime_consumer()
+        self._telemetry_api = telemetry_api
+        self._feature_flag_storage = feature_flag_storage
+        self._segment_storage = segment_storage
+
+    async def synchronize_config(self):
+        """synchronize initial config data classe."""
+        await self._telemetry_api.record_init(await self._telemetry_init_consumer.get_config_stats())
+
+    async def synchronize_stats(self):
+        """synchronize runtime stats class."""
+        await self._telemetry_api.record_stats(await self._build_stats())
+
+    async def _build_stats(self):
+        """
+        Format stats to Dict.
+
+        :returns: formatted stats
+        :rtype: Dict
+        """
+        merged_dict = {
+            'spC': await self._feature_flag_storage.get_splits_count(),
+            'seC': await self._segment_storage.get_segments_count(),
+            'skC': await self._segment_storage.get_segments_keys_count()
+        }
+        merged_dict.update(await self._telemetry_runtime_consumer.pop_formatted_stats())
+        merged_dict.update(await self._telemetry_evaluation_consumer.pop_formatted_stats())
+        return merged_dict
+
 class RedisTelemetrySubmitter(object):
     """Telemetry sumbitter class."""
 
@@ -82,6 +137,21 @@ def synchronize_stats(self):
         pass
 
 
+class RedisTelemetrySubmitterAsync(object):
+    """Telemetry sumbitter class."""
+
+    def __init__(self, telemetry_storage):
+        """Initialize all producer classes."""
+        self._telemetry_storage = telemetry_storage
+
+    async def synchronize_config(self):
+        """synchronize initial config data classe."""
+        await self._telemetry_storage.push_config_stats()
+
+    async def synchronize_stats(self):
+        """No implementation."""
+        pass
+
 class LocalhostTelemetrySubmitter(object):
     """Telemetry sumbitter class."""
 
@@ -92,3 +162,14 @@ def synchronize_config(self):
     def synchronize_stats(self):
         """No implementation."""
         pass
+
+class LocalhostTelemetrySubmitterAsync(object):
+    """Telemetry sumbitter class."""
+
+    async def synchronize_config(self):
+        """No implementation."""
+        pass
+
+    async def synchronize_stats(self):
+        """No implementation."""
+        pass

tests/sync/test_telemetry.py

@@ -1,12 +1,13 @@
 """Telemetry Worker tests."""
 import unittest.mock as mock
-import json
-from splitio.sync.telemetry import TelemetrySynchronizer, InMemoryTelemetrySubmitter
-from splitio.engine.telemetry import TelemetryEvaluationConsumer, TelemetryInitConsumer, TelemetryRuntimeConsumer, TelemetryStorageConsumer
-from splitio.storage.inmemmory import InMemoryTelemetryStorage, InMemorySegmentStorage, InMemorySplitStorage
+import pytest
+
+from splitio.sync.telemetry import TelemetrySynchronizer, TelemetrySynchronizerAsync, InMemoryTelemetrySubmitter, InMemoryTelemetrySubmitterAsync
+from splitio.engine.telemetry import TelemetryStorageConsumer, TelemetryStorageConsumerAsync
+from splitio.storage.inmemmory import InMemoryTelemetryStorage, InMemoryTelemetryStorageAsync, InMemorySegmentStorage, InMemorySegmentStorageAsync, InMemorySplitStorage, InMemorySplitStorageAsync
 from splitio.models.splits import Split, Status
 from splitio.models.segments import Segment
-from splitio.models.telemetry import StreamingEvents
+from splitio.models.telemetry import StreamingEvents, StreamingEventsAsync
 from splitio.api.telemetry import TelemetryAPI
 
 class TelemetrySynchronizerTests(object):
@@ -24,6 +25,31 @@ def test_synchronize_stats(self, mocker):
         telemetry_synchronizer.synchronize_stats()
         assert(mocker.called)
 
+
+class TelemetrySynchronizerAsyncTests(object):
+    """Telemetry synchronizer async test cases."""
+
+    @pytest.mark.asyncio
+    async def test_synchronize_config(self, mocker):
+        telemetry_synchronizer = TelemetrySynchronizerAsync(InMemoryTelemetrySubmitterAsync(mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock()))
+        self.called = False
+        async def synchronize_config():
+            self.called = True
+        telemetry_synchronizer.synchronize_config = synchronize_config
+        await telemetry_synchronizer.synchronize_config()
+        assert(self.called)
+
+    @pytest.mark.asyncio
+    async def test_synchronize_stats(self, mocker):
+        telemetry_synchronizer = TelemetrySynchronizer(InMemoryTelemetrySubmitter(mocker.Mock(), mocker.Mock(), mocker.Mock(), mocker.Mock()))
+        self.called = False
+        async def synchronize_stats():
+            self.called = True
+        telemetry_synchronizer.synchronize_stats = synchronize_stats
+        await telemetry_synchronizer.synchronize_stats()
+        assert(self.called)
+
+
 class TelemetrySubmitterTests(object):
     """Telemetry submitter test cases."""
 
@@ -136,3 +162,118 @@ def record_stats(*args, **kwargs):
             "skC": 0,
             "t": ['tag1']
         })
+
+
+class TelemetrySubmitterAsyncTests(object):
+    """Telemetry submitter async test cases."""
+
+    @pytest.mark.asyncio
+    async def test_synchronize_telemetry(self, mocker):
+        api = mocker.Mock(spec=TelemetryAPI)
+        telemetry_storage = await InMemoryTelemetryStorageAsync.create()
+        telemetry_consumer = TelemetryStorageConsumerAsync(telemetry_storage)
+        split_storage = InMemorySplitStorageAsync()
+        await split_storage.put(Split('split1', 1234, 1, False, 'user', Status.ACTIVE, 123))
+        segment_storage = InMemorySegmentStorageAsync()
+        await segment_storage.put(Segment('segment1', [], 123))
+        telemetry_submitter = InMemoryTelemetrySubmitterAsync(telemetry_consumer, split_storage, segment_storage, api)
+
+        telemetry_storage._counters._impressions_queued = 100
+        telemetry_storage._counters._impressions_deduped = 30
+        telemetry_storage._counters._impressions_dropped = 0
+        telemetry_storage._counters._events_queued = 20
+        telemetry_storage._counters._events_dropped = 10
+        telemetry_storage._counters._auth_rejections = 1
+        telemetry_storage._counters._token_refreshes = 3
+        telemetry_storage._counters._session_length = 3
+
+        telemetry_storage._method_exceptions._treatment =  10
+        telemetry_storage._method_exceptions._treatments = 1
+        telemetry_storage._method_exceptions._treatment_with_config = 5
+        telemetry_storage._method_exceptions._treatments_with_config = 1
+        telemetry_storage._method_exceptions._track = 3
+
+        telemetry_storage._last_synchronization._split = 5
+        telemetry_storage._last_synchronization._segment = 3
+        telemetry_storage._last_synchronization._impression = 10
+        telemetry_storage._last_synchronization._impression_count = 0
+        telemetry_storage._last_synchronization._event = 4
+        telemetry_storage._last_synchronization._telemetry = 0
+        telemetry_storage._last_synchronization._token = 3
+
+        telemetry_storage._http_sync_errors._split = {'500': 3, '501': 2}
+        telemetry_storage._http_sync_errors._segment = {'401': 1}
+        telemetry_storage._http_sync_errors._impression = {'500': 1}
+        telemetry_storage._http_sync_errors._impression_count = {'401': 5}
+        telemetry_storage._http_sync_errors._event = {'404': 10}
+        telemetry_storage._http_sync_errors._telemetry = {'501': 3}
+        telemetry_storage._http_sync_errors._token = {'505': 11}
+
+        telemetry_storage._streaming_events = await StreamingEventsAsync.create()
+        telemetry_storage._tags = ['tag1']
+
+        telemetry_storage._method_latencies._treatment = [1] + [0] * 22
+        telemetry_storage._method_latencies._treatments = [0] * 23
+        telemetry_storage._method_latencies._treatment_with_config = [0] * 23
+        telemetry_storage._method_latencies._treatments_with_config = [0] * 23
+        telemetry_storage._method_latencies._track = [0] * 23
+
+        telemetry_storage._http_latencies._split = [1] + [0] * 22
+        telemetry_storage._http_latencies._segment = [0] * 23
+        telemetry_storage._http_latencies._impression = [0] * 23
+        telemetry_storage._http_latencies._impression_count = [0] * 23
+        telemetry_storage._http_latencies._event = [0] * 23
+        telemetry_storage._http_latencies._telemetry = [0] * 23
+        telemetry_storage._http_latencies._token =  [0] * 23
+
+        await telemetry_storage.record_config({'operationMode': 'inmemory',
+                                         'storageType': None,
+                                        'streamingEnabled': True,
+                                        'impressionsQueueSize': 100,
+                                        'eventsQueueSize': 200,
+                                        'impressionsMode': 'DEBUG',
+                                        'impressionListener': None,
+                                        'featuresRefreshRate': 30,
+                                        'segmentsRefreshRate': 30,
+                                        'impressionsRefreshRate': 60,
+                                        'eventsPushRate': 60,
+                                        'metricsRefreshRate': 10,
+                                        'activeFactoryCount': 1,
+                                        'notReady': 0,
+                                        'timeUntilReady': 1
+                                       }, {}
+        )
+        self.formatted_config = ""
+        async def record_init(*args, **kwargs):
+            self.formatted_config = args[0]
+        api.record_init = record_init
+
+        await telemetry_submitter.synchronize_config()
+        assert(self.formatted_config == await telemetry_submitter._telemetry_init_consumer.get_config_stats())
+
+        async def record_stats(*args, **kwargs):
+            self.formatted_stats = args[0]
+        api.record_stats = record_stats
+
+        await telemetry_submitter.synchronize_stats()
+        assert(self.formatted_stats == {
+            "iQ": 100,
+            "iDe": 30,
+            "iDr": 0,
+            "eQ": 20,
+            "eD": 10,
+            "lS": {"sp": 5, "se": 3, "im": 10, "ic": 0, "ev": 4, "te": 0, "to": 3},
+            "t": ["tag1"],
+            "hE": {"sp": {"500": 3, "501": 2}, "se": {"401": 1}, "im": {"500": 1}, "ic": {"401": 5}, "ev": {"404": 10}, "te": {"501": 3}, "to": {"505": 11}},
+            "hL": {"sp": [1] + [0] * 22, "se": [0] * 23, "im": [0] * 23, "ic": [0] * 23, "ev": [0] * 23, "te": [0] * 23, "to": [0] * 23},
+            "aR": 1,
+            "tR": 3,
+            "sE": [],
+            "sL": 3,
+            "mE": {"t": 10, "ts": 1, "tc": 5, "tcs": 1, "tr": 3},
+            "mL": {"t": [1] + [0] * 22, "ts": [0] * 23, "tc": [0] * 23, "tcs": [0] * 23, "tr": [0] * 23},
+            "spC": 1,
+            "seC": 1,
+            "skC": 0,
+            "t": ['tag1']
+        })