diff --git a/chart/templates/nodewright-crd.yaml b/chart/templates/nodewright-crd.yaml index 68d871d35..7b194121a 100644 --- a/chart/templates/nodewright-crd.yaml +++ b/chart/templates/nodewright-crd.yaml @@ -847,6 +847,13 @@ spec: image: description: Image for the package type: string + interruptPending: + description: |- + InterruptPending is set at apply or upgrade for a package that declares its own interrupt, + and cleared once that interrupt has completed or the package reaches post-interrupt. While set, it is still owed on this + node, so a config change that matches no configInterrupts does not let it finish without it. + Entries written before this field existed read as not pending. + type: boolean name: description: Name is the name of the package type: string diff --git a/chart/templates/skyhook-crd.yaml b/chart/templates/skyhook-crd.yaml index 73452dd42..62e770ad2 100644 --- a/chart/templates/skyhook-crd.yaml +++ b/chart/templates/skyhook-crd.yaml @@ -816,6 +816,10 @@ spec: image: description: Image for the package type: string + interruptPending: + description: InterruptPending mirrors the nodewright.nvidia.com + field of the same name. + type: boolean name: description: Name is the name of the package type: string diff --git a/docs/architecture/lifecycle.md b/docs/architecture/lifecycle.md index 5a7590651..e8a9757b8 100644 --- a/docs/architecture/lifecycle.md +++ b/docs/architecture/lifecycle.md @@ -377,7 +377,7 @@ the spec and it converges again: |---|---| | Package `version` increases | Package re-enters at `upgrade` | | Package `version` decreases | With `uninstall.enabled`, rejected unless the package was explicitly uninstalled first; without it, the new version applies | -| A `configMap` key | Nothing happens until every node is complete or one is erroring at `config`, `interrupt` or `post-interrupt`. Then every node at `config`, `interrupt`, `post-interrupt` or complete re-enters at `config`, and its unfinished Jobs are cleared; `configInterrupts` decides whether that costs an interrupt. Nodes at `uninstall`, `apply` or `upgrade`, or without the package, are not reset and pick up the new ConfigMap when they reach `config` | +| A `configMap` key | Nothing happens until every node is complete or one is erroring at `config`, `interrupt` or `post-interrupt`. Then every node at `config`, `interrupt`, `post-interrupt` or complete re-enters at `config`, and its unfinished Jobs are cleared; the matching `configInterrupts` entries, plus the package's own `interrupt` on any node that has not yet run it, decide whether that costs an interrupt. Nodes at `uninstall`, `apply` or `upgrade`, or without the package, are not reset and pick up the new ConfigMap when they reach `config` | | `nodeSelectors` | Newly matching nodes are enrolled; newly excluded nodes leave scope — their host changes remain | | `image` or `containerSHA` only | No stage change — `version` is the ordering key | diff --git a/docs/user-guide/custom-resource.md b/docs/user-guide/custom-resource.md index 5fc15c947..a361b052d 100644 --- a/docs/user-guide/custom-resource.md +++ b/docs/user-guide/custom-resource.md @@ -506,6 +506,9 @@ not cost you a node reboot, while the key that does need one still gets it. literal key that does not exist, or a glob that matches nothing, is rejected by the webhook — this catches the typo that would otherwise silently skip an interrupt. +- A config change adds to the package's own [`interrupt`](#interrupt), it does not + replace it: on a node where the package's own interrupt has not yet run, that + interrupt still runs, merged with the matching `configInterrupts` entries. For secrets, do **not** put them in `configMap`. See [Providing Secrets to Packages](providing-secrets.md). @@ -644,7 +647,7 @@ To end a failing interrupt sooner, or to run it again after the verdict: A failing `uninstall-interrupt` cannot be ended by a rerun or a reset. Both remove the package's entry, an absent entry reads as already uninstalled, and so the operator abandons the uninstall interrupt rather than running it again; `package rerun --stage` cannot target `uninstall-interrupt` either. Fix the cause and delete the failed Job instead. -A configMap change does not retry the interrupt either. When the package's configMap changes while it is `erroring` at `config`, `interrupt` or `post-interrupt`, the operator deletes its interrupt Job and sends the package back through `config` with the changed keys recorded. From then on the package's own `interrupt` is not consulted: only its `configInterrupts` entries matching the changed keys decide whether an interrupt runs. With no match the package completes at `config` without its interrupt ever having succeeded, and the node is uncordoned; with a match, that entry runs instead. A configMap change is not applied at all while the package is mid-uninstall. +A configMap change retries the interrupt only on a node that still owes it. When the package's configMap changes while it is `erroring` at `config`, `interrupt` or `post-interrupt`, the operator deletes its interrupt Job and sends the package back through `config` with the changed keys recorded. On a node where the package's own `interrupt` has not yet run (the node's `interruptPending` flag is still set; it is cleared once the interrupt completes or the package reaches `post-interrupt`), that interrupt still runs after `config`, merged with any `configInterrupts` entry matching the changed keys; the highest-priority one wins (a `reboot` over a service restart). On a node where it already ran, only the matching `configInterrupts` entries decide whether an interrupt runs, so a change that matches none completes at `config` without one. A package whose node entry predates the `interruptPending` flag reads as already done. A configMap change is not applied at all while the package is mid-uninstall. ### `uninstall` diff --git a/operator/api/nodewright/v1alpha1/nodewright_types.go b/operator/api/nodewright/v1alpha1/nodewright_types.go index 2b93e682e..1cab46672 100644 --- a/operator/api/nodewright/v1alpha1/nodewright_types.go +++ b/operator/api/nodewright/v1alpha1/nodewright_types.go @@ -548,8 +548,17 @@ type NodeWrightStatus struct { type NodeState map[string]PackageStatus -// Upsert adds or updates specified state for package in the node state +// Upsert adds or updates specified state for package in the node state, carrying the entry's +// InterruptPending flag through unchanged apart from clearing it once the interrupt completes +// or at post-interrupt. func (ns *NodeState) Upsert(_package PackageRef, image string, state State, stage Stage, restarts int32, containerSHA string) bool { + return ns.UpsertOwing(_package, image, state, stage, restarts, containerSHA, false) +} + +// UpsertOwing is Upsert for a package that declares its own interrupt when ownInterrupt is set: +// an entry written at apply or upgrade then owes that interrupt until it completes or the +// package reaches post-interrupt. +func (ns *NodeState) UpsertOwing(_package PackageRef, image string, state State, stage Stage, restarts int32, containerSHA string, ownInterrupt bool) bool { if *ns == nil { *ns = make(map[string]PackageStatus) @@ -566,6 +575,15 @@ func (ns *NodeState) Upsert(_package PackageRef, image string, state State, stag } existing, ok := (*ns)[_package.GetUniqueName()] + + status.InterruptPending = ok && existing.InterruptPending + switch { + case ownInterrupt && (stage == StageApply || stage == StageUpgrade): + status.InterruptPending = true + case stage == StagePostInterrupt, stage == StageInterrupt && state == StateComplete: + status.InterruptPending = false + } + if ok && status.Equal(&existing) { return false } @@ -575,6 +593,16 @@ func (ns *NodeState) Upsert(_package PackageRef, image string, state State, stag return true } +// OwesOwnInterrupt reports whether the package declares an interrupt that has not yet run on +// this node: the entry is missing or still pending. +func (ns *NodeState) OwesOwnInterrupt(_package Package) bool { + if !_package.HasInterrupt() { + return false + } + status, ok := (*ns)[_package.GetUniqueName()] + return !ok || status.InterruptPending +} + // RemoveState removes specified package from Node State func (ns *NodeState) RemoveState(_package PackageRef) bool { if *ns == nil { @@ -653,17 +681,15 @@ func (ns *NodeState) IsComplete(packages Packages, interrupt map[string][]*Inter return false } -// HasInterrupt determines whether package has an interrupt based off of config updates and interrupts passed to it +// HasInterrupt determines whether package has an interrupt based off of config updates and interrupts passed to it. +// With config updates recorded, the interrupts matching the changed keys apply, plus the package's own +// while this node still owes it, so a config change cannot finish a package that never ran its interrupt. func (ns *NodeState) HasInterrupt(_package Package, interrupt map[string][]*Interrupt, config map[string][]string) bool { - var hasInterrupt bool - if len(config[_package.Name]) > 0 { - hasInterrupt = len(interrupt[_package.Name]) > 0 - } else { - hasInterrupt = _package.HasInterrupt() + return len(interrupt[_package.Name]) > 0 || ns.OwesOwnInterrupt(_package) } - return hasInterrupt + return _package.HasInterrupt() } func (ns *NodeState) NextStage(_package *Package, interrupt map[string][]*Interrupt, config map[string][]string) *Stage { @@ -787,6 +813,7 @@ func (ns *NodeState) ProgressSkipped(packages Packages, interrupt map[string][]* if s.Stage == StageInterrupt && s.State == StateSkipped { s.State = StateComplete + s.InterruptPending = false // nothing is left to run at (interrupt, complete) (*ns)[f.GetUniqueName()] = s ret = true } @@ -825,9 +852,16 @@ type PackageStatus struct { // Restarts are the number of times a package restarted Restarts int32 `json:"restarts,omitempty"` + + // InterruptPending is set at apply or upgrade for a package that declares its own interrupt, + // and cleared once that interrupt has completed or the package reaches post-interrupt. While set, it is still owed on this + // node, so a config change that matches no configInterrupts does not let it finish without it. + // Entries written before this field existed read as not pending. + //+optional + InterruptPending bool `json:"interruptPending,omitempty"` } -// Equal checks name, version, image, containerSHA, stage, state, and restarts +// Equal checks name, version, image, containerSHA, stage, state, restarts, and interruptPending func (left *PackageStatus) Equal(right *PackageStatus) bool { return left.Name == right.Name && left.Version == right.Version && @@ -835,7 +869,8 @@ func (left *PackageStatus) Equal(right *PackageStatus) bool { left.ContainerSHA == right.ContainerSHA && left.Stage == right.Stage && left.State == right.State && - left.Restarts == right.Restarts + left.Restarts == right.Restarts && + left.InterruptPending == right.InterruptPending } // IsInterruptStage reports whether the package is in one of the two interrupt-phase diff --git a/operator/api/nodewright/v1alpha1/nodewright_types_test.go b/operator/api/nodewright/v1alpha1/nodewright_types_test.go index acfb996f7..6c7193975 100644 --- a/operator/api/nodewright/v1alpha1/nodewright_types_test.go +++ b/operator/api/nodewright/v1alpha1/nodewright_types_test.go @@ -218,6 +218,104 @@ var _ = Describe("NodeWrightSpec BuildGraph", func() { var _ = Describe("NodeState", func() { + Describe("owed own interrupt", func() { + ref := PackageRef{Name: "foo", Version: "1.0.0"} + withReboot := Package{PackageRef: ref, Interrupt: &Interrupt{Type: REBOOT}} + plain := Package{PackageRef: ref} + unmatched := map[string][]string{"foo": {"app.conf"}} + noMatch := map[string][]*Interrupt{} + + pending := func(ns NodeState) bool { return ns["foo|1.0.0"].InterruptPending } + + It("sets it on apply and upgrade, carries it through config and interrupt, clears it at post-interrupt", func() { + ns := NodeState{} + ns.UpsertOwing(ref, "img", StateInProgress, StageApply, 0, "", true) + Expect(pending(ns)).To(BeTrue()) + ns.Upsert(ref, "img", StateComplete, StageConfig, 0, "") + Expect(pending(ns)).To(BeTrue()) + ns.Upsert(ref, "img", StateInProgress, StageInterrupt, 0, "") + Expect(pending(ns)).To(BeTrue()) + ns.Upsert(ref, "img", StateInProgress, StagePostInterrupt, 0, "") + Expect(pending(ns)).To(BeFalse()) + + // a config reset keeps it cleared + ns.Upsert(ref, "img", StateInProgress, StageConfig, 0, "") + Expect(pending(ns)).To(BeFalse()) + + // a new apply or upgrade owes it again + ns.UpsertOwing(ref, "img", StateInProgress, StageUpgrade, 0, "", true) + Expect(pending(ns)).To(BeTrue()) + }) + + It("clears it once the interrupt completes, so a config reset at (interrupt, complete) does not rerun it", func() { + ns := NodeState{} + ns.UpsertOwing(ref, "img", StateInProgress, StageApply, 0, "", true) + ns.Upsert(ref, "img", StateInProgress, StageInterrupt, 0, "") + ns.Upsert(ref, "img", StateComplete, StageInterrupt, 0, "") + Expect(pending(ns)).To(BeFalse()) + + ns.Upsert(ref, "img", StateInProgress, StageConfig, 0, "") + Expect(ns.HasInterrupt(withReboot, noMatch, unmatched)).To(BeFalse()) + Expect(ns.IsPackageComplete(withReboot, noMatch, unmatched)).To(BeFalse()) // in_progress at config, not complete + }) + + It("clears it when a skipped interrupt is promoted to complete", func() { + ns := NodeState{"foo|1.0.0": {Name: "foo", Version: "1.0.0", Stage: StageInterrupt, State: StateSkipped, InterruptPending: true}} + Expect(ns.ProgressSkipped(Packages{"foo": withReboot}, noMatch, unmatched)).To(BeTrue()) + Expect(pending(ns)).To(BeFalse()) + Expect(ns["foo|1.0.0"].State).To(Equal(StateComplete)) + }) + + It("leaves a package without its own interrupt unflagged and keeps a carried flag on plain upserts", func() { + ns := NodeState{} + ns.UpsertOwing(ref, "img", StateInProgress, StageApply, 0, "", false) + Expect(pending(ns)).To(BeFalse()) + + ns = NodeState{} + ns.UpsertOwing(ref, "img", StateInProgress, StageApply, 0, "", true) + ns.Upsert(ref, "img", StateComplete, StageApply, 0, "") + Expect(pending(ns)).To(BeTrue()) + }) + + It("reports a change when only the flag differs", func() { + ns := NodeState{"foo|1.0.0": PackageStatus{Name: "foo", Version: "1.0.0", Image: "img", Stage: StageApply, State: StateInProgress}} + Expect(ns.UpsertOwing(ref, "img", StateInProgress, StageApply, 0, "", true)).To(BeTrue()) + Expect(ns.UpsertOwing(ref, "img", StateInProgress, StageApply, 0, "", true)).To(BeFalse()) + }) + + It("owes it for a missing or pending entry and not for a settled one", func() { + Expect((&NodeState{}).OwesOwnInterrupt(withReboot)).To(BeTrue()) + Expect((&NodeState{"foo|1.0.0": {InterruptPending: true}}).OwesOwnInterrupt(withReboot)).To(BeTrue()) + Expect((&NodeState{"foo|1.0.0": {}}).OwesOwnInterrupt(withReboot)).To(BeFalse()) + Expect((&NodeState{"foo|1.0.0": {InterruptPending: true}}).OwesOwnInterrupt(plain)).To(BeFalse()) + }) + + It("keeps the own interrupt when a config change matches no configInterrupts", func() { + ns := NodeState{"foo|1.0.0": {Name: "foo", Version: "1.0.0", Stage: StageConfig, State: StateComplete, InterruptPending: true}} + Expect(ns.HasInterrupt(withReboot, noMatch, unmatched)).To(BeTrue()) + Expect(ns.IsPackageComplete(withReboot, noMatch, unmatched)).To(BeFalse()) + Expect(*ns.NextStage(&withReboot, noMatch, unmatched)).To(Equal(StageInterrupt)) + }) + + It("finishes at config once the own interrupt already ran", func() { + ns := NodeState{"foo|1.0.0": {Name: "foo", Version: "1.0.0", Stage: StageConfig, State: StateComplete}} + Expect(ns.HasInterrupt(withReboot, noMatch, unmatched)).To(BeFalse()) + Expect(ns.IsPackageComplete(withReboot, noMatch, unmatched)).To(BeTrue()) + }) + + It("still runs a matching configInterrupt after the own interrupt ran", func() { + ns := NodeState{"foo|1.0.0": {Name: "foo", Version: "1.0.0", Stage: StageConfig, State: StateComplete}} + matched := map[string][]*Interrupt{"foo": {{Type: SERVICE}}} + Expect(ns.HasInterrupt(withReboot, matched, unmatched)).To(BeTrue()) + }) + + It("ignores the flag when there are no config updates", func() { + ns := NodeState{"foo|1.0.0": {Name: "foo", Version: "1.0.0", Stage: StageConfig, State: StateComplete}} + Expect(ns.HasInterrupt(withReboot, noMatch, nil)).To(BeTrue()) + Expect(ns.HasInterrupt(plain, noMatch, nil)).To(BeFalse()) + }) + }) + Describe("RemoveState", func() { It("should report no change when the state map is nil", func() { var ns NodeState diff --git a/operator/api/v1alpha1/nodewright_conversion.go b/operator/api/v1alpha1/nodewright_conversion.go index 270f85068..17663ad17 100644 --- a/operator/api/v1alpha1/nodewright_conversion.go +++ b/operator/api/v1alpha1/nodewright_conversion.go @@ -363,6 +363,8 @@ func convertPackageStatus(in PackageStatus) nwv1.PackageStatus { Stage: nwv1.Stage(in.Stage), State: nwv1.State(in.State), Restarts: in.Restarts, + + InterruptPending: in.InterruptPending, } } diff --git a/operator/api/v1alpha1/nodewright_conversion_test.go b/operator/api/v1alpha1/nodewright_conversion_test.go index 465ba4ab8..445b9458f 100644 --- a/operator/api/v1alpha1/nodewright_conversion_test.go +++ b/operator/api/v1alpha1/nodewright_conversion_test.go @@ -129,13 +129,14 @@ func fullSkyhook() *Skyhook { NodeState: map[string]NodeState{ "node-1": { "tuning|1.2.3": PackageStatus{ - Name: "tuning", - Version: "1.2.3", - Image: "alpine", - ContainerSHA: "sha256:deadbeef", - Stage: StageConfig, - State: StateComplete, - Restarts: 2, + Name: "tuning", + Version: "1.2.3", + Image: "alpine", + ContainerSHA: "sha256:deadbeef", + Stage: StageConfig, + State: StateComplete, + Restarts: 2, + InterruptPending: true, }, }, }, diff --git a/operator/api/v1alpha1/skyhook_types.go b/operator/api/v1alpha1/skyhook_types.go index 96eb51b9e..e8a678c49 100644 --- a/operator/api/v1alpha1/skyhook_types.go +++ b/operator/api/v1alpha1/skyhook_types.go @@ -792,6 +792,10 @@ type PackageStatus struct { // Restarts are the number of times a package restarted Restarts int32 `json:"restarts,omitempty"` + + // InterruptPending mirrors the nodewright.nvidia.com field of the same name. + //+optional + InterruptPending bool `json:"interruptPending,omitempty"` } // Equal checks name, version, image, containerSHA, stage, state, and restarts diff --git a/operator/cmd/cli/app/package/package_rerun.go b/operator/cmd/cli/app/package/package_rerun.go index da6a748fa..701e7aed5 100644 --- a/operator/cmd/cli/app/package/package_rerun.go +++ b/operator/cmd/cli/app/package/package_rerun.go @@ -207,7 +207,7 @@ func rerunPackage(ctx context.Context, cmd *cobra.Command, kubeClient *client.Cl } // Update each node's annotation to trigger re-run - successCount, updateErrors := updateNodeAnnotations(ctx, kubeClient, nodesToUpdate, nodeStates, packageKey, annotationKey, opts) + successCount, updateErrors := updateNodeAnnotations(ctx, kubeClient, nodesToUpdate, nodeStates, packageKey, annotationKey, pkg.HasInterrupt(), opts) printRerunResults(cmd, packageName, successCount, updateErrors) @@ -297,6 +297,7 @@ func updateNodeAnnotations( nodesToUpdate []string, nodeStates map[string]v1alpha1.NodeState, packageKey, annotationKey string, + ownInterrupt bool, opts *rerunOptions, ) (int, []string) { var updateErrors []string @@ -309,6 +310,12 @@ func updateNodeAnnotations( pkgStatus := nodeState[packageKey] pkgStatus.Stage = v1alpha1.Stage(opts.stage) pkgStatus.State = v1alpha1.StateInProgress + switch pkgStatus.Stage { + case v1alpha1.StageApply, v1alpha1.StageUpgrade: + pkgStatus.InterruptPending = ownInterrupt // a fresh apply owes the package's own interrupt again + case v1alpha1.StagePostInterrupt: + pkgStatus.InterruptPending = false + } nodeState[packageKey] = pkgStatus } else { delete(nodeState, packageKey) diff --git a/operator/cmd/cli/app/package/package_rerun_test.go b/operator/cmd/cli/app/package/package_rerun_test.go index 436a06995..f7046df94 100644 --- a/operator/cmd/cli/app/package/package_rerun_test.go +++ b/operator/cmd/cli/app/package/package_rerun_test.go @@ -222,6 +222,48 @@ var _ = Describe("Package Rerun Command", func() { }) }) + Describe("updateNodeAnnotations interruptPending", func() { + const key = "pkg1|1.0" + annotation := nodeStateAnnotationKey(testSkyhookNameRerun) + + rerunStage := func(stage string, ownInterrupt, pending bool) bool { + fakeKube := fake.NewClientset() + state := v1alpha1.NodeState{key: {Name: "pkg1", Version: "1.0", Stage: v1alpha1.StageConfig, State: v1alpha1.StateComplete, InterruptPending: pending}} + raw, err := json.Marshal(state) + Expect(err).NotTo(HaveOccurred()) + _, err = fakeKube.CoreV1().Nodes().Create(gocontext.Background(), &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "n1", Annotations: map[string]string{annotation: string(raw)}}, + }, metav1.CreateOptions{}) + Expect(err).NotTo(HaveOccurred()) + + kubeClient := client.NewWithClientsAndConfig(fakeKube, &mockdynamic.Interface{}, nil) + count, errs := updateNodeAnnotations(gocontext.Background(), kubeClient, []string{"n1"}, map[string]v1alpha1.NodeState{"n1": state}, key, annotation, ownInterrupt, &rerunOptions{stage: stage}) + Expect(errs).To(BeEmpty()) + Expect(count).To(Equal(1)) + + got, err := fakeKube.CoreV1().Nodes().Get(gocontext.Background(), "n1", metav1.GetOptions{}) + Expect(err).NotTo(HaveOccurred()) + var ns v1alpha1.NodeState + Expect(json.Unmarshal([]byte(got.Annotations[annotation]), &ns)).To(Succeed()) + return ns[key].InterruptPending + } + + It("owes the interrupt again on apply and upgrade only when the package declares one", func() { + Expect(rerunStage("apply", true, false)).To(BeTrue()) + Expect(rerunStage("upgrade", true, false)).To(BeTrue()) + Expect(rerunStage("apply", false, false)).To(BeFalse()) + }) + + It("clears the flag on post-interrupt", func() { + Expect(rerunStage("post-interrupt", true, true)).To(BeFalse()) + }) + + It("leaves the flag alone on config and interrupt", func() { + Expect(rerunStage("config", true, true)).To(BeTrue()) + Expect(rerunStage("interrupt", true, true)).To(BeTrue()) + }) + }) + Describe("printRerunResults", func() { var ( cmd *cobra.Command diff --git a/operator/cmd/cli/app/update_state.go b/operator/cmd/cli/app/update_state.go index fcf079cbb..f9c108fd1 100644 --- a/operator/cmd/cli/app/update_state.go +++ b/operator/cmd/cli/app/update_state.go @@ -196,6 +196,8 @@ func buildUpdateStateTarget( Image: pkg.Image, Stage: stage, State: state, + + InterruptPending: pkg.HasInterrupt() && (stage == v1alpha1.StageApply || stage == v1alpha1.StageUpgrade), } } else { if !hasEntry { @@ -209,6 +211,16 @@ func buildUpdateStateTarget( } entry.Stage = stage entry.State = state + switch stage { + case v1alpha1.StageApply, v1alpha1.StageUpgrade: + entry.InterruptPending = pkg.HasInterrupt() // a fresh apply owes the package's own interrupt again + case v1alpha1.StagePostInterrupt: + entry.InterruptPending = false + case v1alpha1.StageInterrupt: + if state == v1alpha1.StateComplete { + entry.InterruptPending = false + } + } ns[uniqueName] = entry } diff --git a/operator/cmd/cli/app/update_state_test.go b/operator/cmd/cli/app/update_state_test.go index ebbc196e3..03157e10e 100644 --- a/operator/cmd/cli/app/update_state_test.go +++ b/operator/cmd/cli/app/update_state_test.go @@ -332,4 +332,44 @@ var _ = Describe("UpdateState Command", func() { }) }) }) + + Describe("buildUpdateStateTarget interruptPending", func() { + pkgWith := func(interrupt *v1alpha1.Interrupt) v1alpha1.Package { + return v1alpha1.Package{PackageRef: v1alpha1.PackageRef{Name: "pkg1", Version: "1.0"}, Interrupt: interrupt} + } + pending := func(pkg v1alpha1.Package, addMode bool) bool { + states := map[string]v1alpha1.NodeState{} + if !addMode { + states["n1"] = v1alpha1.NodeState{"pkg1|1.0": {Name: "pkg1", Version: "1.0", Stage: v1alpha1.StagePostInterrupt, State: v1alpha1.StateComplete}} + } + target, ok := buildUpdateStateTarget(&cobra.Command{}, "n1", states, pkg, v1alpha1.StageApply, v1alpha1.StateInProgress, addMode) + Expect(ok).To(BeTrue()) + var ns v1alpha1.NodeState + Expect(json.Unmarshal([]byte(target.nextRaw), &ns)).To(Succeed()) + return ns["pkg1|1.0"].InterruptPending + } + + It("sets the flag on apply only for a package that declares an interrupt", func() { + withInterrupt := pkgWith(&v1alpha1.Interrupt{Type: v1alpha1.REBOOT}) + Expect(pending(withInterrupt, false)).To(BeTrue()) + Expect(pending(withInterrupt, true)).To(BeTrue()) + Expect(pending(pkgWith(nil), false)).To(BeFalse()) + Expect(pending(pkgWith(nil), true)).To(BeFalse()) + }) + + It("clears the flag on post-interrupt and on interrupt complete, and keeps it otherwise", func() { + pendingAt := func(stage v1alpha1.Stage, state v1alpha1.State) bool { + states := map[string]v1alpha1.NodeState{"n1": {"pkg1|1.0": {Name: "pkg1", Version: "1.0", Stage: v1alpha1.StageConfig, State: v1alpha1.StateComplete, InterruptPending: true}}} + target, ok := buildUpdateStateTarget(&cobra.Command{}, "n1", states, pkgWith(&v1alpha1.Interrupt{Type: v1alpha1.REBOOT}), stage, state, false) + Expect(ok).To(BeTrue()) + var ns v1alpha1.NodeState + Expect(json.Unmarshal([]byte(target.nextRaw), &ns)).To(Succeed()) + return ns["pkg1|1.0"].InterruptPending + } + Expect(pendingAt(v1alpha1.StagePostInterrupt, v1alpha1.StateInProgress)).To(BeFalse()) + Expect(pendingAt(v1alpha1.StageInterrupt, v1alpha1.StateComplete)).To(BeFalse()) + Expect(pendingAt(v1alpha1.StageInterrupt, v1alpha1.StateInProgress)).To(BeTrue()) + Expect(pendingAt(v1alpha1.StageConfig, v1alpha1.StateInProgress)).To(BeTrue()) + }) + }) }) diff --git a/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml b/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml index aaeb97f38..f0e5431a4 100644 --- a/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml +++ b/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml @@ -852,6 +852,13 @@ spec: image: description: Image for the package type: string + interruptPending: + description: |- + InterruptPending is set at apply or upgrade for a package that declares its own interrupt, + and cleared once that interrupt has completed or the package reaches post-interrupt. While set, it is still owed on this + node, so a config change that matches no configInterrupts does not let it finish without it. + Entries written before this field existed read as not pending. + type: boolean name: description: Name is the name of the package type: string diff --git a/operator/config/crd/bases/skyhook.nvidia.com_skyhooks.yaml b/operator/config/crd/bases/skyhook.nvidia.com_skyhooks.yaml index 2583fe6cc..fdae8f5cf 100644 --- a/operator/config/crd/bases/skyhook.nvidia.com_skyhooks.yaml +++ b/operator/config/crd/bases/skyhook.nvidia.com_skyhooks.yaml @@ -818,6 +818,10 @@ spec: image: description: Image for the package type: string + interruptPending: + description: InterruptPending mirrors the nodewright.nvidia.com + field of the same name. + type: boolean name: description: Name is the name of the package type: string diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index 6c52e9864..a82bfeb09 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -1780,7 +1780,7 @@ func (r *SkyhookReconciler) RunSkyhookPackages(ctx context.Context, clusterState return nil, fmt.Errorf("node %s: selecting packages to run: %w", node.GetNode().Name, err) } - interrupt, pack := fudgeInterruptWithPriority(toRun, skyhook.GetSkyhook().GetConfigUpdates(), skyhook.GetSkyhook().GetConfigInterrupts()) + interrupt, pack := nodeInterrupt(node, toRun, skyhook.GetSkyhook()) for _, f := range toRun { @@ -3236,8 +3236,14 @@ func (r *SkyhookReconciler) Interrupt(ctx context.Context, skyhookNode wrapper.S return nil } +// nodeInterrupt picks the interrupt to run on a node for the packages about to run, counting a +// package's own interrupt only where that node still owes it. +func nodeInterrupt(node wrapper.SkyhookNode, toRun []*v1alpha1.Package, skyhook *wrapper.Skyhook) (*v1alpha1.Interrupt, string) { + return fudgeInterruptWithPriority(toRun, skyhook.GetConfigUpdates(), skyhook.GetConfigInterrupts(), node.OwesOwnInterrupt) +} + // fudgeInterruptWithPriority takes a list of packages, interrupts, and configUpdates and returns the correct merged interrupt to run to handle all the packages -func fudgeInterruptWithPriority(next []*v1alpha1.Package, configUpdates map[string][]string, interrupts map[string][]*v1alpha1.Interrupt) (*v1alpha1.Interrupt, string) { +func fudgeInterruptWithPriority(next []*v1alpha1.Package, configUpdates map[string][]string, interrupts map[string][]*v1alpha1.Interrupt, owesOwn func(*v1alpha1.Package) bool) (*v1alpha1.Interrupt, string) { var ret *v1alpha1.Interrupt var pack string @@ -3257,6 +3263,10 @@ func fudgeInterruptWithPriority(next []*v1alpha1.Package, configUpdates map[stri if _package.HasInterrupt() { interrupts[_package.Name] = append(interrupts[_package.Name], _package.Interrupt) } + } else if owesOwn != nil && owesOwn(_package) { + // a config change only brings the interrupts its keys match; a package whose own + // interrupt never ran on this node still owes it + interrupts[_package.Name] = append(interrupts[_package.Name], _package.Interrupt) } } diff --git a/operator/internal/controller/skyhook_controller_test.go b/operator/internal/controller/skyhook_controller_test.go index 892a11037..7c3476ae0 100644 --- a/operator/internal/controller/skyhook_controller_test.go +++ b/operator/internal/controller/skyhook_controller_test.go @@ -229,6 +229,53 @@ var _ = Describe("skyhook controller tests", func() { }) }) + It("should add a package's own interrupt on a config change only where this node still owes it", func() { + reboot := &v1alpha1.Interrupt{Type: v1alpha1.REBOOT} + packages := []*v1alpha1.Package{ + {PackageRef: v1alpha1.PackageRef{Name: "foo", Version: "1.0.0"}, Interrupt: reboot}, + {PackageRef: v1alpha1.PackageRef{Name: "bar", Version: "1.0.0"}, Interrupt: reboot}, + } + // keys on both packages, and none of them match a configInterrupts entry + configUpdates := map[string][]string{"foo": {"a.conf"}, "bar": {"b.conf"}} + owesFoo := func(p *v1alpha1.Package) bool { return p.Name == "foo" } + + interrupt, pack := fudgeInterruptWithPriority(packages, configUpdates, map[string][]*v1alpha1.Interrupt{}, owesFoo) + Expect(interrupt).To(Equal(reboot)) + Expect(pack).To(Equal("foo")) + + interrupt, pack = fudgeInterruptWithPriority(packages, configUpdates, map[string][]*v1alpha1.Interrupt{}, func(*v1alpha1.Package) bool { return false }) + Expect(interrupt).To(BeNil()) + Expect(pack).To(BeEmpty()) + }) + + It("should pick a package's own interrupt for a real node only while that node owes it", func() { + reboot := &v1alpha1.Interrupt{Type: v1alpha1.REBOOT} + pkg := v1alpha1.Package{PackageRef: v1alpha1.PackageRef{Name: "foo", Version: "1.0.0"}, Interrupt: reboot} + skyhook := &v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: "sk"}, + Spec: v1alpha1.NodeWrightSpec{Packages: v1alpha1.Packages{"foo": pkg}}, + Status: v1alpha1.NodeWrightStatus{ConfigUpdates: map[string][]string{"foo": {"a.conf"}}}, // matches no configInterrupts + } + nodeWith := func(pending bool) wrapper.SkyhookNode { + raw, err := json.Marshal(v1alpha1.NodeState{"foo|1.0.0": {Name: "foo", Version: "1.0.0", Stage: v1alpha1.StageConfig, State: v1alpha1.StateInProgress, InterruptPending: pending}}) + Expect(err).ToNot(HaveOccurred()) + node, err := wrapper.NewSkyhookNode(&corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Annotations: map[string]string{fmt.Sprintf("%s/nodeState_%s", v1alpha1.METADATA_PREFIX, skyhook.Name): string(raw)}, + }}, skyhook) + Expect(err).ToNot(HaveOccurred()) + return node + } + + interrupt, pack := nodeInterrupt(nodeWith(true), []*v1alpha1.Package{&pkg}, wrapper.NewSkyhookWrapper(skyhook)) + Expect(interrupt).To(Equal(reboot)) + Expect(pack).To(Equal("foo")) + + interrupt, pack = nodeInterrupt(nodeWith(false), []*v1alpha1.Package{&pkg}, wrapper.NewSkyhookWrapper(skyhook)) + Expect(interrupt).To(BeNil()) + Expect(pack).To(BeEmpty()) + }) + It("should merge interrupts", func() { packages := []*v1alpha1.Package{ { @@ -273,7 +320,7 @@ var _ = Describe("skyhook controller tests", func() { }, } configUpdates := make(map[string][]string) - interrupt, _package := fudgeInterruptWithPriority(packages, configUpdates, interrupts) + interrupt, _package := fudgeInterruptWithPriority(packages, configUpdates, interrupts, nil) Expect(interrupt).ToNot(BeNil()) Expect(interrupt.Services).To(BeEquivalentTo([]string{"bar", "ducks", "foo", "kittens", "log"})) Expect(_package).To(BeEquivalentTo("bar")) @@ -318,7 +365,7 @@ var _ = Describe("skyhook controller tests", func() { }, } - interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts) + interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts, nil) Expect(interrupt).ToNot(BeNil()) Expect(_package).To(BeEquivalentTo("omg")) Expect(interrupt.Type).To(BeEquivalentTo(v1alpha1.REBOOT)) @@ -388,7 +435,7 @@ var _ = Describe("skyhook controller tests", func() { }, } - interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts) + interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts, nil) Expect(interrupt).ToNot(BeNil()) Expect(_package).To(BeEquivalentTo("bar")) Expect(interrupt.Type).To(BeEquivalentTo(v1alpha1.SERVICE)) @@ -459,7 +506,7 @@ var _ = Describe("skyhook controller tests", func() { } // configUpdate matches package so config interrupts are now added but not package interrupts - interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts) + interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts, nil) Expect(interrupt).ToNot(BeNil()) Expect(_package).To(BeEquivalentTo("bar")) Expect(interrupt.Type).To(BeEquivalentTo(v1alpha1.SERVICE)) @@ -1867,7 +1914,7 @@ var _ = Describe("skyhook controller tests", func() { interrupts := make(map[string][]*v1alpha1.Interrupt) configUpdates := make(map[string][]string) - interrupt, _package := fudgeInterruptWithPriority(packages, configUpdates, interrupts) + interrupt, _package := fudgeInterruptWithPriority(packages, configUpdates, interrupts, nil) Expect(interrupt).ToNot(BeNil()) Expect(interrupt.Type).To(BeEquivalentTo(v1alpha1.NOOP)) Expect(_package).To(BeEquivalentTo("foo")) @@ -1881,7 +1928,7 @@ var _ = Describe("skyhook controller tests", func() { }, }) - interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts) + interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts, nil) Expect(interrupt).ToNot(BeNil()) Expect(_package).To(BeEquivalentTo("bar")) Expect(interrupt.Type).To(BeEquivalentTo(v1alpha1.SERVICE)) @@ -1895,7 +1942,7 @@ var _ = Describe("skyhook controller tests", func() { }, }) - interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts) + interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts, nil) Expect(interrupt).ToNot(BeNil()) Expect(_package).To(BeEquivalentTo("baz")) Expect(interrupt.Type).To(BeEquivalentTo(v1alpha1.RESTART_ALL_SERVICES)) @@ -1909,7 +1956,7 @@ var _ = Describe("skyhook controller tests", func() { }, }) - interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts) + interrupt, _package = fudgeInterruptWithPriority(packages, configUpdates, interrupts, nil) Expect(interrupt).ToNot(BeNil()) Expect(_package).To(BeEquivalentTo("omg")) Expect(interrupt.Type).To(BeEquivalentTo(v1alpha1.REBOOT)) diff --git a/operator/internal/wrapper/mock/SkyhookNode.go b/operator/internal/wrapper/mock/SkyhookNode.go index b7472980a..656b4e53b 100644 --- a/operator/internal/wrapper/mock/SkyhookNode.go +++ b/operator/internal/wrapper/mock/SkyhookNode.go @@ -808,6 +808,57 @@ func (_c *MockSkyhookNode_NextStage_Call) RunAndReturn(run func(_package *v1alph return _c } +// OwesOwnInterrupt provides a mock function for the type MockSkyhookNode +func (_mock *MockSkyhookNode) OwesOwnInterrupt(_package *v1alpha1.Package) bool { + ret := _mock.Called(_package) + + if len(ret) == 0 { + panic("no return value specified for OwesOwnInterrupt") + } + + var r0 bool + if returnFunc, ok := ret.Get(0).(func(*v1alpha1.Package) bool); ok { + r0 = returnFunc(_package) + } else { + r0 = ret.Get(0).(bool) + } + return r0 +} + +// MockSkyhookNode_OwesOwnInterrupt_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'OwesOwnInterrupt' +type MockSkyhookNode_OwesOwnInterrupt_Call struct { + *mock.Call +} + +// OwesOwnInterrupt is a helper method to define mock.On call +// - _package *v1alpha1.Package +func (_e *MockSkyhookNode_Expecter) OwesOwnInterrupt(_package any) *MockSkyhookNode_OwesOwnInterrupt_Call { + return &MockSkyhookNode_OwesOwnInterrupt_Call{Call: _e.mock.On("OwesOwnInterrupt", _package)} +} + +func (_c *MockSkyhookNode_OwesOwnInterrupt_Call) Run(run func(_package *v1alpha1.Package)) *MockSkyhookNode_OwesOwnInterrupt_Call { + _c.Call.Run(func(args mock.Arguments) { + var arg0 *v1alpha1.Package + if args[0] != nil { + arg0 = args[0].(*v1alpha1.Package) + } + run( + arg0, + ) + }) + return _c +} + +func (_c *MockSkyhookNode_OwesOwnInterrupt_Call) Return(b bool) *MockSkyhookNode_OwesOwnInterrupt_Call { + _c.Call.Return(b) + return _c +} + +func (_c *MockSkyhookNode_OwesOwnInterrupt_Call) RunAndReturn(run func(_package *v1alpha1.Package) bool) *MockSkyhookNode_OwesOwnInterrupt_Call { + _c.Call.Return(run) + return _c +} + // PackageStatus provides a mock function for the type MockSkyhookNode func (_mock *MockSkyhookNode) PackageStatus(name string) (*v1alpha1.PackageStatus, bool) { ret := _mock.Called(name) diff --git a/operator/internal/wrapper/node.go b/operator/internal/wrapper/node.go index 16f5c3e7f..fdf5d5efc 100644 --- a/operator/internal/wrapper/node.go +++ b/operator/internal/wrapper/node.go @@ -60,6 +60,8 @@ type SkyhookNode interface { NextStage(_package *v1alpha1.Package) *v1alpha1.Stage // HasInterrupt reports whether the package has an interrupt (e.g. wait-for-input) that blocks progression. HasInterrupt(_package v1alpha1.Package) bool + // OwesOwnInterrupt reports whether the package's own interrupt has not yet run on this node. + OwesOwnInterrupt(_package *v1alpha1.Package) bool // UpdateCondition refreshes Skyhook-related node conditions (NotReady and Erroring) from current package state. UpdateCondition() // HasSkyhookAnnotations reports whether the node has any Skyhook operator annotations. @@ -441,7 +443,13 @@ func (node *skyhookNode) RemoveState(_package v1alpha1.PackageRef) error { // Upsert creates or updates state for a package (image, state, stage, restarts, containerSHA) and persists it. func (node *skyhookNode) Upsert(_package v1alpha1.PackageRef, image string, state v1alpha1.State, stage v1alpha1.Stage, restarts int32, containerSHA string) error { - changed := node.nodeState.Upsert(_package, image, state, stage, restarts, containerSHA) + ownInterrupt := false + if node.skyhook != nil { + if pkg, ok := node.skyhook.Spec.Packages[_package.Name]; ok { + ownInterrupt = pkg.HasInterrupt() + } + } + changed := node.nodeState.UpsertOwing(_package, image, state, stage, restarts, containerSHA, ownInterrupt) if changed { if node.skyhook != nil { node.skyhook.Updated = true @@ -511,6 +519,11 @@ func (node *skyhookNode) Changed() bool { return node.updated } +// OwesOwnInterrupt reports whether the package's own interrupt has not yet run on this node. +func (node *skyhookNode) OwesOwnInterrupt(_package *v1alpha1.Package) bool { + return node.nodeState.OwesOwnInterrupt(*_package) +} + // HasInterrupt reports whether the package has an interrupt (e.g. wait-for-input) that blocks progression. func (node *skyhookNode) HasInterrupt(_package v1alpha1.Package) bool { return node.nodeState.HasInterrupt(_package, node.skyhook.GetConfigInterrupts(), node.skyhook.GetConfigUpdates()) diff --git a/operator/internal/wrapper/node_test.go b/operator/internal/wrapper/node_test.go index d749e1306..999c9ac02 100644 --- a/operator/internal/wrapper/node_test.go +++ b/operator/internal/wrapper/node_test.go @@ -32,6 +32,38 @@ import ( ) var _ = Describe("SkyhookNode", func() { + Context("Upsert and the owed own interrupt", func() { + It("flags only a package that declares its own interrupt", func() { + node := corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "test-node"}} + skyhook := v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: "test-skyhook"}, + Spec: v1alpha1.NodeWrightSpec{ + Packages: map[string]v1alpha1.Package{ + "with": { + PackageRef: v1alpha1.PackageRef{Name: "with", Version: "1.0"}, + Image: "img", + Interrupt: &v1alpha1.Interrupt{Type: v1alpha1.REBOOT}, + }, + "without": {PackageRef: v1alpha1.PackageRef{Name: "without", Version: "1.0"}, Image: "img"}, + }, + }, + } + skyhookNode, err := NewSkyhookNode(&node, &skyhook) + Expect(err).NotTo(HaveOccurred()) + + for _, name := range []string{"with", "without"} { + Expect(skyhookNode.Upsert(v1alpha1.PackageRef{Name: name, Version: "1.0"}, "img", v1alpha1.StateInProgress, v1alpha1.StageApply, 0, "")).To(Succeed()) + } + + with, _ := skyhookNode.PackageStatus("with|1.0") + without, _ := skyhookNode.PackageStatus("without|1.0") + Expect(with.InterruptPending).To(BeTrue()) + Expect(without.InterruptPending).To(BeFalse()) + withPkg := skyhook.Spec.Packages["with"] + Expect(skyhookNode.OwesOwnInterrupt(&withPkg)).To(BeTrue()) + }) + }) + Context("RunNext", func() { It("should return packages in deterministic order and respect dependencies", func() { node := corev1.Node{