added logic for bypassing cdn

Author: Matias Melograno

CHANGES.txt

@@ -1,3 +1,7 @@
+9.1.0 (Jul XX, 2021)
+- Added Cache-Control header for on-demand requests to sdk-server.
+- Updated the synchronization flow to be more reliable in the event of an edge case generating delay in cache purge propagation, keeping the SDK cache properly synced.
+
 9.0.0 (May 3, 2021)
  - BREAKING CHANGE: Removed splitSdkMachineIp and splitSdkMachineName configs.
  - BREAKING CHANGE: Deprecated `redisCharset` config.

splitio/api/__init__.py

@@ -41,3 +41,38 @@ def headers_from_metadata(sdk_metadata, client_key=None):
         metadata['SplitSDKClientKey'] = client_key
 
     return metadata
+
+
+class FetchOptions(object):
+    """Fetch Options object."""
+
+    def __init__(self, cache_control_headers=False, change_number=None):
+        """
+        Class constructor.
+
+        :param cache_control_headers: Flag for Cache-Control header
+        :type cache_control_headers: bool
+
+        :param change_number: ChangeNumber to use for bypassing CDN in request.
+        :type change_number: int
+        """
+        self._cache_control_headers = cache_control_headers
+        self._change_number = change_number
+
+    @property
+    def cache_control_headers(self):
+        """Return cache control headers."""
+        return self._cache_control_headers
+
+    @property
+    def change_number(self):
+        """Return change number."""
+        return self._change_number
+
+    def __eq__(self, other):
+        """Match between other options."""
+        if self._cache_control_headers != other._cache_control_headers:
+            return False
+        if self._change_number != other._change_number:
+            return False
+        return True

splitio/api/splits.py

@@ -10,6 +10,10 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+_CACHE_CONTROL = 'Cache-Control'
+_CACHE_CONTROL_NO_CACHE = 'no-cache'
+
+
 class SplitsAPI(object):  # pylint: disable=too-few-public-methods
     """Class that uses an httpClient to communicate with the splits API."""
 
@@ -28,23 +32,50 @@ def __init__(self, client, apikey, sdk_metadata):
         self._apikey = apikey
         self._metadata = headers_from_metadata(sdk_metadata)
 
-    def fetch_splits(self, change_number):
+    def _build_fetch(self, change_number, fetch_options):
+        """
+        Build fetch with new flags if that is the case.
+
+        :param change_number: Last known timestamp of a split modification.
+        :type change_number: int
+
+        :param fetch_options: Fetch options for getting split definitions.
+        :type fetch_options: splitio.api.FetchOptions
+
+        :return: Objects for fetch
+        :rtype: dict, dict
+        """
+        query = {'since': change_number}
+        extra_headers = self._metadata
+        if fetch_options is None:
+            return query, extra_headers
+        if fetch_options.cache_control_headers:
+            extra_headers[_CACHE_CONTROL] = _CACHE_CONTROL_NO_CACHE
+        if fetch_options.change_number is not None:
+            query['till'] = fetch_options.change_number
+        return query, extra_headers
+
+    def fetch_splits(self, change_number, fetch_options):
         """
         Fetch splits from backend.
 
-        :param changeNumber: Last known timestamp of a split modification.
-        :type changeNumber: int
+        :param change_number: Last known timestamp of a split modification.
+        :type change_number: int
+
+        :param fetch_options: Fetch options for getting split definitions.
+        :type fetch_options: splitio.api.FetchOptions
 
         :return: Json representation of a splitChanges response.
         :rtype: dict
         """
         try:
+            query, extra_headers = self._build_fetch(change_number, fetch_options)
             response = self._client.get(
                 'sdk',
                 '/splitChanges',
                 self._apikey,
-                extra_headers=self._metadata,
-                query={'since': change_number}
+                extra_headers=extra_headers,
+                query=query,
             )
             if 200 <= response.status_code < 300:
                 return json.loads(response.body)

splitio/sync/split.py

@@ -3,9 +3,11 @@
 import re
 import itertools
 import yaml
+import time
 
-from splitio.api import APIException
+from splitio.api import APIException, FetchOptions
 from splitio.models import splits
+from splitio.util.backoff import Backoff
 
 
 _LEGACY_COMMENT_LINE_RE = re.compile(r'^#.*$')
