Merge pull request #181 from splitio/impression_observer

add impression observer

Author: mredolatti

splitio/engine/cache/__init__.py

@@ -0,0 +1,121 @@
+"""Simple test-and-set LRU Cache."""
+import threading
+
+
+DEFAULT_MAX_SIZE = 5000
+
+
+class SimpleLruCache(object):  #pylint: disable=too-many-instance-attributes
+    """
+    Key/Value local memory cache. with expiration & LRU eviction.
+
+    LRU double-linked-list format:
+
+    {
+        'key1'---------------------------------------------------------------
+        'key2'------------------------------------                           |
+        'key3'------------                        |                          |
+    }                     |                       |                          |
+                          V                       V                          V
+                     || MRU  ||  -previous-> ||   X  || ... -previous-> || LRU  || -previous-> None
+    None <---next--- || node ||  <---next--- || node || ... <---next--- || node ||
+    """
+
+    class _Node(object):  #pylint: disable=too-few-public-methods
+        """Links to previous an next items in the circular list."""
+
+        def __init__(self, key, value, previous_element, next_element):
+            """Class constructor."""
+            self.key = key  # we also keep the key for O(1) access when removing the LRU.
+            self.value = value
+            self.previous = previous_element
+            self.next = next_element
+
+        def __str__(self):
+            """Return string representation."""
+            return '(%s, %s)' % (self.key, self.value)
+
+    def __init__(self, max_size=DEFAULT_MAX_SIZE):
+        """Class constructor."""
+        self._data = {}
+        self._lock = threading.Lock()
+        self._max_size = max_size
+        self._lru = None
+        self._mru = None
+
+    def test_and_set(self, key, value):
+        """
+        Set an item in the cache and return the previous value.
+
+        :param key: object key
+        :type args: object
+        :param value: object value
+        :type kwargs: object
+
+        :return: previous value if any. None otherwise
+        :rtype: object
+        """
+        with self._lock:
+            node = self._data.get(key)
+            to_return = node.value if node else None
+            if node is None:
+                node = SimpleLruCache._Node(key, value, None, None)
+            node = self._bubble_up(node)
+            self._data[key] = node
+            self._rollover()
+            return to_return
+
+    def clear(self):
+        """Clear the cache."""
+        self._data = {}
+        self._lru = None
+        self._mru = None
+
+    def _bubble_up(self, node):
+        """Send node to the top of the list (mark it as the MRU)."""
+        if node is None:
+            return None
+
+        # First item, just set lru & mru
+        if not self._data:
+            self._lru = node
+            self._mru = node
+            return node
+
+        # MRU, just return it
+        if node is self._mru:
+            return node
+
+        # LRU, update pointer and end-of-list
+        if node is self._lru:
+            self._lru = node.next
+            self._lru.previous = None
+
+        if node.previous is not None:
+            node.previous.next = node.next
+        if node.next is not None:
+            node.next.previous = node.previous
+
+        node.previous = self._mru
+        node.previous.next = node
+        node.next = None
+        self._mru = node
+
+        return node
+
+    def _rollover(self):
+        """Check we're within the size limit. Otherwise drop the LRU."""
+        if len(self._data) > self._max_size:
+            next_item = self._lru.next
+            del self._data[self._lru.key]
+            self._lru = next_item
+            self._lru.previous = None
+
+    def __str__(self):
+        """User friendly representation of cache."""
+        nodes = []
+        node = self._mru
+        while node is not None:
+            nodes.append('\t<%s: %s>  -->' % (node.key, node.value))
+            node = node.previous
+        return '<MRU>\n' + '\n'.join(nodes) + '\n<LRU>'

splitio/engine/cache/lru.py

