From 9dcd7152cdfa1e38121b3373f78d300b63863069 Mon Sep 17 00:00:00 2001 From: tuhinkanti Date: Wed, 22 Jul 2026 15:01:59 -0700 Subject: [PATCH] feat(flagd): support custom gRPC metadata on in-process SyncFlags Add a `sync_metadata` option to FlagdProvider / Config that appends user-supplied gRPC metadata headers to every in-process SyncFlags call, alongside the provider-managed flagd-selector header. This lets callers inject infrastructure-specific headers on the long-lived sync stream -- e.g. `x-envoy-upstream-rq-timeout-ms: 0` to disable a proxy request timeout that would otherwise sever the stream after its default deadline. Keep `fatal_status_codes` as the last positional parameter in both public constructors so existing positional callers are unaffected, and add a regression test covering that ordering. Signed-off-by: Tuhin Sharma --- .../contrib/provider/flagd/config.py | 9 +++++ .../contrib/provider/flagd/provider.py | 5 +++ .../process/connector/grpc_watcher.py | 12 ++++-- .../tests/test_config.py | 18 +++++++++ .../tests/test_grpc_watcher.py | 39 +++++++++++++++++++ 5 files changed, 79 insertions(+), 4 deletions(-) diff --git a/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/config.py b/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/config.py index 75a2cb4b..d1beb694 100644 --- a/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/config.py +++ b/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/config.py @@ -105,6 +105,7 @@ def __init__( # noqa: PLR0913, PLR0915 channel_credentials: grpc.ChannelCredentials | None = None, sync_metadata_disabled: bool | None = None, fatal_status_codes: list[str] | None = None, + sync_metadata: typing.Sequence[tuple[str, str]] | None = None, ): self.host = env_or_default(ENV_VAR_HOST, DEFAULT_HOST) if host is None else host @@ -278,3 +279,11 @@ def __init__( # noqa: PLR0913, PLR0915 # Disabling will prevent static context from flagd being used in evaluations. # GetMetadata and this option will be removed. self.sync_metadata_disabled = sync_metadata_disabled + + # Additional gRPC metadata headers sent on every in-process SyncFlags call. + # Useful for injecting infrastructure-specific headers, e.g. disabling a + # proxy/mesh request timeout on the long-lived sync stream. These are merged + # with any headers the provider sets itself (such as ``flagd-selector``). + self.sync_metadata: tuple[tuple[str, str], ...] = ( + tuple(sync_metadata) if sync_metadata is not None else () + ) diff --git a/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/provider.py b/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/provider.py index 0cbadbf7..ce5e4218 100644 --- a/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/provider.py +++ b/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/provider.py @@ -66,6 +66,7 @@ def __init__( # noqa: PLR0913 channel_credentials: grpc.ChannelCredentials | None = None, sync_metadata_disabled: bool | None = None, fatal_status_codes: list[str] | None = None, + sync_metadata: typing.Sequence[tuple[str, str]] | None = None, ): """ Create an instance of the FlagdProvider @@ -83,6 +84,9 @@ def __init__( # noqa: PLR0913 :param stream_deadline_ms: the maximum time to wait before a request times out :param keep_alive_time: the number of milliseconds to keep alive :param resolver_type: the type of resolver to use + :param sync_metadata: additional gRPC metadata headers (key-value tuples) sent on + every in-process SyncFlags call, e.g. to disable a proxy/mesh + request timeout on the long-lived sync stream """ if deadline_ms is None and timeout is not None: deadline_ms = timeout * 1000 @@ -112,6 +116,7 @@ def __init__( # noqa: PLR0913 default_authority=default_authority, channel_credentials=channel_credentials, sync_metadata_disabled=sync_metadata_disabled, + sync_metadata=sync_metadata, fatal_status_codes=fatal_status_codes, ) self.enriched_context: dict = {} diff --git a/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/grpc_watcher.py b/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/grpc_watcher.py index e625ae33..8bde9217 100644 --- a/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/grpc_watcher.py +++ b/providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/grpc_watcher.py @@ -212,17 +212,21 @@ def _create_request_args(self) -> dict: return request_args - def _create_metadata(self) -> tuple[tuple[str, str]] | None: + def _create_metadata(self) -> tuple[tuple[str, str], ...] | None: """Create gRPC metadata headers for the request. Returns gRPC metadata as a tuples of tuples containing header key-value pairs. The selector is passed via the 'flagd-selector' header per flagd v0.11.0+ specification, while also being included in the request body for backward compatibility with older flagd versions. + Any user-configured ``sync_metadata`` headers are appended, allowing callers to inject + infrastructure-specific headers (e.g. proxy/mesh timeout overrides) on the sync stream. """ - if self.selector is None: - return None + metadata: tuple[tuple[str, str], ...] = () + if self.selector is not None: + metadata += (("flagd-selector", self.selector),) + metadata += self.config.sync_metadata - return (("flagd-selector", self.selector),) + return metadata if metadata else None def _fetch_metadata(self) -> sync_pb2.GetMetadataResponse | None: if self.config.sync_metadata_disabled: diff --git a/providers/openfeature-provider-flagd/tests/test_config.py b/providers/openfeature-provider-flagd/tests/test_config.py index a671f8c1..e2a57d3b 100644 --- a/providers/openfeature-provider-flagd/tests/test_config.py +++ b/providers/openfeature-provider-flagd/tests/test_config.py @@ -44,6 +44,24 @@ def test_return_default_values_rpc(): assert config.retry_backoff_ms == DEFAULT_RETRY_BACKOFF assert config.stream_deadline_ms == DEFAULT_STREAM_DEADLINE assert config.tls is DEFAULT_TLS + assert config.sync_metadata == () + + +def test_sync_metadata_passthrough(): + metadata = [("x-envoy-upstream-rq-timeout-ms", "0")] + config = Config(resolver=ResolverType.IN_PROCESS, sync_metadata=metadata) + assert config.sync_metadata == (("x-envoy-upstream-rq-timeout-ms", "0"),) + + +def test_positional_fatal_status_codes_backwards_compatible(): + # fatal_status_codes must stay the last positional parameter so callers that + # passed it positionally before sync_metadata was added keep working. + # It is the 22nd positional parameter (21 parameters precede it). + leading_args = [None] * 21 + config = Config(*leading_args, ["UNAVAILABLE", "DATA_LOSS"]) + assert config.fatal_status_codes == ["UNAVAILABLE", "DATA_LOSS"] + # The positional value must not leak into sync_metadata. + assert config.sync_metadata == () def test_return_default_values_in_process(): diff --git a/providers/openfeature-provider-flagd/tests/test_grpc_watcher.py b/providers/openfeature-provider-flagd/tests/test_grpc_watcher.py index 395dd6a2..8e855366 100644 --- a/providers/openfeature-provider-flagd/tests/test_grpc_watcher.py +++ b/providers/openfeature-provider-flagd/tests/test_grpc_watcher.py @@ -36,6 +36,7 @@ def setUp(self): config.host = "localhost" config.port = 5000 config.sync_metadata_disabled = False + config.sync_metadata = () flag_store = Mock(spec=FlagStore) flag_store.update.return_value = None @@ -160,3 +161,41 @@ def test_selector_passed_via_both_metadata_and_body(self): self.assertIn("metadata", kwargs) metadata = kwargs["metadata"] self.assertEqual(metadata, (("flagd-selector", "test-selector"),)) + + def test_custom_sync_metadata_appended(self): + """User-configured sync_metadata headers are sent on the SyncFlags call.""" + self.grpc_watcher.selector = "test-selector" + self.grpc_watcher.config.sync_metadata = ( + ("x-envoy-upstream-rq-timeout-ms", "0"), + ) + mock_stream = iter( + [SyncFlagsResponse(flag_configuration='{"flag_key": "flag_value"}')] + ) + self.mock_stub.SyncFlags = Mock(return_value=mock_stream) + + self.run_listen_and_shutdown_after() + + metadata = self.mock_stub.SyncFlags.call_args.kwargs["metadata"] + self.assertEqual( + metadata, + ( + ("flagd-selector", "test-selector"), + ("x-envoy-upstream-rq-timeout-ms", "0"), + ), + ) + + def test_custom_sync_metadata_without_selector(self): + """sync_metadata is sent even when no selector is configured.""" + self.grpc_watcher.selector = None + self.grpc_watcher.config.sync_metadata = ( + ("x-envoy-upstream-rq-timeout-ms", "0"), + ) + mock_stream = iter( + [SyncFlagsResponse(flag_configuration='{"flag_key": "flag_value"}')] + ) + self.mock_stub.SyncFlags = Mock(return_value=mock_stream) + + self.run_listen_and_shutdown_after() + + metadata = self.mock_stub.SyncFlags.call_args.kwargs["metadata"] + self.assertEqual(metadata, (("x-envoy-upstream-rq-timeout-ms", "0"),))