Updated Synching Splits to detect if there is a segment in rules, and sync the segment if it does not exist in storage

Author: chillaq

splitio/sync/segment.py

@@ -145,17 +145,18 @@ def synchronize_segment(self, segment_name, till=None):
         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
+            return True
         else:
             _LOGGER.debug('No changes fetched after %d attempts with CDN bypassed.',
                           without_cdn_attempts)
+            return False
 
     def synchronize_segments(self):
         """
@@ -168,3 +169,18 @@ def synchronize_segments(self):
         for segment_name in segment_names:
             self._worker_pool.submit_work(segment_name)
         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
+        """
+        if self._segment_storage.get(segment_name) != None: 
+            return True
+        else:
+            return False

splitio/sync/split.py

@@ -1,4 +1,5 @@
 """Splits synchronization logic."""
+from ast import Not
 import logging
 import re
 import itertools
@@ -41,6 +42,7 @@ def __init__(self, split_api, split_storage):
         self._backoff = Backoff(
                                 _ON_DEMAND_FETCH_BACKOFF_BASE,
                                 _ON_DEMAND_FETCH_BACKOFF_MAX_WAIT)
+        self.segment_list = []
 
     def _fetch_until(self, fetch_options, till=None):
         """
@@ -69,13 +71,15 @@ def _fetch_until(self, fetch_options, till=None):
                 _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:
+#                    _LOGGER.debug('split details: '+str(split))                        
                     self._split_storage.put(splits.from_raw(split))
                 else:
                     self._split_storage.remove(split['name'])
-
+            self.segment_list = self._split_storage.get_segment_names()
+                
             self._split_storage.set_change_number(split_changes['till'])
             if split_changes['till'] == split_changes['since']:
                 return split_changes['till']
@@ -118,14 +122,14 @@ def synchronize_splits(self, till=None):
         attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if successful_sync:  # succedeed sync
             _LOGGER.debug('Refresh completed in %d attempts.', attempts)
-            return
+            return self.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_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 self.segment_list
         else:
             _LOGGER.debug('No changes fetched after %d attempts with CDN bypassed.',
                           without_cdn_attempts)

splitio/sync/synchronizer.py

@@ -205,6 +205,7 @@ def __init__(self, split_synchronizers, split_tasks):
     def _synchronize_segments(self):
         _LOGGER.debug('Starting segments synchronization')
         return self._split_synchronizers.segment_sync.synchronize_segments()
+        return
 
     def synchronize_segment(self, segment_name, till):
         """
@@ -232,8 +233,19 @@ def synchronize_splits(self, till):
         :rtype: bool
         """
         _LOGGER.debug('Starting splits synchronization')
+        self._split_synchronizers.split_sync.segment_list = []
         try:
             self._split_synchronizers.split_sync.synchronize_splits(till)
+            for segment in self._split_synchronizers.split_sync.segment_list:
+                _LOGGER.debug('Found segment: %s', segment)
+                if not self._split_synchronizers.segment_sync.segment_exist_in_storage(segment):
+                    _LOGGER.debug('Segment does not exist, syncing now.')
+                    success = self.synchronize_segment(segment, -1)    
+                    if not success:
+                        _LOGGER.error('Failed to sync segment.')
+                    else:
+                        _LOGGER.debug('Segment synced.')
+            
             return True
         except APIException:
             _LOGGER.error('Failed syncing splits')
@@ -248,11 +260,6 @@ def sync_all(self):
                 if not self.synchronize_splits(None):
                     attempts -= 1
                     continue
-
-                # Only retrying splits, since segments may trigger too many calls.
-                if not self._synchronize_segments():
-                    _LOGGER.warning('Segments failed to synchronize.')
-
                 # All is good
                 return
             except Exception as exc:  # pylint:disable=broad-except

tests/sync/test_synchronizer.py

@@ -15,6 +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):
@@ -66,40 +67,68 @@ 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_sync_all(self, mocker):
-        split_storage = mocker.Mock(spec=SplitStorage)
-        split_storage.get_change_number.return_value = 123
-        split_storage.get_segment_names.return_value = ['segmentA']
+    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 = mocker.Mock(spec=SegmentStorage)
-        segment_storage.get_change_number.return_value = 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}
+                                                  'key3'], 'removed': [], 'since': -1, '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'
+        
+        inserted_segment = segment_storage.get('segmentA')
+        assert inserted_segment.name == 'segmentA'
+        assert inserted_segment.keys == {'key1', 'key2', 'key3'}
 
+    def test_sync_all(self, mocker):
+        split_storage = mocker.Mock(spec=SplitStorage)
+        split_storage.get_change_number.return_value = 123
+        split_storage.get_segment_names.return_value = ['segmentA']
+        split_api = mocker.Mock()
+        split_api.fetch_splits.return_value = {'splits': self.splits, 'since': 123,
+                                               'till': 123}
+        split_sync = SplitSynchronizer(split_api, split_storage)
+        
+        split_synchronizers = SplitSynchronizers(split_sync, mocker.Mock(), mocker.Mock(),
+                                                 mocker.Mock(), mocker.Mock())
         synchronizer = Synchronizer(split_synchronizers, mocker.Mock(spec=SplitTasks))
+
         synchronizer.sync_all()
 
         inserted_split = split_storage.put.mock_calls[0][1][0]
         assert isinstance(inserted_split, Split)
         assert inserted_split.name == 'some_name'
-
-        inserted_segment = segment_storage.update.mock_calls[0][1]
-        assert inserted_segment[0] == 'segmentA'
-        assert inserted_segment[1] == ['key1', 'key2', 'key3']
-        assert inserted_segment[2] == []
-
+        
     def test_start_periodic_fetching(self, mocker):
         split_task = mocker.Mock(spec=SplitSynchronizationTask)
         segment_task = mocker.Mock(spec=SegmentSynchronizationTask)
@@ -221,7 +250,7 @@ def sync_segments(*_):
 
         synchronizer.sync_all()
         assert counts['splits'] == 1
-        assert counts['segments'] == 1
+#        assert counts['segments'] == 1
 
     def test_sync_all_split_attempts(self, mocker):
         """Test that 3 attempts are done before failing."""
@@ -254,5 +283,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