Merge pull request #339 from splitio/development

Release 9.4.1

Author: chillaq

CHANGES.txt

@@ -1,3 +1,6 @@
+9.4.1 (Apr 18, 2023)
+- Fixed storing incorrect Telemetry method latency data
+
 9.4.0 (Mar 1, 2023)
 - Added support to use JSON files in localhost mode.
 - Updated default periodic telemetry post time to one hour.

setup.py

@@ -12,7 +12,8 @@
     'pytest-cov',
     'importlib-metadata==4.2',
     'tomli==1.2.3',
-    'iniconfig==1.1.1'
+    'iniconfig==1.1.1',
+    'attrs==22.1.0'
 ]
 
 INSTALL_REQUIRES = [

splitio/client/client.py

@@ -404,7 +404,7 @@ def track(self, key, traffic_type, event_type, value=None, properties=None):
             return_flag = self._recorder.record_track_stats([EventWrapper(
                 event=event,
                 size=size,
-            )], get_current_epoch_time_ms() - start)
+            )], get_latency_bucket_index(get_current_epoch_time_ms() - start))
             return return_flag
         except Exception:  # pylint: disable=broad-except
             self._telemetry_evaluation_producer.record_exception(MethodExceptionsAndLatencies.TRACK)

splitio/client/config.py

@@ -10,7 +10,7 @@
 
 
 DEFAULT_CONFIG = {
-    'operationMode': 'in-memory',
+    'operationMode': 'standalone',
     'connectionTimeout': 1500,
     'streamingEnabled': True,
     'featuresRefreshRate': 30,
@@ -56,29 +56,43 @@
     'localhostRefreshEnabled': False,
     'preforkedInitialization': False,
     'dataSampling': DEFAULT_DATA_SAMPLING,
+    'storageWrapper': None,
+    'storagePrefix': None,
+    'storageType': None
 }
 
 
 def _parse_operation_mode(apikey, config):
     """
-    Process incoming config to determine operation mode.
+    Process incoming config to determine operation mode and storage type
 
     :param config: user supplied config
     :type config: dict
 
-    :returns: operation mode
-    :rtype: str
+    :returns: operation mode and storage type
+    :rtype: Tuple (str, str)
     """
     if apikey == 'localhost':
-        return 'localhost-standalone'
+        _LOGGER.debug('Using Localhost operation mode')
+        return 'localhost', 'localhost'
 
     if 'redisHost' in config or 'redisSentinels' in config:
-        return 'redis-consumer'
+        _LOGGER.debug('Using Redis storage operation mode')
+        return 'consumer', 'redis'
 
-    return 'inmemory-standalone'
+    if  config.get('storageType') is not None:
+        if config.get('storageType').lower() == 'pluggable':
+            _LOGGER.debug('Using Pluggable storage operation mode')
+            return 'consumer', 'pluggable'
 
+        _LOGGER.warning('You passed an invalid storageType, acceptable value is '
+                            '`pluggable`. Defaulting storage to In-Memory mode.')
 
-def _sanitize_impressions_mode(mode, refresh_rate=None):
+    _LOGGER.debug('Using In-Memory operation mode')
+    return 'standalone', 'memory'
+
+
+def _sanitize_impressions_mode(storage_type, mode, refresh_rate=None):
     """
     Check supplied impressions mode and adjust refresh rate.
 
@@ -92,10 +106,10 @@ def _sanitize_impressions_mode(mode, refresh_rate=None):
         try:
             mode = ImpressionsMode(mode.upper())
         except (ValueError, AttributeError):
-            _LOGGER.warning('You passed an invalid impressionsMode, impressionsMode should be '
-                            'one of the following values: `debug`, `none` or `optimized`. '
-                            'Defaulting to `optimized` mode.')
             mode = ImpressionsMode.OPTIMIZED
+            _LOGGER.warning('You passed an invalid impressionsMode, impressionsMode should be ' \
+                            'one of the following values: `debug`, `none` or `optimized`. '
+                            ' Defaulting to `optimized` mode.')
 
     if mode == ImpressionsMode.DEBUG:
         refresh_rate = max(1, refresh_rate) if refresh_rate is not None else 60
@@ -118,10 +132,10 @@ def sanitize(apikey, config):
     :returns: sanitized config
     :rtype: dict
     """
-    config['operationMode'] = _parse_operation_mode(apikey, config)
+    config['operationMode'], config['storageType'] = _parse_operation_mode(apikey, config)
     processed = DEFAULT_CONFIG.copy()
     processed.update(config)
