Updating redis-telemetry PR

Author: chillaq

splitio/api/auth.py

@@ -5,7 +5,7 @@
 
 from splitio.api import APIException
 from splitio.api.commons import headers_from_metadata, record_telemetry
-from splitio.util.time import get_current_epoch_time
+from splitio.util.time import get_current_epoch_time_ms
 from splitio.api.client import HttpClientException
 from splitio.models.token import from_raw
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
@@ -39,15 +39,15 @@ def authenticate(self):
         :return: Json representation of an authentication.
         :rtype: splitio.models.token.Token
         """
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             response = self._client.get(
                 'auth',
                 '/v2/auth',
                 self._apikey,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.TOKEN, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time_ms() - start, HTTPExceptionsAndLatencies.TOKEN, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 payload = json.loads(response.body)
                 return from_raw(payload)

splitio/api/commons.py

@@ -1,5 +1,5 @@
 """Commons module."""
-from splitio.util.time import get_current_epoch_time
+from splitio.util.time import get_current_epoch_time_ms
 
 _CACHE_CONTROL = 'Cache-Control'
 _CACHE_CONTROL_NO_CACHE = 'no-cache'
@@ -50,7 +50,7 @@ def record_telemetry(status_code, elapsed, metric_name, telemetry_runtime_produc
     """
     telemetry_runtime_producer.record_sync_latency(metric_name, elapsed)
     if 200 <= status_code < 300:
