policy-controller: exit if the lease watch task dies (#15506)

[pujitha24]

Problem

The status controller's leader-election logic panics if its lease
claim watch is ever dropped ("Claims watch must not be dropped"),
on the assumption from #10584 that Kubernetes will then restart the
container. But that consumer (policy-controller/k8s/status/index.rs)
runs inside a detached tokio::spawned task that nothing ever joins,
so the panic is only logged--it never reaches the process. Worse,
the task that owns the watch::Sender (spawned by kubert's
spawn_lease) was itself never joined either (its JoinHandle was
discarded as _task), so if that task ever dies (e.g. after
exhausting retries against a flaky Kubernetes API), the container is
left running with a permanently broken status controller:
restarts=0, ready=true, and no replica ever holds the write lease
again. New/changed Server, HTTPRoute, and AuthorizationPolicy
resources then stop getting their status patched.

To be clear about scope: this does not change any user-visible
behavior while the lease task is healthy, and it does not itself
prevent the underlying watch loss (e.g. a flaky apiserver). It closes
the specific gap where that failure was invisible to Kubernetes.

Solution

Keep the JoinHandle returned by spawn_lease and spawn a watchdog
task that awaits it. If the lease task ever ends--whether by
returning an error, returning Ok unexpectedly, or panicking--log why
and call std::process::exit(1) so Kubernetes actually restarts the
container, fulfilling the original intent of #10584.

Validation

cargo build -p linkerd-policy-controller-runtime
cargo test -p linkerd-policy-controller-runtime -p linkerd-policy-controller-k8s-status
cargo clippy -p linkerd-policy-controller-runtime --all-targets
cargo fmt -p linkerd-policy-controller-runtime -- --check

All of the above pass with no warnings or diffs. No new automated
test was added: verifying this change requires observing an actual
process exit, which would need a subprocess-spawning test harness
that this repo doesn't currently have; the change was instead traced
against kubert's lease implementation and the status index/controller
call sites to confirm the watchdog cannot fire during a normal
SIGINT/SIGTERM graceful shutdown.

Fixes #15506

Signed-off-by: Pujitha Paladugu 10557236+pujitha24@users.noreply.github.com

[Copilot]

Pull request overview

This PR ensures the policy-controller process reliably terminates when the underlying Kubernetes lease-maintenance task stops, so Kubernetes can restart the container and avoid silently broken leader-election/status reconciliation.

Changes:

• Keep the JoinHandle returned by runtime.spawn_lease(...) rather than discarding it.
• Add a watchdog task that awaits the lease task and calls std::process::exit(1) if it ever returns/errors/panics.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[adleong]

Great catch with this and I think this is on the right track but I'm a bit worried about using process::exit inside a worker task like this. What do you think about a similar solution where we monitor the lease task handle at a higher level? i.e.

diff --git a/policy-controller/runtime/src/args.rs b/policy-controller/runtime/src/args.rs
index 81f812b9b..bd1fe02fa 100644
--- a/policy-controller/runtime/src/args.rs
+++ b/policy-controller/runtime/src/args.rs
@@ -181,7 +181,7 @@ impl Args {
         let hostname =
             std::env::var("HOSTNAME").expect("Failed to fetch `HOSTNAME` environment variable");
 
-        let claims = lease::init(
+        let (claims, lease_task) = lease::init(
             &runtime,
             &control_plane_namespace,
             &policy_deployment_name,
@@ -416,10 +416,30 @@ impl Args {
 
         let runtime = runtime.spawn_server(Admission::new);
 
-        // Block the main thread on the shutdown signal. Once it fires, wait for the background tasks to
-        // complete before exiting.
-        if runtime.run().await.is_err() {
-            bail!("Aborted");
+        // Run until shutdown or until the lease task terminates unexpectedly.
+        tokio::select! {
+            biased;
+            res = lease_task => {
+                match res {
+                    Ok(Err(error)) => {
+                        bail!("Lease task failed: {error}");
+                    }
+                    Ok(Ok(())) => {
+                        bail!("Lease task exited unexpectedly");
+                    }
+                    Err(error) if error.is_panic() => {
+                        std::panic::resume_unwind(error.into_panic());
+                    }
+                    Err(error) => {
+                        bail!("Lease task was cancelled: {error}");
+                    }
+                }
+            }
+            res = runtime.run() => {
+                if res.is_err() {
+                    bail!("Aborted");
+                }
+            }
         }
 
         Ok(())
diff --git a/policy-controller/runtime/src/lease.rs b/policy-controller/runtime/src/lease.rs
index ff326b239..e974a3adb 100644
--- a/policy-controller/runtime/src/lease.rs
+++ b/policy-controller/runtime/src/lease.rs
@@ -15,7 +15,10 @@ pub async fn init<T>(
     namespace: &str,
     deployment_name: &str,
     claimant: &str,
-) -> Result<watch::Receiver<Arc<kubert::lease::Claim>>> {
+) -> Result<(
+    watch::Receiver<Arc<kubert::lease::Claim>>,
+    tokio::task::JoinHandle<Result<(), kubert::lease::Error>>,
+)> {
     let params = kubert::LeaseParams {
         name: LEASE_NAME.to_string(),
         namespace: namespace.to_string(),
@@ -109,6 +112,6 @@ pub async fn init<T>(
         time::sleep(time::Duration::from_secs(1)).await;
     }
 
-    let (claim, _task) = runtime.spawn_lease(params).await?;
-    Ok(claim)
+    let (claim, task) = runtime.spawn_lease(params).await?;
+    Ok((claim, task))
 }

[pujitha24]

Thanks for the feedback and the concrete diff — that's a much better shape for this than calling process::exit from inside a worker task. I moved the watchdog into Args::run in commit 653164f: lease::init now returns the JoinHandle alongside the claim, and a tokio::select! races it against runtime.run(), routing failures through the normal anyhow::Result/bail! path (and re-raising panics via std::panic::resume_unwind instead of masking them).

That change also picks up the Copilot review comment on lease.rs: the new code checks JoinError::is_panic() explicitly before treating an error as a panic, so a cancelled task is no longer mislabeled as "panicked".

Build, tests (cargo test -p linkerd-policy-controller-runtime -p linkerd-policy-controller-k8s-status), and cargo fmt --check all pass locally. Let me know if this matches what you had in mind.

[adleong]

Thanks for this contribution!