Merge pull request #325 from splitio/refactor-om-and-st

updated operationMode and storageType params values

Author: chillaq

splitio/client/config.py

@@ -10,7 +10,7 @@
 
 
 DEFAULT_CONFIG = {
-    'operationMode': 'in-memory',
+    'operationMode': 'standalone',
     'connectionTimeout': 1500,
     'streamingEnabled': True,
     'featuresRefreshRate': 30,
@@ -64,34 +64,35 @@
 
 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':
         _LOGGER.debug('Using Localhost operation mode')
-        return 'localhost-standalone'
+        return 'localhost', 'localhost'
 
     if 'redisHost' in config or 'redisSentinels' in config:
         _LOGGER.debug('Using Redis storage operation mode')
-        return 'redis-consumer'
+        return 'consumer', 'redis'
 
-    if 'storageType' in config:
+    if  config.get('storageType') is not None:
         if config.get('storageType').lower() == 'pluggable':
             _LOGGER.debug('Using Pluggable storage operation mode')
-            return 'pluggable'
+            return 'consumer', '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'
+    return 'standalone', 'memory'
 
 
-def _sanitize_impressions_mode(operation_mode, mode, refresh_rate=None):
+def _sanitize_impressions_mode(storage_type, mode, refresh_rate=None):
     """
     Check supplied impressions mode and adjust refresh rate.
 
@@ -110,7 +111,7 @@ def _sanitize_impressions_mode(operation_mode, mode, refresh_rate=None):
                             'one of the following values: `debug`, `none` or `optimized`. '
                             ' Defaulting to `optimized` mode.')
 
-    if operation_mode == 'pluggable' and mode != ImpressionsMode.DEBUG:
+    if storage_type == 'pluggable' and mode != ImpressionsMode.DEBUG:
         mode = ImpressionsMode.DEBUG
         _LOGGER.warning('`pluggable` storageMode only support `debug` impressionMode, adjusting impressionsMode to `debug`. ')
 
@@ -135,10 +136,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['operationMode'], 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

@@ -671,11 +671,11 @@ 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['operationMode'] == 'pluggable':
+    elif config['storageType'] == 'pluggable':
         split_factory = _build_pluggable_factory(api_key, config)
     else:
         split_factory = _build_in_memory_factory(

splitio/models/telemetry.py

@@ -123,14 +123,13 @@ class StorageType(Enum):
     """Storage types constants"""
     MEMORY = 'memory'
     REDIS = 'redis'
-    LOCALHOST = 'localhost'
     PLUGGABLE = 'pluggable'
 
 class OperationMode(Enum):
     """Storage modes constants"""
-    MEMORY = 'inmemory'
-    REDIS = 'redis-consumer'
-    PLUGGABLE = 'pluggable'
+    STANDALONE = 'standalone'
+    CONSUMER = 'consumer'
+    PARTIAL_CONSUMER = 'partial_consumer'
 
 def get_latency_bucket_index(micros):
     """
@@ -727,7 +726,7 @@ def record_config(self, config, extra_config):
         Record configurations.
 
         :param config: config dict: {
-            'operationMode': string, 'storageType': string, 'streamingEnabled': boolean,
+            'operationMode': int, 'storageType': string, 'streamingEnabled': boolean,
             'refreshRate' : {
                 'featuresRefreshRate': int,
                 'segmentsRefreshRate': int,
@@ -746,7 +745,7 @@ def record_config(self, config, extra_config):
         """
         with self._lock:
             self._operation_mode = self._get_operation_mode(config[ConfigParams.OPERATION_MODE.value])
-            self._storage_type = self._get_storage_type(config[ConfigParams.OPERATION_MODE.value])
+            self._storage_type = self._get_storage_type(config[ConfigParams.OPERATION_MODE.value], config[ConfigParams.STORAGE_TYPE.value])
             self._streaming_enabled = config[ConfigParams.STREAMING_ENABLED.value]
             self._refresh_rate = self._get_refresh_rates(config)
             self._url_override = self._get_url_overrides(extra_config)
@@ -853,14 +852,14 @@ def _get_operation_mode(self, op_mode):
         :rtype: int
         """
         with self._lock:
-            if OperationMode.MEMORY.value in op_mode:
+            if op_mode == OperationMode.STANDALONE.value:
                 return 0
-            elif op_mode == OperationMode.REDIS.value:
+            elif op_mode == OperationMode.CONSUMER.value:
                 return 1
             else:
                 return 2
 
-    def _get_storage_type(self, op_mode):
+    def _get_storage_type(self, op_mode, st_type):
         """
         Get storage type from operation mode
 
@@ -871,9 +870,9 @@ def _get_storage_type(self, op_mode):
         :rtype: str
         """
         with self._lock:
-            if OperationMode.MEMORY.value in op_mode:
+            if op_mode == OperationMode.STANDALONE.value:
                 return StorageType.MEMORY.value
-            elif StorageType.REDIS.value in op_mode:
+            elif st_type == StorageType.REDIS.value:
                 return StorageType.REDIS.value
             else:
                 return StorageType.PLUGGABLE.value

tests/client/test_config.py

@@ -10,27 +10,31 @@ class ConfigSanitizationTests(object):
 
     def test_parse_operation_mode(self):
         """Make sure operation mode is correctly captured."""
-        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'
+        assert (config._parse_operation_mode('some', {})) == ('standalone', 'memory')
+        assert (config._parse_operation_mode('localhost', {})) == ('localhost', 'localhost')
+        assert (config._parse_operation_mode('some', {'redisHost': 'x'})) == ('consumer', 'redis')
+        assert (config._parse_operation_mode('some', {'storageType': 'pluggable'})) == ('consumer', 'pluggable')
+        assert (config._parse_operation_mode('some', {'storageType': 'custom2'})) == ('standalone', 'memory')
 
     def test_sanitize_imp_mode(self):
         """Test sanitization of impressions mode."""
-        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'OPTIMIZED', 1)
+        mode, rate = config._sanitize_impressions_mode('memory', 'OPTIMIZED', 1)
         assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 60
 
-        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'DEBUG', 1)
+        mode, rate = config._sanitize_impressions_mode('memory', 'DEBUG', 1)
         assert mode == ImpressionsMode.DEBUG
         assert rate == 1
 
-        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'debug', 1)
+        mode, rate = config._sanitize_impressions_mode('redis', 'OPTIMIZED', 1)
+        assert mode == ImpressionsMode.OPTIMIZED
+        assert rate == 60
+
+        mode, rate = config._sanitize_impressions_mode('redis', 'debug', 1)
         assert mode == ImpressionsMode.DEBUG
         assert rate == 1
 
