Merge pull request #564 from splitio/rbs-old-spec-fetcher

Added support for old spec in fetcher

Author: chillaq

splitio/api/client.py

@@ -92,6 +92,25 @@ def proxy_headers(self, proxy):
 class HttpClientBase(object, metaclass=abc.ABCMeta):
     """HttpClient wrapper template."""
 
+    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, telemetry_url=None):
+        """
+        Class constructor.
+
+        :param timeout: How many milliseconds to wait until the server responds.
+        :type timeout: int
+        :param sdk_url: Optional alternative sdk URL.
+        :type sdk_url: str
+        :param events_url: Optional alternative events URL.
+        :type events_url: str
+        :param auth_url: Optional alternative auth URL.
+        :type auth_url: str
+        :param telemetry_url: Optional alternative telemetry URL.
+        :type telemetry_url: str
+        """
+        _LOGGER.debug("Initializing httpclient")
+        self._timeout = timeout/1000 if timeout else None # Convert ms to seconds.
+        self._urls = _construct_urls(sdk_url, events_url, auth_url, telemetry_url)
+
     @abc.abstractmethod
     def get(self, server, path, apikey):
         """http get request"""
@@ -113,6 +132,9 @@ def set_telemetry_data(self, metric_name, telemetry_runtime_producer):
         self._telemetry_runtime_producer = telemetry_runtime_producer
         self._metric_name = metric_name
 
+    def is_sdk_endpoint_overridden(self):
+        return self._urls['sdk'] == SDK_URL
+
     def _get_headers(self, extra_headers, sdk_key):
         headers = _build_basic_headers(sdk_key)
         if extra_headers is not None:
@@ -154,10 +176,8 @@ def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, t
         :param telemetry_url: Optional alternative telemetry URL.
         :type telemetry_url: str
         """
-        _LOGGER.debug("Initializing httpclient")
-        self._timeout = timeout/1000 if timeout else None # Convert ms to seconds.
-        self._urls = _construct_urls(sdk_url, events_url, auth_url, telemetry_url)
-
+        HttpClientBase.__init__(self, timeout, sdk_url, events_url, auth_url, telemetry_url)
+        
     def get(self, server, path, sdk_key, query=None, extra_headers=None):  # pylint: disable=too-many-arguments
         """
         Issue a get request.
@@ -241,8 +261,7 @@ def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, t
         :param telemetry_url: Optional alternative telemetry URL.
         :type telemetry_url: str
         """
-        self._timeout = timeout/1000 if timeout else None  # Convert ms to seconds.
-        self._urls = _construct_urls(sdk_url, events_url, auth_url, telemetry_url)
+        HttpClientBase.__init__(self, timeout, sdk_url, events_url, auth_url, telemetry_url)
         self._session = aiohttp.ClientSession()
 
     async def get(self, server, path, apikey, query=None, extra_headers=None):  # pylint: disable=too-many-arguments

splitio/api/splits.py

@@ -4,14 +4,17 @@
 import json
 
 from splitio.api import APIException, headers_from_metadata
-from splitio.api.commons import build_fetch
+from splitio.api.commons import build_fetch, FetchOptions
 from splitio.api.client import HttpClientException
 from splitio.models.telemetry import HTTPExceptionsAndLatencies
+from splitio.util.time import utctime_ms
+from splitio.spec import SPEC_VERSION
 
 _LOGGER = logging.getLogger(__name__)
+_SPEC_1_1 = "1.1"
+_PROXY_CHECK_INTERVAL_MILLISECONDS_SS =  24 * 60 * 60 * 1000
 
-
-class SplitsAPI(object):  # pylint: disable=too-few-public-methods
+class SplitsAPIBase(object):  # pylint: disable=too-few-public-methods
     """Class that uses an httpClient to communicate with the splits API."""
 
     def __init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer):
@@ -30,6 +33,35 @@ def __init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer):
         self._metadata = headers_from_metadata(sdk_metadata)
         self._telemetry_runtime_producer = telemetry_runtime_producer
         self._client.set_telemetry_data(HTTPExceptionsAndLatencies.SPLIT, self._telemetry_runtime_producer)
+        self._spec_version = SPEC_VERSION
+        self._last_proxy_check_timestamp = 0
+        self.clear_storage = False
+
+    def _convert_to_new_spec(self, body):
+        return {"ff": {"d": body["splits"], "s": body["since"], "t": body["till"]}, 
+                "rbs": {"d": [], "s": -1, "t": -1}}
+
+    def _check_last_proxy_check_timestamp(self):
+        if self._spec_version == _SPEC_1_1 and ((utctime_ms() - self._last_proxy_check_timestamp) >= _PROXY_CHECK_INTERVAL_MILLISECONDS_SS):
+            _LOGGER.info("Switching to new Feature flag spec (%s) and fetching.", SPEC_VERSION);
+            self._spec_version = SPEC_VERSION
+    
+
+class SplitsAPI(SplitsAPIBase):  # pylint: disable=too-few-public-methods
+    """Class that uses an httpClient to communicate with the splits API."""
+
+    def __init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer):
+        """
+        Class constructor.
+
+        :param client: HTTP Client responsble for issuing calls to the backend.
+        :type client: HttpClient
+        :param sdk_key: User sdk_key token.
+        :type sdk_key: string
+        :param sdk_metadata: SDK version & machine name & IP.
+        :type sdk_metadata: splitio.client.util.SdkMetadata
+        """
+        SplitsAPIBase.__init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer)
 
     def fetch_splits(self, change_number, rbs_change_number, fetch_options):
         """
