Merge pull request #215 from splitio/development

mredolatti · web-flow · commit 64f1084fe658 · 2020-11-20T13:36:56.000-03:00
Development
diff --git a/CHANGES.txt b/CHANGES.txt
@@ -1,3 +1,7 @@
+8.3.1 (Nov 20, 2020)
+ - Fixed error handling when split server fails, so that it doesn't bring streaming down.
+ - Added SDK Metadata headers to split & segments API clients
+
 8.3.0 (Nov 4, 2020)
  - Added local impressions deduping. Defaulting to optimized
  - Added support for the new Split streaming architecture. When enabled (default), the SDK will not poll for updates but instead receive notifications every time there's a change in your environments, allowing to process those much quicker. If disabled or in the event of an issue, the SDK will fallback to the known polling mechanism to provide a seamless experience.
diff --git a/splitio/api/segments.py b/splitio/api/segments.py
@@ -5,7 +5,7 @@
 
 from future.utils import raise_from
 
-from splitio.api import APIException
+from splitio.api import APIException, headers_from_metadata
 from splitio.api.client import HttpClientException
 
 
@@ -15,17 +15,21 @@
 class SegmentsAPI(object):  # pylint: disable=too-few-public-methods
     """Class that uses an httpClient to communicate with the segments API."""
 
-    def __init__(self, http_client, apikey):
+    def __init__(self, http_client, apikey, sdk_metadata):
         """
         Class constructor.
 
         :param client: HTTP Client responsble for issuing calls to the backend.
         :type client: client.HttpClient
         :param apikey: User apikey token.
         :type apikey: string
+        :param sdk_metadata: SDK version & machine name & IP.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
+
         """
         self._client = http_client
         self._apikey = apikey
+        self._metadata = headers_from_metadata(sdk_metadata)
 
     def fetch_segment(self, segment_name, change_number):
         """
@@ -44,7 +48,8 @@ def fetch_segment(self, segment_name, change_number):
                 'sdk',
                 '/segmentChanges/{segment_name}'.format(segment_name=segment_name),
                 self._apikey,
-                {'since': change_number}
+                extra_headers=self._metadata,
+                query={'since': change_number}
             )
 
             if 200 <= response.status_code < 300:
diff --git a/splitio/api/splits.py b/splitio/api/splits.py
@@ -5,7 +5,7 @@
 
 from future.utils import raise_from
 
-from splitio.api import APIException
+from splitio.api import APIException, headers_from_metadata
 from splitio.api.client import HttpClientException
 
 
@@ -15,17 +15,20 @@
 class SplitsAPI(object):  # pylint: disable=too-few-public-methods
     """Class that uses an httpClient to communicate with the splits API."""
 
-    def __init__(self, client, apikey):
+    def __init__(self, client, apikey, sdk_metadata):
         """
         Class constructor.
 
         :param client: HTTP Client responsble for issuing calls to the backend.
         :type client: HttpClient
         :param apikey: User apikey token.
         :type apikey: string
+        :param sdk_metadata: SDK version & machine name & IP.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
         """
         self._client = client
         self._apikey = apikey
+        self._metadata = headers_from_metadata(sdk_metadata)
 
     def fetch_splits(self, change_number):
         """
@@ -42,7 +45,8 @@ def fetch_splits(self, change_number):
                 'sdk',
                 '/splitChanges',
                 self._apikey,
