Merge pull request #577 from splitio/feature/rule-based-segment

Feature/rule based segment

Author: chillaq

.github/workflows/ci.yml

@@ -16,7 +16,7 @@ concurrency:
 jobs:
   test:
     name: Test
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
     services:
       redis:
         image: redis
@@ -35,6 +35,7 @@ jobs:
 
       - name: Install dependencies
         run: |
+            sudo apt update
             sudo apt-get install -y libkrb5-dev
             pip install -U setuptools pip wheel
             pip install -e .[cpphash,redis,uwsgi]

CHANGES.txt

@@ -1,3 +1,7 @@
+10.3.0 (Jun xx, 2025)
+- Added support for rule-based segments. These segments determine membership at runtime by evaluating their configured rules against the user attributes provided to the SDK.
+- Added support for feature flag prerequisites. This allows customers to define dependency conditions between flags, which are evaluated before any allowlists or targeting rules.
+
 10.2.0 (Jan 17, 2025)
 - Added support for the new impressions tracking toggle available on feature flags, both respecting the setting and including the new field being returned on SplitView type objects. Read more in our docs.
 

setup.py

@@ -8,7 +8,7 @@
     'flake8',
     'pytest==7.0.1',
     'pytest-mock==3.11.1',
-    'coverage',
+    'coverage==7.0.0',
     'pytest-cov==4.1.0',
     'importlib-metadata==6.7',
     'tomli==1.2.3',
@@ -17,7 +17,8 @@
     'pytest-asyncio==0.21.0',
     'aiohttp>=3.8.4',
     'aiofiles>=23.1.0',
