feat: add odp manager (#405)

* bump up py version in gitactions to 3.10
* feat: add odp_manager
* add update config event manager signal

Co-authored-by: Andy Leap <andrew.leap@optimizely.com>

Author: Mat001

optimizely/exceptions.py

@@ -64,3 +64,21 @@ class UnsupportedDatafileVersionException(Exception):
     """ Raised when provided version in datafile is not supported. """
 
     pass
+
+
+class OdpNotEnabled(Exception):
+    """ Raised when Optimizely Data Platform (ODP) is not enabled. """
+
+    pass
+
+
+class OdpNotIntegrated(Exception):
+    """ Raised when Optimizely Data Platform (ODP) is not integrated. """
+
+    pass
+
+
+class OdpInvalidData(Exception):
+    """ Raised when passing invalid ODP data. """
+
+    pass

optimizely/helpers/enums.py

@@ -123,8 +123,9 @@ class Errors:
     INVALID_SEGMENT_IDENTIFIER: Final = 'Audience segments fetch failed (invalid identifier).'
     FETCH_SEGMENTS_FAILED: Final = 'Audience segments fetch failed ({}).'
     ODP_EVENT_FAILED: Final = 'ODP event send failed ({}).'
-    ODP_NOT_ENABLED: Final = 'ODP is not enabled.'
     ODP_NOT_INTEGRATED: Final = 'ODP is not integrated.'
+    ODP_NOT_ENABLED: Final = 'ODP is not enabled.'
+    ODP_INVALID_DATA: Final = 'ODP data is not valid.'
 
 
 class ForcedDecisionLogs:
@@ -214,3 +215,15 @@ class OdpEventManagerConfig:
     DEFAULT_BATCH_SIZE: Final = 10
     DEFAULT_FLUSH_INTERVAL: Final = 1
     DEFAULT_RETRY_COUNT: Final = 3
+
+
+class OdpManagerConfig:
+    """ODP Manager configs."""
+    KEY_FOR_USER_ID: Final = 'fs_user_id'
+    EVENT_TYPE: Final = 'fullstack'
+
+
+class OdpSegmentsCacheConfig:
+    """ODP Segment Cache configs."""
+    DEFAULT_CAPACITY: Final = 10_000
+    DEFAULT_TIMEOUT_SECS: Final = 600

optimizely/odp/odp_event_manager.py

@@ -12,23 +12,25 @@
 # limitations under the License.
 
 from __future__ import annotations
+
+import time
 from enum import Enum
+from queue import Empty, Queue, Full
 from threading import Thread
 from typing import Optional
-import time
-from queue import Empty, Queue, Full
 
 from optimizely import logger as _logging
-from .odp_event import OdpEvent, OdpDataDict
+from optimizely.helpers.enums import OdpEventManagerConfig, Errors, OdpManagerConfig
 from .odp_config import OdpConfig, OdpConfigState
+from .odp_event import OdpEvent, OdpDataDict
 from .zaius_rest_api_manager import ZaiusRestApiManager
-from optimizely.helpers.enums import OdpEventManagerConfig, Errors
 
 
 class Signal(Enum):
     """Enum for sending signals to the event queue."""
     SHUTDOWN = 1
     FLUSH = 2
+    UPDATE_CONFIG = 3
 
 
 class OdpEventManager:
@@ -55,7 +57,11 @@ def __init__(
         """
         self.logger = logger or _logging.NoOpLogger()
         self.zaius_manager = api_manager or ZaiusRestApiManager(self.logger)
+
         self.odp_config = odp_config
+        self.api_key = odp_config.get_api_key()
+        self.api_host = odp_config.get_api_host()
+
         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
@@ -101,7 +107,11 @@ def _run(self) -> None:
                     self.logger.debug('ODP event queue: received flush signal.')
                     self._flush_batch()
                     self.event_queue.task_done()
-                    continue
+
+                elif item == Signal.UPDATE_CONFIG:
+                    self.logger.debug('ODP event queue: received update config signal.')
+                    self._update_config()
+                    self.event_queue.task_done()
 
                 elif isinstance(item, OdpEvent):
                     self._add_to_batch(item)
@@ -136,10 +146,7 @@ def _flush_batch(self) -> None:
             self.logger.debug('ODP event queue: nothing to flush.')
             return
 
-        api_key = self.odp_config.get_api_key()
-        api_host = self.odp_config.get_api_host()
-
-        if not api_key or not api_host:
+        if not self.api_key or not self.api_host:
             self.logger.debug(Errors.ODP_NOT_INTEGRATED)
             self._current_batch.clear()
             return
@@ -149,7 +156,7 @@ def _flush_batch(self) -> None:
 
         for i in range(1 + self.retry_count):
             try:
-                should_retry = self.zaius_manager.send_odp_events(api_key, api_host, self._current_batch)
+                should_retry = self.zaius_manager.send_odp_events(self.api_key, self.api_host, self._current_batch)
             except Exception as error:
                 should_retry = False
                 self.logger.error(Errors.ODP_EVENT_FAILED.format(f'Error: {error} {self._current_batch}'))
@@ -236,3 +243,22 @@ def dispatch(self, event: OdpEvent) -> None:
             self.event_queue.put_nowait(event)
         except Full:
             self.logger.warning(Errors.ODP_EVENT_FAILED.format("Queue is full"))
+
+    def identify_user(self, user_id: str) -> None:
+        self.send_event(OdpManagerConfig.EVENT_TYPE, 'identified',
+                        {OdpManagerConfig.KEY_FOR_USER_ID: user_id}, {})
+
+    def update_config(self) -> None:
+        """Adds update config signal to event_queue."""
+        try:
+            self.event_queue.put_nowait(Signal.UPDATE_CONFIG)
+        except Full:
+            self.logger.error("Error updating ODP config for the event queue")
+
+    def _update_config(self) -> None:
+        """Updates the configuration used to send events."""
+        if len(self._current_batch) > 0:
+            self._flush_batch()
+
+        self.api_host = self.odp_config.get_api_host()
+        self.api_key = self.odp_config.get_api_key()

optimizely/odp/odp_manager.py

@@ -0,0 +1,133 @@
+# Copyright 2022, Optimizely
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from typing import Optional, Any
+
+from optimizely import exceptions as optimizely_exception
+from optimizely import logger as optimizely_logger
+from optimizely.helpers.enums import Errors, OdpManagerConfig, OdpSegmentsCacheConfig
+from optimizely.helpers.validator import are_odp_data_types_valid
+from optimizely.odp.lru_cache import OptimizelySegmentsCache, LRUCache
+from optimizely.odp.odp_config import OdpConfig, OdpConfigState
+from optimizely.odp.odp_event_manager import OdpEventManager
+from optimizely.odp.odp_segment_manager import OdpSegmentManager
+from optimizely.odp.zaius_graphql_api_manager import ZaiusGraphQLApiManager
+
+
+class OdpManager:
+    """Orchestrates segment manager, event manager and odp config."""
+
+    def __init__(
+        self,
+        disable: bool,
+        segments_cache: Optional[OptimizelySegmentsCache] = None,
+        segment_manager: Optional[OdpSegmentManager] = None,
+        event_manager: Optional[OdpEventManager] = None,
+        logger: Optional[optimizely_logger.Logger] = None
+    ) -> None:
+
+        self.enabled = not disable
+        self.odp_config = OdpConfig()
+        self.logger = logger or optimizely_logger.NoOpLogger()
+
+        self.segment_manager = segment_manager
+        self.event_manager = event_manager
+
+        if not self.enabled:
+            self.logger.info('ODP is disabled.')
+            return
+
+        if not self.segment_manager:
+            if not segments_cache:
+                segments_cache = LRUCache(
+                    OdpSegmentsCacheConfig.DEFAULT_CAPACITY,
+                    OdpSegmentsCacheConfig.DEFAULT_TIMEOUT_SECS
+                )
+            self.segment_manager = OdpSegmentManager(
+                self.odp_config,
+                segments_cache,
+                ZaiusGraphQLApiManager(logger), logger
+            )
+        else:
+            self.segment_manager.odp_config = self.odp_config
+
+        if event_manager:
+            event_manager.odp_config = self.odp_config
+            self.event_manager = event_manager
+        else:
+            self.event_manager = OdpEventManager(self.odp_config, logger)
+
+        self.event_manager.start()
+
+    def fetch_qualified_segments(self, user_id: str, options: list[str]) -> Optional[list[str]]:
+        if not self.enabled or not self.segment_manager:
+            self.logger.error(Errors.ODP_NOT_ENABLED)
+            return None
+
+        user_key = OdpManagerConfig.KEY_FOR_USER_ID
+        user_value = user_id
+
+        return self.segment_manager.fetch_qualified_segments(user_key, user_value, options)
+
+    def identify_user(self, user_id: str) -> None:
+        if not self.enabled or not self.event_manager:
+            self.logger.debug('ODP identify event is not dispatched (ODP disabled).')
+            return
+        if self.odp_config.odp_state() == OdpConfigState.NOT_INTEGRATED:
+            self.logger.debug('ODP identify event is not dispatched (ODP not integrated).')
+            return
+
+        self.event_manager.identify_user(user_id)
+
+    def send_event(self, type: str, action: str, identifiers: dict[str, str], data: dict[str, Any]) -> None:
+        """
+        Send an event to the ODP server.
+
+        Args:
+            type: The event type.
+            action: The event action name.
+            identifiers: A dictionary for identifiers.
+            data: A dictionary for associated data. The default event data will be added to this data
+            before sending to the ODP server.
+
+        Raises custom exception if error is detected.
+        """
+        if not self.enabled or not self.event_manager:
+            raise optimizely_exception.OdpNotEnabled(Errors.ODP_NOT_ENABLED)
+
+        if self.odp_config.odp_state() == OdpConfigState.NOT_INTEGRATED:
+            raise optimizely_exception.OdpNotIntegrated(Errors.ODP_NOT_INTEGRATED)
+
+        if not are_odp_data_types_valid(data):
+            raise optimizely_exception.OdpInvalidData(Errors.ODP_INVALID_DATA)
+
+        self.event_manager.send_event(type, action, identifiers, data)
+
+    def update_odp_config(self, api_key: Optional[str], api_host: Optional[str],
+                          segments_to_check: list[str]) -> None:
+        if not self.enabled:
+            return
+
+        config_changed = self.odp_config.update(api_key, api_host, segments_to_check)
+        if not config_changed:
+            self.logger.debug('Odp config was not changed.')
+            return
+
+        # reset segments cache when odp integration or segments to check are changed
+        if self.segment_manager:
+            self.segment_manager.reset()
+
+        if self.event_manager:
+            self.event_manager.update_config()

optimizely/odp/odp_segment_manager.py

@@ -26,17 +26,21 @@
 class OdpSegmentManager:
     """Schedules connections to ODP for audience segmentation and caches the results."""
 
-    def __init__(self, odp_config: OdpConfig, segments_cache: OptimizelySegmentsCache,
-                 zaius_manager: ZaiusGraphQLApiManager,
-                 logger: Optional[optimizely_logger.Logger] = None) -> None:
+    def __init__(
+        self,
+        odp_config: OdpConfig,
+        segments_cache: OptimizelySegmentsCache,
+        zaius_manager: ZaiusGraphQLApiManager,
+        logger: Optional[optimizely_logger.Logger] = None
+    ) -> None:
 
         self.odp_config = odp_config
         self.segments_cache = segments_cache
         self.zaius_manager = zaius_manager
         self.logger = logger or optimizely_logger.NoOpLogger()
 
-    def fetch_qualified_segments(self, user_key: str, user_value: str, options: list[str]) -> \
-            Optional[list[str]]:
+    def fetch_qualified_segments(self, user_key: str, user_value: str, options: list[str]
+                                 ) -> Optional[list[str]]:
         """
         Args:
             user_key: The key for identifying the id type.
@@ -64,7 +68,7 @@ def fetch_qualified_segments(self, user_key: str, user_value: str, options: list
         reset_cache = OptimizelyOdpOption.RESET_CACHE in options
 
         if reset_cache:
-            self._reset()
+            self.reset()
 
         if not ignore_cache and not reset_cache:
             segments = self.segments_cache.lookup(cache_key)
@@ -83,7 +87,7 @@ def fetch_qualified_segments(self, user_key: str, user_value: str, options: list
 
         return segments
 
-    def _reset(self) -> None:
+    def reset(self) -> None:
         self.segments_cache.reset()
 
     def make_cache_key(self, user_key: str, user_value: str) -> str:

tests/test_odp_event_manager.py

@@ -411,6 +411,7 @@ def test_odp_event_manager_events_before_odp_ready(self, *args):
             event_manager.send_event(**self.events[1])
 
             odp_config.update(self.api_key, self.api_host, [])
+            event_manager.update_config()
             event_manager.event_queue.join()
 
             event_manager.send_event(**self.events[0])
@@ -423,6 +424,7 @@ def test_odp_event_manager_events_before_odp_ready(self, *args):
         mock_logger.debug.assert_has_calls([
             mock.call('ODP event queue: cannot send before the datafile has loaded.'),
             mock.call('ODP event queue: cannot send before the datafile has loaded.'),
+            mock.call('ODP event queue: received update config signal.'),
             mock.call('ODP event queue: adding event.'),
             mock.call('ODP event queue: adding event.'),
             mock.call('ODP event queue: received flush signal.'),
@@ -442,6 +444,7 @@ def test_odp_event_manager_events_before_odp_disabled(self, *args):
             event_manager.send_event(**self.events[1])
 
             odp_config.update(None, None, [])
+            event_manager.update_config()
             event_manager.event_queue.join()
 
             event_manager.send_event(**self.events[0])
@@ -453,6 +456,7 @@ def test_odp_event_manager_events_before_odp_disabled(self, *args):
         mock_logger.debug.assert_has_calls([
             mock.call('ODP event queue: cannot send before the datafile has loaded.'),
             mock.call('ODP event queue: cannot send before the datafile has loaded.'),
+            mock.call('ODP event queue: received update config signal.'),
             mock.call(Errors.ODP_NOT_INTEGRATED),
             mock.call(Errors.ODP_NOT_INTEGRATED)
         ])
@@ -496,20 +500,25 @@ def test_odp_event_manager_disabled_after_events_in_queue(self, *args):
         odp_config = OdpConfig(self.api_key, self.api_host)
 
         event_manager = OdpEventManager(odp_config, mock_logger)
-        event_manager.batch_size = 2
+        event_manager.batch_size = 3
 
         with mock.patch('optimizely.odp.odp_event_manager.OdpEventManager.is_running', True):
             event_manager.send_event(**self.events[0])
             event_manager.send_event(**self.events[1])
-
-        with mock.patch.object(event_manager.zaius_manager, 'send_odp_events') as mock_send:
             odp_config.update(None, None, [])
+            event_manager.update_config()
+
+        with mock.patch.object(
+            event_manager.zaius_manager, 'send_odp_events', new_callable=CopyingMock, return_value=False
+        ) as mock_send:
             event_manager.start()
+            event_manager.send_event(**self.events[0])
             event_manager.send_event(**self.events[1])
+            event_manager.send_event(**self.events[0])
             event_manager.event_queue.join()
 
         self.assertEqual(len(event_manager._current_batch), 0)
         mock_logger.debug.assert_any_call(Errors.ODP_NOT_INTEGRATED)
         mock_logger.error.assert_not_called()
-        mock_send.assert_not_called()
+        mock_send.assert_called_once_with(self.api_key, self.api_host, self.processed_events)
         event_manager.stop()