From 271616dafdf8cd7e5a50dfcadda67e0dd7ec590a Mon Sep 17 00:00:00 2001 From: Noah Kontur Date: Mon, 12 Oct 2026 10:50:44 -0400 Subject: [PATCH 1/5] fix(operator): do not record a config Job that ran on replaced ConfigMap data A config-update reset puts a node's entry back at (config, in_progress), which is indistinguishable from the run an older config Job did. If that Job's completion was still in flight, it was recorded as the new run's, and the node finished on the old data with no diff left to retry. Package pods read each key once at start, so nothing corrects it afterwards. Stamp a config-stage Job with a hash of the ConfigMap data it mounts at creation. When recording a completion, compare it with the live ConfigMap; on a mismatch, delete the Job instead so the stage re-runs on current data. Jobs without the stamp, and Jobs whose ConfigMap is gone, record as before. Follow-up to the unrecorded-Job clearing in #788; part of #776. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Noah Kontur --- .../2026-07-10-package-execution-as-jobs.md | 2 +- .../internal/controller/job_controller.go | 51 +++++++++++++ .../controller/job_controller_test.go | 71 +++++++++++++++++++ .../internal/controller/skyhook_controller.go | 12 ++++ operator/internal/controller/swap_test.go | 26 +++++++ 5 files changed, 161 insertions(+), 1 deletion(-) diff --git a/docs/designs/2026-07-10-package-execution-as-jobs.md b/docs/designs/2026-07-10-package-execution-as-jobs.md index d0fae1cd0..30d23d8ce 100644 --- a/docs/designs/2026-07-10-package-execution-as-jobs.md +++ b/docs/designs/2026-07-10-package-execution-as-jobs.md @@ -120,7 +120,7 @@ The pseudo-controller pattern is kept (issue #223 option A): a Jobs watch maps e Because completed pods now linger, pod-deletion can no longer be the processed-once marker. A persisted Job annotation, `nodewright.nvidia.com/state-recorded: "true"`, replaces it. On each Job event: -- **`Complete`, not yet marked** → read the `Succeeded` child pod for the container name (selected by the Job's controller UID, not by name — a same-named prior Job's pod can still be terminating), record completion through the existing node-state path (upgrade/uninstall/interrupt special cases unchanged), then one Job Update sets `state-recorded` and the success TTL. If the child pod was already GC'd, fall back to the Job's own stage/node/interrupt labels. Already-marked Jobs are skipped, so duplicate events are harmless. The recorded attempt count now comes from the Job's failed-pod count — a user-visible improvement over today's usually-zero restart count. +- **`Complete`, not yet marked** → read the `Succeeded` child pod for the container name (selected by the Job's controller UID, not by name — a same-named prior Job's pod can still be terminating), record completion through the existing node-state path (upgrade/uninstall/interrupt special cases unchanged), then one Job Update sets `state-recorded` and the success TTL. If the child pod was already GC'd, fall back to the Job's own stage/node/interrupt labels. Already-marked Jobs are skipped, so duplicate events are harmless. A config-stage Job also carries `nodewright.nvidia.com/config-hash`, a hash of the ConfigMap data it was created against; if the live ConfigMap no longer matches, the completion is not recorded and the Job is deleted so the stage re-runs on the current data (a config-update reset puts the entry back at `config`, which otherwise looks identical to the run the Job did). The recorded attempt count now comes from the Job's failed-pod count — a user-visible improvement over today's usually-zero restart count. - **`Failed`, on genuine evidence** → the stage is out of attempts (`BackoffLimitExceeded` with at least one retained archive that really failed) or hit the Job-level ceiling (`DeadlineExceeded`, which needs no archive — see Stage deadline, below). Record `erroring`, mark processed with the failure TTL, and **leave the Job in place as the timeout marker**. - **`Failed`, without it** → every retained attempt was a pod the package never ran in, the kubelet-admission case below. No node-state write (matching today's silent disruption behavior); one Update carries the marker and failure TTL, and the sweep then clears the Job so the stage re-runs. - **Node gone** → if the node `Get` returns NotFound, there is no state to record: mark the Job processed and stop, rather than error-looping on a node that will never return. diff --git a/operator/internal/controller/job_controller.go b/operator/internal/controller/job_controller.go index b0cc3958a..2550d2524 100644 --- a/operator/internal/controller/job_controller.go +++ b/operator/internal/controller/job_controller.go @@ -20,8 +20,11 @@ package controller import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" "sort" + "strings" "time" "github.com/NVIDIA/nodewright/operator/api/nodewright/v1alpha1" @@ -69,6 +72,11 @@ const ( // interrupt Job created before the bound existed, is never ended by it. annotationRestartAllowance = v1alpha1.METADATA_PREFIX + "/restart-allowance" + // annotationConfigHash is stamped at creation on a config-stage Job: a hash of the package + // ConfigMap data the pod mounts. The pod reads each key once at start, so a Job that ran + // against data since replaced must not have its completion recorded as the new data's. + annotationConfigHash = v1alpha1.METADATA_PREFIX + "/config-hash" + // interruptRestartLimitAction is the event action for an interrupt Job ended that way. interruptRestartLimitAction = "RestartLimitExceeded" @@ -226,6 +234,15 @@ func (r *JobReconciler) handleCompleteJob(ctx context.Context, job *batchv1.Job) return ctrl.Result{}, r.markJobProcessed(ctx, job, r.opts.JobTTLSucceeded) } + // The entry is back at config after a config-update reset, which looks the same as the run this + // Job did, so recording would complete the node on data it never saw. Drop the Job instead; + // the stage is re-created from the entry against the current ConfigMap. + if superseded, err := r.configSuperseded(ctx, job, pkg); err != nil { + return ctrl.Result{}, err + } else if superseded { + return ctrl.Result{}, deleteJobForeground(ctx, r.Client, job, "config changed while the config stage ran") + } + if err := r.recordJobCompletion(ctx, job, pkg); err != nil { return ctrl.Result{}, err } @@ -233,6 +250,40 @@ func (r *JobReconciler) handleCompleteJob(ctx context.Context, job *batchv1.Job) return ctrl.Result{}, r.markJobProcessed(ctx, job, r.opts.JobTTLSucceeded) } +// configSuperseded reports a config-stage Job whose stamped ConfigMap hash no longer matches the +// live ConfigMap. A Job without the stamp (created by an older operator) or a ConfigMap that is +// gone is never superseded, so those still record as before. +func (r *JobReconciler) configSuperseded(ctx context.Context, job *batchv1.Job, pkg *PackageSkyhook) (bool, error) { + stamped, ok := job.Annotations[annotationConfigHash] + if !ok || pkg.Stage != v1alpha1.StageConfig { + return false, nil + } + cm := &corev1.ConfigMap{} + key := types.NamespacedName{Namespace: job.Namespace, Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", pkg.Skyhook, pkg.Name, pkg.Version))} + if err := r.Get(ctx, key, cm); err != nil { + if apierrors.IsNotFound(err) { + return false, nil + } + return false, fmt.Errorf("getting config map %s for job %s: %w", key.Name, job.Name, err) + } + return configDataHash(cm.Data) != stamped, nil +} + +// configDataHash is a stable digest of ConfigMap data: keys sorted, each key and value +// length-prefixed so neither can bleed into the next. +func configDataHash(data map[string]string) string { + keys := make([]string, 0, len(data)) + for k := range data { + keys = append(keys, k) + } + sort.Strings(keys) + h := sha256.New() + for _, k := range keys { + _, _ = fmt.Fprintf(h, "%d:%s%d:%s", len(k), k, len(data[k]), data[k]) + } + return hex.EncodeToString(h.Sum(nil)) +} + // shouldRecordCompletion is the re-serve guard. It records a completion only when the // package's node-state entry is still at this Job's stage and not yet complete: a valid // forward transition. An absent entry (uninstalled or node reset), an already-complete diff --git a/operator/internal/controller/job_controller_test.go b/operator/internal/controller/job_controller_test.go index a6a6e514c..61c2bfc8b 100644 --- a/operator/internal/controller/job_controller_test.go +++ b/operator/internal/controller/job_controller_test.go @@ -21,6 +21,7 @@ package controller import ( "context" "fmt" + "strings" "time" "github.com/NVIDIA/nodewright/operator/api/nodewright/v1alpha1" @@ -196,6 +197,76 @@ var _ = Describe("JobReconcile", func() { Expect(getNodeState(r)[pkgRef.GetUniqueName()].Stage).To(Equal(v1alpha1.StageConfig)) }) + // A config-update reset puts the entry back at (config, in_progress), the same shape the old + // run had, so the stamped ConfigMap hash is the only thing that says which data a Job saw. + Describe("config Job completion after the ConfigMap changed", func() { + configMap := func(data map[string]string) *corev1.ConfigMap { + return &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhookName, pkgRef.Name, pkgRef.Version)), + }, Data: data} + } + configJob := func(stamped map[string]string) *batchv1.Job { + job := packageJob(v1alpha1.StageConfig, false, trueCondition(batchv1.JobComplete, "")) + if stamped != nil { + job.Annotations[annotationConfigHash] = configDataHash(stamped) + } + return job + } + + It("does not record a Job that ran on data since replaced, and drops the Job", func() { + node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) + job := configJob(map[string]string{"a.properties": "old"}) + r := newReconciler(node, job, configMap(map[string]string{"a.properties": "new"})) + + _, err := r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateInProgress)) + err = r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: job.Name}, &batchv1.Job{}) + Expect(apierrors.IsNotFound(err)).To(BeTrue(), "the stale Job is deleted so the stage re-runs") + }) + + It("records a Job whose data still matches", func() { + node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) + data := map[string]string{"a.properties": "new"} + job := configJob(data) + r := newReconciler(node, job, configMap(data)) + + _, err := r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete)) + }) + + It("records a Job created without the stamp", func() { + node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) + job := configJob(nil) + r := newReconciler(node, job, configMap(map[string]string{"a.properties": "new"})) + + _, err := r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete)) + }) + + It("records when the ConfigMap is gone", func() { + node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) + job := configJob(map[string]string{"a.properties": "old"}) + r := newReconciler(node, job) + + _, err := r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateComplete)) + }) + + It("hashes the data independent of key order and without key/value bleed", func() { + Expect(configDataHash(map[string]string{"a": "1", "b": "2"})).To(Equal(configDataHash(map[string]string{"b": "2", "a": "1"}))) + Expect(configDataHash(map[string]string{"a": "bc"})).ToNot(Equal(configDataHash(map[string]string{"ab": "c"}))) + }) + }) + It("does not regress a stage when a completion is re-served after the node advanced", func() { // The apply Job re-fires, but node state has already moved on to config. node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index eeea02f1d..f65383a77 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -3873,6 +3873,18 @@ func (r *SkyhookReconciler) ApplyPackage(ctx context.Context, logger logr.Logger return fmt.Errorf("error setting package on job: %w", err) } + // Stamp the data the config pod will mount, so a completion that outlives a later config change + // is not recorded as the new data's (see JobReconciler.configSuperseded). + if stage == v1alpha1.StageConfig && len(_package.ConfigMap) > 0 { + cm := &corev1.ConfigMap{} + key := client.ObjectKey{Namespace: r.opts.Namespace, Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhookNode.GetSkyhook().Name, _package.Name, _package.Version))} + if err := r.Get(ctx, key, cm); err == nil { + job.Annotations[annotationConfigHash] = configDataHash(cm.Data) + } else if !apierrors.IsNotFound(err) { + return fmt.Errorf("error getting config map %s for job: %w", key.Name, err) + } + } + // setup ownership of the job we created so it and its child pods GC with the Skyhook CR if err := ctrl.SetControllerReference(skyhookNode.GetSkyhook().NodeWright, job, r.scheme); err != nil { return fmt.Errorf("error setting ownership: %w", err) diff --git a/operator/internal/controller/swap_test.go b/operator/internal/controller/swap_test.go index 6c13f921d..0350c1334 100644 --- a/operator/internal/controller/swap_test.go +++ b/operator/internal/controller/swap_test.go @@ -156,6 +156,32 @@ var _ = Describe("Jobs execution swap", func() { Expect(pods.Items).To(BeEmpty()) }) + It("stamps a config Job with a hash of the ConfigMap data it will mount", func() { + data := map[string]string{"a.properties": "new"} + cmPkg := pkg.DeepCopy() + cmPkg.ConfigMap = data + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + seeded, err := wrapper.NewSkyhookNodeOnly(node, skyhookName) + Expect(err).ToNot(HaveOccurred()) + Expect(seeded.Upsert(cmPkg.PackageRef, image, v1alpha1.StateInProgress, v1alpha1.StageConfig, 0, "")).To(Succeed()) + scr := &v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: skyhookName, Namespace: namespace, Generation: 1}, + Spec: v1alpha1.NodeWrightSpec{Packages: v1alpha1.Packages{"tuning": *cmPkg}}, + } + cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "gpu-init-tuning-1.0.0"}, Data: data} + r, c := newReconciler(node, scr, cm) + skyhookNode, err := wrapper.NewSkyhookNode(node, scr) + Expect(err).ToNot(HaveOccurred()) + + Expect(r.ApplyPackage(ctx, GinkgoLogr, nil, skyhookNode, cmPkg, false)).To(Succeed()) + + var jobs batchv1.JobList + Expect(c.List(ctx, &jobs, client.InNamespace(namespace))).To(Succeed()) + Expect(jobs.Items).To(HaveLen(1)) + Expect(jobs.Items[0].Labels).To(HaveKeyWithValue(stageLabel, string(v1alpha1.StageConfig))) + Expect(jobs.Items[0].Annotations).To(HaveKeyWithValue(annotationConfigHash, configDataHash(data))) + }) + It("starts one package per node per pass under serial", func() { nodes := []corev1.Node{{ObjectMeta: metav1.ObjectMeta{Name: "worker-a"}}, {ObjectMeta: metav1.ObjectMeta{Name: "worker-b"}}} other := v1alpha1.Package{PackageRef: v1alpha1.PackageRef{Name: "other", Version: "1.0.0"}, Image: image} From 3048e1930efb82b58365fc6a584e6926754faf54 Mon Sep 17 00:00:00 2001 From: Noah Kontur Date: Mon, 12 Oct 2026 12:32:15 -0400 Subject: [PATCH 2/5] fix(operator): read the config map uncached for the config-hash check A stale cache could match the old hash or miss the ConfigMap, letting a superseded completion record. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Noah Kontur --- operator/internal/controller/job_controller.go | 7 ++++++- operator/internal/controller/skyhook_controller.go | 7 ++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/operator/internal/controller/job_controller.go b/operator/internal/controller/job_controller.go index 2550d2524..c75da71b6 100644 --- a/operator/internal/controller/job_controller.go +++ b/operator/internal/controller/job_controller.go @@ -260,7 +260,12 @@ func (r *JobReconciler) configSuperseded(ctx context.Context, job *batchv1.Job, } cm := &corev1.ConfigMap{} key := types.NamespacedName{Namespace: job.Namespace, Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", pkg.Skyhook, pkg.Name, pkg.Version))} - if err := r.Get(ctx, key, cm); err != nil { + // uncached: a stale cache could match the old hash and let a superseded completion record + var reader client.Reader = r.Client + if r.uncached != nil { + reader = r.uncached + } + if err := reader.Get(ctx, key, cm); err != nil { if apierrors.IsNotFound(err) { return false, nil } diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index f65383a77..b574095d3 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -3878,7 +3878,12 @@ func (r *SkyhookReconciler) ApplyPackage(ctx context.Context, logger logr.Logger if stage == v1alpha1.StageConfig && len(_package.ConfigMap) > 0 { cm := &corev1.ConfigMap{} key := client.ObjectKey{Namespace: r.opts.Namespace, Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhookNode.GetSkyhook().Name, _package.Name, _package.Version))} - if err := r.Get(ctx, key, cm); err == nil { + // uncached: a stale cache could stamp old data or miss the ConfigMap, skipping the check later + var reader client.Reader = r.Client + if r.uncached != nil { + reader = r.uncached + } + if err := reader.Get(ctx, key, cm); err == nil { job.Annotations[annotationConfigHash] = configDataHash(cm.Data) } else if !apierrors.IsNotFound(err) { return fmt.Errorf("error getting config map %s for job: %w", key.Name, err) From 90a64209887dbf92825bc144cfc13f3a717feb17 Mon Sep 17 00:00:00 2001 From: Noah Kontur Date: Wed, 21 Oct 2026 13:38:13 -0400 Subject: [PATCH 3/5] fix(operator): close the in-flight config completion window and share the ConfigMap lookups Rewrite the package's node entry to config, in progress from a fresh read once the ConfigMap is replaced, so a completion recorded after the pass read the node cannot stand on the old data. Share the ConfigMap name and reader fallback between the Job builder, the stamp and the check, pin the uncached reads with specs that split the cached and live clients, and bring the design doc's Job-retention rules in line. Part of #776 Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Noah Kontur --- .../2026-07-10-package-execution-as-jobs.md | 4 +- operator/internal/controller/job_builder.go | 19 ++++- .../internal/controller/job_controller.go | 13 ++-- .../controller/job_controller_test.go | 21 ++++- .../controller/node_state_merge_test.go | 77 +++++++++++++++++++ .../internal/controller/skyhook_controller.go | 57 +++++++++++--- operator/internal/controller/swap_test.go | 31 ++++++++ 7 files changed, 199 insertions(+), 23 deletions(-) diff --git a/docs/designs/2026-07-10-package-execution-as-jobs.md b/docs/designs/2026-07-10-package-execution-as-jobs.md index b60d0d53f..88987b231 100644 --- a/docs/designs/2026-07-10-package-execution-as-jobs.md +++ b/docs/designs/2026-07-10-package-execution-as-jobs.md @@ -223,14 +223,14 @@ Scale: succeeded pods are retained for the success window — roughly `nodes × Job names are deterministic (same inputs as today's pod names), which makes creation idempotent — the name *is* the claim. But a finished Job keeps its name until TTL, so every "run this stage again" path must first clear the name. Five rules replace today's delete-on-success; "processed" means `state-recorded` is set, for both outcomes, and Job deletions are always **foreground** so a deterministic name frees only after its child pods are gone. 1. **Validation is scoped to unfinished Jobs.** The spec-mismatch / not-in-node-state / stage-mismatch checks apply only to unfinished Jobs; applying them to finished Jobs would delete every retained Job the moment the node progresses, gutting Goal 2. -2. **The rerun predicate.** A processed finished Job for stage `S` is deleted once node state no longer records `S` as done (entry absent, or progress at/before `S` without `S` complete — compared via the state machine's stage ordering, not string equality). **Timeout exception:** a `Failed` Job is *not* deleted while the entry sits at (`S`, `erroring`) — that pair is the timed-out state; every other value clears it, which is what sweeps away a failure the operator judged non-genuine and never wrote state for. **Unprocessed** finished Jobs are never deleted except by the two sweeps — for `Complete` that protects an unrecorded completion, and for `Failed` it keeps this sweep from racing the erroring write, a window a finite `backoffLimit` shrinks from an hour to seconds. +2. **The rerun predicate.** A processed finished Job for stage `S` is deleted once node state no longer records `S` as done (entry absent, or progress at/before `S` without `S` complete — compared via the state machine's stage ordering, not string equality). **Timeout exception:** a `Failed` Job is *not* deleted while the entry sits at (`S`, `erroring`) — that pair is the timed-out state; every other value clears it, which is what sweeps away a failure the operator judged non-genuine and never wrote state for. **Unprocessed** finished Jobs are never deleted except by the two sweeps and two exceptions (the config-hash check on a config-stage completion, above, and the config-update reset's deletion of unfinished or failed Jobs, which #788 added) — for `Complete` that protects an unrecorded completion, and for `Failed` it keeps this sweep from racing the erroring write, a window a finite `backoffLimit` shrinks from an hour to seconds. 3. **`AlreadyExists` on create is not blindly benign.** GET the Job: unfinished and matching → another pass won the race, return. Timed out → return; the timeout is doing its job (this branch is what absorbs level-triggered recreate attempts on an erroring stage). Not yet marked → don't delete, whatever the outcome: a `Complete` one holds an unrecorded completion, and a `Failed` one has not yet had its chance to write `erroring`, so deleting it would take its retained attempts with it and restart the stage on a fresh budget. This path reaches that window first, because a finished Job does not satisfy the existence gate. Otherwise processed, or unfinished-but-mismatched → foreground-delete and requeue, without recording an `in_progress` that isn't happening. 4. **Orphaned-node sweep.** Jobs whose node no longer exists are deleted regardless of status — an unfinished one would otherwise churn PodGC↔replacement forever against the missing node, and a finished one has no node state left to claim it. 5. **Reboot-reset sweep.** When `REAPPLY_ON_REBOOT` resets a node's state, that node's Jobs for the Skyhook are deleted regardless of status, including unprocessed `Complete` ones — a completion from a previous boot must not land on freshly-reset state. Rule 2's protection of unprocessed completions yields to this sweep. Two retention carve-outs follow and are accepted: a successful no-interrupt **uninstall** Job's completion *is* entry removal, so it satisfies the absent-entry branch immediately and gets no retention window; and because uninstall precedes apply in the ordering, starting an uninstall also clears the package's retained apply/config Jobs. Uninstall is the "remove all trace" operation, so prompt cleanup is coherent — and crash-loop *failures* of uninstall still retain logs (Active Job). The cost is that a *successful* uninstall's output is unrecoverable, since the Job and its pod are gone before the success TTL applies: a known limitation, tracked in [#443](https://github.com/NVIDIA/nodewright/issues/443) and deferred to a later release. -The existence gates move with these rules: the "is this stage in flight" check and the "wait until nothing is running" interrupt gate both count **unfinished Jobs only** (a retained Succeeded pod must not hold an interrupt hostage — node cordoned and drained — until its TTL expires), and config-update handling deletes the package's **unfinished or terminally failed** Job rather than its child pod (which would just trigger a same-config replacement) — a config change invalidates that failure history anyway, and the deterministic name has to be freed for the rerun. +The existence gates move with these rules: the "is this stage in flight" check and the "wait until nothing is running" interrupt gate both count **unfinished Jobs only** (a retained Succeeded pod must not hold an interrupt hostage — node cordoned and drained — until its TTL expires), and config-update handling deletes the package's **unfinished, terminally failed or unrecorded config-stage `Complete`** Job rather than its child pod (it clears executors on the reset nodes; the `Complete` one is also dropped by the config-hash check when the ConfigMap has since changed) (which would just trigger a same-config replacement) — a config change invalidates that failure history anyway, and the deterministic name has to be freed for the rerun. ### Labels diff --git a/operator/internal/controller/job_builder.go b/operator/internal/controller/job_builder.go index e7acb8b9b..1609266b4 100644 --- a/operator/internal/controller/job_builder.go +++ b/operator/internal/controller/job_builder.go @@ -34,6 +34,7 @@ import ( apiequality "k8s.io/apimachinery/pkg/api/equality" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" ) // doneContainerName replaces the forever-running pause main container. A Job records @@ -298,7 +299,7 @@ func createPodFromPackage(opts SkyhookOperatorOptions, _package *v1alpha1.Packag VolumeSource: corev1.VolumeSource{ ConfigMap: &corev1.ConfigMapVolumeSource{ LocalObjectReference: corev1.LocalObjectReference{ - Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.Name, _package.Name, _package.Version)), + Name: packageConfigMapName(skyhook.Name, _package.Name, _package.Version), }, }, }, @@ -655,3 +656,19 @@ func podMatchesPackage(opts SkyhookOperatorOptions, _package *v1alpha1.Package, return true } + +// packageConfigMapName is the name of a package's ConfigMap, and the one the package pod mounts it +// from. The config-hash check looks the ConfigMap up by this name, so a copy that drifted would turn +// the check off without an error. +func packageConfigMapName(skyhook, pkg, version string) string { + return strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook, pkg, version)) +} + +// uncachedReader is the apiserver reader when one is configured, else the (cached) client, which is +// what the fake-client tests use. +func uncachedReader(uncached client.Reader, c client.Client) client.Reader { + if uncached != nil { + return uncached + } + return c +} diff --git a/operator/internal/controller/job_controller.go b/operator/internal/controller/job_controller.go index cb930de26..4b16aedb2 100644 --- a/operator/internal/controller/job_controller.go +++ b/operator/internal/controller/job_controller.go @@ -25,7 +25,6 @@ import ( "fmt" "slices" "sort" - "strings" "time" "github.com/NVIDIA/nodewright/operator/api/nodewright/v1alpha1" @@ -134,7 +133,9 @@ const ( type JobReconciler struct { client.Client // uncached reads straight from the apiserver: a Node re-read after a patch conflict (see - // patchNodeState), and a terminal Job's child pods (see packageJobRestarts). + // patchNodeState), a terminal Job's child pods (see packageJobRestarts), and the package + // ConfigMap behind the config-hash check (see configSuperseded), where a stale cache would + // match the old hash. uncached client.Reader recorder events.EventRecorder opts JobOperatorOptions @@ -318,13 +319,9 @@ func (r *JobReconciler) configSuperseded(ctx context.Context, job *batchv1.Job, return false, nil } cm := &corev1.ConfigMap{} - key := types.NamespacedName{Namespace: job.Namespace, Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", pkg.Skyhook, pkg.Name, pkg.Version))} + key := types.NamespacedName{Namespace: job.Namespace, Name: packageConfigMapName(pkg.Skyhook, pkg.Name, pkg.Version)} // uncached: a stale cache could match the old hash and let a superseded completion record - var reader client.Reader = r.Client - if r.uncached != nil { - reader = r.uncached - } - if err := reader.Get(ctx, key, cm); err != nil { + if err := uncachedReader(r.uncached, r.Client).Get(ctx, key, cm); err != nil { if apierrors.IsNotFound(err) { return false, nil } diff --git a/operator/internal/controller/job_controller_test.go b/operator/internal/controller/job_controller_test.go index 81ca0529d..b2901f8ae 100644 --- a/operator/internal/controller/job_controller_test.go +++ b/operator/internal/controller/job_controller_test.go @@ -243,7 +243,7 @@ var _ = Describe("JobReconcile", func() { configMap := func(data map[string]string) *corev1.ConfigMap { return &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{ Namespace: namespace, - Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhookName, pkgRef.Name, pkgRef.Version)), + Name: packageConfigMapName(skyhookName, pkgRef.Name, pkgRef.Version), }, Data: data} } configJob := func(stamped map[string]string) *batchv1.Job { @@ -267,6 +267,25 @@ var _ = Describe("JobReconcile", func() { Expect(apierrors.IsNotFound(err)).To(BeTrue(), "the stale Job is deleted so the stage re-runs") }) + It("judges the Job by the apiserver's ConfigMap, not a cache that still holds the old one", func() { + node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) + job := configJob(map[string]string{"a.properties": "old"}) + + scheme := runtime.NewScheme() + Expect(corev1.AddToScheme(scheme)).To(Succeed()) + Expect(batchv1.AddToScheme(scheme)).To(Succeed()) + Expect(v1alpha1.AddToScheme(scheme)).To(Succeed()) + // the cache still serves the data the Job ran on; only the live reader has the new one + cached := fake.NewClientBuilder().WithScheme(scheme).WithObjects(node, job, configMap(map[string]string{"a.properties": "old"})).Build() + live := fake.NewClientBuilder().WithScheme(scheme).WithObjects(configMap(map[string]string{"a.properties": "new"})).Build() + r := NewJobReconciler(cached, live, k8sfake.NewClientset(), events.NewFakeRecorder(50), validOpts().JobOperatorOptions) + + _, err := r.JobReconcile(ctx, job) + Expect(err).ToNot(HaveOccurred()) + + Expect(getNodeState(r)[pkgRef.GetUniqueName()].State).To(Equal(v1alpha1.StateInProgress)) + }) + It("records a Job whose data still matches", func() { node := nodeWithState(v1alpha1.StateInProgress, v1alpha1.StageConfig) data := map[string]string{"a.properties": "new"} diff --git a/operator/internal/controller/node_state_merge_test.go b/operator/internal/controller/node_state_merge_test.go index fd2155496..e79d0b121 100644 --- a/operator/internal/controller/node_state_merge_test.go +++ b/operator/internal/controller/node_state_merge_test.go @@ -463,3 +463,80 @@ var _ = Describe("saveNodeChanges conflict retry", func() { Expect(reader.calls).To(BeNumerically(">=", 1)) }) }) + +var _ = Describe("restartConfigStage", func() { + const skyhookName = "gpu-init" + const nodeName = "worker-1" + key := nodeStateAnnotationKey(skyhookName) + pkg := &v1alpha1.Package{PackageRef: v1alpha1.PackageRef{Name: "a", Version: "1.0.0"}, Image: "img"} + + build := func(entry v1alpha1.PackageStatus) (*SkyhookReconciler, client.Client) { + raw, err := json.Marshal(v1alpha1.NodeState{"a|1.0.0": entry}) + Expect(err).ToNot(HaveOccurred()) + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName, Annotations: map[string]string{key: string(raw)}}} + + scheme := runtime.NewScheme() + Expect(corev1.AddToScheme(scheme)).To(Succeed()) + Expect(batchv1.AddToScheme(scheme)).To(Succeed()) + Expect(v1alpha1.AddToScheme(scheme)).To(Succeed()) + + scr := &v1alpha1.NodeWright{ObjectMeta: metav1.ObjectMeta{Name: skyhookName, Namespace: "skyhook"}} + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(node, scr).Build() + r, err := NewSkyhookReconciler(scheme, c, nil, k8sfake.NewClientset(), events.NewFakeRecorder(10), SkyhookOperatorOptions{ + Namespace: "skyhook", CopyDirRoot: "/var/lib/skyhook", AgentLogRoot: "/var/log/skyhook", + RuntimeRequiredTaint: "skyhook.nvidia.com=runtime-required:NoSchedule", + AgentImage: "ghcr.io/nvidia/skyhook/agent:1.2.3", + PauseImage: "registry.k8s.io/pause:3.10", MaxInterval: 10 * time.Minute, + JobOperatorOptions: JobOperatorOptions{JobTTLSucceeded: time.Hour, JobTTLFailed: 24 * time.Hour, JobStageTimeout: time.Hour}, + }) + Expect(err).ToNot(HaveOccurred()) + return r, c + } + entry := func(stage v1alpha1.Stage, state v1alpha1.State) v1alpha1.PackageStatus { + return v1alpha1.PackageStatus{Name: "a", Version: "1.0.0", Image: "img", Stage: stage, State: state} + } + after := func(c client.Client) v1alpha1.PackageStatus { + var got corev1.Node + Expect(c.Get(ctx, types.NamespacedName{Name: nodeName}, &got)).To(Succeed()) + state, err := parseNodeState(&got, key) + Expect(err).ToNot(HaveOccurred()) + return state["a|1.0.0"] + } + + DescribeTable("rewrites an entry a concurrent writer moved on", + func(stage v1alpha1.Stage, state v1alpha1.State) { + r, c := build(entry(stage, state)) + Expect(r.restartConfigStage(ctx, nodeName, skyhookName, pkg)).To(Succeed()) + got := after(c) + Expect(got.Stage).To(Equal(v1alpha1.StageConfig)) + Expect(got.State).To(Equal(v1alpha1.StateInProgress)) + }, + Entry("a completion recorded at config", v1alpha1.StageConfig, v1alpha1.StateComplete), + Entry("an erroring config", v1alpha1.StageConfig, v1alpha1.StateErroring), + Entry("an interrupt in progress", v1alpha1.StageInterrupt, v1alpha1.StateInProgress), + Entry("a finished post-interrupt", v1alpha1.StagePostInterrupt, v1alpha1.StateComplete), + ) + + DescribeTable("leaves the entry alone", + func(stage v1alpha1.Stage, state v1alpha1.State) { + r, c := build(entry(stage, state)) + Expect(r.restartConfigStage(ctx, nodeName, skyhookName, pkg)).To(Succeed()) + Expect(after(c)).To(Equal(entry(stage, state))) + }, + Entry("already at config in progress", v1alpha1.StageConfig, v1alpha1.StateInProgress), + Entry("still applying", v1alpha1.StageApply, v1alpha1.StateInProgress), + Entry("upgrading", v1alpha1.StageUpgrade, v1alpha1.StateComplete), + Entry("uninstalling", v1alpha1.StageUninstall, v1alpha1.StateInProgress), + ) + + It("does not create an entry that is gone", func() { + r, c := build(entry(v1alpha1.StageConfig, v1alpha1.StateComplete)) + other := &v1alpha1.Package{PackageRef: v1alpha1.PackageRef{Name: "b", Version: "1.0.0"}, Image: "img"} + Expect(r.restartConfigStage(ctx, nodeName, skyhookName, other)).To(Succeed()) + var got corev1.Node + Expect(c.Get(ctx, types.NamespacedName{Name: nodeName}, &got)).To(Succeed()) + state, err := parseNodeState(&got, key) + Expect(err).ToNot(HaveOccurred()) + Expect(state).ToNot(HaveKey("b|1.0.0")) + }) +}) diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index 4c65eac52..9d97977b6 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -340,8 +340,9 @@ func NewSkyhookReconciler(schema *runtime.Scheme, c client.Client, uncached clie type SkyhookReconciler struct { client.Client // uncached reads straight from the apiserver when a Node patch conflict requires the - // mutation to be recomputed from current state. Nil falls back to the cached client, - // which is what the fake-client tests use. + // mutation to be recomputed from current state, and for the package ConfigMap the config-hash + // stamp is taken from. Nil falls back to the cached client, which is what the fake-client + // tests use. uncached client.Reader scheme *runtime.Scheme recorder events.EventRecorder @@ -2394,6 +2395,16 @@ func (r *SkyhookReconciler) HandleConfigUpdates(ctx context.Context, clusterStat return false, false, fmt.Errorf("error updating config map [%s]: %w", newConfigMap.Name, err) } + // A node already at (config, in_progress) is unchanged by the reset above, so nothing was + // written for it, and a completion JobReconciler recorded after this pass read the node + // would stand on the old data. Now that the ConfigMap is replaced, any later completion + // fails the config-hash check, so rewriting from a fresh read closes the rest. + for name := range resettable { + if err := r.restartConfigStage(ctx, name, skyhook.GetSkyhook().Name, &_package); err != nil { + return false, false, err + } + } + return true, false, nil } @@ -2406,6 +2417,34 @@ func (r *SkyhookReconciler) HandleConfigUpdates(ctx context.Context, clusterStat return false, false, nil } +// restartConfigStage puts the package back at (config, in_progress) on a node from a fresh read, if a +// concurrent writer moved it on from the reset HandleConfigUpdates made. An entry that is gone, or at +// a stage the reset does not own (uninstall, apply, upgrade), is left alone. +func (r *SkyhookReconciler) restartConfigStage(ctx context.Context, nodeName, skyhookName string, _package *v1alpha1.Package) error { + return patchNodeState(ctx, r.dal, r.uncached, r.Client, nodeName, func(node *corev1.Node) (bool, error) { + skyhookNode, err := wrapper.NewSkyhookNodeOnly(node, skyhookName) + if err != nil { + return false, fmt.Errorf("creating node wrapper for %s: %w", nodeName, err) + } + status, found := skyhookNode.PackageStatus(_package.GetUniqueName()) + if !found { + return false, nil + } + switch status.Stage { + case v1alpha1.StageConfig, v1alpha1.StageInterrupt, v1alpha1.StagePostInterrupt: + default: + return false, nil + } + if status.Stage == v1alpha1.StageConfig && status.State == v1alpha1.StateInProgress { + return false, nil + } + if err := skyhookNode.Upsert(_package.PackageRef, _package.Image, v1alpha1.StateInProgress, v1alpha1.StageConfig, 0, _package.ContainerSHA); err != nil { + return false, fmt.Errorf("restarting config stage on node %s: %w", nodeName, err) + } + return true, nil + }) +} + func (r *SkyhookReconciler) UpsertConfigmaps(ctx context.Context, skyhook SkyhookNodes, clusterState *clusterState) (bool, bool, error) { updated := false pendingSync := false @@ -2424,7 +2463,7 @@ func (r *SkyhookReconciler) UpsertConfigmaps(ctx context.Context, skyhook Skyhoo // clean up from an update shouldExist := make(map[string]struct{}) for _, _package := range skyhook.GetSkyhook().Spec.Packages { - shouldExist[strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version))] = struct{}{} + shouldExist[packageConfigMapName(skyhook.GetSkyhook().Name, _package.Name, _package.Version)] = struct{}{} } for k, v := range existingCMs { @@ -2466,7 +2505,7 @@ func (r *SkyhookReconciler) UpsertConfigmaps(ctx context.Context, skyhook Skyhoo newCM := &corev1.ConfigMap{ ObjectMeta: metav1.ObjectMeta{ - Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version)), + Name: packageConfigMapName(skyhook.GetSkyhook().Name, _package.Name, _package.Version), Namespace: r.opts.Namespace, Labels: map[string]string{ fmt.Sprintf("%s/name", v1alpha1.METADATA_PREFIX): skyhook.GetSkyhook().Name, @@ -2484,7 +2523,7 @@ func (r *SkyhookReconciler) UpsertConfigmaps(ctx context.Context, skyhook Skyhoo return false, false, fmt.Errorf("error setting ownership of cm: %w", err) } - if existingCM, ok := existingCMs[strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhook.GetSkyhook().Name, _package.Name, _package.Version))]; ok { + if existingCM, ok := existingCMs[packageConfigMapName(skyhook.GetSkyhook().Name, _package.Name, _package.Version)]; ok { updatedConfigMap, pendingConfigSync, err := r.HandleConfigUpdates(ctx, clusterState, skyhook, _package, &existingCM, newCM) if err != nil { return false, false, fmt.Errorf("error updating config map [%s]: %w", newCM.Name, err) @@ -3913,13 +3952,9 @@ func (r *SkyhookReconciler) ApplyPackage(ctx context.Context, logger logr.Logger // is not recorded as the new data's (see JobReconciler.configSuperseded). if stage == v1alpha1.StageConfig && len(_package.ConfigMap) > 0 { cm := &corev1.ConfigMap{} - key := client.ObjectKey{Namespace: r.opts.Namespace, Name: strings.ToLower(fmt.Sprintf("%s-%s-%s", skyhookNode.GetSkyhook().Name, _package.Name, _package.Version))} + key := client.ObjectKey{Namespace: r.opts.Namespace, Name: packageConfigMapName(skyhookNode.GetSkyhook().Name, _package.Name, _package.Version)} // uncached: a stale cache could stamp old data or miss the ConfigMap, skipping the check later - var reader client.Reader = r.Client - if r.uncached != nil { - reader = r.uncached - } - if err := reader.Get(ctx, key, cm); err == nil { + if err := uncachedReader(r.uncached, r.Client).Get(ctx, key, cm); err == nil { job.Annotations[annotationConfigHash] = configDataHash(cm.Data) } else if !apierrors.IsNotFound(err) { return fmt.Errorf("error getting config map %s for job: %w", key.Name, err) diff --git a/operator/internal/controller/swap_test.go b/operator/internal/controller/swap_test.go index 187f72723..77640be8a 100644 --- a/operator/internal/controller/swap_test.go +++ b/operator/internal/controller/swap_test.go @@ -182,6 +182,37 @@ var _ = Describe("Jobs execution swap", func() { Expect(jobs.Items[0].Annotations).To(HaveKeyWithValue(annotationConfigHash, configDataHash(data))) }) + It("stamps the hash from the apiserver's ConfigMap, not a cache that still holds the old one", func() { + oldData := map[string]string{"a.properties": "old"} + newData := map[string]string{"a.properties": "new"} + cmPkg := pkg.DeepCopy() + cmPkg.ConfigMap = newData + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + seeded, err := wrapper.NewSkyhookNodeOnly(node, skyhookName) + Expect(err).ToNot(HaveOccurred()) + Expect(seeded.Upsert(cmPkg.PackageRef, image, v1alpha1.StateInProgress, v1alpha1.StageConfig, 0, "")).To(Succeed()) + scr := &v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: skyhookName, Namespace: namespace, Generation: 1}, + Spec: v1alpha1.NodeWrightSpec{Packages: v1alpha1.Packages{"tuning": *cmPkg}}, + } + cmName := packageConfigMapName(skyhookName, cmPkg.Name, cmPkg.Version) + r, c := newReconciler(node, scr, &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: cmName}, Data: oldData}) + + scheme := runtime.NewScheme() + Expect(corev1.AddToScheme(scheme)).To(Succeed()) + r.uncached = fake.NewClientBuilder().WithScheme(scheme). + WithObjects(&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: cmName}, Data: newData}).Build() + + skyhookNode, err := wrapper.NewSkyhookNode(node, scr) + Expect(err).ToNot(HaveOccurred()) + Expect(r.ApplyPackage(ctx, GinkgoLogr, nil, skyhookNode, cmPkg, false)).To(Succeed()) + + var jobs batchv1.JobList + Expect(c.List(ctx, &jobs, client.InNamespace(namespace))).To(Succeed()) + Expect(jobs.Items).To(HaveLen(1)) + Expect(jobs.Items[0].Annotations).To(HaveKeyWithValue(annotationConfigHash, configDataHash(newData))) + }) + It("starts one package per node per pass under serial", func() { nodes := []corev1.Node{{ObjectMeta: metav1.ObjectMeta{Name: "worker-a"}}, {ObjectMeta: metav1.ObjectMeta{Name: "worker-b"}}} other := v1alpha1.Package{PackageRef: v1alpha1.PackageRef{Name: "other", Version: "1.0.0"}, Image: image} From 03e975f4fbea0ae19c58b8fcb1b1f11f44381cb1 Mon Sep 17 00:00:00 2001 From: Noah Kontur Date: Thu, 22 Oct 2026 13:40:04 -0400 Subject: [PATCH 4/5] fix(controller): restart config stage on every reset node before returning an error Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Noah Kontur --- docs/designs/2026-07-10-package-execution-as-jobs.md | 2 +- operator/internal/controller/skyhook_controller.go | 7 ++++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/docs/designs/2026-07-10-package-execution-as-jobs.md b/docs/designs/2026-07-10-package-execution-as-jobs.md index 88987b231..b4b348c0c 100644 --- a/docs/designs/2026-07-10-package-execution-as-jobs.md +++ b/docs/designs/2026-07-10-package-execution-as-jobs.md @@ -223,7 +223,7 @@ Scale: succeeded pods are retained for the success window — roughly `nodes × Job names are deterministic (same inputs as today's pod names), which makes creation idempotent — the name *is* the claim. But a finished Job keeps its name until TTL, so every "run this stage again" path must first clear the name. Five rules replace today's delete-on-success; "processed" means `state-recorded` is set, for both outcomes, and Job deletions are always **foreground** so a deterministic name frees only after its child pods are gone. 1. **Validation is scoped to unfinished Jobs.** The spec-mismatch / not-in-node-state / stage-mismatch checks apply only to unfinished Jobs; applying them to finished Jobs would delete every retained Job the moment the node progresses, gutting Goal 2. -2. **The rerun predicate.** A processed finished Job for stage `S` is deleted once node state no longer records `S` as done (entry absent, or progress at/before `S` without `S` complete — compared via the state machine's stage ordering, not string equality). **Timeout exception:** a `Failed` Job is *not* deleted while the entry sits at (`S`, `erroring`) — that pair is the timed-out state; every other value clears it, which is what sweeps away a failure the operator judged non-genuine and never wrote state for. **Unprocessed** finished Jobs are never deleted except by the two sweeps and two exceptions (the config-hash check on a config-stage completion, above, and the config-update reset's deletion of unfinished or failed Jobs, which #788 added) — for `Complete` that protects an unrecorded completion, and for `Failed` it keeps this sweep from racing the erroring write, a window a finite `backoffLimit` shrinks from an hour to seconds. +2. **The rerun predicate.** A processed finished Job for stage `S` is deleted once node state no longer records `S` as done (entry absent, or progress at/before `S` without `S` complete — compared via the state machine's stage ordering, not string equality). **Timeout exception:** a `Failed` Job is *not* deleted while the entry sits at (`S`, `erroring`) — that pair is the timed-out state; every other value clears it, which is what sweeps away a failure the operator judged non-genuine and never wrote state for. **Unprocessed** finished Jobs are never deleted except by the two sweeps and two exceptions (the config-hash check on a config-stage completion, above, and the config-update reset's deletion of unfinished, failed, or unrecorded config-stage `Complete` Jobs, which #788 added) — for `Complete` that protects an unrecorded completion, and for `Failed` it keeps this sweep from racing the erroring write, a window a finite `backoffLimit` shrinks from an hour to seconds. 3. **`AlreadyExists` on create is not blindly benign.** GET the Job: unfinished and matching → another pass won the race, return. Timed out → return; the timeout is doing its job (this branch is what absorbs level-triggered recreate attempts on an erroring stage). Not yet marked → don't delete, whatever the outcome: a `Complete` one holds an unrecorded completion, and a `Failed` one has not yet had its chance to write `erroring`, so deleting it would take its retained attempts with it and restart the stage on a fresh budget. This path reaches that window first, because a finished Job does not satisfy the existence gate. Otherwise processed, or unfinished-but-mismatched → foreground-delete and requeue, without recording an `in_progress` that isn't happening. 4. **Orphaned-node sweep.** Jobs whose node no longer exists are deleted regardless of status — an unfinished one would otherwise churn PodGC↔replacement forever against the missing node, and a finished one has no node state left to claim it. 5. **Reboot-reset sweep.** When `REAPPLY_ON_REBOOT` resets a node's state, that node's Jobs for the Skyhook are deleted regardless of status, including unprocessed `Complete` ones — a completion from a previous boot must not land on freshly-reset state. Rule 2's protection of unprocessed completions yields to this sweep. diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index 9d97977b6..775c03976 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -2399,11 +2399,16 @@ func (r *SkyhookReconciler) HandleConfigUpdates(ctx context.Context, clusterStat // written for it, and a completion JobReconciler recorded after this pass read the node // would stand on the old data. Now that the ConfigMap is replaced, any later completion // fails the config-hash check, so rewriting from a fresh read closes the rest. + // Try every node: the next pass sees no diff, so a node skipped here is not retried. + var restartErrs []error for name := range resettable { if err := r.restartConfigStage(ctx, name, skyhook.GetSkyhook().Name, &_package); err != nil { - return false, false, err + restartErrs = append(restartErrs, err) } } + if len(restartErrs) > 0 { + return false, false, utilerrors.NewAggregate(restartErrs) + } return true, false, nil } From 462ff47e65d4f39cc14616cc4eb6fa2a7fad04a7 Mon Sep 17 00:00:00 2001 From: Noah Kontur Date: Fri, 23 Oct 2026 13:00:45 -0400 Subject: [PATCH 5/5] docs(design): tidy the config-update deletion note and describe the fresh-read rewrite Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Noah Kontur --- docs/designs/2026-07-10-package-execution-as-jobs.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/designs/2026-07-10-package-execution-as-jobs.md b/docs/designs/2026-07-10-package-execution-as-jobs.md index b4b348c0c..a6e98f194 100644 --- a/docs/designs/2026-07-10-package-execution-as-jobs.md +++ b/docs/designs/2026-07-10-package-execution-as-jobs.md @@ -230,7 +230,7 @@ Job names are deterministic (same inputs as today's pod names), which makes crea Two retention carve-outs follow and are accepted: a successful no-interrupt **uninstall** Job's completion *is* entry removal, so it satisfies the absent-entry branch immediately and gets no retention window; and because uninstall precedes apply in the ordering, starting an uninstall also clears the package's retained apply/config Jobs. Uninstall is the "remove all trace" operation, so prompt cleanup is coherent — and crash-loop *failures* of uninstall still retain logs (Active Job). The cost is that a *successful* uninstall's output is unrecoverable, since the Job and its pod are gone before the success TTL applies: a known limitation, tracked in [#443](https://github.com/NVIDIA/nodewright/issues/443) and deferred to a later release. -The existence gates move with these rules: the "is this stage in flight" check and the "wait until nothing is running" interrupt gate both count **unfinished Jobs only** (a retained Succeeded pod must not hold an interrupt hostage — node cordoned and drained — until its TTL expires), and config-update handling deletes the package's **unfinished, terminally failed or unrecorded config-stage `Complete`** Job rather than its child pod (it clears executors on the reset nodes; the `Complete` one is also dropped by the config-hash check when the ConfigMap has since changed) (which would just trigger a same-config replacement) — a config change invalidates that failure history anyway, and the deterministic name has to be freed for the rerun. +The existence gates move with these rules: the "is this stage in flight" check and the "wait until nothing is running" interrupt gate both count **unfinished Jobs only** (a retained Succeeded pod must not hold an interrupt hostage — node cordoned and drained — until its TTL expires), and config-update handling deletes the package's **unfinished, terminally failed or unrecorded config-stage `Complete`** Job on the reset nodes, rather than just its child pod, which would only trigger a same-config replacement. A config change invalidates that failure history anyway, and the deterministic name has to be freed for the rerun. The `Complete` one is also dropped by the config-hash check when the ConfigMap has since changed; and once the ConfigMap is replaced, each reset node's entry is rewritten from a fresh read (`restartConfigStage`), so a completion recorded between the pass's node read and the ConfigMap update does not survive on the old data. ### Labels