Merge pull request #399 from splitio/async-redis-event-storage

added redis event async storage class

Author: chillaq

splitio/storage/redis.py

@@ -686,24 +686,12 @@ async def put(self, impressions):
             return False
 
 
-class RedisEventsStorage(EventStorage):
-    """Redis based event storage class."""
+class RedisEventsStorageBase(EventStorage):
+    """Redis based event storage base class."""
 
     _EVENTS_KEY_TEMPLATE = 'SPLITIO.events'
     _EVENTS_KEY_DEFAULT_TTL = 3600
 
-    def __init__(self, redis_client, sdk_metadata):
-        """
-        Class constructor.
-
-        :param redis_client: Redis client or compliant interface.
-        :type redis_client: splitio.storage.adapters.redis.RedisAdapter
-        :param sdk_metadata: SDK & Machine information.
-        :type sdk_metadata: splitio.client.util.SdkMetadata
-        """
-        self._redis = redis_client
-        self._sdk_metadata = sdk_metadata
-
     def add_events_to_pipe(self, events, pipe):
         """
         Add put operation to pipeline
@@ -738,6 +726,59 @@ def _wrap_events(self, events):
         for e in events
     ]
 
+    def put(self, events):
+        """
+        Add an event to the redis storage.
+
+        :param event: Event to add to the queue.
+        :type event: splitio.models.events.Event
+
+        :return: Whether the event has been added or not.
+        :rtype: bool
+        """
+        pass
+
+    def pop_many(self, count):
+        """
+        Pop the oldest N events from storage.
+
+        :param count: Number of events to pop.
+        :type count: int
+        """
+        raise NotImplementedError('Only redis-consumer mode is supported.')
+
+    def clear(self):
+        """
+        Clear data.
+        """
+        raise NotImplementedError('Not supported for redis.')
+
+    def expire_keys(self, total_keys, inserted):
+        """
+        Set expire
+
+        :param total_keys: length of keys.
+        :type total_keys: int
+        :param inserted: added keys.
+        :type inserted: int
+        """
+        pass
+
+class RedisEventsStorage(RedisEventsStorageBase):
+    """Redis based event storage class."""
+
+    def __init__(self, redis_client, sdk_metadata):
+        """
+        Class constructor.
+
+        :param redis_client: Redis client or compliant interface.
+        :type redis_client: splitio.storage.adapters.redis.RedisAdapter
+        :param sdk_metadata: SDK & Machine information.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
+        """
+        self._redis = redis_client
+        self._sdk_metadata = sdk_metadata
+
     def put(self, events):
         """
         Add an event to the redis storage.
@@ -760,22 +801,57 @@ def put(self, events):
             _LOGGER.debug('Error: ', exc_info=True)
             return False
 
-    def pop_many(self, count):
+    def expire_keys(self, total_keys, inserted):
         """
-        Pop the oldest N events from storage.
+        Set expire
 
-        :param count: Number of events to pop.
-        :type count: int
+        :param total_keys: length of keys.
+        :type total_keys: int
+        :param inserted: added keys.
+        :type inserted: int
         """
-        raise NotImplementedError('Only redis-consumer mode is supported.')
+        if total_keys == inserted:
+            self._redis.expire(self._EVENTS_KEY_TEMPLATE, self._EVENTS_KEY_DEFAULT_TTL)
 
-    def clear(self):
+
+class RedisEventsStorageAsync(RedisEventsStorageBase):
+    """Redis based event async storage class."""
+
+    def __init__(self, redis_client, sdk_metadata):
         """
-        Clear data.
+        Class constructor.
+
+        :param redis_client: Redis client or compliant interface.
+        :type redis_client: splitio.storage.adapters.redis.RedisAdapter
+        :param sdk_metadata: SDK & Machine information.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
         """
-        raise NotImplementedError('Not supported for redis.')
+        self._redis = redis_client
+        self._sdk_metadata = sdk_metadata
 
-    def expire_keys(self, total_keys, inserted):
+    async def put(self, events):
+        """
+        Add an event to the redis storage.
+
+        :param event: Event to add to the queue.
+        :type event: splitio.models.events.Event
+
+        :return: Whether the event has been added or not.
+        :rtype: bool
+        """
+        key = self._EVENTS_KEY_TEMPLATE
+        to_store = self._wrap_events(events)
+        try:
+            _LOGGER.debug("Adding Events to redis key %s" % (key))
+            _LOGGER.debug(to_store)
+            await self._redis.rpush(key, *to_store)
+            return True
+        except RedisAdapterException:
+            _LOGGER.error('Something went wrong when trying to add event to redis')
+            _LOGGER.debug('Error: ', exc_info=True)
+            return False
+
+    async def expire_keys(self, total_keys, inserted):
         """
         Set expire
 
@@ -785,7 +861,8 @@ def expire_keys(self, total_keys, inserted):
         :type inserted: int
         """
         if total_keys == inserted:
