Merge pull request #268 from splitio/NoneImplementation

None implementation

Author: chillaq

splitio/api/telemetry.py

@@ -6,9 +6,6 @@
 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."""
 
@@ -34,8 +31,8 @@ def record_unique_keys(self, uniques):
         """
         try:
             response = self._client.post(
-                'keys',
-                '/ss',
+                'telemetry',
+                '/keys/ss',
                 self._apikey,
                 body=uniques,
                 extra_headers=self._metadata

splitio/client/config.py

@@ -91,7 +91,7 @@ def _sanitize_impressions_mode(mode, refresh_rate=None):
             mode = ImpressionsMode(mode.upper())
         except (ValueError, AttributeError):
             _LOGGER.warning('You passed an invalid impressionsMode, impressionsMode should be '
-                            'one of the following values: `debug` or `optimized`. '
+                            'one of the following values: `debug`, `none` or `optimized`. '
                             'Defaulting to `optimized` mode.')
             mode = ImpressionsMode.OPTIMIZED
 

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.strategies.strategy_none_mode import StrategyNoneMode
 from splitio.engine.sender_adapters.in_memory_sender_adapter import InMemorySenderAdapter
 
 # Storage
@@ -326,19 +327,21 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     }
     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
-
-    strategies = {
-        ImpressionsMode.OPTIMIZED : StrategyOptimizedMode(imp_counter),
-        ImpressionsMode.DEBUG : StrategyDebugMode(),
-    }
-    imp_strategy = strategies[cfg['impressionsMode']]
+    unique_keys_synchronizer = None
+    clear_filter_sync = None
+    unique_keys_task = None
+    clear_filter_task = None
+    if cfg['impressionsMode'] == ImpressionsMode.NONE:
+        imp_strategy = StrategyNoneMode(imp_counter)
+        clear_filter_sync = ClearFilterSynchronizer(imp_strategy._unique_keys_tracker)
+        unique_keys_synchronizer = UniqueKeysSynchronizer(InMemorySenderAdapter(apis['telemetry']), imp_strategy._unique_keys_tracker)
+        unique_keys_task = UniqueKeysSyncTask(unique_keys_synchronizer.send_all)
+        clear_filter_task = ClearFilterSyncTask(clear_filter_sync.clear_all)
+        imp_strategy._unique_keys_tracker.set_queue_full_hook(unique_keys_task.flush)
+    elif cfg['impressionsMode'] == ImpressionsMode.DEBUG:
+        imp_strategy = StrategyDebugMode()
+    else:
+        imp_strategy = StrategyOptimizedMode(imp_counter)
 
     imp_manager = ImpressionsManager(
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
@@ -350,9 +353,10 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         ImpressionSynchronizer(apis['impressions'], storages['impressions'],
                                cfg['impressionsBulkSize']),
         EventSynchronizer(apis['events'], storages['events'], cfg['eventsBulkSize']),
-        ImpressionsCountSynchronizer(apis['impressions'], imp_manager),
+        ImpressionsCountSynchronizer(apis['impressions'], imp_counter),
+        unique_keys_synchronizer,
+        clear_filter_sync
     )
-    imp_count_sync_task = ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters) if cfg['impressionsMode'] == 'OPTIMIZED' else None
 
     tasks = SplitTasks(
         SplitSynchronizationTask(
@@ -369,6 +373,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         ),
         EventsSyncTask(synchronizers.events_sync.synchronize_events, cfg['eventsPushRate']),
         ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters),
+        unique_keys_task,
+        clear_filter_task
     )
 
     synchronizer = Synchronizer(synchronizers, tasks)
@@ -438,6 +444,7 @@ def _build_redis_factory(api_key, cfg):
         recorder,
     )
 
