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 efedf10c7e3 [improvement](ci) Harden Litefuse OTLP reliability (#67416)
efedf10c7e3 is described below
commit efedf10c7e35877bc0aa36573cd1f2af621367b9
Author: shuke <[email protected]>
AuthorDate: Mon Sep 7 11:21:19 2026 +0800
[improvement](ci) Harden Litefuse OTLP reliability (#67416)
### What problem does this PR solve?
Issue Number: None
Related PR: #67413 (merged)
Problem Summary:
Large code reviews repeatedly serialize growing batches during Litefuse
pre-chunking, transient HTTP failures stop export immediately, and the
main review trace can pass read-back verification after only a small
subset of its exported observations becomes visible.
This PR makes three focused changes:
1. Makes legacy pre-chunk size accounting linear in the encoded event
data while preserving event order, exact encoded sizes, and the existing
OTLP-aware truncation and adaptive HTTP 413 handling.
2. Retries HTTP 429, 502, 503, and 504 within the configured per-export
retry budget and delay, with an HTTP retry counter alongside existing
transport and payload-size counters.
3. Strengthens read-back verification of the **main review trace**: it
must expose at least its exported observation count in unique, non-empty
IDs, with no duplicate or missing IDs, in addition to the existing I/O
and step checks.
The existing legacy-observations-first read order, v2 and trace-detail
fallbacks, and pagination remain unchanged. Main and subagent exports
both use the chunking and transport improvements; read-back verification
remains scoped to the main trace. Verifying separately exported subagent
traces is outside this PR's scope.
---
.github/scripts/emit_litefuse_otel_io.py | 69 ++++++--
.github/scripts/test_emit_litefuse_otel_io.py | 223 ++++++++++++++++++++++++++
2 files changed, 280 insertions(+), 12 deletions(-)
diff --git a/.github/scripts/emit_litefuse_otel_io.py
b/.github/scripts/emit_litefuse_otel_io.py
index 463876a89e8..e3a33f377b3 100644
--- a/.github/scripts/emit_litefuse_otel_io.py
+++ b/.github/scripts/emit_litefuse_otel_io.py
@@ -29,6 +29,9 @@ import urllib.parse
import urllib.request
+OTLP_RETRYABLE_HTTP_STATUS_CODES = frozenset((429, 502, 503, 504))
+
+
def read_text(path, max_chars, tail=False, optional=False):
try:
with open(path, "r", encoding="utf-8", errors="replace") as handle:
@@ -1138,22 +1141,22 @@ def chunk_payload(payload, max_payload_bytes,
shrink_oversized=True):
batch = payload.get("batch") or []
chunks = []
current = []
- current_bytes = json_payload_bytes({"batch": current})
+ empty_payload_bytes = json_payload_bytes({"batch": []})
+ current_bytes = empty_payload_bytes
for event in batch:
- event_payload = {"batch": [event]}
- event_bytes = json_payload_bytes(event_payload)
- if event_bytes > max_payload_bytes and shrink_oversized:
+ event_json_bytes = len(compact_json_bytes(event))
+ event_payload_bytes = empty_payload_bytes + event_json_bytes
+ if event_payload_bytes > max_payload_bytes and shrink_oversized:
event = shrink_event_for_payload(event, max_payload_bytes)
- event_payload = {"batch": [event]}
- event_bytes = json_payload_bytes(event_payload)
+ event_json_bytes = len(compact_json_bytes(event))
+ event_payload_bytes = empty_payload_bytes + event_json_bytes
- candidate = {"batch": current + [event]}
- candidate_bytes = json_payload_bytes(candidate)
+ candidate_bytes = current_bytes + event_json_bytes + (1 if current
else 0)
if current and candidate_bytes > max_payload_bytes:
chunks.append(({"batch": current}, current_bytes))
current = [event]
- current_bytes = event_bytes
+ current_bytes = event_payload_bytes
else:
current.append(event)
current_bytes = candidate_bytes
@@ -1490,6 +1493,7 @@ def post_payload(
request_sizes = []
payload_too_large_retry_count = 0
transport_retry_count = 0
+ http_retry_count = 0
post_attempt_count = 0
singleton_413_attempts = {}
singleton_413_sources = {}
@@ -1503,6 +1507,17 @@ def post_payload(
endpoint, public_key, secret_key, otlp_chunk, timeout_seconds
)
except urllib.error.HTTPError as exc:
+ if exc.code in OTLP_RETRYABLE_HTTP_STATUS_CODES:
+ http_retry_count += 1
+ if http_retry_count <= retry_attempts:
+ time.sleep(retry_sleep_seconds)
+ chunks.insert(0, (chunk, otlp_chunk, request_size))
+ continue
+ error_body = exc.read().decode("utf-8", errors="replace")
+ raise RuntimeError(
+ "Litefuse OTLP ingestion failed after HTTP retries: "
+ f"HTTP {exc.code}: {truncate_text(error_body, 4_000)}"
+ ) from exc
if exc.code != 413:
error_body = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(
@@ -1572,6 +1587,7 @@ def post_payload(
"max_request_size": max(request_sizes) if request_sizes else 0,
"payload_too_large_retries": payload_too_large_retry_count,
"transport_retries": transport_retry_count,
+ "http_retries": http_retry_count,
"success_count": success_count,
}
@@ -1696,8 +1712,13 @@ def context_events_readback_ok(input_object):
return event_count == 0 and events_value in (None, "", {})
-def verify_trace(args, public_key, secret_key, trace_id):
+def verify_trace(
+ args, public_key, secret_key, trace_id, expected_observation_count
+):
last_diagnostic = {}
+ required_observation_count = max(
+ args.min_observations, expected_observation_count
+ )
for _ in range(args.verify_attempts):
legacy_trace_error = ""
try:
@@ -1726,6 +1747,14 @@ def verify_trace(args, public_key, secret_key, trace_id):
for observation in observations
if not (observation.get("input") and observation.get("output"))
]
+ observation_ids = [
+ str(observation.get("id"))
+ for observation in observations
+ if observation.get("id")
+ ]
+ unique_observation_count = len(set(observation_ids))
+ observations_missing_id_count = len(observations) -
len(observation_ids)
+ duplicate_observation_count = len(observation_ids) -
unique_observation_count
step_observations = [
observation
for observation in observations
@@ -1799,6 +1828,10 @@ def verify_trace(args, public_key, secret_key, trace_id):
"trace_input": bool(trace_input),
"trace_output": bool(trace_output),
"observation_count": len(observations),
+ "required_observation_count": required_observation_count,
+ "unique_observation_count": unique_observation_count,
+ "observations_missing_id_count": observations_missing_id_count,
+ "duplicate_observation_count": duplicate_observation_count,
"step_observation_count": len(step_observations),
"agent_message_count": len(agent_message_observations),
"agent_message_input_keys": agent_message_input_keys,
@@ -1817,7 +1850,9 @@ def verify_trace(args, public_key, secret_key, trace_id):
[
trace_input,
trace_output,
- len(observations) >= args.min_observations,
+ unique_observation_count >= required_observation_count,
+ observations_missing_id_count == 0,
+ duplicate_observation_count == 0,
len(step_observations) >= args.min_step_observations,
not observations_missing_io,
not agent_message_observations or agent_message_structure_ok,
@@ -1829,6 +1864,10 @@ def verify_trace(args, public_key, secret_key, trace_id):
"trace_output": True,
"read_source": read_source,
"observation_count": len(observations),
+ "required_observation_count": required_observation_count,
+ "unique_observation_count": unique_observation_count,
+ "observations_missing_id_count": observations_missing_id_count,
+ "duplicate_observation_count": duplicate_observation_count,
"step_observation_count": len(step_observations),
"agent_message_count": len(agent_message_observations),
"agent_message_input_keys": agent_message_input_keys,
@@ -1992,7 +2031,13 @@ def main():
if args.verify:
try:
- result["verified"] = verify_trace(args, public_key, secret_key,
trace_id)
+ result["verified"] = verify_trace(
+ args,
+ public_key,
+ secret_key,
+ trace_id,
+ observation_count,
+ )
except Exception as exc:
result["verification_error"] = str(exc)
print(json.dumps(result, sort_keys=True))
diff --git a/.github/scripts/test_emit_litefuse_otel_io.py
b/.github/scripts/test_emit_litefuse_otel_io.py
index 2b4aea839de..3ccdb7b6d60 100644
--- a/.github/scripts/test_emit_litefuse_otel_io.py
+++ b/.github/scripts/test_emit_litefuse_otel_io.py
@@ -629,5 +629,228 @@ class LitefuseOtelExporterTest(unittest.TestCase):
)
+ def test_prechunking_encodes_each_legacy_event_once(self):
+ span_events = [
+ self.span_event(f"{index:032x}", output_size=200)
+ for index in range(1, 101)
+ ]
+ original_compact_json_bytes = MODULE.compact_json_bytes
+ encode_count = 0
+
+ def recording_compact_json_bytes(value):
+ nonlocal encode_count
+ encode_count += 1
+ return original_compact_json_bytes(value)
+
+ with mock.patch.object(
+ MODULE,
+ "compact_json_bytes",
+ side_effect=recording_compact_json_bytes,
+ ):
+ chunks = MODULE.chunk_payload(
+ {"batch": span_events}, max_payload_bytes=5_000
+ )
+
+ self.assertEqual(encode_count, len(span_events))
+ self.assertEqual(
+ [event for chunk, _size in chunks for event in chunk["batch"]],
+ span_events,
+ )
+ self.assertTrue(
+ all(
+ size == MODULE.json_payload_bytes(chunk)
+ for chunk, size in chunks
+ )
+ )
+
+ def test_retries_retryable_otlp_http_status(self):
+ trace_event = {"type": "trace-create", "body": self.trace_body()}
+ payload = {"batch": [trace_event, self.span_event("2" * 32)]}
+ unavailable = urllib.error.HTTPError(
+ "https://litefuse.example/api/public/otel/v1/traces";,
+ 503,
+ "Service Unavailable",
+ {},
+ io.BytesIO(b'{"message":"temporarily unavailable"}'),
+ )
+
+ with mock.patch.object(
+ MODULE.urllib.request,
+ "urlopen",
+ side_effect=[unavailable, FakeResponse()],
+ ):
+ status = MODULE.post_payload(
+ "https://litefuse.example/api/public/otel/v1/traces";,
+ "public",
+ "secret",
+ payload,
+ 10_000,
+ 30,
+ 3,
+ 0,
+ )
+
+ self.assertEqual(status["http_retries"], 1)
+ self.assertEqual(status["request_count"], 1)
+ self.assertEqual(status["success_count"], 1)
+
+ def test_verify_complete_main_trace_uses_legacy_observations(self):
+ args = mock.Mock(
+ base_url="https://litefuse.example";,
+ verify_attempts=1,
+ verify_sleep_seconds=0,
+ min_observations=3,
+ min_step_observations=1,
+ )
+ observations = [
+ {
+ "id": "review",
+ "name": "codex.review",
+ "input": {"prompt": "p"},
+ "output": {"text": "o"},
+ },
+ {
+ "id": "turn",
+ "name": "codex.turn",
+ "input": {"prompt": "p"},
+ "output": {"text": "o"},
+ },
+ {
+ "id": "command",
+ "name": "codex.command",
+ "input": {"command": "pwd"},
+ "output": {"status": "ok"},
+ },
+ ]
+
+ with mock.patch.object(MODULE, "fetch_trace", return_value={}),
mock.patch.object(
+ MODULE, "fetch_observations_legacy", return_value={"data":
observations}
+ ), mock.patch.object(MODULE, "fetch_observations_v2") as v2_fetch:
+ result = MODULE.verify_trace(
+ args, "public", "secret", "trace-id", len(observations)
+ )
+
+ self.assertEqual(result["read_source"], "legacy_observations")
+ self.assertEqual(result["required_observation_count"],
len(observations))
+ v2_fetch.assert_not_called()
+
+ def test_verify_rejects_partially_visible_trace(self):
+ args = mock.Mock(
+ base_url="https://litefuse.example";,
+ verify_attempts=1,
+ verify_sleep_seconds=0,
+ min_observations=1,
+ min_step_observations=1,
+ )
+ observations = [
+ {
+ "id": "review",
+ "name": "codex.review",
+ "input": {"prompt": "p"},
+ "output": {"text": "o"},
+ },
+ {
+ "id": "command",
+ "name": "codex.command",
+ "input": {"command": "pwd"},
+ "output": {"status": "ok"},
+ },
+ ]
+
+ with mock.patch.object(MODULE, "fetch_trace", return_value={}),
mock.patch.object(
+ MODULE, "fetch_observations_legacy", return_value={"data":
observations}
+ ), mock.patch.object(MODULE, "fetch_observations_v2"):
+ with self.assertRaisesRegex(
+ RuntimeError, '"required_observation_count": 3'
+ ):
+ MODULE.verify_trace(
+ args, "public", "secret", "trace-id", 3
+ )
+
+ def test_verify_rejects_duplicate_observation_ids(self):
+ args = mock.Mock(
+ base_url="https://litefuse.example";,
+ verify_attempts=1,
+ verify_sleep_seconds=0,
+ min_observations=1,
+ min_step_observations=1,
+ )
+ observations = [
+ {
+ "id": "review",
+ "name": "codex.review",
+ "input": {"prompt": "p"},
+ "output": {"text": "o"},
+ },
+ {
+ "id": "command",
+ "name": "codex.command",
+ "input": {"command": "pwd"},
+ "output": {"status": "ok"},
+ },
+ {
+ "id": "command",
+ "name": "codex.command",
+ "input": {"command": "pwd"},
+ "output": {"status": "ok"},
+ },
+ ]
+
+ with mock.patch.object(MODULE, "fetch_trace", return_value={}),
mock.patch.object(
+ MODULE, "fetch_observations_legacy", return_value={"data":
observations}
+ ), mock.patch.object(MODULE, "fetch_observations_v2"):
+ with self.assertRaisesRegex(
+ RuntimeError, '"duplicate_observation_count": 1'
+ ):
+ MODULE.verify_trace(
+ args, "public", "secret", "trace-id", 2
+ )
+
+ def test_verify_rejects_missing_or_empty_observation_ids(self):
+ args = SimpleNamespace(
+ base_url="https://litefuse.example";,
+ verify_attempts=1,
+ verify_sleep_seconds=0,
+ min_observations=1,
+ min_step_observations=1,
+ )
+ complete_observations = [
+ {
+ "id": "review",
+ "name": "codex.review",
+ "input": {"prompt": "p"},
+ "output": {"text": "o"},
+ },
+ {
+ "id": "command",
+ "name": "codex.command",
+ "input": {"command": "pwd"},
+ "output": {"status": "ok"},
+ },
+ ]
+ for id_fields in ({}, {"id": None}, {"id": ""}):
+ with self.subTest(id_fields=id_fields):
+ observations = [
+ *complete_observations,
+ {
+ **id_fields,
+ "name": "codex.command",
+ "input": {"command": "git status"},
+ "output": {"status": "ok"},
+ },
+ ]
+ with mock.patch.object(
+ MODULE, "fetch_trace", return_value={}
+ ), mock.patch.object(
+ MODULE, "fetch_observations_legacy", return_value={"data":
observations}
+ ), mock.patch.object(MODULE, "fetch_observations_v2"):
+ # The two valid IDs already satisfy the count requirement;
+ # the additional ID-less row must independently fail
verification.
+ with self.assertRaisesRegex(
+ RuntimeError, '"observations_missing_id_count": 1'
+ ):
+ MODULE.verify_trace(args, "public", "secret",
"trace-id", 2)
+
+
if __name__ == "__main__":
unittest.main()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
