github-actions[bot] commented on code in PR #67413:
URL: https://github.com/apache/doris/pull/67413#discussion_r3910285357
##########
.github/scripts/emit_litefuse_otel_io.py:
##########
@@ -1189,38 +1421,54 @@ def post_payload(
request_sizes = []
payload_too_large_retry_count = 0
transport_retry_count = 0
- chunks = chunk_payload(payload, max_payload_bytes)
+ trace_body = trace_body_from_payload(payload)
+ chunks = otlp_chunks(payload, max_payload_bytes, trace_body)
while chunks:
- chunk, request_size = chunks.pop(0)
+ chunk, otlp_chunk, request_size = chunks.pop(0)
try:
status = post_payload_once(
- endpoint, public_key, secret_key, chunk, timeout_seconds
+ endpoint, public_key, secret_key, otlp_chunk, timeout_seconds
)
except urllib.error.HTTPError as exc:
if exc.code != 413:
- raise
+ error_body = exc.read().decode("utf-8", errors="replace")
+ raise RuntimeError(
+ "Litefuse OTLP ingestion returned "
+ f"HTTP {exc.code}: {truncate_text(error_body, 4_000)}"
+ ) from exc
payload_too_large_retry_count += 1
- chunks = (
- retry_payload_chunks_after_413(
- chunk, request_size, max_payload_bytes
- )
- + chunks
- )
+ chunk_events = chunk.get("batch") or []
+ if len(chunk_events) > 1:
+ chunks = split_otlp_chunk(
+ chunk, max_payload_bytes, trace_body
+ ) + chunks
+ continue
+ next_limit = max(1_000, min(max_payload_bytes - 1, request_size //
2))
Review Comment:
[P1] Do not make half the rejected size a hard OTLP retry ceiling
For a singleton 413, `request_size // 2` becomes the maximum for the
complete re-encoded OTLP envelope, including fixed resource/scope and repeated
trace attributes that `shrink_event_for_payload` cannot reduce. I reproduced a
4,112-byte request rejected by a 3,912-byte server cap: a moderately reduced
3,912-byte request would fit, but this path instead required the whole envelope
to fit 2,056 bytes and raised before making a second request. Please shrink the
observation content stepwise while measuring the full envelope, retry only
requests that are strictly smaller than the rejected one, and fail explicitly
at the fixed-envelope floor instead of imposing an arbitrary half-size ceiling;
add a singleton threshold regression.
##########
.github/scripts/emit_litefuse_otel_io.py:
##########
@@ -1126,6 +1127,247 @@ def chunk_payload(payload, max_payload_bytes):
return chunks
+def otel_id(value, byte_count):
+ expected_length = byte_count * 2
+ normalized = str(value or "").lower()
+ if len(normalized) == expected_length and all(
+ char in "0123456789abcdef" for char in normalized
+ ):
+ return normalized
+ return hashlib.blake2b(normalized.encode(),
digest_size=byte_count).hexdigest()
+
+
+def unix_nanos(timestamp):
+ parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+ if parsed.tzinfo is None:
+ raise ValueError(f"OpenTelemetry timestamp has no timezone:
{timestamp}")
+ delta = parsed.astimezone(timezone.utc) - datetime(1970, 1, 1,
tzinfo=timezone.utc)
+ seconds = delta.days * 86_400 + delta.seconds
+ return str(seconds * 1_000_000_000 + delta.microseconds * 1_000)
+
+
+def otel_any_value(value):
+ if isinstance(value, bool):
+ return {"boolValue": value}
+ if isinstance(value, int):
+ return {"intValue": str(value)}
+ if isinstance(value, float):
+ return {"doubleValue": value}
+ if isinstance(value, str):
+ return {"stringValue": value}
+ if isinstance(value, list) and all(
+ isinstance(item, (bool, int, float, str)) for item in value
+ ):
+ return {"arrayValue": {"values": [otel_any_value(item) for item in
value]}}
+ return {"stringValue": json_attr(value)}
+
+
+def otel_attributes(values):
+ return [
+ {"key": key, "value": otel_any_value(value)}
+ for key, value in values.items()
+ if value is not None
+ ]
+
+
+def serialized_otel_value(value):
+ if isinstance(value, str):
+ return value
+ return json_attr(value)
+
+
+def metadata_otel_attributes(prefix, metadata):
+ if not isinstance(metadata, dict):
+ return {prefix: serialized_otel_value(metadata)} if metadata is not
None else {}
+ return {
+ f"{prefix}.{key}": (
+ value if isinstance(value, (str, int)) else
serialized_otel_value(value)
+ )
+ for key, value in metadata.items()
+ if value is not None
+ }
+
+
+def trace_body_from_payload(payload):
+ for event in payload.get("batch") or []:
+ if event.get("type") == "trace-create" and
isinstance(event.get("body"), dict):
+ return event["body"]
+ raise RuntimeError("Litefuse payload is missing its trace-create context
event")
+
+
+def legacy_event_to_otel_span(event, trace_body):
+ event_type = event.get("type")
+ if event_type not in ("span-create", "generation-create"):
+ raise RuntimeError(f"Unsupported trace event for OTLP conversion:
{event_type}")
+ body = event.get("body") if isinstance(event.get("body"), dict) else {}
+ trace_id = otel_id(body.get("traceId"), 16)
+ span_id = otel_id(body.get("id"), 8)
+ parent_id = body.get("parentObservationId")
+ start_time = body.get("startTime") or event.get("timestamp")
+ end_time = body.get("endTime") or start_time
+
+ attributes = {
+ "langfuse.trace.name": trace_body.get("name"),
+ "session.id": trace_body.get("sessionId"),
+ "langfuse.trace.tags": trace_body.get("tags"),
+ "langfuse.environment": body.get("environment")
+ or trace_body.get("environment"),
+ "langfuse.observation.type": (
+ "generation" if event_type == "generation-create" else "span"
+ ),
+ "langfuse.observation.input": (
+ serialized_otel_value(body["input"])
+ if body.get("input") is not None
+ else None
+ ),
+ "langfuse.observation.output": (
+ serialized_otel_value(body["output"])
+ if body.get("output") is not None
+ else None
+ ),
+ "langfuse.observation.level": body.get("level"),
+ "langfuse.observation.status_message": body.get("statusMessage"),
+ }
+ attributes.update(
+ metadata_otel_attributes(
+ "langfuse.trace.metadata", trace_body.get("metadata") or {}
+ )
+ )
+ attributes.update(
+ metadata_otel_attributes(
+ "langfuse.observation.metadata", body.get("metadata") or {}
+ )
+ )
+ if event_type == "generation-create":
+ attributes["langfuse.observation.model.name"] = body.get("model")
+ if body.get("usageDetails") is not None:
+ attributes["langfuse.observation.usage_details"] =
serialized_otel_value(
+ body["usageDetails"]
+ )
+ if not parent_id:
+ attributes["langfuse.internal.is_app_root"] = True
+
+ span = {
+ "traceId": trace_id,
+ "spanId": span_id,
+ "name": body.get("name") or "codex.unknown",
+ "kind": 1,
+ "startTimeUnixNano": unix_nanos(start_time),
+ "endTimeUnixNano": unix_nanos(end_time),
+ "attributes": otel_attributes(attributes),
+ "status": {
+ "code": 2 if body.get("level") == "ERROR" else 1,
+ **(
+ {"message": body["statusMessage"]}
+ if body.get("statusMessage")
+ else {}
+ ),
+ },
+ "flags": 1,
+ }
+ if parent_id:
+ span["parentSpanId"] = otel_id(parent_id, 8)
+ return span
+
+
+def otlp_payload(trace_body, events):
+ spans = [legacy_event_to_otel_span(event, trace_body) for event in events]
+ return {
+ "resourceSpans": [
+ {
+ "resource": {
+ "attributes": otel_attributes(
+ {
+ "service.name": "doris-code-review",
+ "langfuse.environment":
trace_body.get("environment"),
+ }
+ )
+ },
+ "scopeSpans": [
+ {
+ "scope": {"name": "doris-litefuse-exporter",
"version": "2"},
+ "spans": spans,
+ }
+ ],
+ }
+ ]
+ }
+
+
+def otlp_span_count(payload):
+ return sum(
+ len(scope_spans.get("spans") or [])
+ for resource_spans in payload.get("resourceSpans") or []
+ for scope_spans in resource_spans.get("scopeSpans") or []
+ )
+
+
+def otlp_chunks(payload, max_payload_bytes, trace_body=None):
+ trace_body = trace_body or trace_body_from_payload(payload)
+ events = [
+ event
+ for event in payload.get("batch") or []
+ if event.get("type") in ("span-create", "generation-create")
+ ]
+ chunks = []
+
+ def add_chunk(candidate_events):
+ candidate_payload = otlp_payload(trace_body, candidate_events)
+ request_size = json_payload_bytes(candidate_payload)
+ if request_size <= max_payload_bytes:
+ chunks.append(
+ (
+ {"batch": candidate_events},
+ candidate_payload,
+ request_size,
+ )
+ )
+ return
+ if len(candidate_events) > 1:
+ middle = len(candidate_events) // 2
+ add_chunk(candidate_events[:middle])
+ add_chunk(candidate_events[middle:])
+ return
+
+ event = candidate_events[0]
+ for divisor in (2, 4, 8, 16, 32, 64):
+ target_size = max(1_000, max_payload_bytes // divisor)
+ shrunk_event = shrink_event_for_payload(event, target_size)
+ shrunk_payload = otlp_payload(trace_body, [shrunk_event])
+ shrunk_size = json_payload_bytes(shrunk_payload)
+ if shrunk_size <= max_payload_bytes:
+ chunks.append(
+ ({"batch": [shrunk_event]}, shrunk_payload, shrunk_size)
+ )
+ return
+ raise RuntimeError(
+ "Litefuse OTLP span is too large after truncation: "
+ f"{request_size} bytes > {max_payload_bytes} bytes; "
+ f"name={(event.get('body') or {}).get('name')}"
+ )
+
+ if not events:
+ raise RuntimeError("Litefuse payload contains no spans for OTLP
ingestion")
+ prechunk_limit = max(1_000, max_payload_bytes // 2)
Review Comment:
[P1] Do not shrink a span against the half-size prechunk limit
Passing `max_payload_bytes // 2` into `chunk_payload` makes it truncate a
single legacy event before `add_chunk` measures the real OTLP request against
the configured limit. I reproduced a 2,101,028-byte OTLP span under the
workflow's 4 MB cap being reduced to about 3 KB with nearly all I/O replaced by
`truncated_json`. This is reachable when an agent-message input accumulates
many context events. Please use the pre-pass only to partition multi-event
batches; measure each individual OTLP span against the full active limit before
shrinking it, and add a half-limit/full-limit boundary test.
##########
.github/scripts/emit_litefuse_otel_io.py:
##########
@@ -1126,6 +1127,247 @@ def chunk_payload(payload, max_payload_bytes):
return chunks
+def otel_id(value, byte_count):
+ expected_length = byte_count * 2
+ normalized = str(value or "").lower()
+ if len(normalized) == expected_length and all(
+ char in "0123456789abcdef" for char in normalized
+ ):
+ return normalized
+ return hashlib.blake2b(normalized.encode(),
digest_size=byte_count).hexdigest()
+
+
+def unix_nanos(timestamp):
+ parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+ if parsed.tzinfo is None:
+ raise ValueError(f"OpenTelemetry timestamp has no timezone:
{timestamp}")
+ delta = parsed.astimezone(timezone.utc) - datetime(1970, 1, 1,
tzinfo=timezone.utc)
+ seconds = delta.days * 86_400 + delta.seconds
+ return str(seconds * 1_000_000_000 + delta.microseconds * 1_000)
+
+
+def otel_any_value(value):
+ if isinstance(value, bool):
+ return {"boolValue": value}
+ if isinstance(value, int):
+ return {"intValue": str(value)}
+ if isinstance(value, float):
+ return {"doubleValue": value}
+ if isinstance(value, str):
+ return {"stringValue": value}
+ if isinstance(value, list) and all(
+ isinstance(item, (bool, int, float, str)) for item in value
+ ):
+ return {"arrayValue": {"values": [otel_any_value(item) for item in
value]}}
+ return {"stringValue": json_attr(value)}
+
+
+def otel_attributes(values):
+ return [
+ {"key": key, "value": otel_any_value(value)}
+ for key, value in values.items()
+ if value is not None
+ ]
+
+
+def serialized_otel_value(value):
+ if isinstance(value, str):
+ return value
+ return json_attr(value)
+
+
+def metadata_otel_attributes(prefix, metadata):
+ if not isinstance(metadata, dict):
+ return {prefix: serialized_otel_value(metadata)} if metadata is not
None else {}
+ return {
+ f"{prefix}.{key}": (
+ value if isinstance(value, (str, int)) else
serialized_otel_value(value)
+ )
+ for key, value in metadata.items()
+ if value is not None
+ }
+
+
+def trace_body_from_payload(payload):
+ for event in payload.get("batch") or []:
+ if event.get("type") == "trace-create" and
isinstance(event.get("body"), dict):
+ return event["body"]
+ raise RuntimeError("Litefuse payload is missing its trace-create context
event")
+
+
+def legacy_event_to_otel_span(event, trace_body):
+ event_type = event.get("type")
+ if event_type not in ("span-create", "generation-create"):
+ raise RuntimeError(f"Unsupported trace event for OTLP conversion:
{event_type}")
+ body = event.get("body") if isinstance(event.get("body"), dict) else {}
+ trace_id = otel_id(body.get("traceId"), 16)
+ span_id = otel_id(body.get("id"), 8)
+ parent_id = body.get("parentObservationId")
+ start_time = body.get("startTime") or event.get("timestamp")
+ end_time = body.get("endTime") or start_time
+
+ attributes = {
+ "langfuse.trace.name": trace_body.get("name"),
+ "session.id": trace_body.get("sessionId"),
+ "langfuse.trace.tags": trace_body.get("tags"),
+ "langfuse.environment": body.get("environment")
+ or trace_body.get("environment"),
+ "langfuse.observation.type": (
+ "generation" if event_type == "generation-create" else "span"
+ ),
+ "langfuse.observation.input": (
Review Comment:
[P1] Preserve the subagent task on the v4 root observation
This always sources OTLP input from the span body, but
`build_subagent_session_payload` stores the actual first user message only in
`trace-create.body.input`; its parentless `codex.subagent.review` span contains
just `session_file` and `thread_id`. Since the trace event is discarded, the v4
root/evaluation surface loses the overall task even though the text survives on
a child user-message observation and the placeholder root input lets read-back
look non-empty. [Langfuse v4 requires overall request/response on the root
observation](https://langfuse.com/integrations/native/opentelemetry/migration-to-v4).
Please put the task input on the subagent root while retaining the file/thread
identifiers as metadata, and cover the producer-to-OTLP path in a test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
