update telemetry calls

Author: chillaq

splitio/storage/redis.py

@@ -5,7 +5,7 @@
 
 from splitio.models.impressions import Impression
 from splitio.models import splits, segments
-from splitio.models.telemetry import MethodExceptions, MethodLatencies, TelemetryConfig, get_latency_bucket_index
+from splitio.models.telemetry import TelemetryConfig, get_latency_bucket_index, TelemetryConfigAsync
 from splitio.storage import SplitStorage, SegmentStorage, ImpressionStorage, EventStorage, \
     ImpressionPipelinedStorage, TelemetryStorage
 from splitio.storage.adapters.redis import RedisAdapterException
@@ -623,7 +623,7 @@ def record_config(self, config, extra_config):
         :param congif: factory configuration parameters
         :type config: splitio.client.config
         """
-        self._tel_config.record_config(config, extra_config)
+        pass
 
     def pop_config_tags(self):
         """Get and reset tags."""
@@ -633,9 +633,8 @@ def push_config_stats(self):
         """push config stats to redis."""
         pass
 
-    def _format_config_stats(self, tags):
+    def _format_config_stats(self, config_stats, tags):
         """format only selected config stats to json"""
-        config_stats = self._tel_config.get_stats()
         return json.dumps({
             'aF': config_stats['aF'],
             'rF': config_stats['rF'],
@@ -646,7 +645,7 @@ def _format_config_stats(self, tags):
 
     def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
         """Record active and redundant factories."""
-        self._tel_config.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
+        pass
 
     def add_latency_to_pipe(self, method, bucket, pipe):
         """
@@ -728,8 +727,6 @@ def __init__(self, redis_client, sdk_metadata):
         self._reset_config_tags()
         self._redis_client = redis_client
         self._sdk_metadata = sdk_metadata
-        self._method_latencies = MethodLatencies()
-        self._method_exceptions = MethodExceptions()
         self._tel_config = TelemetryConfig()
         self._make_pipe = redis_client.pipeline
 
@@ -744,6 +741,15 @@ def add_config_tag(self, tag):
             if len(self._config_tags) < MAX_TAGS:
                 self._config_tags.append(tag)
 
+    def record_config(self, config, extra_config):
+        """
+        initilize telemetry objects
+
+        :param congif: factory configuration parameters
+        :type config: splitio.client.config
+        """
+        self._tel_config.record_config(config, extra_config)
+
     def pop_config_tags(self):
         """Get and reset tags."""
         with self._lock:
@@ -754,8 +760,8 @@ def pop_config_tags(self):
     def push_config_stats(self):
         """push config stats to redis."""
         _LOGGER.debug("Adding Config stats to redis key %s" % (self._TELEMETRY_CONFIG_KEY))
-        _LOGGER.debug(str(self._format_config_stats(self.pop_config_tags())))
-        self._redis_client.hset(self._TELEMETRY_CONFIG_KEY, self._sdk_metadata.sdk_version + '/' + self._sdk_metadata.instance_name + '/' + self._sdk_metadata.instance_ip, str(self._format_config_stats(self.pop_config_tags())))
+        _LOGGER.debug(str(self._format_config_stats(self._tel_config.get_stats(), self.pop_config_tags())))
+        self._redis_client.hset(self._TELEMETRY_CONFIG_KEY, self._sdk_metadata.sdk_version + '/' + self._sdk_metadata.instance_name + '/' + self._sdk_metadata.instance_ip, str(self._format_config_stats(self._tel_config.get_stats(), self.pop_config_tags())))
 
     def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
         """Record active and redundant factories."""
@@ -777,6 +783,10 @@ def record_exception(self, method):
         result = pipe.execute()
         self.expire_keys(self._TELEMETRY_EXCEPTIONS_KEY, self._TELEMETRY_KEY_DEFAULT_TTL, 1, result[0])
 
