//! `swarmctl agent create`, `swarmctl agent mint-identity` and `swarmctl agent //! mint-forge-token` — queue work on the swarm-controller's job graph. //! //! Each POSTs and returns as soon as the work is *inserted*. Both verbs //! print the queued node id and stop: creation'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, } /// Body of `POST /api/agents/{name}/identity`. Mirrors the controller's own /// `MintAgentIdentityRequest` — see this module's doc comment. #[derive(Serialize)] struct MintIdentityRequest<'a> { hive: &'a str, } /// Success body of `POST /api/agents/{name}/identity`. #[derive(Deserialize)] struct MintIdentityResponse { node_id: u64, } /// Success body of `POST /api/agents/{name}/forge-token`, which takes no body. #[derive(Deserialize)] struct MintForgeTokenResponse { node_id: u64, } /// 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 { hive_types::Ident::parse(value) .map(hive_types::Ident::into_string) .map_err(|reason| anyhow::anyhow!("invalid {what} {value:?}: {reason}")) } /// Run `swarmctl agent mint-identity`. /// /// Synchronous for the same reason [`create`] is, and built on the same /// round trip. pub(crate) fn mint_identity(socket: &Path, name: &str, hive: &str) -> Result<()> { // Client-side first, so a typo is a local error rather than a 400 the // operator waits for. The controller validates both again. 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: MintIdentityResponse = rt.block_on(post( socket, &format!("/api/agents/{name}/identity"), &MintIdentityRequest { hive: &hive }, "mint-identity", ))?; println!("queued: job node {}", resp.node_id); println!( "agent {name:?} will have its identity re-minted on hive {hive:?} once the job graph \ runs; `swarmctl` does not wait for it. An existing queue secret is kept as it is; the \ store certificate is re-minted and the agent picks the new one up on its next boot" ); Ok(()) } /// Run `swarmctl agent mint-forge-token`. /// /// Synchronous for the same reason [`create`] is, and built on the same /// round trip. No `--hive`: the token's store path has no hive in it. pub(crate) fn mint_forge_token(socket: &Path, name: &str) -> Result<()> { let name = parse_ident(name, "agent name")?; let rt = tokio::runtime::Builder::new_current_thread() .enable_io() .build() .context("starting a tokio runtime for the controller request")?; let resp: MintForgeTokenResponse = rt.block_on(post( socket, &format!("/api/agents/{name}/forge-token"), &serde_json::json!({}), "mint-forge-token", ))?; println!("queued: job node {}", resp.node_id); println!( "agent {name:?}'s forge token will be checked once the job graph runs, and minted \ only if it is missing or stale; `swarmctl` does not wait for it. The agent picks a \ new token up on its next fetch" ); Ok(()) } /// One `POST /api/agents` round trip over the controller's unix socket. async fn post_create(socket: &Path, name: &str, hive: &str) -> Result { post( socket, "/api/agents", &CreateAgentRequest { name, hive }, "create-agent", ) .await } /// One JSON `POST` to `uri` over the controller's unix socket. /// /// `what` names the request in error messages — the operator needs to know /// which call failed, and every other part of this function is identical /// between the two verbs. async fn post( socket: &Path, uri: &str, request: &Req, what: &str, ) -> Result { let body = serde_json::to_vec(request).with_context(|| format!("serialising the {what} 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(uri) // 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))) .with_context(|| format!("building the {what} request"))?; let resp = sender .send_request(req) .await .with_context(|| format!("sending the {what} 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 {what} 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::(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" ); } }