@@ -0,0 +1,89 @@
+"""Split evaluator module."""
+import logging
+
+from splitio.models.impressions import Impression
+from splitio.engine.hashfns import murmur_128
+from splitio.engine.cache.lru import SimpleLruCache
+
+
+class Hasher(object):
+    """Impression hasher."""
+
+    _PATTERN = "%s:%s:%s:%s:%d"
+
+    def __init__(self, hash_fn=murmur_128, seed=0):
+        """
+        Class constructor.
+
+        :param hash_fn: Hash function to apply (str, int) -> int
+        :type hash_fn: callable
+
+        :param seed: seed to be provided when hashing
+        :type seed: int
+        """
+        self._hash_fn = hash_fn
+        self._seed = seed
+
+    def _stringify(self, impression):
+        """
+        Stringify an impression.
+
+        :param impression: Impression to stringify using _PATTERN
+        :type impression: splitio.models.impressions.Impression
+
+        :returns: a string representation of the impression
+        :rtype: str
+        """
+        return self._PATTERN % (impression.matching_key,
+                                impression.feature_name,
+                                impression.treatment,
+                                impression.label,
+                                impression.change_number)
+
+    def process(self, impression):
+        """
+        Hash an impression.
+
+        :param impression: Impression to hash.
+        :type impression: splitio.models.impressions.Impression
+
+        :returns: a hash of the supplied impression's relevant fields.
+        :rtype: int
+        """
+        return self._hash_fn(self._stringify(impression), self._seed)
+
+
+class Observer(object):
+    """Observe impression and add a previous time if applicable."""
+
+    def __init__(self, size):
+        """Class constructor."""
+        self._hasher = Hasher()
+        self._cache = SimpleLruCache(size)
+
+    def test_and_set(self, impression):
+        """
+        Examine an impression to determine and set it's previous time accordingly.
+
+        :param impression: Impression to track
+        :type impression: splitio.models.impressions.Impression
+
+        :returns: Impression with populated previous time
+        :rtype: splitio.models.impressions.Impression
+        """
+        previous_time = self._cache.test_and_set(self._hasher.process(impression), impression.time)
+        return Impression(impression.matching_key,
+                          impression.feature_name,
+                          impression.treatment,
+                          impression.label,
+                          impression.change_number,
+                          impression.bucketing_key,
+                          impression.time,
+                          previous_time)
+
+
+class Manager(object):
+    """Impression manager."""
+
+    #TODO: implement
+    pass

splitio/engine/impmanager.py

@@ -14,10 +14,14 @@
         'label',
         'change_number',
         'bucketing_key',
-        'time'
+        'time',
+        'previous_time'
     ]
 )
 
+# pre-python3.7 hack to make previous_time optional
+Impression.__new__.__defaults__ = (None,)
+
 
 class Label(object):  # pylint: disable=too-few-public-methods
     """Impressions labels."""

splitio/models/impressions.py

@@ -0,0 +1,38 @@
+"""LRU Cache unit tests."""
+
+from splitio.engine.cache.lru import SimpleLruCache
+
+class SimpleLruCacheTests(object):
+    """Test SimpleLruCache."""
+
+    def test_basic_usage(self, mocker):
+        """Test that a missing split logs and returns CONTROL."""
+        cache = SimpleLruCache(5)
+        assert cache.test_and_set('a', 1) is None
+        assert cache.test_and_set('b', 2) is None
+        assert cache.test_and_set('c', 3) is None
+        assert cache.test_and_set('d', 4) is None
+        assert cache.test_and_set('e', 5) is None
+
+        assert cache.test_and_set('a', 10) is 1
+        assert cache.test_and_set('b', 20) is 2
+        assert cache.test_and_set('c', 30) is 3
+        assert cache.test_and_set('d', 40) is 4
+        assert cache.test_and_set('e', 50) is 5
+        assert len(cache._data) is 5
+
+    def test_lru_eviction(self, mocker):
+        """Test that a missing split logs and returns CONTROL."""
+        cache = SimpleLruCache(5)
+        assert cache.test_and_set('a', 1) is None
+        assert cache.test_and_set('b', 2) is None
+        assert cache.test_and_set('c', 3) is None
+        assert cache.test_and_set('d', 4) is None
+        assert cache.test_and_set('e', 5) is None
+        assert cache.test_and_set('f', 6) is None
+        assert cache.test_and_set('g', 7) is None
+        assert cache.test_and_set('h', 8) is None
+        assert cache.test_and_set('i', 9) is None
+        assert cache.test_and_set('j', 0) is None
+        assert len(cache._data) is 5
+        assert set(cache._data.keys()) == set(['f', 'g', 'h', 'i', 'j'])

