Merge branch 'development' into invalid-apikey-timeout

Author: chillaq

CHANGES.txt

@@ -1,4 +1,4 @@
-9.2.1 (Nov 29, 2022)
+9.2.1 (Dec 1, 2022)
 - Changed redis record type for impressions counts from list using rpush to hashed key using hincrby.
 - Apply Timeout Exception when incorrect SDK API Key is used.
 - Changed potential initial fetching segment Warning to Debug in logging.

splitio/engine/impressions/adapters.py

@@ -66,6 +66,8 @@ def __init__(self, redis_client):
         :type telemtry_http_client: splitio.api.telemetry.TelemetryAPI
         """
         self._redis_client = redis_client
+        self.pipe = self._redis_client.pipeline()
+
 
     def record_unique_keys(self, uniques):
         """
@@ -88,13 +90,17 @@ def flush_counters(self, to_send):
         """
         post the impression counters to redis.
 
-        :param uniques: unique keys disctionary
-        :type uniques: Dictionary {'feature1': set(), 'feature2': set(), .. }
+        :param to_send: unique keys disctionary
+        :type to_send: Dictionary {'feature1': set(), 'feature2': set(), .. }
         """
-        bulk_counts = self._build_counters(to_send)
         try:
-            inserted = self._redis_client.rpush(self.IMP_COUNT_QUEUE_KEY, *bulk_counts)
-            self._expire_keys(self.IMP_COUNT_QUEUE_KEY, self.IMP_COUNT_KEY_DEFAULT_TTL, inserted, len(bulk_counts))
+            resulted = 0
+            counted = 0
+            for pf_count in to_send:
+                self.pipe.hincrby(self.IMP_COUNT_QUEUE_KEY, pf_count.feature + "::" + str(pf_count.timeframe), pf_count.count)
+                counted += pf_count.count
+            resulted = sum(self.pipe.execute())
+            self._expire_keys(self.IMP_COUNT_QUEUE_KEY, self.IMP_COUNT_KEY_DEFAULT_TTL, resulted, counted)
             return True
         except RedisAdapterException:
             _LOGGER.error('Something went wrong when trying to add counters to redis')
@@ -124,23 +130,3 @@ def _uniques_formatter(self, uniques):
         :rtype: json
         """
         return [json.dumps({'f': feature, 'ks': list(keys)}) for feature, keys in uniques.items()]
-
-    def _build_counters(self, counters):
-        """
-        Build an impression bulk formatted as the API expects it.
-
-        :param counters: List of impression counters per feature.
-        :type counters: list[splitio.engine.impressions.Counter.CountPerFeature]
-
-        :return: dict with list of impression count dtos
-        :rtype: dict
-        """
-        return json.dumps({
-            'pf': [
-                {
-                    'f': pf_count.feature,
-                    'm': pf_count.timeframe,
-                    'rc': pf_count.count
-                } for pf_count in counters
-            ]
-        })

splitio/storage/adapters/redis.py

@@ -241,6 +241,13 @@ def hget(self, name, key):
         except RedisError as exc:
             raise RedisAdapterException('Error executing hget operation') from exc
 
+    def hincrby(self, name, key, amount=1):
+        """Mimic original redis function but using user custom prefix."""
+        try:
+            return self._decorated.hincrby(self._prefix_helper.add_prefix(name), key, amount)
+        except RedisError as exc:
+            raise RedisAdapterException('Error executing hincrby operation') from exc
+
     def incr(self, name, amount=1):
         """Mimic original redis function but using user custom prefix."""
         try:
@@ -323,6 +330,10 @@ def incr(self, name, amount=1):
         """Mimic original redis function but using user custom prefix."""
         self._pipe.incr(self._prefix_helper.add_prefix(name), amount)
 
+    def hincrby(self, name, key, amount=1):
+        """Mimic original redis function but using user custom prefix."""
+        self._pipe.hincrby(self._prefix_helper.add_prefix(name), key, amount)
+
     def execute(self):
         """Mimic original redis function but using user custom prefix."""
         try:

tests/client/test_factory.py

@@ -104,7 +104,7 @@ def test_redis_client_creation(self, mocker):
         assert adapter == factory._get_storage('impressions')._redis
         assert adapter == factory._get_storage('events')._redis
 
-        assert strict_redis_mock.mock_calls == [mocker.call(
+        assert strict_redis_mock.mock_calls[0] == mocker.call(
             host='some_host',
             port=1234,
             db=1,
@@ -125,8 +125,8 @@ def test_redis_client_creation(self, mocker):
             ssl_certfile='some_cert_file',
             ssl_cert_reqs='some_cert_req',
             ssl_ca_certs='some_ca_cert',
-            max_connections=999
-        )]
+            max_connections=999,
+        )
         assert factory._labels_enabled is False
         assert isinstance(factory._recorder, PipelinedRecorder)
         assert isinstance(factory._recorder._impressions_manager, ImpressionsManager)

tests/engine/test_send_adapters.py

@@ -56,22 +56,6 @@ def test_uniques_formatter(self, mocker):
         for i in range(0,1):
             assert(sorted(ast.literal_eval(sender_adapter._uniques_formatter(uniques)[i])["ks"]) == sorted(formatted[i]["ks"]))
 
-    def test_build_counters(self, mocker):
-        """Test formatting counters dict to json."""
-
-        counters = [
-            Counter.CountPerFeature('f1', 123, 2),
-            Counter.CountPerFeature('f2', 123, 123),
-        ]
-        formatted = [
-            {'f': 'f1', 'm': 123, 'rc': 2},
-            {'f': 'f2', 'm': 123, 'rc': 123},
-        ]
-
-        sender_adapter = RedisSenderAdapter(mocker.Mock())
-        for i in range(0,1):
-            assert(sorted(ast.literal_eval(sender_adapter._build_counters(counters))['pf'][i]) == sorted(formatted[i]))
-
     @mock.patch('splitio.storage.adapters.redis.RedisAdapter.rpush')
     def test_record_unique_keys(self, mocker):
         """Test sending unique keys."""
@@ -85,7 +69,7 @@ def test_record_unique_keys(self, mocker):
 
         assert(mocker.called)
 
-    @mock.patch('splitio.storage.adapters.redis.RedisAdapter.rpush')
+    @mock.patch('splitio.storage.adapters.redis.RedisPipelineAdapter.hincrby')
     def test_flush_counters(self, mocker):
         """Test sending counters."""