From cc7e8c616b41949781714d8480f6246f1dcf6eee Mon Sep 17 00:00:00 2001 From: Rawad Hossain Date: Thu, 23 Jul 2026 00:51:45 +0600 Subject: [PATCH 1/3] Improve failure metrics Signed-off-by: Rawad Hossain --- docs/book/src/operations/monitoring.md | 2 +- internal/controller/node_controller.go | 6 ++++++ internal/controller/nodereadinessrule_controller.go | 12 ++++++++++-- 3 files changed, 17 insertions(+), 3 deletions(-) diff --git a/docs/book/src/operations/monitoring.md b/docs/book/src/operations/monitoring.md index fe2b6d1b..4980bf81 100644 --- a/docs/book/src/operations/monitoring.md +++ b/docs/book/src/operations/monitoring.md @@ -67,7 +67,7 @@ Total number of failure events recorded by the controller. | Label | Description | Values | | --- | --- | --- | | `rule` | `NodeReadinessRule` name | Any rule name | -| `reason` | Failure label recorded by the controller | `EvaluationError`, `AddTaintError`, `RemoveTaintError` | +| `reason` | Failure label recorded by the controller | `EvaluationError`, `AddTaintError`, `RemoveTaintError`, `AddTaintConflictExhausted`, `RemoveTaintConflictExhausted`, `StatusPatchError`, `StatusPatchConflictExhausted` | ### `node_readiness_build_info` diff --git a/internal/controller/node_controller.go b/internal/controller/node_controller.go index 86b48a6e..3132fbff 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" @@ -214,6 +215,11 @@ func (r *RuleReadinessController) processNodeAgainstAllRules(ctx context.Context }) if err != nil { + reason := "StatusPatchError" + if apierrors.IsConflict(err) { + reason = "StatusPatchConflictExhausted" + } + metrics.Failures.WithLabelValues(rule.Name, reason).Inc() log.Error(err, "Failed to update rule status after node evaluation", "node", node.Name, "rule", rule.Name, diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index eb854306..34dd7b8e 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -443,7 +443,11 @@ func (r *RuleReadinessController) evaluateRuleForNode(ctx context.Context, rule err = r.removeTaintBySpec(ctx, node, rule.Spec.Taint, rule.Name) } if err != nil { - metrics.Failures.WithLabelValues(rule.Name, string(metrics.FailureReasonRemoveTaintError)).Inc() + reason := string(metrics.FailureReasonRemoveTaintError) + if apierrors.IsConflict(err) { + reason = "RemoveTaintConflictExhausted" + } + metrics.Failures.WithLabelValues(rule.Name, reason).Inc() return fmt.Errorf("failed to remove taint: %w", err) } @@ -473,7 +477,11 @@ func (r *RuleReadinessController) evaluateRuleForNode(ctx context.Context, rule var added bool if added, err = r.addTaintBySpec(ctx, node, rule); err != nil { - metrics.Failures.WithLabelValues(rule.Name, string(metrics.FailureReasonAddTaintError)).Inc() + reason := string(metrics.FailureReasonAddTaintError) + if apierrors.IsConflict(err) { + reason = "AddTaintConflictExhausted" + } + metrics.Failures.WithLabelValues(rule.Name, reason).Inc() return fmt.Errorf("failed to add taint: %w", err) } From f1bc8756f68d8944252605a5fa0c4111c036b2f1 Mon Sep 17 00:00:00 2001 From: Rawad Hossain Date: Sun, 23 Aug 2026 00:40:15 +0600 Subject: [PATCH 2/3] record errors and logs --- go.mod | 2 +- internal/controller/node_controller.go | 7 ++ internal/controller/node_controller_test.go | 89 ++++++++++++++++++++- 3 files changed, 95 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index ff9da5d4..55c4e30f 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module sigs.k8s.io/node-readiness-controller go 1.26.0 require ( + github.com/go-logr/logr v1.4.4 github.com/onsi/ginkgo/v2 v2.32.0 github.com/onsi/gomega v1.40.0 github.com/prometheus/client_golang v1.24.0 @@ -29,7 +30,6 @@ require ( github.com/felixge/httpsnoop v1.1.0 // indirect github.com/fsnotify/fsnotify v1.10.1 // indirect github.com/fxamacker/cbor/v2 v2.9.2 // indirect - github.com/go-logr/logr v1.4.4 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-logr/zapr v1.3.0 // indirect github.com/go-openapi/jsonpointer v1.0.0 // indirect diff --git a/internal/controller/node_controller.go b/internal/controller/node_controller.go index 3132fbff..f3520f86 100644 --- a/internal/controller/node_controller.go +++ b/internal/controller/node_controller.go @@ -305,6 +305,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) { + log.V(1).Info("Conflict adding taint to node", "rule", rule.Name, "operation", "add_taint") + } return err } @@ -357,6 +360,7 @@ func (r *RuleReadinessController) removeTaintAndCompleteBootstrap(ctx context.Co // conflict error if the node was modified concurrently, allowing the // controller to retry with fresh state. func (r *RuleReadinessController) removeTaint(ctx context.Context, node *corev1.Node, taintSpec corev1.Taint, ruleName string, annotations map[string]string) (bool, error) { + log := ctrl.LoggerFrom(ctx) hasNewAnnotations := false err := retry.RetryOnConflict(retry.DefaultRetry, func() error { // Fetch latest node state @@ -396,6 +400,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) { + log.V(1).Info("Conflict removing taint from node", "rule", ruleName, "operation", "remove_taint") + } return err } diff --git a/internal/controller/node_controller_test.go b/internal/controller/node_controller_test.go index 3eeba977..6a6efa41 100644 --- a/internal/controller/node_controller_test.go +++ b/internal/controller/node_controller_test.go @@ -19,9 +19,11 @@ package controller import ( "context" "fmt" + "sync" "sync/atomic" "time" + "github.com/go-logr/logr" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" dto "github.com/prometheus/client_model/go" @@ -35,12 +37,62 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake" "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + logf "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/reconcile" nodereadinessiov1alpha1 "sigs.k8s.io/node-readiness-controller/api/v1alpha1" "sigs.k8s.io/node-readiness-controller/internal/metrics" ) +type capturedLogEntry struct { + level int + msg string + keysAndValues []any +} + +type capturingLogSink struct { + mu *sync.Mutex + entries *[]capturedLogEntry +} + +func newCapturingLogger() (logr.Logger, *[]capturedLogEntry) { + entries := &[]capturedLogEntry{} + sink := &capturingLogSink{mu: &sync.Mutex{}, entries: entries} + return logr.New(sink), entries +} + +func (s *capturingLogSink) Init(_ logr.RuntimeInfo) {} + +func (s *capturingLogSink) Enabled(_ int) bool { return true } + +func (s *capturingLogSink) Info(level int, msg string, keysAndValues ...any) { + s.mu.Lock() + defer s.mu.Unlock() + *s.entries = append(*s.entries, capturedLogEntry{level: level, msg: msg, keysAndValues: keysAndValues}) +} + +func (s *capturingLogSink) Error(_ error, msg string, keysAndValues ...any) { + s.mu.Lock() + defer s.mu.Unlock() + *s.entries = append(*s.entries, capturedLogEntry{level: -1, msg: msg, keysAndValues: keysAndValues}) +} + +func (s *capturingLogSink) WithValues(_ ...any) logr.LogSink { return s } + +func (s *capturingLogSink) WithName(_ string) logr.LogSink { return s } + +func fieldsOf(kvs []any) map[string]any { + fields := make(map[string]any, len(kvs)/2) + for i := 0; i+1 < len(kvs); i += 2 { + key, ok := kvs[i].(string) + if !ok { + continue + } + fields[key] = kvs[i+1] + } + return fields +} + var _ = Describe("Node Controller", func() { const ( nodeName = "node-controller-test-node" @@ -859,7 +911,10 @@ var _ = Describe("Node Controller", func() { Expect(fc.Get(ctx, types.NamespacedName{Name: node.Name}, node)).To(Succeed()) - err := controller.removeTaintBySpec(ctx, node, corev1.Taint{ + logger, logEntries := newCapturingLogger() + loggedCtx := logf.IntoContext(ctx, logger) + + err := controller.removeTaintBySpec(loggedCtx, node, corev1.Taint{ Key: "readiness.k8s.io/test", Effect: corev1.TaintEffectNoSchedule, }, "test-rule") @@ -883,6 +938,19 @@ var _ = Describe("Node Controller", func() { // Verify that the patch was attempted twice (first failed, second succeeded) Expect(patchCount.Load()).To(BeNumerically(">=", 2)) + + var conflictLogs []capturedLogEntry + for _, e := range *logEntries { + if e.msg == "Conflict removing taint from node" { + conflictLogs = append(conflictLogs, e) + } + } + + Expect(conflictLogs).To(HaveLen(1)) + Expect(conflictLogs[0].level).To(Equal(1)) + fields := fieldsOf(conflictLogs[0].keysAndValues) + Expect(fields).To(HaveKeyWithValue("rule", "test-rule")) + Expect(fields).To(HaveKeyWithValue("operation", "remove_taint")) }) It("should retry and succeed when addTaintBySpec encounters a conflict", func() { @@ -934,7 +1002,11 @@ var _ = Describe("Node Controller", func() { EnforcementMode: nodereadinessiov1alpha1.EnforcementModeContinuous, }, } - added, err := controller.addTaintBySpec(ctx, node, addRule) + + logger, logEntries := newCapturingLogger() + loggedCtx := logf.IntoContext(ctx, logger) + + added, err := controller.addTaintBySpec(loggedCtx, node, addRule) // Should succeed after retry Expect(err).NotTo(HaveOccurred()) @@ -956,6 +1028,19 @@ var _ = Describe("Node Controller", func() { // Verify that the patch was attempted twice (first failed, second succeeded) Expect(patchCount.Load()).To(BeNumerically(">=", 2)) + + var conflictLogs []capturedLogEntry + for _, e := range *logEntries { + if e.msg == "Conflict adding taint to node" { + conflictLogs = append(conflictLogs, e) + } + } + + Expect(conflictLogs).To(HaveLen(1)) + Expect(conflictLogs[0].level).To(Equal(1)) + fields := fieldsOf(conflictLogs[0].keysAndValues) + Expect(fields).To(HaveKeyWithValue("rule", "test-rule")) + Expect(fields).To(HaveKeyWithValue("operation", "add_taint")) }) It("should not mark bootstrap completed when the rule taints concurrently", func() { From 4d3181b5e2d85c157d82ded9b38dabde5f275972 Mon Sep 17 00:00:00 2001 From: Rawad Hossain Date: Sun, 6 Sep 2026 14:51:46 +0600 Subject: [PATCH 3/3] add api conflicts metric --- docs/book/src/operations/monitoring.md | 31 +- .../controller/api_conflicts_metric_test.go | 677 ++++++++++++++++++ internal/controller/node_controller.go | 28 +- .../nodereadinessrule_controller.go | 87 ++- internal/metrics/metrics.go | 34 +- 5 files changed, 818 insertions(+), 39 deletions(-) create mode 100644 internal/controller/api_conflicts_metric_test.go diff --git a/docs/book/src/operations/monitoring.md b/docs/book/src/operations/monitoring.md index 4980bf81..3ceed82e 100644 --- a/docs/book/src/operations/monitoring.md +++ b/docs/book/src/operations/monitoring.md @@ -67,7 +67,7 @@ Total number of failure events recorded by the controller. | Label | Description | Values | | --- | --- | --- | | `rule` | `NodeReadinessRule` name | Any rule name | -| `reason` | Failure label recorded by the controller | `EvaluationError`, `AddTaintError`, `RemoveTaintError`, `AddTaintConflictExhausted`, `RemoveTaintConflictExhausted`, `StatusPatchError`, `StatusPatchConflictExhausted` | +| `reason` | Failure label recorded by the controller | `EvaluationError`, `AddTaintError`, `RemoveTaintError`, `AddTaintConflictExhausted`, `RemoveTaintConflictExhausted`, `StatusPatchError`, `StatusPatchConflictExhausted`, `RuleStatusRuleSweepConflictExhausted` | ### `node_readiness_build_info` @@ -141,6 +141,35 @@ Total number of nodes that have completed bootstrap. | --- | --- | --- | | `rule` | `NodeReadinessRule` name | Any rule name | +### `node_readiness_api_conflicts_total` + +Total number of conflicts encountered on API writes, counted once per failed attempt, including attempts that later succeed on retry. + +| Property | Value | +| --- | --- | +| Type | `counter` | +| Labels | `rule`, `operation` | +| Recorded when | Controller retries an API write after a conflicting update from another writer | + +#### Labels + +| Label | Description | Values | +| --- | --- | --- | +| `rule` | `NodeReadinessRule` name | Any rule name | +| `operation` | Which write conflicted | See table below | + +#### Operation values + +| `operation` | Triggered by | +| --- | --- | +| `add_taint` | Adding a rule's taint to a Node (`addTaintBySpec`). | +| `remove_taint` | Removing a rule's taint from a Node (`removeTaint`). | +| `mark_bootstrap_completed` | Marking bootstrap as completed when no taint needs to be removed (`markBootstrapCompleted`). | +| `finalizer_add` | Adding the controller finalizer to a `NodeReadinessRule` (`ensureFinalizer`). | +| `finalizer_remove` | Removing the controller finalizer during rule deletion (`reconcileDelete`). | +| `rule_status_node_write` | Updating a single node's evaluation in rule `status` (`processNodeAgainstAllRules`). | +| `rule_status_rule_sweep` | Updating rule status for all nodes or removing deleted nodes (`updateRuleStatus`, `cleanupDeletedNodes`). | + ## Reporter Metrics The `readiness-condition-reporter` serves its own Prometheus metrics on `/metrics`, on the address configured by `METRICS_BIND_ADDRESS`. See [Reporter Configuration](../reference/reporter-configuration.md) for deployment details. diff --git a/internal/controller/api_conflicts_metric_test.go b/internal/controller/api_conflicts_metric_test.go new file mode 100644 index 00000000..3a9c3cb6 --- /dev/null +++ b/internal/controller/api_conflicts_metric_test.go @@ -0,0 +1,677 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "context" + "fmt" + "sync/atomic" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + 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" + "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" + + nodereadinessiov1alpha1 "sigs.k8s.io/node-readiness-controller/api/v1alpha1" + "sigs.k8s.io/node-readiness-controller/internal/metrics" +) + +// apiConflictsValue reads the current value of APIConflicts{rule, operation}. +func apiConflictsValue(rule string, op metrics.ConflictOperation) float64 { + return counterValue(metrics.APIConflicts.WithLabelValues(rule, string(op))) +} + +// failuresValue reads the current value of Failures{rule, reason}. +func failuresValue(rule string, reason metrics.FailureReason) float64 { + return counterValue(metrics.Failures.WithLabelValues(rule, string(reason))) +} + +var _ = Describe("node_readiness_api_conflicts_total", func() { + var ( + ctx context.Context + testScheme *runtime.Scheme + ) + + BeforeEach(func() { + ctx = context.Background() + testScheme = runtime.NewScheme() + Expect(corev1.AddToScheme(testScheme)).To(Succeed()) + Expect(nodereadinessiov1alpha1.AddToScheme(testScheme)).To(Succeed()) + }) + + It("records an add_taint conflict and retries", func() { + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-add-node", Labels: map[string]string{"role": "worker"}}, + } + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-add-rule"}, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{{Type: "Ready", RequiredStatus: corev1.ConditionTrue}}, + Taint: corev1.Taint{Key: "readiness.k8s.io/api-conflict-add", 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). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*corev1.Node); ok && patchCount.Add(1) == 1 { + // Simulate another writer changing the node between the Get and the Patch. + current := &corev1.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: obj.GetName()}, current)).To(Succeed()) + current.Spec.Taints = append(current.Spec.Taints, corev1.Taint{ + Key: "other-controller/taint", Effect: corev1.TaintEffectNoSchedule, + }) + Expect(c.Update(ctx, current)).To(Succeed()) + } + return c.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), + } + + Expect(fc.Get(ctx, types.NamespacedName{Name: node.Name}, node)).To(Succeed()) + before := apiConflictsValue(rule.Name, metrics.ConflictOperationAddTaint) + + added, err := controller.addTaintBySpec(ctx, node, rule) + Expect(err).NotTo(HaveOccurred()) + Expect(added).To(BeTrue()) + Expect(patchCount.Load()).To(BeNumerically(">=", 2), "the first patch conflicts, so it should retry") + + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationAddTaint)). + To(BeNumerically("==", before+1)) + }) + + It("records a finalizer_add conflict and retries", func() { + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-finalizer-rule"}, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{{Type: "Ready", RequiredStatus: corev1.ConditionTrue}}, + Taint: corev1.Taint{Key: "readiness.k8s.io/api-conflict-finalizer", 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(rule). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*nodereadinessiov1alpha1.NodeReadinessRule); ok && patchCount.Add(1) == 1 { + current := &nodereadinessiov1alpha1.NodeReadinessRule{} + Expect(c.Get(ctx, types.NamespacedName{Name: obj.GetName()}, current)).To(Succeed()) + if current.Labels == nil { + current.Labels = map[string]string{} + } + current.Labels["bumped-by"] = "competing-writer" + Expect(c.Update(ctx, current)).To(Succeed()) + } + return c.Patch(ctx, obj, patch, opts...) + }, + }). + Build() + + reconciler := &RuleReconciler{Client: fc, Scheme: testScheme} + + Expect(fc.Get(ctx, types.NamespacedName{Name: rule.Name}, rule)).To(Succeed()) + before := apiConflictsValue(rule.Name, metrics.ConflictOperationFinalizerAdd) + + added, err := reconciler.ensureFinalizer(ctx, rule, finalizerName) + Expect(err).NotTo(HaveOccurred()) + Expect(added).To(BeTrue()) + Expect(patchCount.Load()).To(BeNumerically(">=", 2), "the first patch conflicts, so it should retry") + + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationFinalizerAdd)). + To(BeNumerically("==", before+1)) + }) + + It("records a finalizer_remove conflict during delete and retries", func() { + deletionTS := metav1.Now() + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{ + Name: "api-conflict-finalizer-remove-rule", + Finalizers: []string{finalizerName}, + DeletionTimestamp: &deletionTS, + }, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{{Type: "Ready", RequiredStatus: corev1.ConditionTrue}}, + Taint: corev1.Taint{Key: "readiness.k8s.io/api-conflict-finalizer-remove", 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(rule). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*nodereadinessiov1alpha1.NodeReadinessRule); ok && patchCount.Add(1) == 1 { + current := &nodereadinessiov1alpha1.NodeReadinessRule{} + Expect(c.Get(ctx, types.NamespacedName{Name: obj.GetName()}, current)).To(Succeed()) + if current.Labels == nil { + current.Labels = map[string]string{} + } + current.Labels["bumped-by"] = "competing-writer" + Expect(c.Update(ctx, current)).To(Succeed()) + } + return c.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), + } + reconciler := &RuleReconciler{Client: fc, Scheme: testScheme, Controller: controller} + + fetched := &nodereadinessiov1alpha1.NodeReadinessRule{} + Expect(fc.Get(ctx, types.NamespacedName{Name: rule.Name}, fetched)).To(Succeed()) + controller.updateRuleCache(ctx, fetched) + + before := apiConflictsValue(rule.Name, metrics.ConflictOperationFinalizerRemove) + + result, err := reconciler.reconcileDelete(ctx, fetched, &corev1.NodeList{}) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RequeueAfter).To(BeZero(), "a successful delete should not requeue") + Expect(patchCount.Load()).To(BeNumerically(">=", 2), "the first patch conflicts, so it should retry") + + remaining := &nodereadinessiov1alpha1.NodeReadinessRule{} + err = fc.Get(ctx, types.NamespacedName{Name: rule.Name}, remaining) + if err == nil { + Expect(remaining.Finalizers).NotTo(ContainElement(finalizerName)) + } else { + Expect(apierrors.IsNotFound(err)).To(BeTrue(), "the rule may be deleted once its last finalizer is gone") + } + + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationFinalizerRemove)). + To(BeNumerically("==", before+1)) + }) + + It("records a remove_taint conflict and retries", func() { + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-remove-rule"}, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{{Type: "Ready", RequiredStatus: corev1.ConditionTrue}}, + Taint: corev1.Taint{Key: "readiness.k8s.io/api-conflict-remove", Effect: corev1.TaintEffectNoSchedule}, + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"role": "worker"}}, + EnforcementMode: nodereadinessiov1alpha1.EnforcementModeContinuous, + }, + } + + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-remove-node", Labels: map[string]string{"role": "worker"}}, + Spec: corev1.NodeSpec{Taints: []corev1.Taint{ + {Key: "readiness.k8s.io/api-conflict-remove", Effect: corev1.TaintEffectNoSchedule}, + {Key: "other-controller/taint", Effect: corev1.TaintEffectNoSchedule}, + }}, + Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: "Ready", Status: corev1.ConditionTrue}}}, + } + + var patchCount atomic.Int32 + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*corev1.Node); ok && patchCount.Add(1) == 1 { + current := &corev1.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: obj.GetName()}, current)).To(Succeed()) + current.Spec.Taints = append(current.Spec.Taints, corev1.Taint{ + Key: "third-controller/taint", Effect: corev1.TaintEffectNoSchedule, + }) + Expect(c.Update(ctx, current)).To(Succeed()) + } + return c.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), + } + + Expect(fc.Get(ctx, types.NamespacedName{Name: node.Name}, node)).To(Succeed()) + before := apiConflictsValue(rule.Name, metrics.ConflictOperationRemoveTaint) + + err := controller.removeTaintBySpec(ctx, node, rule.Spec.Taint, rule.Name) + Expect(err).NotTo(HaveOccurred()) + Expect(patchCount.Load()).To(BeNumerically(">=", 2), "the first patch conflicts, so it should retry") + + updated := &corev1.Node{} + Expect(fc.Get(ctx, types.NamespacedName{Name: node.Name}, updated)).To(Succeed()) + Expect(controller.hasTaintBySpec(updated, rule.Spec.Taint)).To(BeFalse(), "the taint should be gone") + + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRemoveTaint)). + To(BeNumerically("==", before+1)) + }) + + It("records a mark_bootstrap_completed conflict and retries", func() { + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{ + Name: "api-conflict-bootstrap-rule", + UID: types.UID("9a9a9a9a-9a9a-9a9a-9a9a-9a9a9a9a9a9a"), + }, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{{Type: "Ready", RequiredStatus: corev1.ConditionTrue}}, + Taint: corev1.Taint{Key: "readiness.k8s.io/api-conflict-bootstrap", Effect: corev1.TaintEffectNoSchedule}, + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"role": "worker"}}, + EnforcementMode: nodereadinessiov1alpha1.EnforcementModeBootstrapOnly, + }, + } + + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-bootstrap-node", Labels: map[string]string{"role": "worker"}}, + Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: "Ready", Status: corev1.ConditionTrue}}}, + } + + var patchCount atomic.Int32 + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*corev1.Node); ok && patchCount.Add(1) == 1 { + current := &corev1.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: obj.GetName()}, current)).To(Succeed()) + if current.Labels == nil { + current.Labels = map[string]string{} + } + current.Labels["bumped-by"] = "competing-writer" + Expect(c.Update(ctx, current)).To(Succeed()) + } + return c.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), + } + + before := apiConflictsValue(rule.Name, metrics.ConflictOperationMarkBootstrapCompleted) + + controller.markBootstrapCompleted(ctx, node.Name, rule) + Expect(patchCount.Load()).To(BeNumerically(">=", 2), "the first patch conflicts, so it should retry") + + updated := &corev1.Node{} + Expect(fc.Get(ctx, types.NamespacedName{Name: node.Name}, updated)).To(Succeed()) + Expect(updated.Annotations).To(HaveKey(bootstrapAnnotationKey(rule.GetUID())), + "the retry should mark bootstrap complete") + + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationMarkBootstrapCompleted)). + To(BeNumerically("==", before+1)) + }) + + conflictOnEveryNodePatch := func(patchCount *atomic.Int32) interceptor.Funcs { + return interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*corev1.Node); ok { + n := patchCount.Add(1) + current := &corev1.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: obj.GetName()}, current)).To(Succeed()) + if current.Labels == nil { + current.Labels = map[string]string{} + } + current.Labels["bumped-by"] = fmt.Sprintf("competing-writer-%d", n) + Expect(c.Update(ctx, current)).To(Succeed()) + } + return c.Patch(ctx, obj, patch, opts...) + }, + } + } + + It("records AddTaintConflictExhausted when retries run out", func() { + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-add-exhaust-rule"}, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{{Type: "Ready", RequiredStatus: corev1.ConditionTrue}}, + Taint: corev1.Taint{Key: "readiness.k8s.io/api-conflict-add-exhaust", Effect: corev1.TaintEffectNoSchedule}, + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"role": "worker"}}, + EnforcementMode: nodereadinessiov1alpha1.EnforcementModeContinuous, + }, + } + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-add-exhaust-node", Labels: map[string]string{"role": "worker"}}, + Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: "Ready", Status: corev1.ConditionFalse}}}, + } + + var patchCount atomic.Int32 + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithInterceptorFuncs(conflictOnEveryNodePatch(&patchCount)). + Build() + + controller := &RuleReadinessController{ + Client: fc, + Scheme: testScheme, + clientset: fake.NewSimpleClientset(), + ruleCache: make(map[string]*nodereadinessiov1alpha1.NodeReadinessRule), + EventRecorder: events.NewFakeRecorder(10), + } + + Expect(fc.Get(ctx, types.NamespacedName{Name: node.Name}, node)).To(Succeed()) + beforeExhausted := failuresValue(rule.Name, metrics.FailureReasonAddTaintConflictExhausted) + beforeAddError := failuresValue(rule.Name, metrics.FailureReasonAddTaintError) + beforeConflicts := apiConflictsValue(rule.Name, metrics.ConflictOperationAddTaint) + + err := controller.evaluateRuleForNode(ctx, rule, node) + Expect(err).To(HaveOccurred()) + Expect(apierrors.IsConflict(err)).To(BeTrue(), "the last conflict should be returned") + Expect(patchCount.Load()).To(BeNumerically(">=", 5), "all retries should conflict") + + Expect(failuresValue(rule.Name, metrics.FailureReasonAddTaintConflictExhausted)). + To(BeNumerically("==", beforeExhausted+1), + "should record AddTaintConflictExhausted") + Expect(failuresValue(rule.Name, metrics.FailureReasonAddTaintError)). + To(BeNumerically("==", beforeAddError), "should not record AddTaintError") + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationAddTaint)). + To(BeNumerically("==", beforeConflicts+float64(patchCount.Load())), + "each conflict should be counted") + }) + + It("records RemoveTaintConflictExhausted when retries run out", func() { + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-remove-exhaust-rule"}, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{{Type: "Ready", RequiredStatus: corev1.ConditionTrue}}, + Taint: corev1.Taint{Key: "readiness.k8s.io/api-conflict-remove-exhaust", Effect: corev1.TaintEffectNoSchedule}, + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"role": "worker"}}, + EnforcementMode: nodereadinessiov1alpha1.EnforcementModeContinuous, + }, + } + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "api-conflict-remove-exhaust-node", Labels: map[string]string{"role": "worker"}}, + Spec: corev1.NodeSpec{Taints: []corev1.Taint{ + {Key: "readiness.k8s.io/api-conflict-remove-exhaust", Effect: corev1.TaintEffectNoSchedule}, + }}, + Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: "Ready", Status: corev1.ConditionTrue}}}, + } + + var patchCount atomic.Int32 + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithInterceptorFuncs(conflictOnEveryNodePatch(&patchCount)). + Build() + + controller := &RuleReadinessController{ + Client: fc, + Scheme: testScheme, + clientset: fake.NewSimpleClientset(), + ruleCache: make(map[string]*nodereadinessiov1alpha1.NodeReadinessRule), + EventRecorder: events.NewFakeRecorder(10), + } + + Expect(fc.Get(ctx, types.NamespacedName{Name: node.Name}, node)).To(Succeed()) + beforeExhausted := failuresValue(rule.Name, metrics.FailureReasonRemoveTaintConflictExhausted) + beforeRemoveError := failuresValue(rule.Name, metrics.FailureReasonRemoveTaintError) + beforeConflicts := apiConflictsValue(rule.Name, metrics.ConflictOperationRemoveTaint) + + err := controller.evaluateRuleForNode(ctx, rule, node) + Expect(err).To(HaveOccurred()) + Expect(apierrors.IsConflict(err)).To(BeTrue(), "the last conflict should be returned") + Expect(patchCount.Load()).To(BeNumerically(">=", 5), "all retries should conflict") + + Expect(failuresValue(rule.Name, metrics.FailureReasonRemoveTaintConflictExhausted)). + To(BeNumerically("==", beforeExhausted+1), + "should record RemoveTaintConflictExhausted") + Expect(failuresValue(rule.Name, metrics.FailureReasonRemoveTaintError)). + To(BeNumerically("==", beforeRemoveError), "should not record RemoveTaintError") + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRemoveTaint)). + To(BeNumerically("==", beforeConflicts+float64(patchCount.Load())), + "each conflict should be counted") + }) + + Context("status patch conflicts use the caller's operation label", func() { + newRuleWithNode := func(ruleName, nodeName string) (*nodereadinessiov1alpha1.NodeReadinessRule, *corev1.Node) { + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: nodeName, Labels: map[string]string{"role": "worker"}}, + Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: "Ready", Status: corev1.ConditionTrue}}}, + } + rule := &nodereadinessiov1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: ruleName}, + Spec: nodereadinessiov1alpha1.NodeReadinessRuleSpec{ + Conditions: []nodereadinessiov1alpha1.ConditionRequirement{{Type: "Ready", RequiredStatus: corev1.ConditionTrue}}, + Taint: corev1.Taint{Key: "readiness.k8s.io/" + ruleName, Effect: corev1.TaintEffectNoSchedule}, + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"role": "worker"}}, + EnforcementMode: nodereadinessiov1alpha1.EnforcementModeContinuous, + }, + } + return rule, node + } + + conflictOnFirstStatusPatch := func(ruleName string, patchCount *atomic.Int32) interceptor.Funcs { + return 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 == ruleName && patchCount.Add(1) == 1 { + 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: "competing-node", + TaintStatus: nodereadinessiov1alpha1.TaintStatusAbsent, + }) + Expect(c.Status().Update(ctx, current)).To(Succeed()) + } + return c.SubResource(subResourceName).Patch(ctx, obj, patch, opts...) + }, + } + } + + conflictOnEveryStatusPatch := func(ruleName string, patchCount *atomic.Int32) interceptor.Funcs { + return 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 == ruleName { + n := patchCount.Add(1) + competing := &nodereadinessiov1alpha1.NodeReadinessRule{} + Expect(c.Get(ctx, types.NamespacedName{Name: r.Name}, competing)).To(Succeed()) + competing.Status.NodeEvaluations = append(competing.Status.NodeEvaluations, nodereadinessiov1alpha1.NodeEvaluation{ + NodeName: fmt.Sprintf("competing-node-%d", n), + TaintStatus: nodereadinessiov1alpha1.TaintStatusAbsent, + }) + Expect(c.Status().Update(ctx, competing)).To(Succeed()) + } + return c.SubResource(subResourceName).Patch(ctx, obj, patch, opts...) + }, + } + } + + It("uses rule_status_node_write for NodeReconciler status updates", func() { + rule, node := newRuleWithNode("api-conflict-node-write-rule", "api-conflict-node-write-node") + + var patchCount atomic.Int32 + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithStatusSubresource(rule). + WithInterceptorFuncs(conflictOnFirstStatusPatch(rule.Name, &patchCount)). + Build() + + controller := &RuleReadinessController{ + Client: fc, + Scheme: testScheme, + clientset: fake.NewSimpleClientset(), + ruleCache: map[string]*nodereadinessiov1alpha1.NodeReadinessRule{rule.Name: rule}, + EventRecorder: events.NewFakeRecorder(10), + } + + beforeNodeWrite := apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusNodeWrite) + beforeSweep := apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusRuleSweep) + + Expect(controller.processNodeAgainstAllRules(ctx, node)).To(Succeed()) + Expect(patchCount.Load()).To(BeNumerically(">=", 2), "the first patch conflicts, so it should retry") + + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusNodeWrite)). + To(BeNumerically("==", beforeNodeWrite+1), "should use rule_status_node_write") + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusRuleSweep)). + To(BeNumerically("==", beforeSweep), "should not use rule_status_rule_sweep") + }) + + It("uses rule_status_rule_sweep for RuleReconciler status updates", func() { + rule, node := newRuleWithNode("api-conflict-sweep-rule", "api-conflict-sweep-node") + + var patchCount atomic.Int32 + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithStatusSubresource(rule). + WithInterceptorFuncs(conflictOnFirstStatusPatch(rule.Name, &patchCount)). + Build() + + controller := &RuleReadinessController{ + Client: fc, + Scheme: testScheme, + clientset: fake.NewSimpleClientset(), + ruleCache: make(map[string]*nodereadinessiov1alpha1.NodeReadinessRule), + EventRecorder: events.NewFakeRecorder(10), + } + controller.updateRuleCache(ctx, rule) + + nodeList := &corev1.NodeList{Items: []corev1.Node{*node}} + delta, err := controller.processAllNodesForRule(ctx, rule, nodeList) + Expect(err).NotTo(HaveOccurred()) + + beforeSweep := apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusRuleSweep) + beforeNodeWrite := apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusNodeWrite) + + Expect(controller.updateRuleStatus(ctx, rule, delta)).To(Succeed()) + Expect(patchCount.Load()).To(BeNumerically(">=", 2), "the first patch conflicts, so it should retry") + + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusRuleSweep)). + To(BeNumerically("==", beforeSweep+1), "should use rule_status_rule_sweep") + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusNodeWrite)). + To(BeNumerically("==", beforeNodeWrite), "should not use rule_status_node_write") + }) + + It("uses rule_status_rule_sweep when cleaning up deleted nodes", func() { + rule, node := newRuleWithNode("api-conflict-cleanup-rule", "api-conflict-cleanup-node") + rule.Status.NodeEvaluations = []nodereadinessiov1alpha1.NodeEvaluation{ + {NodeName: "deleted-node", TaintStatus: nodereadinessiov1alpha1.TaintStatusAbsent}, + } + + var patchCount atomic.Int32 + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithStatusSubresource(rule). + WithInterceptorFuncs(conflictOnFirstStatusPatch(rule.Name, &patchCount)). + Build() + + controller := &RuleReadinessController{ + Client: fc, + Scheme: testScheme, + clientset: fake.NewSimpleClientset(), + ruleCache: make(map[string]*nodereadinessiov1alpha1.NodeReadinessRule), + EventRecorder: events.NewFakeRecorder(10), + } + + beforeSweep := apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusRuleSweep) + beforeNodeWrite := apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusNodeWrite) + + nodeList := &corev1.NodeList{Items: []corev1.Node{*node}} + Expect(controller.cleanupDeletedNodes(ctx, rule, nodeList)).To(Succeed()) + Expect(patchCount.Load()).To(BeNumerically(">=", 2), "the first patch conflicts, so it should retry") + + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusRuleSweep)). + To(BeNumerically("==", beforeSweep+1), "should use rule_status_rule_sweep") + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusNodeWrite)). + To(BeNumerically("==", beforeNodeWrite), "should not use rule_status_node_write") + }) + + It("records RuleStatusRuleSweepConflictExhausted when retries run out", func() { + rule, node := newRuleWithNode("api-conflict-exhaust-rule", "api-conflict-exhaust-node") + + var patchCount atomic.Int32 + fc := fakeclient.NewClientBuilder(). + WithScheme(testScheme). + WithObjects(node, rule). + WithStatusSubresource(rule). + WithInterceptorFuncs(conflictOnEveryStatusPatch(rule.Name, &patchCount)). + Build() + + controller := &RuleReadinessController{ + Client: fc, + Scheme: testScheme, + clientset: fake.NewSimpleClientset(), + ruleCache: make(map[string]*nodereadinessiov1alpha1.NodeReadinessRule), + EventRecorder: events.NewFakeRecorder(10), + } + controller.updateRuleCache(ctx, rule) + + nodeList := &corev1.NodeList{Items: []corev1.Node{*node}} + delta, err := controller.processAllNodesForRule(ctx, rule, nodeList) + Expect(err).NotTo(HaveOccurred()) + + beforeSweepExhausted := failuresValue(rule.Name, metrics.FailureReasonRuleStatusRuleSweepConflictExhausted) + beforeNodeExhausted := failuresValue(rule.Name, metrics.FailureReasonStatusPatchConflictExhausted) + beforePatchError := failuresValue(rule.Name, metrics.FailureReasonStatusPatchError) + beforeSweepConflicts := apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusRuleSweep) + + err = controller.updateRuleStatus(ctx, rule, delta) + Expect(err).To(HaveOccurred()) + Expect(apierrors.IsConflict(err)).To(BeTrue(), "the last conflict should be returned") + Expect(patchCount.Load()).To(BeNumerically(">=", 5), "all retries should conflict") + + Expect(failuresValue(rule.Name, metrics.FailureReasonRuleStatusRuleSweepConflictExhausted)). + To(BeNumerically("==", beforeSweepExhausted+1), + "should record RuleStatusRuleSweepConflictExhausted") + Expect(failuresValue(rule.Name, metrics.FailureReasonStatusPatchConflictExhausted)). + To(BeNumerically("==", beforeNodeExhausted), + "should not record StatusPatchConflictExhausted") + Expect(failuresValue(rule.Name, metrics.FailureReasonStatusPatchError)). + To(BeNumerically("==", beforePatchError), "should not record StatusPatchError") + Expect(apiConflictsValue(rule.Name, metrics.ConflictOperationRuleStatusRuleSweep)). + To(BeNumerically("==", beforeSweepConflicts+float64(patchCount.Load())), + "each conflict should be counted") + }) + }) +}) diff --git a/internal/controller/node_controller.go b/internal/controller/node_controller.go index f3520f86..fac2602f 100644 --- a/internal/controller/node_controller.go +++ b/internal/controller/node_controller.go @@ -209,17 +209,13 @@ func (r *RuleReadinessController) processNodeAgainstAllRules(ctx context.Context } } - err := r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, func(latestRule *readinessv1alpha1.NodeReadinessRule) { - applyNodeStatusDelta(latestRule, delta) - successfullyPatchedRule = latestRule - }) + err := r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, metrics.ConflictOperationRuleStatusNodeWrite, + func(latestRule *readinessv1alpha1.NodeReadinessRule) { + applyNodeStatusDelta(latestRule, delta) + successfullyPatchedRule = latestRule + }) if err != nil { - reason := "StatusPatchError" - if apierrors.IsConflict(err) { - reason = "StatusPatchConflictExhausted" - } - metrics.Failures.WithLabelValues(rule.Name, reason).Inc() log.Error(err, "Failed to update rule status after node evaluation", "node", node.Name, "rule", rule.Name, @@ -306,7 +302,9 @@ func (r *RuleReadinessController) addTaintBySpec(ctx context.Context, node *core 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) { - log.V(1).Info("Conflict adding taint to node", "rule", rule.Name, "operation", "add_taint") + metrics.APIConflicts.WithLabelValues(rule.Name, string(metrics.ConflictOperationAddTaint)).Inc() + log.V(1).Info("Conflict adding taint to node", + "rule", rule.Name, "operation", string(metrics.ConflictOperationAddTaint)) } return err } @@ -401,7 +399,9 @@ func (r *RuleReadinessController) removeTaint(ctx context.Context, node *corev1. } if err := r.Patch(ctx, latestNode, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { if apierrors.IsConflict(err) { - log.V(1).Info("Conflict removing taint from node", "rule", ruleName, "operation", "remove_taint") + metrics.APIConflicts.WithLabelValues(ruleName, string(metrics.ConflictOperationRemoveTaint)).Inc() + log.V(1).Info("Conflict removing taint from node", + "rule", ruleName, "operation", string(metrics.ConflictOperationRemoveTaint)) } return err } @@ -477,6 +477,12 @@ 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.APIConflicts.WithLabelValues(rule.Name, string(metrics.ConflictOperationMarkBootstrapCompleted)).Inc() + log.V(1).Info("Conflict marking bootstrap completed on node", + "node", nodeName, "rule", rule.Name, + "operation", string(metrics.ConflictOperationMarkBootstrapCompleted)) + } return err } diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index 34dd7b8e..e9843a65 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -213,7 +213,15 @@ 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{})) + if err := r.Patch(ctx, latest, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { + if apierrors.IsConflict(err) { + metrics.APIConflicts.WithLabelValues(rule.Name, string(metrics.ConflictOperationFinalizerRemove)).Inc() + log.V(1).Info("Conflict removing finalizer from rule", + "rule", rule.Name, "operation", string(metrics.ConflictOperationFinalizerRemove)) + } + return err + } + return nil }) if err != nil { return ctrl.Result{}, err @@ -266,16 +274,17 @@ func (r *RuleReadinessController) cleanupDeletedNodes(ctx context.Context, rule "after", len(newNodeEvaluations)) // Use an optimistic-locked patch to avoid race conditions from concurrent node updates. - return r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, func(fresh *readinessv1alpha1.NodeReadinessRule) { - freshNodeEvaluations, freshFailedNodes := filterStatusForExistingNodes( - existingNodes, - fresh.Status.NodeEvaluations, - fresh.Status.FailedNodes, - ) + return r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, metrics.ConflictOperationRuleStatusRuleSweep, + func(fresh *readinessv1alpha1.NodeReadinessRule) { + freshNodeEvaluations, freshFailedNodes := filterStatusForExistingNodes( + existingNodes, + fresh.Status.NodeEvaluations, + fresh.Status.FailedNodes, + ) - fresh.Status.NodeEvaluations = freshNodeEvaluations - fresh.Status.FailedNodes = freshFailedNodes - }) + fresh.Status.NodeEvaluations = freshNodeEvaluations + fresh.Status.FailedNodes = freshFailedNodes + }) } // processAllNodesForRule processes all nodes when a rule changes. It mutates rule.Status in place @@ -443,11 +452,11 @@ func (r *RuleReadinessController) evaluateRuleForNode(ctx context.Context, rule err = r.removeTaintBySpec(ctx, node, rule.Spec.Taint, rule.Name) } if err != nil { - reason := string(metrics.FailureReasonRemoveTaintError) + reason := metrics.FailureReasonRemoveTaintError if apierrors.IsConflict(err) { - reason = "RemoveTaintConflictExhausted" + reason = metrics.FailureReasonRemoveTaintConflictExhausted } - metrics.Failures.WithLabelValues(rule.Name, reason).Inc() + metrics.Failures.WithLabelValues(rule.Name, string(reason)).Inc() return fmt.Errorf("failed to remove taint: %w", err) } @@ -477,11 +486,11 @@ func (r *RuleReadinessController) evaluateRuleForNode(ctx context.Context, rule var added bool if added, err = r.addTaintBySpec(ctx, node, rule); err != nil { - reason := string(metrics.FailureReasonAddTaintError) + reason := metrics.FailureReasonAddTaintError if apierrors.IsConflict(err) { - reason = "AddTaintConflictExhausted" + reason = metrics.FailureReasonAddTaintConflictExhausted } - metrics.Failures.WithLabelValues(rule.Name, reason).Inc() + metrics.Failures.WithLabelValues(rule.Name, string(reason)).Inc() return fmt.Errorf("failed to add taint: %w", err) } @@ -746,9 +755,11 @@ func (r *RuleReadinessController) removeRuleFromCache(ctx context.Context, ruleN func (r *RuleReadinessController) patchRuleStatusWithOptimisticLock( ctx context.Context, ruleName string, + operation metrics.ConflictOperation, mutate func(latest *readinessv1alpha1.NodeReadinessRule), ) error { - return retry.RetryOnConflict(retry.DefaultRetry, func() error { + log := ctrl.LoggerFrom(ctx) + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { latestRule := &readinessv1alpha1.NodeReadinessRule{} if err := r.Get(ctx, client.ObjectKey{Name: ruleName}, latestRule); err != nil { return err @@ -761,8 +772,29 @@ func (r *RuleReadinessController) patchRuleStatusWithOptimisticLock( return nil } - return r.Status().Patch(ctx, latestRule, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})) + if err := r.Status().Patch(ctx, latestRule, client.MergeFromWithOptions(stored, client.MergeFromWithOptimisticLock{})); err != nil { + if apierrors.IsConflict(err) { + metrics.APIConflicts.WithLabelValues(ruleName, string(operation)).Inc() + log.V(1).Info("Conflict patching rule status", + "rule", ruleName, "operation", string(operation)) + } + return err + } + return nil }) + if err != nil { + reason := metrics.FailureReasonStatusPatchError + if apierrors.IsConflict(err) { + reason = metrics.FailureReasonStatusPatchConflictExhausted + if operation == metrics.ConflictOperationRuleStatusRuleSweep { + reason = metrics.FailureReasonRuleStatusRuleSweepConflictExhausted + } + log.V(1).Info("Rule status patch exhausted all retries", + "rule", ruleName, "operation", string(operation), "reason", string(reason)) + } + metrics.Failures.WithLabelValues(ruleName, string(reason)).Inc() + } + return err } // updateRuleStatus updates the status of a NodeReadinessRule. delta carries the per-node @@ -778,12 +810,13 @@ func (r *RuleReadinessController) updateRuleStatus(ctx context.Context, rule *re "nodeEvaluations", len(rule.Status.NodeEvaluations), "appliedNodes", len(rule.Status.AppliedNodes)) - err := r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, func(latestRule *readinessv1alpha1.NodeReadinessRule) { - applyNodeStatusDelta(latestRule, delta) - latestRule.Status.AppliedNodes = rule.Status.AppliedNodes - latestRule.Status.ObservedGeneration = rule.Status.ObservedGeneration - latestRule.Status.DryRunResults = rule.Status.DryRunResults - }) + err := r.patchRuleStatusWithOptimisticLock(ctx, rule.Name, metrics.ConflictOperationRuleStatusRuleSweep, + func(latestRule *readinessv1alpha1.NodeReadinessRule) { + applyNodeStatusDelta(latestRule, delta) + latestRule.Status.AppliedNodes = rule.Status.AppliedNodes + latestRule.Status.ObservedGeneration = rule.Status.ObservedGeneration + latestRule.Status.DryRunResults = rule.Status.DryRunResults + }) if err != nil { log.V(1).Info("Failed to patch rule status", "rule", rule.Name, "error", err.Error()) return err @@ -911,6 +944,7 @@ func (r *RuleReadinessController) cleanupTaintsForRule(ctx context.Context, rule } func (r *RuleReconciler) ensureFinalizer(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, finalizer string) (finalizerAdded bool, err error) { + log := ctrl.LoggerFrom(ctx) // Finalizers can only be added when the deletionTimestamp is not set. if !rule.GetDeletionTimestamp().IsZero() { return false, nil @@ -932,6 +966,11 @@ 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.APIConflicts.WithLabelValues(rule.Name, string(metrics.ConflictOperationFinalizerAdd)).Inc() + log.V(1).Info("Conflict adding finalizer to rule", + "rule", rule.Name, "operation", string(metrics.ConflictOperationFinalizerAdd)) + } return err } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index db418e48..56f3bbcd 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -27,9 +27,27 @@ import ( type FailureReason string const ( - FailureReasonEvaluationError FailureReason = "EvaluationError" - FailureReasonAddTaintError FailureReason = "AddTaintError" - FailureReasonRemoveTaintError FailureReason = "RemoveTaintError" + FailureReasonEvaluationError FailureReason = "EvaluationError" + FailureReasonAddTaintError FailureReason = "AddTaintError" + FailureReasonRemoveTaintError FailureReason = "RemoveTaintError" + FailureReasonAddTaintConflictExhausted FailureReason = "AddTaintConflictExhausted" + FailureReasonRemoveTaintConflictExhausted FailureReason = "RemoveTaintConflictExhausted" + FailureReasonStatusPatchError FailureReason = "StatusPatchError" + FailureReasonStatusPatchConflictExhausted FailureReason = "StatusPatchConflictExhausted" + FailureReasonRuleStatusRuleSweepConflictExhausted FailureReason = "RuleStatusRuleSweepConflictExhausted" +) + +// ConflictOperation identifies which optimistic-locked API write hit a 409. +type ConflictOperation string + +const ( + ConflictOperationAddTaint ConflictOperation = "add_taint" + ConflictOperationRemoveTaint ConflictOperation = "remove_taint" + ConflictOperationMarkBootstrapCompleted ConflictOperation = "mark_bootstrap_completed" + ConflictOperationFinalizerAdd ConflictOperation = "finalizer_add" + ConflictOperationFinalizerRemove ConflictOperation = "finalizer_remove" + ConflictOperationRuleStatusNodeWrite ConflictOperation = "rule_status_node_write" + ConflictOperationRuleStatusRuleSweep ConflictOperation = "rule_status_rule_sweep" ) // TaintOperation represents a taint operation. @@ -102,6 +120,15 @@ var ( []string{"rule", "reason"}, ) + // APIConflicts counts API write conflicts for each retry attempt. + APIConflicts = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "node_readiness_api_conflicts_total", + Help: "Total number of API write conflicts encountered per retry attempt", + }, + []string{"rule", "operation"}, + ) + // BootstrapCompleted tracks the number of nodes that have completed bootstrap. BootstrapCompleted = prometheus.NewCounterVec( prometheus.CounterOpts{ @@ -182,6 +209,7 @@ func init() { metrics.Registry.MustRegister(TaintOperations) metrics.Registry.MustRegister(EvaluationDuration) metrics.Registry.MustRegister(Failures) + metrics.Registry.MustRegister(APIConflicts) metrics.Registry.MustRegister(BootstrapCompleted) metrics.Registry.MustRegister(BootstrapDuration) metrics.Registry.MustRegister(ReconciliationLatency)