data_throttlong for imp

Author: Matias Melograno

splitio/client/config.py

@@ -6,6 +6,7 @@
 
 
 _LOGGER = logging.getLogger(__name__)
+DEFAULT_DATA_THROTTLING = 1
 
 
 DEFAULT_CONFIG = {
@@ -52,7 +53,7 @@
     'machineIp': None,
     'splitFile': os.path.join(os.path.expanduser('~'), '.split'),
     'preforkedInitialization': False,
-    'dataThrotling': 1,
+    'dataThrottling': DEFAULT_DATA_THROTTLING,
 }
 
 

splitio/client/factory.py

@@ -8,7 +8,7 @@
 from splitio.client.client import Client
 from splitio.client import input_validator
 from splitio.client.manager import SplitManager
-from splitio.client.config import sanitize as sanitize_config
+from splitio.client.config import sanitize as sanitize_config, DEFAULT_DATA_THROTTLING
 from splitio.client import util
 from splitio.client.listener import ImpressionListenerWrapper
 from splitio.engine.impressions import Manager as ImpressionsManager
@@ -57,6 +57,7 @@
 _LOGGER = logging.getLogger(__name__)
 _INSTANTIATED_FACTORIES = Counter()
 _INSTANTIATED_FACTORIES_LOCK = threading.RLock()
+_MIN_DEFAULT_DATA_THROTTLING_ALLOWED = 0.1  # 10%
 
 
 class Status(Enum):
@@ -401,13 +402,19 @@ def _build_redis_factory(api_key, cfg):
         'events': RedisEventsStorage(redis_adapter, sdk_metadata),
         'telemetry': RedisTelemetryStorage(redis_adapter, sdk_metadata)
     }
+    data_throttling = cfg.get('dataThrottling', DEFAULT_DATA_THROTTLING)
+    if data_throttling < _MIN_DEFAULT_DATA_THROTTLING_ALLOWED:
+        _LOGGER.warning("dataThrottling cannot be less than %f, defaulting to minimum",
+                        _MIN_DEFAULT_DATA_THROTTLING_ALLOWED)
+        data_throttling = _MIN_DEFAULT_DATA_THROTTLING_ALLOWED
     recorder = PipelinedRecorder(
         redis_adapter.pipeline,
         ImpressionsManager(cfg['impressionsMode'], False,
                            _wrap_impression_listener(cfg['impressionListener'], sdk_metadata)),
         storages['telemetry'],
         storages['events'],
         storages['impressions'],
+        data_throttling,
     )
     return SplitFactory(
         api_key,

splitio/recorder/recorder.py

@@ -4,8 +4,10 @@
 import random
 
 
+from splitio.client.config import DEFAULT_DATA_THROTTLING
+
+
 _LOGGER = logging.getLogger(__name__)
-_MIN_THROTLING = 1
 
 
 class StatsRecorder(object, metaclass=abc.ABCMeta):
@@ -89,7 +91,8 @@ def record_track_stats(self, event):
 class PipelinedRecorder(StatsRecorder):
     """PipelinedRecorder class."""
 
-    def __init__(self, pipe, impressions_manager, telemetry_storage, event_storage, impression_storage, data_throtling):
+    def __init__(self, pipe, impressions_manager, telemetry_storage, event_storage,
+                 impression_storage, data_throttling=DEFAULT_DATA_THROTTLING):
         """
         Class constructor.
 
@@ -103,13 +106,15 @@ def __init__(self, pipe, impressions_manager, telemetry_storage, event_storage,
         :type event_storage: splitio.storage.EventStorage
         :param impression_storage: impression storage instance
         :type impression_storage: splitio.storage.redis.RedisImpressionsStorage
+        :param data_throttling: data throttling factor
+        :type data_throttling: number
         """
         self._make_pipe = pipe
         self._impressions_manager = impressions_manager
         self._telemetry_storage = telemetry_storage
         self._event_sotrage = event_storage
         self._impression_storage = impression_storage
-        self._data_trothling = data_throtling
+        self._data_throttling = data_throttling
 
     def record_treatment_stats(self, impressions, latency, operation):
         """
@@ -123,11 +128,12 @@ def record_treatment_stats(self, impressions, latency, operation):
         :type operation: str
         """
         try:
+            # TODO @matias.melograno
             # Changing logic until TelemetryV2 released to avoid using pipelined operations
             # Deprecated Old Telemetry
-            if self._data_trothling < _MIN_THROTLING:
+            if self._data_throttling < DEFAULT_DATA_THROTTLING:
                 rnumber = random.uniform(0, 1)
-                if self._data_trothling > rnumber:
+                if self._data_throttling > rnumber:
                     return
             impressions = self._impressions_manager.process_impressions(impressions)
             self._impression_storage.put(impressions)

tests/recorder/test_recorder.py

@@ -39,10 +39,12 @@ def test_pipelined_recorder(self, mocker):
         impmanager.process_impressions.return_value = impressions
         telemetry = mocker.Mock(spec=TelemetryPipelinedStorage)
         event = mocker.Mock(spec=EventStorage)
-        impression = mocker.Mock(spec=ImpressionPipelinedStorage)
+        impression = mocker.Mock(spec=ImpressionStorage)
         recorder = PipelinedRecorder(redis, impmanager, telemetry, event, impression)
         recorder.record_treatment_stats(impressions, 1, 'some')
 
-        assert recorder._impression_storage.add_impressions_to_pipe.mock_calls[0][1][0] == impressions
-        assert recorder._telemetry_storage.add_latency_to_pipe.mock_calls[0][1][0] == 'some'
-        assert recorder._telemetry_storage.add_latency_to_pipe.mock_calls[0][1][1] == 1
+        # TODO @matias.melograno Commented until we implement TelemetryV2
+        # assert recorder._impression_storage.add_impressions_to_pipe.mock_calls[0][1][0] == impressions
+        # assert recorder._telemetry_storage.add_latency_to_pipe.mock_calls[0][1][0] == 'some'
+        # assert recorder._telemetry_storage.add_latency_to_pipe.mock_calls[0][1][1] == 1
+        assert recorder._impression_storage.put.mock_calls[0][1][0] == impressions