Merge pull request #95 from splitio/feature/RedisSentinelSupport

[SDKS-70]: FEATURE - Add support for Redis Sentinel

Author: mredolatti

.gitignore

@@ -23,6 +23,7 @@ var/
 *.egg-info/
 .installed.cfg
 *.egg
+.vscode
 
 # PyInstaller
 #  Usually these files are written by a python script from a template

splitio/brokers.py

@@ -513,8 +513,7 @@ def __init__(self, redis):
         split_fetcher = CacheBasedSplitFetcher(split_cache)
 
         impressions_cache = RedisImpressionsCache(redis)
-        delegate_treatment_log = CacheBasedTreatmentLog(impressions_cache)
-        treatment_log = AsyncTreatmentLog(delegate_treatment_log)
+        treatment_log = CacheBasedTreatmentLog(impressions_cache)
 
         metrics_cache = RedisMetricsCache(redis)
         delegate_metrics = CacheBasedMetrics(metrics_cache)

splitio/factories.py

@@ -40,7 +40,7 @@ def __init__(self, api_key, **kwargs):
             config = kwargs['config']
 
         labels_enabled = config.get('labelsEnabled', True)
-        if 'redisHost' in config:
+        if 'redisHost' in config or 'redisSentinels' in config:
             broker = get_redis_broker(api_key, **kwargs)
             self._client = Client(broker, labels_enabled)
             self._manager = RedisSplitManager(broker)
@@ -91,7 +91,6 @@ def client(self):  # pragma: no cover
         """
         return self._client
 
-
     def manager(self):  # pragma: no cover
         """Get the split manager implementation.
         :return: The split manager implementation.

splitio/redis_support.py

@@ -11,6 +11,7 @@
 try:
     from jsonpickle import decode, encode
     from redis import StrictRedis
+    from redis.sentinel import Sentinel
 except ImportError:
     def missing_redis_dependencies(*args, **kwargs):
         raise NotImplementedError('Missing Redis support dependencies.')
@@ -38,6 +39,10 @@ def missing_redis_dependencies(*args, **kwargs):
 }
 
 
+class SentinelConfigurationException(Exception):
+    pass
+
+
 class RedisSegmentCache(SegmentCache):
     '''
     '''
@@ -66,21 +71,25 @@ def register_segment(self, segment_name):
         :param segment_name: Name of the segment.
         :type segment_name: str
         '''
-        self._redis.sadd(
-            RedisSegmentCache._KEY_TEMPLATE.format(suffix='registered'),
-            segment_name
-        )
+        # self._redis.sadd(
+        #     RedisSegmentCache._KEY_TEMPLATE.format(suffix='registered'),
+        #     segment_name
+        # )
+        # @TODO The Segment logic for redis should be removed.
+        pass
 
     def unregister_segment(self, segment_name):
         '''
         Unregister a segment from the automatic update process.
         :param segment_name: Name of the segment.
         :type segment_name: str
         '''
-        self._redis.srem(
-            RedisSegmentCache._KEY_TEMPLATE.format(suffix='registered'),
-            segment_name
-        )
+        # self._redis.srem(
+        #     RedisSegmentCache._KEY_TEMPLATE.format(suffix='registered'),
+        #     segment_name
+        # )
+        # @TODO The Segment logic for redis should be removed.
+        pass
 
     def get_registered_segments(self):
         '''
@@ -799,8 +808,11 @@ def get_redis(config):
             config['redisFactory'], 'redisFactory'
         )
         return redis_factory()
-
-    return default_redis_factory(config)
+    else:
+        if 'redisSentinels' in config:
+            return default_redis_sentinel_factory(config)
+        else:
+            return default_redis_factory(config)
 
 
 def default_redis_factory(config):
@@ -860,3 +872,82 @@ def default_redis_factory(config):
         max_connections=max_connections
     )
     return PrefixDecorator(redis, prefix=prefix)
