agentsight_capture/sinks/
otel.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2026 eunomia-bpf org.
3
4//! OpenTelemetry GenAI exporter.
5//!
6//! Maps completed materialized `llm_call` rows onto
7//! OpenTelemetry **GenAI semantic-convention** (`gen_ai.*`) spans and ships them
8//! to an OpenTelemetry Collector via **OTLP/HTTP (JSON)** — the standard wire
9//! format understood by the OTel Collector, Jaeger, Grafana Tempo, and the major
10//! observability vendors. Because AgentSight captures the traffic at the kernel
11//! (eBPF) level, this produces vendor-neutral GenAI telemetry for *any* agent
12//! binary with **zero in-process instrumentation**.
13//!
14//! Spec: <https://opentelemetry.io/docs/specs/semconv/gen-ai/>
15//!
16//! Each completed call becomes one `chat {model}` CLIENT span. Correlation and
17//! token extraction happen before this sink sees the row; this sink only maps
18//! stable view data to OTLP.
19//!
20use crate::analyzers::AnalyzerError;
21use crate::model::{LlmCallRow, ViewResult, ViewSink};
22use crate::view::llm::provider_from_host;
23use http_body_util::{BodyExt, Full};
24use hyper::body::Bytes;
25use hyper_util::client::legacy::Client;
26use hyper_util::rt::TokioExecutor;
27use serde_json::{Value, json};
28use std::collections::HashMap;
29use std::sync::Arc;
30
31/// Default OTLP/HTTP receiver endpoint (OpenTelemetry Collector).
32const DEFAULT_OTLP_ENDPOINT: &str = "http://localhost:4318";
33
34#[derive(Clone)]
35struct SpanInput {
36    start_unix_nano: u128,
37    provider: String,
38    server_address: String,
39    model: Option<String>,
40    conversation_id: Option<String>,
41    max_tokens: Option<i64>,
42    temperature: Option<f64>,
43    top_p: Option<f64>,
44    /// Opt-in: the request messages, captured only when content capture is on.
45    input_messages: Option<String>,
46}
47
48/// Exports GenAI spans for LLM HTTP exchanges via OTLP/HTTP (JSON).
49pub struct OtelExporter {
50    /// Full traces URL, e.g. `http://localhost:4318/v1/traces`.
51    traces_url: String,
52    /// `service.name` reported on the OTLP Resource.
53    service_name: String,
54    /// Whether to attach prompt/completion content (`gen_ai.{input,output}.messages`).
55    capture_content: bool,
56    /// Trace IDs assigned to conversation/session identifiers in this recording.
57    trace_ids: HashMap<String, String>,
58    /// Recording-scoped trace used when a call has no correlation identifier.
59    fallback_trace_id: String,
60    client: Arc<Client<hyper_util::client::legacy::connect::HttpConnector, Full<Bytes>>>,
61}
62
63impl OtelExporter {
64    /// Create an exporter. `endpoint` is the OTLP/HTTP base (e.g.
65    /// `http://collector:4318`); when `None`, `OTEL_EXPORTER_OTLP_ENDPOINT` is
66    /// honored, falling back to `http://localhost:4318`. A full traces endpoint
67    /// in `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` takes precedence and is used as-is.
68    pub fn new(endpoint: Option<String>, capture_content: bool) -> Self {
69        let traces_url = if let Ok(full) = std::env::var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") {
70            full
71        } else {
72            let base = endpoint
73                .or_else(|| std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok())
74                .unwrap_or_else(|| DEFAULT_OTLP_ENDPOINT.to_string());
75            format!("{}/v1/traces", base.trim_end_matches('/'))
76        };
77
78        let service_name =
79            std::env::var("OTEL_SERVICE_NAME").unwrap_or_else(|_| "agentsight".to_string());
80
81        Self {
82            traces_url,
83            service_name,
84            capture_content,
85            trace_ids: HashMap::new(),
86            fallback_trace_id: new_trace_id(),
87            client: Arc::new(Client::builder(TokioExecutor::new()).build_http()),
88        }
89    }
90
91    fn trace_id_for(&mut self, conversation_id: Option<&str>, session_id: Option<&str>) -> String {
92        let key = conversation_id
93            .filter(|id| !id.is_empty())
94            .map(|id| format!("conversation:{id}"))
95            .or_else(|| {
96                session_id
97                    .filter(|id| !id.is_empty())
98                    .map(|id| format!("session:{id}"))
99            });
100        let Some(key) = key else {
101            return self.fallback_trace_id.clone();
102        };
103        self.trace_ids
104            .entry(key)
105            .or_insert_with(new_trace_id)
106            .clone()
107    }
108}
109
110/// Pull an integer token count from a usage object, tolerating both the
111/// OpenAI (`prompt_tokens`/`completion_tokens`) and Anthropic/Responses
112/// (`input_tokens`/`output_tokens`) field names.
113fn usage_int(usage: &Value, names: &[&str]) -> Option<i64> {
114    names
115        .iter()
116        .find_map(|n| usage.get(*n).and_then(|v| v.as_i64()))
117}
118
119/// Extract finish/stop reasons from a response body across provider shapes.
120fn finish_reasons(body: &Value) -> Vec<String> {
121    // OpenAI chat/completions: choices[].finish_reason
122    if let Some(choices) = body.get("choices").and_then(|c| c.as_array()) {
123        let reasons: Vec<String> = choices
124            .iter()
125            .filter_map(|c| {
126                c.get("finish_reason")
127                    .and_then(|r| r.as_str())
128                    .map(String::from)
129            })
130            .collect();
131        if !reasons.is_empty() {
132            return reasons;
133        }
134    }
135    // Anthropic messages: stop_reason
136    if let Some(stop) = body.get("stop_reason").and_then(|v| v.as_str()) {
137        return vec![stop.to_string()];
138    }
139    Vec::new()
140}
141
142/// Build an OTLP AnyValue JSON object for a string.
143fn av_str(s: &str) -> Value {
144    json!({ "stringValue": s })
145}
146
147/// Build an OTLP key/value attribute with a string value.
148fn attr_str(key: &str, s: &str) -> Value {
149    json!({ "key": key, "value": av_str(s) })
150}
151
152/// Build an OTLP key/value attribute with an int value (int64 is JSON-encoded as
153/// a string per the OTLP/JSON spec).
154fn attr_int(key: &str, n: i64) -> Value {
155    json!({ "key": key, "value": { "intValue": n.to_string() } })
156}
157
158/// Build an OTLP key/value attribute with a double value.
159fn attr_double(key: &str, n: f64) -> Value {
160    json!({ "key": key, "value": { "doubleValue": n } })
161}
162
163/// Build an OTLP key/value attribute holding an array of strings.
164fn attr_str_array(key: &str, items: &[String]) -> Value {
165    let values: Vec<Value> = items.iter().map(|s| av_str(s)).collect();
166    json!({ "key": key, "value": { "arrayValue": { "values": values } } })
167}
168
169/// Construct the OTLP/HTTP JSON `ExportTraceServiceRequest` body for a single
170/// span built from a request/response pair.
171fn build_otlp_payload(
172    service_name: &str,
173    trace_id: &str,
174    span_id: &str,
175    req: &SpanInput,
176    end_unix_nano: u128,
177    status_code: Option<u16>,
178    response_body: Option<&Value>,
179    capture_content: bool,
180) -> Value {
181    let model_name = req.model.as_deref().unwrap_or("unknown");
182
183    let mut attributes = vec![
184        attr_str("gen_ai.operation.name", "chat"),
185        attr_str("gen_ai.provider.name", &req.provider),
186        attr_str("server.address", &req.server_address),
187    ];
188    if let Some(conversation_id) = &req.conversation_id {
189        attributes.push(attr_str("gen_ai.conversation.id", conversation_id));
190    }
191    if let Some(model) = &req.model {
192        attributes.push(attr_str("gen_ai.request.model", model));
193    }
194    if let Some(mt) = req.max_tokens {
195        attributes.push(attr_int("gen_ai.request.max_tokens", mt));
196    }
197    if let Some(t) = req.temperature {
198        attributes.push(attr_double("gen_ai.request.temperature", t));
199    }
200    if let Some(p) = req.top_p {
201        attributes.push(attr_double("gen_ai.request.top_p", p));
202    }
203
204    // Response-derived attributes.
205    let mut span_status = json!({ "code": 1 }); // STATUS_CODE_OK
206    if let Some(body) = response_body {
207        if let Some(rmodel) = body.get("model").and_then(|v| v.as_str()) {
208            attributes.push(attr_str("gen_ai.response.model", rmodel));
209        }
210        if let Some(id) = body.get("id").and_then(|v| v.as_str()) {
211            attributes.push(attr_str("gen_ai.response.id", id));
212        }
213        if let Some(usage) = body.get("usage") {
214            if let Some(input) = usage_int(usage, &["input_tokens", "prompt_tokens"]) {
215                attributes.push(attr_int("gen_ai.usage.input_tokens", input));
216            }
217            if let Some(output) = usage_int(usage, &["output_tokens", "completion_tokens"]) {
218                attributes.push(attr_int("gen_ai.usage.output_tokens", output));
219            }
220        }
221        let reasons = finish_reasons(body);
222        if !reasons.is_empty() {
223            attributes.push(attr_str_array("gen_ai.response.finish_reasons", &reasons));
224        }
225        if capture_content {
226            attributes.push(attr_str("gen_ai.output.messages", &body.to_string()));
227        }
228    }
229
230    // HTTP error status -> span ERROR.
231    if let Some(code) = status_code {
232        attributes.push(attr_int("http.response.status_code", code as i64));
233        if code >= 400 {
234            span_status = json!({ "code": 2, "message": format!("HTTP {}", code) });
235        }
236    }
237
238    if capture_content && let Some(msgs) = &req.input_messages {
239        attributes.push(attr_str("gen_ai.input.messages", msgs));
240    }
241
242    json!({
243        "resourceSpans": [{
244            "resource": {
245                "attributes": [ attr_str("service.name", service_name) ]
246            },
247            "scopeSpans": [{
248                "scope": { "name": "agentsight", "version": env!("CARGO_PKG_VERSION") },
249                "spans": [{
250                    "traceId": trace_id,
251                    "spanId": span_id,
252                    "name": format!("chat {}", model_name),
253                    "kind": 3, // SPAN_KIND_CLIENT
254                    "startTimeUnixNano": req.start_unix_nano.to_string(),
255                    "endTimeUnixNano": end_unix_nano.to_string(),
256                    "attributes": attributes,
257                    "status": span_status
258                }]
259            }]
260        }]
261    })
262}
263
264fn new_trace_id() -> String {
265    uuid::Uuid::new_v4().simple().to_string()
266}
267
268fn new_span_id() -> String {
269    uuid::Uuid::new_v4().simple().to_string()[..16].to_string()
270}
271
272impl SpanInput {
273    fn from_call(call: &LlmCallRow, capture_content: bool) -> Self {
274        let request = &call.request;
275        let host = call.host.as_deref().unwrap_or_default();
276        Self {
277            start_unix_nano: (call.start_timestamp_ms as u128) * 1_000_000,
278            provider: call
279                .provider
280                .clone()
281                .unwrap_or_else(|| provider_from_host(host)),
282            server_address: host.to_string(),
283            // The materialized row may carry a generic per-response ID here.
284            // Only fields explicitly identified as a conversation or thread
285            // are safe to use for grouping.
286            conversation_id: explicit_conversation_id(request)
287                .or_else(|| explicit_conversation_id(&call.response)),
288            model: call.model.clone().or_else(|| {
289                request
290                    .get("model")
291                    .and_then(Value::as_str)
292                    .map(String::from)
293            }),
294            max_tokens: request
295                .get("max_tokens")
296                .or_else(|| request.get("max_output_tokens"))
297                .and_then(Value::as_i64),
298            temperature: request.get("temperature").and_then(Value::as_f64),
299            top_p: request.get("top_p").and_then(Value::as_f64),
300            input_messages: capture_content
301                .then(|| {
302                    request
303                        .get("messages")
304                        .or_else(|| request.get("input"))
305                        .map(Value::to_string)
306                })
307                .flatten(),
308        }
309    }
310}
311
312impl ViewSink for OtelExporter {
313    fn llm_call(&mut self, call: &LlmCallRow) -> ViewResult<()> {
314        let Some(end_ms) = call.end_timestamp_ms else {
315            return Ok(());
316        };
317        let span_input = SpanInput::from_call(call, self.capture_content);
318        let trace_id = self.trace_id_for(
319            span_input.conversation_id.as_deref(),
320            call.session_id.as_deref(),
321        );
322        let span_id = new_span_id();
323        let payload = build_otlp_payload(
324            &self.service_name,
325            &trace_id,
326            &span_id,
327            &span_input,
328            (end_ms as u128) * 1_000_000,
329            call.status_code,
330            Some(&call.response),
331            self.capture_content,
332        );
333        let client = self.client.clone();
334        let url = self.traces_url.clone();
335        tokio::spawn(async move {
336            if let Err(e) = post_otlp(&client, &url, payload).await {
337                log::warn!("OtelExporter: failed to export span: {}", e);
338            }
339        });
340        Ok(())
341    }
342}
343
344fn explicit_conversation_id(body: &Value) -> Option<String> {
345    [
346        "/conversation_id",
347        "/conversationId",
348        "/thread_id",
349        "/threadId",
350        "/metadata/conversation_id",
351        "/metadata/conversationId",
352        "/metadata/thread_id",
353        "/metadata/threadId",
354        "/conversation/id",
355        "/thread/id",
356        "/response/conversation_id",
357        "/response/conversationId",
358        "/response/thread_id",
359        "/response/threadId",
360        "/response/conversation/id",
361        "/response/thread/id",
362    ]
363    .iter()
364    .filter_map(|path| body.pointer(path).and_then(Value::as_str))
365    .find(|value| !value.is_empty())
366    .map(str::to_string)
367}
368
369/// POST an OTLP/HTTP JSON trace payload to the collector.
370async fn post_otlp(
371    client: &Client<hyper_util::client::legacy::connect::HttpConnector, Full<Bytes>>,
372    url: &str,
373    payload: Value,
374) -> Result<(), AnalyzerError> {
375    let body = serde_json::to_vec(&payload)?;
376    let req = hyper::Request::builder()
377        .method(hyper::Method::POST)
378        .uri(url)
379        .header(hyper::header::CONTENT_TYPE, "application/json")
380        .body(Full::new(Bytes::from(body)))?;
381
382    let resp = client.request(req).await?;
383    let status = resp.status();
384    if !status.is_success() {
385        let bytes = resp.into_body().collect().await?.to_bytes();
386        let text = String::from_utf8_lossy(&bytes);
387        return Err(format!("collector returned {}: {}", status, text.trim()).into());
388    }
389    Ok(())
390}
391
392#[cfg(test)]
393mod tests {
394    use super::*;
395
396    fn completed_call(
397        id: &str,
398        session_id: Option<&str>,
399        conversation_id: Option<&str>,
400        request: Value,
401    ) -> LlmCallRow {
402        LlmCallRow {
403            id: id.to_string(),
404            session_id: session_id.map(str::to_string),
405            conversation_id: conversation_id.map(str::to_string),
406            start_timestamp_ms: 1,
407            end_timestamp_ms: Some(2),
408            pid: Some(1),
409            comm: Some("agent".to_string()),
410            provider: Some("openai".to_string()),
411            model: Some("model".to_string()),
412            call_kind: Some("chat".to_string()),
413            status: "complete".to_string(),
414            error_type: None,
415            finish_reason: None,
416            host: Some("api.openai.com".to_string()),
417            path: Some("/v1/chat/completions".to_string()),
418            status_code: Some(200),
419            input_tokens: 0,
420            output_tokens: 0,
421            total_tokens: 0,
422            request,
423            response: Value::Null,
424        }
425    }
426
427    #[test]
428    fn maps_providers() {
429        assert_eq!(provider_from_host("api.openai.com"), "openai");
430        assert_eq!(provider_from_host("api.anthropic.com"), "anthropic");
431        assert_eq!(
432            provider_from_host("generativelanguage.googleapis.com"),
433            "gcp.gen_ai"
434        );
435        assert_eq!(
436            provider_from_host("my-resource.openai.azure.com"),
437            "azure.ai.openai"
438        );
439        // Unknown OpenAI-compatible host is reported as the host itself.
440        assert_eq!(provider_from_host("localhost:8443"), "localhost:8443");
441    }
442
443    #[test]
444    fn parses_usage_both_shapes() {
445        let openai = json!({ "usage": { "prompt_tokens": 12, "completion_tokens": 7 } });
446        assert_eq!(
447            usage_int(&openai["usage"], &["input_tokens", "prompt_tokens"]),
448            Some(12)
449        );
450        assert_eq!(
451            usage_int(&openai["usage"], &["output_tokens", "completion_tokens"]),
452            Some(7)
453        );
454
455        let anthropic = json!({ "usage": { "input_tokens": 30, "output_tokens": 15 } });
456        assert_eq!(
457            usage_int(&anthropic["usage"], &["input_tokens", "prompt_tokens"]),
458            Some(30)
459        );
460    }
461
462    #[test]
463    fn extracts_finish_reasons() {
464        let openai = json!({ "choices": [{ "finish_reason": "stop" }] });
465        assert_eq!(finish_reasons(&openai), vec!["stop".to_string()]);
466        let anthropic = json!({ "stop_reason": "end_turn" });
467        assert_eq!(finish_reasons(&anthropic), vec!["end_turn".to_string()]);
468        let none = json!({ "foo": 1 });
469        assert!(finish_reasons(&none).is_empty());
470    }
471
472    #[test]
473    fn builds_payload_with_gen_ai_attributes() {
474        let req = SpanInput {
475            start_unix_nano: 1_000_000_000,
476            provider: "openai".to_string(),
477            server_address: "api.openai.com".to_string(),
478            model: Some("gpt-4o".to_string()),
479            conversation_id: Some("conv_123".to_string()),
480            max_tokens: Some(256),
481            temperature: Some(0.7),
482            top_p: None,
483            input_messages: None,
484        };
485        let response = json!({
486            "model": "gpt-4o-2024",
487            "usage": { "prompt_tokens": 10, "completion_tokens": 5 },
488            "choices": [{ "finish_reason": "stop" }]
489        });
490        let payload = build_otlp_payload(
491            "agentsight",
492            "0123456789abcdef0123456789abcdef",
493            "0123456789abcdef",
494            &req,
495            2_000_000_000,
496            Some(200),
497            Some(&response),
498            false,
499        );
500
501        let span = &payload["resourceSpans"][0]["scopeSpans"][0]["spans"][0];
502        assert_eq!(span["name"], "chat gpt-4o");
503        assert_eq!(span["kind"], 3);
504        assert_eq!(span["startTimeUnixNano"], "1000000000");
505        assert_eq!(span["endTimeUnixNano"], "2000000000");
506
507        let attrs = span["attributes"].as_array().unwrap();
508        let find = |k: &str| attrs.iter().find(|a| a["key"] == k).cloned();
509        assert_eq!(
510            find("gen_ai.operation.name").unwrap()["value"]["stringValue"],
511            "chat"
512        );
513        assert_eq!(
514            find("gen_ai.provider.name").unwrap()["value"]["stringValue"],
515            "openai"
516        );
517        assert_eq!(
518            find("gen_ai.request.model").unwrap()["value"]["stringValue"],
519            "gpt-4o"
520        );
521        assert_eq!(
522            find("gen_ai.conversation.id").unwrap()["value"]["stringValue"],
523            "conv_123"
524        );
525        assert_eq!(
526            explicit_conversation_id(&json!({ "conversation": { "id": "conv_123" } })).as_deref(),
527            Some("conv_123")
528        );
529        assert_eq!(
530            explicit_conversation_id(&json!({ "session_id": "sid_123" })),
531            None
532        );
533        assert_eq!(
534            find("gen_ai.request.max_tokens").unwrap()["value"]["intValue"],
535            "256"
536        );
537        assert_eq!(
538            find("gen_ai.usage.input_tokens").unwrap()["value"]["intValue"],
539            "10"
540        );
541        assert_eq!(
542            find("gen_ai.usage.output_tokens").unwrap()["value"]["intValue"],
543            "5"
544        );
545        assert_eq!(span["status"]["code"], 1);
546        // Content not captured by default.
547        assert!(find("gen_ai.input.messages").is_none());
548    }
549
550    #[test]
551    fn error_status_marks_span_error() {
552        let req = SpanInput {
553            start_unix_nano: 1,
554            provider: "openai".to_string(),
555            server_address: "api.openai.com".to_string(),
556            model: Some("gpt-4o".to_string()),
557            conversation_id: None,
558            max_tokens: None,
559            temperature: None,
560            top_p: None,
561            input_messages: None,
562        };
563        let payload = build_otlp_payload("agentsight", "t", "s", &req, 2, Some(429), None, false);
564        let span = &payload["resourceSpans"][0]["scopeSpans"][0]["spans"][0];
565        assert_eq!(span["status"]["code"], 2);
566    }
567
568    #[test]
569    fn correlates_trace_ids_by_explicit_conversation_then_session() {
570        let mut exporter = OtelExporter::new(Some("http://localhost:4318".to_string()), false);
571
572        let conversation_a = exporter.trace_id_for(Some("conversation-a"), Some("session-a"));
573        assert_eq!(
574            conversation_a,
575            exporter.trace_id_for(Some("conversation-a"), Some("session-b"))
576        );
577        assert_ne!(
578            conversation_a,
579            exporter.trace_id_for(Some("conversation-b"), Some("session-a"))
580        );
581
582        let session_a = exporter.trace_id_for(None, Some("session-a"));
583        assert_eq!(session_a, exporter.trace_id_for(None, Some("session-a")));
584        assert_ne!(session_a, exporter.trace_id_for(None, Some("session-b")));
585    }
586
587    #[test]
588    fn per_response_conversation_ids_do_not_split_one_session() {
589        let mut exporter = OtelExporter::new(Some("http://localhost:4318".to_string()), false);
590        let first = completed_call("first", Some("session-a"), Some("chatcmpl-a"), json!({}));
591        let second = completed_call("second", Some("session-a"), Some("chatcmpl-b"), json!({}));
592
593        let first_input = SpanInput::from_call(&first, false);
594        let second_input = SpanInput::from_call(&second, false);
595        assert!(first_input.conversation_id.is_none());
596        assert_eq!(
597            exporter.trace_id_for(
598                first_input.conversation_id.as_deref(),
599                first.session_id.as_deref()
600            ),
601            exporter.trace_id_for(
602                second_input.conversation_id.as_deref(),
603                second.session_id.as_deref()
604            )
605        );
606    }
607
608    #[test]
609    fn explicit_conversation_overrides_different_sessions() {
610        let mut exporter = OtelExporter::new(Some("http://localhost:4318".to_string()), false);
611        let first = completed_call(
612            "first",
613            Some("session-a"),
614            None,
615            json!({ "metadata": { "conversation_id": "conversation-a" } }),
616        );
617        let mut second = completed_call("second", Some("session-b"), None, json!({}));
618        second.response = json!({ "thread": { "id": "conversation-a" } });
619
620        let first_input = SpanInput::from_call(&first, false);
621        let second_input = SpanInput::from_call(&second, false);
622        assert_eq!(
623            first_input.conversation_id.as_deref(),
624            Some("conversation-a")
625        );
626        assert_eq!(first_input.conversation_id, second_input.conversation_id);
627        assert_eq!(
628            exporter.trace_id_for(
629                first_input.conversation_id.as_deref(),
630                first.session_id.as_deref()
631            ),
632            exporter.trace_id_for(
633                second_input.conversation_id.as_deref(),
634                second.session_id.as_deref()
635            )
636        );
637    }
638
639    #[test]
640    fn uses_one_recording_trace_without_correlation_ids() {
641        let mut exporter = OtelExporter::new(Some("http://localhost:4318".to_string()), false);
642        assert_eq!(
643            exporter.trace_id_for(None, None),
644            exporter.trace_id_for(None, None)
645        );
646        assert_ne!(new_span_id(), new_span_id());
647    }
648}
agentsight_capture/sinks/otel.rs

agentsight_capture/sinks/
otel.rs
