Merge pull request #280 from splitio/telemetry-redis

Telemetry redis

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
@@ -66,7 +66,7 @@
 _INSTANTIATED_FACTORIES = Counter()
 _INSTANTIATED_FACTORIES_LOCK = threading.RLock()
 _MIN_DEFAULT_DATA_SAMPLING_ALLOWED = 0.1  # 10%
-
+_MAX_RETRY_SYNC_ALL = 3
 
 class Status(Enum):
     """Factory Status."""
@@ -152,6 +152,9 @@ def _start_status_updater(self):
             ready_updater.start()
         else:
             self._status = Status.READY
+            #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."""
@@ -330,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(
@@ -352,9 +355,6 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata, telemetry_runtime_producer),
     }
 
-    if not input_validator.validate_apikey_type(apis['segments']):
-        return None
-
     storages = {
         'splits': InMemorySplitStorage(),
         'segments': InMemorySegmentStorage(),
@@ -417,10 +417,11 @@ 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:
-        synchronizer.sync_all()
+        synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
         synchronizer._split_synchronizers._segment_sync.shutdown()
         return SplitFactory(api_key, storages, cfg['labelsEnabled'],
                             recorder, manager, None, telemetry_producer, telemetry_consumer.get_telemetry_init_consumer(), apis['telemetry'], preforked_initialization=preforked_initialization)
@@ -440,16 +441,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:
@@ -487,6 +488,7 @@ def _build_redis_factory(api_key, cfg):
         imp_manager,
         storages['events'],
         storages['impressions'],
+        telemetry_storage,
         data_sampling,
     )
 
@@ -495,13 +497,16 @@ def _build_redis_factory(api_key, cfg):
     initialization_thread.setDaemon(True)
     initialization_thread.start()
 
+    telemetry_producer.get_telemetry_init_producer().record_config(cfg, {})
+
     return SplitFactory(
         api_key,
         storages,
         cfg['labelsEnabled'],
         recorder,
         manager,
         sdk_ready_flag=None,
+        telemetry_api=redis_adapter,
         telemetry_producer=telemetry_producer,
         telemetry_init_consumer=telemetry_consumer.get_telemetry_init_consumer()
     )
@@ -512,6 +517,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(),
@@ -541,6 +547,7 @@ def _build_localhost_factory(cfg):
         ImpressionsManager(StrategyDebugMode(), telemetry_runtime_producer),
         storages['events'],
         storages['impressions'],
+        telemetry_evaluation_producer
     )
     return SplitFactory(
         'localhost',

splitio/client/input_validator.py

@@ -448,31 +448,6 @@ def filter(self, record):
         return record.name not in ('SegmentsAPI', 'HttpClient')
 
 
-def validate_apikey_type(segment_api):
-    """
-    Try to guess if the apikey is of browser type and let the user know.
-
-    :param segment_api: Segments API client.
-    :type segment_api: splitio.api.segments.SegmentsAPI
-    """
-    api_messages_filter = _ApiLogFilter()
-    _logger = logging.getLogger('splitio.api.segments')
-    try:
-        _logger.addFilter(api_messages_filter)  # pylint: disable=protected-access
-        segment_api.fetch_segment('__SOME_INVALID_SEGMENT__', -1, FetchOptions())
-    except APIException as exc:
-        if exc.status_code == 403:
-            _LOGGER.error('factory instantiation: you passed a browser type '
-                          + 'api_key, please grab an api key from the Split '
-                          + 'console that is of type sdk')
-            return False
-    finally:
-        _logger.removeFilter(api_messages_filter)  # pylint: disable=protected-access
-
-    # True doesn't mean that the APIKEY is right, only that it's not of type "browser"
-    return True
-
-
 def validate_factory_instantiation(apikey):
     """
     Check if the factory if being instantiated with the appropriate arguments.

splitio/client/util.py

@@ -1,42 +1,41 @@
 """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
+
+_MAP_METHOD_TO_ENUM = {'treatment': MethodExceptionsAndLatencies.TREATMENT,
+                       'treatments': MethodExceptionsAndLatencies.TREATMENTS,
+                        'treatment_with_config': MethodExceptionsAndLatencies.TREATMENT_WITH_CONFIG,
+                        'treatments_with_config': MethodExceptionsAndLatencies.TREATMENTS_WITH_CONFIG,
+                        'track': MethodExceptionsAndLatencies.TRACK
+                       }
 
 SdkMetadata = namedtuple(
     'SdkMetadata',
     ['sdk_version', 'instance_name', 'instance_ip']
 )
 
+def _get_hostname_and_ip(config):
+    """
+    Get current hostname and IP address if config parameters are not set.
 
-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('.', '-')
-
+    :param config: User supplied config augmented with defaults.
+    :type config: dict
 
-def _get_hostname_and_ip(config):
+    :return: IP address and Hostname
+    :rtype: Tuple (str, str)
+    """
     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 +49,12 @@ 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):
