swarm-metrics-auth: remember a failed lookup briefly so a store outage costs one lookup per burst
After a failed or timed-out store read the responder denies for LOOKUP_FAILURE_WINDOW (3 s) without a new lookup, so a burst of requests during an outage costs one lookup instead of one per request. Three edits besides the code in main.rs belong to this change: - swarm-metrics-auth/Cargo.toml gains tokio's test-util dev-dependency because the burst test runs on paused time (start_paused) to let a hanging store sleep past LOOKUP_TIMEOUT without a real 5 s wait. - docs/swarm/services.md gains the sentence that a failed or timed-out read is denied for three seconds without a retry, which documents the window this change adds. - docs/swarm/credentials.md gains the clause "every three seconds while the store fails" on the Grafana bearer row, which documents the same window in the secrets table.
This commit is contained in:
atlas
mara
parent
6971b4e4ed
commit
70a0435814
4 changed files with 101 additions and 18 deletions
|
|
@ -76,7 +76,7 @@ of the cell says how.
|
|||
| `swarm/hives/<hive>/oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | hive-c0re, under the hive's own certificate, when it first connects to the queue | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ✅ read until the first connect succeeds, retried with a backoff of up to five minutes while it's missing; then held in memory and read again when authelia refuses it |
|
||||
| `swarm/services/<clientId>/oidc/client` | authelia | that service's own reader, under its own certificate — the queue responder's under `swarm-nats-auth`'s, the homeserver's under `swarm-matrix-ctl`'s, the forge's under `swarm-forge-publisher`'s, the review classifier's under `swarm-review-classifier`'s, the CI runner's, Grafana's, the log store's, the metrics store's, the forge's and the homeserver's journal forwarders' under `swarm-ci-runner`'s, `swarm-grafana-oidc`'s, `swarm-logs-auth`'s, `swarm-metrics-auth`'s, `swarm-forge-publisher`'s and `swarm-matrix-ctl`'s, authelia's journal forwarder's under `swarm-secret-publisher`'s; the bao UI's by the store's granter | must be stated | must be stated |
|
||||
| `swarm/services/ci-runner/registration-token` | `swarm-controller`, from the forge's admin API, at start and in a five-minute pass; its grant on the path is write-only | `swarm-bao-ci-runner` on the runner's host, under the runner's own certificate, into a file under the host's `/run` that only `swarm-ci-runner-register` in the runner's container loads; no other principal's policy reads it | ❌ the forge hands out one token until an admin resets it; the controller writes a reset token within five minutes | ✅ `swarm-bao-ci-runner` reads it at boot, on deploy and every ten minutes, and restarts the runner when the value changed |
|
||||
| `swarm/services/<grafana client id>/metrics/bearer` | `swarm-bao-grafana-metrics-bearer` on Grafana's host, under Grafana's certificate, when the store holds none | `swarm-metrics-auth` on the metrics store's host, under its own certificate, at most once a minute, to check the bearer Grafana's datasource presents at the store's read API, and `swarm-logs-grafana-auth` on the log store's host, under `swarm-logs-auth`'s certificate, to check the same bearer at the log store's query API | ❌ nothing re-mints it; an operator deletes it and restarts `swarm-bao-grafana-metrics-bearer` | ✅ `swarm-metrics-auth` reads it again once its copy is a minute old. ❌ Grafana reads its file copy only as it starts; the bearer unit restarts Grafana when the copy changes |
|
||||
| `swarm/services/<grafana client id>/metrics/bearer` | `swarm-bao-grafana-metrics-bearer` on Grafana's host, under Grafana's certificate, when the store holds none | `swarm-metrics-auth` on the metrics store's host, under its own certificate, at most once a minute (every three seconds while the store fails), to check the bearer Grafana's datasource presents at the store's read API, and `swarm-logs-grafana-auth` on the log store's host, under `swarm-logs-auth`'s certificate, to check the same bearer at the log store's query API | ❌ nothing re-mints it; an operator deletes it and restarts `swarm-bao-grafana-metrics-bearer` | ✅ `swarm-metrics-auth` reads it again once its copy is a minute old. ❌ Grafana reads its file copy only as it starts; the bearer unit restarts Grafana when the copy changes |
|
||||
| `swarm/services/review-classifier/anthropic-api-key` | the operator, by hand in the store's UI, field `api_key` | `swarm-review-classifier` in its container, under its own certificate, at the start of every API key run; held for that run and handed to `claude`'s environment only; no other principal's policy reads it | ❌ an operator's key; replaced by writing a new one | ✅ read on every run, so a written or removed key takes effect on the next run |
|
||||
| `swarm/services/review-classifier/acp-provider` | the operator, by hand in the store's UI, field `api_key` | `swarm-review-classifier` in its container, under its own certificate, at the start of every ACP run; held for that run and handed to `opencode`'s environment only; no other principal's policy reads it | ❌ an operator's key; replaced by writing a new one | ✅ read on every run, so a written or removed key takes effect on the next run |
|
||||
| `swarm/services/review-classifier/forge/token` | `swarm-controller`, with the forge's admin API, for its `review-classifier` account, at start and in a five-minute pass; the secret publisher's policy denies it | `swarm-controller` under its own certificate, before it decides whether to mint, and on every review it posts as the account; the classifier service's policy doesn't read it | ✅ the controller's pass re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes) | ✅ the controller reads the store on every review it posts |
|
||||
|
|
|
|||
|
|
@ -238,7 +238,8 @@ Where the secret store runs on the same host, these default to the leaf it
|
|||
mints. On any other host you place a leaf whose CN is
|
||||
`deploy.bao.metricsAuthCommonName`, and evaluation refuses the config until
|
||||
you set both. `swarm-metrics-auth` reads the stored bearer at most once a
|
||||
minute, so a replaced bearer starts working up to a minute later.
|
||||
minute, so a replaced bearer starts working up to a minute later. After a
|
||||
failed or timed-out read it denies for three seconds without trying again.
|
||||
`swarm-metrics-agent-auth` is the log store's `swarm-logs-auth` binary, with
|
||||
the same grant TTL and rate limit, under `deploy.victoriametrics.agentAuth*`
|
||||
(table above).
|
||||
|
|
|
|||
|
|
@ -23,3 +23,8 @@ tracing-subscriber.workspace = true
|
|||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
# `test-util` for `#[tokio::test(start_paused = true)]`: the lookup-timeout
|
||||
# test waits out `LOOKUP_TIMEOUT` on paused time.
|
||||
[dev-dependencies]
|
||||
tokio = { workspace = true, features = ["test-util"] }
|
||||
|
|
|
|||
|
|
@ -12,8 +12,10 @@
|
|||
//!
|
||||
//! The stored value is kept for `--cache-ttl` seconds, and no request reaches
|
||||
//! the store within that window: a replaced bearer is refused, and the one it
|
||||
//! replaced still accepted, for up to that long. Lookups are serialised, so a
|
||||
//! burst of requests against an expired copy costs one.
|
||||
//! replaced still accepted, for up to that long. Lookups are serialised, and a
|
||||
//! failed or timed-out one is remembered for `LOOKUP_FAILURE_WINDOW`, so a
|
||||
//! burst of requests against an expired copy costs one lookup, success or
|
||||
//! failure.
|
||||
|
||||
use std::future::Future;
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
|
|
@ -33,6 +35,13 @@ use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings};
|
|||
/// rather than a gateway timeout.
|
||||
const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
|
||||
/// How long requests are denied without a lookup after one failed or timed
|
||||
/// out. Each request queued behind a failing lookup would otherwise run its
|
||||
/// own, and the k-th would wait k × `LOOKUP_TIMEOUT`, past nginx's 60s read
|
||||
/// timeout. Short against `--cache-ttl`, so a recovered store is used again
|
||||
/// within seconds.
|
||||
const LOOKUP_FAILURE_WINDOW: Duration = Duration::from_secs(3);
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(about = "auth_request responder: checks Grafana's metrics bearer against the store")]
|
||||
struct Args {
|
||||
|
|
@ -47,7 +56,8 @@ struct Args {
|
|||
#[arg(long)]
|
||||
store_path: String,
|
||||
/// Seconds the stored bearer is used without reading the store again;
|
||||
/// 0 reads it on every request.
|
||||
/// 0 reads it on every request, except for a few seconds after a read
|
||||
/// failed.
|
||||
#[arg(long, default_value_t = 60)]
|
||||
cache_ttl: u64,
|
||||
}
|
||||
|
|
@ -97,10 +107,19 @@ struct Cached {
|
|||
read_at: Instant,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Cache {
|
||||
stored: Option<Cached>,
|
||||
/// Until when requests are denied without a lookup, after one failed or
|
||||
/// timed out. Stamped when that lookup ends: the requests queued behind
|
||||
/// it carry the time they arrived, which is earlier.
|
||||
failed_until: Option<Instant>,
|
||||
}
|
||||
|
||||
struct Responder<S> {
|
||||
source: S,
|
||||
ttl: Duration,
|
||||
cached: tokio::sync::Mutex<Option<Cached>>,
|
||||
cached: tokio::sync::Mutex<Cache>,
|
||||
}
|
||||
|
||||
/// The answer for one subrequest's `Authorization` header, at `now`.
|
||||
|
|
@ -115,17 +134,23 @@ async fn decide(
|
|||
};
|
||||
// Held across the lookup, so concurrent requests wait for one read
|
||||
// rather than each starting their own.
|
||||
let mut cached = responder.cached.lock().await;
|
||||
let fresh = cached
|
||||
let mut cache = responder.cached.lock().await;
|
||||
let fresh = cache
|
||||
.stored
|
||||
.as_ref()
|
||||
.is_some_and(|c| now.saturating_duration_since(c.read_at) < responder.ttl);
|
||||
if !fresh {
|
||||
if cache.failed_until.is_some_and(|until| now < until) {
|
||||
tracing::warn!("bearer lookup failed a moment ago; denying without another");
|
||||
return StatusCode::UNAUTHORIZED;
|
||||
}
|
||||
match tokio::time::timeout(LOOKUP_TIMEOUT, responder.source.lookup()).await {
|
||||
Ok(Ok(Some(value))) if !value.is_empty() => {
|
||||
*cached = Some(Cached {
|
||||
cache.stored = Some(Cached {
|
||||
value,
|
||||
read_at: now,
|
||||
});
|
||||
cache.failed_until = None;
|
||||
}
|
||||
Ok(Ok(_)) => {
|
||||
tracing::warn!("no bearer is stored; denying");
|
||||
|
|
@ -133,15 +158,17 @@ async fn decide(
|
|||
}
|
||||
Ok(Err(e)) => {
|
||||
tracing::warn!(error = format!("{e:#}"), "bearer lookup failed; denying");
|
||||
cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW);
|
||||
return StatusCode::UNAUTHORIZED;
|
||||
}
|
||||
Err(_) => {
|
||||
tracing::warn!("bearer lookup timed out; denying");
|
||||
cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW);
|
||||
return StatusCode::UNAUTHORIZED;
|
||||
}
|
||||
}
|
||||
}
|
||||
let Some(stored) = cached.as_ref() else {
|
||||
let Some(stored) = cache.stored.as_ref() else {
|
||||
return StatusCode::UNAUTHORIZED;
|
||||
};
|
||||
// Constant time, so the time to refuse says nothing about how much of the
|
||||
|
|
@ -183,7 +210,7 @@ async fn main() -> anyhow::Result<()> {
|
|||
path: args.store_path,
|
||||
},
|
||||
ttl: Duration::from_secs(args.cache_ttl),
|
||||
cached: tokio::sync::Mutex::new(None),
|
||||
cached: tokio::sync::Mutex::new(Cache::default()),
|
||||
});
|
||||
|
||||
// `RuntimeDirectoryPreserve=yes` can leave the previous run's socket; the
|
||||
|
|
@ -212,10 +239,11 @@ mod tests {
|
|||
use std::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
/// A store answering `stored` (or failing when it is `Err`), counting
|
||||
/// lookups.
|
||||
/// A store answering `stored` (or failing when it is `Err`) after `delay`,
|
||||
/// counting lookups.
|
||||
struct Fake {
|
||||
stored: Mutex<Result<Option<String>, ()>>,
|
||||
delay: Duration,
|
||||
lookups: AtomicUsize,
|
||||
}
|
||||
|
||||
|
|
@ -223,10 +251,18 @@ mod tests {
|
|||
fn holding(value: &str) -> Self {
|
||||
Self {
|
||||
stored: Mutex::new(Ok(Some(value.to_owned()))),
|
||||
delay: Duration::ZERO,
|
||||
lookups: AtomicUsize::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
fn hanging() -> Self {
|
||||
Self {
|
||||
delay: Duration::from_hours(1),
|
||||
..Self::holding(SECRET)
|
||||
}
|
||||
}
|
||||
|
||||
fn set(&self, stored: Result<Option<&str>, ()>) {
|
||||
*self.stored.lock().unwrap() = stored.map(|v| v.map(str::to_owned));
|
||||
}
|
||||
|
|
@ -239,6 +275,7 @@ mod tests {
|
|||
impl BearerSource for &Fake {
|
||||
async fn lookup(&self) -> anyhow::Result<Option<String>> {
|
||||
self.lookups.fetch_add(1, Ordering::SeqCst);
|
||||
tokio::time::sleep(self.delay).await;
|
||||
self.stored
|
||||
.lock()
|
||||
.unwrap()
|
||||
|
|
@ -247,6 +284,12 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
impl BearerSource for Arc<Fake> {
|
||||
async fn lookup(&self) -> anyhow::Result<Option<String>> {
|
||||
(&**self).lookup().await
|
||||
}
|
||||
}
|
||||
|
||||
const TTL: Duration = Duration::from_mins(1);
|
||||
const SECRET: &str = "0f1e2d3c4b5a";
|
||||
const GOOD: &str = "Bearer 0f1e2d3c4b5a";
|
||||
|
|
@ -255,7 +298,7 @@ mod tests {
|
|||
Responder {
|
||||
source: store,
|
||||
ttl: TTL,
|
||||
cached: tokio::sync::Mutex::new(None),
|
||||
cached: tokio::sync::Mutex::new(Cache::default()),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -338,17 +381,51 @@ mod tests {
|
|||
assert_eq!(store.lookups(), 2);
|
||||
}
|
||||
|
||||
/// A failed read leaves nothing usable cached, so the next request reads
|
||||
/// the store again rather than waiting out a TTL.
|
||||
/// A failed read is remembered for `LOOKUP_FAILURE_WINDOW`, not for a TTL:
|
||||
/// requests inside it are denied without a lookup, the first one after it
|
||||
/// reads the store again.
|
||||
#[tokio::test]
|
||||
async fn a_failed_lookup_is_retried_on_the_next_request() {
|
||||
async fn a_failed_lookup_is_retried_after_the_failure_window() {
|
||||
let store = Fake::holding(SECRET);
|
||||
store.set(Err(()));
|
||||
let r = responder(&store);
|
||||
let t0 = Instant::now();
|
||||
assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED);
|
||||
store.set(Ok(Some(SECRET)));
|
||||
assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK);
|
||||
assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED);
|
||||
assert_eq!(store.lookups(), 1);
|
||||
let after = t0 + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1);
|
||||
assert_eq!(decide(&r, Some(GOOD), after).await, StatusCode::OK);
|
||||
assert_eq!(store.lookups(), 2);
|
||||
}
|
||||
|
||||
/// A burst against a store that errors, or that sleeps past
|
||||
/// `LOOKUP_TIMEOUT`, costs one lookup, and every request is denied
|
||||
/// within that one timeout rather than queueing behind one timeout each.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn a_burst_against_a_failing_store_costs_one_lookup() {
|
||||
const BURST: usize = 20;
|
||||
let erroring = Fake::holding(SECRET);
|
||||
erroring.set(Err(()));
|
||||
for (name, store) in [("erroring", erroring), ("hanging", Fake::hanging())] {
|
||||
let store = Arc::new(store);
|
||||
let r = Arc::new(Responder {
|
||||
source: Arc::clone(&store),
|
||||
ttl: TTL,
|
||||
cached: tokio::sync::Mutex::new(Cache::default()),
|
||||
});
|
||||
let started = tokio::time::Instant::now();
|
||||
let now = Instant::now();
|
||||
let mut burst = tokio::task::JoinSet::new();
|
||||
for _ in 0..BURST {
|
||||
let r = Arc::clone(&r);
|
||||
burst.spawn(async move { decide(&r, Some(GOOD), now).await });
|
||||
}
|
||||
while let Some(status) = burst.join_next().await {
|
||||
assert_eq!(status.unwrap(), StatusCode::UNAUTHORIZED, "{name}");
|
||||
}
|
||||
assert_eq!(store.lookups(), 1, "{name}");
|
||||
assert!(started.elapsed() <= LOOKUP_TIMEOUT, "{name}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue