diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index d6cf23ed..5118956f 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -491,9 +491,13 @@ caps. swarm-controller alone writes it: | `PUT` | `/api/agents//limits` | `{"cpu_quota": "400%", "memory_max": "8G"}`, every field optional, replacing the stored value | | `DELETE` | `/api/agents//limits` | —; answers 204 | -`PUT` refuses an agent no hive's wanted state declares, unknown fields, zero -and anything systemd doesn't take, and stores memory the way systemd spells -it (`8gb` as `8G`). +`PUT` refuses an agent no hive's wanted state declares, unknown fields, and +any value systemd would ignore, which would leave the agent uncapped. A +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 agent it has and stores the result in `meta/swarm-resource-limits.json`. Each diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index 478a92db..3d33a900 100644 --- a/swarm-controller/src/agent_limits.rs +++ b/swarm-controller/src/agent_limits.rs @@ -7,9 +7,9 @@ //! hive's own answer for that field, and a field left unset stays with the //! hive. //! -//! The hive hands these values to systemd verbatim, and a value systemd -//! rejects stops the agent's container from starting, so they are validated -//! here as well as on the hive. +//! The hive hands these values to systemd verbatim, and systemd ignores a value +//! it can't parse, leaving the agent uncapped, so they are checked here against +//! systemd's own syntax (`Caps`) as well as on the hive. use axum::Json; use axum::extract::{Path, State}; @@ -27,16 +27,23 @@ use crate::{AppState, error_problem, wanted_error_status}; const MAX_ATTEMPTS: usize = 5; /// 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)] #[serde(deny_unknown_fields)] 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")] #[schema(example = "400%")] cpu_quota: Option, - /// systemd `MemoryMax=`: a size above zero with an optional `K`/`M`/`G`/`T` - /// suffix, a percentage of physical memory up to `100%`, or `infinity`. - /// Stored in the one spelling systemd accepts (`8gb` is stored as `8G`). + /// systemd `MemoryMax=`: `infinity`; a percentage of physical memory + /// written as for `cpu_quota`, above zero and at most `100%`; or a size: + /// 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")] #[schema(example = "8G")] memory_max: Option, @@ -61,60 +68,97 @@ impl From for Caps { } } -// The accepted shapes are hive-c0re's `resource_limits::validate_cpu_quota` / -// `validate_memory_max`, the operator's own input path for the same drop-in, -// with zero and an over-100% memory share refused as well. +// The accepted forms are a subset of what systemd parses (checked against +// `systemd-analyze verify`, systemd 260) and stricter than hive-c0re's own +// `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 { - match value.strip_suffix('%') { - Some(n) if is_positive_number(n) => Ok(value.to_owned()), + match hundredths(value) { + Some(h) if (1..=CPU_QUOTA_MAX_HUNDREDTHS).contains(&h) => Ok(value.to_owned()), _ => Err(format!( - "invalid cpu_quota {value:?}: expected a percentage above zero such as \"200%\" \ - (100% = one full core)" + "invalid cpu_quota {value:?}: expected a percentage above zero with at most two \ + decimals, such as \"200%\" (100% = one full core), up to \"21474836.47%\"" )), } } fn memory_max(value: &str) -> Result { - if value == "infinity" { - return Ok(value.to_owned()); + let applies = value == "infinity" + || 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::().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 -/// with an optional trailing `B`, in the form systemd takes: `8gb` → `8G`, a -/// bare `1048576B` kept as it is. -fn byte_size(value: &str) -> Option { - const UNITS: [char; 8] = ['K', 'M', 'G', 'T', 'k', 'm', 'g', 't']; - let (body, b) = match value.strip_suffix(['B', 'b']) { - Some(body) => (body, true), - None => (value, false), +/// `value` as systemd reads a percentage, in hundredths of a percent: digits, +/// optionally `.` and one or two digits, then `%`. `None` for anything else. +fn hundredths(value: &str) -> Option { + let number = value.strip_suffix('%')?; + let (int, frac) = match number.split_once('.') { + Some((int, frac)) if (1..=2).contains(&frac.len()) && digits(frac) => (int, frac), + Some(_) => return None, + None => (number, ""), }; - let (mantissa, unit) = match body.strip_suffix(UNITS) { - Some(mantissa) => (mantissa, body[mantissa.len()..].to_ascii_uppercase()), - None if b => (body, "B".to_owned()), - None => (body, String::new()), + if !digits(int) { + return None; + } + let frac = match frac.len() { + 0 => 0, + 1 => frac.parse::().ok()? * 10, + _ => frac.parse::().ok()?, }; - is_positive_number(mantissa).then(|| format!("{mantissa}{unit}")) + int.parse::().ok()?.checked_mul(100)?.checked_add(frac) } -/// Digits with at most one `.` between digits, and not zero. No sign or -/// exponent, which systemd does not take either. -fn is_positive_number(value: &str) -> bool { - let (int, frac) = value.split_once('.').unwrap_or((value, "1")); - let digits = |s: &str| !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()); - digits(int) && digits(frac) && value.bytes().any(|b| (b'1'..=b'9').contains(&b)) +/// `value` as one term of systemd's `parse_size` reads it, in bytes: digits, +/// optionally `.` and digits, then nothing, `B`, or `K`/`M`/`G`/`T` (1024-based, +/// case-sensitive). `P` and `E`, which systemd also reads, are left out +/// because hive-c0re's validator refuses them. `None` for anything else, and for what +/// `parse_size` refuses as out of range: a whole part or fraction digits past +/// `u64`, or a whole part that, plus one when there is a fraction, is past +/// `u64` once scaled. +fn size_bytes(value: &str) -> Option { + 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::().ok()?); + let frac_value = if frac.is_empty() { + 0 + } else { + u128::from(frac.parse::().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 { @@ -128,7 +172,7 @@ fn queue_problem(e: &anyhow::Error) -> ProblemDetails { error_problem(wanted_error_status(e), &format!("{e:#}")) } -async fn open( +async fn open_or_create( client: &async_nats::Client, ) -> Result { agent_limits::open_or_create(client) @@ -157,7 +201,12 @@ pub(crate) async fn get_agent_limits( ) -> Result, ProblemDetails> { let agent = agent_name(&name).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::(&store, &agent) .await .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"), )); } - let store = open(&client).await?; + let store = open_or_create(&client).await?; for _ in 0..MAX_ATTEMPTS { let (current, revision) = read::(&store, &agent) .await @@ -241,7 +290,7 @@ pub(crate) async fn delete_agent_limits( ) -> Result { let agent = agent_name(&name).map_err(refused)?; let client = queue(&state).map_err(refused)?; - let store = open(&client).await?; + let store = open_or_create(&client).await?; store .delete(&agent) .await @@ -262,35 +311,110 @@ mod tests { use std::sync::Arc; 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] 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)); } 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"); } } + /// 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] - fn memory_max_takes_a_size_a_share_or_infinity_and_stores_the_systemd_spelling() { - for (given, stored) in [ - ("8G", "8G"), - ("8gb", "8G"), - ("512m", "512M"), - ("1048576", "1048576"), - ("1048576B", "1048576B"), - ("1.5T", "1.5T"), - ("50%", "50%"), - ("100%", "100%"), - ("infinity", "infinity"), + fn memory_max_takes_a_size_a_share_or_infinity_that_systemd_applies() { + for ok in [ + "8G", + "512M", + "4096K", + "1048576", + "1048576B", + "1.5B", + "0.5K", + "1.5T", + "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 [ - "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"); } @@ -389,7 +513,7 @@ mod tests { let client = nats.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 .expect("caps are set"); assert_eq!(answer.0, caps(Some("400%"), Some("8G"))); @@ -407,6 +531,23 @@ mod tests { 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 /// that does not exist yet. #[tokio::test] @@ -500,6 +641,27 @@ mod tests { 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 /// misspelt `memory_max` must not read as "no memory cap". #[tokio::test] diff --git a/swarm-queue-client/src/agent_limits.rs b/swarm-queue-client/src/agent_limits.rs index 91b857af..77cf3b7f 100644 --- a/swarm-queue-client/src/agent_limits.rs +++ b/swarm-queue-client/src/agent_limits.rs @@ -71,6 +71,30 @@ pub async fn open_or_create( .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, 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. #[cfg(feature = "agent-limits")] #[derive(Debug)] diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index 451039c5..16681441 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/swarm-queue-client/src/wanted.rs @@ -156,7 +156,7 @@ pub async fn open_read_only( /// `get_key_value` reports a missing bucket as `GetBucket` wrapping the /// stream lookup's `STREAM_NOT_FOUND`; every other kind is a real failure. #[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}; e.kind() == KeyValueErrorKind::GetBucket