+    """
+    Get method name mapped to the Method Enum object
+
+    :return: method name
+    :rtype: str
+    """
+    return _MAP_METHOD_TO_ENUM[method]

splitio/engine/impressions/__init__.py

@@ -14,19 +14,21 @@ def set_classes(storage_mode, impressions_mode, api_adapter):
     clear_filter_task = None
     impressions_count_sync = None
     impressions_count_task = None
+    sender_adapter = None
     if storage_mode == 'REDIS':
-        redis_sender_adapter = RedisSenderAdapter(api_adapter)
-        api_telemetry_adapter = redis_sender_adapter
-        api_impressions_adapter = redis_sender_adapter
+        sender_adapter = RedisSenderAdapter(api_adapter)
+        api_telemetry_adapter = sender_adapter
+        api_impressions_adapter = sender_adapter
     else:
         api_telemetry_adapter = api_adapter['telemetry']
         api_impressions_adapter = api_adapter['impressions']
+        sender_adapter = InMemorySenderAdapter(api_telemetry_adapter)
 
     if impressions_mode == ImpressionsMode.NONE:
         imp_counter = ImpressionsCounter()
         imp_strategy = StrategyNoneMode(imp_counter)
         clear_filter_sync = ClearFilterSynchronizer(imp_strategy.get_unique_keys_tracker())
-        unique_keys_synchronizer = UniqueKeysSynchronizer(InMemorySenderAdapter(api_telemetry_adapter), imp_strategy.get_unique_keys_tracker())
+        unique_keys_synchronizer = UniqueKeysSynchronizer(sender_adapter, imp_strategy.get_unique_keys_tracker())
         unique_keys_task = UniqueKeysSyncTask(unique_keys_synchronizer.send_all)
         clear_filter_task = ClearFilterSyncTask(clear_filter_sync.clear_all)
         imp_strategy.get_unique_keys_tracker().set_queue_full_hook(unique_keys_task.flush)

splitio/engine/impressions/adapters.py

@@ -91,10 +91,16 @@ def flush_counters(self, to_send):
         :param uniques: unique keys disctionary
         :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
         """
-        bulk_counts = self._build_counters(to_send)
         try:
-            inserted = self._redis_client.rpush(self.IMP_COUNT_QUEUE_KEY, *bulk_counts)
-            self._expire_keys(self.IMP_COUNT_QUEUE_KEY, self.IMP_COUNT_KEY_DEFAULT_TTL, inserted, len(bulk_counts))
+            resulted = 0
+            counted = 0
+            pipe = self._redis_client.pipeline()
+            for pf_count in to_send:
+                pipe.hincrby(self.IMP_COUNT_QUEUE_KEY, pf_count.feature + "::" + str(pf_count.timeframe), pf_count.count)
+                counted += pf_count.count
+            resulted = sum(pipe.execute())
+            self._expire_keys(self.IMP_COUNT_QUEUE_KEY,
+                              self.IMP_COUNT_KEY_DEFAULT_TTL, resulted, counted)
             return True
         except RedisAdapterException:
             _LOGGER.error('Something went wrong when trying to add counters to redis')
@@ -124,23 +130,3 @@ def _uniques_formatter(self, uniques):
         :rtype: json
         """
         return [json.dumps({'f': feature, 'ks': list(keys)}) for feature, keys in uniques.items()]
-
-    def _build_counters(self, counters):
-        """
-        Build an impression bulk formatted as the API expects it.
-
-        :param counters: List of impression counters per feature.
-        :type counters: list[splitio.engine.impressions.Counter.CountPerFeature]
-
-        :return: dict with list of impression count dtos
-        :rtype: dict
-        """
-        return json.dumps({
-            'pf': [
-                {
-                    'f': pf_count.feature,
-                    'm': pf_count.timeframe,
-                    'rc': pf_count.count
-                } for pf_count in counters
-            ]
-        })