diff --git a/manifests/php.yml b/manifests/php.yml index 14f93e3e087..78d43b2b769 100644 --- a/manifests/php.yml +++ b/manifests/php.yml @@ -717,7 +717,6 @@ manifest: - weblog_declaration: "*": v1.21.0-dev laravel11x: missing_feature (pending to be implemented on this weblog) - tests/ffe/test_exposures.py::Test_FFE_Exposure_Events::test_ffe_multiple_remote_config_files: flaky (FFL-2676) tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446) tests/ffe/test_flag_eval_metrics.py: - weblog_declaration: diff --git a/tests/ffe/test_exposures.py b/tests/ffe/test_exposures.py index a90a1a6ed89..28c8cec765c 100644 --- a/tests/ffe/test_exposures.py +++ b/tests/ffe/test_exposures.py @@ -17,6 +17,8 @@ EXPOSURES_PATH = "/api/v2/exposures" EXPOSURE_WAIT_TIMEOUT_SECONDS = 30 +WaitResult = tuple[bool, str] + def exposure_events_from_data( data: dict, flag_keys: set[str] | None = None, subject_id: str | None = None @@ -59,26 +61,84 @@ def find_exposure_events(flag_key: str, subject_id: str | None = None) -> list[d return events -def wait_for_exposure_event(flag_keys: set[str], subject_id: str | None = None) -> None: - """Wait until the agent receives an exposure event for one of the given flags.""" - assert interfaces.agent.wait_for( - lambda data: bool(exposure_events_from_data(data, flag_keys, subject_id)), +def exposure_flag_keys_seen(flag_keys: set[str], subject_id: str | None = None) -> set[str]: + """Return matching flag keys already captured by the agent.""" + seen = set() + for data in interfaces.agent.get_data(path_filters=EXPOSURES_PATH): + for event in exposure_events_from_data(data, flag_keys, subject_id): + flag = event.get("flag") + flag_key = flag.get("key") if isinstance(flag, dict) else None + if isinstance(flag_key, str): + seen.add(flag_key) + return seen + + +def flush_ffe_payloads() -> WaitResult: + """Flush buffered payloads when the weblog exposes the common flush endpoint.""" + response = weblog.get("/flush", timeout=10) + if response.status_code in (200, 404, 405): + return True, "" + + return False, f"Failed to flush FFE payloads with status {response.status_code}: {response.text}" + + +def wait_for_exposure_event(flag_keys: set[str], subject_id: str | None = None) -> WaitResult: + """Wait until the agent receives exposure events for all of the given flags.""" + flush_result = flush_ffe_payloads() + if not flush_result[0]: + return flush_result + + sorted_flag_keys = sorted(flag_keys) + seen_flag_keys = exposure_flag_keys_seen(flag_keys, subject_id) + + def collect_matching_flags(data: dict) -> bool: + for event in exposure_events_from_data(data, flag_keys, subject_id): + flag = event.get("flag") + flag_key = flag.get("key") if isinstance(flag, dict) else None + if isinstance(flag_key, str): + seen_flag_keys.add(flag_key) + return flag_keys <= seen_flag_keys + + if flag_keys <= seen_flag_keys or interfaces.agent.wait_for( + collect_matching_flags, timeout=EXPOSURE_WAIT_TIMEOUT_SECONDS, - ), f"Timed out waiting for exposure event for flags {sorted(flag_keys)} and subject {subject_id!r}" + ): + return True, "" + + missing_flag_keys = sorted(flag_keys - seen_flag_keys) + return ( + False, + f"Timed out waiting for exposure events for flags {sorted_flag_keys} " + f"and subject {subject_id!r}; missing {missing_flag_keys}", + ) -def wait_for_min_exposure_count(flag_key: str, expected: int, subject_id: str | None = None) -> int: - """Wait until enough matching exposure events are available, then return the current count.""" +def wait_for_min_exposure_count(flag_key: str, expected: int, subject_id: str | None = None) -> WaitResult: + """Wait until enough matching exposure events are available.""" + flush_result = flush_ffe_payloads() + if not flush_result[0]: + return flush_result + + if count_exposure_events(flag_key, subject_id) >= expected: + return True, "" + + if interfaces.agent.wait_for( + lambda _: count_exposure_events(flag_key, subject_id) >= expected, + timeout=EXPOSURE_WAIT_TIMEOUT_SECONDS, + ): + return True, "" + count = count_exposure_events(flag_key, subject_id) + return ( + False, + f"Timed out waiting for exposure count >= {expected} for flag {flag_key} " + f"and subject {subject_id!r}; observed {count}", + ) - if count < expected: - assert interfaces.agent.wait_for( - lambda _: count_exposure_events(flag_key, subject_id) >= expected, - timeout=EXPOSURE_WAIT_TIMEOUT_SECONDS, - ), f"Timed out waiting for exposure count >= {expected} for flag {flag_key} and subject {subject_id!r}" - count = count_exposure_events(flag_key, subject_id) - return count +def assert_wait_results(*wait_results: WaitResult) -> None: + for success, failure_message in wait_results: + assert success, failure_message # Simple UFC fixture for testing with doLog: true @@ -132,11 +192,12 @@ def setup_ffe_exposure_event_generation(self): "attributes": attributes, }, ) + self.exposure_ready = wait_for_exposure_event({self.flag}, self.targeting_key) def test_ffe_exposure_event_generation(self): """Test that FFE generates exposure events when flags are evaluated via weblog.""" assert self.r.status_code == 200, f"Flag evaluation failed: {self.r.text}" - wait_for_exposure_event({self.flag}, self.targeting_key) + assert_wait_results(self.exposure_ready) # Search for our specific flag in all exposure events matching_event = None @@ -192,9 +253,13 @@ def test_ffe_exposure_event_generation(self): f"Expected subject '{self.targeting_key}', got '{matching_event['subject']['id']}'" ) - def setup_ffe_multiple_remote_config_files(self): - """Set up FFE with multiple remote config files across different target paths.""" - # Set up multiple Remote Config files with different config IDs + def setup_ffe_remote_config_last_received_wins(self): + """Set up FFE where a later UFC config replaces (does not merge with) the earlier one. + + FFE_FLAGS delivers a single canonical UFC document: the last received config wins and + multiple config files are not merged into one flag set (confirmed across the Go, PHP, + Java and libdatadog SDKs). + """ config_id_1 = "ffe-test-config-1" config_id_2 = "ffe-test-config-2" @@ -244,29 +309,30 @@ def setup_ffe_multiple_remote_config_files(self): }, } - # Apply both configurations - rc.tracer_rc_state.reset().set_config(f"{RC_PATH}/{config_id_1}/config", rc_config_1).set_config( - f"{RC_PATH}/{config_id_2}/config", rc_config_2 - ).apply() - - # Evaluate both feature flags self.flag_1 = "test-flag-1" self.flag_2 = "test-flag-2" + self.flag_1_variation_value = "on" # value test-flag-1 resolves to under config 1 + self.flag_1_default = "last-received-wins-default" self.targeting_key = "test-user-multi" + self.recheck_key = "test-user-last-received-wins" - # Evaluate first flag + # Apply config 1 and evaluate flag 1 while config 1 is the current config. + rc.tracer_rc_state.reset().set_config(f"{RC_PATH}/{config_id_1}/config", rc_config_1).apply() self.r1 = weblog.post( "/ffe", json={ "flag": self.flag_1, "variationType": "STRING", - "defaultValue": "default", + "defaultValue": self.flag_1_default, "targetingKey": self.targeting_key, "attributes": {}, }, ) + # Store setup-time wait results; tests run after teardown and cannot produce new payloads. + self.exposure_ready_1 = wait_for_exposure_event({self.flag_1}, self.targeting_key) - # Evaluate second flag + # Replace config 1 with config 2. Configs are not merged; the last received wins. + rc.tracer_rc_state.reset().set_config(f"{RC_PATH}/{config_id_2}/config", rc_config_2).apply() self.r2 = weblog.post( "/ffe", json={ @@ -277,47 +343,59 @@ def setup_ffe_multiple_remote_config_files(self): "attributes": {}, }, ) + self.exposure_ready_2 = wait_for_exposure_event({self.flag_2}, self.targeting_key) - def test_ffe_multiple_remote_config_files(self): - """Test that FFE correctly handles multiple remote config files with different flags.""" + # Re-evaluate flag 1 now that config 2 is current. Because config 1 was replaced (not + # merged), flag 1 is no longer known and must fall through to the caller default. + self.r1_after_replace = weblog.post( + "/ffe", + json={ + "flag": self.flag_1, + "variationType": "STRING", + "defaultValue": self.flag_1_default, + "targetingKey": self.recheck_key, + "attributes": {}, + }, + ) + + def test_ffe_remote_config_last_received_wins(self): + """A later UFC config replaces the earlier one; FFE_FLAGS files are not merged.""" assert self.r1.status_code == 200, f"First flag evaluation failed: {self.r1.text}" assert self.r2.status_code == 200, f"Second flag evaluation failed: {self.r2.text}" - wait_for_exposure_event({self.flag_1, self.flag_2}, self.targeting_key) - - # Collect all exposure events for our specific flags - flags_found = set() + assert self.r1_after_replace.status_code == 200, ( + f"Flag re-evaluation after config replacement failed: {self.r1_after_replace.text}" + ) + assert_wait_results(self.exposure_ready_1, self.exposure_ready_2) - for data in interfaces.agent.get_data(path_filters=EXPOSURES_PATH): - exposure_data = data["request"]["content"] - assert exposure_data is not None, "No exposure events were sent to agent" + # Each flag produced an exposure while its own config was the current config. + flag_1_events = find_exposure_events(self.flag_1, self.targeting_key) + flag_2_events = find_exposure_events(self.flag_2, self.targeting_key) + assert flag_1_events, f"Expected an exposure for '{self.flag_1}' while config 1 was current" + assert flag_2_events, f"Expected an exposure for '{self.flag_2}' while config 2 was current" - # Validate context - assert "context" in exposure_data, "Response missing 'context' field" - context = exposure_data["context"] - assert context.get("service") == "weblog", f"Expected service_name 'weblog', got '{context}'" + for event in flag_1_events + flag_2_events: + assert event["subject"]["id"] == self.targeting_key, ( + f"Expected subject '{self.targeting_key}', got '{event['subject']['id']}'" + ) - # Validate exposures array - assert "exposures" in exposure_data, "Response missing 'exposures' field" - assert isinstance(exposure_data["exposures"], list), "Exposures should be a list" + # last-received-wins: after config 2 replaced config 1, flag 1 must no longer resolve to + # its config-1 variation and must fall through to the caller default. Had the configs been + # merged, flag 1 would still return its config-1 value. + value_after_replace = json.loads(self.r1_after_replace.text).get("value") + assert value_after_replace != self.flag_1_variation_value, ( + f"'{self.flag_1}' still resolved to its config-1 value {value_after_replace!r} after its " + f"config was replaced; FFE_FLAGS configs must not be merged (last received wins)" + ) + assert value_after_replace == self.flag_1_default, ( + f"Expected '{self.flag_1}' to fall through to default {self.flag_1_default!r} after config " + f"replacement, got {value_after_replace!r}" + ) - # Collect flag keys and validate events for our test flags - for event in exposure_data["exposures"]: - assert "flag" in event, "Exposure event missing 'flag' field" - assert "key" in event["flag"], "Flag missing 'key' field" - flag_key = event["flag"]["key"] - - # Only validate events for our test flags with our specific targeting_key - if flag_key in (self.flag_1, self.flag_2) and event.get("subject", {}).get("id") == self.targeting_key: - flags_found.add(flag_key) - # Validate subject for our test events - assert "subject" in event, "Exposure event missing 'subject' field" - assert event["subject"]["id"] == self.targeting_key, ( - f"Expected subject '{self.targeting_key}', got '{event['subject']['id']}'" - ) - - # Verify that both flags were evaluated and sent exposure events - assert self.flag_1 in flags_found or self.flag_2 in flags_found, ( - f"Expected to find flags '{self.flag_1}' or '{self.flag_2}' in exposure events, found: {flags_found}" + # No new exposure should be emitted for the replaced flag under the recheck subject. + replaced_flag_events = find_exposure_events(self.flag_1, self.recheck_key) + assert not replaced_flag_events, ( + f"Expected no exposure for replaced flag '{self.flag_1}' under subject '{self.recheck_key}', " + f"but found {len(replaced_flag_events)}" ) @@ -410,6 +488,7 @@ def setup_ffe_malformed_remote_config_rejection(self): "attributes": {}, }, ) + self.exposure_ready = wait_for_exposure_event({self.flag}, self.targeting_key) # Now update with a malformed config (missing allocations and variationType) malformed_rc_config = { @@ -448,7 +527,7 @@ def test_ffe_malformed_remote_config_rejection(self): """Test that FFE rejects malformed remote config and preserves the old valid configuration.""" assert self.r1.status_code == 200, f"First flag evaluation failed: {self.r1.text}" assert self.r2.status_code == 200, f"Second flag evaluation failed: {self.r2.text}" - wait_for_exposure_event({self.flag}, self.targeting_key) + assert_wait_results(self.exposure_ready) # Verify that exposure events are still generated for both requests # and the flag configuration remained valid despite the malformed update @@ -532,17 +611,21 @@ def setup_ffe_exposure_caching_same_subject(self): }, ) self.responses.append(r) + self.exposure_ready = wait_for_min_exposure_count(self.flag_key, 1, self.targeting_key) def test_ffe_exposure_caching_same_subject(self): """Test that multiple evaluations for the same subject generate at most one exposure event.""" # Verify all requests succeeded for i, r in enumerate(self.responses): assert r.status_code == 200, f"Request {i + 1} failed: {r.text}" + assert_wait_results(self.exposure_ready) + + for i, r in enumerate(self.responses): result = json.loads(r.text) assert result["value"] == "value-a", f"Request {i + 1}: expected 'value-a', got '{result['value']}'" # Count exposure events for this specific subject - exposure_count = wait_for_min_exposure_count(self.flag_key, 1, self.targeting_key) + exposure_count = count_exposure_events(self.flag_key, self.targeting_key) # The exposure cache should deduplicate events - we expect exactly 1 exposure # for the same (subject, allocation, variant) tuple @@ -583,19 +666,21 @@ def setup_ffe_exposure_caching_different_subjects(self): }, ) self.responses.append(r) + self.exposure_wait_results = [ + wait_for_min_exposure_count(self.flag_key, 1, subject) for subject in self.subjects + ] def test_ffe_exposure_caching_different_subjects(self): """Test that each unique subject generates exactly one exposure event.""" # Verify all requests succeeded for i, r in enumerate(self.responses): assert r.status_code == 200, f"Request {i + 1} failed: {r.text}" + assert_wait_results(*self.exposure_wait_results) + + for i, r in enumerate(self.responses): result = json.loads(r.text) assert result["value"] == "value-a", f"Request {i + 1}: expected 'value-a', got '{result['value']}'" - # Wait for each subject to be observed before asserting exact totals. - for subject in self.subjects: - wait_for_min_exposure_count(self.flag_key, 1, subject) - # Count total exposure events for this flag total_exposure_count = count_exposure_events(self.flag_key) @@ -678,6 +763,7 @@ def setup_ffe_exposure_caching_allocation_cycle(self): "attributes": {}, }, ) + self.exposure_ready = wait_for_min_exposure_count(self.flag_key, 3, self.targeting_key) def test_ffe_exposure_caching_allocation_cycle(self): """Test that allocation-a → allocation-b → allocation-a generates 3 exposures.""" @@ -700,7 +786,8 @@ def test_ffe_exposure_caching_allocation_cycle(self): # - Exposure #1: default-allocation # - Exposure #2: different-allocation (allocation changed) # - Exposure #3: default-allocation (allocation changed back) - exposure_count = wait_for_min_exposure_count(self.flag_key, 3, self.targeting_key) + assert_wait_results(self.exposure_ready) + exposure_count = count_exposure_events(self.flag_key, self.targeting_key) assert exposure_count == 3, ( f"Expected exactly 3 exposure events for subject '{self.targeting_key}' " @@ -773,6 +860,7 @@ def setup_ffe_exposure_caching_variant_cycle(self): "attributes": {}, }, ) + self.exposure_ready = wait_for_min_exposure_count(self.flag_key, 3, self.targeting_key) def test_ffe_exposure_caching_variant_cycle(self): """Test that variant-a → variant-b → variant-a generates 3 exposures.""" @@ -795,7 +883,8 @@ def test_ffe_exposure_caching_variant_cycle(self): # - Exposure #1: variant-a # - Exposure #2: variant-b (variant changed) # - Exposure #3: variant-a (variant changed back) - exposure_count = wait_for_min_exposure_count(self.flag_key, 3, self.targeting_key) + assert_wait_results(self.exposure_ready) + exposure_count = count_exposure_events(self.flag_key, self.targeting_key) assert exposure_count == 3, ( f"Expected exactly 3 exposure events for subject '{self.targeting_key}' " @@ -921,6 +1010,8 @@ def test_ffe_exposure_dolog_false(self): # Verify all requests succeeded and returned the expected value for i, r in enumerate(self.responses): assert r.status_code == 200, f"Request {i + 1} failed: {r.text}" + + for i, r in enumerate(self.responses): result = json.loads(r.text) assert result["value"] == "value-a", f"Request {i + 1}: expected 'value-a', got '{result['value']}'" @@ -962,14 +1053,15 @@ def setup_ffe_exp_5_missing_targeting_key(self): "attributes": {}, }, ) + self.exposure_ready = wait_for_exposure_event({self.flag_key}, "") def test_ffe_exp_5_missing_targeting_key(self): """EXP.5: Test that empty targeting key generates exposure with subject.id = ''.""" assert self.response.status_code == 200, f"Flag evaluation failed: {self.response.text}" + assert_wait_results(self.exposure_ready) result = json.loads(self.response.text) assert result["value"] == "value-a", f"Expected 'value-a', got '{result['value']}'" - wait_for_exposure_event({self.flag_key}, "") # Search for exposure event with empty subject.id matching_event = None diff --git a/tests/ffe/test_flag_eval_metrics.py b/tests/ffe/test_flag_eval_metrics.py index 245bfb169a5..2a79e731570 100644 --- a/tests/ffe/test_flag_eval_metrics.py +++ b/tests/ffe/test_flag_eval_metrics.py @@ -12,6 +12,8 @@ RC_PRODUCT = "FFE_FLAGS" RC_PATH = f"datadog/2/{RC_PRODUCT}" +EVAL_METRIC_WAIT_TIMEOUT_SECONDS = 30 +EXPECTED_EVAL_METRIC_FLAGS: set[str] = set() def find_eval_metrics(flag_key: str | None = None): @@ -34,6 +36,23 @@ def find_eval_metrics(flag_key: str | None = None): return results +def wait_for_expected_eval_metrics() -> None: + """Wait for metric payloads that are known to arrive near the end of setup.""" + missing = {flag_key for flag_key in EXPECTED_EVAL_METRIC_FLAGS if len(find_eval_metrics(flag_key)) == 0} + + def has_expected_metrics(_: dict) -> bool: + missing.difference_update(flag_key for flag_key in list(missing) if len(find_eval_metrics(flag_key)) > 0) + return not missing + + if not missing: + return + + interfaces.agent.wait_for( + has_expected_metrics, + timeout=EVAL_METRIC_WAIT_TIMEOUT_SECONDS, + ) + + def get_tag_value(tags: list[str], key: str): """Extract a tag value from a list of 'key:value' strings.""" prefix = f"{key}:" @@ -617,6 +636,7 @@ def setup_ffe_eval_config_exists_flag_missing(self): "attributes": {}, }, ) + EXPECTED_EVAL_METRIC_FLAGS.add(self.flag_key) def test_ffe_eval_config_exists_flag_missing(self): """Test that missing flag (with config loaded) produces error.type=flag_not_found.""" @@ -979,6 +999,7 @@ def setup_ffe_eval_nested_attributes_ignored(self): "attributes": {"nested": {"inner": "value"}, "flat": "value"}, # Nested dict should be ignored }, ) + EXPECTED_EVAL_METRIC_FLAGS.add(self.flag_key) def test_ffe_eval_nested_attributes_ignored(self): """Test that nested attributes are ignored and evaluation succeeds (OF.3).""" @@ -1027,6 +1048,7 @@ def setup_ffe_lowercase_reason(self): "attributes": {}, }, ) + EXPECTED_EVAL_METRIC_FLAGS.add(self.flag_key) def test_ffe_lowercase_reason(self): """Test that reason values are lowercase.""" @@ -1060,6 +1082,8 @@ def setup_ffe_lowercase_error_type(self): "attributes": {}, }, ) + EXPECTED_EVAL_METRIC_FLAGS.add(self.error_flag_key) + wait_for_expected_eval_metrics() def test_ffe_lowercase_error_type(self): """Test that error.type values are lowercase.""" diff --git a/utils/build/docker/php/common/ffe.php b/utils/build/docker/php/common/ffe.php index 1fae050cbec..88aa42fea7c 100644 --- a/utils/build/docker/php/common/ffe.php +++ b/utils/build/docker/php/common/ffe.php @@ -19,6 +19,15 @@ function dd_ffe_error_response($statusCode, $errorCode, $errorMessage) )); } +function dd_ffe_flush_exposures() +{ + if (function_exists('DDTrace\\Testing\\flush_ffe_exposures')) { + return \DDTrace\Testing\flush_ffe_exposures(); + } + + return null; +} + function dd_ffe_read_payload() { $rawBody = file_get_contents('php://input'); @@ -186,6 +195,7 @@ function dd_ffe_details_payload($details) if ($details !== null) { $response = dd_ffe_details_payload($details); $response['count'] = count($targetingKeys); + $response['exposuresFlushed'] = dd_ffe_flush_exposures(); dd_ffe_json_response(200, $response); return; }