Merge branch 'main' into jb/sdk-2922/float-v3-contract-tests

Author: jsonbailey

ldclient/async_config.py

@@ -8,6 +8,7 @@
     compatibility guarantees.
 """
 
+from dataclasses import dataclass
 from typing import Callable, List, Optional, Set
 
 from ldclient.async_feature_store import AsyncInMemoryFeatureStore
@@ -17,8 +18,8 @@
     DEFAULT_STREAM_URI,
     GET_LATEST_FEATURES_PATH,
     STREAM_FLAGS_PATH,
+    DataSourceBuilder,
     DataSourceBuilderConfig,
-    DataSystemConfig,
     HTTPConfig,
     PrivateAttributesConfig
 )
@@ -34,7 +35,10 @@
     AsyncDataSourceUpdateSink,
     AsyncEventProcessor,
     AsyncFeatureStore,
-    AsyncUpdateProcessor
+    AsyncInitializer,
+    AsyncSynchronizer,
+    AsyncUpdateProcessor,
+    DataStoreMode
 )
 from ldclient.plugin import AsyncPlugin
 
@@ -91,6 +95,39 @@ def stale_after(self) -> float:
         return self.__stale_after
 
 
+@dataclass(frozen=True)
+class AsyncDataSystemConfig:
+    """Configuration for the async SDK's data acquisition strategy.
+
+    This mirrors :class:`ldclient.config.DataSystemConfig` for the async client.
+    Its data sources are async builders and its data store is an async store.
+
+    .. caution::
+        This feature is experimental and should NOT be considered ready for production
+        use. It may change or be removed without notice and is not subject to backwards
+        compatibility guarantees.
+    """
+
+    initializers: Optional[List[DataSourceBuilder[AsyncInitializer]]] = None
+    """The initializers for the data system."""
+
+    synchronizers: Optional[List[DataSourceBuilder[AsyncSynchronizer]]] = None
+    """
+    The synchronizers for the data system, ordered by preference.
+    The first synchronizer is the most preferred, with subsequent synchronizers
+    serving as fallbacks in order of decreasing preference.
+    """
+
+    data_store_mode: DataStoreMode = DataStoreMode.READ_WRITE
+    """The data store mode specifies the mode in which the persistent store will operate, if present."""
+
+    data_store: Optional[AsyncFeatureStore] = None
+    """The (optional) async persistent data store instance."""
+
+    fdv1_fallback_synchronizer: Optional[DataSourceBuilder[AsyncSynchronizer]] = None
+    """An optional fallback synchronizer that will read from FDv1"""
+
+
 class AsyncConfig(DataSourceBuilderConfig, PrivateAttributesConfig):
     """Advanced configuration options for the async SDK client.
 
@@ -138,7 +175,7 @@ def __init__(
         enable_event_compression: bool = False,
         omit_anonymous_contexts: bool = False,
         payload_filter_key: Optional[str] = None,
-        datasystem_config: Optional[DataSystemConfig] = None,
+        datasystem_config: Optional[AsyncDataSystemConfig] = None,
     ):
         """
         :param sdk_key: The SDK key for your LaunchDarkly account. This is always required.
@@ -466,7 +503,7 @@ def data_source_update_sink(self) -> Optional[AsyncDataSourceUpdateSink]:
         return self._data_source_update_sink
 
     @property
-    def datasystem_config(self) -> Optional[DataSystemConfig]:
+    def datasystem_config(self) -> Optional[AsyncDataSystemConfig]:
         """
         Configuration for the upcoming enhanced data system design. This is
         experimental and should not be set without direction from LaunchDarkly

ldclient/client.py

@@ -4,13 +4,12 @@
 
 import threading
 import traceback
-from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple
+from typing import Any, Callable, List, Optional, Tuple
 from uuid import uuid4
 
 from ldclient.config import Config
 from ldclient.context import Context
 from ldclient.evaluation import EvaluationDetail, FeatureFlagsState