-            self._redis.expire(self._EVENTS_KEY_TEMPLATE, self._EVENTS_KEY_DEFAULT_TTL)
+            await self._redis.expire(self._EVENTS_KEY_TEMPLATE, self._EVENTS_KEY_DEFAULT_TTL)
+
 
 class RedisTelemetryStorageBase(TelemetryStorage):
     """Redis based telemetry storage class."""

tests/storage/test_redis.py

@@ -8,8 +8,9 @@
 import pytest
 
 from splitio.client.util import get_metadata, SdkMetadata
+from splitio.storage.adapters.redis import RedisAdapter, RedisAdapterAsync, RedisAdapterException, build
 from splitio.optional.loaders import asyncio
-from splitio.storage.redis import RedisEventsStorage, RedisImpressionsStorage, RedisImpressionsStorageAsync, \
+from splitio.storage.redis import RedisEventsStorage, RedisEventsStorageAsync, RedisImpressionsStorage, RedisImpressionsStorageAsync, \
     RedisSegmentStorage, RedisSegmentStorageAsync, RedisSplitStorage, RedisSplitStorageAsync, RedisTelemetryStorage, RedisTelemetryStorageAsync
 from splitio.storage.adapters.redis import RedisAdapter, RedisAdapterException, build
 from redis.asyncio.client import Redis as aioredis
@@ -627,6 +628,103 @@ def _raise_exc(*_):
         adapter.rpush.side_effect = _raise_exc
         assert storage.put(events) is False
 
+    def test_expire_keys(self, mocker):
+        adapter = mocker.Mock(spec=RedisAdapter)
+        metadata = get_metadata({})
+        storage = RedisEventsStorage(adapter, metadata)
+
+        self.key = None
+        self.ttl = None
+        def expire(key, ttl):
+            self.key = key
+            self.ttl = ttl
+        adapter.expire = expire
+
+        storage.expire_keys(2, 2)
+        assert self.key == 'SPLITIO.events'
+        assert self.ttl == 3600
+
+        self.key = None
+        storage.expire_keys(2, 1)
+        assert self.key == None
+
+class RedisEventsStorageAsyncTests(object):  # pylint: disable=too-few-public-methods
+    """Redis Impression async storage test cases."""
+
+    @pytest.mark.asyncio
+    async def test_add_events(self, mocker):
+        """Test that adding impressions to storage works."""
+        adapter = mocker.Mock(spec=RedisAdapterAsync)
+        metadata = get_metadata({})
+
+        storage = RedisEventsStorageAsync(adapter, metadata)
+
+        events = [
+            EventWrapper(event=Event('key1', 'user', 'purchase', 10, 123456, None),  size=32768),
+            EventWrapper(event=Event('key2', 'user', 'purchase', 10, 123456, None),  size=32768),
+            EventWrapper(event=Event('key3', 'user', 'purchase', 10, 123456, None),  size=32768),
+            EventWrapper(event=Event('key4', 'user', 'purchase', 10, 123456, None),  size=32768),
+        ]
+        self.key = None
+        self.events = None
+        async def rpush(key, *events):
+            self.key = key
+            self.events = events
+        adapter.rpush = rpush
+
+        assert await storage.put(events) is True
+
+        list_of_raw_events = [json.dumps({
+            'e': {  # EVENT PORTION
+                'key': e.event.key,
+                'trafficTypeName': e.event.traffic_type_name,
+                'eventTypeId': e.event.event_type_id,
+                'value': e.event.value,
+                'timestamp': e.event.timestamp,
+                'properties': e.event.properties,
+            },
+            'm': {  # METADATA PORTION
+                's': metadata.sdk_version,
+                'n': metadata.instance_name,
+                'i': metadata.instance_ip,
+            }
+        }) for e in events]
+
+        assert self.events == tuple(list_of_raw_events)
+        assert self.key == 'SPLITIO.events'
+        assert storage._wrap_events(events) == list_of_raw_events
+
+        # Assert that if an exception is thrown it's caught and False is returned
+        adapter.reset_mock()
+
+        async def rpush2(key, *events):
+            raise RedisAdapterException('something')
+        adapter.rpush = rpush2
+        assert await storage.put(events) is False
+
+
+    @pytest.mark.asyncio
+    async def test_expire_keys(self, mocker):
+        adapter = mocker.Mock(spec=RedisAdapterAsync)
+        metadata = get_metadata({})
+        storage = RedisEventsStorageAsync(adapter, metadata)
+
+        self.key = None
+        self.ttl = None
+        async def expire(key, ttl):
+            self.key = key
+            self.ttl = ttl
+        adapter.expire = expire
+
+        await storage.expire_keys(2, 2)
+        assert self.key == 'SPLITIO.events'
+        assert self.ttl == 3600
+
+        self.key = None
+        await storage.expire_keys(2, 1)
+        assert self.key == None
+
+
 class RedisTelemetryStorageTests(object):
     """Redis Telemetry storage test cases."""