Fixed several bugs and tested locally against staging, fixed existing tests.

TBD: creating new tests

Author: chillaq

splitio/api/telemetry.py

@@ -8,7 +8,6 @@
 
 _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 +33,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

@@ -12,9 +12,12 @@
 from splitio.client import util
 from splitio.client.listener import ImpressionListenerWrapper
 from splitio.engine.impressions import Manager as ImpressionsManager
+from splitio.engine.impressions import ImpressionsMode
+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
 from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
@@ -310,7 +313,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),
+        'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata),
     }
 
     if not input_validator.validate_apikey_type(apis['segments']):
@@ -322,13 +325,12 @@ 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 = Counter() if cfg['impressionsMode'] == 'OPTIMIZED' else None
+    imp_counter = ImpressionsCounter() if cfg['impressionsMode'] != ImpressionsMode.DEBUG else None
 
     strategies = {
-        'OPTIMIZED': StrategyOptimizedMode(imp_counter),
-        'DEBUG' : StrategyDebugMode(),
-        'NONE' : StrategyNoneMode(imp_counter),
+        ImpressionsMode.OPTIMIZED : StrategyOptimizedMode(imp_counter),
+        ImpressionsMode.DEBUG : StrategyDebugMode(),
+        ImpressionsMode.NONE : StrategyNoneMode(imp_counter),
     }
     imp_strategy = strategies[cfg['impressionsMode']]
 
@@ -362,9 +364,9 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters),
     )
 
-    if cfg['impressionsMode'] == 'NONE':
+    if cfg['impressionsMode'] == ImpressionsMode.NONE:
         synchronizers.set_none_syncs(
-            UniqueKeysSynchronizer(imp_strategy._unique_keys_tracker),
+            UniqueKeysSynchronizer(InMemorySenderAdapter(apis['telemetry']), imp_strategy._unique_keys_tracker),
             ClearFilterSynchronizer(imp_strategy._unique_keys_tracker),
         )
         tasks.set_none_tasks(
@@ -382,7 +384,8 @@ 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)
+    if cfg['impressionsMode'] == ImpressionsMode.NONE:
+        imp_strategy._unique_keys_tracker.set_queue_full_hook(tasks._unique_keys_task.flush)
 
     recorder = StandardRecorder(
         imp_manager,
@@ -421,10 +424,14 @@ def _build_redis_factory(api_key, cfg):
         _LOGGER.warning("dataSampling cannot be less than %.2f, defaulting to minimum",
                         _MIN_DEFAULT_DATA_SAMPLING_ALLOWED)
         data_sampling = _MIN_DEFAULT_DATA_SAMPLING_ALLOWED
+
+    imp_manager = ImpressionsManager(
+        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
+        StrategyDebugMode())
+
     recorder = PipelinedRecorder(
         redis_adapter.pipeline,
-        ImpressionsManager(cfg['impressionsMode'], False,
-                           _wrap_impression_listener(cfg['impressionListener'], sdk_metadata)),
+        imp_manager,
         storages['events'],
         storages['impressions'],
         data_sampling,
@@ -464,7 +471,7 @@ def _build_localhost_factory(cfg):
     manager = Manager(ready_event, synchronizer, None, False, sdk_metadata)
     manager.start()
     recorder = StandardRecorder(
-        ImpressionsManager(cfg['impressionsMode'], True, None),
+        ImpressionsManager(None, StrategyDebugMode()),
         storages['events'],
         storages['impressions'],
     )
@@ -513,7 +520,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,7 +8,7 @@ class ImpressionsMode(Enum):
 
     OPTIMIZED = "OPTIMIZED"
     DEBUG = "DEBUG"
-    NONE = 'NONE'
+    NONE = "NONE"
 
 class Manager(object):  # pylint:disable=too-few-public-methods
     """Impression manager."""

splitio/engine/sender_adapters/__init__.py

@@ -10,11 +10,3 @@ def record_unique_keys(self, data):
 
         """
         pass
-
-    @abc.abstractmethod
-    def record_impressions_count(self, data):
-        """
-        No Return value
-
-        """
-        pass

splitio/engine/sender_adapters/in_memory_sender_adapter.py

@@ -33,9 +33,14 @@ def _uniques_formatter(self, uniques):
         :return: unique keys JSON
         :rtype: json
         """
-        formatted_uniques = json.load('{keys: []}')
+        formatted_uniques = json.loads('{"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
+            return json.loads('{"keys": []}')
+
+        return {
+            'keys':  [{'f': feature, 'ks': list(keys)} for feature, keys in uniques.items()]
+        }
+#        for key in uniques:
+#            formatted_uniques["keys"].append(json.loads('{"f":"' + key +'", "ks":' + json.dumps(list(uniques[key])) + '}'))
+
+#        return formatted_uniques

splitio/engine/strategies/strategy_none_mode.py

@@ -34,6 +34,6 @@ def process_impressions(self, impressions):
         imps = [(self._observer.test_and_set(imp), attrs) for imp, attrs in impressions]
         self._counter.track([imp for imp, _ in imps])
         this_hour = truncate_time(util.utctime_ms())
-        [self._unique_keys_tracker(i.matching_key, i.feature_name) for i, _ in imps if i.previous_time is None or i.previous_time < this_hour], imps
+        [self._unique_keys_tracker.track(i.matching_key, i.feature_name) for i, _ in imps if i.previous_time is None or i.previous_time < this_hour]
 
-        return []
+        return [], imps

splitio/engine/strategies/strategy_optimized_mode.py

@@ -7,14 +7,15 @@
 class StrategyOptimizedMode(BaseStrategy):
     """Optimized mode strategy."""
 
-    def __init__(self, counter=None):
+    def __init__(self, counter):
         """
         Construct a strategy instance for optimized mode.
 
         """
         self._observer = Observer(_IMPRESSION_OBSERVER_CACHE_SIZE)
         self._counter = counter
 
+
     def process_impressions(self, impressions):
         """
         Process impressions.

splitio/engine/unique_keys_tracker.py

@@ -2,7 +2,6 @@
 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__)
 
@@ -32,6 +31,7 @@ def __init__(self, cache_size=30000):
         self._lock = threading.RLock()
         self._cache = {}
         self._queue_full_hook = None
+        self._current_cache_size = 0
 
     def track(self, key, feature_name):
         """
@@ -45,33 +45,23 @@ def track(self, key, feature_name):
         :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)
+            self._current_cache_size = self._current_cache_size + 1
 
-        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()
+        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):
-        """
-        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._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):
         """
         Add the feature_name+key to both bloom filter and dictionary.
@@ -93,3 +83,20 @@ def set_queue_full_hook(self, hook):
         """
         if callable(hook):
             self._queue_full_hook = hook
+
+    def filter_pop_all(self):
+        """
+        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

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