Watch
0
0
Fork
You've already forked hyperhive
0

swarm-controller: refuse a cap systemd would ignore; GET creates no bucket

systemd ignores a CPUQuota= or MemoryMax= it can't parse ("Invalid ...,
ignoring") and runs the unit without that limit. A swarm cap replaces
the hive's answer, so such a value left the agent uncapped while the
route, GET and the dashboard all said it was capped. argus on #4268 (R3)
found values the route took that systemd ignores: 12.345%, 400.123%,
0.001%, 17000000T, 99999999999999999999T and 99999999999999999999%.

The validators now accept a subset of what systemd parses, measured
against `systemd-analyze verify` (systemd 260):

- a percentage takes at most two decimals and is above zero, up to 100%
  for memory_max and 21474836.47% for cpu_quota (systemd keeps the quota
  in hundredths of a percent in a C int, not a u64)
- a size is one parse_size term: digits, an optional fraction, then
  nothing, B, K, M, G or T, case-sensitive. The whole part and the
  fraction digits fit u64, the whole part (plus one with a fraction)
  fits u64 once scaled, and the byte count is at least 1 and below
  2^64 - 1, as systemd checks them.

P and E, which systemd reads, stay refused because hive-c0re's validator
refuses them, and one value the hive refuses stops every cap update on
that hive. Values are stored as given: lower-case suffixes and a B
after a multiplier are refused rather than normalized, as systemd
refuses them.

The newline-injection values from the hive's test are in both bad lists
and go through the router in a_cap_with_a_second_line_is_a_400_and_the_caps_stay.

GET reads through a new agent_limits::open that returns None for a
missing bucket, answered as {}, instead of creating the bucket.
This commit is contained in:
atlas 2026-10-12 12:03:49 +02:00 • committed by mara
commit 028ca1bf1f
4 changed files with 259 additions and 69 deletions

View file

@ -491,9 +491,13 @@ caps. swarm-controller alone writes it:
| `PUT` | `/api/agents/<agent>/limits` | `{"cpu_quota": "400%", "memory_max": "8G"}`, every field optional, replacing the stored value | | `PUT` | `/api/agents/<agent>/limits` | `{"cpu_quota": "400%", "memory_max": "8G"}`, every field optional, replacing the stored value |
| `DELETE` | `/api/agents/<agent>/limits` | —; answers 204 | | `DELETE` | `/api/agents/<agent>/limits` | —; answers 204 |
`PUT` refuses an agent no hive's wanted state declares, unknown fields, zero `PUT` refuses an agent no hive's wanted state declares, unknown fields, and
and anything systemd doesn't take, and stores memory the way systemd spells any value systemd would ignore, which would leave the agent uncapped. A
it (`8gb` as `8G`). percentage has at most two decimals and is above zero, up to `100%` for
`memory_max` and `21474836.47%` for `cpu_quota`. A size is digits with an
optional fraction and an optional `B`, `K`, `M`, `G` or `T` (upper case), at
least one byte and below 2^64 - 1 bytes. The controller stores each value as
given.
Every 30 seconds hive-c0re (`workers::swarm_limits`) reads the key of each Every 30 seconds hive-c0re (`workers::swarm_limits`) reads the key of each
agent it has and stores the result in `meta/swarm-resource-limits.json`. Each agent it has and stores the result in `meta/swarm-resource-limits.json`. Each

View file

