Merge pull request #323 from splitio/factory-integration

Factory integration for pluggable storage and tests

Author: chillaq

splitio/client/config.py

@@ -56,6 +56,9 @@
     'localhostRefreshEnabled': False,
     'preforkedInitialization': False,
     'dataSampling': DEFAULT_DATA_SAMPLING,
+    'storageWrapper': None,
+    'storagePrefix': None,
+    'storageType': None
 }
 
 
@@ -70,15 +73,25 @@ def _parse_operation_mode(apikey, config):
     :rtype: str
     """
     if apikey == 'localhost':
+        _LOGGER.debug('Using Localhost operation mode')
         return 'localhost-standalone'
 
     if 'redisHost' in config or 'redisSentinels' in config:
+        _LOGGER.debug('Using Redis storage operation mode')
         return 'redis-consumer'
 
+    if 'storageType' in config:
+        if config.get('storageType').lower() == 'pluggable':
+            _LOGGER.debug('Using Pluggable storage operation mode')
+            return 'pluggable'
+        _LOGGER.warning('You passed an invalid storageType, acceptable value is '
+                            '`pluggable`. Defaulting storage to In-Memory mode.')
+
+    _LOGGER.debug('Using In-Memory operation mode')
     return 'inmemory-standalone'
 
 
-def _sanitize_impressions_mode(mode, refresh_rate=None):
+def _sanitize_impressions_mode(operation_mode, mode, refresh_rate=None):
     """
     Check supplied impressions mode and adjust refresh rate.
 
@@ -92,10 +105,14 @@ 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 operation_mode == 'pluggable' and mode != ImpressionsMode.DEBUG:
+        mode = ImpressionsMode.DEBUG
+        _LOGGER.warning('`pluggable` storageMode only support `debug` impressionMode, adjusting impressionsMode to `debug`. ')
 
     if mode == ImpressionsMode.DEBUG:
         refresh_rate = max(1, refresh_rate) if refresh_rate is not None else 60
@@ -121,7 +138,7 @@ def sanitize(apikey, config):
     config['operationMode'] = _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['operationMode'], 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,69 @@ 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)
+
+    imp_manager = ImpressionsManager(
+        imp_strategy,
+        telemetry_runtime_producer,
+        _wrap_impression_listener(cfg['impressionListener'], sdk_metadata),
+        )
+
+    synchronizer = PluggableSynchronizer()
+    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()
@@ -610,6 +675,8 @@ def get_factory(api_key, **kwargs):
         split_factory =  _build_localhost_factory(config)
     elif config['operationMode'] == 'redis-consumer':
         split_factory = _build_redis_factory(api_key, config)