-        telemetry_runtime_producer.record_successful_sync(metric_name, get_current_epoch_time())
+        telemetry_runtime_producer.record_successful_sync(metric_name, get_current_epoch_time_ms())
         return
     telemetry_runtime_producer.record_sync_error(metric_name, status_code)
 

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
-from splitio.util.time import get_current_epoch_time
+from splitio.util.time import get_current_epoch_time_ms
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 
@@ -65,7 +65,7 @@ def flush_events(self, events):
         :rtype: bool
         """
         bulk = self._build_bulk(events)
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             response = self._client.post(
                 'events',
@@ -74,7 +74,7 @@ def flush_events(self, events):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.EVENT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time_ms() - 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

@@ -6,7 +6,7 @@
 from splitio.api import APIException
 from splitio.api.client import HttpClientException
 from splitio.api.commons import headers_from_metadata, record_telemetry
-from splitio.util.time import get_current_epoch_time
+from splitio.util.time import get_current_epoch_time_ms
 from splitio.engine.impressions import ImpressionsMode
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
@@ -94,7 +94,7 @@ def flush_impressions(self, impressions):
         :type impressions: list
         """
         bulk = self._build_bulk(impressions)
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             response = self._client.post(
                 'events',
@@ -103,7 +103,7 @@ def flush_impressions(self, impressions):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.IMPRESSION, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time_ms() - 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:
@@ -121,7 +121,7 @@ def flush_counters(self, counters):
         :type impressions: list
         """
         bulk = self._build_counters(counters)
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             response = self._client.post(
                 'events',
@@ -130,7 +130,7 @@ def flush_counters(self, counters):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.IMPRESSION_COUNT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time_ms() - 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

@@ -6,7 +6,7 @@
 
 from splitio.api import APIException
 from splitio.api.commons import headers_from_metadata, build_fetch, record_telemetry
-from splitio.util.time import get_current_epoch_time
+from splitio.util.time import get_current_epoch_time_ms
 from splitio.api.client import HttpClientException
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
@@ -50,7 +50,7 @@ def fetch_segment(self, segment_name, change_number, fetch_options):
         :return: Json representation of a segmentChange response.
         :rtype: dict
         """
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
             response = self._client.get(
@@ -60,7 +60,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, HTTPExceptionsAndLatencies.SEGMENT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time_ms() - start, HTTPExceptionsAndLatencies.SEGMENT, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)
             else:

splitio/api/splits.py

@@ -6,7 +6,7 @@
 
 from splitio.api import APIException
 from splitio.api.commons import headers_from_metadata, build_fetch, record_telemetry
-from splitio.util.time import get_current_epoch_time
+from splitio.util.time import get_current_epoch_time_ms
 from splitio.api.client import HttpClientException
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
@@ -45,7 +45,7 @@ def fetch_splits(self, change_number, fetch_options):
         :return: Json representation of a splitChanges response.
         :rtype: dict
         """
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
             response = self._client.get(
@@ -55,7 +55,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, HTTPExceptionsAndLatencies.SPLIT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time_ms() - 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
-from splitio.util.time import get_current_epoch_time
+from splitio.util.time import get_current_epoch_time_ms
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
 
 _LOGGER = logging.getLogger(__name__)
@@ -34,7 +34,7 @@ def record_unique_keys(self, uniques):
         :param uniques: Unique Keys
         :type json
         """
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             response = self._client.post(
                 'telemetry',
@@ -43,7 +43,7 @@ def record_unique_keys(self, uniques):
                 body=uniques,
                 extra_headers=self._metadata
             )
-            record_telemetry(response.status_code,  get_current_epoch_time() - start, HTTPExceptionsAndLatencies.TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code,  get_current_epoch_time_ms() - 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:
@@ -60,7 +60,7 @@ def record_init(self, configs):
         :param configs: configs
         :type json
         """
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             response = self._client.post(
                 'telemetry',
@@ -69,7 +69,7 @@ def record_init(self, configs):
                 body=configs,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time_ms() - 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:
@@ -86,7 +86,7 @@ def record_stats(self, stats):
         :param stats: stats
         :type json
         """
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         try:
             response = self._client.post(
                 'telemetry',
@@ -95,7 +95,7 @@ def record_stats(self, stats):
                 body=stats,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, get_current_epoch_time() - start, HTTPExceptionsAndLatencies.TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time_ms() - 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

@@ -7,7 +7,7 @@
 from splitio.models.events import Event, EventWrapper
 from splitio.models.telemetry import get_latency_bucket_index, MethodExceptionsAndLatencies
 from splitio.client import input_validator
-from splitio.util.time import get_current_epoch_time, utctime_ms
+from splitio.util.time import get_current_epoch_time_ms, utctime_ms
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -92,7 +92,7 @@ def _make_evaluation(self, key, feature, attributes, method_name, metric_name):
                 _LOGGER.error("Client is not ready - no calls possible")
                 return CONTROL, None
 
-            start = get_current_epoch_time()
+            start = get_current_epoch_time_ms()
 
             matching_key, bucketing_key = input_validator.validate_key(key, method_name)
             feature = input_validator.validate_feature_name(
@@ -148,7 +148,7 @@ def _make_evaluations(self, key, features, attributes, method_name, metric_name)
             _LOGGER.error("Client is not ready - no calls possible")
             return input_validator.generate_control_treatments(features, method_name)
 
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
 
         matching_key, bucketing_key = input_validator.validate_key(key, method_name)
         if matching_key is None and bucketing_key is None:
@@ -208,7 +208,7 @@ def _make_evaluations(self, key, features, attributes, method_name, metric_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_latency(self._get_method_constant(method_name[4:]), get_current_epoch_time() - start)
+            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:]))
@@ -346,7 +346,7 @@ def _record_stats(self, impressions, start, operation, method_name=None):
         :param operation: operation performed.
         :type operation: str
         """
-        end = get_current_epoch_time()
+        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:
@@ -381,7 +381,7 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
             _LOGGER.warn("track: the SDK is not ready, results may be incorrect. Make sure to wait for SDK readiness before using this method")
             self._telemetry_init_producer.record_not_ready_usage()
 
-        start = get_current_epoch_time()
+        start = get_current_epoch_time_ms()
         key = input_validator.validate_track_key(key)
         event_type = input_validator.validate_event_type(event_type)
         should_validate_existance = self.ready and self._factory._apikey != 'localhost'  # pylint: disable=protected-access
@@ -412,7 +412,7 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
                 event=event,
                 size=size,
             )])
-            self._telemetry_evaluation_producer.record_latency(MethodExceptionsAndLatencies.TRACK, get_current_epoch_time() - start)
+            self._telemetry_evaluation_producer.record_latency(MethodExceptionsAndLatencies.TRACK, get_current_epoch_time_ms() - start)
             return return_flag
         except Exception:  # pylint: disable=broad-except
             self._telemetry_evaluation_producer.record_exception(MethodExceptionsAndLatencies.TRACK)

splitio/client/factory.py

@@ -33,7 +33,7 @@
 from splitio.api.events import EventsAPI
 from splitio.api.auth import AuthAPI
 from splitio.api.telemetry import TelemetryAPI, LocalhostTelemetryAPI
-from splitio.util.time import get_current_epoch_time
+from splitio.util.time import get_current_epoch_time_ms
 
 # Tasks
 from splitio.tasks.split_sync import SplitSynchronizationTask
@@ -130,7 +130,7 @@ def __init__(  # pylint: disable=too-many-arguments
         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()
+        self._ready_time = get_current_epoch_time_ms()
         self._start_status_updater()
 
     def _start_status_updater(self):
@@ -157,22 +157,21 @@ def _start_status_updater(self):
             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)
+        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()
+        config_post_thread.start()        
 
     def _update_status_when_ready(self):
         """Wait until the sdk is ready and update the status."""
         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)
+        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)
 
@@ -463,7 +462,7 @@ def _build_redis_factory(api_key, cfg):
     telemetry_storage = InMemoryTelemetryStorage()
     telemetry_producer = TelemetryStorageProducer(telemetry_storage)
     telemetry_consumer = TelemetryStorageConsumer(telemetry_storage)
-    telemetry_runtime_producer=telemetry_producer.get_telemetry_runtime_producer()
+    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:
@@ -508,7 +507,7 @@ 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(
@@ -618,7 +617,7 @@ def _get_active_and_redundant_count():
     active_factory_count = 0
     _INSTANTIATED_FACTORIES_LOCK.acquire()
     for item in _INSTANTIATED_FACTORIES:
-        redundant_factory_count = redundant_factory_count + _INSTANTIATED_FACTORIES[item] - 1
-        active_factory_count = active_factory_count + _INSTANTIATED_FACTORIES[item]
+        redundant_factory_count += _INSTANTIATED_FACTORIES[item] - 1
+        active_factory_count += _INSTANTIATED_FACTORIES[item]
     _INSTANTIATED_FACTORIES_LOCK.release()
     return redundant_factory_count, active_factory_count

splitio/engine/impressions/impressions.py

@@ -14,7 +14,7 @@ class ImpressionsMode(Enum):
 class Manager(object):  # pylint:disable=too-few-public-methods
     """Impression manager."""
 
-    def __init__(self, strategy, telemetry_runtime_producer=None, listener=None):
+    def __init__(self, strategy, telemetry_runtime_producer, listener=None):
         """
         Construct a manger to track and forward impressions to the queue.