//! `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, } /// 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}")) } /// One `POST /api/agents` round trip over the controller's unix socket. async fn post_create(socket: &Path, name: &str, hive: &str) -> Result { 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::(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" ); } }