+    elif config['operationMode'] == 'pluggable':
+        split_factory = _build_pluggable_factory(api_key, config)
     else:
         split_factory = _build_in_memory_factory(
         api_key,

splitio/client/input_validator.py

@@ -529,21 +529,21 @@ def validate_pluggable_adapter(config):
     :return: True if no issue found otherwise False
     :rtype: bool
     """
-    if config.get('storageType') != 'PLUGGABLE':
+    if config.get('storageType') != 'pluggable':
         return True
 
     if config.get('storageWrapper') is None:
-        _LOGGER.error("Expecting custom storage `wrapper` in options, but no valid wrapper instance was provided.")
+        _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("Custom storage prefix should be string type only")
+            _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("Custom storage instance is not inherted from object class")
+        _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,

splitio/engine/impressions/__init__.py

@@ -15,7 +15,10 @@ def set_classes(storage_mode, impressions_mode, api_adapter):
     impressions_count_sync = None
     impressions_count_task = None
     sender_adapter = None
-    if storage_mode == 'REDIS':
+    if storage_mode == 'PLUGGABLE':
+        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

splitio/models/telemetry.py

@@ -124,11 +124,13 @@ class StorageType(Enum):
     MEMORY = 'memory'
     REDIS = 'redis'
     LOCALHOST = 'localhost'
+    PLUGGABLE = 'pluggable'
 
 class OperationMode(Enum):
     """Storage modes constants"""
     MEMORY = 'inmemory'
     REDIS = 'redis-consumer'
+    PLUGGABLE = 'pluggable'
 
 def get_latency_bucket_index(micros):
     """
@@ -874,7 +876,7 @@ def _get_storage_type(self, op_mode):
             elif StorageType.REDIS.value in op_mode:
                 return StorageType.REDIS.value
             else:
-                return StorageType.LOCALHOST.value
+                return StorageType.PLUGGABLE.value
 
     def _get_refresh_rates(self, config):
         """

splitio/storage/pluggable.py

@@ -771,7 +771,7 @@ def record_active_and_redundant_factories(self, active_factory_count, redundant_
         """
         self._tel_config.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
 
-    def record_latency(self, method, latency):
+    def record_latency(self, method, bucket):
         """
         record latency data
 
@@ -780,7 +780,6 @@ def record_latency(self, method, latency):
         :param latency: latency
         :type latency: int64
         """
-        bucket = get_latency_bucket_index(latency)
         latency_key = self._telemetry_latencies_key + '::' + self._sdk_metadata + '/' + method.value + '/' + str(bucket)
         result = self._pluggable_adapter.increment(latency_key, 1)
         self.expire_keys(latency_key, self._TELEMETRY_KEY_DEFAULT_TTL, 1, result)

splitio/storage/redis.py

@@ -649,7 +649,7 @@ def record_active_and_redundant_factories(self, active_factory_count, redundant_
         """Record active and redundant factories."""
         self._tel_config.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
 
-    def add_latency_to_pipe(self, method, latency, pipe):
+    def add_latency_to_pipe(self, method, bucket, pipe):
         """
         record latency data
 
@@ -660,7 +660,6 @@ def add_latency_to_pipe(self, method, latency, pipe):
         :param pipe: Redis pipe.
         :type pipe: redis.pipe
         """
-        bucket = get_latency_bucket_index(latency)
         pipe.hincrby(self._TELEMETRY_LATENCIES_KEY, self._sdk_metadata.sdk_version + '/' + self._sdk_metadata.instance_name + '/' + self._sdk_metadata.instance_ip + '/' +
             method.value + '/' + str(bucket), 1)
 

splitio/sync/synchronizer.py

@@ -589,3 +589,69 @@ def shutdown(self, blocking):
         :type blocking: bool
         """
         self.stop_periodic_fetching()
+
+
+class PluggableSynchronizer(BaseSynchronizer):
+    """Plugable Synchronizer."""
+
+    def synchronize_segment(self, segment_name, till):
+        """
+        Synchronize particular segment.
+
+        :param segment_name: segment associated
+        :type segment_name: str
+        :param till: to fetch
+        :type till: int
+        """
+        pass
+
+    def synchronize_splits(self, till):
+        """
+        Synchronize all splits.
+
+        :param till: to fetch
+        :type till: int
+        """
+        pass
+
+    def sync_all(self):
+        """Synchronize all split data."""
+        pass
+
+    def start_periodic_fetching(self):
+        """Start fetchers for splits and segments."""
+        pass
+
+    def stop_periodic_fetching(self):
+        """Stop fetchers for splits and segments."""
+        pass
+
+    def start_periodic_data_recording(self):
+        """Start recorders."""
+        pass
+
+    def stop_periodic_data_recording(self, blocking):
+        """Stop recorders."""
+        pass
+
+    def kill_split(self, split_name, default_treatment, change_number):
+        """
+        Kill a split locally.
+
+        :param split_name: name of the split to perform kill
+        :type split_name: str
+        :param default_treatment: name of the default treatment to return
+        :type default_treatment: str
+        :param change_number: change_number
+        :type change_number: int
+        """
+        pass
+
+    def shutdown(self, blocking):
+        """
+        Stop tasks.
+
+        :param blocking:flag to wait until tasks are stopped
+        :type blocking: bool
+        """
+        pass

tests/client/test_config.py

@@ -13,34 +13,48 @@ def test_parse_operation_mode(self):
         assert config._parse_operation_mode('some', {}) == 'inmemory-standalone'
         assert config._parse_operation_mode('localhost', {}) == 'localhost-standalone'
         assert config._parse_operation_mode('some', {'redisHost': 'x'}) == 'redis-consumer'
+        assert config._parse_operation_mode('some', {'storageType': 'pluggable'}) == 'pluggable'
+        assert config._parse_operation_mode('some', {'storageType': 'custom2'}) == 'inmemory-standalone'
 
     def test_sanitize_imp_mode(self):
         """Test sanitization of impressions mode."""
-        mode, rate = config._sanitize_impressions_mode('OPTIMIZED', 1)
+        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'OPTIMIZED', 1)
         assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 60
 
-        mode, rate = config._sanitize_impressions_mode('DEBUG', 1)
+        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'DEBUG', 1)
         assert mode == ImpressionsMode.DEBUG
         assert rate == 1
 
-        mode, rate = config._sanitize_impressions_mode('debug', 1)
+        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'debug', 1)
         assert mode == ImpressionsMode.DEBUG
         assert rate == 1
 
-        mode, rate = config._sanitize_impressions_mode('ANYTHING', 200)
+        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'ANYTHING', 200)
         assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 200
 
-        mode, rate = config._sanitize_impressions_mode(43, -1)
+        mode, rate = config._sanitize_impressions_mode('pluggable', 'ANYTHING', 200)
+        assert mode == ImpressionsMode.DEBUG
+        assert rate == 200
+
+        mode, rate = config._sanitize_impressions_mode('pluggable', 'NONE', 200)
+        assert mode == ImpressionsMode.DEBUG
+        assert rate == 200
+
+        mode, rate = config._sanitize_impressions_mode('pluggable', 'OPTIMIZED', 200)
+        assert mode == ImpressionsMode.DEBUG
+        assert rate == 200
+
+        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 43, -1)
         assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 60
 
-        mode, rate = config._sanitize_impressions_mode('OPTIMIZED')
+        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'OPTIMIZED')
         assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 300
 
-        mode, rate = config._sanitize_impressions_mode('DEBUG')
+        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'DEBUG')
         assert mode == ImpressionsMode.DEBUG
         assert rate == 60