Added telemetery pluggable storage with tests

Author: chillaq

splitio/models/telemetry.py

@@ -17,6 +17,7 @@
 MAX_LATENCY = 7481828
 MAX_LATENCY_BUCKET_COUNT = 23
 MAX_STREAMING_EVENTS = 20
+MAX_TAGS = 10
 
 class CounterConstants(Enum):
     """Impressions and events counters constants"""

splitio/storage/pluggable.py

@@ -2,10 +2,12 @@
 
 import logging
 import json
+import threading
 
 from splitio.models import splits, segments
 from splitio.models.impressions import Impression
-from splitio.storage import SplitStorage, SegmentStorage, ImpressionStorage, EventStorage
+from splitio.models.telemetry import MethodExceptions, MethodLatencies, TelemetryConfig, MAX_TAGS
+from splitio.storage import SplitStorage, SegmentStorage, ImpressionStorage, EventStorage, TelemetryStorage
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -15,7 +17,14 @@ class PluggableSplitStorage(SplitStorage):
     _SPLIT_NAME_LENGTH = 12
 
     def __init__(self, pluggable_adapter, prefix=None):
-        """Constructor."""
+        """
+        Class constructor.
+
+        :param pluggable_adapter: Storage client or compliant interface.
+        :type pluggable_adapter: TBD
+        :param prefix: optional, prefix to storage keys
+        :type prefix: str
+        """
         self._pluggable_adapter = pluggable_adapter
         self._prefix = "SPLITIO.split.{split_name}"
         self._traffic_type_prefix = "SPLITIO.trafficType.{traffic_type_name}"
@@ -302,7 +311,14 @@ class PluggableSegmentStorage(SegmentStorage):
     _TILL_LENGTH = 4
 
     def __init__(self, pluggable_adapter, prefix=None):
-        """Constructor."""
+        """
+        Class constructor.
+
+        :param pluggable_adapter: Storage client or compliant interface.
+        :type pluggable_adapter: TBD
+        :param prefix: optional, prefix to storage keys
+        :type prefix: str
+        """
         self._pluggable_adapter = pluggable_adapter
         self._prefix = "SPLITIO.segment.{segment_name}"
         self._segment_till_prefix = "SPLITIO.segment.{segment_name}.till"
@@ -475,6 +491,7 @@ def put(self, segment):
 
 
 class PluggableImpressionsStorage(ImpressionStorage):
+    """Pluggable Impressions storage class."""
 
     IMPRESSIONS_KEY_DEFAULT_TTL = 3600
 
@@ -486,6 +503,8 @@ def __init__(self, pluggable_adapter, sdk_metadata, prefix=None):
         :type pluggable_adapter: TBD
         :param sdk_metadata: SDK & Machine information.
         :type sdk_metadata: splitio.client.util.SdkMetadata
+        :param prefix: optional, prefix to storage keys
+        :type prefix: str
         """
         self._pluggable_adapter = pluggable_adapter
         self._sdk_metadata = {
@@ -573,18 +592,20 @@ def clear(self):
 
 
 class PluggableEventsStorage(EventStorage):
-    """Redis based event storage class."""
+    """Pluggable Event storage class."""
 
     _EVENTS_KEY_DEFAULT_TTL = 3600
 
     def __init__(self, pluggable_adapter, sdk_metadata, prefix=None):
         """
         Class constructor.
 
-        :param redis_client: Redis client or compliant interface.
-        :type redis_client: splitio.storage.adapters.redis.RedisAdapter
+        :param pluggable_adapter: Storage client or compliant interface.
+        :type pluggable_adapter: TBD
         :param sdk_metadata: SDK & Machine information.
         :type sdk_metadata: splitio.client.util.SdkMetadata
+        :param prefix: optional, prefix to storage keys
+        :type prefix: str
         """
         self._pluggable_adapter = pluggable_adapter
         self._sdk_metadata = {
@@ -657,3 +678,167 @@ def clear(self):
         Clear data.
         """
         raise NotImplementedError('Not supported for redis.')