@@ -48,6 +80,7 @@ def fetch_splits(self, change_number, rbs_change_number, fetch_options):
         :rtype: dict
         """
         try:
+            self._check_last_proxy_check_timestamp()
             query, extra_headers = build_fetch(change_number, fetch_options, self._metadata, rbs_change_number)
             response = self._client.get(
                 'sdk',
@@ -57,19 +90,32 @@ def fetch_splits(self, change_number, rbs_change_number, fetch_options):
                 query=query,
             )
             if 200 <= response.status_code < 300:
+                if self._spec_version == _SPEC_1_1:
+                    return self._convert_to_new_spec(json.loads(response.body))
+                
+                self.clear_storage = self._last_proxy_check_timestamp != 0
+                self._last_proxy_check_timestamp = 0
                 return json.loads(response.body)
 
             else:
                 if response.status_code == 414:
                     _LOGGER.error('Error fetching feature flags; the amount of flag sets provided are too big, causing uri length error.')
+                    
+                if self._client.is_sdk_endpoint_overridden() and response.status_code == 400 and self._spec_version == SPEC_VERSION:
+                    _LOGGER.warning('Detected proxy response error, changing spec version from %s to %s and re-fetching.', self._spec_version, _SPEC_1_1)
+                    self._spec_version = _SPEC_1_1
+                    self._last_proxy_check_timestamp = utctime_ms()                    
+                    return self.fetch_splits(change_number, None, FetchOptions(fetch_options.cache_control_headers, fetch_options.change_number,
+                                                                               None, fetch_options.sets, self._spec_version))
+                    
                 raise APIException(response.body, response.status_code)
+            
         except HttpClientException as exc:
             _LOGGER.error('Error fetching feature flags because an exception was raised by the HTTPClient')
             _LOGGER.debug('Error: ', exc_info=True)
             raise APIException('Feature flags not fetched correctly.') from exc
 
-
-class SplitsAPIAsync(object):  # pylint: disable=too-few-public-methods
+class SplitsAPIAsync(SplitsAPIBase):  # pylint: disable=too-few-public-methods
     """Class that uses an httpClient to communicate with the splits API."""
 
     def __init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer):
@@ -83,11 +129,7 @@ def __init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer):
         :param sdk_metadata: SDK version & machine name & IP.
         :type sdk_metadata: splitio.client.util.SdkMetadata
         """
-        self._client = client
-        self._sdk_key = sdk_key
-        self._metadata = headers_from_metadata(sdk_metadata)
-        self._telemetry_runtime_producer = telemetry_runtime_producer
-        self._client.set_telemetry_data(HTTPExceptionsAndLatencies.SPLIT, self._telemetry_runtime_producer)
+        SplitsAPIBase.__init__(self, client, sdk_key, sdk_metadata, telemetry_runtime_producer)
 
     async def fetch_splits(self, change_number, rbs_change_number, fetch_options):
         """
@@ -106,6 +148,7 @@ async def fetch_splits(self, change_number, rbs_change_number, fetch_options):
         :rtype: dict
         """
         try:
