diff --git a/internal/controller/node_controller.go b/internal/controller/node_controller.go index c6692af9..0bfeb357 100644 --- a/internal/controller/node_controller.go +++ b/internal/controller/node_controller.go @@ -31,7 +31,9 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" readinessv1alpha1 "sigs.k8s.io/node-readiness-controller/api/v1alpha1" "sigs.k8s.io/node-readiness-controller/internal/metrics" @@ -46,6 +48,12 @@ type NodeReconciler struct { } // SetupWithManager sets up the controller with the Manager. +// +// The node controller serves as the sole writer for taint mutations. It is +// triggered by two event sources: +// +// 1. Node changes (conditions / taints / labels). +// 2. NodeReadinessRule changes. func (r *NodeReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error { concurrency := max(r.MaxConcurrentReconciles, 1) return ctrl.NewControllerManagedBy(mgr). @@ -84,9 +92,51 @@ func (r *NodeReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) return shouldReconcile }, })). + Watches( + &readinessv1alpha1.NodeReadinessRule{}, + handler.EnqueueRequestsFromMapFunc(r.ruleToNodeRequests), + builder.WithPredicates(predicate.GenerationChangedPredicate{}), + ). Complete(r) } +// ruleToNodeRequests maps a NodeReadinessRule event to reconcile requests for +// every Node that matches the rule's nodeSelector. +func (r *NodeReconciler) ruleToNodeRequests(ctx context.Context, obj client.Object) []reconcile.Request { + log := ctrl.LoggerFrom(ctx) + + rule, ok := obj.(*readinessv1alpha1.NodeReadinessRule) + if !ok { + return nil + } + + if !rule.DeletionTimestamp.IsZero() { + return nil + } + + selector, err := metav1.LabelSelectorAsSelector(&rule.Spec.NodeSelector) + if err != nil { + log.Error(err, "invalid nodeSelector on rule, skipping node fan-out", "rule", rule.Name) + return nil + } + + nodeList := &corev1.NodeList{} + if err := r.List(ctx, nodeList, client.MatchingLabelsSelector{Selector: selector}); err != nil { + log.Error(err, "failed to list matching nodes for rule fan-out", "rule", rule.Name) + return nil + } + + requests := make([]reconcile.Request, len(nodeList.Items)) + for i, node := range nodeList.Items { + requests[i] = reconcile.Request{ + NamespacedName: types.NamespacedName{Name: node.Name}, + } + } + + log.V(4).Info("Enqueuing node reconciles for rule change", "rule", rule.Name, "matchingNodes", len(requests)) + return requests +} + // +kubebuilder:rbac:groups=core,resources=nodes,verbs=get;list;watch;update;patch // +kubebuilder:rbac:groups=core,resources=nodes/status,verbs=get @@ -152,12 +202,13 @@ func (r *RuleReadinessController) processNodeAgainstAllRules(ctx context.Context "rule", rule.Name, "ruleResourceVersion", rule.ResourceVersion) - if err := r.evaluateRuleForNode(ctx, rule, node); err != nil { - log.Error(err, "Failed to evaluate rule for node", + evalErr := r.evaluateRuleForNode(ctx, rule, node) + if evalErr != nil { + log.Error(evalErr, "Failed to evaluate rule for node", "node", node.Name, "rule", rule.Name) // Continue with other rules even if one fails - r.recordNodeFailure(rule, node.Name, "EvaluationError", err.Error()) - errs = append(errs, err) + r.recordNodeFailure(rule, node.Name, "EvaluationError", evalErr.Error()) + errs = append(errs, evalErr) metrics.Failures.WithLabelValues(rule.Name, "EvaluationError").Inc() } @@ -167,9 +218,11 @@ func (r *RuleReadinessController) processNodeAgainstAllRules(ctx context.Context "rule", rule.Name, "resourceVersion", rule.ResourceVersion) + evalSucceeded := evalErr == nil + var successfullyPatchedRule *readinessv1alpha1.NodeReadinessRule - err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + patchErr := retry.RetryOnConflict(retry.DefaultRetry, func() error { latestRule := &readinessv1alpha1.NodeReadinessRule{} if err := r.Get(ctx, client.ObjectKey{Name: rule.Name}, latestRule); err != nil { return err @@ -201,6 +254,24 @@ func (r *RuleReadinessController) processNodeAgainstAllRules(ctx context.Context ) } + // Maintain AppliedNodes: add the node on success, remove on failure. + var newApplied []string + alreadyPresent := false + for _, n := range latestRule.Status.AppliedNodes { + if n == node.Name { + alreadyPresent = true + if evalSucceeded { + newApplied = append(newApplied, n) + } + } else { + newApplied = append(newApplied, n) + } + } + if evalSucceeded && !alreadyPresent { + newApplied = append(newApplied, node.Name) + } + latestRule.Status.AppliedNodes = newApplied + // handle status.FailedNodes for this node var updatedFailedNodes []readinessv1alpha1.NodeFailure for _, failure := range latestRule.Status.FailedNodes { @@ -223,13 +294,13 @@ func (r *RuleReadinessController) processNodeAgainstAllRules(ctx context.Context return nil }) - if err != nil { - log.Error(err, "Failed to update rule status after node evaluation", + if patchErr != nil { + log.Error(patchErr, "Failed to update rule status after node evaluation", "node", node.Name, "rule", rule.Name, "resourceVersion", rule.ResourceVersion) // continue with other rules - errs = append(errs, err) + errs = append(errs, patchErr) } else { log.V(4).Info("Successfully persisted rule status from node reconciler", "node", node.Name, diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index f51d7187..35251d47 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -125,8 +125,15 @@ func (r *RuleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. return ctrl.Result{RequeueAfter: time.Second}, nil } + // Build the label selector once so all downstream callers (dry-run, cleanup, + // deletion) share the same filtered list. + selector, err := metav1.LabelSelectorAsSelector(&rule.Spec.NodeSelector) + if err != nil { + return ctrl.Result{}, fmt.Errorf("invalid nodeSelector on rule %s: %w", rule.Name, err) + } + nodeList := &corev1.NodeList{} - if err := r.List(ctx, nodeList); err != nil { + if err := r.List(ctx, nodeList, client.MatchingLabelsSelector{Selector: selector}); err != nil { return ctrl.Result{}, err } @@ -138,25 +145,22 @@ func (r *RuleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. // Update rule cache (after cleanup) r.Controller.updateRuleCache(ctx, rule) - // Handle dry run + // Handle dry run: simulate the evaluation inline because dry-run never + // mutates nodes and its results belong on the rule status, not on a node. if rule.Spec.DryRun { if err := r.Controller.processDryRun(ctx, rule, nodeList); err != nil { log.Error(err, "Failed to process dry run", "rule", rule.Name) return ctrl.Result{RequeueAfter: time.Minute}, err } } else { - // Clear previous dry run results + // Not a dry-run: the node fan-out via NodeReconciler handles all taint + // mutations. The rule reconciler only needs to advance ObservedGeneration + // and clear any stale dry-run results so the status stays consistent. + rule.Status.ObservedGeneration = rule.Generation rule.Status.DryRunResults = readinessv1alpha1.DryRunResults{} - - // Process all applicable nodes for this rule - if err := r.Controller.processAllNodesForRule(ctx, rule, nodeList); err != nil { - log.Error(err, "Failed to process nodes for rule", "rule", rule.Name) - return ctrl.Result{RequeueAfter: time.Minute}, err - } } - // Update rule status - if err := r.Controller.updateRuleStatus(ctx, rule); err != nil { + if err := r.Controller.updateRuleObservedStatus(ctx, rule); err != nil { log.Error(err, "Failed to update rule status", "rule", rule.Name) return ctrl.Result{RequeueAfter: time.Minute}, err } @@ -557,14 +561,11 @@ func (r *RuleReadinessController) removeRuleFromCache(ctx context.Context, ruleN log.Info("Removed rule from cache", "rule", ruleName, "totalRules", len(r.ruleCache)) } -// updateRuleStatus updates the status of a NodeReadinessRule. -func (r *RuleReadinessController) updateRuleStatus(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule) error { +// updateRuleObservedStatus persists only the rule-level fields that the rule +// reconciler owns: ObservedGeneration and DryRunResults. +func (r *RuleReadinessController) updateRuleObservedStatus(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule) error { log := ctrl.LoggerFrom(ctx) - - log.V(1).Info("Updating rule status", - "rule", rule.Name, - "nodeEvaluations", len(rule.Status.NodeEvaluations), - "appliedNodes", len(rule.Status.AppliedNodes)) + log.V(1).Info("Updating rule observed status", "rule", rule.Name, "observedGeneration", rule.Status.ObservedGeneration) return retry.RetryOnConflict(retry.DefaultRetry, func() error { latestRule := &readinessv1alpha1.NodeReadinessRule{} @@ -573,21 +574,15 @@ func (r *RuleReadinessController) updateRuleStatus(ctx context.Context, rule *re } patch := client.MergeFrom(latestRule.DeepCopy()) - - latestRule.Status.NodeEvaluations = rule.Status.NodeEvaluations - latestRule.Status.AppliedNodes = rule.Status.AppliedNodes - latestRule.Status.FailedNodes = rule.Status.FailedNodes latestRule.Status.ObservedGeneration = rule.Status.ObservedGeneration latestRule.Status.DryRunResults = rule.Status.DryRunResults if err := r.Status().Patch(ctx, latestRule, patch); err != nil { - log.V(1).Info("Status patch conflict, will retry", - "rule", rule.Name, - "error", err.Error()) + log.V(1).Info("Status patch conflict, will retry", "rule", rule.Name, "error", err.Error()) return err } - log.V(1).Info("Successfully patched rule status", "rule", rule.Name) + log.V(1).Info("Successfully patched rule observed status", "rule", rule.Name) return nil }) } @@ -600,10 +595,6 @@ func (r *RuleReadinessController) processDryRun(ctx context.Context, rule *readi var summaryParts []string for _, node := range nodeList.Items { - if !r.ruleAppliesTo(ctx, rule, &node) { - continue - } - affectedNodes++ // Simulate rule evaluation @@ -672,10 +663,6 @@ func (r *RuleReadinessController) cleanupTaintsForRule(ctx context.Context, rule var errors []string for _, node := range nodeList.Items { - if !r.ruleAppliesTo(ctx, rule, &node) { - continue - } - // Check if node has the taint managed by this rule if r.hasTaintBySpec(&node, rule.Spec.Taint) { log.Info("Removing taint from node during rule cleanup", diff --git a/internal/controller/nodereadinessrule_controller_test.go b/internal/controller/nodereadinessrule_controller_test.go index b551ebb3..7c81213d 100644 --- a/internal/controller/nodereadinessrule_controller_test.go +++ b/internal/controller/nodereadinessrule_controller_test.go @@ -289,6 +289,12 @@ var _ = Describe("NodeReadinessRule Controller", func() { }) Expect(err).NotTo(HaveOccurred()) + // Rule reconciler updated the cache; now NodeReconciler owns taint work. + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: "immediate-test-node"}, + }) + Expect(err).NotTo(HaveOccurred()) + // Verify the node gets tainted immediately due to unmet condition Eventually(func() bool { updatedNode := &corev1.Node{} @@ -925,6 +931,12 @@ var _ = Describe("NodeReadinessRule Controller", func() { }) Expect(err).NotTo(HaveOccurred()) + // Rule reconciler updated the cache; NodeReconciler removes the taint. + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: nodeName}, + }) + Expect(err).NotTo(HaveOccurred()) + // The pre-existing taint should be removed because the absent condition // is satisfied via defaultStatus:False. Eventually(func() bool { @@ -1356,6 +1368,10 @@ var _ = Describe("NodeReadinessRule Controller", func() { _, err := ruleReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "db-rule"}}) Expect(err).NotTo(HaveOccurred()) + // Rule reconciler updated the cache; NodeReconciler owns taint work. + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "node1"}}) + Expect(err).NotTo(HaveOccurred()) + // Verify that the taint has been added to the node Eventually(func() bool { updatedNode := &corev1.Node{} @@ -1495,6 +1511,12 @@ var _ = Describe("NodeReadinessRule Controller", func() { }) Expect(err).NotTo(HaveOccurred()) + // Rule reconciler updated the cache; NodeReconciler owns taint work. + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: "new-node"}, + }) + Expect(err).NotTo(HaveOccurred()) + // Verify that the rule's status is updated to include the new node Eventually(func() []string { updatedRule := &nodereadinessiov1alpha1.NodeReadinessRule{} @@ -1658,10 +1680,16 @@ var _ = Describe("NodeReadinessRule Controller", func() { }) It("should remove the node from the rule's status", func() { - // Initial reconcile to populate status + // Initial reconcile to populate cache, then drive node reconciles to + // seed NodeEvaluations for both nodes. _, err := ruleReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "delete-node-rule"}}) Expect(err).NotTo(HaveOccurred()) + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "node1"}}) + Expect(err).NotTo(HaveOccurred()) + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "node2"}}) + Expect(err).NotTo(HaveOccurred()) + Eventually(func() int { updatedRule := &nodereadinessiov1alpha1.NodeReadinessRule{} _ = k8sClient.Get(ctx, types.NamespacedName{Name: "delete-node-rule"}, updatedRule) @@ -2083,6 +2111,12 @@ var _ = Describe("NodeReadinessRule Controller", func() { }) Expect(err).NotTo(HaveOccurred()) + // Rule reconciler updated the cache; drive NodeReconciler for matching nodes. + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "applied-node-1"}}) + Expect(err).NotTo(HaveOccurred()) + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "applied-node-2"}}) + Expect(err).NotTo(HaveOccurred()) + By("Verifying AppliedNodes contains only matching nodes") Eventually(func() []string { updatedRule := &nodereadinessiov1alpha1.NodeReadinessRule{} @@ -2102,6 +2136,12 @@ var _ = Describe("NodeReadinessRule Controller", func() { }) Expect(err).NotTo(HaveOccurred()) + // Rule reconciler updated the cache; drive NodeReconciler for matching nodes. + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "applied-node-1"}}) + Expect(err).NotTo(HaveOccurred()) + _, err = nodeReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: "applied-node-2"}}) + Expect(err).NotTo(HaveOccurred()) + By("Verifying NodeEvaluations exist for all AppliedNodes") Eventually(func() bool { updatedRule := &nodereadinessiov1alpha1.NodeReadinessRule{}