src/promote/release.rs
//! **Release operations** — install/upgrade, set-image, wait for rollout, roll
//! back on failure. Over the Kubernetes API, **never a shell**.
//!
//! # On Helm, honestly
//!
//! The tool this generalises shells `helm upgrade` — and its `--dry-run` still
//! spawned `helm version`, so a preview touched the world. Neither is acceptable
//! here. But the honest reason is worth stating plainly rather than being quietly
//! designed around: **rendering a Helm chart natively in Rust is not a small
//! thing.** A chart is Go templates plus the Sprig function library plus
//! `.Capabilities` plus `lookup` against a live cluster; there is no
//! production-grade pure-Rust implementation of that stack, and writing one is a
//! project, not a module.
//!
//! So `promote` does **not** reimplement Helm. It defines a seam with two sides:
//!
//! * [`ReleaseSource::Manifests`] — the **native** path. Objects the caller
//! already has (this crate's own [`crate::k8s::KubernetesRenderer`] produces
//! them, and so does `helm template` run by whoever owns the chart) are applied
//! server-side through `kube`. This is the sanctioned path, and it is what
//! [`KubeReleaseOps::apply`] implements.
//! * [`ReleaseSource::Chart`] — a **declaration**, not an execution. It records
//! which chart, which release name and which values a slot is supposed to be
//! running. The ship carries it, compares against it, and refuses to pretend it
//! applied it. What a chart path needs in order to become manifests — a
//! renderer, whether `helm template` in the caller's own pipeline or a future
//! engine — stays outside this crate.
//!
//! What *is* generic, shell-free and worth owning is the **read** side:
//! [`stored_release_values`] pulls a Helm release's stored user-supplied values
//! straight out of its `sh.helm.release.v1.*` Secret (base64 around a gzip stream
//! around JSON). That is exactly what `helm get values` would answer, obtained
//! without a Helm binary on `PATH`, without a subprocess whose stdout carries the
//! slot's secrets, and therefore runnable as a plain Kubernetes Job. It needs no
//! chart, only the API.
//!
//! # Every mutation demands a clearance
//!
//! The mutating methods take `&`[`Cleared`] — the token
//! [`Guard::assert`](super::Guard::assert) returns and nothing else can build.
//! "Did anyone check the fence?" is answered by the signature instead of by
//! reading each call site, and [`Cleared::require_covers`] stops a clearance for
//! one slot being spent on another.
use std::path::PathBuf;
use std::time::Duration;
use anyhow::Result;
use serde::{Deserialize, Serialize};
use super::Fut;
use super::guard::Cleared;
/// Which workload, in which namespace, and which container inside it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReleaseRef {
pub namespace: String,
/// The Deployment/StatefulSet name.
pub workload: String,
/// The container within the pod that carries the image.
pub container: String,
}
impl ReleaseRef {
pub fn new(namespace: &str, workload: &str, container: &str) -> Self {
ReleaseRef {
namespace: namespace.to_string(),
workload: workload.to_string(),
container: container.to_string(),
}
}
}
/// A chart the caller declares a slot is running. **Declarative only** — see the
/// module docs; the ship never renders or applies this.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChartRef {
/// Where the chart lives, in the caller's world.
pub path: PathBuf,
/// The chart's own name, as it appears in `chart.metadata.name` of a stored
/// release — the discriminator [`stored_release_values`] filters on when a
/// namespace carries more than one release.
pub chart_name: String,
/// The Helm release name.
pub release: String,
/// Values files, in order.
#[serde(default)]
pub values: Vec<PathBuf>,
}
/// Where a release's objects come from.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case", tag = "source")]
pub enum ReleaseSource {
/// Rendered manifests as YAML documents. The native path.
Manifests { docs: Vec<String> },
/// A chart, declared but not executed here.
Chart(ChartRef),
}
/// The cluster operations a promotion needs. One trait, so the sequencing above
/// it is testable without a cluster.
///
/// Async by way of [`Fut`] — a boxed `std` future — so the trait is object-safe
/// and the core pulls no async runtime.
pub trait ReleaseOps: Send + Sync {
/// The image a workload's container is running right now. `None` when the
/// workload or the container is not there. **Never a guess.**
fn current_image<'a>(&'a self, r: &'a ReleaseRef) -> Fut<'a, Result<Option<String>>>;
/// Patch the container's image.
fn set_image<'a>(
&'a self,
cleared: &'a Cleared,
r: &'a ReleaseRef,
image: &'a str,
) -> Fut<'a, Result<()>>;
/// Wait until the rollout is complete — updated == ready == available ==
/// desired — or time out.
fn wait_rollout<'a>(&'a self, r: &'a ReleaseRef, timeout: Duration) -> Fut<'a, Result<()>>;
/// Set a workload's replica count.
fn scale<'a>(
&'a self,
cleared: &'a Cleared,
namespace: &'a str,
workload: &'a str,
replicas: i32,
) -> Fut<'a, Result<()>>;
/// Wait until a workload reports `replicas` ready.
fn wait_replicas<'a>(
&'a self,
namespace: &'a str,
workload: &'a str,
replicas: i32,
timeout: Duration,
) -> Fut<'a, Result<()>>;
/// Every workload in a namespace — what `sleep`/`wake` operate on.
fn workloads<'a>(&'a self, namespace: &'a str) -> Fut<'a, Result<Vec<String>>>;
/// Install or upgrade a release. Returns one line per object applied.
fn apply<'a>(
&'a self,
cleared: &'a Cleared,
namespace: &'a str,
source: &'a ReleaseSource,
) -> Fut<'a, Result<Vec<String>>>;
/// Which namespace currently owns the route for `host`, if any.
fn host_owner<'a>(&'a self, host: &'a str) -> Fut<'a, Result<Option<String>>>;
/// Point `host` at a service in `namespace`.
fn route_host<'a>(
&'a self,
cleared: &'a Cleared,
host: &'a str,
namespace: &'a str,
service: &'a str,
port: i32,
tls_secret: &'a str,
) -> Fut<'a, Result<()>>;
/// Remove the route for `host` from `namespace`.
fn drop_route<'a>(
&'a self,
cleared: &'a Cleared,
host: &'a str,
namespace: &'a str,
) -> Fut<'a, Result<()>>;
}
/// What a rollout should do.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RolloutPlan {
pub target: ReleaseRef,
pub image: String,
/// How long to wait for the new pods.
pub wait: Duration,
/// …and for the old ones to come back if it has to roll back.
pub rollback_wait: Duration,
}
impl RolloutPlan {
pub fn new(target: ReleaseRef, image: &str) -> Self {
RolloutPlan {
target,
image: image.to_string(),
wait: Duration::from_secs(240),
rollback_wait: Duration::from_secs(180),
}
}
pub fn wait(mut self, d: Duration) -> Self {
self.wait = d;
self
}
pub fn rollback_wait(mut self, d: Duration) -> Self {
self.rollback_wait = d;
self
}
/// The dry run: what this would do, as lines. **Pure** — no cluster call, no
/// subprocess, no filesystem read. A preview cannot touch the world because
/// there is no code path here that could.
pub fn describe(&self) -> Vec<String> {
vec![
format!(
"read the current image of {}/{} [{}] — the rollback anchor",
self.target.namespace, self.target.workload, self.target.container
),
format!("set that container's image to {}", self.image),
format!(
"wait for the rollout to complete (≤{}s)",
self.wait.as_secs()
),
format!(
"on failure: put the anchor back and wait ≤{}s; if there was no anchor, \
say so rather than pretend a rollback happened",
self.rollback_wait.as_secs()
),
]
}
}
/// How a rollout ended. **Only [`RolloutOutcome::Rolled`] is success**, and the
/// three failure shapes are kept apart because they mean different things to the
/// person on call.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case", tag = "outcome")]
pub enum RolloutOutcome {
/// The new image is live.
Rolled { from: Option<String>, to: String },
/// The new image failed and the previous one is back. The environment is
/// serving; the deploy is not done.
RolledBack {
to_previous: String,
failure: String,
},
/// The new image failed and the rollback failed too. **The environment may be
/// down.** This is the case that must never be collapsed into "rollout
/// failed".
RollbackFailed {
failure: String,
rollback_failure: String,
},
/// The new image failed and there was nothing to roll back to — no anchor was
/// readable. Distinct from a successful rollback: nothing was restored.
NoAnchor { failure: String },
}
impl RolloutOutcome {
pub fn is_success(&self) -> bool {
matches!(self, RolloutOutcome::Rolled { .. })
}
/// Collapse to a `Result` for a caller that only wants "did it work".
pub fn into_result(self) -> Result<Self> {
match &self {
RolloutOutcome::Rolled { .. } => Ok(self),
RolloutOutcome::RolledBack {
to_previous,
failure,
} => anyhow::bail!(
"rollout failed and was rolled back to '{to_previous}'. \
The environment is serving the previous image.\n cause: {failure}"
),
RolloutOutcome::RollbackFailed {
failure,
rollback_failure,
} => anyhow::bail!(
"rollout failed AND the rollback failed — the environment may be DOWN.\n \
rollout cause: {failure}\n rollback cause: {rollback_failure}"
),
RolloutOutcome::NoAnchor { failure } => anyhow::bail!(
"rollout failed and there was no previous image to restore — nothing was \
rolled back.\n cause: {failure}"
),
}
}
/// One line for an operator.
pub fn line(&self) -> String {
match self {
RolloutOutcome::Rolled { from, to } => {
format!("rolled {} → {to}", from.as_deref().unwrap_or("(unknown)"))
}
RolloutOutcome::RolledBack {
to_previous,
failure,
} => {
format!("FAILED ({failure}) — rolled back to {to_previous}")
}
RolloutOutcome::RollbackFailed {
failure,
rollback_failure,
} => {
format!("FAILED ({failure}) and the ROLLBACK FAILED ({rollback_failure})")
}
RolloutOutcome::NoAnchor { failure } => {
format!("FAILED ({failure}) — no anchor, nothing restored")
}
}
}
}
/// **Set-image → wait → roll back on failure**, over any [`ReleaseOps`].
///
/// The anchor is read *first*, before anything is patched, because a rollback
/// needs to know where it is going back to and the moment after the patch is too
/// late to ask. An unreadable anchor is not fatal on its own — it becomes
/// [`RolloutOutcome::NoAnchor`] if the rollout then fails, which is an honest
/// statement that nothing was restored rather than a rollback that never happened
/// being reported as one.
pub async fn rollout(
ops: &dyn ReleaseOps,
cleared: &Cleared,
plan: &RolloutPlan,
) -> Result<RolloutOutcome> {
cleared.require_covers(&plan.target.namespace)?;
let anchor = ops.current_image(&plan.target).await.ok().flatten();
ops.set_image(cleared, &plan.target, &plan.image).await?;
let failure = match ops.wait_rollout(&plan.target, plan.wait).await {
Ok(()) => {
return Ok(RolloutOutcome::Rolled {
from: anchor,
to: plan.image.clone(),
});
}
Err(e) => format!("{e:#}"),
};
let Some(previous) = anchor else {
return Ok(RolloutOutcome::NoAnchor { failure });
};
if let Err(e) = ops.set_image(cleared, &plan.target, &previous).await {
return Ok(RolloutOutcome::RollbackFailed {
failure,
rollback_failure: format!("{e:#}"),
});
}
if let Err(e) = ops.wait_rollout(&plan.target, plan.rollback_wait).await {
return Ok(RolloutOutcome::RollbackFailed {
failure,
rollback_failure: format!("{e:#}"),
});
}
Ok(RolloutOutcome::RolledBack {
to_previous: previous,
failure,
})
}
/// Execute a generation plan ([`super::generation`]) over any [`ReleaseOps`].
///
/// The plan decided the order; this only walks it. Splitting them is what lets
/// the ordering — the part that has actually caused outages — be asserted in a
/// test with no cluster in sight.
pub async fn run_generation_plan(
ops: &dyn ReleaseOps,
cleared: &Cleared,
steps: &[super::generation::GenerationStep],
) -> Result<Vec<String>> {
use super::generation::GenerationStep as S;
let mut log = Vec::new();
for step in steps {
match step {
S::Scale {
namespace,
replicas,
} => {
cleared.require_covers(namespace)?;
for w in ops.workloads(namespace).await? {
ops.scale(cleared, namespace, &w, *replicas).await?;
log.push(format!("{namespace}/{w} → {replicas}"));
}
}
S::WaitReplicas {
namespace,
replicas,
timeout_secs,
} => {
for w in ops.workloads(namespace).await? {
ops.wait_replicas(namespace, &w, *replicas, Duration::from_secs(*timeout_secs))
.await?;
}
log.push(format!("{namespace}: {replicas} ready"));
}
S::PointHost {
host,
namespace,
service,
port,
tls_secret,
} => {
cleared.require_covers(namespace)?;
ops.route_host(cleared, host, namespace, service, *port, tls_secret)
.await?;
log.push(format!("{host} → {namespace}/{service}:{port}"));
}
S::DropHost { host, namespace } => {
// Deliberately NOT fatal. Traffic already goes to the new
// generation; a leftover duplicate route is something the ingress
// controller picks between, not a danger. It must still be said.
match ops.drop_route(cleared, host, namespace).await {
Ok(()) => log.push(format!("removed the route for {host} from {namespace}")),
Err(e) => log.push(format!(
"⚠ could not remove the route for {host} from {namespace} ({e:#}) — \
remove it by hand; traffic is already going the right way"
)),
}
}
S::NoOp { because } => log.push(format!("nothing to do — {because}")),
}
}
Ok(log)
}
// ─────────────────────────── the live implementation ───────────────────────────
/// The real thing, over `kube`. Feature `promote-k8s`.
#[cfg(feature = "promote-k8s")]
pub mod kube_ops {
use std::time::Duration;
use anyhow::{Context, Result, anyhow};
use k8s_openapi::api::apps::v1::Deployment;
use k8s_openapi::api::core::v1::Secret;
use k8s_openapi::api::networking::v1::Ingress;
use kube::Client;
use kube::api::{Api, DeleteParams, DynamicObject, ListParams, Patch, PatchParams};
use kube::core::GroupVersionKind;
use kube::discovery::{Discovery, Scope};
use serde_json::json;
use super::super::Fut;
use super::super::guard::Cleared;
use super::{ReleaseOps, ReleaseRef, ReleaseSource};
/// A connected client — kubeconfig on a laptop, in-cluster ServiceAccount as a
/// Job.
pub struct KubeReleaseOps {
client: Client,
discovery: Discovery,
/// The field manager server-side apply attributes changes to.
manager: String,
}
impl KubeReleaseOps {
pub async fn connect() -> Result<Self> {
let client = Client::try_default()
.await
.context("kube client (kubeconfig / in-cluster ServiceAccount)")?;
let discovery = Discovery::new(client.clone())
.run()
.await
.context("kube API discovery")?;
Ok(KubeReleaseOps {
client,
discovery,
manager: "skidbladnir-promote".to_string(),
})
}
/// Use an already-built client (a caller that owns its own connection).
pub async fn with_client(client: Client) -> Result<Self> {
let discovery = Discovery::new(client.clone())
.run()
.await
.context("kube API discovery")?;
Ok(KubeReleaseOps {
client,
discovery,
manager: "skidbladnir-promote".to_string(),
})
}
fn deployments(&self, ns: &str) -> Api<Deployment> {
Api::namespaced(self.client.clone(), ns)
}
}
async fn ready(api: &Api<Deployment>, name: &str) -> Result<(i32, i32, i32, i32)> {
let d = api
.get(name)
.await
.with_context(|| format!("get deployment/{name}"))?;
let desired = d.spec.as_ref().and_then(|s| s.replicas).unwrap_or(1);
let st = d.status.unwrap_or_default();
Ok((
desired,
st.updated_replicas.unwrap_or(0),
st.ready_replicas.unwrap_or(0),
st.available_replicas.unwrap_or(0),
))
}
impl ReleaseOps for KubeReleaseOps {
fn current_image<'a>(&'a self, r: &'a ReleaseRef) -> Fut<'a, Result<Option<String>>> {
Box::pin(async move {
let d = match self.deployments(&r.namespace).get(&r.workload).await {
Ok(d) => d,
Err(_) => return Ok(None),
};
Ok(d.spec
.and_then(|s| s.template.spec)
.and_then(|p| p.containers.into_iter().find(|c| c.name == r.container))
.and_then(|c| c.image))
})
}
fn set_image<'a>(
&'a self,
cleared: &'a Cleared,
r: &'a ReleaseRef,
image: &'a str,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
cleared.require_covers(&r.namespace)?;
let patch = json!({
"spec": { "template": { "spec": { "containers": [
{ "name": r.container, "image": image }
]}}}
});
self.deployments(&r.namespace)
.patch(
&r.workload,
&PatchParams::default(),
&Patch::Strategic(patch),
)
.await
.with_context(|| {
format!(
"set image on {}/{} [{}]",
r.namespace, r.workload, r.container
)
})?;
Ok(())
})
}
fn wait_rollout<'a>(&'a self, r: &'a ReleaseRef, timeout: Duration) -> Fut<'a, Result<()>> {
Box::pin(async move {
let api = self.deployments(&r.namespace);
let deadline = tokio::time::Instant::now() + timeout;
// Deliberately uninitialised: every path through the loop below
// sets it before the deadline check reads it, and a placeholder
// here would be a value that can never appear in a message.
let mut last: String;
loop {
match ready(&api, &r.workload).await {
Ok((desired, updated, rdy, avail)) => {
if updated == desired && rdy == desired && avail == desired {
return Ok(());
}
last = format!(
"desired={desired} updated={updated} ready={rdy} available={avail}"
);
}
Err(e) => last = format!("{e:#}"),
}
if tokio::time::Instant::now() >= deadline {
return Err(anyhow!(
"rollout of {}/{} did not complete within {}s — last seen: {last}",
r.namespace,
r.workload,
timeout.as_secs()
));
}
tokio::time::sleep(Duration::from_secs(3)).await;
}
})
}
fn scale<'a>(
&'a self,
cleared: &'a Cleared,
namespace: &'a str,
workload: &'a str,
replicas: i32,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
cleared.require_covers(namespace)?;
self.deployments(namespace)
.patch(
workload,
&PatchParams::default(),
&Patch::Merge(json!({ "spec": { "replicas": replicas } })),
)
.await
.with_context(|| format!("scale {namespace}/{workload} to {replicas}"))?;
Ok(())
})
}
fn wait_replicas<'a>(
&'a self,
namespace: &'a str,
workload: &'a str,
replicas: i32,
timeout: Duration,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
let api = self.deployments(namespace);
let deadline = tokio::time::Instant::now() + timeout;
loop {
let (_, _, rdy, _) = ready(&api, workload).await?;
if rdy == replicas {
return Ok(());
}
if tokio::time::Instant::now() >= deadline {
return Err(anyhow!(
"{namespace}/{workload}: {rdy} ready, wanted {replicas}, after {}s",
timeout.as_secs()
));
}
tokio::time::sleep(Duration::from_secs(2)).await;
}
})
}
fn workloads<'a>(&'a self, namespace: &'a str) -> Fut<'a, Result<Vec<String>>> {
Box::pin(async move {
let list = self
.deployments(namespace)
.list(&ListParams::default())
.await
.with_context(|| format!("list deployments in {namespace}"))?;
Ok(list.into_iter().filter_map(|d| d.metadata.name).collect())
})
}
fn apply<'a>(
&'a self,
cleared: &'a Cleared,
namespace: &'a str,
source: &'a ReleaseSource,
) -> Fut<'a, Result<Vec<String>>> {
Box::pin(async move {
cleared.require_covers(namespace)?;
let docs = match source {
ReleaseSource::Manifests { docs } => docs,
// The honest refusal. See the module docs: rendering a chart
// is a Go-template problem, and the ship will not shell out to
// solve it or pretend it did.
ReleaseSource::Chart(c) => {
return Err(anyhow!(
"cannot apply chart '{}' ({}): this crate does not render Helm \
charts and will not shell out to `helm`.\n \
TO FIX: render the chart where it is owned (`helm template` in \
your own pipeline, or this crate's KubernetesRenderer) and pass \
the result as ReleaseSource::Manifests. The chart declaration is \
still carried, compared and reported — it is simply not executed \
here.",
c.release,
c.path.display()
));
}
};
let pp = PatchParams::apply(&self.manager).force();
let mut applied = Vec::with_capacity(docs.len());
for doc in docs {
let obj: DynamicObject =
serde_yaml::from_str(doc).context("parse manifest document")?;
let tm = obj
.types
.clone()
.ok_or_else(|| anyhow!("manifest document has no apiVersion/kind"))?;
let gvk = match tm.api_version.split_once('/') {
Some((g, v)) => GroupVersionKind::gvk(g, v, &tm.kind),
None => GroupVersionKind::gvk("", &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 has no 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 {} {name}", tm.kind))?;
applied.push(format!("{} {name}", tm.kind));
}
Ok(applied)
})
}
fn host_owner<'a>(&'a self, host: &'a str) -> Fut<'a, Result<Option<String>>> {
Box::pin(async move {
let api: Api<Ingress> = Api::all(self.client.clone());
let list = api
.list(&ListParams::default())
.await
.context("list ingresses")?;
for ing in list {
let ns = ing.metadata.namespace.clone();
let hit = ing
.spec
.as_ref()
.and_then(|s| s.rules.as_ref())
.is_some_and(|rules| {
rules.iter().any(|r| {
r.host
.as_deref()
.is_some_and(|h| h.eq_ignore_ascii_case(host))
})
});
if hit {
return Ok(ns);
}
}
Ok(None)
})
}
fn route_host<'a>(
&'a self,
cleared: &'a Cleared,
host: &'a str,
namespace: &'a str,
service: &'a str,
port: i32,
tls_secret: &'a str,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
cleared.require_covers(namespace)?;
let api: Api<Ingress> = Api::namespaced(self.client.clone(), namespace);
let obj: Ingress = serde_json::from_value(json!({
"apiVersion": "networking.k8s.io/v1",
"kind": "Ingress",
"metadata": { "name": namespace, "namespace": namespace },
"spec": {
"tls": [ { "hosts": [host], "secretName": tls_secret } ],
"rules": [ { "host": host, "http": { "paths": [ {
"path": "/", "pathType": "Prefix",
"backend": { "service": {
"name": service, "port": { "number": port }
}}
} ]}} ]
}
}))?;
api.patch(
namespace,
&PatchParams::apply(&self.manager).force(),
&Patch::Apply(&obj),
)
.await
.with_context(|| format!("point {host} at {namespace}/{service}:{port}"))?;
Ok(())
})
}
fn drop_route<'a>(
&'a self,
cleared: &'a Cleared,
host: &'a str,
namespace: &'a str,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
let _ = host;
cleared.require_covers(namespace)?;
let api: Api<Ingress> = Api::namespaced(self.client.clone(), namespace);
api.delete(namespace, &DeleteParams::default())
.await
.with_context(|| format!("delete ingress {namespace}/{namespace}"))?;
Ok(())
})
}
}
/// **Read a Helm release's stored user-supplied values, without Helm.**
///
/// Helm 3 keeps every revision in a Secret named `sh.helm.release.v1.<release>.v<N>`
/// whose `release` field is base64 around a gzip stream around JSON. Reading it
/// directly is exactly what `helm get values` would answer, and it makes the
/// check runnable as a plain Kubernetes Job: no Helm binary in the image, no
/// `PATH` question, and no subprocess whose stdout carries the slot's secrets.
///
/// Only the **deployed** revision is read, and only releases whose chart is
/// `chart_name` — a namespace routinely carries more than one release, and the
/// others are different applications.
pub async fn stored_release_values(
client: Client,
namespace: &str,
chart_name: &str,
) -> Result<Option<serde_json::Value>> {
use std::io::Read;
let api: Api<Secret> = Api::namespaced(client, namespace);
let list = api
.list(&ListParams::default().labels("owner=helm,status=deployed"))
.await
.with_context(|| format!("list helm releases in {namespace}"))?;
let mut best: Option<(u64, serde_json::Value)> = None;
for s in list {
let Some(row) = s.data.as_ref().and_then(|d| d.get("release")) else {
continue;
};
// The outer base64 is Kubernetes' own and `kube` already stripped it;
// what is left is Helm's base64 around the gzip stream.
let Ok(gz) = base64_decode(&row.0) else {
continue;
};
let mut json = Vec::new();
if flate2::read::GzDecoder::new(&gz[..])
.read_to_end(&mut json)
.is_err()
{
continue;
}
let Ok(rel) = serde_json::from_slice::<serde_json::Value>(&json) else {
continue;
};
if rel.pointer("/chart/metadata/name").and_then(|v| v.as_str()) != Some(chart_name) {
continue;
}
let version = rel
.pointer("/version")
.and_then(|v| v.as_u64())
.unwrap_or(0);
let config = rel.get("config").cloned().unwrap_or(serde_json::json!({}));
if best.as_ref().is_none_or(|(v, _)| version > *v) {
best = Some((version, config));
}
}
Ok(best.map(|(_, c)| c))
}
fn base64_decode(bytes: &[u8]) -> Result<Vec<u8>> {
use base64::Engine as _;
base64::engine::general_purpose::STANDARD
.decode(
bytes
.iter()
.copied()
.filter(|b| !b.is_ascii_whitespace())
.collect::<Vec<u8>>(),
)
.context("helm release payload is not valid base64")
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::promote::generation::{GenerationStep, switch_plan};
use crate::promote::guard::Guard;
use crate::promote::registry::{SlotRegistry, TargetRef};
use crate::promote::schema::{Slot, SlotRole};
use std::cell::RefCell;
use std::future::Future;
use std::task::{Context, Poll, Waker};
/// A four-line executor. The fake below never yields, so every future is
/// `Ready` on its first poll — this needs no runtime, no threads, and
/// therefore does not put a `tokio` dependency into the lean core just to run
/// a test.
fn block_on<F: Future>(mut f: F) -> F::Output {
let mut f = unsafe { std::pin::Pin::new_unchecked(&mut f) };
let waker = Waker::noop();
let mut cx = Context::from_waker(waker);
loop {
if let Poll::Ready(v) = f.as_mut().poll(&mut cx) {
return v;
}
std::hint::spin_loop();
}
}
#[derive(Default)]
struct Fake {
calls: RefCell<Vec<String>>,
image: RefCell<Option<String>>,
/// Images whose rollout fails.
bad_images: Vec<String>,
/// Fail the rollback's wait too.
rollback_also_fails: bool,
workloads: Vec<String>,
host_owner: Option<String>,
drop_route_fails: bool,
}
// The fake is single-threaded and only ever touched from `block_on` above;
// `RefCell` is enough, and the unsafe impls keep the trait's `Send + Sync`
// bound (which the real, network-backed implementation genuinely needs).
unsafe impl Send for Fake {}
unsafe impl Sync for Fake {}
impl Fake {
fn log(&self, s: String) {
self.calls.borrow_mut().push(s);
}
fn calls(&self) -> Vec<String> {
self.calls.borrow().clone()
}
}
impl ReleaseOps for Fake {
fn current_image<'a>(&'a self, _r: &'a ReleaseRef) -> Fut<'a, Result<Option<String>>> {
let v = self.image.borrow().clone();
self.log(format!("current_image -> {v:?}"));
Box::pin(async move { Ok(v) })
}
fn set_image<'a>(
&'a self,
c: &'a Cleared,
r: &'a ReleaseRef,
image: &'a str,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
c.require_covers(&r.namespace)?;
self.log(format!("set_image {image}"));
*self.image.borrow_mut() = Some(image.to_string());
Ok(())
})
}
fn wait_rollout<'a>(&'a self, _r: &'a ReleaseRef, _t: Duration) -> Fut<'a, Result<()>> {
Box::pin(async move {
let cur = self.image.borrow().clone().unwrap_or_default();
let bad = self.bad_images.contains(&cur);
self.log(format!(
"wait_rollout {cur} -> {}",
if bad { "FAIL" } else { "ok" }
));
if bad || (self.rollback_also_fails && !self.bad_images.contains(&cur)) {
anyhow::bail!("pods never became ready on {cur}");
}
Ok(())
})
}
fn scale<'a>(
&'a self,
c: &'a Cleared,
ns: &'a str,
w: &'a str,
replicas: i32,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
c.require_covers(ns)?;
self.log(format!("scale {ns}/{w}={replicas}"));
Ok(())
})
}
fn wait_replicas<'a>(
&'a self,
ns: &'a str,
w: &'a str,
replicas: i32,
_t: Duration,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
self.log(format!("wait_replicas {ns}/{w}={replicas}"));
Ok(())
})
}
fn workloads<'a>(&'a self, ns: &'a str) -> Fut<'a, Result<Vec<String>>> {
let w = self.workloads.clone();
Box::pin(async move {
self.log(format!("workloads {ns}"));
Ok(w)
})
}
fn apply<'a>(
&'a self,
c: &'a Cleared,
ns: &'a str,
source: &'a ReleaseSource,
) -> Fut<'a, Result<Vec<String>>> {
Box::pin(async move {
c.require_covers(ns)?;
match source {
ReleaseSource::Manifests { docs } => {
self.log(format!("apply {} manifests to {ns}", docs.len()));
Ok(docs.iter().map(|_| "applied".to_string()).collect())
}
ReleaseSource::Chart(_) => anyhow::bail!("charts are not rendered here"),
}
})
}
fn host_owner<'a>(&'a self, _host: &'a str) -> Fut<'a, Result<Option<String>>> {
let o = self.host_owner.clone();
Box::pin(async move { Ok(o) })
}
fn route_host<'a>(
&'a self,
c: &'a Cleared,
host: &'a str,
ns: &'a str,
svc: &'a str,
port: i32,
_tls: &'a str,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
c.require_covers(ns)?;
self.log(format!("route {host} -> {ns}/{svc}:{port}"));
Ok(())
})
}
fn drop_route<'a>(
&'a self,
c: &'a Cleared,
host: &'a str,
ns: &'a str,
) -> Fut<'a, Result<()>> {
Box::pin(async move {
c.require_covers(ns)?;
self.log(format!("drop_route {host} from {ns}"));
if self.drop_route_fails {
anyhow::bail!("forbidden");
}
Ok(())
})
}
}
fn registry() -> SlotRegistry {
SlotRegistry::from_slots(vec![
Slot::new("prod", "njord", SlotRole::Production, true).host("g11.se"),
Slot::new("test1", "test1", SlotRole::Test, false)
.host("test1.example")
.promotes_to("prod"),
])
.unwrap()
}
fn cleared(r: &SlotRegistry) -> Cleared {
Guard::new(r).assert(&TargetRef::Slot("test1")).unwrap()
}
fn target() -> ReleaseRef {
ReleaseRef::new("test1", "test1-web", "web")
}
#[test]
fn a_good_rollout_reads_the_anchor_first_then_sets_then_waits() {
let r = registry();
let c = cleared(&r);
let fake = Fake {
image: RefCell::new(Some("app:v1".into())),
..Default::default()
};
let plan = RolloutPlan::new(target(), "app:v2");
let out = block_on(rollout(&fake, &c, &plan)).unwrap();
assert_eq!(
out,
RolloutOutcome::Rolled {
from: Some("app:v1".into()),
to: "app:v2".into()
}
);
assert!(out.is_success());
let calls = fake.calls();
assert!(
calls[0].starts_with("current_image"),
"the anchor must be read BEFORE anything is patched: {calls:?}"
);
assert_eq!(calls[1], "set_image app:v2");
assert!(calls[2].starts_with("wait_rollout app:v2"));
}
#[test]
fn a_failed_rollout_puts_the_anchor_back() {
let r = registry();
let c = cleared(&r);
let fake = Fake {
image: RefCell::new(Some("app:v1".into())),
bad_images: vec!["app:v2".into()],
..Default::default()
};
let out = block_on(rollout(&fake, &c, &RolloutPlan::new(target(), "app:v2"))).unwrap();
assert_eq!(
out,
RolloutOutcome::RolledBack {
to_previous: "app:v1".into(),
failure: "pods never became ready on app:v2".into()
}
);
assert!(!out.is_success(), "a rollback is not a successful deploy");
assert!(out.into_result().is_err());
assert_eq!(fake.image.borrow().as_deref(), Some("app:v1"));
}
/// The case that must never be collapsed into "rollout failed": the
/// environment may be down.
#[test]
fn a_failed_rollback_is_its_own_outcome() {
let r = registry();
let c = cleared(&r);
let fake = Fake {
image: RefCell::new(Some("app:v1".into())),
bad_images: vec!["app:v2".into(), "app:v1".into()],
..Default::default()
};
let out = block_on(rollout(&fake, &c, &RolloutPlan::new(target(), "app:v2"))).unwrap();
assert!(
matches!(out, RolloutOutcome::RollbackFailed { .. }),
"{out:?}"
);
let err = format!("{:#}", out.into_result().unwrap_err());
assert!(err.contains("may be DOWN"), "{err}");
}
/// No anchor is not a rollback. Reporting it as one would be a lie about what
/// is running.
#[test]
fn a_failure_with_no_readable_anchor_says_nothing_was_restored() {
let r = registry();
let c = cleared(&r);
let fake = Fake {
image: RefCell::new(None),
bad_images: vec!["app:v2".into()],
..Default::default()
};
let out = block_on(rollout(&fake, &c, &RolloutPlan::new(target(), "app:v2"))).unwrap();
assert!(matches!(out, RolloutOutcome::NoAnchor { .. }), "{out:?}");
assert!(out.line().contains("nothing restored"));
}
/// The clearance is not transferable — a token for `test1` cannot roll a
/// guarded namespace.
#[test]
fn a_clearance_for_one_slot_cannot_be_spent_on_another() {
let r = registry();
let c = cleared(&r);
let fake = Fake::default();
let plan = RolloutPlan::new(ReleaseRef::new("njord", "njord-web", "web"), "app:v2");
let err = block_on(rollout(&fake, &c, &plan)).expect_err("must refuse");
assert!(format!("{err:#}").contains("clearance mismatch"), "{err:#}");
assert!(
fake.calls().is_empty(),
"nothing may be touched before the check: {:?}",
fake.calls()
);
}
#[test]
fn the_dry_run_is_pure_and_says_what_it_would_do() {
let d = RolloutPlan::new(target(), "app:v2").describe();
assert!(d[0].contains("rollback anchor"));
assert!(
d.iter()
.any(|l| l.contains("set that container's image to app:v2"))
);
assert!(d.iter().any(|l| l.contains("on failure")));
}
// ── the generation plan, executed ────────────────────────────────────────
#[test]
fn a_switch_plan_executes_in_its_declared_order() {
let r = registry();
let c = cleared(&r);
let fake = Fake {
workloads: vec!["web".into()],
..Default::default()
};
let steps = switch_plan(
"test1",
Some("test1-20260715-0900"),
"test1-20260721-1432",
"test1.example",
"test1-20260721-1432",
3000,
"tls",
)
.unwrap();
block_on(run_generation_plan(&fake, &c, &steps)).unwrap();
let calls = fake.calls();
let idx = |needle: &str| calls.iter().position(|c| c.contains(needle)).expect(needle);
assert!(idx("scale test1-20260721-1432/web=1") < idx("wait_replicas"));
assert!(idx("wait_replicas") < idx("route test1.example"));
assert!(idx("route test1.example") < idx("drop_route"));
}
/// Removing the old route is best-effort — traffic is already going the right
/// way — but it must be *said*, not swallowed.
#[test]
fn a_failed_route_removal_is_reported_and_not_fatal() {
let r = registry();
let c = cleared(&r);
let fake = Fake {
workloads: vec!["web".into()],
drop_route_fails: true,
..Default::default()
};
let steps = switch_plan(
"test1",
Some("test1-20260715-0900"),
"test1-20260721-1432",
"test1.example",
"test1-20260721-1432",
3000,
"tls",
)
.unwrap();
let log = block_on(run_generation_plan(&fake, &c, &steps)).expect("not fatal");
assert!(
log.iter()
.any(|l| l.contains("could not remove the route") && l.contains("by hand")),
"{log:?}"
);
}
#[test]
fn a_no_op_plan_says_so_rather_than_looking_like_work() {
let r = registry();
let c = cleared(&r);
let fake = Fake::default();
let steps = vec![GenerationStep::NoOp {
because: "already there".into(),
}];
let log = block_on(run_generation_plan(&fake, &c, &steps)).unwrap();
assert_eq!(log, vec!["nothing to do — already there".to_string()]);
assert!(fake.calls().is_empty());
}
#[test]
fn a_chart_release_source_is_carried_but_never_executed() {
let r = registry();
let c = cleared(&r);
let fake = Fake::default();
let src = ReleaseSource::Chart(ChartRef {
path: PathBuf::from("deploy/helm/app"),
chart_name: "app".into(),
release: "test1".into(),
values: vec![],
});
assert!(block_on(fake.apply(&c, "test1", &src)).is_err());
// …and it round-trips, because carrying the declaration is the point.
let back: ReleaseSource =
serde_json::from_str(&serde_json::to_string(&src).unwrap()).unwrap();
assert_eq!(back, src);
}
#[test]
fn outcomes_round_trip_so_a_viewer_or_rpc_can_carry_them() {
let o = RolloutOutcome::RolledBack {
to_previous: "app:v1".into(),
failure: "timeout".into(),
};
let back: RolloutOutcome =
serde_json::from_str(&serde_json::to_string(&o).unwrap()).unwrap();
assert_eq!(o, back);
}
}