+
+
+def default_redis_sentinel_factory(config):
+    '''
+    Default redis client factory for sentinel mode.
+    :param config: A dict with the Redis configuration parameters
+    :type config: dict
+    :return: A Sentinel object using the provided config values
+    :rtype: Sentinel
+    '''
+    sentinels = config.get('redisSentinels')
+
+    if (sentinels is None):
+        raise SentinelConfigurationException('redisSentinels must be specified.')
+    if (not isinstance(sentinels, list)):
+        raise SentinelConfigurationException('Sentinels must be an array of elements in the form of'
+                                             ' [(ip, port)].')
+    if (len(sentinels) == 0):
+        raise SentinelConfigurationException('It must be at least one sentinel.')
+    if not all(isinstance(s, tuple) for s in sentinels):
+        raise SentinelConfigurationException('Sentinels must respect the tuple structure'
+                                             '[(ip, port)].')
+
+    master_service = config.get('redisMasterService')
+
+    if (master_service is None):
+        raise SentinelConfigurationException('redisMasterService must be specified.')
+
+    db = config.get('redisDb', 0)
+    password = config.get('redisPassword', None)
+    socket_timeout = config.get('redisSocketTimeout', None)
+    socket_connect_timeout = config.get('redisSocketConnectTimeout', None)
+    socket_keepalive = config.get('redisSocketKeepalive', None)
+    socket_keepalive_options = config.get('redisSocketKeepaliveOptions', None)
+    connection_pool = config.get('redisConnectionPool', None)
+    unix_socket_path = config.get('redisUnixSocketPath', None)
+    encoding = config.get('redisEncoding', 'utf-8')
+    encoding_errors = config.get('redisEncodingErrors', 'strict')
+    charset = config.get('redisCharset', None)
+    errors = config.get('redisErrors', None)
+    decode_responses = config.get('redisDecodeResponses', False)
+    retry_on_timeout = config.get('redisRetryOnTimeout', False)
+    ssl = config.get('redisSsl', False)
+    ssl_keyfile = config.get('redisSslKeyfile', None)
+    ssl_certfile = config.get('redisSslCertfile', None)
+    ssl_cert_reqs = config.get('redisSslCertReqs', None)
+    ssl_ca_certs = config.get('redisSslCaCerts', None)
+    max_connections = config.get('redisMaxConnections', None)
+    prefix = config.get('redisPrefix')
+
+    sentinel = Sentinel(
+        sentinels,
+        0,
+        {
+            'db': db,
+            'password': password,
+            'socket_timeout': socket_timeout,
+            'socket_connect_timeout': socket_connect_timeout,
+            'socket_keepalive': socket_keepalive,
+            'socket_keepalive_options': socket_keepalive_options,
+            'connection_pool': connection_pool,
+            'unix_socket_path': unix_socket_path,
+            'encoding': encoding,
+            'encoding_errors': encoding_errors,
+            'charset': charset,
+            'errors': errors,
+            'decode_responses': decode_responses,
+            'retry_on_timeout': retry_on_timeout,
+            'ssl': ssl,
+            'ssl_keyfile': ssl_keyfile,
+            'ssl_certfile': ssl_certfile,
+            'ssl_cert_reqs': ssl_cert_reqs,
+            'ssl_ca_certs': ssl_ca_certs,
+            'max_connections': max_connections
+        }
+    )
+
+    redis = sentinel.master_for(master_service)
+    return PrefixDecorator(redis, prefix=prefix)

splitio/tests/splitChangesReadOnly.json

@@ -0,0 +1,46 @@
+{
+    "splits": [
+        {
+            "orgId": null,
+            "environment": null,
+            "trafficTypeId": null,
+            "trafficTypeName": null,
+            "name": "test_read_only_1",
+            "seed": -1329591480,
+            "status": "ACTIVE",
+            "killed": false,
+            "defaultTreatment": "off",
+            "changeNumber": 1325599980,
+            "conditions": [
+                {
+                    "matcherGroup": {
+                        "combiner": "AND",
+                        "matchers": [
+                            {
+                                "keySelector": null,
+                                "matcherType": "WHITELIST",
+                                "negate": false,
+                                "userDefinedSegmentMatcherData": null,
+                                "whitelistMatcherData": {
+                                    "whitelist": [
+                                    "valid"
+                                    ]
+                                },
+                                "unaryNumericMatcherData": null,
+                                "betweenMatcherData": null
+                            }
+                        ]
+                    },
+                    "partitions": [
+                        {
+                            "treatment": "on",
+                            "size": 100
+                        }
+                    ]
+                }
+            ]
+        }
+    ],
+    "since": -1,
+    "till": 1461957424937
+  }

splitio/tests/test_clients.py

@@ -1165,6 +1165,7 @@ def setUp(self):
             'splitio.tests.test_clients.LocalhostBroker._build_split_fetcher')
 
         self.open_mock = self.patch_builtin('open')
+        self.threading_mock = self.patch('threading.Thread')
         self.broker = LocalhostBroker()
 
     def test_skips_comment_lines(self):

splitio/tests/test_factories.py

@@ -0,0 +1,61 @@
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+try:
+    from unittest import mock
+except ImportError:
+    # Python 2
+    import mock
+
+from unittest import TestCase
+
+from splitio import get_factory
+from splitio.redis_support import SentinelConfigurationException
+
+
+class RedisSentinelFactory(TestCase):
+    def test_redis_factory_with_empty_sentinels_array(self):
+        config = {
+            'redisDb': 0,
+            'redisPrefix': 'test',
+            'redisSentinels': [],
+            'redisMasterService': 'mymaster',
+            'redisSocketTimeout': 3
+        }
+
+        with self.assertRaises(SentinelConfigurationException):
+            get_factory('abc', config=config)
+
+    def test_redis_factory_with_wrong_type_in_sentinels_array(self):
+        config = {
+            'redisDb': 0,
+            'redisPrefix': 'test',
+            'redisSentinels': 'abc',
+            'redisMasterService': 'mymaster',
+            'redisSocketTimeout': 3
+        }
+
+        with self.assertRaises(SentinelConfigurationException):
+            get_factory('abc', config=config)
+
+    def test_redis_factory_with_wrong_data_in_sentinels_array(self):
+        config = {
+            'redisDb': 0,
+            'redisPrefix': 'test',
+            'redisSentinels': ['asdasd'],
+            'redisMasterService': 'mymaster',
+            'redisSocketTimeout': 3
+        }
+
+        with self.assertRaises(SentinelConfigurationException):
+            get_factory('abc', config=config)
+
+    def test_redis_factory_with_without_master_service(self):
+        config = {
+            'redisDb': 0,
+            'redisPrefix': 'test',
+            'redisSentinels': [('test', 1234)],
+            'redisSocketTimeout': 3
+        }
+
+        with self.assertRaises(SentinelConfigurationException):
+            get_factory('abc', config=config)

