Merge pull request #282 from splitio/fix-redis-impressions-count

fixed impressions.count key and ttl in redis

Author: chillaq

CHANGES.txt

@@ -1,3 +1,6 @@
+9.2.1 (Nov 29, 2022)
+- Changed redis record type for impressions counts from list using rpush to hashed key using hincrby
+
 9.2.0 (Oct 14, 2022)
 - Added a new impressions mode for the SDK called NONE , to be used in factory when there is no desire to capture impressions on an SDK factory to feed Split's analytics engine. Running NONE mode, the SDK will only capture unique keys evaluated for a particular feature flag instead of full blown impressions
 

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:

splitio/version.py

@@ -1 +1 @@
-__version__ = '9.2.0'
+__version__ = '9.2.1-rc'

tests/client/test_factory.py

@@ -100,7 +100,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,
@@ -121,8 +121,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."""