Merge branch 'development' of github.com:splitio/python-client into development

Author: mmelograno

setup.py

@@ -19,6 +19,7 @@
     'pyyaml>=5.4',
     'docopt>=0.6.2',
     'enum34;python_version<"3.4"',
+    'bloom-filter2>=2.0.0',
 ]
 
 with open(path.join(path.abspath(path.dirname(__file__)), 'splitio', 'version.py')) as f:

splitio/api/client.py

@@ -25,8 +25,9 @@ class HttpClient(object):
     SDK_URL = 'https://sdk.split.io/api'
     EVENTS_URL = 'https://events.split.io/api'
     AUTH_URL = 'https://auth.split.io/api'
+    TELEMETRY_URL = 'https://telemetry.split.io/api'
 
-    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None):
+    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, telemetry_url=None):
         """
         Class constructor.
 
@@ -38,12 +39,15 @@ def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None):
         :type events_url: str
         :param auth_url: Optional alternative auth URL.
         :type auth_url: str
+        :param telemetry_url: Optional alternative telemetry URL.
+        :type telemetry_url: str
         """
         self._timeout = timeout/1000 if timeout else None  # Convert ms to seconds.
         self._urls = {
             'sdk': sdk_url if sdk_url is not None else self.SDK_URL,
             'events': events_url if events_url is not None else self.EVENTS_URL,
             'auth': auth_url if auth_url is not None else self.AUTH_URL,
+            'telemetry': telemetry_url if telemetry_url is not None else self.TELEMETRY_URL,
         }
 
     def _build_url(self, server, path):

splitio/api/impressions.py

@@ -6,7 +6,7 @@
 from splitio.api import APIException
 from splitio.api.client import HttpClientException
 from splitio.api.commons import headers_from_metadata
-from splitio.engine.impressions import ImpressionsMode
+from splitio.engine.impressions.impressions import ImpressionsMode
 
 
 _LOGGER = logging.getLogger(__name__)

splitio/api/telemetry.py

@@ -0,0 +1,48 @@
+"""Impressions API module."""
+import logging
+
+from splitio.api import APIException
+from splitio.api.client import HttpClientException
+from splitio.api.commons import headers_from_metadata
+
+_LOGGER = logging.getLogger(__name__)
+
+class TelemetryAPI(object):  # pylint: disable=too-few-public-methods
+    """Class that uses an httpClient to communicate with the Telemetry API."""
+
+    def __init__(self, client, apikey, sdk_metadata):
+        """
+        Class constructor.
+
+        :param client: HTTP Client responsble for issuing calls to the backend.
+        :type client: HttpClient
+        :param apikey: User apikey token.
+        :type apikey: string
+        """
+        self._client = client
+        self._apikey = apikey
+        self._metadata = headers_from_metadata(sdk_metadata)
+
+    def record_unique_keys(self, uniques):
+        """
+        Send unique keys to the backend.
+
+        :param uniques: Unique Keys
+        :type json
+        """
+        try:
+            response = self._client.post(
+                'telemetry',
+                '/keys/ss',
+                self._apikey,
+                body=uniques,
+                extra_headers=self._metadata
+            )
+            if not 200 <= response.status_code < 300:
+                raise APIException(response.body, response.status_code)
+        except HttpClientException as exc:
+            _LOGGER.error(
+                'Error posting unique keys because an exception was raised by the HTTPClient'
+            )
+            _LOGGER.debug('Error: ', exc_info=True)
+            raise APIException('Unique keys not flushed properly.') from exc

splitio/client/config.py

@@ -2,7 +2,7 @@
 import os.path
 import logging
 
-from splitio.engine.impressions import ImpressionsMode
+from splitio.engine.impressions.impressions import ImpressionsMode
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -91,7 +91,7 @@ def _sanitize_impressions_mode(mode, refresh_rate=None):
             mode = ImpressionsMode(mode.upper())
         except (ValueError, AttributeError):
             _LOGGER.warning('You passed an invalid impressionsMode, impressionsMode should be '
-                            'one of the following values: `debug` or `optimized`. '
+                            'one of the following values: `debug`, `none` or `optimized`. '
                             'Defaulting to `optimized` mode.')
             mode = ImpressionsMode.OPTIMIZED
 

splitio/client/factory.py

@@ -11,7 +11,12 @@
 from splitio.client.config import sanitize as sanitize_config, DEFAULT_DATA_SAMPLING
 from splitio.client import util
 from splitio.client.listener import ImpressionListenerWrapper