+
+
+class PluggableTelemetryStorage(TelemetryStorage):
+    """Pluggable telemetry storage class."""
+
+    _TELEMETRY_KEY_DEFAULT_TTL = 3600
+
+    def __init__(self, pluggable_adapter, sdk_metadata, prefix=None):
+        """
+        Class constructor.
+
+        :param pluggable_adapter: Storage client or compliant interface.
+        :type pluggable_adapter: TBD
+        :param sdk_metadata: SDK & Machine information.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
+        :param prefix: optional, prefix to storage keys
+        :type prefix: str
+        """
+        self._lock = threading.RLock()
+        self._reset_config_tags()
+        self._pluggable_adapter = pluggable_adapter
+        self._sdk_metadata = sdk_metadata.sdk_version + '/' + sdk_metadata.instance_name + '/' + sdk_metadata.instance_ip
+        self._method_latencies = MethodLatencies()
+        self._method_exceptions = MethodExceptions()
+        self._tel_config = TelemetryConfig()
+        self._telemetry_config_key = 'SPLITIO.telemetry.init'
+        self._telemetry_latencies_key = 'SPLITIO.telemetry.latencies'
+        self._telemetry_exceptions_key = 'SPLITIO.telemetry.exceptions'
+        if prefix is not None:
+            self._telemetry_config_key = prefix + "." + self._telemetry_config_key
+            self._telemetry_latencies_key = prefix + "." + self._telemetry_latencies_key
+            self._telemetry_exceptions_key = prefix + "." + self._telemetry_exceptions_key
+
+    def _reset_config_tags(self):
+        """Reset config tags."""
+        with self._lock:
+            self._config_tags = []
+
+    def add_config_tag(self, tag):
+        """
+        Record tag string.
+
+        :param tag: tag to be added
+        :type tag: str
+        """
+        with self._lock:
+            if len(self._config_tags) < MAX_TAGS:
+                self._config_tags.append(tag)
+
+    def record_config(self, config, extra_config):
+        """
+        initilize telemetry objects
+
+        :param config: factory configuration parameters
+        :type config: Dict
+        :param extra_config: any extra configs
+        :type extra_config: Dict
+        """
+        self._tel_config.record_config(config, extra_config)
+
+    def pop_config_tags(self):
+        """Get and reset configs."""
+        with self._lock:
+            tags = self._config_tags
+            self._reset_config_tags()
+            return tags
+
+    def push_config_stats(self):
+        """push config stats to storage."""
+        self._pluggable_adapter.set(self._telemetry_config_key + "::" + self._sdk_metadata, str(self._format_config_stats()))
+
+    def _format_config_stats(self):
+        """format only selected config stats to json"""
+        config_stats = self._tel_config.get_stats()
+        return json.dumps({
+            'aF': config_stats['aF'],
+            'rF': config_stats['rF'],
+            'sT': config_stats['sT'],
+            'oM': config_stats['oM'],
+            't': self.pop_config_tags()
+        })
+
+    def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
+        """
+        Record active and redundant factories.
+
+        :param active_factory_count: active factory count
+        :type active_factory_count: int
+        :param redundant_factory_count: redundant factory count
+        :type redundant_factory_count: int
+        """
+        self._tel_config.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
+
+    def record_latency(self, method, latency):
+        """
+        record latency data
+
+        :param method: method name
+        :type method: string
+        :param latency: latency
+        :type latency: int64
+        """
+        self._method_latencies.add_latency(method, latency)
+        latencies = self._method_latencies.pop_all()['methodLatencies']
+        values = latencies[method.value]
+        total_keys = 0
+        bucket_number = 0
+        for bucket in values:
+            if bucket > 0:
+                latency_key = self._telemetry_latencies_key + '::' + self._sdk_metadata + '/' + method.value + '/' + str(bucket_number)
+                result = self._pluggable_adapter.increment(latency_key, bucket)
+                self.expire_keys(latency_key, self._TELEMETRY_KEY_DEFAULT_TTL, 1, result)
+                total_keys += 1
+            bucket_number = bucket_number + 0
+
+    def record_exception(self, method):
+        """
+        record an exception
+
+        :param method: method name
+        :type method: string
+        """
+        except_key = self._telemetry_exceptions_key + "::" + self._sdk_metadata + '/' + method.value
+        result = self._pluggable_adapter.increment(except_key, 1)
+        self.expire_keys(except_key, self._TELEMETRY_KEY_DEFAULT_TTL, 1, result)
+
+    def record_not_ready_usage(self):
+        """Not implemented"""
+        pass
+
+    def record_bur_time_out(self):
+        """Not implemented"""
+        pass
+
+    def record_impression_stats(self, data_type, count):
+        """Not implemented"""
+        pass
+
+    def expire_latency_keys(self, total_keys, inserted):
+        """
+        Set expire ttl for a latency key in storage
+
+        :param total_keys: length of keys.
+        :type total_keys: int
+        :param inserted: added keys.
+        :type inserted: int
+        """
+        self.expire_keys(self._telemetry_latencies_key, self._TELEMETRY_KEY_DEFAULT_TTL, total_keys, inserted)
+
+    def expire_keys(self, queue_key, key_default_ttl, total_keys, inserted):
+        """
+        Set expire ttl for a key in storage if total keys equal inserted
+
+        :param queue_keys: key to be set
+        :type queue_keys: str
+        :param ey_default_ttl: ttl value
+        :type ey_default_ttl: int
+        :param total_keys: length of keys.
+        :type total_keys: int
+        :param inserted: added keys.
+        :type inserted: int
+        """
+        if total_keys == inserted:
+            self._pluggable_adapter.expire(queue_key, key_default_ttl)

