added support for properties in track method

Author: Matias Melograno

splitio/api/events.py

@@ -7,7 +7,7 @@
 from splitio.api.client import HttpClientException
 
 
-class EventsAPI(object):  #pylint: disable=too-few-public-methods
+class EventsAPI(object):  # pylint: disable=too-few-public-methods
     """Class that uses an httpClient to communicate with the events API."""
 
     def __init__(self, http_client, apikey, sdk_metadata):
@@ -43,7 +43,14 @@ def _build_bulk(events):
                 'trafficTypeName': event.traffic_type_name,
                 'eventTypeId': event.event_type_id,
                 'value': event.value,
-                'timestamp': event.timestamp
+                'timestamp': event.timestamp,
+                'properties': event.properties,
+            } if event.properties is not None else {
+                'key': event.key,
+                'trafficTypeName': event.traffic_type_name,
+                'eventTypeId': event.event_type_id,
+                'value': event.value,
+                'timestamp': event.timestamp,
             }
             for event in events
         ]

splitio/api/impressions.py

@@ -9,7 +9,7 @@
 from splitio.api.client import HttpClientException
 
 
-class ImpressionsAPI(object): # pylint: disable=too-few-public-methods
+class ImpressionsAPI(object):  # pylint: disable=too-few-public-methods
     """Class that uses an httpClient to communicate with the impressions API."""
 
     def __init__(self, client, apikey, sdk_metadata):

splitio/client/client.py

@@ -8,7 +8,7 @@
 from splitio.engine.evaluator import Evaluator, CONTROL
 from splitio.engine.splitters import Splitter
 from splitio.models.impressions import Impression, Label
-from splitio.models.events import Event
+from splitio.models.events import Event, EventWrapper
 from splitio.models.telemetry import get_latency_bucket_index
 from splitio.client import input_validator
 from splitio.client.listener import ImpressionListenerException
@@ -340,7 +340,7 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
         event_type = input_validator.validate_event_type(event_type)
         traffic_type = input_validator.validate_traffic_type(traffic_type)
         value = input_validator.validate_value(value)
-        valid, properties = input_validator.valid_properties(properties)
+        valid, properties, size = input_validator.valid_properties(properties)
 
         if key is None or event_type is None or traffic_type is None or value is False \
            or valid is False:
@@ -354,4 +354,7 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
             timestamp=int(time.time()*1000),
             properties=properties,
         )
-        return self._events_storage.put([event])
+        return self._events_storage.put([EventWrapper(
+            event=event,
+            size=size,
+        )])

splitio/client/factory.py

@@ -304,6 +304,9 @@ def _build_in_memory_factory(api_key, config, sdk_url=None, events_url=None):  #
     tasks['events'].start()
     tasks['telemetry'].start()
 
+    storages['events'].set_queue_full_hook(tasks['events'].flush())
+    storages['impressions'].set_queue_full_hook(tasks['impressions'].flush())
+
     def split_ready_task():
         """Wait for splits to be ready and start fetching segments."""
         splits_ready_flag.wait()

splitio/client/input_validator.py

@@ -8,7 +8,6 @@
 import math
 
 import six
-import sys
 
 from splitio.api import APIException
 from splitio.client.key import Key
@@ -476,15 +475,16 @@ def valid_properties(properties):
     :param properties: dict
     :type properties: dict
     :return: tuple
-    :rtype: (bool,dict)
+    :rtype: (bool,dict,int)
     """
+    size = 1024  # We assume 1kb events without properties (750 bytes avg measured)
+
     if properties is None:
-        return True, None
+        return True, None, size
     if not isinstance(properties, dict):
         _LOGGER.error('track: properties must be of type dictionary.')
-        return False, None
+        return False, None, 0
 
-    size = 1024  # We assume 1kb events without properties (750 bytes avg measured)
     valid_properties = None
 
     for property, element in properties.items():
@@ -495,7 +495,7 @@ def valid_properties(properties):
             valid_properties = dict()
 
         valid_properties[property] = None
-        size += sys.getsizeof(property)
+        size += len(property)
 
         if element is None:
             continue
@@ -506,17 +506,18 @@ def valid_properties(properties):
             element = None
 
         valid_properties[property] = element
-        size += sys.getsizeof(str(element))
+
+        if isinstance(element, six.string_types):
+            size += len(element)
 
         if size > MAX_PROPERTIES_LENGTH_BYTES:
             _LOGGER.error(
                 'The maximum size allowed for the properties is 32768 bytes. ' +
                 'Current one is ' + str(size) + ' bytes. Event not queued'
             )
-            return False, None
+            return False, None, size
 
     if isinstance(valid_properties, dict) and len(valid_properties.keys()) > 300:
         _LOGGER.warning('Event has more than 300 properties. Some of them will be trimmed' +
                         ' when processed')
-
-    return True, valid_properties
+    return True, valid_properties, size

splitio/models/events.py

@@ -16,3 +16,8 @@
     'timestamp',
     'properties',
 ])
+
+EventWrapper = namedtuple('EventWrapper', [
+    'event',
+    'size',
+])

splitio/storage/inmemmory.py

@@ -8,6 +8,8 @@
 from splitio.storage import SplitStorage, SegmentStorage, ImpressionStorage, EventStorage, \
     TelemetryStorage
 
+MAX_SIZE_BYTES = 5 * 1024 * 1024
+
 
 class InMemorySplitStorage(SplitStorage):
     """InMemory implementation of a split storage."""
@@ -149,7 +151,7 @@ def update(self, segment_name, to_add, to_remove, change_number=None):
         :type to_remove: Set
         """
         with self._lock:
