vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

hive status publishing: read the hive's OIDC secret from the store, retry until it is there

Browse source 
hive-c0re publishes this hive's status to the swarm queue as the OIDC
client `hive-<name>`. Its secret came from
`deploy.hive-controller.statusPublish.clientSecretFile`, which defaulted
to authelia's minted file only on a host running both authelia and the
queue. Every other hive had no secret and never published, so the
controller showed it `never_reported`.

Now `swarm-secret-publish`, beside authelia, also copies each hive's
`hive-<name>` secret to `swarm/hives/<hive>/oidc/client`. Its policy
already allows `swarm/hives/*`. hive-c0re reads that path under the
hive's own store leaf, which the existing `swarm/hives/<hive>/*` grant
covers, and holds the value in memory.

The shared queue client no longer remembers a failed first connect for
the life of the process. A failed attempt is retried with a backoff
from 5s, doubling to a cap of 5 minutes; the dashboard banner stays up
until an attempt succeeds. The two status publishers wait for the
client instead of giving up on the first `None`, so a secret published
after boot is picked up without a restart.

Removed: the `statusPublish.clientSecretFile` option (both spellings,
now removed options with a message), its same-host default, its
LoadCredential, the assertion tying it to the other two coordinates,
and `swarm_queue_client::QueueConfig::from_env`, whose only caller was
this read. Publishing is now on wherever the queue URL and token
endpoint are set and the hive holds a store leaf.

Refs #5015
This commit is contained in:
atlas 2026-10-09 23:12:28 +02:00 • committed by mara
commit c2c61b6146
19 changed files with 643 additions and 394 deletions
Download patch file Download diff file Expand all files Collapse all files

3
hive-c0re/src/lifecycle/host_config.rs
View file

