feat: Introduce flag change tracker api (#229)

The client instance will now provide access to a `flag_tracker`. This
tracker allows developers to be notified when a flag configuration
changes (or optionally when the /value/ of a flag changes for a
particular context).

Author: keelerm84

ldclient/client.py

@@ -29,10 +29,11 @@
 from ldclient.impl.listeners import Listeners
 from ldclient.impl.stubs import NullEventProcessor, NullUpdateProcessor
 from ldclient.impl.util import check_uwsgi, log
-from ldclient.interfaces import BigSegmentStoreStatusProvider, DataSourceStatusProvider, FeatureRequester, FeatureStore
+from ldclient.interfaces import BigSegmentStoreStatusProvider, DataSourceStatusProvider, FeatureRequester, FeatureStore, FlagTracker
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind
 from ldclient.feature_store import FeatureStore
 from ldclient.migrations import Stage, OpTracker
+from ldclient.impl.flag_tracker import FlagTrackerImpl
 
 from threading import Lock
 
@@ -103,9 +104,13 @@ def __init__(self, config: Config, start_wait: float=5):
 
         store = _FeatureStoreClientWrapper(self._config.feature_store)
 
-        listeners = Listeners()
-        self._config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners)
-        self.__data_source_status_provider = DataSourceStatusProviderImpl(listeners, self._config._data_source_update_sink)
+        data_source_listeners = Listeners()
+        flag_change_listeners = Listeners()
+
+        self.__flag_tracker = FlagTrackerImpl(flag_change_listeners, lambda key, context: self.variation(key, context, None))
+
+        self._config._data_source_update_sink = DataSourceUpdateSinkImpl(store, data_source_listeners, flag_change_listeners)
+        self.__data_source_status_provider = DataSourceStatusProviderImpl(data_source_listeners, self._config._data_source_update_sink)
         self._store = store  # type: FeatureStore
 
         big_segment_store_manager = BigSegmentStoreManager(self._config.big_segments)
