added cdn bypass segments

Author: Matias Melograno

splitio/api/__init__.py

@@ -1,6 +1,10 @@
 """Split API module."""
 
 
+_CACHE_CONTROL = 'Cache-Control'
+_CACHE_CONTROL_NO_CACHE = 'no-cache'
+
+
 class APIException(Exception):
     """Exception to raise when an API call fails."""
 
@@ -76,3 +80,30 @@ def __eq__(self, other):
         if self._change_number != other._change_number:
             return False
         return True
+
+
+def build_fetch(change_number, fetch_options, metadata):
+    """
+    Build fetch with new flags if that is the case.
+
+    :param change_number: Last known timestamp of definition.
+    :type change_number: int
+
+    :param fetch_options: Fetch options for getting definitions.
+    :type fetch_options: splitio.api.FetchOptions
+
+    :param metadata: Metadata Headers.
+    :type metadata: dict
+
+    :return: Objects for fetch
+    :rtype: dict, dict
+    """
+    query = {'since': change_number}
+    extra_headers = 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

splitio/api/segments.py

@@ -3,7 +3,7 @@
 import json
 import logging
 
-from splitio.api import APIException, headers_from_metadata
+from splitio.api import APIException, headers_from_metadata, build_fetch
 from splitio.api.client import HttpClientException
 
 
@@ -29,25 +29,30 @@ def __init__(self, http_client, apikey, sdk_metadata):
         self._apikey = apikey
         self._metadata = headers_from_metadata(sdk_metadata)
 
-    def fetch_segment(self, segment_name, change_number):
+    def fetch_segment(self, segment_name, change_number, fetch_options):
         """
         Fetch splits from backend.
 
         :param segment_name: Name of the segment to fetch changes for.
         :type segment_name: str
-        :param change_number: Last known timestamp of a split modification.
+
+        :param change_number: Last known timestamp of a segment modification.
         :type change_number: int
 
+        :param fetch_options: Fetch options for getting segment definitions.
+        :type fetch_options: splitio.api.FetchOptions
+
         :return: Json representation of a segmentChange response.
         :rtype: dict
         """
         try:
+            query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
             response = self._client.get(
                 'sdk',
                 '/segmentChanges/{segment_name}'.format(segment_name=segment_name),
                 self._apikey,
-                extra_headers=self._metadata,
-                query={'since': change_number}
+                extra_headers=extra_headers,
+                query=query,
             )
 
             if 200 <= response.status_code < 300:

splitio/api/splits.py

@@ -3,17 +3,13 @@
 import logging
 import json
 
-from splitio.api import APIException, headers_from_metadata
+from splitio.api import APIException, headers_from_metadata, build_fetch
 from splitio.api.client import HttpClientException
 
 
 _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."""
 
@@ -32,29 +28,6 @@ def __init__(self, client, apikey, sdk_metadata):
         self._apikey = apikey
         self._metadata = headers_from_metadata(sdk_metadata)
 
-    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.
@@ -69,7 +42,7 @@ def fetch_splits(self, change_number, fetch_options):
         :rtype: dict
         """
         try:
