basic telemetry classes

Author: chillaq

splitio/api/telemetry.py

@@ -6,6 +6,8 @@
 from splitio.api.client import HttpClientException
 from splitio.api.commons import headers_from_metadata
 
+_LOGGER = logging.getLogger(__name__)
+
 class TelemetryAPI(object):  # pylint: disable=too-few-public-methods
     """Class that uses an httpClient to communicate with the Telemetry API."""
 
@@ -45,3 +47,51 @@ def record_unique_keys(self, uniques):
             )
             _LOGGER.debug('Error: ', exc_info=True)
             raise APIException('Unique keys not flushed properly.') from exc
+
+    def record_init(self, configs):
+        """
+        Send init config data to the backend.
+
+        :param configs: configs
+        :type json
+        """
+        try:
+            response = self._client.post(
+                'metrics',
+                '/config',
+                self._apikey,
+                body=configs,
+                extra_headers=self._metadata
+            )
+            if not 200 <= response.status_code < 300:
+                raise APIException(response.body, response.status_code)
+        except HttpClientException as exc:
+            _LOGGER.error(
+                'Error posting init config because an exception was raised by the HTTPClient'
+            )
+            _LOGGER.debug('Error: ', exc_info=True)
+            raise APIException('Init config data not flushed properly.') from exc
+
+    def record_stats(self, stats):
+        """
+        Send runtime stats to the backend.
+
+        :param configs: configs
+        :type json
+        """
+        try:
+            response = self._client.post(
+                'metrics',
+                '/usage',
+                self._apikey,
+                body=stats,
+                extra_headers=self._metadata
+            )
+            if not 200 <= response.status_code < 300:
+                raise APIException(response.body, response.status_code)
+        except HttpClientException as exc:
+            _LOGGER.error(
+                'Error posting runtime stats because an exception was raised by the HTTPClient'
+            )
+            _LOGGER.debug('Error: ', exc_info=True)
+            raise APIException('Runtime stats not flushed properly.') from exc

splitio/client/factory.py

@@ -297,6 +297,12 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     if not input_validator.validate_factory_instantiation(api_key):
         return None
 
