flag as processed after request forwarded to next service

Author: jameswnl

collector/host_inventory.py

@@ -109,6 +109,7 @@ def worker(_: str, source_id: str, dest: str, b64_identity: str) -> None:
     prometheus_metrics.METRICS['posts'].inc()
     try:
         utils.retryable('post', dest, json=data, headers=headers)
+        utils.set_processed(account_id)
         prometheus_metrics.METRICS['post_successes'].inc()
     except utils.RetryFailedError as exception:
         LOGGER.error(

collector/utils.py

@@ -17,12 +17,14 @@
 REDIS = redis.Redis(**json.loads(REDIS_ENV), password=REDIS_PASSWORD)
 
 
-def processed(account_id: str) -> bool:
+def processed(key: str) -> bool:
     """If an account has been processed within the window."""
-    if REDIS.incr(account_id) == 1:
-        REDIS.expire(account_id, PROCESS_WINDOW)
-        return False
-    return True
+    return REDIS.get(key)
+
+
+def set_processed(key: str) -> None:
+    """Flag an account as processed with TTL."""
+    REDIS.set(key, 1, ex=PROCESS_WINDOW)
 
 
 class RetryFailedError(requests.HTTPError):

tests/collector/test_host_inventory.py

@@ -112,17 +112,25 @@ class TestRedisLookup:
     def test_not_processed_yet(self, mocker):
         """When account not processed before."""
         redis = mocker.MagicMock()
-        redis.incr.return_value = 1
+        redis.get.return_value = None
         mocker.patch.object(collector.utils, 'REDIS', redis)
         assert not collector.utils.processed('abc')
-        redis.expire.assert_called_once_with(
-            'abc',
-            collector.utils.PROCESS_WINDOW
-        )
 
     def test_processed_before(self, mocker):
         """When account processed before."""
         redis = mocker.MagicMock()
-        redis.incr.return_value = 2
+        redis.get.return_value = 1
         mocker.patch.object(collector.utils, 'REDIS', redis)
         assert collector.utils.processed('abc')
+
+    def test_set_processed(self, mocker):
+        """Flag an account as processed before."""
+        redis = mocker.MagicMock()
+        redis.set.return_value = True
+        mocker.patch.object(collector.utils, 'REDIS', redis)
+        collector.utils.set_processed('abc')
+        redis.set.assert_called_once_with(
+            'abc',
+            1,
+            ex=collector.utils.PROCESS_WINDOW,
+        )