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

Compare commits

base: vinzenz:3464a912bfd224207ff4cb9430ab7f4c776af9b6
Branches Tags
vinzenz:main
..
compare: vinzenz:daaf920e06a0a3c9c326b4677c6bb0b2234a00aa
Branches Tags
vinzenz:main
6 changed files with 43 additions and 325 deletions
Download patch file Download diff file Expand all files Collapse all files

3
docs/scheduler/coordinator.md
View file

@ -491,9 +491,6 @@ caps. swarm-controller alone writes it:
| `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 |
A stored value the controller can't decode, such as one a newer build wrote,
answers `{}`, with a warning in its log, until the next `PUT` replaces it.
`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, no leading zero in its whole part (it

4
docs/swarm/inference.md
View file

@ -88,9 +88,7 @@ which swarm-controller alone writes:
Every field is optional. The agent runs the same with `{}` as with no key;
`GET` answers `"configured": null` for no key (or a cleared one) and `{}` for
a stored value with no fields. A stored value the controller can't decode,
such as one a newer build wrote, also answers `"configured": null`, with a
warning in its log, until the next `PUT` replaces it.
a stored value with no fields.
| method | path | body |
| -------- | ------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |

10
docs/swarm/schedules.md
View file

@ -93,16 +93,10 @@ involved.
`schedule-definitions` holds every prompt as one JSON document under the key
`all`. Agents hold no grant on this bucket. Each change re-reads the key and
writes against its revision, so a prompt exists for every one of its agents or for
none of them, and two changes at once don't lose one. The controller leaves a
prompt it can't decode, such as one a newer build wrote, out of the list, with
a warning in its log. An agent's key that already held the prompt keeps it as
it was, so it goes on firing, and no other key gets it. The controller keeps
the prompt in the document as it was, after the others, and won't reuse its
id. A document that isn't an object with a list of `schedules` is an error.
none of them, and two changes at once don't lose one.
`agent-schedules` holds one key per agent: the prompts naming that agent,
without their names or other agents. A key the controller can't decode counts
as no prompts, with a warning in its log, and the next rewrite replaces it.
without their names or other agents.
```json
{

38
swarm-controller/src/agent_limits.rs
View file

@ -19,7 +19,7 @@ use serde::{Deserialize, Serialize};
use swarm_queue_client::agent_limits::{self, AgentCaps};
use utoipa::ToSchema;
use crate::schedules::{live, queue, read_or_default, refused, write};
use crate::schedules::{live, queue, read, refused, write};
use crate::{AppState, error_problem, wanted_error_status};
/// How many times a losing writer re-reads and re-applies, as in
@ -209,7 +209,7 @@ pub(crate) async fn get_agent_limits(
else {
return Ok(Json(Caps::default()));
};
let (current, _, _) = read_or_default::<AgentCaps>(&store, &agent)
let (current, _) = read::<AgentCaps>(&store, &agent)
.await
.map_err(|e| queue_problem(&e))?;
Ok(Json(current.into()))
@ -255,10 +255,10 @@ pub(crate) async fn put_agent_limits(
}
let store = open_or_create(&client).await?;
for _ in 0..MAX_ATTEMPTS {
let (current, revision, undecodable) = read_or_default::<AgentCaps>(&store, &agent)
let (current, revision) = read::<AgentCaps>(&store, &agent)
.await
.map_err(|e| queue_problem(&e))?;
if (current == value && !undecodable)
if current == value
|| write(&store, &agent, &value, revision)
.await
.map_err(|e| queue_problem(&e))?
@ -302,7 +302,7 @@ pub(crate) async fn delete_agent_limits(
}
#[cfg(test)]
pub(crate) mod tests {
mod tests {
use super::{
Caps, cpu_quota, delete_agent_limits, get_agent_limits, memory_max, put_agent_limits,
};
@ -464,7 +464,7 @@ pub(crate) mod tests {
}
/// A state on a real server whose `pr1ma` declares `atlas`.
pub(crate) async fn declared(client: &async_nats::Client) -> crate::AppState {
async fn declared(client: &async_nats::Client) -> crate::AppState {
WantedWriter::new(client.clone())
.set("pr1ma", "atlas", AgentState::Up)
.await
@ -698,30 +698,4 @@ pub(crate) mod tests {
);
}
}
/// A stored value this build cannot decode, such as one a newer build
/// wrote, reads as no caps, and a `PUT`, even of no caps, replaces it.
#[tokio::test]
async fn an_undecodable_value_reads_as_no_caps_until_a_put() {
let nats = TestNats::start();
let client = nats.client().await;
let state = declared(&client).await;
let store = swarm_queue_client::agent_limits::open_or_create(&client)
.await
.expect("bucket");
store
.put(
"atlas",
br#"{"cpu_quota":{"later":"400%"}}"#.to_vec().into(),
)
.await
.expect("written");
assert_eq!(get(&state, "atlas").await, Caps::default());
let answer = put(&state, "atlas", Caps::default())
.await
.expect("caps are set");
assert_eq!(answer.0, Caps::default());
assert_eq!(published(&client, "atlas").await.as_deref(), Some("{}"));
}
}

44
swarm-controller/src/inference.rs
View file

@ -25,7 +25,7 @@ use serde::{Deserialize, Serialize};
use swarm_queue_client::agent_inference::{self, AgentInference, Effective, Effort, Source};
use utoipa::ToSchema;
use crate::schedules::{live, queue, read_or_default, refused, write};
use crate::schedules::{live, queue, read, refused, write};
use crate::{AppState, error_problem, wanted_error_status};
/// How many times a losing writer re-reads and re-applies, as in
@ -242,9 +242,9 @@ pub(crate) async fn get_agent_inference(
let agent = agent_name(&name).map_err(refused)?;
let client = queue(&state).map_err(refused)?;
let store = open(&client).await?;
// `Option`'s default is the `None` an absent, deleted or undecodable key
// reads as; a stored `{}` decodes to `Some`.
let (current, _, _) = read_or_default::<Option<AgentInference>>(&store, &agent)
// `Option`'s default is the `None` an absent or deleted key reads as; a
// stored `{}` decodes to `Some`.
let (current, _) = read::<Option<AgentInference>>(&store, &agent)
.await
.map_err(|e| queue_problem(&e))?;
Ok(Json(AgentInferenceView {
@ -296,10 +296,10 @@ pub(crate) async fn put_agent_inference(
}
let store = open(&client).await?;
for _ in 0..MAX_ATTEMPTS {
let (current, revision, undecodable) = read_or_default::<AgentInference>(&store, &agent)
let (current, revision) = read::<AgentInference>(&store, &agent)
.await
.map_err(|e| queue_problem(&e))?;
if (current == value && !undecodable)
if current == value
|| write(&store, &agent, &value, revision)
.await
.map_err(|e| queue_problem(&e))?
@ -506,36 +506,4 @@ mod tests {
Some(AgentInference::default())
);
}
/// A stored value this build cannot decode, such as an effort a newer
/// build added, reads as no settings, and a `PUT`, even of no fields,
/// replaces it.
#[tokio::test]
async fn an_undecodable_value_reads_as_unset_until_a_put() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
let state = crate::agent_limits::tests::declared(&client).await;
let store = agent_inference::open_or_create(&client)
.await
.expect("bucket");
store
.put("atlas", br#"{"effort":"later"}"#.to_vec().into())
.await
.expect("written");
let view = get_agent_inference(State(state.clone()), Path("atlas".to_owned()))
.await
.expect("read");
assert_eq!(view.0.configured, None);
let stored = put_agent_inference(
State(state),
Path("atlas".to_owned()),
Json(InferenceSettings::default()),
)
.await
.expect("set");
assert_eq!(stored.0, InferenceSettings::default());
let entry = store.get("atlas").await.expect("read").expect("a value");
assert_eq!(&entry[..], b"{}");
}
}

269
swarm-controller/src/schedules.rs
View file

@ -52,82 +52,10 @@ const ID_LEN: usize = 8;
const RECONCILE_INTERVAL: Duration = Duration::from_mins(1);
/// Every schedule in the swarm: the value under [`DEFINITIONS_KEY`].
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
struct Definitions {
#[serde(default)]
schedules: Vec<Definition>,
/// Stored definitions this build cannot decode, such as ones a newer
/// build wrote. Not listed, edited or projected, and written back as they
/// were read, after the others, so a newer build reads them again after a
/// roll-forward.
unreadable: Vec<serde_json::Value>,
}
impl Serialize for Definitions {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
#[derive(Serialize)]
#[serde(untagged)]
enum Entry<'a> {
Readable(&'a Definition),
Unreadable(&'a serde_json::Value),
}
#[derive(Serialize)]
struct Stored<'a> {
schedules: Vec<Entry<'a>>,
}
let readable = self.schedules.iter().map(Entry::Readable);
let unreadable = self.unreadable.iter().map(Entry::Unreadable);
Stored {
schedules: readable.chain(unreadable).collect(),
}
.serialize(serializer)
}
}
impl Definitions {
/// `bytes`, the stored document, with a warning when any definition in
/// it does not decode. A document that is not an object with a list of
/// `schedules` is an error: reading it as no schedules would empty every
/// agent's key.
fn decode(bytes: &[u8]) -> anyhow::Result<Self> {
#[derive(Deserialize)]
struct Stored {
#[serde(default)]
schedules: Vec<serde_json::Value>,
}
let stored: Stored = serde_json::from_slice(bytes)
.with_context(|| format!("{DEFINITIONS_KEY} is not decodable"))?;
let mut all = Self::default();
let mut first_error = None;
for value in stored.schedules {
match Definition::deserialize(&value) {
Ok(d) => all.schedules.push(d),
Err(e) => {
first_error.get_or_insert(e);
all.unreadable.push(value);
}
}
}
if let Some(error) = first_error {
tracing::warn!(
bucket = DEFINITIONS_BUCKET, key = DEFINITIONS_KEY,
unreadable = all.unreadable.len(), %error,
"schedules this build cannot decode are left out and kept"
);
}
Ok(all)
}
/// Whether a schedule, readable or not, has `id`.
fn taken(&self, id: &str) -> bool {
self.schedules.iter().any(|d| d.id == id) || self.unreadable_ids().any(|i| i == id)
}
/// The ids of the [`Self::unreadable`] definitions that have one.
fn unreadable_ids(&self) -> impl Iterator<Item = &str> {
self.unreadable
.iter()
.filter_map(|v| v.get("id").and_then(serde_json::Value::as_str))
}
}
/// One schedule as the operator defines it.
@ -170,23 +98,15 @@ impl Definition {
}
}
/// `agent`'s key: the schedules that name it, in definition order, then the
/// entries of `current`, its key as stored, whose definition this build
/// cannot decode. Those stay as they were, so they keep firing, and are never
/// added to a key that did not hold them.
fn projection(definitions: &Definitions, agent: &str, current: &AgentSchedules) -> AgentSchedules {
let readable = definitions
.schedules
.iter()
.filter(|d| d.agents.contains(agent))
.map(Definition::for_agent);
let kept = current
.schedules
.iter()
.filter(|s| definitions.unreadable_ids().any(|id| id == s.id))
.cloned();
/// `agent`'s key: the schedules that name it, in definition order.
fn projection(definitions: &Definitions, agent: &str) -> AgentSchedules {
AgentSchedules {
schedules: readable.chain(kept).collect(),
schedules: definitions
.schedules
.iter()
.filter(|d| d.agents.contains(agent))
.map(Definition::for_agent)
.collect(),
}
}
@ -425,7 +345,7 @@ fn mint_id(taken: &Definitions) -> anyhow::Result<String> {
getrandom::fill(&mut bytes)
.map_err(|e| anyhow::anyhow!("drawing a schedule id from the kernel's CSPRNG: {e}"))?;
let id: String = bytes.iter().map(|b| char::from(b'a' + b % 26)).collect();
if !taken.taken(&id) {
if !taken.schedules.iter().any(|d| d.id == id) {
return Ok(id);
}
}
@ -483,36 +403,18 @@ async fn open_or_create_definitions(client: &async_nats::Client) -> anyhow::Resu
}
/// The value of `key` decoded, with the revision to write it back against.
/// An absent or deleted key is the default value. So is a document that does
/// not decode, such as one a newer build wrote: it is logged, and the `bool`
/// is `true`, so a writer replaces it even with the default value.
pub(crate) async fn read_or_default<T: Default + serde::de::DeserializeOwned>(
/// An absent or deleted key is the default value.
pub(crate) async fn read<T: Default + serde::de::DeserializeOwned>(
store: &Store,
key: &str,
) -> anyhow::Result<(T, Option<u64>, bool)> {
let ((value, undecodable), revision) = read_with(store, key, |bytes| {
Ok(match serde_json::from_slice(bytes) {
Ok(value) => (value, false),
Err(e) => {
tracing::warn!(
bucket = %store.name, key, error = %e,
"a stored document is not decodable; reading it as unset until the next write"
);
(T::default(), true)
}
})
) -> anyhow::Result<(T, Option<u64>)> {
read_with(store, key, |bytes| {
serde_json::from_slice(bytes).with_context(|| format!("{key} is not decodable"))
})
.await?;
Ok((value, revision, undecodable))
.await
}
/// [`DEFINITIONS_KEY`] in `store`, with the revision to write it back against.
async fn read_definitions(store: &Store) -> anyhow::Result<(Definitions, Option<u64>)> {
read_with(store, DEFINITIONS_KEY, Definitions::decode).await
}
/// The value of `key` as `decode` reads its bytes, with the revision to write
/// it back against. An absent or deleted key is the default value.
/// As [`read`], with the stored bytes decoded by `decode`.
pub(crate) async fn read_with<T: Default>(
store: &Store,
key: &str,
@ -561,7 +463,7 @@ async fn commit<T>(
mut change: impl FnMut(&mut Definitions) -> Result<T, Failure>,
) -> Result<T, Failure> {
for _ in 0..MAX_ATTEMPTS {
let (mut definitions, revision) = read_definitions(store).await?;
let (mut definitions, revision) = read::<Definitions>(store, DEFINITIONS_KEY).await?;
let out = change(&mut definitions)?;
if write(store, DEFINITIONS_KEY, &definitions, revision).await? {
return Ok(out);
@ -581,11 +483,10 @@ async fn commit<T>(
/// is redone from fresh ones.
async fn project(definitions: &Store, agents: &Store, agent: &str) -> anyhow::Result<()> {
for _ in 0..MAX_ATTEMPTS {
let (current, revision, undecodable) =
read_or_default::<AgentSchedules>(agents, agent).await?;
let (all, _) = read_definitions(definitions).await?;
let wanted = projection(&all, agent, &current);
if current == wanted && !undecodable {
let (current, revision) = read::<AgentSchedules>(agents, agent).await?;
let (all, _) = read::<Definitions>(definitions, DEFINITIONS_KEY).await?;
let wanted = projection(&all, agent);
if current == wanted {
return Ok(());
}
if write(agents, agent, &wanted, revision).await? {
@ -643,7 +544,7 @@ pub(crate) fn spawn_reconcile(
continue;
}
};
let all = match read_definitions(&store).await {
let all = match read::<Definitions>(&store, DEFINITIONS_KEY).await {
Ok((all, _)) => all,
Err(e) => {
tracing::warn!(error = %format!("{e:#}"), "schedules: reconcile could not read the definitions");
@ -711,7 +612,7 @@ pub(crate) async fn list_schedules(
let store = open_or_create_definitions(&client)
.await
.map_err(|e| problem(e.into()))?;
let (mut all, _) = read_definitions(&store)
let (mut all, _) = read::<Definitions>(&store, DEFINITIONS_KEY)
.await
.map_err(|e| problem(e.into()))?;
all.schedules
@ -856,7 +757,6 @@ mod tests {
fn all() -> Definitions {
Definitions {
schedules: vec![standup()],
unreadable: Vec::new(),
}
}
@ -873,17 +773,10 @@ mod tests {
"xyz".clone_into(&mut other.id);
other.agents = ["scribe"].map(str::to_owned).into();
defs.schedules.push(other);
let atlas = projection(&defs, "atlas", &AgentSchedules::default());
let atlas = projection(&defs, "atlas");
assert_eq!(atlas.schedules, vec![standup().for_agent()]);
assert_eq!(
projection(&defs, "scribe", &AgentSchedules::default()).schedules[0].id,
"xyz"
);
assert!(
projection(&defs, "argus", &AgentSchedules::default())
.schedules
.is_empty()
);
assert_eq!(projection(&defs, "scribe").schedules[0].id, "xyz");
assert!(projection(&defs, "argus").schedules.is_empty());
let encoded = serde_json::to_string(&atlas).unwrap();
assert!(!encoded.contains("\"name\""), "{encoded}");
assert!(!encoded.contains("iris"), "{encoded}");
@ -1019,112 +912,6 @@ mod tests {
);
}
/// A definition from a newer build among readable ones: listing, a change
/// and projecting keep working on the others, and the unreadable one is
/// written back as it was and keeps its id. An agent key that does not
/// decode is rewritten, even when the agent has no schedules.
#[tokio::test]
async fn an_undecodable_definition_is_kept_and_the_others_still_work() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
let store = open_or_create_definitions(&client).await.expect("bucket");
let mut newer = serde_json::to_value(standup()).unwrap();
newer["id"] = "new".into();
newer["anchor_unix"] = "a later cadence".into();
let doc = serde_json::json!({"schedules": [newer, standup()]});
store
.put(DEFINITIONS_KEY, doc.to_string().into())
.await
.expect("written");
let (read, _) = read_definitions(&store).await.expect("readable");
assert_eq!(read.schedules, vec![standup()]);
assert!(read.taken("new"));
let mut other = standup();
"xyz".clone_into(&mut other.id);
commit(&store, |all| {
all.schedules.push(other.clone());
Ok(())
})
.await
.unwrap_or_else(|_| panic!("committed"));
let entry = store
.get(DEFINITIONS_KEY)
.await
.expect("read")
.expect("a doc");
let after: serde_json::Value = serde_json::from_slice(&entry).unwrap();
assert_eq!(
after["schedules"],
serde_json::json!([standup(), other, newer]),
"kept untouched"
);
let agents = agent_schedules::open_or_create(&client)
.await
.expect("bucket");
for agent in ["iris", "scribe"] {
agents
.put(agent, br#"{"schedules":"later"}"#.to_vec().into())
.await
.expect("written");
project(&store, &agents, agent).await.expect("projected");
let entry = agents.get(agent).await.expect("read").expect("a key");
let (all, _) = read_definitions(&store).await.expect("readable");
assert_eq!(
serde_json::from_slice::<AgentSchedules>(&entry).expect("decodes"),
projection(&all, agent, &AgentSchedules::default()),
"{agent}"
);
}
}
/// A projected schedule whose definition stops decoding stays in the
/// agent's key through a reconcile, so it keeps firing; one deleted from
/// the definitions leaves it.
#[tokio::test]
async fn a_projected_schedule_that_stops_decoding_keeps_firing() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
let store = open_or_create_definitions(&client).await.expect("bucket");
let agents = agent_schedules::open_or_create(&client)
.await
.expect("bucket");
let mut deleted = standup();
"xyz".clone_into(&mut deleted.id);
let doc = serde_json::json!({"schedules": [standup(), deleted]});
store
.put(DEFINITIONS_KEY, doc.to_string().into())
.await
.expect("written");
let iris = BTreeSet::from(["iris".to_owned()]);
let key = || async {
let entry = agents.get("iris").await.expect("read").expect("a key");
serde_json::from_slice::<AgentSchedules>(&entry).expect("decodes")
};
project_all(&client, &store, &iris).await;
assert_eq!(key().await.schedules.len(), 2);
let mut newer = serde_json::to_value(standup()).unwrap();
newer["anchor_unix"] = "a later cadence".into();
let doc = serde_json::json!({"schedules": [newer]});
store
.put(DEFINITIONS_KEY, doc.to_string().into())
.await
.expect("written");
project_all(&client, &store, &iris).await;
assert_eq!(key().await.schedules, vec![standup().for_agent()]);
}
/// Reading a document with no list of schedules as none would empty
/// every agent's key.
#[test]
fn a_document_with_no_list_of_schedules_is_an_error() {
assert!(Definitions::decode(br#"{"schedules":{}}"#).is_err());
assert!(Definitions::decode(br#""later""#).is_err());
assert_eq!(Definitions::decode(b"{}").unwrap(), Definitions::default());
}
/// Letters only: hive-c0re keys its todos `schedule:<number>`, and a
/// swarm id that could be all digits could collapse into one of them.
#[test]