From 7a964c9c5cfbe30e7c49ee100cef323f37afa689 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 25 Oct 2026 21:49:51 +0100 Subject: [PATCH] notify router: log each delivered forge event and each sink outcome --- docs/swarm/notify-router.md | 15 ++ swarm-notify-router/src/log.rs | 37 +++++ swarm-notify-router/src/main.rs | 256 ++++++++++++++++++++++++++++---- 3 files changed, 282 insertions(+), 26 deletions(-) diff --git a/docs/swarm/notify-router.md b/docs/swarm/notify-router.md index 252b5f9d..1721f721 100644 --- a/docs/swarm/notify-router.md +++ b/docs/swarm/notify-router.md @@ -218,6 +218,21 @@ so a retry shows twice, and the page says so under the table. A row that takes longer than 2 seconds to write counts as failed: the router logs a warning and the delivery goes on. +**The journal.** The router also logs each delivery it handles, so +`swarm-logs` answers whether a flow ran for a given PR. The lines carry the +event and sender, never the payload body: + +| Message | Level | Fields | +| -------------------------------------------- | ----- | ----------------------------------------------------------------------------------------- | +| `notify router: forge event routed` | info | `event_type`, `action`, `repo`, `number`, `sender`, `delivery_id`, `flows` (at least one) | +| `notify router: forge event matched no flow` | debug | the same fields, with `flows` of 0 | +| `notify router: sink ran` | info | `flow`, `repo`, `number`, `kind`, `target`, `outcome` | +| `notify router: sink failed` | warn | `flow`, `repo`, `number`, `kind`, `target`, `error` | + +`flows` counts the flows that produced a todo op or a review request. +`delivery_id` is the forge's delivery id, or `none` when the event carried no +id. A review the controller skipped is a `sink ran` line, as in the log rows. + **The dry run.** The router runs a flows document over the newest 50 events of the `SWARM_EVENTS` stream and returns, for each flow the document turns on, how many of those events reached each node. A `match` node counts the events diff --git a/swarm-notify-router/src/log.rs b/swarm-notify-router/src/log.rs index 1675c47c..f7849a28 100644 --- a/swarm-notify-router/src/log.rs +++ b/swarm-notify-router/src/log.rs @@ -80,6 +80,43 @@ impl Cause { } } +/// Log one routed delivery: at info when `flows` flows produced a todo or a +/// review request, at debug when none did, so deliveries nothing wants do not +/// flood the log. It names the sender and never the body. +pub fn delivery( + event_type: &str, + delivery: &crate::event::Delivery, + delivery_id: Option<&str>, + flows: usize, +) { + let (repo, number, sender, action) = ( + &delivery.repo, + delivery.number, + &delivery.sender, + &delivery.action, + ); + let delivery_id = delivery_id.unwrap_or("none"); + if flows == 0 { + tracing::debug!(event_type, %action, %repo, number, %sender, delivery_id, flows, "notify router: forge event matched no flow"); + } else { + tracing::info!(event_type, %action, %repo, number, %sender, delivery_id, flows, "notify router: forge event routed"); + } +} + +/// Log what one sink of `flow` did for `target` on the issue or PR `cause` names: info when it went through, +/// warn with the error when it did not. +pub fn sink(cause: &Cause, flow: &str, kind: Kind, target: &str, result: &Result) { + let (repo, number) = (&cause.repo, cause.number); + match result { + Ok(outcome) => { + tracing::info!(%flow, %repo, number, ?kind, %target, %outcome, "notify router: sink ran"); + } + Err(error) => { + tracing::warn!(%flow, %repo, number, ?kind, %target, %error, "notify router: sink failed"); + } + } +} + /// Append `row`; a failure, or an append that takes longer than /// `APPEND_WITHIN`, is logged and dropped. It is awaited in place, so rows /// keep the order of the deliveries. diff --git a/swarm-notify-router/src/main.rs b/swarm-notify-router/src/main.rs index 9f0bd709..a7cb9784 100644 --- a/swarm-notify-router/src/main.rs +++ b/swarm-notify-router/src/main.rs @@ -214,6 +214,18 @@ impl Router { let routed = route::route(&event_type, &delivery, &snapshot, &self.flows.active, |a| { roster.contains(a) }); + let flows: BTreeSet<&str> = routed + .actions + .iter() + .map(|a| a.flow.as_str()) + .chain(routed.requests.iter().map(|r| r.flow.as_str())) + .collect(); + let delivery_id = message + .headers + .as_ref() + .and_then(|h| h.get(async_nats::header::NATS_MESSAGE_ID)) + .map(async_nats::HeaderValue::as_str); + log::delivery(&event_type, &delivery, delivery_id, flows.len()); let cause = log::Cause { event: event_type.clone(), repo: delivery.repo.clone(), @@ -247,36 +259,20 @@ impl Router { let sent = async_nats::Request::new() .payload(body.into()) .timeout(Some(router_sink::REPLY_DEADLINE)); - let (flow, repo, number, reviewer) = ( - &request.flow, - &request.repo, - request.number, - &request.reviewer, - ); + let (flow, reviewer) = (&request.flow, &request.reviewer); let result = match self .nats .send_request(router_sink::REQUEST_REVIEWER_SUBJECT, sent) .await { Ok(reply) => match Outcome::decode(&String::from_utf8_lossy(&reply.payload)) { - Outcome::Requested => { - tracing::info!(%flow, %repo, number, %reviewer, "notify router: review requested"); - Ok("requested".to_owned()) - } - Outcome::Skipped(why) => { - tracing::info!(%flow, %repo, number, %reviewer, %why, "notify router: review not requested"); - Ok(format!("skipped: {why}")) - } - Outcome::Error(why) => { - tracing::warn!(%flow, %repo, number, %reviewer, %why, "notify router: the controller could not request the review"); - Err(why) - } + Outcome::Requested => Ok("requested".to_owned()), + Outcome::Skipped(why) => Ok(format!("skipped: {why}")), + Outcome::Error(why) => Err(why), }, - Err(e) => { - tracing::warn!(%flow, %repo, number, %reviewer, error = %e, "notify router: no answer to a review request; not retried"); - Err(format!("no answer: {e}")) - } + Err(e) => Err(format!("no answer: {e}")), }; + log::sink(cause, flow, Kind::RequestReviewer, reviewer, &result); let row = cause.row(flow, &request.node, Kind::RequestReviewer, reviewer, result); log::append(&self.js, &row).await; } @@ -289,6 +285,7 @@ impl Router { Ok(()) => Ok(format!("{} {}", op_name(action.op.op), action.op.key)), Err(e) => Err(format!("{e:#}")), }; + log::sink(cause, &action.flow, Kind::Todo, &action.agent, &result); let row = cause.row( &action.flow, &action.node, @@ -313,7 +310,6 @@ impl Router { .with_context(|| format!("publishing to {subject}"))? .await .with_context(|| format!("waiting for {subject}'s ack"))?; - tracing::info!(agent = %action.agent, flow = %action.flow, key = %action.op.key, op = ?action.op.op, "notify router: todo op queued"); Ok(()) } @@ -399,7 +395,7 @@ mod tests { use super::*; use crate::test_nats::TestNats; - fn cause() -> log::Cause { + pub(super) fn cause() -> log::Cause { log::Cause { event: "issue_assign".to_owned(), repo: "o/r".to_owned(), @@ -407,7 +403,7 @@ mod tests { } } - fn action() -> flow::Action { + pub(super) fn action() -> flow::Action { let flow = builtins().remove(0); let (agent, op) = ( "atlas".to_owned(), @@ -427,7 +423,7 @@ mod tests { } } - fn review() -> flow::ReviewRequest { + pub(super) fn review() -> flow::ReviewRequest { flow::ReviewRequest { flow: "head-classifier".to_owned(), node: "request".to_owned(), @@ -535,3 +531,211 @@ mod tests { ); } } + +#[cfg(test)] +mod log_tests { + use std::collections::BTreeMap; + use std::sync::{Arc, Mutex}; + + use tracing_subscriber::layer::SubscriberExt as _; + + use super::tests::{action, cause, review}; + use super::*; + use crate::test_nats::TestNats; + + type Event = (tracing::Level, BTreeMap); + + /// The level and fields of every event, fields as their debug text. + #[derive(Clone, Default)] + struct Events(Arc>>); + + impl tracing_subscriber::Layer for Events { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + let mut fields = BTreeMap::new(); + event.record( + &mut |field: &tracing::field::Field, value: &dyn std::fmt::Debug| { + fields.insert(field.name().to_owned(), format!("{value:?}")); + }, + ); + self.0 + .lock() + .expect("unpoisoned") + .push((*event.metadata().level(), fields)); + } + } + + impl Events { + /// The events whose message is `message`, with their levels. + fn named(&self, message: &str) -> Vec { + let want = format!("notify router: {message}"); + self.0 + .lock() + .expect("unpoisoned") + .iter() + .filter(|(_, f)| f.get("message").is_some_and(|m| *m == want)) + .cloned() + .collect() + } + } + + fn capture() -> (Events, tracing::subscriber::DefaultGuard) { + let events = Events::default(); + let guard = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); + (events, guard) + } + + fn body(action: &str) -> String { + format!( + r#"{{"action":"{action}","issue":{{"number":9,"title":"t","html_url":"u","assignees":[{{"login":"atlas"}}]}},"repository":{{"full_name":"o/r","id":1}},"sender":{{"login":"mara"}}}}"# + ) + } + + /// Publish one delivery and hand it to the router, as the pull would. + async fn deliver(router: &mut Router, event_type: &str, id: &str, body: String) { + let ack = router + .js + .send_publish( + forge_events::subject(event_type), + PublishMessage::build().payload(body.into()).message_id(id), + ) + .await + .expect("sent") + .await + .expect("acked"); + let mut batch = router + .events + .batch() + .max_messages(1) + .expires(Duration::from_secs(5)) + .messages() + .await + .expect("batch"); + let message = batch.next().await.expect("a message").expect("pulled"); + assert_eq!(message.info().expect("info").stream_sequence, ack.sequence); + assert!(router.handle(&message).await.is_ok()); + } + + /// A delivery a flow wants is one info line with its type, action, repo, + /// number, sender, delivery id and flow count; one nothing wants is a + /// debug line, and neither carries the body. + #[tokio::test] + async fn a_delivery_is_logged_at_info_when_a_flow_matched_and_at_debug_when_none() { + let nats = TestNats::start(); + let client = nats.client().await; + let js = jetstream::new(client.clone()); + let bucket = swarm_wire::agent_status::BUCKET; + js.create_key_value(kv::Config { + bucket: bucket.to_owned(), + ..Default::default() + }) + .await + .expect("bucket") + .put("hive.atlas", "{}".into()) + .await + .expect("an agent"); + let mut router = Router::open(&client).await.expect("opens"); + let (events, _guard) = capture(); + + // Control: nothing is logged before a delivery. + assert!(events.named("forge event routed").is_empty()); + + deliver(&mut router, "issue_assign", "d-1", body("assigned")).await; + let info = events.named("forge event routed"); + let [(level, f)] = info.as_slice() else { + panic!("one line, got {info:?}"); + }; + assert_eq!(*level, tracing::Level::INFO); + let got = |k: &str| f.get(k).map(String::as_str); + assert_eq!(got("event_type"), Some("\"issue_assign\"")); + assert_eq!(got("action"), Some("assigned")); + assert_eq!(got("repo"), Some("o/r")); + assert_eq!(got("number"), Some("9")); + assert_eq!(got("sender"), Some("mara")); + assert_eq!(got("delivery_id"), Some("\"d-1\"")); + assert_eq!(got("flows"), Some("1")); + assert!( + !format!("{f:?}").contains("html_url"), + "the body is not logged" + ); + assert!(events.named("forge event matched no flow").is_empty()); + + deliver(&mut router, "issue_comment", "d-2", body("created")).await; + let none = events.named("forge event matched no flow"); + let [(level, f)] = none.as_slice() else { + panic!("one line, got {none:?}"); + }; + assert_eq!(*level, tracing::Level::DEBUG); + assert_eq!(f.get("flows").map(String::as_str), Some("0")); + assert_eq!(events.named("forge event routed").len(), 1); + } + + /// A sink that went through is an info line and one that failed is a warn + /// carrying the error text, for a review request and for a todo. + #[tokio::test] + async fn a_failing_sink_is_a_warn_with_the_error_and_a_working_one_is_info() { + let nats = TestNats::start(); + let client = nats.client().await; + let router = Router::open(&client).await.expect("opens"); + let (events, _guard) = capture(); + + let mut hears = client + .subscribe(router_sink::REQUEST_REVIEWER_SUBJECT) + .await + .expect("subscribed"); + let answering = client.clone(); + tokio::spawn(async move { + for line in ["requested", "error:Doer can't choose reviewer"] { + let m = hears.next().await.expect("a request"); + answering + .publish(m.reply.expect("a reply subject"), line.into()) + .await + .expect("answered"); + } + }); + router.request_review(&cause(), &review()).await; + let ran = events.named("sink ran"); + let [(level, f)] = ran.as_slice() else { + panic!("one line, got {ran:?}"); + }; + assert_eq!(*level, tracing::Level::INFO); + assert_eq!(f["flow"], "head-classifier"); + assert_eq!((f["repo"].as_str(), f["number"].as_str()), ("o/r", "9")); + assert_eq!(f["kind"], "RequestReviewer"); + assert_eq!(f["target"], "review-classifier"); + assert_eq!(f["outcome"], "requested"); + assert!(events.named("sink failed").is_empty()); + + router.request_review(&cause(), &review()).await; + let failed = events.named("sink failed"); + let [(level, f)] = failed.as_slice() else { + panic!("one line, got {failed:?}"); + }; + assert_eq!(*level, tracing::Level::WARN); + assert_eq!(f["target"], "review-classifier"); + assert_eq!(f["error"], "Doer can't choose reviewer"); + assert_eq!((f["repo"].as_str(), f["number"].as_str()), ("o/r", "9")); + + router.queue(5, &cause(), &action()).await.expect("queued"); + assert_eq!(events.named("sink ran").len(), 2); + router + .js + .delete_stream(router_todo::STREAM) + .await + .expect("deleted"); + router + .queue(6, &cause(), &action()) + .await + .expect_err("no stream takes it"); + let failed = events.named("sink failed"); + assert_eq!(failed.len(), 2); + assert_eq!(failed[1].0, tracing::Level::WARN); + assert_eq!(failed[1].1["kind"], "Todo"); + assert_eq!(failed[1].1["repo"], "o/r"); + assert_eq!(failed[1].1["number"], "9"); + } +}