-    imp_mode, imp_rate = _sanitize_impressions_mode(config.get('impressionsMode'),
+    imp_mode, imp_rate = _sanitize_impressions_mode(config['storageType'], config.get('impressionsMode'),
                                                     config.get('impressionsRefreshRate'))
     processed['impressionsMode'] = imp_mode
     processed['impressionsRefreshRate'] = imp_rate

splitio/client/factory.py

@@ -24,6 +24,8 @@
 from splitio.storage.adapters import redis
 from splitio.storage.redis import RedisSplitStorage, RedisSegmentStorage, RedisImpressionsStorage, \
     RedisEventsStorage, RedisTelemetryStorage
+from splitio.storage.pluggable import PluggableEventsStorage, PluggableImpressionsStorage, PluggableSegmentStorage, \
+    PluggableSplitStorage, PluggableTelemetryStorage
 
 # APIs
 from splitio.api.client import HttpClient
@@ -45,7 +47,7 @@
 
 # Synchronizer
 from splitio.sync.synchronizer import SplitTasks, SplitSynchronizers, Synchronizer, \
-    LocalhostSynchronizer, RedisSynchronizer
+    LocalhostSynchronizer, RedisSynchronizer, PluggableSynchronizer
 from splitio.sync.manager import Manager, RedisManager
 from splitio.sync.split import SplitSynchronizer, LocalSplitSynchronizer, LocalhostMode
 from splitio.sync.segment import SegmentSynchronizer, LocalSegmentSynchronizer
@@ -512,6 +514,84 @@ def _build_redis_factory(api_key, cfg):
     return split_factory
 
 
+def _build_pluggable_factory(api_key, cfg):
+    """Build and return a split factory with pluggable storage."""
+    sdk_metadata = util.get_metadata(cfg)
+    if not input_validator.validate_pluggable_adapter(cfg):
+        raise Exception("Pluggable Adapter validation failed, exiting")
+
+    pluggable_adapter = cfg.get('storageWrapper')
+    storage_prefix = cfg.get('storagePrefix')
+    storages = {
+        'splits': PluggableSplitStorage(pluggable_adapter, storage_prefix),
+        'segments': PluggableSegmentStorage(pluggable_adapter, storage_prefix),
+        'impressions': PluggableImpressionsStorage(pluggable_adapter, sdk_metadata, storage_prefix),
+        'events': PluggableEventsStorage(pluggable_adapter, sdk_metadata, storage_prefix),
+        'telemetry': PluggableTelemetryStorage(pluggable_adapter, sdk_metadata, storage_prefix)
+    }
+    telemetry_producer = TelemetryStorageProducer(storages['telemetry'])
+    telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+    telemetry_init_producer = telemetry_producer.get_telemetry_init_producer()
+    # Using same class as redis
+    telemetry_submitter = RedisTelemetrySubmitter(storages['telemetry'])
+
+    unique_keys_synchronizer, clear_filter_sync, unique_keys_task, \
+    clear_filter_task, impressions_count_sync, impressions_count_task, \
+    imp_strategy = set_classes('PLUGGABLE', cfg['impressionsMode'], pluggable_adapter, storage_prefix)
+
+    imp_manager = ImpressionsManager(
+        imp_strategy,
+        telemetry_runtime_producer,
+        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
+        )
+
+    synchronizers = SplitSynchronizers(None, None, None, None,
+        impressions_count_sync,
+        None,
+        unique_keys_synchronizer,
+        clear_filter_sync
+    )
+
+    tasks = SplitTasks(None, None, None, None,
+        impressions_count_task,
+        None,
+        unique_keys_task,
+        clear_filter_task
+    )
+
+    # Using same class as redis for consumer mode only
+    synchronizer = RedisSynchronizer(synchronizers, tasks)
+    recorder = StandardRecorder(
+        imp_manager,
+        storages['events'],
+        storages['impressions'],
+        storages['telemetry']
+    )
+
+    # Using same class as redis for consumer mode only
+    manager = RedisManager(synchronizer)
+    initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True)
+    initialization_thread.start()
+
+    telemetry_init_producer.record_config(cfg, {})
+
+    split_factory = SplitFactory(
+        api_key,
+        storages,
+        cfg['labelsEnabled'],
+        recorder,
+        manager,
+        sdk_ready_flag=None,
+        telemetry_producer=telemetry_producer,
+        telemetry_init_producer=telemetry_init_producer
+    )
+    redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
+    storages['telemetry'].record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
+    telemetry_submitter.synchronize_config()
+
+    return split_factory
+
+
 def _build_localhost_factory(cfg):
     """Build and return a localhost factory for testing/development purposes."""
     telemetry_storage = LocalhostTelemetryStorage()