-                {'since': change_number}
+                extra_headers=self._metadata,
+                query={'since': change_number}
             )
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)
diff --git a/splitio/client/factory.py b/splitio/client/factory.py
@@ -261,8 +261,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     sdk_metadata = util.get_metadata(cfg)
     apis = {
         'auth': AuthAPI(http_client, api_key, sdk_metadata),
-        'splits': SplitsAPI(http_client, api_key),
-        'segments': SegmentsAPI(http_client, api_key),
+        'splits': SplitsAPI(http_client, api_key, sdk_metadata),
+        'segments': SegmentsAPI(http_client, api_key, sdk_metadata),
         'impressions': ImpressionsAPI(http_client, api_key, sdk_metadata, cfg['impressionsMode']),
         'events': EventsAPI(http_client, api_key, sdk_metadata),
         'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata)
diff --git a/splitio/push/manager.py b/splitio/push/manager.py
@@ -66,7 +66,6 @@ def update_workers_status(self, enabled):
         """
         self._processor.update_workers_status(enabled)
 
-
     def start(self):
         """Start a new connection if not already running."""
         if self._running:
diff --git a/splitio/sync/synchronizer.py b/splitio/sync/synchronizer.py
@@ -235,28 +235,51 @@ def synchronize_segment(self, segment_name, till):
         :type till: int
         """
         _LOGGER.debug('Synchronizing segment %s', segment_name)
-        return self._split_synchronizers.segment_sync.synchronize_segment(segment_name, till)
+        success = self._split_synchronizers.segment_sync.synchronize_segment(segment_name, till)
+        if not success:
+            _LOGGER.error('Failed to sync some segments.')
+        return success
 
     def synchronize_splits(self, till):
         """
         Synchronize all splits.
 
         :param till: to fetch
         :type till: int
+
+        :returns: whether the synchronization was successful or not.
+        :rtype: bool
         """
         _LOGGER.debug('Starting splits synchronization')
-        return self._split_synchronizers.split_sync.synchronize_splits(till)
+        try:
+            self._split_synchronizers.split_sync.synchronize_splits(till)
+            return True
+        except APIException:
+            _LOGGER.error('Failed syncing splits')
+            _LOGGER.debug('Error: ', exc_info=True)
+            return False
 
     def sync_all(self):
         """Synchronize all split data."""
-        try:
-            self.synchronize_splits(None)
-            if not self._synchronize_segments():
-                _LOGGER.error('Failed syncing segments')
-                raise RuntimeError('Failed syncing segments')
-        except APIException as exc:
-            _LOGGER.error('Failed syncing splits')
-            raise_from(APIException('Failed to sync splits'), exc)
+        attempts = 3
+        while attempts > 0:
+            try:
+                if not self.synchronize_splits(None):
+                    attempts -= 1
+                    continue
+
+                # Only retrying splits, since segments may trigger too many calls.
+                if not self._synchronize_segments():
+                    _LOGGER.warn('Segments failed to synchronize.')
+
+                # All is good
+                return
+            except Exception as exc:  # pylint:disable=broad-except
+                attempts -= 1
+                _LOGGER.error("Exception caught when trying to sync all data: %s", str(exc))
+                _LOGGER.debug('Error: ', exc_info=True)
+
+        _LOGGER.error("Could not correctly synchronize splits and segments after 3 attempts.")
 
     def shutdown(self, blocking):
         """
diff --git a/splitio/tasks/uwsgi_wrappers.py b/splitio/tasks/uwsgi_wrappers.py
@@ -45,10 +45,12 @@ def uwsgi_update_splits(user_config):
     :type user_config: dict
     """
     config = _get_config(user_config)
+    metadata = get_metadata(config)
     seconds = config['featuresRefreshRate']
     split_sync = SplitSynchronizer(
         SplitsAPI(
-            HttpClient(1500, config.get('sdk_url'), config.get('events_url')), config['apikey']
+            HttpClient(1500, config.get('sdk_url'), config.get('events_url')), config['apikey'],
+            metadata
         ),
         UWSGISplitStorage(get_uwsgi()),
     )
@@ -71,9 +73,11 @@ def uwsgi_update_segments(user_config):
     """
     config = _get_config(user_config)
     seconds = config['segmentsRefreshRate']
+    metadata = get_metadata(config)
     segment_sync = SegmentSynchronizer(
         SegmentsAPI(
-            HttpClient(1500, config.get('sdk_url'), config.get('events_url')), config['apikey']
+            HttpClient(1500, config.get('sdk_url'), config.get('events_url')), config['apikey'],
+            metadata
         ),
         UWSGISplitStorage(get_uwsgi()),
         UWSGISegmentStorage(get_uwsgi()),
diff --git a/splitio/version.py b/splitio/version.py
@@ -1 +1 @@
-__version__ = '8.3.0'
+__version__ = '8.3.1'
diff --git a/tests/api/test_segments_api.py b/tests/api/test_segments_api.py
@@ -2,6 +2,7 @@
 
 import pytest
 from splitio.api import segments, client, APIException
+from splitio.client.util import SdkMetadata
 
 
 class SegmentAPITests(object):
@@ -11,11 +12,17 @@ def test_fetch_segment_changes(self, mocker):
         """Test segment changes fetching API call."""
         httpclient = mocker.Mock(spec=client.HttpClient)
         httpclient.get.return_value = client.HttpResponse(200, '{"prop1": "value1"}')
-        segment_api = segments.SegmentsAPI(httpclient, 'some_api_key')
+        segment_api = segments.SegmentsAPI(httpclient, 'some_api_key', SdkMetadata('1.0', 'some', '1.2.3.4'))
         response = segment_api.fetch_segment('some_segment', 123)
 
         assert response['prop1'] == 'value1'
-        assert httpclient.get.mock_calls == [mocker.call('sdk', '/segmentChanges/some_segment', 'some_api_key', {'since': 123})]
+        assert httpclient.get.mock_calls == [mocker.call('sdk', '/segmentChanges/some_segment', 'some_api_key', 
+                                                         extra_headers={
+                                                             'SplitSDKVersion': '1.0', 
+                                                             'SplitSDKMachineIP': '1.2.3.4', 
+                                                             'SplitSDKMachineName': 'some'
+                                                         },
+                                                         query={'since': 123})]
 
         httpclient.reset_mock()
         def raise_exception(*args, **kwargs):
diff --git a/tests/api/test_splits_api.py b/tests/api/test_splits_api.py
@@ -2,6 +2,7 @@
 
 import pytest
 from splitio.api import splits, client, APIException
+from splitio.client.util import SdkMetadata
 
 
 class SplitAPITests(object):
@@ -11,11 +12,17 @@ def test_fetch_split_changes(self, mocker):
         """Test split changes fetching API call."""
         httpclient = mocker.Mock(spec=client.HttpClient)
         httpclient.get.return_value = client.HttpResponse(200, '{"prop1": "value1"}')
-        split_api = splits.SplitsAPI(httpclient, 'some_api_key')
+        split_api = splits.SplitsAPI(httpclient, 'some_api_key', SdkMetadata('1.0', 'some', '1.2.3.4'))
         response = split_api.fetch_splits(123)
 
         assert response['prop1'] == 'value1'
-        assert httpclient.get.mock_calls == [mocker.call('sdk', '/splitChanges', 'some_api_key', {'since': 123})]
+        assert httpclient.get.mock_calls == [mocker.call('sdk', '/splitChanges', 'some_api_key', 
+                                                         extra_headers={
+                                                             'SplitSDKVersion': '1.0', 
+                                                             'SplitSDKMachineIP': '1.2.3.4', 
+                                                             'SplitSDKMachineName': 'some'
+                                                         },
+                                                         query={'since': 123})]
 
         httpclient.reset_mock()
         def raise_exception(*args, **kwargs):
diff --git a/tests/sync/test_manager.py b/tests/sync/test_manager.py
@@ -45,8 +45,7 @@ def run(x):
         synchronizer = Synchronizer(synchronizers, split_tasks)
         manager = Manager(threading.Event(), synchronizer,  mocker.Mock(), False)
 
-        with pytest.raises(APIException):
-            manager.start()
+        manager.start()  # should not throw!
 
     def test_start_streaming_false(self, mocker):
         splits_ready_event = threading.Event()
diff --git a/tests/sync/test_synchronizer.py b/tests/sync/test_synchronizer.py
@@ -33,10 +33,9 @@ def run(x):
                                                  mocker.Mock(), mocker.Mock(), mocker.Mock())
         sychronizer = Synchronizer(split_synchronizers, mocker.Mock(spec=SplitTasks))
 
-        with pytest.raises(APIException):
-            sychronizer.synchronize_splits(None)
-        with pytest.raises(APIException):
-            sychronizer.sync_all()
+        sychronizer.synchronize_splits(None)  # APIExceptions are handled locally and should not be propagated!
+
+        sychronizer.sync_all()  # sync_all should not throw!
 
     def test_sync_all_failed_segments(self, mocker):
         api = mocker.Mock()
@@ -55,8 +54,7 @@ def run(x, y):
                                                  mocker.Mock(), mocker.Mock(), mocker.Mock())
         sychronizer = Synchronizer(split_synchronizers, mocker.Mock(spec=SplitTasks))
 
-        with pytest.raises(RuntimeError):
-            sychronizer.sync_all()
+        sychronizer.sync_all()  # SyncAll should not throw!
         assert not sychronizer._synchronize_segments()
 
     splits = [{
@@ -210,3 +208,61 @@ def stop_mock_2():
         assert len(impression_count_task.stop.mock_calls) == 1
         assert len(event_task.stop.mock_calls) == 1
         assert len(telemetry_task.stop.mock_calls) == 1
+
+    def test_sync_all_ok(self, mocker):
+        """Test that 3 attempts are done before failing."""
+        split_synchronizers = mocker.Mock(spec=SplitSynchronizers)
+        counts = {'splits': 0, 'segments': 0}
+
+        def sync_splits(*_):
+            """Sync Splits."""
+            counts['splits'] += 1
+            return True
+
+        def sync_segments(*_):
+            """Sync Segments."""
+            counts['segments'] += 1
+            return True
+
+        split_synchronizers.split_sync.synchronize_splits.side_effect = sync_splits
+        split_synchronizers.segment_sync.synchronize_segments.side_effect = sync_segments
+        split_tasks = mocker.Mock(spec=SplitTasks)
+        synchronizer = Synchronizer(split_synchronizers, split_tasks)
+
+        synchronizer.sync_all()
+        assert counts['splits'] == 1
+        assert counts['segments'] == 1
+
+    def test_sync_all_split_attempts(self, mocker):
+        """Test that 3 attempts are done before failing."""
+        split_synchronizers = mocker.Mock(spec=SplitSynchronizers)
+        counts = {'splits': 0, 'segments': 0}
+        def sync_splits(*_):
+            """Sync Splits."""
+            counts['splits'] += 1
+            raise Exception('sarasa')
+
+        split_synchronizers.split_sync.synchronize_splits.side_effect = sync_splits
+        split_tasks = mocker.Mock(spec=SplitTasks)
+        synchronizer = Synchronizer(split_synchronizers, split_tasks)
+
+        synchronizer.sync_all()
+        assert counts['splits'] == 3
+
+    def test_sync_all_segment_attempts(self, mocker):
+        """Test that segments don't trigger retries."""
+        split_synchronizers = mocker.Mock(spec=SplitSynchronizers)
+        counts = {'splits': 0, 'segments': 0}
+
+        def sync_segments(*_):
+            """Sync Splits."""
+
+            counts['segments'] += 1
+            return False
+
+        split_synchronizers.segment_sync.synchronize_segments.side_effect = sync_segments
+        split_tasks = mocker.Mock(spec=SplitTasks)
+        synchronizer = Synchronizer(split_synchronizers, split_tasks)
+
+        synchronizer.sync_all()
+        assert counts['segments'] == 1

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-__version__ = '8.3.0'`
	`1`	`+__version__ = '8.3.1'`