Merge pull request #326 from splitio/expire-for-imps-and-events

Added expire to impressions and events keys

Author: chillaq

splitio/storage/pluggable.py

@@ -556,7 +556,8 @@ def put(self, impressions):
         """
         bulk_impressions = self._wrap_impressions(impressions)
         try:
-            self._pluggable_adapter.push_items(self._impressions_queue_key, *bulk_impressions)
+            total_keys = self._pluggable_adapter.push_items(self._impressions_queue_key, *bulk_impressions)
+            self.expire_key(total_keys, len(bulk_impressions))
             return True
         except Exception:
             _LOGGER.error('Something went wrong when trying to add impression to storage')
@@ -645,7 +646,8 @@ def put(self, events):
         """
         to_store = self._wrap_events(events)
         try:
-            self._pluggable_adapter.push_items(self._events_queue_key, *to_store)
+            total_keys = self._pluggable_adapter.push_items(self._events_queue_key, *to_store)
+            self.expire_key(total_keys, len(to_store))
             return True
         except Exception:
             _LOGGER.error('Something went wrong when trying to add event to redis')

tests/storage/test_pluggable.py

@@ -17,6 +17,7 @@
 class StorageMockAdapter(object):
     def __init__(self):
         self._keys = {}
+        self._expire = {}
         self._lock = threading.RLock()
 
     def get(self, key):
@@ -42,6 +43,7 @@ def push_items(self, key, *value):
                 items = self._keys[key]
             [items.append(item) for item in value]
             self._keys[key] = items
+            return len(self._keys[key])
 
     def delete(self, key):
         with self._lock:
@@ -56,7 +58,6 @@ def pop_items(self, key):
             del self._keys[key]
             return items
 
-
     def increment(self, key, value):
         with self._lock:
             if key not in self._keys:
@@ -116,8 +117,12 @@ def get_items_count(self, key):
             return None
 
     def expire(self, key, ttl):
-        #Not needed for Memory storage
-        pass
+        with self._lock:
+            if key in self._expire:
+                self._expire[key] = -1
+            else:
+                self._expire[key] = ttl
+            # should only be called once per key.
 
 class PluggableSplitStorageTests(object):
     """In memory split storage test cases."""
@@ -381,15 +386,16 @@ def test_put(self):
             Impression('key3', 'feature2', 'on', 'some_label', 123456, 'buck1', 321654),
             Impression('key4', 'feature1', 'on', 'some_label', 123456, 'buck1', 321654)
         ]
-        self.pluggable_imp_storage.put(impressions)
+        assert(self.pluggable_imp_storage.put(impressions))
         assert(self.pluggable_imp_storage._impressions_queue_key in self.mock_adapter._keys)
         assert(self.mock_adapter._keys["myprefix.SPLITIO.impressions"] == self.pluggable_imp_storage._wrap_impressions(impressions))
+        assert(self.mock_adapter._expire["myprefix.SPLITIO.impressions"] == PluggableImpressionsStorage.IMPRESSIONS_KEY_DEFAULT_TTL)
 
         impressions2 = [
             Impression('key5', 'feature1', 'off', 'some_label', 123456, 'buck1', 321654),
             Impression('key6', 'feature2', 'off', 'some_label', 123456, 'buck1', 321654),
         ]
-        self.pluggable_imp_storage.put(impressions2)
+        assert(self.pluggable_imp_storage.put(impressions2))
         assert(self.mock_adapter._keys["myprefix.SPLITIO.impressions"] == self.pluggable_imp_storage._wrap_impressions(impressions + impressions2))
 
     def test_wrap_impressions(self):
@@ -480,15 +486,16 @@ def test_put(self):
             EventWrapper(event=Event('key3', 'user', 'purchase', 10, 123456, None),  size=32768),
             EventWrapper(event=Event('key4', 'user', 'purchase', 10, 123456, None),  size=32768),
         ]
-        self.pluggable_events_storage.put(events)
+        assert(self.pluggable_events_storage.put(events))
         assert(self.pluggable_events_storage._events_queue_key in self.mock_adapter._keys)
         assert(self.mock_adapter._keys["myprefix.SPLITIO.events"] == self.pluggable_events_storage._wrap_events(events))
+        assert(self.mock_adapter._expire["myprefix.SPLITIO.events"] == PluggableEventsStorage._EVENTS_KEY_DEFAULT_TTL)
 
         events2 = [
             EventWrapper(event=Event('key5', 'user', 'purchase', 10, 123456, None),  size=32768),
             EventWrapper(event=Event('key6', 'user', 'purchase', 10, 123456, None),  size=32768),
         ]
-        self.pluggable_events_storage.put(events2)
+        assert(self.pluggable_events_storage.put(events2))
         assert(self.mock_adapter._keys["myprefix.SPLITIO.events"] == self.pluggable_events_storage._wrap_events(events + events2))
 
     def test_wrap_events(self):