diff --git a/.github/workflows/run-end-to-end.yml b/.github/workflows/run-end-to-end.yml index 477831b4bd6..c959da4ad41 100644 --- a/.github/workflows/run-end-to-end.yml +++ b/.github/workflows/run-end-to-end.yml @@ -311,6 +311,12 @@ jobs: - name: Run FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS scenario if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS"') run: ./run.sh FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS + - name: Run FEATURE_FLAGGING_CAPABILITY_REFRESH scenario + if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_CAPABILITY_REFRESH"') + run: ./run.sh FEATURE_FLAGGING_CAPABILITY_REFRESH + - name: Run FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED scenario + if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED"') + run: ./run.sh FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED - name: Run FEATURE_FLAGGING_EVP_STRESS scenario if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_EVP_STRESS"') run: ./run.sh FEATURE_FLAGGING_EVP_STRESS diff --git a/manifests/cpp.yml b/manifests/cpp.yml index bc2ba4b839d..469a92863b0 100644 --- a/manifests/cpp.yml +++ b/manifests/cpp.yml @@ -25,6 +25,7 @@ manifest: tests/debugger/test_debugger_probe_snapshot.py::Test_Debugger_Line_Probe_Snaphots::test_process_tags_snapshot_svc: missing_feature (Not yet implemented) tests/ffe/test_evaluation_telemetry_opt_out.py: irrelevant (only parametric tests are run for cpp) tests/ffe/test_flag_eval_evp.py: irrelevant (only parametric tests are run for cpp) + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234) tests/integrations/crossed_integrations/test_sqs.py::Test_SQS_PROPAGATION_VIA_AWS_XRAY_HEADERS: irrelevant (Localstack SQS does not support AWS Xray Header parsing) tests/integrations/test_base_service.py::Test_BaseService_SqlSpan: irrelevant (/rasp/sqli endpoint is not available) diff --git a/manifests/cpp_nginx.yml b/manifests/cpp_nginx.yml index e5bb1c735d2..4513f000a60 100644 --- a/manifests/cpp_nginx.yml +++ b/manifests/cpp_nginx.yml @@ -295,6 +295,7 @@ manifest: tests/ffe/test_exposures_datadog_agent.py: missing_feature tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446) tests/ffe/test_flag_eval_metrics.py: missing_feature + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234) tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka: missing_feature tests/integrations/crossed_integrations/test_kinesis.py::Test_Kinesis_PROPAGATION_VIA_MESSAGE_ATTRIBUTES: missing_feature diff --git a/manifests/dotnet.yml b/manifests/dotnet.yml index 5a1cc7af3a1..fa308185c62 100644 --- a/manifests/dotnet.yml +++ b/manifests/dotnet.yml @@ -891,6 +891,7 @@ manifest: excluded_weblog: [poc, uds] tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446) tests/ffe/test_flag_eval_metrics.py: v3.44.0 + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234) tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka: v2.0.0-prerelease tests/integrations/crossed_integrations/test_kinesis.py::Test_Kinesis_PROPAGATION_VIA_MESSAGE_ATTRIBUTES: missing_feature diff --git a/manifests/golang.yml b/manifests/golang.yml index 81b952fa63c..bd4b2963217 100644 --- a/manifests/golang.yml +++ b/manifests/golang.yml @@ -1874,6 +1874,7 @@ manifest: tests/ffe/test_flag_eval_metrics.py: v2.8.0 tests/ffe/test_flag_eval_metrics.py::Test_FFE_Eval_Metric_Parse_Error_Invalid_Regex: irrelevant (Go validates regex at config load time) tests/ffe/test_flag_eval_metrics.py::Test_FFE_Eval_Nested_Attributes_Ignored: irrelevant (FFL-1980) + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234) tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka: - weblog_declaration: diff --git a/manifests/java.yml b/manifests/java.yml index 32e8421e217..6c7fe06eb72 100644 --- a/manifests/java.yml +++ b/manifests/java.yml @@ -3537,6 +3537,7 @@ manifest: - weblog_declaration: "*": irrelevant spring-boot: v1.63.0 + tests/ffe/test_rc_capability_refresh.py: missing_feature (Provisional dependent-flag capability coverage; enable after the SDK release) tests/integration_frameworks/llm/anthropic/test_anthropic_ai_guard.py: missing_feature (APPSEC-68977, AI Guard/Anthropic integration not yet wired into INTEGRATION_FRAMEWORKS) tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234) tests/integration_frameworks/llm/openai/test_openai_ai_guard.py: missing_feature (APPSEC-68977, AI Guard/OpenAI integration not yet wired into INTEGRATION_FRAMEWORKS) diff --git a/manifests/nodejs.yml b/manifests/nodejs.yml index 5a81d06c711..19b22b4221c 100644 --- a/manifests/nodejs.yml +++ b/manifests/nodejs.yml @@ -1993,6 +1993,7 @@ manifest: weblog: [express5] - declaration: bug (FFL-1730) excluded_weblog: [express5] + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks/llm/anthropic/test_anthropic_ai_guard.py::TestAnthropicAiGuard: # TODO: a lower version might be supported - declaration: missing_feature (APPSEC-69630, AI Guard/Anthropic integration not yet wired into INTEGRATION_FRAMEWORKS) component_version: <6.19.0 diff --git a/manifests/php.yml b/manifests/php.yml index 128eb676088..ce326bee274 100644 --- a/manifests/php.yml +++ b/manifests/php.yml @@ -1326,6 +1326,7 @@ manifest: "*": v1.21.0-dev laravel11x: incomplete_test_app symfony7x: incomplete_test_app + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234) tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka: missing_feature tests/integrations/crossed_integrations/test_kinesis.py::Test_Kinesis_PROPAGATION_VIA_MESSAGE_ATTRIBUTES: missing_feature diff --git a/manifests/python.yml b/manifests/python.yml index 76d91e0c9c6..4beb0debf76 100644 --- a/manifests/python.yml +++ b/manifests/python.yml @@ -2375,6 +2375,7 @@ manifest: - declaration: irrelevant (FFL-1980) - weblog_declaration: aiohttp: missing_feature (new weblog, tests are enabled with easy wins) + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks: - weblog_declaration: aiohttp: missing_feature (new weblog, tests are enabled with easy wins) diff --git a/manifests/ruby.yml b/manifests/ruby.yml index e7111d5e851..4f0f1ddef9f 100644 --- a/manifests/ruby.yml +++ b/manifests/ruby.yml @@ -2197,6 +2197,7 @@ manifest: - weblog_declaration: "*": irrelevant rails72: v2.32.0 + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234) tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka: - weblog_declaration: diff --git a/manifests/rust.yml b/manifests/rust.yml index 8de4581130c..128a0bc665c 100644 --- a/manifests/rust.yml +++ b/manifests/rust.yml @@ -29,6 +29,7 @@ manifest: tests/ffe/test_exposures_datadog_agent.py: missing_feature tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446) tests/ffe/test_flag_eval_metrics.py: missing_feature + tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription) tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234) tests/integrations/crossed_integrations/: incomplete_test_app (crossed integration endpoints are not implemented) tests/integrations/crossed_integrations/test_sqs.py::Test_SQS_PROPAGATION_VIA_AWS_XRAY_HEADERS: irrelevant (Localstack SQS does not support AWS Xray Header parsing) diff --git a/tests/ffe/README.md b/tests/ffe/README.md index bf46dcbf00a..987732aacb4 100644 --- a/tests/ffe/README.md +++ b/tests/ffe/README.md @@ -419,3 +419,39 @@ The tests use `interfaces.mocked_backend.get_metrics()` to retrieve the metrics 4. **Enable tests** in manifest file 5. **Run locally** with `./run.sh` to iterate quickly 6. **Push to CI** only when local tests pass + + +### Delayed provider capability registration (real Agent) + +`FEATURE_FLAGGING_CAPABILITY_REFRESH` and `FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED` +exercise the SDK → real Agent → mocked HTTPS RC backend path. They require Python 3.12 +and the Java `spring-boot` weblog, whose provider starts lazily on the first `/ffe` request. +They are provisional scenarios selected explicitly, rather than scheduled for every SDK. + +Publish base and dependent configurations once, observe Java ACK the base with capability +46 but without 50, then initialize the provider. The same RC client must add 50 without +resetting its targets version, receive and ACK the companion, and evaluate a dependent +flag successfully. The preloaded case first caches the companion through another capable +protocol client, with a legacy client kept active so the selected file union stays +unchanged when Java upgrades. + +The mocked backend renews signed director metadata on its ordinary Agent polling cadence. +It models backend capability selection and client-ID predicates; configuration repository +metadata, file hashes, and publication version stay fixed. A second publication is rejected. +This guards the real Agent and SDK transition, including the Java replacement failure in +[dd-trace-java#13258](https://github.com/DataDog/dd-trace-java/pull/13258). It does not verify +that the production backend continues renewing director metadata; that needs a backend +contract test or a live environment test. + +Use a Java build with dependent capability reporting, dependent evaluation, and the RC +replacement fix in `binaries/` (see [custom binaries](../../docs/execute/binaries.md)): + +```sh +./build.sh java -w spring-boot +TEST_LIBRARY=java ./run.sh FEATURE_FLAGGING_CAPABILITY_REFRESH -F tests/ffe/test_rc_capability_refresh.py +TEST_LIBRARY=java ./run.sh FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED -F tests/ffe/test_rc_capability_refresh.py +``` + +`-F` makes failures blocking while the manifest remains provisional. Enable the Java +manifest entry for the confirmed SDK release once the changes ship. Other weblogs need +an equivalent delayed provider lifecycle before this scenario can cover them. diff --git a/tests/ffe/test_rc_capability_refresh.py b/tests/ffe/test_rc_capability_refresh.py new file mode 100644 index 00000000000..b20c60951ed --- /dev/null +++ b/tests/ffe/test_rc_capability_refresh.py @@ -0,0 +1,154 @@ +"""Delayed capability registration through a real Agent, with no UFC republish.""" + +import base64 +import json +import time +from collections.abc import Callable +from typing import TYPE_CHECKING, cast + +from utils import context, features, scenarios, weblog +from utils.mocked_backend.rc_capabilities import BASE_PATH, DEPENDENT_PATH, capability_mask +from utils.proxy.ports import ProxyPorts + +from .utils.rc_capability_refresh import CachePreloader, acknowledged, configurations, sdk_polls + + +if TYPE_CHECKING: + from utils._context._scenarios.feature_flagging import FeatureFlaggingCapabilityRefresh + + +def wait_until(predicate: Callable[[], bool]) -> bool: + deadline = time.monotonic() + 40 + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.2) + return False + + +@scenarios.feature_flagging_capability_refresh +@scenarios.feature_flagging_capability_refresh_preloaded +@features.feature_flags_dynamic_evaluation +class Test_FFE_RC_Capability_Refresh: + def setup_delayed_provider(self) -> None: + scenario = cast("FeatureFlaggingCapabilityRefresh", context.scenario) + backend = scenario.mocked_backend_container.client + base, dependent = configurations() + self.published = backend.rc_publish_capability_variants(base, dependent) + preloader = None + legacy = None + self.cache_ready = True + if scenario is scenarios.feature_flagging_capability_refresh_preloaded: + port = scenario.proxy_container.runtime_container.attrs["NetworkSettings"]["Ports"][ + f"{ProxyPorts.weblog}/tcp" + ][0]["HostPort"] + legacy = CachePreloader(f"http://127.0.0.1:{port}", dependent=False) + legacy.poll() + preloader = CachePreloader(f"http://127.0.0.1:{port}") + self.cache_ready = wait_until(lambda: DEPENDENT_PATH in preloader.poll()) + preloader.poll() # ACK the companion before Java registers its provider. + self.preloaded = preloader is not None + self.base_ready = wait_until( + lambda: any( + acknowledged(data["request"]["content"]["client"], BASE_PATH) + for data in sdk_polls() + if not data["request"]["content"]["client"]["id"].startswith("cache-preloader-") + ) + ) + self.before_status = backend.rc_capability_status() + # Spring's @Lazy Client initializes the Datadog provider on the first /ffe request. + self.initial = weblog.post( + "/ffe", + json={ + "flag": "parent", + "variationType": "STRING", + "defaultValue": "caller-default", + "targetingKey": "test-user", + "attributes": {}, + }, + ) + + def companion_acknowledged() -> bool: + # Keep both cache clients live throughout the wait. Their expiry must not + # change the target union and accidentally unstick a stale director. + for cache_client in (preloader, legacy): + if cache_client is not None: + cache_client.poll() + return any( + acknowledged(data["request"]["content"]["client"], DEPENDENT_PATH) + for data in sdk_polls() + if not data["request"]["content"]["client"]["id"].startswith("cache-preloader-") + ) + + self.dependent_ready = wait_until(companion_acknowledged) + # Keep the preloader active, so the preloaded case retains the same union of files. + if preloader is not None: + preloader.poll() + if legacy is not None: + legacy.poll() + self.evaluation = weblog.post( + "/ffe", + json={ + "flag": "child", + "variationType": "STRING", + "defaultValue": "caller-default", + "targetingKey": "test-user", + "attributes": {}, + }, + ) + self.after = sdk_polls() + self.after_status = backend.rc_capability_status() + + def test_delayed_provider(self) -> None: + assert self.cache_ready, "The second client must cache the companion before provider initialization" + assert self.base_ready, "Java must ACK the base config before provider initialization" + assert self.dependent_ready, "Java must ACK the companion after provider initialization" + polls = [ + data for data in self.after if not data["request"]["content"]["client"]["id"].startswith("cache-preloader-") + ] + clients = [data["request"]["content"]["client"] for data in polls] + assert len({client["id"] for client in clients}) == 1, "Provider startup must retain the RC client identity" + transition = next( + index for index, client in enumerate(clients) if capability_mask(client["capabilities"]) & (1 << 50) + ) + assert transition > 0, "Must observe polls before the provider registers capability 50" + previous = clients[transition - 1] + upgraded = clients[transition] + assert capability_mask(previous["capabilities"]) & (1 << 46) + assert not capability_mask(previous["capabilities"]) & (1 << 50) + assert acknowledged(previous, BASE_PATH) + previous_version = previous["state"]["targets_version"] + response = polls[transition - 1]["response"].get("content", {}) + if isinstance(response, dict) and response.get("targets"): + previous_version = json.loads(base64.b64decode(response["targets"]))["signed"]["version"] + assert upgraded["state"]["targets_version"] == previous_version > 0, ( + "Capabilities must change without resetting SDK state" + ) + versions = [client["state"]["targets_version"] for client in clients[transition:]] + assert versions == sorted(versions), "SDK state must not decrease to recover delivery" + assert all(version > 0 for version in versions), "SDK state must not reset to recover delivery" + assert any(acknowledged(client, DEPENDENT_PATH) for client in clients[transition:]) + deliveries = [data["response"].get("content", {}) for data in polls[transition:]] + assert any( + isinstance(response, dict) and response.get("client_configs") == [DEPENDENT_PATH] for response in deliveries + ) + assert self.initial.status_code == 200 + assert json.loads(self.initial.text)["value"] == "parent-value" + assert self.evaluation.status_code == 200 + assert json.loads(self.evaluation.text)["value"] == "dependent-match", ( + "ACK alone does not prove the provider retained the replacement" + ) + assert self.published["hashes"] == self.after_status["hashes"] + assert self.after_status["publications"] == self.after_status["config_version"] == 1 + client_id = upgraded["id"] + before = next( + entry for entry in reversed(self.before_status["history"]) if entry["selected"].get(client_id) == BASE_PATH + ) + after = next( + entry for entry in self.after_status["history"] if entry["selected"].get(client_id) == DEPENDENT_PATH + ) + assert after["version"] > before["version"] + if self.preloaded: + assert set(before["selected"].values()) == set(after["selected"].values()) == {BASE_PATH, DEPENDENT_PATH} + else: + assert set(before["selected"].values()) == {BASE_PATH} diff --git a/tests/ffe/utils/rc_capability_refresh.py b/tests/ffe/utils/rc_capability_refresh.py new file mode 100644 index 00000000000..c401cceb0b4 --- /dev/null +++ b/tests/ffe/utils/rc_capability_refresh.py @@ -0,0 +1,127 @@ +"""Wire fixtures and a second RC client for the capability-refresh scenarios.""" + +import base64 +from copy import deepcopy +import hashlib +import json +import uuid + +import requests + +from utils import interfaces + + +def configurations() -> tuple[dict, dict]: + def flag(key: str, value: str) -> dict: + return { + "key": key, + "enabled": True, + "variationType": "STRING", + "variations": {"approved": {"key": "approved", "value": value}}, + "allocations": [ + {"key": "default", "rules": [], "splits": [{"variationKey": "approved", "shards": []}], "doLog": False} + ], + } + + parent = flag("parent", "parent-value") + child = flag("child", "dependent-match") + child["allocations"][0]["rules"] = [ + { + "conditions": [ + { + "flagEvaluation": {"key": "parent"}, + "operator": "ONE_OF", + "value": ["approved"], + } + ] + } + ] + base: dict = { + "createdAt": "2026-01-01T00:00:00Z", + "format": "SERVER", + "environment": {"name": "system-tests"}, + "flags": {"parent": parent}, + } + dependent = deepcopy(base) + dependent["flags"]["child"] = child + return base, dependent + + +def sdk_polls() -> list[dict]: + return [ + data + for data in interfaces.library.get_data(path_filters="/v0.7/config") + if isinstance(data.get("request", {}).get("content"), dict) + and "FFE_FLAGS" in data["request"]["content"].get("client", {}).get("products", []) + ] + + +def acknowledged(client: dict, path: str) -> bool: + return any( + state.get("product") == "FFE_FLAGS" and state.get("id") == path.split("/")[3] and state.get("apply_state") == 2 + for state in client.get("state", {}).get("config_states", []) + ) + + +class CachePreloader: + """A second capable client populates the real Agent cache using the SDK protocol.""" + + def __init__(self, url: str, *, dependent: bool = True) -> None: + self.url = url + self.client: dict = { + "id": f"cache-preloader-{uuid.uuid4()}", + "is_tracer": True, + "products": ["FFE_FLAGS"], + "capabilities": base64.b64encode(((1 << 46) | ((1 << 50) if dependent else 0)).to_bytes(7, "big")).decode(), + "client_tracer": { + "runtime_id": str(uuid.uuid4()), + "language": "java", + "tracer_version": "1.69.0", + "service": "cache-preloader", + "env": "system-tests", + "app_version": "1", + }, + "state": { + "root_version": 1, + "targets_version": 0, + "config_states": [], + "has_error": False, + "backend_client_state": "", + }, + } + self.cache: dict[str, dict] = {} + self.selected: list[str] = [] + + def poll(self) -> list[str]: + response = requests.post( + self.url + "/v0.7/config", + json={ + "client": self.client, + "cached_target_files": list(self.cache.values()), + }, + timeout=10, + ) + response.raise_for_status() + payload = response.json() + if "targets" in payload: + targets = json.loads(base64.b64decode(payload["targets"]))["signed"] + self.client["state"]["targets_version"] = targets["version"] + self.client["state"]["backend_client_state"] = targets.get("custom", {}).get("opaque_backend_state", "") + self.selected = payload.get("client_configs", []) + self.client["state"]["config_states"] = [ + { + "id": path.split("/")[3], + "product": "FFE_FLAGS", + "version": targets["targets"][path]["custom"]["v"], + "apply_state": 2, + } + for path in self.selected + ] + for file in payload.get("target_files", []): + raw = base64.b64decode(file["raw"]) + self.cache[file["path"]] = { + "path": file["path"], + "length": len(raw), + "hashes": [{"algorithm": "sha256", "hash": hashlib.sha256(raw).hexdigest()}], + } + return self.selected diff --git a/tests/test_the_test/test_minimal_number_of_scenarios.py b/tests/test_the_test/test_minimal_number_of_scenarios.py index af6ee86d03e..3ad995c622c 100644 --- a/tests/test_the_test/test_minimal_number_of_scenarios.py +++ b/tests/test_the_test/test_minimal_number_of_scenarios.py @@ -42,6 +42,10 @@ ("APPSEC_RUNTIME_ACTIVATION", "REMOTE_CONFIG_MOCKED_BACKEND_ASM_FEATURES_NOCACHE"): "TODO", ("APPSEC_RUNTIME_ACTIVATION", "REMOTE_CONFIG_MOCKED_BACKEND_LIVE_DEBUGGING"): "TODO", ("APPSEC_RUNTIME_ACTIVATION", "TRACING_CONFIG_NONDEFAULT_4"): "TODO", + ( + "FEATURE_FLAGGING_CAPABILITY_REFRESH", + "FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED", + ): "Each cache state needs a fresh Agent and an uninitialized provider; the transition runs once per process", ( "FEATURE_FLAGGING_AND_EXPERIMENTATION", "FEATURE_FLAGGING_EVP_STRESS", diff --git a/tests/test_the_test/test_rc_capability_routing.py b/tests/test_the_test/test_rc_capability_routing.py new file mode 100644 index 00000000000..e929ab3816c --- /dev/null +++ b/tests/test_the_test/test_rc_capability_routing.py @@ -0,0 +1,76 @@ +"""Validate the capability-routing fixture independently of SDK/Agent behavior.""" + +import base64 +import json + +import requests +from google.protobuf.json_format import ParseDict +from utils import pytest, scenarios +from utils.mocked_backend import MockedBackendServer +from utils.mocked_backend.rc_capabilities import BASE_PATH, DEPENDENT_PATH +from utils.proxy._decoders.protobuf_schemas import LatestConfigsRequest, LatestConfigsResponse + + +def client(client_id: str, *bits: int) -> dict: + return { + "id": client_id, + "products": ["FFE_FLAGS"], + "capabilities": base64.b64encode(sum(1 << bit for bit in bits).to_bytes(7, "big")).decode(), + } + + +@scenarios.test_the_test +@pytest.mark.parametrize("preloaded", [False, True]) +def test_rc_routing_changes_without_republishing(*, preloaded: bool) -> None: + server = MockedBackendServer() + try: + publication = server.client.rc_publish_capability_variants({"flags": {}}, {"flags": {"child": {}}}) + with pytest.raises(requests.HTTPError, match="409"): + server.client.rc_publish_capability_variants({"flags": {}}, {"flags": {}}) + + def poll(clients: list[dict]) -> dict: + # Exercise the actual protobuf request decoder and HTTP route, including signed capabilities. + request = ParseDict({"active_clients": clients}, LatestConfigsRequest()) + response = requests.post( + server.base_url + "/api/v0.1/configurations", + data=request.SerializeToString(), + headers={"Content-Type": "application/x-protobuf"}, + timeout=5, + ) + response.raise_for_status() + parsed = LatestConfigsResponse.FromString(response.content) + return { + "director": json.loads(parsed.director_metas.targets.raw), # type: ignore[attr-defined] + "config": parsed.config_metas.SerializeToString(), # type: ignore[attr-defined] + "files": {file.path: file.raw for file in parsed.target_files}, # type: ignore[attr-defined] + } + + others = [client("preloaded", 46, 50), client("legacy", 46)] if preloaded else [] + initial = poll([client("java", 46), *others]) + stable = poll([client("java", 46), *others]) + upgraded = poll([client("java", 46, 50), *others]) + assert initial["config"] == stable["config"] == upgraded["config"] + assert ( + initial["director"]["signed"]["version"] + < stable["director"]["signed"]["version"] + < upgraded["director"]["signed"]["version"] + ) + base = initial["director"]["signed"]["targets"][BASE_PATH]["custom"] + assert {predicate["clientID"] for predicate in base["tracer-predicates"]["tracer_predicates_v1"]} == { + "java", + *(["legacy"] if preloaded else []), + } + dependent = upgraded["director"]["signed"]["targets"][DEPENDENT_PATH]["custom"] + assert {predicate["clientID"] for predicate in dependent["tracer-predicates"]["tracer_predicates_v1"]} == { + "java", + *(["preloaded"] if preloaded else []), + } + if preloaded: + assert set(initial["files"]) == set(upgraded["files"]) == {BASE_PATH, DEPENDENT_PATH} + assert (DEPENDENT_PATH in initial["files"]) == preloaded + assert initial["files"][BASE_PATH] == stable["files"][BASE_PATH] + unsupported = poll([client("unsupported", 50)]) + assert unsupported["director"]["signed"]["targets"] == {} + assert server.client.rc_capability_status()["hashes"] == publication["hashes"] + finally: + server.close() diff --git a/utils/_context/_scenarios/__init__.py b/utils/_context/_scenarios/__init__.py index 205dd030727..41a074de126 100644 --- a/utils/_context/_scenarios/__init__.py +++ b/utils/_context/_scenarios/__init__.py @@ -124,6 +124,8 @@ ) from .feature_flagging import ( FeatureFlaggingAndExperimentation, + FeatureFlaggingCapabilityRefresh, + FeatureFlaggingCapabilityRefreshPreloaded, FeatureFlaggingEvpStress, FeatureFlaggingEvaluationTelemetryOptOut, FeatureFlaggingEvaluationTelemetryOptOutDisabled, @@ -321,6 +323,8 @@ class _Scenarios: feature_flagging_evaluation_telemetry_opt_out = FeatureFlaggingEvaluationTelemetryOptOut() feature_flagging_evaluation_telemetry_opt_out_disabled = FeatureFlaggingEvaluationTelemetryOptOutDisabled() feature_flagging_evp_stress = FeatureFlaggingEvpStress() + feature_flagging_capability_refresh = FeatureFlaggingCapabilityRefresh() + feature_flagging_capability_refresh_preloaded = FeatureFlaggingCapabilityRefreshPreloaded() remote_config_mocked_backend_asm_features_nocache = RemoteConfigMockedBackendAsmFeaturesNocache() # APM tracing end-to-end scenarios apm_tracing_e2e_otel = ApmTracingE2eOtel() diff --git a/utils/_context/_scenarios/feature_flagging.py b/utils/_context/_scenarios/feature_flagging.py index 711d7402497..1e50559cda1 100644 --- a/utils/_context/_scenarios/feature_flagging.py +++ b/utils/_context/_scenarios/feature_flagging.py @@ -1,5 +1,7 @@ import pytest +from utils.proxy.ports import ProxyPorts + from ._core import DdTraceEndToEndScenario, ScenarioGroup, scenario_groups # Shared by the ordinary FFE scenario, the isolated EVP stress scenario, and both @@ -88,3 +90,23 @@ def configure(self, config: pytest.Config) -> None: # Allow an EVP flush (Go: 10 s) plus the telemetry heartbeat (2 s) before stopping. # Preserve longer language-specific waits. self.library_interface_timeout = max(self.library_interface_timeout, 15) + + +class FeatureFlaggingCapabilityRefresh(_ProvisionalFeatureFlaggingScenario): + def __init__(self) -> None: + super().__init__( + "FEATURE_FLAGGING_CAPABILITY_REFRESH", + rc_api_enabled=True, + rc_backend_enabled=True, + weblog_env={**_FFE_WEBLOG_ENV, "DD_FEATURE_FLAGS_ENABLED": "true"}, + doc="Delayed provider capability registration through a real Agent, without republishing UFC.", + ) + self.weblog_categories = [] + # A second protocol client uses this listener to preload the real Agent cache. + self.proxy_container.ports[f"{ProxyPorts.weblog}/tcp"] = ("127.0.0.1", None) + + +class FeatureFlaggingCapabilityRefreshPreloaded(FeatureFlaggingCapabilityRefresh): + def __init__(self) -> None: + super().__init__() + self.name = "FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED" diff --git a/utils/build/docker/java/spring-boot/src/main/java/com/datadoghq/system_tests/springboot/featureflag/FeatureFlagEvaluatorController.java b/utils/build/docker/java/spring-boot/src/main/java/com/datadoghq/system_tests/springboot/featureflag/FeatureFlagEvaluatorController.java index 023360ae5d1..c3d08a752a7 100644 --- a/utils/build/docker/java/spring-boot/src/main/java/com/datadoghq/system_tests/springboot/featureflag/FeatureFlagEvaluatorController.java +++ b/utils/build/docker/java/spring-boot/src/main/java/com/datadoghq/system_tests/springboot/featureflag/FeatureFlagEvaluatorController.java @@ -2,7 +2,6 @@ import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE; -import datadog.trace.api.openfeature.Provider; import dev.openfeature.sdk.Client; import dev.openfeature.sdk.EvaluationContext; import dev.openfeature.sdk.FeatureProvider; @@ -33,6 +32,25 @@ public class FeatureFlagEvaluatorController { @Configuration public static class FeatureFlagEvaluatorConfig { + private static FeatureProvider datadogProvider() { + // The standalone provider moved packages; retain support for older SDK binaries. + Class providerClass; + try { + providerClass = Class.forName("com.datadog.openfeature.Provider"); + } catch (ClassNotFoundException absent) { + try { + providerClass = Class.forName("datadog.trace.api.openfeature.Provider"); + } catch (ClassNotFoundException missing) { + throw new IllegalStateException("Datadog OpenFeature provider is missing", missing); + } + } + try { + return (FeatureProvider) providerClass.getDeclaredConstructor().newInstance(); + } catch (ReflectiveOperationException failure) { + throw new IllegalStateException("Cannot initialize Datadog OpenFeature provider", failure); + } + } + @Lazy @Bean public Client client() { @@ -44,7 +62,7 @@ public Client client() { || Boolean.parseBoolean(System.getenv("DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED")); final FeatureProvider provider; if (featureFlaggingConfigured) { - provider = new Provider(); + provider = datadogProvider(); } else { provider = new NoOpProvider() { @Override diff --git a/utils/build/docker/java/weblog_metadata.yml b/utils/build/docker/java/weblog_metadata.yml index aab68288d0c..9f4a800e836 100644 --- a/utils/build/docker/java/weblog_metadata.yml +++ b/utils/build/docker/java/weblog_metadata.yml @@ -16,6 +16,7 @@ resteasy-netty3: categories: [dd_trace] spring-boot: categories: [dd_trace] + supported_scenarios: [FEATURE_FLAGGING_CAPABILITY_REFRESH, FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED] spring-boot-3-native: categories: [dd_trace] spring-boot-jetty: diff --git a/utils/mocked_backend/_backend_rc.py b/utils/mocked_backend/_backend_rc.py index 8cb9e3f1448..2fdd7a9bf92 100644 --- a/utils/mocked_backend/_backend_rc.py +++ b/utils/mocked_backend/_backend_rc.py @@ -5,7 +5,11 @@ from __future__ import annotations from http import HTTPStatus -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any +import threading + +from ._utils import CONTROL_URL_PREFIX +from .rc_capabilities import CapabilityRoutingFixture from utils.proxy.rc_response_builder import ( build_org_data_protobuf, @@ -15,6 +19,7 @@ if TYPE_CHECKING: + from collections.abc import Mapping from utils.mocked_backend._request_handler import HttpMethod, MockedBackendRequestHandler, MockedBackendServerCore _Base = MockedBackendServerCore @@ -30,12 +35,45 @@ class MockedBackendServerRcMixin(_Base): # - POST /api/v0.1/configurations def _rc_configure(self) -> None: + self._rc_routing: CapabilityRoutingFixture | None = None + self._rc_lock = threading.Lock() + self.add_handler("POST", f"{CONTROL_URL_PREFIX}/rc/capability-routing", self._rc_publish) + self.add_handler("GET", f"{CONTROL_URL_PREFIX}/rc/capability-routing", self._rc_status) self._rc_add_static_handler("GET", "/api/v0.1/org", build_org_data_protobuf(), "application/x-protobuf") self._rc_add_static_handler("GET", "/api/v0.1/status", build_org_status_protobuf(), "application/x-protobuf") self._rc_add_static_handler("GET", "/api/v0.2/echo-test", b"ok", "text/plain") - self._rc_add_static_handler( - "POST", "/api/v0.1/configurations", build_rc_configurations_protobuf(None), "application/x-protobuf" - ) + self.add_handler("POST", "/api/v0.1/configurations", self._rc_poll) + + def _rc_publish( + self, request: MockedBackendRequestHandler, payload: bytes | dict | None + ) -> tuple[Mapping[str, Any], bytes]: + if ( + not isinstance(payload, dict) + or set(payload) != {"base", "dependent"} + or not all(isinstance(value, dict) for value in payload.values()) + ): + return request.write_json(HTTPStatus.BAD_REQUEST, {"error": "Expected base and dependent configurations"}) + with self._rc_lock: + if self._rc_routing is not None: + return request.write_json(HTTPStatus.CONFLICT, {"error": "Configurations are already published"}) + self._rc_routing = CapabilityRoutingFixture(payload["base"], payload["dependent"]) + return request.write_json(HTTPStatus.OK, self._rc_routing.status()) + + def _rc_status( + self, request: MockedBackendRequestHandler, _: bytes | dict | None + ) -> tuple[Mapping[str, Any], bytes]: + return request.write_json(HTTPStatus.OK, self._rc_routing.status() if self._rc_routing else {}) + + def _rc_poll(self, request: MockedBackendRequestHandler, payload: bytes | dict | None) -> tuple[bytes, bytes]: + if self._rc_routing is None: + content = build_rc_configurations_protobuf(None) + else: + if not isinstance(payload, dict): + raise ValueError("Expected a decoded LatestConfigsRequest") + content = self._rc_routing.respond(payload.get("active_clients", [])) + request.send_response(HTTPStatus.OK) + request.send_header("Content-Type", "application/x-protobuf") + return content, content def _rc_add_static_handler(self, method: HttpMethod, path: str, content: bytes, content_type: str) -> None: def handler(request: MockedBackendRequestHandler, _: bytes | dict | None) -> tuple[bytes, bytes]: diff --git a/utils/mocked_backend/_client.py b/utils/mocked_backend/_client.py index 707e6a7aacc..c2bfe6d07dc 100644 --- a/utils/mocked_backend/_client.py +++ b/utils/mocked_backend/_client.py @@ -57,3 +57,22 @@ def ffe_get_status(self) -> MockedBackendFFEStatus: response = requests.get(f"{self.base_url}{_FFE_CONTROL_PATH}/status", timeout=5, verify=self.verify) response.raise_for_status() return cast("MockedBackendFFEStatus", response.json()) + + def rc_publish_capability_variants(self, base: dict, dependent: dict) -> dict: + response = requests.post( + f"{self.base_url}{CONTROL_URL_PREFIX}/rc/capability-routing", + json={"base": base, "dependent": dependent}, + timeout=5, + verify=self.verify, + ) + response.raise_for_status() + return cast("dict", response.json()) + + def rc_capability_status(self) -> dict: + response = requests.get( + f"{self.base_url}{CONTROL_URL_PREFIX}/rc/capability-routing", + timeout=5, + verify=self.verify, + ) + response.raise_for_status() + return cast("dict", response.json()) diff --git a/utils/mocked_backend/rc_capabilities.py b/utils/mocked_backend/rc_capabilities.py new file mode 100644 index 00000000000..c5a480ed0a5 --- /dev/null +++ b/utils/mocked_backend/rc_capabilities.py @@ -0,0 +1,91 @@ +"""Capability-aware RC director fixture. Configuration publication is immutable. + +The real backend selects by capabilities, then targets client IDs in director metadata. +This models that contract; it does not test the production backend's implementation. +""" + +from __future__ import annotations + +import base64 +from copy import deepcopy +import hashlib +import json +import threading + +from utils.proxy._decoders.protobuf_schemas import LatestConfigsResponse +from utils.proxy.rc_response_builder import build_rc_configurations_protobuf +from utils.proxy.tuf import build_signed_targets + +BASE_PATH = "datadog/2/FFE_FLAGS/capability-base/config" +DEPENDENT_PATH = "datadog/2/FFE_FLAGS/capability-dependent/config" + + +def capability_mask(value: str | list[int]) -> int: + raw = base64.b64decode(value) if isinstance(value, str) else bytes(item & 255 for item in value) + return int.from_bytes(raw, "big") + + +class CapabilityRoutingFixture: + """Publish once, and renew signed director metadata on ordinary Agent polls. + + The configuration repository, file hashes and file versions never change. + Director versions advance even when the union of selected files is unchanged, + as happens when a second client already populated the Agent's cache. + """ + + def __init__(self, base: dict, dependent: dict) -> None: + self._lock = threading.Lock() + self._version = 0 + self._history: list[dict] = [] + self._files = { + path: json.dumps(payload, separators=(",", ":")).encode() + for path, payload in ((BASE_PATH, base), (DEPENDENT_PATH, dependent)) + } + self._metadata: dict[str, dict] = { + path: {"length": len(raw), "hashes": {"sha256": hashlib.sha256(raw).hexdigest()}, "custom": {"v": 1}} + for path, raw in self._files.items() + } + published = LatestConfigsResponse.FromString(self._response(1, self._metadata)) + self._config_metas = published.config_metas # type: ignore[attr-defined] + + def _response(self, version: int, targets: dict) -> bytes: + return build_rc_configurations_protobuf( + { + "targets": base64.b64encode(json.dumps(build_signed_targets(version, targets)).encode()).decode(), + "target_files": [ + {"path": path, "raw": base64.b64encode(self._files[path]).decode()} for path in targets + ], + } + ) + + def respond(self, active_clients: list[dict]) -> bytes: + with self._lock: + targets: dict = {} + selected: dict[str, str] = {} + for client in active_clients: + bits = capability_mask(client.get("capabilities", "")) + if "FFE_FLAGS" not in client.get("products", []) or not bits & (1 << 46): + continue + path = DEPENDENT_PATH if bits & (1 << 50) else BASE_PATH + selected[client["id"]] = path + if path not in targets: + targets[path] = deepcopy(self._metadata[path]) + targets[path]["custom"]["c"] = [] + targets[path]["custom"]["tracer-predicates"] = {"tracer_predicates_v1": []} + targets[path]["custom"]["c"].append(client["id"]) + targets[path]["custom"]["tracer-predicates"]["tracer_predicates_v1"].append({"clientID": client["id"]}) + self._version += 1 + response = LatestConfigsResponse.FromString(self._response(self._version, targets)) + response.config_metas.CopyFrom(self._config_metas) # type: ignore[attr-defined] + self._history.append({"version": self._version, "selected": selected}) + self._history = self._history[-100:] + return response.SerializeToString() + + def status(self) -> dict: + with self._lock: + return { + "publications": 1, + "config_version": 1, + "hashes": {path: meta["hashes"]["sha256"] for path, meta in self._metadata.items()}, + "history": deepcopy(self._history), + } diff --git a/utils/proxy/_decoders/protobuf_schemas.py b/utils/proxy/_decoders/protobuf_schemas.py index 56af04c97f0..95c9f31f647 100644 --- a/utils/proxy/_decoders/protobuf_schemas.py +++ b/utils/proxy/_decoders/protobuf_schemas.py @@ -28,6 +28,7 @@ def _get_mesages(filename: str) -> dict[str, type[message.Message]]: MetricsV3Payload = _metrics_v3_messages["datadog.dogstatsdhttp.Payload"] _remoteconfig_messages = _get_mesages("remoteconfig.descriptor") +LatestConfigsRequest = _remoteconfig_messages["datadog.config.LatestConfigsRequest"] LatestConfigsResponse = _remoteconfig_messages["datadog.config.LatestConfigsResponse"] ConfigMetas = _remoteconfig_messages["datadog.config.ConfigMetas"] DirectorMetas = _remoteconfig_messages["datadog.config.DirectorMetas"] diff --git a/utils/proxy/_decoders/remoteconfig.descriptor b/utils/proxy/_decoders/remoteconfig.descriptor index 450e54c0a8b..85ece88fc81 100644 --- a/utils/proxy/_decoders/remoteconfig.descriptor +++ b/utils/proxy/_decoders/remoteconfig.descriptor @@ -1,6 +1,13 @@ -Ö -(utils/proxy/_decoders/remoteconfig.protodatadog.config"¬ +‡ + +(utils/proxy/_decoders/remoteconfig.protodatadog.config"U +LatestConfigsRequest= +active_clients ( 2.datadog.config.ClientR activeClients"X +Client +id ( Rid +products ( Rproducts" + capabilities ( R capabilities"¬ ConfigMetas- roots ( 2.datadog.config.TopMetaRroots5 timestamp ( 2.datadog.config.TopMetaR timestamp3 diff --git a/utils/proxy/_decoders/remoteconfig.proto b/utils/proxy/_decoders/remoteconfig.proto index 97f50ac05dc..cd006dd74fc 100644 --- a/utils/proxy/_decoders/remoteconfig.proto +++ b/utils/proxy/_decoders/remoteconfig.proto @@ -7,6 +7,17 @@ syntax = "proto3"; package datadog.config; +// Request subset used to route published configurations to active SDK clients. +// Field numbers match the Agent schema; unrelated fields are ignored. +message LatestConfigsRequest { + repeated Client active_clients = 6; +} +message Client { + string id = 2; + repeated string products = 3; + bytes capabilities = 11; +} + // Backend TUF metadata structures message ConfigMetas { diff --git a/utils/proxy/_deserializer.py b/utils/proxy/_deserializer.py index 6c34f81f317..483beffc10d 100644 --- a/utils/proxy/_deserializer.py +++ b/utils/proxy/_deserializer.py @@ -26,7 +26,14 @@ ExportLogsServiceRequest, ExportLogsServiceResponse, ) -from ._decoders.protobuf_schemas import MetricPayload, TracePayload, SketchPayload, BackendResponsePayload +from ._decoders.protobuf_schemas import ( + MetricPayload, + TracePayload, + SketchPayload, + BackendResponsePayload, + LatestConfigsRequest, + LatestConfigsResponse, +) from ._decoders.metrics_v3 import decode_metrics_v3, decode_sketches_v3 from .trace_bytes_decoding import decode_trace_bytes_ascii, unpack_trace_bytes_msgpack from .traces.trace_v1 import deserialize_v1_trace, _uncompress_agent_v1_trace, decode_appsec_s_value @@ -196,6 +203,9 @@ def json_load(): if content_type == "application/x-protobuf": assert isinstance(content, bytes) + if path == "/api/v0.1/configurations": + schema = LatestConfigsRequest if "status_code" not in message else LatestConfigsResponse + return MessageToDict(schema.FromString(content), preserving_proto_field_name=True) dd_protocol = get_header_value("dd-protocol", message["headers"]) if dd_protocol == "otlp" and "traces" in path: return deserialize_otlp_v1_trace(