+    def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
+        """Record active and redundant factories."""
+        self._tel_config.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
+
     def expire_latency_keys(self, total_keys, inserted):
         """
         Expire lstency keys
@@ -820,9 +830,7 @@ async def create(redis_client, sdk_metadata):
         await self._reset_config_tags()
         self._redis_client = redis_client
         self._sdk_metadata = sdk_metadata
-        self._method_latencies = MethodLatencies() # to be changed to async version class
-        self._method_exceptions = MethodExceptions() # to be changed to async version class
-        self._tel_config = TelemetryConfig() # to be changed to async version class
+        self._tel_config = await TelemetryConfigAsync.create()
         self._make_pipe = redis_client.pipeline
         return self
 
@@ -835,6 +843,15 @@ async def add_config_tag(self, tag):
         if len(self._config_tags) < MAX_TAGS:
             self._config_tags.append(tag)
 
+    async def record_config(self, config, extra_config):
+        """
+        initilize telemetry objects
+
+        :param congif: factory configuration parameters
+        :type config: splitio.client.config
+        """
+        await self._tel_config.record_config(config, extra_config)
+
     async def pop_config_tags(self):
         """Get and reset tags."""
         tags = self._config_tags
@@ -844,8 +861,8 @@ async def pop_config_tags(self):
     async def push_config_stats(self):
         """push config stats to redis."""
         _LOGGER.debug("Adding Config stats to redis key %s" % (self._TELEMETRY_CONFIG_KEY))
-        _LOGGER.debug(str(await self._format_config_stats(await self.pop_config_tags())))
-        await self._redis_client.hset(self._TELEMETRY_CONFIG_KEY, self._sdk_metadata.sdk_version + '/' + self._sdk_metadata.instance_name + '/' + self._sdk_metadata.instance_ip, str(await self._format_config_stats(await self.pop_config_tags())))
+        _LOGGER.debug(str(await self._format_config_stats(await self._tel_config.get_stats(), await self.pop_config_tags())))
+        await self._redis_client.hset(self._TELEMETRY_CONFIG_KEY, self._sdk_metadata.sdk_version + '/' + self._sdk_metadata.instance_name + '/' + self._sdk_metadata.instance_ip, str(await self._format_config_stats(await self._tel_config.get_stats(), await self.pop_config_tags())))
 
     async def record_exception(self, method):
         """
@@ -863,6 +880,10 @@ async def record_exception(self, method):
         result = await pipe.execute()
         await self.expire_keys(self._TELEMETRY_EXCEPTIONS_KEY, self._TELEMETRY_KEY_DEFAULT_TTL, 1, result[0])
 
