refactor redis telemetry storage

Author: chillaq

splitio/client/client.py

@@ -1,6 +1,7 @@
 """A module for Split.io SDK API clients."""
 import logging
-import time
+
+from splitio.client.util import get_method_constant
 from splitio.engine.evaluator import Evaluator, CONTROL
 from splitio.engine.splitters import Splitter
 from splitio.models.impressions import Impression, Label
@@ -123,7 +124,7 @@ def _make_evaluation(self, key, feature, attributes, method_name, metric_name):
         except Exception:  # pylint: disable=broad-except
             _LOGGER.error('Error getting treatment for feature')
             _LOGGER.debug('Error: ', exc_info=True)
-            self._telemetry_evaluation_producer.record_exception(self._get_method_constant(method_name[4:]))
+            self._telemetry_evaluation_producer.record_exception(get_method_constant(method_name[4:]))
             try:
                 impression = self._build_impression(
                     matching_key,
@@ -200,18 +201,18 @@ def _make_evaluations(self, key, features, attributes, method_name, metric_name)
                     self._record_stats(
                         [(i, attributes) for i in bulk_impressions],
                         start,
-                        metric_name
+                        metric_name,
+                        method_name
                     )
             except Exception:  # pylint: disable=broad-except
                 _LOGGER.error('%s: An exception when trying to store '
                               'impressions.' % method_name)
                 _LOGGER.debug('Error: ', exc_info=True)
-                self._telemetry_evaluation_producer.record_exception(self._get_method_constant(method_name[4:]))
+                self._telemetry_evaluation_producer.record_exception(get_method_constant(method_name[4:]))
 
-            self._telemetry_evaluation_producer.record_latency(self._get_method_constant(method_name[4:]), get_current_epoch_time_ms() - start)
             return treatments
         except Exception:  # pylint: disable=broad-except
-            self._telemetry_evaluation_producer.record_exception(self._get_method_constant(method_name[4:]))
+            self._telemetry_evaluation_producer.record_exception(get_method_constant(method_name[4:]))
             _LOGGER.error('Error getting treatment for features')
             _LOGGER.debug('Error: ', exc_info=True)
         return input_validator.generate_control_treatments(list(features), method_name)
@@ -348,10 +349,7 @@ def _record_stats(self, impressions, start, operation, method_name=None):
         """
         end = get_current_epoch_time_ms()
         self._recorder.record_treatment_stats(impressions, get_latency_bucket_index(end - start),
-                                              operation)
-        if method_name is not None:
-            self._telemetry_evaluation_producer.record_latency(self._get_method_constant(method_name[4:]), end - start)
-
+                                              operation, method_name)
 
     def track(self, key, traffic_type, event_type, value=None, properties=None):
         """
@@ -411,21 +409,11 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
             return_flag = self._recorder.record_track_stats([EventWrapper(
                 event=event,
                 size=size,
-            )])
-            self._telemetry_evaluation_producer.record_latency(MethodExceptionsAndLatencies.TRACK, get_current_epoch_time_ms() - start)
+            )], get_current_epoch_time_ms() - start)
             return return_flag
         except Exception:  # pylint: disable=broad-except
             self._telemetry_evaluation_producer.record_exception(MethodExceptionsAndLatencies.TRACK)
             _LOGGER.error('Error processing track event')
             _LOGGER.debug('Error: ', exc_info=True)
             return False
 
-    def _get_method_constant(self, method):
-        if method == 'treatment':
-            return MethodExceptionsAndLatencies.TREATMENT
-        elif method == 'treatments':
-            return MethodExceptionsAndLatencies.TREATMENTS
-        elif method == 'treatment_with_config':
-            return MethodExceptionsAndLatencies.TREATMENT_WITH_CONFIG
-        elif method == 'treatments_with_config':
-            return MethodExceptionsAndLatencies.TREATMENTS_WITH_CONFIG

splitio/client/factory.py

@@ -23,7 +23,7 @@
     InMemoryImpressionStorage, InMemoryEventStorage, InMemoryTelemetryStorage, LocalhostTelemetryStorage
 from splitio.storage.adapters import redis
 from splitio.storage.redis import RedisSplitStorage, RedisSegmentStorage, RedisImpressionsStorage, \
-    RedisEventsStorage
+    RedisEventsStorage, RedisTelemetryStorage
 
 # APIs
 from splitio.api.client import HttpClient
@@ -152,19 +152,9 @@ 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_ms() - 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()        
+            #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)
 
     def _update_status_when_ready(self):
         """Wait until the sdk is ready and update the status."""
@@ -343,8 +333,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     telemetry_storage = InMemoryTelemetryStorage()
     telemetry_producer = TelemetryStorageProducer(telemetry_storage)
     telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
