Skip to content

Preserve ASIM canary visibility diagnostics on shared deadline expiry #15488

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
adbhogeGit wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

Preserve ASIM canary visibility diagnostics on shared deadline expiry #15488

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions .script/tests/asimParsersTest/ingestTimeParser/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
147 changes: 87 additions & 60 deletions .script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
205 changes: 205 additions & 0 deletions .script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -349,6 +349,211 @@
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(
[
Expand Down
Loading