-            if not segment_name in self._segments:
+            if segment_name not in self._segments:
                 self._segments[segment_name] = Segment(segment_name, to_add, change_number)
                 return
 
@@ -167,7 +169,7 @@ def get_change_number(self, segment_name):
         :rtype: int
         """
         with self._lock:
-            if not segment_name in self._segments:
+            if segment_name not in self._segments:
                 return None
             return self._segments[segment_name].change_number
 
@@ -181,7 +183,7 @@ def set_change_number(self, segment_name, new_change_number):
         :type new_change_number: int
         """
         with self._lock:
-            if not segment_name in self._segments:
+            if segment_name not in self._segments:
                 return
             self._segments[segment_name].change_number = new_change_number
 
@@ -198,7 +200,7 @@ def segment_contains(self, segment_name, key):
         :rtype: bool
         """
         with self._lock:
-            if not segment_name in self._segments:
+            if segment_name not in self._segments:
                 self._logger.warning(
                     "Tried to query members for nonexistant segment %s. Returning False",
                     segment_name
@@ -283,6 +285,7 @@ def __init__(self, eventsQueueSize):
         self._lock = threading.Lock()
         self._events = queue.Queue(maxsize=eventsQueueSize)
         self._queue_full_hook = None
+        self._size = 0
 
     def set_queue_full_hook(self, hook):
         """
@@ -295,21 +298,27 @@ def set_queue_full_hook(self, hook):
 
     def put(self, events):
         """
-        Add an avent to storage.
+        Add an event to storage.
 
         :param event: Event to be added in the storage
         """
         try:
             with self._lock:
                 for event in events:
-                    self._events.put(event, False)
+                    self._size += event.size
+
+                    if self._size >= MAX_SIZE_BYTES:
+                        self._queue_full_hook()
+                        return False
+
+                    self._events.put(event.event, False)
             return True
         except queue.Full:
             if self._queue_full_hook is not None and callable(self._queue_full_hook):
                 self._queue_full_hook()
             self._logger.warning(
                 'Events queue is full, failing to add more events. \n'
-                'Consider increasing parameter `impressionsQueueSize` in configuration'
+                'Consider increasing parameter `eventsQueueSize` in configuration'
             )
             return False
 
@@ -324,6 +333,7 @@ def pop_many(self, count):
             while not self._events.empty() and count > 0:
                 events.append(self._events.get(False))
                 count -= 1
+        self._size = 0
         return events
 
 

splitio/storage/redis.py

@@ -371,19 +371,20 @@ def put(self, events):
         to_store = [
             json.dumps({
                 'e': {
-                    'key': event.key,
-                    'trafficTypeName': event.traffic_type_name,
-                    'eventTypeId': event.event_type_id,
-                    'value': event.value,
-                    'timestamp': event.timestamp
+                    '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': {
                     's': self._sdk_metadata.sdk_version,
                     'n': self._sdk_metadata.instance_name,
                     'i': self._sdk_metadata.instance_ip,
                 }
             })
-            for event in events
+            for e in events
         ]
         try:
             self._redis.rpush(key, *to_store)

splitio/storage/uwsgi.py

@@ -9,7 +9,7 @@
 from splitio.models.events import Event
 from splitio.storage.adapters.uwsgi_cache import _SPLITIO_CHANGE_NUMBERS, \
     _SPLITIO_EVENTS_CACHE_NAMESPACE, _SPLITIO_IMPRESSIONS_CACHE_NAMESPACE, \
-     _SPLITIO_METRICS_CACHE_NAMESPACE, _SPLITIO_MISC_NAMESPACE, UWSGILock, \
+    _SPLITIO_METRICS_CACHE_NAMESPACE, _SPLITIO_MISC_NAMESPACE, UWSGILock, \
     _SPLITIO_SEGMENTS_CACHE_NAMESPACE, _SPLITIO_SPLITS_CACHE_NAMESPACE, \
     _SPLITIO_LOCK_CACHE_NAMESPACE
 
@@ -113,7 +113,7 @@ def remove(self, split_name):
             self._KEY_TEMPLATE.format(suffix=split_name),
             _SPLITIO_SPLITS_CACHE_NAMESPACE
         )