-    telemetry_runtime_producer=telemetry_producer.get_telemetry_runtime_producer()
-#    telemetry_evaluation_producer=telemetry_producer.get_telemetry_evaluation_producer()
+    telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+    telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
 
 
     http_client = HttpClient(
@@ -430,6 +420,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         imp_manager,
         storages['events'],
         storages['impressions'],
+        telemetry_evaluation_producer
     )
 
     if preforked_initialization:
@@ -453,16 +444,16 @@ def _build_redis_factory(api_key, cfg):
     redis_adapter = redis.build(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_consumer = TelemetryStorageConsumer(telemetry_storage)
+    telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
     storages = {
         'splits': RedisSplitStorage(redis_adapter, cache_enabled, cache_ttl),
         'segments': RedisSegmentStorage(redis_adapter),
         'impressions': RedisImpressionsStorage(redis_adapter, sdk_metadata),
-        'events': RedisEventsStorage(redis_adapter, sdk_metadata),
+        'events': RedisEventsStorage(redis_adapter, sdk_metadata)
     }
-    telemetry_storage = InMemoryTelemetryStorage()
-    telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-    telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
-    telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
 
     data_sampling = cfg.get('dataSampling', DEFAULT_DATA_SAMPLING)
     if data_sampling < _MIN_DEFAULT_DATA_SAMPLING_ALLOWED:
@@ -482,14 +473,14 @@ def _build_redis_factory(api_key, cfg):
 
     synchronizers = SplitSynchronizers(None, None, None, None,
         impressions_count_sync,
-        TelemetrySynchronizer(telemetry_consumer, storages['splits'], storages['segments'], redis_adapter),
+        None,
         unique_keys_synchronizer,
         clear_filter_sync
     )
 
     tasks = SplitTasks(None, None, None, None,
         impressions_count_task,
-        TelemetrySyncTask(synchronizers.telemetry_sync.synchronize_stats, cfg['metricsRefreshRate']),
+        None,
         unique_keys_task,
         clear_filter_task
     )
@@ -500,6 +491,7 @@ def _build_redis_factory(api_key, cfg):
         imp_manager,
         storages['events'],
         storages['impressions'],
+        telemetry_storage,
         data_sampling,
     )
 
