diff --git a/ldclient/async_config.py b/ldclient/async_config.py index 9a37a133..2af0a1fb 100644 --- a/ldclient/async_config.py +++ b/ldclient/async_config.py @@ -1,6 +1,6 @@ """ -Configuration types for the async SDK client. Currently this holds -:class:`AsyncBigSegmentsConfig`; the full :class:`AsyncConfig` lands in a later slice. +This submodule contains the :class:`AsyncConfig` class for custom configuration of +the async SDK client. .. caution:: This feature is experimental and should NOT be considered ready for production @@ -8,9 +8,35 @@ compatibility guarantees. """ -from typing import Optional +from typing import Callable, List, Optional, Set -from ldclient.interfaces import AsyncBigSegmentStore +from ldclient.async_feature_store import AsyncInMemoryFeatureStore +from ldclient.config import ( + DEFAULT_BASE_URI, + DEFAULT_EVENTS_URI, + DEFAULT_STREAM_URI, + GET_LATEST_FEATURES_PATH, + STREAM_FLAGS_PATH, + DataSourceBuilderConfig, + DataSystemConfig, + HTTPConfig, + PrivateAttributesConfig +) +from ldclient.hook import AsyncHook +from ldclient.impl.aio.concurrency import AsyncEvent +from ldclient.impl.util import ( + log, + validate_application_info, + validate_sdk_key_format +) +from ldclient.interfaces import ( + AsyncBigSegmentStore, + AsyncDataSourceUpdateSink, + AsyncEventProcessor, + AsyncFeatureStore, + AsyncUpdateProcessor +) +from ldclient.plugin import AsyncPlugin class AsyncBigSegmentsConfig: @@ -63,3 +89,394 @@ def status_poll_interval(self) -> float: @property def stale_after(self) -> float: return self.__stale_after + + +class AsyncConfig(DataSourceBuilderConfig, PrivateAttributesConfig): + """Advanced configuration options for the async SDK client. + + To use these options, create an instance of ``AsyncConfig`` and pass it to the + :class:`ldclient.async_client.AsyncLDClient` constructor. + + .. 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. + """ + + def __init__( + self, + sdk_key: str, + base_uri: str = DEFAULT_BASE_URI, + events_uri: str = DEFAULT_EVENTS_URI, + events_max_pending: int = 10000, + flush_interval: float = 5, + stream_uri: str = DEFAULT_STREAM_URI, + stream: bool = True, + initial_reconnect_delay: float = 1, + defaults: dict = {}, + send_events: Optional[bool] = None, + update_processor_class: Optional[Callable[['AsyncConfig', AsyncFeatureStore, AsyncEvent], AsyncUpdateProcessor]] = None, + poll_interval: float = 30, + use_ldd: bool = False, + feature_store: Optional[AsyncFeatureStore] = None, + feature_requester_class=None, + event_processor_class: Optional[Callable[['AsyncConfig'], AsyncEventProcessor]] = None, + private_attributes: Set[str] = set(), + all_attributes_private: bool = False, + offline: bool = False, + context_keys_capacity: int = 1000, + context_keys_flush_interval: float = 300, + diagnostic_opt_out: bool = False, + diagnostic_recording_interval: int = 900, + wrapper_name: Optional[str] = None, + wrapper_version: Optional[str] = None, + http: HTTPConfig = HTTPConfig(), + big_segments: Optional[AsyncBigSegmentsConfig] = None, + application: Optional[dict] = None, + hooks: Optional[List[AsyncHook]] = None, + plugins: Optional[List[AsyncPlugin]] = None, + enable_event_compression: bool = False, + omit_anonymous_contexts: bool = False, + payload_filter_key: Optional[str] = None, + datasystem_config: Optional[DataSystemConfig] = None, + ): + """ + :param sdk_key: The SDK key for your LaunchDarkly account. This is always required. + :param base_uri: The base URL for the LaunchDarkly server. Most users should use the default + value. + :param events_uri: The URL for the LaunchDarkly events server. Most users should use the + default value. + :param events_max_pending: The capacity of the events buffer. The client buffers up to this many + events in memory before flushing. If the capacity is exceeded before the buffer is flushed, events + will be discarded. + :param flush_interval: The number of seconds in between flushes of the events buffer. Decreasing + the flush interval means that the event buffer is less likely to reach capacity. + :param stream_uri: The URL for the LaunchDarkly streaming events server. Most users should + use the default value. + :param stream: Whether or not the streaming API should be used to receive flag updates. By + default, it is enabled. Streaming should only be disabled on the advice of LaunchDarkly support. + :param initial_reconnect_delay: The initial reconnect delay (in seconds) for the streaming + connection. The streaming service uses a backoff algorithm (with jitter) every time the connection needs + to be reestablished. The delay for the first reconnection will start near this value, and then + increase exponentially for any subsequent connection failures. + :param send_events: Whether or not to send events back to LaunchDarkly. This differs from + ``offline`` in that it affects only the sending of client-side events, not streaming or polling for + events from the server. By default, events will be sent. + :param offline: Whether the client should be initialized in offline mode. In offline mode, + default values are returned for all flags and no remote network requests are made. By default, + this is false. + :param poll_interval: The number of seconds between polls for flag updates if streaming is off. + :param use_ldd: Whether you are using the LaunchDarkly Relay Proxy in daemon mode. In this + configuration, the client will not use a streaming connection to listen for updates, but instead + will get feature state from a Redis instance. The ``stream`` and ``poll_interval`` options will be + ignored if this option is set to true. By default, this is false. + For more information, read the LaunchDarkly + documentation: https://docs.launchdarkly.com/home/relay-proxy/using#using-daemon-mode + :param array private_attributes: Marks a set of attributes private. Any users sent to LaunchDarkly + with this configuration active will have these attributes removed. Each item can be either the + name of an attribute ("email"), or a slash-delimited path ("/address/street") to mark a + property within a JSON object value as private. + :param all_attributes_private: If true, all user attributes (other than the key) will be + private, not just the attributes specified in ``private_attributes``. + :param feature_store: An AsyncFeatureStore implementation + :param context_keys_capacity: The number of context keys that the event processor can remember at any + one time, so that duplicate context details will not be sent in analytics events. + :param context_keys_flush_interval: The interval in seconds at which the event processor will + reset its set of known context keys. + :param feature_requester_class: A factory for a FeatureRequester implementation taking the sdk key and config + :param event_processor_class: A factory for an AsyncEventProcessor implementation taking the config + :param update_processor_class: A factory for an UpdateProcessor implementation taking the config, an + AsyncFeatureStore implementation, and an AsyncEvent to signal readiness. + :param diagnostic_opt_out: Unless this field is set to True, the client will send + some diagnostics data to the LaunchDarkly servers in order to assist in the development of future SDK + improvements. These diagnostics consist of an initial payload containing some details of SDK in use, + the SDK's configuration, and the platform the SDK is being run on, as well as periodic information + on irregular occurrences such as dropped events. + :param diagnostic_recording_interval: The interval in seconds at which periodic diagnostic data is + sent. The default is 900 seconds (every 15 minutes) and the minimum value is 60 seconds. + :param wrapper_name: For use by wrapper libraries to set an identifying name for the wrapper + being used. This will be sent in HTTP headers during requests to the LaunchDarkly servers to allow + recording metrics on the usage of these wrapper libraries. + :param wrapper_version: For use by wrapper libraries to report the version of the library in + use. If ``wrapper_name`` is not set, this field will be ignored. Otherwise the version string will + be included in the HTTP headers along with the ``wrapper_name`` during requests to the LaunchDarkly + servers. + :param http: Optional properties for customizing the client's HTTP/HTTPS behavior. See + :class:`ldclient.config.HTTPConfig`. + :param application: Optional properties for setting application metadata. See :py:attr:`~application` + :param hooks: Hooks provide entrypoints which allow for observation of SDK functions. + :param plugins: A list of plugins to be used with the SDK. Plugin support is currently experimental and subject to change. + :param enable_event_compression: Whether or not to enable GZIP compression for outgoing events. + :param omit_anonymous_contexts: Sets whether anonymous contexts should be omitted from index and identify events. + :param payload_filter_key: The payload filter is used to selectively limited the flags and segments delivered in the data source payload. + :param datasystem_config: Configuration for the upcoming enhanced data system design. This is experimental and should not be set without direction from LaunchDarkly support. + """ + self.__sdk_key = validate_sdk_key_format(sdk_key, log) + + self.__base_uri = base_uri.rstrip('/') + self.__events_uri = events_uri.rstrip('/') + self.__stream_uri = stream_uri.rstrip('/') + self.__update_processor_class = update_processor_class + self.__stream = stream + self.__initial_reconnect_delay = initial_reconnect_delay + self.__poll_interval = max(poll_interval, 30.0) + self.__use_ldd = use_ldd + self.__feature_store = AsyncInMemoryFeatureStore() if not feature_store else feature_store + self.__event_processor_class = event_processor_class + self.__feature_requester_class = feature_requester_class + self.__events_max_pending = events_max_pending + self.__flush_interval = flush_interval + self.__defaults = defaults + if offline is True: + send_events = False + self.__send_events = True if send_events is None else send_events + self.__private_attributes = private_attributes + self.__all_attributes_private = all_attributes_private + self.__offline = offline + self.__context_keys_capacity = context_keys_capacity + self.__context_keys_flush_interval = context_keys_flush_interval + self.__diagnostic_opt_out = diagnostic_opt_out + self.__diagnostic_recording_interval = max(diagnostic_recording_interval, 60) + self.__wrapper_name = wrapper_name + self.__wrapper_version = wrapper_version + self.__http = http + self.__big_segments = AsyncBigSegmentsConfig() if not big_segments else big_segments + self.__application = validate_application_info(application or {}, log) + self.__hooks = [hook for hook in hooks if isinstance(hook, AsyncHook)] if hooks else [] + self.__plugins = [plugin for plugin in plugins if isinstance(plugin, AsyncPlugin)] if plugins else [] + self.__enable_event_compression = enable_event_compression + self.__omit_anonymous_contexts = omit_anonymous_contexts + self.__payload_filter_key = payload_filter_key + self._data_source_update_sink: Optional[AsyncDataSourceUpdateSink] = None + self.__instance_id: Optional[str] = None + self._datasystem_config = datasystem_config + + # 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] + + @property + def sdk_key(self) -> Optional[str]: + return self.__sdk_key + + @property + def base_uri(self) -> str: + return self.__base_uri + + # for internal use only - also no longer used, will remove + @property + def get_latest_flags_uri(self): + return self.__base_uri + GET_LATEST_FEATURES_PATH + + # for internal use only + @property + def events_base_uri(self): + return self.__events_uri + + # for internal use only - should construct the URL path in the events code, not here + @property + def events_uri(self): + return self.__events_uri + '/bulk' + + # for internal use only + @property + def stream_base_uri(self) -> str: + return self.__stream_uri + + # for internal use only - should construct the URL path in the streaming code, not here + @property + def stream_uri(self): + return self.__stream_uri + STREAM_FLAGS_PATH + + @property + def update_processor_class(self) -> Optional[Callable[['AsyncConfig', AsyncFeatureStore, AsyncEvent], AsyncUpdateProcessor]]: + return self.__update_processor_class + + @property + def stream(self) -> bool: + return self.__stream + + @property + def initial_reconnect_delay(self) -> float: + return self.__initial_reconnect_delay + + @property + def poll_interval(self) -> float: + return self.__poll_interval + + @property + def use_ldd(self) -> bool: + return self.__use_ldd + + @property + def feature_store(self) -> AsyncFeatureStore: + return self.__feature_store + + @property + def event_processor_class(self) -> Optional[Callable[['AsyncConfig'], AsyncEventProcessor]]: + return self.__event_processor_class + + @property + def feature_requester_class(self) -> Optional[Callable]: + return self.__feature_requester_class + + @property + def send_events(self) -> bool: + return self.__send_events + + @property + def events_max_pending(self) -> int: + return self.__events_max_pending + + @property + def flush_interval(self) -> float: + return self.__flush_interval + + @property + def private_attributes(self) -> List[str]: + return list(self.__private_attributes) + + @property + def all_attributes_private(self) -> bool: + return self.__all_attributes_private + + @property + def offline(self) -> bool: + return self.__offline + + @property + def context_keys_capacity(self) -> int: + return self.__context_keys_capacity + + @property + def context_keys_flush_interval(self) -> float: + return self.__context_keys_flush_interval + + @property + def diagnostic_opt_out(self) -> bool: + return self.__diagnostic_opt_out + + @property + def diagnostic_recording_interval(self) -> int: + return self.__diagnostic_recording_interval + + @property + def wrapper_name(self) -> Optional[str]: + return self.__wrapper_name + + @property + def wrapper_version(self) -> Optional[str]: + return self.__wrapper_version + + @property + def http(self) -> HTTPConfig: + return self.__http + + @property + def big_segments(self) -> AsyncBigSegmentsConfig: + return self.__big_segments + + @property + def application(self) -> dict: + """ + An object that allows configuration of application metadata. + + Application metadata may be used in LaunchDarkly analytics or other + product features, but does not affect feature flag evaluations. + + If you want to set non-default values for any of these fields, provide + the appropriately configured dict to the {AsyncConfig} object. + """ + return self.__application + + @property + def hooks(self) -> List[AsyncHook]: + """ + Initial set of hooks for the client. + + Hooks provide entrypoints which allow for observation of SDK functions. + + LaunchDarkly provides integration packages, and most applications will + not need to implement their own hooks. + """ + return self.__hooks + + @property + def plugins(self) -> List[AsyncPlugin]: + """ + Initial set of plugins for the client. + + LaunchDarkly provides plugin packages, and most applications will + not need to implement their own plugins. + """ + return self.__plugins + + @property + def enable_event_compression(self) -> bool: + return self.__enable_event_compression + + @property + def omit_anonymous_contexts(self) -> bool: + """ + Determines whether or not anonymous contexts will be omitted from index and identify events. + """ + return self.__omit_anonymous_contexts + + @property + def payload_filter_key(self) -> Optional[str]: + """ + LaunchDarkly Server SDKs historically downloaded all flag configuration + and segments for a particular environment during initialization. + + For some customers, this is an unacceptably large amount of data, and + has contributed to performance issues within their products. + + Filtered environments aim to solve this problem. By allowing customers + to specify subsets of an environment's flags using a filter key, SDKs + will initialize faster and use less memory. + + This payload filter key only applies to the default streaming and + polling data sources. It will not affect TestData or FileData data + sources, nor will it be applied to any data source provided through the + {#data_source} config property. + """ + return self.__payload_filter_key + + @property + def _instance_id(self) -> Optional[str]: + """The instance ID included in request headers. Set by the SDK.""" + return self.__instance_id + + @_instance_id.setter + def _instance_id(self, value: Optional[str]) -> None: + self.__instance_id = value + + @property + def data_source_update_sink(self) -> Optional[AsyncDataSourceUpdateSink]: + """ + Returns the component that allows a data source to push data into the SDK. + + This property should only be set by the SDK. Long term access of this + property is not supported; it is temporarily being exposed to maintain + backwards compatibility while the SDK structure is updated. + + Custom data source implementations should integrate with this sink if + they want to provide support for data source status listeners. + """ + return self._data_source_update_sink + + @property + def datasystem_config(self) -> Optional[DataSystemConfig]: + """ + Configuration for the upcoming enhanced data system design. This is + experimental and should not be set without direction from LaunchDarkly + support. + """ + return self._datasystem_config + + def _validate(self): + if self.offline is False and self.sdk_key == '': + log.warning("Missing or blank SDK key") + + +__all__ = ['AsyncConfig', 'AsyncBigSegmentsConfig'] diff --git a/ldclient/config.py b/ldclient/config.py index e2e38ef2..a58cc70b 100644 --- a/ldclient/config.py +++ b/ldclient/config.py @@ -31,6 +31,10 @@ GET_LATEST_FEATURES_PATH = '/sdk/latest-flags' STREAM_FLAGS_PATH = '/flags' +DEFAULT_BASE_URI = 'https://app.launchdarkly.com' +DEFAULT_EVENTS_URI = 'https://events.launchdarkly.com' +DEFAULT_STREAM_URI = 'https://stream.launchdarkly.com' + class BigSegmentsConfig: """Configuration options related to Big Segments. @@ -284,11 +288,11 @@ class Config(DataSourceBuilderConfig, PrivateAttributesConfig): def __init__( self, sdk_key: str, - base_uri: str = 'https://app.launchdarkly.com', - events_uri: str = 'https://events.launchdarkly.com', + base_uri: str = DEFAULT_BASE_URI, + events_uri: str = DEFAULT_EVENTS_URI, events_max_pending: int = 10000, flush_interval: float = 5, - stream_uri: str = 'https://stream.launchdarkly.com', + stream_uri: str = DEFAULT_STREAM_URI, stream: bool = True, initial_reconnect_delay: float = 1, defaults: dict = {}, diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index 6f7db48c..e016f6ce 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -456,6 +456,37 @@ def initialized(self) -> bool: # type: ignore[empty-body] """ +class AsyncUpdateProcessor(ABC): + """ + Async interface for the component that obtains feature flag data and passes it to an + :class:`AsyncFeatureStore`, for use with :class:`ldclient.async_client.AsyncLDClient`. It mirrors + :class:`UpdateProcessor`, except ``stop`` is a coroutine. The built-in implementations are the + client's standard streaming or polling behavior. + + .. 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. + """ + + def start(self): + """ + Starts the update processor in the background. Should return immediately and not block. + """ + pass + + async def stop(self): + """ + Stops the update processor. Awaiting the result ensures background work has stopped. + """ + pass + + def initialized(self) -> bool: # type: ignore[empty-body] + """ + Returns whether the update processor has received feature flags and has initialized its feature store. + """ + + class EventProcessor(ABC): """ Interface for the component that buffers analytics events and sends them to LaunchDarkly. @@ -484,6 +515,47 @@ def stop(self): """ +class AsyncEventProcessor(ABC): + """ + Async interface for the component that buffers analytics events and sends them to LaunchDarkly, + for use with :class:`ldclient.async_client.AsyncLDClient`. It mirrors :class:`EventProcessor`, + except ``stop`` is a coroutine. The default implementation can be replaced for testing. + + .. 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. + """ + + @abstractmethod + def send_event(self, event): + """ + Processes an event to be sent at some point. + """ + + @abstractmethod + def flush(self): + """ + Specifies that any buffered events should be sent as soon as possible, rather than waiting + for the next flush interval. This method is not awaitable and does not wait for delivery; + use ``flush_and_wait`` to wait for delivery to complete. + """ + + @abstractmethod + async def flush_and_wait(self, timeout: float) -> bool: + """ + Flushes any buffered events and waits for delivery to complete, up to ``timeout`` seconds. + Returns True if delivery completed within the timeout, or False if it timed out. Unlike + ``stop``, this does not shut down the event processor. + """ + + @abstractmethod + async def stop(self): + """ + Shuts down the event processor after first delivering all pending events. + """ + + class FeatureRequester(ABC): """ Interface for the component that acquires feature flag data in polling mode. The default