Tests for the log's fields and the metrics Grafana reads (#403)

Three tests for what the dashboard and the alert rules read and nothing
checked: every ipx_ metric named in grafana/dashboard.py and alerts.py
(read out of them at compile time) is served by render, with the ones
written only once known (counts, the last scan, a job under way) set
first; ipx's JSON log lines for a scan's events carry ev, feed, new,
bytes, msg and error.type, and inside a span trace_id and span_id; the
directory's snapshot line carries instance, accepted and the refused_
counts.

The JSON layer was built the same way in both services' setup; it is
otel::json_layer now, so the tests run the layer production does.

The log tests set the global subscriber, as production does: under a
scoped one WithTrace finds no current span (#408).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/directory/src/api.rs

@@ -393,6 +393,46 @@ mod tests {
         assert_eq!(call(&app, "DELETE", &format!("/feed/{id}"), None, None).await.0, 405);
     }
 
+    /// What the log layer wrote, kept for the test to read.
+    struct Lines(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+
+    impl std::io::Write for Lines {
+        fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
+            self.0.lock().unwrap().extend_from_slice(b);
+            Ok(b.len())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn a_snapshots_log_line_has_the_fields_grafana_reads() {
+        use tracing_subscriber::layer::SubscriberExt;
+        let buf = std::sync::Arc::new(std::sync::Mutex::new(vec![]));
+        let b = buf.clone();
+        // Global, as the directory's own is; other tests' snapshots land here too, and each of
+        // their lines must carry the same fields.
+        tracing::subscriber::set_global_default(
+            tracing_subscriber::registry().with(crate::otel::json_layer(move || Lines(b.clone()), false)),
+        )
+        .unwrap();
+        let (app, _dir) = app().await;
+        let snap = serde_json::json!([{ "url": "https://x.test/log.xml", "subscribers": 1 }]);
+        assert_eq!(call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap)).await.0, 200);
+
+        let text = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
+        let lines: Vec<serde_json::Value> = text.lines().map(|l| serde_json::from_str(l).expect(l)).collect();
+        let snaps: Vec<&serde_json::Value> = lines.iter().filter(|l| l["message"] == "snapshot").collect();
+        assert!(!snaps.is_empty(), "no snapshot line in {text}");
+        for l in snaps {
+            assert!(l["instance"].is_string(), "{l}");
+            for n in ["accepted", "refused_private", "refused_blocked", "refused_invalid"] {
+                assert!(l[n].is_u64(), "{n} in {l}");
+            }
+        }
+    }
+
     #[tokio::test]
     async fn each_access_level_does_what_it_may_and_no_more() {
         let (app, dir) = app().await;

crates/directory/src/main.rs

@@ -85,21 +85,9 @@ async fn serve(path: &Path) -> Result<()> {
     let otel = ipx_directory::otel::provider("ipx-directory")?;
     tracing_subscriber::registry()
         .with((!json).then(|| tracing_subscriber::fmt::layer().with_writer(std::io::stderr).with_filter(filter())))
-        .with(json.then(|| {
-            tracing_subscriber::fmt::layer()
-                .fmt_fields(tracing_subscriber::fmt::format::JsonFields::new())
-                // Each line's own fields say all its span did, and the list said it again (#363);
-                // the trace and span ids lead to the trace instead, as ipx's do (#375).
-                .event_format(ipx_directory::otel::WithTrace(
-                    tracing_subscriber::fmt::format()
-                        .json()
-                        .flatten_event(true)
-                        .with_current_span(false)
-                        .with_span_list(false),
-                ))
-                .with_writer(std::io::stderr)
-                .with_filter(filter())
-        }))
+        // Each line's own fields say all its span did, and the span said it again (#363); the trace
+        // and span ids lead to the trace instead, as ipx's do (#375).
+        .with(json.then(|| ipx_directory::otel::json_layer(std::io::stderr, false).with_filter(filter())))
         .with(otel.as_ref().map(|p| {
             use opentelemetry::trace::TracerProvider;
             tracing_opentelemetry::layer()

crates/directory/src/otel.rs

@@ -95,6 +95,27 @@ pub fn with_ids<R>(
 /// The JSON log line with the trace and span it belongs to (#91), so a line in Loki leads to its
 /// trace in Tempo. The JSON formatter cannot take a field of its own, so the ids go on the end of
 /// the object it writes; the directory's lines carry them too (#375). A line outside any traced span, or with no trace exporter, is unchanged.
+/// One JSON object a line for Loki, each event's fields its own and the trace and span ids beside
+/// them, for ipx and the directory both: the dashboard and the alert rules read these fields.
+/// `current_span` adds the span's own fields under `span`, which ipx keeps and the directory
+/// dropped (#363).
+pub fn json_layer<S, W>(writer: W, current_span: bool) -> impl tracing_subscriber::Layer<S>
+where
+    S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
+    W: for<'w> tracing_subscriber::fmt::MakeWriter<'w> + Send + Sync + 'static,
+{
+    tracing_subscriber::fmt::layer()
+        .fmt_fields(tracing_subscriber::fmt::format::JsonFields::new())
+        .event_format(WithTrace(
+            tracing_subscriber::fmt::format()
+                .json()
+                .flatten_event(true)
+                .with_current_span(current_span)
+                .with_span_list(false),
+        ))
+        .with_writer(writer)
+}
+
 pub struct WithTrace<F>(pub F);
 
 impl<S, N, F> tracing_subscriber::fmt::FormatEvent<S, N> for WithTrace<F>

crates/ipx/src/logging.rs

@@ -1,10 +1,9 @@
 //! Log lines as JSON for Loki, carrying the trace and span they belong to.
 
 use crate::logbuf;
-use ipx_directory::otel::WithTrace;
+use ipx_directory::otel::json_layer;
 use std::io::IsTerminal;
 use tracing_subscriber::EnvFilter;
-use tracing_subscriber::fmt::format::JsonFields;
 
 /// Sets up logging: to stderr, as text or as JSON (`IPX_LOG_FORMAT`), into the ring the page's log
 /// reads, and to the trace exporter when there is one.
@@ -30,19 +29,7 @@ pub(crate) fn init(otel: Option<&opentelemetry_sdk::trace::SdkTracerProvider>) {
                 .with_ansi(IsTerminal::is_terminal(&std::io::stderr()))
                 .with_filter(stderr_filter())
         }))
-        .with(json.then(|| {
-            tracing_subscriber::fmt::layer()
-                .fmt_fields(JsonFields::new())
-                .event_format(WithTrace(
-                    tracing_subscriber::fmt::format()
-                        .json()
-                        .flatten_event(true)
-                        .with_current_span(true)
-                        .with_span_list(false),
-                ))
-                .with_writer(std::io::stderr)
-                .with_filter(stderr_filter())
-        }))
+        .with(json.then(|| json_layer(std::io::stderr, true).with_filter(stderr_filter())))
         .with(logbuf::RingLayer.with_filter(ui_filter))
         .with(otel.as_ref().map(|p| {
             use opentelemetry::trace::TracerProvider;
@@ -52,3 +39,74 @@ pub(crate) fn init(otel: Option<&opentelemetry_sdk::trace::SdkTracerProvider>) {
         }))
         .init();
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ipc::{Emitter, Event};
+    use std::sync::{Arc, Mutex};
+
+    /// What the layer wrote, kept for the test to read.
+    struct Lines(Arc<Mutex<Vec<u8>>>);
+
+    impl std::io::Write for Lines {
+        fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
+            self.0.lock().unwrap().extend_from_slice(b);
+            Ok(b.len())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn a_scans_events_are_json_lines_with_the_fields_grafana_reads() {
+        use opentelemetry::trace::TracerProvider;
+        use tracing_subscriber::layer::SubscriberExt;
+        let buf = Arc::new(Mutex::new(vec![]));
+        let b = buf.clone();
+        let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder().build();
+        let sub = tracing_subscriber::registry()
+            .with(json_layer(move || Lines(b.clone()), true))
+            .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("test")));
+        // Global, as production's is: under a scoped one (with_default), Span::current() inside the
+        // formatter finds no dispatcher, so no line would name its trace. Other tests' events
+        // land here too, hence feed names only this test uses.
+        tracing::subscriber::set_global_default(sub).unwrap();
+        {
+            let out = Emitter::terminal();
+            out.emit(Event::FeedDone { feed: "log-test-f".into(), new: 2, downloaded: 1, failed: 0, torrents: 0 });
+            let scan = tracing::info_span!("scan");
+            let _in = scan.enter();
+            out.emit(Event::DownloadDone {
+                feed: "log-test-f".into(),
+                enclosure: 7,
+                url: "https://x.test/a.mp3".into(),
+                path: "/downloads/a.mp3".into(),
+                bytes: 1234,
+            });
+            out.emit(Event::FeedError { feed: "log-test-g".into(), msg: "HTTP 404 Not Found".into() });
+        }
+        let text = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
+        let lines: Vec<serde_json::Value> = text.lines().map(|l| serde_json::from_str(l).expect(l)).collect();
+        let find = |ev: &str| {
+            lines
+                .iter()
+                .find(|l| l["ev"] == ev && l["feed"].as_str().is_some_and(|f| f.starts_with("log-test-")))
+                .unwrap_or_else(|| panic!("no {ev} in {text}"))
+        };
+
+        let done = find("feed_done");
+        assert_eq!((done["feed"].as_str(), done["new"].as_u64()), (Some("log-test-f"), Some(2)), "{done}");
+        assert!(done.get("trace_id").is_none(), "outside a span there is no trace to name: {done}");
+        let saved = find("download_done");
+        assert_eq!(saved["bytes"].as_u64(), Some(1234), "{saved}");
+        let failed = find("feed_error");
+        assert_eq!((failed["feed"].as_str(), failed["msg"].as_str()), (Some("log-test-g"), Some("HTTP 404 Not Found")));
+        assert!(failed["error.type"].is_string(), "the failure's kind: {failed}");
+        for l in [saved, failed] {
+            let (t, s) = (l["trace_id"].as_str().unwrap_or(""), l["span_id"].as_str().unwrap_or(""));
+            assert!(t.len() == 32 && s.len() == 16, "inside a span a line names its trace: {l}");
+        }
+    }
+}

crates/ipx/src/metrics.rs

@@ -544,6 +544,54 @@ fn write(out: &mut String, kind: &str, name: &str, help: &str, samples: &[(&str,
 mod tests {
     use super::*;
 
+    /// The names the dashboard and the alert rules query, read out of what generates them.
+    fn queried() -> std::collections::BTreeSet<String> {
+        let src = [include_str!("../../../grafana/dashboard.py"), include_str!("../../../grafana/alerts.py")].concat();
+        src.match_indices("ipx_")
+            .filter(|(i, _)| !src[..*i].ends_with(|c: char| c.is_ascii_alphanumeric() || c == '_'))
+            .map(|(i, _)| {
+                let rest = &src[i..];
+                let end = rest.find(|c: char| !(c.is_ascii_lowercase() || c == '_')).unwrap_or(rest.len());
+                (&rest[..end], rest[end..].starts_with("::"))
+            })
+            // A log target (ipx_directory::http) is not a metric; the coverage numbers are
+            // node-exporter's, written by scripts/coverage-metrics.ts.
+            .filter(|(n, target)| !target && !n.starts_with("ipx_coverage_"))
+            .map(|(n, _)| n.to_string())
+            .collect()
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn every_metric_grafana_queries_is_served() {
+        let ctx = Arc::new(crate::test_util::test_ctx(crate::config::Config::default()).await);
+        let s = Sources {
+            ctx,
+            cmds: tokio::sync::mpsc::channel(1).0,
+            events: tokio::sync::broadcast::channel(1).0,
+            playing: Default::default(),
+            runtime: Handle::current(),
+            started: Instant::now(),
+        };
+        // Written only once known (a count taken, a scan finished, a job under way): know them.
+        for v in [&STATS.feeds, &STATS.pending, &STATS.downloaded, &STATS.last_scan_done, &STATS.signals_last_pass] {
+            v.fetch_max(1, Relaxed);
+        }
+        let _job = STATS.job("scan");
+        let out = render(&s);
+        let served: std::collections::BTreeSet<&str> =
+            out.lines().filter_map(|l| l.strip_prefix("# TYPE ")).filter_map(|l| l.split(' ').next()).collect();
+        let names = queried();
+        assert!(names.len() > 30, "the dashboard's names were not found: {names:?}");
+        let missing: Vec<&String> = names
+            .iter()
+            .filter(|n| {
+                let base = ["_bucket", "_count", "_sum"].iter().find_map(|x| n.strip_suffix(x)).unwrap_or(n);
+                !served.contains(n.as_str()) && !served.contains(base)
+            })
+            .collect();
+        assert!(missing.is_empty(), "queried by grafana/ but not served by /metrics: {missing:?}");
+    }
+
     #[test]
     fn gauges_are_written_as_prometheus_reads_them() {
         let mut out = String::new();