comindly_engine/
telemetry.rs

1//! The process's logs and traces: `tracing` events rendered as readable lines or, with
2//! `COMINDLY_LOG_FORMAT=json`, one JSON object per line; spans sent over OTLP when configured.
3
4use std::fmt::Debug;
5use std::sync::{Arc, OnceLock};
6
7use console::style;
8use serde_json::{Map, Value};
9use tracing::field::{Field, Visit};
10use tracing::{span, Level, Subscriber};
11use tracing_subscriber::layer::{Context, Layer, SubscriberExt};
12use tracing_subscriber::registry::LookupSpan;
13
14/// Kinds of lines not logged; starts from `COMINDLY_LOG_SKIP` (e.g. `access,tool`), changed live.
15fn skip() -> &'static std::sync::RwLock<Vec<String>> {
16    static SKIP: std::sync::OnceLock<std::sync::RwLock<Vec<String>>> = std::sync::OnceLock::new();
17    SKIP.get_or_init(|| {
18        std::sync::RwLock::new(parse_skip(
19            &std::env::var("COMINDLY_LOG_SKIP").unwrap_or_default(),
20        ))
21    })
22}
23
24/// A comma-separated list of kinds.
25pub fn parse_skip(text: &str) -> Vec<String> {
26    text.split(',')
27        .map(|k| k.trim().to_lowercase())
28        .filter(|k| !k.is_empty())
29        .collect()
30}
31
32/// Whether lines of `kind` are logged.
33pub fn logs(kind: &str) -> bool {
34    !skip().read().is_ok_and(|s| s.iter().any(|k| k == kind))
35}
36
37/// Log or drop `kind` from now on.
38pub fn set_logs(kind: &str, on: bool) {
39    if let Ok(mut s) = skip().write() {
40        s.retain(|k| k != kind);
41        if !on {
42            s.push(kind.to_string());
43        }
44    }
45}
46
47/// The kinds not logged, comma-separated, as `COMINDLY_LOG_SKIP` takes them.
48pub fn skipped() -> String {
49    skip().read().map(|s| s.join(",")).unwrap_or_default()
50}
51
52/// Set up logging (and OTLP export when `OTEL_EXPORTER_OTLP_ENDPOINT` is set) for this process,
53/// to stdout when `stdout`, else stderr. Keep the guard until the process ends.
54pub fn init(stdout: bool) -> Guard {
55    let _ = STDOUT.set(stdout);
56    let level = match std::env::var("COMINDLY_LOG_LEVEL").as_deref() {
57        Ok("error") => Level::ERROR,
58        Ok("warn") => Level::WARN,
59        Ok("debug") => Level::DEBUG,
60        Ok("trace") => Level::TRACE,
61        _ => Level::INFO,
62    };
63    let lines = Lines::to(crate::ui::json_logs(), write).with_filter(
64        tracing_subscriber::filter::filter_fn(move |m| {
65            ours(m.target()) && (m.is_span() || *m.level() <= level)
66        }),
67    );
68    let registry = tracing_subscriber::registry().with(lines);
69    #[cfg(feature = "otel")]
70    let registry = registry.with(otel::layer());
71    #[cfg(not(feature = "otel"))]
72    if [
73        "OTEL_EXPORTER_OTLP_ENDPOINT",
74        "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
75    ]
76    .iter()
77    .any(|k| std::env::var(k).is_ok_and(|v| !v.trim().is_empty()))
78    {
79        crate::ui::warn("traces will not be exported: this build has no OTLP support; use the container image or build with `--features otel`");
80    }
81    let _ = tracing::subscriber::set_global_default(registry);
82    Guard(())
83}
84
85static STDOUT: OnceLock<bool> = OnceLock::new();
86
87/// Write one log line to the process's log stream: stdout on a server, stderr elsewhere.
88pub fn write(line: &str) {
89    if STDOUT.get() == Some(&true) {
90        println!("{line}");
91    } else {
92        eprintln!("{line}");
93    }
94}
95
96/// Sends what is queued for export; for when the process is about to exit.
97pub fn flush() {
98    #[cfg(feature = "otel")]
99    otel::flush();
100}
101
102pub struct Guard(());
103
104impl Drop for Guard {
105    fn drop(&mut self) {
106        #[cfg(feature = "otel")]
107        otel::shutdown();
108    }
109}
110
111fn ours(target: &str) -> bool {
112    target.starts_with("comindly")
113}
114
115/// What a span tells the lines logged inside it.
116#[derive(Default)]
117struct Scope {
118    request_id: Option<String>,
119    job_id: Option<String>,
120}
121
122impl Visit for Scope {
123    fn record_str(&mut self, f: &Field, v: &str) {
124        match f.name() {
125            "request_id" => self.request_id = Some(v.to_string()),
126            "job_id" => self.job_id = Some(v.to_string()),
127            _ => {}
128        }
129    }
130
131    fn record_debug(&mut self, f: &Field, v: &dyn Debug) {
132        self.record_str(f, &format!("{v:?}"))
133    }
134}
135
136struct Fields(Map<String, Value>);
137
138impl Visit for Fields {
139    fn record_str(&mut self, f: &Field, v: &str) {
140        self.0.insert(f.name().into(), v.into());
141    }
142    fn record_u64(&mut self, f: &Field, v: u64) {
143        self.0.insert(f.name().into(), v.into());
144    }
145    fn record_i64(&mut self, f: &Field, v: i64) {
146        self.0.insert(f.name().into(), v.into());
147    }
148    fn record_f64(&mut self, f: &Field, v: f64) {
149        self.0.insert(f.name().into(), v.into());
150    }
151    fn record_bool(&mut self, f: &Field, v: bool) {
152        self.0.insert(f.name().into(), v.into());
153    }
154    fn record_debug(&mut self, f: &Field, v: &dyn Debug) {
155        self.0.insert(f.name().into(), format!("{v:?}").into());
156    }
157}
158
159/// Renders events as comindly's log lines, with the request and job they happened in.
160pub struct Lines {
161    json: bool,
162    out: Arc<dyn Fn(&str) + Send + Sync>,
163}
164
165impl Lines {
166    /// Lines handed to `out`.
167    pub fn to(json: bool, out: impl Fn(&str) + Send + Sync + 'static) -> Self {
168        Lines {
169            json,
170            out: Arc::new(out),
171        }
172    }
173}
174
175impl<S: Subscriber + for<'a> LookupSpan<'a>> Layer<S> for Lines {
176    fn on_new_span(&self, attrs: &span::Attributes<'_>, id: &span::Id, ctx: Context<'_, S>) {
177        let mut scope = Scope::default();
178        attrs.record(&mut scope);
179        if let Some(span) = ctx.span(id) {
180            span.extensions_mut().insert(scope);
181        }
182    }
183
184    fn on_record(&self, id: &span::Id, values: &span::Record<'_>, ctx: Context<'_, S>) {
185        if let Some(span) = ctx.span(id) {
186            if let Some(scope) = span.extensions_mut().get_mut::<Scope>() {
187                values.record(scope);
188            }
189        }
190    }
191
192    fn on_event(&self, event: &tracing::Event<'_>, ctx: Context<'_, S>) {
193        let mut fields = Fields(Map::new());
194        event.record(&mut fields);
195        let mut fields = fields.0;
196        for span in ctx.event_scope(event).into_iter().flatten() {
197            if let Some(s) = span.extensions().get::<Scope>() {
198                for (k, v) in [("request_id", &s.request_id), ("job_id", &s.job_id)] {
199                    if let Some(v) = v {
200                        fields.entry(k).or_insert_with(|| v.clone().into());
201                    }
202                }
203            }
204        }
205        (self.out)(&render(*event.metadata().level(), fields, self.json));
206    }
207}
208
209/// One log line: a JSON object (`timestamp`, `level`, `kind`, `message`, then the other fields),
210/// or the readable `line` when given, else one styled by `kind`.
211pub(crate) fn render(level: Level, mut f: Map<String, Value>, json: bool) -> String {
212    let readable = f.shift_remove("line");
213    let text = |v: Option<Value>| match v {
214        Some(Value::String(s)) => s,
215        Some(v) => v.to_string(),
216        None => String::new(),
217    };
218    let message = text(f.shift_remove("message"));
219    let kind = match f.shift_remove("kind") {
220        Some(k) => text(Some(k)),
221        None if level == Level::ERROR => "error".into(),
222        None if level == Level::WARN => "warning".into(),
223        None => "note".into(),
224    };
225    if json {
226        let mut out = Map::new();
227        out.insert(
228            "timestamp".into(),
229            chrono::Utc::now()
230                .to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
231                .into(),
232        );
233        out.insert("level".into(), level_name(level).into());
234        out.insert("kind".into(), kind.into());
235        out.insert("message".into(), message.into());
236        for k in ["step", "job_id", "request_id"] {
237            if let Some(v) = f.shift_remove(k) {
238                out.insert(k.into(), v);
239            }
240        }
241        out.extend(f);
242        return Value::Object(out).to_string();
243    }
244    let mut line = match readable {
245        Some(r) => text(Some(r)),
246        None => match kind.as_str() {
247            "step" => format!("\n{}", style(&message).bold().cyan()),
248            "done" => format!("{} {message}", style("✓").green().bold()),
249            "warning" => format!("{} {message}", style("!").yellow().bold()),
250            "error" => format!("{} {message}", style("✗").red().bold()),
251            "access" => format!("{} {message}", style("·").dim()),
252            _ => format!("  {}", style(&message).dim()),
253        },
254    };
255    let tagged = level <= Level::WARN || kind == "access";
256    if let Some(id) = f
257        .get("request_id")
258        .and_then(Value::as_str)
259        .filter(|_| tagged)
260    {
261        let who = f.get("email").and_then(Value::as_str);
262        let suffix = who.map_or(format!("request {id}"), |w| format!("{w} · request {id}"));
263        line.push_str(&format!(" {}", style(format!("({suffix})")).dim()));
264    }
265    line
266}
267
268fn level_name(l: Level) -> &'static str {
269    match l {
270        Level::ERROR => "error",
271        Level::WARN => "warn",
272        Level::INFO => "info",
273        Level::DEBUG => "debug",
274        Level::TRACE => "trace",
275    }
276}
277
278#[cfg(feature = "otel")]
279mod otel {
280    use std::sync::OnceLock;
281
282    use opentelemetry::trace::TracerProvider as _;
283    use opentelemetry_sdk::trace::SdkTracerProvider;
284    use tracing::Subscriber;
285    use tracing_subscriber::registry::LookupSpan;
286    use tracing_subscriber::Layer;
287
288    static PROVIDER: OnceLock<SdkTracerProvider> = OnceLock::new();
289
290    /// Spans of comindly's own requests and jobs, to the OTLP/HTTP endpoint in the environment.
291    pub(super) fn layer<S>() -> Option<impl Layer<S>>
292    where
293        S: Subscriber + for<'a> LookupSpan<'a>,
294    {
295        let set = |k: &str| std::env::var(k).is_ok_and(|v| !v.trim().is_empty());
296        if !set("OTEL_EXPORTER_OTLP_ENDPOINT") && !set("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") {
297            return None;
298        }
299        let exporter = match opentelemetry_otlp::SpanExporter::builder()
300            .with_http()
301            .build()
302        {
303            Ok(e) => e,
304            Err(e) => {
305                crate::ui::warn(&format!("traces will not be exported: {e}"));
306                return None;
307            }
308        };
309        let mut resource = opentelemetry_sdk::Resource::builder().with_attribute(
310            opentelemetry::KeyValue::new("service.version", env!("CARGO_PKG_VERSION")),
311        );
312        if !set("OTEL_SERVICE_NAME") {
313            resource = resource.with_service_name("comindly");
314        }
315        let provider = SdkTracerProvider::builder()
316            .with_batch_exporter(exporter)
317            .with_resource(resource.build())
318            .build();
319        let tracer = provider.tracer("comindly");
320        let _ = PROVIDER.set(provider);
321        let only_ours = tracing_subscriber::filter::filter_fn(|m| {
322            super::ours(m.target()) && *m.level() <= tracing::Level::INFO
323        });
324        Some(
325            tracing_opentelemetry::layer()
326                .with_tracer(tracer)
327                .with_filter(only_ours),
328        )
329    }
330
331    /// How long exiting waits for a collector; an unreachable one would otherwise hold it for the export timeout.
332    const AT_EXIT: std::time::Duration = std::time::Duration::from_secs(2);
333
334    pub(super) fn flush() {
335        if let Some(p) = PROVIDER.get() {
336            bounded(move || drop(p.force_flush()));
337        }
338    }
339
340    pub(super) fn shutdown() {
341        if let Some(p) = PROVIDER.get() {
342            bounded(move || drop(p.shutdown()));
343        }
344    }
345
346    /// Runs `f` on its own thread and waits for it at most [`AT_EXIT`].
347    fn bounded(f: impl FnOnce() + Send + 'static) {
348        let (done, wait) = std::sync::mpsc::channel();
349        std::thread::spawn(move || {
350            f();
351            let _ = done.send(());
352        });
353        let _ = wait.recv_timeout(AT_EXIT);
354    }
355}
356
357#[cfg(test)]
358mod tests {
359    use super::*;
360
361    #[test]
362    fn json_lines_keep_their_field_names() {
363        let mut f = Map::new();
364        f.insert("message".into(), "hello".into());
365        f.insert("kind".into(), "done".into());
366        f.insert("job_id".into(), "j1".into());
367        f.insert("request_id".into(), "r1".into());
368        let v: Value = serde_json::from_str(&render(Level::INFO, f, true)).unwrap();
369        assert_eq!(v["level"], "info");
370        assert_eq!(v["kind"], "done");
371        assert_eq!(v["message"], "hello");
372        assert_eq!(v["job_id"], "j1");
373        assert_eq!(v["request_id"], "r1");
374        assert!(v["timestamp"].is_string());
375    }
376}
comindly_engine/telemetry.rs

comindly_engine/
telemetry.rs
