Merge pull request #261 from splitio/development

Development

Author: chillaq

CHANGES.txt

@@ -1,3 +1,6 @@
+9.1.3 (July 25, 2022)
+- Fixed synching missed segment(s) after receiving split update
+
 9.1.2 (April 6, 2022)
 - Updated pyyaml dependency for vulnerability CVE-2020-14343.
 

setup.py

@@ -8,7 +8,7 @@
     'flake8',
     'pytest==7.0.1',
     'pytest-mock>=3.5.1',
-    'coverage',
+    'coverage==6.2',
     'pytest-cov',
     'importlib-metadata==4.2',
     'tomli==1.2.3',

splitio/sync/segment.py

@@ -139,32 +139,56 @@ def synchronize_segment(self, segment_name, till=None):
         :param till: ChangeNumber received.
         :type till: int
 
+        :return: True if no error occurs. False otherwise.
+        :rtype: bool
         """
         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 %d attempts.', attempts)
-            return
+            return True
         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 %d attempts.',
                           without_cdn_attempts)
-            return
-        else:
-            _LOGGER.debug('No changes fetched after %d attempts with CDN bypassed.',
-                          without_cdn_attempts)
+            return True
+        _LOGGER.debug('No changes fetched after %d attempts with CDN bypassed.',
+                        without_cdn_attempts)
+        return False
 
-    def synchronize_segments(self):
+    def synchronize_segments(self, segment_names = None, dont_wait = False):
         """
-        Submit all current segments and wait for them to finish, then set the ready flag.
+        Submit all current segments and wait for them to finish depend on dont_wait flag, then set the ready flag.
 
-        :return: True if no error occurs. False otherwise.
+        :param segment_names: Optional, array of segment names to update.
+        :type segment_name: {str}
+
+        :param dont_wait: Optional, instruct the function to not wait for task completion
+        :type segment_name: boolean
+
+        :return: True if no error occurs or dont_wait flag is True. False otherwise.
         :rtype: bool
         """
-        segment_names = self._split_storage.get_segment_names()
+        if segment_names is None:
+            segment_names = self._split_storage.get_segment_names()
+            
         for segment_name in segment_names:
             self._worker_pool.submit_work(segment_name)
+        if (dont_wait):
+            return True
         return not self._worker_pool.wait_for_completion()
+    
+    def segment_exist_in_storage(self, segment_name):
+        """
+        Check if a segment exists in the storage
+
+        :param segment_name: Name of the segment
+        :type segment_name: str
+
+        :return: True if segment exist. False otherwise.
+        :rtype: bool
+        """
+        return self._segment_storage.get(segment_name) != None

splitio/sync/split.py

@@ -55,13 +55,14 @@ def _fetch_until(self, fetch_options, till=None):
         :return: last change number
         :rtype: int
         """
+        segment_list = set()
         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
-                return change_number
+                return change_number, segment_list
 
             try:
                 split_changes = self._api.fetch_splits(change_number, fetch_options)
@@ -72,13 +73,14 @@ def _fetch_until(self, fetch_options, till=None):
 
             for split in split_changes.get('splits', []):
                 if split['status'] == splits.Status.ACTIVE.value:
-                    self._split_storage.put(splits.from_raw(split))
+                    parsed = splits.from_raw(split)
+                    self._split_storage.put(parsed)
+                    segment_list.update(set(parsed.get_segment_names()))
                 else:
                     self._split_storage.remove(split['name'])
-
             self._split_storage.set_change_number(split_changes['till'])
             if split_changes['till'] == split_changes['since']:
-                return split_changes['till']
+                return split_changes['till'], segment_list
 
     def _attempt_split_sync(self, fetch_options, till=None):
         """
@@ -94,14 +96,16 @@ def _attempt_split_sync(self, fetch_options, till=None):
         :rtype: bool, int, int
         """
         self._backoff.reset()
+        final_segment_list = set()
         remaining_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES
         while True:
             remaining_attempts -= 1
-            change_number = self._fetch_until(fetch_options, till)
+            change_number, segment_list = self._fetch_until(fetch_options, till)
+            final_segment_list.update(segment_list)
             if till is None or till <= change_number:
-                return True, remaining_attempts, change_number
+                return True, remaining_attempts, change_number, final_segment_list
             elif remaining_attempts <= 0:
