Merge pull request #267 from splitio/ImpressionsSender

Impressions sender

Author: chillaq

splitio/api/client.py

@@ -25,8 +25,9 @@ class HttpClient(object):
     SDK_URL = 'https://sdk.split.io/api'
     EVENTS_URL = 'https://events.split.io/api'
     AUTH_URL = 'https://auth.split.io/api'
+    TELEMETRY_URL = 'https://telemetry.split.io/api'
 
-    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None):
+    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, telemetry_url=None):
         """
         Class constructor.
 
@@ -38,12 +39,15 @@ def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None):
         :type events_url: str
         :param auth_url: Optional alternative auth URL.
         :type auth_url: str
+        :param telemetry_url: Optional alternative telemetry URL.
+        :type telemetry_url: str
         """
         self._timeout = timeout/1000 if timeout else None  # Convert ms to seconds.
         self._urls = {
             'sdk': sdk_url if sdk_url is not None else self.SDK_URL,
             'events': events_url if events_url is not None else self.EVENTS_URL,
             'auth': auth_url if auth_url is not None else self.AUTH_URL,
+            'telemetry': telemetry_url if telemetry_url is not None else self.TELEMETRY_URL,
         }
 
     def _build_url(self, server, path):

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

@@ -16,6 +16,7 @@
 from splitio.engine.strategies import Counter as ImpressionsCounter
 from splitio.engine.strategies.strategy_debug_mode import StrategyDebugMode
 from splitio.engine.strategies.strategy_optimized_mode import StrategyOptimizedMode
+from splitio.engine.sender_adapters.in_memory_sender_adapter import InMemorySenderAdapter
 
 # Storage
 from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
@@ -31,12 +32,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, \
@@ -46,6 +49,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
@@ -287,7 +292,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
@@ -296,6 +301,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')
     )
 
@@ -306,6 +312,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),
+        'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata),
     }
 
     if not input_validator.validate_apikey_type(apis['segments']):
@@ -317,6 +324,13 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'impressions': InMemoryImpressionStorage(cfg['impressionsQueueSize']),
         'events': InMemoryEventStorage(cfg['eventsQueueSize']),
     }
+    imp_counter = ImpressionsCounter() if cfg['impressionsMode'] != ImpressionsMode.DEBUG else None
+
+    strategies = {
+        ImpressionsMode.OPTIMIZED : StrategyOptimizedMode(imp_counter),
+        ImpressionsMode.DEBUG : StrategyDebugMode(),
+    }
+    imp_strategy = strategies[cfg['impressionsMode']]
 
     imp_counter = ImpressionsCounter() if cfg['impressionsMode'] != ImpressionsMode.DEBUG else None
 
@@ -354,7 +368,7 @@ 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']),
-        imp_count_sync_task
+        ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters),
     )
 
     synchronizer = Synchronizer(synchronizers, tasks)
@@ -452,7 +466,7 @@ def _build_localhost_factory(cfg):
     manager.start()
 
     recorder = StandardRecorder(
-        ImpressionsManager(cfg['impressionsMode'], StrategyDebugMode()),
+        ImpressionsManager(None, StrategyDebugMode()),
         storages['events'],
         storages['impressions'],
     )
@@ -501,7 +515,8 @@ def get_factory(api_key, **kwargs):
             kwargs.get('sdk_api_base_url'),
             kwargs.get('events_api_base_url'),
             kwargs.get('auth_api_base_url'),
-            kwargs.get('streaming_api_base_url')
+            kwargs.get('streaming_api_base_url'),
+            kwargs.get('telemetry_api_base_url')
         )
     finally:
         _INSTANTIATED_FACTORIES.update([api_key])

splitio/engine/impressions.py

@@ -8,6 +8,7 @@ class ImpressionsMode(Enum):
 
     OPTIMIZED = "OPTIMIZED"
     DEBUG = "DEBUG"
+    NONE = "NONE"
 
 class Manager(object):  # pylint:disable=too-few-public-methods
     """Impression manager."""

splitio/engine/sender_adapters/__init__.py

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

splitio/engine/sender_adapters/in_memory_sender_adapter.py

@@ -0,0 +1,42 @@
+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.loads('{"keys": []}')
+        if len(uniques) == 0:
+            return json.loads('{"keys": []}')
+
+        return {
+            'keys':  [{'f': feature, 'ks': list(keys)} for feature, keys in uniques.items()]
+        }

splitio/engine/strategies/strategy_optimized_mode.py

@@ -7,7 +7,7 @@
 class StrategyOptimizedMode(BaseStrategy):
     """Optimized mode strategy."""
 
-    def __init__(self, counter=None):
+    def __init__(self, counter):
         """
         Construct a strategy instance for optimized mode.
 

splitio/engine/unique_keys_tracker.py

@@ -16,74 +16,87 @@ 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
+        self._current_cache_size = 0
 
     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
+        with self._lock:
+            if self._filter.contains(feature_name+key):
+                return False
 
         with self._lock:
             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
-
+            self._current_cache_size = self._current_cache_size + 1
+
+        if self._current_cache_size > self._cache_size:
+            _LOGGER.info(
+                'Unique Keys queue is full, flushing the current queue now.'
+            )
+            if self._queue_full_hook is not None and callable(self._queue_full_hook):
+                self._queue_full_hook()
         return True
 
-    def _get_dict_size(self):
-        total_size = 0
-        for key in self._cache:
-            total_size = total_size + len(self._cache[key])
-        return total_size
-
     def _add_or_update(self, feature_name, key):
+        """
+        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 start(self):
+    def set_queue_full_hook(self, hook):
         """
-        TODO: Add start 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
 
-    def stop(self):
+    def filter_pop_all(self):
         """
-        TODO: Add stop posting impressions job in next PR
+        Delete the filter items
 
         """
+        with self._lock:
+            self._filter.clear()
+
+    def get_cache_info_and_pop_all(self):
+        with self._lock:
+            temp_cach = self._cache.copy()
+            temp_cache_size = self._current_cache_size
+            self._cache = {}
+            self._current_cache_size = 0
+
+            return temp_cach, temp_cache_size

splitio/sync/impression.py

@@ -84,7 +84,7 @@ def __init__(self, impressions_api, impressions_counter):
 
     def synchronize_counters(self):
         """Send impressions from both the failed and new queues."""
-        to_send = self._impressions_counter.pop_all()
+        to_send = self._impressions_manager._strategy._counter.pop_all()
         if not to_send:
             return