@@ -15,6 +17,11 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+_ON_DEMAND_FETCH_BACKOFF_BASE = 10  # backoff base starting at 10 seconds
+_ON_DEMAND_FETCH_BACKOFF_MAX_WAIT = 60  # don't sleep for more than 1 minute
+_ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES = 10
+
+
 class SplitSynchronizer(object):
     """Split changes synchronizer."""
 
@@ -30,39 +37,83 @@ def __init__(self, split_api, split_storage):
         """
         self._api = split_api
         self._split_storage = split_storage
+        self._backoff = Backoff(
+                                _ON_DEMAND_FETCH_BACKOFF_BASE,
+                                _ON_DEMAND_FETCH_BACKOFF_MAX_WAIT)
 
-    def synchronize_splits(self, till=None):
+    def attempt_split_sync(self, fetch_options, till=None):
         """
         Hit endpoint, update storage and return True if sync is complete.
 
+        :param fetch_options Fetch options for getting split definitions.
+        :type fetch_options splitio.api.FetchOptions
+
         :param till: Passed till from Streaming.
         :type till: int
+
+        :return: Flags to check if it should perform bypass or operation ended
+        :rtype: bool, int, int
         """
+        self._backoff.reset()
+        remaining_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES
         while True:
-            change_number = self._split_storage.get_change_number()
-            if change_number is None:
-                change_number = -1
-            if till is not None and till < change_number:
-                # the passed till is less than change_number, no need to perform updates
-                return
-
-            try:
-                split_changes = self._api.fetch_splits(change_number)
-            except APIException as exc:
-                _LOGGER.error('Exception raised while fetching splits')
-                _LOGGER.debug('Exception information: ', exc_info=True)
-                raise exc
-
-            for split in split_changes.get('splits', []):
-                if split['status'] == splits.Status.ACTIVE.value:
-                    self._split_storage.put(splits.from_raw(split))
-                else:
-                    self._split_storage.remove(split['name'])
-
-            self._split_storage.set_change_number(split_changes['till'])
-            if split_changes['till'] == split_changes['since'] \
-               and (till is None or split_changes['till'] >= till):
-                return
+            remaining_attempts -= 1
+            while True:  # Fetch until since==till
+                change_number = self._split_storage.get_change_number()
+                if change_number is None:
+                    change_number = -1
+                if till is not None and till < change_number:
+                    # the passed till is less than change_number, no need to perform updates
+                    break
+
+                try:
+                    split_changes = self._api.fetch_splits(change_number, fetch_options)
+                except APIException as exc:
+                    _LOGGER.error('Exception raised while fetching splits')
+                    _LOGGER.debug('Exception information: ', exc_info=True)
+                    raise exc
+
+                for split in split_changes.get('splits', []):
+                    if split['status'] == splits.Status.ACTIVE.value:
+                        self._split_storage.put(splits.from_raw(split))
+                    else:
+                        self._split_storage.remove(split['name'])
+
+                self._split_storage.set_change_number(split_changes['till'])
+                if split_changes['till'] == split_changes['since']:
+                    break
+
+            if till is None or till <= change_number:
+                return True, remaining_attempts, change_number
+            elif remaining_attempts <= 0:
+                return False, remaining_attempts, change_number
+            how_long = self._backoff.get()
+            time.sleep(how_long)
+
+    def synchronize_splits(self, till=None):
+        """
+        Hit endpoint, update storage and return True if sync is complete.
+
+        :param till: Passed till from Streaming.
+        :type till: int
+        """
+        fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
+        successful_sync, remaining_attempts, change_number = self.attempt_split_sync(fetch_options,
+                                                                                     till)
+        attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
+        if successful_sync:  # succedeed sync
+            _LOGGER.debug('Refresh completed in %s attempts.', attempts)
+            return
+        with_cdn_bypass = FetchOptions(True, change_number)  # Set flag for bypassing CDN
+        without_cdn_successful_sync, remaining_attempts, change_number = self.attempt_split_sync(with_cdn_bypass, till)
+        without_cdn_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
+        if without_cdn_successful_sync:
+            _LOGGER.debug('Refresh completed bypassing the CDN in %s attempts.',
+                          without_cdn_attempts)
+            return
+        else:
+            _LOGGER.debug('No changes fetched after %s attempts with CDN bypassed.',
+                          without_cdn_attempts)
 
     def kill_split(self, split_name, default_treatment, change_number):
         """