-from splitio.engine.impressions import Manager as ImpressionsManager
+from splitio.engine.impressions.impressions import Manager as ImpressionsManager
+from splitio.engine.impressions.impressions import ImpressionsMode
+from splitio.engine.impressions.manager import Counter as ImpressionsCounter
+from splitio.engine.impressions.strategies import StrategyNoneMode, StrategyDebugMode, StrategyOptimizedMode
+from splitio.engine.impressions.adapters import InMemorySenderAdapter, RedisSenderAdapter
+from splitio.engine.impressions import set_classes
 
 # Storage
 from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
@@ -27,21 +32,25 @@
 from splitio.api.impressions import ImpressionsAPI
 from splitio.api.events import EventsAPI
 from splitio.api.auth import AuthAPI
+from splitio.api.telemetry import TelemetryAPI
 
 # Tasks
 from splitio.tasks.split_sync import SplitSynchronizationTask
 from splitio.tasks.segment_sync import SegmentSynchronizationTask
 from splitio.tasks.impressions_sync import ImpressionsSyncTask, ImpressionsCountSyncTask
 from splitio.tasks.events_sync import EventsSyncTask
+from splitio.tasks.unique_keys_sync import UniqueKeysSyncTask, ClearFilterSyncTask
 
 # Synchronizer
 from splitio.sync.synchronizer import SplitTasks, SplitSynchronizers, Synchronizer, \
-    LocalhostSynchronizer
-from splitio.sync.manager import Manager
+    LocalhostSynchronizer, RedisSynchronizer
+from splitio.sync.manager import Manager, RedisManager
 from splitio.sync.split import SplitSynchronizer, LocalSplitSynchronizer
 from splitio.sync.segment import SegmentSynchronizer
 from splitio.sync.impression import ImpressionSynchronizer, ImpressionsCountSynchronizer
 from splitio.sync.event import EventSynchronizer
+from splitio.sync.unique_keys import UniqueKeysSynchronizer, ClearFilterSynchronizer
+
 
 # Recorder
 from splitio.recorder.recorder import StandardRecorder, PipelinedRecorder
@@ -207,6 +216,7 @@ def destroy(self, destroyed_event=None):
             return
 
         try:
+            _LOGGER.info('Factory destroy called, stopping tasks.')
             if self._sync_manager is not None:
                 if destroyed_event is not None:
 
@@ -283,7 +293,7 @@ def _wrap_impression_listener(listener, metadata):
 
 
 def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pylint:disable=too-many-arguments,too-many-locals
-                             auth_api_base_url=None, streaming_api_base_url=None):
+                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None):
     """Build and return a split factory tailored to the supplied config."""
     if not input_validator.validate_factory_instantiation(api_key):
         return None
@@ -292,6 +302,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         sdk_url=sdk_url,
         events_url=events_url,
         auth_url=auth_api_base_url,
+        telemetry_url=telemetry_api_base_url,
         timeout=cfg.get('connectionTimeout')
     )
 
@@ -302,6 +313,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'segments': SegmentsAPI(http_client, api_key, sdk_metadata),
         'impressions': ImpressionsAPI(http_client, api_key, sdk_metadata, cfg['impressionsMode']),
         'events': EventsAPI(http_client, api_key, sdk_metadata),
+        'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata),
     }
 
     if not input_validator.validate_apikey_type(apis['segments']):
@@ -314,18 +326,23 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'events': InMemoryEventStorage(cfg['eventsQueueSize']),
     }
 
+    unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
+    clear_filter_task, impressions_count_sync, impressions_count_task, \
+    imp_strategy = set_classes('MEMORY', cfg['impressionsMode'], apis)
+
     imp_manager = ImpressionsManager(
-        cfg['impressionsMode'],
-        True,
-        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata))
+        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
+        imp_strategy)
 
     synchronizers = SplitSynchronizers(
         SplitSynchronizer(apis['splits'], storages['splits']),
         SegmentSynchronizer(apis['segments'], storages['splits'], storages['segments']),
         ImpressionSynchronizer(apis['impressions'], storages['impressions'],
                                cfg['impressionsBulkSize']),
         EventSynchronizer(apis['events'], storages['events'], cfg['eventsBulkSize']),
-        ImpressionsCountSynchronizer(apis['impressions'], imp_manager),
+        impressions_count_sync,
+        unique_keys_synchronizer,
+        clear_filter_sync
     )
 
     tasks = SplitTasks(
@@ -342,7 +359,9 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
             cfg['impressionsRefreshRate'],
         ),
         EventsSyncTask(synchronizers.events_sync.synchronize_events, cfg['eventsPushRate']),
-        ImpressionsCountSyncTask(synchronizers.impressions_count_sync.synchronize_counters)
+        impressions_count_task,
+        unique_keys_task,
+        clear_filter_task
     )
 
     synchronizer = Synchronizer(synchronizers, tasks)