-                return False, remaining_attempts, change_number
+                return False, remaining_attempts, change_number, final_segment_list
             how_long = self._backoff.get()
             time.sleep(how_long)
 
@@ -112,20 +116,23 @@ def synchronize_splits(self, till=None):
         :param till: Passed till from Streaming.
         :type till: int
         """
+        final_segment_list = set()
         fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
-        successful_sync, remaining_attempts, change_number = self._attempt_split_sync(fetch_options,
+        successful_sync, remaining_attempts, change_number, segment_list = self._attempt_split_sync(fetch_options,
                                                                                       till)
+        final_segment_list.update(segment_list)
         attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if successful_sync:  # succedeed sync
             _LOGGER.debug('Refresh completed in %d attempts.', attempts)
-            return
+            return final_segment_list
         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_successful_sync, remaining_attempts, change_number, segment_list = self._attempt_split_sync(with_cdn_bypass, till)
+        final_segment_list.update(segment_list)
         without_cdn_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if without_cdn_successful_sync:
             _LOGGER.debug('Refresh completed bypassing the CDN in %d attempts.',
                           without_cdn_attempts)
-            return
+            return final_segment_list
         else:
             _LOGGER.debug('No changes fetched after %d attempts with CDN bypassed.',
                           without_cdn_attempts)

splitio/sync/synchronizer.py

@@ -221,7 +221,7 @@ def synchronize_segment(self, segment_name, till):
             _LOGGER.error('Failed to sync some segments.')
         return success
 
-    def synchronize_splits(self, till):
+    def synchronize_splits(self, till, sync_segments=True):
         """
         Synchronize all splits.
 
