diff --git a/crates/directory/src/api.rs b/crates/directory/src/api.rs index 118a102..0036f59 100644 --- a/crates/directory/src/api.rs +++ b/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>>); + + impl std::io::Write for Lines { + fn write(&mut self, b: &[u8]) -> std::io::Result { + 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 = 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; diff --git a/crates/directory/src/main.rs b/crates/directory/src/main.rs index 8229853..3f2a545 100644 --- a/crates/directory/src/main.rs +++ b/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() diff --git a/crates/directory/src/otel.rs b/crates/directory/src/otel.rs index fb47679..3c36d31 100644 --- a/crates/directory/src/otel.rs +++ b/crates/directory/src/otel.rs @@ -95,6 +95,27 @@ pub fn with_ids( /// 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(writer: W, current_span: bool) -> impl tracing_subscriber::Layer +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(pub F); impl tracing_subscriber::fmt::FormatEvent for WithTrace diff --git a/crates/ipx/src/logging.rs b/crates/ipx/src/logging.rs index 190c209..7dc57f0 100644 --- a/crates/ipx/src/logging.rs +++ b/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>>); + + impl std::io::Write for Lines { + fn write(&mut self, b: &[u8]) -> std::io::Result { + 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 = 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}"); + } + } +} diff --git a/crates/ipx/src/metrics.rs b/crates/ipx/src/metrics.rs index f0e7951..0f2910d 100644 --- a/crates/ipx/src/metrics.rs +++ b/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 { + 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();