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

@@ -158,14 +158,19 @@ def synchronize_segment(self, segment_name, till=None):
                           without_cdn_attempts)
             return False
 
-    def synchronize_segments(self):
+    def synchronize_segments(self, segment_names = None):
         """
         Submit all current segments and wait for them to finish, then set the ready flag.
 
+        :param segment_names: Optional, array of segment names to update.
+        :type segment_name: [str]
+
         :return: True if no error occurs. 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)
         return not self._worker_pool.wait_for_completion()

splitio/sync/split.py

@@ -3,6 +3,8 @@
 import logging
 import re
 import itertools
+from numpy import append
+from pyparsing import Each
 import yaml
 import time
 
@@ -42,9 +44,8 @@ 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):
+    def _fetch_until(self, fetch_options, till=None, segment_sync=None):
         """
         Hit endpoint, update storage and return when since==till.
 
@@ -57,13 +58,14 @@ def _fetch_until(self, fetch_options, till=None):
         :return: last change number
         :rtype: int
         """
+        segment_list = []
         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)
@@ -74,17 +76,21 @@ def _fetch_until(self, fetch_options, till=None):
 
             for split in split_changes.get('splits', []):
                 if split['status'] == splits.Status.ACTIVE.value:
-#                    _LOGGER.debug('split details: '+str(split))                        
+                    _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()
+                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']
+                return split_changes['till'], segment_list
 
-    def _attempt_split_sync(self, fetch_options, till=None):
+    def _attempt_split_sync(self, fetch_options, till=None, segment_sync=None):
         """
         Hit endpoint, update storage and return True if sync is complete.
 
@@ -101,35 +107,35 @@ def _attempt_split_sync(self, fetch_options, till=None):
         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, segment_sync)
             if till is None or till <= change_number:
-                return True, remaining_attempts, change_number
+                return True, remaining_attempts, change_number, segment_list
             elif remaining_attempts <= 0:
-                return False, remaining_attempts, change_number
+                return False, remaining_attempts, change_number, segment_list
             how_long = self._backoff.get()
             time.sleep(how_long)
 
-    def synchronize_splits(self, till=None):
+    def synchronize_splits(self, till=None, segment_sync=None):
         """
         Hit endpoint, update storage and return True if sync is complete.
 
         :param till: Passed till from Streaming.
         :type till: int
         """
         fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
-        successful_sync, remaining_attempts, change_number = self._attempt_split_sync(fetch_options,
-                                                                                      till)
+        successful_sync, remaining_attempts, change_number, segment_list = self._attempt_split_sync(fetch_options,
+                                                                                      till, segment_sync)
         attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
         if successful_sync:  # succedeed sync
             _LOGGER.debug('Refresh completed in %d attempts.', attempts)
-            return self.segment_list
+            return 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, segment_sync)
         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 self.segment_list
+            return segment_list
         else:
             _LOGGER.debug('No changes fetched after %d attempts with CDN bypassed.',
                           without_cdn_attempts)

splitio/sync/synchronizer.py

@@ -205,7 +205,6 @@ 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):
         """
@@ -233,19 +232,14 @@ 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.')
-            
+            segment_list = self._split_synchronizers.split_sync.synchronize_splits(till, self._split_synchronizers.segment_sync)
+            if segment_list != []:
+                success = self._split_synchronizers.segment_sync.synchronize_segments(segment_list)
+                if not success:
+                    _LOGGER.error('Failed to sync segment.')
+                else:
+                    _LOGGER.debug('Segment synced.')            
             return True
         except APIException:
             _LOGGER.error('Failed syncing splits')
@@ -260,6 +254,11 @@ 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_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

@@ -17,7 +17,6 @@
 from splitio.models.segments import Segment
 from splitio.storage.inmemmory import InMemorySegmentStorage, InMemorySplitStorage
 
-
 class SynchronizerTests(object):
     def test_sync_all_failed_splits(self, mocker):
         api = mocker.Mock()
@@ -95,13 +94,14 @@ def test_synchronize_splits(self, mocker):
         segment_storage = InMemorySegmentStorage()
         segment_api = mocker.Mock()
         segment_api.fetch_segment.return_value = {'name': 'segmentA', 'added': ['key1', 'key2',
-                                                  'key3'], 'removed': [], 'since': -1, 'till': 123}
+                                                  '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)
 
+        synchronizer.synchronize_splits(123)
+        
         inserted_split = split_storage.get('some_name')
         assert isinstance(inserted_split, Split)
         assert inserted_split.name == 'some_name'
@@ -118,17 +118,29 @@ def test_sync_all(self, mocker):
         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(),
+
+        segment_storage = mocker.Mock(spec=SegmentStorage)
+        segment_storage.get_change_number.return_value = 123
+        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 = 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)
@@ -250,7 +262,7 @@ def sync_segments(*_):
 
         synchronizer.sync_all()
         assert counts['splits'] == 1
-#        assert counts['segments'] == 1
+        assert counts['segments'] == 2
 
     def test_sync_all_split_attempts(self, mocker):
         """Test that 3 attempts are done before failing."""

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 = [{