Added fetch segment with IFF split update

Author: chillaq

splitio/push/processor.py

@@ -19,7 +19,7 @@ def __init__(self, synchronizer, telemetry_runtime_producer):
         self._feature_flag_queue = Queue()
         self._segments_queue = Queue()
         self._synchronizer = synchronizer
-        self._feature_flag_worker = SplitWorker(synchronizer.synchronize_splits, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage, telemetry_runtime_producer)
+        self._feature_flag_worker = SplitWorker(synchronizer.synchronize_splits, synchronizer.synchronize_segment, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage, synchronizer.segment_storage, telemetry_runtime_producer)
         self._segments_worker = SegmentWorker(synchronizer.synchronize_segment, self._segments_queue)
         self._handlers = {
             UpdateType.SPLIT_UPDATE: self._handle_feature_flag_update,

splitio/push/splitworker.py

@@ -26,21 +26,35 @@ class SplitWorker(object):
 
     _centinel = object()
 
-    def __init__(self, synchronize_feature_flag, feature_flag_queue, feature_flag_storage, telemetry_runtime_producer):
+    def __init__(self, synchronize_feature_flag, synchronize_segment, feature_flag_queue, feature_flag_storage, segment_storage, telemetry_runtime_producer):
         """
         Class constructor.
 
         :param synchronize_feature_flag: handler to perform feature flag synchronization on incoming event
         :type synchronize_feature_flag: callable
 
+        :param synchronize_segment: handler to perform segment synchronization on incoming event
+        :type synchronize_segment: function
+
         :param feature_flag_queue: queue with feature flag updates notifications
         :type feature_flag_queue: queue
+
+        :param feature_flag_storage: feature flag storage instance
+        :type feature_flag_storage: splitio.storage.inmemory.InMemorySplitStorage
+
+        :param segment_storage: segment storage instance
+        :type segment_storage: splitio.storage.inmemory.InMemorySegmentStorage
+
+        :param telemetry_runtime_producer: Telemetry runtime producer instance
+        :type telemetry_runtime_producer: splitio.engine.telemetry.TelemetryRuntimeProducer
         """
         self._feature_flag_queue = feature_flag_queue
         self._handler = synchronize_feature_flag
+        self._segment_handler = synchronize_segment
         self._running = False
         self._worker = None
         self._feature_flag_storage = feature_flag_storage
+        self._segment_storage = segment_storage
         self._compression_handlers = {
             CompressionMode.NO_COMPRESSION: lambda event: base64.b64decode(event.feature_flag_definition),
             CompressionMode.GZIP_COMPRESSION: lambda event: gzip.decompress(base64.b64decode(event.feature_flag_definition)).decode('utf-8'),
@@ -62,7 +76,6 @@ def _check_instant_ff_update(self, event):
             return True
         return False
 
-
     def _run(self):
         """Run worker handler."""
         while self.is_running():
@@ -76,9 +89,14 @@ def _run(self):
                 if self._check_instant_ff_update(event):
                     try:
                         new_split = from_raw(json.loads(self._get_feature_flag_definition(event)))
+                        _LOGGER.debug(new_split)
                         if new_split.status == Status.ACTIVE:
                             self._feature_flag_storage.put(new_split)
                             _LOGGER.debug('Feature flag %s is updated', new_split.name)
+                            for segment_name in new_split.get_segment_names():
+                                if self._segment_storage.get(segment_name) is None:
+                                    _LOGGER.debug('Fetching new segment %s', segment_name)
+                                    self._segment_handler(segment_name, event.change_number)
                         else:
                             self._feature_flag_storage.remove(new_split.name)
                         self._feature_flag_storage.set_change_number(event.change_number)

splitio/sync/synchronizer.py

@@ -256,6 +256,10 @@ def __init__(self, split_synchronizers, split_tasks):
     def split_sync(self):
         return self._split_synchronizers.split_sync
 
+    @property
+    def segment_storage(self):
+        return self._split_synchronizers.segment_sync._segment_storage
+
     def _synchronize_segments(self):
         _LOGGER.debug('Starting segments synchronization')
         return self._split_synchronizers.segment_sync.synchronize_segments()

tests/push/test_split_worker.py

@@ -7,8 +7,7 @@
 from splitio.push.splitworker import SplitWorker
 from splitio.push.parser import SplitChangeUpdate
 from splitio.engine.telemetry import TelemetryStorageProducer
-from splitio.storage.inmemmory import InMemoryTelemetryStorage
-
+from splitio.storage.inmemmory import InMemoryTelemetryStorage, InMemorySplitStorage, InMemorySegmentStorage
 change_number_received = None
 
 
@@ -22,14 +21,10 @@ class SplitWorkerTests(object):
 
     def test_on_error(self, mocker):
         q = queue.Queue()
-        telemetry_storage = InMemoryTelemetryStorage()
-        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
-
         def handler_sync(change_number):
             raise APIException('some')
 
-        split_worker = SplitWorker(handler_sync, q, mocker.Mock(), telemetry_runtime_producer)
+        split_worker = SplitWorker(handler_sync, mocker.Mock(), q, mocker.Mock(), mocker.Mock(), mocker.Mock())
         split_worker.start()
         assert split_worker.is_running()
 
@@ -46,10 +41,7 @@ def handler_sync(change_number):
 
     def test_handler(self, mocker):
         q = queue.Queue()
-        telemetry_storage = InMemoryTelemetryStorage()
-        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
-        split_worker = SplitWorker(handler_sync, q, mocker.Mock(), telemetry_runtime_producer)
+        split_worker = SplitWorker(handler_sync, mocker.Mock(), q, mocker.Mock(), mocker.Mock(), mocker.Mock())
 
         global change_number_received
         assert not split_worker.is_running()
@@ -101,7 +93,7 @@ def test_compression(self, mocker):
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
-        split_worker = SplitWorker(handler_sync, q, mocker.Mock(), telemetry_runtime_producer)
+        split_worker = SplitWorker(handler_sync, mocker.Mock(), q, mocker.Mock(), mocker.Mock(), telemetry_runtime_producer)
         global change_number_received
         split_worker.start()
         def get_change_number():
@@ -148,10 +140,7 @@ def remove(feature_flag):
 
     def test_edge_cases(self, mocker):
         q = queue.Queue()
-        telemetry_storage = InMemoryTelemetryStorage()
-        telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
-        split_worker = SplitWorker(handler_sync, q, mocker.Mock(), telemetry_runtime_producer)
+        split_worker = SplitWorker(handler_sync, mocker.Mock(), q, mocker.Mock(), mocker.Mock(), mocker.Mock())
         global change_number_received
         split_worker.start()
 
@@ -190,4 +179,28 @@ def put(feature_flag):
         change_number_received = 0
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456, 2345, None, 1))
         time.sleep(0.1)
