src/k8s_apply.rs
//! Async Kubernetes applier — the **network half** of the k8s-unfold target.
//!
//! Feature-gated (`kubernetes`) so the lean core stays sync + dep-free: with the
//! feature off this file is not compiled and neither `kube` nor `tokio` enter the
//! build (mirrors how `redfish`/`terraform`/`kvm` gate their network clients).
//!
//! It drives the manifests + verbs the **sync** [`crate::k8s::KubernetesRenderer`]
//! produced over the real Kubernetes API — server-side apply, scale, rollout-wait
//! — the systemd-over-D-Bus analog for k8s. There is **no `kubectl`/`helm`
//! shelling**: manifests apply through the typed/dynamic client.
//!
//! Scope: **platform bring-up only** (apply the workload/service/pvc + wait
//! ready). The APP DATA load (DB `COPY` / S3 objects) is `tillsynia-admin`'s job,
//! not Skidbladnir's.
use std::time::Duration;
use anyhow::{Context, Result, anyhow};
use k8s_openapi::api::apps::v1::{Deployment, StatefulSet};
use k8s_openapi::api::batch::v1::Job;
use kube::Client;
use kube::api::{Api, DeleteParams, DynamicObject, Patch, PatchParams, PostParams};
use kube::core::GroupVersionKind;
use kube::discovery::{Discovery, Scope};
use serde_json::{Value, json};
use crate::service::RenderedService;
/// The GVK of a rendered manifest, from its `apiVersion`/`kind`. `apiVersion`
/// splits `group/version` (core group is the empty string, e.g. `v1`).
fn gvk_of(api_version: &str, kind: &str) -> GroupVersionKind {
match api_version.split_once('/') {
Some((g, v)) => GroupVersionKind::gvk(g, v, kind),
None => GroupVersionKind::gvk("", api_version, kind),
}
}
/// Parse a rendered service's manifest files into typed-erased objects.
/// **Pure — no I/O**, so it is unit-tested without a cluster.
pub fn manifests(rendered: &RenderedService) -> Result<Vec<DynamicObject>> {
let mut out = Vec::with_capacity(rendered.files.len());
for f in &rendered.files {
let obj: DynamicObject = serde_yaml::from_str(&f.contents)
.with_context(|| format!("parse manifest {}", f.path))?;
out.push(obj);
}
Ok(out)
}
/// A connected applier (kubeconfig on a laptop, or in-cluster ServiceAccount
/// when it runs as a Job).
pub struct K8sApplier {
client: Client,
discovery: Discovery,
}
impl K8sApplier {
/// Connect + run API discovery (so any kind resolves to its resource).
pub async fn connect() -> Result<Self> {
let client = Client::try_default()
.await
.context("kube client (kubeconfig / in-cluster ServiceAccount)")?;
Self::from_client(client).await
}
/// Build an applier over an ALREADY-connected client (running API discovery on
/// it). The seam a consumer that owns its own [`Client`] — the Njord deployer,
/// which creates one with `Client::try_default()` for its whole apply path —
/// uses to drive this apply engine without opening a second connection.
pub async fn from_client(client: Client) -> Result<Self> {
let discovery = Discovery::new(client.clone())
.run()
.await
.context("kube API discovery")?;
Ok(Self { client, discovery })
}
/// Server-side apply one already-parsed manifest under `pp` (any kind,
/// resolved to its resource via discovery). The shared per-object core of
/// [`Self::apply`] and [`Self::apply_json_manifests`].
async fn apply_one(
&self,
obj: &DynamicObject,
namespace: &str,
pp: &PatchParams,
) -> Result<()> {
let tm = obj
.types
.clone()
.ok_or_else(|| anyhow!("manifest missing apiVersion/kind"))?;
let gvk = gvk_of(&tm.api_version, &tm.kind);
let (ar, caps) = self
.discovery
.resolve_gvk(&gvk)
.ok_or_else(|| anyhow!("cluster does not know kind {}", tm.kind))?;
let name = obj
.metadata
.name
.clone()
.ok_or_else(|| anyhow!("manifest {} missing metadata.name", tm.kind))?;
let api: Api<DynamicObject> = if caps.scope == Scope::Namespaced {
Api::namespaced_with(self.client.clone(), namespace, &ar)
} else {
Api::all_with(self.client.clone(), &ar)
};
api.patch(&name, pp, &Patch::Apply(obj))
.await
.with_context(|| format!("apply {} {}", tm.kind, name))?;
Ok(())
}
/// Server-side apply every manifest (any kind, resolved via discovery).
pub async fn apply(&self, rendered: &RenderedService, namespace: &str) -> Result<()> {
let pp = PatchParams::apply("skidbladnir").force();
for obj in manifests(rendered)? {
self.apply_one(&obj, namespace, &pp).await?;
}
Ok(())
}
/// Server-side apply a set of manifests supplied as `serde_json::Value`s (the
/// shape a JSON renderer emits), under a caller-chosen field `manager`.
///
/// This is the reusable apply engine a consumer with its own renderer drives
/// instead of re-implementing the server-side-apply loop: each object is
/// resolved to its resource via discovery and applied with `force`, exactly
/// like [`Self::apply`] — the only difference is that field ownership is
/// attributed to the caller's `manager` (e.g. `"njord-deployer"`), not to
/// `skidbladnir`, so the consumer stays the honest owner of the fields it
/// wrote. Resolving each kind through discovery sidesteps the client-side
/// plural-guessing pitfall (Ingress → `ingresses`, not `ingresss`).
pub async fn apply_json_manifests(
&self,
namespace: &str,
manager: &str,
objs: &[Value],
) -> Result<()> {
let pp = PatchParams::apply(manager).force();
for v in objs {
let obj: DynamicObject = serde_json::from_value(v.clone())
.context("parse manifest JSON into a typed-erased object")?;
self.apply_one(&obj, namespace, &pp).await?;
}
Ok(())
}
/// Patch a Deployment/StatefulSet replica count (the renderer's start/stop verbs).
pub async fn scale(
&self,
stateful: bool,
name: &str,
namespace: &str,
replicas: i32,
) -> Result<()> {
let patch = Patch::Merge(json!({ "spec": { "replicas": replicas } }));
let pp = PatchParams::default();
if stateful {
let api: Api<StatefulSet> = Api::namespaced(self.client.clone(), namespace);
api.patch(name, &pp, &patch)
.await
.with_context(|| format!("scale statefulset/{name}"))?;
} else {
let api: Api<Deployment> = Api::namespaced(self.client.clone(), namespace);
api.patch(name, &pp, &patch)
.await
.with_context(|| format!("scale deployment/{name}"))?;
}
Ok(())
}
async fn ready_replicas(&self, stateful: bool, name: &str, namespace: &str) -> Result<i32> {
if stateful {
let api: Api<StatefulSet> = Api::namespaced(self.client.clone(), namespace);
Ok(api
.get(name)
.await?
.status
.and_then(|s| s.ready_replicas)
.unwrap_or(0))
} else {
let api: Api<Deployment> = Api::namespaced(self.client.clone(), namespace);
Ok(api
.get(name)
.await?
.status
.and_then(|s| s.ready_replicas)
.unwrap_or(0))
}
}
/// Poll until `ready_replicas >= desired` or `timeout` elapses (the `status`
/// verb: rollout-wait).
pub async fn wait_ready(
&self,
stateful: bool,
name: &str,
namespace: &str,
desired: i32,
timeout: Duration,
) -> Result<()> {
let deadline = tokio::time::Instant::now() + timeout;
loop {
let ready = self.ready_replicas(stateful, name, namespace).await?;
if ready >= desired {
return Ok(());
}
if tokio::time::Instant::now() >= deadline {
return Err(anyhow!(
"timed out waiting for {name}: {ready}/{desired} ready"
));
}
tokio::time::sleep(Duration::from_secs(3)).await;
}
}
/// Delete every manifest (best-effort; unknown kinds / already-gone ignored).
pub async fn delete(&self, rendered: &RenderedService, namespace: &str) -> Result<()> {
for obj in manifests(rendered)? {
let Some(tm) = obj.types.clone() else {
continue;
};
let gvk = gvk_of(&tm.api_version, &tm.kind);
let Some((ar, caps)) = self.discovery.resolve_gvk(&gvk) else {
continue;
};
let Some(name) = obj.metadata.name.clone() else {
continue;
};
let api: Api<DynamicObject> = if caps.scope == Scope::Namespaced {
Api::namespaced_with(self.client.clone(), namespace, &ar)
} else {
Api::all_with(self.client.clone(), &ar)
};
let _ = api.delete(&name, &Default::default()).await; // best-effort
}
Ok(())
}
}
/// The terminal state of a [`run_job`] wait.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobOutcome {
/// The Job reported `status.succeeded >= 1`.
Succeeded,
/// The Job reported `status.failed >= 1`.
Failed,
/// Neither state was reached before the timeout elapsed.
TimedOut,
}
/// Run a one-shot Job to completion, replacing any prior Job of the same name
/// first — the "run a Job and WAIT for it before proceeding" step (e.g. a DB
/// migrate Job that must succeed before the app rolls out), the ordering a Helm
/// `pre-upgrade` hook gave for free.
///
/// It deletes an existing Job of the same name (foreground, so its pods go with
/// it) and waits for it to disappear so the create cannot `409`, creates this
/// one, then polls until it succeeds/fails or `timeout` elapses. Reusable by
/// design: it returns the [`JobOutcome`] rather than embedding any one product's
/// failure policy — the caller decides what `Failed`/`TimedOut` means (fetch pod
/// logs, bail with its own message, …).
pub async fn run_job(
client: &Client,
namespace: &str,
job: Value,
timeout: Duration,
) -> Result<JobOutcome> {
let name = job["metadata"]["name"]
.as_str()
.ok_or_else(|| anyhow!("Job manifest missing metadata.name"))?
.to_string();
let api: Api<Job> = Api::namespaced(client.clone(), namespace);
// before-hook-creation: remove the previous Job (and its pods) first.
if api.get_opt(&name).await?.is_some() {
let _ = api.delete(&name, &DeleteParams::foreground()).await;
// wait for it to actually be gone so create() doesn't 409.
for _ in 0..60 {
if api.get_opt(&name).await?.is_none() {
break;
}
tokio::time::sleep(Duration::from_secs(1)).await;
}
}
let obj: Job = serde_json::from_value(job).context("deserialize Job manifest")?;
api.create(&PostParams::default(), &obj)
.await
.with_context(|| format!("create Job/{name}"))?;
let deadline = tokio::time::Instant::now() + timeout;
loop {
let j = api
.get(&name)
.await
.with_context(|| format!("read Job/{name}"))?;
if let Some(st) = &j.status {
if st.succeeded.unwrap_or(0) >= 1 {
return Ok(JobOutcome::Succeeded);
}
if st.failed.unwrap_or(0) >= 1 {
return Ok(JobOutcome::Failed);
}
}
if tokio::time::Instant::now() >= deadline {
return Ok(JobOutcome::TimedOut);
}
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
/// Thin platform-install entrypoint: apply the rendered manifests and wait for
/// the workload to become ready. (App-data load is `tillsynia-admin`'s job.)
pub async fn k8s_install(
rendered: &RenderedService,
namespace: &str,
name: &str,
stateful: bool,
replicas: i32,
timeout: Duration,
) -> Result<()> {
let applier = K8sApplier::connect().await?;
applier.apply(rendered, namespace).await?;
applier
.wait_ready(stateful, name, namespace, replicas, timeout)
.await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::k8s::KubernetesRenderer;
use crate::service::{ServiceRenderer, ServiceSpec, Socket, Storage};
// Pure parse test — no cluster: the renderer's YAML round-trips into typed
// DynamicObjects with the right kind + name (what `apply` feeds the API).
#[test]
fn manifests_parse_web_deployment_and_service() {
let spec = ServiceSpec::new("web", "Tillsynia web", "/unused")
.image("registry.local:5000/njord-web:v1")
.namespace("njord-lab")
.replicas(2)
.socket(Socket::tcp(3000));
let rendered = KubernetesRenderer.render(&spec).unwrap();
let objs = manifests(&rendered).expect("parse");
assert_eq!(objs.len(), 2);
let kinds: Vec<String> = objs
.iter()
.map(|o| o.types.as_ref().unwrap().kind.clone())
.collect();
assert!(kinds.contains(&"Deployment".to_string()));
assert!(kinds.contains(&"Service".to_string()));
assert!(
objs.iter()
.all(|o| o.metadata.name.as_deref() == Some("web"))
);
}
#[test]
fn manifests_parse_stateful_with_pvc() {
let spec = ServiceSpec::new("postgres", "Postgres", "/unused")
.image("postgres:16")
.namespace("njord-lab")
.stateful(true)
.socket(Socket::tcp(5432))
.storage(Storage::new("8Gi", "/var/lib/postgresql/data"));
let rendered = KubernetesRenderer.render(&spec).unwrap();
let objs = manifests(&rendered).expect("parse");
assert_eq!(objs.len(), 3);
let kinds: Vec<String> = objs
.iter()
.map(|o| o.types.as_ref().unwrap().kind.clone())
.collect();
assert!(kinds.contains(&"StatefulSet".to_string()));
assert!(kinds.contains(&"Service".to_string()));
assert!(kinds.contains(&"PersistentVolumeClaim".to_string()));
}
#[test]
fn gvk_splits_group_version() {
assert_eq!(
gvk_of("apps/v1", "Deployment"),
GroupVersionKind::gvk("apps", "v1", "Deployment")
);
assert_eq!(
gvk_of("v1", "Service"),
GroupVersionKind::gvk("", "v1", "Service")
);
}
// Live smoke — requires a real cluster (the lab worker node) + kubeconfig.
// Run with: cargo test --features kubernetes -- --ignored k8s_live
#[tokio::test]
#[ignore]
async fn k8s_live_apply_scale_delete() {
let spec = ServiceSpec::new("skidbladnir-probe", "probe", "/unused")
.image("registry.k8s.io/pause:3.9")
.namespace("default")
.replicas(1);
let rendered = KubernetesRenderer.render(&spec).unwrap();
let a = K8sApplier::connect().await.expect("connect");
a.apply(&rendered, "default").await.expect("apply");
a.wait_ready(
false,
"skidbladnir-probe",
"default",
1,
Duration::from_secs(120),
)
.await
.expect("ready");
a.delete(&rendered, "default").await.expect("delete");
}
}