github-actions[bot] commented on code in PR #67413:
URL: https://github.com/apache/doris/pull/67413#discussion_r3910636121
##########
.github/scripts/emit_litefuse_otel_io.py:
##########
@@ -1126,6 +1127,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"]}
+ 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 [])
Review Comment:
[P1] Preserve near-limit singleton payloads
When an OTLP singleton is only slightly over `max_payload_bytes`, the first
fallback target is `max_payload_bytes // 2`, so a 6,022-byte request under a
6,000-byte cap gets re-encoded around 3,075 bytes and loses most of its output.
Please choose a target close to the active limit (for example,
search/binary-search the minimum truncation needed), and add a just-over-limit
regression; the current 50,000-byte oversized test will not catch this data
loss.
##########
.github/scripts/emit_litefuse_otel_io.py:
##########
@@ -19,6 +19,7 @@
import argparse
import base64
from datetime import datetime, timedelta, timezone
+import hashlib
import json
Review Comment:
[P1] Paginate v2 readback before declaring verification failure
The v2 endpoint is cursor-paginated and returns newest observations first. A
trace with more than 100 observations therefore omits the oldest `codex.review`
root from this single-page request; when the legacy endpoint is unavailable,
`verify_trace` then reports failure even though ingestion succeeded because it
cannot recover root input/output. Follow `meta.cursor` (or query the root
explicitly) and add a >100-observation regression; the reader is
`fetch_observations_v2` at source line 1507.
--
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]