+    async def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
+        """Record active and redundant factories."""
+        await self._tel_config.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
+
     async def expire_latency_keys(self, total_keys, inserted):
         """
         Expire lstency keys

tests/storage/test_redis.py

@@ -14,7 +14,7 @@
 from splitio.models.segments import Segment
 from splitio.models.impressions import Impression
 from splitio.models.events import Event, EventWrapper
-from splitio.models.telemetry import MethodExceptions, MethodLatencies, TelemetryConfig, MethodExceptionsAndLatencies
+from splitio.models.telemetry import MethodExceptions, MethodLatencies, TelemetryConfig, MethodExceptionsAndLatencies, TelemetryConfigAsync
 
 
 class RedisSplitStorageTests(object):
@@ -496,20 +496,18 @@ async def test_init(self, mocker):
         redis_telemetry = await RedisTelemetryStorageAsync.create(mocker.Mock(), mocker.Mock())
         assert(redis_telemetry._redis_client is not None)
         assert(redis_telemetry._sdk_metadata is not None)
-        assert(isinstance(redis_telemetry._method_latencies, MethodLatencies))
-        assert(isinstance(redis_telemetry._method_exceptions, MethodExceptions))
-        assert(isinstance(redis_telemetry._tel_config, TelemetryConfig))
+        assert(isinstance(redis_telemetry._tel_config, TelemetryConfigAsync))
         assert(redis_telemetry._make_pipe is not None)
 
     @pytest.mark.asyncio
     async def test_record_config(self, mocker):
         redis_telemetry = await RedisTelemetryStorageAsync.create(mocker.Mock(), mocker.Mock())
         self.called = False
-        def record_config(*args):
+        async def record_config(*args):
             self.called = True
         redis_telemetry._tel_config.record_config = record_config
 
-        redis_telemetry.record_config(mocker.Mock(), mocker.Mock())
+        await redis_telemetry.record_config(mocker.Mock(), mocker.Mock())
         assert(self.called)
 
     @pytest.mark.asyncio
@@ -523,7 +521,7 @@ async def hset(key, hash, val):
             self.hash = hash
 
         adapter.hset = hset
-        async def format_config_stats(tags):
+        async def format_config_stats(stats, tags):
             return ""
         redis_telemetry._format_config_stats = format_config_stats
         await redis_telemetry.push_config_stats()
@@ -533,8 +531,8 @@ async def format_config_stats(tags):
     @pytest.mark.asyncio
     async def test_format_config_stats(self, mocker):
         redis_telemetry = await RedisTelemetryStorageAsync.create(mocker.Mock(), mocker.Mock())
-        json_value = redis_telemetry._format_config_stats([])
-        stats = redis_telemetry._tel_config.get_stats()
+        json_value = redis_telemetry._format_config_stats({'aF': 0, 'rF': 0, 'sT': None, 'oM': None}, [])
+        stats = await redis_telemetry._tel_config.get_stats()
         assert(json_value == json.dumps({
             'aF': stats['aF'],
             'rF': stats['rF'],
@@ -548,7 +546,7 @@ async def test_record_active_and_redundant_factories(self, mocker):
         redis_telemetry = await RedisTelemetryStorageAsync.create(mocker.Mock(), mocker.Mock())
         active_factory_count = 1
         redundant_factory_count = 2
-        redis_telemetry.record_active_and_redundant_factories(1, 2)
+        await redis_telemetry.record_active_and_redundant_factories(1, 2)
         assert (redis_telemetry._tel_config._active_factory_count == active_factory_count)
         assert (redis_telemetry._tel_config._redundant_factory_count == redundant_factory_count)
 
@@ -577,18 +575,26 @@ def _mocked_hincrby2(*args, **kwargs):
 
     @pytest.mark.asyncio
     async def test_record_exception(self, mocker):
-        async def _mocked_hincrby(*args, **kwargs):
+        self.called = False
+        def _mocked_hincrby(*args, **kwargs):
+            self.called = True
             assert(args[1] == RedisTelemetryStorageAsync._TELEMETRY_EXCEPTIONS_KEY)
             assert(args[2] == 'python-1.1.1/hostname/ip/treatment')
             assert(args[3] == 1)
 
-        adapter = build({})
+        self.called2 = False
+        async def _mocked_execute(*args):
+            self.called2 = True
+            return [1]
+
+        adapter = await aioredis.from_url("redis://localhost")
         metadata = SdkMetadata('python-1.1.1', 'hostname', 'ip')
         redis_telemetry = await RedisTelemetryStorageAsync.create(adapter, metadata)
-        with mock.patch('redis.client.Pipeline.hincrby', _mocked_hincrby):
-            with mock.patch('redis.client.Pipeline.execute') as mock_method:
-                mock_method.return_value = [1]
-                redis_telemetry.record_exception(MethodExceptionsAndLatencies.TREATMENT)
+        with mock.patch('redis.asyncio.client.Pipeline.hincrby', _mocked_hincrby):
+            with mock.patch('redis.asyncio.client.Pipeline.execute', _mocked_execute):
+                await redis_telemetry.record_exception(MethodExceptionsAndLatencies.TREATMENT)
+                assert self.called
+                assert self.called2
 
     @pytest.mark.asyncio
     async def test_expire_latency_keys(self, mocker):