Telemetry implementation for redis

Author: chillaq

splitio/client/factory.py

@@ -152,6 +152,20 @@ def _start_status_updater(self):
             ready_updater.start()
         else:
             self._status = Status.READY
+            ready_updater = threading.Thread(target=self._update_redis_telemetry_config,
+                                             name='SDKRedisTelemetryConfig')
+            ready_updater.setDaemon(True)
+            ready_updater.start()
+
+
+    def _update_redis_telemetry_config(self):
+        """Push Config Telemetry into storage."""
+        self._telemetry_init_producer.record_ready_time(get_current_epoch_time() - self._ready_time)
+        redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
+        self._telemetry_init_producer.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
+        config_post_thread = threading.Thread(target=self._telemetry_api.record_init(self._telemetry_init_consumer.get_config_stats()), name="PostConfigData")
+        config_post_thread.setDaemon(True)
+        config_post_thread.start()
 
     def _update_status_when_ready(self):
         """Wait until the sdk is ready and update the status."""
@@ -469,14 +483,14 @@ def _build_redis_factory(api_key, cfg):
 
     synchronizers = SplitSynchronizers(None, None, None, None,
         impressions_count_sync,
-        None,
+        TelemetrySynchronizer(telemetry_consumer, storages['splits'], storages['segments'], redis_adapter),
         unique_keys_synchronizer,
         clear_filter_sync
     )
 
     tasks = SplitTasks(None, None, None, None,
         impressions_count_task,
-        None,
+        TelemetrySyncTask(synchronizers.telemetry_sync.synchronize_stats, cfg['metricsRefreshRate']),
         unique_keys_task,
         clear_filter_task
     )
@@ -494,6 +508,8 @@ def _build_redis_factory(api_key, cfg):
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer")
     initialization_thread.setDaemon(True)
     initialization_thread.start()
+    
+    telemetry_producer.get_telemetry_init_producer().record_config(cfg, {})
 
     return SplitFactory(
         api_key,
@@ -502,6 +518,7 @@ def _build_redis_factory(api_key, cfg):
         recorder,
         manager,
         sdk_ready_flag=None,
+        telemetry_api=redis_adapter,
         telemetry_producer=telemetry_producer,
         telemetry_init_consumer=telemetry_consumer.get_telemetry_init_consumer()
     )

splitio/storage/adapters/redis.py

@@ -1,5 +1,8 @@
 """Redis client wrapper with prefix support."""
 from builtins import str
+import socket
+import logging
+from splitio.version import __version__
 
 try:
     from redis import StrictRedis
@@ -14,6 +17,10 @@ def missing_redis_dependencies(*_, **__):
         )
     StrictRedis = Sentinel = missing_redis_dependencies
 
+_LOGGER = logging.getLogger(__name__)
+TELEMETRY_CONFIG_KEY = 'SPLITIO.telemetry.init'
+TELEMETRY_EXCEPTIONS_KEY = 'SPLITIO.telemetry.exceptions'
+TELEMETRY_LATENCIES_KEY = 'SPLITIO.telemetry.latencies'
 
 class RedisAdapterException(Exception):
     """Exception to be thrown when a redis command fails with an exception."""
@@ -241,6 +248,13 @@ def hget(self, name, key):
         except RedisError as exc:
             raise RedisAdapterException('Error executing hget operation') from exc
 
+    def hincrby(self, name, key, amount=1):
+        """Mimic original redis function but using user custom prefix."""
+        try:
+            return self._decorated.hincrby(self._prefix_helper.add_prefix(name), key, amount)
+        except RedisError as exc:
+            raise RedisAdapterException('Error executing hincrby operation') from exc
+
     def incr(self, name, amount=1):
         """Mimic original redis function but using user custom prefix."""
         try:
@@ -297,6 +311,54 @@ def pipeline(self):
         except RedisError as exc:
             raise RedisAdapterException('Error executing ttl operation') from exc
 