@@ -606,10 +686,12 @@ def get_factory(api_key, **kwargs):
 
     config = sanitize_config(api_key, kwargs.get('config', {}))
 
-    if config['operationMode'] == 'localhost-standalone':
+    if config['operationMode'] == 'localhost':
         split_factory =  _build_localhost_factory(config)
-    elif config['operationMode'] == 'redis-consumer':
+    elif config['storageType'] == 'redis':
         split_factory = _build_redis_factory(api_key, config)
+    elif config['storageType'] == 'pluggable':
+        split_factory = _build_pluggable_factory(api_key, config)
     else:
         split_factory = _build_in_memory_factory(
         api_key,

splitio/client/input_validator.py

@@ -3,6 +3,7 @@
 import logging
 import re
 import math
+import inspect
 
 from splitio.api import APIException
 from splitio.api.commons import FetchOptions
@@ -517,3 +518,51 @@ def valid_properties(properties):
         _LOGGER.warning('Event has more than 300 properties. Some of them will be trimmed' +
                         ' when processed')
     return True, valid_properties if len(valid_properties) else None, size
+
+def validate_pluggable_adapter(config):
+    """
+    Check if pluggable adapter contains the expected method signature
+
+    :param config: config parameters
+    :type config: Dict
+
+    :return: True if no issue found otherwise False
+    :rtype: bool
+    """
+    if config.get('storageType') != 'pluggable':
+        return True
+
+    if config.get('storageWrapper') is None:
+        _LOGGER.error("Expecting pluggable storage `wrapper` in options, but no valid wrapper instance was provided.")
+        return False
+
+    if config.get('storagePrefix') is not None:
+        if not isinstance(config.get('storagePrefix'), str):
+            _LOGGER.error("Pluggable storage prefix should be string type only")
+            return False
+
+    pluggable_adapter = config.get('storageWrapper')
+    if not isinstance(pluggable_adapter, object):
+        _LOGGER.error("Pluggable storage instance is not inherted from object class")
+        return False
+
+    expected_methods = {'get': 1, 'get_items': 1, 'get_many': 1, 'set': 2, 'push_items': 2,
+                        'delete': 1, 'increment': 2, 'decrement': 2, 'get_keys_by_prefix': 1,
+                        'get_many': 1, 'add_items' : 2, 'remove_items': 2, 'item_contains': 2,
+                        'get_items_count': 1, 'expire': 2}
+    methods = inspect.getmembers(pluggable_adapter, predicate=inspect.ismethod)
+    for exp_method in expected_methods:
+        method_found = False
+        get_method_args = set()
+        for method in methods:
+            if exp_method == method[0]:
+                method_found = True
+                get_method_args = inspect.signature(method[1]).parameters
+                break
+        if not method_found:
+            _LOGGER.error("Pluggable adapter does not have required method: %s" % exp_method)
+            return False
+        if len(get_method_args) < expected_methods[exp_method]:
+            _LOGGER.error("Pluggable adapter method %s has less than required arguments count: %s : " % (exp_method, len(get_method_args)))
+            return False
+    return True

splitio/engine/impressions/__init__.py

@@ -1,21 +1,25 @@
 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.adapters import InMemorySenderAdapter, RedisSenderAdapter, PluggableSenderAdapter
 from splitio.tasks.unique_keys_sync import UniqueKeysSyncTask, ClearFilterSyncTask
 from splitio.sync.unique_keys import UniqueKeysSynchronizer, ClearFilterSynchronizer
 from splitio.sync.impression import ImpressionsCountSynchronizer
 from splitio.tasks.impressions_sync import ImpressionsCountSyncTask
 
-def set_classes(storage_mode, impressions_mode, api_adapter):
+def set_classes(storage_mode, impressions_mode, api_adapter, prefix=None):
     unique_keys_synchronizer = None
     clear_filter_sync = None
     unique_keys_task = None
     clear_filter_task = None
     impressions_count_sync = None
     impressions_count_task = None
     sender_adapter = None
-    if storage_mode == 'REDIS':
+    if storage_mode == 'PLUGGABLE':
+        sender_adapter = PluggableSenderAdapter(api_adapter, prefix)
+        api_telemetry_adapter = sender_adapter
+        api_impressions_adapter = sender_adapter
+    elif storage_mode == 'REDIS':
         sender_adapter = RedisSenderAdapter(api_adapter)
         api_telemetry_adapter = sender_adapter
         api_impressions_adapter = sender_adapter