tests/engine/cache/test_lru.py

@@ -0,0 +1,52 @@
+"""Impression manager, observer & hasher tests."""
+
+from splitio.engine.impmanager import Hasher, Observer, Manager
+from splitio.models.impressions import Impression
+
+
+class ImpressionHasherTests(object):
+    """Test ImpressionHasher behavior."""
+
+    def test_changes_are_reflected(self):
+        """Test that change in any field changes the resulting hash."""
+        total = set()
+        hasher = Hasher()
+        total.add(hasher.process(Impression('key1', 'feature1', 'on', 'killed', 123, None, 456)))
+        total.add(hasher.process(Impression('key2', 'feature1', 'on', 'killed', 123, None, 456)))
+        total.add(hasher.process(Impression('key1', 'feature2', 'on', 'killed', 123, None, 456)))
+        total.add(hasher.process(Impression('key1', 'feature1', 'off', 'killed', 123, None, 456)))
+        total.add(hasher.process(Impression('key1', 'feature1', 'on', 'not killed', 123, None, 456)))
+        total.add(hasher.process(Impression('key1', 'feature1', 'on', 'killed', 321, None, 456)))
+        assert len(total) == 6
+
+        # Re-adding the first-one should not increase the number of different hashes
+        total.add(hasher.process(Impression('key1', 'feature1', 'on', 'killed', 123, None, 456)))
+        assert len(total) == 6
+
+
+class ImpressionObserverTests(object):
+    """Test impression observer behaviour."""
+
+    def test_previous_time_properly_calculated(self):
+        """Test that the previous time is properly set."""
+        observer = Observer(5)
+        assert (observer.test_and_set(Impression('key1', 'f1', 'on', 'killed', 123, None, 456))
+                == Impression('key1', 'f1', 'on', 'killed', 123, None, 456))
+        assert (observer.test_and_set(Impression('key1', 'f1', 'on', 'killed', 123, None, 457))
+                == Impression('key1', 'f1', 'on', 'killed', 123, None, 457, 456))
+
+        # Add 5 new impressions to evict the first one and check that previous time is None again
+        assert (observer.test_and_set(Impression('key2', 'f1', 'on', 'killed', 123, None, 456))
+                == Impression('key2', 'f1', 'on', 'killed', 123, None, 456))
+        assert (observer.test_and_set(Impression('key3', 'f1', 'on', 'killed', 123, None, 456))
+                == Impression('key3', 'f1', 'on', 'killed', 123, None, 456))
+        assert (observer.test_and_set(Impression('key4', 'f1', 'on', 'killed', 123, None, 456))
+                == Impression('key4', 'f1', 'on', 'killed', 123, None, 456))
+        assert (observer.test_and_set(Impression('key5', 'f1', 'on', 'killed', 123, None, 456))
+                == Impression('key5', 'f1', 'on', 'killed', 123, None, 456))
+        assert (observer.test_and_set(Impression('key6', 'f1', 'on', 'killed', 123, None, 456))
+                == Impression('key6', 'f1', 'on', 'killed', 123, None, 456))
+
+        # Re-process the first-one
+        assert (observer.test_and_set(Impression('key1', 'f1', 'on', 'killed', 123, None, 456))
+                == Impression('key1', 'f1', 'on', 'killed', 123, None, 456))