ported fixes from MTK and updated tests

Author: chillaq

splitio/client/factory.py

@@ -11,11 +11,12 @@
 from splitio.client.config import sanitize as sanitize_config, DEFAULT_DATA_SAMPLING
 from splitio.client import util
 from splitio.client.listener import ImpressionListenerWrapper
-from splitio.engine.impressions import Manager as ImpressionsManager
+from splitio.engine.impressions.impressions import Manager as ImpressionsManager
 from splitio.engine.impressions import ImpressionsMode
-from splitio.engine.manager import Counter as ImpressionsCounter
-from splitio.engine.strategies import StrategyNoneMode, StrategyDebugMode, StrategyOptimizedMode
-from splitio.engine.adapters import InMemorySenderAdapter, RedisSenderAdapter
+from splitio.engine.impressions.manager import Counter as ImpressionsCounter
+from splitio.engine.impressions.strategies import StrategyNoneMode, StrategyDebugMode, StrategyOptimizedMode
+from splitio.engine.impressions.adapters import InMemorySenderAdapter, RedisSenderAdapter
+from splitio.engine.impressions import set_classes
 
 # Storage
 from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
@@ -291,17 +292,17 @@ def _wrap_impression_listener(listener, metadata):
     return None
 
 
-def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pylint:disable=too-many-arguments,too-many-locals
+def _build_in_memory_factory(api_key, cfg, extra_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, 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
 
-    cfg['sdk_url'] = sdk_url
-    cfg['events_url'] = events_url
-    cfg['auth_url'] = auth_api_base_url
-    cfg['streaming_url'] = streaming_api_base_url
-    cfg['telemetry_api_url'] = telemetry_api_base_url
+    extra_cfg['sdk_url'] = sdk_url
+    extra_cfg['events_url'] = events_url
+    extra_cfg['auth_url'] = auth_api_base_url
+    extra_cfg['streaming_url'] = streaming_api_base_url
+    extra_cfg['telemetry_api_url'] = telemetry_api_base_url
 
     http_client = HttpClient(
         sdk_url=sdk_url,
@@ -330,30 +331,10 @@ 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
-
-    unique_keys_synchronizer = None
-    clear_filter_sync = None
-    unique_keys_task = None
-    clear_filter_task = None
-    impressions_count_sync = None
-    impressions_count_task = None
-
-    if cfg['impressionsMode'] == ImpressionsMode.NONE:
-        imp_strategy = StrategyNoneMode(imp_counter)
-        clear_filter_sync = ClearFilterSynchronizer(imp_strategy.get_unique_keys_tracker())
-        unique_keys_synchronizer = UniqueKeysSynchronizer(InMemorySenderAdapter(apis['telemetry']), imp_strategy.get_unique_keys_tracker())
-        unique_keys_task = UniqueKeysSyncTask(unique_keys_synchronizer.send_all)
-        clear_filter_task = ClearFilterSyncTask(clear_filter_sync.clear_all)
-        imp_strategy.get_unique_keys_tracker().set_queue_full_hook(unique_keys_task.flush)
-        impressions_count_sync = ImpressionsCountSynchronizer(apis['impressions'], imp_counter)
-        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
-    elif cfg['impressionsMode'] == ImpressionsMode.DEBUG:
-        imp_strategy = StrategyDebugMode()
-    else:
-        imp_strategy = StrategyOptimizedMode(imp_counter)
-        impressions_count_sync = ImpressionsCountSynchronizer(apis['impressions'], imp_counter)
-        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
+
+    unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
+    clear_filter_task, impressions_count_sync, impressions_count_task, \
+    imp_strategy = set_classes('MEMORY', cfg['impressionsMode'], apis)
 
     imp_manager = ImpressionsManager(
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
@@ -438,31 +419,9 @@ def _build_redis_factory(api_key, cfg):
                         _MIN_DEFAULT_DATA_SAMPLING_ALLOWED)
         data_sampling = _MIN_DEFAULT_DATA_SAMPLING_ALLOWED
 
-    unique_keys_synchronizer = None
-    clear_filter_sync = None
-    unique_keys_task = None
-    clear_filter_task = None
-    impressions_count_sync = None
-    impressions_count_task = None
-    redis_sender_adapter = RedisSenderAdapter(redis_adapter)
-
-    if cfg['impressionsMode'] == ImpressionsMode.NONE:
-        imp_counter = ImpressionsCounter()
-        imp_strategy = StrategyNoneMode(imp_counter)
-        clear_filter_sync = ClearFilterSynchronizer(imp_strategy.get_unique_keys_tracker())
-        unique_keys_synchronizer = UniqueKeysSynchronizer(redis_sender_adapter, imp_strategy.get_unique_keys_tracker())
-        unique_keys_task = UniqueKeysSyncTask(unique_keys_synchronizer.send_all)
-        clear_filter_task = ClearFilterSyncTask(clear_filter_sync.clear_all)
-        imp_strategy.get_unique_keys_tracker().set_queue_full_hook(unique_keys_task.flush)
-        impressions_count_sync = ImpressionsCountSynchronizer(redis_sender_adapter, imp_counter)
-        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
-    elif cfg['impressionsMode'] == ImpressionsMode.DEBUG:
-        imp_strategy = StrategyDebugMode()
-    else:
-        imp_counter = ImpressionsCounter()
-        imp_strategy = StrategyOptimizedMode(imp_counter)
-        impressions_count_sync = ImpressionsCountSynchronizer(redis_sender_adapter, imp_counter)
-        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
+    unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
+    clear_filter_task, impressions_count_sync, impressions_count_task, \
+    imp_strategy = set_classes('REDIS', cfg['impressionsMode'], redis_adapter)
 
     imp_manager = ImpressionsManager(
         _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
@@ -543,17 +502,12 @@ def _build_localhost_factory(cfg):
         ready_event
     )
 
-
 def get_factory(api_key, **kwargs):
     """Build and return the appropriate factory."""
     try:
-        active_factory_count = 0
-        redundant_factory_count = 0
         _INSTANTIATED_FACTORIES_LOCK.acquire()
         if _INSTANTIATED_FACTORIES:
-            active_factory_count = active_factory_count + 1
             if api_key in _INSTANTIATED_FACTORIES:
-                redundant_factory_count = redundant_factory_count + 1
                 _LOGGER.warning(
                     "factory instantiation: You already have %d %s with this API Key. "
                     "We recommend keeping only one instance of the factory at all times "
@@ -570,8 +524,7 @@ def get_factory(api_key, **kwargs):
                 )
 
         config = sanitize_config(api_key, kwargs.get('config', {}))
-        config['redundantFactoryCount'] = redundant_factory_count
-        config['activeFactoryCount'] = active_factory_count
+        extra_config = {}
 
         if config['operationMode'] == 'localhost-standalone':
             return _build_localhost_factory(config)
@@ -582,12 +535,20 @@ def get_factory(api_key, **kwargs):
         return _build_in_memory_factory(
             api_key,
             config,
+            extra_config,
             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('telemetry_api_base_url')
         )
     finally:
+        redundant_factory_count = 0
+        active_factory_count = 0
         _INSTANTIATED_FACTORIES.update([api_key])
+        for item in _INSTANTIATED_FACTORIES:
+            redundant_factory_count = redundant_factory_count + _INSTANTIATED_FACTORIES[item] - 1
+            active_factory_count = active_factory_count + _INSTANTIATED_FACTORIES[item]
+        extra_config['redundant_factory_count'] = redundant_factory_count
+        extra_config['active_factory_count'] = active_factory_count
         _INSTANTIATED_FACTORIES_LOCK.release()

splitio/engine/filters.py

@@ -1,4 +1,5 @@
 import abc
+import threading
 
 from bloom_filter2 import BloomFilter as BloomFilter2
 
@@ -45,6 +46,7 @@ def __init__(self, max_elements=5000, error_rate=0.01):
         self._max_elements = max_elements
         self._error_rate = error_rate
         self._imps_bloom_filter = BloomFilter2(max_elements=self._max_elements, error_rate=self._error_rate)
+        self._lock = threading.RLock()
 
     def add(self, data):
         """
@@ -56,8 +58,9 @@ def add(self, data):
         :return: True if successful
         :rtype: boolean
         """
-        self._imps_bloom_filter.add(data)
-        return data in self._imps_bloom_filter
+        with self._lock:
+            self._imps_bloom_filter.add(data)
+            return data in self._imps_bloom_filter
 
     def contains(self, data):
         """
@@ -69,12 +72,14 @@ def contains(self, data):
         :return: True if exist
         :rtype: boolean
         """
-        return data in self._imps_bloom_filter
+        with self._lock:
+            return data in self._imps_bloom_filter
 
     def clear(self):
         """
         Destroy the current filter instance and create new one.
 
         """
-        self._imps_bloom_filter.close()
-        self._imps_bloom_filter = BloomFilter2(max_elements=self._max_elements, error_rate=self._error_rate)
+        with self._lock:
+            self._imps_bloom_filter.close()
+            self._imps_bloom_filter = BloomFilter2(max_elements=self._max_elements, error_rate=self._error_rate)

splitio/engine/impressions/__init__.py

@@ -0,0 +1,45 @@
+from splitio.engine.impressions.impressions import ImpressionsMode
+from splitio.engine.impressions.manager import Counter as ImpressionsCounter
+from splitio.engine.impressions.strategies import StrategyNoneMode, StrategyDebugMode, StrategyOptimizedMode
+from splitio.engine.impressions.adapters import InMemorySenderAdapter, RedisSenderAdapter
+from splitio.tasks.unique_keys_sync import UniqueKeysSyncTask, ClearFilterSyncTask
+from splitio.sync.unique_keys import UniqueKeysSynchronizer, ClearFilterSynchronizer
+from splitio.sync.impression import ImpressionsCountSynchronizer
+from splitio.tasks.impressions_sync import ImpressionsCountSyncTask
+
+
+def set_classes(storage_mode, impressions_mode, api_adapter):
+    unique_keys_synchronizer = None
+    clear_filter_sync = None
+    unique_keys_task = None
+    clear_filter_task = None
+    impressions_count_sync = None
+    impressions_count_task = None
+    if storage_mode == 'REDIS':
+        redis_sender_adapter = RedisSenderAdapter(api_adapter)
+        api_telemetry_adapter = redis_sender_adapter
+        api_impressions_adapter = redis_sender_adapter
+    else:
+        api_telemetry_adapter = api_adapter['telemetry']
+        api_impressions_adapter = api_adapter['impressions']
+
+    if impressions_mode == ImpressionsMode.NONE:
+        imp_counter = ImpressionsCounter()
+        imp_strategy = StrategyNoneMode(imp_counter)
+        clear_filter_sync = ClearFilterSynchronizer(imp_strategy.get_unique_keys_tracker())
+        unique_keys_synchronizer = UniqueKeysSynchronizer(InMemorySenderAdapter(api_telemetry_adapter), imp_strategy.get_unique_keys_tracker())
+        unique_keys_task = UniqueKeysSyncTask(unique_keys_synchronizer.send_all)
+        clear_filter_task = ClearFilterSyncTask(clear_filter_sync.clear_all)
+        imp_strategy.get_unique_keys_tracker().set_queue_full_hook(unique_keys_task.flush)
+        impressions_count_sync = ImpressionsCountSynchronizer(api_impressions_adapter, imp_counter)
+        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
+    elif impressions_mode == ImpressionsMode.DEBUG:
+        imp_strategy = StrategyDebugMode()
+    else:
+        imp_counter = ImpressionsCounter()
+        imp_strategy = StrategyOptimizedMode(imp_counter)
+        impressions_count_sync = ImpressionsCountSynchronizer(api_impressions_adapter, imp_counter)
+        impressions_count_task = ImpressionsCountSyncTask(impressions_count_sync.synchronize_counters)
+
+    return unique_keys_synchronizer, clear_filter_sync, unique_keys_task, clear_filter_task, \
+            impressions_count_sync, impressions_count_task, imp_strategy

splitio/engine/adapters.py splitio/engine/impressions/adapters.pysplitio/engine/adapters.py renamed to splitio/engine/impressions/adapters.py

@@ -135,12 +135,12 @@ def _build_counters(self, counters):
         :return: dict with list of impression count dtos
         :rtype: dict
         """
-        return [json.dumps({
+        return json.dumps({
             'pf': [
                 {
                     'f': pf_count.feature,
                     'm': pf_count.timeframe,
                     'rc': pf_count.count
                 } for pf_count in counters
             ]
-        })]
+        })

splitio/engine/impressions.py splitio/engine/impressions/impressions.pysplitio/engine/impressions.py renamed to splitio/engine/impressions/impressions.py

@@ -17,14 +17,11 @@ def __init__(self, listener=None, strategy=None):
         """
         Construct a manger to track and forward impressions to the queue.
 
-        :param mode: Impressions capturing mode.
-        :type mode: ImpressionsMode
-
-        :param standalone: whether the SDK is running in standalone sending impressions by itself
-        :type standalone: bool
-
         :param listener: Optional impressions listener that will capture all seen impressions.
         :type listener: splitio.client.listener.ImpressionListenerWrapper
+
+        :param strategy: Impressions stragetgy instance
+        :type strategy: (BaseStrategy)
         """
 
         self._strategy = strategy

splitio/engine/manager.py splitio/engine/impressions/manager.pysplitio/engine/manager.py renamed to splitio/engine/impressions/manager.py

@@ -1,7 +1,7 @@
 import abc
 
-from splitio.engine.manager import Observer, truncate_impressions_time, Counter, truncate_time
-from splitio.engine.unique_keys_tracker import UniqueKeysTracker
+from splitio.engine.impressions.manager import Observer, truncate_impressions_time, Counter, truncate_time
+from splitio.engine.impressions.unique_keys_tracker import UniqueKeysTracker
 from splitio import util
 
 _IMPRESSION_OBSERVER_CACHE_SIZE = 500000

splitio/engine/strategies.py splitio/engine/impressions/strategies.pysplitio/engine/strategies.py renamed to splitio/engine/impressions/strategies.py

@@ -48,8 +48,6 @@ def track(self, key, feature_name):
         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
@@ -72,9 +70,11 @@ def _add_or_update(self, feature_name, key):
         :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)
+
+        with self._lock:
+            if feature_name not in self._cache:
+                self._cache[feature_name] = set()
+            self._cache[feature_name].add(key)
 
     def set_queue_full_hook(self, hook):
         """
@@ -85,7 +85,7 @@ def set_queue_full_hook(self, hook):
         if callable(hook):
             self._queue_full_hook = hook
 
-    def filter_pop_all(self):
+    def clear_filter(self):
         """
         Delete the filter items
 

splitio/engine/unique_keys_tracker.py …ngine/impressions/unique_keys_tracker.pysplitio/engine/unique_keys_tracker.py renamed to splitio/engine/impressions/unique_keys_tracker.py splitio/engine/unique_keys_tracker.py renamed to splitio/engine/impressions/unique_keys_tracker.py

@@ -129,7 +129,7 @@ def record_treatment_stats(self, impressions, latency, operation):
                 if self._data_sampling < rnumber:
                     return
             impressions = self._impressions_manager.process_impressions(impressions)
-            if impressions == []:
+            if not impressions:
                 return
             # pipe = self._make_pipe()
             # self._impression_storage.add_impressions_to_pipe(impressions, pipe)

splitio/recorder/recorder.py

@@ -229,9 +229,9 @@ def __init__(self, split_synchronizers, split_tasks):
         ]
         if self._split_tasks.impressions_count_task:
             self._periodic_data_recording_tasks.append(self._split_tasks.impressions_count_task)
-        if self._split_tasks.unique_keys_task is not None:
+        if self._split_tasks.unique_keys_task:
             self._periodic_data_recording_tasks.append(self._split_tasks.unique_keys_task)
-        if self._split_tasks.clear_filter_task is not None:
+        if self._split_tasks.clear_filter_task:
             self._periodic_data_recording_tasks.append(self._split_tasks.clear_filter_task)