Merge pull request #97 from splitio/development

Development

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

CHANGES.txt

@@ -1,3 +1,9 @@
+6.0.0 (Aug 29, 2018)
+ - Add support for redis sentinel
+ - UWSGI performance boost (breaking change)
+ - Add support for more c++ compilers in windows for mumur hash extension
+5.5.0 (Feb 28, 2018)
+ - Add support for .track
 5.4.3 (Jan 7, 2018)
  - Move impressions listener to it's own thread
  - Fix bug in localhost client

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)
@@ -591,6 +590,7 @@ def __init__(self, uwsgi, config=None):
         self._treatment_log = treatment_log
         self._metrics = metrics
 
+
     def get_split_fetcher(self):
         """
         Get the split fetcher implementation for the broker.

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/tasks.py

@@ -80,30 +80,31 @@ def update_splits(split_cache, split_change_fetcher, split_parser):
     If an exception is raised, the process is stopped and it won't try to
     update splits again until enabled_updates is called on the splits cache.
     """
+    added_features = []
+    removed_features = []
     try:
         till = split_cache.get_change_number()
 
         while True:
             response = split_change_fetcher.fetch(till)
 
             if 'till' not in response:
-                return
+                break
 
             if till >= response['till']:
                 _logger.debug("change_number is greater or equal than 'till'")
-                return
+                break
 
             if 'splits' in response and len(response['splits']) > 0:
                 _logger.debug(
                     "Splits field in response. response = %s",
                     response
                 )
-                added_features = []
-                removed_features = []
 
                 for split_change in response['splits']:
                     if Status(split_change['status']) != Status.ACTIVE:
                         split_cache.remove_split(split_change['name'])
+                        removed_features.append(split_change['name'])
                         continue
 
                     parsed_split = split_parser.parse(split_change)
@@ -129,6 +130,9 @@ def update_splits(split_cache, split_change_fetcher, split_parser):
     except:
         _logger.exception('Exception caught updating split definitions')
         split_cache.disable()
+        return [], []
+
+    return added_features, removed_features
 
 
 def report_impressions(impressions_cache, sdk_api, listener=None):

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)