Skip to content

Changes to handle Node taint mutation only through the node controller #389

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Karthik-K-N wants to merge 1 commit into
base: main
Choose a base branch
from
+141 −43
Draft

Changes to handle Node taint mutation only through the node controller #389

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 79 additions & 8 deletions internal/controller/node_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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).
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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()
}

Expand All @@ -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
Expand Down Expand Up @@ -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 {
Expand All @@ -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,
Expand Down
55 changes: 21 additions & 34 deletions internal/controller/nodereadinessrule_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand All @@ -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
}
Expand Down Expand Up @@ -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{}
Expand All @@ -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
})
}
Expand All @@ -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
Expand Down Expand Up @@ -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",
Expand Down
42 changes: 41 additions & 1 deletion internal/controller/nodereadinessrule_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{}
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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{}
Expand Down Expand Up @@ -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{}
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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{}
Expand All @@ -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{}
Expand Down