-        if not result is False:
+        if result is not False:
             self._logger.warning("Trying to retrieve nonexistant split %s. Ignoring.", split_name)
         return result
 
@@ -149,8 +149,8 @@ def get_split_names(self):
                 return json.loads(
                     self._uwsgi.cache_get(self._KEY_FEATURE_LIST, _SPLITIO_MISC_NAMESPACE)
                 )
-            except TypeError: # Thrown by json.loads when passing none
-                pass # Fall back to default return statement (empty list)
+            except TypeError:  # Thrown by json.loads when passing none
+                pass  # Fall back to default return statement (empty list)
         return []
 
     def get_all_splits(self):
@@ -251,7 +251,6 @@ def put(self, segment):
         )
         self.set_change_number(segment.name, segment.change_number)
 
-
     def get_change_number(self, segment_name):
         """
         Retrieve latest change number for a segment.
@@ -424,7 +423,7 @@ def put(self, events):
                 current = []
             self._uwsgi.cache_update(
                 self._EVENTS_KEY,
-                json.dumps(current + [e._asdict() for e in events]),
+                json.dumps(current + [e.event._asdict() for e in events]),
                 0,
                 _SPLITIO_EVENTS_CACHE_NAMESPACE
             )
@@ -457,8 +456,17 @@ def pop_many(self, count):
                 event['traffic_type_name'],
                 event['event_type_id'],
                 event['value'],
-                event['timestamp']
-            )   for event in current[:count]
+                event['timestamp'],
+                event['properties']
+            ) if 'properties' in event else
+            Event(
+                event['key'],
+                event['traffic_type_name'],
+                event['event_type_id'],
+                event['value'],
+                event['timestamp'],
+            )
+            for event in current[:count]
         ]
 
     def request_flush(self):
@@ -501,7 +509,6 @@ def __init__(self, uwsgi_entrypoint):
         self._uwsgi = uwsgi_entrypoint
         self._logger = logging.getLogger(self.__class__.__name__)
 
-
     def inc_latency(self, name, bucket):
         """
         Add a latency.
@@ -516,7 +523,8 @@ def inc_latency(self, name, bucket):
             return
 
         with UWSGILock(self._uwsgi, self._LATENCIES_LOCK_KEY):
-            latencies_raw = self._uwsgi.cache_get(self._LATENCIES_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
+            latencies_raw = self._uwsgi.cache_get(
+                self._LATENCIES_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
             latencies = json.loads(latencies_raw) if latencies_raw else {}
             to_update = latencies.get(name, [0] * 22)
             to_update[bucket] += 1
@@ -536,7 +544,8 @@ def inc_counter(self, name):
         :type name: str
         """
         with UWSGILock(self._uwsgi, self._COUNTERS_LOCK_KEY):
-            counters_raw = self._uwsgi.cache_get(self._COUNTERS_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
+            counters_raw = self._uwsgi.cache_get(
+                self._COUNTERS_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
             counters = json.loads(counters_raw) if counters_raw else {}
             value = counters.get(name, 0)
             value += 1
@@ -575,7 +584,8 @@ def pop_counters(self):
         :rtype: list
         """
         with UWSGILock(self._uwsgi, self._COUNTERS_LOCK_KEY):
-            counters_raw = self._uwsgi.cache_get(self._COUNTERS_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
+            counters_raw = self._uwsgi.cache_get(
+                self._COUNTERS_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
             self._uwsgi.cache_del(self._COUNTERS_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
             return json.loads(counters_raw) if counters_raw else {}
 
@@ -598,6 +608,7 @@ def pop_latencies(self):
         :rtype: list
         """
         with UWSGILock(self._uwsgi, self._LATENCIES_LOCK_KEY):
-            latencies_raw = self._uwsgi.cache_get(self._LATENCIES_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
+            latencies_raw = self._uwsgi.cache_get(
+                self._LATENCIES_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
             self._uwsgi.cache_del(self._LATENCIES_KEY, _SPLITIO_METRICS_CACHE_NAMESPACE)
             return json.loads(latencies_raw) if latencies_raw else {}