github-actions[bot] commented on code in PR #67413:
URL: https://github.com/apache/doris/pull/67413#discussion_r3922577631
##########
.github/scripts/emit_litefuse_otel_io.py:
##########
@@ -1126,6 +1146,249 @@ 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"]}
Review Comment:
[P2] Include failed-turn status in payload shrinking
`statusMessage` is copied unchanged into both the Langfuse attribute and
OTLP `status.message`, but the singleton shrinker only reduces input, output,
and metadata. I reproduced a 2,050,000-character `turn.failed.error.message`:
the legacy batch is 2,051,990 bytes (below the configured 4 MB ceiling), while
the resulting `codex.turn` OTLP singleton is 4,102,802 bytes and `otlp_chunks`
raises before any POST. Please bound/shrink the status text while retaining a
truncation marker (and avoid carrying the full value twice if possible), with a
failed-turn near-limit regression.
##########
.github/scripts/emit_litefuse_otel_io.py:
##########
@@ -1189,38 +1442,52 @@ 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
+ # Retry a strictly smaller complete OTLP envelope. Using half of
the
+ # rejected size can be smaller than the fixed resource/trace
attributes
+ # and prevent a viable reduced observation from being attempted.
+ next_limit = request_size - 1
Review Comment:
[P1] Bound singleton 413 retry progression
`request_size - 1` is used as the next complete-envelope ceiling, but these
413s never consume `retry_attempts`. I reproduced a 9,486-byte singleton
against an 8,486-byte server ceiling taking 268 POSTs, and a 6,000-byte ceiling
taking 903. With the workflow allowing 4 MB requests inside a five-minute step,
a material server/local limit mismatch can hammer the endpoint until the job
times out. Please reduce adaptively toward the measured fixed-envelope floor
and enforce a 413 attempt/deadline budget; add a materially-lower-cap and
always-413 regression rather than only the current one-byte mismatch.
--
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]
