140 lines
5 KiB
Rust
140 lines
5 KiB
Rust
//! Publishing lifecycle notices onto the swarm queue.
|
|
//!
|
|
//! Replaces the old `push_todo(MANAGER_AGENT, ...)` fallback the
|
|
//! lifecycle-notice call sites used to reach for when there was nobody
|
|
//! else to tell. Every hive is swarm-controlled now, so there is no case
|
|
//! left that needs a manager-agent recipient — this module has no such
|
|
//! fallback, on purpose, not by omission.
|
|
//!
|
|
//! **Why a stream and not the [`crate::swarm_status`] KV bucket shape**:
|
|
//! a status snapshot has a current value a late reader can always ask
|
|
//! for; a lifecycle notice ("container crashed at 04:12") does not — miss
|
|
//! it and there is nothing left to read later that says it happened. See
|
|
//! [`swarm_queue_client::notices`] for the stream this publishes into.
|
|
//!
|
|
//! **Best-effort, never fatal to the caller.** A hive with no queue
|
|
//! configured is a silent no-op (the ordinary case). A hive whose queue
|
|
//! is unreachable loses the swarm's visibility of the notice, not the
|
|
//! host's — `warn!` fires on every failed attempt regardless, and the
|
|
//! dashboard banners only after [`FAILURES_BEFORE_BANNER`] consecutive
|
|
//! misses, the same debounce shape [`crate::swarm_status`] uses and for
|
|
//! the same reason: a `warn!` that fires every call for three weeks is
|
|
//! indistinguishable from silence in practice.
|
|
|
|
use anyhow::{Context as _, Result};
|
|
use tokio::sync::{Mutex, OnceCell};
|
|
|
|
use crate::stats::sweep_health::{self, SweepHealth};
|
|
|
|
/// Consecutive failed publishes before the dashboard banners — same
|
|
/// value [`crate::swarm_status`] uses and for the same reason: a debounce
|
|
/// against one blip flapping a banner an operator learns to ignore.
|
|
const FAILURES_BEFORE_BANNER: u32 = 3;
|
|
|
|
static HEALTH: OnceCell<Mutex<SweepHealth>> = OnceCell::const_new();
|
|
|
|
async fn health() -> tokio::sync::MutexGuard<'static, SweepHealth> {
|
|
HEALTH
|
|
.get_or_init(|| async {
|
|
Mutex::new(SweepHealth::new(
|
|
"swarm_notices_publish",
|
|
"warn",
|
|
FAILURES_BEFORE_BANNER,
|
|
))
|
|
})
|
|
.await
|
|
.lock()
|
|
.await
|
|
}
|
|
|
|
/// Publish one lifecycle notice for this hive.
|
|
///
|
|
/// `subsystem`/`key`/`summary`/`source` carry the same meaning and the
|
|
/// same owned-`String` shape they did as `push_todo` arguments — a
|
|
/// drop-in replacement for that call, minus the recipient (there is
|
|
/// none) and `reopen_if_acked` (an inbox-todo concept with no equivalent
|
|
/// on an append-only stream).
|
|
pub async fn notify(subsystem: &str, key: Option<String>, summary: String, source: Option<String>) {
|
|
let Some(client) = crate::swarm_queue::client().await else {
|
|
return;
|
|
};
|
|
// Same absent-name condition `swarm_status` bails on — that module
|
|
// already banners it under `swarm_status_config` the first time
|
|
// either of us hits it; nothing more to add from here.
|
|
let Some(hive) = crate::container_view::hive_swarm_names().0 else {
|
|
return;
|
|
};
|
|
|
|
match publish(
|
|
&client,
|
|
&hive,
|
|
subsystem,
|
|
key.as_deref(),
|
|
&summary,
|
|
source.as_deref(),
|
|
)
|
|
.await
|
|
{
|
|
Ok(()) => health().await.record_ok(),
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, subsystem, summary, "swarm notice: publish failed");
|
|
let err = format!("{e:#}");
|
|
health().await.record_err(|ctx| {
|
|
let age = ctx.since_last_ok.map_or_else(
|
|
|| "no success this session".to_owned(),
|
|
|d| format!("last ok {} ago", sweep_health::fmt_age(d)),
|
|
);
|
|
format!(
|
|
"swarm notice publishing is failing ({} consecutive, {age}) \
|
|
— notices are being lost, not just delayed: {err}",
|
|
ctx.consecutive
|
|
)
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(serde::Serialize)]
|
|
struct Notice<'a> {
|
|
subsystem: &'a str,
|
|
key: Option<&'a str>,
|
|
summary: &'a str,
|
|
source: Option<&'a str>,
|
|
}
|
|
|
|
async fn publish(
|
|
client: &async_nats::Client,
|
|
hive: &str,
|
|
subsystem: &str,
|
|
key: Option<&str>,
|
|
summary: &str,
|
|
source: Option<&str>,
|
|
) -> Result<()> {
|
|
// An unconnected client does not fail a JetStream request, it hangs
|
|
// on it — see `ensure_connected`'s own doc comment for why this has
|
|
// to run before every such request, not just the first one.
|
|
swarm_queue_client::ensure_connected(client)?;
|
|
|
|
// Ensures the stream exists; the handle itself is unused below —
|
|
// `Context::publish` routes by subject, it does not need the
|
|
// `Stream` object in hand.
|
|
swarm_queue_client::notices::open_or_create(client)
|
|
.await
|
|
.context("opening the notices stream")?;
|
|
|
|
let payload = serde_json::to_vec(&Notice {
|
|
subsystem,
|
|
key,
|
|
summary,
|
|
source,
|
|
})
|
|
.context("serialising the notice")?;
|
|
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
js.publish(swarm_queue_client::notices::subject(hive), payload.into())
|
|
.await
|
|
.context("publishing the notice")?
|
|
.await
|
|
.context("awaiting the notice's ack")?;
|
|
Ok(())
|
|
}
|