+    cfg['sdk_url'] = sdk_url if sdk_url is not None else None
+    cfg['events_url'] = events_url if events_url is not None else None
+    cfg['auth_url'] = auth_api_base_url if auth_api_base_url is not None else None
+    cfg['streaming_url'] = streaming_api_base_url if streaming_api_base_url is not None else None
+    cfg['telemetry_api_url'] = telemetry_api_base_url if telemetry_api_base_url is not None else None
+
     http_client = HttpClient(
         sdk_url=sdk_url,
         events_url=events_url,
@@ -541,9 +547,13 @@ def _build_localhost_factory(cfg):
 def get_factory(api_key, **kwargs):
     """Build and return the appropriate factory."""
     try:
+        active_factory_count = 0
+        redundant_factory_count = 0
         _INSTANTIATED_FACTORIES_LOCK.acquire()
         if _INSTANTIATED_FACTORIES:
             if api_key in _INSTANTIATED_FACTORIES:
+                redundant_factory_count = redundant_factory_count + 1
+                active_factory_count = active_factory_count + 1
                 _LOGGER.warning(
                     "factory instantiation: You already have %d %s with this API Key. "
                     "We recommend keeping only one instance of the factory at all times "
@@ -552,6 +562,7 @@ def get_factory(api_key, **kwargs):
                     'factory' if _INSTANTIATED_FACTORIES[api_key] == 1 else 'factories'
                 )
             else:
+                active_factory_count = active_factory_count + 1
                 _LOGGER.warning(
                     "factory instantiation: You already have an instance of the Split factory. "
                     "Make sure you definitely want this additional instance. "
@@ -560,6 +571,8 @@ def get_factory(api_key, **kwargs):
                 )
 
         config = sanitize_config(api_key, kwargs.get('config', {}))
+        config['redundantFactoryCount'] = redundant_factory_count
+        config['activeFactoryCount'] = active_factory_count + 1
 
         if config['operationMode'] == 'localhost-standalone':
             return _build_localhost_factory(config)

splitio/engine/telemetry.py

@@ -0,0 +1,210 @@
+"""Telemetry engine classes."""
+from  splitio.storage.inmemmory import InMemoryTelemetryStorage
+
+class TelemetryStorageProducer(object):
+    """Telemetry storage producer class."""
+
+    def __init__(self, telemetry_storage):
+        """Initialize all producer classes."""
+        self._telemetry_init_producer = TelemetryInitProducer(telemetry_storage)
+        self._telemetry_evaluation_producer = TelemetryEvaluationProducer(telemetry_storage)
+        self._telemetry_runtime_producer = TelemetryRuntimeProducer(telemetry_storage)
+
+    def get_telemetry_init_producer(self):
+        """get init producer instance."""
+        return self._telemetry_init_producer
+
+    def get_telemetry_evaluation_producer(self):
+        """get evaluation producer instance."""
+        return self._telemetry_evaluation_producer
+
+    def get_telemetry_runtime_producer(self):
+        """get runtime producer instance."""
+        return self._telemetry_runtime_producer
+
+class TelemetryInitProducer(object):
+    """Telemetry init producer class."""
+
+    def __init__(self, telemetry_storage):
+        """Constructor."""
+        self._telemetry_storage = telemetry_storage
+
+    def record_config(self, config):
+        """Record configurations."""
+        self._telemetry_storage.record_config(config)
+
+    def record_ready_time(self, ready_time):
+        """Record ready time."""
+        self._telemetry_storage.record_ready_time(ready_time)
+
+    def record_bur_timeout(self):
+        """Record block until ready timeout."""
+        self._telemetry_storage.record_bur_timeout()
+
+    def record_non_ready_usage(self):
+        """record non-ready usage."""
+        self._telemetry_storage.record_non_ready_usage()
+
+class TelemetryEvaluationProducer(object):
+    """Telemetry evaluation producer class."""
+
+    def __init__(self, telemetry_storage):
+        """Constructor."""
+        self._telemetry_storage = telemetry_storage
+
+    def record_latency(self, method, latency):
+        """Record method latency time."""
+        self._telemetry_storage.record_latency(method, latency)
+
+    def record_exception(self, method):
+        """Record method exception time."""
+        self._telemetry_storage.record_exception(method)
+
+class TelemetryRuntimeProducer(object):
+    """Telemetry runtime producer class."""
+
+    def __init__(self, telemetry_storage):
+        """Constructor."""
+        self._telemetry_storage = telemetry_storage
+
+    def add_tag(self, tag):
+        """Record tag string."""
+        self._telemetry_storage.add_tag(tag)
+
+    def record_impression_stats(self, data_type, count):
+        """Record impressions stats."""
+        self._telemetry_storage.record_impression_stats(data_type, count)
+
+    def record_event_stats(self, data_type, count):
+        """Record events stats."""
+        self._telemetry_storage.record_event_stats(data_type, count)
+
+    def record_suceessful_sync(self, resource, time):
+        """Record successful sync."""
+        self._telemetry_storage.record_suceessful_sync(resource, time)
+
+    def record_sync_error(self, resource, status):
+        """Record sync error."""
+        self._telemetry_storage.record_sync_error(resource, status)
+
+    def record_sync_latency(self, resource, latency):
+        """Record latency time."""
+        self._telemetry_storage.record_sync_latency(resource, latency)
+
+    def record_auth_rejections(self):
+        """Record auth rejection."""
+        self._telemetry_storage.record_auth_rejections()
+
+    def record_token_refreshes(self):
+        """Record sse token refresh."""
+        self._telemetry_storage.record_token_refreshes()
+
+    def record_streaming_event(self, streaming_event):
+        """Record incoming streaming event."""
+        self._telemetry_storage.record_streaming_event(streaming_event)
+
+    def record_session_length(self, session):
+        """Record session length."""
+        self._telemetry_storage.record_session_length(session)
+
+class TelemetryStorageConsumer(object):
+    """Telemetry storage consumer class."""
+
+    def __init__(self, telemetry_storage):
+        """Initialize all consumer classes."""
+        self._telemetry_init_consumer = TelemetryInitConsumer(telemetry_storage)
+        self._telemetry_evaluation_consumer = TelemetryEvaluationConsumer(telemetry_storage)
+        self._telemetry_runtime_consumer = TelemetryRuntimeConsumer(telemetry_storage)
+
+    def get_telemetry_init_consumer(self):
+        """Get telemetry init instance"""
+        return self._telemetry_init_consumer
+
+    def get_telemetry_evaluation_consumer(self):
+        """Get telemetry evaluation instance"""
+        return self._telemetry_evaluation_consumer
+
+    def get_telemetry_runtime_consumer(self):
+        """Get telemetry runtime instance"""
+        return self._telemetry_runtime_consumer
+
+class TelemetryInitConsumer(object):
+    """Telemetry init consumer class."""
+
+    def __init__(self, telemetry_storage):
+        """Constructor."""
+        self._telemetry_storage = telemetry_storage
+
+    def get_bur_timeouts(self):
+        """Get block until ready timeout."""
+        return self._telemetry_storage.get_bur_timeouts()
+
+    def get_non_ready_usage(self):
+        """Get none-ready usage."""
+        return self._telemetry_storage.get_non_ready_usage()
+
+    def get_config_stats(self):
+        """Get none-ready usage."""
+        return self._telemetry_storage.get_config_stats()
+
+class TelemetryEvaluationConsumer(object):
+    """Telemetry evaluation consumer class."""
+
+    def __init__(self, telemetry_storage):
+        """Constructor."""
+        self._telemetry_storage = telemetry_storage
+
+    def pop_exceptions(self):
+        """Get and reset method exceptions."""
+        return self._telemetry_storage.pop_exceptions()
+
+    def pop_latencies(self):
+        """Get and reset eval latencies."""
+        return self._telemetry_storage.pop_latencies()
+
+class TelemetryRuntimeConsumer(object):
+    """Telemetry runtime consumer class."""
+
+    def __init__(self, telemetry_storage):
+        """Constructor."""
+        self._telemetry_storage = telemetry_storage
+
+    def get_impressions_stats(self, type):
+        """Get impressions stats"""
+        return self._telemetry_storage.get_impressions_stats(type)
+
+    def get_events_stats(self, type):
+        """Get events stats"""
+        return self._telemetry_storage.get_events_stats(type)
+
+    def get_last_synchronization(self):
+        """Get last sync"""
+        return self._telemetry_storage.get_last_synchronization()
+
+    def pop_tags(self):
+        """Get and reset http errors."""
+        return self._telemetry_storage.pop_tags()
+
+    def pop_http_errors(self):
+        """Get and reset http errors."""
+        return self._telemetry_storage.pop_http_errors()
+
+    def pop_http_latencies(self):
+        """Get and reset http latencies."""
+        return self._telemetry_storage.pop_http_latencies()
+
+    def pop_auth_rejections(self):
+        """Get and reset auth rejections."""
+        return self._telemetry_storage.pop_auth_rejections()
+
+    def pop_token_refreshes(self):
+        """Get and reset token refreshes."""
+        return self._telemetry_storage.pop_token_refreshes()
+
+    def pop_streaming_events(self):
+        """Get and reset streaming events."""
+        return self._telemetry_storage.pop_streaming_events()
+
+    def get_session_length(self):
+        """Get session length"""
+        return self._telemetry_storage.get_session_length()

splitio/storage/__init__.py

@@ -283,3 +283,54 @@ def clear(self):
         Clear data.
         """
         pass
+
+class TelemetryStorage(object, metaclass=abc.ABCMeta):
+    """Telemetry storage interface."""
+
+    @abc.abstractmethod
+    def record_init(self, config):
+        """
+        initilize telemetry objects
+
+        :param congif: factory configuration parameters
+        :type config: splitio.client.config
+        """
+        pass
+
+    @abc.abstractmethod
+    def record_latency(self, method, latency):
+        """
+        record latency data
+
+        :param method: method name
+        :type method: string
+        :param latency: latency
+        :type latency: int64
+        """
+        pass
+
+    @abc.abstractmethod
+    def record_exception(self, method):
+        """
+        record an exception
+
+        :param method: method name
+        :type method: string
+        """
+        pass
+
+    @abc.abstractmethod
+    def record_not_ready_usage(self):
+        """
+        record not ready time
+
+        """
+        pass
+
+    @abc.abstractmethod
+    def record_bur_time_out(self):
+        """
+        record BUR timeouts
+
+        """
+        pass