+    def record_init(self, *values):
+        try:
+            host_name, host_ip = self._get_host_info()
+            return self.hset(TELEMETRY_CONFIG_KEY, 'python-' + __version__ + '/' + host_name+ '/' + host_ip, str(*values))
+        except RedisError as exc:
+            raise RedisAdapterException('Error pushing telemetry config operation') from exc
+
+    def _get_host_info(self):
+        host_name = 'Unknown'
+        host_ip = 'Unknown'
+        try:
+            host_name = socket.gethostname()
+            host_ip = socket.gethostbyname(socket.gethostname())
+        except:
+            _LOGGER.debug("Could not get hostname or ip")
+            pass
+        return host_name, host_ip
+
+    def record_stats(self, values):
+        try:
+            host_name, host_ip = self._get_host_info()
+            for item in values['mL']:
+                bucket_number = 0
+                for bucket in values['mL'][item]:
+                    if bucket > 0:
+                        self.hincrby(TELEMETRY_LATENCIES_KEY, 'python-' + __version__ + '/' + host_name+ '/' + host_ip + '/' +
+                             self._get_method_name(item) + '/' + str(bucket_number), bucket)
+                    bucket_number = bucket_number + 0
+            for item in values['mE']:
+                if values['mE'][item] > 0:
+                    self.hincrby(TELEMETRY_EXCEPTIONS_KEY, 'python-' + __version__ + '/' + host_name+ '/' + host_ip + '/' +
+                            self._get_method_name(item), values['mE'][item])
+        except RedisError as exc:
+            raise RedisAdapterException('Error pushing telemetry evaluation operation') from exc
+
+    def _get_method_name(self, item):
+        if item == 't':
+            return 'treatment'
+        elif item == 'ts':
+            return 'treatments'
+        elif item == 'tc':
+            return 'treatment_with_config'
+        elif item == 'tcs':
+            return 'treatments_with_config'
+        elif item == 'tr':
+            return 'track'
+        else:
+            return ''
 
 class RedisPipelineAdapter(object):
     """

splitio/storage/redis.py

@@ -180,6 +180,23 @@ def get_split_names(self):
             _LOGGER.debug('Error: ', exc_info=True)
             return []
 
+    def get_splits_count(self):
+        """
+        Return splits count.
+
+        :rtype: int
+        """
+        return 0
+
+    def get_all_splits(self):
+        """
+        Return all the splits in cache.
+
+        :return: 0
+        :rtype: int
+        """
+        return 0
+
     def get_all_splits(self):
         """
         Return all the splits in cache.
@@ -345,6 +362,22 @@ def segment_contains(self, segment_name, key):
             _LOGGER.debug('Error: ', exc_info=True)
             return None
 
+    def get_segments_count(self):
+        """
+        Return segment count.
+
+        :return: 0
+        :rtype: int
+        """
+        return 0
+
+    def get_segments_keys_count(self):
+        """
+        Return segment count.
+
+        :rtype: int
+        """
+        return 0
 
 class RedisImpressionsStorage(ImpressionStorage, ImpressionPipelinedStorage):
     """Redis based event storage class."""

splitio/sync/synchronizer.py

@@ -400,13 +400,16 @@ def __init__(self, split_synchronizers, split_tasks):
         :type split_tasks: splitio.sync.synchronizer.SplitTasks
         """
         self._split_synchronizers = split_synchronizers
-        self._tasks = []
+        self._tasks = [split_tasks.telemetry_task]
         if split_tasks.impressions_count_task is not None:
             self._tasks.append(split_tasks.impressions_count_task)
         if split_tasks.unique_keys_task is not None:
             self._tasks.append(split_tasks.unique_keys_task)
         if split_tasks.clear_filter_task is not None:
             self._tasks.append(split_tasks.clear_filter_task)
+        self._periodic_data_recording_tasks = [
+            split_tasks.telemetry_task
+        ]
 
     def sync_all(self):
         """

splitio/sync/telemetry.py

@@ -1,6 +1,4 @@
 import json
-import logging
-_LOGGER = logging.getLogger(__name__)
 
 from splitio.api.telemetry import TelemetryAPI
 from splitio.engine.telemetry import TelemetryStorageConsumer
@@ -49,4 +47,4 @@ def _build_stats(self):
         }
         merged_dict.update(self._telemetry_runtime_consumer.pop_formatted_stats())
         merged_dict.update(self._telemetry_evaluation_consumer.pop_formatted_stats())
-        return merged_dict
+        return merged_dict