@ -7,9 +7,9 @@
//! hive's own answer for that field, and a field left unset stays with the //! hive's own answer for that field, and a field left unset stays with the
//! hive. //! hive.
//! //!
//! The hive hands these values to systemd verbatim, and a value systemd //! The hive hands these values to systemd verbatim, and systemd ignores a value
//! rejects stops the agent's container from starting, so they are validated //! it can't parse, leaving the agent uncapped, so they are checked here against
//! here as well as on the hive. //! systemd's own syntax (`Caps`) as well as on the hive.
use axum::Json; use axum::Json;
use axum::extract::{Path, State}; use axum::extract::{Path, State};
@ -27,16 +27,23 @@ use crate::{AppState, error_problem, wanted_error_status};
const MAX_ATTEMPTS: usize = 5; const MAX_ATTEMPTS: usize = 5;
/// One agent's caps. A field left out is not capped by the swarm. /// One agent's caps. A field left out is not capped by the swarm.
///
/// Each value is stored as given, and only in a form systemd applies: a value
/// systemd ignores ("Invalid …, ignoring") would leave the agent uncapped,
/// since a swarm cap replaces the hive's answer.
#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] #[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct Caps { pub struct Caps {
/// systemd `CPUQuota=`: a percentage above zero, `"100%"` being one core. /// systemd `CPUQuota=`: digits, optionally `.` and one or two digits, then
/// `%`; above zero and at most `21474836.47%`. `"100%"` is one core.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
#[schema(example = "400%")] #[schema(example = "400%")]
cpu_quota: Option<String>, cpu_quota: Option<String>,
/// systemd `MemoryMax=`: a size above zero with an optional `K`/`M`/`G`/`T` /// systemd `MemoryMax=`: `infinity`; a percentage of physical memory
/// suffix, a percentage of physical memory up to `100%`, or `infinity`. /// written as for `cpu_quota`, above zero and at most `100%`; or a size:
/// Stored in the one spelling systemd accepts (`8gb` is stored as `8G`). /// digits, optionally `.` and digits, then nothing, `B` or one of
/// `K`/`M`/`G`/`T` (1024-based, upper case), at least one byte and below
/// 2^64 - 1 bytes.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
#[schema(example = "8G")] #[schema(example = "8G")]
memory_max: Option<String>, memory_max: Option<String>,
@ -61,60 +68,97 @@ impl From<AgentCaps> for Caps {
} }
} }
// The accepted shapes are hive-c0re's `resource_limits::validate_cpu_quota` / // The accepted forms are a subset of what systemd parses (checked against
// `validate_memory_max`, the operator's own input path for the same drop-in, // `systemd-analyze verify`, systemd 260) and stricter than hive-c0re's own
// with zero and an over-100% memory share refused as well. // `validate_cpu_quota` / `validate_memory_max`, so the hive never refuses a
// stored value.
/// systemd keeps `CPUQuota=` in hundredths of a percent in a C `int`.
const CPU_QUOTA_MAX_HUNDREDTHS: u64 = 2_147_483_647;
fn cpu_quota(value: &str) -> Result<String, String> { fn cpu_quota(value: &str) -> Result<String, String> {
match value.strip_suffix('%') { match hundredths(value) {
Some(n) if is_positive_number(n) => Ok(value.to_owned()), Some(h) if (1..=CPU_QUOTA_MAX_HUNDREDTHS).contains(&h) => Ok(value.to_owned()),
_ => Err(format!( _ => Err(format!(
"invalid cpu_quota {value:?}: expected a percentage above zero such as \"200%\" \ "invalid cpu_quota {value:?}: expected a percentage above zero with at most two \
(100% = one full core)" decimals, such as \"200%\" (100% = one full core), up to \"21474836.47%\""
)), )),
} }
} }
fn memory_max(value: &str) -> Result<String, String> { fn memory_max(value: &str) -> Result<String, String> {
if value == "infinity" { let applies = value == "infinity"
return Ok(value.to_owned()); || if value.ends_with('%') {
hundredths(value).is_some_and(|h| (1..=10_000).contains(&h))
} else {
size_bytes(value).is_some_and(|b| b >= 1 && b < u128::from(u64::MAX))
};
if applies {
Ok(value.to_owned())
} else {
Err(format!(
"invalid memory_max {value:?}: expected a size above zero such as \"8G\" \
(suffix B, K, M, G or T, upper case), a percentage up to \"100%\" with at \
most two decimals, or \"infinity\""
))
} }
let normalized = match value.strip_suffix('%') {
Some(n) => (is_positive_number(n) && n.parse::<f64>().is_ok_and(|p| p <= 100.0))
.then(|| value.to_owned()),
None => byte_size(value),
};
normalized.ok_or_else(|| {
format!(
"invalid memory_max {value:?}: expected a size above zero such as \"8G\", a \
percentage up to \"100%\", or \"infinity\""
)
})
} }
/// A byte count with an optional `K`/`M`/`G`/`T` multiplier, in any case and /// `value` as systemd reads a percentage, in hundredths of a percent: digits,
/// with an optional trailing `B`, in the form systemd takes: `8gb` → `8G`, a /// optionally `.` and one or two digits, then `%`. `None` for anything else.
/// bare `1048576B` kept as it is. fn hundredths(value: &str) -> Option<u64> {
fn byte_size(value: &str) -> Option<String> { let number = value.strip_suffix('%')?;
const UNITS: [char; 8] = ['K', 'M', 'G', 'T', 'k', 'm', 'g', 't']; let (int, frac) = match number.split_once('.') {
let (body, b) = match value.strip_suffix(['B', 'b']) { Some((int, frac)) if (1..=2).contains(&frac.len()) && digits(frac) => (int, frac),
Some(body) => (body, true), Some(_) => return None,
None => (value, false), None => (number, ""),
}; };
let (mantissa, unit) = match body.strip_suffix(UNITS) { if !digits(int) {
Some(mantissa) => (mantissa, body[mantissa.len()..].to_ascii_uppercase()), return None;
None if b => (body, "B".to_owned()), }
None => (body, String::new()), let frac = match frac.len() {
0 => 0,
1 => frac.parse::<u64>().ok()? * 10,
_ => frac.parse::<u64>().ok()?,
}; };
is_positive_number(mantissa).then(|| format!("{mantissa}{unit}")) int.parse::<u64>().ok()?.checked_mul(100)?.checked_add(frac)
} }
/// Digits with at most one `.` between digits, and not zero. No sign or /// `value` as one term of systemd's `parse_size` reads it, in bytes: digits,
/// exponent, which systemd does not take either. /// optionally `.` and digits, then nothing, `B`, or `K`/`M`/`G`/`T` (1024-based,
fn is_positive_number(value: &str) -> bool { /// case-sensitive). `P` and `E`, which systemd also reads, are left out
let (int, frac) = value.split_once('.').unwrap_or((value, "1")); /// because hive-c0re's validator refuses them. `None` for anything else, and for what
let digits = |s: &str| !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()); /// `parse_size` refuses as out of range: a whole part or fraction digits past
digits(int) && digits(frac) && value.bytes().any(|b| (b'1'..=b'9').contains(&b)) /// `u64`, or a whole part that, plus one when there is a fraction, is past
/// `u64` once scaled.
fn size_bytes(value: &str) -> Option<u128> {
let (number, exponent) = match value.char_indices().last()? {
(i, 'B') => (&value[..i], 0),
(i, c) => match "KMGT".find(c) {
Some(k) => (&value[..i], k + 1),
None => (value, 0),
},
};
let scale = 1024u128.pow(u32::try_from(exponent).ok()?);
let (int, frac) = number.split_once('.').unwrap_or((number, ""));
if !digits(int) || (number.contains('.') && !digits(frac)) {
return None;
}
let whole = u128::from(int.parse::<u64>().ok()?);
let frac_value = if frac.is_empty() {
0
} else {
u128::from(frac.parse::<u64>().ok()?)
};
if whole + u128::from(frac_value > 0) > u128::from(u64::MAX) / scale {
return None;
}
let frac_bytes = frac_value * scale / 10u128.checked_pow(u32::try_from(frac.len()).ok()?)?;
Some(whole * scale + frac_bytes)
}
fn digits(s: &str) -> bool {
!s.is_empty() && s.bytes().all(|b| b.is_ascii_digit())
} }
fn agent_name(name: &str) -> Result<String, (StatusCode, String)> { fn agent_name(name: &str) -> Result<String, (StatusCode, String)> {
@ -128,7 +172,7 @@ fn queue_problem(e: &anyhow::Error) -> ProblemDetails {
error_problem(wanted_error_status(e), &format!("{e:#}")) error_problem(wanted_error_status(e), &format!("{e:#}"))
} }
async fn open( async fn open_or_create(
client: &async_nats::Client, client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, ProblemDetails> { ) -> Result<async_nats::jetstream::kv::Store, ProblemDetails> {
agent_limits::open_or_create(client) agent_limits::open_or_create(client)
@ -157,7 +201,12 @@ pub(crate) async fn get_agent_limits(
) -> Result<Json<Caps>, ProblemDetails> { ) -> Result<Json<Caps>, ProblemDetails> {
let agent = agent_name(&name).map_err(refused)?; let agent = agent_name(&name).map_err(refused)?;
let client = queue(&state).map_err(refused)?; let client = queue(&state).map_err(refused)?;
let store = open(&client).await?; let Some(store) = agent_limits::open(&client)
.await
.map_err(|e| queue_problem(&anyhow::Error::new(e)))?
else {
return Ok(Json(Caps::default()));
};
let (current, _) = read::<AgentCaps>(&store, &agent) let (current, _) = read::<AgentCaps>(&store, &agent)
.await .await
.map_err(|e| queue_problem(&e))?; .map_err(|e| queue_problem(&e))?;
@ -202,7 +251,7 @@ pub(crate) async fn put_agent_limits(
&format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"), &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"),
)); ));
} }
let store = open(&client).await?; let store = open_or_create(&client).await?;
for _ in 0..MAX_ATTEMPTS { for _ in 0..MAX_ATTEMPTS {
let (current, revision) = read::<AgentCaps>(&store, &agent) let (current, revision) = read::<AgentCaps>(&store, &agent)
.await .await
@ -241,7 +290,7 @@ pub(crate) async fn delete_agent_limits(
) -> Result<StatusCode, ProblemDetails> { ) -> Result<StatusCode, ProblemDetails> {
let agent = agent_name(&name).map_err(refused)?; let agent = agent_name(&name).map_err(refused)?;
let client = queue(&state).map_err(refused)?; let client = queue(&state).map_err(refused)?;
let store = open(&client).await?; let store = open_or_create(&client).await?;
store store
.delete(&agent) .delete(&agent)
.await .await
@ -262,35 +311,110 @@ mod tests {
use std::sync::Arc; use std::sync::Arc;
use swarm_queue_client::wanted::AgentState; use swarm_queue_client::wanted::AgentState;
/// Of the `bad` values, systemd parses `""` (no quota) and `+5%`, and the
/// last adds a directive on a second line. It ignores every other one
/// ("Invalid CPU quota …, ignoring", checked with `systemd-analyze
/// verify`), which would leave the agent uncapped.
#[test] #[test]
fn cpu_quota_takes_a_percentage_above_zero() { fn cpu_quota_takes_a_percentage_above_zero() {
for ok in ["400%", "100%", "0.5%", "1500%"] { for ok in [
"400%",
"100%",
"0.5%",
"0.01%",
"1500%",
"400.12%",
"05%",
"21474836.47%",
] {
assert_eq!(cpu_quota(ok).as_deref(), Ok(ok)); assert_eq!(cpu_quota(ok).as_deref(), Ok(ok));
} }
for bad in [ for bad in [
"0%", "0.0%", "-50%", "400", "4 00%", "%", "lots", "", "1e3%", "+5%", "0%",
"0.0%",
"0.00%",
"-50%",
"400",
"4 00%",
"%",
"lots",
"",
"1e3%",
"+5%",
"400.123%",
"0.001%",
"99999999999999999999%",
"21474836.48%",
"1.%",
".5%",
"400%\nExecStartPre=/bin/sh -c id #%",
] { ] {
assert!(cpu_quota(bad).is_err(), "{bad:?} was accepted"); assert!(cpu_quota(bad).is_err(), "{bad:?} was accepted");
} }
} }
/// As above for `MemoryMax=`. Of the `bad` values, systemd also parses `3P`
/// and `2E`, which hive-c0re's validator refuses, and `8 G` and `1.G`.
#[test] #[test]
fn memory_max_takes_a_size_a_share_or_infinity_and_stores_the_systemd_spelling() { fn memory_max_takes_a_size_a_share_or_infinity_that_systemd_applies() {
for (given, stored) in [ for ok in [
("8G", "8G"), "8G",
("8gb", "8G"), "512M",
("512m", "512M"), "4096K",
("1048576", "1048576"), "1048576",
("1048576B", "1048576B"), "1048576B",
("1.5T", "1.5T"), "1.5B",
("50%", "50%"), "0.5K",
("100%", "100%"), "1.5T",
("infinity", "infinity"), "1.23456789G",
"16383T",
"16777215T",
"18446744073709551614",
"18446744073709551614.5",
"1.00000000000000000001G",
"50%",
"12.34%",
"0.01%",
"100%",
"100.00%",
"infinity",
] { ] {
assert_eq!(memory_max(given).as_deref(), Ok(stored), "{given:?}"); assert_eq!(memory_max(ok).as_deref(), Ok(ok), "{ok:?}");
} }
for bad in [ for bad in [
"0", "0G", "0.0G", "0%", "-8G", "150%", "8X", "G", "8 G", "", "eight", "8GG", "0",
"0G",
"0.0G",
"0.5",
"0%",
"-8G",
"150%",
"100.01%",
"12.345%",
"8X",
"G",
"8 G",
"",
"eight",
"8GG",
"8gb",
"8g",
"8GB",
"4096k",
"1048576b",
"3P",
"2E",
".5G",
"1.G",
"Infinity",
"17000000T",
"16777216T",
"16777215.5T",
"99999999999999999999T",
"18446744073709551615",
"18446744073709551616",
"1.123456789012345678901234567890G",
"8G\nDevicePolicy=auto",
] { ] {
assert!(memory_max(bad).is_err(), "{bad:?} was accepted"); assert!(memory_max(bad).is_err(), "{bad:?} was accepted");
} }
@ -389,7 +513,7 @@ mod tests {
let client = nats.client().await; let client = nats.client().await;
let state = declared(&client).await; let state = declared(&client).await;
let answer = put(&state, "atlas", caps(Some("400%"), Some("8gb"))) let answer = put(&state, "atlas", caps(Some("400%"), Some("8G")))
.await .await
.expect("caps are set"); .expect("caps are set");
assert_eq!(answer.0, caps(Some("400%"), Some("8G"))); assert_eq!(answer.0, caps(Some("400%"), Some("8G")));
@ -407,6 +531,23 @@ mod tests {
assert_eq!(get(&state, "atlas").await, Caps::default()); assert_eq!(get(&state, "atlas").await, Caps::default());
} }
/// A read before any cap was set answers `{}` and leaves the bucket
/// uncreated.
#[tokio::test]
async fn a_read_before_any_cap_answers_empty_and_creates_no_bucket() {
let nats = TestNats::start();
let client = nats.client().await;
let state = state(&client);
assert_eq!(get(&state, "atlas").await, Caps::default());
assert!(
swarm_queue_client::agent_limits::open(&client)
.await
.expect("the queue answers")
.is_none()
);
}
/// A name no hive declares gets no key, so a typo cannot cap an agent /// A name no hive declares gets no key, so a typo cannot cap an agent
/// that does not exist yet. /// that does not exist yet.
#[tokio::test] #[tokio::test]
@ -500,6 +641,27 @@ mod tests {
assert_eq!(published(&client, "atlas").await.as_deref(), Some("{}")); assert_eq!(published(&client, "atlas").await.as_deref(), Some("{}"));
} }
/// A value carrying a directive on a second line is a 400 through the real
/// router, and the caps already set stay.
#[tokio::test]
async fn a_cap_with_a_second_line_is_a_400_and_the_caps_stay() {
let nats = TestNats::start();
let client = nats.client().await;
let app = capped_router(&client).await;
for body in [
r#"{"cpu_quota":"400%\nExecStartPre=/bin/sh -c id #%"}"#,
r#"{"memory_max":"8G\nDevicePolicy=auto"}"#,
] {
assert_eq!(put_raw(&app, body).await, StatusCode::BAD_REQUEST, "{body}");
assert_eq!(
published(&client, "atlas").await.as_deref(),
Some(CAPPED),
"{body}"
);
}
}
/// A field the body does not define is a 422 and writes nothing: a /// A field the body does not define is a 422 and writes nothing: a
/// misspelt `memory_max` must not read as "no memory cap". /// misspelt `memory_max` must not read as "no memory cap".
#[tokio::test] #[tokio::test]

View file

@ -71,6 +71,30 @@ pub async fn open_or_create(
.await .await
} }
/// Open the caps bucket for reading, or `None` when nothing has created it
/// yet, which reads the same as no caps.
///
/// **Controller-side**, for reads that must not create the bucket.
///
/// # Errors
///
/// [`crate::Error::OpenBucket`] for every failure other than the bucket not
/// existing.
#[cfg(feature = "kv")]
pub async fn open(
client: &async_nats::Client,
) -> Result<Option<async_nats::jetstream::kv::Store>, crate::Error> {
let js = async_nats::jetstream::new(client.clone());
match js.get_key_value(BUCKET).await {
Ok(store) => Ok(Some(store)),
Err(e) if crate::wanted::is_bucket_not_found(&e) => Ok(None),
Err(source) => Err(crate::Error::OpenBucket {
bucket: BUCKET.to_owned(),
source,
}),
}
}
/// What reading one agent's key found. /// What reading one agent's key found.
#[cfg(feature = "agent-limits")] #[cfg(feature = "agent-limits")]
#[derive(Debug)] #[derive(Debug)]

View file

@ -156,7 +156,7 @@ pub async fn open_read_only(
/// `get_key_value` reports a missing bucket as `GetBucket` wrapping the /// `get_key_value` reports a missing bucket as `GetBucket` wrapping the
/// stream lookup's `STREAM_NOT_FOUND`; every other kind is a real failure. /// stream lookup's `STREAM_NOT_FOUND`; every other kind is a real failure.
#[cfg(feature = "kv")] #[cfg(feature = "kv")]
fn is_bucket_not_found(e: &async_nats::jetstream::context::KeyValueError) -> bool { pub(crate) fn is_bucket_not_found(e: &async_nats::jetstream::context::KeyValueError) -> bool {
use async_nats::jetstream::context::{GetStreamError, GetStreamErrorKind, KeyValueErrorKind}; use async_nats::jetstream::context::{GetStreamError, GetStreamErrorKind, KeyValueErrorKind};
e.kind() == KeyValueErrorKind::GetBucket e.kind() == KeyValueErrorKind::GetBucket