Merge pull request #277 from splitio/development

Version 9.2.0

Author: chillaq

CHANGES.txt

@@ -1,3 +1,6 @@
+9.2.0 (Oct 14, 2022)
+- Added a new impressions mode for the SDK called NONE , to be used in factory when there is no desire to capture impressions on an SDK factory to feed Split's analytics engine. Running NONE mode, the SDK will only capture unique keys evaluated for a particular feature flag instead of full blown impressions
+
 9.1.3 (July 25, 2022)
 - Fixed synching missed segment(s) after receiving split update
 

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',
+                '/v1/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)