+
 def _build_localhost_factory(cfg):
     """Build and return a localhost factory for testing/development purposes."""
     storages = {
@@ -464,7 +471,6 @@ def _build_localhost_factory(cfg):
     synchronizer = LocalhostSynchronizer(synchronizers, tasks)
     manager = Manager(ready_event, synchronizer, None, False, sdk_metadata)
     manager.start()
-
     recorder = StandardRecorder(
         ImpressionsManager(None, StrategyDebugMode()),
         storages['events'],

splitio/engine/sender_adapters/in_memory_sender_adapter.py

@@ -33,7 +33,6 @@ def _uniques_formatter(self, uniques):
         :return: unique keys JSON
         :rtype: json
         """
-        formatted_uniques = json.loads('{"keys": []}')
         if len(uniques) == 0:
             return json.loads('{"keys": []}')
 

splitio/engine/strategies/strategy_none_mode.py

@@ -0,0 +1,35 @@
+from splitio.engine.strategies.base_strategy import BaseStrategy
+from splitio.engine.strategies import  Counter, truncate_time
+from splitio.engine.unique_keys_tracker import UniqueKeysTracker
+from splitio import util
+
+_IMPRESSION_OBSERVER_CACHE_SIZE = 500000
+_UNIQUE_KEYS_CACHE_SIZE = 30000
+
+class StrategyNoneMode(BaseStrategy):
+    """Debug mode strategy."""
+
+    def __init__(self, counter):
+        """
+        Construct a strategy instance for none mode.
+
+        """
+        self._counter = counter
+        self._unique_keys_tracker = UniqueKeysTracker(_UNIQUE_KEYS_CACHE_SIZE)
+
+    def process_impressions(self, impressions):
+        """
+        Process impressions.
+
+        Impressions are analyzed to see if they've been seen before and counted.
+        Unique keys tracking are updated.
+
+        :param impressions: List of impression objects with attributes
+        :type impressions: list[tuple[splitio.models.impression.Impression, dict]]
+
+        :returns: Empty list, no impressions to post
+        :rtype: list[]
+        """
+        self._counter.track([imp for imp, _ in impressions])
+        [self._unique_keys_tracker.track(i.matching_key, i.feature_name) for i, _ in impressions]
+        return [], impressions

splitio/sync/impression.py

@@ -4,7 +4,6 @@
 from splitio.api import APIException
 from splitio.engine.strategies import Counter
 
-
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -69,22 +68,26 @@ def synchronize_impressions(self):
 
 
 class ImpressionsCountSynchronizer(object):
-    def __init__(self, impressions_api, impressions_counter):
+    def __init__(self, impressions_api, imp_counter):
         """
         Class constructor.
 
         :param impressions_api: Impressions Api object to send data to the backend
         :type impressions_api: splitio.api.impressions.ImpressionsAPI
         :param impressions_manager: Impressions manager instance
-        :type impressions_counter: splitio.engine.strategies
+        :type impressions_manager: splitio.engine.impressions.Manager
 
         """
         self._impressions_api = impressions_api
-        self._impressions_counter = impressions_counter
+        self._impressions_counter = imp_counter
 
     def synchronize_counters(self):
         """Send impressions from both the failed and new queues."""
-        to_send = self._impressions_manager._strategy._counter.pop_all()
+
+        if not isinstance(self._impressions_counter, Counter):
+            return
+
+        to_send = self._impressions_counter.pop_all()
         if not to_send:
             return
 

splitio/sync/synchronizer.py

@@ -14,7 +14,7 @@ class SplitSynchronizers(object):
     """SplitSynchronizers."""
 
     def __init__(self, split_sync, segment_sync, impressions_sync, events_sync,  # pylint:disable=too-many-arguments
-                 impressions_count_sync):
+                 impressions_count_sync, unique_keys_sync = None, clear_filter_sync = None):
         """
         Class constructor.
 
@@ -34,6 +34,8 @@ def __init__(self, split_sync, segment_sync, impressions_sync, events_sync,  # p
         self._impressions_sync = impressions_sync
         self._events_sync = events_sync
         self._impressions_count_sync = impressions_count_sync
+        self._unique_keys_sync = unique_keys_sync
+        self._clear_filter_sync = clear_filter_sync
 
     @property
     def split_sync(self):
@@ -60,11 +62,21 @@ def impressions_count_sync(self):
         """Return impressions count synchonizer."""
         return self._impressions_count_sync
 
+    @property
+    def unique_keys_sync(self):
+        """Return unique keys synchonizer."""
+        return self._unique_keys_sync
+
+    @property
+    def clear_filter_sync(self):
+        """Return clear filter synchonizer."""
+        return self._clear_filter_sync
+
 class SplitTasks(object):
     """SplitTasks."""
 
     def __init__(self, split_task, segment_task, impressions_task, events_task,  # pylint:disable=too-many-arguments
-                 impressions_count_task):
+                 impressions_count_task, unique_keys_task = None, clear_filter_task = None):
         """
         Class constructor.
 
@@ -84,6 +96,8 @@ def __init__(self, split_task, segment_task, impressions_task, events_task,  # p
         self._impressions_task = impressions_task
         self._events_task = events_task
         self._impressions_count_task = impressions_count_task
+        self._unique_keys_task = unique_keys_task
+        self._clear_filter_task = clear_filter_task
 
     @property
     def split_task(self):
@@ -110,6 +124,16 @@ def impressions_count_task(self):
         """Return impressions count sync task."""
         return self._impressions_count_task
 
+    @property
+    def unique_keys_task(self):
+        """Return unique keys sync task."""
+        return self._unique_keys_task
+
+    @property
+    def clear_filter_task(self):
+        """Return clear filter sync task."""
+        return self._clear_filter_task
+
 class BaseSynchronizer(object, metaclass=abc.ABCMeta):
     """Synchronizer interface."""
 
@@ -300,8 +324,11 @@ def start_periodic_data_recording(self):
         _LOGGER.debug('Starting periodic data recording')
         self._split_tasks.impressions_task.start()
         self._split_tasks.events_task.start()
-        if self._split_tasks.impressions_count_task is not None:
-            self._split_tasks.impressions_count_task.start()
+        self._split_tasks.impressions_count_task.start()
+        if self._split_tasks.unique_keys_task is not None:
+            self._split_tasks.unique_keys_task.start()
+        if self._split_tasks.clear_filter_task is not None:
+            self._split_tasks.clear_filter_task.start()
 
     def stop_periodic_data_recording(self, blocking):
         """
@@ -316,6 +343,11 @@ def stop_periodic_data_recording(self, blocking):
             tasks = [self._split_tasks.impressions_task,
                     self._split_tasks.events_task,
                     self._split_tasks.impressions_count_task]
+            if self._split_tasks.unique_keys_task is not None:
+                tasks.append(self._split_tasks.unique_keys_task)
+            if self._split_tasks.clear_filter_task is not None:
+                tasks.append(self._split_tasks.clear_filter_task)
+
             for task in tasks:
                 stop_event = threading.Event()
                 task.stop(stop_event)
@@ -325,8 +357,11 @@ def stop_periodic_data_recording(self, blocking):
         else:
             self._split_tasks.impressions_task.stop()
             self._split_tasks.events_task.stop()
-            if self._split_tasks.impressions_count_task is not None:
-                self._split_tasks.impressions_count_task.stop()
+            self._split_tasks.impressions_count_task.stop()
+            if self._split_tasks.unique_keys_task is not None:
+                self._split_tasks.unique_keys_task.stop()
+            if self._split_tasks.clear_filter_task is not None:
+                self._split_tasks.clear_filter_task.stop()
 
     def kill_split(self, split_name, default_treatment, change_number):
         """
@@ -342,8 +377,6 @@ def kill_split(self, split_name, default_treatment, change_number):
         self._split_synchronizers.split_sync.kill_split(split_name, default_treatment,
                                                         change_number)
 
-
-
 class LocalhostSynchronizer(BaseSynchronizer):
     """LocalhostSynchronizer."""