tests/storage/test_pluggable.py

@@ -6,8 +6,9 @@
 from splitio.models.segments import Segment
 from splitio.models.impressions import Impression
 from splitio.models.events import Event, EventWrapper
-from splitio.storage.pluggable import PluggableSplitStorage, PluggableSegmentStorage, PluggableImpressionsStorage, PluggableEventsStorage
+from splitio.storage.pluggable import PluggableSplitStorage, PluggableSegmentStorage, PluggableImpressionsStorage, PluggableEventsStorage, PluggableTelemetryStorage
 from splitio.client.util import get_metadata, SdkMetadata
+from splitio.models.telemetry import MAX_TAGS, MethodExceptionsAndLatencies, OperationMode
 
 from tests.integration import splits_json
 import pytest
@@ -432,7 +433,7 @@ def mock_expire(impressions_queue_key, ttl):
 
 
 class PluggableEventsStorageTests(object):
-    """In memory events storage test cases."""
+    """Pluggable events storage test cases."""
 
     def setup_method(self):
         """Prepare storages with test data."""
@@ -526,3 +527,109 @@ def mock_expire(impressions_event_key, ttl):
         assert(self.expired_called)
         assert(self.key == "myprefix.SPLITIO.events")
         assert(self.ttl == self.pluggable_events_storage._EVENTS_KEY_DEFAULT_TTL)
