From 0459f155e72d61c6d927ee8ca9ae73dc0d5a62b1 Mon Sep 17 00:00:00 2001 From: Ben Howard Date: Mon, 10 Aug 2026 09:57:16 -0600 Subject: [PATCH 1/2] Cleanup: fix deletion hang Part 1 in a serious of deletion handling. Deleting a RestateDeployment hangs forever due to conflating is active to mean: * "is a service pointed at this deployment" * "are there active invocations" This change seperates those two facts and considers them differently depending on why the drain is happening (rollout vs deletion). Additionally, a few other issues feel out: 1. The drain requeue deadline was being thrown away do an error policy wrapping the finalizer. 2. A draining version kept its autoscaler. HPA removal has been moved to the point where we actually scale to zero. References #172 --- Cargo.lock | 107 ++++------ Cargo.toml | 6 +- .../172-restatedeployment-deletion-hang.md | 81 ++++++++ src/controllers/restatedeployment/cleanup.rs | 184 ++++++++++++++++++ .../restatedeployment/controller.rs | 175 ++++++++++++++--- src/controllers/restatedeployment/mod.rs | 1 + .../restatedeployment/reconcilers/knative.rs | 33 ++-- .../reconcilers/replicaset.rs | 54 ++--- src/lib.rs | 6 +- 9 files changed, 501 insertions(+), 146 deletions(-) create mode 100644 release-notes/unreleased/172-restatedeployment-deletion-hang.md create mode 100644 src/controllers/restatedeployment/cleanup.rs diff --git a/Cargo.lock b/Cargo.lock index 47c1284..5ae27d0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -279,7 +279,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -290,7 +290,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -845,7 +845,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1287,7 +1287,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.3", + "socket2 0.5.10", "tokio", "tower-service", "tracing", @@ -1800,7 +1800,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1836,20 +1836,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" -[[package]] -name = "opentelemetry" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0" -dependencies = [ - "futures-core", - "futures-sink", - "js-sys", - "pin-project-lite", - "thiserror 2.0.18", - "tracing", -] - [[package]] name = "opentelemetry" version = "0.32.0" @@ -1864,66 +1850,36 @@ dependencies = [ "tracing", ] -[[package]] -name = "opentelemetry-http" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d" -dependencies = [ - "async-trait", - "bytes", - "http 1.4.0", - "opentelemetry 0.31.0", - "reqwest", -] - [[package]] name = "opentelemetry-otlp" -version = "0.31.1" +version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" +checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" dependencies = [ "http 1.4.0", - "opentelemetry 0.31.0", - "opentelemetry-http", + "opentelemetry", "opentelemetry-proto", - "opentelemetry_sdk 0.31.0", + "opentelemetry_sdk", "prost", - "reqwest", "thiserror 2.0.18", "tokio", "tonic", - "tracing", + "tonic-types", ] [[package]] name = "opentelemetry-proto" -version = "0.31.0" +version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" +checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" dependencies = [ - "opentelemetry 0.31.0", - "opentelemetry_sdk 0.31.0", + "opentelemetry", + "opentelemetry_sdk", "prost", "tonic", "tonic-prost", ] -[[package]] -name = "opentelemetry_sdk" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd" -dependencies = [ - "futures-channel", - "futures-executor", - "futures-util", - "opentelemetry 0.31.0", - "percent-encoding", - "rand 0.9.4", - "thiserror 2.0.18", -] - [[package]] name = "opentelemetry_sdk" version = "0.32.1" @@ -1933,7 +1889,7 @@ dependencies = [ "futures-channel", "futures-executor", "futures-util", - "opentelemetry 0.32.0", + "opentelemetry", "percent-encoding", "portable-atomic", "rand 0.9.4", @@ -2256,7 +2212,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2 0.6.3", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -2293,7 +2249,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.3", + "socket2 0.5.10", "tracing", "windows-sys 0.60.2", ] @@ -2437,9 +2393,7 @@ checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ "base64", "bytes", - "futures-channel", "futures-core", - "futures-util", "http 1.4.0", "http-body", "http-body-util", @@ -2482,9 +2436,9 @@ dependencies = [ "http 1.4.0", "k8s-openapi", "kube", - "opentelemetry 0.31.0", + "opentelemetry", "opentelemetry-otlp", - "opentelemetry_sdk 0.32.1", + "opentelemetry_sdk", "prometheus", "prost", "regex", @@ -2547,7 +2501,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2878,7 +2832,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2940,7 +2894,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3218,6 +3172,17 @@ dependencies = [ "tonic-build", ] +[[package]] +name = "tonic-types" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a875a902255423d34c1f20838ab374126db8eb41625b7947a1d54113b0b7399" +dependencies = [ + "prost", + "prost-types", + "tonic", +] + [[package]] name = "tower" version = "0.5.3" @@ -3316,12 +3281,12 @@ dependencies = [ [[package]] name = "tracing-opentelemetry" -version = "0.32.1" +version = "0.33.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ac28f2d093c6c477eaa76b23525478f38de514fa9aeb1285738d4b97a9552fc" +checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26" dependencies = [ "js-sys", - "opentelemetry 0.31.0", + "opentelemetry", "smallvec", "tracing", "tracing-core", diff --git a/Cargo.toml b/Cargo.toml index 91acb4d..d7ebbbf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,9 +60,9 @@ fnv = "1.0.7" futures = "0.3.31" k8s-openapi = { version = "0.26.1", features = ["latest", "schemars"] } kube = { version = "2.0.1", features = ["runtime", "client", "derive", "unstable-runtime", ] } -opentelemetry = { version = "0.31.0", features = ["trace"] } +opentelemetry = { version = "0.32.0", features = ["trace"] } opentelemetry_sdk = "0.32.1" -opentelemetry-otlp = { version = "0.31.0", features = ["tokio", "grpc-tonic"], optional = true } +opentelemetry-otlp = { version = "0.32.0", default-features = false, features = ["tokio", "grpc-tonic", "trace"], optional = true } prometheus = "0.14.0" prost = "0.14.1" regex = "1.12.2" @@ -80,7 +80,7 @@ tonic = { version = "0.14.2", features = ["channel"] } tonic-prost = "0.14.2" tracing = "0.1.44" tracing-subscriber = { version = "0.3.22", features = ["json", "env-filter"] } -tracing-opentelemetry = "0.32.0" +tracing-opentelemetry = "0.33.0" url = { version = "2.5.7", features = ["serde"] } [build-dependencies] diff --git a/release-notes/unreleased/172-restatedeployment-deletion-hang.md b/release-notes/unreleased/172-restatedeployment-deletion-hang.md new file mode 100644 index 0000000..80dee47 --- /dev/null +++ b/release-notes/unreleased/172-restatedeployment-deletion-hang.md @@ -0,0 +1,81 @@ +# Release Notes for Issue #172: RestateDeployment deletion no longer hangs forever + +## Bug Fix + +### What Changed + +Deleting a `RestateDeployment` no longer blocks indefinitely on its own latest version. + +Cleanup previously asked Restate a single question — "is this deployment active?" — where +"active" meant *either* that a service still pointed at the deployment *or* that it had +unfinished invocations. During a rollout that conflation is harmless: a newer version +eventually takes over as the service's endpoint and the old one goes inactive. During a +deletion it was fatal. Nothing is coming to supersede the endpoint, so the latest version +stayed "active" forever, cleanup never deregistered it, `active_count` never reached zero, +and the finalizer requeued every 30 seconds without end. + +The operator now tracks the two facts separately and weighs them by why cleanup is +running. Being a service's current endpoint holds a version through a rollout but is +ignored during a deletion; only unfinished invocations — which drain on their own — can +hold a deletion. The version then goes through the normal drain, deregistration and +teardown path. + +Alongside that: + +- **Unfinished invocations not yet bound to a deployment now count.** Paused, queued and + scheduled work carries no `pinned_deployment_id`, so the old query scored it as zero and + a deletion could tear the endpoint out from under it. It is now attributed through the + target service. +- **A blocked deletion says what is blocking it.** The `DeploymentInUse` event now names + each version and its pinned/unpinned invocation counts instead of a generic message. +- **Drain deadlines are now honoured.** The requeue interval derived from a version's + remove-at time was being discarded, because errors reach the controller's error policy + wrapped by the finalizer machinery; a short `drainDelaySeconds` cost up to 30 seconds per + version regardless. It is now unwrapped, and floored at one second so a sub-second + deadline cannot spin the reconciler. +- **A draining version keeps its autoscaler.** Removal of an inactive version's + operator-managed HPA moved to the point where the version is actually scaled to zero. + Previously a deletion stripped the HPA from every version on the first reconcile, while + those versions were still serving traffic for the whole drain window. + +### Why This Matters + +Before this fix, `kubectl delete restatedeployment` never returned for a +`RestateDeployment` whose services were still registered — which is the normal state of any +healthy deployment. The only workaround was to remove the finalizer by hand, which left the +deployment registered in Restate with no pods behind it. Namespace deletion inherited the +same hang. + +### Impact on Users + +- **Existing deployments:** no configuration change. Deletions that were previously wedged + will proceed the next time the operator reconciles them; ones whose finalizer was removed + by hand may have left a stale registration in Restate that needs deleting via the admin + API. +- **Deletion now takes at least `spec.restate.drainDelaySeconds` (default 300s).** The + latest version is put through the same drain as any superseded version, so teardown waits + out the drain window even when the deployment never served an invocation. This is the + interval that was previously unbounded. +- **Deletion still waits on unfinished invocations, and that wait has no upper bound.** + Scheduled invocations are the sharp edge: a delayed call whose execution time is days out + counts as unfinished and holds the deletion until it fires. The `DeploymentInUse` event + reports the counts so this is diagnosable; a `deletionPolicy` field to bound or skip the + wait is planned. +- **New deployments:** no impact. + +### Migration Guidance + +None required. If a `RestateDeployment` is currently stuck deleting, upgrading the operator +is sufficient — no manual finalizer edits. + +To check what is holding a deletion: + +```bash +kubectl describe restatedeployment -n +# Warning FailedReconcile ... This RestateDeployment is backing active versions in +# Restate: greeter-7f9c4d (0 pinned, 3 unpinned invocations). ... +``` + +### Related Issues + +- Issue #172: RestateDeployment finalizer never completes diff --git a/src/controllers/restatedeployment/cleanup.rs b/src/controllers/restatedeployment/cleanup.rs new file mode 100644 index 0000000..8c71200 --- /dev/null +++ b/src/controllers/restatedeployment/cleanup.rs @@ -0,0 +1,184 @@ +//! Policy decisions shared by the ReplicaSet and Knative cleanup reconcilers: what +//! Restate still needs a registered deployment for, and how long we keep a drained +//! version around. + +use std::collections::HashMap; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum CleanupMode { + Rollout, + Deleting, +} + +impl CleanupMode { + pub(crate) fn for_rsd(rsd: &crate::resources::restatedeployments::RestateDeployment) -> Self { + if rsd.metadata.deletion_timestamp.is_some() { + Self::Deleting + } else { + Self::Rollout + } + } + + pub(crate) fn is_deleting(self) -> bool { + matches!(self, Self::Deleting) + } +} + +/// What Restate believes about one registered deployment. Kept as separate facts +/// because cleanup weighs them differently during a rollout and during a deletion. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct DeploymentUsage { + /// At least one service still points here. + pub latest_for_service: bool, + /// Unfinished invocations already bound to this deployment. + pub pinned_invocations: u64, + /// Unfinished invocations not yet bound to any deployment + pub unpinned_invocations: u64, +} + +pub(crate) type DeploymentUsageMap = HashMap; + +impl DeploymentUsage { + /// Keep whichever value argues more strongly for holding on to the deployment. + pub(crate) fn merge(&mut self, other: Self) { + self.latest_for_service |= other.latest_for_service; + self.pinned_invocations = self.pinned_invocations.max(other.pinned_invocations); + self.unpinned_invocations = self.unpinned_invocations.max(other.unpinned_invocations); + } + + /// Work that would be stranded by removing this deployment. Unlike being a service's + /// endpoint, this falls to zero on its own, so it is safe to wait on. + pub(crate) fn in_flight_invocations(&self) -> u64 { + self.pinned_invocations + self.unpinned_invocations + } + + /// Whether Restate still needs this deployment. + pub(crate) fn is_active(&self, mode: CleanupMode) -> bool { + match mode { + CleanupMode::Deleting => self.in_flight_invocations() > 0, + CleanupMode::Rollout => self.latest_for_service || self.in_flight_invocations() > 0, + } + } +} + +/// A version that cleanup could not remove because Restate still needs it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct BlockingVersion { + /// The ReplicaSet or Configuration name. + pub name: String, + pub usage: DeploymentUsage, +} + +/// Render blocking versions for the `DeploymentInUse` error, so a stuck deletion says +/// which version is holding it and what kind of work to go looking for. +pub(crate) fn describe_blocking_versions(blocking: &[BlockingVersion]) -> String { + blocking + .iter() + .map(|BlockingVersion { name, usage }| { + format!( + "{name} ({} pinned, {} unpinned invocations)", + usage.pinned_invocations, usage.unpinned_invocations + ) + }) + .collect::>() + .join(", ") +} + +/// Whether a drained, zero-scaled version should be kept for rollback. +pub(crate) fn retain_for_rollback( + mode: CleanupMode, + historic_count: i32, + revision_history_limit: i32, +) -> bool { + !mode.is_deleting() && historic_count < revision_history_limit +} + +#[cfg(test)] +mod tests { + use super::*; + + fn usage(latest: bool, pinned: u64, unpinned: u64) -> DeploymentUsage { + DeploymentUsage { + latest_for_service: latest, + pinned_invocations: pinned, + unpinned_invocations: unpinned, + } + } + + #[test] + fn latest_endpoint_is_active_during_rollout_but_not_during_deletion() { + let u = usage(true, 0, 0); + assert!(u.is_active(CleanupMode::Rollout)); + assert!(!u.is_active(CleanupMode::Deleting)); + } + + #[test] + fn pinned_invocations_block_deletion() { + let u = usage(true, 3, 0); + assert!(u.is_active(CleanupMode::Rollout)); + assert!(u.is_active(CleanupMode::Deleting)); + } + + #[test] + fn unpinned_invocations_block_deletion() { + // a pinned-only count reads paused/queued work as zero and deletes out from under it + let u = usage(true, 0, 80); + assert!(u.is_active(CleanupMode::Rollout)); + assert!(u.is_active(CleanupMode::Deleting)); + } + + #[test] + fn fully_drained_version_is_inactive_either_way() { + let u = usage(false, 0, 0); + assert!(!u.is_active(CleanupMode::Rollout)); + assert!(!u.is_active(CleanupMode::Deleting)); + } + + #[test] + fn superseded_version_with_in_flight_work_stays_active() { + let u = usage(false, 1, 0); + assert!(u.is_active(CleanupMode::Rollout)); + assert!(u.is_active(CleanupMode::Deleting)); + } + + #[test] + fn merge_takes_the_conservative_view() { + let mut u = usage(false, 1, 0); + u.merge(usage(true, 0, 5)); + assert_eq!(u, usage(true, 1, 5)); + } + + #[test] + fn rollback_retention_applies_during_normal_reconcile() { + assert!(retain_for_rollback(CleanupMode::Rollout, 0, 10)); + assert!(retain_for_rollback(CleanupMode::Rollout, 9, 10)); + assert!(!retain_for_rollback(CleanupMode::Rollout, 10, 10)); + } + + #[test] + fn rollback_retention_is_disabled_while_deleting() { + // retaining here would skip the deregistration below it and leak the registration + assert!(!retain_for_rollback(CleanupMode::Deleting, 0, 10)); + assert!(!retain_for_rollback(CleanupMode::Deleting, 9, 10)); + } + + #[test] + fn blocking_versions_name_the_work_holding_the_deletion() { + let described = describe_blocking_versions(&[ + BlockingVersion { + name: "greeter-abc123".into(), + usage: usage(true, 2, 0), + }, + BlockingVersion { + name: "greeter-def456".into(), + usage: usage(false, 0, 7), + }, + ]); + + assert_eq!( + described, + "greeter-abc123 (2 pinned, 0 unpinned invocations), \ + greeter-def456 (0 pinned, 7 unpinned invocations)" + ); + } +} diff --git a/src/controllers/restatedeployment/controller.rs b/src/controllers/restatedeployment/controller.rs index ee13d56..25eb6ac 100644 --- a/src/controllers/restatedeployment/controller.rs +++ b/src/controllers/restatedeployment/controller.rs @@ -46,6 +46,9 @@ use crate::telemetry; use crate::{Error, Result}; // Import our reconcilers +use crate::controllers::restatedeployment::cleanup::{ + CleanupMode, DeploymentUsage, DeploymentUsageMap, describe_blocking_versions, +}; use crate::controllers::restatedeployment::reconcilers; use super::reconcilers::replicaset::{ @@ -207,8 +210,31 @@ async fn reconcile(rs: Arc, ctx: Arc) -> Result(_rs: Arc, _: &Error, _ctx: C) -> Action { - Action::requeue(Duration::from_secs(30)) +/// Look through the `finalizer` wrapper `reconcile` puts on every error. +/// +/// Errors raised inside the finalizer closure reach `error_policy` as +/// `FinalizerError(CleanupFailed(..))`, so matching on the reconciler's own variants only +/// works after unwrapping — otherwise every arm below the wrapper is dead. +fn root_cause(err: &Error) -> &Error { + match err { + Error::FinalizerError(inner) => match inner.as_ref() { + kube::runtime::finalizer::Error::ApplyFailed(err) + | kube::runtime::finalizer::Error::CleanupFailed(err) => root_cause(err), + _ => err, + }, + err => err, + } +} + +fn error_policy(_rs: Arc, err: &Error, _ctx: C) -> Action { + match root_cause(err) { + // A drain knows its own deadline; the blanket interval would make a short + // drainDelaySeconds cost up to 30s per version anyway. + Error::DeploymentDraining { + requeue_after: Some(requeue_after), + } => Action::requeue(*requeue_after), + _ => Action::requeue(Duration::from_secs(30)), + } } impl RestateDeployment { @@ -432,7 +458,8 @@ impl RestateDeployment { )?, }; - let mut deployments = self.list_deployments(&ctx).await?; + // this path only runs for a live RestateDeployment; deletion goes to `cleanup`. + let mut deployments = self.list_deployments(&ctx, CleanupMode::Rollout).await?; let existing_deployment_id = replicaset .annotations() @@ -442,8 +469,7 @@ impl RestateDeployment { if existing_deployment_id.is_none_or(|existing_deployment_id| { !deployments .get(existing_deployment_id) - .cloned() - .unwrap_or_default() + .is_some_and(|usage| usage.is_active(CleanupMode::Rollout)) }) { let valid = async { if let Some(cluster_name) = &self.spec.restate.register.cluster { @@ -496,7 +522,13 @@ impl RestateDeployment { .await?; // if registration succeeded, treat this as an active endpoint // if we fail after this point we will re-register and should get the same deployment id - deployments.insert(deployment_id.clone(), true); + deployments.insert( + deployment_id.clone(), + DeploymentUsage { + latest_for_service: true, + ..Default::default() + }, + ); debug!( "Updating deployment-id annotation of ReplicaSet/Service {versioned_name} in namespace {namespace}" @@ -940,23 +972,47 @@ impl RestateDeployment { Ok(resp.id) } - pub(super) async fn list_deployments(&self, ctx: &Context) -> Result> { - // This query finds deployments, noting those that are the latest for a particular service, or have an active invocation - let sql_query = r#" - WITH active_deployments AS ( - SELECT DISTINCT deployment_id as id + pub(super) async fn list_deployments( + &self, + ctx: &Context, + mode: CleanupMode, + ) -> Result { + // The unpinned count is attributed through `sys_service.deployment_id`, the same + // column that sets `latest_for_service`, so a non-zero count implies that flag. + let count_unpinned = if mode.is_deleting() { "1 = 1" } else { "1 = 0" }; + + let sql_query = format!( + r#" + WITH latest AS ( + SELECT DISTINCT deployment_id AS id FROM sys_service WHERE deployment_id IS NOT NULL - UNION - SELECT DISTINCT pinned_deployment_id as id + ), + pinned AS ( + SELECT pinned_deployment_id AS id, COUNT(*) AS n FROM sys_invocation_status WHERE pinned_deployment_id IS NOT NULL AND status != 'completed' + GROUP BY pinned_deployment_id + ), + unpinned AS ( + SELECT s.deployment_id AS id, COUNT(*) AS n + FROM sys_invocation_status i + JOIN sys_service s ON s.name = i.target_service_name + WHERE {count_unpinned} + AND i.pinned_deployment_id IS NULL + AND i.status != 'completed' + GROUP BY s.deployment_id ) - SELECT d.id as deployment_id, - a.id IS NOT NULL as active + SELECT d.id AS deployment_id, + l.id IS NOT NULL AS latest_for_service, + COALESCE(p.n, 0) AS pinned_invocations, + COALESCE(u.n, 0) AS unpinned_invocations FROM sys_deployment d - LEFT JOIN active_deployments a ON d.id = a.id; - "#; + LEFT JOIN latest l ON d.id = l.id + LEFT JOIN pinned p ON d.id = p.id + LEFT JOIN unpinned u ON d.id = u.id; + "# + ); #[derive(Deserialize)] struct DeploymentQueryResult { @@ -966,7 +1022,9 @@ impl RestateDeployment { #[derive(Deserialize)] struct DeploymentQueryResultRow { deployment_id: String, - active: bool, + latest_for_service: bool, + pinned_invocations: i64, + unpinned_invocations: i64, } let resp = ctx @@ -984,19 +1042,25 @@ impl RestateDeployment { .await .map_err(Error::AdminCallFailed)?; - let mut endpoints = HashMap::with_capacity(response.rows.len()); + let mut endpoints: DeploymentUsageMap = HashMap::with_capacity(response.rows.len()); for row in response.rows { + let usage = DeploymentUsage { + latest_for_service: row.latest_for_service, + pinned_invocations: row.pinned_invocations.max(0) as u64, + unpinned_invocations: row.unpinned_invocations.max(0) as u64, + }; + match endpoints.entry(row.deployment_id) { std::collections::hash_map::Entry::Occupied(mut entry) => { - // two rows with same deployment id shouldnt happen... - // we treat the deployment as active if any row is active - if !entry.get() { - entry.insert(row.active); - } + // two rows for one deployment id shouldnt happen: `latest` is DISTINCT, + // `pinned`/`unpinned` are grouped, sys_deployment holds one row per id, + // and all three are LEFT JOINed. we take the most conservative view of + // each fact if a future query shape ever does produce duplicates. + entry.get_mut().merge(usage); } std::collections::hash_map::Entry::Vacant(entry) => { - entry.insert(row.active); + entry.insert(usage); } } } @@ -1038,7 +1102,7 @@ impl RestateDeployment { }; } - let deployments = self.list_deployments(&ctx).await?; + let deployments = self.list_deployments(&ctx, CleanupMode::Deleting).await?; let my_uid = self.uid().expect("RestateDeployment to have a uid"); @@ -1048,7 +1112,7 @@ impl RestateDeployment { Some(crate::resources::restatedeployments::DeploymentMode::Knative) ); - let (active_count, next_removal) = if is_knative { + let (blocking, next_removal) = if is_knative { // Knative cleanup path reconcilers::knative::cleanup_old_configurations( namespace, @@ -1073,13 +1137,20 @@ impl RestateDeployment { .await? }; - if active_count > 0 { + if !blocking.is_empty() { + let blocked_by = describe_blocking_versions(&blocking); + debug!( - "Cannot process deletion of RestateDeployment '{}' from Restate as there are {} active deployments that rely on it", + "Cannot process deletion of RestateDeployment '{}' from Restate as {} version(s) still have unfinished invocations: {blocked_by}", self.name_any(), - active_count + blocking.len(), ); - return Err(Error::DeploymentInUse); + + // Named versions and counts rather than the bare message: `reconcile` publishes + // this as a Warning event, and it is the only place a stuck deletion explains + // itself. Unpinned work includes scheduled invocations, whose execution time + // can be arbitrarily far out, so "wait for it" is not always sound advice. + return Err(Error::DeploymentInUse { blocked_by }); } if let Some(next_removal) = next_removal { @@ -1088,7 +1159,10 @@ impl RestateDeployment { self.name_any() ); - let secs_until_next_removal = (next_removal - chrono::Utc::now()).num_seconds().max(0); + // Floor at 1s: `num_seconds` truncates, so a deadline under a second away would + // otherwise requeue with no delay and spin the reconciler — each turn of which + // re-runs the admin query above — until the deadline passes. + let secs_until_next_removal = (next_removal - chrono::Utc::now()).num_seconds().max(1); return Err(Error::DeploymentDraining { requeue_after: Some(Duration::from_secs(secs_until_next_removal as u64)), @@ -1422,6 +1496,43 @@ mod tests { use super::*; use serde_json::json; + /// `reconcile` hands every error to the error policy wrapped by the finalizer + /// machinery, so a policy that matches on the reconciler's own variants only fires if + /// it looks through the wrapper first. + #[test] + fn drain_requeue_survives_the_finalizer_wrapper() { + let draining = || Error::DeploymentDraining { + requeue_after: Some(Duration::from_secs(7)), + }; + + let wrapped = Error::FinalizerError(Box::new( + kube::runtime::finalizer::Error::CleanupFailed(draining()), + )); + + assert_eq!( + error_policy(Arc::new(()), &wrapped, ()), + Action::requeue(Duration::from_secs(7)) + ); + assert_eq!( + error_policy(Arc::new(()), &draining(), ()), + Action::requeue(Duration::from_secs(7)) + ); + } + + #[test] + fn unrelated_errors_keep_the_blanket_interval() { + let wrapped = Error::FinalizerError(Box::new( + kube::runtime::finalizer::Error::CleanupFailed(Error::DeploymentInUse { + blocked_by: "greeter-abc123 (1 pinned, 0 unpinned invocations)".into(), + }), + )); + + assert_eq!( + error_policy(Arc::new(()), &wrapped, ()), + Action::requeue(Duration::from_secs(30)) + ); + } + /// Build a minimal ReplicaSet-mode RestateDeployment for selector tests. fn make_rsd(match_labels: Option<&[(&str, &str)]>, image: &str) -> RestateDeployment { let selector = match_labels.map(|labels| { diff --git a/src/controllers/restatedeployment/mod.rs b/src/controllers/restatedeployment/mod.rs index 90f83d8..163dea6 100644 --- a/src/controllers/restatedeployment/mod.rs +++ b/src/controllers/restatedeployment/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod cleanup; pub mod controller; mod reconcilers; diff --git a/src/controllers/restatedeployment/reconcilers/knative.rs b/src/controllers/restatedeployment/reconcilers/knative.rs index 4ea5993..8b3503c 100644 --- a/src/controllers/restatedeployment/reconcilers/knative.rs +++ b/src/controllers/restatedeployment/reconcilers/knative.rs @@ -8,6 +8,9 @@ use serde_json::json; use tracing::*; use url::Url; +use crate::controllers::restatedeployment::cleanup::{ + BlockingVersion, CleanupMode, DeploymentUsageMap, retain_for_rollback, +}; use crate::controllers::restatedeployment::controller::{ Context, RESTATE_DEPLOYMENT_ID_ANNOTATION, }; @@ -220,7 +223,8 @@ pub async fn reconcile_knative( annotate_configuration(ctx, namespace, &config, &deployment_id).await?; // Cleanup old Configurations (mirrors ReplicaSet cleanup pattern) - let deployments = rsd.list_deployments(ctx).await?; + // this path only runs for a live RestateDeployment; deletion goes to `cleanup`. + let deployments = rsd.list_deployments(ctx, CleanupMode::Rollout).await?; let rsd_uid = rsd .uid() .ok_or_else(|| Error::InvalidRestateConfig("RestateDeployment must have UID".into()))?; @@ -743,9 +747,9 @@ pub async fn cleanup_old_configurations( ctx: &Context, rsd_uid: &str, rsd: &RestateDeployment, - deployments: &std::collections::HashMap, + deployments: &DeploymentUsageMap, active_tag: Option<&str>, -) -> Result<(i32, Option>)> { +) -> Result<(Vec, Option>)> { // Use reflector cache instead of API list() call let configurations_cell = std::cell::Cell::new(Vec::new()); @@ -800,14 +804,17 @@ pub async fn cleanup_old_configurations( .cmp(&a.metadata.creation_timestamp) }); - // keep track of how many configurations there are that are still in-use by restate (active services or invocations) - let mut active_count = 0; + // keep track of the configurations that are still in-use by restate (active services or invocations) + let mut blocking = Vec::new(); // Keep track of how many zero-scaled configurations there are (for revision history limit) let mut historic_count = 0; let mut next_removal = None; let now = chrono::Utc::now(); + // As in `cleanup_old_replicasets`. + let mode = CleanupMode::for_rsd(rsd); + for config in configurations { let config_name = config.name_any(); @@ -819,12 +826,14 @@ pub async fn cleanup_old_configurations( // Skip active deployments let deployment = config_deployment_id - .and_then(|config_deployment_id| deployments.get(config_deployment_id).cloned()); + .and_then(|config_deployment_id| deployments.get(config_deployment_id).copied()); let deployment_exists = deployment.is_some(); - let deployment_active = deployment.unwrap_or(false); - if deployment_active { - active_count += 1; + if let Some(usage) = deployment.filter(|usage| usage.is_active(mode)) { + blocking.push(BlockingVersion { + name: config_name.clone(), + usage, + }); if config .metadata @@ -887,7 +896,7 @@ pub async fn cleanup_old_configurations( ) { (_, true, _) | (_, _, false) => { // we are past the remove-at time, or the endpoint was removed by other means; can now delete it (subject to the history limit) - if historic_count < rsd.spec.revision_history_limit { + if retain_for_rollback(mode, historic_count, rsd.spec.revision_history_limit) { historic_count += 1; trace!( "Keeping old Configuration {} in namespace {namespace} (within revision history limit: {}/{})", @@ -987,7 +996,7 @@ pub async fn cleanup_old_configurations( // If there are active old deployments still draining but no removal is yet scheduled, // requeue on a short poll interval to detect drain completion promptly. - if active_count > 0 && next_removal.is_none() { + if !blocking.is_empty() && next_removal.is_none() { let poll_seconds = 10; next_removal = Some( chrono::Utc::now() @@ -996,7 +1005,7 @@ pub async fn cleanup_old_configurations( ); } - Ok((active_count, next_removal)) + Ok((blocking, next_removal)) } /// Get tag from Configuration annotation diff --git a/src/controllers/restatedeployment/reconcilers/replicaset.rs b/src/controllers/restatedeployment/reconcilers/replicaset.rs index 23266d4..da28452 100644 --- a/src/controllers/restatedeployment/reconcilers/replicaset.rs +++ b/src/controllers/restatedeployment/reconcilers/replicaset.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, HashMap}; +use std::collections::BTreeMap; use k8s_openapi::api::apps::v1::ReplicaSet; use k8s_openapi::api::autoscaling::v2::HorizontalPodAutoscaler; @@ -15,6 +15,9 @@ use reqwest::Method; use serde_json::json; use tracing::*; +use crate::controllers::restatedeployment::cleanup::{ + BlockingVersion, CleanupMode, DeploymentUsageMap, retain_for_rollback, +}; use crate::controllers::restatedeployment::controller::{ APP_MANAGED_BY_LABEL, Context, OWNED_BY_LABEL, RESTATE_DEPLOYMENT_ID_ANNOTATION, }; @@ -274,9 +277,9 @@ pub async fn cleanup_old_replicasets( rs_api: &Api, rsd_uid: &str, rsd: &RestateDeployment, - deployments: &HashMap, + deployments: &DeploymentUsageMap, except_rs: Option<&str>, -) -> Result<(i32, Option>)> { +) -> Result<(Vec, Option>)> { let replicasets_cell = std::cell::Cell::new(Vec::new()); let _ = ctx.replicasets_store.find(|rs| { @@ -322,14 +325,16 @@ pub async fn cleanup_old_replicasets( .cmp(&a.metadata.creation_timestamp) }); - // keep track of how many rs there are that are still in-use by restate (active services or invocations) - let mut active_count = 0; + // keep track of the rs that are still in-use by restate (active services or invocations) + let mut blocking = Vec::new(); // Keep track of how many zero-scaled rs there are (for revision history limit) let mut historic_count = 0; let mut next_removal = None; let now = chrono::Utc::now(); + let mode = CleanupMode::for_rsd(rsd); + for rs in replicasets { let rs_name = rs.name_any(); @@ -337,12 +342,14 @@ pub async fn cleanup_old_replicasets( // Skip active deployments let deployment = rs_deployment_id - .and_then(|rs_deployment_id| deployments.get(rs_deployment_id).cloned()); + .and_then(|rs_deployment_id| deployments.get(rs_deployment_id).copied()); let deployment_exists = deployment.is_some(); - let deployment_active = deployment.unwrap_or(false); - if deployment_active { - active_count += 1; + if let Some(usage) = deployment.filter(|usage| usage.is_active(mode)) { + blocking.push(BlockingVersion { + name: rs_name.clone(), + usage, + }); // Per-version autoscaling: a non-latest version has an operator HPA // iff it is still active and autoscaling is configured. (See @@ -456,19 +463,6 @@ pub async fn cleanup_old_replicasets( continue; } - // Non-active version: remove any operator HPA before scaling it down. Its - // minReplicas floor (>= 1) would otherwise fight the operator scaling the - // ReplicaSet to zero, and would hold the version at the floor through the - // revision-history retention window. Gate on the owned-HPA cache so we - // only call the API when an HPA actually exists (avoids a wasted delete - // per inactive ReplicaSet on every reconcile); this still fires whenever - // an HPA is present, including during RestateDeployment deletion, so - // teardown is never contended. - let hpa_ref = ObjectRef::::new(&rs_name).within(namespace); - if ctx.hpa_store.get(&hpa_ref).is_some() { - super::autoscaling::delete_version_hpa(&ctx.client, namespace, &rs_name).await?; - } - let current_remove_at = rs .annotations() .get(RESTATE_REMOVE_VERSION_AT_ANNOTATION) @@ -488,6 +482,16 @@ pub async fn cleanup_old_replicasets( (_, true, _) | (_, _, false) => { // we are past the remove at time, or the endpoint was removed by other means; can now scale it down + // Remove any operator HPA first: its minReplicas floor (>= 1) would fight + // the scale to zero below, and would hold the version at the floor through + // the revision-history retention window. Gate on the owned-HPA cache so we + // only call the API when an HPA actually exists. + let hpa_ref = ObjectRef::::new(&rs_name).within(namespace); + if ctx.hpa_store.get(&hpa_ref).is_some() { + super::autoscaling::delete_version_hpa(&ctx.client, namespace, &rs_name) + .await?; + } + // If this version has active pods, scale it down to 0 first if rs .spec @@ -516,7 +520,7 @@ pub async fn cleanup_old_replicasets( } // If we are here, there is a 0 sized replicaset which should be subject to the history limit - if historic_count < rsd.spec.revision_history_limit { + if retain_for_rollback(mode, historic_count, rsd.spec.revision_history_limit) { historic_count += 1; // we haven't hit that limit yet, so we don't need to delete this rs continue; @@ -606,7 +610,7 @@ pub async fn cleanup_old_replicasets( // If there are active old deployments still draining but no removal is yet scheduled, // requeue on a short poll interval to detect drain completion promptly. - if active_count > 0 && next_removal.is_none() { + if !blocking.is_empty() && next_removal.is_none() { let poll_seconds = 10; next_removal = Some( chrono::Utc::now() @@ -615,7 +619,7 @@ pub async fn cleanup_old_replicasets( ); } - Ok((active_count, next_removal)) + Ok((blocking, next_removal)) } #[cfg(test)] diff --git a/src/lib.rs b/src/lib.rs index 0c0f28e..321dce6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -68,9 +68,9 @@ pub enum Error { HashCollision, #[error( - "This RestateDeployment is backing active versions in Restate. If you want to delete the RestateDeployment, either register new endpoints for the relevant services or delete the Restate versions." + "This RestateDeployment is backing active versions in Restate: {blocked_by}. If you want to delete the RestateDeployment, either register new endpoints for the relevant services or cancel/complete the outstanding invocations." )] - DeploymentInUse, + DeploymentInUse { blocked_by: String }, #[error( "This RestateDeployment is backing recently-active versions in Restate. It will be removed after the drain delay period." @@ -114,7 +114,7 @@ impl Error { Error::AdminCallFailed(_) => "AdminCallFailed", Error::AdminCallRejected { .. } => "AdminCallRejected", Error::HashCollision => "HashCollision", - Error::DeploymentInUse => "DeploymentInUse", + Error::DeploymentInUse { .. } => "DeploymentInUse", Error::DeploymentDraining { .. } => "DeploymentDraining", Error::ConfigurationNotReady { .. } => "ConfigurationNotReady", Error::RouteNotReady { .. } => "RouteNotReady", From fb57caa06dc4b7efc00560fcafa6c249f5e7c359 Mon Sep 17 00:00:00 2001 From: Ben Howard Date: Mon, 10 Aug 2026 12:59:47 -0600 Subject: [PATCH 2/2] Cleanup: address review on #176 - Build the deployment-usage query in Rust rather than sharing one string behind a `WHERE 1 = 0` guard. DataFusion does fold the guard away (EXPLAIN against 1.7.2 shows the unpinned scan disappearing), but the reconcile path should not depend on an optimiser pass across server versions we don't pin. The rollout flavour now scans sys_invocation_status once, with projection [status, pinned_deployment_id] -- no target_service_name decode. - Restore the note explaining the COALESCE/LEFT JOIN pairing, and move the row parsing next to the query it belongs to. - Back off a blocked deletion: retries stretch from 30s to a 5m cap as the wait grows, instead of re-running the deleting query every 30s forever. - Cover the teardown ordering with mocked-apiserver tests: the HPA comes off immediately before the scale to zero, and not while a version is still waiting out its drain deadline (in a rollout or a deletion). - Run cargo test in CI, which was not running tests at all. --- .github/workflows/ci.yml | 12 + justfile | 3 + .../172-restatedeployment-deletion-hang.md | 15 +- src/controllers/restatedeployment/cleanup.rs | 180 ++++++++- .../restatedeployment/controller.rs | 125 +++---- .../reconcilers/replicaset.rs | 353 ++++++++++++++++++ src/lib.rs | 5 +- 7 files changed, 607 insertions(+), 86 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ccf079b..15c3291 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,6 +22,18 @@ jobs: - uses: Swatinem/rust-cache@v2 - run: cargo clippy --all-targets --all-features -- -D warnings + test: + name: Test + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - name: Install pinned Rust toolchain + run: rustup show active-toolchain || rustup toolchain install + - name: Install protoc + uses: restatedev/restate/.github/actions/install-protoc@main + - uses: Swatinem/rust-cache@v2 + - run: cargo test --all-targets + build-image: name: Build release Docker image runs-on: ubuntu-latest diff --git a/justfile b/justfile index 2f94645..b890196 100644 --- a/justfile +++ b/justfile @@ -123,4 +123,7 @@ fmt: lint: cargo clippy +test: + cargo test --all-targets + check: fmt lint diff --git a/release-notes/unreleased/172-restatedeployment-deletion-hang.md b/release-notes/unreleased/172-restatedeployment-deletion-hang.md index 80dee47..65fe6cb 100644 --- a/release-notes/unreleased/172-restatedeployment-deletion-hang.md +++ b/release-notes/unreleased/172-restatedeployment-deletion-hang.md @@ -25,9 +25,15 @@ Alongside that: - **Unfinished invocations not yet bound to a deployment now count.** Paused, queued and scheduled work carries no `pinned_deployment_id`, so the old query scored it as zero and a deletion could tear the endpoint out from under it. It is now attributed through the - target service. + target service. Only a deletion asks: that attribution costs a second scan of + `sys_invocation_status` and a per-row decode of the invocation target, and during a + rollout it can only ever name the deployment that is already the service's endpoint. The + reconcile path's query is unchanged. - **A blocked deletion says what is blocking it.** The `DeploymentInUse` event now names each version and its pinned/unpinned invocation counts instead of a generic message. +- **A blocked deletion backs off.** Retries start at the usual 30 seconds and stretch to + five minutes the longer the wait runs, so a deletion parked behind a scheduled invocation + days out stops re-running that query twice a minute for the duration. - **Drain deadlines are now honoured.** The requeue interval derived from a version's remove-at time was being discarded, because errors reach the controller's error policy wrapped by the finalizer machinery; a short `drainDelaySeconds` cost up to 30 seconds per @@ -50,8 +56,7 @@ same hang. - **Existing deployments:** no configuration change. Deletions that were previously wedged will proceed the next time the operator reconciles them; ones whose finalizer was removed - by hand may have left a stale registration in Restate that needs deleting via the admin - API. + by hand may have left a stale registration behind in Restate. - **Deletion now takes at least `spec.restate.drainDelaySeconds` (default 300s).** The latest version is put through the same drain as any superseded version, so teardown waits out the drain window even when the deployment never served an invocation. This is the @@ -59,8 +64,8 @@ same hang. - **Deletion still waits on unfinished invocations, and that wait has no upper bound.** Scheduled invocations are the sharp edge: a delayed call whose execution time is days out counts as unfinished and holds the deletion until it fires. The `DeploymentInUse` event - reports the counts so this is diagnosable; a `deletionPolicy` field to bound or skip the - wait is planned. + reports the counts so this is diagnosable. Bounding or skipping the wait is the job of the + planned `deletionPolicy` field, not of manual intervention. - **New deployments:** no impact. ### Migration Guidance diff --git a/src/controllers/restatedeployment/cleanup.rs b/src/controllers/restatedeployment/cleanup.rs index 8c71200..d2222e2 100644 --- a/src/controllers/restatedeployment/cleanup.rs +++ b/src/controllers/restatedeployment/cleanup.rs @@ -1,8 +1,11 @@ //! Policy decisions shared by the ReplicaSet and Knative cleanup reconcilers: what -//! Restate still needs a registered deployment for, and how long we keep a drained -//! version around. +//! Restate still needs a registered deployment for, how we ask it, and how long we keep a +//! drained version around. use std::collections::HashMap; +use std::time::Duration; + +use serde::Deserialize; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum CleanupMode { @@ -61,6 +64,123 @@ impl DeploymentUsage { } } +/// The Restate SQL behind [`DeploymentUsageMap`]: one row per registered deployment. +/// +/// The unpinned count is the expensive half. Attributing an invocation that carries no +/// `pinned_deployment_id` needs `target_service_name`, which is nested inside the encoded +/// invocation target and so costs a decode per scanned row, plus a join against +/// `sys_service` — on top of a second pass over `sys_invocation_status` across every +/// partition. Only a deletion needs the answer: unpinned work is attributed through +/// `sys_service.deployment_id`, the same column that sets `latest_for_service`, so during a +/// rollout a non-zero count could only ever name a deployment that flag is already holding. +/// The rollout query therefore doesn't ask, and selects a constant so both modes return one +/// row shape. +/// +/// DataFusion does fold a shared query guarded by `WHERE 1 = 0` down to the same plan +/// (checked with `EXPLAIN` against Restate 1.7.2 — the `unpinned` scan disappears +/// entirely), but building the two strings here says what we mean rather than leaving the +/// cost of the reconcile path resting on an optimiser pass across server versions we don't +/// pin. +pub(crate) fn deployment_usage_query(mode: CleanupMode) -> String { + let (unpinned_cte, unpinned_count, unpinned_join) = if mode.is_deleting() { + ( + r#", + unpinned AS ( + SELECT s.deployment_id AS id, COUNT(*) AS n + FROM sys_invocation_status i + JOIN sys_service s ON s.name = i.target_service_name + WHERE i.pinned_deployment_id IS NULL + AND i.status != 'completed' + GROUP BY s.deployment_id + )"#, + "COALESCE(u.n, 0)", + "\n LEFT JOIN unpinned u ON d.id = u.id", + ) + } else { + ("", "0", "") + }; + + // The counts are LEFT JOINed onto `sys_deployment` so that a deployment with no + // matching row still appears; COALESCE turns that missing row back into a zero count + // rather than a NULL the row struct can't hold. + format!( + r#" + WITH latest AS ( + SELECT DISTINCT deployment_id AS id + FROM sys_service + WHERE deployment_id IS NOT NULL + ), + pinned AS ( + SELECT pinned_deployment_id AS id, COUNT(*) AS n + FROM sys_invocation_status + WHERE pinned_deployment_id IS NOT NULL AND status != 'completed' + GROUP BY pinned_deployment_id + ){unpinned_cte} + SELECT d.id AS deployment_id, + l.id IS NOT NULL AS latest_for_service, + COALESCE(p.n, 0) AS pinned_invocations, + {unpinned_count} AS unpinned_invocations + FROM sys_deployment d + LEFT JOIN latest l ON d.id = l.id + LEFT JOIN pinned p ON d.id = p.id{unpinned_join} + "# + ) +} + +/// One row of [`deployment_usage_query`], as Restate's `/query` endpoint returns it. +#[derive(Debug, Deserialize)] +pub(crate) struct DeploymentUsageRow { + pub deployment_id: String, + pub latest_for_service: bool, + // DataFusion counts are signed; the domain isn't, so these are clamped on the way in. + pub pinned_invocations: i64, + pub unpinned_invocations: i64, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct DeploymentUsageRows { + pub rows: Vec, +} + +impl DeploymentUsageRows { + pub(crate) fn into_map(self) -> DeploymentUsageMap { + let mut usage_by_deployment: DeploymentUsageMap = HashMap::with_capacity(self.rows.len()); + + for row in self.rows { + let usage = DeploymentUsage { + latest_for_service: row.latest_for_service, + pinned_invocations: row.pinned_invocations.max(0) as u64, + unpinned_invocations: row.unpinned_invocations.max(0) as u64, + }; + + usage_by_deployment + .entry(row.deployment_id) + // two rows for one deployment id shouldnt happen: `latest` is DISTINCT, + // `pinned`/`unpinned` are grouped, sys_deployment holds one row per id, and + // all three are LEFT JOINed. we take the most conservative view of each + // fact if a future query shape ever does produce duplicates. + .and_modify(|existing| existing.merge(usage)) + .or_insert(usage); + } + + usage_by_deployment + } +} + +/// How long to wait before re-checking a deletion that in-flight invocations are holding. +/// +/// Every retry re-runs [`deployment_usage_query`] in its deleting flavour, so a deletion +/// parked behind a scheduled invocation days out would otherwise pay that query's two +/// all-partition scans every 30 seconds for as long as it waits. Most drains finish in the +/// first minutes, so poll at the floor early and stretch towards a cap the longer the wait +/// has already run. +pub(crate) fn blocked_deletion_requeue(blocked_for: Duration) -> Duration { + const FLOOR: Duration = Duration::from_secs(30); + const CEILING: Duration = Duration::from_secs(300); + + (blocked_for / 4).clamp(FLOOR, CEILING) +} + /// A version that cleanup could not remove because Restate still needs it. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct BlockingVersion { @@ -162,6 +282,62 @@ mod tests { assert!(!retain_for_rollback(CleanupMode::Deleting, 9, 10)); } + #[test] + fn rollout_query_does_not_pay_for_the_unpinned_count() { + let query = deployment_usage_query(CleanupMode::Rollout); + + // no second scan of sys_invocation_status, and no decode of the nested + // target_service_name to join it against sys_service + assert_eq!(query.matches("sys_invocation_status").count(), 1); + assert!(!query.contains("target_service_name")); + assert!(!query.contains("unpinned AS")); + + // ...but the column is still projected, so one row struct parses both modes + assert!(query.contains("AS unpinned_invocations")); + assert!(query.contains("AS pinned_invocations")); + assert!(query.contains("AS latest_for_service")); + } + + #[test] + fn deleting_query_counts_unpinned_work_through_the_service() { + let query = deployment_usage_query(CleanupMode::Deleting); + + assert_eq!(query.matches("sys_invocation_status").count(), 2); + assert!(query.contains("JOIN sys_service s ON s.name = i.target_service_name")); + assert!(query.contains("COALESCE(u.n, 0) AS unpinned_invocations")); + assert!(query.contains("LEFT JOIN unpinned u ON d.id = u.id")); + } + + #[test] + fn both_query_flavours_exclude_completed_invocations() { + // completed invocations linger in sys_invocation_status for the retention window; + // counting them would hold a deletion open for a full day of drained history + for mode in [CleanupMode::Rollout, CleanupMode::Deleting] { + let query = deployment_usage_query(mode); + assert_eq!( + query.matches("status != 'completed'").count(), + query.matches("sys_invocation_status").count(), + "{mode:?}" + ); + } + } + + #[test] + fn blocked_deletion_polls_hard_early_and_backs_off_later() { + let requeue = |secs| blocked_deletion_requeue(Duration::from_secs(secs)); + + // the common case -- a drain that finishes in the first minutes -- keeps the + // 30s cadence it had before the backoff + assert_eq!(requeue(0), Duration::from_secs(30)); + assert_eq!(requeue(120), Duration::from_secs(30)); + + assert_eq!(requeue(240), Duration::from_secs(60)); + + // a deletion held by a scheduled invocation days out settles on the cap + assert_eq!(requeue(1200), Duration::from_secs(300)); + assert_eq!(requeue(86_400), Duration::from_secs(300)); + } + #[test] fn blocking_versions_name_the_work_holding_the_deletion() { let described = describe_blocking_versions(&[ diff --git a/src/controllers/restatedeployment/controller.rs b/src/controllers/restatedeployment/controller.rs index 25eb6ac..3e58dc5 100644 --- a/src/controllers/restatedeployment/controller.rs +++ b/src/controllers/restatedeployment/controller.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, HashMap}; +use std::collections::BTreeMap; use std::sync::Arc; use std::time::Duration; @@ -47,7 +47,8 @@ use crate::{Error, Result}; // Import our reconcilers use crate::controllers::restatedeployment::cleanup::{ - CleanupMode, DeploymentUsage, DeploymentUsageMap, describe_blocking_versions, + CleanupMode, DeploymentUsage, DeploymentUsageMap, DeploymentUsageRows, + blocked_deletion_requeue, deployment_usage_query, describe_blocking_versions, }; use crate::controllers::restatedeployment::reconcilers; @@ -229,9 +230,15 @@ fn root_cause(err: &Error) -> &Error { fn error_policy(_rs: Arc, err: &Error, _ctx: C) -> Action { match root_cause(err) { // A drain knows its own deadline; the blanket interval would make a short - // drainDelaySeconds cost up to 30s per version anyway. + // drainDelaySeconds cost up to 30s per version anyway. A deletion blocked on + // in-flight invocations sets its own interval too, backing off as the wait grows + // so a long one stops re-running the admin query twice a minute forever. Error::DeploymentDraining { requeue_after: Some(requeue_after), + } + | Error::DeploymentInUse { + requeue_after: Some(requeue_after), + .. } => Action::requeue(*requeue_after), _ => Action::requeue(Duration::from_secs(30)), } @@ -977,55 +984,7 @@ impl RestateDeployment { ctx: &Context, mode: CleanupMode, ) -> Result { - // The unpinned count is attributed through `sys_service.deployment_id`, the same - // column that sets `latest_for_service`, so a non-zero count implies that flag. - let count_unpinned = if mode.is_deleting() { "1 = 1" } else { "1 = 0" }; - - let sql_query = format!( - r#" - WITH latest AS ( - SELECT DISTINCT deployment_id AS id - FROM sys_service - WHERE deployment_id IS NOT NULL - ), - pinned AS ( - SELECT pinned_deployment_id AS id, COUNT(*) AS n - FROM sys_invocation_status - WHERE pinned_deployment_id IS NOT NULL AND status != 'completed' - GROUP BY pinned_deployment_id - ), - unpinned AS ( - SELECT s.deployment_id AS id, COUNT(*) AS n - FROM sys_invocation_status i - JOIN sys_service s ON s.name = i.target_service_name - WHERE {count_unpinned} - AND i.pinned_deployment_id IS NULL - AND i.status != 'completed' - GROUP BY s.deployment_id - ) - SELECT d.id AS deployment_id, - l.id IS NOT NULL AS latest_for_service, - COALESCE(p.n, 0) AS pinned_invocations, - COALESCE(u.n, 0) AS unpinned_invocations - FROM sys_deployment d - LEFT JOIN latest l ON d.id = l.id - LEFT JOIN pinned p ON d.id = p.id - LEFT JOIN unpinned u ON d.id = u.id; - "# - ); - - #[derive(Deserialize)] - struct DeploymentQueryResult { - rows: Vec, - } - - #[derive(Deserialize)] - struct DeploymentQueryResultRow { - deployment_id: String, - latest_for_service: bool, - pinned_invocations: i64, - unpinned_invocations: i64, - } + let sql_query = deployment_usage_query(mode); let resp = ctx .request(Method::POST, &self.spec.restate.register, "/query")? @@ -1036,36 +995,24 @@ impl RestateDeployment { .send() .await .map_err(Error::AdminCallFailed)?; - let response: DeploymentQueryResult = check_admin_response(resp) + let response: DeploymentUsageRows = check_admin_response(resp) .await? .json() .await .map_err(Error::AdminCallFailed)?; - let mut endpoints: DeploymentUsageMap = HashMap::with_capacity(response.rows.len()); - - for row in response.rows { - let usage = DeploymentUsage { - latest_for_service: row.latest_for_service, - pinned_invocations: row.pinned_invocations.max(0) as u64, - unpinned_invocations: row.unpinned_invocations.max(0) as u64, - }; - - match endpoints.entry(row.deployment_id) { - std::collections::hash_map::Entry::Occupied(mut entry) => { - // two rows for one deployment id shouldnt happen: `latest` is DISTINCT, - // `pinned`/`unpinned` are grouped, sys_deployment holds one row per id, - // and all three are LEFT JOINed. we take the most conservative view of - // each fact if a future query shape ever does produce duplicates. - entry.get_mut().merge(usage); - } - std::collections::hash_map::Entry::Vacant(entry) => { - entry.insert(usage); - } - } - } + Ok(response.into_map()) + } - Ok(endpoints) + /// How long ago deletion was requested, for pacing the retries of a blocked one. + /// Zero for an object that is not being deleted, and for a clock that has gone + /// backwards since the deletion timestamp was stamped. + fn blocked_for(&self) -> Duration { + self.metadata + .deletion_timestamp + .as_ref() + .and_then(|deleted_at| (chrono::Utc::now() - deleted_at.0).to_std().ok()) + .unwrap_or_default() } // Finalizer cleanup (the object was deleted, ensure nothing is orphaned) @@ -1150,7 +1097,10 @@ impl RestateDeployment { // this as a Warning event, and it is the only place a stuck deletion explains // itself. Unpinned work includes scheduled invocations, whose execution time // can be arbitrarily far out, so "wait for it" is not always sound advice. - return Err(Error::DeploymentInUse { blocked_by }); + return Err(Error::DeploymentInUse { + blocked_by, + requeue_after: Some(blocked_deletion_requeue(self.blocked_for())), + }); } if let Some(next_removal) = next_removal { @@ -1519,16 +1469,35 @@ mod tests { ); } + /// A deletion held by in-flight invocations paces its own retries, so the interval it + /// computed has to survive the same wrapper. #[test] - fn unrelated_errors_keep_the_blanket_interval() { + fn blocked_deletion_backoff_survives_the_finalizer_wrapper() { let wrapped = Error::FinalizerError(Box::new( kube::runtime::finalizer::Error::CleanupFailed(Error::DeploymentInUse { blocked_by: "greeter-abc123 (1 pinned, 0 unpinned invocations)".into(), + requeue_after: Some(Duration::from_secs(120)), }), )); assert_eq!( error_policy(Arc::new(()), &wrapped, ()), + Action::requeue(Duration::from_secs(120)) + ); + } + + #[test] + fn unrelated_errors_keep_the_blanket_interval() { + let wrapped = Error::FinalizerError(Box::new( + kube::runtime::finalizer::Error::CleanupFailed(Error::HashCollision), + )); + + assert_eq!( + error_policy(Arc::new(()), &wrapped, ()), + Action::requeue(Duration::from_secs(30)) + ); + assert_eq!( + error_policy(Arc::new(()), &Error::HashCollision, ()), Action::requeue(Duration::from_secs(30)) ); } diff --git a/src/controllers/restatedeployment/reconcilers/replicaset.rs b/src/controllers/restatedeployment/reconcilers/replicaset.rs index da28452..0dd5bc8 100644 --- a/src/controllers/restatedeployment/reconcilers/replicaset.rs +++ b/src/controllers/restatedeployment/reconcilers/replicaset.rs @@ -812,4 +812,357 @@ mod tests { inject_in_process_tunnel_env(spec.clone(), "greeter-abc123", &test_params()).unwrap(); assert_eq!(injected, spec); } + + // --- cleanup_old_replicasets against a mocked apiserver. The teardown *order* + // is the invariant here, not just which calls happen, so these record every + // request the reconciler makes and assert on the sequence. --- + + mod teardown { + use std::any::Any; + use std::convert::Infallible; + use std::sync::{Arc, Mutex}; + + use http::{Request, Response}; + use kube::client::Body; + use kube::runtime::reflector; + use kube::runtime::watcher; + use serde_json::json; + + use super::super::*; + use crate::controllers::State; + use crate::controllers::restatedeployment::cleanup::DeploymentUsage; + use crate::metrics::Metrics; + use crate::resources::restatedeployments::RestateDeployment; + + const NAMESPACE: &str = "apps"; + const RSD_UID: &str = "uid-123"; + const VERSION: &str = "greeter-old"; + + /// Every request the reconciler made, in order, as "METHOD /path". + #[derive(Clone, Default)] + struct Calls(Arc>>); + + impl Calls { + fn matching(&self, needle: &str) -> Vec { + self.0 + .lock() + .expect("calls are not poisoned") + .iter() + .filter(|call| call.contains(needle)) + .cloned() + .collect() + } + } + + struct Harness { + ctx: Arc, + rs_api: Api, + calls: Calls, + /// Stores read through their `Store` handles; the writers own the data. + _writers: Vec>, + } + + fn store_of(objects: Vec) -> (reflector::Store, Box) + where + K: reflector::Lookup + Clone + 'static, + K::DynamicType: Default + Eq + std::hash::Hash + Clone, + { + let (reader, mut writer) = reflector::store::(); + writer.apply_watcher_event(&watcher::Event::Init); + for object in objects { + writer.apply_watcher_event(&watcher::Event::InitApply(object)); + } + writer.apply_watcher_event(&watcher::Event::InitDone); + (reader, Box::new(writer)) + } + + fn harness(replicasets: Vec, hpas: Vec) -> Harness { + let calls = Calls::default(); + let client = { + let calls = calls.clone(); + let svc = tower::service_fn(move |req: Request| { + let calls = calls.clone(); + async move { + let path = req.uri().path().to_owned(); + calls + .0 + .lock() + .expect("calls are not poisoned") + .push(format!("{} {path}", req.method())); + + // enough of a body for kube to deserialise the return type of + // whichever call this was; none of it is asserted on + let body = if path.ends_with("/scale") { + json!({ + "apiVersion": "autoscaling/v1", "kind": "Scale", + "metadata": { "name": VERSION, "namespace": NAMESPACE }, + "spec": { "replicas": 0 }, + }) + } else if path.contains("horizontalpodautoscalers") { + json!({ + "apiVersion": "autoscaling/v2", "kind": "HorizontalPodAutoscaler", + "metadata": { "name": VERSION, "namespace": NAMESPACE }, + }) + } else { + json!({ + "apiVersion": "apps/v1", "kind": "ReplicaSet", + "metadata": { "name": VERSION, "namespace": NAMESPACE }, + }) + }; + + Ok::<_, Infallible>( + Response::builder() + .status(200) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(&body).unwrap())) + .unwrap(), + ) + } + }); + kube::Client::new(svc, NAMESPACE) + }; + + let (replicasets_store, rs_writer) = store_of(replicasets); + let (hpa_store, hpa_writer) = store_of(hpas); + let (rce_store, rce_writer) = store_of(vec![]); + let (secret_store, secret_writer) = store_of(vec![]); + let (revision_store, revision_writer) = store_of(vec![]); + let (configuration_store, configuration_writer) = store_of(vec![]); + + let ctx = Context::new( + client.clone(), + replicasets_store, + rce_store, + secret_store, + revision_store, + configuration_store, + hpa_store, + Metrics::default(), + State::new( + None, + false, + "restate-operator".into(), + None, + None, + "tunnel:latest".into(), + "cluster.local".into(), + "alpine:3.21".into(), + None, + None, + ), + ); + + Harness { + rs_api: Api::namespaced(client, NAMESPACE), + ctx, + calls, + _writers: vec![ + rs_writer, + hpa_writer, + rce_writer, + secret_writer, + revision_writer, + configuration_writer, + ], + } + } + + fn rsd(autoscaling: bool, deleting: bool) -> RestateDeployment { + let spec = serde_json::from_value(json!({ + "replicas": 3, + "revisionHistoryLimit": 10, + "template": { + "metadata": null, + "spec": { "containers": [{ "name": "app", "image": "greeter:v1" }] } + }, + "restate": { + "register": { "cluster": null, "cloud": null, "service": null, "url": "http://restate:9070/" }, + "servicePath": null, "useHttp11": null, "drainDelaySeconds": null + }, + "autoscaling": autoscaling.then(|| json!({ "minReplicas": 1, "maxReplicas": 5 })), + })) + .expect("test RestateDeploymentSpec deserializes"); + + let mut rsd = RestateDeployment::new("greeter", spec); + rsd.metadata.uid = Some(RSD_UID.into()); + rsd.metadata.namespace = Some(NAMESPACE.into()); + if deleting { + rsd.metadata.deletion_timestamp = Some( + k8s_openapi::apimachinery::pkg::apis::meta::v1::Time(chrono::Utc::now()), + ); + } + rsd + } + + fn version( + deployment_id: &str, + remove_at: Option>, + ) -> ReplicaSet { + let mut annotations = serde_json::Map::new(); + annotations.insert( + RESTATE_DEPLOYMENT_ID_ANNOTATION.into(), + json!(deployment_id), + ); + if let Some(remove_at) = remove_at { + annotations.insert( + RESTATE_REMOVE_VERSION_AT_ANNOTATION.into(), + json!(remove_at.to_rfc3339()), + ); + } + + serde_json::from_value(json!({ + "apiVersion": "apps/v1", + "kind": "ReplicaSet", + "metadata": { + "name": VERSION, + "namespace": NAMESPACE, + "creationTimestamp": "2026-01-01T00:00:00Z", + "annotations": annotations, + "ownerReferences": [{ + "apiVersion": "restate.dev/v1beta1", + "kind": "RestateDeployment", + "name": "greeter", + "uid": RSD_UID, + "controller": true, + }], + }, + "spec": { "replicas": 3 }, + })) + .expect("test ReplicaSet deserializes") + } + + fn version_hpa() -> HorizontalPodAutoscaler { + serde_json::from_value(json!({ + "apiVersion": "autoscaling/v2", + "kind": "HorizontalPodAutoscaler", + "metadata": { "name": VERSION, "namespace": NAMESPACE }, + "spec": { + "scaleTargetRef": { + "apiVersion": "apps/v1", "kind": "ReplicaSet", "name": VERSION, + }, + "minReplicas": 1, "maxReplicas": 5, + }, + })) + .expect("test HorizontalPodAutoscaler deserializes") + } + + fn usage_of(deployment_id: &str, usage: DeploymentUsage) -> DeploymentUsageMap { + [(deployment_id.to_owned(), usage)].into() + } + + /// The autoscaler comes off immediately before the scale to zero, in the same + /// branch: its `minReplicas` floor of 1 would otherwise fight the scale-down and + /// hold the version at that floor for the whole retention window. + #[tokio::test] + async fn autoscaler_is_removed_immediately_before_the_scale_to_zero() { + let rsd = rsd(true, false); + let harness = harness(vec![version("dp_gone", None)], vec![version_hpa()]); + + let (blocking, next_removal) = cleanup_old_replicasets( + NAMESPACE, + &harness.ctx, + &harness.rs_api, + RSD_UID, + &rsd, + // the endpoint was deregistered by other means, so the version is + // scaled down on this pass rather than waiting out a drain + &DeploymentUsageMap::new(), + None, + ) + .await + .expect("cleanup succeeds"); + + assert!(blocking.is_empty()); + assert_eq!(next_removal, None); + assert_eq!( + harness.calls.matching(VERSION), + vec![ + format!( + "DELETE /apis/autoscaling/v2/namespaces/{NAMESPACE}/horizontalpodautoscalers/{VERSION}" + ), + format!( + "PATCH /apis/apps/v1/namespaces/{NAMESPACE}/replicasets/{VERSION}/scale" + ), + ], + ); + } + + /// ...and not before that. A version waiting out its drain deadline is still + /// serving traffic, so it keeps the autoscaler it was given. + #[tokio::test] + async fn draining_version_keeps_its_autoscaler_until_the_deadline() { + let rsd = rsd(true, false); + let remove_at = chrono::Utc::now() + chrono::TimeDelta::seconds(300); + let harness = harness( + vec![version("dp_draining", Some(remove_at))], + vec![version_hpa()], + ); + + let (blocking, next_removal) = cleanup_old_replicasets( + NAMESPACE, + &harness.ctx, + &harness.rs_api, + RSD_UID, + &rsd, + // registered, superseded, and nothing in flight: drained but not yet due + &usage_of("dp_draining", DeploymentUsage::default()), + None, + ) + .await + .expect("cleanup succeeds"); + + assert!(blocking.is_empty()); + assert_eq!(next_removal, Some(remove_at)); + assert_eq!( + harness.calls.matching(VERSION), + Vec::::new(), + "nothing is touched until the deadline passes" + ); + } + + /// The same holds while the RestateDeployment itself is being deleted: deletion + /// puts every version through the drain, and stripping their autoscalers up front + /// would collapse them to `spec.replicas` while they still serve invocations. + #[tokio::test] + async fn deletion_does_not_strip_autoscalers_up_front() { + let rsd = rsd(true, true); + let remove_at = chrono::Utc::now() + chrono::TimeDelta::seconds(300); + let harness = harness( + vec![version("dp_busy", Some(remove_at))], + vec![version_hpa()], + ); + + let busy = DeploymentUsage { + latest_for_service: true, + pinned_invocations: 2, + unpinned_invocations: 0, + }; + + let (blocking, _) = cleanup_old_replicasets( + NAMESPACE, + &harness.ctx, + &harness.rs_api, + RSD_UID, + &rsd, + &usage_of("dp_busy", busy), + None, + ) + .await + .expect("cleanup succeeds"); + + assert_eq!( + blocking, + vec![BlockingVersion { + name: VERSION.into(), + usage: busy, + }], + "in-flight invocations hold the deletion, and say so" + ); + assert_eq!( + harness.calls.matching("horizontalpodautoscalers"), + Vec::::new(), + "an owned HPA is garbage-collected with the RestateDeployment, not by us" + ); + } + } } diff --git a/src/lib.rs b/src/lib.rs index 321dce6..019b396 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -70,7 +70,10 @@ pub enum Error { #[error( "This RestateDeployment is backing active versions in Restate: {blocked_by}. If you want to delete the RestateDeployment, either register new endpoints for the relevant services or cancel/complete the outstanding invocations." )] - DeploymentInUse { blocked_by: String }, + DeploymentInUse { + blocked_by: String, + requeue_after: Option, + }, #[error( "This RestateDeployment is backing recently-active versions in Restate. It will be removed after the drain delay period."