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::sync::Arc;
29
30/// Default OTLP/HTTP receiver endpoint (OpenTelemetry Collector).
31const DEFAULT_OTLP_ENDPOINT: &str = "http://localhost:4318";
32
33#[derive(Clone)]
34struct SpanInput {
35    start_unix_nano: u128,
36    provider: String,
37    server_address: String,
38    model: Option<String>,
39    conversation_id: Option<String>,
40    max_tokens: Option<i64>,
41    temperature: Option<f64>,
42    top_p: Option<f64>,
43    /// Opt-in: the request messages, captured only when content capture is on.
44    input_messages: Option<String>,
45}
46
47/// Exports GenAI spans for LLM HTTP exchanges via OTLP/HTTP (JSON).
48pub struct OtelExporter {
49    /// Full traces URL, e.g. `http://localhost:4318/v1/traces`.
50    traces_url: String,
51    /// `service.name` reported on the OTLP Resource.
52    service_name: String,
53    /// Whether to attach prompt/completion content (`gen_ai.{input,output}.messages`).
54    capture_content: bool,
55    client: Arc<Client<hyper_util::client::legacy::connect::HttpConnector, Full<Bytes>>>,
56}
57
58impl OtelExporter {
59    /// Create an exporter. `endpoint` is the OTLP/HTTP base (e.g.
60    /// `http://collector:4318`); when `None`, `OTEL_EXPORTER_OTLP_ENDPOINT` is
61    /// honored, falling back to `http://localhost:4318`. A full traces endpoint
62    /// in `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` takes precedence and is used as-is.
63    pub fn new(endpoint: Option<String>, capture_content: bool) -> Self {
64        let traces_url = if let Ok(full) = std::env::var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") {
65            full
66        } else {
67            let base = endpoint
68                .or_else(|| std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok())
69                .unwrap_or_else(|| DEFAULT_OTLP_ENDPOINT.to_string());
70            format!("{}/v1/traces", base.trim_end_matches('/'))
71        };
72
73        let service_name =
74            std::env::var("OTEL_SERVICE_NAME").unwrap_or_else(|_| "agentsight".to_string());
75
76        Self {
77            traces_url,
78            service_name,
79            capture_content,
80            client: Arc::new(Client::builder(TokioExecutor::new()).build_http()),
81        }
82    }
83}
84
85/// Pull an integer token count from a usage object, tolerating both the
86/// OpenAI (`prompt_tokens`/`completion_tokens`) and Anthropic/Responses
87/// (`input_tokens`/`output_tokens`) field names.
88fn usage_int(usage: &Value, names: &[&str]) -> Option<i64> {
89    names
90        .iter()
91        .find_map(|n| usage.get(*n).and_then(|v| v.as_i64()))
92}
93
94/// Extract finish/stop reasons from a response body across provider shapes.
95fn finish_reasons(body: &Value) -> Vec<String> {
96    // OpenAI chat/completions: choices[].finish_reason
97    if let Some(choices) = body.get("choices").and_then(|c| c.as_array()) {
98        let reasons: Vec<String> = choices
99            .iter()
100            .filter_map(|c| {
101                c.get("finish_reason")
102                    .and_then(|r| r.as_str())
103                    .map(String::from)
104            })
105            .collect();
106        if !reasons.is_empty() {
107            return reasons;
108        }
109    }
110    // Anthropic messages: stop_reason
111    if let Some(stop) = body.get("stop_reason").and_then(|v| v.as_str()) {
112        return vec![stop.to_string()];
113    }
114    Vec::new()
115}
116
117/// Build an OTLP AnyValue JSON object for a string.
118fn av_str(s: &str) -> Value {
119    json!({ "stringValue": s })
120}
121
122/// Build an OTLP key/value attribute with a string value.
123fn attr_str(key: &str, s: &str) -> Value {
124    json!({ "key": key, "value": av_str(s) })
125}
126
127/// Build an OTLP key/value attribute with an int value (int64 is JSON-encoded as
128/// a string per the OTLP/JSON spec).
129fn attr_int(key: &str, n: i64) -> Value {
130    json!({ "key": key, "value": { "intValue": n.to_string() } })
131}
132
133/// Build an OTLP key/value attribute with a double value.
134fn attr_double(key: &str, n: f64) -> Value {
135    json!({ "key": key, "value": { "doubleValue": n } })
136}
137
138/// Build an OTLP key/value attribute holding an array of strings.
139fn attr_str_array(key: &str, items: &[String]) -> Value {
140    let values: Vec<Value> = items.iter().map(|s| av_str(s)).collect();
141    json!({ "key": key, "value": { "arrayValue": { "values": values } } })
142}
143
144/// Construct the OTLP/HTTP JSON `ExportTraceServiceRequest` body for a single
145/// span built from a request/response pair.
146fn build_otlp_payload(
147    service_name: &str,
148    trace_id: &str,
149    span_id: &str,
150    req: &SpanInput,
151    end_unix_nano: u128,
152    status_code: Option<u16>,
153    response_body: Option<&Value>,
154    capture_content: bool,
155) -> Value {
156    let model_name = req.model.as_deref().unwrap_or("unknown");
157
158    let mut attributes = vec![
159        attr_str("gen_ai.operation.name", "chat"),
160        attr_str("gen_ai.provider.name", &req.provider),
161        attr_str("server.address", &req.server_address),
162    ];
163    if let Some(conversation_id) = &req.conversation_id {
164        attributes.push(attr_str("gen_ai.conversation.id", conversation_id));
165    }
166    if let Some(model) = &req.model {
167        attributes.push(attr_str("gen_ai.request.model", model));
168    }
169    if let Some(mt) = req.max_tokens {
170        attributes.push(attr_int("gen_ai.request.max_tokens", mt));
171    }
172    if let Some(t) = req.temperature {
173        attributes.push(attr_double("gen_ai.request.temperature", t));
174    }
175    if let Some(p) = req.top_p {
176        attributes.push(attr_double("gen_ai.request.top_p", p));
177    }
178
179    // Response-derived attributes.
180    let mut span_status = json!({ "code": 1 }); // STATUS_CODE_OK
181    if let Some(body) = response_body {
182        if let Some(rmodel) = body.get("model").and_then(|v| v.as_str()) {
183            attributes.push(attr_str("gen_ai.response.model", rmodel));
184        }
185        if let Some(id) = body.get("id").and_then(|v| v.as_str()) {
186            attributes.push(attr_str("gen_ai.response.id", id));
187        }
188        if let Some(usage) = body.get("usage") {
189            if let Some(input) = usage_int(usage, &["input_tokens", "prompt_tokens"]) {
190                attributes.push(attr_int("gen_ai.usage.input_tokens", input));
191            }
192            if let Some(output) = usage_int(usage, &["output_tokens", "completion_tokens"]) {
193                attributes.push(attr_int("gen_ai.usage.output_tokens", output));
194            }
195        }
196        let reasons = finish_reasons(body);
197        if !reasons.is_empty() {
198            attributes.push(attr_str_array("gen_ai.response.finish_reasons", &reasons));
199        }
200        if capture_content {
201            attributes.push(attr_str("gen_ai.output.messages", &body.to_string()));
202        }
203    }
204
205    // HTTP error status -> span ERROR.
206    if let Some(code) = status_code {
207        attributes.push(attr_int("http.response.status_code", code as i64));
208        if code >= 400 {
209            span_status = json!({ "code": 2, "message": format!("HTTP {}", code) });
210        }
211    }
212
213    if capture_content && let Some(msgs) = &req.input_messages {
214        attributes.push(attr_str("gen_ai.input.messages", msgs));
215    }
216
217    json!({
218        "resourceSpans": [{
219            "resource": {
220                "attributes": [ attr_str("service.name", service_name) ]
221            },
222            "scopeSpans": [{
223                "scope": { "name": "agentsight", "version": env!("CARGO_PKG_VERSION") },
224                "spans": [{
225                    "traceId": trace_id,
226                    "spanId": span_id,
227                    "name": format!("chat {}", model_name),
228                    "kind": 3, // SPAN_KIND_CLIENT
229                    "startTimeUnixNano": req.start_unix_nano.to_string(),
230                    "endTimeUnixNano": end_unix_nano.to_string(),
231                    "attributes": attributes,
232                    "status": span_status
233                }]
234            }]
235        }]
236    })
237}
238
239/// Generate a 32-hex-char trace id and 16-hex-char span id.
240fn new_ids() -> (String, String) {
241    let trace = uuid::Uuid::new_v4().simple().to_string(); // 32 hex chars
242    let span = uuid::Uuid::new_v4().simple().to_string()[..16].to_string();
243    (trace, span)
244}
245
246impl SpanInput {
247    fn from_call(call: &LlmCallRow, capture_content: bool) -> Self {
248        let request = &call.request;
249        let host = call.host.as_deref().unwrap_or_default();
250        Self {
251            start_unix_nano: (call.start_timestamp_ms as u128) * 1_000_000,
252            provider: call
253                .provider
254                .clone()
255                .unwrap_or_else(|| provider_from_host(host)),
256            server_address: host.to_string(),
257            conversation_id: conversation_id_from_request(request),
258            model: call.model.clone().or_else(|| {
259                request
260                    .get("model")
261                    .and_then(Value::as_str)
262                    .map(String::from)
263            }),
264            max_tokens: request
265                .get("max_tokens")
266                .or_else(|| request.get("max_output_tokens"))
267                .and_then(Value::as_i64),
268            temperature: request.get("temperature").and_then(Value::as_f64),
269            top_p: request.get("top_p").and_then(Value::as_f64),
270            input_messages: capture_content
271                .then(|| {
272                    request
273                        .get("messages")
274                        .or_else(|| request.get("input"))
275                        .map(Value::to_string)
276                })
277                .flatten(),
278        }
279    }
280}
281
282impl ViewSink for OtelExporter {
283    fn llm_call(&mut self, call: &LlmCallRow) -> ViewResult<()> {
284        let Some(end_ms) = call.end_timestamp_ms else {
285            return Ok(());
286        };
287        let span_input = SpanInput::from_call(call, self.capture_content);
288        let (trace_id, span_id) = new_ids();
289        let payload = build_otlp_payload(
290            &self.service_name,
291            &trace_id,
292            &span_id,
293            &span_input,
294            (end_ms as u128) * 1_000_000,
295            call.status_code,
296            Some(&call.response),
297            self.capture_content,
298        );
299        let client = self.client.clone();
300        let url = self.traces_url.clone();
301        tokio::spawn(async move {
302            if let Err(e) = post_otlp(&client, &url, payload).await {
303                log::warn!("OtelExporter: failed to export span: {}", e);
304            }
305        });
306        Ok(())
307    }
308}
309
310fn conversation_id_from_request(request: &Value) -> Option<String> {
311    ["conversation_id", "conversationId", "thread_id", "threadId"]
312        .iter()
313        .filter_map(|key| request.get(*key).and_then(Value::as_str))
314        .find(|value| !value.is_empty())
315        .or_else(|| request.pointer("/conversation/id").and_then(Value::as_str))
316        .filter(|value| !value.is_empty())
317        .map(str::to_string)
318}
319
320/// POST an OTLP/HTTP JSON trace payload to the collector.
321async fn post_otlp(
322    client: &Client<hyper_util::client::legacy::connect::HttpConnector, Full<Bytes>>,
323    url: &str,
324    payload: Value,
325) -> Result<(), AnalyzerError> {
326    let body = serde_json::to_vec(&payload)?;
327    let req = hyper::Request::builder()
328        .method(hyper::Method::POST)
329        .uri(url)
330        .header(hyper::header::CONTENT_TYPE, "application/json")
331        .body(Full::new(Bytes::from(body)))?;
332
333    let resp = client.request(req).await?;
334    let status = resp.status();
335    if !status.is_success() {
336        let bytes = resp.into_body().collect().await?.to_bytes();
337        let text = String::from_utf8_lossy(&bytes);
338        return Err(format!("collector returned {}: {}", status, text.trim()).into());
339    }
340    Ok(())
341}
342
343#[cfg(test)]
344mod tests {
345    use super::*;
346
347    #[test]
348    fn maps_providers() {
349        assert_eq!(provider_from_host("api.openai.com"), "openai");
350        assert_eq!(provider_from_host("api.anthropic.com"), "anthropic");
351        assert_eq!(
352            provider_from_host("generativelanguage.googleapis.com"),
353            "gcp.gen_ai"
354        );
355        assert_eq!(
356            provider_from_host("my-resource.openai.azure.com"),
357            "azure.ai.openai"
358        );
359        // Unknown OpenAI-compatible host is reported as the host itself.
360        assert_eq!(provider_from_host("localhost:8443"), "localhost:8443");
361    }
362
363    #[test]
364    fn parses_usage_both_shapes() {
365        let openai = json!({ "usage": { "prompt_tokens": 12, "completion_tokens": 7 } });
366        assert_eq!(
367            usage_int(&openai["usage"], &["input_tokens", "prompt_tokens"]),
368            Some(12)
369        );
370        assert_eq!(
371            usage_int(&openai["usage"], &["output_tokens", "completion_tokens"]),
372            Some(7)
373        );
374
375        let anthropic = json!({ "usage": { "input_tokens": 30, "output_tokens": 15 } });
376        assert_eq!(
377            usage_int(&anthropic["usage"], &["input_tokens", "prompt_tokens"]),
378            Some(30)
379        );
380    }
381
382    #[test]
383    fn extracts_finish_reasons() {
384        let openai = json!({ "choices": [{ "finish_reason": "stop" }] });
385        assert_eq!(finish_reasons(&openai), vec!["stop".to_string()]);
386        let anthropic = json!({ "stop_reason": "end_turn" });
387        assert_eq!(finish_reasons(&anthropic), vec!["end_turn".to_string()]);
388        let none = json!({ "foo": 1 });
389        assert!(finish_reasons(&none).is_empty());
390    }
391
392    #[test]
393    fn builds_payload_with_gen_ai_attributes() {
394        let req = SpanInput {
395            start_unix_nano: 1_000_000_000,
396            provider: "openai".to_string(),
397            server_address: "api.openai.com".to_string(),
398            model: Some("gpt-4o".to_string()),
399            conversation_id: Some("conv_123".to_string()),
400            max_tokens: Some(256),
401            temperature: Some(0.7),
402            top_p: None,
403            input_messages: None,
404        };
405        let response = json!({
406            "model": "gpt-4o-2024",
407            "usage": { "prompt_tokens": 10, "completion_tokens": 5 },
408            "choices": [{ "finish_reason": "stop" }]
409        });
410        let payload = build_otlp_payload(
411            "agentsight",
412            "0123456789abcdef0123456789abcdef",
413            "0123456789abcdef",
414            &req,
415            2_000_000_000,
416            Some(200),
417            Some(&response),
418            false,
419        );
420
421        let span = &payload["resourceSpans"][0]["scopeSpans"][0]["spans"][0];
422        assert_eq!(span["name"], "chat gpt-4o");
423        assert_eq!(span["kind"], 3);
424        assert_eq!(span["startTimeUnixNano"], "1000000000");
425        assert_eq!(span["endTimeUnixNano"], "2000000000");
426
427        let attrs = span["attributes"].as_array().unwrap();
428        let find = |k: &str| attrs.iter().find(|a| a["key"] == k).cloned();
429        assert_eq!(
430            find("gen_ai.operation.name").unwrap()["value"]["stringValue"],
431            "chat"
432        );
433        assert_eq!(
434            find("gen_ai.provider.name").unwrap()["value"]["stringValue"],
435            "openai"
436        );
437        assert_eq!(
438            find("gen_ai.request.model").unwrap()["value"]["stringValue"],
439            "gpt-4o"
440        );
441        assert_eq!(
442            find("gen_ai.conversation.id").unwrap()["value"]["stringValue"],
443            "conv_123"
444        );
445        assert_eq!(
446            conversation_id_from_request(&json!({ "conversation": { "id": "conv_123" } }))
447                .as_deref(),
448            Some("conv_123")
449        );
450        assert_eq!(
451            conversation_id_from_request(&json!({ "session_id": "sid_123" })),
452            None
453        );
454        assert_eq!(
455            find("gen_ai.request.max_tokens").unwrap()["value"]["intValue"],
456            "256"
457        );
458        assert_eq!(
459            find("gen_ai.usage.input_tokens").unwrap()["value"]["intValue"],
460            "10"
461        );
462        assert_eq!(
463            find("gen_ai.usage.output_tokens").unwrap()["value"]["intValue"],
464            "5"
465        );
466        assert_eq!(span["status"]["code"], 1);
467        // Content not captured by default.
468        assert!(find("gen_ai.input.messages").is_none());
469    }
470
471    #[test]
472    fn error_status_marks_span_error() {
473        let req = SpanInput {
474            start_unix_nano: 1,
475            provider: "openai".to_string(),
476            server_address: "api.openai.com".to_string(),
477            model: Some("gpt-4o".to_string()),
478            conversation_id: None,
479            max_tokens: None,
480            temperature: None,
481            top_p: None,
482            input_messages: None,
483        };
484        let payload = build_otlp_payload("agentsight", "t", "s", &req, 2, Some(429), None, false);
485        let span = &payload["resourceSpans"][0]["scopeSpans"][0]["spans"][0];
486        assert_eq!(span["status"]["code"], 2);
487    }
488}
agentsight_capture/sinks/otel.rs

agentsight_capture/sinks/
otel.rs
