Created impressions sender, integrated with unique_keys_tracker, added synchronizer and task classes, added telemetry api endpoint.

Author: chillaq

splitio/api/telemetry.py

@@ -0,0 +1,50 @@
+"""Impressions API module."""
+
+import logging
+
+from splitio.api import APIException
+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."""
+
+    def __init__(self, client, apikey, sdk_metadata):
+        """
+        Class constructor.
+
+        :param client: HTTP Client responsble for issuing calls to the backend.
+        :type client: HttpClient
+        :param apikey: User apikey token.
+        :type apikey: string
+        """
+        self._client = client
+        self._apikey = apikey
+        self._metadata = headers_from_metadata(sdk_metadata)
+
+    def record_unique_keys(self, uniques):
+        """
+        Send unique keys to the backend.
+
+        :param uniques: Unique Keys
+        :type json
+        """
+        try:
+            response = self._client.post(
+                'keys',
+                '/ss',
+                self._apikey,
+                body=uniques,
+                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 unique keys because an exception was raised by the HTTPClient'
+            )
+            _LOGGER.debug('Error: ', exc_info=True)
+            raise APIException('Unique keys not flushed properly.') from exc

splitio/client/factory.py

@@ -27,12 +27,14 @@
 from splitio.api.impressions import ImpressionsAPI
 from splitio.api.events import EventsAPI
 from splitio.api.auth import AuthAPI
+from splitio.api.telemetry import TelemetryAPI
 
 # Tasks
 from splitio.tasks.split_sync import SplitSynchronizationTask
 from splitio.tasks.segment_sync import SegmentSynchronizationTask
 from splitio.tasks.impressions_sync import ImpressionsSyncTask, ImpressionsCountSyncTask
 from splitio.tasks.events_sync import EventsSyncTask
+from splitio.tasks.unique_keys_sync import UniqueKeysSyncTask, ClearFilterSyncTask
 
 # Synchronizer
 from splitio.sync.synchronizer import SplitTasks, SplitSynchronizers, Synchronizer, \
@@ -42,6 +44,8 @@
 from splitio.sync.segment import SegmentSynchronizer
 from splitio.sync.impression import ImpressionSynchronizer, ImpressionsCountSynchronizer
 from splitio.sync.event import EventSynchronizer
+from splitio.sync.unique_keys import UniqueKeysSynchronizer, ClearFilterSynchronizer
+
 
 # Recorder
 from splitio.recorder.recorder import StandardRecorder, PipelinedRecorder
@@ -283,7 +287,7 @@ def _wrap_impression_listener(listener, metadata):
 
 
 def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pylint:disable=too-many-arguments,too-many-locals
-                             auth_api_base_url=None, streaming_api_base_url=None):
+                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None):
     """Build and return a split factory tailored to the supplied config."""
     if not input_validator.validate_factory_instantiation(api_key):
         return None
@@ -292,6 +296,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         sdk_url=sdk_url,
         events_url=events_url,
         auth_url=auth_api_base_url,
+        telemetry_url=telemetry_api_base_url,
         timeout=cfg.get('connectionTimeout')
     )
 
@@ -302,6 +307,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'segments': SegmentsAPI(http_client, api_key, sdk_metadata),
         'impressions': ImpressionsAPI(http_client, api_key, sdk_metadata, cfg['impressionsMode']),
         'events': EventsAPI(http_client, api_key, sdk_metadata),
+        'telemtery': TelemetryAPI(http_client, api_key, sdk_metadata),
     }
 
     if not input_validator.validate_apikey_type(apis['segments']):
@@ -326,6 +332,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
                                cfg['impressionsBulkSize']),
         EventSynchronizer(apis['events'], storages['events'], cfg['eventsBulkSize']),
         ImpressionsCountSynchronizer(apis['impressions'], imp_manager),
+        UniqueKeysSynchronizer(), # TODO: Pass the UniqueKeysTracker instance fetched from Strategy instance created above.
+        ClearFilterSynchronizer(), # TODO: Pass the UniqueKeysTracker instance fetched from Strategy instance created above.
     )
 
     tasks = SplitTasks(
@@ -342,7 +350,9 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
             cfg['impressionsRefreshRate'],
         ),
         EventsSyncTask(synchronizers.events_sync.synchronize_events, cfg['eventsPushRate']),
-        ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters)
+        ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters),
+        UniqueKeysSyncTask(synchronizers.unique_keys_sync.SendAll),
+        ClearFilterSyncTask(synchronizers.clear_filter_sync.clearAll)
     )
 
     synchronizer = Synchronizer(synchronizers, tasks)
@@ -355,6 +365,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
 
     storages['events'].set_queue_full_hook(tasks.events_task.flush)
     storages['impressions'].set_queue_full_hook(tasks.impressions_task.flush)
+    # TODO: Add unique_keys_tracker.set_queue_full_hook(tasks.unique_keys.flush)
 
     recorder = StandardRecorder(
         imp_manager,

splitio/engine/sender_adapters/__init__.py

@@ -0,0 +1,20 @@
+import abc
+
+class ImpressionsSenderAdapter(object, metaclass=abc.ABCMeta):
+    """Impressions Sender Adapter interface."""
+
+    @abc.abstractmethod
+    def record_unique_keys(self, data):
+        """
+        No Return value
+
+        """
+        pass
+
+    @abc.abstractmethod
+    def record_impressions_count(self, data):
+        """
+        No Return value
+
+        """
+        pass

splitio/engine/sender_adapters/in_memory_sender_adapter.py

@@ -0,0 +1,41 @@
+import json
+
+from splitio.engine.sender_adapters import ImpressionsSenderAdapter
+
+class InMemorySenderAdapter(ImpressionsSenderAdapter):
+    """In Memory Impressions Sender Adapter class."""
+
+    def __init__(self, telemtry_http_client):
+        """
+        Initialize In memory sender adapter instance
+
+        :param telemtry_http_client: instance of telemetry http api
+        :type telemtry_http_client: splitio.api.telemetry.TelemetryAPI
+        """
+        self._telemtry_http_client = telemtry_http_client
+
+    def record_unique_keys(self, uniques):
+        """
+        post the unique keys to split back end.
+
+        :param uniques: unique keys disctionary
+        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+        """
+        self._telemtry_http_client.record_unique_keys(self._uniques_formatter(uniques))
+
+    def _uniques_formatter(self, uniques):
+        """
+        Format the unique keys dictionary to a JSON body
+
+        :param uniques: unique keys disctionary
+        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+
+        :return: unique keys JSON
+        :rtype: json
+        """
+        formatted_uniques = json.load('{keys: []}')
+        if len(uniques) == 0:
+            return formatted_uniques
+        for key in uniques:
+            formatted_uniques['keys'].append('{"f":"' + key +'", "ks:['+ json.dump(uniques[key])+']}')
+        return formatted_uniques

splitio/engine/unique_keys_tracker.py

@@ -2,6 +2,7 @@
 import threading
 import logging
 from splitio.engine.filters.bloom_filter import BloomFilter
+from splitio.engine.sender_adapters.in_memory_sender_adapter import InMemorySenderAdapter
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -16,38 +17,33 @@ def track(self, key, feature_name):
         """
         pass
 
