667 lines
27 KiB
Rust
667 lines
27 KiB
Rust
//! Swarm-level controller daemon. Runs as the unprivileged
|
|
//! `swarm-controller` user on whichever host the operator flips
|
|
//! `services.hyperhive.swarm.controller.enable` on, and serves HTTP over a
|
|
//! unix socket that the hive-gateway's nginx proxies to.
|
|
//!
|
|
//! Holds one piece of read-only state: the swarm's hive directory, loaded
|
|
//! once at startup from an env var the NixOS module sets
|
|
//! (`services.hyperhive.swarm.controller`) — see `load_hives`. Still no
|
|
//! persistence and no writes; a config change means a redeploy, same as
|
|
//! every other option this process reads.
|
|
//!
|
|
//! Distinct from `hive-c0re`, which is per-hive: c0re owns the agents on
|
|
//! one host, this owns what is true across hives.
|
|
//!
|
|
//! `OpenAPI` spec generation mirrors `hive-c0re/src/dashboard/mod.rs`
|
|
//! exactly: `#[utoipa::path(...)]` per handler, an `ApiDoc` root, and a
|
|
//! raw JSON route at `/api/openapi.json`. Swagger UI itself is
|
|
//! nginx-hosted from the nix store, same shape as the per-hive
|
|
//! dashboard's (`nix/host-modules/hive-gateway/vhosts.nix`'s
|
|
//! `swarmUiVhost` — a swagger-ui-theme dist under `/api/docs/`, no
|
|
//! fallback to this daemon). Only annotated routes appear in the spec;
|
|
//! an unannotated one just doesn't show up, nothing breaks.
|
|
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
use std::path::PathBuf;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use anyhow::{Context, Result};
|
|
use axum::{Json, extract::State, routing::get};
|
|
use hive_jobq_wire::GraphWire as _;
|
|
use serde::{Deserialize, Serialize};
|
|
use utoipa::{OpenApi, ToSchema};
|
|
use utoipa_axum::{router::OpenApiRouter, routes};
|
|
|
|
mod status;
|
|
|
|
/// Placeholder node payload for the swarm-level job graph — uninhabited on
|
|
/// purpose, and named `Swarm*` rather than the bare `NodeKind`/`Resource`
|
|
/// `hive-c0re::job_queue::model` already uses, so a grep for either doesn't
|
|
/// land on both crates. The *scheduler loop* below is real and running
|
|
/// (`spawn_jobq_worker`, mirroring `hive-c0re/src/job_queue/scheduler.rs`'s
|
|
/// `run_worker`) — what's still missing is a real job to give it: no
|
|
/// variant exists yet, so nothing is ever inserted into the graph and
|
|
/// `claim_next` always returns `None`. Giving this real variants (starting
|
|
/// with `CreateRepo`) is the next slice, landing together with
|
|
/// `swarm-controller::forge`, the client those nodes will call. `WireNode`
|
|
/// is trivially satisfiable on an empty enum (`match *self {}`), so the
|
|
/// wire machinery below is real and typechecked today, with nothing yet to
|
|
/// put in it.
|
|
#[derive(Clone, Debug)]
|
|
enum SwarmNodeKind {}
|
|
|
|
impl hive_jobq_wire::WireNode for SwarmNodeKind {
|
|
fn label(&self) -> String {
|
|
match *self {}
|
|
}
|
|
|
|
fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value {
|
|
match *self {}
|
|
}
|
|
}
|
|
|
|
/// Placeholder resource name — same rationale and same "no variants until a
|
|
/// real node needs one" shape as [`SwarmNodeKind`].
|
|
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
|
|
enum SwarmResourceKind {}
|
|
|
|
impl hive_jobq_wire::WireResource for SwarmResourceKind {
|
|
fn name(&self) -> String {
|
|
match *self {}
|
|
}
|
|
}
|
|
|
|
/// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/
|
|
/// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind`
|
|
/// variant turns into a real effect. Trivially exhaustive today
|
|
/// (`match kind {}`) because the enum has no variants yet; the first
|
|
/// real arm (`CreateRepo`, calling `swarm-controller::forge`) lands
|
|
/// alongside that variant, not before.
|
|
async fn run_swarm_node(
|
|
_id: hive_jobq::NodeId,
|
|
kind: SwarmNodeKind,
|
|
builder: hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
|
|
) -> (
|
|
hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
|
|
hive_jobq::scheduler::Outcome,
|
|
) {
|
|
let _ = builder;
|
|
match kind {}
|
|
}
|
|
|
|
/// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/
|
|
/// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node,
|
|
/// spawn the future that runs + completes it, loop again immediately if
|
|
/// something started (more may now be runnable), otherwise back off
|
|
/// briefly before re-polling.
|
|
///
|
|
/// No shutdown signal to wire in — unlike `hive-c0re`'s `coord.shutdown_rx()`,
|
|
/// this daemon has no graceful-shutdown machinery at all yet (`main`'s
|
|
/// `axum::serve` runs unconditionally to process exit), so this loop
|
|
/// matches that: it rides the runtime down with the process, same as
|
|
/// every in-flight HTTP request does.
|
|
///
|
|
/// Cheap to run with an empty graph: `claim_next` on a graph nothing was
|
|
/// ever inserted into just returns `None` every poll, so this is a
|
|
/// harmless idle loop until the first real node kind exists.
|
|
fn spawn_jobq_worker(
|
|
sched: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
|
) {
|
|
tokio::spawn(async move {
|
|
loop {
|
|
let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, run_swarm_node);
|
|
match runner {
|
|
Some(runner) => {
|
|
tokio::spawn(async move {
|
|
let (id, grew) = runner.await;
|
|
if let Err(e) = grew {
|
|
tracing::warn!(
|
|
node = id.get(),
|
|
error = %e,
|
|
"swarm jobq: grown job rejected"
|
|
);
|
|
}
|
|
});
|
|
// Something just started — more may be runnable right
|
|
// now, so loop again immediately rather than sleeping.
|
|
}
|
|
None => {
|
|
// Nothing runnable. Bounded poll rather than an event
|
|
// wake (unlike hive-c0re's `notify.notify_one()`,
|
|
// there is no completion-signal channel here yet) —
|
|
// fine at this daemon's scale (one graph, no
|
|
// submitters yet); revisit if/when that stops holding.
|
|
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`.
|
|
///
|
|
/// A compiled-in default is legitimate here and is *not* the mistake that
|
|
/// a hardcoded remote address would be: this is a path this process
|
|
/// **creates**, not an address it hopes to find something at. systemd's
|
|
/// `RuntimeDirectory=swarm-controller` makes the parent exist before
|
|
/// `ExecStart`, so the default names a directory the unit just produced.
|
|
///
|
|
/// The directory is its own — deliberately not shared with hive-c0re's
|
|
/// `/run/hyperhive`. The socket is `0666`, so its directory is the only
|
|
/// access control it has; co-locating it with c0re's admin socket would
|
|
/// put both within reach of whatever can reach either. nginx runs on the
|
|
/// host, so nothing narrows its reach for you.
|
|
const DEFAULT_SOCKET: &str = "/run/swarm-controller/controller.sock";
|
|
|
|
fn socket_path() -> PathBuf {
|
|
std::env::var_os("SWARM_CONTROLLER_SOCKET")
|
|
.map_or_else(|| PathBuf::from(DEFAULT_SOCKET), PathBuf::from)
|
|
}
|
|
|
|
/// Root of the auto-generated `OpenAPI` spec, served raw at
|
|
/// `/api/openapi.json` — see the module doc comment above. Tag list
|
|
/// grows alongside the swarm-level surfaces this daemon picks up, same
|
|
/// as `hive-c0re::dashboard::ApiDoc`'s tag list did.
|
|
#[derive(OpenApi)]
|
|
#[openapi(
|
|
info(
|
|
title = "hyperhive swarm-controller API",
|
|
description = "swarm-controller's HTTP surface, served over its unix \
|
|
socket behind the gateway's swarm-UI vhost."
|
|
),
|
|
tags(
|
|
(name = "health", description = "liveness probe"),
|
|
(name = "hives", description = "the swarm's hive directory"),
|
|
(name = "links", description = "swarm service quick links"),
|
|
(name = "jobq", description = "the swarm-level job graph"),
|
|
)
|
|
)]
|
|
struct ApiDoc;
|
|
|
|
/// Liveness probe. Returns the build's version so an operator can tell
|
|
/// *which* controller answered without shelling onto the host.
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/health",
|
|
responses((status = 200, description = "process is up, body is \"swarm-controller <version>\"", body = String)),
|
|
tag = "health"
|
|
)]
|
|
async fn health() -> &'static str {
|
|
concat!("swarm-controller ", env!("CARGO_PKG_VERSION"), "\n")
|
|
}
|
|
|
|
/// One hive in the swarm's directory — the same `name`/`domain` pair
|
|
/// `services.hyperhive.swarm.hives` (nix/host-modules/swarm.nix) declares,
|
|
/// carried across unchanged rather than reshaped, so this stays a direct
|
|
/// mirror of the nix source of truth instead of a second vocabulary for
|
|
/// the same two fields.
|
|
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)]
|
|
struct HiveEntry {
|
|
name: String,
|
|
domain: String,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct AppState {
|
|
/// Loaded once at startup (`load_hives`); never mutated, so an
|
|
/// `Arc` clone per request is the whole synchronization story.
|
|
hives: Arc<Vec<HiveEntry>>,
|
|
/// Loaded once at startup (`load_links`); same synchronization story
|
|
/// as `hives`.
|
|
links: Arc<Vec<ServiceLink>>,
|
|
/// `None` when this deployment wired up no swarm queue — the only
|
|
/// state in which `/api/hives/status` cannot answer at all. A queue
|
|
/// that is merely *unreachable* still yields a reader, because
|
|
/// `async-nats` reconnects underneath it.
|
|
status: Option<Arc<status::StatusReader>>,
|
|
/// The swarm-level job graph, wrapped in its
|
|
/// [`hive_jobq::scheduler::Scheduler`] now that something drives it
|
|
/// (`spawn_jobq_worker`) — the graph alone was enough for the
|
|
/// read-only endpoints, the scheduler is what a `claim_next` loop
|
|
/// needs. `std::sync::Mutex`, not `tokio`'s — every lock scope below
|
|
/// is synchronous (no `.await` while held). Always present, never
|
|
/// gated on the swarm queue: this is process state, not something
|
|
/// read over the network.
|
|
jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
|
}
|
|
|
|
/// Env var the controller's NixOS module sets from
|
|
/// `services.hyperhive.swarm.hives`, JSON-encoded — the full directory
|
|
/// (this daemon has no "self" to exclude) rather than peers-minus-self
|
|
/// (`services.hyperhive.swarm.peerHives`, which other consumers use).
|
|
const HIVES_ENV: &str = "SWARM_CONTROLLER_HIVES";
|
|
|
|
/// Parses [`HIVES_ENV`] into the swarm's hive directory. Unset or
|
|
/// unparseable both fall back to an empty list with a warning rather than
|
|
/// failing startup — a swarm-controller that can't yet see its own config
|
|
/// (dev run, a module not wired up yet) should still serve `/health`.
|
|
fn load_hives() -> Vec<HiveEntry> {
|
|
let Some(raw) = std::env::var_os(HIVES_ENV) else {
|
|
return Vec::new();
|
|
};
|
|
match serde_json::from_str(&raw.to_string_lossy()) {
|
|
Ok(hives) => hives,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, env = HIVES_ENV, "failed to parse hive directory, serving an empty list");
|
|
Vec::new()
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The swarm's hive directory — every hive, including whichever one this
|
|
/// controller instance happens to run on (there is no "self" to exclude
|
|
/// at the swarm level, unlike `hive-c0re`'s peer list).
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/hives",
|
|
responses((status = 200, description = "every hive in the swarm", body = Vec<HiveEntry>)),
|
|
tag = "hives"
|
|
)]
|
|
async fn get_hives(State(state): State<AppState>) -> Json<Vec<HiveEntry>> {
|
|
Json((*state.hives).clone())
|
|
}
|
|
|
|
/// One quick link to a swarm-wide service (authelia, matrix, forge, this
|
|
/// daemon's own swagger UI, …). Deliberately generic rather than named
|
|
/// fields per service: each service's own nix module contributes its own
|
|
/// entry to `services.hyperhive.swarm.controller.links` (same list-merge
|
|
/// idiom `services.hyperhive.gateway.localNames` already uses), so adding
|
|
/// a new one is a nix-only change — no new field here, no swarm-ui change.
|
|
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)]
|
|
struct ServiceLink {
|
|
label: String,
|
|
/// Emoji or short glyph. Empty string, not `Option`, when a
|
|
/// contributing module has none — one fewer null-vs-absent case for
|
|
/// the frontend to handle, and every real contributor sets one today.
|
|
icon: String,
|
|
url: String,
|
|
}
|
|
|
|
/// Env var the controller's NixOS module sets from the merged
|
|
/// `services.hyperhive.swarm.controller.links` list, JSON-encoded — same
|
|
/// shape/rationale as [`HIVES_ENV`]. Consumed by `GET /api/links`.
|
|
const LINKS_ENV: &str = "SWARM_CONTROLLER_LINKS";
|
|
|
|
/// Parses [`LINKS_ENV`] into the swarm's service-link list. Same
|
|
/// fall-back-to-empty rationale as `load_hives`: a daemon that can't yet
|
|
/// see this config should still serve `/health` rather than fail startup.
|
|
fn load_links() -> Vec<ServiceLink> {
|
|
let Some(raw) = std::env::var_os(LINKS_ENV) else {
|
|
return Vec::new();
|
|
};
|
|
match serde_json::from_str(&raw.to_string_lossy()) {
|
|
Ok(links) => links,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, env = LINKS_ENV, "failed to parse service links, serving an empty list");
|
|
Vec::new()
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Quick links to swarm-wide services (authelia, matrix, forge, this
|
|
/// daemon's own swagger UI, …), as contributed by each service's own nix
|
|
/// module. Empty when nothing was configured to contribute — a caller
|
|
/// renders 0 links the same way it renders any other count, no special
|
|
/// "not configured" case.
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/links",
|
|
responses((status = 200, description = "swarm service quick links", body = Vec<ServiceLink>)),
|
|
tag = "links"
|
|
)]
|
|
async fn get_links(State(state): State<AppState>) -> Json<Vec<ServiceLink>> {
|
|
Json((*state.links).clone())
|
|
}
|
|
|
|
/// Why the status route answers 503 rather than an empty list.
|
|
///
|
|
/// "I cannot reach the store" and "every hive is silent" are different
|
|
/// answers, and rendering the second when the first is true is exactly
|
|
/// the smoothing this endpoint exists to avoid — a caller would draw a
|
|
/// swarm-wide outage out of a local one. The cause is carried in the
|
|
/// body because a bare 503 on an operator-facing diagnostic is how a
|
|
/// misconfiguration costs an afternoon; it is a queue/JetStream error
|
|
/// string, and this surface is already behind the swarm's SSO.
|
|
struct StatusUnavailable(String);
|
|
|
|
impl axum::response::IntoResponse for StatusUnavailable {
|
|
fn into_response(self) -> axum::response::Response {
|
|
(axum::http::StatusCode::SERVICE_UNAVAILABLE, self.0).into_response()
|
|
}
|
|
}
|
|
|
|
/// What each hive last said about itself, read from the swarm queue at
|
|
/// request time.
|
|
///
|
|
/// Every hive in the roster gets a row whether or not it has ever
|
|
/// reported — see the `status` module for why absence, not presence, is
|
|
/// the case this is built around.
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/hives/status",
|
|
responses(
|
|
(status = 200, description = "a row per hive, freshness derived now", body = Vec<status::HiveStatus>),
|
|
(status = 503, description = "no swarm queue is configured here, or its store could not be read", body = String),
|
|
),
|
|
tag = "hives"
|
|
)]
|
|
async fn get_hives_status(
|
|
State(state): State<AppState>,
|
|
) -> Result<Json<Vec<status::HiveStatus>>, StatusUnavailable> {
|
|
let Some(reader) = state.status.as_ref() else {
|
|
return Err(StatusUnavailable(
|
|
"no swarm queue is configured on this host".to_owned(),
|
|
));
|
|
};
|
|
match reader
|
|
.view(&state.hives, std::time::SystemTime::now())
|
|
.await
|
|
{
|
|
Ok(rows) => Ok(Json(rows)),
|
|
Err(e) => {
|
|
let detail = format!("{e:#}");
|
|
tracing::warn!(error = %detail, "reading the swarm status bucket failed");
|
|
Err(StatusUnavailable(detail))
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Query params for `GET /api/jobq/graph` — `?states=` narrows to root
|
|
/// groups in the named states, same shape `hive_jobq_wire::parse_states`
|
|
/// parses.
|
|
#[derive(Deserialize, utoipa::IntoParams)]
|
|
struct JobqGraphQuery {
|
|
states: Option<String>,
|
|
}
|
|
|
|
/// Every node of every root group in the swarm-level job graph. Nothing
|
|
/// filters "done and old" here — with zero nodes ever submitted there is
|
|
/// nothing to bound yet, so `graph.roots()` (everything) is the whole
|
|
/// roster passed to [`hive_jobq_wire::GraphWire::wire_snapshot`].
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/jobq/graph",
|
|
params(JobqGraphQuery),
|
|
responses((status = 200, description = "every node of every root group, as generic \
|
|
`hive_jobq` graph nodes. `?states=` narrows to root groups in the named states.",
|
|
body = Vec<hive_jobq_wire::GraphNode>)),
|
|
tag = "jobq"
|
|
)]
|
|
async fn get_jobq_graph(
|
|
State(state): State<AppState>,
|
|
axum::extract::Query(q): axum::extract::Query<JobqGraphQuery>,
|
|
) -> Json<Vec<hive_jobq_wire::GraphNode>> {
|
|
let states = hive_jobq_wire::parse_states(q.states.as_deref());
|
|
let sched = state
|
|
.jobq
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
|
let graph = sched.graph();
|
|
let roots: Vec<hive_jobq::NodeId> = graph.roots().map(|n| n.id).collect();
|
|
let nodes = graph.wire_snapshot(roots);
|
|
Json(hive_jobq_wire::filter_nodes_by_state(
|
|
nodes,
|
|
states.as_deref(),
|
|
))
|
|
}
|
|
|
|
/// Counts by lifecycle state over the same groups `/api/jobq/graph` serves.
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/jobq/rollup",
|
|
responses((status = 200, description = "counts by lifecycle state", body = Vec<hive_jobq_wire::StateCount>)),
|
|
tag = "jobq"
|
|
)]
|
|
async fn get_jobq_rollup(State(state): State<AppState>) -> Json<Vec<hive_jobq_wire::StateCount>> {
|
|
let sched = state
|
|
.jobq
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
|
let graph = sched.graph();
|
|
let roots: Vec<hive_jobq::NodeId> = graph.roots().map(|n| n.id).collect();
|
|
Json(hive_jobq_wire::state_rollup(graph, roots))
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
tracing_subscriber::fmt()
|
|
.with_env_filter(
|
|
tracing_subscriber::EnvFilter::try_from_default_env()
|
|
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
|
|
)
|
|
.init();
|
|
|
|
let path = socket_path();
|
|
|
|
// `RuntimeDirectoryPreserve=yes` keeps the directory across a restart,
|
|
// so a socket file from the previous run can outlive the process that
|
|
// owned it and `bind` would fail with EADDRINUSE. Unlinking a stale
|
|
// socket is safe precisely because the directory is ours alone: nothing
|
|
// else can have put a file at this path.
|
|
if let Err(e) = std::fs::remove_file(&path)
|
|
&& e.kind() != std::io::ErrorKind::NotFound
|
|
{
|
|
return Err(e).with_context(|| format!("clearing stale socket at {}", path.display()));
|
|
}
|
|
|
|
let listener = tokio::net::UnixListener::bind(&path)
|
|
.with_context(|| format!("binding {}", path.display()))?;
|
|
|
|
// `bind` leaves the socket 0755, and connecting needs write — the
|
|
// gateway's nginx is a different user, so it would be locked out.
|
|
// 0666 matches how hive-c0re publishes the per-agent sockets
|
|
// (`socket_server::start`), and rests on the same argument: **the
|
|
// containing directory is the access control, not the socket mode.**
|
|
// This directory holds one socket and is bind-mounted into exactly
|
|
// one container. That is also why it must not be shared with
|
|
// hive-c0re's `/run/hyperhive` — with a 0666 socket, a directory
|
|
// that carries more than it should is the whole vulnerability.
|
|
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o666))
|
|
.with_context(|| format!("chmod {}", path.display()))?;
|
|
tracing::info!(socket = %path.display(), "swarm-controller listening");
|
|
|
|
// Connect to the swarm queue when this deployment wired one up.
|
|
//
|
|
// Deliberately NOT fatal on failure: the controller's HTTP surface is
|
|
// useful without the queue, and a hive that cannot be read from renders
|
|
// as `unknown` rather than as an outage of this daemon. What IS fatal is
|
|
// a half-set environment — `QueueConfig::from_env` refuses that, because
|
|
// silently behaving like an unconfigured host is how every hive ends up
|
|
// reading `never_reported` with nothing to point at.
|
|
let status = match swarm_queue_client::QueueConfig::from_env("SWARM_CONTROLLER")? {
|
|
None => {
|
|
tracing::info!("no swarm queue configured; status aggregation is off");
|
|
None
|
|
}
|
|
Some(cfg) => match swarm_queue_client::connect(cfg).await {
|
|
Ok(client) => {
|
|
// NOT "connected": `retry_on_initial_connect` returns a client
|
|
// before any connection has been established, so claiming a
|
|
// connection here would put "connected to the swarm queue" in
|
|
// the journal moments before every request 503s with "not
|
|
// connected" — and a reader would rightly distrust the second
|
|
// line rather than the first. The connection's real state is
|
|
// reported by the status endpoint, which checks it per request.
|
|
tracing::info!("swarm queue configured; connecting in the background");
|
|
Some(Arc::new(status::StatusReader::new(
|
|
client,
|
|
status::StatusReader::stale_after_from_env(),
|
|
)))
|
|
}
|
|
Err(e) => {
|
|
// `chain`, not `{:#}`: this is the queue client's own
|
|
// error type, and thiserror's Display ignores the
|
|
// alternate flag — the source would be dropped silently.
|
|
tracing::warn!(
|
|
error = swarm_queue_client::chain(&e),
|
|
"swarm queue unreachable"
|
|
);
|
|
None
|
|
}
|
|
},
|
|
};
|
|
|
|
let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new(
|
|
hive_jobq::Graph::new(),
|
|
hive_jobq::resources::ResourceTable::new(),
|
|
)));
|
|
spawn_jobq_worker(Arc::clone(&jobq));
|
|
|
|
let state = AppState {
|
|
hives: Arc::new(load_hives()),
|
|
links: Arc::new(load_links()),
|
|
status,
|
|
jobq,
|
|
};
|
|
|
|
let (router, api) = OpenApiRouter::<AppState>::with_openapi(ApiDoc::openapi())
|
|
.routes(routes!(health))
|
|
.routes(routes!(get_hives))
|
|
.routes(routes!(get_hives_status))
|
|
.routes(routes!(get_links))
|
|
.routes(routes!(get_jobq_graph))
|
|
.routes(routes!(get_jobq_rollup))
|
|
.split_for_parts();
|
|
// Just the JSON, not the UI — Swagger UI itself is nginx-hosted from
|
|
// the nix store (see the module doc comment above). `api` is
|
|
// `Clone`; each request gets its own owned copy for `Json` to
|
|
// serialize, same as `hive-c0re::dashboard::serve`.
|
|
let app = router
|
|
.route(
|
|
"/api/openapi.json",
|
|
get(move || async move { Json(api.clone()) }),
|
|
)
|
|
.with_state(state);
|
|
axum::serve(listener, app)
|
|
.await
|
|
.context("serving swarm-controller")
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{
|
|
DEFAULT_SOCKET, HIVES_ENV, HiveEntry, LINKS_ENV, ServiceLink, load_hives, load_links,
|
|
};
|
|
use std::path::Path;
|
|
|
|
/// The socket must not share a directory with anything else, because
|
|
/// the socket is `0666` and the directory is therefore the only access
|
|
/// control it has. `/run/hyperhive` in particular holds hive-c0re's
|
|
/// **admin** socket, and nginx — a host service — is bounded by
|
|
/// nothing but the directory itself.
|
|
///
|
|
/// A test rather than a comment: the failure this guards against is a
|
|
/// one-word edit that looks tidier and reads fine in review.
|
|
#[test]
|
|
fn socket_lives_in_its_own_runtime_dir() {
|
|
let parent = Path::new(DEFAULT_SOCKET)
|
|
.parent()
|
|
.expect("socket path has a parent directory");
|
|
assert_eq!(
|
|
parent,
|
|
Path::new("/run/swarm-controller"),
|
|
"the socket's directory is its access control — moving it under a shared \
|
|
directory (notably /run/hyperhive, which holds the host admin socket) \
|
|
exposes everything else in that directory to the gateway's nginx"
|
|
);
|
|
}
|
|
|
|
/// One test, not three, deliberately: `HIVES_ENV` is a real process
|
|
/// env var, and `cargo test`'s default parallel runner would race
|
|
/// separate missing/malformed/valid tests against each other. Driving
|
|
/// all three states sequentially inside one test needs no mutex and
|
|
/// still proves each branch of `load_hives`.
|
|
///
|
|
/// SAFETY: single-threaded mutation of a process env var no other
|
|
/// test in this crate reads; restored (removed) before returning.
|
|
#[test]
|
|
fn load_hives_covers_missing_malformed_and_valid() {
|
|
unsafe {
|
|
std::env::remove_var(HIVES_ENV);
|
|
}
|
|
assert_eq!(
|
|
load_hives(),
|
|
Vec::<HiveEntry>::new(),
|
|
"unset env var is an empty directory, not a startup failure"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(HIVES_ENV, "not json");
|
|
}
|
|
assert_eq!(
|
|
load_hives(),
|
|
Vec::<HiveEntry>::new(),
|
|
"unparseable env var falls back to empty rather than panicking — \
|
|
/health must still answer even if this daemon's own config is wrong"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(
|
|
HIVES_ENV,
|
|
r#"[{"name":"pr1ma","domain":"pr1ma.example.com"},{"name":"umbra","domain":"umbra.example.com"}]"#,
|
|
);
|
|
}
|
|
assert_eq!(
|
|
load_hives(),
|
|
vec![
|
|
HiveEntry {
|
|
name: "pr1ma".to_string(),
|
|
domain: "pr1ma.example.com".to_string(),
|
|
},
|
|
HiveEntry {
|
|
name: "umbra".to_string(),
|
|
domain: "umbra.example.com".to_string(),
|
|
},
|
|
]
|
|
);
|
|
|
|
unsafe {
|
|
std::env::remove_var(HIVES_ENV);
|
|
}
|
|
}
|
|
|
|
/// Same three-state coverage as `load_hives_covers_missing_malformed_and_valid`,
|
|
/// same reason for one test rather than three (a shared process env var).
|
|
///
|
|
/// SAFETY: single-threaded mutation of a process env var no other test
|
|
/// in this crate reads; restored (removed) before returning.
|
|
#[test]
|
|
fn load_links_covers_missing_malformed_and_valid() {
|
|
unsafe {
|
|
std::env::remove_var(LINKS_ENV);
|
|
}
|
|
assert_eq!(
|
|
load_links(),
|
|
Vec::<ServiceLink>::new(),
|
|
"unset env var is an empty list, not a startup failure"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(LINKS_ENV, "not json");
|
|
}
|
|
assert_eq!(
|
|
load_links(),
|
|
Vec::<ServiceLink>::new(),
|
|
"unparseable env var falls back to empty rather than panicking"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(
|
|
LINKS_ENV,
|
|
r#"[{"label":"Authelia","icon":"🔑","url":"https://auth.example.com/"}]"#,
|
|
);
|
|
}
|
|
assert_eq!(
|
|
load_links(),
|
|
vec![ServiceLink {
|
|
label: "Authelia".to_string(),
|
|
icon: "🔑".to_string(),
|
|
url: "https://auth.example.com/".to_string(),
|
|
}]
|
|
);
|
|
|
|
unsafe {
|
|
std::env::remove_var(LINKS_ENV);
|
|
}
|
|
}
|
|
}
|