+
+class PluggableTelemetryStorageTests(object):
+    """Pluggable telemetry storage test cases."""
+
+    def setup_method(self):
+        """Prepare storages with test data."""
+        self.mock_adapter = StorageMockAdapter()
+        self.sdk_metadata = SdkMetadata('python-1.1.1', 'hostname', 'ip')
+        self.pluggable_telemetry_storage = PluggableTelemetryStorage(self.mock_adapter, self.sdk_metadata, 'myprefix')
+
+    def test_init(self):
+        assert(self.pluggable_telemetry_storage._telemetry_config_key == 'myprefix.SPLITIO.telemetry.init')
+        assert(self.pluggable_telemetry_storage._telemetry_latencies_key == 'myprefix.SPLITIO.telemetry.latencies')
+        assert(self.pluggable_telemetry_storage._telemetry_exceptions_key == 'myprefix.SPLITIO.telemetry.exceptions')
+        assert(self.pluggable_telemetry_storage._sdk_metadata == self.sdk_metadata.sdk_version + '/' + self.sdk_metadata.instance_name + '/' + self.sdk_metadata.instance_ip)
+        assert(self.pluggable_telemetry_storage._config_tags == [])
+
+        pluggable2 = PluggableTelemetryStorage(self.mock_adapter, self.sdk_metadata)
+        assert(pluggable2._telemetry_config_key == 'SPLITIO.telemetry.init')
+        assert(pluggable2._telemetry_latencies_key == 'SPLITIO.telemetry.latencies')
+        assert(pluggable2._telemetry_exceptions_key == 'SPLITIO.telemetry.exceptions')
+
+    def test_reset_config_tags(self):
+        self.pluggable_telemetry_storage._config_tags = ['a']
+        self.pluggable_telemetry_storage._reset_config_tags()
+        assert(self.pluggable_telemetry_storage._config_tags == [])
+
+    def test_add_config_tag(self):
+        self.pluggable_telemetry_storage.add_config_tag('q')
+        assert(self.pluggable_telemetry_storage._config_tags == ['q'])
+
+        self.pluggable_telemetry_storage._config_tags = []
+        for i in range(0, 20):
+            self.pluggable_telemetry_storage.add_config_tag('q' + str(i))
+        assert(len(self.pluggable_telemetry_storage._config_tags) == MAX_TAGS)
+        assert(self.pluggable_telemetry_storage._config_tags == ['q' + str(i)  for i in range(0, MAX_TAGS)])
+
+    def test_record_config(self):
+        self.config = {}
+        self.extra_config = {}
+        def record_config_mock(config, extra_config):
+            self.config = config
+            self.extra_config = extra_config
+
+        self.pluggable_telemetry_storage.record_config = record_config_mock
+        self.pluggable_telemetry_storage.record_config({'item': 'value'}, {'item2': 'value2'})
+        assert(self.config == {'item': 'value'})
+        assert(self.extra_config == {'item2': 'value2'})
+
+    def test_pop_config_tags(self):
+        self.pluggable_telemetry_storage._config_tags = ['a']
+        self.pluggable_telemetry_storage.pop_config_tags()
+        assert(self.pluggable_telemetry_storage._config_tags == [])
+
+    def test_record_active_and_redundant_factories(self):
+        self.active_factory_count = 0
+        self.redundant_factory_count = 0
+        def record_active_and_redundant_factories_mock(active_factory_count, redundant_factory_count):
+            self.active_factory_count = active_factory_count
+            self.redundant_factory_count = redundant_factory_count
+
+        self.pluggable_telemetry_storage.record_active_and_redundant_factories = record_active_and_redundant_factories_mock
+        self.pluggable_telemetry_storage.record_active_and_redundant_factories(2, 1)
+        assert(self.active_factory_count == 2)
+        assert(self.redundant_factory_count == 1)
+
+    def test_record_latency(self):
+        def expire_keys_mock(*args, **kwargs):
+            assert(args[0] == self.pluggable_telemetry_storage._telemetry_latencies_key + '::python-1.1.1/hostname/ip/treatment/0')
+            assert(args[1] == self.pluggable_telemetry_storage._TELEMETRY_KEY_DEFAULT_TTL)
+            assert(args[2] == 1)
+            assert(args[3] == 1)
+
+        self.pluggable_telemetry_storage.expire_keys = expire_keys_mock
+        self.pluggable_telemetry_storage.record_latency(MethodExceptionsAndLatencies.TREATMENT, 10)
+        assert(self.mock_adapter._keys[self.pluggable_telemetry_storage._telemetry_latencies_key + '::python-1.1.1/hostname/ip/treatment/0'] == 1)
+
+    def test_record_exception(self):
+        def expire_keys_mock(*args, **kwargs):
+            assert(args[0] == self.pluggable_telemetry_storage._telemetry_exceptions_key + '::python-1.1.1/hostname/ip/treatment')
+            assert(args[1] == self.pluggable_telemetry_storage._TELEMETRY_KEY_DEFAULT_TTL)
+            assert(args[2] == 1)
+            assert(args[3] == 1)
+
+        self.pluggable_telemetry_storage.expire_keys = expire_keys_mock
+        self.pluggable_telemetry_storage.record_exception(MethodExceptionsAndLatencies.TREATMENT)
+        assert(self.mock_adapter._keys[self.pluggable_telemetry_storage._telemetry_exceptions_key + '::python-1.1.1/hostname/ip/treatment'] == 1)
+
+    def test_push_config_stats(self):
+        self.pluggable_telemetry_storage.record_config(
+                    {'operationMode': 'inmemory',
+                  'streamingEnabled': True,
+                  'impressionsQueueSize': 100,
+                  'eventsQueueSize': 200,
+                  'impressionsMode': 'DEBUG',''
+                  'impressionListener': None,
+                  'featuresRefreshRate': 30,
+                  'segmentsRefreshRate': 30,
+                  'impressionsRefreshRate': 60,
+                  'eventsPushRate': 60,
+                  'metricsRefreshRate': 10,
+                  }, {}
+        )
+        self.pluggable_telemetry_storage.record_active_and_redundant_factories(2, 1)
+        self.pluggable_telemetry_storage.push_config_stats()
+        assert(self.mock_adapter._keys[self.pluggable_telemetry_storage._telemetry_config_key + "::" + self.pluggable_telemetry_storage._sdk_metadata] == '{"aF": 2, "rF": 1, "sT": "memory", "oM": 0, "t": []}')