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 317ca22ce77 [improvement](ci) Compress Litefuse review uploads with
gzip (#69008)
317ca22ce77 is described below
commit 317ca22ce77b44c8ceff3a42cf357fc380bdfa91
Author: shuke <[email protected]>
AuthorDate: Tue Oct 13 16:11:04 2026 +0800
[improvement](ci) Compress Litefuse review uploads with gzip (#69008)
Problem Summary:
Code Review Runner currently sends uncompressed OTLP JSON to Litefuse.
Large reviews can upload tens of megabytes: run 38323941693 sent
61,506,031 bytes across 26 successful POSTs. Enable gzip for the
runner's main and subagent uploads to reduce transferred bytes.
The exporter adds an opt-in `--gzip` flag using gzip level 1. Chunk
limits and HTTP 413 splitting continue to use uncompressed sizes.
Existing retry behavior and rejection checks, including HTTP 200
responses with rejected spans, are preserved. Per-trace results now
report compression, successful request body sizes on the wire, and
upload duration (including chunk preparation and retries, excluding
readback).
Litefuse remains optional telemetry: its workflow step retains
`continue-on-error: true`. Upload or readback errors are still surfaced
in logs and do not determine the review result or aggregate gate.
Uploads remain serial, and readback settings and timeouts are unchanged.
Validation:
- 40 exporter unit tests pass, including gzip round trips with Unicode
IO, uncompressed chunk limits, main/subagent readback, transport
retries, HTTP 413 splitting/shrinking, partial rejection, and HTTP
415/429/503 failures.
- `actionlint -shellcheck= -pyflakes=` passes for the runner workflow;
no whitespace errors.
- Live compatibility check against the configured Litefuse test project:
one explicitly synthetic gzip trace returned HTTP 200 and its complete
input/output matched on readback. Request body: 1,543 bytes before
compression, 576 bytes on the wire. This verifies compatibility, not
production speedup; the added metrics support measurement after rollout.
---
.github/scripts/emit_litefuse_otel_io.py | 30 ++++++-
.github/scripts/test_emit_litefuse_otel_io.py | 108 +++++++++++++++++++++++++-
.github/workflows/code-review-runner.yml | 2 +
3 files changed, 133 insertions(+), 7 deletions(-)
diff --git a/.github/scripts/emit_litefuse_otel_io.py
b/.github/scripts/emit_litefuse_otel_io.py
index 3c6cf3846cb..5f6a44817dc 100644
--- a/.github/scripts/emit_litefuse_otel_io.py
+++ b/.github/scripts/emit_litefuse_otel_io.py
@@ -18,6 +18,7 @@
import argparse
import base64
+import gzip
from datetime import datetime, timedelta, timezone
import hashlib
import json
@@ -1458,11 +1459,16 @@ def shrink_singleton_otlp_retry(
return otlp_chunks(chunk, next_limit, trace_body)
-def post_payload_once(endpoint, public_key, secret_key, payload,
timeout_seconds):
+def post_payload_once(
+ endpoint, public_key, secret_key, payload, timeout_seconds,
gzip_enabled=False
+):
auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
+ data = compact_json_bytes(payload)
+ if gzip_enabled:
+ data = gzip.compress(data, compresslevel=1, mtime=0)
request = urllib.request.Request(
endpoint,
- data=compact_json_bytes(payload),
+ data=data,
headers={
"Content-Type": "application/json",
"Authorization": f"Basic {auth}",
@@ -1472,6 +1478,8 @@ def post_payload_once(endpoint, public_key, secret_key,
payload, timeout_seconds
},
method="POST",
)
+ if gzip_enabled:
+ request.add_header("Content-Encoding", "gzip")
with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
body = response.read().decode()
detail = json.loads(body) if body else {}
@@ -1493,6 +1501,7 @@ def post_payload_once(endpoint, public_key, secret_key,
payload, timeout_seconds
return {
"status": response.status,
"success_count": otlp_span_count(payload),
+ "wire_bytes": len(data),
}
@@ -1505,10 +1514,13 @@ def post_payload(
timeout_seconds,
retry_attempts,
retry_sleep_seconds,
+ gzip_enabled=False,
):
+ upload_started = time.monotonic()
statuses = []
success_count = 0
request_sizes = []
+ wire_request_sizes = []
payload_too_large_retry_count = 0
transport_retry_count = 0
http_retry_count = 0
@@ -1522,7 +1534,8 @@ def post_payload(
post_attempt_count += 1
try:
status = post_payload_once(
- endpoint, public_key, secret_key, otlp_chunk, timeout_seconds
+ endpoint, public_key, secret_key, otlp_chunk, timeout_seconds,
+ gzip_enabled=gzip_enabled,
)
except urllib.error.HTTPError as exc:
if exc.code in OTLP_RETRYABLE_HTTP_STATUS_CODES:
@@ -1597,11 +1610,16 @@ def post_payload(
statuses.append(status["status"])
success_count += int(status.get("success_count") or 0)
request_sizes.append(request_size)
+ wire_request_sizes.append(status["wire_bytes"])
return {
+ "compression": "gzip" if gzip_enabled else "none",
+ "upload_seconds": round(time.monotonic() - upload_started, 3),
"statuses": statuses,
"request_count": len(statuses),
"post_attempt_count": post_attempt_count,
"request_sizes": request_sizes,
+ # Like request_sizes, these sizes cover successful requests only.
+ "wire_request_sizes": wire_request_sizes,
"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,
@@ -2024,6 +2042,10 @@ def parse_args():
parser.add_argument("--post-timeout-seconds", type=int, default=120)
parser.add_argument("--post-retry-attempts", type=int, default=5)
parser.add_argument("--post-retry-sleep-seconds", type=int, default=5)
+ parser.add_argument(
+ "--gzip", action="store_true",
+ help="Gzip OTLP uploads; payload size limits still apply before
compression",
+ )
parser.add_argument("--verify", action="store_true")
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--verify-attempts", type=int, default=24)
@@ -2113,6 +2135,7 @@ def main():
args.post_timeout_seconds,
args.post_retry_attempts,
args.post_retry_sleep_seconds,
+ gzip_enabled=args.gzip,
)
result["subagent_traces"] = []
for subagent_payload in subagent_payloads:
@@ -2125,6 +2148,7 @@ def main():
args.post_timeout_seconds,
args.post_retry_attempts,
args.post_retry_sleep_seconds,
+ gzip_enabled=args.gzip,
)
result["subagent_traces"].append(
{
diff --git a/.github/scripts/test_emit_litefuse_otel_io.py
b/.github/scripts/test_emit_litefuse_otel_io.py
index 6c7b31960ac..389e42a933b 100644
--- a/.github/scripts/test_emit_litefuse_otel_io.py
+++ b/.github/scripts/test_emit_litefuse_otel_io.py
@@ -18,6 +18,7 @@
import importlib.util
from contextlib import redirect_stdout
+import gzip
import io
import json
from pathlib import Path
@@ -135,8 +136,11 @@ class LitefuseOtelExporterTest(unittest.TestCase):
def reject_payloads_above(self, server_limit, request_sizes):
def fake_urlopen(request, timeout):
- request_sizes.append(len(request.data))
- if len(request.data) > server_limit:
+ data = request.data
+ if request.get_header("Content-encoding") == "gzip":
+ data = gzip.decompress(data)
+ request_sizes.append(len(data))
+ if len(data) > server_limit:
raise urllib.error.HTTPError(
request.full_url,
413,
@@ -450,11 +454,67 @@ class LitefuseOtelExporterTest(unittest.TestCase):
headers = {key.lower(): value for key, value in
captured["request"].header_items()}
self.assertEqual(headers["content-type"], "application/json")
+ self.assertNotIn("content-encoding", headers)
+ self.assertEqual(json.loads(captured["request"].data), payload)
self.assertEqual(headers["x-langfuse-ingestion-version"], "4")
self.assertEqual(headers["x-langfuse-sdk-name"], "doris-code-review")
self.assertEqual(captured["timeout"], 30)
self.assertEqual(status["success_count"], 1)
+ def test_gzip_preserves_io_and_uncompressed_chunk_limits(self):
+ events = [
+ self.span_event(f"{index:032x}", output_size=2_000)
+ for index in range(1, 5)
+ ]
+ for event in events:
+ event["body"]["input"] = {"command": "检查 gzip 上传 🌏"}
+ payload = {"batch": [{"type": "trace-create", "body":
self.trace_body()}, *events]}
+ requests = []
+
+ def fake_urlopen(request, timeout):
+ requests.append(request)
+ return FakeResponse()
+
+ with mock.patch.object(MODULE.urllib.request, "urlopen",
side_effect=fake_urlopen):
+ status = MODULE.post_payload(
+ "https://litefuse.example/api/public/otel/v1/traces";,
+ "public", "secret", payload, 8_000, 30, 3, 0,
gzip_enabled=True,
+ )
+
+ self.assertGreater(len(requests), 1)
+ raw_sizes = []
+ spans = []
+ for request in requests:
+ self.assertEqual(request.get_header("Content-encoding"), "gzip")
+ self.assertEqual(request.get_header("Content-type"),
"application/json")
+ raw = gzip.decompress(request.data)
+ raw_sizes.append(len(raw))
+ self.assertLessEqual(len(raw), 8_000)
+ self.assertLess(len(request.data), len(raw))
+ for resource in json.loads(raw)["resourceSpans"]:
+ for scope in resource["scopeSpans"]:
+ spans.extend(scope["spans"])
+ self.assertEqual(len(spans), len(events))
+ self.assertEqual(
+ [span["spanId"] for span in spans],
+ [MODULE.otel_id(event["body"]["id"], 8) for event in events],
+ )
+ for span in spans:
+ attrs = attribute_values(span)
+ self.assertEqual(
+ json.loads(attrs["langfuse.observation.input"]),
+ {"command": "检查 gzip 上传 🌏"},
+ )
+ self.assertEqual(
+ json.loads(attrs["langfuse.observation.output"])["text"], "x"
* 2_000
+ )
+ self.assertEqual(status["compression"], "gzip")
+ self.assertEqual(status["request_sizes"], raw_sizes)
+ self.assertEqual(
+ status["wire_request_sizes"], [len(request.data) for request in
requests]
+ )
+ self.assertGreaterEqual(status["upload_seconds"], 0)
+
def test_paginates_v2_observations_until_root_is_visible(self):
first_page = [{"id": f"child-{index}"} for index in range(1_000)]
responses = [
@@ -516,6 +576,7 @@ class LitefuseOtelExporterTest(unittest.TestCase):
"secret",
payload,
30,
+ gzip_enabled=True,
)
def test_splits_multi_span_chunk_after_transport_error(self):
@@ -546,12 +607,39 @@ class LitefuseOtelExporterTest(unittest.TestCase):
30,
3,
0,
+ gzip_enabled=True,
)
self.assertEqual(status["transport_retries"], 1)
self.assertEqual(status["request_count"], 2)
self.assertEqual(status["success_count"], 2)
+ def test_gzip_upload_failures_are_not_reported_as_success(self):
+ payload = {"batch": [
+ {"type": "trace-create", "body": self.trace_body()},
+ self.span_event("2" * 32),
+ ]}
+ for code, attempts in ((415, 1), (429, 2), (503, 2)):
+ with self.subTest(code=code):
+ errors = [urllib.error.HTTPError(
+ "https://litefuse.example/api/public/otel/v1/traces";,
+ code, "ingestion failed", {}, io.BytesIO(b"ingestion
failed"),
+ ) for _ in range(attempts)]
+ with mock.patch.object(
+ MODULE.urllib.request, "urlopen", side_effect=errors
+ ) as send:
+ with self.assertRaisesRegex(RuntimeError, f"HTTP {code}"):
+ MODULE.post_payload(
+
"https://litefuse.example/api/public/otel/v1/traces";,
+ "public", "secret", payload, 10_000, 30, 1, 0,
+ gzip_enabled=True,
+ )
+ self.assertEqual(send.call_count, attempts)
+ for call in send.call_args_list:
+ self.assertEqual(
+ call.args[0].get_header("Content-encoding"), "gzip"
+ )
+
def test_splits_multi_span_chunk_after_http_413(self):
trace_event = {"type": "trace-create", "body": self.trace_body()}
payload = {
@@ -583,6 +671,7 @@ class LitefuseOtelExporterTest(unittest.TestCase):
30,
3,
0,
+ gzip_enabled=True,
)
self.assertEqual(status["payload_too_large_retries"], 1)
@@ -615,6 +704,7 @@ class LitefuseOtelExporterTest(unittest.TestCase):
30,
3,
0,
+ gzip_enabled=True,
)
self.assertEqual(status["payload_too_large_retries"], 1)
@@ -652,6 +742,7 @@ class LitefuseOtelExporterTest(unittest.TestCase):
30,
5,
0,
+ gzip_enabled=True,
)
self.assertEqual(initial_size, 9_486)
@@ -687,6 +778,7 @@ class LitefuseOtelExporterTest(unittest.TestCase):
30,
3,
0,
+ gzip_enabled=True,
)
self.assertEqual(len(request_sizes), 4)
@@ -757,6 +849,7 @@ class LitefuseOtelExporterTest(unittest.TestCase):
30,
3,
0,
+ gzip_enabled=True,
)
self.assertEqual(status["http_retries"], 1)
@@ -975,7 +1068,7 @@ class LitefuseOtelExporterTest(unittest.TestCase):
self.assertEqual(timeouts, [5, 2])
def
test_main_exports_and_verifies_all_traces_with_failure_diagnostics(self):
- for mode in ("success", "missing_subagent", "main_only", "no_verify",
"dry_run"):
+ for mode in ("success", "gzip", "missing_subagent", "main_only",
"no_verify", "dry_run"):
with self.subTest(mode=mode):
argv = [
"exporter", "--input-file", "prompt", "--events-file",
"events",
@@ -988,6 +1081,8 @@ class LitefuseOtelExporterTest(unittest.TestCase):
argv.append("--verify")
if mode == "dry_run":
argv.append("--dry-run")
+ if mode == "gzip":
+ argv.append("--gzip")
with mock.patch.object(sys, "argv", argv):
args = MODULE.parse_args()
events = [
@@ -1011,7 +1106,12 @@ class LitefuseOtelExporterTest(unittest.TestCase):
def fake_urlopen(request, timeout):
requests.append(request)
if request.method == "POST":
- payload = json.loads(request.data)
+ if mode == "gzip":
+
self.assertEqual(request.get_header("Content-encoding"), "gzip")
+ payload = json.loads(gzip.decompress(request.data))
+ else:
+
self.assertIsNone(request.get_header("Content-encoding"))
+ payload = json.loads(request.data)
for resource in payload["resourceSpans"]:
for scope in resource["scopeSpans"]:
for span in scope["spans"]:
diff --git a/.github/workflows/code-review-runner.yml
b/.github/workflows/code-review-runner.yml
index 23ffcefa2c9..dc8826ac6cc 100644
--- a/.github/workflows/code-review-runner.yml
+++ b/.github/workflows/code-review-runner.yml
@@ -1137,6 +1137,7 @@ jobs:
- name: Record review I/O to Litefuse
if: ${{ always() }}
+ # Telemetry errors must not change the code review result.
continue-on-error: true
timeout-minutes: 5
env:
@@ -1188,6 +1189,7 @@ jobs:
--reasoning-effort "xhigh" \
--environment "github-actions" \
--max-payload-bytes 4000000 \
+ --gzip \
--min-observations "$min_observations" \
--min-step-observations "$min_step_observations" \
--verify-attempts 24 \
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
