This is an automated email from the ASF dual-hosted git repository.
hello-stephen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 0cfc81c6163 [fix](ci) Avoid duplicate Litefuse verification reads
(#69013)
0cfc81c6163 is described below
commit 0cfc81c6163b79a6717e165e4c9dc1c8e27faa60
Author: shuke <[email protected]>
AuthorDate: Wed Oct 14 10:26:46 2026 +0800
[fix](ci) Avoid duplicate Litefuse verification reads (#69013)
Related PR: #69008 (independent gzip upload improvement)
Problem Summary:
Litefuse verification fetches trace detail and then unconditionally
downloads the same observations through the paginated API. When detail
already contains all observation IO, these extra reads consume the
shared 120-second verification budget and can leave later traces
unpolled.
Extract the existing completeness checks into a shared validator.
Validate trace-detail observations first and finish that trace
immediately when all checks pass. Incomplete detail still falls back to
the existing legacy/v2 observation readers. The observation count, IDs,
duplicates, IO, subagent expected IDs, root and agent-message context
checks are unchanged.
Verification remains enabled with the same shared timeout and polling
behavior. The workflow's existing `continue-on-error: true` still
isolates telemetry failures from review status. This PR changes only the
exporter and its tests; it is independent of gzip and introduces no
concurrency.
Validation:
- Reproduced on the original implementation: a deterministic nine-trace
test with eight seconds of simulated latency per request exhausts the
120-second budget and leaves the last trace unpolled. The fix verifies
all nine traces with nine detail requests in 72 simulated seconds.
- 42 exporter tests pass, including incomplete detail fallback,
IDs/duplicates/IO, agent-message context, existing pagination fallbacks
and deadline checks. The regression tests fail on the original
implementation.
- An AST comparison confirms the extracted validation body is unchanged.
- Offline replay of five saved traces (1,850 observations) from run
38323941693 verifies the same traces while reducing GET requests from 26
to 5. Serialized replay response bytes fall from 40,094,490 to
20,054,297. These are replay measurements, not a live latency benchmark.
- Live read-only check of that run's 315-observation main trace passes
complete verification with one detail GET (2.319 seconds for this
sample).
- The two affected files merge cleanly with #69008; all 44 combined
exporter tests pass. No whitespace errors.
---
.github/scripts/emit_litefuse_otel_io.py | 88 ++++++++++++--------
.github/scripts/test_emit_litefuse_otel_io.py | 112 ++++++++++++++++++++++++++
2 files changed, 168 insertions(+), 32 deletions(-)
diff --git a/.github/scripts/emit_litefuse_otel_io.py
b/.github/scripts/emit_litefuse_otel_io.py
index 5f6a44817dc..a2e8f8b3d69 100644
--- a/.github/scripts/emit_litefuse_otel_io.py
+++ b/.github/scripts/emit_litefuse_otel_io.py
@@ -1805,44 +1805,15 @@ def observation_has_expected_io(observation, target):
return all(observation.get(field) is not None for field in fields)
-def inspect_trace(args, public_key, secret_key, target, deadline):
+def inspect_observations(
+ args, target, observations, legacy_detail, read_source, legacy_trace_error
+):
result = target["result"]
- trace_id = result["trace_id"]
subagent = target["subagent"]
root_name = "codex.subagent.review" if subagent else "codex.review"
required_observation_count = max(
1 if subagent else args.min_observations, result["observation_count"]
)
- legacy_trace_error = ""
- try:
- legacy_detail = fetch_trace(
- args.base_url, public_key, secret_key, trace_id, deadline=deadline
- )
- except Exception as exc:
- legacy_detail = {}
- legacy_trace_error = type(exc).__name__
- try:
- observations_payload = fetch_observations_legacy(
- args.base_url, public_key, secret_key, trace_id,
- max_pages=max(10, required_observation_count // 100 + 1),
- deadline=deadline,
- )
- observations = observation_rows_from_v2(observations_payload)
- read_source = "legacy_observations"
- except Exception as exc:
- try:
- observations_payload = fetch_observations_v2(
- args.base_url, public_key, secret_key, trace_id,
- max_pages=max(10, required_observation_count // 1000 + 1),
- deadline=deadline,
- start_time=target.get("start_time"),
- end_time=target.get("end_time"),
- )
- observations = observation_rows_from_v2(observations_payload)
- read_source = "v2_observations"
- except Exception:
- observations = legacy_detail.get("observations") or []
- read_source = f"legacy_trace_fallback:{type(exc).__name__}"
observations_missing_io = [
observation
for observation in observations
@@ -1974,6 +1945,59 @@ def inspect_trace(args, public_key, secret_key, target,
deadline):
return ok, last_diagnostic
+def inspect_trace(args, public_key, secret_key, target, deadline):
+ result = target["result"]
+ trace_id = result["trace_id"]
+ required_observation_count = max(
+ 1 if target["subagent"] else args.min_observations,
result["observation_count"]
+ )
+ legacy_trace_error = ""
+ try:
+ legacy_detail = fetch_trace(
+ args.base_url, public_key, secret_key, trace_id, deadline=deadline
+ )
+ except Exception as exc:
+ legacy_detail = {}
+ legacy_trace_error = type(exc).__name__
+
+ # Some deployments return all observation IO in trace detail. Apply the
+ # same complete validation before downloading those observations again.
+ ok, detail_diagnostic = inspect_observations(
+ args, target, legacy_detail.get("observations") or [], legacy_detail,
+ "legacy_trace", legacy_trace_error,
+ )
+ if ok:
+ return ok, detail_diagnostic
+
+ try:
+ observations_payload = fetch_observations_legacy(
+ args.base_url, public_key, secret_key, trace_id,
+ max_pages=max(10, required_observation_count // 100 + 1),
+ deadline=deadline,
+ )
+ observations = observation_rows_from_v2(observations_payload)
+ read_source = "legacy_observations"
+ except Exception as exc:
+ try:
+ observations_payload = fetch_observations_v2(
+ args.base_url, public_key, secret_key, trace_id,
+ max_pages=max(10, required_observation_count // 1000 + 1),
+ deadline=deadline,
+ start_time=target.get("start_time"),
+ end_time=target.get("end_time"),
+ )
+ observations = observation_rows_from_v2(observations_payload)
+ read_source = "v2_observations"
+ except Exception:
+ return False, {
+ **detail_diagnostic,
+ "read_source": f"legacy_trace_fallback:{type(exc).__name__}",
+ }
+ return inspect_observations(
+ args, target, observations, legacy_detail, read_source,
legacy_trace_error
+ )
+
+
def verify_traces(args, public_key, secret_key, targets):
# All exported traces share polling rounds and one elapsed-time budget.
# Never restart the wait budget for each subagent.
diff --git a/.github/scripts/test_emit_litefuse_otel_io.py
b/.github/scripts/test_emit_litefuse_otel_io.py
index 389e42a933b..01ab250c7d3 100644
--- a/.github/scripts/test_emit_litefuse_otel_io.py
+++ b/.github/scripts/test_emit_litefuse_otel_io.py
@@ -872,6 +872,118 @@ class LitefuseOtelExporterTest(unittest.TestCase):
self.assertEqual(target["result"]["verified"]["observation_count"], 3)
self.assertNotIn("verification_diagnostic", target["result"])
+ def test_complete_trace_detail_skips_observation_endpoints(self):
+ for subagent in (False, True):
+ with self.subTest(subagent=subagent):
+ target, rows, _ = self.verification_fixture(subagent=subagent)
+ with mock.patch.object(
+ MODULE, "fetch_trace", return_value={"observations": rows}
+ ) as detail, mock.patch.object(
+ MODULE, "fetch_observations_legacy", return_value={"data":
rows}
+ ) as legacy, mock.patch.object(MODULE,
"fetch_observations_v2") as v2:
+ MODULE.verify_traces(
+ self.verification_args(), "public", "secret", [target]
+ )
+ detail.assert_called_once()
+ legacy.assert_not_called()
+ v2.assert_not_called()
+ self.assertEqual(target["result"]["verified"]["read_source"],
"legacy_trace")
+
self.assertEqual(target["result"]["verified"]["observation_count"], 3)
+
+ def
test_incomplete_trace_detail_falls_back_to_validated_observations(self):
+ for subagent in (False, True):
+ defects = ("count", "root", "duplicate", "empty_id", "input",
"output")
+ if subagent:
+ defects += ("wrong_id",)
+ for defect in defects:
+ with self.subTest(subagent=subagent, defect=defect):
+ target, rows, _ =
self.verification_fixture(subagent=subagent)
+ incomplete = [dict(row) for row in rows]
+ if defect == "count":
+ incomplete.pop()
+ elif defect == "root":
+ incomplete[0]["name"] = "codex.command"
+ elif defect == "duplicate":
+ incomplete[1]["id"] = incomplete[2]["id"]
+ elif defect == "empty_id":
+ incomplete[1]["id"] = ""
+ elif defect == "wrong_id":
+ incomplete[1]["id"] = "unexpected"
+ else:
+ incomplete[1][defect] = None
+ with mock.patch.object(
+ MODULE, "fetch_trace", return_value={"observations":
incomplete}
+ ), mock.patch.object(
+ MODULE, "fetch_observations_legacy",
return_value={"data": rows}
+ ) as legacy:
+ MODULE.verify_traces(
+ self.verification_args(), "public", "secret",
[target]
+ )
+ legacy.assert_called_once()
+ self.assertEqual(
+ target["result"]["verified"]["read_source"],
"legacy_observations"
+ )
+
+ def
test_trace_detail_does_not_bypass_agent_message_context_validation(self):
+ for complete in (False, True):
+ with self.subTest(complete=complete):
+ target, rows, _ = self.verification_fixture()
+ rows[1]["name"] = "codex.agent_message"
+ rows[1]["input"] = {"text": "missing context"}
+ if complete:
+ rows[1]["input"] = {
+ "context_window": {"event_count": 0},
+ "events_since_previous_agent_message": [],
+ "turn_input": {"prompt": "p"},
+ }
+ with mock.patch.object(
+ MODULE, "fetch_trace", return_value={"observations": rows}
+ ), mock.patch.object(
+ MODULE, "fetch_observations_legacy", return_value={"data":
rows}
+ ) as legacy:
+ if complete:
+ MODULE.verify_traces(
+ self.verification_args(), "public", "secret",
[target]
+ )
+ legacy.assert_not_called()
+ else:
+ with self.assertRaisesRegex(RuntimeError, "main"):
+ MODULE.verify_traces(
+ self.verification_args(verify_attempts=1),
+ "public", "secret", [target],
+ )
+ legacy.assert_called_once()
+
+ def test_complete_details_allow_all_traces_within_shared_deadline(self):
+ targets = []
+ replies = {}
+ for index in range(9):
+ target, rows, _ = self.verification_fixture(subagent=index > 0)
+ target["result"]["trace_id"] = f"trace-{index}"
+ targets.append(target)
+ replies[f"trace-{index}"] = rows
+ now = [0]
+ requests = []
+
+ def fake_urlopen(request, timeout):
+ # Model cumulative IO latency without sleeping in the test.
+ now[0] += 8
+ requests.append(request.full_url)
+ if "/api/public/traces/" in request.full_url:
+ trace_id = request.full_url.rsplit("/", 1)[1]
+ return JsonResponse({"observations": replies[trace_id]})
+ query =
urllib.parse.parse_qs(urllib.parse.urlparse(request.full_url).query)
+ trace_id = query["traceId"][0]
+ return JsonResponse({"data": replies[trace_id]})
+
+ with mock.patch.object(MODULE.time, "monotonic", side_effect=lambda:
now[0]), mock.patch.object(
+ MODULE.urllib.request, "urlopen", side_effect=fake_urlopen
+ ):
+ MODULE.verify_traces(self.verification_args(), "public", "secret",
targets)
+ self.assertEqual(len(requests), len(targets))
+ self.assertEqual(now[0], 72)
+ self.assertTrue(all("verified" in target["result"] for target in
targets))
+
def test_reports_all_missing_subagents_and_preserves_main_success(self):
main, main_rows, _ = self.verification_fixture()
children = [self.verification_fixture(subagent=True)[0] for _ in
range(2)]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