@ -272,8 +272,7 @@ fn queue_agent_credentials(agent_name: &str, dir: Option<&Path>) -> Vec<Credenti
};
let secret = dir.join("secret");
let client_id = dir.join("client_id");
// Both or neither, for the same reason `QueueConfig::from_env` refuses a
// half-set environment: a client holding one of the two comes up fine and
// Both or neither: a client holding one of the two comes up fine and
// never connects.
if !secret.is_file() || !client_id.is_file() {
tracing::info!(

6
hive-c0re/src/swarm_agent_status.rs
View file

@ -33,8 +33,8 @@ const FAILURES_BEFORE_BANNER: u32 = 3;
/// Start the publish loop, if this deployment wired up a swarm queue.
///
/// Shares its connect gate with [`crate::swarm_status::spawn`] — both read
/// `crate::swarm_queue::client()`, which connects once per process — so a
/// Shares its connect gate with [`crate::swarm_status::spawn`] — both wait on
/// `crate::swarm_queue::wait_for_client()`, one connection per process — so a
/// hive with no queue coordinates pays for this decision once, not twice.
///
/// Takes no `Coordinator` handle, unlike its sibling: this task only reads
@ -51,7 +51,7 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
};
tokio::spawn(async move {
let Some(client) = crate::swarm_queue::client().await else {
let Some(client) = crate::swarm_queue::wait_for_client().await else {
return;
};

429
hive-c0re/src/swarm_queue.rs
View file

@ -10,63 +10,400 @@
//! crate in the first place, one layer up. This module is that same move
//! made again, this time between two consumers *inside* one process.
//!
//! The queue's address, the token endpoint and the client id come from the
//! environment. The client secret comes from the swarm's secret store, at
//! `swarm_secret_client::queue::hive_client_path`, read under this hive's own
//! certificate and held only in memory. `swarm-secret-publish`, on the host
//! that runs authelia, puts it there.
//!
//! Connects lazily on first use rather than at boot — nothing here
//! blocks `hive-c0re` starting up on hosts with no queue configured,
//! which is the ordinary case.
//! which is the ordinary case. A failed attempt is retried after a
//! backoff capped at [`MAX_RETRY_DELAY`], so a secret published after this
//! daemon started is picked up without a restart.
use tokio::sync::OnceCell;
use std::future::Future;
use std::path::PathBuf;
use std::time::Duration;
/// Env var prefix for this daemon's swarm-queue credentials — see
/// [`swarm_queue_client::QueueConfig::from_env`]. All four or none.
use anyhow::{Context, Result, anyhow, bail};
use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::queue::{self, HiveClientCredential};
use tokio::sync::Mutex;
use tokio::time::Instant;
/// Env var prefix for this daemon's swarm-queue coordinates — see
/// [`coordinates_from`].
const ENV_PREFIX: &str = "HIVE_C0RE";
static CLIENT: OnceCell<Option<async_nats::Client>> = OnceCell::const_new();
/// Wait after the first failed attempt; doubled after each further one.
const FIRST_RETRY_DELAY: Duration = Duration::from_secs(5);
/// The shared swarm-queue client, connecting on first call and memoized
/// for the rest of the process's life.
/// The longest wait between two attempts.
const MAX_RETRY_DELAY: Duration = Duration::from_mins(5);
/// The banner kind a failed attempt raises, cleared by the next success.
const BANNER: &str = "swarm_queue_config";
static CLIENT: Retrying<async_nats::Client> = Retrying::new(FIRST_RETRY_DELAY, MAX_RETRY_DELAY);
/// The shared swarm-queue client, or `None` when there is none right now.
///
/// `None` covers both "no queue configured" (the ordinary case, logged
/// once at `info`) and "config present but connecting failed" (bannered
/// once via [`crate::warnings::set_boot_warning`] the first time this is
/// called) — either way, a caller with `None` should just skip whatever
/// it was about to publish. No caller needs to distinguish the two: both
/// mean "this hive is not offering anything to the swarm right now."
/// `None` covers "no queue configured" (the ordinary case, logged once at
/// `info`), a deployment bug in the coordinates (bannered, and permanent for
/// this process), and an attempt that failed or is still backing off. A
/// caller with `None` should just skip whatever it was about to publish.
/// Never waits out a backoff: see [`wait_for_client`] for that.
pub async fn client() -> Option<async_nats::Client> {
CLIENT.get_or_init(connect_once).await.clone()
match CLIENT.get(connect_once).await {
Outcome::Ready(client) => Some(client),
Outcome::Off | Outcome::NotBefore(_) => None,
}
}
async fn connect_once() -> Option<async_nats::Client> {
let cfg = match swarm_queue_client::QueueConfig::from_env(ENV_PREFIX) {
Ok(Some(cfg)) => cfg,
Ok(None) => {
tracing::info!("no swarm queue configured; this hive offers nothing upward");
return None;
}
Err(e) => {
// A one-shot startup step with no later retry to clear it —
// exactly what `set_boot_warning` is for. `chain`, not
// `{:#}`: this is `swarm_queue_client::Error`, whose
// `Display` ignores the alternate flag (see `chain`'s own
// doc comment), so `{:#}` would drop which env vars are
// actually missing.
crate::warnings::set_boot_warning(
"swarm_queue_config",
"warn",
format!("swarm queue is off: {}", swarm_queue_client::chain(&e)),
);
return None;
}
};
match swarm_queue_client::connect(cfg).await {
Ok(client) => Some(client),
Err(e) => {
crate::warnings::set_boot_warning(
"swarm_queue_config",
"warn",
format!("swarm queue is off: {}", swarm_queue_client::chain(&e)),
);
None
/// The shared swarm-queue client, waiting through every backoff until there
/// is one. `None` only when this hive publishes nothing to a queue at all.
pub async fn wait_for_client() -> Option<async_nats::Client> {
loop {
match CLIENT.get(connect_once).await {
Outcome::Ready(client) => return Some(client),
Outcome::Off => return None,
Outcome::NotBefore(at) => tokio::time::sleep_until(at).await,
}
}
}
/// What one connection attempt came to.
enum Attempt<T> {
/// Nothing to connect to, for the life of the process.
Off,
Ready(T),
/// Worth trying again later.
Failed(anyhow::Error),
}
/// What [`Retrying::get`] answers.
enum Outcome<T> {
Off,
Ready(T),
/// The last attempt failed; the next one is not tried before this.
NotBefore(Instant),
}
enum Slot<T> {
Off,
Ready(T),
/// `failures` consecutive attempts have failed. `banner` holds the
/// dashboard warning until the slot leaves this state.
Failing {
failures: u32,
not_before: Instant,
banner: Option<crate::warnings::WarningGuard>,
},
}
/// A value made by an attempt that can fail: a success or an `Off` is kept
/// for the life of the process, a failure only until its backoff runs out.
struct Retrying<T> {
slot: Mutex<Option<Slot<T>>>,
first_delay: Duration,
max_delay: Duration,
}
impl<T: Clone> Retrying<T> {
const fn new(first_delay: Duration, max_delay: Duration) -> Self {
Self {
slot: Mutex::const_new(None),
first_delay,
max_delay,
}
}
/// The kept value, or one new `attempt` if the last one's backoff has run
/// out. Concurrent callers wait on one attempt rather than each making
/// their own.
async fn get<F, Fut>(&self, attempt: F) -> Outcome<T>
where
F: FnOnce() -> Fut,
Fut: Future<Output = Attempt<T>>,
{
let mut slot = self.slot.lock().await;
let failures = match &*slot {
None => 0,
Some(Slot::Off) => return Outcome::Off,
Some(Slot::Ready(value)) => return Outcome::Ready(value.clone()),
Some(Slot::Failing {
failures,
not_before,
..
}) => {
if Instant::now() < *not_before {
return Outcome::NotBefore(*not_before);
}
*failures
}
};
match attempt().await {
Attempt::Off => {
*slot = Some(Slot::Off);
Outcome::Off
}
Attempt::Ready(value) => {
if failures > 0 {
tracing::info!(failures, "swarm queue: connected after earlier failures");
}
*slot = Some(Slot::Ready(value.clone()));
Outcome::Ready(value)
}
Attempt::Failed(e) => {
let failures = failures.saturating_add(1);
let delay = self.delay_after(failures);
let not_before = Instant::now() + delay;
let message = format!("swarm queue is off, retrying in {delay:?}: {e:#}");
tracing::warn!(failures, "{message}");
// Updating the held guard keeps the banner up without a
// moment where it is cleared and re-raised.
let banner = match slot.take() {
Some(Slot::Failing {
banner: Some(banner),
..
}) => {
banner.update("warn", message);
banner
}
_ => crate::warnings::set_warning(BANNER, "warn", message),
};
*slot = Some(Slot::Failing {
failures,
not_before,
banner: Some(banner),
});
Outcome::NotBefore(not_before)
}
}
}
/// `first_delay` doubled per failure after the first, capped at
/// `max_delay`.
fn delay_after(&self, failures: u32) -> Duration {
let doublings = failures.saturating_sub(1).min(16);
self.first_delay
.saturating_mul(1 << doublings)
.min(self.max_delay)
}
}
async fn connect_once() -> Attempt<async_nats::Client> {
let coordinates = match coordinates_from(|name| std::env::var(name).ok()) {
Ok(Some(c)) => c,
Ok(None) => {
tracing::info!("no swarm queue configured; this hive offers nothing upward");
return Attempt::Off;
}
Err(e) => {
// A deployment bug, which no retry clears.
crate::warnings::set_boot_warning(BANNER, "warn", format!("swarm queue is off: {e:#}"));
return Attempt::Off;
}
};
let secret = match fetch_secret().await {
Ok(secret) => secret,
Err(e) => return Attempt::Failed(e),
};
match swarm_queue_client::connect(coordinates.with_secret(secret)).await {
Ok(client) => Attempt::Ready(client),
// `chain`, not `{:#}`: this is `swarm_queue_client::Error`, whose
// `Display` ignores the alternate flag.
Err(e) => Attempt::Failed(anyhow!(swarm_queue_client::chain(&e))),
}
}
/// Everything in a [`QueueConfig`] except the secret.
#[derive(Debug, PartialEq, Eq)]
struct Coordinates {
url: String,
token_endpoint: String,
client_id: String,
ca_file: Option<PathBuf>,
}
impl Coordinates {
fn with_secret(self, secret: String) -> QueueConfig {
QueueConfig {
url: self.url,
token_endpoint: self.token_endpoint,
client_id: self.client_id,
client_secret: ClientSecret::Value(secret),
ca_file: self.ca_file,
}
}
}
/// `<ENV_PREFIX>_NATS_URL`, `_OIDC_TOKEN_ENDPOINT` and `_OIDC_CLIENT_ID`, all
/// or none; `_OIDC_CA_FILE` optionally.
fn coordinates_from(var: impl Fn(&str) -> Option<String>) -> Result<Option<Coordinates>> {
let url = var(&format!("{ENV_PREFIX}_NATS_URL"));
let token_endpoint = var(&format!("{ENV_PREFIX}_OIDC_TOKEN_ENDPOINT"));
let client_id = var(&format!("{ENV_PREFIX}_OIDC_CLIENT_ID"));
let ca_file = var(&format!("{ENV_PREFIX}_OIDC_CA_FILE")).map(PathBuf::from);
match (url, token_endpoint, client_id) {
(None, None, None) => Ok(None),
(Some(url), Some(token_endpoint), Some(client_id)) => Ok(Some(Coordinates {
url,
token_endpoint,
client_id,
ca_file,
})),
_ => bail!(
"swarm queue is half-configured: {ENV_PREFIX}_NATS_URL, \
{ENV_PREFIX}_OIDC_TOKEN_ENDPOINT and {ENV_PREFIX}_OIDC_CLIENT_ID must be set \
together or not at all"
),
}
}
/// One login and one read, under this hive's own certificate. An empty value
/// is refused like an absent one: it would reach authelia as the secret and
/// come back as `invalid_client`, which names neither the store nor the
/// publisher.
async fn fetch_secret() -> Result<String> {
let Some(hive) = crate::container_view::hive_swarm_names().0 else {
bail!("HYPERHIVE_HIVE_NAME is unset, so there is no store path for this hive's secret");
};
let path = queue::hive_client_path(&hive)?;
let store = swarm_secret_client::SecretStore::from_env(&hive)
.await
.context("logging in to the swarm secret store")?;
let stored: Option<HiveClientCredential> = store
.read_optional(&path)
.await
.with_context(|| format!("reading {path}"))?;
match stored {
Some(c) if !c.value.trim().is_empty() => Ok(c.value.trim().to_owned()),
_ => bail!(
"nothing at {path}: swarm-secret-publish, on the host that runs authelia, has not \
published this hive's client secret"
),
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use super::*;
fn env(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option<String> {
let map: HashMap<String, String> = pairs
.iter()
.map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
.collect();
move |name| map.get(name).cloned()
}
const FULL: [(&str, &str); 3] = [
("HIVE_C0RE_NATS_URL", "tls://nats.example:4222"),
(
"HIVE_C0RE_OIDC_TOKEN_ENDPOINT",
"https://auth.example/api/oidc/token",
),
("HIVE_C0RE_OIDC_CLIENT_ID", "hive-alpha"),
];
#[test]
fn no_coordinates_is_no_queue() {
assert_eq!(
coordinates_from(env(&[])).expect("absent is not an error"),
None
);
}
#[test]
fn a_half_set_environment_is_an_error_naming_the_variables() {
let err = coordinates_from(env(&FULL[..2])).expect_err("a partial set is not absent");
assert!(
format!("{err}").contains("HIVE_C0RE_OIDC_CLIENT_ID"),
"{err}"
);
}
#[test]
fn a_full_environment_carries_the_stored_secret_as_a_value() {
let cfg = coordinates_from(env(&FULL))
.expect("complete")
.expect("configured")
.with_secret("s3cret".to_owned());
assert_eq!(cfg.client_id, "hive-alpha");
assert!(matches!(&cfg.client_secret, ClientSecret::Value(v) if v == "s3cret"));
}
/// The secret is not in the store when this daemon first asks, and is
/// later: the later value is used, with no restart in between.
#[tokio::test]
async fn a_failed_first_read_is_retried_and_its_later_secret_used() {
let shared = Retrying::<String>::new(Duration::ZERO, Duration::ZERO);
let first = shared
.get(|| async { Attempt::Failed(anyhow!("nothing at the path yet")) })
.await;
assert!(
matches!(first, Outcome::NotBefore(_)),
"the first read failed"
);
let second = shared
.get(|| async { Attempt::Ready("published-later".to_owned()) })
.await;
assert!(
matches!(&second, Outcome::Ready(s) if s == "published-later"),
"the read after the failure is made and its secret used"
);
let kept = shared
.get(|| async { panic!("a success is kept, not re-read") })
.await;
assert!(matches!(&kept, Outcome::Ready(s) if s == "published-later"));
}
#[tokio::test(start_paused = true)]
async fn a_failure_is_not_retried_before_its_backoff_runs_out() {
let shared = Retrying::<String>::new(Duration::from_secs(5), Duration::from_mins(5));
let failed = shared
.get(|| async { Attempt::Failed(anyhow!("down")) })
.await;
let Outcome::NotBefore(at) = failed else {
panic!("a failure backs off");
};
assert_eq!(at - Instant::now(), Duration::from_secs(5));
let early = shared
.get(|| async { panic!("no attempt inside the backoff") })
.await;
assert!(matches!(early, Outcome::NotBefore(t) if t == at));
tokio::time::advance(Duration::from_secs(5)).await;
let due = shared
.get(|| async { Attempt::Ready("up".to_owned()) })
.await;
assert!(matches!(&due, Outcome::Ready(s) if s == "up"));
}
#[test]
fn the_backoff_doubles_up_to_its_cap() {
let shared = Retrying::<String>::new(Duration::from_secs(5), Duration::from_mins(5));
let delays: Vec<u64> = (1..=8).map(|n| shared.delay_after(n).as_secs()).collect();
assert_eq!(delays, [5, 10, 20, 40, 80, 160, 300, 300]);
assert_eq!(shared.delay_after(u32::MAX), Duration::from_mins(5));
}
#[tokio::test]
async fn off_is_kept_for_the_life_of_the_process() {
let shared = Retrying::<String>::new(Duration::ZERO, Duration::ZERO);
assert!(matches!(
shared.get(|| async { Attempt::Off }).await,
Outcome::Off
));
let again = shared.get(|| async { panic!("off is not re-tried") }).await;
assert!(matches!(again, Outcome::Off));
}
}

7
hive-c0re/src/swarm_status.rs
View file

@ -75,10 +75,9 @@ pub fn spawn(
};
tokio::spawn(async move {
let Some(client) = crate::swarm_queue::client().await else {
// Absent or failed — either way already handled (an `info`
// log or a `swarm_queue_config` banner) by the shared
// connector; nothing left to report here.
let Some(client) = crate::swarm_queue::wait_for_client().await else {
// No queue configured, already logged or bannered by the
// shared connector; nothing left to report here.
return;
};