-        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'ANYTHING', 200)
+        mode, rate = config._sanitize_impressions_mode('memory', 'ANYTHING', 200)
         assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 200
 
@@ -46,15 +50,15 @@ def test_sanitize_imp_mode(self):
         assert mode == ImpressionsMode.DEBUG
         assert rate == 200
 
-        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 43, -1)
+        mode, rate = config._sanitize_impressions_mode('memory', 43, -1)
         assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 60
 
-        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'OPTIMIZED')
+        mode, rate = config._sanitize_impressions_mode('memory', 'OPTIMIZED')
         assert mode == ImpressionsMode.OPTIMIZED
         assert rate == 300
 
-        mode, rate = config._sanitize_impressions_mode('inmemory-standalone', 'DEBUG')
+        mode, rate = config._sanitize_impressions_mode('memory', 'DEBUG')
         assert mode == ImpressionsMode.DEBUG
         assert rate == 60
 

tests/engine/test_telemetry.py

@@ -19,9 +19,6 @@ def test_instances(self):
         assert(telemetry_producer._telemetry_init_producer == telemetry_producer.get_telemetry_init_producer())
         assert(telemetry_producer._telemetry_runtime_producer == telemetry_producer.get_telemetry_runtime_producer())
 
-class TelemetryInitProducerTest(object):
-    """TelemetryInitProducer test."""
-
     def test_record_config(self, mocker):
         telemetry_storage = mocker.Mock()
         telemetry_init_producer = TelemetryInitProducer(telemetry_storage)