@@ -510,5 +515,17 @@ def data_source_status_provider(self) -> DataSourceStatusProvider:
         """
         return self.__data_source_status_provider
 
+    @property
+    def flag_tracker(self) -> FlagTracker:
+        """
+        Returns an interface for tracking changes in feature flag configurations.
+
+        The :class:`ldclient.interfaces.FlagTracker` contains methods for
+        requesting notifications about feature flag changes using an event
+        listener model.
+        """
+        return self.__flag_tracker
+
+
 
 __all__ = ['LDClient', 'Config']

ldclient/impl/datasource/status.py

@@ -1,16 +1,21 @@
+from ldclient.versioned_data_kind import FEATURES, SEGMENTS
+from ldclient.impl.dependency_tracker import DependencyTracker
 from ldclient.impl.listeners import Listeners
-from ldclient.interfaces import DataSourceStatusProvider, DataSourceUpdateSink, DataSourceStatus, FeatureStore, DataSourceState, DataSourceErrorInfo, DataSourceErrorKind
+from ldclient.interfaces import DataSourceStatusProvider, DataSourceUpdateSink, DataSourceStatus, FeatureStore, DataSourceState, DataSourceErrorInfo, DataSourceErrorKind, FlagChange
 from ldclient.impl.rwlock import ReadWriteLock
 from ldclient.versioned_data_kind import VersionedDataKind
+from ldclient.impl.dependency_tracker import KindAndKey
 
 import time
-from typing import Callable, Mapping, Optional
+from typing import Callable, Mapping, Optional, Set
 
 
 class DataSourceUpdateSinkImpl(DataSourceUpdateSink):
-    def __init__(self, store: FeatureStore, listeners: Listeners):
+    def __init__(self, store: FeatureStore, status_listeners: Listeners, flag_change_listeners: Listeners):
         self.__store = store
-        self.__listeners = listeners
+        self.__status_listeners = status_listeners
+        self.__flag_change_listeners = flag_change_listeners
+        self.__tracker = DependencyTracker()
 
         self.__lock = ReadWriteLock()
         self.__status = DataSourceStatus(
@@ -28,13 +33,38 @@ def status(self) -> DataSourceStatus:
             self.__lock.runlock()
 
     def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]):
-        self.__monitor_store_update(lambda: self.__store.init(all_data))
+        old_data = None
+
+        def init_store():
+            nonlocal old_data
+            if self.__flag_change_listeners.has_listeners():
+                old_data = {}
+                for kind in [FEATURES, SEGMENTS]:
+                    old_data[kind] = self.__store.all(kind, lambda x: x)
+
+            self.__store.init(all_data)
+
+        self.__monitor_store_update(init_store)
+        self.__reset_tracker_with_new_data(all_data)
+
+        if old_data is None:
+            return
+
+        self.__send_change_events(
+            self.__compute_changed_items_for_full_data_set(old_data, all_data)
+        )
 
     def upsert(self, kind: VersionedDataKind, item: dict):
         self.__monitor_store_update(lambda: self.__store.upsert(kind, item))
 
+        # TODO(sc-212471): We only want to do this if the store successfully
+        # updates the record.
+        key = item.get('key', '')
+        self.__update_dependency_for_single_item(kind, key, item)
+
     def delete(self, kind: VersionedDataKind, key: str, version: int):
         self.__monitor_store_update(lambda: self.__store.delete(kind, key, version))
+        self.__update_dependency_for_single_item(kind, key, None)
 
     def update_status(self, new_state: DataSourceState, new_error: Optional[DataSourceErrorInfo]):
         status_to_broadcast = None
@@ -60,7 +90,7 @@ def update_status(self, new_state: DataSourceState, new_error: Optional[DataSour
             self.__lock.unlock()
 
         if status_to_broadcast is not None:
-            self.__listeners.notify(status_to_broadcast)
+            self.__status_listeners.notify(status_to_broadcast)
 
     def __monitor_store_update(self, fn: Callable[[], None]):
         try:
@@ -75,6 +105,46 @@ def __monitor_store_update(self, fn: Callable[[], None]):
             self.update_status(DataSourceState.INTERRUPTED, error_info)
             raise
 
+    def __update_dependency_for_single_item(self, kind: VersionedDataKind, key: str, item: Optional[dict]):
+        self.__tracker.update_dependencies_from(kind, key, item)
+        if self.__flag_change_listeners.has_listeners():
+            affected_items: Set[KindAndKey] = set()
+            self.__tracker.add_affected_items(affected_items, KindAndKey(kind=kind, key=key))
+            self.__send_change_events(affected_items)
+
+    def __reset_tracker_with_new_data(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]):
+        self.__tracker.reset()
+
+        for kind, items in all_data.items():
+            for key, item in items.items():
+                self.__tracker.update_dependencies_from(kind, key, item)
+
+    def __send_change_events(self, affected_items: Set[KindAndKey]):
+        for item in affected_items:
+            if item.kind == FEATURES:
+                self.__flag_change_listeners.notify(FlagChange(item.key))
+
+    def __compute_changed_items_for_full_data_set(self, old_data: Mapping[VersionedDataKind, Mapping[str, dict]], new_data: Mapping[VersionedDataKind, Mapping[str, dict]]):
+        affected_items: Set[KindAndKey] = set()
+
+        for kind in [FEATURES, SEGMENTS]:
+            old_items = old_data.get(kind, {})
+            new_items = new_data.get(kind, {})
+
+            keys: Set[str] = set()
+
+            for key in keys.union(old_items.keys(), new_items.keys()):
+                old_item = old_items.get(key)
+                new_item = new_items.get(key)
+
+                if old_item is None and new_item is None:
+                    continue
+
+                if old_item is None or new_item is None or old_item['version'] < new_item['version']:
+                    self.__tracker.add_affected_items(affected_items, KindAndKey(kind=kind, key=key))
+
+        return affected_items
+
 
 class DataSourceStatusProviderImpl(DataSourceStatusProvider):
     def __init__(self, listeners: Listeners, updates_sink: DataSourceUpdateSinkImpl):

ldclient/impl/dependency_tracker.py

@@ -0,0 +1,119 @@
+from ldclient.impl.model.feature_flag import FeatureFlag
+from ldclient.impl.model.segment import Segment
+from ldclient.impl.model.clause import Clause
+from ldclient.versioned_data_kind import VersionedDataKind, SEGMENTS, FEATURES
+
+from typing import Set, List, Dict, NamedTuple, Union, Optional
+
+
+class KindAndKey(NamedTuple):
+    kind: VersionedDataKind
+    key: str
+
+
+class DependencyTracker:
+    """
+    The DependencyTracker is responsible for tracking both up and downstream
+    dependency relationships. Managing a bi-directional mapping allows us to
+    more easily perform updates to the tracker, and to determine affected items
+    when a downstream item is modified.
+    """
+
+    def __init__(self):
+        self.__children: Dict[KindAndKey, Set[KindAndKey]] = {}
+        self.__parents: Dict[KindAndKey, Set[KindAndKey]] = {}
+
+    def update_dependencies_from(self, from_kind: VersionedDataKind, from_key: str, from_item: Optional[Union[dict, FeatureFlag, Segment]]):
+        """
+        Updates the dependency graph when an item has changed.
+
+        :param from_kind: the changed item's kind
+        :param from_key: the changed item's key
+        :param from_item: the changed item
+
+        """
+        from_what = KindAndKey(kind=from_kind, key=from_key)
+        updated_dependencies = DependencyTracker.compute_dependencies_from(from_kind, from_item)
+
+        old_children_set = self.__children.get(from_what)
+
+        if old_children_set is not None:
+            for kind_and_key in old_children_set:
+                parents_of_this_old_dep = self.__parents.get(kind_and_key, set())
+                if from_what in parents_of_this_old_dep:
+                    parents_of_this_old_dep.remove(from_what)
+
+        self.__children[from_what] = updated_dependencies
+        for kind_and_key in updated_dependencies:
+            parents_of_this_new_dep = self.__parents.get(kind_and_key)
+            if parents_of_this_new_dep is None:
+                parents_of_this_new_dep = set()
+                self.__parents[kind_and_key] = parents_of_this_new_dep
+
+            parents_of_this_new_dep.add(from_what)
+
+    def add_affected_items(self, items_out: Set[KindAndKey], initial_modified_item: KindAndKey):
+        """
+
+        Populates the given set with the union of the initial item and all items that directly or indirectly
+        depend on it (based on the current state of the dependency graph).
+
+        @param items_out [Set]
+        @param initial_modified_item [Object]
+
+        """
+
+        if initial_modified_item in items_out:
+            return
+
+        items_out.add(initial_modified_item)
+
+        parents = self.__parents.get(initial_modified_item)
+        if parents is None:
+            return
+
+        for parent in parents:
+            self.add_affected_items(items_out, parent)
+
+    def reset(self):
+        """
+        Clear any tracked dependencies and reset the tracking state to a clean slate.
+        """
+        self.__children.clear()
+        self.__parents.clear()
+
+    @staticmethod
+    def compute_dependencies_from(from_kind: VersionedDataKind, from_item: Optional[Union[dict, FeatureFlag, Segment]]) -> Set[KindAndKey]:
+        """
+        @param from_kind [String]
+        @param from_item [LaunchDarkly::Impl::Model::FeatureFlag, LaunchDarkly::Impl::Model::Segment]
+        @return [Set]
+        """
+        if from_item is None:
+            return set()
+
+        from_item = from_kind.decode(from_item) if isinstance(from_item, dict) else from_item
+
+        if from_kind == FEATURES and isinstance(from_item, FeatureFlag):
+            prereq_keys = [KindAndKey(kind=from_kind, key=p.key) for p in from_item.prerequisites]
+            segment_keys = [kindAndKey for rule in from_item.rules for kindAndKey in DependencyTracker.segment_keys_from_clauses(rule.clauses)]
+
+            results = set(prereq_keys)
+            results.update(segment_keys)
+
+            return results
+        elif from_kind == SEGMENTS and isinstance(from_item, Segment):
+            kind_and_keys = [key for rule in from_item.rules for key in DependencyTracker.segment_keys_from_clauses(rule.clauses)]
+            return set(kind_and_keys)
+        else:
+            return set()
+
+    @staticmethod
+    def segment_keys_from_clauses(clauses: List[Clause]) -> List[KindAndKey]:
+        results = []
+        for clause in clauses:
+            if clause.op == 'segmentMatch':
+                pairs = [KindAndKey(kind=SEGMENTS, key=value) for value in clause.values]
+                results.extend(pairs)
+
+        return results

ldclient/impl/flag_tracker.py

@@ -0,0 +1,50 @@
+from ldclient.interfaces import FlagTracker, FlagChange, FlagValueChange
+from ldclient.impl.listeners import Listeners
+from ldclient.context import Context
+from ldclient.impl.rwlock import ReadWriteLock
+
+from typing import Callable
+
+
+class FlagValueChangeListener:
+    def __init__(self, key: str, context: Context, listener: Callable[[FlagValueChange], None], eval_fn: Callable):
+        self.__key = key
+        self.__context = context
+        self.__listener = listener
+        self.__eval_fn = eval_fn
+
+        self.__lock = ReadWriteLock()
+        self.__value = eval_fn(key, context)
+
+    def __call__(self, flag_change: FlagChange):
+        if flag_change.key != self.__key:
+            return
+
+        new_value = self.__eval_fn(self.__key, self.__context)
+
+        self.__lock.lock()
+        old_value, self.__value = self.__value, new_value
+        self.__lock.unlock()
+
+        if new_value == old_value:
+            return
+
+        self.__listener(FlagValueChange(self.__key, old_value, new_value))
+
+
+class FlagTrackerImpl(FlagTracker):
+    def __init__(self, listeners: Listeners, eval_fn: Callable):
+        self.__listeners = listeners
+        self.__eval_fn = eval_fn
+
+    def add_listener(self, listener: Callable[[FlagChange], None]):
+        self.__listeners.add(listener)
+
+    def remove_listener(self, listener: Callable[[FlagChange], None]):
+        self.__listeners.remove(listener)
+
+    def add_flag_value_change_listener(self, key: str, context: Context, fn: Callable[[FlagValueChange], None]) -> Callable[[FlagChange], None]:
+        listener = FlagValueChangeListener(key, context, fn, self.__eval_fn)
+        self.add_listener(listener)
+
+        return listener

ldclient/impl/listeners.py

@@ -3,26 +3,32 @@
 from threading import RLock
 from typing import Any, Callable
 
+
 class Listeners:
     """
     Simple abstraction for a list of callbacks that can receive a single value. Callbacks are
     done synchronously on the caller's thread.
     """
+
     def __init__(self):
         self.__listeners = []
         self.__lock = RLock()
-    
+
+    def has_listeners(self) -> bool:
+        with self.__lock:
+            return len(self.__listeners) > 0
+
     def add(self, listener: Callable):
         with self.__lock:
             self.__listeners.append(listener)
-    
+
     def remove(self, listener: Callable):
         with self.__lock:
             try:
                 self.__listeners.remove(listener)
             except ValueError:
-                pass # removing a listener that wasn't in the list is a no-op
-    
+                pass  # removing a listener that wasn't in the list is a no-op
+
     def notify(self, value: Any):
         with self.__lock:
             listeners_copy = self.__listeners.copy()