@@ -233,7 +233,18 @@ def synchronize_splits(self, till):
         """
         _LOGGER.debug('Starting splits synchronization')
         try:
-            self._split_synchronizers.split_sync.synchronize_splits(till)
+            new_segments = []
+            for segment in self._split_synchronizers.split_sync.synchronize_splits(till):
+                    if not self._split_synchronizers.segment_sync.segment_exist_in_storage(segment):
+                        new_segments.append(segment)
+            if sync_segments and len(new_segments) != 0:
+                _LOGGER.debug('Synching Segments: %s', ','.join(new_segments))
+                success = self._split_synchronizers.segment_sync.synchronize_segments(new_segments, True)
+                if not success:
+                    _LOGGER.error('Failed to schedule sync one or all segment(s) below.')
+                    _LOGGER.error(','.join(new_segments))
+                else:
+                    _LOGGER.debug('Segment sync scheduled.')
             return True
         except APIException:
             _LOGGER.error('Failed syncing splits')
@@ -245,7 +256,7 @@ def sync_all(self):
         attempts = 3
         while attempts > 0:
             try:
-                if not self.synchronize_splits(None):
+                if not self.synchronize_splits(None, False):
                     attempts -= 1
                     continue
 

splitio/version.py

@@ -1 +1 @@
-__version__ = '9.1.2'
+__version__ = '9.1.3'

tests/sync/test_splits_synchronizer.py

@@ -40,6 +40,7 @@ def change_number_mock():
             return 123
         change_number_mock._calls = 0
         storage.get_change_number.side_effect = change_number_mock
+        storage.get_segment_names.return_value = []
 
         api = mocker.Mock()
         splits = [{
@@ -113,6 +114,7 @@ def test_not_called_on_till(self, mocker):
         def change_number_mock():
             return 2
         storage.get_change_number.side_effect = change_number_mock
+        storage.get_segment_names.return_value = []
 
         def get_changes(*args, **kwargs):
             get_changes.called += 1
@@ -147,6 +149,7 @@ def change_number_mock():
             return 12345 # Return proper cn for CDN Bypass
         change_number_mock._calls = 0
         storage.get_change_number.side_effect = change_number_mock
+        storage.get_segment_names.return_value = []
 
         api = mocker.Mock()
         splits = [{

tests/sync/test_synchronizer.py

@@ -15,7 +15,7 @@
 from splitio.api import APIException
 from splitio.models.splits import Split
 from splitio.models.segments import Segment
-
+from splitio.storage.inmemmory import InMemorySegmentStorage, InMemorySplitStorage
 
 class SynchronizerTests(object):
     def test_sync_all_failed_splits(self, mocker):
@@ -66,9 +66,74 @@ def run(x, y):
         'killed': False,
         'defaultTreatment': 'off',
         'algo': 2,
-        'conditions': []
+        'conditions': [{
+            'conditionType': 'WHITELIST',
+            'matcherGroup':{
+                'combiner': 'AND',
+                'matchers':[{
+                    'matcherType': 'IN_SEGMENT',
+                    'negate': False,
+                    'userDefinedSegmentMatcherData': {
+                        'segmentName': 'segmentA'
+                    }
+                }]
+            },
+            'partitions': [{
+                'size': 100,
+                'treatment': 'on'
+            }]
+        }]
     }]
 
+    def test_synchronize_splits(self, mocker):
+        split_storage = InMemorySplitStorage()
+        split_api = mocker.Mock()
+        split_api.fetch_splits.return_value = {'splits': self.splits, 'since': 123,
+                                               'till': 123}
+        split_sync = SplitSynchronizer(split_api, split_storage)
+        segment_storage = InMemorySegmentStorage()
+        segment_api = mocker.Mock()
+        segment_api.fetch_segment.return_value = {'name': 'segmentA', 'added': ['key1', 'key2',
+                                                  'key3'], 'removed': [], 'since': 123, 'till': 123}
+        segment_sync = SegmentSynchronizer(segment_api, split_storage, segment_storage)
+        split_synchronizers = SplitSynchronizers(split_sync, segment_sync, mocker.Mock(),
+                                                 mocker.Mock(), mocker.Mock())
+        synchronizer = Synchronizer(split_synchronizers, mocker.Mock(spec=SplitTasks))
+
+        synchronizer.synchronize_splits(123)
+
+        inserted_split = split_storage.get('some_name')
+        assert isinstance(inserted_split, Split)
+        assert inserted_split.name == 'some_name'
+
+        if not segment_sync._worker_pool.wait_for_completion():
+            inserted_segment = segment_storage.get('segmentA')
+            assert inserted_segment.name == 'segmentA'
+            assert inserted_segment.keys == {'key1', 'key2', 'key3'}
+
+    def test_synchronize_splits_calling_segment_sync_once(self, mocker):
+        split_storage = InMemorySplitStorage()
+        split_api = mocker.Mock()
+        split_api.fetch_splits.return_value = {'splits': self.splits, 'since': 123,
+                                               'till': 123}
+        split_sync = SplitSynchronizer(split_api, split_storage)
+        counts = {'segments': 0}
+
+        def sync_segments(*_):
+            """Sync Segments."""
+            counts['segments'] += 1
+            return True
+
+        segment_sync = mocker.Mock()
+        segment_sync.synchronize_segments.side_effect = sync_segments
+        segment_sync.segment_exist_in_storage.return_value = False
+        split_synchronizers = SplitSynchronizers(split_sync, segment_sync, mocker.Mock(),
+                                                 mocker.Mock(), mocker.Mock())
+        synchronizer = Synchronizer(split_synchronizers, mocker.Mock(spec=SplitTasks))
+        synchronizer.synchronize_splits(123, True)
+
+        assert counts['segments'] == 1
+
     def test_sync_all(self, mocker):
         split_storage = mocker.Mock(spec=SplitStorage)
         split_storage.get_change_number.return_value = 123
@@ -207,7 +272,7 @@ def test_sync_all_ok(self, mocker):
         def sync_splits(*_):
             """Sync Splits."""
             counts['splits'] += 1
-            return True
+            return []
 
         def sync_segments(*_):
             """Sync Segments."""
@@ -254,5 +319,5 @@ def sync_segments(*_):
         split_tasks = mocker.Mock(spec=SplitTasks)
         synchronizer = Synchronizer(split_synchronizers, split_tasks)
 
-        synchronizer.sync_all()
+        synchronizer._synchronize_segments()
         assert counts['segments'] == 1

tests/tasks/test_split_sync.py

@@ -24,6 +24,7 @@ def change_number_mock():
             return 123
         change_number_mock._calls = 0
         storage.get_change_number.side_effect = change_number_mock
+        storage.get_segment_names.return_value = []
 
         api = mocker.Mock()
         splits = [{