superfluid_daemon/
otlp.rs

1use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
2use std::sync::mpsc::{sync_channel, Receiver, RecvTimeoutError, SyncSender, TrySendError};
3use std::sync::{Arc, Condvar, Mutex, OnceLock};
4use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
5
6use serde_json::{json, Value};
7use tracing::field::{Field, Visit};
8use tracing::span::{Attributes, Id, Record};
9use tracing::Subscriber;
10use tracing_subscriber::layer::Context;
11use tracing_subscriber::registry::LookupSpan;
12use tracing_subscriber::Layer;
13
14use crate::export;
15
16pub static EXPORTED: AtomicU64 = AtomicU64::new(0);
17pub static DROPPED_QUEUE_FULL: AtomicU64 = AtomicU64::new(0);
18pub static DROPPED_REJECTED: AtomicU64 = AtomicU64::new(0);
19pub static DROPPED_EXPORT_FAILED: AtomicU64 = AtomicU64::new(0);
20
21pub const DEFAULT_FILTER: &str = "info,superfluid_daemon::scheduler=debug";
22
23#[derive(Debug, Clone)]
24pub struct OtlpConfig {
25    pub endpoint: String,
26    pub headers: Vec<(String, String)>,
27    pub service_name: String,
28    pub resource: Vec<(String, String)>,
29    pub filter: String,
30    pub queue: usize,
31    pub batch_max: usize,
32    pub interval: Duration,
33    pub timeout: Duration,
34}
35
36impl Default for OtlpConfig {
37    fn default() -> Self {
38        OtlpConfig {
39            endpoint: String::new(),
40            headers: Vec::new(),
41            service_name: "superfluid".into(),
42            resource: Vec::new(),
43            filter: DEFAULT_FILTER.into(),
44            queue: 8192,
45            batch_max: 512,
46            interval: Duration::from_millis(1000),
47            timeout: Duration::from_millis(5000),
48        }
49    }
50}
51
52impl OtlpConfig {
53    pub fn traces_url(&self) -> String {
54        let e = self.endpoint.trim_end_matches('/');
55        if e.ends_with("/v1/traces") {
56            e.to_string()
57        } else {
58            format!("{e}/v1/traces")
59        }
60    }
61
62    pub fn validate(&self) -> Result<(), String> {
63        validate_http_url(&self.endpoint)?;
64        for (k, v) in &self.headers {
65            ureq::http::HeaderName::from_bytes(k.as_bytes())
66                .map_err(|_| format!("--otlp-header: bad header name {k:?}"))?;
67            ureq::http::HeaderValue::from_str(v)
68                .map_err(|_| format!("--otlp-header: bad value for {k:?}"))?;
69        }
70        tracing_subscriber::EnvFilter::try_new(&self.filter)
71            .map_err(|e| format!("--otlp-filter {:?}: {e}", self.filter))?;
72        if self.queue == 0 || self.batch_max == 0 {
73            return Err("--otlp-queue and the batch size must be > 0".into());
74        }
75        if self.interval.is_zero() {
76            return Err("--otlp-batch-ms must be > 0".into());
77        }
78        if self.timeout.is_zero() {
79            return Err("--otlp-timeout-ms must be > 0".into());
80        }
81        Ok(())
82    }
83
84    pub fn is_loopback(&self) -> bool {
85        let rest = self.endpoint.trim_start_matches("http://");
86        let host = rest.split('/').next().unwrap_or("");
87        let host = match host.strip_prefix('[') {
88            Some(v6) => v6.split(']').next().unwrap_or(""),
89            None => host.rsplit_once(':').map_or(host, |(h, _)| h),
90        };
91        host == "localhost"
92            || host
93                .parse::<std::net::IpAddr>()
94                .map(|ip| ip.is_loopback())
95                .unwrap_or(false)
96    }
97}
98
99pub fn validate_http_url(e: &str) -> Result<(), String> {
100    if e.starts_with("https://") {
101        return Err(format!(
102            "{e}: https is not supported (this build's HTTP client has no TLS) — \
103             point it at a local OpenTelemetry collector over http:// and let it \
104             forward with TLS"
105        ));
106    }
107    if !e.starts_with("http://") {
108        return Err(format!("{e}: expected an http:// URL"));
109    }
110    let uri: ureq::http::Uri = e
111        .parse()
112        .map_err(|err| format!("{e}: not a valid URL: {err}"))?;
113    let auth = uri
114        .authority()
115        .ok_or_else(|| format!("{e}: missing host"))?;
116    if auth.host().is_empty() {
117        return Err(format!("{e}: missing host"));
118    }
119    if let Some((_, port)) = auth.as_str().rsplit_once(':') {
120        if !auth.as_str().ends_with(']') && port.parse::<u16>().is_err() {
121            return Err(format!("{e}: bad port {port:?}"));
122        }
123    }
124    if uri.query().is_some() {
125        return Err(format!("{e}: a query string is not allowed"));
126    }
127    Ok(())
128}
129
130#[derive(Debug, Clone, Default)]
131pub struct TraceScope {
132    pub model: String,
133    pub ns: u64,
134}
135
136impl TraceScope {
137    pub fn model(model: &str) -> TraceScope {
138        TraceScope {
139            model: model.to_string(),
140            ns: 0,
141        }
142    }
143}
144
145const STRING_FIELDS: &[&str] = &["model"];
146
147#[derive(Debug, Clone)]
148enum Val {
149    I(i64),
150    U(u64),
151    F(f64),
152    B(bool),
153    S(String),
154}
155
156struct SpanData {
157    trace: [u8; 16],
158    span: [u8; 8],
159    parent: Option<[u8; 8]>,
160    start_ns: u64,
161    attrs: Vec<(&'static str, Val)>,
162    links: Vec<([u8; 16], [u8; 8])>,
163}
164
165struct Finished {
166    name: &'static str,
167    target: &'static str,
168    end_ns: u64,
169    data: SpanData,
170}
171
172struct Fields<'a>(&'a mut Vec<(&'static str, Val)>, bool);
173
174fn daemon_owned(target: &str) -> bool {
175    target == "superfluid_daemon" || target.starts_with("superfluid_daemon::")
176}
177
178impl Visit for Fields<'_> {
179    fn record_i64(&mut self, f: &Field, v: i64) {
180        self.0.push((f.name(), Val::I(v)));
181    }
182    fn record_u64(&mut self, f: &Field, v: u64) {
183        self.0.push((f.name(), Val::U(v)));
184    }
185    fn record_f64(&mut self, f: &Field, v: f64) {
186        self.0.push((f.name(), Val::F(v)));
187    }
188    fn record_bool(&mut self, f: &Field, v: bool) {
189        self.0.push((f.name(), Val::B(v)));
190    }
191    fn record_i128(&mut self, f: &Field, v: i128) {
192        if let Ok(v) = i64::try_from(v) {
193            self.record_i64(f, v);
194        }
195    }
196    fn record_u128(&mut self, f: &Field, v: u128) {
197        if let Ok(v) = u64::try_from(v) {
198            self.record_u64(f, v);
199        }
200    }
201    fn record_str(&mut self, f: &Field, v: &str) {
202        if self.1 && STRING_FIELDS.contains(&f.name()) {
203            self.0.push((f.name(), Val::S(v.to_string())));
204        }
205    }
206    fn record_debug(&mut self, _: &Field, _: &dyn std::fmt::Debug) {}
207}
208
209fn now_ns() -> u64 {
210    SystemTime::now()
211        .duration_since(UNIX_EPOCH)
212        .map(|d| d.as_nanos() as u64)
213        .unwrap_or(0)
214}
215
216fn fresh_u64() -> u64 {
217    static SEED: OnceLock<u64> = OnceLock::new();
218    static N: AtomicU64 = AtomicU64::new(0);
219    let seed = *SEED.get_or_init(|| now_ns() ^ ((std::process::id() as u64) << 32));
220    let n = N.fetch_add(1, Ordering::Relaxed);
221    xxhash_rust::xxh3::xxh3_64_with_seed(&n.to_le_bytes(), seed).max(1)
222}
223
224fn fresh_trace() -> [u8; 16] {
225    let mut t = [0u8; 16];
226    t[..8].copy_from_slice(&fresh_u64().to_be_bytes());
227    t[8..].copy_from_slice(&fresh_u64().to_be_bytes());
228    t
229}
230
231pub struct OtlpLayer {
232    tx: SyncSender<Finished>,
233}
234
235impl<S> Layer<S> for OtlpLayer
236where
237    S: Subscriber + for<'a> LookupSpan<'a>,
238{
239    fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
240        let Some(span) = ctx.span(id) else {
241            return;
242        };
243        let mut fields = Vec::new();
244        let owned = daemon_owned(span.metadata().target());
245        attrs.record(&mut Fields(&mut fields, owned));
246        let parent = span
247            .parent()
248            .and_then(|p| p.extensions().get::<SpanData>().map(|d| (d.trace, d.span)));
249        let num = |name: &str| {
250            fields.iter().find_map(|(k, v)| match (*k == name, v) {
251                (true, Val::U(x)) => Some(*x),
252                (true, Val::I(x)) if *x >= 0 => Some(*x as u64),
253                _ => None,
254            })
255        };
256        let session = num("session")
257            .filter(|s| *s != 0)
258            .zip(num("store_ns").filter(|ns| *ns != 0 && owned));
259        let (trace, parent, links) = match (parent, session) {
260            (Some((t, p)), _) => (t, Some(p), Vec::new()),
261            (None, Some((s, ns))) => (
262                fresh_trace(),
263                None,
264                vec![(
265                    export::trace_id_bytes_ns(ns, s),
266                    export::span_id_bytes_ns(ns, s, export::ROOT),
267                )],
268            ),
269            (None, None) => (fresh_trace(), None, Vec::new()),
270        };
271        span.extensions_mut().insert(SpanData {
272            trace,
273            span: fresh_u64().to_be_bytes(),
274            parent,
275            start_ns: now_ns(),
276            attrs: fields,
277            links,
278        });
279    }
280
281    fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) {
282        if let Some(span) = ctx.span(id) {
283            if let Some(d) = span.extensions_mut().get_mut::<SpanData>() {
284                let owned = daemon_owned(span.metadata().target());
285                values.record(&mut Fields(&mut d.attrs, owned));
286            }
287        }
288    }
289
290    fn on_follows_from(&self, id: &Id, follows: &Id, ctx: Context<'_, S>) {
291        if id == follows {
292            return;
293        }
294        let Some(target) = ctx
295            .span(follows)
296            .and_then(|s| s.extensions().get::<SpanData>().map(|d| (d.trace, d.span)))
297        else {
298            return;
299        };
300        if let Some(span) = ctx.span(id) {
301            if let Some(d) = span.extensions_mut().get_mut::<SpanData>() {
302                d.links.push(target);
303            }
304        }
305    }
306
307    fn on_close(&self, id: Id, ctx: Context<'_, S>) {
308        let Some(span) = ctx.span(&id) else {
309            return;
310        };
311        let Some(data) = span.extensions_mut().remove::<SpanData>() else {
312            return;
313        };
314        let meta = span.metadata();
315        let fin = Finished {
316            name: meta.name(),
317            target: meta.target(),
318            end_ns: now_ns(),
319            data,
320        };
321        match self.tx.try_send(fin) {
322            Ok(()) => {}
323            Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => {
324                DROPPED_QUEUE_FULL.fetch_add(1, Ordering::Relaxed);
325                crate::telemetry::DROPPED.fetch_add(1, Ordering::Relaxed);
326            }
327        }
328    }
329}
330
331fn rejected_spans(body: &str) -> u64 {
332    let Ok(v) = serde_json::from_str::<Value>(body) else {
333        return 0;
334    };
335    let r = &v["partialSuccess"]["rejectedSpans"];
336    r.as_u64()
337        .or_else(|| r.as_str().and_then(|s| s.parse().ok()))
338        .unwrap_or(0)
339}
340
341fn attr_key(field: &str) -> String {
342    match field {
343        "session" => "superfluid.session_id".into(),
344        f => format!("superfluid.{f}"),
345    }
346}
347
348fn val_json(v: &Val) -> Value {
349    match *v {
350        Val::I(i) => json!(i),
351        Val::U(u) => json!(u as i64),
352        Val::F(f) => json!(f),
353        Val::B(b) => json!(b),
354        Val::S(ref s) => json!(s),
355    }
356}
357
358fn span_json(s: &Finished) -> Value {
359    let d = &s.data;
360    let mut attrs = Vec::with_capacity(d.attrs.len() + 1);
361    attrs.push(export::attr("superfluid.target", json!(s.target)));
362    for (k, v) in &d.attrs {
363        attrs.push(export::attr(&attr_key(k), val_json(v)));
364    }
365    let mut o = json!({
366        "traceId": export::hex(&d.trace),
367        "spanId": export::hex(&d.span),
368        "name": s.name,
369        "kind": 1,
370        "startTimeUnixNano": d.start_ns.to_string(),
371        "endTimeUnixNano": s.end_ns.max(d.start_ns).to_string(),
372        "attributes": attrs,
373    });
374    if let Some(p) = d.parent {
375        o["parentSpanId"] = json!(export::hex(&p));
376    }
377    if !d.links.is_empty() {
378        o["links"] = Value::Array(
379            d.links
380                .iter()
381                .map(|(t, sp)| json!({"traceId": export::hex(t), "spanId": export::hex(sp)}))
382                .collect(),
383        );
384    }
385    o
386}
387
388fn encode(cfg: &OtlpConfig, host: &str, batch: &[Finished]) -> String {
389    let mut resource = vec![
390        export::attr("service.name", json!(cfg.service_name)),
391        export::attr("service.version", json!(env!("CARGO_PKG_VERSION"))),
392        export::attr("host.name", json!(host)),
393        export::attr("superfluid.signal", json!("operational")),
394        export::attr("superfluid.content_included", json!(false)),
395    ];
396    for (k, v) in &cfg.resource {
397        resource.push(export::attr(k, json!(v)));
398    }
399    let spans = batch.iter().map(span_json).collect();
400    export::trace_request(resource, "superfluid.operational", spans).to_string()
401}
402
403struct Shared {
404    stop: AtomicBool,
405    done: Mutex<bool>,
406    cv: Condvar,
407}
408
409#[derive(Clone)]
410pub struct OtlpHandle {
411    shared: Arc<Shared>,
412}
413
414impl OtlpHandle {
415    pub fn shutdown(&self, bound: Duration) -> bool {
416        self.shared.stop.store(true, Ordering::SeqCst);
417        let done = self.shared.done.lock().unwrap_or_else(|e| e.into_inner());
418        let (done, _) = self
419            .shared
420            .cv
421            .wait_timeout_while(done, bound, |d| !*d)
422            .unwrap_or_else(|e| e.into_inner());
423        *done
424    }
425}
426
427static GLOBAL: OnceLock<OtlpHandle> = OnceLock::new();
428
429pub(crate) fn set_global(h: OtlpHandle) {
430    let _ = GLOBAL.set(h);
431}
432
433pub fn shutdown(bound: Duration) -> bool {
434    GLOBAL.get().map(|h| h.shutdown(bound)).unwrap_or(true)
435}
436
437pub fn start(cfg: OtlpConfig) -> Result<(OtlpLayer, OtlpHandle), String> {
438    cfg.validate()?;
439    let (tx, rx) = sync_channel::<Finished>(cfg.queue);
440    let shared = Arc::new(Shared {
441        stop: AtomicBool::new(false),
442        done: Mutex::new(false),
443        cv: Condvar::new(),
444    });
445    let sh = Arc::clone(&shared);
446    std::thread::Builder::new()
447        .name("superfluid-otlp".into())
448        .spawn(move || {
449            run_exporter(&cfg, &rx, &sh);
450            *sh.done.lock().unwrap_or_else(|e| e.into_inner()) = true;
451            sh.cv.notify_all();
452        })
453        .map_err(|e| e.to_string())?;
454    Ok((OtlpLayer { tx }, OtlpHandle { shared }))
455}
456
457const STOP_POLL: Duration = Duration::from_millis(50);
458
459fn run_exporter(cfg: &OtlpConfig, rx: &Receiver<Finished>, sh: &Shared) {
460    let url = cfg.traces_url();
461    let host = crate::metrics::hostname();
462    let agent = ureq::Agent::config_builder()
463        .timeout_global(Some(cfg.timeout))
464        .build()
465        .new_agent();
466    let mut failing = false;
467    let mut flush = |batch: &mut Vec<Finished>| {
468        if batch.is_empty() {
469            return;
470        }
471        let n = batch.len() as u64;
472        let body = encode(cfg, &host, batch);
473        batch.clear();
474        let mut req = agent.post(&url).header("content-type", "application/json");
475        for (k, v) in &cfg.headers {
476            req = req.header(k.as_str(), v.as_str());
477        }
478        match req.send(body.as_bytes()) {
479            Ok(mut resp) => {
480                let text = resp
481                    .body_mut()
482                    .with_config()
483                    .limit(64 * 1024)
484                    .read_to_string()
485                    .unwrap_or_default();
486                let rejected = rejected_spans(&text).min(n);
487                EXPORTED.fetch_add(n - rejected, Ordering::Relaxed);
488                if rejected > 0 {
489                    DROPPED_REJECTED.fetch_add(rejected, Ordering::Relaxed);
490                    crate::telemetry::DROPPED.fetch_add(rejected, Ordering::Relaxed);
491                }
492                if failing {
493                    failing = false;
494                    tracing::info!(target: "superfluid_daemon::otlp", "otlp trace export recovered");
495                }
496            }
497            Err(e) => {
498                DROPPED_EXPORT_FAILED.fetch_add(n, Ordering::Relaxed);
499                crate::telemetry::DROPPED.fetch_add(n, Ordering::Relaxed);
500                if !failing {
501                    failing = true;
502                    tracing::warn!(target: "superfluid_daemon::otlp", error = %e, "otlp trace export failing; spans are dropped until it recovers");
503                }
504            }
505        }
506    };
507    let mut batch: Vec<Finished> = Vec::with_capacity(cfg.batch_max);
508    let mut deadline = Instant::now() + cfg.interval;
509    loop {
510        if sh.stop.load(Ordering::SeqCst) {
511            while let Ok(s) = rx.try_recv() {
512                batch.push(s);
513                if batch.len() >= cfg.batch_max {
514                    flush(&mut batch);
515                }
516            }
517            flush(&mut batch);
518            return;
519        }
520        let now = Instant::now();
521        let wait = deadline.saturating_duration_since(now).min(STOP_POLL);
522        match rx.recv_timeout(wait) {
523            Ok(s) => {
524                batch.push(s);
525                if batch.len() >= cfg.batch_max || Instant::now() >= deadline {
526                    flush(&mut batch);
527                    deadline = Instant::now() + cfg.interval;
528                }
529            }
530            Err(RecvTimeoutError::Timeout) => {
531                if Instant::now() >= deadline {
532                    flush(&mut batch);
533                    deadline = Instant::now() + cfg.interval;
534                }
535            }
536            Err(RecvTimeoutError::Disconnected) => {
537                flush(&mut batch);
538                return;
539            }
540        }
541    }
542}
543
544#[cfg(test)]
545mod tests {
546    use super::*;
547
548    #[test]
549    fn traces_url_and_validation() {
550        let c = |e: &str| OtlpConfig {
551            endpoint: e.into(),
552            ..Default::default()
553        };
554        assert_eq!(c("http://h:4318").traces_url(), "http://h:4318/v1/traces");
555        assert_eq!(c("http://h:4318/").traces_url(), "http://h:4318/v1/traces");
556        assert_eq!(
557            c("http://h:4318/v1/traces").traces_url(),
558            "http://h:4318/v1/traces"
559        );
560        assert!(c("http://h:4318").validate().is_ok());
561        assert!(c("https://h:4318").validate().unwrap_err().contains("TLS"));
562        assert!(c("h:4318").validate().is_err());
563        assert!(c("http://").validate().is_err());
564        assert!(c("http://localhost:notaport")
565            .validate()
566            .unwrap_err()
567            .contains("port"));
568        assert!(c("http://localhost:99999").validate().is_err());
569        assert!(c("http://h:4318/v1/traces?x=1").validate().is_err());
570        assert!(c("http://[::1]:4318").validate().is_ok());
571        assert!(c("http://collector/otlp").validate().is_ok());
572        let mut bad = c("http://h:1");
573        bad.headers.push(("x-auth".into(), "a\nb".into()));
574        assert!(bad.validate().unwrap_err().contains("bad value"));
575        assert!(c("http://127.0.0.1:4318").is_loopback());
576        assert!(c("http://localhost:4318").is_loopback());
577        assert!(c("http://[::1]:4318/x").is_loopback());
578        assert!(!c("http://10.0.0.2:4318").is_loopback());
579        assert!(!c("http://collector:4318").is_loopback());
580        let mut bad = c("http://h:1");
581        bad.headers.push(("x auth".into(), "v".into()));
582        assert!(bad.validate().is_err());
583        let mut bad = c("http://h:1");
584        bad.filter = "not a directive[".into();
585        assert!(bad.validate().unwrap_err().contains("--otlp-filter"));
586        let mut bad = c("http://h:1");
587        bad.interval = Duration::ZERO;
588        assert!(bad.validate().unwrap_err().contains("--otlp-batch-ms"));
589        let mut bad = c("http://h:1");
590        bad.timeout = Duration::ZERO;
591        assert!(bad.validate().unwrap_err().contains("--otlp-timeout-ms"));
592    }
593
594    #[test]
595    fn partial_success_counts_rejected_spans() {
596        assert_eq!(rejected_spans("{}"), 0);
597        assert_eq!(rejected_spans(""), 0);
598        assert_eq!(rejected_spans(r#"{"partialSuccess":{}}"#), 0);
599        assert_eq!(
600            rejected_spans(r#"{"partialSuccess":{"rejectedSpans":"3"}}"#),
601            3
602        );
603        assert_eq!(
604            rejected_spans(r#"{"partialSuccess":{"rejectedSpans":4}}"#),
605            4
606        );
607    }
608
609    #[test]
610    fn store_namespaces_session_ids() {
611        use crate::store::StoreId;
612        assert_eq!(export::trace_id_bytes_ns(0, 7), export::trace_id_bytes(7));
613        assert_eq!(export::span_id_bytes_ns(0, 7, 3), export::span_id_bytes(7, 3));
614        let a = export::store_trace_ns(&StoreId([1; 16]));
615        let b = export::store_trace_ns(&StoreId([2; 16]));
616        assert!(a != 0 && b != 0 && a != b);
617        assert!(a <= i64::MAX as u64 && b <= i64::MAX as u64, "int64 domain");
618        assert_eq!(a, export::store_trace_ns(&StoreId([1; 16])), "stable");
619        assert_ne!(export::trace_id_bytes_ns(a, 1), export::trace_id_bytes_ns(b, 1));
620        assert_ne!(export::trace_id_bytes_ns(a, 1), export::trace_id_bytes(1));
621    }
622}
superfluid_daemon/otlp.rs

superfluid_daemon/
otlp.rs
