agentd/obs/
otel.rs

1// SPDX-License-Identifier: AGPL-3.0-only
2//! OTLP span export with the GenAI semantic conventions. [feature: otel]
3//!
4//! Hand-rolled OTLP-over-HTTP/**JSON** — no `opentelemetry` crate, no protobuf.
5//! It reuses what agentd already has: the W3C trace/span ids on every run
6//! ([`crate::obs::trace`]), `serde_json`, and the hand-rolled HTTP client
7//! ([`crate::net::http`]). So `--features otel` stays dependency-free.
8//!
9//! Off the default path: [`RunSpan`] is a **no-op handle unless built
10//! `--features otel`** (so loop call sites stay clean and the default build pays
11//! nothing). With the feature, a run records a `chat` span per model call and an
12//! `execute_tool` span per tool call, then flushes the whole trace — the
13//! `invoke_agent` run span (GenAI semconv) plus those children, one OTLP batch —
14//! to `OTEL_EXPORTER_OTLP_ENDPOINT` when it finishes. Best-effort: an export
15//! failure is logged-and-dropped; telemetry never fails a run.
16
17use std::time::{SystemTime, UNIX_EPOCH};
18
19/// Current time as unix nanoseconds — a span start/end stamp.
20pub fn now_unix_nanos() -> u128 {
21    SystemTime::now()
22        .duration_since(UNIX_EPOCH)
23        .map(|d| d.as_nanos())
24        .unwrap_or(0)
25}
26
27/// A run's span recorder. Begin it once (mints the `invoke_agent` span id under
28/// the run trace), record a `chat`/`execute_tool` child as each completes, then
29/// `finish` to flush the whole trace. Every method is a **no-op without the
30/// `otel` feature** (or without `OTEL_EXPORTER_OTLP_ENDPOINT`), so the loop wires
31/// it unconditionally with no `cfg` at the call sites.
32pub struct RunSpan {
33    #[cfg(feature = "otel")]
34    inner: Option<imp::RunSpan>,
35}
36
37/// Begin the run span. `trace_id` is the run's W3C trace id; `start_unix_nanos`
38/// stamps the `invoke_agent` span start.
39pub fn run_begin(trace_id: Option<&str>, start_unix_nanos: u128) -> RunSpan {
40    #[cfg(feature = "otel")]
41    {
42        RunSpan {
43            inner: imp::RunSpan::begin(trace_id, start_unix_nanos),
44        }
45    }
46    #[cfg(not(feature = "otel"))]
47    {
48        let _ = (trace_id, start_unix_nanos);
49        RunSpan {}
50    }
51}
52
53impl RunSpan {
54    /// Record a `chat` child span for one model call (parent = the run span).
55    pub fn record_chat(
56        &mut self,
57        model: &str,
58        input_tokens: u64,
59        output_tokens: u64,
60        ok: bool,
61        start_unix_nanos: u128,
62    ) {
63        #[cfg(feature = "otel")]
64        if let Some(i) = self.inner.as_mut() {
65            i.record_chat(model, input_tokens, output_tokens, ok, start_unix_nanos);
66        }
67        #[cfg(not(feature = "otel"))]
68        let _ = (model, input_tokens, output_tokens, ok, start_unix_nanos);
69    }
70
71    /// Record an `execute_tool` child span for one tool call (parent = the run span).
72    pub fn record_tool(&mut self, tool_name: &str, ok: bool, start_unix_nanos: u128) {
73        #[cfg(feature = "otel")]
74        if let Some(i) = self.inner.as_mut() {
75            i.record_tool(tool_name, ok, start_unix_nanos);
76        }
77        #[cfg(not(feature = "otel"))]
78        let _ = (tool_name, ok, start_unix_nanos);
79    }
80
81    /// Close the `invoke_agent` span and export the run trace (run span + every
82    /// recorded child) as one OTLP batch. No-op without the feature/endpoint.
83    pub fn finish(self, model: &str, input_tokens: u64, output_tokens: u64, ok: bool) {
84        #[cfg(feature = "otel")]
85        if let Some(i) = self.inner {
86            i.finish(model, input_tokens, output_tokens, ok);
87        }
88        #[cfg(not(feature = "otel"))]
89        let _ = (model, input_tokens, output_tokens, ok);
90    }
91}
92
93/// Buffer one log record for OTLP export. A no-op unless
94/// [`arm_logs`] installed the exporter (and always a no-op without `otel`).
95pub fn capture_log(unix_nanos: u128, level: &str, event: &str, fields: &serde_json::Value) {
96    #[cfg(feature = "otel")]
97    imp::capture_log(unix_nanos, level, event, fields);
98    #[cfg(not(feature = "otel"))]
99    let _ = (unix_nanos, level, event, fields);
100}
101
102/// Arm the OTLP **logs** exporter: a bounded buffer drained by a background
103/// thread to `<endpoint>/v1/logs`. Idempotent; a no-op without `otel`.
104pub fn arm_logs(endpoint: &str, service: &str, version: &str) {
105    #[cfg(feature = "otel")]
106    imp::arm_logs(endpoint, service, version);
107    #[cfg(not(feature = "otel"))]
108    let _ = (endpoint, service, version);
109}
110
111#[cfg(feature = "otel")]
112mod imp {
113    use crate::net::http::{self, Url};
114    use serde_json::{Value, json};
115    use std::time::Duration;
116
117    /// A finished span, ready to encode as OTLP. Times are unix nanoseconds.
118    pub(super) struct Span {
119        pub trace_id: String,
120        pub span_id: String,
121        pub parent_span_id: Option<String>,
122        pub name: String,
123        pub start_unix_nanos: u128,
124        pub end_unix_nanos: u128,
125        pub ok: bool,
126        pub attrs: Vec<(&'static str, Value)>,
127    }
128
129    /// The live recorder for one run: the `invoke_agent` span identity + the
130    /// child spans collected so far + where to ship them.
131    pub(super) struct RunSpan {
132        trace_id: String,
133        span_id: String,
134        start_unix_nanos: u128,
135        endpoint: String,
136        children: Vec<Span>,
137    }
138
139    /// OTLP `AnyValue` for a string.
140    pub(super) fn str_val(s: impl Into<String>) -> Value {
141        json!({ "stringValue": s.into() })
142    }
143
144    /// OTLP `AnyValue` for an integer (OTLP ints are stringified on the wire).
145    pub(super) fn int_val(n: u64) -> Value {
146        json!({ "intValue": n.to_string() })
147    }
148
149    impl RunSpan {
150        /// Begin a run span, or `None` if there's no endpoint / trace to anchor it.
151        pub(super) fn begin(trace_id: Option<&str>, start_unix_nanos: u128) -> Option<RunSpan> {
152            let endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
153                .ok()
154                .filter(|s| !s.is_empty())?;
155            let trace_id = trace_id?.to_string();
156            Some(RunSpan {
157                span_id: crate::obs::trace::new_span_id(),
158                trace_id,
159                start_unix_nanos,
160                endpoint,
161                children: Vec::new(),
162            })
163        }
164
165        pub(super) fn record_chat(
166            &mut self,
167            model: &str,
168            input_tokens: u64,
169            output_tokens: u64,
170            ok: bool,
171            start_unix_nanos: u128,
172        ) {
173            self.children.push(Span {
174                trace_id: self.trace_id.clone(),
175                span_id: crate::obs::trace::new_span_id(),
176                parent_span_id: Some(self.span_id.clone()),
177                name: "chat".into(),
178                start_unix_nanos,
179                end_unix_nanos: super::now_unix_nanos(),
180                ok,
181                attrs: vec![
182                    ("gen_ai.operation.name", str_val("chat")),
183                    ("gen_ai.request.model", str_val(model)),
184                    ("gen_ai.usage.input_tokens", int_val(input_tokens)),
185                    ("gen_ai.usage.output_tokens", int_val(output_tokens)),
186                ],
187            });
188        }
189
190        pub(super) fn record_tool(&mut self, tool_name: &str, ok: bool, start_unix_nanos: u128) {
191            self.children.push(Span {
192                trace_id: self.trace_id.clone(),
193                span_id: crate::obs::trace::new_span_id(),
194                parent_span_id: Some(self.span_id.clone()),
195                name: "execute_tool".into(),
196                start_unix_nanos,
197                end_unix_nanos: super::now_unix_nanos(),
198                ok,
199                attrs: vec![
200                    ("gen_ai.operation.name", str_val("execute_tool")),
201                    ("gen_ai.tool.name", str_val(tool_name)),
202                ],
203            });
204        }
205
206        /// Close the run span and export the whole trace as one OTLP batch.
207        pub(super) fn finish(
208            mut self,
209            model: &str,
210            input_tokens: u64,
211            output_tokens: u64,
212            ok: bool,
213        ) {
214            let run = Span {
215                trace_id: self.trace_id.clone(),
216                span_id: self.span_id.clone(),
217                parent_span_id: None,
218                name: "invoke_agent".into(),
219                start_unix_nanos: self.start_unix_nanos,
220                end_unix_nanos: super::now_unix_nanos(),
221                ok,
222                attrs: vec![
223                    ("gen_ai.operation.name", str_val("invoke_agent")),
224                    ("gen_ai.request.model", str_val(model)),
225                    ("gen_ai.usage.input_tokens", int_val(input_tokens)),
226                    ("gen_ai.usage.output_tokens", int_val(output_tokens)),
227                ],
228            };
229            self.children.push(run);
230            // Best-effort: telemetry export must never fail the run.
231            let _ = export(
232                &self.endpoint,
233                &to_otlp_json(&self.children, "agentd", crate::VERSION),
234            );
235        }
236    }
237
238    /// Encode spans as an OTLP `ExportTraceServiceRequest` body (`resourceSpans`).
239    pub(super) fn to_otlp_json(spans: &[Span], service: &str, version: &str) -> Value {
240        let encoded: Vec<Value> = spans.iter().map(encode_span).collect();
241        json!({
242            "resourceSpans": [{
243                "resource": { "attributes": [
244                    { "key": "service.name", "value": str_val(service) },
245                    { "key": "service.version", "value": str_val(version) },
246                ]},
247                "scopeSpans": [{ "scope": { "name": "agentd" }, "spans": encoded }]
248            }]
249        })
250    }
251
252    fn encode_span(s: &Span) -> Value {
253        let attrs: Vec<Value> = s
254            .attrs
255            .iter()
256            .map(|(k, v)| json!({ "key": k, "value": v }))
257            .collect();
258        let mut span = json!({
259            "traceId": s.trace_id,
260            "spanId": s.span_id,
261            "name": s.name,
262            "kind": 1, // SPAN_KIND_INTERNAL
263            "startTimeUnixNano": s.start_unix_nanos.to_string(),
264            "endTimeUnixNano": s.end_unix_nanos.to_string(),
265            "status": { "code": if s.ok { 1 } else { 2 } }, // OK / ERROR
266            "attributes": attrs,
267        });
268        if let Some(p) = &s.parent_span_id {
269            span["parentSpanId"] = json!(p);
270        }
271        span
272    }
273
274    /// POST the OTLP body to `<endpoint>/v1/traces` (OTLP/HTTP, JSON). `http://`
275    /// only in the default build; an `https://` collector needs `--features tls`.
276    fn export(endpoint: &str, body: &Value) -> Result<(), String> {
277        let base = endpoint.trim_end_matches('/');
278        let target = if base.ends_with("/v1/traces") {
279            base.to_string()
280        } else {
281            format!("{base}/v1/traces")
282        };
283        let url = Url::parse(&target).map_err(|e| format!("otel: bad endpoint '{target}': {e}"))?;
284        if url.is_tls() {
285            return Err("otel: https OTLP endpoints need --features tls".into());
286        }
287        let bytes = serde_json::to_vec(body).map_err(|e| e.to_string())?;
288        let mut stream = http::connect_tcp(&url.host, url.port, Duration::from_secs(5))
289            .map_err(|e| e.to_string())?;
290        let headers = [("content-type", "application/json")];
291        let resp = http::send(
292            &mut stream,
293            &url.host_header(),
294            "POST",
295            &url.path,
296            &headers,
297            &bytes,
298        )
299        .map_err(|e| e.to_string())?;
300        if resp.is_success() {
301            Ok(())
302        } else {
303            Err(format!("otel: collector returned HTTP {}", resp.status))
304        }
305    }
306
307    // ---- OTLP logs (optional) ----------------------------------------------
308    // A bounded buffer drained by a background thread to `<endpoint>/v1/logs`.
309    // The JSON-lines log surface is the primary one; this is the OTLP mirror.
310
311    use std::sync::{Mutex, OnceLock};
312
313    static LOGS: OnceLock<Mutex<Vec<Value>>> = OnceLock::new();
314
315    /// Buffer one log record (no-op unless [`arm_logs`] ran; bounded at 8192 so a
316    /// stalled/absent collector can never grow memory without bound).
317    pub(super) fn capture_log(unix_nanos: u128, level: &str, event: &str, fields: &Value) {
318        let Some(buf) = LOGS.get() else { return };
319        let mut b = buf.lock().unwrap_or_else(|e| e.into_inner());
320        if b.len() >= 8192 {
321            return;
322        }
323        b.push(json!({
324            "timeUnixNano": unix_nanos.to_string(),
325            "severityText": level.to_ascii_uppercase(),
326            "body": { "stringValue": event },
327            "attributes": [ { "key": "log.fields", "value": { "stringValue": fields.to_string() } } ],
328        }));
329    }
330
331    /// Arm the OTLP logs exporter: install the buffer + spawn a background flush
332    /// thread. Idempotent (a second call no-ops).
333    pub(super) fn arm_logs(endpoint: &str, service: &str, version: &str) {
334        if LOGS.set(Mutex::new(Vec::new())).is_err() {
335            return; // already armed
336        }
337        let (endpoint, service, version) = (
338            endpoint.to_string(),
339            service.to_string(),
340            version.to_string(),
341        );
342        std::thread::Builder::new()
343            .name("otel-logs".into())
344            .spawn(move || {
345                loop {
346                    std::thread::sleep(Duration::from_secs(5));
347                    let batch = {
348                        let mut b = LOGS
349                            .get()
350                            .expect("armed")
351                            .lock()
352                            .unwrap_or_else(|e| e.into_inner());
353                        std::mem::take(&mut *b)
354                    };
355                    if batch.is_empty() {
356                        continue;
357                    }
358                    let body = to_otlp_logs_json(&batch, &service, &version);
359                    let _ = export_signal(&endpoint, "logs", &body);
360                }
361            })
362            .ok();
363    }
364
365    pub(super) fn to_otlp_logs_json(records: &[Value], service: &str, version: &str) -> Value {
366        json!({
367            "resourceLogs": [{
368                "resource": { "attributes": [
369                    { "key": "service.name", "value": { "stringValue": service } },
370                    { "key": "service.version", "value": { "stringValue": version } },
371                ]},
372                "scopeLogs": [{ "scope": { "name": "agentd" }, "logRecords": records }]
373            }]
374        })
375    }
376
377    /// POST an OTLP body to `<endpoint>/v1/<signal>` (`traces` | `logs`).
378    fn export_signal(endpoint: &str, signal: &str, body: &Value) -> Result<(), String> {
379        let base = endpoint.trim_end_matches('/');
380        let suffix = format!("/v1/{signal}");
381        let target = if base.ends_with(&suffix) {
382            base.to_string()
383        } else {
384            format!("{base}{suffix}")
385        };
386        let url = Url::parse(&target).map_err(|e| format!("otel: bad endpoint '{target}': {e}"))?;
387        if url.is_tls() {
388            return Err("otel: https OTLP endpoints need --features tls".into());
389        }
390        let bytes = serde_json::to_vec(body).map_err(|e| e.to_string())?;
391        let mut stream = http::connect_tcp(&url.host, url.port, Duration::from_secs(5))
392            .map_err(|e| e.to_string())?;
393        let headers = [("content-type", "application/json")];
394        let resp = http::send(
395            &mut stream,
396            &url.host_header(),
397            "POST",
398            &url.path,
399            &headers,
400            &bytes,
401        )
402        .map_err(|e| e.to_string())?;
403        if resp.is_success() {
404            Ok(())
405        } else {
406            Err(format!("otel: collector returned HTTP {}", resp.status))
407        }
408    }
409
410    #[cfg(test)]
411    mod tests {
412        use super::*;
413
414        fn span() -> Span {
415            Span {
416                trace_id: "4bf92f3577b34da6a3ce929d0e0e4736".into(),
417                span_id: "00f067aa0ba902b7".into(),
418                parent_span_id: None,
419                name: "invoke_agent".into(),
420                start_unix_nanos: 1_700_000_000_000_000_000,
421                end_unix_nanos: 1_700_000_001_000_000_000,
422                ok: true,
423                attrs: vec![
424                    ("gen_ai.operation.name", str_val("invoke_agent")),
425                    ("gen_ai.usage.input_tokens", int_val(1234)),
426                ],
427            }
428        }
429
430        #[test]
431        fn otlp_json_has_the_expected_shape() {
432            let v = to_otlp_json(&[span()], "agentd", "0.1.0");
433            let rs = &v["resourceSpans"][0];
434            assert_eq!(
435                rs["resource"]["attributes"][0]["value"]["stringValue"],
436                "agentd"
437            );
438            let sp = &rs["scopeSpans"][0]["spans"][0];
439            assert_eq!(sp["traceId"], "4bf92f3577b34da6a3ce929d0e0e4736");
440            assert_eq!(sp["name"], "invoke_agent");
441            assert_eq!(sp["status"]["code"], 1); // OK
442            assert_eq!(sp["startTimeUnixNano"], "1700000000000000000"); // stringified
443            assert_eq!(sp["attributes"][1]["value"]["intValue"], "1234"); // stringified int
444            assert!(sp.get("parentSpanId").is_none());
445        }
446
447        #[test]
448        fn error_span_sets_status_error_and_parent() {
449            let mut s = span();
450            s.ok = false;
451            s.parent_span_id = Some("aaaaaaaaaaaaaaaa".into());
452            let sp = to_otlp_json(&[s], "agentd", "0.1.0");
453            let sp = &sp["resourceSpans"][0]["scopeSpans"][0]["spans"][0];
454            assert_eq!(sp["status"]["code"], 2); // ERROR
455            assert_eq!(sp["parentSpanId"], "aaaaaaaaaaaaaaaa");
456        }
457
458        #[test]
459        fn a_run_records_chat_and_tool_children_under_the_run_span() {
460            let mut run = RunSpan {
461                trace_id: "4bf92f3577b34da6a3ce929d0e0e4736".into(),
462                span_id: "00f067aa0ba902b7".into(),
463                start_unix_nanos: 1_700_000_000_000_000_000,
464                endpoint: "http://127.0.0.1:4318".into(),
465                children: Vec::new(),
466            };
467            run.record_chat("m", 10, 20, true, 1_700_000_000_000_000_000);
468            run.record_tool("resource.read", true, 1_700_000_000_500_000_000);
469            assert_eq!(run.children.len(), 2);
470
471            // The batch the exporter would ship: 2 children + the run span, all
472            // sharing the trace, children parented to the run span.
473            let mut batch = std::mem::take(&mut run.children);
474            batch.push(Span {
475                trace_id: run.trace_id.clone(),
476                span_id: run.span_id.clone(),
477                parent_span_id: None,
478                name: "invoke_agent".into(),
479                start_unix_nanos: run.start_unix_nanos,
480                end_unix_nanos: 1_700_000_001_000_000_000,
481                ok: true,
482                attrs: vec![],
483            });
484            let v = to_otlp_json(&batch, "agentd", "0.1.0");
485            let spans = &v["resourceSpans"][0]["scopeSpans"][0]["spans"];
486            assert_eq!(spans[0]["name"], "chat");
487            assert_eq!(spans[0]["parentSpanId"], "00f067aa0ba902b7");
488            assert_eq!(spans[0]["attributes"][1]["value"]["stringValue"], "m");
489            assert_eq!(spans[1]["name"], "execute_tool");
490            assert_eq!(
491                spans[1]["attributes"][1]["value"]["stringValue"],
492                "resource.read"
493            );
494            assert_eq!(spans[1]["parentSpanId"], "00f067aa0ba902b7");
495            assert_eq!(spans[2]["name"], "invoke_agent");
496            assert!(spans[2].get("parentSpanId").is_none()); // root
497            // every child shares the run trace id
498            assert_eq!(spans[0]["traceId"], spans[2]["traceId"]);
499        }
500
501        #[test]
502        fn otlp_logs_json_has_the_expected_shape() {
503            let rec = json!({
504                "timeUnixNano": "1700000000000000000",
505                "severityText": "INFO",
506                "body": { "stringValue": "turn.start" },
507                "attributes": [ { "key": "log.fields", "value": { "stringValue": "{}" } } ],
508            });
509            let v = to_otlp_logs_json(&[rec], "agentd", "2.0.0");
510            let rl = &v["resourceLogs"][0];
511            assert_eq!(
512                rl["resource"]["attributes"][0]["value"]["stringValue"],
513                "agentd"
514            );
515            let lr = &rl["scopeLogs"][0]["logRecords"][0];
516            assert_eq!(lr["body"]["stringValue"], "turn.start");
517            assert_eq!(lr["severityText"], "INFO");
518            assert_eq!(lr["timeUnixNano"], "1700000000000000000");
519        }
520    }
521}
agentd/obs/otel.rs

agentd/obs/
otel.rs
