From cb0fc60295d04faff6a26c217ed193a27d2765bd Mon Sep 17 00:00:00 2001 From: bharqav Date: Sat, 5 Sep 2026 12:18:34 +0530 Subject: [PATCH 1/7] fix(operator): surface Blocked condition and event when non-interrupt pods block drain Previously, when spec.podNonInterruptLabels held a node at the pre-drain barrier, the operator gave no indication why or which pods were responsible - administrators only saw Ready=False/Progressing indefinitely, with no way to tell a deliberate wait from a stuck node. - HasNonInterruptWork now returns the names of the blocking pods instead of discarding them. - EnsureNodeIsReadyForInterrupt sets a Blocked condition (reason NonInterruptPodsRunning) naming the blocking pods (up to ReadyConditionNodeListLimit, else a count with the full list logged at info level), and emits a single Warning event when the barrier is first entered - not on every reconcile pass. - The condition clears once the barrier lifts, without disturbing an unrelated Blocked condition set for another reason (e.g. DependencyUninstalled). - UpdateBlockedCondition no longer unconditionally clears all Blocked conditions regardless of reason, which previously caused this condition to flap every reconcile. Fixes #542 Signed-off-by: bharqav --- docs/architecture/interrupt-flow.md | 5 +- operator/RELEASE_NOTES.md | 18 +++++ .../nodewright/v1alpha1/nodewright_types.go | 3 +- .../nodewright.nvidia.com_nodewrights.yaml | 5 +- .../internal/controller/cluster_state_v2.go | 5 +- .../internal/controller/skyhook_controller.go | 66 ++++++++++++++++--- .../controller/skyhook_controller_test.go | 36 +++++++++- .../internal/wrapper/skyhook_conditions.go | 38 +++++++++++ .../wrapper/skyhook_conditions_test.go | 56 ++++++++++++++++ 9 files changed, 217 insertions(+), 15 deletions(-) diff --git a/docs/architecture/interrupt-flow.md b/docs/architecture/interrupt-flow.md index a4b95c221..f4e73f572 100644 --- a/docs/architecture/interrupt-flow.md +++ b/docs/architecture/interrupt-flow.md @@ -163,7 +163,10 @@ matching more closely: the unschedulable toleration check uses Kubernetes owner reference, and mirror/static pods are ignored. `podNonInterruptLabels` remains a pre-drain barrier. Matching pods must finish -or move away before the operator starts the configurable drain step. +or move away before the operator starts the configurable drain step. This wait +is unbounded and drain timeout does not apply. While this barrier holds, +administrators will see a `Blocked` condition with reason `NonInterruptPodsRunning` +reporting which pods are causing the hold. ### When Drain Is Complete diff --git a/operator/RELEASE_NOTES.md b/operator/RELEASE_NOTES.md index 7f98c0e37..0f874107c 100644 --- a/operator/RELEASE_NOTES.md +++ b/operator/RELEASE_NOTES.md @@ -7,6 +7,24 @@ For the full commit-level log see CHANGELOG.md. ### Bug Fixes +- **A `Blocked` status condition (reason `NonInterruptPodsRunning`) and a Warning event are + now surfaced when `spec.podNonInterruptLabels` blocks node drain.** Previously, + the operator held the node in `Ready=False` / `Progressing` with no condition or + event indicating why or which pods were causing the hold. When a node's interrupt + is held at this barrier: + - A `Blocked` condition (status `True`, reason `NonInterruptPodsRunning`) is set on the + NodeWright. Its message identifies the blocking pods on that node: up to 10 pods + (`wrapper.ReadyConditionNodeListLimit`) are listed by name (e.g. `Pod [x] is running. Waiting.` + or `Pods [x, y] are running. Waiting.`). If more than 10 pods are running on the node, + the full list is logged at info level and the condition message is summarized as + ` pods are running. Waiting.`. + - A Warning event with reason `Drain` (`EventsReasonSkyhookDrain`) is emitted on the NodeWright + object when this barrier is first entered, reporting the node and package being held + (`drain blocked by non-interrupt pods for node [] package [:]`). + - Once all matching non-interrupt pods finish or terminate on the node, the + `NonInterruptPodsRunning` condition is removed and drain proceeds, preserving any unrelated + `Blocked` condition that may also be active (such as `DependencyUninstalled`). + - **Adding and removing the finalizer from a natively authored NodeWright no longer rewrites its spec.** Both paths now use optimistic, metadata-only merge patches, preserving concurrent finalizer changes and user-authored resource diff --git a/operator/api/nodewright/v1alpha1/nodewright_types.go b/operator/api/nodewright/v1alpha1/nodewright_types.go index bbdcf0546..f98df25d2 100644 --- a/operator/api/nodewright/v1alpha1/nodewright_types.go +++ b/operator/api/nodewright/v1alpha1/nodewright_types.go @@ -45,7 +45,8 @@ type NodeWrightSpec struct { //+kubebuilder:default=false Serial bool `json:"serial,omitempty"` - // PodNonInterruptLabels are a set of labels we want to monitor pods for whether they Interruptible + // PodNonInterruptLabels are a set of labels we want to monitor pods for whether they Interruptible. + // A Blocked condition with reason NonInterruptPodsRunning is surfaced while this barrier holds. PodNonInterruptLabels metav1.LabelSelector `json:"podNonInterruptLabels,omitempty"` // NodeSelector are a set of labels we want to monitor nodes for applying packages too diff --git a/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml b/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml index 82aefd610..49bb56e82 100644 --- a/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml +++ b/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml @@ -591,8 +591,9 @@ spec: description: Packages are the DAG of packages to be applied to nodes. type: object podNonInterruptLabels: - description: PodNonInterruptLabels are a set of labels we want to - monitor pods for whether they Interruptible + description: |- + PodNonInterruptLabels are a set of labels we want to monitor pods for whether they Interruptible. + A Blocked condition with reason NonInterruptPodsRunning is surfaced while this barrier holds. properties: matchExpressions: description: matchExpressions is a list of label selector requirements. diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go index ddc40b3a1..f71b5189e 100644 --- a/operator/internal/controller/cluster_state_v2.go +++ b/operator/internal/controller/cluster_state_v2.go @@ -614,7 +614,10 @@ func (s *skyhookNodes) UpdateBlockedCondition() error { Message: strings.Join(blockedMsgs, "; "), }) } else { - wrapper.RemoveSkyhookConditionTypes(s.skyhook, wrapper.SkyhookConditionBlocked) + existing := wrapper.FindSkyhookCondition(s.skyhook, wrapper.SkyhookConditionBlocked) + if existing != nil && existing.Reason != wrapper.SkyhookReasonNonInterruptPodsRunning { + wrapper.RemoveSkyhookConditionTypes(s.skyhook, wrapper.SkyhookConditionBlocked) + } } return nil diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index 9c0276666..07e320aa2 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -2412,16 +2412,16 @@ func (r *SkyhookReconciler) HandleFinalizer(ctx context.Context, skyhook Skyhook return false, nil } -// HasNonInterruptWork returns true if pods are running on the node that are either packages, or matches the SCR selector -func (r *SkyhookReconciler) HasNonInterruptWork(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, error) { +// HasNonInterruptWork returns true and the list of running or pending pod names if pods are running on the node that match the SCR selector +func (r *SkyhookReconciler) HasNonInterruptWork(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, []string, error) { selector, err := metav1.LabelSelectorAsSelector(&skyhookNode.GetSkyhook().Spec.PodNonInterruptLabels) if err != nil { - return false, fmt.Errorf("error creating selector: %w", err) + return false, nil, fmt.Errorf("error creating selector: %w", err) } if selector.Empty() { // when selector is empty it does not do any selecting, ie will return all pods on node. - return false, nil + return false, nil, nil } pods, err := r.dal.GetPods(ctx, @@ -2431,21 +2431,27 @@ func (r *SkyhookReconciler) HasNonInterruptWork(ctx context.Context, skyhookNode }, ) if err != nil { - return false, fmt.Errorf("error getting pods: %w", err) + return false, nil, fmt.Errorf("error getting pods: %w", err) } if pods == nil || len(pods.Items) == 0 { - return false, nil + return false, nil, nil } + var podNames []string for _, pod := range pods.Items { switch pod.Status.Phase { case corev1.PodRunning, corev1.PodPending: - return true, nil + podNames = append(podNames, pod.Name) } } - return false, nil + if len(podNames) > 0 { + sort.Strings(podNames) + return true, podNames, nil + } + + return false, nil, nil } func (r *SkyhookReconciler) HasRunningPackages(ctx context.Context, skyhookNode wrapper.SkyhookNode) (bool, error) { @@ -3260,14 +3266,56 @@ func (r *SkyhookReconciler) EnsureNodeIsReadyForInterrupt(ctx context.Context, s return false, nil } - hasWork, err := r.HasNonInterruptWork(ctx, skyhookNode) + hasWork, podNames, err := r.HasNonInterruptWork(ctx, skyhookNode) if err != nil { return false, err } if hasWork { // keep waiting... + logger := log.FromContext(ctx) + podLabel := "Pod" + podVerb := "is" + if len(podNames) > 1 { + podLabel = "Pods" + podVerb = "are" + } + message := fmt.Sprintf("%s [%s] %s running. Waiting.", podLabel, strings.Join(podNames, ", "), podVerb) + if len(podNames) > wrapper.ReadyConditionNodeListLimit { + logger.Info("Condition message truncated for non-interrupt pods", "nodewright", skyhookNode.GetSkyhook().Name, "pods", podNames) + message = fmt.Sprintf("%d pods are running. Waiting.", len(podNames)) + } + + existing := wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + if existing == nil || existing.Reason != wrapper.SkyhookReasonNonInterruptPodsRunning { + pkgName := "" + pkgVersion := "" + if _package != nil { + pkgName = _package.Name + pkgVersion = _package.Version + } + r.recorder.Eventf(skyhookNode.GetSkyhook().NodeWright, nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, wrapper.SkyhookReasonNonInterruptPodsRunning, + "drain blocked by non-interrupt pods for node [%s] package [%s:%s]", + skyhookNode.GetNode().Name, + pkgName, + pkgVersion, + ) + } + + if existing == nil || existing.Reason == wrapper.SkyhookReasonNonInterruptPodsRunning { + wrapper.AddSkyhookCondition(skyhookNode.GetSkyhook(), metav1.Condition{ + Type: wrapper.SkyhookConditionBlocked, + Status: metav1.ConditionTrue, + Reason: wrapper.SkyhookReasonNonInterruptPodsRunning, + Message: message, + ObservedGeneration: skyhookNode.GetSkyhook().Generation, + LastTransitionTime: metav1.Now(), + }) + } + return false, nil } + wrapper.RemoveSkyhookConditionTypeAndReason(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked, wrapper.SkyhookReasonNonInterruptPodsRunning) + ready, err := r.DrainNode(ctx, skyhookNode, _package) if err != nil { return false, fmt.Errorf("error draining node [%s]: %w", skyhookNode.GetNode().Name, err) diff --git a/operator/internal/controller/skyhook_controller_test.go b/operator/internal/controller/skyhook_controller_test.go index 26db5a9c5..e2073a240 100644 --- a/operator/internal/controller/skyhook_controller_test.go +++ b/operator/internal/controller/skyhook_controller_test.go @@ -801,7 +801,8 @@ var _ = Describe("skyhook controller tests", func() { }, }) - r, err := NewSkyhookReconciler(testClient.Scheme(), testClient, testClient, k8sfake.NewClientset(), events.NewFakeRecorder(10), opts) + recorder := events.NewFakeRecorder(10) + r, err := NewSkyhookReconciler(testClient.Scheme(), testClient, testClient, k8sfake.NewClientset(), recorder, opts) Expect(err).ToNot(HaveOccurred()) // Already cordoned in the API, so this spec exercises the podNonInterruptLabels @@ -844,6 +845,39 @@ var _ = Describe("skyhook controller tests", func() { drainStartedAt, err := skyhookNode.DrainStartedAt() Expect(err).ToNot(HaveOccurred()) Expect(drainStartedAt).To(BeNil()) + + cond := wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Status).To(Equal(metav1.ConditionTrue)) + Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) + Expect(cond.Message).To(Equal("Pod [golden] is running. Waiting.")) + Eventually(recorder.Events).Should(Receive(ContainSubstring("Warning Drain drain blocked by non-interrupt pods for node [node-a] package [pkg:1.0.0]"))) + + Expect(testClient.Delete(ctx, goldenPod)).To(Succeed()) + + _, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, &v1alpha1.Package{ + PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, + }) + Expect(err).ToNot(HaveOccurred()) + cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + Expect(cond).To(BeNil()) + + // Verify that an unrelated Blocked condition (e.g. DependencyUninstalled) is preserved + wrapper.AddSkyhookCondition(skyhookNode.GetSkyhook(), metav1.Condition{ + Type: wrapper.SkyhookConditionBlocked, + Status: metav1.ConditionTrue, + Reason: "DependencyUninstalled", + Message: "package pkg is blocked: dependency dep has been uninstalled", + ObservedGeneration: skyhookNode.GetSkyhook().Generation, + LastTransitionTime: metav1.Now(), + }) + _, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, &v1alpha1.Package{ + PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, + }) + Expect(err).ToNot(HaveOccurred()) + cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Reason).To(Equal("DependencyUninstalled")) }) It("should mark the node erroring when drain timeout expires", func() { diff --git a/operator/internal/wrapper/skyhook_conditions.go b/operator/internal/wrapper/skyhook_conditions.go index 8ce86c509..f368e9888 100644 --- a/operator/internal/wrapper/skyhook_conditions.go +++ b/operator/internal/wrapper/skyhook_conditions.go @@ -41,6 +41,8 @@ const ( SkyhookConditionNodeStateMalformed = "NodeStateMalformed" SkyhookConditionDeletionBlocked = "DeletionBlocked" + SkyhookReasonNonInterruptPodsRunning = "NonInterruptPodsRunning" + skyhookReadyReasonNodesConverged = "NodesConverged" skyhookReadyReasonProgressing = "Progressing" skyhookReadyReasonBlocked = "Blocked" @@ -221,6 +223,42 @@ func RemoveSkyhookConditionTypes(skyhook *Skyhook, conditionTypes ...string) boo return changed } +// RemoveSkyhookConditionTypeAndReason removes any condition matching both conditionType and reason. +// Returns true if the condition slice was modified. +func RemoveSkyhookConditionTypeAndReason(skyhook *Skyhook, conditionType, reason string) bool { + if len(skyhook.Status.Conditions) == 0 { + return false + } + + conditions := skyhook.Status.Conditions[:0] + changed := false + for _, condition := range skyhook.Status.Conditions { + if condition.Type == conditionType && condition.Reason == reason { + changed = true + continue + } + conditions = append(conditions, condition) + } + + if changed { + skyhook.Status.Conditions = conditions + skyhook.Updated = true + } + + return changed +} + +// FindSkyhookCondition searches skyhook.Status.Conditions for a condition with the given conditionType. +// Returns a pointer to the matching condition if present, or nil otherwise. +func FindSkyhookCondition(skyhook *Skyhook, conditionType string) *metav1.Condition { + for i := range skyhook.Status.Conditions { + if skyhook.Status.Conditions[i].Type == conditionType { + return &skyhook.Status.Conditions[i] + } + } + return nil +} + func HasTrueSkyhookCondition(skyhook *Skyhook, conditionTypes ...string) bool { for _, condition := range skyhook.Status.Conditions { for _, conditionType := range conditionTypes { diff --git a/operator/internal/wrapper/skyhook_conditions_test.go b/operator/internal/wrapper/skyhook_conditions_test.go index c55b58eb8..c5220f88f 100644 --- a/operator/internal/wrapper/skyhook_conditions_test.go +++ b/operator/internal/wrapper/skyhook_conditions_test.go @@ -194,6 +194,62 @@ var _ = Describe("Skyhook condition helpers", func() { }, []string{SkyhookConditionNodesIgnored, SkyhookConditionApplyPackage}, []string{SkyhookConditionReady, SkyhookConditionTaintNotTolerable}, true), ) + DescribeTable("removeSkyhookConditionTypeAndReason", func(existing []metav1.Condition, conditionType, reason string, expectedConditions []metav1.Condition, changed bool) { + skyhook := &Skyhook{ + NodeWright: &v1alpha1.NodeWright{ + Status: v1alpha1.NodeWrightStatus{ + Conditions: existing, + }, + }, + } + + Expect(RemoveSkyhookConditionTypeAndReason(skyhook, conditionType, reason)).To(Equal(changed)) + if expectedConditions == nil { + Expect(skyhook.Status.Conditions).To(BeEmpty()) + } else { + Expect(skyhook.Status.Conditions).To(Equal(expectedConditions)) + } + Expect(skyhook.Updated).To(Equal(changed)) + }, + Entry("returns false when there are no conditions", nil, SkyhookConditionBlocked, SkyhookReasonNonInterruptPodsRunning, nil, false), + Entry("removes condition when type and reason match", []metav1.Condition{ + {Type: SkyhookConditionBlocked, Reason: SkyhookReasonNonInterruptPodsRunning}, + {Type: SkyhookConditionReady, Reason: "NodesConverged"}, + }, SkyhookConditionBlocked, SkyhookReasonNonInterruptPodsRunning, []metav1.Condition{ + {Type: SkyhookConditionReady, Reason: "NodesConverged"}, + }, true), + Entry("preserves condition when type matches but reason differs", []metav1.Condition{ + {Type: SkyhookConditionBlocked, Reason: "DependencyUninstalled"}, + }, SkyhookConditionBlocked, SkyhookReasonNonInterruptPodsRunning, []metav1.Condition{ + {Type: SkyhookConditionBlocked, Reason: "DependencyUninstalled"}, + }, false), + Entry("preserves condition when reason matches but type differs", []metav1.Condition{ + {Type: SkyhookConditionReady, Reason: SkyhookReasonNonInterruptPodsRunning}, + }, SkyhookConditionBlocked, SkyhookReasonNonInterruptPodsRunning, []metav1.Condition{ + {Type: SkyhookConditionReady, Reason: SkyhookReasonNonInterruptPodsRunning}, + }, false), + ) + + Describe("FindSkyhookCondition", func() { + It("finds an existing condition and returns nil when not found", func() { + skyhook := &Skyhook{ + NodeWright: &v1alpha1.NodeWright{ + Status: v1alpha1.NodeWrightStatus{ + Conditions: []metav1.Condition{ + {Type: SkyhookConditionBlocked, Reason: SkyhookReasonNonInterruptPodsRunning}, + }, + }, + }, + } + + cond := FindSkyhookCondition(skyhook, SkyhookConditionBlocked) + Expect(cond).NotTo(BeNil()) + Expect(cond.Reason).To(Equal(SkyhookReasonNonInterruptPodsRunning)) + + Expect(FindSkyhookCondition(skyhook, SkyhookConditionReady)).To(BeNil()) + }) + }) + DescribeTable("hasTrueSkyhookCondition", func(conditions []metav1.Condition, conditionTypes []string, expected bool) { skyhook := &Skyhook{ NodeWright: &v1alpha1.NodeWright{ From ddee2aed432ce65a1192a70479a542f7c59f076b Mon Sep 17 00:00:00 2001 From: bharqav Date: Sat, 5 Sep 2026 12:25:46 +0530 Subject: [PATCH 2/7] chore: retrigger signature check Signed-off-by: bharqav From fdff3b4da79b9e995f0a9c7a85bb00d93e1fa0ea Mon Sep 17 00:00:00 2001 From: bharqav Date: Sat, 5 Sep 2026 13:09:14 +0530 Subject: [PATCH 3/7] fix(operator): defer drain-blocked event/condition when another Blocked reason is active MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses CodeRabbit review on PR #582 — the event was firing on every reconcile whenever an unrelated Blocked reason (e.g. DependencyUninstalled) already owned the condition slot, since the transition guard never resolved to false in that case. Also documents the full podNonInterruptLabels barrier contract per review feedback. Signed-off-by: bharqav --- operator/RELEASE_NOTES.md | 8 +- .../nodewright/v1alpha1/nodewright_types.go | 5 +- .../nodewright.nvidia.com_nodewrights.yaml | 5 +- .../internal/controller/skyhook_controller.go | 30 ++--- .../controller/skyhook_controller_test.go | 105 ++++++++++++++++++ 5 files changed, 131 insertions(+), 22 deletions(-) diff --git a/operator/RELEASE_NOTES.md b/operator/RELEASE_NOTES.md index 0f874107c..4b20a7030 100644 --- a/operator/RELEASE_NOTES.md +++ b/operator/RELEASE_NOTES.md @@ -21,9 +21,11 @@ For the full commit-level log see CHANGELOG.md. - A Warning event with reason `Drain` (`EventsReasonSkyhookDrain`) is emitted on the NodeWright object when this barrier is first entered, reporting the node and package being held (`drain blocked by non-interrupt pods for node [] package [:]`). - - Once all matching non-interrupt pods finish or terminate on the node, the - `NonInterruptPodsRunning` condition is removed and drain proceeds, preserving any unrelated - `Blocked` condition that may also be active (such as `DependencyUninstalled`). + - If another `Blocked` reason is already active (such as `DependencyUninstalled`), the + `NonInterruptPodsRunning` condition and Warning event are deferred and will only appear + once that other condition clears. Once all matching non-interrupt pods finish or + terminate on the node, the `NonInterruptPodsRunning` condition is removed and drain proceeds, + preserving any unrelated `Blocked` condition that may also be active. - **Adding and removing the finalizer from a natively authored NodeWright no longer rewrites its spec.** Both paths now use optimistic, metadata-only merge diff --git a/operator/api/nodewright/v1alpha1/nodewright_types.go b/operator/api/nodewright/v1alpha1/nodewright_types.go index f98df25d2..0e8f8abc3 100644 --- a/operator/api/nodewright/v1alpha1/nodewright_types.go +++ b/operator/api/nodewright/v1alpha1/nodewright_types.go @@ -45,8 +45,9 @@ type NodeWrightSpec struct { //+kubebuilder:default=false Serial bool `json:"serial,omitempty"` - // PodNonInterruptLabels are a set of labels we want to monitor pods for whether they Interruptible. - // A Blocked condition with reason NonInterruptPodsRunning is surfaced while this barrier holds. + // PodNonInterruptLabels are a set of labels we want to monitor pods for whether they are interruptible. + // Matching Pending or Running pods block pre-drain progress indefinitely; spec.drainConfig.timeout does not + // apply to this barrier. A Blocked condition with reason NonInterruptPodsRunning is surfaced while this barrier holds. PodNonInterruptLabels metav1.LabelSelector `json:"podNonInterruptLabels,omitempty"` // NodeSelector are a set of labels we want to monitor nodes for applying packages too diff --git a/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml b/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml index 49bb56e82..efdee8749 100644 --- a/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml +++ b/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml @@ -592,8 +592,9 @@ spec: type: object podNonInterruptLabels: description: |- - PodNonInterruptLabels are a set of labels we want to monitor pods for whether they Interruptible. - A Blocked condition with reason NonInterruptPodsRunning is surfaced while this barrier holds. + PodNonInterruptLabels are a set of labels we want to monitor pods for whether they are interruptible. + Matching Pending or Running pods block pre-drain progress indefinitely; spec.drainConfig.timeout does not + apply to this barrier. A Blocked condition with reason NonInterruptPodsRunning is surfaced while this barrier holds. properties: matchExpressions: description: matchExpressions is a list of label selector requirements. diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index 07e320aa2..aa6485b5a 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -3285,22 +3285,22 @@ func (r *SkyhookReconciler) EnsureNodeIsReadyForInterrupt(ctx context.Context, s } existing := wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) - if existing == nil || existing.Reason != wrapper.SkyhookReasonNonInterruptPodsRunning { - pkgName := "" - pkgVersion := "" - if _package != nil { - pkgName = _package.Name - pkgVersion = _package.Version - } - r.recorder.Eventf(skyhookNode.GetSkyhook().NodeWright, nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, wrapper.SkyhookReasonNonInterruptPodsRunning, - "drain blocked by non-interrupt pods for node [%s] package [%s:%s]", - skyhookNode.GetNode().Name, - pkgName, - pkgVersion, - ) - } - if existing == nil || existing.Reason == wrapper.SkyhookReasonNonInterruptPodsRunning { + if existing == nil { + pkgName := "" + pkgVersion := "" + if _package != nil { + pkgName = _package.Name + pkgVersion = _package.Version + } + r.recorder.Eventf(skyhookNode.GetSkyhook().NodeWright, nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, wrapper.SkyhookReasonNonInterruptPodsRunning, + "drain blocked by non-interrupt pods for node [%s] package [%s:%s]", + skyhookNode.GetNode().Name, + pkgName, + pkgVersion, + ) + } + wrapper.AddSkyhookCondition(skyhookNode.GetSkyhook(), metav1.Condition{ Type: wrapper.SkyhookConditionBlocked, Status: metav1.ConditionTrue, diff --git a/operator/internal/controller/skyhook_controller_test.go b/operator/internal/controller/skyhook_controller_test.go index e2073a240..8019f7a55 100644 --- a/operator/internal/controller/skyhook_controller_test.go +++ b/operator/internal/controller/skyhook_controller_test.go @@ -880,6 +880,111 @@ var _ = Describe("skyhook controller tests", func() { Expect(cond.Reason).To(Equal("DependencyUninstalled")) }) + It("suppresses drain warning events and preserves condition while DependencyUninstalled is active, then emits exactly once when cleared", func() { + goldenPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "golden", + Namespace: "default", + Labels: map[string]string{ + "workload": "golden", + }, + }, + Spec: corev1.PodSpec{ + NodeName: "node-a", + Containers: []corev1.Container{ + {Name: "workload", Image: "busybox"}, + }, + }, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + } + + testClient := fakeDrainClient(goldenPod) + recorder := events.NewFakeRecorder(10) + r, err := NewSkyhookReconciler(testClient.Scheme(), testClient, testClient, k8sfake.NewClientset(), recorder, opts) + Expect(err).ToNot(HaveOccurred()) + + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "node-a", + Annotations: map[string]string{ + fmt.Sprintf("%s/cordon_%s", v1alpha1.METADATA_PREFIX, "drain-dep"): "true", + }, + }, + Spec: corev1.NodeSpec{Unschedulable: true}, + } + skyhook := &v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: "drain-dep"}, + Spec: v1alpha1.NodeWrightSpec{ + PodNonInterruptLabels: metav1.LabelSelector{ + MatchLabels: map[string]string{ + "workload": "golden", + }, + }, + DrainConfig: &v1alpha1.DrainConfig{ + DisableEviction: ptr(true), + }, + Packages: v1alpha1.Packages{}, + }, + } + skyhookNode, err := wrapper.NewSkyhookNode(node, skyhook) + Expect(err).ToNot(HaveOccurred()) + + // Pre-seed unrelated Blocked condition + wrapper.AddSkyhookCondition(skyhookNode.GetSkyhook(), metav1.Condition{ + Type: wrapper.SkyhookConditionBlocked, + Status: metav1.ConditionTrue, + Reason: "DependencyUninstalled", + Message: "package pkg is blocked: dependency dep has been uninstalled", + ObservedGeneration: skyhookNode.GetSkyhook().Generation, + LastTransitionTime: metav1.Now(), + }) + + pkg := &v1alpha1.Package{ + PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, + } + + // (a) Reconcile 1: non-interrupt pods present, but DependencyUninstalled owns the Blocked slot + ready, err := r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, pkg) + Expect(err).ToNot(HaveOccurred()) + Expect(ready).To(BeFalse()) + cond := wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Reason).To(Equal("DependencyUninstalled")) + Expect(cond.Message).To(Equal("package pkg is blocked: dependency dep has been uninstalled")) + Expect(recorder.Events).To(BeEmpty()) + + // (a) Reconcile 2: second pass with DependencyUninstalled still active + ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, pkg) + Expect(err).ToNot(HaveOccurred()) + Expect(ready).To(BeFalse()) + cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Reason).To(Equal("DependencyUninstalled")) + Expect(recorder.Events).To(BeEmpty()) + + // (b) DependencyUninstalled is cleared, non-interrupt pods still present + wrapper.RemoveSkyhookConditionTypes(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + + // Reconcile 3: immediately takes ownership of the Blocked slot and emits exactly one event + ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, pkg) + Expect(err).ToNot(HaveOccurred()) + Expect(ready).To(BeFalse()) + cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) + Expect(cond.Message).To(Equal("Pod [golden] is running. Waiting.")) + Eventually(recorder.Events).Should(Receive(ContainSubstring("Warning Drain drain blocked by non-interrupt pods for node [node-a] package [pkg:1.0.0]"))) + + // Reconcile 4: subsequent pass with non-interrupt pods still present emits no duplicate event + ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, pkg) + Expect(err).ToNot(HaveOccurred()) + Expect(ready).To(BeFalse()) + cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) + Expect(recorder.Events).To(BeEmpty()) + }) + It("should mark the node erroring when drain timeout expires", func() { pod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ From cd8cd1f4030d0c74e5a79d83cf46380b0eafd2e4 Mon Sep 17 00:00:00 2001 From: bharqav Date: Sat, 12 Sep 2026 08:07:36 +0530 Subject: [PATCH 4/7] fix(operator): aggregate drain-blocked state across nodes per reconcile pass Addresses ayuskauskas's review on PR #582. The prior per-node approach let EnsureNodeIsReadyForInterrupt read-modify-write a single shared Blocked condition once per node per reconcile, causing three real bugs: an unblocked node could clobber a blocked node's condition and re-trigger its Warning event every pass; a paused, deleted, or deselected node could leave the condition permanently stuck True since only the per-node path ever cleared it; and blocking pod names were reported without their namespace. - updateDrainBlockedCondition now runs once per reconcile pass in refreshSkyhookConditions, computing the full set of currently-blocked nodes fresh from live state and writing a single aggregate condition and at most one Warning event per genuine transition into the blocked state. - EnsureNodeIsReadyForInterrupt no longer touches the shared condition; it still gates per-node drain and now also emits a supplementary Warning event on the Node object itself, matching the existing DrainTimeout convention. - HasNonInterruptWork now reports pod names as namespace/name. - Collapsed RemoveSkyhookConditionTypeAndReason into RemoveSkyhookConditionTypes via a shared predicate helper, and replaced FindSkyhookCondition with the existing upstream meta.FindStatusCondition. - Mirrored the CRD field description into chart/templates, documented the Blocked/NonInterruptPodsRunning reason and the cordon-persists-throughout behavior, and updated RELEASE_NOTES.md to match the new aggregate behavior. - Added a two-node envtest regression covering the exact contention bug: one blocked node, one unblocked, across multiple reconcile passes, including coexistence with an unrelated DependencyUninstalled condition. Signed-off-by: bharqav --- chart/templates/nodewright-crd.yaml | 6 +- chart/templates/skyhook-crd.yaml | 6 +- docs/architecture/interrupt-flow.md | 10 +- docs/architecture/operator-status.md | 3 + operator/RELEASE_NOTES.md | 23 +- .../internal/controller/cluster_state_v2.go | 5 +- .../internal/controller/skyhook_controller.go | 165 ++++++++--- .../controller/skyhook_controller_test.go | 271 ++++++++++++++---- .../internal/wrapper/skyhook_conditions.go | 61 ++-- .../wrapper/skyhook_conditions_test.go | 22 +- 10 files changed, 390 insertions(+), 182 deletions(-) diff --git a/chart/templates/nodewright-crd.yaml b/chart/templates/nodewright-crd.yaml index e14bfc675..c95ca3684 100644 --- a/chart/templates/nodewright-crd.yaml +++ b/chart/templates/nodewright-crd.yaml @@ -586,8 +586,10 @@ spec: description: Packages are the DAG of packages to be applied to nodes. type: object podNonInterruptLabels: - description: PodNonInterruptLabels are a set of labels we want to - monitor pods for whether they Interruptible + description: |- + PodNonInterruptLabels are a set of labels we want to monitor pods for whether they are interruptible. + Matching Pending or Running pods block pre-drain progress indefinitely; spec.drainConfig.timeout does not + apply to this barrier. A Blocked condition with reason NonInterruptPodsRunning is surfaced while this barrier holds. properties: matchExpressions: description: matchExpressions is a list of label selector requirements. diff --git a/chart/templates/skyhook-crd.yaml b/chart/templates/skyhook-crd.yaml index c0d3f7034..7b3a42ff1 100644 --- a/chart/templates/skyhook-crd.yaml +++ b/chart/templates/skyhook-crd.yaml @@ -567,8 +567,10 @@ spec: description: Packages are the DAG of packages to be applied to nodes. type: object podNonInterruptLabels: - description: PodNonInterruptLabels are a set of labels we want to - monitor pods for whether they Interruptible + description: |- + PodNonInterruptLabels are a set of labels we want to monitor pods for whether they are interruptible. + Matching Pending or Running pods block pre-drain progress indefinitely; spec.drainConfig.timeout does not + apply to this barrier. A Blocked condition with reason NonInterruptPodsRunning is surfaced while this barrier holds. properties: matchExpressions: description: matchExpressions is a list of label selector requirements. diff --git a/docs/architecture/interrupt-flow.md b/docs/architecture/interrupt-flow.md index f4e73f572..4be725da8 100644 --- a/docs/architecture/interrupt-flow.md +++ b/docs/architecture/interrupt-flow.md @@ -163,10 +163,12 @@ matching more closely: the unschedulable toleration check uses Kubernetes owner reference, and mirror/static pods are ignored. `podNonInterruptLabels` remains a pre-drain barrier. Matching pods must finish -or move away before the operator starts the configurable drain step. This wait -is unbounded and drain timeout does not apply. While this barrier holds, -administrators will see a `Blocked` condition with reason `NonInterruptPodsRunning` -reporting which pods are causing the hold. +or move away before the operator starts the configurable drain step. The node is cordoned +before entering this barrier and remains cordoned throughout the wait so no replacement +workloads can schedule on it. This wait is unbounded and drain timeout does not apply. +While this barrier holds, administrators will see a `Blocked` condition with reason +`NonInterruptPodsRunning` reporting which nodes are held, along with Warning events on +both the NodeWright and affected Node objects detailing the hold. ### When Drain Is Complete diff --git a/docs/architecture/operator-status.md b/docs/architecture/operator-status.md index b71155f78..373899ca2 100644 --- a/docs/architecture/operator-status.md +++ b/docs/architecture/operator-status.md @@ -83,6 +83,9 @@ The truncation cap exists to keep condition payloads bounded for etcd object siz The operator also sets additional condition types that may be useful for troubleshooting: +- `Blocked`: rollout progress is blocked on one or more nodes. Reasons include: + - `NonInterruptPodsRunning`: node drain is held because pods matching `spec.podNonInterruptLabels` are still running or pending on selected nodes + - `DependencyUninstalled`: a required package dependency is being or has been uninstalled - `TaintNotTolerable`: selected nodes are skipped because their taints are not tolerated by the NodeWright - `NodesIgnored`: selected nodes are skipped because they have the ignore label set - `ApplyPackage`: the controller is applying a package to a node diff --git a/operator/RELEASE_NOTES.md b/operator/RELEASE_NOTES.md index 4b20a7030..3e9752fd0 100644 --- a/operator/RELEASE_NOTES.md +++ b/operator/RELEASE_NOTES.md @@ -10,21 +10,22 @@ For the full commit-level log see CHANGELOG.md. - **A `Blocked` status condition (reason `NonInterruptPodsRunning`) and a Warning event are now surfaced when `spec.podNonInterruptLabels` blocks node drain.** Previously, the operator held the node in `Ready=False` / `Progressing` with no condition or - event indicating why or which pods were causing the hold. When a node's interrupt - is held at this barrier: + event indicating why or which pods were causing the hold. When nodes with packages + requiring interrupt are held at this barrier: - A `Blocked` condition (status `True`, reason `NonInterruptPodsRunning`) is set on the - NodeWright. Its message identifies the blocking pods on that node: up to 10 pods - (`wrapper.ReadyConditionNodeListLimit`) are listed by name (e.g. `Pod [x] is running. Waiting.` - or `Pods [x, y] are running. Waiting.`). If more than 10 pods are running on the node, - the full list is logged at info level and the condition message is summarized as - ` pods are running. Waiting.`. + NodeWright once per reconcile pass. Its message identifies the blocked nodes using standard + node list formatting: up to 10 nodes (`wrapper.ReadyConditionNodeListLimit`) are listed by + name (e.g. `1 node blocked by non-interrupt pods (node-a). Waiting.` or `2 nodes blocked by non-interrupt pods (node-a, node-b). Waiting.`). + If more than 10 nodes are blocked, the full list is logged at info level once per pass and the + condition message is summarized with `(list truncated; see controller logs)`. - A Warning event with reason `Drain` (`EventsReasonSkyhookDrain`) is emitted on the NodeWright - object when this barrier is first entered, reporting the node and package being held - (`drain blocked by non-interrupt pods for node [] package [:]`). + object on transition from unblocked to blocked, reporting the held nodes. + - A supplementary Warning event is also emitted on each affected Node object detailing the + namespace-qualified blocking pods and package being held. - If another `Blocked` reason is already active (such as `DependencyUninstalled`), the - `NonInterruptPodsRunning` condition and Warning event are deferred and will only appear + `NonInterruptPodsRunning` condition and NodeWright Warning event are deferred and will only appear once that other condition clears. Once all matching non-interrupt pods finish or - terminate on the node, the `NonInterruptPodsRunning` condition is removed and drain proceeds, + terminate, the `NonInterruptPodsRunning` condition is removed and drain proceeds, preserving any unrelated `Blocked` condition that may also be active. - **Adding and removing the finalizer from a natively authored NodeWright no diff --git a/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go index f71b5189e..9777bb59e 100644 --- a/operator/internal/controller/cluster_state_v2.go +++ b/operator/internal/controller/cluster_state_v2.go @@ -31,6 +31,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" @@ -614,7 +615,9 @@ func (s *skyhookNodes) UpdateBlockedCondition() error { Message: strings.Join(blockedMsgs, "; "), }) } else { - existing := wrapper.FindSkyhookCondition(s.skyhook, wrapper.SkyhookConditionBlocked) + // Preserving NonInterruptPodsRunning here allows updateDrainBlockedCondition to own + // and clear its own reason; clearing it here would clobber in-flight drain waits. + existing := meta.FindStatusCondition(s.skyhook.Status.Conditions, wrapper.SkyhookConditionBlocked) if existing != nil && existing.Reason != wrapper.SkyhookReasonNonInterruptPodsRunning { wrapper.RemoveSkyhookConditionTypes(s.skyhook, wrapper.SkyhookConditionBlocked) } diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index aa6485b5a..e27aad09b 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -44,6 +44,7 @@ import ( corev1 "k8s.io/api/core/v1" policyv1 "k8s.io/api/policy/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -608,12 +609,118 @@ func (r *SkyhookReconciler) refreshSkyhookConditions(ctx context.Context, cluste if err := skyhook.UpdateBlockedCondition(); err != nil { return fmt.Errorf("error updating blocked condition: %w", err) } + if err := r.updateDrainBlockedCondition(ctx, skyhook); err != nil { + return fmt.Errorf("error updating drain blocked condition: %w", err) + } if err := skyhook.UpdateUninstallConditions(); err != nil { return fmt.Errorf("error updating uninstall conditions: %w", err) } return nil } +// nodeNeedsInterruptDrain reports whether the node has a runnable package with an interrupt +// that is currently at the pre-drain apply or uninstall stage, matching ProcessInterrupt's entry gate. +func nodeNeedsInterruptDrain(node wrapper.SkyhookNode) bool { + if node.IsComplete() { + return false + } + toRun, err := node.RunNext() + if err != nil || len(toRun) == 0 { + return false + } + for _, pkg := range toRun { + if !node.HasInterrupt(*pkg) { + continue + } + stage := v1alpha1.StageApply + if nextStage := node.NextStage(pkg); nextStage != nil { + stage = *nextStage + } + if stage == v1alpha1.StageApply || stage == v1alpha1.StageUninstall { + return true + } + } + return false +} + +// updateDrainBlockedCondition aggregates non-interrupt pod blocking state across all in-scope +// nodes once per reconcile pass. It updates the NodeWright-level Blocked condition and emits a +// Warning event on the NodeWright only on a genuine transition from unblocked to blocked. +func (r *SkyhookReconciler) updateDrainBlockedCondition(ctx context.Context, skyhook SkyhookNodes) error { + logger := log.FromContext(ctx) + + selector, err := metav1.LabelSelectorAsSelector(&skyhook.GetSkyhook().Spec.PodNonInterruptLabels) + if err != nil { + return fmt.Errorf("error creating selector: %w", err) + } + if selector.Empty() { + wrapper.RemoveSkyhookConditionTypeAndReason(skyhook.GetSkyhook(), wrapper.SkyhookConditionBlocked, wrapper.SkyhookReasonNonInterruptPodsRunning) + return nil + } + + // Defer NonInterruptPodsRunning if another Blocked reason (e.g. DependencyUninstalled) + // currently holds the single Blocked condition slot on the NodeWright. + existing := meta.FindStatusCondition(skyhook.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) + if existing != nil && existing.Reason != wrapper.SkyhookReasonNonInterruptPodsRunning { + return nil + } + + var blockedNodes []string + for _, node := range skyhook.GetNodes() { + if !nodeNeedsInterruptDrain(node) { + continue + } + + hasWork, _, err := r.HasNonInterruptWork(ctx, node) + if err != nil { + return fmt.Errorf("checking non-interrupt work for node [%s]: %w", node.GetNode().Name, err) + } + if hasWork { + blockedNodes = append(blockedNodes, node.GetNode().Name) + } + } + + if len(blockedNodes) > 0 { + sort.Strings(blockedNodes) + + if len(blockedNodes) > wrapper.ReadyConditionNodeListLimit { + logger.Info("Condition message truncated for non-interrupt pods", "nodewright", skyhook.GetSkyhook().Name, "nodes", blockedNodes) + } + + nodeLabel := "node" + if len(blockedNodes) > 1 { + nodeLabel = "nodes" + } + message := fmt.Sprintf("%d %s blocked by non-interrupt pods%s. Waiting.", + len(blockedNodes), + nodeLabel, + wrapper.FormatNodeList(blockedNodes), + ) + + if existing == nil { + r.recorder.Eventf(skyhook.GetSkyhook().NodeWright, nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, wrapper.SkyhookReasonNonInterruptPodsRunning, + "drain blocked by non-interrupt pods on %d %s%s", + len(blockedNodes), + nodeLabel, + wrapper.FormatNodeList(blockedNodes), + ) + } + + wrapper.AddSkyhookCondition(skyhook.GetSkyhook(), metav1.Condition{ + Type: wrapper.SkyhookConditionBlocked, + Status: metav1.ConditionTrue, + Reason: wrapper.SkyhookReasonNonInterruptPodsRunning, + Message: message, + ObservedGeneration: skyhook.GetSkyhook().Generation, + LastTransitionTime: metav1.Now(), + }) + } else if existing != nil && existing.Reason == wrapper.SkyhookReasonNonInterruptPodsRunning { + wrapper.RemoveSkyhookConditionTypeAndReason(skyhook.GetSkyhook(), wrapper.SkyhookConditionBlocked, wrapper.SkyhookReasonNonInterruptPodsRunning) + } + + return nil +} + // processSkyhooksPerNode processes all skyhooks for nodes that are ready (per-node priority ordering). // A node is ready for a skyhook if all higher-priority skyhooks are complete on that specific node. func (r *SkyhookReconciler) processSkyhooksPerNode(ctx context.Context, clusterState *clusterState, nodePicker *NodePicker, logger logr.Logger) (*ctrl.Result, error) { @@ -2442,7 +2549,11 @@ func (r *SkyhookReconciler) HasNonInterruptWork(ctx context.Context, skyhookNode for _, pod := range pods.Items { switch pod.Status.Phase { case corev1.PodRunning, corev1.PodPending: - podNames = append(podNames, pod.Name) + podName := pod.Name + if pod.Namespace != "" { + podName = fmt.Sprintf("%s/%s", pod.Namespace, pod.Name) + } + podNames = append(podNames, podName) } } @@ -3271,51 +3382,19 @@ func (r *SkyhookReconciler) EnsureNodeIsReadyForInterrupt(ctx context.Context, s return false, err } if hasWork { // keep waiting... - logger := log.FromContext(ctx) - podLabel := "Pod" - podVerb := "is" - if len(podNames) > 1 { - podLabel = "Pods" - podVerb = "are" - } - message := fmt.Sprintf("%s [%s] %s running. Waiting.", podLabel, strings.Join(podNames, ", "), podVerb) - if len(podNames) > wrapper.ReadyConditionNodeListLimit { - logger.Info("Condition message truncated for non-interrupt pods", "nodewright", skyhookNode.GetSkyhook().Name, "pods", podNames) - message = fmt.Sprintf("%d pods are running. Waiting.", len(podNames)) - } - - existing := wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) - if existing == nil || existing.Reason == wrapper.SkyhookReasonNonInterruptPodsRunning { - if existing == nil { - pkgName := "" - pkgVersion := "" - if _package != nil { - pkgName = _package.Name - pkgVersion = _package.Version - } - r.recorder.Eventf(skyhookNode.GetSkyhook().NodeWright, nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, wrapper.SkyhookReasonNonInterruptPodsRunning, - "drain blocked by non-interrupt pods for node [%s] package [%s:%s]", - skyhookNode.GetNode().Name, - pkgName, - pkgVersion, - ) - } - - wrapper.AddSkyhookCondition(skyhookNode.GetSkyhook(), metav1.Condition{ - Type: wrapper.SkyhookConditionBlocked, - Status: metav1.ConditionTrue, - Reason: wrapper.SkyhookReasonNonInterruptPodsRunning, - Message: message, - ObservedGeneration: skyhookNode.GetSkyhook().Generation, - LastTransitionTime: metav1.Now(), - }) - } - + // Condition and NodeWright-level event are managed once per reconcile pass by + // updateDrainBlockedCondition. Here we emit a supplementary event on the Node itself + // (matching DrainTimeout behavior) so node inspection reflects why drain is waiting. + r.recorder.Eventf(skyhookNode.GetNode(), nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, wrapper.SkyhookReasonNonInterruptPodsRunning, + "drain blocked by non-interrupt pods [%s] for package [%s:%s] from [nodewright:%s]", + strings.Join(podNames, ", "), + _package.Name, + _package.Version, + skyhookNode.GetSkyhook().Name, + ) return false, nil } - wrapper.RemoveSkyhookConditionTypeAndReason(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked, wrapper.SkyhookReasonNonInterruptPodsRunning) - ready, err := r.DrainNode(ctx, skyhookNode, _package) if err != nil { return false, fmt.Errorf("error draining node [%s]: %w", skyhookNode.GetNode().Name, err) diff --git a/operator/internal/controller/skyhook_controller_test.go b/operator/internal/controller/skyhook_controller_test.go index 8019f7a55..d740656be 100644 --- a/operator/internal/controller/skyhook_controller_test.go +++ b/operator/internal/controller/skyhook_controller_test.go @@ -36,6 +36,7 @@ import ( "github.com/stretchr/testify/mock" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -846,38 +847,22 @@ var _ = Describe("skyhook controller tests", func() { Expect(err).ToNot(HaveOccurred()) Expect(drainStartedAt).To(BeNil()) - cond := wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) - Expect(cond).ToNot(BeNil()) - Expect(cond.Status).To(Equal(metav1.ConditionTrue)) - Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) - Expect(cond.Message).To(Equal("Pod [golden] is running. Waiting.")) - Eventually(recorder.Events).Should(Receive(ContainSubstring("Warning Drain drain blocked by non-interrupt pods for node [node-a] package [pkg:1.0.0]"))) + Eventually(recorder.Events).Should(Receive(ContainSubstring("Warning Drain drain blocked by non-interrupt pods [default/golden] for package [pkg:1.0.0] from [nodewright:drain-golden]"))) Expect(testClient.Delete(ctx, goldenPod)).To(Succeed()) - _, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, &v1alpha1.Package{ + ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, &v1alpha1.Package{ PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, }) Expect(err).ToNot(HaveOccurred()) - cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) - Expect(cond).To(BeNil()) + Expect(ready).To(BeFalse(), "evictable pod is still being drained") + Expect(deleteCalled).To(BeTrue(), "DrainNode should delete evictable pod") - // Verify that an unrelated Blocked condition (e.g. DependencyUninstalled) is preserved - wrapper.AddSkyhookCondition(skyhookNode.GetSkyhook(), metav1.Condition{ - Type: wrapper.SkyhookConditionBlocked, - Status: metav1.ConditionTrue, - Reason: "DependencyUninstalled", - Message: "package pkg is blocked: dependency dep has been uninstalled", - ObservedGeneration: skyhookNode.GetSkyhook().Generation, - LastTransitionTime: metav1.Now(), - }) - _, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, &v1alpha1.Package{ + ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, &v1alpha1.Package{ PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, }) Expect(err).ToNot(HaveOccurred()) - cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) - Expect(cond).ToNot(BeNil()) - Expect(cond.Reason).To(Equal("DependencyUninstalled")) + Expect(ready).To(BeTrue(), "node should be ready once all pods are drained") }) It("suppresses drain warning events and preserves condition while DependencyUninstalled is active, then emits exactly once when cleared", func() { @@ -905,16 +890,18 @@ var _ = Describe("skyhook controller tests", func() { node := &corev1.Node{ ObjectMeta: metav1.ObjectMeta{ - Name: "node-a", - Annotations: map[string]string{ - fmt.Sprintf("%s/cordon_%s", v1alpha1.METADATA_PREFIX, "drain-dep"): "true", - }, + Name: "node-a", + Labels: map[string]string{"drain-test": "true"}, }, - Spec: corev1.NodeSpec{Unschedulable: true}, + } + pkg := v1alpha1.Package{ + PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, + Interrupt: &v1alpha1.Interrupt{Type: "reboot"}, } skyhook := &v1alpha1.NodeWright{ ObjectMeta: metav1.ObjectMeta{Name: "drain-dep"}, Spec: v1alpha1.NodeWrightSpec{ + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"drain-test": "true"}}, PodNonInterruptLabels: metav1.LabelSelector{ MatchLabels: map[string]string{ "workload": "golden", @@ -923,63 +910,59 @@ var _ = Describe("skyhook controller tests", func() { DrainConfig: &v1alpha1.DrainConfig{ DisableEviction: ptr(true), }, - Packages: v1alpha1.Packages{}, + Packages: v1alpha1.Packages{ + pkg.Name: pkg, + }, }, } - skyhookNode, err := wrapper.NewSkyhookNode(node, skyhook) + + clusterState, err := BuildState( + &v1alpha1.NodeWrightList{Items: []v1alpha1.NodeWright{*skyhook}}, + &corev1.NodeList{Items: []corev1.Node{*node}}, + &v1alpha1.DeploymentPolicyList{}, + ) Expect(err).ToNot(HaveOccurred()) + sn := clusterState.skyhooks[0] // Pre-seed unrelated Blocked condition - wrapper.AddSkyhookCondition(skyhookNode.GetSkyhook(), metav1.Condition{ + wrapper.AddSkyhookCondition(sn.GetSkyhook(), metav1.Condition{ Type: wrapper.SkyhookConditionBlocked, Status: metav1.ConditionTrue, Reason: "DependencyUninstalled", Message: "package pkg is blocked: dependency dep has been uninstalled", - ObservedGeneration: skyhookNode.GetSkyhook().Generation, + ObservedGeneration: sn.GetSkyhook().Generation, LastTransitionTime: metav1.Now(), }) - pkg := &v1alpha1.Package{ - PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, - } - - // (a) Reconcile 1: non-interrupt pods present, but DependencyUninstalled owns the Blocked slot - ready, err := r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, pkg) - Expect(err).ToNot(HaveOccurred()) - Expect(ready).To(BeFalse()) - cond := wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + // (a) Pass 1: non-interrupt pods present, but DependencyUninstalled owns the Blocked slot + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + cond := meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) Expect(cond).ToNot(BeNil()) Expect(cond.Reason).To(Equal("DependencyUninstalled")) Expect(cond.Message).To(Equal("package pkg is blocked: dependency dep has been uninstalled")) Expect(recorder.Events).To(BeEmpty()) - // (a) Reconcile 2: second pass with DependencyUninstalled still active - ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, pkg) - Expect(err).ToNot(HaveOccurred()) - Expect(ready).To(BeFalse()) - cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + // (a) Pass 2: second pass with DependencyUninstalled still active + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + cond = meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) Expect(cond).ToNot(BeNil()) Expect(cond.Reason).To(Equal("DependencyUninstalled")) Expect(recorder.Events).To(BeEmpty()) // (b) DependencyUninstalled is cleared, non-interrupt pods still present - wrapper.RemoveSkyhookConditionTypes(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + wrapper.RemoveSkyhookConditionTypes(sn.GetSkyhook(), wrapper.SkyhookConditionBlocked) - // Reconcile 3: immediately takes ownership of the Blocked slot and emits exactly one event - ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, pkg) - Expect(err).ToNot(HaveOccurred()) - Expect(ready).To(BeFalse()) - cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + // Pass 3: immediately takes ownership of the Blocked slot and emits exactly one event + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + cond = meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) Expect(cond).ToNot(BeNil()) Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) - Expect(cond.Message).To(Equal("Pod [golden] is running. Waiting.")) - Eventually(recorder.Events).Should(Receive(ContainSubstring("Warning Drain drain blocked by non-interrupt pods for node [node-a] package [pkg:1.0.0]"))) + Expect(cond.Message).To(Equal("1 node blocked by non-interrupt pods (node-a). Waiting.")) + Eventually(recorder.Events).Should(Receive(ContainSubstring("Warning Drain drain blocked by non-interrupt pods on 1 node (node-a)"))) - // Reconcile 4: subsequent pass with non-interrupt pods still present emits no duplicate event - ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, pkg) - Expect(err).ToNot(HaveOccurred()) - Expect(ready).To(BeFalse()) - cond = wrapper.FindSkyhookCondition(skyhookNode.GetSkyhook(), wrapper.SkyhookConditionBlocked) + // Pass 4: subsequent pass with non-interrupt pods still present emits no duplicate event + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + cond = meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) Expect(cond).ToNot(BeNil()) Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) Expect(recorder.Events).To(BeEmpty()) @@ -5089,3 +5072,175 @@ var _ = Describe("HandleFinalizer merge patch", func() { Expect(live.Labels).To(HaveKeyWithValue("concurrent-label", "applied")) }) }) + +var _ = Describe("drain blocked by non-interrupt pods multi-node reconcile", func() { + It("correctly reflects blocked nodes across passes without duplicate events and respects DependencyUninstalled precedence", func() { + const nodeAName = "multi-drain-node-a" + const nodeBName = "multi-drain-node-b" + const podName = "multi-drain-golden-a" + const skyhookName = "two-node-drain" + testLabels := map[string]string{"multi-drain-test": "two-node"} + + cordonAnnoKey := fmt.Sprintf("%s/cordon_%s", v1alpha1.METADATA_PREFIX, skyhookName) + nodeA := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: nodeAName, + Labels: testLabels, + Annotations: map[string]string{cordonAnnoKey: "true"}, + }, + Spec: corev1.NodeSpec{Unschedulable: true}, + } + nodeB := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: nodeBName, + Labels: testLabels, + Annotations: map[string]string{cordonAnnoKey: "true"}, + }, + Spec: corev1.NodeSpec{Unschedulable: true}, + } + + Expect(k8sClient.Create(ctx, nodeA)).To(Succeed()) + DeferCleanup(func() { _ = client.IgnoreNotFound(k8sClient.Delete(ctx, nodeA)) }) + + Expect(k8sClient.Create(ctx, nodeB)).To(Succeed()) + DeferCleanup(func() { _ = client.IgnoreNotFound(k8sClient.Delete(ctx, nodeB)) }) + + goldenPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: "default", + Labels: map[string]string{ + "workload": "golden", + }, + }, + Spec: corev1.PodSpec{ + NodeName: nodeAName, + Containers: []corev1.Container{ + {Name: "workload", Image: "busybox"}, + }, + }, + } + Expect(k8sClient.Create(ctx, goldenPod)).To(Succeed()) + DeferCleanup(func() { _ = client.IgnoreNotFound(k8sClient.Delete(ctx, goldenPod)) }) + + goldenPod.Status.Phase = corev1.PodRunning + Expect(k8sClient.Status().Update(ctx, goldenPod)).To(Succeed()) + + pkg := v1alpha1.Package{ + PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, + Interrupt: &v1alpha1.Interrupt{Type: "reboot"}, + } + + skyhook := &v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: skyhookName}, + Spec: v1alpha1.NodeWrightSpec{ + NodeSelector: metav1.LabelSelector{ + MatchLabels: testLabels, + }, + PodNonInterruptLabels: metav1.LabelSelector{ + MatchLabels: map[string]string{ + "workload": "golden", + }, + }, + DrainConfig: &v1alpha1.DrainConfig{ + DisableEviction: ptr(true), + }, + Packages: v1alpha1.Packages{ + pkg.Name: pkg, + }, + }, + } + + clusterState, err := BuildState( + &v1alpha1.NodeWrightList{Items: []v1alpha1.NodeWright{*skyhook}}, + &corev1.NodeList{Items: []corev1.Node{*nodeA, *nodeB}}, + &v1alpha1.DeploymentPolicyList{}, + ) + Expect(err).ToNot(HaveOccurred()) + Expect(clusterState.skyhooks).To(HaveLen(1)) + sn := clusterState.skyhooks[0] + Expect(sn.GetNodes()).To(HaveLen(2)) + + recorder := events.NewFakeRecorder(20) + r := &SkyhookReconciler{ + Client: k8sClient, + uncached: k8sClient, + dal: dal.New(k8sClient, nil), + recorder: recorder, + opts: opts, + scheme: k8sClient.Scheme(), + } + + // Pass 1: node-a blocked by non-interrupt pod, node-b has no blocking pods. + // Condition reflects only node-a; exactly one Warning event on NodeWright; supplementary event on node-a. + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + + cond := meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Status).To(Equal(metav1.ConditionTrue)) + Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) + Expect(cond.Message).To(Equal(fmt.Sprintf("1 node blocked by non-interrupt pods (%s). Waiting.", nodeAName))) + Eventually(recorder.Events).Should(Receive(ContainSubstring(fmt.Sprintf("Warning Drain drain blocked by non-interrupt pods on 1 node (%s)", nodeAName)))) + + _, nodeWrapperA := sn.GetNode(nodeAName) + _, nodeWrapperB := sn.GetNode(nodeBName) + Expect(nodeWrapperA).ToNot(BeNil()) + Expect(nodeWrapperB).ToNot(BeNil()) + + readyA, err := r.EnsureNodeIsReadyForInterrupt(ctx, nodeWrapperA, &pkg) + Expect(err).ToNot(HaveOccurred()) + Expect(readyA).To(BeFalse()) + Eventually(recorder.Events).Should(Receive(ContainSubstring(fmt.Sprintf("Warning Drain drain blocked by non-interrupt pods [default/%s] for package [pkg:1.0.0] from [nodewright:%s]", podName, skyhookName)))) + + readyB, err := r.EnsureNodeIsReadyForInterrupt(ctx, nodeWrapperB, &pkg) + Expect(err).ToNot(HaveOccurred()) + Expect(readyB).To(BeTrue(), "node-b has no non-interrupt work and should be ready for interrupt") + + // Pass 2: subsequent reconcile with condition unchanged emits zero duplicate events. + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + cond = meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) + Expect(cond.Message).To(Equal(fmt.Sprintf("1 node blocked by non-interrupt pods (%s). Waiting.", nodeAName))) + Expect(recorder.Events).To(BeEmpty()) + + // Pass 3: inject DependencyUninstalled — it takes precedence, suppressing NonInterruptPodsRunning without event spam. + wrapper.AddSkyhookCondition(sn.GetSkyhook(), metav1.Condition{ + Type: wrapper.SkyhookConditionBlocked, + Status: metav1.ConditionTrue, + Reason: "DependencyUninstalled", + Message: "package pkg is blocked: dependency dep has been uninstalled", + ObservedGeneration: sn.GetSkyhook().Generation, + LastTransitionTime: metav1.Now(), + }) + + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + cond = meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Reason).To(Equal("DependencyUninstalled")) + Expect(cond.Message).To(Equal("package pkg is blocked: dependency dep has been uninstalled")) + Expect(recorder.Events).To(BeEmpty()) + + // Pass 4: DependencyUninstalled is cleared — NonInterruptPodsRunning immediately restored with 1 Warning event. + wrapper.RemoveSkyhookConditionTypes(sn.GetSkyhook(), wrapper.SkyhookConditionBlocked) + + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + cond = meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) + Expect(cond).ToNot(BeNil()) + Expect(cond.Reason).To(Equal(wrapper.SkyhookReasonNonInterruptPodsRunning)) + Expect(cond.Message).To(Equal(fmt.Sprintf("1 node blocked by non-interrupt pods (%s). Waiting.", nodeAName))) + Eventually(recorder.Events).Should(Receive(ContainSubstring(fmt.Sprintf("Warning Drain drain blocked by non-interrupt pods on 1 node (%s)", nodeAName)))) + Expect(recorder.Events).To(BeEmpty()) + + // Pass 5: delete the blocking pod — Blocked condition is cleared and node-a becomes ready. + Expect(k8sClient.Delete(ctx, goldenPod, client.GracePeriodSeconds(0))).To(Succeed()) + + Expect(r.updateDrainBlockedCondition(ctx, sn)).To(Succeed()) + cond = meta.FindStatusCondition(sn.GetSkyhook().Status.Conditions, wrapper.SkyhookConditionBlocked) + Expect(cond).To(BeNil(), "Blocked condition must be removed when no nodes are blocked") + + readyA, err = r.EnsureNodeIsReadyForInterrupt(ctx, nodeWrapperA, &pkg) + Expect(err).ToNot(HaveOccurred()) + Expect(readyA).To(BeTrue(), "node-a should be ready for interrupt once non-interrupt pods are gone") + }) +}) diff --git a/operator/internal/wrapper/skyhook_conditions.go b/operator/internal/wrapper/skyhook_conditions.go index f368e9888..9509edc8a 100644 --- a/operator/internal/wrapper/skyhook_conditions.go +++ b/operator/internal/wrapper/skyhook_conditions.go @@ -195,20 +195,17 @@ func addOrUpdateSkyhookCondition(conditions []metav1.Condition, condition metav1 return conditions, true } -func RemoveSkyhookConditionTypes(skyhook *Skyhook, conditionTypes ...string) bool { +// removeSkyhookConditions removes any condition matching predicate. +// Returns true if the condition slice was modified. +func removeSkyhookConditions(skyhook *Skyhook, shouldRemove func(metav1.Condition) bool) bool { if len(skyhook.Status.Conditions) == 0 { return false } - remove := make(map[string]struct{}, len(conditionTypes)) - for _, conditionType := range conditionTypes { - remove[conditionType] = struct{}{} - } - conditions := skyhook.Status.Conditions[:0] changed := false for _, condition := range skyhook.Status.Conditions { - if _, ok := remove[condition.Type]; ok { + if shouldRemove(condition) { changed = true continue } @@ -223,40 +220,24 @@ func RemoveSkyhookConditionTypes(skyhook *Skyhook, conditionTypes ...string) boo return changed } -// RemoveSkyhookConditionTypeAndReason removes any condition matching both conditionType and reason. -// Returns true if the condition slice was modified. -func RemoveSkyhookConditionTypeAndReason(skyhook *Skyhook, conditionType, reason string) bool { - if len(skyhook.Status.Conditions) == 0 { - return false - } - - conditions := skyhook.Status.Conditions[:0] - changed := false - for _, condition := range skyhook.Status.Conditions { - if condition.Type == conditionType && condition.Reason == reason { - changed = true - continue - } - conditions = append(conditions, condition) - } - - if changed { - skyhook.Status.Conditions = conditions - skyhook.Updated = true +func RemoveSkyhookConditionTypes(skyhook *Skyhook, conditionTypes ...string) bool { + remove := make(map[string]struct{}, len(conditionTypes)) + for _, conditionType := range conditionTypes { + remove[conditionType] = struct{}{} } - return changed + return removeSkyhookConditions(skyhook, func(condition metav1.Condition) bool { + _, ok := remove[condition.Type] + return ok + }) } -// FindSkyhookCondition searches skyhook.Status.Conditions for a condition with the given conditionType. -// Returns a pointer to the matching condition if present, or nil otherwise. -func FindSkyhookCondition(skyhook *Skyhook, conditionType string) *metav1.Condition { - for i := range skyhook.Status.Conditions { - if skyhook.Status.Conditions[i].Type == conditionType { - return &skyhook.Status.Conditions[i] - } - } - return nil +// RemoveSkyhookConditionTypeAndReason removes any condition matching both conditionType and reason. +// Returns true if the condition slice was modified. +func RemoveSkyhookConditionTypeAndReason(skyhook *Skyhook, conditionType, reason string) bool { + return removeSkyhookConditions(skyhook, func(condition metav1.Condition) bool { + return condition.Type == conditionType && condition.Reason == reason + }) } func HasTrueSkyhookCondition(skyhook *Skyhook, conditionTypes ...string) bool { @@ -301,7 +282,7 @@ func SkyhookReadyConditionMessageTruncated(byStatus map[v1alpha1.Status][]string func skyhookReadyConditionMessageFromStatusGroups(byStatus map[v1alpha1.Status][]string, total int) string { complete := len(byStatus[v1alpha1.StatusComplete]) - parts := []string{fmt.Sprintf("%d/%d nodes complete%s", complete, total, formatNodeList(byStatus[v1alpha1.StatusComplete]))} + parts := []string{fmt.Sprintf("%d/%d nodes complete%s", complete, total, FormatNodeList(byStatus[v1alpha1.StatusComplete]))} for _, status := range []v1alpha1.Status{ v1alpha1.StatusInProgress, @@ -316,7 +297,7 @@ func skyhookReadyConditionMessageFromStatusGroups(byStatus map[v1alpha1.Status][ if len(nodes) == 0 { continue } - parts = append(parts, fmt.Sprintf("%d %s%s", len(nodes), nodeProgressStatusLabel(status), formatNodeList(nodes))) + parts = append(parts, fmt.Sprintf("%d %s%s", len(nodes), nodeProgressStatusLabel(status), FormatNodeList(nodes))) } return strings.Join(parts, ", ") @@ -331,7 +312,7 @@ func nodeProgressStatusLabel(status v1alpha1.Status) string { } } -func formatNodeList(nodes []string) string { +func FormatNodeList(nodes []string) string { if len(nodes) == 0 { return "" } diff --git a/operator/internal/wrapper/skyhook_conditions_test.go b/operator/internal/wrapper/skyhook_conditions_test.go index c5220f88f..7857bbd95 100644 --- a/operator/internal/wrapper/skyhook_conditions_test.go +++ b/operator/internal/wrapper/skyhook_conditions_test.go @@ -230,26 +230,6 @@ var _ = Describe("Skyhook condition helpers", func() { }, false), ) - Describe("FindSkyhookCondition", func() { - It("finds an existing condition and returns nil when not found", func() { - skyhook := &Skyhook{ - NodeWright: &v1alpha1.NodeWright{ - Status: v1alpha1.NodeWrightStatus{ - Conditions: []metav1.Condition{ - {Type: SkyhookConditionBlocked, Reason: SkyhookReasonNonInterruptPodsRunning}, - }, - }, - }, - } - - cond := FindSkyhookCondition(skyhook, SkyhookConditionBlocked) - Expect(cond).NotTo(BeNil()) - Expect(cond.Reason).To(Equal(SkyhookReasonNonInterruptPodsRunning)) - - Expect(FindSkyhookCondition(skyhook, SkyhookConditionReady)).To(BeNil()) - }) - }) - DescribeTable("hasTrueSkyhookCondition", func(conditions []metav1.Condition, conditionTypes []string, expected bool) { skyhook := &Skyhook{ NodeWright: &v1alpha1.NodeWright{ @@ -293,6 +273,6 @@ var _ = Describe("Skyhook condition helpers", func() { nodes = append(nodes, fmt.Sprintf("node-%02d", i)) } - Expect(formatNodeList(nodes)).To(Equal(" (list truncated; see controller logs)")) + Expect(FormatNodeList(nodes)).To(Equal(" (list truncated; see controller logs)")) }) }) From 8f9edd279c67d87de440a50aa478c5d370aee046 Mon Sep 17 00:00:00 2001 From: bharqav Date: Sat, 12 Sep 2026 08:28:18 +0530 Subject: [PATCH 5/7] fix(operator): bound pod list in node-level drain warning event Cap the pod list in the Node-level Warning event emitted from EnsureNodeIsReadyForInterrupt at wrapper.ReadyConditionNodeListLimit, matching the existing truncation behavior in updateDrainBlockedCondition. The full sorted pod list is logged at info level when truncation occurs. Addresses CodeRabbit finding on PR #582. Signed-off-by: bharqav --- .../internal/controller/skyhook_controller.go | 8 ++- .../controller/skyhook_controller_test.go | 71 +++++++++++++++++++ 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index e27aad09b..9a1905f94 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -3382,12 +3382,18 @@ func (r *SkyhookReconciler) EnsureNodeIsReadyForInterrupt(ctx context.Context, s return false, err } if hasWork { // keep waiting... + displayPods := podNames + if len(podNames) > wrapper.ReadyConditionNodeListLimit { + logger := log.FromContext(ctx) + logger.Info("Event message truncated for non-interrupt pods", "node", skyhookNode.GetNode().Name, "nodewright", skyhookNode.GetSkyhook().Name, "pods", podNames) + displayPods = podNames[:wrapper.ReadyConditionNodeListLimit] + } // Condition and NodeWright-level event are managed once per reconcile pass by // updateDrainBlockedCondition. Here we emit a supplementary event on the Node itself // (matching DrainTimeout behavior) so node inspection reflects why drain is waiting. r.recorder.Eventf(skyhookNode.GetNode(), nil, corev1.EventTypeWarning, EventsReasonSkyhookDrain, wrapper.SkyhookReasonNonInterruptPodsRunning, "drain blocked by non-interrupt pods [%s] for package [%s:%s] from [nodewright:%s]", - strings.Join(podNames, ", "), + strings.Join(displayPods, ", "), _package.Name, _package.Version, skyhookNode.GetSkyhook().Name, diff --git a/operator/internal/controller/skyhook_controller_test.go b/operator/internal/controller/skyhook_controller_test.go index d740656be..268951ae5 100644 --- a/operator/internal/controller/skyhook_controller_test.go +++ b/operator/internal/controller/skyhook_controller_test.go @@ -22,6 +22,7 @@ import ( "context" "encoding/json" "fmt" + "strings" "testing" "time" @@ -865,6 +866,76 @@ var _ = Describe("skyhook controller tests", func() { Expect(ready).To(BeTrue(), "node should be ready once all pods are drained") }) + It("truncates pod list in node warning event when matching pods exceed ReadyConditionNodeListLimit", func() { + numPods := wrapper.ReadyConditionNodeListLimit + 5 + var pods []client.Object + var expectedPodNames []string + for i := 1; i <= numPods; i++ { + name := fmt.Sprintf("golden-%02d", i) + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + Labels: map[string]string{ + "workload": "golden", + }, + }, + Spec: corev1.PodSpec{ + NodeName: "node-a", + Containers: []corev1.Container{ + {Name: "golden", Image: "busybox"}, + }, + }, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + } + pods = append(pods, pod) + if i <= wrapper.ReadyConditionNodeListLimit { + expectedPodNames = append(expectedPodNames, fmt.Sprintf("default/%s", name)) + } + } + + testClient := fakeDrainClient(pods...) + recorder := events.NewFakeRecorder(10) + r, err := NewSkyhookReconciler(testClient.Scheme(), testClient, testClient, k8sfake.NewClientset(), recorder, opts) + Expect(err).ToNot(HaveOccurred()) + + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "node-a", + Annotations: map[string]string{ + fmt.Sprintf("%s/cordon_%s", v1alpha1.METADATA_PREFIX, "drain-golden"): "true", + }, + }, + Spec: corev1.NodeSpec{Unschedulable: true}, + } + skyhook := &v1alpha1.NodeWright{ + ObjectMeta: metav1.ObjectMeta{Name: "drain-golden"}, + Spec: v1alpha1.NodeWrightSpec{ + PodNonInterruptLabels: metav1.LabelSelector{ + MatchLabels: map[string]string{ + "workload": "golden", + }, + }, + DrainConfig: &v1alpha1.DrainConfig{ + DisableEviction: ptr(true), + }, + Packages: v1alpha1.Packages{}, + }, + } + skyhookNode, err := wrapper.NewSkyhookNode(node, skyhook) + Expect(err).ToNot(HaveOccurred()) + + ready, err := r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, &v1alpha1.Package{ + PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(ready).To(BeFalse()) + + expectedSubstring := fmt.Sprintf("Warning Drain drain blocked by non-interrupt pods [%s] for package [pkg:1.0.0] from [nodewright:drain-golden]", + strings.Join(expectedPodNames, ", ")) + Eventually(recorder.Events).Should(Receive(ContainSubstring(expectedSubstring))) + }) + It("suppresses drain warning events and preserves condition while DependencyUninstalled is active, then emits exactly once when cleared", func() { goldenPod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ From f3beb19d3d2f73833a24e26dc5ffe46be41ca29d Mon Sep 17 00:00:00 2001 From: bharqav Date: Tue, 15 Sep 2026 15:28:46 +0530 Subject: [PATCH 6/7] fix(operator): log discarded RunNext error in nodeNeedsInterruptDrain nodeNeedsInterruptDrain previously treated a RunNext() error identically to an empty task list, silently returning false with no trace. This could cause a node with a real dependency-graph error to be silently excluded from the NonInterruptPodsRunning aggregate, in reconcile paths (paused, disabled, node-picker skip) that never reach RunSkyhookPackages' own error handling for the same call in the same pass. Now logs the error via log.FromContext before returning false, matching this file's existing logging conventions. Addresses CodeRabbit finding on PR #582. Signed-off-by: bharqav --- operator/internal/controller/skyhook_controller.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index 825544bde..c7174dd08 100644 --- a/operator/internal/controller/skyhook_controller.go +++ b/operator/internal/controller/skyhook_controller.go @@ -620,12 +620,17 @@ func (r *SkyhookReconciler) refreshSkyhookConditions(ctx context.Context, cluste // nodeNeedsInterruptDrain reports whether the node has a runnable package with an interrupt // that is currently at the pre-drain apply or uninstall stage, matching ProcessInterrupt's entry gate. -func nodeNeedsInterruptDrain(node wrapper.SkyhookNode) bool { +func nodeNeedsInterruptDrain(ctx context.Context, node wrapper.SkyhookNode) bool { if node.IsComplete() { return false } toRun, err := node.RunNext() - if err != nil || len(toRun) == 0 { + if err != nil { + logger := log.FromContext(ctx) + logger.Error(err, "error getting next packages to run", "node", node.GetNode().Name, "nodewright", node.GetSkyhook().Name) + return false + } + if len(toRun) == 0 { return false } for _, pkg := range toRun { @@ -667,7 +672,7 @@ func (r *SkyhookReconciler) updateDrainBlockedCondition(ctx context.Context, sky var blockedNodes []string for _, node := range skyhook.GetNodes() { - if !nodeNeedsInterruptDrain(node) { + if !nodeNeedsInterruptDrain(ctx, node) { continue } From eb774a168d491735e9dea5ea2069a116c13e41f2 Mon Sep 17 00:00:00 2001 From: bharqav Date: Thu, 17 Sep 2026 06:28:12 +0530 Subject: [PATCH 7/7] docs(operator): fix contradictory timeout guidance for podNonInterruptLabels barrier custom-resource.md incorrectly suggested pairing this barrier with drainConfig.timeout to bound the wait. The barrier is unbounded by design: EnsureNodeIsReadyForInterrupt returns before DrainNode is ever reached while non-interrupt pods are running, so drainStart_ (written only inside StartDrain, called only from DrainNode) is never set and no timeout clock exists. Corrected to match the existing, accurate wording already used in docs/architecture/interrupt-flow.md and the CRD field description. Addresses review feedback on PR #582. Signed-off-by: bharqav --- docs/user-guide/custom-resource.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/user-guide/custom-resource.md b/docs/user-guide/custom-resource.md index 583721e11..db998aa53 100644 --- a/docs/user-guide/custom-resource.md +++ b/docs/user-guide/custom-resource.md @@ -319,7 +319,7 @@ spec: This is a barrier that runs *before* the configurable drain — it is not a drain exclusion. The operator will wait indefinitely for a matching pod that never -finishes, so pair it with `drainConfig.timeout` if you need a bound. +finishes. This wait is unbounded and drain timeout does not apply. **Note the asymmetry with `nodeSelectors`:** an empty `podNonInterruptLabels` is special-cased to mean *no pods are protected*, not *all of them*. The two fields