feat: fenced, progress-preserving driver restarts (FencedRestart, alpha)

[gangavhi]

Purpose of this PR

Add fenced, progress-preserving driver restarts behind a new alpha
feature gate FencedRestart (opt-in per application via
spec.restartPolicy.recovery).

Closes #3040

Problem. The driver is the SPoF of Spark on Kubernetes. Today
restartPolicy: OnFailure/Always has two gaps:

1. Correctness: after a node partition, a rerun can race a still-running
   zombie of the old driver on non-transactional sinks (plain files, JDBC,
   custom sinks). Nothing fences the old driver at submission time.
2. Economics: the rerun starts from zero even when the job had durably
   completed most of its work, because the operator gives applications no
   standard place to record and read back progress.

Solution. With spec.restartPolicy.recovery set:

• Before each rerun the controller atomically advances a monotonic epoch
  in an external state store (Redis in this PR; small pluggable interface in
  pkg/statestore). All store writes on behalf of a driver are epoch-guarded
  and atomic (Lua check-and-write), so once fenced, the old driver can never
  commit again — correctness never depends on the old pod actually being
  dead. Order: fence → delete → resubmit (native spark-submit path).
• The existing pod webhook injects a spark-recovery-agent sidecar (a
  new agent subcommand of the operator binary — no new image) that
  heartbeats to the store, serves a localhost progress-marker API to the
  driver, and restores the latest committed marker before the restarted
  driver boots (SPARK_RECOVERY_* driver env carries the contract).
• status.recoveryStatus reports epoch / restoredFromEpoch; a
  warning event fires when a recovery-enabled app restarts having never
  committed a marker — a fenced restart-from-zero must never masquerade as
  progress preservation.

Explicitly not claimed: Spark has no driver checkpoint/restore;
executors, shuffle, and scheduler state die with the driver regardless.
This is restart minus redone work plus a fencing guarantee — never
mid-stage resume. docs/fenced-restart.md includes a "you don't need this
if…" section.

Design doc: proposals/fenced-progress-preserving-restart.md (included).

Change Category

• Bugfix (non-breaking change which fixes an issue)
• Feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that could affect existing functionality)
• Documentation update

Gate off (default): zero behavior change; the validating webhook rejects
spec.restartPolicy.recovery with a clear message.

Rationale

See the included proposal for goals/non-goals, the fencing protocol,
alternatives considered (external companion operator, CRIU/process
checkpointing, Lease-based fencing, sink-side-transactions-only), and
graduation criteria. Highlights of the integration approach:

• Reuses the existing restart machinery — one fencing step added at the
  FAILING → PENDING_RERUN transition; rerun is the native submission
  path, so the restarted driver is a first-class operator-managed driver.
• Reuses the existing pod mutating webhook — no new deployment, no new
  cert path.
• OnFailureRetries / OnFailureRetryInterval keep their semantics and
  bound fenced recoveries too: one attempt-tracking system.

Testing evidence (local kind cluster, long-running job)

Reproducible via hack/fenced-restart-demo/run-demo-kind.sh (builds the
image from this branch, creates a kind cluster, installs the chart,
deploys Redis + a 20-stage resumable PySpark job, kills the driver
mid-run, and collects all artifacts referenced below into
evidence-20260712-112419/).

• Environment: kind (Docker Desktop, macOS/arm64), Kubernetes v1.35.0,
  Spark 4.0.1.
• Webhook injected the sidecar: 01-driver-containers.txt shows
  spark-kubernetes-driver + spark-recovery-agent.
• Run 1 executed stages 0..3 (4 stages, 32s of compute — a deterministic
  8s/stage floor was added to the demo job so timing doesn't race real
  Spark compute, which is sub-second on this hardware; see "Also fixed"
  below), committing a marker after each stage (02-driver-run1.log).
• Driver pod force-deleted → operator fenced (epoch 0 → 1, event
  SparkApplicationEpochAdvanced in 04-events.txt), deleted resources,
  resubmitted natively.
• Fence verified directly: a write guarded by the stale epoch (0) is
  rejected with FENCED (06-stale-write-rejected.txt).