+            self._check_last_proxy_check_timestamp()            
             query, extra_headers = build_fetch(change_number, fetch_options, self._metadata, rbs_change_number)
             response = await self._client.get(
                 'sdk',
@@ -115,12 +158,26 @@ async def fetch_splits(self, change_number, rbs_change_number, fetch_options):
                 query=query,
             )
             if 200 <= response.status_code < 300:
+                if self._spec_version == _SPEC_1_1:
+                    return self._convert_to_new_spec(json.loads(response.body))
+                
+                self.clear_storage = self._last_proxy_check_timestamp != 0
+                self._last_proxy_check_timestamp = 0
                 return json.loads(response.body)
 
             else:
                 if response.status_code == 414:
                     _LOGGER.error('Error fetching feature flags; the amount of flag sets provided are too big, causing uri length error.')
+                    
+                if self._client.is_sdk_endpoint_overridden() and response.status_code == 400 and self._spec_version == SPEC_VERSION:
+                    _LOGGER.warning('Detected proxy response error, changing spec version from %s to %s and re-fetching.', self._spec_version, _SPEC_1_1)
+                    self._spec_version = _SPEC_1_1
+                    self._last_proxy_check_timestamp = utctime_ms()                    
+                    return await self.fetch_splits(change_number, None, FetchOptions(fetch_options.cache_control_headers, fetch_options.change_number,
+                                                                               None, fetch_options.sets, self._spec_version))
+
                 raise APIException(response.body, response.status_code)
+            
         except HttpClientException as exc:
             _LOGGER.error('Error fetching feature flags because an exception was raised by the HTTPClient')
             _LOGGER.debug('Error: ', exc_info=True)

splitio/models/rule_based_segments.py

@@ -5,6 +5,7 @@
 from splitio.models import MatcherNotFoundException
 from splitio.models.splits import _DEFAULT_CONDITIONS_TEMPLATE
 from splitio.models.grammar import condition
+from splitio.models.splits import Status
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -31,9 +32,12 @@ def __init__(self, name, traffic_type_name, change_number, status, conditions, e
         self._name = name
         self._traffic_type_name = traffic_type_name
         self._change_number = change_number
-        self._status = status
         self._conditions = conditions
         self._excluded = excluded
+        try:
+            self._status = Status(status)
+        except ValueError:
+            self._status = Status.ARCHIVED
 
     @property
     def name(self):
@@ -71,7 +75,7 @@ def to_json(self):
             'changeNumber': self.change_number,
             'trafficTypeName': self.traffic_type_name,
             'name': self.name,
-            'status': self.status,
+            'status': self.status.value,
             'conditions': [c.to_json() for c in self.conditions],
             'excluded': self.excluded.to_json()
         }

splitio/storage/inmemmory.py

@@ -116,6 +116,14 @@ def __init__(self):
         self._rule_based_segments = {}
         self._change_number = -1
 
+    def clear(self):
+        """
+        Clear storage
+        """
+        with self._lock:
+            self._rule_based_segments = {}
+            self._change_number = -1
+
     def get(self, segment_name):
         """
         Retrieve a rule based segment.
@@ -231,6 +239,14 @@ def __init__(self):
         self._rule_based_segments = {}
         self._change_number = -1
 
+    async def clear(self):
+        """
+        Clear storage
+        """
+        with self._lock:
+            self._rule_based_segments = {}
+            self._change_number = -1
+
     async def get(self, segment_name):
         """
         Retrieve a rule based segment.
@@ -466,6 +482,16 @@ def __init__(self, flag_sets=[]):
         self.flag_set = FlagSets(flag_sets)
         self.flag_set_filter = FlagSetsFilter(flag_sets)
 