@@ -30,7 +27,7 @@ def record_config(*args, **kwargs):
             self.passed_config = args[0]
 
         telemetry_storage.record_config.side_effect = record_config
-        telemetry_init_producer.record_config({'bT':0, 'nR':0, 'uC': 0})
+        telemetry_init_producer.record_config({'bT':0, 'nR':0, 'uC': 0}, {})
         assert(self.passed_config == {'bT':0, 'nR':0, 'uC': 0})
 
     def test_record_ready_time(self, mocker):
@@ -58,9 +55,6 @@ def test_record_not_ready_usage(self, mocker):
         telemetry_init_producer.record_not_ready_usage()
         assert(mocker.called)
 
-class TelemetryEvaluationProducerTest(object):
-    """Telemetry evaluation producer test class."""
-
     def test_record_latency(self, mocker):
         telemetry_storage = mocker.Mock()
         telemetry_evaluation_producer = TelemetryEvaluationProducer(telemetry_storage)
@@ -84,10 +78,6 @@ def record_exception(*args, **kwargs):
         telemetry_evaluation_producer.record_exception('method')
         assert(self.passed_method == 'method')
 
-
-class TelemetryRuntimeProducerTest(object):
-    """Telemetry runtime producer test."""
-
     def test_add_tag(self, mocker):
         telemetry_storage = mocker.Mock()
         telemetry_runtime_producer = TelemetryRuntimeProducer(telemetry_storage)
@@ -210,9 +200,6 @@ def test_instances(self):
         assert(telemetry_consumer._telemetry_init_consumer == telemetry_consumer.get_telemetry_init_consumer())
         assert(telemetry_consumer._telemetry_runtime_consumer == telemetry_consumer.get_telemetry_runtime_consumer())
 
-class TelemetryInitConsumerTest(object):
-    """TelemetryInitConsumer test."""
-
     @mock.patch('splitio.storage.inmemmory.InMemoryTelemetryStorage.get_bur_time_outs')
     def test_get_bur_time_outs(self, mocker):
         telemetry_storage = InMemoryTelemetryStorage()
@@ -234,9 +221,6 @@ def get_not_ready_usage(self, mocker):
         telemetry_init_consumer.get_config_stats()
         assert(mocker.called)
 
-class TelemetryEvaluationConsumerTest(object):
-    """TelemetryEvaluationConsumer test."""
-
     @mock.patch('splitio.storage.inmemmory.InMemoryTelemetryStorage.pop_exceptions')
     def pop_exceptions(self, mocker):
         telemetry_storage = InMemoryTelemetryStorage()
@@ -251,9 +235,6 @@ def pop_latencies(self, mocker):
         telemetry_evaluation_consumer.pop_latencies()
         assert(mocker.called)
 
-class TelemetryRuntimeConsumerTest(object):
-    """TelemetryRuntimeConsumer test."""
-
     def test_get_impressions_stats(self, mocker):
         telemetry_storage = mocker.Mock()
         telemetry_runtime_consumer = TelemetryRuntimeConsumer(telemetry_storage)

tests/models/test_telemetry_model.py

@@ -37,11 +37,10 @@ def test_latency_bucket_index(self):
             assert(result_bucket == ModelTelemetry.get_latency_bucket_index(latency))
 
     def test_storage_type_and_operation_mode(self, mocker):
-        assert(StorageType.LOCALHOST.value == 'localhost')
         assert(StorageType.MEMORY.value == 'memory')
         assert(StorageType.REDIS.value == 'redis')