-        assert self._feature_flag == None
+        assert self._feature_flag == None
+
+    def test_fetch_segment(self, mocker):
+        q = queue.Queue()
+        split_storage = InMemorySplitStorage()
+        segment_storage = InMemorySegmentStorage()
+
+        self.segment_name = None
+        def segment_handler_sync(segment_name, change_number):
+            self.segment_name = segment_name
+            return
+        split_worker = SplitWorker(handler_sync, segment_handler_sync, q, split_storage, segment_storage, mocker.Mock())
+        split_worker.start()
+
+        def get_change_number():
+            return 2345
+        split_worker._feature_flag_storage.get_change_number = get_change_number
+
+        def check_instant_ff_update(event):
+            return True
+        split_worker._check_instant_ff_update = check_instant_ff_update
+
+        q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 1675095324253, 2345, 'eyJjaGFuZ2VOdW1iZXIiOiAxNjc1MDk1MzI0MjUzLCAidHJhZmZpY1R5cGVOYW1lIjogInVzZXIiLCAibmFtZSI6ICJiaWxhbF9zcGxpdCIsICJ0cmFmZmljQWxsb2NhdGlvbiI6IDEwMCwgInRyYWZmaWNBbGxvY2F0aW9uU2VlZCI6IC0xMzY0MTE5MjgyLCAic2VlZCI6IC02MDU5Mzg4NDMsICJzdGF0dXMiOiAiQUNUSVZFIiwgImtpbGxlZCI6IGZhbHNlLCAiZGVmYXVsdFRyZWF0bWVudCI6ICJvZmYiLCAiYWxnbyI6IDIsICJjb25kaXRpb25zIjogW3siY29uZGl0aW9uVHlwZSI6ICJST0xMT1VUIiwgIm1hdGNoZXJHcm91cCI6IHsiY29tYmluZXIiOiAiQU5EIiwgIm1hdGNoZXJzIjogW3sia2V5U2VsZWN0b3IiOiB7InRyYWZmaWNUeXBlIjogInVzZXIiLCAiYXR0cmlidXRlIjogbnVsbH0sICJtYXRjaGVyVHlwZSI6ICJJTl9TRUdNRU5UIiwgIm5lZ2F0ZSI6IGZhbHNlLCAidXNlckRlZmluZWRTZWdtZW50TWF0Y2hlckRhdGEiOiB7InNlZ21lbnROYW1lIjogImJpbGFsX3NlZ21lbnQifSwgIndoaXRlbGlzdE1hdGNoZXJEYXRhIjogbnVsbCwgInVuYXJ5TnVtZXJpY01hdGNoZXJEYXRhIjogbnVsbCwgImJldHdlZW5NYXRjaGVyRGF0YSI6IG51bGwsICJkZXBlbmRlbmN5TWF0Y2hlckRhdGEiOiBudWxsLCAiYm9vbGVhbk1hdGNoZXJEYXRhIjogbnVsbCwgInN0cmluZ01hdGNoZXJEYXRhIjogbnVsbH1dfSwgInBhcnRpdGlvbnMiOiBbeyJ0cmVhdG1lbnQiOiAib24iLCAic2l6ZSI6IDB9LCB7InRyZWF0bWVudCI6ICJvZmYiLCAic2l6ZSI6IDEwMH1dLCAibGFiZWwiOiAiaW4gc2VnbWVudCBiaWxhbF9zZWdtZW50In0sIHsiY29uZGl0aW9uVHlwZSI6ICJST0xMT1VUIiwgIm1hdGNoZXJHcm91cCI6IHsiY29tYmluZXIiOiAiQU5EIiwgIm1hdGNoZXJzIjogW3sia2V5U2VsZWN0b3IiOiB7InRyYWZmaWNUeXBlIjogInVzZXIiLCAiYXR0cmlidXRlIjogbnVsbH0sICJtYXRjaGVyVHlwZSI6ICJBTExfS0VZUyIsICJuZWdhdGUiOiBmYWxzZSwgInVzZXJEZWZpbmVkU2VnbWVudE1hdGNoZXJEYXRhIjogbnVsbCwgIndoaXRlbGlzdE1hdGNoZXJEYXRhIjogbnVsbCwgInVuYXJ5TnVtZXJpY01hdGNoZXJEYXRhIjogbnVsbCwgImJldHdlZW5NYXRjaGVyRGF0YSI6IG51bGwsICJkZXBlbmRlbmN5TWF0Y2hlckRhdGEiOiBudWxsLCAiYm9vbGVhbk1hdGNoZXJEYXRhIjogbnVsbCwgInN0cmluZ01hdGNoZXJEYXRhIjogbnVsbH1dfSwgInBhcnRpdGlvbnMiOiBbeyJ0cmVhdG1lbnQiOiAib24iLCAic2l6ZSI6IDUwfSwgeyJ0cmVhdG1lbnQiOiAib2ZmIiwgInNpemUiOiA1MH1dLCAibGFiZWwiOiAiZGVmYXVsdCBydWxlIn1dLCAiY29uZmlndXJhdGlvbnMiOiB7fX0=', 0))
+        time.sleep(0.1)
+        assert self.segment_name == "bilal_segment"