+    def clear(self):
+        """
+        Clear storage
+        """
+        with self._lock:
+            self._feature_flags = {}
+            self._change_number = -1
+            self._traffic_types = Counter()
+            self.flag_set = FlagSets(self.flag_set_filter.flag_sets)
+        
     def get(self, feature_flag_name):
         """
         Retrieve a feature flag.
@@ -672,6 +698,16 @@ def __init__(self, flag_sets=[]):
         self.flag_set = FlagSets(flag_sets)
         self.flag_set_filter = FlagSetsFilter(flag_sets)
 
+    async def clear(self):
+        """
+        Clear storage
+        """
+        with self._lock:
+            self._feature_flags = {}
+            self._change_number = -1
+            self._traffic_types = Counter()
+            self.flag_set = FlagSets(self.flag_set_filter.flag_sets)
+
     async def get(self, feature_flag_name):
         """
         Retrieve a feature flag.

splitio/sync/split.py

@@ -135,10 +135,10 @@ def _fetch_until(self, fetch_options, till=None, rbs_till=None):
                 raise exc
 
             fetched_rule_based_segments = [(rule_based_segments.from_raw(rule_based_segment)) for rule_based_segment in feature_flag_changes.get('rbs').get('d', [])]
-            rbs_segment_list = update_rule_based_segment_storage(self._rule_based_segment_storage, fetched_rule_based_segments, feature_flag_changes.get('rbs')['t'])
+            rbs_segment_list = update_rule_based_segment_storage(self._rule_based_segment_storage, fetched_rule_based_segments, feature_flag_changes.get('rbs')['t'], self._api.clear_storage)
 
             fetched_feature_flags = [(splits.from_raw(feature_flag)) for feature_flag in feature_flag_changes.get('ff').get('d', [])]
-            segment_list = update_feature_flag_storage(self._feature_flag_storage, fetched_feature_flags, feature_flag_changes.get('ff')['t'])
+            segment_list = update_feature_flag_storage(self._feature_flag_storage, fetched_feature_flags, feature_flag_changes.get('ff')['t'], self._api.clear_storage)
             segment_list.update(rbs_segment_list)
 
             if feature_flag_changes.get('ff')['t'] == feature_flag_changes.get('ff')['s'] and feature_flag_changes.get('rbs')['t'] == feature_flag_changes.get('rbs')['s']:
@@ -294,10 +294,10 @@ async def _fetch_until(self, fetch_options, till=None, rbs_till=None):
                 raise exc
 
             fetched_rule_based_segments = [(rule_based_segments.from_raw(rule_based_segment)) for rule_based_segment in feature_flag_changes.get('rbs').get('d', [])]
-            rbs_segment_list = await update_rule_based_segment_storage_async(self._rule_based_segment_storage, fetched_rule_based_segments, feature_flag_changes.get('rbs')['t'])
+            rbs_segment_list = await update_rule_based_segment_storage_async(self._rule_based_segment_storage, fetched_rule_based_segments, feature_flag_changes.get('rbs')['t'], self._api.clear_storage)
 
             fetched_feature_flags = [(splits.from_raw(feature_flag)) for feature_flag in feature_flag_changes.get('ff').get('d', [])]
-            segment_list = await update_feature_flag_storage_async(self._feature_flag_storage, fetched_feature_flags, feature_flag_changes.get('ff')['t'])
+            segment_list = await update_feature_flag_storage_async(self._feature_flag_storage, fetched_feature_flags, feature_flag_changes.get('ff')['t'], self._api.clear_storage)
             segment_list.update(rbs_segment_list)
 
             if feature_flag_changes.get('ff')['t'] == feature_flag_changes.get('ff')['s'] and feature_flag_changes.get('rbs')['t'] == feature_flag_changes.get('rbs')['s']:
@@ -541,7 +541,7 @@ def _sanitize_rb_segment_elements(self, parsed_rb_segments):
                 _LOGGER.warning("A rule based segment in json file does not have (Name) or property is empty, skipping.")
                 continue
             for element in [('trafficTypeName', 'user', None, None, None, None),
-                            ('status', 'ACTIVE', None, None, ['ACTIVE', 'ARCHIVED'], None),
+                            ('status', splits.Status.ACTIVE, None, None, [splits.Status.ACTIVE, splits.Status.ARCHIVED], None),
                             ('changeNumber', 0, 0, None, None, None)]:
                 rb_segment = util._sanitize_object_element(rb_segment, 'rule based segment', element[0], element[1], lower_value=element[2], upper_value=element[3], in_list=element[4], not_in_list=element[5])
             rb_segment = self._sanitize_condition(rb_segment)