Minor fixes and updated/created tests

Author: chillaq

splitio/api/auth.py

@@ -7,7 +7,7 @@
 from splitio.api.commons import headers_from_metadata, record_telemetry, get_current_epoch_time
 from splitio.api.client import HttpClientException
 from splitio.models.token import from_raw
-from splitio.models.telemetry import TOKEN
+from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -46,7 +46,7 @@ def authenticate(self):
                 self._apikey,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, TOKEN, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.TOKEN, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 payload = json.loads(response.body)
                 return from_raw(payload)

splitio/api/client.py

@@ -6,7 +6,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 HttpResponse = namedtuple('HttpResponse', ['status_code', 'body'])
-
+HTTP_TIMEOUT = 1500
 
 class HttpClientException(Exception):
     """HTTP Client exception."""
@@ -44,7 +44,7 @@ def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, t
         :param telemetry_url: Optional alternative telemetry URL.
         :type telemetry_url: str
         """
-        self._timeout = timeout/1000 if timeout else None  # Convert ms to seconds.
+        self._timeout = timeout/1000 if timeout else HTTP_TIMEOUT  # Convert ms to seconds.
         self._urls = {
             'sdk': sdk_url if sdk_url is not None else self.SDK_URL,
             'events': events_url if events_url is not None else self.EVENTS_URL,

splitio/api/events.py

@@ -5,7 +5,7 @@
 from splitio.api import APIException
 from splitio.api.client import HttpClientException
 from splitio.api.commons import headers_from_metadata, record_telemetry, get_current_epoch_time
-from splitio.models.telemetry import EVENT
+from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -73,7 +73,7 @@ def flush_events(self, events):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, EVENT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.EVENT, self._telemetry_runtime_producer)
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:

splitio/api/impressions.py

@@ -7,7 +7,7 @@
 from splitio.api.client import HttpClientException
 from splitio.api.commons import headers_from_metadata, record_telemetry, get_current_epoch_time
 from splitio.engine.impressions import ImpressionsMode
-from splitio.models.telemetry import IMPRESSION, IMPRESSION_COUNT
+from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -102,7 +102,7 @@ def flush_impressions(self, impressions):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, IMPRESSION, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.IMPRESSION, self._telemetry_runtime_producer)
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:
@@ -129,7 +129,7 @@ def flush_counters(self, counters):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, IMPRESSION_COUNT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.IMPRESSION_COUNT, self._telemetry_runtime_producer)
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:

splitio/api/segments.py

@@ -7,7 +7,7 @@
 from splitio.api import APIException
 from splitio.api.commons import headers_from_metadata, build_fetch, record_telemetry, get_current_epoch_time
 from splitio.api.client import HttpClientException
-from splitio.models.telemetry import SEGMENT
+from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -59,7 +59,7 @@ def fetch_segment(self, segment_name, change_number, fetch_options):
                 extra_headers=extra_headers,
                 query=query,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, SEGMENT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.SEGMENT, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)
             else:

splitio/api/splits.py

@@ -7,7 +7,7 @@
 from splitio.api import APIException
 from splitio.api.commons import headers_from_metadata, build_fetch, record_telemetry, get_current_epoch_time
 from splitio.api.client import HttpClientException
-from splitio.models.telemetry import SPLIT
+from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -54,7 +54,7 @@ def fetch_splits(self, change_number, fetch_options):
                 extra_headers=extra_headers,
                 query=query,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, SPLIT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.SPLIT, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)
             else:

splitio/api/telemetry.py

@@ -5,7 +5,7 @@
 from splitio.api import APIException
 from splitio.api.client import HttpClientException
 from splitio.api.commons import headers_from_metadata, record_telemetry, get_current_epoch_time
-from splitio.models.telemetry import TELEMETRY
+from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -42,7 +42,7 @@ def record_unique_keys(self, uniques):
                 body=uniques,
                 extra_headers=self._metadata
             )
-            record_telemetry(response.status_code,  get_current_epoch_time() - start, TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code,  get_current_epoch_time() - start, HTTPExceptionsAndLatencies.TELEMETRY, self._telemetry_runtime_producer)
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:
@@ -68,7 +68,7 @@ def record_init(self, configs):
                 body=configs,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.TELEMETRY, self._telemetry_runtime_producer)
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:
@@ -94,7 +94,7 @@ def record_stats(self, stats):
                 body=stats,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.TELEMETRY, self._telemetry_runtime_producer)
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:

splitio/client/client.py

@@ -5,7 +5,7 @@
 from splitio.engine.splitters import Splitter
 from splitio.models.impressions import Impression, Label
 from splitio.models.events import Event, EventWrapper
-from splitio.models.telemetry import get_latency_bucket_index, TRACK
+from splitio.models.telemetry import get_latency_bucket_index, MethodExceptionsAndLatencies
 from splitio.client import input_validator
 from splitio.util import utctime_ms
 from splitio.api.commons import get_current_epoch_time
@@ -124,7 +124,8 @@ 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(method_name[4:])
+            if not self._telemetry_evaluation_producer == None:
+                self._telemetry_evaluation_producer.record_exception(method_name[4:])
             try:
                 impression = self._build_impression(
                     matching_key,
@@ -207,12 +208,15 @@ def _make_evaluations(self, key, features, attributes, method_name, metric_name)
                 _LOGGER.error('%s: An exception when trying to store '
                               'impressions.' % method_name)
                 _LOGGER.debug('Error: ', exc_info=True)
-                self._telemetry_evaluation_producer.record_exception(method_name[4:])
+                if not self._telemetry_evaluation_producer == None:
+                    self._telemetry_evaluation_producer.record_exception(method_name[4:])
 
-            self._telemetry_evaluation_producer.record_latency(method_name[4:], get_current_epoch_time() - start)
+            if not self._telemetry_evaluation_producer == None:
+                self._telemetry_evaluation_producer.record_latency(method_name[4:], get_current_epoch_time() - start)
             return treatments
         except Exception:  # pylint: disable=broad-except
-            self._telemetry_evaluation_producer.record_exception(method_name)
+            if not self._telemetry_evaluation_producer == None:
+                self._telemetry_evaluation_producer.record_exception(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)
@@ -350,7 +354,7 @@ def _record_stats(self, impressions, start, operation, method_name=None):
         end = get_current_epoch_time()
         self._recorder.record_treatment_stats(impressions, get_latency_bucket_index(end - start),
                                               operation)
-        if not method_name == None:
+        if not method_name == None and not self._telemetry_evaluation_producer == None:
             self._telemetry_evaluation_producer.record_latency(method_name[4:], end - start)
 
 
@@ -413,9 +417,11 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
                 event=event,
                 size=size,
             )])
-            self._telemetry_evaluation_producer.record_latency(TRACK, get_current_epoch_time() - start)
+            if not self._telemetry_evaluation_producer == None:
+                self._telemetry_evaluation_producer.record_latency(MethodExceptionsAndLatencies.TRACK, get_current_epoch_time() - start)
         except Exception:  # pylint: disable=broad-except
-            self._telemetry_evaluation_producer.record_exception(TRACK)
+            if not self._telemetry_evaluation_producer == None:
+                self._telemetry_evaluation_producer.record_exception(MethodExceptionsAndLatencies.TRACK)
             _LOGGER.error('Error processing track event')
             _LOGGER.debug('Error: ', exc_info=True)
 

splitio/client/factory.py

@@ -124,8 +124,11 @@ def __init__(  # pylint: disable=too-many-arguments
         self._sdk_internal_ready_flag = sdk_ready_flag
         self._recorder = recorder
         self._preforked_initialization = preforked_initialization
-        self._telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
-        self._telemetry_init_producer = telemetry_producer.get_telemetry_init_producer()
+        self._telemetry_evaluation_producer = None
+        self._telemetry_init_producer = None
+        if not telemetry_producer == None:
+            self._telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
+            self._telemetry_init_producer = telemetry_producer.get_telemetry_init_producer()
         self._telemetry_init_consumer = telemetry_init_consumer
         self._telemetry_api = telemetry_api
         self._ready_time = get_current_epoch_time()
@@ -156,13 +159,14 @@ def _update_status_when_ready(self):
         self._sdk_internal_ready_flag.wait()
         self._status = Status.READY
         self._sdk_ready_flag.set()
-        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)
+        if not self._telemetry_init_producer == None:
+            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()
+            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 _get_storage(self, name):
@@ -329,7 +333,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     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_evaluation_producer=telemetry_producer.get_telemetry_evaluation_producer()
 
 
     http_client = HttpClient(
@@ -365,8 +369,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     imp_strategy = set_classes('MEMORY', cfg['impressionsMode'], apis)
 
     imp_manager = ImpressionsManager(
-        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
-        imp_strategy, telemetry_runtime_producer)
+        imp_strategy, telemetry_runtime_producer,
+        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata))
 
     synchronizers = SplitSynchronizers(
         SplitSynchronizer(apis['splits'], storages['splits']),
@@ -375,9 +379,9 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
                                cfg['impressionsBulkSize']),
         EventSynchronizer(apis['events'], storages['events'], cfg['eventsBulkSize']),
         impressions_count_sync,
+        TelemetrySynchronizer(telemetry_consumer, storages['splits'], storages['segments'], apis['telemetry']),
         unique_keys_synchronizer,
         clear_filter_sync,
-        TelemetrySynchronizer(telemetry_consumer, storages['splits'], storages['segments'], apis['telemetry'])
     )
 
     tasks = SplitTasks(
@@ -395,9 +399,9 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         ),
         EventsSyncTask(synchronizers.events_sync.synchronize_events, cfg['eventsPushRate']),
         impressions_count_task,
+        TelemetrySyncTask(synchronizers.telemetry_sync.synchronize_stats, cfg['metricsRefreshRate']),
         unique_keys_task,
         clear_filter_task,
-        TelemetrySyncTask(synchronizers.telemetry_sync.synchronize_stats, cfg['metricsRefreshRate']),
     )
 
     synchronizer = Synchronizer(synchronizers, tasks)
@@ -421,7 +425,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         synchronizer.sync_all()
         synchronizer._split_synchronizers._segment_sync.shutdown()
         return SplitFactory(api_key, storages, cfg['labelsEnabled'],
-                            recorder, manager, preforked_initialization=preforked_initialization)
+                            recorder, manager, None, telemetry_producer, telemetry_consumer.get_telemetry_init_consumer(), apis['telemetry'], preforked_initialization=preforked_initialization)
 
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer")
     initialization_thread.setDaemon(True)
@@ -444,6 +448,11 @@ def _build_redis_factory(api_key, cfg):
         'impressions': RedisImpressionsStorage(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:
         _LOGGER.warning("dataSampling cannot be less than %.2f, defaulting to minimum",
@@ -455,17 +464,21 @@ def _build_redis_factory(api_key, cfg):
     imp_strategy = set_classes('REDIS', cfg['impressionsMode'], redis_adapter)
 
     imp_manager = ImpressionsManager(
+        imp_strategy,
+        telemetry_runtime_producer,
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
-        imp_strategy)
+        )
 
     synchronizers = SplitSynchronizers(None, None, None, None,
         impressions_count_sync,
+        None,
         unique_keys_synchronizer,
         clear_filter_sync
     )
 
     tasks = SplitTasks(None, None, None, None,
         impressions_count_task,
+        None,
         unique_keys_task,
         clear_filter_task
     )
@@ -490,9 +503,11 @@ def _build_redis_factory(api_key, cfg):
         cfg['labelsEnabled'],
         recorder,
         manager,
+        sdk_ready_flag=None,
+        telemetry_producer=telemetry_producer,
+        telemetry_init_consumer=telemetry_consumer.get_telemetry_init_consumer()
     )
 
-
 def _build_localhost_factory(cfg):
     """Build and return a localhost factory for testing/development purposes."""
     storages = {
@@ -520,7 +535,7 @@ def _build_localhost_factory(cfg):
     manager = Manager(ready_event, synchronizer, None, False, sdk_metadata)
     manager.start()
     recorder = StandardRecorder(
-        ImpressionsManager(None, StrategyDebugMode()),
+        ImpressionsManager(StrategyDebugMode()),
         storages['events'],
         storages['impressions'],
     )
@@ -530,7 +545,7 @@ def _build_localhost_factory(cfg):
         False,
         recorder,
         manager,
-        ready_event
+        ready_event,
     )
 
 def get_factory(api_key, **kwargs):