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/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/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 new file mode 100644 index 0000000..65fe6cb --- /dev/null +++ b/release-notes/unreleased/172-restatedeployment-deletion-hang.md @@ -0,0 +1,86 @@ +# 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. 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 + 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 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 + 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. Bounding or skipping the wait is the job of the + planned `deletionPolicy` field, not of manual intervention. +- **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..d2222e2 --- /dev/null +++ b/src/controllers/restatedeployment/cleanup.rs @@ -0,0 +1,360 @@ +//! Policy decisions shared by the ReplicaSet and Knative cleanup reconcilers: what +//! 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 { + 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, + } + } +} + +/// 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 { + /// 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 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(&[ + 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..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; @@ -46,6 +46,10 @@ use crate::telemetry; use crate::{Error, Result}; // Import our reconcilers +use crate::controllers::restatedeployment::cleanup::{ + CleanupMode, DeploymentUsage, DeploymentUsageMap, DeploymentUsageRows, + blocked_deletion_requeue, deployment_usage_query, describe_blocking_versions, +}; use crate::controllers::restatedeployment::reconcilers; use super::reconcilers::replicaset::{ @@ -207,8 +211,37 @@ 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. 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)), + } } impl RestateDeployment { @@ -432,7 +465,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 +476,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 +529,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,34 +979,12 @@ 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 - FROM sys_service - WHERE deployment_id IS NOT NULL - UNION - SELECT DISTINCT pinned_deployment_id as id - FROM sys_invocation_status - WHERE pinned_deployment_id IS NOT NULL AND status != 'completed' - ) - SELECT d.id as deployment_id, - a.id IS NOT NULL as active - FROM sys_deployment d - LEFT JOIN active_deployments a ON d.id = a.id; - "#; - - #[derive(Deserialize)] - struct DeploymentQueryResult { - rows: Vec, - } - - #[derive(Deserialize)] - struct DeploymentQueryResultRow { - deployment_id: String, - active: bool, - } + pub(super) async fn list_deployments( + &self, + ctx: &Context, + mode: CleanupMode, + ) -> Result { + let sql_query = deployment_usage_query(mode); let resp = ctx .request(Method::POST, &self.spec.restate.register, "/query")? @@ -978,30 +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 = HashMap::with_capacity(response.rows.len()); - - for row in response.rows { - 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); - } - } - std::collections::hash_map::Entry::Vacant(entry) => { - entry.insert(row.active); - } - } - } + 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) @@ -1038,7 +1049,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 +1059,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 +1084,23 @@ 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, + requeue_after: Some(blocked_deletion_requeue(self.blocked_for())), + }); } if let Some(next_removal) = next_removal { @@ -1088,7 +1109,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 +1446,62 @@ 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)) + ); + } + + /// A deletion held by in-flight invocations paces its own retries, so the interval it + /// computed has to survive the same wrapper. + #[test] + 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)) + ); + } + /// 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..0dd5bc8 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)] @@ -808,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 0c0f28e..019b396 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -68,9 +68,12 @@ 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, + requeue_after: Option, + }, #[error( "This RestateDeployment is backing recently-active versions in Restate. It will be removed after the drain delay period." @@ -114,7 +117,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",