1- Moved segment in split check logic to synchronizer.py

2- Created test to verify segment_sync is called once from split_sync
3- disable waiting for segment sync workerpool job when synching segments from splits.
4- Other general cleanup.

Author: chillaq

splitio/sync/segment.py

@@ -139,6 +139,8 @@ 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)
@@ -157,21 +159,26 @@ def synchronize_segment(self, segment_name, till=None):
                         without_cdn_attempts)
         return False
 
-    def synchronize_segments(self, segment_names = None):
+    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.
 
         :param segment_names: Optional, array of segment names to update.
-        :type segment_name: [str]
+        :type segment_name: {str}
 
-        :return: True if no error occurs. False otherwise.
+        :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
         """
         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):
@@ -184,6 +191,4 @@ def segment_exist_in_storage(self, segment_name):
         :return: True if segment exist. False otherwise.
         :rtype: bool
         """
-        if self._segment_storage.get(segment_name) != None: 
-            return True
-        return False
+        return self._segment_storage.get(segment_name) != None

splitio/sync/split.py

@@ -42,7 +42,7 @@ def __init__(self, split_api, split_storage):
                                 _ON_DEMAND_FETCH_BACKOFF_BASE,
                                 _ON_DEMAND_FETCH_BACKOFF_MAX_WAIT)
 
-    def _fetch_until(self, fetch_options, till=None, segment_sync=None):
+    def _fetch_until(self, fetch_options, till=None):
         """
         Hit endpoint, update storage and return when since==till.
 
@@ -55,7 +55,7 @@ def _fetch_until(self, fetch_options, till=None, segment_sync=None):
         :return: last change number
         :rtype: int
         """
-        segment_list = []
+        segment_list = set()
         while True:  # Fetch until since==till
             change_number = self._split_storage.get_change_number()
             if change_number is None:
@@ -70,23 +70,19 @@ def _fetch_until(self, fetch_options, till=None, segment_sync=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:
-                    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'])
-                for segment in self._split_storage.get_segment_names():
-                    _LOGGER.debug('Found segment: %s', segment)
-                    if not segment_sync.segment_exist_in_storage(segment):
-                        _LOGGER.debug('Segment %s does not exist, syncing.', segment)
-                        segment_list.append(segment)
-                
             self._split_storage.set_change_number(split_changes['till'])
             if split_changes['till'] == split_changes['since']:
                 return split_changes['till'], segment_list
 
-    def _attempt_split_sync(self, fetch_options, till=None, segment_sync=None):
+    def _attempt_split_sync(self, fetch_options, till=None):
         """
         Hit endpoint, update storage and return True if sync is complete.
 
@@ -103,15 +99,15 @@ def _attempt_split_sync(self, fetch_options, till=None, segment_sync=None):
         remaining_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES
         while True:
             remaining_attempts -= 1
-            change_number, segment_list = self._fetch_until(fetch_options, till, segment_sync)
+            change_number, segment_list = self._fetch_until(fetch_options, till)
             if till is None or till <= change_number:
                 return True, remaining_attempts, change_number, segment_list
             elif remaining_attempts <= 0:
                 return False, remaining_attempts, change_number, segment_list
             how_long = self._backoff.get()
             time.sleep(how_long)
 
-    def synchronize_splits(self, till=None, segment_sync=None):
+    def synchronize_splits(self, till=None):
         """
         Hit endpoint, update storage and return True if sync is complete.
 
@@ -120,13 +116,13 @@ def synchronize_splits(self, till=None, segment_sync=None):
         """
         fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
         successful_sync, remaining_attempts, change_number, segment_list = self._attempt_split_sync(fetch_options,
-                                                                                      till, segment_sync)
+                                                                                      till)
         attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if successful_sync:  # succedeed sync
             _LOGGER.debug('Refresh completed in %d attempts.', attempts)
             return segment_list
         with_cdn_bypass = FetchOptions(True, change_number)  # Set flag for bypassing CDN
-        without_cdn_successful_sync, remaining_attempts, change_number, segment_list = self._attempt_split_sync(with_cdn_bypass, till, segment_sync)
+        without_cdn_successful_sync, remaining_attempts, change_number, segment_list = 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.',

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,13 +233,17 @@ def synchronize_splits(self, till):
         """
         _LOGGER.debug('Starting splits synchronization')
         try:
-            segment_list = self._split_synchronizers.split_sync.synchronize_splits(till, self._split_synchronizers.segment_sync)
-            if len(segment_list) != 0:
-                success = self._split_synchronizers.segment_sync.synchronize_segments(segment_list)
+            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:
+                success = self._split_synchronizers.segment_sync.synchronize_segments(new_segments, True)
                 if not success:
-                    _LOGGER.error('Failed to sync segment.')
+                    _LOGGER.error('Failed to schedule sync one or all segment(s) below.')
+                    _LOGGER.error(','.join(new_segments))
                 else:
-                    _LOGGER.debug('Segment synced.')            
+                    _LOGGER.debug('Segment sync scheduled.')
             return True
         except APIException:
             _LOGGER.error('Failed syncing splits')
@@ -251,14 +255,14 @@ 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
-                
+
                 # 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

@@ -90,7 +90,7 @@ def test_synchronize_splits(self, mocker):
         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_sync = SplitSynchronizer(split_api, split_storage)
         segment_storage = InMemorySegmentStorage()
         segment_api = mocker.Mock()
         segment_api.fetch_segment.return_value = {'name': 'segmentA', 'added': ['key1', 'key2',
@@ -101,14 +101,38 @@ def test_synchronize_splits(self, mocker):
         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'}
+
+        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)
@@ -140,7 +164,7 @@ def test_sync_all(self, mocker):
         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)