-            query, extra_headers = self._build_fetch(change_number, fetch_options)
+            query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
             response = self._client.get(
                 'sdk',
                 '/splitChanges',

splitio/client/input_validator.py

@@ -4,7 +4,7 @@
 import re
 import math
 
-from splitio.api import APIException
+from splitio.api import APIException, FetchOptions
 from splitio.client.key import Key
 from splitio.engine.evaluator import CONTROL
 
@@ -458,7 +458,7 @@ def validate_apikey_type(segment_api):
     _logger = logging.getLogger('splitio.api.segments')
     try:
         _logger.addFilter(api_messages_filter)  # pylint: disable=protected-access
-        segment_api.fetch_segment('__SOME_INVALID_SEGMENT__', -1)
+        segment_api.fetch_segment('__SOME_INVALID_SEGMENT__', -1, FetchOptions())
     except APIException as exc:
         if exc.status_code == 403:
             _LOGGER.error('factory instantiation: you passed a browser type '

splitio/sync/segment.py

@@ -1,13 +1,20 @@
 import logging
+import time
 
-from splitio.api import APIException
+from splitio.api import APIException, FetchOptions
 from splitio.tasks.util import workerpool
 from splitio.models import segments
+from splitio.util.backoff import Backoff
 
 
 _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 SegmentSynchronizer(object):
     def __init__(self, segment_api, split_storage, segment_storage):
         """
@@ -28,6 +35,9 @@ def __init__(self, segment_api, split_storage, segment_storage):
         self._segment_storage = segment_storage
         self._worker_pool = workerpool.WorkerPool(10, self.synchronize_segment)
         self._worker_pool.start()
+        self._backoff = Backoff(
+                                _ON_DEMAND_FETCH_BACKOFF_BASE,
+                                _ON_DEMAND_FETCH_BACKOFF_MAX_WAIT)
 
     def recreate(self):
         """
@@ -44,27 +54,33 @@ def shutdown(self):
         """
         self._worker_pool.stop()
 
-    def synchronize_segment(self, segment_name, till=None):
+    def _fetch_until(self, segment_name, fetch_options, till=None):
         """
-        Update a segment from queue
+        Hit endpoint, update storage and return when since==till.
 
         :param segment_name: Name of the segment to update.
         :type segment_name: str
 
-        :param till: ChangeNumber received.
+        :param fetch_options Fetch options for getting segment definitions.
+        :type fetch_options splitio.api.FetchOptions
+
+        :param till: Passed till from Streaming.
         :type till: int
 
+        :return: last change number
+        :rtype: int
         """
-        while True:
+        while True:  # Fetch until since==till
             change_number = self._segment_storage.get_change_number(segment_name)
             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
+                return change_number
 
             try:
-                segment_changes = self._api.fetch_segment(segment_name, change_number)
+                segment_changes = self._api.fetch_segment(segment_name, change_number,
+                                                          fetch_options)
             except APIException as exc:
                 _LOGGER.error('Exception raised while fetching segment %s', segment_name)
                 _LOGGER.debug('Exception information: ', exc_info=True)
@@ -82,7 +98,63 @@ def synchronize_segment(self, segment_name, till=None):
                 )
 
             if segment_changes['till'] == segment_changes['since']:
-                return
+                return segment_changes['till']
+
+    def _attempt_segment_sync(self, segment_name, fetch_options, till=None):
+        """
+        Hit endpoint, update storage and return True if sync is complete.
+
+        :param segment_name: Name of the segment to update.
+        :type segment_name: str
+
+        :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:
+            remaining_attempts -= 1
+            change_number = self._fetch_until(segment_name, fetch_options, till)
+            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_segment(self, segment_name, till=None):
+        """
+        Update a segment from queue
+
+        :param segment_name: Name of the segment to update.
+        :type segment_name: str
+
+        :param till: ChangeNumber received.
+        :type till: int
+
+        """
+        fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
+        successful_sync, remaining_attempts, change_number = self._attempt_segment_sync(segment_name, 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_segment_sync(segment_name, 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 synchronize_segments(self):
         """

tests/api/test_segments_api.py

@@ -1,7 +1,7 @@
 """Segment API tests module."""
 
 import pytest
-from splitio.api import segments, client, APIException
+from splitio.api import segments, client, APIException, FetchOptions
 from splitio.client.util import SdkMetadata
 
 
@@ -13,8 +13,8 @@ def test_fetch_segment_changes(self, mocker):
         httpclient = mocker.Mock(spec=client.HttpClient)
         httpclient.get.return_value = client.HttpResponse(200, '{"prop1": "value1"}')
         segment_api = segments.SegmentsAPI(httpclient, 'some_api_key', SdkMetadata('1.0', 'some', '1.2.3.4'))
-        response = segment_api.fetch_segment('some_segment', 123)
 
+        response = segment_api.fetch_segment('some_segment', 123, FetchOptions())
         assert response['prop1'] == 'value1'
         assert httpclient.get.mock_calls == [mocker.call('sdk', '/segmentChanges/some_segment', 'some_api_key', 
                                                          extra_headers={
@@ -24,11 +24,35 @@ def test_fetch_segment_changes(self, mocker):
                                                          },
                                                          query={'since': 123})]
 
+        httpclient.reset_mock()
+        response = segment_api.fetch_segment('some_segment', 123, FetchOptions(True))
+        assert response['prop1'] == 'value1'
+        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',
+                                                             'Cache-Control': 'no-cache'
+                                                         },
+                                                         query={'since': 123})]
+
+        httpclient.reset_mock()
+        response = segment_api.fetch_segment('some_segment', 123, FetchOptions(True, 123))
+        assert response['prop1'] == 'value1'
+        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',
+                                                             '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 = segment_api.fetch_segment('some_segment', 123)
+            response = segment_api.fetch_segment('some_segment', 123, FetchOptions())
             assert exc_info.type == APIException
             assert exc_info.value.message == 'some_message'