diff --git a/chart/templates/nodewright-crd.yaml b/chart/templates/nodewright-crd.yaml index 4a495e45..c695eb33 100644 --- a/chart/templates/nodewright-crd.yaml +++ b/chart/templates/nodewright-crd.yaml @@ -588,8 +588,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 2810b4ae..b76fb6e2 100644 --- a/chart/templates/skyhook-crd.yaml +++ b/chart/templates/skyhook-crd.yaml @@ -569,8 +569,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 a4b95c22..4be725da 100644 --- a/docs/architecture/interrupt-flow.md +++ b/docs/architecture/interrupt-flow.md @@ -163,7 +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. +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 d5ab821f..a12926eb 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/docs/user-guide/custom-resource.md b/docs/user-guide/custom-resource.md index 583721e1..db998aa5 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 diff --git a/operator/RELEASE_NOTES.md b/operator/RELEASE_NOTES.md index 7f98c0e3..3e9752fd 100644 --- a/operator/RELEASE_NOTES.md +++ b/operator/RELEASE_NOTES.md @@ -7,6 +7,27 @@ 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 nodes with packages + requiring interrupt are held at this barrier: + - A `Blocked` condition (status `True`, reason `NonInterruptPodsRunning`) is set on the + 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 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 NodeWright Warning event are deferred and will only appear + once that other condition clears. Once all matching non-interrupt pods finish or + 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 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 bbdcf054..0e8f8abc 100644 --- a/operator/api/nodewright/v1alpha1/nodewright_types.go +++ b/operator/api/nodewright/v1alpha1/nodewright_types.go @@ -45,7 +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 + // 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 20d86571..7866181e 100644 --- a/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml +++ b/operator/config/crd/bases/nodewright.nvidia.com_nodewrights.yaml @@ -593,8 +593,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/operator/internal/controller/cluster_state_v2.go b/operator/internal/controller/cluster_state_v2.go index 29d59411..6104f828 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,12 @@ func (s *skyhookNodes) UpdateBlockedCondition() error { Message: strings.Join(blockedMsgs, "; "), }) } else { - wrapper.RemoveSkyhookConditionTypes(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) + } } return nil diff --git a/operator/internal/controller/skyhook_controller.go b/operator/internal/controller/skyhook_controller.go index c2469d8e..c7174dd0 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,123 @@ 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(ctx context.Context, node wrapper.SkyhookNode) bool { + if node.IsComplete() { + return false + } + toRun, err := node.RunNext() + 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 { + 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(ctx, 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) { @@ -2410,16 +2522,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, @@ -2429,21 +2541,31 @@ 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 + podName := pod.Name + if pod.Namespace != "" { + podName = fmt.Sprintf("%s/%s", pod.Namespace, pod.Name) + } + podNames = append(podNames, podName) } } - 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) { @@ -3258,11 +3380,27 @@ 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... + 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(displayPods, ", "), + _package.Name, + _package.Version, + skyhookNode.GetSkyhook().Name, + ) return false, nil } diff --git a/operator/internal/controller/skyhook_controller_test.go b/operator/internal/controller/skyhook_controller_test.go index 998a672f..9745d9b2 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" @@ -36,6 +37,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" @@ -801,7 +803,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 +847,196 @@ var _ = Describe("skyhook controller tests", func() { drainStartedAt, err := skyhookNode.DrainStartedAt() Expect(err).ToNot(HaveOccurred()) Expect(drainStartedAt).To(BeNil()) + + 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()) + + 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(), "evictable pod is still being drained") + Expect(deleteCalled).To(BeTrue(), "DrainNode should delete evictable pod") + + ready, err = r.EnsureNodeIsReadyForInterrupt(ctx, skyhookNode, &v1alpha1.Package{ + PackageRef: v1alpha1.PackageRef{Name: "pkg", Version: "1.0.0"}, + }) + Expect(err).ToNot(HaveOccurred()) + 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{ + 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", + Labels: map[string]string{"drain-test": "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", + }, + }, + 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{*node}}, + &v1alpha1.DeploymentPolicyList{}, + ) + Expect(err).ToNot(HaveOccurred()) + sn := clusterState.skyhooks[0] + + // Pre-seed unrelated Blocked 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: sn.GetSkyhook().Generation, + LastTransitionTime: metav1.Now(), + }) + + // (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) 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(sn.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("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)"))) + + // 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()) }) It("should mark the node erroring when drain timeout expires", func() { @@ -5059,3 +5252,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 8ce86c50..9509edc8 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" @@ -193,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 } @@ -221,6 +220,26 @@ func RemoveSkyhookConditionTypes(skyhook *Skyhook, conditionTypes ...string) boo return changed } +func RemoveSkyhookConditionTypes(skyhook *Skyhook, conditionTypes ...string) bool { + remove := make(map[string]struct{}, len(conditionTypes)) + for _, conditionType := range conditionTypes { + remove[conditionType] = struct{}{} + } + + return removeSkyhookConditions(skyhook, func(condition metav1.Condition) bool { + _, ok := remove[condition.Type] + return ok + }) +} + +// 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 { for _, condition := range skyhook.Status.Conditions { for _, conditionType := range conditionTypes { @@ -263,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, @@ -278,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, ", ") @@ -293,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 c55b58eb..7857bbd9 100644 --- a/operator/internal/wrapper/skyhook_conditions_test.go +++ b/operator/internal/wrapper/skyhook_conditions_test.go @@ -194,6 +194,42 @@ 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), + ) + DescribeTable("hasTrueSkyhookCondition", func(conditions []metav1.Condition, conditionTypes []string, expected bool) { skyhook := &Skyhook{ NodeWright: &v1alpha1.NodeWright{ @@ -237,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)")) }) })