-from ldclient.feature_store import _FeatureStoreDataSetSorter
 from ldclient.hook import (
     EvaluationSeriesContext,
     Hook,
@@ -24,15 +23,7 @@
 from ldclient.impl.client_common import secure_mode_hash as _secure_mode_hash
 from ldclient.impl.datasource.feature_requester import FeatureRequesterImpl
 from ldclient.impl.datasource.polling import PollingUpdateProcessor
-from ldclient.impl.datasource.status import (
-    DataSourceStatusProviderImpl,
-    DataSourceUpdateSinkImpl
-)
 from ldclient.impl.datasource.streaming import StreamingUpdateProcessor
-from ldclient.impl.datastore.status import (
-    DataStoreStatusProviderImpl,
-    DataStoreUpdateSinkImpl
-)
 from ldclient.impl.datasystem import DataAvailability, DataSystem
 from ldclient.impl.datasystem.fdv2 import FDv2
 from ldclient.impl.evaluator import Evaluator, error_reason
@@ -43,146 +34,23 @@
 from ldclient.impl.events.event_processor import DefaultEventProcessor
 from ldclient.impl.events.types import EventFactory
 from ldclient.impl.flag_tracker import FlagTrackerImpl
-from ldclient.impl.listeners import Listeners
 from ldclient.impl.model.feature_flag import FeatureFlag
-from ldclient.impl.repeating_task import RepeatingTask
 from ldclient.impl.rwlock import ReadWriteLock
 from ldclient.impl.stubs import NullEventProcessor, NullUpdateProcessor
 from ldclient.impl.util import check_uwsgi, log
 from ldclient.interfaces import (
     BigSegmentStoreStatusProvider,
     DataSourceStatusProvider,
-    DataStoreStatus,
     DataStoreStatusProvider,
-    DataStoreUpdateSink,
-    FeatureStore,
-    FlagTracker,
-    ReadOnlyStore
+    FlagTracker
 )
 from ldclient.migrations import OpTracker, Stage
 from ldclient.plugin import EnvironmentMetadata
-from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind
+from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
 from .impl import AnyNum
 
 
-class _FeatureStoreClientWrapper(FeatureStore):
-    """Provides additional behavior that the client requires before or after feature store operations.
-    Currently this just means sorting the data set for init() and dealing with data store status listeners.
-    """
-
-    def __init__(self, store: FeatureStore, store_update_sink: DataStoreUpdateSink):
-        self.store = store
-        self.__store_update_sink = store_update_sink
-        self.__monitoring_enabled = self.is_monitoring_enabled()
-
-        # Covers the following variables
-        self.__lock = ReadWriteLock()
-        self.__last_available = True
-        self.__poller: Optional[RepeatingTask] = None
-
-    def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, Dict[Any, Any]]]):
-        return self.__wrapper(lambda: self.store.init(_FeatureStoreDataSetSorter.sort_all_collections(all_data)))
-
-    def get(self, kind, key, callback):
-        return self.__wrapper(lambda: self.store.get(kind, key, callback))
-
-    def all(self, kind, callback):
-        return self.__wrapper(lambda: self.store.all(kind, callback))
-
-    def delete(self, kind, key, version):
-        return self.__wrapper(lambda: self.store.delete(kind, key, version))
-
-    def upsert(self, kind, item):
-        return self.__wrapper(lambda: self.store.upsert(kind, item))
-
-    @property
-    def initialized(self) -> bool:
-        return self.store.initialized
-
-    def __wrapper(self, fn: Callable):
-        try:
-            return fn()
-        except BaseException:
-            if self.__monitoring_enabled:
-                self.__update_availability(False)
-            raise
-
-    def __update_availability(self, available: bool):
-        with self.__lock.write():
-            if available == self.__last_available:
-                return
-            self.__last_available = available
-
-        status = DataStoreStatus(available, False)
-
-        if available:
-            log.warn("Persistent store is available again")
-
-        self.__store_update_sink.update_status(status)
-
-        if available:
-            with self.__lock.write():
-                if self.__poller is not None:
-                    self.__poller.stop()
-                    self.__poller = None
-
-            return
-
-        log.warn("Detected persistent store unavailability; updates will be cached until it recovers")
-        task = RepeatingTask("ldclient.check-availability", 0.5, 0, self.__check_availability)
-
-        with self.__lock.write():
-            self.__poller = task
-            self.__poller.start()
-
-    def __check_availability(self):
-        try:
-            if self.store.is_available():
-                self.__update_availability(True)
-        except BaseException as e:
-            log.error("Unexpected error from data store status function: %s", e)
-
-    def is_monitoring_enabled(self) -> bool:
-        """
-        This methods determines whether the wrapped store can support enabling monitoring.
-
-        The wrapped store must provide a monitoring_enabled method, which must
-        be true. But this alone is not sufficient.
-
-        Because this class wraps all interactions with a provided store, it can
-        technically "monitor" any store. However, monitoring also requires that
-        we notify listeners when the store is available again.
-
-        We determine this by checking the store's `available?` method, so this
-        is also a requirement for monitoring support.
-
-        These extra checks won't be necessary once `available` becomes a part
-        of the core interface requirements and this class no longer wraps every
-        feature store.
-        """
-
-        if not hasattr(self.store, 'is_monitoring_enabled'):
-            return False
-
-        if not hasattr(self.store, 'is_available'):
-            return False
-
-        monitoring_enabled = getattr(self.store, 'is_monitoring_enabled')
-        if not callable(monitoring_enabled):
-            return False
-
-        return monitoring_enabled()
-
-
-def _get_store_item(store, kind: VersionedDataKind, key: str) -> Any:
-    # This decorator around store.get provides backward compatibility with any custom data
-    # store implementation that might still be returning a dict, instead of our data model
-    # classes like FeatureFlag.
-    item = store.get(kind, key, lambda x: x)
-    return kind.decode(item) if isinstance(item, dict) else item
-
-
 class LDClient:
     """The LaunchDarkly SDK client object.
 
@@ -268,8 +136,8 @@ def __start_up(self, start_wait: float):
         self.__big_segment_store_manager = big_segment_store_manager
 
         self._evaluator = Evaluator(
-            lambda key: _get_store_item(self._data_system.store, FEATURES, key),
-            lambda key: _get_store_item(self._data_system.store, SEGMENTS, key),
+            lambda key: self._data_system.store.get(FEATURES, key),
+            lambda key: self._data_system.store.get(SEGMENTS, key),
             lambda key: big_segment_store_manager.get_user_membership(key),
             log,
         )
@@ -554,7 +422,7 @@ def _evaluate_internal(self, key: str, context: Context, default: Any, event_fac
             return EvaluationDetail(default, None, error_reason('USER_NOT_SPECIFIED')), None
 
         try:
-            flag = _get_store_item(self._data_system.store, FEATURES, key)
+            flag = self._data_system.store.get(FEATURES, key)
         except Exception as e:
             log.error("Unexpected error while retrieving feature flag \"%s\": %s" % (key, repr(e)))
             log.debug(traceback.format_exc())

ldclient/config.py

@@ -4,6 +4,7 @@
 Note that the same class can also be imported from the ``ldclient.client`` submodule.
 """
 