-    @abc.abstractmethod
-    def start(self):
-        """
-        No return value
+class UniqueKeysTracker(BaseUniqueKeysTracker):
+    """Unique Keys Tracker class."""
 
+    def __init__(self, cache_size=30000):
         """
-        pass
+        Initialize unique keys tracker instance
 
-    @abc.abstractmethod
-    def stop(self):
+        :param cache_size: The size of the unique keys dictionary
+        :type key: int
         """
-        No return value
-
-        """
-        pass
-
-class UniqueKeysTracker(BaseUniqueKeysTracker):
-    """Unique Keys Tracker class."""
-
-    def __init__(self, cache_size=30000, max_bulk_size=5000, task_refresh_rate = 24):
         self._cache_size = cache_size
-        self._max_bulk_size = max_bulk_size
-        self._task_refresh_rate = task_refresh_rate
         self._filter = BloomFilter(cache_size)
         self._lock = threading.RLock()
         self._cache = {}
-        # TODO: initialize impressions sender adapter and task referesh rate in next PR
+        self._queue_full_hook = None
 
     def track(self, key, feature_name):
         """
         Return a boolean flag
 
+        :param key: key to be added to MTK list
+        :type key: int
+        :param feature_name: split name associated with the key
+        :type feature_name: str
+
+        :return: True if successful
+        :rtype: boolean
         """
         if self._filter.contains(feature_name+key):
             return False
@@ -56,34 +52,44 @@ def track(self, key, feature_name):
             self._add_or_update(feature_name, key)
             self._filter.add(feature_name+key)
 
-        if len(self._cache[feature_name]) == self._cache_size:
-            _LOGGER.warn("MTK Cache size for Split [%s] has reach maximum unique keys [%d], flushing data now.", feature_name, self._cache_size)
-#            TODO: Flush the data and reset split cache in next PR
-        if self._get_dict_size() >= self._max_bulk_size:
-            _LOGGER.info("Bulk MTK cache size has reach maximum, flushing data now.")
-#            TODO: Flush the data and reset split cache in next PR
-
+        if self._get_dict_size() > self._cache_size:
+            if self._queue_full_hook is not None and callable(self._queue_full_hook):
+                self._queue_full_hook()
+            _LOGGER.info(
+                'Unique Keys queue is full, flushing the current queue now.'
+            )
         return True
 
     def _get_dict_size(self):
+        """
+        Return the size of unique keys dictionary (number of keys in all features)
+
+        :return: dictionary set() items count
+        :rtype: int
+        """
         total_size = 0
-        for key in self._cache:
-            total_size = total_size + len(self._cache[key])
+        for key in self._uniqe_keys_tracker._cache:
+            total_size = total_size + len(self._uniqe_keys_tracker._cache[key])
         return total_size
 
     def _add_or_update(self, feature_name, key):
-        if feature_name not in self._cache:
-            self._cache[feature_name] = set()
-        self._cache[feature_name].add(key)
-
-    def start(self):
         """
-        TODO: Add start posting impressions job in next PR
+        Add the feature_name+key to both bloom filter and dictionary.
 
+        :param feature_name: split name associated with the key
+        :type feature_name: str
+        :param key: key to be added to MTK list
+        :type key: int
         """
+        if feature_name not in self._cache:
+            self._cache[feature_name] = set()
+        self._cache[feature_name].add(key)
 
-    def stop(self):
+    def set_queue_full_hook(self, hook):
         """
-        TODO: Add stop posting impressions job in next PR
+        Set a hook to be called when the queue is full.
 
+        :param h: Hook to be called when the queue is full
         """
+        if callable(hook):
+            self._queue_full_hook = hook