@@ -393,19 +412,47 @@ def _build_redis_factory(api_key, cfg):
         _LOGGER.warning("dataSampling cannot be less than %.2f, defaulting to minimum",
                         _MIN_DEFAULT_DATA_SAMPLING_ALLOWED)
         data_sampling = _MIN_DEFAULT_DATA_SAMPLING_ALLOWED
+
+    unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
+    clear_filter_task, impressions_count_sync, impressions_count_task, \
+    imp_strategy = set_classes('REDIS', cfg['impressionsMode'], redis_adapter)
+
+    imp_manager = ImpressionsManager(
+        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
+        imp_strategy)
+
+    synchronizers = SplitSynchronizers(None, None, None, None,
+        impressions_count_sync,
+        unique_keys_synchronizer,
+        clear_filter_sync
+    )
+
+    tasks = SplitTasks(None, None, None, None,
+        impressions_count_task,
+        unique_keys_task,
+        clear_filter_task
+    )
+
+    synchronizer = RedisSynchronizer(synchronizers, tasks)
     recorder = PipelinedRecorder(
         redis_adapter.pipeline,
-        ImpressionsManager(cfg['impressionsMode'], False,
-                           _wrap_impression_listener(cfg['impressionListener'], sdk_metadata)),
+        imp_manager,
         storages['events'],
         storages['impressions'],
         data_sampling,
     )
+
+    manager = RedisManager(synchronizer)
+    initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer")
+    initialization_thread.setDaemon(True)
+    initialization_thread.start()
+
     return SplitFactory(
         api_key,
         storages,
         cfg['labelsEnabled'],
         recorder,
+        manager,
     )
 
 
@@ -436,7 +483,7 @@ def _build_localhost_factory(cfg):
     manager = Manager(ready_event, synchronizer, None, False, sdk_metadata)
     manager.start()
     recorder = StandardRecorder(
-        ImpressionsManager(cfg['impressionsMode'], True, None),
+        ImpressionsManager(None, StrategyDebugMode()),
         storages['events'],
         storages['impressions'],
     )
@@ -485,7 +532,8 @@ def get_factory(api_key, **kwargs):
             kwargs.get('sdk_api_base_url'),
             kwargs.get('events_api_base_url'),
             kwargs.get('auth_api_base_url'),
-            kwargs.get('streaming_api_base_url')
+            kwargs.get('streaming_api_base_url'),
+            kwargs.get('telemetry_api_base_url')
         )
     finally:
         _INSTANTIATED_FACTORIES.update([api_key])

splitio/engine/filters.py

@@ -0,0 +1,85 @@
+import abc
+import threading
+
+from bloom_filter2 import BloomFilter as BloomFilter2
+
+class BaseFilter(object, metaclass=abc.ABCMeta):
+    """Impressions Filter interface."""
+
+    @abc.abstractmethod
+    def add(self, data):
+        """
+        Return a boolean flag
+
+        """
+        pass
+
+    @abc.abstractmethod
+    def contains(self, data):
+        """
+        Return a boolean flag
+
+        """
+        pass
+
+    @abc.abstractmethod
+    def clear(self):
+        """
+        No return
+
+        """
+        pass
+
+class BloomFilter(BaseFilter):
+    """Optimized mode strategy."""
+
+    def __init__(self, max_elements=5000, error_rate=0.01):
+        """
+        Construct a bloom filter instance.
+
+        :param max_element: maximum elements in the filter
+        :type string:
+
+        :param error_rate: error rate for the false positives, reduce it will consume more memory
+        :type numeric:
+        """
+        self._max_elements = max_elements
+        self._error_rate = error_rate
+        self._imps_bloom_filter = BloomFilter2(max_elements=self._max_elements, error_rate=self._error_rate)
+        self._lock = threading.RLock()
+
+    def add(self, data):
+        """
+        Add an item to the bloom filter instance.
+
+        :param data: element to be added
+        :type string:
+
+        :return: True if successful
+        :rtype: boolean
+        """
+        with self._lock:
+            self._imps_bloom_filter.add(data)
+            return data in self._imps_bloom_filter
+
+    def contains(self, data):
+        """
+        Check if an item exist in the bloom filter instance.
+
+        :param data: element to be checked
+        :type string:
+
+        :return: True if exist
+        :rtype: boolean
+        """
+        with self._lock:
+            return data in self._imps_bloom_filter
+
+    def clear(self):
+        """
+        Destroy the current filter instance and create new one.
+
+        """
+        with self._lock:
+            self._imps_bloom_filter.close()
+            self._imps_bloom_filter = BloomFilter2(max_elements=self._max_elements, error_rate=self._error_rate)