splitio/util/backoff.py

@@ -6,14 +6,18 @@ class Backoff(object):
 
     MAX_ALLOWED_WAIT = 30 * 60  # half an hour
 
-    def __init__(self, base=1):
+    def __init__(self, base=1, max_allowed=MAX_ALLOWED_WAIT):
         """
         Class constructor.
 
         :param base: basic unit to be multiplied on each iteration (seconds)
         :param base: float
+
+        :param max_allowed: max seconds to wait
+        :param max_allowed: int
         """
         self._base = base
+        self._max_allowed = max_allowed
         self._attempt = 0
 
     def get(self):
@@ -23,7 +27,7 @@ def get(self):
         :returns: time to wait until next retry.
         :rtype: float
         """
-        to_return = min(self._base * (2 ** self._attempt), self.MAX_ALLOWED_WAIT)
+        to_return = min(self._base * (2 ** self._attempt), self._max_allowed)
         self._attempt += 1
         return to_return
 

splitio/version.py

@@ -1 +1 @@
-__version__ = '9.0.0'
+__version__ = '9.1.0-rc1'

tests/api/test_splits_api.py

@@ -1,7 +1,7 @@
 """Split API tests module."""
 
 import pytest
-from splitio.api import splits, client, APIException
+from splitio.api import splits, client, APIException, FetchOptions
 from splitio.client.util import SdkMetadata
 
 
@@ -13,22 +13,46 @@ def test_fetch_split_changes(self, mocker):
         httpclient = mocker.Mock(spec=client.HttpClient)
         httpclient.get.return_value = client.HttpResponse(200, '{"prop1": "value1"}')
         split_api = splits.SplitsAPI(httpclient, 'some_api_key', SdkMetadata('1.0', 'some', '1.2.3.4'))
-        response = split_api.fetch_splits(123)
 
+        response = split_api.fetch_splits(123, FetchOptions())
         assert response['prop1'] == 'value1'
-        assert httpclient.get.mock_calls == [mocker.call('sdk', '/splitChanges', 'some_api_key', 
+        assert httpclient.get.mock_calls == [mocker.call('sdk', '/splitChanges', 'some_api_key',
                                                          extra_headers={
-                                                             'SplitSDKVersion': '1.0', 
-                                                             'SplitSDKMachineIP': '1.2.3.4', 
+                                                             'SplitSDKVersion': '1.0',
+                                                             'SplitSDKMachineIP': '1.2.3.4',
                                                              'SplitSDKMachineName': 'some'
                                                          },
                                                          query={'since': 123})]
 
+        httpclient.reset_mock()
+        response = split_api.fetch_splits(123, FetchOptions(True))
+        assert response['prop1'] == 'value1'
+        assert httpclient.get.mock_calls == [mocker.call('sdk', '/splitChanges', 'some_api_key',
+                                                         extra_headers={
+                                                             'SplitSDKVersion': '1.0',
+                                                             'SplitSDKMachineIP': '1.2.3.4',
+                                                             'SplitSDKMachineName': 'some',
+                                                             'Cache-Control': 'no-cache'
+                                                         },
+                                                         query={'since': 123})]
+
+        httpclient.reset_mock()
+        response = split_api.fetch_splits(123, FetchOptions(True, 123))
+        assert response['prop1'] == 'value1'
+        assert httpclient.get.mock_calls == [mocker.call('sdk', '/splitChanges', 'some_api_key',
+                                                         extra_headers={
+                                                             'SplitSDKVersion': '1.0',
+                                                             'SplitSDKMachineIP': '1.2.3.4',
+                                                             'SplitSDKMachineName': 'some',
+                                                             'Cache-Control': 'no-cache'
+                                                         },
+                                                         query={'since': 123, 'till': 123})]
+
         httpclient.reset_mock()
         def raise_exception(*args, **kwargs):
             raise client.HttpClientException('some_message')
         httpclient.get.side_effect = raise_exception
         with pytest.raises(APIException) as exc_info:
-            response = split_api.fetch_splits(123)
+            response = split_api.fetch_splits(123, FetchOptions())
             assert exc_info.type == APIException
             assert exc_info.value.message == 'some_message'