feat: Add experimental support for flag overrides

contract-tests/async_client_entity.py

@@ -27,6 +27,7 @@
 )
 from ldclient.impl.util import Result
 from ldclient.integrations import Redis
+from ldclient.integrations.overrides import FileOverrideSourceBuilder
 from ldclient.interfaces import DataStoreMode
 from ldclient.migrations import (
     AsyncMigratorBuilder,
@@ -82,8 +83,10 @@ async def start(self):
             opts['events_uri'] = endpoints['events']
 
         datasystem_config = config_params.get('dataSystem')
+        if config_params.get('overrides') is not None and datasystem_config is None:
+            raise ValueError("flag overrides require the data system to be configured")
         if datasystem_config is not None:
-            opts["datasystem_config"] = _build_async_data_system(datasystem_config, opts)
+            opts["datasystem_config"] = _build_async_data_system(datasystem_config, opts, config_params.get('overrides'))
         elif config_params.get("streaming") is not None:
             streaming = config_params["streaming"]
             if streaming.get("baseUri") is not None:
@@ -320,13 +323,13 @@ def _set_optional_value(params_in: dict, name_in: str, func: Callable[[Any], Any
         func(params_in[name_in])
 
 
-def _build_async_data_system(datasystem_config: dict, opts: dict) -> AsyncDataSystemConfig:
+def _build_async_data_system(datasystem_config: dict, opts: dict, overrides_config: Optional[dict] = None) -> AsyncDataSystemConfig:
     """Build an AsyncDataSystemConfig from the harness's dataSystem config.
 
     Wires the FDv2 initializers, the ordered synchronizer chain, the FDv1
-    fallback synchronizer, the payload filter, and an optional async
-    persistent store. The async client injects its shared aiohttp session
-    into these builders when it starts.
+    fallback synchronizer, the payload filter, an optional async persistent
+    store, and the optional file-based override source. The async client
+    injects its shared aiohttp session into these builders when it starts.
     """
     initializers: Optional[list] = None
     init_configs = datasystem_config.get('initializers')
@@ -379,6 +382,9 @@ def _build_async_data_system(datasystem_config: dict, opts: dict) -> AsyncDataSy
         "fdv1_fallback_synchronizer": fdv1_fallback_synchronizer,
     }
 
+    if overrides_config is not None:
+        ds_kwargs["override_source"] = _make_override_source(overrides_config)
+
     store_config = datasystem_config.get("store")
     if store_config is not None:
         persistent_store_config = store_config.get("persistentDataStore")
@@ -393,6 +399,18 @@ def _build_async_data_system(datasystem_config: dict, opts: dict) -> AsyncDataSy
     return AsyncDataSystemConfig(**ds_kwargs)
 
 
+def _make_override_source(params: dict) -> FileOverrideSourceBuilder:
+    """
+    Builds the file-based override source from the harness's overrides configuration. The
+    files are written by the harness on a file system shared with this service.
+    """
+    builder = FileOverrideSourceBuilder(params["filePaths"])
+    _set_optional_value(params, "duplicateKeysHandling", builder.duplicate_keys_handling)
+    _set_optional_value(params, "changeDetection", builder.change_detection)
+    _set_optional_time(params, "pollIntervalMs", builder.poll_interval)
+    return builder
+
+
 def _create_async_persistent_store(persistent_store_config: dict):
     """Create an async persistent feature store from the harness config.
 

contract-tests/async_service.py

@@ -69,6 +69,7 @@ async def handle_status(request: aiohttp.web.Request) -> aiohttp.web.Response:
             'migrations',
             'persistent-data-store-redis',
             'fdv1-fallback',
+            'flag-overrides',
             'retry-conformance-fdv1-streaming',
             'retry-conformance-fdv1-polling',
             'service-endpoints',

contract-tests/client_entity.py

@@ -28,6 +28,7 @@
 from ldclient.feature_store import CacheConfig
 from ldclient.impl.datasourcev2.polling import PollingDataSourceBuilder
 from ldclient.integrations import Consul, DynamoDB, Redis
+from ldclient.integrations.overrides import FileOverrideSourceBuilder
 from ldclient.interfaces import DataStoreMode
 
 
@@ -69,6 +70,8 @@ def __init__(self, tag, config):
             opts['events_uri'] = endpoints['events']
 
         datasystem_config = config.get('dataSystem')
+        if config.get('overrides') is not None and datasystem_config is None:
+            raise ValueError("flag overrides require the data system to be configured")
         if datasystem_config is not None:
             datasystem = custom()
 
@@ -132,6 +135,10 @@ def __init__(self, tag, config):
 
                     datasystem.data_store(store, store_mode)
 
+            overrides_config = config.get('overrides')
+            if overrides_config is not None:
+                datasystem.overrides(_make_override_source(overrides_config))
+
             opts["datasystem_config"] = datasystem.build()
 
         elif config.get("streaming") is not None:
@@ -352,6 +359,18 @@ def _set_optional_value(params_in: dict, name_in: str, func: Callable[[Any], Any
         func(params_in[name_in])
 
 
+def _make_override_source(params: dict) -> FileOverrideSourceBuilder:
+    """
+    Builds the file-based override source from the harness's overrides configuration. The
+    files are written by the harness on a file system shared with this service.
+    """
+    builder = FileOverrideSourceBuilder(params["filePaths"])
+    _set_optional_value(params, "duplicateKeysHandling", builder.duplicate_keys_handling)
+    _set_optional_value(params, "changeDetection", builder.change_detection)
+    _set_optional_time(params, "pollIntervalMs", builder.poll_interval)
+    return builder
+
+
 def _create_persistent_store(persistent_store_config: dict):
     """
     Creates a persistent store instance based on the configuration.

contract-tests/service.py

@@ -86,6 +86,7 @@ def status():
             'flag-change-listeners',
             'flag-value-change-listeners',
             'fdv1-fallback',
+            'flag-overrides',
             'retry-conformance-fdv1-streaming',
             'retry-conformance-fdv1-polling',
             'service-endpoints',

docs/api-integrations.rst

@@ -8,3 +8,13 @@ ldclient.integrations module
     :members:
     :special-members: __init__
     :show-inheritance:
+
+ldclient.integrations.overrides module
+--------------------------------------
+
+The entry point for this feature is :class:`ldclient.integrations.overrides.FileOverrideSourceBuilder`.
+
+.. automodule:: ldclient.integrations.overrides
+    :members:
+    :special-members: __init__
+    :show-inheritance:

ldclient/async_client.py

@@ -4,7 +4,7 @@
 
 import asyncio
 import traceback
-from typing import Any, Callable, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple
 from uuid import uuid4
 
 from ldclient.async_config import AsyncConfig
@@ -28,7 +28,7 @@
 )
 from ldclient.impl.client_common import secure_mode_hash as _secure_mode_hash
 from ldclient.impl.datasystem import AsyncDataSystem, DataAvailability
-from ldclient.impl.evaluator_common import error_reason
+from ldclient.impl.evaluator_common import error_reason, is_override_definition
 from ldclient.impl.events.async_event_processor import (
     DefaultAsyncEventProcessor
 )
@@ -95,6 +95,7 @@ def __init__(self, config: AsyncConfig):
         # these flags, so plain booleans are safe.
         self._eval_cached_data_warned = False
         self._all_flags_cached_data_warned = False
+        self._all_flags_overrides_only_warned = False
 
         # Build the object graph here (loop-free). start() supplies the loop-bound
         # resources: the HTTP session (created lazily), the data source, the
@@ -485,7 +486,12 @@ async def evaluate():
                     tracker = OpTracker(key, flag, context, detail, default_stage)
                     return _EvaluationWithHookResult(evaluation_detail=detail, results={'default_stage': stage, 'tracker': tracker})
 
-            detail = EvaluationDetail(default_stage.value, None, error_reason('WRONG_TYPE'))
+            # The type mismatch replaces the reason. The evaluation read the same definitions,
+            # so the new reason keeps the override-affected marking.
+            reason = error_reason('WRONG_TYPE')
+            if detail.reason.get('overrideAffected') is True:
+                reason['overrideAffected'] = True
+            detail = EvaluationDetail(default_stage.value, None, reason)
             tracker = OpTracker(key, flag, context, detail, default_stage)
             return _EvaluationWithHookResult(evaluation_detail=detail, results={'default_stage': default_stage, 'tracker': tracker})
 
@@ -498,30 +504,42 @@ async def _evaluate_internal(self, key: str, context: Context, default: Any, eve
         if self._config.offline:
             return EvaluationDetail(default, None, error_reason('CLIENT_NOT_READY')), None
 
+        def not_ready() -> Tuple[EvaluationDetail, Optional[FeatureFlag]]:
+            log.warning("Feature Flag evaluation attempted before client has initialized! Feature store unavailable - returning default: " + str(default) + " for feature key: " + key)
+            reason = error_reason('CLIENT_NOT_READY')
+            self._send_event(event_factory.new_unknown_flag_event(key, context, default, reason))
+            return EvaluationDetail(default, None, reason), None
+
+        flag: Any = None
         availability = await self._data_system.data_availability()
         if availability != DataAvailability.REFRESHED:
             if availability == DataAvailability.CACHED:
                 if not self._eval_cached_data_warned:
                     self._eval_cached_data_warned = True
                     log.warning("Feature Flag evaluation attempted before client has initialized - using last known values from feature store for feature key: " + key + ". This message is logged once.")
             else:
-                log.warning("Feature Flag evaluation attempted before client has initialized! Feature store unavailable - returning default: " + str(default) + " for feature key: " + key)
-                reason = error_reason('CLIENT_NOT_READY')
-                self._send_event(event_factory.new_unknown_flag_event(key, context, default, reason))
-                return EvaluationDetail(default, None, reason), None
+                # No data from LaunchDarkly is available. A flag that the override layer holds
+                # is served anyway. A flag the layer does not hold gets the not-ready default,
+                # as it does without an override source.
+                layer = self._data_system.override_layer
+                if layer is not None:
+                    flag = layer.get(FEATURES, key)
+                if flag is None:
+                    return not_ready()
 
         if not context.valid:
             log.warning("Context was invalid for flag evaluation (%s); returning default value" % context.error)
             return EvaluationDetail(default, None, error_reason('USER_NOT_SPECIFIED')), None
 
-        try:
-            flag = await 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())
-            reason = error_reason('EXCEPTION')
-            self._send_event(event_factory.new_unknown_flag_event(key, context, default, reason))
-            return EvaluationDetail(default, None, reason), None
+        if flag is None:
+            try:
+                flag = await 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())
+                reason = error_reason('EXCEPTION')
+                self._send_event(event_factory.new_unknown_flag_event(key, context, default, reason))
+                return EvaluationDetail(default, None, reason), None
         if not flag:
             reason = error_reason('FLAG_NOT_FOUND')
             self._send_event(event_factory.new_unknown_flag_event(key, context, default, reason))
