From 1c460aa71f9381a205f0001c2bcb37ddd733513b Mon Sep 17 00:00:00 2001 From: Aditi Bhoge Date: Thu, 29 Oct 2026 00:42:01 -0600 Subject: [PATCH] Preserve ASIM canary visibility diagnostics on shared deadline expiry Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ingestTimeParser/README.md | 8 + .../asim_ingestion/azure/client.py | 147 ++++++++----- .../tests/test_asim_ingestion_harness.py | 205 ++++++++++++++++++ 3 files changed, 300 insertions(+), 60 deletions(-) diff --git a/.script/tests/asimParsersTest/ingestTimeParser/README.md b/.script/tests/asimParsersTest/ingestTimeParser/README.md index 10f9b803123..4577520d3a5 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/README.md +++ b/.script/tests/asimParsersTest/ingestTimeParser/README.md @@ -76,6 +76,14 @@ must use the supported stream-name format. available in `validation-result.json` when visibility detects unexpected, duplicate, or incomplete rows. +Exact-row visibility timeouts include `expectedCounts`, the last successful +`observedCounts`, `missingTimestamps` (timestamps with fewer rows than expected), +and `observedQueryTable` in the error details, including when the shared deadline +expires between polls or during settling. Before any successful observation, +the observed counts are empty and the query table is null. Shared-deadline +errors retain their original stage and message; query failures remain unchanged. +These diagnostics do not resolve the underlying canary ingestion timeout. + QTP parity stores the complete `qtp_schema`, `qtp_rows`, `itp_schema`, and `itp_rows` under `outputValidation`. It derives `intersect_schema` from shared column names, excluding the platform-managed `Type` column, and records both diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py index 176f785960d..c620add63b7 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py @@ -497,69 +497,96 @@ def wait_for_exact_rows( raise ValueError("Expected counts cannot be negative.") poll_index = 0 stable_once = False - while True: - query_table = self.query_exact_rows(table_name, timestamps) - raw_observed_counts = _query_timestamp_counts( - query_table, stage - ) - unexpected = { - timestamp: count - for timestamp, count in raw_observed_counts.items() - if timestamp not in expected_counts - } - if unexpected: - raise HarnessError( - stage, - "Exact timestamp query returned unexpected timestamps.", - details={ - "unexpected": unexpected, - "observedQueryTable": query_table, - }, + query_table = None + observed_counts: dict[str, int] = {} + try: + while True: + query_table = self.query_exact_rows(table_name, timestamps) + raw_observed_counts = _query_timestamp_counts( + query_table, stage ) - observed_counts = { - timestamp: raw_observed_counts.get(timestamp, 0) - for timestamp in expected_counts - } - overages = { - timestamp: { - "expected": expected_counts.get(timestamp, 0), - "observed": observed, + unexpected = { + timestamp: count + for timestamp, count in raw_observed_counts.items() + if timestamp not in expected_counts + } + if unexpected: + raise HarnessError( + stage, + "Exact timestamp query returned unexpected timestamps.", + details={ + "unexpected": unexpected, + "observedQueryTable": query_table, + }, + ) + observed_counts = { + timestamp: raw_observed_counts.get(timestamp, 0) + for timestamp in expected_counts } - for timestamp, observed in observed_counts.items() - if observed > expected_counts.get(timestamp, 0) + overages = { + timestamp: { + "expected": expected_counts.get(timestamp, 0), + "observed": observed, + } + for timestamp, observed in observed_counts.items() + if observed > expected_counts.get(timestamp, 0) + } + if overages: + raise HarnessError( + stage, + "Exact timestamp slice contains more rows than " + "expected for one or more timestamps.", + details={ + "overages": overages, + "observedQueryTable": query_table, + }, + ) + if observed_counts == expected_counts: + if stable_once: + return query_table + stable_once = True + self._sleep(self.settling_delay_seconds) + continue + stable_once = False + if self._remaining() <= 0: + raise HarnessError( + stage, + "Timed out waiting for exact table rows.", + details={ + "expectedCounts": expected_counts, + "observedCounts": observed_counts, + "missingTimestamps": [ + timestamp + for timestamp, count in expected_counts.items() + if observed_counts.get(timestamp, 0) < count + ], + "observedQueryTable": query_table, + }, + ) + delay = self.poll_delays[ + min(poll_index, len(self.poll_delays) - 1) + ] + poll_index += 1 + self._sleep(delay) + except HarnessError as exc: + # Only enrich locally generated deadline errors, not query failures. + if ( + exc.stage not in {"query", "deadline"} + or str(exc) != "Validation deadline expired." + or exc.status_code is not None + ): + raise + exc.details = { + "expectedCounts": expected_counts, + "observedCounts": observed_counts, + "missingTimestamps": [ + timestamp + for timestamp, count in expected_counts.items() + if observed_counts.get(timestamp, 0) < count + ], + "observedQueryTable": query_table, } - if overages: - raise HarnessError( - stage, - "Exact timestamp slice contains more rows than " - "expected for one or more timestamps.", - details={ - "overages": overages, - "observedQueryTable": query_table, - }, - ) - if observed_counts == expected_counts: - if stable_once: - return query_table - stable_once = True - self._sleep(self.settling_delay_seconds) - continue - stable_once = False - if self._remaining() <= 0: - raise HarnessError( - stage, - "Timed out waiting for exact table rows.", - details={ - "expectedCounts": expected_counts, - "observedCounts": observed_counts, - "observedQueryTable": query_table, - }, - ) - delay = self.poll_delays[ - min(poll_index, len(self.poll_delays) - 1) - ] - poll_index += 1 - self._sleep(delay) + raise def upload_rows( self, diff --git a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py index af2c134a1f3..83c36d536fc 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py @@ -349,6 +349,211 @@ def test_wait_for_rows_canonicalizes_returned_timestamp(self): self.assertEqual(1, len(table["rows"])) self.assertEqual([5], clock.sleeps) + def test_wait_for_rows_deadline_preserves_latest_observation(self): + first = "2026-10-16T19:55:00.1234567Z" + second = "2026-10-16T19:54:59.1234567Z" + third = "2026-10-16T19:54:58.1234567Z" + filtered = "2026-10-16T19:54:57.1234567Z" + columns = [{"name": "TimeGenerated", "type": "datetime"}] + latest_table = query_payload(columns, [[first], [second]])["tables"][0] + expected = {first: 1, second: 2, third: 1, filtered: 0} + harness, session, clock = self.harness( + [ + FakeResponse(200, query_payload(columns, [])), + FakeResponse(200, {"tables": [latest_table]}), + ], + deadline=3, + ) + + with self.assertRaises(HARNESS.HarnessError) as caught: + harness.wait_for_exact_rows( + "Contoso_CL", list(expected), expected, + stage="target-visibility", + ) + + self.assertEqual("query", caught.exception.stage) + self.assertEqual( + "Validation deadline expired.", str(caught.exception) + ) + self.assertIsNone(caught.exception.status_code) + self.assertEqual( + { + "expectedCounts": expected, + "observedCounts": {first: 1, second: 1, third: 0, filtered: 0}, + "missingTimestamps": [second, third], + "observedQueryTable": latest_table, + }, + caught.exception.details, + ) + self.assertEqual([1, 2], clock.sleeps) + self.assertEqual(2, len(session.calls)) + self.assertEqual(2, len(harness.request_attempts)) + + def test_wait_for_rows_deadline_before_first_query(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + harness, session, _ = self.harness([], deadline=0) + + with self.assertRaises(HARNESS.HarnessError) as caught: + harness.wait_for_exact_rows( + "Contoso_CL", [timestamp], {timestamp: 1}, + stage="target-visibility", + ) + + self.assertEqual( + { + "expectedCounts": {timestamp: 1}, + "observedCounts": {}, + "missingTimestamps": [timestamp], + "observedQueryTable": None, + }, + caught.exception.details, + ) + self.assertEqual([], session.calls) + + def test_wait_for_rows_deadline_during_settling(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + payload = query_payload( + [{"name": "TimeGenerated", "type": "datetime"}], + [[timestamp]], + ) + harness, session, _ = self.harness( + [FakeResponse(200, payload)], deadline=1, + ) + + with self.assertRaises(HARNESS.HarnessError) as caught: + harness.wait_for_exact_rows( + "Contoso_CL", [timestamp], {timestamp: 1}, + stage="target-visibility", + ) + + self.assertEqual( + {timestamp: 1}, caught.exception.details["observedCounts"] + ) + self.assertEqual([], caught.exception.details["missingTimestamps"]) + self.assertEqual( + payload["tables"][0], + caught.exception.details["observedQueryTable"], + ) + self.assertEqual(1, len(session.calls)) + + def test_wait_for_rows_deadline_after_final_response(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + payload = query_payload( + [{"name": "TimeGenerated", "type": "datetime"}], [], + ) + for rows in ([], [[timestamp]]): + with self.subTest(rows=rows): + payload["tables"][0]["rows"] = rows + harness, _, clock = self.harness([], deadline=1) + + def final_response(method, url, kwargs): + clock.value = 1 + return FakeResponse(200, payload) + + harness.session.responses.append(final_response) + with self.assertRaises(HARNESS.HarnessError) as caught: + harness.wait_for_exact_rows( + "Contoso_CL", [timestamp], {timestamp: 1}, + stage="target-visibility", + ) + + self.assertEqual( + [timestamp] if not rows else [], + caught.exception.details["missingTimestamps"], + ) + self.assertEqual( + {timestamp: len(rows)}, + caught.exception.details["observedCounts"], + ) + self.assertEqual( + payload["tables"][0], + caught.exception.details["observedQueryTable"], + ) + self.assertEqual( + "Timed out waiting for exact table rows." + if not rows else "Validation deadline expired.", + str(caught.exception), + ) + + def test_wait_for_rows_preserves_query_errors_at_deadline(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + for status, payload, message in ( + ( + 400, + {"error": {"code": "BadArgumentError"}}, + "POST failed with HTTP 400.", + ), + ( + 403, + {"error": {"code": "Forbidden"}}, + "POST failed with HTTP 403.", + ), + ( + 200, + {"tables": []}, + "Log Analytics returned no result table.", + ), + ): + with self.subTest(status=status): + harness, _, clock = self.harness( + [ + FakeResponse( + 200, + query_payload( + [{"name": "TimeGenerated", "type": "datetime"}], + [], + ), + ), + ], + deadline=2, + ) + + def query_error(method, url, kwargs): + clock.value = 2 + return FakeResponse(status, payload) + + harness.session.responses.append(query_error) + with self.assertRaises(HARNESS.HarnessError) as caught: + harness.wait_for_exact_rows( + "Contoso_CL", [timestamp], {timestamp: 1}, + stage="target-visibility", + ) + + self.assertEqual("query", caught.exception.stage) + self.assertEqual(message, str(caught.exception)) + self.assertEqual( + status if status != 200 else None, + caught.exception.status_code, + ) + self.assertEqual( + payload if status != 200 else None, + caught.exception.details, + ) + + def test_wait_for_rows_enriches_deadline_error_without_replacing_it(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + harness, _, _ = self.harness([]) + cause = requests.Timeout("query transport timed out") + error = HARNESS.HarnessError( + "deadline", "Validation deadline expired." + ) + error.__cause__ = cause + with unittest.mock.patch.object( + harness, "query_exact_rows", side_effect=error, + ): + with self.assertRaises(HARNESS.HarnessError) as caught: + harness.wait_for_exact_rows( + "Contoso_CL", [timestamp], {timestamp: 1}, + stage="target-visibility", + ) + + self.assertIs(error, caught.exception) + self.assertIs(cause, caught.exception.__cause__) + self.assertEqual("deadline", caught.exception.stage) + self.assertEqual( + {timestamp: 1}, caught.exception.details["expectedCounts"] + ) + def test_wait_for_rows_fails_on_contamination(self): harness, _, _ = self.harness( [