renamed and added test

Author: Matias Melograno

splitio/client/config.py

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

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, DEFAULT_DATA_THROTTLING
+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
@@ -57,7 +57,7 @@
 _LOGGER = logging.getLogger(__name__)
 _INSTANTIATED_FACTORIES = Counter()
 _INSTANTIATED_FACTORIES_LOCK = threading.RLock()
-_MIN_DEFAULT_DATA_THROTTLING_ALLOWED = 0.1  # 10%
+_MIN_DEFAULT_DATA_SAMPLING_ALLOWED = 0.1  # 10%
 
 
 class Status(Enum):
@@ -402,19 +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
+    data_sampling = cfg.get('dataSampling', DEFAULT_DATA_SAMPLING)
+    if data_sampling < _MIN_DEFAULT_DATA_SAMPLING_ALLOWED:
+        _LOGGER.warning("dataSampling cannot be less than %f, defaulting to minimum",
+                        _MIN_DEFAULT_DATA_SAMPLING_ALLOWED)
+        data_sampling = _MIN_DEFAULT_DATA_SAMPLING_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,
+        data_sampling,
     )
     return SplitFactory(
         api_key,

splitio/recorder/recorder.py

@@ -4,7 +4,7 @@
 import random
 
 
-from splitio.client.config import DEFAULT_DATA_THROTTLING
+from splitio.client.config import DEFAULT_DATA_SAMPLING
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -92,7 +92,7 @@ class PipelinedRecorder(StatsRecorder):
     """PipelinedRecorder class."""
 
     def __init__(self, pipe, impressions_manager, telemetry_storage, event_storage,
-                 impression_storage, data_throttling=DEFAULT_DATA_THROTTLING):
+                 impression_storage, data_sampling=DEFAULT_DATA_SAMPLING):
         """
         Class constructor.
 
@@ -106,15 +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
+        :param data_sampling: data sampling factor
+        :type data_sampling: 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_throttling = data_throttling
+        self._data_sampling = data_sampling
 
     def record_treatment_stats(self, impressions, latency, operation):
         """
@@ -131,9 +131,9 @@ def record_treatment_stats(self, impressions, latency, operation):
             # TODO @matias.melograno
             # Changing logic until TelemetryV2 released to avoid using pipelined operations
             # Deprecated Old Telemetry
-            if self._data_throttling < DEFAULT_DATA_THROTTLING:
+            if self._data_sampling < DEFAULT_DATA_SAMPLING:
                 rnumber = random.uniform(0, 1)
-                if self._data_throttling > rnumber:
+                if self._data_sampling < rnumber:
                     return
             impressions = self._impressions_manager.process_impressions(impressions)
             self._impression_storage.put(impressions)

tests/recorder/test_recorder.py

@@ -48,3 +48,33 @@ def test_pipelined_recorder(self, mocker):
         # 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
+
+    def test_sampled_recorder(self, mocker):
+        impressions = [
+            Impression('k1', 'f1', 'on', 'l1', 123, None, None),
+            Impression('k1', 'f2', 'on', 'l1', 123, None, None)
+        ]
+        redis = mocker.Mock(spec=RedisAdapter)
+        impmanager = mocker.Mock(spec=ImpressionsManager)
+        impmanager.process_impressions.return_value = impressions
+        telemetry = mocker.Mock(spec=TelemetryPipelinedStorage)
+        event = mocker.Mock(spec=EventStorage)
+        impression = mocker.Mock(spec=ImpressionStorage)
+        recorder = PipelinedRecorder(redis, impmanager, telemetry, event, impression, 0.5)
+
+        def put(x):
+            return
+
+        recorder._impression_storage.put.side_effect = put
+
+        for _ in range(100):
+            recorder.record_treatment_stats(impressions, 1, 'some')
+        print(recorder._impression_storage.put.call_count)
+        assert recorder._impression_storage.put.call_count < 80
+
+
+        # 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