-        assert(OperationMode.MEMORY.value == 'inmemory')
-        assert(OperationMode.REDIS.value == 'redis-consumer')
+        assert(OperationMode.STANDALONE.value == 'standalone')
+        assert(OperationMode.CONSUMER.value == 'consumer')
 
     def test_method_latencies(self, mocker):
         method_latencies = MethodLatencies()
@@ -238,7 +237,7 @@ def test_streaming_events(self, mocker):
 
     def test_telemetry_config(self):
         telemetry_config = TelemetryConfig()
-        config = {'operationMode': 'inmemory',
+        config = {'operationMode': 'standalone',
                   'streamingEnabled': True,
                   'impressionsQueueSize': 100,
                   'eventsQueueSize': 200,
@@ -249,10 +248,11 @@ def test_telemetry_config(self):
                   'impressionsRefreshRate': 60,
                   'eventsPushRate': 60,
                   'metricsRefreshRate': 10,
+                  'storageType': None
                 }
         telemetry_config.record_config(config, {})
         assert(telemetry_config.get_stats() == {'oM': 0,
-            'sT': telemetry_config._get_storage_type(config['operationMode']),
+            'sT': telemetry_config._get_storage_type(config['operationMode'], config['storageType']),
             'sE': config['streamingEnabled'],
             'rR': {'sp': 30, 'se': 30, 'im': 60, 'ev': 60, 'te': 10},
             'uO':  {'s': False, 'e': False, 'a': False, 'st': False, 't': False},

tests/storage/test_inmemory_storage.py

@@ -477,7 +477,7 @@ def test_resets(self):
 
     def test_record_config(self):
         storage = InMemoryTelemetryStorage()
-        config = {'operationMode': 'inmemory',
+        config = {'operationMode': 'standalone',
                   'streamingEnabled': True,
                   'impressionsQueueSize': 100,
                   'eventsQueueSize': 200,
@@ -488,11 +488,12 @@ def test_record_config(self):
                   'impressionsRefreshRate': 60,
                   'eventsPushRate': 60,
                   'metricsRefreshRate': 10,
+                  'storageType': None
                   }
         storage.record_config(config, {})
         storage.record_active_and_redundant_factories(1, 0)
         assert(storage._tel_config.get_stats() == {'oM': 0,
-            'sT': storage._tel_config._get_storage_type(config['operationMode']),
+            'sT': storage._tel_config._get_storage_type(config['operationMode'], config['storageType']),
             'sE': config['streamingEnabled'],
             'rR': {'sp': 30, 'se': 30, 'im': 60, 'ev': 60, 'te': 10},
             'uO':  {'s': False, 'e': False, 'a': False, 'st': False, 't': False},

tests/storage/test_pluggable.py

@@ -657,7 +657,7 @@ def expire_keys_mock(*args, **kwargs):
 
     def test_push_config_stats(self):
         self.pluggable_telemetry_storage.record_config(
-                    {'operationMode': 'inmemory',
+                    {'operationMode': 'standalone',
                   'streamingEnabled': True,
                   'impressionsQueueSize': 100,
                   'eventsQueueSize': 200,
@@ -668,6 +668,7 @@ def test_push_config_stats(self):
                   'impressionsRefreshRate': 60,
                   'eventsPushRate': 60,
                   'metricsRefreshRate': 10,
+                  'storageType': None
                   }, {}
         )
         self.pluggable_telemetry_storage.record_active_and_redundant_factories(2, 1)

tests/sync/test_telemetry.py

@@ -86,6 +86,7 @@ def test_synchronize_telemetry(self, mocker):
         telemetry_storage._http_latencies._token =  [0] * 23
 
         telemetry_storage.record_config({'operationMode': 'inmemory',
+                                         'storageType': None,
                                         'streamingEnabled': True,
                                         'impressionsQueueSize': 100,
                                         'eventsQueueSize': 200,