+import copy
 import warnings
 from dataclasses import dataclass
 from threading import Event
@@ -472,6 +473,25 @@ def copy_with_new_sdk_key(self, new_sdk_key: str) -> 'Config':
             big_segments=self.__big_segments,
         )
 
+    def with_wrapper_information(self, wrapper_name: Optional[str], wrapper_version: Optional[str] = None) -> 'Config':
+        """Returns a new ``Config`` instance that is the same as this one, except for having different wrapper information.
+
+        This is intended for use by wrapper libraries, such as the LaunchDarkly OpenFeature providers, which need to
+        identify themselves rather than the application that configured the client.
+
+        The new instance is a shallow copy: it shares the objects the original configuration references, such as the
+        feature store, the logger, and the HTTP configuration. Mutating one of those objects affects both
+        configurations.
+
+        :param wrapper_name: an identifying name for the wrapper being used; see :py:attr:`~wrapper_name`
+        :param wrapper_version: the version of the wrapper being used; see :py:attr:`~wrapper_version`
+        """
+        updated = copy.copy(self)
+        updated.__wrapper_name = wrapper_name
+        updated.__wrapper_version = wrapper_version
+
+        return updated
+
     # for internal use only - probably should be part of the client logic
     def get_default(self, key, default):
         return default if key not in self.__defaults else self.__defaults[key]

ldclient/impl/datasourcev2/async_polling.py

@@ -242,9 +242,12 @@ async def close(self) -> None:
         await self._http.close()
 
 
-class AsyncPollingDataSourceBuilder(DataSourceBuilder):
+class AsyncPollingDataSourceBuilder(DataSourceBuilder[AsyncPollingDataSource]):
     """
     Builder for a AsyncPollingDataSource.
+
+    The built polling data source implements both :class:`AsyncInitializer` and
+    :class:`AsyncSynchronizer`, so this builder can be used in either role.
     """
 
     def __init__(self):
@@ -298,7 +301,7 @@ def build(self, config: DataSourceBuilderConfig) -> AsyncPollingDataSource:
         )
 
 
-class AsyncFallbackToFDv1PollingDataSourceBuilder(DataSourceBuilder):
+class AsyncFallbackToFDv1PollingDataSourceBuilder(DataSourceBuilder[AsyncPollingDataSource]):
     """
     Builder for a AsyncPollingDataSource that falls back to Flag Delivery v1.
     """

ldclient/impl/datasourcev2/async_streaming.py

@@ -284,7 +284,7 @@ async def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[O
         return (decision.update, decision.should_continue)
 
 
-class AsyncStreamingDataSourceBuilder(DataSourceBuilder):
+class AsyncStreamingDataSourceBuilder(DataSourceBuilder[AsyncStreamingDataSource]):
     """
     Builder for a AsyncStreamingDataSource.
     """

ldclient/impl/datastore/status.py

@@ -12,7 +12,7 @@
 )
 
 if TYPE_CHECKING:
-    from ldclient.client import _FeatureStoreClientWrapper
+    from ldclient.impl.datasystem.fdv1 import _FeatureStoreClientWrapper
 
 
 class DataStoreUpdateSinkImpl(DataStoreUpdateSink):