Telemetry producer refactor

Author: chillaq

splitio/client/factory.py

@@ -16,7 +16,7 @@
 from splitio.engine.impressions.manager import Counter as ImpressionsCounter
 from splitio.engine.impressions.strategies import StrategyNoneMode, StrategyDebugMode, StrategyOptimizedMode
 from splitio.engine.impressions.adapters import InMemorySenderAdapter, RedisSenderAdapter
-from splitio.engine.telemetry import TelemetryStorageProducer, TelemetryStorageConsumer
+from splitio.engine.telemetry import TelemetryStorageProducer, TelemetryStorageConsumer, RedisTelemetryInitProducer, RedisTelemetryStorageProducer
 
 # Storage
 from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
@@ -161,7 +161,7 @@ def _update_redis_init(self):
         """Push Config Telemetry into redis storage"""
         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)
-        self._telemetry_init_producer.get_telemetry_storage().push_config_stats()
+        self._telemetry_init_producer.push_config()
 
     def _update_status_when_ready(self):
         """Wait until the sdk is ready and update the status."""
@@ -449,7 +449,7 @@ def _build_redis_factory(api_key, cfg):
     cache_enabled = cfg.get('redisLocalCacheEnabled', False)
     cache_ttl = cfg.get('redisLocalCacheTTL', 5)
     telemetry_storage = RedisTelemetryStorage(redis_adapter, sdk_metadata)
-    telemetry_producer = TelemetryStorageProducer(telemetry_storage)
+    telemetry_producer = RedisTelemetryStorageProducer(telemetry_storage)
     telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
     telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
     storages = {

splitio/engine/telemetry.py

@@ -51,9 +51,6 @@ def record_not_ready_usage(self):
     def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
         self._telemetry_storage.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
 
-    def get_telemetry_storage(self):
-        return self._telemetry_storage
-
 class TelemetryEvaluationProducer(object):
     """Telemetry evaluation producer class."""
 
@@ -295,3 +292,54 @@ def pop_formatted_stats(self):
                       } for event in self.pop_streaming_events()['streamingEvents']],
             'sL': self.get_session_length()
         }
+
+class RedisTelemetryStorageProducer(object):
+    """Telemetry storage producer class."""
+
+    def __init__(self, telemetry_storage):
+        """Initialize all producer classes."""
+        self._telemetry_init_producer = RedisTelemetryInitProducer(telemetry_storage)
+        self._telemetry_evaluation_producer = TelemetryEvaluationProducer(telemetry_storage)
+        self._telemetry_runtime_producer = TelemetryRuntimeProducer(telemetry_storage)
+
+    def get_telemetry_init_producer(self):
+        """get init producer instance."""
+        return self._telemetry_init_producer
+
+    def get_telemetry_evaluation_producer(self):
+        """get evaluation producer instance."""
+        return self._telemetry_evaluation_producer
+
+    def get_telemetry_runtime_producer(self):
+        """get runtime producer instance."""
+        return self._telemetry_runtime_producer
+
+class RedisTelemetryInitProducer(object):
+    """Telemetry init producer class."""
+
+    def __init__(self, telemetry_storage):
+        """Constructor."""
+        self._telemetry_storage = telemetry_storage
+
+    def record_config(self, config, extra_config):
+        """Record configurations."""
+        self._telemetry_storage.record_config(config, extra_config)
+
+    def push_config(self):
+        """Record configurations."""
+        self._telemetry_storage.push_config_stats()
+
+    def record_ready_time(self, ready_time):
+        """Record ready time."""
+        pass
+
+    def record_bur_time_out(self):
+        """Record block until ready timeout."""
+        pass
+
+    def record_not_ready_usage(self):
+        """record non-ready usage."""
+        pass
+
+    def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
+        self._telemetry_storage.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)

splitio/storage/adapters/redis.py

@@ -1,11 +1,7 @@
 """Redis client wrapper with prefix support."""
 from builtins import str
 
-import logging
-_LOGGER = logging.getLogger(__name__)
-
 from splitio.version import __version__
-from splitio.util.host_info import get_ip, get_hostname
 
 try:
     from redis import StrictRedis
@@ -20,8 +16,6 @@ def missing_redis_dependencies(*_, **__):
         )
     StrictRedis = Sentinel = missing_redis_dependencies
 
-TELEMETRY_CONFIG_KEY = 'SPLITIO.telemetry.init'
-
 class RedisAdapterException(Exception):
     """Exception to be thrown when a redis command fails with an exception."""
 
@@ -311,15 +305,6 @@ def pipeline(self):
         except RedisError as exc:
             raise RedisAdapterException('Error executing ttl operation') from exc
 
-    def record_init(self, *values):
-        """Write config init values to redis."""
-        try:
-            host_ip = get_ip()
-            host_name = get_hostname()
-            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
-
 class RedisPipelineAdapter(object):
     """
     Instance decorator for Redis Pipeline.

splitio/storage/redis.py

@@ -583,6 +583,7 @@ def expire_keys(self, total_keys, inserted):
 class RedisTelemetryStorage(TelemetryStorage):
     """Redis based telemetry storage class."""
 
+    _TELEMETRY_CONFIG_KEY = 'SPLITIO.telemetry.init'
     _TELEMETRY_LATENCIES_KEY = 'SPLITIO.telemetry.latencies'
     _TELEMETRY_EXCEPTIONS_KEY = 'SPLITIO.telemetry.exceptions'
     _TELEMETRY_KEY_DEFAULT_TTL = 3600
@@ -616,7 +617,9 @@ def record_config(self, config, extra_config):
 
     def push_config_stats(self):
         """push config stats to redis."""
-        self._redis_client.record_init(self._format_config_stats())
+        host_ip = get_ip()
+        host_name = get_hostname()
+        self._redis_client.hset(self._TELEMETRY_CONFIG_KEY, 'python-' + __version__ + '/' + host_name+ '/' + host_ip, str(self._format_config_stats()))
 
     def _format_config_stats(self):
         """format only selected config stats to json"""