feat(swarmctl): add agent create, queueing the swarm-controller creation DAG
`swarmctl agent create <name> --hive <hive>` POSTs `/api/agents` to swarm-controller over the daemon's unix socket and prints the queued job's node id. It deliberately does not wait. The endpoint queues a DAG whose last node *publishes* a deploy message; the hive's `hive-c0re` then converges on its own clock, out of the controller's sight. So even a fully settled graph would not mean the agent is up, and there is nothing this CLI could wait for that would let it claim otherwise. Printing the id is exactly what the response says and all of what it says. Transport is a bare hyper HTTP/1.1 client handshaked onto a tokio `UnixStream` via `hyper_util::rt::TokioIo` — the same crate family `hivectl/src/watch.rs` and `hive-agent/src/web_ui/proxy.rs` already use, all of it already workspace-pinned. The request/response shapes are a local mirror rather than a shared crate: the controller's own types are private to its binary and this crate does not link it, the same separation `hivectl` keeps from `hive-c0re`. Errors are reduced to one actionable line — the controller answers RFC 9457 problem+json, so an unknown `--hive` reaches the operator as the roster of hives that would have worked rather than a body dump. Response `warnings` are printed when non-empty. The nix module wraps the binary with `SWARM_CONTROLLER_SOCKET`, read from the same `socketPath` the daemon binds. Refs #4399
This commit is contained in:
atlas
mara
parent
a3b672d1d5
commit
30fa54cbc6
9 changed files with 521 additions and 17 deletions
255
swarmctl/src/agent.rs
Normal file
255
swarmctl/src/agent.rs
Normal file
|
|
@ -0,0 +1,255 @@
|
|||
//! `swarmctl agent create` — queue the swarm-controller's agent-creation
|
||||
//! job graph.
|
||||
//!
|
||||
//! `POST /api/agents` inserts a DAG and returns as soon as it is
|
||||
//! *inserted*. This verb prints the queued node id and stops: the graph's
|
||||
//! last node only *publishes* a deploy, after which the hive converges on
|
||||
//! its own clock, so even a settled graph would not mean the agent is up
|
||||
//! and there is nothing here that could be waited on honestly.
|
||||
//!
|
||||
//! Transport is a bare `hyper` HTTP/1.1 client handshaked onto a
|
||||
//! [`tokio::net::UnixStream`] via [`hyper_util::rt::TokioIo`] — the
|
||||
//! controller has no TCP port, so this crate speaks plain HTTP directly
|
||||
//! over the socket rather than pulling in a full client stack for one
|
||||
//! POST.
|
||||
//!
|
||||
//! The two shapes below **mirror** the controller's own private
|
||||
//! `CreateAgentRequest` / `CreateAgentResponse` rather than sharing them:
|
||||
//! this crate does not link `swarm-controller` and there is no wire
|
||||
//! crate between them. The seam is narrow and both ends validate, so
|
||||
//! drift surfaces as a 400 naming the field.
|
||||
//!
|
||||
//! Long-form rationale for all three: `swarmctl/README.md`.
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use anyhow::{Context as _, Result, bail};
|
||||
use http_body_util::{BodyExt as _, Full};
|
||||
use hyper_util::rt::TokioIo;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::net::UnixStream;
|
||||
|
||||
/// Body of `POST /api/agents`. Mirrors the controller's own
|
||||
/// `CreateAgentRequest` — see this module's doc comment.
|
||||
#[derive(Serialize)]
|
||||
struct CreateAgentRequest<'a> {
|
||||
name: &'a str,
|
||||
/// Where the agent's deploy message is addressed. Required by the
|
||||
/// endpoint and not defaulted anywhere, because it is only knowable
|
||||
/// from the operator making the choice.
|
||||
hive: &'a str,
|
||||
}
|
||||
|
||||
/// Success body of `POST /api/agents`.
|
||||
#[derive(Deserialize)]
|
||||
struct CreateAgentResponse {
|
||||
node_id: u64,
|
||||
/// Name collisions the controller **allowed through** — it is the
|
||||
/// "warn now, refuse later" step, so an empty list is the normal case
|
||||
/// and the field is omitted from the JSON entirely when it is empty.
|
||||
#[serde(default)]
|
||||
warnings: Vec<String>,
|
||||
}
|
||||
|
||||
/// Run `swarmctl agent create`.
|
||||
///
|
||||
/// Synchronous on purpose: every other verb in this crate is, and this is
|
||||
/// the only one that needs a reactor at all. A current-thread runtime
|
||||
/// built here keeps that cost inside the one arm that incurs it, the same
|
||||
/// way `PathArgs::resolve` is called only in the arms that need the
|
||||
/// deployment's env vars.
|
||||
pub(crate) fn create(socket: &Path, name: &str, hive: &str) -> Result<()> {
|
||||
// Client-side before the round trip, so a typo is an immediate local
|
||||
// error rather than a 400 the operator has to wait for. The controller
|
||||
// validates both again — that is the gate that matters, this is the
|
||||
// one that is fast.
|
||||
let name = parse_ident(name, "agent name")?;
|
||||
let hive = parse_ident(hive, "hive")?;
|
||||
|
||||
let rt = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_io()
|
||||
.build()
|
||||
.context("starting a tokio runtime for the controller request")?;
|
||||
let resp = rt.block_on(post_create(socket, &name, &hive))?;
|
||||
|
||||
println!("queued: job node {}", resp.node_id);
|
||||
println!(
|
||||
"agent {name:?} will be deployed to hive {hive:?} once the job graph runs; \
|
||||
`swarmctl` does not wait for it"
|
||||
);
|
||||
// stderr, and after the id: the id is the result, these are asides the
|
||||
// operator should still not have to go find in the daemon's journal.
|
||||
for warning in &resp.warnings {
|
||||
eprintln!("warning: {warning}");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Parse a CLI-supplied name into a validated identifier, naming which
|
||||
/// argument was wrong — `invalid hive` and `invalid agent name` send the
|
||||
/// operator to different flags.
|
||||
fn parse_ident(value: &str, what: &str) -> Result<String> {
|
||||
hive_types::Ident::parse(value)
|
||||
.map(hive_types::Ident::into_string)
|
||||
.map_err(|reason| anyhow::anyhow!("invalid {what} {value:?}: {reason}"))
|
||||
}
|
||||
|
||||
/// One `POST /api/agents` round trip over the controller's unix socket.
|
||||
async fn post_create(socket: &Path, name: &str, hive: &str) -> Result<CreateAgentResponse> {
|
||||
let body = serde_json::to_vec(&CreateAgentRequest { name, hive })
|
||||
.context("serialising the create-agent request")?;
|
||||
|
||||
let stream = UnixStream::connect(socket)
|
||||
.await
|
||||
.with_context(|| connect_hint(socket))?;
|
||||
let (mut sender, conn) = hyper::client::conn::http1::handshake(TokioIo::new(stream))
|
||||
.await
|
||||
.with_context(|| format!("HTTP handshake with {}", socket.display()))?;
|
||||
// Drives the connection's I/O; a one-shot POST has nothing to do with
|
||||
// the join handle, and a connection that dies surfaces as an error on
|
||||
// `send_request` or on reading the body.
|
||||
tokio::spawn(conn);
|
||||
|
||||
let req = hyper::Request::builder()
|
||||
.method(hyper::Method::POST)
|
||||
.uri("/api/agents")
|
||||
// A unix socket has no authority of its own, but HTTP/1.1 requires
|
||||
// the header; the controller routes on the path alone.
|
||||
.header(hyper::header::HOST, "localhost")
|
||||
.header(hyper::header::CONTENT_TYPE, "application/json")
|
||||
.body(Full::new(bytes::Bytes::from(body)))
|
||||
.context("building the create-agent request")?;
|
||||
|
||||
let resp = sender
|
||||
.send_request(req)
|
||||
.await
|
||||
.context("sending the create-agent request to swarm-controller")?;
|
||||
let status = resp.status();
|
||||
let body = resp
|
||||
.into_body()
|
||||
.collect()
|
||||
.await
|
||||
.context("reading swarm-controller's response")?
|
||||
.to_bytes();
|
||||
|
||||
if !status.is_success() {
|
||||
bail!("{}", describe_error(status, &body));
|
||||
}
|
||||
serde_json::from_slice(&body).with_context(|| {
|
||||
format!(
|
||||
"swarm-controller answered {status} but its body is not a create-agent response: {}",
|
||||
String::from_utf8_lossy(&body).trim()
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
/// Turn a failed response into one line the operator can act on.
|
||||
///
|
||||
/// The controller answers errors as RFC 9457 `application/problem+json`,
|
||||
/// where the actionable sentence is `detail` — a 400 for an unknown hive
|
||||
/// names the hives that *would* have worked. Printing the raw body instead
|
||||
/// would bury that in JSON punctuation, and printing only the status would
|
||||
/// throw it away. Falls back through `title` to the raw body so a response
|
||||
/// from something that is *not* the controller is still shown rather than
|
||||
/// reduced to a bare number.
|
||||
fn describe_error(status: hyper::StatusCode, body: &[u8]) -> String {
|
||||
let text = String::from_utf8_lossy(body);
|
||||
let text = text.trim();
|
||||
let field = |v: &serde_json::Value, k: &str| {
|
||||
v.get(k)
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(str::to_owned)
|
||||
};
|
||||
let detail = serde_json::from_str::<serde_json::Value>(text)
|
||||
.ok()
|
||||
.and_then(|v| field(&v, "detail").or_else(|| field(&v, "title")));
|
||||
match detail {
|
||||
Some(detail) => format!("swarm-controller refused the request ({status}): {detail}"),
|
||||
None if text.is_empty() => format!("swarm-controller refused the request ({status})"),
|
||||
None => format!("swarm-controller refused the request ({status}): {text}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Message for a socket that could not be dialled.
|
||||
///
|
||||
/// Worth spelling out because the three ways this fails need three
|
||||
/// different fixes and `No such file or directory (os error 2)` names
|
||||
/// none of them: the daemon is off, it binds elsewhere, or this host is
|
||||
/// not the one running it at all. No group-membership case to mention
|
||||
/// here — `swarmctl` already runs as root.
|
||||
fn connect_hint(socket: &Path) -> String {
|
||||
format!(
|
||||
"could not connect to swarm-controller at {} — is it running on this host? \
|
||||
(`systemctl status swarm-controller`). The path comes from \
|
||||
`--controller-socket` or `SWARM_CONTROLLER_SOCKET`, which the \
|
||||
swarm-controller nix module sets from `socketPath`",
|
||||
socket.display()
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{CreateAgentResponse, describe_error, parse_ident};
|
||||
|
||||
/// The endpoint omits `warnings` entirely when there are none, so the
|
||||
/// normal response must still decode — a missing field here would make
|
||||
/// every successful creation fail to parse.
|
||||
#[test]
|
||||
fn a_response_without_warnings_decodes() {
|
||||
let resp: CreateAgentResponse =
|
||||
serde_json::from_str(r#"{"node_id":7}"#).expect("the normal shape decodes");
|
||||
assert_eq!(resp.node_id, 7);
|
||||
assert!(resp.warnings.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn warnings_are_carried_through() {
|
||||
let resp: CreateAgentResponse =
|
||||
serde_json::from_str(r#"{"node_id":9,"warnings":["reserved name"]}"#).expect("decodes");
|
||||
assert_eq!(resp.warnings, vec!["reserved name".to_owned()]);
|
||||
}
|
||||
|
||||
/// The whole point of parsing problem+json: the 400 for an unknown
|
||||
/// hive carries the roster of hives that would have worked, and that
|
||||
/// sentence has to reach the operator's terminal intact.
|
||||
#[test]
|
||||
fn a_problem_json_detail_reaches_the_operator() {
|
||||
let body = r#"{"status":400,"title":"Bad Request","detail":"hive \"tyop\" is not in this swarm — known hives: alpha, beta"}"#;
|
||||
let msg = describe_error(hyper::StatusCode::BAD_REQUEST, body.as_bytes());
|
||||
assert!(msg.contains("known hives: alpha, beta"), "{msg}");
|
||||
assert!(msg.contains("400"), "{msg}");
|
||||
// No JSON punctuation: the operator gets prose, not a body dump.
|
||||
assert!(!msg.contains('{'), "{msg}");
|
||||
}
|
||||
|
||||
/// A response from something that is not the controller (an nginx
|
||||
/// error page, a proxy) must still be shown rather than swallowed.
|
||||
#[test]
|
||||
fn a_non_json_body_is_shown_verbatim() {
|
||||
let msg = describe_error(hyper::StatusCode::BAD_GATEWAY, b" 502 Bad Gateway ");
|
||||
assert!(msg.contains("502 Bad Gateway"), "{msg}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_empty_body_still_names_the_status() {
|
||||
let msg = describe_error(hyper::StatusCode::INTERNAL_SERVER_ERROR, b"");
|
||||
assert!(msg.contains("500"), "{msg}");
|
||||
}
|
||||
|
||||
/// Which argument was wrong has to be in the message: `agent create`
|
||||
/// takes two identifiers and they are fixed in different places.
|
||||
#[test]
|
||||
fn an_invalid_ident_names_the_argument_it_came_from() {
|
||||
let err = parse_ident("Not An Ident", "hive").expect_err("rejected");
|
||||
let msg = err.to_string();
|
||||
assert!(msg.contains("invalid hive"), "{msg}");
|
||||
assert!(msg.contains("Not An Ident"), "{msg}");
|
||||
|
||||
assert!(
|
||||
parse_ident("scribe-01", "agent name").is_ok(),
|
||||
"a normal name must pass"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -11,16 +11,23 @@
|
|||
//! activation) or world-readable password hashes. Both are worse than
|
||||
//! root.
|
||||
//!
|
||||
//! So there is no socket, no HTTP route and no privileged helper here.
|
||||
//! When a verb eventually has to run as a non-root user or from another
|
||||
//! host, the answer is a **group-gated admin socket** — separate from the
|
||||
//! controller's `0666` gateway-facing one — not a widening of what root
|
||||
//! does here.
|
||||
//! So this binary **serves** no socket, publishes no HTTP route and has
|
||||
//! no privileged helper. When a verb eventually has to run as a non-root
|
||||
//! user or from another host, the answer is a **group-gated admin
|
||||
//! socket** — separate from the controller's `0666` gateway-facing one —
|
||||
//! not a widening of what root does here.
|
||||
//!
|
||||
//! Acting directly is not the same as acting *alone*, though: `agent
|
||||
//! create` is a client of the controller's own unix socket, because the
|
||||
//! work it asks for is a job graph only the controller can queue (see
|
||||
//! [`agent`]). That is the opposite direction from the socket the
|
||||
//! paragraph above rules out — nothing here becomes reachable by it.
|
||||
//!
|
||||
//! Distinct from `hivectl`, which drives one hive's `hive-c0re` over its
|
||||
//! admin socket. This crate deliberately does not link `swarm-controller`,
|
||||
//! for the same reason `hivectl` does not link `hive-c0re`.
|
||||
|
||||
mod agent;
|
||||
mod users;
|
||||
|
||||
use std::fs::{self, File, Permissions};
|
||||
|
|
@ -106,6 +113,11 @@ fn missing(env: &str) -> String {
|
|||
/// a derive, several errors away from the actual cause.
|
||||
#[derive(Subcommand)]
|
||||
enum Verb {
|
||||
/// Manage agents across the swarm.
|
||||
Agent {
|
||||
#[command(subcommand)]
|
||||
command: AgentVerb,
|
||||
},
|
||||
/// Manage subjects in the swarm's SSO provider.
|
||||
User {
|
||||
#[command(subcommand)]
|
||||
|
|
@ -139,6 +151,49 @@ enum Verb {
|
|||
},
|
||||
}
|
||||
|
||||
#[derive(Subcommand)]
|
||||
enum AgentVerb {
|
||||
/// Queue creation of a new agent on a hive in this swarm.
|
||||
///
|
||||
/// Asks the swarm-controller to insert its agent-creation job graph —
|
||||
/// SSO identity, forge user, config repo, and the deploy message that
|
||||
/// puts the agent on `--hive` — and prints the queued job's node id.
|
||||
///
|
||||
/// **This returns as soon as the work is queued.** It doesn't wait,
|
||||
/// and a finished graph would not mean the agent is up either: the
|
||||
/// last node publishes a deploy, after which the hive converges on its
|
||||
/// own clock. Watch the swarm UI's job view, or the hive itself, for
|
||||
/// the rest.
|
||||
///
|
||||
/// No approval gate guards this: running this binary already means
|
||||
/// being root on the controller's host.
|
||||
Create(AgentCreateArgs),
|
||||
}
|
||||
|
||||
#[derive(Args)]
|
||||
struct AgentCreateArgs {
|
||||
/// Name for the new agent: 1–63 characters of `[a-z0-9-]`.
|
||||
///
|
||||
/// Becomes an SSO subject, a forge user and a repository name, so
|
||||
/// it's validated here before anything is queued.
|
||||
name: String,
|
||||
/// Hive in this swarm to deploy the agent to.
|
||||
///
|
||||
/// Required, and deliberately not defaulted: it's an *address* — the
|
||||
/// hive a deploy message is sent to — and only the operator knows
|
||||
/// which one they mean. The controller checks it against the swarm's
|
||||
/// hive roster and names the known hives if it misses.
|
||||
#[arg(long, value_name = "HIVE")]
|
||||
hive: String,
|
||||
/// swarm-controller's unix socket.
|
||||
///
|
||||
/// Supplied by the nix module that installs this binary, from the same
|
||||
/// `socketPath` option the daemon binds; falls back to
|
||||
/// `SWARM_CONTROLLER_SOCKET`.
|
||||
#[arg(long, value_name = "PATH")]
|
||||
controller_socket: Option<PathBuf>,
|
||||
}
|
||||
|
||||
#[derive(Subcommand)]
|
||||
enum UserVerb {
|
||||
/// Add a user, generating a password for them.
|
||||
|
|
@ -197,6 +252,16 @@ struct UpdateArgs {
|
|||
fn main() -> Result<()> {
|
||||
let Cli { paths, command } = Cli::parse();
|
||||
match command {
|
||||
// Resolves its own socket path, not `PathArgs`: this verb needs
|
||||
// none of the `SWARMCTL_AUTHELIA_*` values, and requiring them
|
||||
// would make agent creation fail on a controller host that is not
|
||||
// also the swarm's SSO host.
|
||||
Verb::Agent {
|
||||
command: AgentVerb::Create(args),
|
||||
} => {
|
||||
let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?;
|
||||
agent::create(&socket, &args.name, &args.hive)
|
||||
}
|
||||
// Resolved lazily, inside the one arm that actually touches the
|
||||
// deployment env vars — see the `MarkdownDocs` doc comment above
|
||||
// for why an unconditional resolve up front would be wrong.
|
||||
|
|
@ -489,6 +554,69 @@ fn write_atomic(path: &Path, contents: &str) -> Result<()> {
|
|||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// clap's own consistency checks over the whole tree — a duplicated
|
||||
/// long flag or a malformed `value_name` panics at parse time in
|
||||
/// production and is otherwise only found by running the binary.
|
||||
#[test]
|
||||
fn the_clap_tree_is_well_formed() {
|
||||
use clap::CommandFactory as _;
|
||||
Cli::command().debug_assert();
|
||||
}
|
||||
|
||||
/// `--hive` carries the address the agent is deployed to and the
|
||||
/// endpoint has no default for it, so omitting it has to fail at parse
|
||||
/// time rather than reach the controller as an empty string.
|
||||
#[test]
|
||||
fn agent_create_requires_a_hive() {
|
||||
assert!(
|
||||
Cli::try_parse_from(["swarmctl", "agent", "create", "scribe"]).is_err(),
|
||||
"a create with no --hive must not parse"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn agent_create_parses_its_name_hive_and_socket() {
|
||||
let cli = Cli::try_parse_from([
|
||||
"swarmctl",
|
||||
"agent",
|
||||
"create",
|
||||
"scribe",
|
||||
"--hive",
|
||||
"alpha",
|
||||
"--controller-socket",
|
||||
"/run/elsewhere/controller.sock",
|
||||
])
|
||||
.expect("the full form parses");
|
||||
let Verb::Agent {
|
||||
command: AgentVerb::Create(args),
|
||||
} = cli.command
|
||||
else {
|
||||
panic!("expected `agent create`");
|
||||
};
|
||||
assert_eq!(args.name, "scribe");
|
||||
assert_eq!(args.hive, "alpha");
|
||||
assert_eq!(
|
||||
args.controller_socket.as_deref(),
|
||||
Some(Path::new("/run/elsewhere/controller.sock"))
|
||||
);
|
||||
}
|
||||
|
||||
/// The socket is optional on the command line because the nix wrapper
|
||||
/// sets `SWARM_CONTROLLER_SOCKET`; `path_from` is what turns an absent
|
||||
/// pair into an error rather than a guess.
|
||||
#[test]
|
||||
fn an_omitted_socket_flag_leaves_the_env_to_supply_it() {
|
||||
let cli = Cli::try_parse_from(["swarmctl", "agent", "create", "scribe", "--hive", "alpha"])
|
||||
.expect("the minimal form parses");
|
||||
let Verb::Agent {
|
||||
command: AgentVerb::Create(args),
|
||||
} = cli.command
|
||||
else {
|
||||
panic!("expected `agent create`");
|
||||
};
|
||||
assert!(args.controller_socket.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_authelia_hash_output() {
|
||||
let out = "Random Password: hunter2\nDigest: $argon2id$v=19$m=65536$abc\n";
|
||||
|
|
|
|||
Loading…
Reference in a new issue