From b6eab974eafdac3e092ce98a332f3ad12a996f84 Mon Sep 17 00:00:00 2001 From: bhuvan-somisetty Date: Mon, 17 Aug 2026 15:56:54 +0530 Subject: [PATCH 1/3] fix(controller): use optimistic locking on rule status patches RuleReconciler and NodeReconciler both patch NodeReadinessRule.Status concurrently, but every status/finalizer patch used a plain client.MergeFrom with no resourceVersion precondition, wrapped in retry.RetryOnConflict. Since a JSON merge patch never carries that precondition unless MergeFromWithOptimisticLock is used, the API server never returns a conflict and the retry wrapper never actually retries. Worse, updateRuleStatus replaced NodeEvaluations/FailedNodes wholesale from a snapshot computed at the start of a RuleReconciler sweep, so it could silently discard a concurrent NodeReconciler per-node update for a node outside that sweep. Fix this by having processAllNodesForRule return a delta of exactly the per-node changes it made, and merging that delta by node name instead of overwriting the whole slice. Signed-off-by: bhuvan-somisetty --- internal/controller/helper.go | 55 +++++ internal/controller/node_controller.go | 19 +- .../nodereadinessrule_controller.go | 190 ++++++++++++------ .../nodereadinessrule_controller_test.go | 157 ++++++++++++++- 4 files changed, 346 insertions(+), 75 deletions(-) diff --git a/internal/controller/helper.go b/internal/controller/helper.go index 959741a3..f55e9336 100644 --- a/internal/controller/helper.go +++ b/internal/controller/helper.go @@ -19,9 +19,11 @@ package controller import ( "encoding/json" "maps" + "sort" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/types" + readinessv1alpha1 "sigs.k8s.io/node-readiness-controller/api/v1alpha1" ) @@ -133,3 +135,56 @@ func filterStatusForExistingNodes( func labelsEqual(a, b map[string]string) bool { return maps.Equal(a, b) } + +// nodeStatusDelta captures the per-node NodeEvaluation/NodeFailure changes a single +// processAllNodesForRule sweep actually produced, keyed by node name. It intentionally does not +// carry AppliedNodes/ObservedGeneration/DryRunResults: those fields only ever have one writer +// (RuleReconciler), so a plain overwrite of them is safe. +// +// A nil value in failures means "clear any failure recorded for this node" (the node evaluated +// successfully). evaluations only ever holds entries for nodes that were freshly (re-)evaluated +// this sweep; a failed evaluation leaves the node's prior NodeEvaluation untouched. +type nodeStatusDelta struct { + evaluations map[string]readinessv1alpha1.NodeEvaluation + failures map[string]*readinessv1alpha1.NodeFailure +} + +// applyNodeStatusDelta merges delta into rule's NodeEvaluations/FailedNodes, replacing only the +// entries for nodes present in delta and leaving every other node's entry untouched. +// +// This is the crux of fixing the lost-update bug described in #341: a naive full-slice +// replacement of NodeEvaluations/FailedNodes (computed from a nodeList snapshot taken at the +// start of a RuleReconciler sweep) would silently discard any per-node status update written +// concurrently by NodeReconciler for a node this particular sweep didn't touch. Merging by node +// name instead means each writer only ever overwrites the entries it just recomputed. +func applyNodeStatusDelta(rule *readinessv1alpha1.NodeReadinessRule, delta nodeStatusDelta) { + if len(delta.evaluations) > 0 { + merged := make([]readinessv1alpha1.NodeEvaluation, 0, len(rule.Status.NodeEvaluations)+len(delta.evaluations)) + for _, eval := range rule.Status.NodeEvaluations { + if _, changed := delta.evaluations[eval.NodeName]; !changed { + merged = append(merged, eval) + } + } + for _, eval := range delta.evaluations { + merged = append(merged, eval) + } + sort.Slice(merged, func(i, j int) bool { return merged[i].NodeName < merged[j].NodeName }) + rule.Status.NodeEvaluations = merged + } + + if len(delta.failures) > 0 { + merged := make([]readinessv1alpha1.NodeFailure, 0, len(rule.Status.FailedNodes)+len(delta.failures)) + for _, failure := range rule.Status.FailedNodes { + if _, changed := delta.failures[failure.NodeName]; !changed { + merged = append(merged, failure) + } + } + for _, failure := range delta.failures { + if failure != nil { + merged = append(merged, *failure) + } + } + sort.Slice(merged, func(i, j int) bool { return merged[i].NodeName < merged[j].NodeName }) + rule.Status.FailedNodes = merged + } +} diff --git a/internal/controller/node_controller.go b/internal/controller/node_controller.go index 2d1b1c12..0d23fedb 100644 --- a/internal/controller/node_controller.go +++ b/internal/controller/node_controller.go @@ -169,14 +169,7 @@ func (r *RuleReadinessController) processNodeAgainstAllRules(ctx context.Context var successfullyPatchedRule *readinessv1alpha1.NodeReadinessRule - err := retry.RetryOnConflict(retry.DefaultRetry, func() error { - latestRule := &readinessv1alpha1.NodeReadinessRule{} - if err := r.Get(ctx, client.ObjectKey{Name: rule.Name}, latestRule); err != nil { - return err - } - - patch := client.MergeFrom(latestRule.DeepCopy()) - + err := r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, func(latestRule *readinessv1alpha1.NodeReadinessRule) bool { // update only this specific node evaluation status currEval := readinessv1alpha1.NodeEvaluation{} for _, eval := range rule.Status.NodeEvaluations { @@ -215,12 +208,8 @@ func (r *RuleReadinessController) processNodeAgainstAllRules(ctx context.Context } latestRule.Status.FailedNodes = updatedFailedNodes - if err := r.Status().Patch(ctx, latestRule, patch); err != nil { - return err - } - successfullyPatchedRule = latestRule - return nil + return true }) if err != nil { @@ -464,7 +453,7 @@ func (r *RuleReadinessController) markBootstrapCompleted(ctx context.Context, no } deferred = false - patch := client.MergeFromWithOptions(node.DeepCopy(), client.MergeFromWithOptimisticLock{}) + stored := node.DeepCopy() // Initialize annotations map if nil. if node.Annotations == nil { @@ -472,7 +461,7 @@ func (r *RuleReadinessController) markBootstrapCompleted(ctx context.Context, no } node.Annotations[annotationKey] = bootstrapAnnotationValue(rule.Name) - if err := r.Patch(ctx, node, patch); err != nil { + if err := r.Patch(ctx, node, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { return err } diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index 3b79df67..d66193cb 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -139,6 +139,7 @@ func (r *RuleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. r.Controller.updateRuleCache(ctx, rule) // Handle dry run + var delta nodeStatusDelta 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) @@ -149,14 +150,16 @@ func (r *RuleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. rule.Status.DryRunResults = readinessv1alpha1.DryRunResults{} // Process all applicable nodes for this rule - if err := r.Controller.processAllNodesForRule(ctx, rule, nodeList); err != nil { + var err error + delta, err = r.Controller.processAllNodesForRule(ctx, rule, nodeList) + if 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.updateRuleStatus(ctx, rule, delta); err != nil { log.Error(err, "Failed to update rule status", "rule", rule.Name) return ctrl.Result{RequeueAfter: time.Minute}, err } @@ -198,9 +201,19 @@ func (r *RuleReconciler) reconcileDelete(ctx context.Context, rule *readinessv1a r.Controller.removeRuleFromCache(ctx, rule.Name) log.V(3).Info("Removing the finalizer from the rule") - patch := client.MergeFrom(rule.DeepCopy()) - controllerutil.RemoveFinalizer(rule, finalizerName) - err := r.Patch(ctx, rule, patch) + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + latest := &readinessv1alpha1.NodeReadinessRule{} + if err := r.Get(ctx, client.ObjectKey{Name: rule.Name}, latest); err != nil { + return client.IgnoreNotFound(err) + } + if !controllerutil.ContainsFinalizer(latest, finalizerName) { + return nil + } + + stored := latest.DeepCopy() + controllerutil.RemoveFinalizer(latest, finalizerName) + return r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) + }) if err != nil { return ctrl.Result{}, err } @@ -251,13 +264,8 @@ func (r *RuleReadinessController) cleanupDeletedNodes(ctx context.Context, rule "before", len(rule.Status.NodeEvaluations), "after", len(newNodeEvaluations)) - // Use retry on conflict to update status to avoid race conditions from node updates - return retry.RetryOnConflict(retry.DefaultRetry, func() error { - fresh := &readinessv1alpha1.NodeReadinessRule{} - if err := r.Get(ctx, client.ObjectKey{Name: rule.Name}, fresh); err != nil { - return err - } - + // Use an optimistic-locked patch to avoid race conditions from concurrent node updates. + return r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, func(fresh *readinessv1alpha1.NodeReadinessRule) bool { freshNodeEvaluations, freshFailedNodes := filterStatusForExistingNodes( existingNodes, fresh.Status.NodeEvaluations, @@ -266,41 +274,66 @@ func (r *RuleReadinessController) cleanupDeletedNodes(ctx context.Context, rule if len(freshNodeEvaluations) == len(fresh.Status.NodeEvaluations) && len(freshFailedNodes) == len(fresh.Status.FailedNodes) { - return nil + return false } - patch := client.MergeFrom(fresh.DeepCopy()) fresh.Status.NodeEvaluations = freshNodeEvaluations fresh.Status.FailedNodes = freshFailedNodes - return r.Status().Patch(ctx, fresh, patch) + return true }) } -// processAllNodesForRule processes all nodes when a rule changes. +// processAllNodesForRule processes all nodes when a rule changes. It mutates rule.Status in place +// (as before) and additionally returns a nodeStatusDelta describing exactly which nodes' status +// this sweep changed, so updateRuleStatus can merge those changes into the latest stored status +// instead of replacing NodeEvaluations/FailedNodes wholesale. // //nolint:unparam // Keep error return for future extensibility and API stability. -func (r *RuleReadinessController) processAllNodesForRule(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, nodeList *corev1.NodeList) error { +func (r *RuleReadinessController) processAllNodesForRule(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, nodeList *corev1.NodeList) (nodeStatusDelta, error) { log := ctrl.LoggerFrom(ctx) log.Info("Processing all nodes for rule", "rule", rule.Name, "totalNodes", len(nodeList.Items)) + delta := nodeStatusDelta{ + evaluations: make(map[string]readinessv1alpha1.NodeEvaluation), + failures: make(map[string]*readinessv1alpha1.NodeFailure), + } + var appliedNodes []string for _, node := range nodeList.Items { - if r.ruleAppliesTo(ctx, rule, &node) { - log.Info("Processing node for rule", "rule", rule.Name, "node", node.Name) - if err := r.evaluateRuleForNode(ctx, rule, &node); err != nil { - log.Error(err, "Failed to evaluate node for rule", "rule", rule.Name, "node", node.Name) - r.recordNodeFailure(rule, node.Name, "EvaluationError", err.Error()) - metrics.Failures.WithLabelValues(rule.Name, string(metrics.FailureReasonEvaluationError)).Inc() - } else { - appliedNodes = append(appliedNodes, node.Name) - var updatedFailedNodes []readinessv1alpha1.NodeFailure - for _, f := range rule.Status.FailedNodes { - if f.NodeName != node.Name { - updatedFailedNodes = append(updatedFailedNodes, f) - } + if !r.ruleAppliesTo(ctx, rule, &node) { + continue + } + + log.Info("Processing node for rule", "rule", rule.Name, "node", node.Name) + if err := r.evaluateRuleForNode(ctx, rule, &node); err != nil { + log.Error(err, "Failed to evaluate node for rule", "rule", rule.Name, "node", node.Name) + r.recordNodeFailure(rule, node.Name, "EvaluationError", err.Error()) + metrics.Failures.WithLabelValues(rule.Name, string(metrics.FailureReasonEvaluationError)).Inc() + + for _, f := range rule.Status.FailedNodes { + if f.NodeName == node.Name { + failure := f + delta.failures[node.Name] = &failure + break + } + } + } else { + appliedNodes = append(appliedNodes, node.Name) + var updatedFailedNodes []readinessv1alpha1.NodeFailure + for _, f := range rule.Status.FailedNodes { + if f.NodeName != node.Name { + updatedFailedNodes = append(updatedFailedNodes, f) + } + } + rule.Status.FailedNodes = updatedFailedNodes + delta.failures[node.Name] = nil // clear any previously-recorded failure + + for _, eval := range rule.Status.NodeEvaluations { + if eval.NodeName == node.Name { + delta.evaluations[node.Name] = eval + break } - rule.Status.FailedNodes = updatedFailedNodes } } } @@ -314,7 +347,7 @@ func (r *RuleReadinessController) processAllNodesForRule(ctx context.Context, ru } log.Info("Completed processing nodes for rule", "rule", rule.Name, "processedCount", len(appliedNodes)) - return nil + return delta, nil } // evaluateRuleForNode evaluates a single rule against a single node. @@ -570,8 +603,42 @@ 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 { +// patchRuleStatusWithOptimisticLock fetches the latest NodeReadinessRule, lets mutate apply status +// changes to it, and patches the result back with an optimistic-locked JSON merge patch. mutate +// should return false if it made no changes, to skip an unnecessary Patch call. +// +// We use client.MergeFromWithOptimisticLock here for the same reason addTaintBySpec/ +// removeTaintBySpec do (see node_controller.go): a JSON merge patch replaces slice fields +// (NodeEvaluations, AppliedNodes, FailedNodes) wholesale rather than merging them, so without a +// resourceVersion precondition retry.RetryOnConflict can never observe a genuine conflict and a +// concurrent status write from the other reconciler (RuleReconciler and NodeReconciler both patch +// NodeReadinessRule.Status independently) can be silently overwritten. +func (r *RuleReadinessController) patchRuleStatusWithOptimisticLock( + ctx context.Context, + ruleName string, + mutate func(latest *readinessv1alpha1.NodeReadinessRule) (changed bool), +) error { + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + latestRule := &readinessv1alpha1.NodeReadinessRule{} + if err := r.Get(ctx, client.ObjectKey{Name: ruleName}, latestRule); err != nil { + return err + } + + stored := latestRule.DeepCopy() + if !mutate(latestRule) { + return nil + } + + return r.Status().Patch(ctx, latestRule, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) + }) +} + +// updateRuleStatus updates the status of a NodeReadinessRule. delta carries the per-node +// NodeEvaluations/FailedNodes changes processAllNodesForRule actually produced this reconcile; +// it is merged into the latest stored status by node name (see applyNodeStatusDelta) rather than +// replacing those fields wholesale, so a concurrent per-node update from NodeReconciler +// (processNodeAgainstAllRules) for a node outside this sweep isn't silently discarded. +func (r *RuleReadinessController) updateRuleStatus(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, delta nodeStatusDelta) error { log := ctrl.LoggerFrom(ctx) log.V(1).Info("Updating rule status", @@ -579,30 +646,20 @@ func (r *RuleReadinessController) updateRuleStatus(ctx context.Context, rule *re "nodeEvaluations", len(rule.Status.NodeEvaluations), "appliedNodes", len(rule.Status.AppliedNodes)) - return retry.RetryOnConflict(retry.DefaultRetry, func() error { - latestRule := &readinessv1alpha1.NodeReadinessRule{} - if err := r.Get(ctx, client.ObjectKey{Name: rule.Name}, latestRule); err != nil { - return err - } - - patch := client.MergeFrom(latestRule.DeepCopy()) - - latestRule.Status.NodeEvaluations = rule.Status.NodeEvaluations + err := r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, func(latestRule *readinessv1alpha1.NodeReadinessRule) bool { + applyNodeStatusDelta(latestRule, delta) 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()) - return err - } - - log.V(1).Info("Successfully patched rule status", "rule", rule.Name) - return nil + return true }) + if err != nil { + log.V(1).Info("Failed to patch rule status", "rule", rule.Name, "error", err.Error()) + return err + } + + log.V(1).Info("Successfully patched rule status", "rule", rule.Name) + return nil } // processDryRun processes dry run for a rule. @@ -725,13 +782,30 @@ func (r *RuleReconciler) ensureFinalizer(ctx context.Context, rule *readinessv1a return false, nil } - patch := client.MergeFrom(rule.DeepCopy()) - controllerutil.AddFinalizer(rule, finalizer) - err = r.Patch(ctx, rule, patch) + added := false + err = retry.RetryOnConflict(retry.DefaultRetry, func() error { + latest := &readinessv1alpha1.NodeReadinessRule{} + if err := r.Get(ctx, client.ObjectKey{Name: rule.Name}, latest); err != nil { + return err + } + if controllerutil.ContainsFinalizer(latest, finalizer) { + return nil + } + + stored := latest.DeepCopy() + controllerutil.AddFinalizer(latest, finalizer) + if err := r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { + return err + } + + *rule = *latest + added = true + return nil + }) if err != nil { return false, err } - return true, nil + return added, nil } // getPreviousNodeEvaluation retrieves the previous evaluation result for a specific node from the rule status. diff --git a/internal/controller/nodereadinessrule_controller_test.go b/internal/controller/nodereadinessrule_controller_test.go index 07fb4b30..000889b5 100644 --- a/internal/controller/nodereadinessrule_controller_test.go +++ b/internal/controller/nodereadinessrule_controller_test.go @@ -19,6 +19,7 @@ package controller import ( "context" "fmt" + "sync/atomic" "time" . "github.com/onsi/ginkgo/v2" @@ -33,6 +34,8 @@ import ( "k8s.io/client-go/kubernetes/fake" "k8s.io/client-go/tools/events" "sigs.k8s.io/controller-runtime/pkg/client" + fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" "sigs.k8s.io/controller-runtime/pkg/reconcile" nodereadinessiov1alpha1 "sigs.k8s.io/node-readiness-controller/api/v1alpha1" @@ -68,6 +71,15 @@ func (c *errorInjectingClient) Patch(ctx context.Context, obj client.Object, pat return c.Client.Patch(ctx, obj, patch, opts...) } +func findNodeEvaluation(evals []nodereadinessiov1alpha1.NodeEvaluation, nodeName string) *nodereadinessiov1alpha1.NodeEvaluation { + for i := range evals { + if evals[i].NodeName == nodeName { + return &evals[i] + } + } + return nil +} + var _ = Describe("NodeReadinessRule Controller", func() { var ( ctx context.Context @@ -2361,7 +2373,8 @@ var _ = Describe("NodeReadinessRule Controller", func() { } nodeList := &corev1.NodeList{Items: []corev1.Node{*failNode}} - Expect(failController.processAllNodesForRule(ctx, rule, nodeList)).To(Succeed()) + _, err := failController.processAllNodesForRule(ctx, rule, nodeList) + Expect(err).NotTo(HaveOccurred()) Expect(rule.Status.AppliedNodes).NotTo(ContainElement("fail-path-node")) @@ -2418,7 +2431,8 @@ var _ = Describe("NodeReadinessRule Controller", func() { } nodeList := &corev1.NodeList{Items: []corev1.Node{*successNode}} - Expect(successController.processAllNodesForRule(ctx, rule, nodeList)).To(Succeed()) + _, err := successController.processAllNodesForRule(ctx, rule, nodeList) + Expect(err).NotTo(HaveOccurred()) Expect(rule.Status.AppliedNodes).To(ContainElement("stale-recovery-node")) @@ -2535,4 +2549,143 @@ var _ = Describe("NodeReadinessRule Controller", func() { Expect(readinessController.hasTaintBySpec(anyOfNode, rule.Spec.Taint)).To(BeFalse()) }) }) + + Context("concurrent status writes between RuleReconciler and NodeReconciler", func() { + var testScheme *runtime.Scheme + + BeforeEach(func() { + testScheme = runtime.NewScheme() + Expect(corev1.AddToScheme(testScheme)).To(Succeed()) + Expect(nodereadinessiov1alpha1.AddToScheme(testScheme)).To(Succeed()) + }) + + It("should not discard a NodeReconciler-written evaluation for a node outside the RuleReconciler sweep", func() { + nodeA := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "concurrent-node-a", Labels: map[string]string{"role": "worker"}}, + Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: "Ready", Status: corev1.ConditionTrue}}}, + } + nodeB := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "concurrent-node-b", Labels: map[string]string{"role": "worker"}}, + Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: "Ready", Status: corev1.ConditionTrue}}}, + } + + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "concurrent-status-rule"}, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{ + {Type: "Ready", RequiredStatus: corev1.ConditionTrue}, + }, + Taint: corev1.Taint{Key: "readiness.k8s.io/concurrent-test", Effect: corev1.TaintEffectNoSchedule}, + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"role": "worker"}}, + EnforcementMode: nodereadinessiov1alpha1.EnforcementModeContinuous, + }, + } + + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(nodeA, nodeB, rule). + WithStatusSubresource(rule). + Build() + + controller := &RuleReadinessController{ + Client: fc, + Scheme: testScheme, + clientset: fake.NewSimpleClientset(), + ruleCache: make(map[string]*nodereadinessiov1alpha1.NodeReadinessRule), + EventRecorder: events.NewFakeRecorder(10), + } + controller.updateRuleCache(ctx, rule) + + // Simulate RuleReconciler starting a sweep from a nodeList snapshot that + // only knows about node-a (e.g. node-b joined after the snapshot was taken). + staleNodeList := &corev1.NodeList{Items: []corev1.Node{*nodeA}} + delta, err := controller.processAllNodesForRule(ctx, rule, staleNodeList) + Expect(err).NotTo(HaveOccurred()) + + // While that sweep is "in flight", NodeReconciler independently reconciles + // node-b and persists its own evaluation into the same rule's status. + Expect(controller.processNodeAgainstAllRules(ctx, nodeB)).To(Succeed()) + + afterConcurrentWrite := &nodereadinessiov1alpha1.NodeReadinessRule{} + Expect(fc.Get(ctx, types.NamespacedName{Name: rule.Name}, afterConcurrentWrite)).To(Succeed()) + Expect(findNodeEvaluation(afterConcurrentWrite.Status.NodeEvaluations, "concurrent-node-b")). + NotTo(BeNil(), "NodeReconciler should have persisted an evaluation for node-b") + + // RuleReconciler now finishes its (stale) sweep and patches the rule status. + Expect(controller.updateRuleStatus(ctx, rule, delta)).To(Succeed()) + + final := &nodereadinessiov1alpha1.NodeReadinessRule{} + Expect(fc.Get(ctx, types.NamespacedName{Name: rule.Name}, final)).To(Succeed()) + + Expect(findNodeEvaluation(final.Status.NodeEvaluations, "concurrent-node-a")).NotTo(BeNil(), + "node-a's freshly computed evaluation should be present") + Expect(findNodeEvaluation(final.Status.NodeEvaluations, "concurrent-node-b")).NotTo(BeNil(), + "node-b's evaluation, written concurrently by NodeReconciler outside this sweep, must not be discarded") + }) + + It("should retry updateRuleStatus on a genuine conflict instead of silently clobbering it", func() { + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "conflict-node", Labels: map[string]string{"role": "worker"}}, + Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: "Ready", Status: corev1.ConditionTrue}}}, + } + + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "conflict-rule"}, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{ + {Type: "Ready", RequiredStatus: corev1.ConditionTrue}, + }, + Taint: corev1.Taint{Key: "readiness.k8s.io/conflict-test", Effect: corev1.TaintEffectNoSchedule}, + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"role": "worker"}}, + EnforcementMode: nodereadinessiov1alpha1.EnforcementModeContinuous, + }, + } + + var patchCount atomic.Int32 + + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithStatusSubresource(rule). + WithInterceptorFuncs(interceptor.Funcs{ + SubResourcePatch: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if r, ok := obj.(*nodereadinessiov1alpha1.NodeReadinessRule); ok && r.Name == "conflict-rule" && patchCount.Add(1) == 1 { + // Simulate a concurrent NodeReconciler write for a different + // node landing in between our Get and our Patch. + current := &nodereadinessiov1alpha1.NodeReadinessRule{} + Expect(c.Get(ctx, types.NamespacedName{Name: r.Name}, current)).To(Succeed()) + current.Status.NodeEvaluations = append(current.Status.NodeEvaluations, nodereadinessiov1alpha1.NodeEvaluation{ + NodeName: "conflict-other-node", + TaintStatus: nodereadinessiov1alpha1.TaintStatusAbsent, + }) + Expect(c.Status().Update(ctx, current)).To(Succeed()) + } + return c.SubResource(subResourceName).Patch(ctx, obj, patch, opts...) + }, + }). + Build() + + controller := &RuleReadinessController{ + Client: fc, + Scheme: testScheme, + clientset: fake.NewSimpleClientset(), + ruleCache: make(map[string]*nodereadinessiov1alpha1.NodeReadinessRule), + EventRecorder: events.NewFakeRecorder(10), + } + + nodeList := &corev1.NodeList{Items: []corev1.Node{*node}} + delta, err := controller.processAllNodesForRule(ctx, rule, nodeList) + Expect(err).NotTo(HaveOccurred()) + + Expect(controller.updateRuleStatus(ctx, rule, delta)).To(Succeed()) + Expect(patchCount.Load()).To(BeNumerically(">=", 2), + "the first patch attempt should hit a conflict and force a retry") + + final := &nodereadinessiov1alpha1.NodeReadinessRule{} + Expect(fc.Get(ctx, types.NamespacedName{Name: rule.Name}, final)).To(Succeed()) + Expect(findNodeEvaluation(final.Status.NodeEvaluations, "conflict-node")).NotTo(BeNil()) + Expect(findNodeEvaluation(final.Status.NodeEvaluations, "conflict-other-node")).NotTo(BeNil(), + "the concurrently-written evaluation must survive the retried patch") + }) + }) }) From ccec808d1d032093de9eeaf99e5c2c26e6a59930 Mon Sep 17 00:00:00 2001 From: bhuvan-somisetty Date: Sun, 23 Aug 2026 02:10:13 +0530 Subject: [PATCH 2/3] address review: trim doc comments, add conflict metric Tighten the nodeStatusDelta/applyNodeStatusDelta doc comments per review feedback to be self-describing instead of narrating the issue they fix. Add node_readiness_status_patch_conflicts_total, incremented whenever an optimistic-locked patch (rule status, rule finalizer add/remove, node taint add/remove, node bootstrap annotation) hits a 409, so the 409 rate this optimistic locking can introduce is observable instead of only showing up indirectly as reconcile latency. Document why markBootstrapCompleted still needs its optimistic lock even though the patch itself is annotation-only and merges cleanly: the lock guards the hasTaintBySpec check preceding it, not the annotation merge, which the "should not mark bootstrap completed when the rule taints concurrently" regression test already covers. --- internal/controller/helper.go | 18 +++++------------- internal/controller/node_controller.go | 17 ++++++++++++++++- .../controller/nodereadinessrule_controller.go | 15 +++++++++++++-- internal/metrics/metrics.go | 12 ++++++++++++ 4 files changed, 46 insertions(+), 16 deletions(-) diff --git a/internal/controller/helper.go b/internal/controller/helper.go index f55e9336..b83b46ff 100644 --- a/internal/controller/helper.go +++ b/internal/controller/helper.go @@ -136,14 +136,12 @@ func labelsEqual(a, b map[string]string) bool { return maps.Equal(a, b) } -// nodeStatusDelta captures the per-node NodeEvaluation/NodeFailure changes a single -// processAllNodesForRule sweep actually produced, keyed by node name. It intentionally does not -// carry AppliedNodes/ObservedGeneration/DryRunResults: those fields only ever have one writer -// (RuleReconciler), so a plain overwrite of them is safe. +// nodeStatusDelta captures the per-node NodeEvaluation/NodeFailure changes produced by a single +// processAllNodesForRule sweep, keyed by node name. It excludes AppliedNodes, ObservedGeneration, +// and DryRunResults, which have a single writer and are safe to overwrite directly. // -// A nil value in failures means "clear any failure recorded for this node" (the node evaluated -// successfully). evaluations only ever holds entries for nodes that were freshly (re-)evaluated -// this sweep; a failed evaluation leaves the node's prior NodeEvaluation untouched. +// A nil value in failures clears any failure recorded for that node. evaluations only holds +// entries for nodes freshly (re-)evaluated this sweep. type nodeStatusDelta struct { evaluations map[string]readinessv1alpha1.NodeEvaluation failures map[string]*readinessv1alpha1.NodeFailure @@ -151,12 +149,6 @@ type nodeStatusDelta struct { // applyNodeStatusDelta merges delta into rule's NodeEvaluations/FailedNodes, replacing only the // entries for nodes present in delta and leaving every other node's entry untouched. -// -// This is the crux of fixing the lost-update bug described in #341: a naive full-slice -// replacement of NodeEvaluations/FailedNodes (computed from a nodeList snapshot taken at the -// start of a RuleReconciler sweep) would silently discard any per-node status update written -// concurrently by NodeReconciler for a node this particular sweep didn't touch. Merging by node -// name instead means each writer only ever overwrites the entries it just recomputed. func applyNodeStatusDelta(rule *readinessv1alpha1.NodeReadinessRule, delta nodeStatusDelta) { if len(delta.evaluations) > 0 { merged := make([]readinessv1alpha1.NodeEvaluation, 0, len(rule.Status.NodeEvaluations)+len(delta.evaluations)) diff --git a/internal/controller/node_controller.go b/internal/controller/node_controller.go index 0d23fedb..95c696a3 100644 --- a/internal/controller/node_controller.go +++ b/internal/controller/node_controller.go @@ -22,6 +22,7 @@ import ( "fmt" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -298,6 +299,9 @@ func (r *RuleReadinessController) addTaintBySpec(ctx context.Context, node *core stored := latestNode.DeepCopy() latestNode.Spec.Taints = append(latestNode.Spec.Taints, taintSpec) if err := r.Patch(ctx, latestNode, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { + if apierrors.IsConflict(err) { + metrics.StatusPatchConflicts.WithLabelValues("node_taint", "add").Inc() + } return err } @@ -389,6 +393,9 @@ func (r *RuleReadinessController) removeTaint(ctx context.Context, node *corev1. latestNode.Annotations[key] = annotations[key] } if err := r.Patch(ctx, latestNode, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { + if apierrors.IsConflict(err) { + metrics.StatusPatchConflicts.WithLabelValues("node_taint", "remove").Inc() + } return err } @@ -435,7 +442,12 @@ func (r *RuleReadinessController) markBootstrapCompleted(ctx context.Context, no deferred := false annotationKey := bootstrapAnnotationKey(rule.GetUID()) - // retry to handle conflict with concurrent node updates + // The optimistic lock here isn't guarding the annotation merge itself (that's map-valued + // and merges cleanly against concurrent writers, e.g. Kubelet). It guards the + // hasTaintBySpec check above: without it, a taint added between that check and the Patch + // below would go undetected, and we'd mark bootstrap complete on a node that still carries + // the taint. See the "should not mark bootstrap completed when the rule taints concurrently" + // test for the regression this prevents. err := retry.RetryOnConflict(retry.DefaultRetry, func() error { node := &corev1.Node{} if err := r.Get(ctx, client.ObjectKey{Name: nodeName}, node); err != nil { @@ -462,6 +474,9 @@ func (r *RuleReadinessController) markBootstrapCompleted(ctx context.Context, no node.Annotations[annotationKey] = bootstrapAnnotationValue(rule.Name) if err := r.Patch(ctx, node, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { + if apierrors.IsConflict(err) { + metrics.StatusPatchConflicts.WithLabelValues("node_bootstrap_annotation", "patch").Inc() + } return err } diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index d66193cb..f0682a17 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -212,7 +212,11 @@ func (r *RuleReconciler) reconcileDelete(ctx context.Context, rule *readinessv1a stored := latest.DeepCopy() controllerutil.RemoveFinalizer(latest, finalizerName) - return r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) + err := r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) + if apierrors.IsConflict(err) { + metrics.StatusPatchConflicts.WithLabelValues("rule_finalizer", "remove").Inc() + } + return err }) if err != nil { return ctrl.Result{}, err @@ -629,7 +633,11 @@ func (r *RuleReadinessController) patchRuleStatusWithOptimisticLock( return nil } - return r.Status().Patch(ctx, latestRule, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) + err := r.Status().Patch(ctx, latestRule, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) + if apierrors.IsConflict(err) { + metrics.StatusPatchConflicts.WithLabelValues("rule_status", "patch").Inc() + } + return err }) } @@ -795,6 +803,9 @@ func (r *RuleReconciler) ensureFinalizer(ctx context.Context, rule *readinessv1a stored := latest.DeepCopy() controllerutil.AddFinalizer(latest, finalizer) if err := r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { + if apierrors.IsConflict(err) { + metrics.StatusPatchConflicts.WithLabelValues("rule_finalizer", "add").Inc() + } return err } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 1b7c0158..34c231fa 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -145,6 +145,17 @@ var ( []string{"rule", "condition"}, ) + // StatusPatchConflicts tracks optimistic-lock conflicts on status/annotation patches. + // A rising rate here means concurrent writers are contending for the same object; it may + // show up as increased reconcile latency before it shows up as visible errors. + StatusPatchConflicts = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "node_readiness_status_patch_conflicts_total", + Help: "Total number of optimistic-lock conflicts (HTTP 409) encountered while patching rule status or node annotations", + }, + []string{"resource", "operation"}, + ) + // RuleLastReconciliationTime tracks when a rule was last reconciled. // This provides rule-level visibility for admins to detect stuck rules. RuleLastReconciliationTime = prometheus.NewGaugeVec( @@ -179,6 +190,7 @@ func init() { metrics.Registry.MustRegister(ReconciliationLatency) metrics.Registry.MustRegister(NodesByState) metrics.Registry.MustRegister(ConditionEvaluationFailures) + metrics.Registry.MustRegister(StatusPatchConflicts) metrics.Registry.MustRegister(RuleLastReconciliationTime) metrics.Registry.MustRegister(BuildInfo) } From e1713aa72d4c6f53ecc05bd9a0cccffa339839c7 Mon Sep 17 00:00:00 2001 From: bhuvan-somisetty Date: Sun, 23 Aug 2026 03:44:44 +0530 Subject: [PATCH 3/3] address review: drop StatusPatchConflicts metric from this PR's scope Deferred to #320/#288 per review feedback so conflict metrics get handled holistically instead of piecemeal here. --- internal/controller/node_controller.go | 10 ---------- .../controller/nodereadinessrule_controller.go | 15 ++------------- internal/metrics/metrics.go | 12 ------------ 3 files changed, 2 insertions(+), 35 deletions(-) diff --git a/internal/controller/node_controller.go b/internal/controller/node_controller.go index 95c696a3..4af75163 100644 --- a/internal/controller/node_controller.go +++ b/internal/controller/node_controller.go @@ -22,7 +22,6 @@ import ( "fmt" corev1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -299,9 +298,6 @@ func (r *RuleReadinessController) addTaintBySpec(ctx context.Context, node *core stored := latestNode.DeepCopy() latestNode.Spec.Taints = append(latestNode.Spec.Taints, taintSpec) if err := r.Patch(ctx, latestNode, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { - if apierrors.IsConflict(err) { - metrics.StatusPatchConflicts.WithLabelValues("node_taint", "add").Inc() - } return err } @@ -393,9 +389,6 @@ func (r *RuleReadinessController) removeTaint(ctx context.Context, node *corev1. latestNode.Annotations[key] = annotations[key] } if err := r.Patch(ctx, latestNode, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { - if apierrors.IsConflict(err) { - metrics.StatusPatchConflicts.WithLabelValues("node_taint", "remove").Inc() - } return err } @@ -474,9 +467,6 @@ func (r *RuleReadinessController) markBootstrapCompleted(ctx context.Context, no node.Annotations[annotationKey] = bootstrapAnnotationValue(rule.Name) if err := r.Patch(ctx, node, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { - if apierrors.IsConflict(err) { - metrics.StatusPatchConflicts.WithLabelValues("node_bootstrap_annotation", "patch").Inc() - } return err } diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index f0682a17..d66193cb 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -212,11 +212,7 @@ func (r *RuleReconciler) reconcileDelete(ctx context.Context, rule *readinessv1a stored := latest.DeepCopy() controllerutil.RemoveFinalizer(latest, finalizerName) - err := r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) - if apierrors.IsConflict(err) { - metrics.StatusPatchConflicts.WithLabelValues("rule_finalizer", "remove").Inc() - } - return err + return r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) }) if err != nil { return ctrl.Result{}, err @@ -633,11 +629,7 @@ func (r *RuleReadinessController) patchRuleStatusWithOptimisticLock( return nil } - err := r.Status().Patch(ctx, latestRule, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) - if apierrors.IsConflict(err) { - metrics.StatusPatchConflicts.WithLabelValues("rule_status", "patch").Inc() - } - return err + return r.Status().Patch(ctx, latestRule, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) }) } @@ -803,9 +795,6 @@ func (r *RuleReconciler) ensureFinalizer(ctx context.Context, rule *readinessv1a stored := latest.DeepCopy() controllerutil.AddFinalizer(latest, finalizer) if err := r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { - if apierrors.IsConflict(err) { - metrics.StatusPatchConflicts.WithLabelValues("rule_finalizer", "add").Inc() - } return err } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 34c231fa..1b7c0158 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -145,17 +145,6 @@ var ( []string{"rule", "condition"}, ) - // StatusPatchConflicts tracks optimistic-lock conflicts on status/annotation patches. - // A rising rate here means concurrent writers are contending for the same object; it may - // show up as increased reconcile latency before it shows up as visible errors. - StatusPatchConflicts = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Name: "node_readiness_status_patch_conflicts_total", - Help: "Total number of optimistic-lock conflicts (HTTP 409) encountered while patching rule status or node annotations", - }, - []string{"resource", "operation"}, - ) - // RuleLastReconciliationTime tracks when a rule was last reconciled. // This provides rule-level visibility for admins to detect stuck rules. RuleLastReconciliationTime = prometheus.NewGaugeVec( @@ -190,7 +179,6 @@ func init() { metrics.Registry.MustRegister(ReconciliationLatency) metrics.Registry.MustRegister(NodesByState) metrics.Registry.MustRegister(ConditionEvaluationFailures) - metrics.Registry.MustRegister(StatusPatchConflicts) metrics.Registry.MustRegister(RuleLastReconciliationTime) metrics.Registry.MustRegister(BuildInfo) }