• Run 2 logged epoch=1 RESUMING from stage 4 (skipping 4 completed stages) (07-driver-run2.log) and completed all 20 stages in 128.2s;
  08-final-status.json shows COMPLETED with executionAttempts: 2,
  recoveryStatus: {epoch: 1, restoredFromEpoch: 0}.
• Value demonstrated: 32s of already-completed compute was preserved
  across a hard driver kill; without this feature run 2 would have
  re-executed all 20 stages.

Unit tests: pkg/statestore (fencing monotonicity, stale-epoch rejection,
heartbeat seq, pruning — against miniredis), webhook injection
(idempotency, gate/role/profile guards, driver-env consistency), event
filter/handler (driver-container-termination detection), validator rules,
and submission env injection. go test ./pkg/statestore/... ./internal/webhook/... ./internal/controller/sparkapplication/....

Bugs found and fixed while getting this evidence

Running the feature end-to-end against a real kind cluster (rather than
only the envtest-based unit suite, which passed throughout and did not
catch any of these) surfaced three real bugs, all fixed in this PR:

1. Epoch race between the webhook and status persistence. The webhook
read the sidecar's epoch from app.Status.RecoveryStatus.Epoch, but
submitSparkApplication creates the driver pod (synchronously invoking
spark-submit, which triggers the webhook) before the reconciler's
subsequent updateSparkApplicationStatus call persists the freshly
advanced epoch to the API server. The webhook could therefore observe the
previous epoch and inject a mismatched value into the sidecar.
Observed directly: the recovery agent's first heartbeat (10s in) found
the real, newer epoch in Redis, concluded it was stale, and self-fenced —
crashing a driver that had never actually been re-run. This repeated on
every subsequent rerun until onFailureRetries was exhausted and the job
went FAILED. Fixed in internal/webhook/recovery_sidecar.go to read
SPARK_RECOVERY_EPOCH from the driver container's own env within the
same pod object being mutated — submission.go already sets it from the
identical in-memory epoch used for fencing, so this is race-free by
construction. Regression test added in recovery_sidecar_test.go.

2 & 3. Completion detection breaks once the driver pod has a sidecar
(two layers). Kubernetes only reports pod.Status.Phase = Succeeded
once every container exits. With the recovery-agent sidecar present,
the driver container can exit 0 while the sidecar keeps running, holding
the pod at Running. Two independent layers that gate whether the
operator re-evaluates a Spark pod update — the predicate.Predicate in
internal/controller/sparkapplication/event_filter.go and the
handler.EventHandler in event_handler.go — both only checked
pod.Status.Phase, so neither ever noticed the driver's completion.
Observed directly: the SparkApplication sat reporting RUNNING
indefinitely after the driver had already logged "ALL 20 stages complete"
and exited cleanly (confirmed by inspecting container statuses directly:
driver terminated, agent still running, with no operator reconcile
logged for the app for over a minute). Fixing only event_handler.go was
not sufficient — event_filter.go's predicate was still dropping the
event a layer higher. Fixed both to also let the event through when the
driver container's own terminated state changes, independent of pod
phase (shared helper driverContainerStateUnchanged). Regression tests
added: new file event_handler_test.go, plus a new case in the existing
TestSparkPodEventFilter_Update table in event_filter_test.go.

Checklist

• I have conducted a self-review of my own code.
• I have updated documentation accordingly (docs/fenced-restart.md).
• I have added tests that prove my fix is effective or that my feature works.
• Existing unit tests pass locally with my changes.

Additional Notes

• make generate manifests api-docs and go mod tidy were run after the
  API changes; generated CRD/deepcopy/docs updates are included. New
  dependencies: github.com/redis/go-redis/v9, github.com/alicebob/miniredis/v2
  (test only).
• examples/spark-fenced-restart.yaml's demo job got a deterministic
  STAGE_MIN_SECONDS=8 floor per stage: on fast hardware the original
  300M-row-sum-per-stage compute finished all 20 stages in ~5 seconds,
  too fast to kill mid-run in any reproducible way.
• Follow-ups I'm happy to drive if there's interest (kept out to keep this
  reviewable): S3-compatible store backend via conditional writes, Helm
  chart values for the new flags, suspect detection from heartbeat
  staleness, Prometheus metrics for epoch/recovery duration.

[google-oss-prow]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign andreyvelich for approval. For more information see the Kubernetes Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment