notify router: log each delivered forge event and each sink outcome

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

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<String, String>) {
+    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.

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<String, String>);
+
+    /// The level and fields of every event, fields as their debug text.
+    #[derive(Clone, Default)]
+    struct Events(Arc<Mutex<Vec<Event>>>);
+
+    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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<Event> {
+            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");
+    }
+}