@@ -534,13 +552,19 @@ async def _evaluate_internal(self, key: str, context: Context, default: Any, eve
                 detail = result.detail
                 if detail.is_default_value():
                     detail = EvaluationDetail(default, None, detail.reason)
-                self._send_event(event_factory.new_eval_event(flag, context, detail, default))
+                self._send_event(event_factory.new_eval_event(flag, context, detail, default, None, result.override_affected))
                 return detail, flag
             except Exception as e:
                 log.error("Unexpected error while evaluating feature flag \"%s\": %s" % (key, repr(e)))
                 log.debug(traceback.format_exc())
                 reason = error_reason('EXCEPTION')
-                self._send_event(event_factory.new_default_event(flag, context, default, reason))
+                # The evaluation read the flag definition before it failed. A definition from
+                # the override layer marks the result, so the failure is reported in the
+                # summary counters only, like any other override-affected evaluation.
+                override_affected = is_override_definition(flag)
+                if override_affected:
+                    reason['overrideAffected'] = True
+                self._send_event(event_factory.new_default_event(flag, context, default, reason, override_affected))
                 return EvaluationDetail(default, None, reason), flag
 
     async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState:
@@ -572,56 +596,97 @@ async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState
             log.warning("all_flags_state() called, but client is in offline mode. Returning empty state")
             return FeatureFlagsState(False)
 
+        override_flags: Optional[Dict[str, Any]] = None
         availability = await self._data_system.data_availability()
         if availability != DataAvailability.REFRESHED:
             if availability == DataAvailability.CACHED:
                 if not self._all_flags_cached_data_warned:
                     self._all_flags_cached_data_warned = True
                     log.warning("all_flags_state() called before client has finished initializing! Using last known values from feature store. This message is logged once.")
             else:
-                log.warning("all_flags_state() called before client has finished initializing! Feature store unavailable - returning empty state")
-                return FeatureFlagsState(False)
+                # No data from LaunchDarkly is available. When the override layer holds flags,
+                # the state contains only those, read from the layer itself: a store that is not
+                # initialized can still hold LaunchDarkly data, which the state must not include.
+                # Without override flags the state is unavailable, as it is without an override
+                # source.
+                layer = self._data_system.override_layer
+                override_flags = layer.all(FEATURES) if layer is not None else {}
+                if len(override_flags) == 0:
+                    log.warning("all_flags_state() called before client has finished initializing! Feature store unavailable - returning empty state")
+                    return FeatureFlagsState(False)
 
         if not context.valid:
             log.warning("Context was invalid for all_flags_state (%s); returning default value" % context.error)
             return FeatureFlagsState(False)
 
-        state = FeatureFlagsState(True)
+        valid = True
         client_only = kwargs.get('client_side_only', False)
         with_reasons = kwargs.get('with_reasons', False)
         details_only_if_tracked = kwargs.get('details_only_for_tracked_flags', False)
-        try:
-            flags_map = await self._data_system.store.all(FEATURES)
-            if flags_map is None:
-                raise ValueError("feature store error")
-        except Exception as e:
-            log.error("Unable to read flags for all_flag_state: %s" % repr(e))
-            return FeatureFlagsState(False)
-
+        if override_flags is not None:
+            if not self._all_flags_overrides_only_warned:
+                self._all_flags_overrides_only_warned = True
+                log.warning("all_flags_state() called before client has finished initializing! Returning only flags from the override layer. This message is logged once.")
+            flags_map = override_flags
+        else:
+            try:
+                flags_map = await self._data_system.store.all(FEATURES)
+                if flags_map is None:
+                    raise ValueError("feature store error")
+            except Exception as e:
+                log.error("Unable to read flags for all_flag_state: %s" % repr(e))
+                # Overrides only add to the state. Without them the state is invalid and empty
+                # here, so it holds the override layer's flags and stays invalid. A client-side
+                # SDK bootstrapped from the state uses its flags whatever $valid says (the
+                # JavaScript client SDKs 3.x and 4.x only warn), so the overridden values still
+                # reach it.
+                layer = self._data_system.override_layer
+                flags_map = layer.all(FEATURES) if layer is not None else {}
+                if len(flags_map) == 0:
+                    return FeatureFlagsState(False)
+                valid = False
+
+        state = FeatureFlagsState(valid)
         for key, flag in flags_map.items():
             if client_only and not flag.get('clientSide', False):
                 continue
             try:
                 result = await self._evaluator.evaluate(flag, context, self._event_factory_default)
                 detail = result.detail
                 prerequisites = result.prerequisites
+                override_affected = result.override_affected
             except Exception as e:
                 log.error("Error evaluating flag \"%s\" in all_flags_state: %s" % (key, repr(e)))
                 log.debug(traceback.format_exc())
-                reason = {'kind': 'ERROR', 'errorKind': 'EXCEPTION'}
+                reason = error_reason('EXCEPTION')
+                # The evaluation read the flag definition before it failed, so a definition from
+                # the override layer marks the result.
+                override_affected = is_override_definition(flag)
+                if override_affected:
+                    reason['overrideAffected'] = True
                 detail = EvaluationDetail(None, None, reason)
                 prerequisites = []
             requires_experiment_data = EventFactory.is_experiment(flag, detail.reason)
+            track_events = flag.get('trackEvents', False) or requires_experiment_data
+            track_reason = requires_experiment_data
+            debug_events_until_date = flag.get('debugEventsUntilDate', None)
+            if override_affected:
+                # A consumer of this state sends individual events according to these fields.
+                # An override-affected evaluation produces no individual events, so the state
+                # turns them off for this flag. The flag, its value, and its reason stay.
+                track_events = False
+                track_reason = False
+                debug_events_until_date = None
             flag_state = {
                 'key': flag['key'],
                 'value': detail.value,
                 'variation': detail.variation_index,
                 'reason': detail.reason,
                 'version': flag['version'],
                 'prerequisites': prerequisites,
-                'trackEvents': flag.get('trackEvents', False) or requires_experiment_data,
-                'trackReason': requires_experiment_data,
-                'debugEventsUntilDate': flag.get('debugEventsUntilDate', None),
+                'trackEvents': track_events,
+                'trackReason': track_reason,
+                'debugEventsUntilDate': debug_events_until_date,
             }
 
             state.add_flag(flag_state, with_reasons, details_only_if_tracked)