-    'requests-kerberos>=0.15.0'
+    'requests-kerberos>=0.15.0',
+    'urllib3==2.0.7'
 ]
 
 INSTALL_REQUIRES = [

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.
@@ -187,7 +207,11 @@ def get(self, server, path, sdk_key, query=None, extra_headers=None):  # pylint:
             self._record_telemetry(response.status_code, get_current_epoch_time_ms() - start)
             return HttpResponse(response.status_code, response.text, response.headers)
 
-        except Exception as exc:  # pylint: disable=broad-except
+        except requests.exceptions.ChunkedEncodingError as exc:
+            _LOGGER.error("IncompleteRead exception detected: %s", exc)
+            return HttpResponse(400, "", {})      
+            
+        except Exception as exc:  # pylint: disable=broad-except            
             raise HttpClientException(_EXC_MSG.format(source='request')) from exc
 
     def post(self, server, path, sdk_key, body, query=None, extra_headers=None):  # pylint: disable=too-many-arguments
@@ -241,8 +265,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
@@ -281,6 +304,10 @@ async def get(self, server, path, apikey, query=None, extra_headers=None):  # py
                 await self._record_telemetry(response.status, get_current_epoch_time_ms() - start)
                 return HttpResponse(response.status, body, response.headers)
 
+        except aiohttp.ClientPayloadError as exc:
+                _LOGGER.error("ContentLengthError exception detected: %s", exc)
+                return HttpResponse(400, "", {})      
+            
         except aiohttp.ClientError as exc:  # pylint: disable=broad-except
             raise HttpClientException(_EXC_MSG.format(source='aiohttp')) from exc
 

splitio/api/commons.py

@@ -57,7 +57,7 @@ def record_telemetry(status_code, elapsed, metric_name, telemetry_runtime_produc
 class FetchOptions(object):
     """Fetch Options object."""
 
-    def __init__(self, cache_control_headers=False, change_number=None, sets=None, spec=SPEC_VERSION):
+    def __init__(self, cache_control_headers=False, change_number=None, rbs_change_number=None, sets=None, spec=SPEC_VERSION):
         """
         Class constructor.
 
@@ -72,6 +72,7 @@ def __init__(self, cache_control_headers=False, change_number=None, sets=None, s
         """
         self._cache_control_headers = cache_control_headers
         self._change_number = change_number
+        self._rbs_change_number = rbs_change_number
         self._sets = sets
         self._spec = spec
 
@@ -85,6 +86,11 @@ def change_number(self):
         """Return change number."""
         return self._change_number
 
+    @property
+    def rbs_change_number(self):
+        """Return change number."""
+        return self._rbs_change_number
+
     @property
     def sets(self):
         """Return sets."""
@@ -103,14 +109,19 @@ def __eq__(self, other):
         if self._change_number != other._change_number:
             return False
 
+        if self._rbs_change_number != other._rbs_change_number:
+            return False
+
         if self._sets != other._sets:
             return False
+        
         if self._spec != other._spec:
             return False
+        
         return True
 
 
-def build_fetch(change_number, fetch_options, metadata):
+def build_fetch(change_number, fetch_options, metadata, rbs_change_number=None):
     """
     Build fetch with new flags if that is the case.
 
@@ -123,11 +134,16 @@ def build_fetch(change_number, fetch_options, metadata):
     :param metadata: Metadata Headers.
     :type metadata: dict
 
+    :param rbs_change_number: Last known timestamp of a rule based segment modification.
+    :type rbs_change_number: int
+
     :return: Objects for fetch
     :rtype: dict, dict
     """
     query = {'s': fetch_options.spec} if fetch_options.spec is not None else {}
     query['since'] = change_number
+    if rbs_change_number is not None:
+        query['rbSince'] = rbs_change_number
     extra_headers = metadata
     if fetch_options is None:
         return query, extra_headers

splitio/api/splits.py

@@ -4,14 +4,18 @@
 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
+from splitio.sync import util
 
 _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,22 +34,66 @@ 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
+        self._old_spec_since = None
+
+    def _check_last_proxy_check_timestamp(self, since):
+        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
+            self._old_spec_since = since
+    
+    def _check_old_spec_since(self, change_number):
+        if self._spec_version == _SPEC_1_1 and self._old_spec_since is not None:
+            since = self._old_spec_since
+            self._old_spec_since = None
+            return since
+        return change_number
+    
+
+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, fetch_options):
+    def fetch_splits(self, change_number, rbs_change_number, fetch_options):
         """
         Fetch feature flags from backend.
 
         :param change_number: Last known timestamp of a split modification.
         :type change_number: int
 
+        :param rbs_change_number: Last known timestamp of a rule based segment modification.
+        :type rbs_change_number: int
+
         :param fetch_options: Fetch options for getting feature flag definitions.
         :type fetch_options: splitio.api.commons.FetchOptions
 
         :return: Json representation of a splitChanges response.
         :rtype: dict
         """
         try:
-            query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
+            self._check_last_proxy_check_timestamp(change_number)
+            change_number = self._check_old_spec_since(change_number)
+
+            if self._spec_version == _SPEC_1_1:
+                fetch_options = FetchOptions(fetch_options.cache_control_headers, fetch_options.change_number,
+                                                                               None, fetch_options.sets, self._spec_version)
+                rbs_change_number = None
+            query, extra_headers = build_fetch(change_number, fetch_options, self._metadata, rbs_change_number)
             response = self._client.get(
                 'sdk',
                 'splitChanges',
@@ -54,19 +102,32 @@ def fetch_splits(self, change_number, fetch_options):
                 query=query,
             )
             if 200 <= response.status_code < 300:
+                if self._spec_version == _SPEC_1_1:
+                    return util.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):
@@ -80,18 +141,17 @@ 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, fetch_options):
+    async def fetch_splits(self, change_number, rbs_change_number, fetch_options):
         """
         Fetch feature flags from backend.
 
         :param change_number: Last known timestamp of a split modification.
         :type change_number: int
+        
+        :param rbs_change_number: Last known timestamp of a rule based segment modification.
+        :type rbs_change_number: int
 
         :param fetch_options: Fetch options for getting feature flag definitions.
         :type fetch_options: splitio.api.commons.FetchOptions
@@ -100,7 +160,14 @@ async def fetch_splits(self, change_number, fetch_options):
         :rtype: dict
         """
         try:
-            query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
+            self._check_last_proxy_check_timestamp(change_number)
+            change_number = self._check_old_spec_since(change_number)
+            if self._spec_version == _SPEC_1_1:
+                fetch_options = FetchOptions(fetch_options.cache_control_headers, fetch_options.change_number,
+                                                                               None, fetch_options.sets, self._spec_version)
+                rbs_change_number = None
+            
+            query, extra_headers = build_fetch(change_number, fetch_options, self._metadata, rbs_change_number)
             response = await self._client.get(
                 'sdk',
                 'splitChanges',
@@ -109,12 +176,26 @@ async def fetch_splits(self, change_number, fetch_options):
                 query=query,
             )
             if 200 <= response.status_code < 300:
+                if self._spec_version == _SPEC_1_1:
+                    return util.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)