@@ -528,6 +520,7 @@ def _build_localhost_factory(cfg):
     telemetry_producer = TelemetryStorageProducer(telemetry_storage)
     telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
     telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+    telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
 
     storages = {
         'splits': InMemorySplitStorage(),
@@ -557,6 +550,7 @@ def _build_localhost_factory(cfg):
         ImpressionsManager(StrategyDebugMode(), telemetry_runtime_producer),
         storages['events'],
         storages['impressions'],
+        telemetry_evaluation_producer
     )
     return SplitFactory(
         'localhost',

splitio/client/util.py

@@ -1,42 +1,25 @@
 """General purpose SDK utilities."""
 
-import socket
 from collections import namedtuple
 from splitio.version import __version__
+from splitio.util.host_info import get_hostname, get_ip
+
+from splitio.models.telemetry import MethodExceptionsAndLatencies
 
 SdkMetadata = namedtuple(
     'SdkMetadata',
     ['sdk_version', 'instance_name', 'instance_ip']
 )
 
-
-def _get_ip():
-    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    try:
-        # doesn't even have to be reachable
-        sock.connect(('10.255.255.255', 1))
-        ip_address = sock.getsockname()[0]
-    except Exception:  # pylint: disable=broad-except
-        ip_address = 'unknown'
-    finally:
-        sock.close()
-    return ip_address
-
-
-def _get_hostname(ip_address):
-    return 'unknown' if ip_address == 'unknown' else 'ip-' + ip_address.replace('.', '-')
-
-
 def _get_hostname_and_ip(config):
     if config.get('IPAddressesEnabled') is False:
         return 'NA', 'NA'
     ip_from_config = config.get('machineIp')
     machine_from_config = config.get('machineName')
-    ip_address = ip_from_config if ip_from_config is not None else _get_ip()
-    hostname = machine_from_config if machine_from_config is not None else _get_hostname(ip_address)
+    ip_address = ip_from_config if ip_from_config is not None else get_ip()
+    hostname = machine_from_config if machine_from_config is not None else get_hostname()
     return ip_address, hostname
 
-
 def get_metadata(config):
     """
     Gather SDK metadata and return a tuple with such info.
@@ -50,3 +33,15 @@ def get_metadata(config):
     version = 'python-%s' % __version__
     ip_address, hostname = _get_hostname_and_ip(config)
     return SdkMetadata(version, hostname, ip_address)
+
+def get_method_constant(method):
+    if method == 'treatment':
+        return MethodExceptionsAndLatencies.TREATMENT
+    elif method == 'treatments':
+        return MethodExceptionsAndLatencies.TREATMENTS
+    elif method == 'treatment_with_config':
+        return MethodExceptionsAndLatencies.TREATMENT_WITH_CONFIG
+    elif method == 'treatments_with_config':
+        return MethodExceptionsAndLatencies.TREATMENTS_WITH_CONFIG
+    elif method == 'track':
+        return MethodExceptionsAndLatencies.TRACK

splitio/recorder/recorder.py

@@ -3,9 +3,9 @@
 import logging
 import random
 
-
+from splitio.client.util import get_method_constant
 from splitio.client.config import DEFAULT_DATA_SAMPLING
-
+from splitio.models.telemetry import MethodExceptionsAndLatencies
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -41,7 +41,7 @@ def record_track_stats(self, events):
 class StandardRecorder(StatsRecorder):
     """StandardRecorder class."""
 
-    def __init__(self, impressions_manager, event_storage, impression_storage):
+    def __init__(self, impressions_manager, event_storage, impression_storage, telemetry_evaluation_producer):
         """
         Class constructor.
 
@@ -55,8 +55,9 @@ def __init__(self, impressions_manager, event_storage, impression_storage):
         self._impressions_manager = impressions_manager
         self._event_sotrage = event_storage
         self._impression_storage = impression_storage
+        self._telemetry_evaluation_producer = telemetry_evaluation_producer
 
-    def record_treatment_stats(self, impressions, latency, operation):
+    def record_treatment_stats(self, impressions, latency, operation, method_name):
         """
         Record stats for treatment evaluation.
 
@@ -68,27 +69,30 @@ def record_treatment_stats(self, impressions, latency, operation):
         :type operation: str
         """
         try:
+            if method_name is not None:
+                self._telemetry_evaluation_producer.record_latency(get_method_constant(method_name[4:]), latency)
             impressions = self._impressions_manager.process_impressions(impressions)
             self._impression_storage.put(impressions)
         except Exception:  # pylint: disable=broad-except
             _LOGGER.error('Error recording impressions')
             _LOGGER.debug('Error: ', exc_info=True)
 
-    def record_track_stats(self, event):
+    def record_track_stats(self, event, latency):
         """
         Record stats for tracking events.
 
         :param event: events tracked
         :type event: splitio.models.events.EventWrapper
         """
+        self._telemetry_evaluation_producer.record_latency(MethodExceptionsAndLatencies.TRACK, latency)
         return self._event_sotrage.put(event)
 
 
 class PipelinedRecorder(StatsRecorder):
     """PipelinedRecorder class."""
 
     def __init__(self, pipe, impressions_manager, event_storage,
-                 impression_storage, data_sampling=DEFAULT_DATA_SAMPLING):
+                 impression_storage, telemetry_redis_storage, data_sampling=DEFAULT_DATA_SAMPLING):
         """
         Class constructor.
 
@@ -108,8 +112,9 @@ def __init__(self, pipe, impressions_manager, event_storage,
         self._event_sotrage = event_storage
         self._impression_storage = impression_storage
         self._data_sampling = data_sampling
+        self._telemetry_redis_storage = telemetry_redis_storage
 
-    def record_treatment_stats(self, impressions, latency, operation):
+    def record_treatment_stats(self, impressions, latency, operation, method_name):
         """
         Record stats for treatment evaluation.
 
@@ -131,22 +136,23 @@ def record_treatment_stats(self, impressions, latency, operation):
             impressions = self._impressions_manager.process_impressions(impressions)
             if not impressions:
                 return
-            # pipe = self._make_pipe()
-            # self._impression_storage.add_impressions_to_pipe(impressions, pipe)
-            # self._telemetry_storage.add_latency_to_pipe(operation, latency, pipe)
-            # result = pipe.execute()
-            # if len(result) == 2:
-            #   self._impression_storage.expire_key(result[0], len(impressions))
-            self._impression_storage.put(impressions)
+            pipe = self._make_pipe()
+            self._impression_storage.add_impressions_to_pipe(impressions, pipe)
+            result = pipe.execute()
+            if len(result) == 2:
+                self._impression_storage.expire_key(result[0], len(impressions))
+            if method_name is not None:
+                self._telemetry_redis_storage.record_latency(method_name[4:], latency)
         except Exception:  # pylint: disable=broad-except
             _LOGGER.error('Error recording impressions')
             _LOGGER.debug('Error: ', exc_info=True)
 
-    def record_track_stats(self, event):
+    def record_track_stats(self, event, latency):
         """
         Record stats for tracking events.
 
         :param event: events tracked
         :type event: splitio.models.events.EventWrapper
         """
+        self._telemetry_redis_storage.record_latency(MethodExceptionsAndLatencies.TRACK.value, latency)
         return self._event_sotrage.put(event)