splitio/tests/test_redis_cache.py

@@ -8,9 +8,14 @@
 
 from os.path import dirname, join
 from unittest import TestCase
+from splitio.tests.utils import MockUtilsMixin
 from json import load
 
 from splitio.redis_support import (RedisSplitCache, RedisSegmentCache, get_redis)
+from redis import StrictRedis
+from splitio.clients import Client
+from splitio.prefix_decorator import PrefixDecorator
+from splitio.brokers import RedisBroker
 
 class CacheInterfacesTests(TestCase):
     def setUp(self):
@@ -42,20 +47,58 @@ def test_split_cache_interface(self):
         self._redis_split_cache.set_change_number(1212)
         self.assertEqual(1212, self._redis_split_cache.get_change_number())
 
+    # @TODO This tests should be removed regarding that this is not supported by redis now.
+    # def testSegmentCacheInterface(self):
+    #     with open(self._segment_changes_file_name) as f:
+    #         self._json = load(f)
+    #         segment_name = self._json['name']
+    #         segment_change_number = self._json['till']
+    #         segment_keys = self._json['added']
 
+    #     self._redis_segment_cache.set_change_number(segment_name, segment_change_number)
+    #     self.assertEqual(segment_change_number, self._redis_segment_cache.get_change_number(segment_name))
 
-    def testSegmentCacheInterface(self):
-        with open(self._segment_changes_file_name) as f:
+    #     self._redis_segment_cache.add_keys_to_segment(segment_name, segment_keys)
+    #     self.assertTrue(self._redis_segment_cache.is_in_segment(segment_name, segment_keys[0]))
+
+    #     self._redis_segment_cache.remove_keys_from_segment(segment_name, [segment_keys[0]])
+    #     self.assertFalse(self._redis_segment_cache.is_in_segment(segment_name, segment_keys[0]))
+
+class ReadOnlyRedisMock(PrefixDecorator):
+
+    def __init__(self, *args, **kwargs):
+        """
+        Bases on PrefixDecorator.
+        """
+        PrefixDecorator.__init__(self, *args, **kwargs)
+
+    def sadd(self, name, *values):
+        """
+        Decorated sadd to simulate read only exception error.
+        """
+        raise Exception('ReadOnlyError')
+
+class RedisReadOnlyTest(TestCase, MockUtilsMixin):
+    def setUp(self):
+        self._split_changes_file_name = join(dirname(__file__), 'splitChangesReadOnly.json')
+        
+        with open(self._split_changes_file_name) as f:
             self._json = load(f)
-            segment_name = self._json['name']
-            segment_change_number = self._json['till']
-            segment_keys = self._json['added']
+            split_definition = self._json['splits'][0]
+            split_name = split_definition['name']
+
+        self._redis = get_redis({'redisPrefix': 'test'})
 
-        self._redis_segment_cache.set_change_number(segment_name, segment_change_number)
-        self.assertEqual(segment_change_number, self._redis_segment_cache.get_change_number(segment_name))
+        self._mocked_redis = ReadOnlyRedisMock(self._redis)
+        self._redis_split_cache = RedisSplitCache(self._redis)
+        self._redis_split_cache.add_split(split_name, split_definition)
+        self._client = Client(RedisBroker(self._mocked_redis))
 
-        self._redis_segment_cache.add_keys_to_segment(segment_name, segment_keys)
-        self.assertTrue(self._redis_segment_cache.is_in_segment(segment_name, segment_keys[0]))
+        self._impression = mock.MagicMock()
+        self._start = mock.MagicMock()
+        self._operation = mock.MagicMock()
 
-        self._redis_segment_cache.remove_keys_from_segment(segment_name, [segment_keys[0]])
-        self.assertFalse(self._redis_segment_cache.is_in_segment(segment_name, segment_keys[0]))
+    def test_redis_read_only_mode(self):
+        self.assertEqual(self._client.get_treatment('valid', 'test_read_only_1'), 'on')
+        self.assertEqual(self._client.get_treatment('invalid', 'test_read_only_1'), 'off')
+        self.assertEqual(self._client.get_treatment('valid', 'test_read_only_1_invalid'), 'control')