feat: add odp event flush interval (#414)

* expose odp flush interval to the client

* add odp flush interval and tests

* tests fix

* Delete z_matjaz_play directory

* pr fixes

* add a new test

* cleanup

* rename docstring

Author: Mat001

optimizely/helpers/sdk_settings.py

@@ -31,8 +31,9 @@ def __init__(
             odp_segments_cache: Optional[OptimizelySegmentsCache] = None,
             odp_segment_manager: Optional[OdpSegmentManager] = None,
             odp_event_manager: Optional[OdpEventManager] = None,
-            fetch_segments_timeout: Optional[int] = None,
-            odp_event_timeout: Optional[int] = None
+            odp_segment_request_timeout: Optional[int] = None,
+            odp_event_request_timeout: Optional[int] = None,
+            odp_event_flush_interval: Optional[int] = None
     ) -> None:
         """
         Args:
@@ -47,8 +48,10 @@ def __init__(
             `fetch_qualified_segments(user_key, user_value, options)`.
           odp_event_manager: A custom odp event manager. Required method is:
             `send_event(type:, action:, identifiers:, data:)`
-          fetch_segments_timeout: A fetch segment timeout in seconds (optional).
-          odp_event_timeout: A send odp event timeout in seconds (optional).
+          odp_segment_request_timeout: Time to wait in seconds for fetch_qualified_segments request to
+            send successfully (optional).
+          odp_event_request_timeout: Time to wait in seconds for send_odp_events request to send successfully.
+          odp_event_flush_interval: Time to wait for events to accumulate before sending a batch in seconds (optional).
         """
 
         self.odp_disabled = odp_disabled
@@ -57,5 +60,6 @@ def __init__(
         self.segments_cache = odp_segments_cache
         self.odp_segment_manager = odp_segment_manager
         self.odp_event_manager = odp_event_manager
-        self.fetch_segments_timeout = fetch_segments_timeout
-        self.odp_event_timeout = odp_event_timeout
+        self.fetch_segments_timeout = odp_segment_request_timeout
+        self.odp_event_timeout = odp_event_request_timeout
+        self.odp_flush_interval = odp_event_flush_interval

optimizely/odp/odp_event_manager.py

@@ -40,31 +40,37 @@ class OdpEventManager:
     The OdpEventManager maintains a single consumer thread that pulls events off of
     the queue and buffers them before events are sent to ODP.
     Sends events when the batch size is met or when the flush timeout has elapsed.
+    Flushes the event queue after specified time (seconds).
     """
 
     def __init__(
         self,
         logger: Optional[_logging.Logger] = None,
         api_manager: Optional[OdpEventApiManager] = None,
-        timeout: Optional[int] = None
+        request_timeout: Optional[int] = None,
+        flush_interval: Optional[int] = None
     ):
         """OdpEventManager init method to configure event batching.
 
         Args:
             logger: Optional component which provides a log method to log messages. By default nothing would be logged.
             api_manager: Optional component which sends events to ODP.
-            timeout: Optional event timeout in seconds.
+            request_timeout: Optional event timeout in seconds - wait time for odp platform to respond before failing.
+            flush_interval: Optional time to wait for events to accumulate before sending the batch in seconds.
         """
         self.logger = logger or _logging.NoOpLogger()
-        self.api_manager = api_manager or OdpEventApiManager(self.logger, timeout)
+        self.api_manager = api_manager or OdpEventApiManager(self.logger, request_timeout)
 
         self.odp_config: Optional[OdpConfig] = None
         self.api_key: Optional[str] = None
         self.api_host: Optional[str] = None
 
         self.event_queue: Queue[OdpEvent | Signal] = Queue(OdpEventManagerConfig.DEFAULT_QUEUE_CAPACITY)
-        self.batch_size = OdpEventManagerConfig.DEFAULT_BATCH_SIZE
-        self.flush_interval = OdpEventManagerConfig.DEFAULT_FLUSH_INTERVAL
+        self.batch_size = 0 if flush_interval == 0 else OdpEventManagerConfig.DEFAULT_BATCH_SIZE
+
+        self.flush_interval = OdpEventManagerConfig.DEFAULT_FLUSH_INTERVAL if flush_interval is None \
+            else flush_interval
+
         self._flush_deadline: float = 0
         self.retry_count = OdpEventManagerConfig.DEFAULT_RETRY_COUNT
         self._current_batch: list[OdpEvent] = []

optimizely/odp/odp_manager.py

@@ -35,6 +35,7 @@ def __init__(
         event_manager: Optional[OdpEventManager] = None,
         fetch_segments_timeout: Optional[int] = None,
         odp_event_timeout: Optional[int] = None,
+        odp_flush_interval: Optional[int] = None,
         logger: Optional[optimizely_logger.Logger] = None
     ) -> None:
 
@@ -58,7 +59,8 @@ def __init__(
                 )
             self.segment_manager = OdpSegmentManager(segments_cache, logger=self.logger, timeout=fetch_segments_timeout)
 
-        self.event_manager = self.event_manager or OdpEventManager(self.logger, timeout=odp_event_timeout)
+        self.event_manager = self.event_manager or OdpEventManager(self.logger, request_timeout=odp_event_timeout,
+                                                                   flush_interval=odp_flush_interval)
         self.segment_manager.odp_config = self.odp_config
 
     def fetch_qualified_segments(self, user_id: str, options: list[str]) -> Optional[list[str]]:

optimizely/optimizely.py

@@ -1317,7 +1317,8 @@ def setup_odp(self, sdk_key: Optional[str]) -> None:
             self.sdk_settings.odp_event_manager,
             self.sdk_settings.fetch_segments_timeout,
             self.sdk_settings.odp_event_timeout,
-            self.logger
+            self.sdk_settings.odp_flush_interval,
+            self.logger,
         )
 
         if self.sdk_settings.odp_disabled:

tests/test_odp_event_manager.py

@@ -382,10 +382,10 @@ def test_odp_event_manager_override_default_data(self, *args):
         mock_send.assert_called_once_with(self.api_key, self.api_host, [processed_event])
         event_manager.stop()
 
-    def test_odp_event_manager_flush_timeout(self, *args):
+    def test_odp_event_manager_flush_interval(self, *args):
+        """Verify that both events have been sent together after they have been batched."""
         mock_logger = mock.Mock()
-        event_manager = OdpEventManager(mock_logger)
-        event_manager.flush_interval = .5
+        event_manager = OdpEventManager(mock_logger, flush_interval=.5)
         event_manager.start(self.odp_config)
 
         with mock.patch.object(
@@ -394,13 +394,34 @@ def test_odp_event_manager_flush_timeout(self, *args):
             event_manager.send_event(**self.events[0])
             event_manager.send_event(**self.events[1])
             event_manager.event_queue.join()
-            time.sleep(1)
+            time.sleep(1)   # ensures that the flush interval time has passed
 
         mock_logger.error.assert_not_called()
         mock_logger.debug.assert_any_call('ODP event queue: flushing on interval.')
         mock_send.assert_called_once_with(self.api_key, self.api_host, self.processed_events)
         event_manager.stop()
 
+    def test_odp_event_manager_flush_interval_is_zero(self, *args):
+        """Verify that event is immediately if flush interval is zero."""
+        mock_logger = mock.Mock()
+        event_manager = OdpEventManager(mock_logger, flush_interval=0)
+        event_manager.start(self.odp_config)
+
+        with mock.patch.object(
+                event_manager.api_manager, 'send_odp_events', new_callable=CopyingMock, return_value=False
+        ) as mock_send:
+            event_manager.send_event(**self.events[0])
+            event_manager.send_event(**self.events[1])
+            event_manager.event_queue.join()
+
+        mock_send.assert_has_calls(
+            [mock.call(self.api_key, self.api_host, [self.processed_events[0]]),
+             mock.call(self.api_key, self.api_host, [self.processed_events[1]])]
+        )
+        mock_logger.error.assert_not_called()
+        mock_logger.debug.assert_any_call('ODP event queue: flushing batch size 1.')
+        event_manager.stop()
+
     def test_odp_event_manager_events_before_odp_ready(self, *args):
         mock_logger = mock.Mock()
         odp_config = OdpConfig()

tests/test_optimizely.py

@@ -5140,11 +5140,7 @@ def test_user_context_invalid_user_id(self):
             uc = self.optimizely.create_user_context(u)
             self.assertIsNone(uc, "invalid user id should return none")
 
-    def test_invalid_flag_key(self):
-        """Tests invalid flag key in function get_flag_variation_by_key()."""
-        pass
-
-    def test_send_identify_event_when_called_with_odp_enabled(self):
+    def test_send_identify_event__when_called_with_odp_enabled(self):
         mock_logger = mock.Mock()
         client = optimizely.Optimizely(json.dumps(self.config_dict_with_audience_segments), logger=mock_logger)
         with mock.patch.object(client, 'identify_user') as identify:
@@ -5154,6 +5150,34 @@ def test_send_identify_event_when_called_with_odp_enabled(self):
         mock_logger.error.assert_not_called()
         client.close()
 
+    def test_sdk_settings__accept_zero_for_flush_interval(self):
+        mock_logger = mock.Mock()
+        sdk_settings = OptimizelySdkSettings(odp_event_flush_interval=0)
+        client = optimizely.Optimizely(
+            json.dumps(self.config_dict_with_audience_segments),
+            logger=mock_logger,
+            settings=sdk_settings
+        )
+        flush_interval = client.odp_manager.event_manager.flush_interval
+
+        self.assertEqual(flush_interval, 0)
+        mock_logger.error.assert_not_called()
+        client.close()
+
+    def test_sdk_settings__should_use_default_when_odp_flush_interval_none(self):
+        mock_logger = mock.Mock()
+        sdk_settings = OptimizelySdkSettings(odp_event_flush_interval=None)
+        client = optimizely.Optimizely(
+            json.dumps(self.config_dict_with_audience_segments),
+            logger=mock_logger,
+            settings=sdk_settings
+        )
+        flush_interval = client.odp_manager.event_manager.flush_interval
+        self.assertEqual(flush_interval, enums.OdpEventManagerConfig.DEFAULT_FLUSH_INTERVAL)
+
+        mock_logger.error.assert_not_called()
+        client.close()
+
     def test_sdk_settings__log_info_when_disabled(self):
         mock_logger = mock.Mock()
         sdk_settings = OptimizelySdkSettings(odp_disabled=True)
@@ -5162,6 +5186,7 @@ def test_sdk_settings__log_info_when_disabled(self):
             logger=mock_logger,
             settings=sdk_settings
         )
+
         self.assertIsNone(client.odp_manager.event_manager)
         self.assertIsNone(client.odp_manager.segment_manager)
         mock_logger.info.assert_called_once_with('ODP is disabled.')
@@ -5211,6 +5236,36 @@ def test_sdk_settings__accept_cache_size_and_cache_timeout(self):
         mock_logger.error.assert_not_called()
         client.close()
 
+    def test_sdk_settings__use_default_cache_size_and_timeout_when_odp_flush_interval_none(self):
+        mock_logger = mock.Mock()
+        sdk_settings = OptimizelySdkSettings()
+        client = optimizely.Optimizely(
+            json.dumps(self.config_dict_with_audience_segments),
+            logger=mock_logger,
+            settings=sdk_settings
+        )
+        segments_cache = client.odp_manager.segment_manager.segments_cache
+        self.assertEqual(segments_cache.timeout, enums.OdpSegmentsCacheConfig.DEFAULT_TIMEOUT_SECS)
+        self.assertEqual(segments_cache.capacity, enums.OdpSegmentsCacheConfig.DEFAULT_CAPACITY)
+
+        mock_logger.error.assert_not_called()
+        client.close()
+
+    def test_sdk_settings__accept_zero_cache_size_timeout_and_cache_size(self):
+        mock_logger = mock.Mock()
+        sdk_settings = OptimizelySdkSettings(segments_cache_size=0, segments_cache_timeout_in_secs=0)
+        client = optimizely.Optimizely(
+            json.dumps(self.config_dict_with_audience_segments),
+            logger=mock_logger,
+            settings=sdk_settings
+        )
+        segments_cache = client.odp_manager.segment_manager.segments_cache
+        self.assertEqual(segments_cache.capacity, 0)
+        self.assertEqual(segments_cache.timeout, 0)
+
+        mock_logger.error.assert_not_called()
+        client.close()
+
     def test_sdk_settings__accept_valid_custom_cache(self):
         class CustomCache:
             def reset(self):