Fixed streaming event data and polishing

Author: chillaq

splitio/api/auth.py

@@ -2,10 +2,9 @@
 
 import logging
 import json
-import time
 
 from splitio.api import APIException
-from splitio.api.commons import headers_from_metadata, record_telemetry
+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
@@ -39,20 +38,20 @@ def authenticate(self):
         :return: Json representation of an authentication.
         :rtype: splitio.models.token.Token
         """
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             response = self._client.get(
                 'auth',
                 '/v2/auth',
                 self._apikey,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, int(round(time.time() * 1000)) - start, TOKEN, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, TOKEN, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 payload = json.loads(response.body)
                 return from_raw(payload)
             else:
-                if response.status_code == 401:
+                if (response.status_code >= 400 and response.status_code < 500):
                     self._telemetry_runtime_producer.record_auth_rejections()
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:

splitio/api/client.py

@@ -2,6 +2,8 @@
 from collections import namedtuple
 
 import requests
+import logging
+_LOGGER = logging.getLogger(__name__)
 
 HttpResponse = namedtuple('HttpResponse', ['status_code', 'body'])
 

splitio/api/commons.py

@@ -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_suceessful_sync(metric_name, int(round(time.time() * 1000)))
+        telemetry_runtime_producer.record_successful_sync(metric_name, get_current_epoch_time())
     else:
         telemetry_runtime_producer.record_sync_error(metric_name, status_code)
 
@@ -114,3 +114,12 @@ def build_fetch(change_number, fetch_options, metadata):
     if fetch_options.change_number is not None:
         query['till'] = fetch_options.change_number
     return query, extra_headers
+
+def get_current_epoch_time():
+    """
+    Get current epoch time in milliseconds
+
+    :return: epoch time
+    :rtype: int
+    """
+    return int(round(time.time() * 1000))

splitio/api/events.py

@@ -4,7 +4,7 @@
 
 from splitio.api import APIException
 from splitio.api.client import HttpClientException
-from splitio.api.commons import headers_from_metadata, record_telemetry
+from splitio.api.commons import headers_from_metadata, record_telemetry, get_current_epoch_time
 from splitio.models.telemetry import EVENT
 
 
@@ -64,7 +64,7 @@ def flush_events(self, events):
         :rtype: bool
         """
         bulk = self._build_bulk(events)
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             response = self._client.post(
                 'events',
@@ -73,7 +73,7 @@ def flush_events(self, events):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, int(round(time.time() * 1000)) - start, EVENT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, 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

@@ -2,11 +2,10 @@
 
 import logging
 from itertools import groupby
-import time
 
 from splitio.api import APIException
 from splitio.api.client import HttpClientException
-from splitio.api.commons import headers_from_metadata, record_telemetry
+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
 
@@ -94,7 +93,7 @@ def flush_impressions(self, impressions):
         :type impressions: list
         """
         bulk = self._build_bulk(impressions)
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             response = self._client.post(
                 'events',
@@ -103,7 +102,7 @@ def flush_impressions(self, impressions):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, int(round(time.time() * 1000)) - start, IMPRESSION, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, 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 +120,7 @@ def flush_counters(self, counters):
         :type impressions: list
         """
         bulk = self._build_counters(counters)
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             response = self._client.post(
                 'events',
@@ -130,7 +129,7 @@ def flush_counters(self, counters):
                 body=bulk,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, int(round(time.time() * 1000)) - start, IMPRESSION_COUNT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, 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

@@ -5,7 +5,7 @@
 import time
 
 from splitio.api import APIException
-from splitio.api.commons import headers_from_metadata, build_fetch, record_telemetry
+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
 
@@ -49,7 +49,7 @@ def fetch_segment(self, segment_name, change_number, fetch_options):
         :return: Json representation of a segmentChange response.
         :rtype: dict
         """
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
             response = self._client.get(
@@ -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, int(round(time.time() * 1000)) - start, SEGMENT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, SEGMENT, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)
             else:

splitio/api/splits.py

@@ -5,7 +5,7 @@
 import time
 
 from splitio.api import APIException
-from splitio.api.commons import headers_from_metadata, build_fetch, record_telemetry
+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
 
@@ -44,7 +44,7 @@ def fetch_splits(self, change_number, fetch_options):
         :return: Json representation of a splitChanges response.
         :rtype: dict
         """
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
             response = self._client.get(
@@ -54,7 +54,7 @@ def fetch_splits(self, change_number, fetch_options):
                 extra_headers=extra_headers,
                 query=query,
             )
-            record_telemetry(response.status_code, int(round(time.time() * 1000)) - start, SPLIT, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, SPLIT, self._telemetry_runtime_producer)
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)
             else:

splitio/api/telemetry.py

@@ -4,7 +4,7 @@
 
 from splitio.api import APIException
 from splitio.api.client import HttpClientException
-from splitio.api.commons import headers_from_metadata, record_telemetry
+from splitio.api.commons import headers_from_metadata, record_telemetry, get_current_epoch_time
 from splitio.models.telemetry import TELEMETRY
 
 _LOGGER = logging.getLogger(__name__)
@@ -33,7 +33,7 @@ def record_unique_keys(self, uniques):
         :param uniques: Unique Keys
         :type json
         """
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             response = self._client.post(
                 'telemetry',
@@ -42,7 +42,7 @@ def record_unique_keys(self, uniques):
                 body=uniques,
                 extra_headers=self._metadata
             )
-            record_telemetry(response.status_code,  int(round(time.time() * 1000)) - start, TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code,  get_current_epoch_time() - start, TELEMETRY, self._telemetry_runtime_producer)
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:
@@ -59,7 +59,7 @@ def record_init(self, configs):
         :param configs: configs
         :type json
         """
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             response = self._client.post(
                 'telemetry',
@@ -68,7 +68,7 @@ def record_init(self, configs):
                 body=configs,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, int(round(time.time() * 1000)) - start, TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, TELEMETRY, self._telemetry_runtime_producer)
             if not 200 <= response.status_code < 300:
                 raise APIException(response.body, response.status_code)
         except HttpClientException as exc:
@@ -85,7 +85,7 @@ def record_stats(self, stats):
         :param stats: stats
         :type json
         """
-        start = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         try:
             response = self._client.post(
                 'telemetry',
@@ -94,7 +94,7 @@ def record_stats(self, stats):
                 body=stats,
                 extra_headers=self._metadata,
             )
-            record_telemetry(response.status_code, int(round(time.time() * 1000)) - start, TELEMETRY, self._telemetry_runtime_producer)
+            record_telemetry(response.status_code, get_current_epoch_time() - start, 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

@@ -8,7 +8,7 @@
 from splitio.models.telemetry import get_latency_bucket_index, TRACK
 from splitio.client import input_validator
 from splitio.util import utctime_ms
-
+from splitio.api.commons import get_current_epoch_time
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -93,7 +93,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 = int(round(time.time() * 1000))
+            start = get_current_epoch_time()
 
             matching_key, bucketing_key = input_validator.validate_key(key, method_name)
             feature = input_validator.validate_feature_name(
@@ -149,7 +149,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 = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
 
         matching_key, bucketing_key = input_validator.validate_key(key, method_name)
         if matching_key is None and bucketing_key is None:
@@ -209,7 +209,7 @@ def _make_evaluations(self, key, features, attributes, method_name, metric_name)
                 _LOGGER.debug('Error: ', exc_info=True)
                 self._telemetry_evaluation_producer.record_exception(method_name[4:])
 
-            self._telemetry_evaluation_producer.record_latency(method_name[4:], int(round(time.time() * 1000)) - start)
+            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)
@@ -347,7 +347,7 @@ def _record_stats(self, impressions, start, operation, method_name=None):
         :param operation: operation performed.
         :type operation: str
         """
-        end = int(round(time.time() * 1000))
+        end = get_current_epoch_time()
         self._recorder.record_treatment_stats(impressions, get_latency_bucket_index(end - start),
                                               operation)
         if not method_name == None:
@@ -382,7 +382,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 = int(round(time.time() * 1000))
+        start = get_current_epoch_time()
         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(TRACK, int(round(time.time() * 1000)) - start)
+        self._telemetry_evaluation_producer.record_latency(TRACK, get_current_epoch_time() - start)
         if not return_flag:
             self._telemetry_evaluation_producer.record_exception(TRACK)
 

splitio/client/factory.py

@@ -2,7 +2,6 @@
 import logging
 import threading
 from collections import Counter
-import time
 
 from enum import Enum
 
@@ -34,6 +33,7 @@
 from splitio.api.events import EventsAPI
 from splitio.api.auth import AuthAPI
 from splitio.api.telemetry import TelemetryAPI
+from splitio.api.commons import get_current_epoch_time
 
 # Tasks
 from splitio.tasks.split_sync import SplitSynchronizationTask
@@ -128,7 +128,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 = int(round(time.time() * 1000))
+        self._ready_time = get_current_epoch_time()
         self._start_status_updater()
 
     def _start_status_updater(self):
@@ -156,7 +156,7 @@ 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(int(round(time.time() * 1000)) - self._ready_time)
+        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)