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 47f135a30..a6e98f194 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 @@ Interrupt Jobs are the exception and keep an unbounded limit. Under `OnFailure` 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. For each unprocessed Job on the node: -- **`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. @@ -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, 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. 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 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 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 14d7b1914..4b16aedb2 100644 --- a/operator/internal/controller/job_controller.go +++ b/operator/internal/controller/job_controller.go @@ -20,6 +20,8 @@ package controller import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" "slices" "sort" @@ -74,6 +76,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" @@ -126,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 @@ -285,6 +294,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 } @@ -292,6 +310,41 @@ 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: packageConfigMapName(pkg.Skyhook, pkg.Name, pkg.Version)} + // uncached: a stale cache could match the old hash and let a superseded completion record + if err := uncachedReader(r.uncached, r.Client).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 fd7df76de..b2901f8ae 100644 --- a/operator/internal/controller/job_controller_test.go +++ b/operator/internal/controller/job_controller_test.go @@ -237,6 +237,95 @@ 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: packageConfigMapName(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("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"} + 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/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 6c52e9864..938205a5b 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 @@ -2516,6 +2517,21 @@ 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. + // 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 { + restartErrs = append(restartErrs, err) + } + } + if len(restartErrs) > 0 { + return false, false, utilerrors.NewAggregate(restartErrs) + } + return true, false, nil } @@ -2528,6 +2544,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 @@ -2546,7 +2590,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 { @@ -2588,7 +2632,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, @@ -2606,7 +2650,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) @@ -4034,6 +4078,19 @@ 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: packageConfigMapName(skyhookNode.GetSkyhook().Name, _package.Name, _package.Version)} + // uncached: a stale cache could stamp old data or miss the ConfigMap, skipping the check later + 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) + } + } + // 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 4b7c4171f..a835da9af 100644 --- a/operator/internal/controller/swap_test.go +++ b/operator/internal/controller/swap_test.go @@ -156,6 +156,63 @@ 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("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}