From 0a8a7ff5a27a756cfdfc461c4da81c4a23105f7c Mon Sep 17 00:00:00 2001 From: Rawad Hossain Date: Wed, 19 Aug 2026 18:43:46 +0600 Subject: [PATCH 1/3] add blocked nodes metric Signed-off-by: Rawad Hossain --- docs/book/src/operations/monitoring.md | 19 + internal/controller/collector_bench_test.go | 222 +++++++- .../nodereadinessrule_controller.go | 66 ++- internal/controller/rule_node_states_test.go | 537 ++++++++++++++++-- internal/metrics/collector.go | 60 +- internal/metrics/collector_test.go | 209 ++++++- 6 files changed, 1042 insertions(+), 71 deletions(-) diff --git a/docs/book/src/operations/monitoring.md b/docs/book/src/operations/monitoring.md index 2d568402..e4ad1a83 100644 --- a/docs/book/src/operations/monitoring.md +++ b/docs/book/src/operations/monitoring.md @@ -106,6 +106,25 @@ Number of nodes currently held or released by each `NodeReadinessRule`, collecte | `rule` | `NodeReadinessRule` name | Any non-dry-run rule name with a valid selector | | `state` | Whether matching nodes are still tainted by the rule or have had the taint removed | `held`, `released` | +### `node_readiness_blocked_nodes` + +*Available starting from the v0.6.0 release.* + +Number of currently-held nodes blocked by each unsatisfied condition, per `NodeReadinessRule`, collected at scrape time from the controller cache. + +| Property | Value | +| --- | --- | +| Type | `gauge` | +| Labels | `rule`, `condition` | +| Recorded when | Computed on each Prometheus scrape from the cached node list | + +#### Labels + +| Label | Description | Values | +| --- | --- | --- | +| `rule` | `NodeReadinessRule` name | Any non-dry-run rule name with a valid selector | +| `condition` | Condition type declared in `spec.conditions` | Any condition type declared by the rule | + ### `node_readiness_bootstrap_completed_total` Total number of nodes that have completed bootstrap. diff --git a/internal/controller/collector_bench_test.go b/internal/controller/collector_bench_test.go index 47c8e431..63409768 100644 --- a/internal/controller/collector_bench_test.go +++ b/internal/controller/collector_bench_test.go @@ -29,12 +29,11 @@ import ( ) // buildBenchController creates a fake client with the given nodes and rules for benchmarking. -func buildBenchController(b *testing.B, nodeCount, ruleCount int) *RuleReadinessController { +func buildBenchController(b *testing.B, nodeCount, ruleCount int) (*RuleReadinessController, []corev1.Node) { b.Helper() scheme := newTestScheme(b) - nodes := make([]client.Object, 0, nodeCount) - objs := make([]client.Object, 0, nodeCount+ruleCount) + objs := make([]client.Object, 0, ruleCount) rules := make(map[string]*readinessv1alpha1.NodeReadinessRule, ruleCount) for i := range ruleCount { @@ -55,27 +54,24 @@ func buildBenchController(b *testing.B, nodeCount, ruleCount int) *RuleReadiness objs = append(objs, rule) } + nodes := make([]corev1.Node, nodeCount) for i := range nodeCount { ruleIdx := i % ruleCount ruleName := fmt.Sprintf("rule-%d", ruleIdx) - node := &corev1.Node{ + nodes[i] = corev1.Node{ ObjectMeta: metav1.ObjectMeta{ Name: fmt.Sprintf("node-%d", i), Labels: map[string]string{"rule-index": fmt.Sprintf("%d", ruleIdx)}, }, } if i%3 == 0 { - node.Spec.Taints = []corev1.Taint{rules[ruleName].Spec.Taint} + nodes[i].Spec.Taints = []corev1.Taint{rules[ruleName].Spec.Taint} } - nodes = append(nodes, node) } - objs = append(objs, nodes...) fc := fakeclient.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build() - return &RuleReadinessController{ - Client: fc, - } + return &RuleReadinessController{Client: fc}, nodes } func BenchmarkListRuleNodeStates(b *testing.B) { @@ -85,13 +81,13 @@ func BenchmarkListRuleNodeStates(b *testing.B) { for _, nodeCount := range nodeCounts { for _, ruleCount := range ruleCounts { b.Run(fmt.Sprintf("nodes=%d/rules=%d", nodeCount, ruleCount), func(b *testing.B) { - c := buildBenchController(b, nodeCount, ruleCount) + c, nodes := buildBenchController(b, nodeCount, ruleCount) ctx := b.Context() b.ResetTimer() b.ReportAllocs() for range b.N { - if _, err := c.ListRuleNodeStates(ctx); err != nil { + if _, err := c.ListRuleNodeStates(ctx, nodes); err != nil { b.Fatalf("ListRuleNodeStates failed: %v", err) } } @@ -99,3 +95,205 @@ func BenchmarkListRuleNodeStates(b *testing.B) { } } } + +// buildBlockedNodesBenchController sets up rules and evaluations for benchmarking ListBlockedNodes. +func buildBlockedNodesBenchController(b *testing.B, nodeCount, ruleCount, conditionsPerRule int) (*RuleReadinessController, []corev1.Node) { + b.Helper() + + scheme := newTestScheme(b) + objs := make([]client.Object, 0, ruleCount) + + rules := make([]*readinessv1alpha1.NodeReadinessRule, 0, ruleCount) + for i := range ruleCount { + ruleName := fmt.Sprintf("rule-%d", i) + conditions := make([]readinessv1alpha1.ConditionRequirement, 0, conditionsPerRule) + for j := range conditionsPerRule { + conditions = append(conditions, readinessv1alpha1.ConditionRequirement{ + Type: fmt.Sprintf("Condition-%d", j), + RequiredStatus: corev1.ConditionTrue, + }) + } + rule := &readinessv1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: ruleName}, + Spec: readinessv1alpha1.NodeReadinessRuleSpec{ + NodeSelector: metav1.LabelSelector{ + MatchLabels: map[string]string{"rule-index": fmt.Sprintf("%d", i)}, + }, + Taint: corev1.Taint{ + Key: fmt.Sprintf("readiness.k8s.io/%s", ruleName), + Effect: corev1.TaintEffectNoSchedule, + }, + Conditions: conditions, + }, + } + rules = append(rules, rule) + objs = append(objs, rule) + } + + nodes := make([]corev1.Node, nodeCount) + for i := range nodeCount { + ruleIdx := i % ruleCount + rule := rules[ruleIdx] + + nodeConditions := make([]corev1.NodeCondition, 0, conditionsPerRule) + for k, cond := range rule.Spec.Conditions { + status := corev1.ConditionTrue + if k%2 == 0 { + status = corev1.ConditionFalse + } + nodeConditions = append(nodeConditions, corev1.NodeCondition{ + Type: corev1.NodeConditionType(cond.Type), + Status: status, + }) + } + + nodes[i] = corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("node-%d", i), + Labels: map[string]string{"rule-index": fmt.Sprintf("%d", ruleIdx)}, + }, + Status: corev1.NodeStatus{Conditions: nodeConditions}, + } + if i%3 == 0 { + nodes[i].Spec.Taints = []corev1.Taint{rule.Spec.Taint} + } + } + + fc := fakeclient.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build() + + return &RuleReadinessController{Client: fc}, nodes +} + +// BenchmarkListBlockedNodes measures the performance of ListBlockedNodes. +func BenchmarkListBlockedNodes(b *testing.B) { + nodeCounts := []int{100, 1000, 5000, 15000} + ruleCounts := []int{5, 20, 50} + conditionsPerRuleOptions := []int{1, 4, 8} + + for _, nodeCount := range nodeCounts { + for _, ruleCount := range ruleCounts { + for _, conditionsPerRule := range conditionsPerRuleOptions { + b.Run(fmt.Sprintf("nodes=%d/rules=%d/conditions=%d", nodeCount, ruleCount, conditionsPerRule), func(b *testing.B) { + c, nodes := buildBlockedNodesBenchController(b, nodeCount, ruleCount, conditionsPerRule) + ctx := b.Context() + + b.ResetTimer() + b.ReportAllocs() + for range b.N { + if _, err := c.ListBlockedNodes(ctx, nodes); err != nil { + b.Fatalf("ListBlockedNodes failed: %v", err) + } + } + }) + } + } + } +} + +// buildFullBenchController sets up rules and nodes for benchmarking the full collector path. +func buildFullBenchController(b *testing.B, nodeCount, ruleCount, conditionsPerRule int) *RuleReadinessController { + b.Helper() + + scheme := newTestScheme(b) + objs := make([]client.Object, 0, nodeCount+ruleCount) + + rules := make([]*readinessv1alpha1.NodeReadinessRule, 0, ruleCount) + for i := range ruleCount { + ruleName := fmt.Sprintf("rule-%d", i) + conditions := make([]readinessv1alpha1.ConditionRequirement, 0, conditionsPerRule) + for j := range conditionsPerRule { + conditions = append(conditions, readinessv1alpha1.ConditionRequirement{ + Type: fmt.Sprintf("Condition-%d", j), + RequiredStatus: corev1.ConditionTrue, + }) + } + rule := &readinessv1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: ruleName}, + Spec: readinessv1alpha1.NodeReadinessRuleSpec{ + NodeSelector: metav1.LabelSelector{ + MatchLabels: map[string]string{"rule-index": fmt.Sprintf("%d", i)}, + }, + Taint: corev1.Taint{ + Key: fmt.Sprintf("readiness.k8s.io/%s", ruleName), + Effect: corev1.TaintEffectNoSchedule, + }, + Conditions: conditions, + }, + } + rules = append(rules, rule) + } + + nodes := make([]corev1.Node, nodeCount) + for i := range nodeCount { + ruleIdx := i % ruleCount + rule := rules[ruleIdx] + nodeName := fmt.Sprintf("node-%d", i) + + nodeConditions := make([]corev1.NodeCondition, 0, conditionsPerRule) + for k, cond := range rule.Spec.Conditions { + status := corev1.ConditionTrue + if k%2 == 0 { + status = corev1.ConditionFalse + } + nodeConditions = append(nodeConditions, corev1.NodeCondition{ + Type: corev1.NodeConditionType(cond.Type), + Status: status, + }) + } + + nodes[i] = corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Labels: map[string]string{"rule-index": fmt.Sprintf("%d", ruleIdx)}, + }, + Status: corev1.NodeStatus{Conditions: nodeConditions}, + } + if i%3 == 0 { + nodes[i].Spec.Taints = []corev1.Taint{rule.Spec.Taint} + } + } + + for _, rule := range rules { + objs = append(objs, rule) + } + for i := range nodes { + objs = append(objs, &nodes[i]) + } + + fc := fakeclient.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build() + + return &RuleReadinessController{Client: fc} +} + +// BenchmarkCollectSharedNodeList measures the collector path with a shared Node list for ListRuleNodeStates and ListBlockedNodes. +func BenchmarkCollectSharedNodeList(b *testing.B) { + nodeCounts := []int{100, 1000, 5000, 15000} + ruleCounts := []int{5, 20, 50} + conditionsPerRuleOptions := []int{1, 4, 8} + + for _, nodeCount := range nodeCounts { + for _, ruleCount := range ruleCounts { + for _, conditionsPerRule := range conditionsPerRuleOptions { + b.Run(fmt.Sprintf("nodes=%d/rules=%d/conditions=%d", nodeCount, ruleCount, conditionsPerRule), func(b *testing.B) { + c := buildFullBenchController(b, nodeCount, ruleCount, conditionsPerRule) + ctx := b.Context() + + b.ResetTimer() + b.ReportAllocs() + for range b.N { + nodes, err := c.ListNodes(ctx) + if err != nil { + b.Fatalf("ListNodes failed: %v", err) + } + if _, err := c.ListRuleNodeStates(ctx, nodes); err != nil { + b.Fatalf("ListRuleNodeStates failed: %v", err) + } + if _, err := c.ListBlockedNodes(ctx, nodes); err != nil { + b.Fatalf("ListBlockedNodes failed: %v", err) + } + } + }) + } + } + } +} diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index 278d8d7a..a52ee53c 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -531,15 +531,19 @@ func (r *RuleReadinessController) getApplicableRulesForNode(ctx context.Context, return applicableRules } -// ListRuleNodeStates returns the number of held and released nodes for each rule. -func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context) (map[string]metrics.RuleNodeCounts, error) { - ruleList := &readinessv1alpha1.NodeReadinessRuleList{} - if err := r.List(ctx, ruleList); err != nil { +// ListNodes returns the current list of Nodes. +func (r *RuleReadinessController) ListNodes(ctx context.Context) ([]corev1.Node, error) { + nodeList := &corev1.NodeList{} + if err := r.List(ctx, nodeList); err != nil { return nil, err } + return nodeList.Items, nil +} - nodeList := &corev1.NodeList{} - if err := r.List(ctx, nodeList); err != nil { +// ListRuleNodeStates returns the number of held and released nodes for each rule. +func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context, nodes []corev1.Node) (map[string]metrics.RuleNodeCounts, error) { + ruleList := &readinessv1alpha1.NodeReadinessRuleList{} + if err := r.List(ctx, ruleList); err != nil { return nil, err } @@ -560,8 +564,8 @@ func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context) (map[s } rc := metrics.RuleNodeCounts{} - for i := range nodeList.Items { - node := &nodeList.Items[i] + for i := range nodes { + node := &nodes[i] if !selector.Matches(labels.Set(node.Labels)) { continue } @@ -577,6 +581,52 @@ func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context) (map[s return counts, nil } +// ListBlockedNodes returns the number of blocked nodes for each rule and unsatisfied condition. +func (r *RuleReadinessController) ListBlockedNodes(ctx context.Context, nodes []corev1.Node) (map[string]metrics.RuleBlockedConditions, error) { + ruleList := &readinessv1alpha1.NodeReadinessRuleList{} + if err := r.List(ctx, ruleList); err != nil { + return nil, err + } + + log := ctrl.LoggerFrom(ctx) + + result := make(map[string]metrics.RuleBlockedConditions, len(ruleList.Items)) + for i := range ruleList.Items { + rule := &ruleList.Items[i] + if rule.Spec.DryRun { + continue + } + + selector, err := metav1.LabelSelectorAsSelector(&rule.Spec.NodeSelector) + if err != nil { + log.V(2).Info("Invalid node selector for rule", "rule", rule.Name, "error", err) + continue + } + + counts := make(metrics.RuleBlockedConditions, len(rule.Spec.Conditions)) + for _, cond := range rule.Spec.Conditions { + counts[cond.Type] = 0 + } + + for i := range nodes { + node := &nodes[i] + if !selector.Matches(labels.Set(node.Labels)) || !r.hasTaintBySpec(node, rule.Spec.Taint) { + continue + } + for _, cond := range rule.Spec.Conditions { + effectiveStatus, _ := r.getConditionStatus(node, cond.Type, cond.GetDefaultStatus()) + if effectiveStatus != cond.RequiredStatus { + counts[cond.Type]++ + } + } + } + + result[rule.Name] = counts + } + + return result, nil +} + // ruleAppliesTo checks if a rule applies to a node. func (r *RuleReadinessController) ruleAppliesTo(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, node *corev1.Node) bool { log := ctrl.LoggerFrom(ctx) diff --git a/internal/controller/rule_node_states_test.go b/internal/controller/rule_node_states_test.go index 8c86070e..634a39c4 100644 --- a/internal/controller/rule_node_states_test.go +++ b/internal/controller/rule_node_states_test.go @@ -73,6 +73,66 @@ func gpuNode(name string, tainted bool) *corev1.Node { return n } +// gpuRuleWithConditions returns a GPU-ready rule with the given conditions. +func gpuRuleWithConditions(conditionTypes ...string) *readinessv1alpha1.NodeReadinessRule { + rule := gpuRule() + conditions := make([]readinessv1alpha1.ConditionRequirement, 0, len(conditionTypes)) + for _, ct := range conditionTypes { + conditions = append(conditions, readinessv1alpha1.ConditionRequirement{ + Type: ct, + RequiredStatus: corev1.ConditionTrue, + }) + } + rule.Spec.Conditions = conditions + return rule +} + +func gpuRuleWithConditionReqs(reqs ...readinessv1alpha1.ConditionRequirement) *readinessv1alpha1.NodeReadinessRule { + rule := gpuRule() + rule.Spec.Conditions = reqs + return rule +} + +func withNodeConditions(node *corev1.Node, conds ...corev1.NodeCondition) *corev1.Node { + node.Status.Conditions = conds + return node +} + +func nodeCondition(condType string, status corev1.ConditionStatus) corev1.NodeCondition { + return corev1.NodeCondition{Type: corev1.NodeConditionType(condType), Status: status} +} + +func TestListNodes_Empty(t *testing.T) { + g := NewWithT(t) + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).Build() + c := &RuleReadinessController{ + Client: fc, + } + + nodes, err := c.ListNodes(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(nodes).To(BeEmpty()) +} + +func TestListNodes_ReturnsAllCachedNodes(t *testing.T) { + g := NewWithT(t) + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( + gpuNode("node-a", true), + gpuNode("node-b", false), + ).Build() + c := &RuleReadinessController{ + Client: fc, + } + + nodes, err := c.ListNodes(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + names := make([]string, 0, len(nodes)) + for _, n := range nodes { + names = append(names, n.Name) + } + g.Expect(names).To(ConsistOf("node-a", "node-b")) +} + func TestListRuleNodeStates_NoRules(t *testing.T) { g := NewWithT(t) fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).Build() @@ -80,22 +140,22 @@ func TestListRuleNodeStates_NoRules(t *testing.T) { Client: fc, } - counts, err := c.ListRuleNodeStates(t.Context()) + counts, err := c.ListRuleNodeStates(t.Context(), nil) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(BeEmpty()) } func TestListRuleNodeStates_ZeroMatches(t *testing.T) { g := NewWithT(t) - fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( - gpuRule(), - &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "cpu-node"}}, - ).Build() + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(gpuRule()).Build() c := &RuleReadinessController{ Client: fc, } + nodes := []corev1.Node{ + {ObjectMeta: metav1.ObjectMeta{Name: "cpu-node"}}, + } - counts, err := c.ListRuleNodeStates(t.Context()) + counts, err := c.ListRuleNodeStates(t.Context(), nodes) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 0, Released: 0}, @@ -104,18 +164,18 @@ func TestListRuleNodeStates_ZeroMatches(t *testing.T) { func TestListRuleNodeStates_MixedHeldReleased(t *testing.T) { g := NewWithT(t) - fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( - gpuRule(), - gpuNode("held-1", true), - gpuNode("held-2", true), - gpuNode("released-1", false), - &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "non-matching"}}, - ).Build() + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(gpuRule()).Build() c := &RuleReadinessController{ Client: fc, } + nodes := []corev1.Node{ + *gpuNode("held-1", true), + *gpuNode("held-2", true), + *gpuNode("released-1", false), + {ObjectMeta: metav1.ObjectMeta{Name: "non-matching"}}, + } - counts, err := c.ListRuleNodeStates(t.Context()) + counts, err := c.ListRuleNodeStates(t.Context(), nodes) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 2, Released: 1}, @@ -126,15 +186,13 @@ func TestListRuleNodeStates_DryRunRuleExcluded(t *testing.T) { g := NewWithT(t) rule := gpuRule() rule.Spec.DryRun = true - fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( - rule, - gpuNode("held-1", true), - ).Build() + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() c := &RuleReadinessController{ Client: fc, } + nodes := []corev1.Node{*gpuNode("held-1", true)} - counts, err := c.ListRuleNodeStates(t.Context()) + counts, err := c.ListRuleNodeStates(t.Context(), nodes) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(BeEmpty()) } @@ -145,17 +203,17 @@ func TestListRuleNodeStates_DeletingRuleIncluded(t *testing.T) { now := metav1.Now() rule.DeletionTimestamp = &now rule.Finalizers = []string{"readiness.node.x-k8s.io/cleanup-taints"} - fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( - rule, - gpuNode("held-1", true), - gpuNode("held-2", true), - gpuNode("released-1", false), - ).Build() + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() c := &RuleReadinessController{ Client: fc, } + nodes := []corev1.Node{ + *gpuNode("held-1", true), + *gpuNode("held-2", true), + *gpuNode("released-1", false), + } - counts, err := c.ListRuleNodeStates(t.Context()) + counts, err := c.ListRuleNodeStates(t.Context(), nodes) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 2, Released: 1}, @@ -175,7 +233,7 @@ func TestListRuleNodeStates_DeletingRulePersistsUntilFinalizer(t *testing.T) { Client: fc, } - counts, err := c.ListRuleNodeStates(t.Context()) + counts, err := c.ListRuleNodeStates(t.Context(), nil) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 0, Released: 0}, @@ -203,24 +261,26 @@ func TestListRuleNodeStates_OneRuleHeldOtherReleased(t *testing.T) { } // Node matches both rules' selectors but only carries taintA, not taintB. - node := &corev1.Node{ - ObjectMeta: metav1.ObjectMeta{ - Name: "shared-node", - Labels: map[string]string{"gpu": "true"}, - }, - Spec: corev1.NodeSpec{ - Taints: []corev1.Taint{taintA}, + nodes := []corev1.Node{ + { + ObjectMeta: metav1.ObjectMeta{ + Name: "shared-node", + Labels: map[string]string{"gpu": "true"}, + }, + Spec: corev1.NodeSpec{ + Taints: []corev1.Taint{taintA}, + }, }, } fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( - ruleA, ruleB, node, + ruleA, ruleB, ).Build() c := &RuleReadinessController{ Client: fc, } - counts, err := c.ListRuleNodeStates(t.Context()) + counts, err := c.ListRuleNodeStates(t.Context(), nodes) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "rule-a": {Held: 1, Released: 0}, @@ -248,17 +308,416 @@ func TestListRuleNodeStates_InvalidSelectorSkipped(t *testing.T) { fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( validRule, invalidRule, - gpuNode("held-1", true), - gpuNode("released-1", false), ).Build() c := &RuleReadinessController{ Client: fc, } + nodes := []corev1.Node{ + *gpuNode("held-1", true), + *gpuNode("released-1", false), + } - counts, err := c.ListRuleNodeStates(t.Context()) + counts, err := c.ListRuleNodeStates(t.Context(), nodes) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).NotTo(HaveKey(invalidRule.Name)) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 1, Released: 1}, })) } + +func TestListBlockedNodes_ZeroSeededWhenNoHeldNodes(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady", "CNIReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + + blocked, err := c.ListBlockedNodes(t.Context(), nil) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 0, "CNIReady": 0}, + })) +} + +func TestListBlockedNodes_UnsatisfiedConditionCounted(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1}, + })) +} + +func TestListBlockedNodes_ReleasedNodeExcluded(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("released-1", false), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 0}, + })) +} + +func TestListBlockedNodes_MixedHeldReleased(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + *withNodeConditions(gpuNode("held-2", true), nodeCondition("GPUDriverReady", corev1.ConditionTrue)), + *withNodeConditions(gpuNode("released-1", false), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1}, + })) +} + +func TestListBlockedNodes_MultipleUnsatisfiedConditions(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady", "CNIReady", "DiskReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), + nodeCondition("GPUDriverReady", corev1.ConditionFalse), + nodeCondition("CNIReady", corev1.ConditionFalse), + nodeCondition("DiskReady", corev1.ConditionTrue), + ), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1, "CNIReady": 1, "DiskReady": 0}, + })) +} + +// TestListBlockedNodes_AnyOfNoConditionsSatisfied verifies that all conditions are counted when none are satisfied. +func TestListBlockedNodes_AnyOfNoConditionsSatisfied(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady", "CNIReady") + rule.Spec.ConditionPolicy = readinessv1alpha1.ConditionPolicyAnyOf + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), + nodeCondition("GPUDriverReady", corev1.ConditionFalse), + nodeCondition("CNIReady", corev1.ConditionFalse), + ), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1, "CNIReady": 1}, + })) +} + +// TestListBlockedNodes_AnyOfWithSatisfiedCondition verifies that only unsatisfied conditions are counted. +func TestListBlockedNodes_AnyOfWithSatisfiedCondition(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady", "CNIReady") + rule.Spec.ConditionPolicy = readinessv1alpha1.ConditionPolicyAnyOf + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), + nodeCondition("GPUDriverReady", corev1.ConditionFalse), + nodeCondition("CNIReady", corev1.ConditionTrue), + ), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1, "CNIReady": 0}, + })) +} + +// TestListBlockedNodes_SharedConditionAcrossRules verifies that the same condition type is counted independently for each rule. +func TestListBlockedNodes_SharedConditionAcrossRules(t *testing.T) { + g := NewWithT(t) + taintA := corev1.Taint{Key: "readiness.k8s.io/rule-a", Effect: corev1.TaintEffectNoSchedule} + taintB := corev1.Taint{Key: "readiness.k8s.io/rule-b", Effect: corev1.TaintEffectNoSchedule} + + ruleA := &readinessv1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "rule-a"}, + Spec: readinessv1alpha1.NodeReadinessRuleSpec{ + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"gpu": "true"}}, + Taint: taintA, + Conditions: []readinessv1alpha1.ConditionRequirement{ + {Type: "GPUDriverReady", RequiredStatus: corev1.ConditionTrue}, + }, + }, + } + ruleB := &readinessv1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "rule-b"}, + Spec: readinessv1alpha1.NodeReadinessRuleSpec{ + NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"gpu": "true"}}, + Taint: taintB, + Conditions: []readinessv1alpha1.ConditionRequirement{ + {Type: "GPUDriverReady", RequiredStatus: corev1.ConditionFalse}, + }, + }, + } + + nodes := []corev1.Node{ + *withNodeConditions(&corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "shared-node", + Labels: map[string]string{"gpu": "true"}, + }, + Spec: corev1.NodeSpec{ + Taints: []corev1.Taint{taintA, taintB}, + }, + }, nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + } + + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( + ruleA, ruleB, + ).Build() + c := &RuleReadinessController{ + Client: fc, + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "rule-a": {"GPUDriverReady": 1}, + "rule-b": {"GPUDriverReady": 0}, + })) +} + +func TestListBlockedNodes_DryRunRuleExcluded(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady") + rule.Spec.DryRun = true + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(BeEmpty()) +} + +func TestListBlockedNodes_DeletingRuleIncluded(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady") + now := metav1.Now() + rule.DeletionTimestamp = &now + rule.Finalizers = []string{"readiness.node.x-k8s.io/cleanup-taints"} + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + *withNodeConditions(gpuNode("held-2", true), nodeCondition("GPUDriverReady", corev1.ConditionTrue)), + *withNodeConditions(gpuNode("released-1", false), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1}, + })) +} + +func TestListBlockedNodes_NonMatchingNodeExcluded(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(&corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "drifted", + Labels: map[string]string{"gpu": "false"}, + }, + Spec: corev1.NodeSpec{Taints: []corev1.Taint{gpuTaint()}}, + }, nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 0}, + })) +} + +func TestListBlockedNodes_InvalidSelectorSkipped(t *testing.T) { + g := NewWithT(t) + validRule := gpuRuleWithConditions("GPUDriverReady") + invalidRule := &readinessv1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "invalid-selector-rule"}, + Spec: readinessv1alpha1.NodeReadinessRuleSpec{ + NodeSelector: metav1.LabelSelector{ + MatchExpressions: []metav1.LabelSelectorRequirement{ + {Key: "gpu", Operator: "BogusOperator", Values: []string{"true"}}, + }, + }, + Taint: corev1.Taint{ + Key: "readiness.k8s.io/invalid-selector", + Effect: corev1.TaintEffectNoSchedule, + }, + Conditions: []readinessv1alpha1.ConditionRequirement{ + {Type: "SomeCondition", RequiredStatus: corev1.ConditionTrue}, + }, + }, + } + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( + validRule, + invalidRule, + ).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).NotTo(HaveKey(invalidRule.Name)) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1}, + })) +} + +func TestListBlockedNodes_ZeroDeclaredConditions(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions() + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodes := []corev1.Node{*gpuNode("held-1", true)} + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {}, + })) +} + +func TestListBlockedNodes_NoRules(t *testing.T) { + g := NewWithT(t) + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).Build() + c := &RuleReadinessController{ + Client: fc, + } + + blocked, err := c.ListBlockedNodes(t.Context(), nil) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(BeEmpty()) +} + +// TestListBlockedNodes_DefaultStatusSatisfies verifies that a default status can satisfy an absent condition. +func TestListBlockedNodes_DefaultStatusSatisfies(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditionReqs( + readinessv1alpha1.ConditionRequirement{Type: "CondA", RequiredStatus: corev1.ConditionTrue}, + readinessv1alpha1.ConditionRequirement{Type: "CondB", RequiredStatus: corev1.ConditionTrue, DefaultStatus: corev1.ConditionTrue}, + ) + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + + nodes := []corev1.Node{ + *withNodeConditions(gpuNode("held-1", true), nodeCondition("CondA", corev1.ConditionFalse)), + } + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"CondA": 1, "CondB": 0}, + })) +} + +// TestListBlockedNodes_AbsentConditionCounted verifies that an absent condition without a default status is counted as blocking. +func TestListBlockedNodes_AbsentConditionCounted(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditionReqs( + readinessv1alpha1.ConditionRequirement{Type: "CondB", RequiredStatus: corev1.ConditionTrue}, + ) + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(rule).Build() + c := &RuleReadinessController{ + Client: fc, + } + + nodes := []corev1.Node{*gpuNode("held-1", true)} + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"CondB": 1}, + })) +} + +// TestSharedNodeSnapshot_BothListersAgree verifies that both listers use the same Node snapshot. +func TestSharedNodeSnapshot_BothListersAgree(t *testing.T) { + g := NewWithT(t) + rule := gpuRuleWithConditions("GPUDriverReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( + rule, + withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + gpuNode("released-1", false), + ).Build() + c := &RuleReadinessController{ + Client: fc, + } + + nodes, err := c.ListNodes(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(nodes).To(HaveLen(2)) + + ruleCounts, err := c.ListRuleNodeStates(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(ruleCounts).To(Equal(map[string]metrics.RuleNodeCounts{ + "gpu-ready": {Held: 1, Released: 1}, + })) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1}, + })) +} diff --git a/internal/metrics/collector.go b/internal/metrics/collector.go index 60d0aee0..70b33b76 100644 --- a/internal/metrics/collector.go +++ b/internal/metrics/collector.go @@ -21,12 +21,18 @@ import ( "time" "github.com/prometheus/client_golang/prometheus" + corev1 "k8s.io/api/core/v1" ctrl "sigs.k8s.io/controller-runtime" ) // collectTimeout limits how long a scrape can wait for cached data. const collectTimeout = 5 * time.Second +// NodeLister lists Nodes for the collector. +type NodeLister interface { + ListNodes(ctx context.Context) ([]corev1.Node, error) +} + // RuleNodeCounts holds the number of held and released nodes for a rule. type RuleNodeCounts struct { Held float64 @@ -35,7 +41,22 @@ type RuleNodeCounts struct { // RuleNodeStateLister lists held and released nodes for each rule. type RuleNodeStateLister interface { - ListRuleNodeStates(ctx context.Context) (map[string]RuleNodeCounts, error) + ListRuleNodeStates(ctx context.Context, nodes []corev1.Node) (map[string]RuleNodeCounts, error) +} + +// RuleBlockedConditions holds blocked node counts by condition. +type RuleBlockedConditions map[string]float64 + +// BlockedNodesLister lists blocked node counts for each rule and condition. +type BlockedNodesLister interface { + ListBlockedNodes(ctx context.Context, nodes []corev1.Node) (map[string]RuleBlockedConditions, error) +} + +// ReadinessLister aggregates the scrape-time lookups the collector needs. +type ReadinessLister interface { + NodeLister + RuleNodeStateLister + BlockedNodesLister } var ruleNodesDesc = prometheus.NewDesc( @@ -45,18 +66,26 @@ var ruleNodesDesc = prometheus.NewDesc( nil, ) +var blockedNodesDesc = prometheus.NewDesc( + "node_readiness_blocked_nodes", + "Number of nodes blocked by each required condition.", + []string{"rule", "condition"}, + nil, +) + // ReadinessCollector is a prometheus.Collector that reads at scrape time. type ReadinessCollector struct { - lister RuleNodeStateLister + lister ReadinessLister } -func NewReadinessCollector(lister RuleNodeStateLister) *ReadinessCollector { +func NewReadinessCollector(lister ReadinessLister) *ReadinessCollector { return &ReadinessCollector{lister: lister} } // Describe implements prometheus.Collector. func (c *ReadinessCollector) Describe(ch chan<- *prometheus.Desc) { ch <- ruleNodesDesc + ch <- blockedNodesDesc } // Collect implements prometheus.Collector. @@ -64,14 +93,31 @@ func (c *ReadinessCollector) Collect(ch chan<- prometheus.Metric) { ctx, cancel := context.WithTimeout(context.Background(), collectTimeout) defer cancel() - counts, err := c.lister.ListRuleNodeStates(ctx) + nodes, err := c.lister.ListNodes(ctx) + if err != nil { + ctrl.Log.V(2).Info("Failed to list nodes", "error", err) + return + } + + counts, err := c.lister.ListRuleNodeStates(ctx, nodes) if err != nil { ctrl.Log.V(2).Info("Failed to list rule node states", "error", err) + } else { + for rule, rc := range counts { + ch <- prometheus.MustNewConstMetric(ruleNodesDesc, prometheus.GaugeValue, rc.Held, rule, string(RuleNodeStateHeld)) + ch <- prometheus.MustNewConstMetric(ruleNodesDesc, prometheus.GaugeValue, rc.Released, rule, string(RuleNodeStateReleased)) + } + } + + blocked, err := c.lister.ListBlockedNodes(ctx, nodes) + if err != nil { + ctrl.Log.V(2).Info("Failed to list blocked nodes", "error", err) return } - for rule, rc := range counts { - ch <- prometheus.MustNewConstMetric(ruleNodesDesc, prometheus.GaugeValue, rc.Held, rule, string(RuleNodeStateHeld)) - ch <- prometheus.MustNewConstMetric(ruleNodesDesc, prometheus.GaugeValue, rc.Released, rule, string(RuleNodeStateReleased)) + for rule, conditions := range blocked { + for condition, count := range conditions { + ch <- prometheus.MustNewConstMetric(blockedNodesDesc, prometheus.GaugeValue, count, rule, condition) + } } } diff --git a/internal/metrics/collector_test.go b/internal/metrics/collector_test.go index 39d5df5d..d2824c5a 100644 --- a/internal/metrics/collector_test.go +++ b/internal/metrics/collector_test.go @@ -25,21 +25,54 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" + dto "github.com/prometheus/client_model/go" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) // stubLister is a test double for RuleNodeStateLister. type stubLister struct { + nodes []corev1.Node + nodesErr error + counts map[string]RuleNodeCounts err error + + blocked map[string]RuleBlockedConditions + blockedErr error + + mu sync.Mutex + gotNodesForRuleStates []corev1.Node + gotNodesForBlocked []corev1.Node +} + +func (s *stubLister) ListNodes(_ context.Context) ([]corev1.Node, error) { + if s.nodesErr != nil { + return nil, s.nodesErr + } + return s.nodes, nil } -func (s *stubLister) ListRuleNodeStates(_ context.Context) (map[string]RuleNodeCounts, error) { +func (s *stubLister) ListRuleNodeStates(_ context.Context, nodes []corev1.Node) (map[string]RuleNodeCounts, error) { + s.mu.Lock() + s.gotNodesForRuleStates = nodes + s.mu.Unlock() if s.err != nil { return nil, s.err } return s.counts, nil } +func (s *stubLister) ListBlockedNodes(_ context.Context, nodes []corev1.Node) (map[string]RuleBlockedConditions, error) { + s.mu.Lock() + s.gotNodesForBlocked = nodes + s.mu.Unlock() + if s.blockedErr != nil { + return nil, s.blockedErr + } + return s.blocked, nil +} + func TestReadinessCollector_NoRules(t *testing.T) { c := NewReadinessCollector(&stubLister{counts: map[string]RuleNodeCounts{}}) @@ -109,18 +142,184 @@ func TestReadinessCollector_ListError(t *testing.T) { } } -func TestReadinessCollector_ConcurrentCollect(t *testing.T) { - c := NewReadinessCollector(&stubLister{counts: map[string]RuleNodeCounts{ - "gpu-ready": {Held: 3, Released: 7}, +func TestReadinessCollector_BlockedNodes_NoRules(t *testing.T) { + c := NewReadinessCollector(&stubLister{blocked: map[string]RuleBlockedConditions{}}) + + expected := `` + if err := testutil.CollectAndCompare(c, strings.NewReader(expected), "node_readiness_blocked_nodes"); err != nil { + t.Fatalf("unexpected collect mismatch: %v", err) + } +} + +func TestReadinessCollector_BlockedNodes_MixedAndZero(t *testing.T) { + c := NewReadinessCollector(&stubLister{blocked: map[string]RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 2, "CNIReady": 0}, }}) + expected := ` + # HELP node_readiness_blocked_nodes Number of nodes blocked by each required condition. + # TYPE node_readiness_blocked_nodes gauge + node_readiness_blocked_nodes{condition="GPUDriverReady",rule="gpu-ready"} 2 + node_readiness_blocked_nodes{condition="CNIReady",rule="gpu-ready"} 0 + ` + if err := testutil.CollectAndCompare(c, strings.NewReader(expected), "node_readiness_blocked_nodes"); err != nil { + t.Fatalf("unexpected collect mismatch: %v", err) + } +} + +func TestReadinessCollector_BlockedNodes_ListError(t *testing.T) { + c := NewReadinessCollector(&stubLister{ + counts: map[string]RuleNodeCounts{"gpu-ready": {Held: 1, Released: 0}}, + blockedErr: errors.New("cache not synced"), + }) + + expected := `` + if err := testutil.CollectAndCompare(c, strings.NewReader(expected), "node_readiness_blocked_nodes"); err != nil { + t.Fatalf("unexpected collect mismatch: %v", err) + } +} + +func collectAll(t *testing.T, c *ReadinessCollector) map[string][]*dto.Metric { + t.Helper() + + ch := make(chan prometheus.Metric, 16) + c.Collect(ch) + close(ch) + + out := make(map[string][]*dto.Metric) + for m := range ch { + name := "node_readiness_rule_nodes" + if m.Desc() == blockedNodesDesc { + name = "node_readiness_blocked_nodes" + } + + pb := &dto.Metric{} + if err := m.Write(pb); err != nil { + t.Fatalf("unexpected error writing metric %s: %v", name, err) + } + out[name] = append(out[name], pb) + } + return out +} + +func TestReadinessCollector_RuleNodesErrorDoesNotBlockBlockedNodes(t *testing.T) { + c := NewReadinessCollector(&stubLister{ + err: errors.New("cache not synced"), + blocked: map[string]RuleBlockedConditions{"gpu-ready": {"GPUDriverReady": 2}}, + }) + + got := collectAll(t, c) + + if len(got["node_readiness_rule_nodes"]) != 0 { + t.Fatalf("expected no node_readiness_rule_nodes metrics when ListRuleNodeStates fails, got %v", got["node_readiness_rule_nodes"]) + } + + blockedMetrics := got["node_readiness_blocked_nodes"] + if len(blockedMetrics) != 1 { + t.Fatalf("expected node_readiness_blocked_nodes to still be emitted despite ListRuleNodeStates failing, got %v", blockedMetrics) + } + if got, want := blockedMetrics[0].GetGauge().GetValue(), 2.0; got != want { + t.Fatalf("blocked_nodes value = %v, want %v", got, want) + } +} + +func TestReadinessCollector_BlockedNodesErrorDoesNotBlockRuleNodes(t *testing.T) { + c := NewReadinessCollector(&stubLister{ + counts: map[string]RuleNodeCounts{"gpu-ready": {Held: 3, Released: 1}}, + blockedErr: errors.New("cache not synced"), + }) + + got := collectAll(t, c) + + ruleNodesMetrics := got["node_readiness_rule_nodes"] + if len(ruleNodesMetrics) != 2 { + t.Fatalf("expected node_readiness_rule_nodes (held+released) to still be emitted despite ListBlockedNodes failing, got %v", ruleNodesMetrics) + } + + if len(got["node_readiness_blocked_nodes"]) != 0 { + t.Fatalf("expected no node_readiness_blocked_nodes metrics when ListBlockedNodes fails, got %v", got["node_readiness_blocked_nodes"]) + } +} + +func TestReadinessCollector_NodeListErrorSkipsBothMetrics(t *testing.T) { + stub := &stubLister{ + nodesErr: errors.New("node cache not synced"), + counts: map[string]RuleNodeCounts{"gpu-ready": {Held: 1, Released: 1}}, + blocked: map[string]RuleBlockedConditions{"gpu-ready": {"GPUDriverReady": 1}}, + } + c := NewReadinessCollector(stub) + + got := collectAll(t, c) + + if len(got["node_readiness_rule_nodes"]) != 0 { + t.Fatalf("expected no node_readiness_rule_nodes metrics when ListNodes fails, got %v", got["node_readiness_rule_nodes"]) + } + if len(got["node_readiness_blocked_nodes"]) != 0 { + t.Fatalf("expected no node_readiness_blocked_nodes metrics when ListNodes fails, got %v", got["node_readiness_blocked_nodes"]) + } + + if stub.gotNodesForRuleStates != nil || stub.gotNodesForBlocked != nil { + t.Fatalf("expected Collect to short-circuit before calling either counting method, but ListRuleNodeStates got %v, ListBlockedNodes got %v", + stub.gotNodesForRuleStates, stub.gotNodesForBlocked) + } +} + +func TestReadinessCollector_NodesSharedBetweenBothListers(t *testing.T) { + nodes := []corev1.Node{{ObjectMeta: metav1.ObjectMeta{Name: "node-a"}}} + stub := &stubLister{ + nodes: nodes, + counts: map[string]RuleNodeCounts{}, + blocked: map[string]RuleBlockedConditions{}, + } + c := NewReadinessCollector(stub) + + ch := make(chan prometheus.Metric, 4) + c.Collect(ch) + close(ch) + for range ch { + } + + if len(stub.gotNodesForRuleStates) != 1 || stub.gotNodesForRuleStates[0].Name != "node-a" { + t.Fatalf("ListRuleNodeStates did not receive the shared node snapshot: %v", stub.gotNodesForRuleStates) + } + if len(stub.gotNodesForBlocked) != 1 || stub.gotNodesForBlocked[0].Name != "node-a" { + t.Fatalf("ListBlockedNodes did not receive the shared node snapshot: %v", stub.gotNodesForBlocked) + } +} + +func TestReadinessCollector_CollectAndLint(t *testing.T) { + c := NewReadinessCollector(&stubLister{ + nodes: []corev1.Node{{}}, + counts: map[string]RuleNodeCounts{ + "gpu-ready": {Held: 3, Released: 7}, + }, + blocked: map[string]RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 2, "CNIReady": 0}, + }, + }) + + problems, err := testutil.CollectAndLint(c) + if err != nil { + t.Fatalf("CollectAndLint error: %v", err) + } + for _, p := range problems { + t.Errorf("lint problem: metric=%s text=%s", p.Metric, p.Text) + } +} + +func TestReadinessCollector_ConcurrentCollect(t *testing.T) { + c := NewReadinessCollector(&stubLister{ + counts: map[string]RuleNodeCounts{"gpu-ready": {Held: 3, Released: 7}}, + blocked: map[string]RuleBlockedConditions{"gpu-ready": {"GPUDriverReady": 3}}, + }) + var wg sync.WaitGroup for range 50 { wg.Add(1) go func() { defer wg.Done() // Exercise concurrent Collect calls for race detection. - ch := make(chan prometheus.Metric, 2) + ch := make(chan prometheus.Metric, 4) done := make(chan struct{}) go func() { for range ch { From 367985d999d4ccbcc5e8b7a890a1b24252d614f7 Mon Sep 17 00:00:00 2001 From: Rawad Hossain Date: Fri, 21 Aug 2026 00:39:40 +0600 Subject: [PATCH 2/3] simplify node rule checks --- .../nodereadinessrule_controller.go | 118 +++++++++++------- internal/controller/rule_node_states_test.go | 6 +- internal/metrics/collector.go | 11 +- 3 files changed, 82 insertions(+), 53 deletions(-) diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index a52ee53c..d2a6e1ab 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -540,42 +540,72 @@ func (r *RuleReadinessController) ListNodes(ctx context.Context) ([]corev1.Node, return nodeList.Items, nil } -// ListRuleNodeStates returns the number of held and released nodes for each rule. -func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context, nodes []corev1.Node) (map[string]metrics.RuleNodeCounts, error) { +// forEachRuleNode applies callbacks to nodes matching each rule. +func (r *RuleReadinessController) forEachRuleNode( + ctx context.Context, + nodes []corev1.Node, + skipRule func(rule *readinessv1alpha1.NodeReadinessRule) bool, + onRule func(rule *readinessv1alpha1.NodeReadinessRule), + onNode func(rule *readinessv1alpha1.NodeReadinessRule, node *corev1.Node, held bool), +) error { ruleList := &readinessv1alpha1.NodeReadinessRuleList{} if err := r.List(ctx, ruleList); err != nil { - return nil, err + return err } log := ctrl.LoggerFrom(ctx) - counts := make(map[string]metrics.RuleNodeCounts, len(ruleList.Items)) for i := range ruleList.Items { rule := &ruleList.Items[i] if rule.Spec.DryRun { continue } + if skipRule(rule) { + continue + } // Parse the selector once per rule. - selector, err := metav1.LabelSelectorAsSelector(&rule.Spec.NodeSelector) + selector, err := parseNodeSelector(rule) if err != nil { log.V(2).Info("Invalid node selector for rule", "rule", rule.Name, "error", err) continue } - rc := metrics.RuleNodeCounts{} + onRule(rule) + for i := range nodes { node := &nodes[i] if !selector.Matches(labels.Set(node.Labels)) { continue } - if r.hasTaintBySpec(node, rule.Spec.Taint) { + onNode(rule, node, r.hasTaintBySpec(node, rule.Spec.Taint)) + } + } + + return nil +} + +// ListRuleNodeStates returns the number of held and released nodes for each rule. +func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context, nodes []corev1.Node) (map[string]metrics.RuleNodeCounts, error) { + counts := make(map[string]metrics.RuleNodeCounts) + + err := r.forEachRuleNode(ctx, nodes, + func(rule *readinessv1alpha1.NodeReadinessRule) bool { return false }, + func(rule *readinessv1alpha1.NodeReadinessRule) { + counts[rule.Name] = metrics.RuleNodeCounts{} + }, + func(rule *readinessv1alpha1.NodeReadinessRule, node *corev1.Node, held bool) { + rc := counts[rule.Name] + if held { rc.Held++ } else { rc.Released++ } - } - counts[rule.Name] = rc + counts[rule.Name] = rc + }, + ) + if err != nil { + return nil, err } return counts, nil @@ -583,55 +613,47 @@ func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context, nodes // ListBlockedNodes returns the number of blocked nodes for each rule and unsatisfied condition. func (r *RuleReadinessController) ListBlockedNodes(ctx context.Context, nodes []corev1.Node) (map[string]metrics.RuleBlockedConditions, error) { - ruleList := &readinessv1alpha1.NodeReadinessRuleList{} - if err := r.List(ctx, ruleList); err != nil { - return nil, err - } - - log := ctrl.LoggerFrom(ctx) + result := make(map[string]metrics.RuleBlockedConditions) - result := make(map[string]metrics.RuleBlockedConditions, len(ruleList.Items)) - for i := range ruleList.Items { - rule := &ruleList.Items[i] - if rule.Spec.DryRun { - continue - } - - selector, err := metav1.LabelSelectorAsSelector(&rule.Spec.NodeSelector) - if err != nil { - log.V(2).Info("Invalid node selector for rule", "rule", rule.Name, "error", err) - continue - } - - counts := make(metrics.RuleBlockedConditions, len(rule.Spec.Conditions)) - for _, cond := range rule.Spec.Conditions { - counts[cond.Type] = 0 - } - - for i := range nodes { - node := &nodes[i] - if !selector.Matches(labels.Set(node.Labels)) || !r.hasTaintBySpec(node, rule.Spec.Taint) { - continue + err := r.forEachRuleNode(ctx, nodes, + func(rule *readinessv1alpha1.NodeReadinessRule) bool { return !rule.DeletionTimestamp.IsZero() }, + func(rule *readinessv1alpha1.NodeReadinessRule) { + counts := make(metrics.RuleBlockedConditions, len(rule.Spec.Conditions)) + for _, cond := range rule.Spec.Conditions { + counts[cond.Type] = 0 + } + result[rule.Name] = counts + }, + func(rule *readinessv1alpha1.NodeReadinessRule, node *corev1.Node, held bool) { + if !held { + return } + counts := result[rule.Name] for _, cond := range rule.Spec.Conditions { effectiveStatus, _ := r.getConditionStatus(node, cond.Type, cond.GetDefaultStatus()) if effectiveStatus != cond.RequiredStatus { counts[cond.Type]++ } } - } - - result[rule.Name] = counts + }, + ) + if err != nil { + return nil, err } return result, nil } +// parseNodeSelector parses a rule's NodeSelector into a labels.Selector. +func parseNodeSelector(rule *readinessv1alpha1.NodeReadinessRule) (labels.Selector, error) { + return metav1.LabelSelectorAsSelector(&rule.Spec.NodeSelector) +} + // ruleAppliesTo checks if a rule applies to a node. func (r *RuleReadinessController) ruleAppliesTo(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, node *corev1.Node) bool { log := ctrl.LoggerFrom(ctx) - selector, err := metav1.LabelSelectorAsSelector(&rule.Spec.NodeSelector) + selector, err := parseNodeSelector(rule) if err != nil { log.Error(err, "Invalid node selector for rule", "rule", rule.Name) return false @@ -640,6 +662,15 @@ func (r *RuleReadinessController) ruleAppliesTo(ctx context.Context, rule *readi return selector.Matches(labels.Set(node.Labels)) } +// checks if a rule applies to a node and has its taint. +func (r *RuleReadinessController) ruleAppliesToWithTaint(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, node *corev1.Node) (applies, held bool) { + applies = r.ruleAppliesTo(ctx, rule, node) + if !applies { + return false, false + } + return true, r.hasTaintBySpec(node, rule.Spec.Taint) +} + // updateRuleCache updates the rule cache. func (r *RuleReadinessController) updateRuleCache(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule) { log := ctrl.LoggerFrom(ctx) @@ -788,12 +819,13 @@ func (r *RuleReadinessController) cleanupTaintsForRule(ctx context.Context, rule var errors []string for _, node := range nodeList.Items { - if !r.ruleAppliesTo(ctx, rule, &node) { + applies, held := r.ruleAppliesToWithTaint(ctx, rule, &node) + if !applies { continue } // Check if node has the taint managed by this rule - if r.hasTaintBySpec(&node, rule.Spec.Taint) { + if held { log.Info("Removing taint from node during rule cleanup", "node", node.Name, "rule", rule.Name, diff --git a/internal/controller/rule_node_states_test.go b/internal/controller/rule_node_states_test.go index 634a39c4..d3e10596 100644 --- a/internal/controller/rule_node_states_test.go +++ b/internal/controller/rule_node_states_test.go @@ -535,7 +535,7 @@ func TestListBlockedNodes_DryRunRuleExcluded(t *testing.T) { g.Expect(blocked).To(BeEmpty()) } -func TestListBlockedNodes_DeletingRuleIncluded(t *testing.T) { +func TestListBlockedNodes_DeletingRuleExcluded(t *testing.T) { g := NewWithT(t) rule := gpuRuleWithConditions("GPUDriverReady") now := metav1.Now() @@ -553,9 +553,7 @@ func TestListBlockedNodes_DeletingRuleIncluded(t *testing.T) { blocked, err := c.ListBlockedNodes(t.Context(), nodes) g.Expect(err).NotTo(HaveOccurred()) - g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ - "gpu-ready": {"GPUDriverReady": 1}, - })) + g.Expect(blocked).To(BeEmpty()) } func TestListBlockedNodes_NonMatchingNodeExcluded(t *testing.T) { diff --git a/internal/metrics/collector.go b/internal/metrics/collector.go index 70b33b76..2fa9fdb9 100644 --- a/internal/metrics/collector.go +++ b/internal/metrics/collector.go @@ -112,12 +112,11 @@ func (c *ReadinessCollector) Collect(ch chan<- prometheus.Metric) { blocked, err := c.lister.ListBlockedNodes(ctx, nodes) if err != nil { ctrl.Log.V(2).Info("Failed to list blocked nodes", "error", err) - return - } - - for rule, conditions := range blocked { - for condition, count := range conditions { - ch <- prometheus.MustNewConstMetric(blockedNodesDesc, prometheus.GaugeValue, count, rule, condition) + } else { + for rule, conditions := range blocked { + for condition, count := range conditions { + ch <- prometheus.MustNewConstMetric(blockedNodesDesc, prometheus.GaugeValue, count, rule, condition) + } } } } From 2745dc4ba0f66a267467ef93b4c8b32189e787f1 Mon Sep 17 00:00:00 2001 From: Rawad Hossain Date: Sun, 23 Aug 2026 18:42:35 +0600 Subject: [PATCH 3/3] reuse rule snapshot in collector --- docs/book/src/operations/monitoring.md | 4 +- internal/controller/collector_bench_test.go | 20 ++- .../controller/collector_shared_rules_test.go | 142 +++++++++++++++ .../nodereadinessrule_controller.go | 54 +++--- internal/controller/rule_node_states_test.go | 161 ++++++++++++++---- internal/metrics/collector.go | 24 ++- internal/metrics/collector_test.go | 66 ++++++- 7 files changed, 405 insertions(+), 66 deletions(-) create mode 100644 internal/controller/collector_shared_rules_test.go diff --git a/docs/book/src/operations/monitoring.md b/docs/book/src/operations/monitoring.md index e4ad1a83..fe2b6d1b 100644 --- a/docs/book/src/operations/monitoring.md +++ b/docs/book/src/operations/monitoring.md @@ -110,7 +110,7 @@ Number of nodes currently held or released by each `NodeReadinessRule`, collecte *Available starting from the v0.6.0 release.* -Number of currently-held nodes blocked by each unsatisfied condition, per `NodeReadinessRule`, collected at scrape time from the controller cache. +Number of currently-held nodes against blocking conditions per `NodeReadinessRule`. | Property | Value | | --- | --- | @@ -122,7 +122,7 @@ Number of currently-held nodes blocked by each unsatisfied condition, per `NodeR | Label | Description | Values | | --- | --- | --- | -| `rule` | `NodeReadinessRule` name | Any non-dry-run rule name with a valid selector | +| `rule` | `NodeReadinessRule` name | Any non-dry-run rule name | | `condition` | Condition type declared in `spec.conditions` | Any condition type declared by the rule | ### `node_readiness_bootstrap_completed_total` diff --git a/internal/controller/collector_bench_test.go b/internal/controller/collector_bench_test.go index 63409768..3aa296b5 100644 --- a/internal/controller/collector_bench_test.go +++ b/internal/controller/collector_bench_test.go @@ -83,11 +83,15 @@ func BenchmarkListRuleNodeStates(b *testing.B) { b.Run(fmt.Sprintf("nodes=%d/rules=%d", nodeCount, ruleCount), func(b *testing.B) { c, nodes := buildBenchController(b, nodeCount, ruleCount) ctx := b.Context() + rules, err := c.ListRules(ctx) + if err != nil { + b.Fatalf("ListRules failed: %v", err) + } b.ResetTimer() b.ReportAllocs() for range b.N { - if _, err := c.ListRuleNodeStates(ctx, nodes); err != nil { + if _, err := c.ListRuleNodeStates(ctx, nodes, rules); err != nil { b.Fatalf("ListRuleNodeStates failed: %v", err) } } @@ -176,11 +180,15 @@ func BenchmarkListBlockedNodes(b *testing.B) { b.Run(fmt.Sprintf("nodes=%d/rules=%d/conditions=%d", nodeCount, ruleCount, conditionsPerRule), func(b *testing.B) { c, nodes := buildBlockedNodesBenchController(b, nodeCount, ruleCount, conditionsPerRule) ctx := b.Context() + rules, err := c.ListRules(ctx) + if err != nil { + b.Fatalf("ListRules failed: %v", err) + } b.ResetTimer() b.ReportAllocs() for range b.N { - if _, err := c.ListBlockedNodes(ctx, nodes); err != nil { + if _, err := c.ListBlockedNodes(ctx, nodes, rules); err != nil { b.Fatalf("ListBlockedNodes failed: %v", err) } } @@ -285,10 +293,14 @@ func BenchmarkCollectSharedNodeList(b *testing.B) { if err != nil { b.Fatalf("ListNodes failed: %v", err) } - if _, err := c.ListRuleNodeStates(ctx, nodes); err != nil { + rules, err := c.ListRules(ctx) + if err != nil { + b.Fatalf("ListRules failed: %v", err) + } + if _, err := c.ListRuleNodeStates(ctx, nodes, rules); err != nil { b.Fatalf("ListRuleNodeStates failed: %v", err) } - if _, err := c.ListBlockedNodes(ctx, nodes); err != nil { + if _, err := c.ListBlockedNodes(ctx, nodes, rules); err != nil { b.Fatalf("ListBlockedNodes failed: %v", err) } } diff --git a/internal/controller/collector_shared_rules_test.go b/internal/controller/collector_shared_rules_test.go new file mode 100644 index 00000000..f5e79496 --- /dev/null +++ b/internal/controller/collector_shared_rules_test.go @@ -0,0 +1,142 @@ +/* +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" + "sync/atomic" + "testing" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + readinessv1alpha1 "sigs.k8s.io/node-readiness-controller/api/v1alpha1" + "sigs.k8s.io/node-readiness-controller/internal/metrics" +) + +func countRuleListCalls(fc client.WithWatch, onList func(ctx context.Context, c client.WithWatch)) (client.WithWatch, *atomic.Int32) { + var calls atomic.Int32 + wrapped := interceptor.NewClient(fc, interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + if _, ok := list.(*readinessv1alpha1.NodeReadinessRuleList); ok { + calls.Add(1) + + if err := c.List(ctx, list, opts...); err != nil { + return err + } + if onList != nil { + onList(ctx, c) + } + return nil + } + return c.List(ctx, list, opts...) + }, + }) + return wrapped, &calls +} + +func collectMetrics(c *metrics.ReadinessCollector) []prometheus.Metric { + ch := make(chan prometheus.Metric, 32) + c.Collect(ch) + close(ch) + var out []prometheus.Metric + for m := range ch { + out = append(out, m) + } + return out +} + +func TestCollect_ListsRulesExactlyOnce(t *testing.T) { + g := NewWithT(t) + + rule := gpuRuleWithConditions("GPUDriverReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( + rule, + withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + gpuNode("released-1", false), + ).Build() + + wrapped, ruleListCalls := countRuleListCalls(fc, nil) + rc := &RuleReadinessController{Client: wrapped} + collector := metrics.NewReadinessCollector(rc) + + collectMetrics(collector) + + g.Expect(ruleListCalls.Load()).To(Equal(int32(1)), + "Collect() must list NodeReadinessRuleList exactly once; got %d calls", ruleListCalls.Load()) +} + +func TestSharedRuleSnapshot_ConcurrentMutationCannotDiverge(t *testing.T) { + g := NewWithT(t) + + rule := gpuRuleWithConditions("GPUDriverReady") + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects( + rule, + withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), + ).Build() + + wrapped, ruleListCalls := countRuleListCalls(fc, func(ctx context.Context, c client.WithWatch) { + current := &readinessv1alpha1.NodeReadinessRule{} + g.Expect(c.Get(ctx, types.NamespacedName{Name: "gpu-ready"}, current)).To(Succeed()) + current.Finalizers = []string{"readiness.node.x-k8s.io/cleanup-taints"} + g.Expect(c.Update(ctx, current)).To(Succeed()) + g.Expect(c.Delete(ctx, current)).To(Succeed()) + }) + + rc := &RuleReadinessController{Client: wrapped} + collector := metrics.NewReadinessCollector(rc) + + metricsOut := collectMetrics(collector) + + g.Expect(ruleListCalls.Load()).To(Equal(int32(1))) + + var sawRuleNodes, sawBlockedNodes bool + for _, m := range metricsOut { + var pb dto.Metric + g.Expect(m.Write(&pb)).To(Succeed()) + + var isGPURule, hasStateLabel, hasConditionLabel bool + for _, l := range pb.GetLabel() { + switch l.GetName() { + case "rule": + isGPURule = l.GetValue() == "gpu-ready" + case "state": + hasStateLabel = true + case "condition": + hasConditionLabel = true + } + } + if !isGPURule { + continue + } + if hasStateLabel { + sawRuleNodes = true + } + if hasConditionLabel { + sawBlockedNodes = true + } + } + g.Expect(sawRuleNodes).To(BeTrue(), "expected gpu-ready to appear in node_readiness_rule_nodes") + g.Expect(sawBlockedNodes).To(BeTrue(), "expected gpu-ready to still appear in node_readiness_blocked_nodes despite the concurrent deletion racing the fetch") +} diff --git a/internal/controller/nodereadinessrule_controller.go b/internal/controller/nodereadinessrule_controller.go index d2a6e1ab..84d67571 100644 --- a/internal/controller/nodereadinessrule_controller.go +++ b/internal/controller/nodereadinessrule_controller.go @@ -540,23 +540,33 @@ func (r *RuleReadinessController) ListNodes(ctx context.Context) ([]corev1.Node, return nodeList.Items, nil } +// ListRules returns the current list of NodeReadinessRules. +func (r *RuleReadinessController) ListRules(ctx context.Context) ([]*readinessv1alpha1.NodeReadinessRule, error) { + ruleList := &readinessv1alpha1.NodeReadinessRuleList{} + if err := r.List(ctx, ruleList); err != nil { + return nil, err + } + rules := make([]*readinessv1alpha1.NodeReadinessRule, len(ruleList.Items)) + for i := range ruleList.Items { + rules[i] = &ruleList.Items[i] + } + return rules, nil +} + // forEachRuleNode applies callbacks to nodes matching each rule. +// +//nolint:unparam // keep error return for future extensibility and API stability. func (r *RuleReadinessController) forEachRuleNode( ctx context.Context, nodes []corev1.Node, + rules []*readinessv1alpha1.NodeReadinessRule, skipRule func(rule *readinessv1alpha1.NodeReadinessRule) bool, onRule func(rule *readinessv1alpha1.NodeReadinessRule), onNode func(rule *readinessv1alpha1.NodeReadinessRule, node *corev1.Node, held bool), ) error { - ruleList := &readinessv1alpha1.NodeReadinessRuleList{} - if err := r.List(ctx, ruleList); err != nil { - return err - } - log := ctrl.LoggerFrom(ctx) - for i := range ruleList.Items { - rule := &ruleList.Items[i] + for _, rule := range rules { if rule.Spec.DryRun { continue } @@ -586,10 +596,10 @@ func (r *RuleReadinessController) forEachRuleNode( } // ListRuleNodeStates returns the number of held and released nodes for each rule. -func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context, nodes []corev1.Node) (map[string]metrics.RuleNodeCounts, error) { +func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context, nodes []corev1.Node, rules []*readinessv1alpha1.NodeReadinessRule) (map[string]metrics.RuleNodeCounts, error) { counts := make(map[string]metrics.RuleNodeCounts) - err := r.forEachRuleNode(ctx, nodes, + err := r.forEachRuleNode(ctx, nodes, rules, func(rule *readinessv1alpha1.NodeReadinessRule) bool { return false }, func(rule *readinessv1alpha1.NodeReadinessRule) { counts[rule.Name] = metrics.RuleNodeCounts{} @@ -612,11 +622,11 @@ func (r *RuleReadinessController) ListRuleNodeStates(ctx context.Context, nodes } // ListBlockedNodes returns the number of blocked nodes for each rule and unsatisfied condition. -func (r *RuleReadinessController) ListBlockedNodes(ctx context.Context, nodes []corev1.Node) (map[string]metrics.RuleBlockedConditions, error) { +func (r *RuleReadinessController) ListBlockedNodes(ctx context.Context, nodes []corev1.Node, rules []*readinessv1alpha1.NodeReadinessRule) (map[string]metrics.RuleBlockedConditions, error) { result := make(map[string]metrics.RuleBlockedConditions) - err := r.forEachRuleNode(ctx, nodes, - func(rule *readinessv1alpha1.NodeReadinessRule) bool { return !rule.DeletionTimestamp.IsZero() }, + err := r.forEachRuleNode(ctx, nodes, rules, + func(rule *readinessv1alpha1.NodeReadinessRule) bool { return false }, func(rule *readinessv1alpha1.NodeReadinessRule) { counts := make(metrics.RuleBlockedConditions, len(rule.Spec.Conditions)) for _, cond := range rule.Spec.Conditions { @@ -662,15 +672,6 @@ func (r *RuleReadinessController) ruleAppliesTo(ctx context.Context, rule *readi return selector.Matches(labels.Set(node.Labels)) } -// checks if a rule applies to a node and has its taint. -func (r *RuleReadinessController) ruleAppliesToWithTaint(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, node *corev1.Node) (applies, held bool) { - applies = r.ruleAppliesTo(ctx, rule, node) - if !applies { - return false, false - } - return true, r.hasTaintBySpec(node, rule.Spec.Taint) -} - // updateRuleCache updates the rule cache. func (r *RuleReadinessController) updateRuleCache(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule) { log := ctrl.LoggerFrom(ctx) @@ -817,15 +818,20 @@ func (r *RuleReadinessController) processDryRun(ctx context.Context, rule *readi func (r *RuleReadinessController) cleanupTaintsForRule(ctx context.Context, rule *readinessv1alpha1.NodeReadinessRule, nodeList *corev1.NodeList) error { log := ctrl.LoggerFrom(ctx) + selector, err := parseNodeSelector(rule) + if err != nil { + log.Error(err, "Invalid node selector for rule during cleanup", "rule", rule.Name) + return nil + } + var errors []string for _, node := range nodeList.Items { - applies, held := r.ruleAppliesToWithTaint(ctx, rule, &node) - if !applies { + if !selector.Matches(labels.Set(node.Labels)) { continue } // Check if node has the taint managed by this rule - if held { + if r.hasTaintBySpec(&node, rule.Spec.Taint) { log.Info("Removing taint from node during rule cleanup", "node", node.Name, "rule", rule.Name, diff --git a/internal/controller/rule_node_states_test.go b/internal/controller/rule_node_states_test.go index d3e10596..d743c64c 100644 --- a/internal/controller/rule_node_states_test.go +++ b/internal/controller/rule_node_states_test.go @@ -140,7 +140,10 @@ func TestListRuleNodeStates_NoRules(t *testing.T) { Client: fc, } - counts, err := c.ListRuleNodeStates(t.Context(), nil) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + counts, err := c.ListRuleNodeStates(t.Context(), nil, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(BeEmpty()) } @@ -155,7 +158,10 @@ func TestListRuleNodeStates_ZeroMatches(t *testing.T) { {ObjectMeta: metav1.ObjectMeta{Name: "cpu-node"}}, } - counts, err := c.ListRuleNodeStates(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + counts, err := c.ListRuleNodeStates(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 0, Released: 0}, @@ -175,7 +181,10 @@ func TestListRuleNodeStates_MixedHeldReleased(t *testing.T) { {ObjectMeta: metav1.ObjectMeta{Name: "non-matching"}}, } - counts, err := c.ListRuleNodeStates(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + counts, err := c.ListRuleNodeStates(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 2, Released: 1}, @@ -192,7 +201,10 @@ func TestListRuleNodeStates_DryRunRuleExcluded(t *testing.T) { } nodes := []corev1.Node{*gpuNode("held-1", true)} - counts, err := c.ListRuleNodeStates(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + counts, err := c.ListRuleNodeStates(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(BeEmpty()) } @@ -213,7 +225,10 @@ func TestListRuleNodeStates_DeletingRuleIncluded(t *testing.T) { *gpuNode("released-1", false), } - counts, err := c.ListRuleNodeStates(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + counts, err := c.ListRuleNodeStates(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 2, Released: 1}, @@ -233,7 +248,10 @@ func TestListRuleNodeStates_DeletingRulePersistsUntilFinalizer(t *testing.T) { Client: fc, } - counts, err := c.ListRuleNodeStates(t.Context(), nil) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + counts, err := c.ListRuleNodeStates(t.Context(), nil, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 0, Released: 0}, @@ -280,7 +298,10 @@ func TestListRuleNodeStates_OneRuleHeldOtherReleased(t *testing.T) { Client: fc, } - counts, err := c.ListRuleNodeStates(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + counts, err := c.ListRuleNodeStates(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ "rule-a": {Held: 1, Released: 0}, @@ -317,7 +338,10 @@ func TestListRuleNodeStates_InvalidSelectorSkipped(t *testing.T) { *gpuNode("released-1", false), } - counts, err := c.ListRuleNodeStates(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + counts, err := c.ListRuleNodeStates(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(counts).NotTo(HaveKey(invalidRule.Name)) g.Expect(counts).To(Equal(map[string]metrics.RuleNodeCounts{ @@ -325,6 +349,32 @@ func TestListRuleNodeStates_InvalidSelectorSkipped(t *testing.T) { })) } +func TestCleanupTaintsForRule_InvalidSelectorReturnsNil(t *testing.T) { + g := NewWithT(t) + invalidRule := &readinessv1alpha1.NodeReadinessRule{ + ObjectMeta: metav1.ObjectMeta{Name: "invalid-selector-rule"}, + Spec: readinessv1alpha1.NodeReadinessRuleSpec{ + NodeSelector: metav1.LabelSelector{ + MatchExpressions: []metav1.LabelSelectorRequirement{ + {Key: "gpu", Operator: "BogusOperator", Values: []string{"true"}}, + }, + }, + Taint: gpuTaint(), + }, + } + + fc := fakeclient.NewClientBuilder().WithScheme(newTestScheme(t)).Build() + c := &RuleReadinessController{ + Client: fc, + } + nodeList := &corev1.NodeList{ + Items: []corev1.Node{*gpuNode("held-1", true)}, + } + + err := c.cleanupTaintsForRule(t.Context(), invalidRule, nodeList) + g.Expect(err).NotTo(HaveOccurred()) +} + func TestListBlockedNodes_ZeroSeededWhenNoHeldNodes(t *testing.T) { g := NewWithT(t) rule := gpuRuleWithConditions("GPUDriverReady", "CNIReady") @@ -333,7 +383,10 @@ func TestListBlockedNodes_ZeroSeededWhenNoHeldNodes(t *testing.T) { Client: fc, } - blocked, err := c.ListBlockedNodes(t.Context(), nil) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nil, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 0, "CNIReady": 0}, @@ -351,7 +404,10 @@ func TestListBlockedNodes_UnsatisfiedConditionCounted(t *testing.T) { *withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 1}, @@ -369,7 +425,10 @@ func TestListBlockedNodes_ReleasedNodeExcluded(t *testing.T) { *withNodeConditions(gpuNode("released-1", false), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 0}, @@ -389,7 +448,10 @@ func TestListBlockedNodes_MixedHeldReleased(t *testing.T) { *withNodeConditions(gpuNode("released-1", false), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 1}, @@ -411,7 +473,10 @@ func TestListBlockedNodes_MultipleUnsatisfiedConditions(t *testing.T) { ), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 1, "CNIReady": 1, "DiskReady": 0}, @@ -434,7 +499,10 @@ func TestListBlockedNodes_AnyOfNoConditionsSatisfied(t *testing.T) { ), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 1, "CNIReady": 1}, @@ -457,7 +525,10 @@ func TestListBlockedNodes_AnyOfWithSatisfiedCondition(t *testing.T) { ), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 1, "CNIReady": 0}, @@ -510,7 +581,10 @@ func TestListBlockedNodes_SharedConditionAcrossRules(t *testing.T) { Client: fc, } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "rule-a": {"GPUDriverReady": 1}, @@ -530,12 +604,15 @@ func TestListBlockedNodes_DryRunRuleExcluded(t *testing.T) { *withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(BeEmpty()) } -func TestListBlockedNodes_DeletingRuleExcluded(t *testing.T) { +func TestListBlockedNodes_DeletingRuleIncluded(t *testing.T) { g := NewWithT(t) rule := gpuRuleWithConditions("GPUDriverReady") now := metav1.Now() @@ -551,9 +628,14 @@ func TestListBlockedNodes_DeletingRuleExcluded(t *testing.T) { *withNodeConditions(gpuNode("released-1", false), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) g.Expect(err).NotTo(HaveOccurred()) - g.Expect(blocked).To(BeEmpty()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ + "gpu-ready": {"GPUDriverReady": 1}, + })) } func TestListBlockedNodes_NonMatchingNodeExcluded(t *testing.T) { @@ -573,7 +655,10 @@ func TestListBlockedNodes_NonMatchingNodeExcluded(t *testing.T) { }, nodeCondition("GPUDriverReady", corev1.ConditionFalse)), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 0}, @@ -611,7 +696,10 @@ func TestListBlockedNodes_InvalidSelectorSkipped(t *testing.T) { *withNodeConditions(gpuNode("held-1", true), nodeCondition("GPUDriverReady", corev1.ConditionFalse)), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).NotTo(HaveKey(invalidRule.Name)) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ @@ -628,7 +716,10 @@ func TestListBlockedNodes_ZeroDeclaredConditions(t *testing.T) { } nodes := []corev1.Node{*gpuNode("held-1", true)} - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {}, @@ -642,7 +733,10 @@ func TestListBlockedNodes_NoRules(t *testing.T) { Client: fc, } - blocked, err := c.ListBlockedNodes(t.Context(), nil) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nil, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(BeEmpty()) } @@ -663,7 +757,10 @@ func TestListBlockedNodes_DefaultStatusSatisfies(t *testing.T) { *withNodeConditions(gpuNode("held-1", true), nodeCondition("CondA", corev1.ConditionFalse)), } - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"CondA": 1, "CondB": 0}, @@ -683,14 +780,17 @@ func TestListBlockedNodes_AbsentConditionCounted(t *testing.T) { nodes := []corev1.Node{*gpuNode("held-1", true)} - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"CondB": 1}, })) } -// TestSharedNodeSnapshot_BothListersAgree verifies that both listers use the same Node snapshot. +// TestSharedNodeSnapshot_BothListersAgree verifies that both listers use the same Node and rule snapshot. func TestSharedNodeSnapshot_BothListersAgree(t *testing.T) { g := NewWithT(t) rule := gpuRuleWithConditions("GPUDriverReady") @@ -707,13 +807,16 @@ func TestSharedNodeSnapshot_BothListersAgree(t *testing.T) { g.Expect(err).NotTo(HaveOccurred()) g.Expect(nodes).To(HaveLen(2)) - ruleCounts, err := c.ListRuleNodeStates(t.Context(), nodes) + rules, err := c.ListRules(t.Context()) + g.Expect(err).NotTo(HaveOccurred()) + + ruleCounts, err := c.ListRuleNodeStates(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(ruleCounts).To(Equal(map[string]metrics.RuleNodeCounts{ "gpu-ready": {Held: 1, Released: 1}, })) - blocked, err := c.ListBlockedNodes(t.Context(), nodes) + blocked, err := c.ListBlockedNodes(t.Context(), nodes, rules) g.Expect(err).NotTo(HaveOccurred()) g.Expect(blocked).To(Equal(map[string]metrics.RuleBlockedConditions{ "gpu-ready": {"GPUDriverReady": 1}, diff --git a/internal/metrics/collector.go b/internal/metrics/collector.go index 2fa9fdb9..e61d339f 100644 --- a/internal/metrics/collector.go +++ b/internal/metrics/collector.go @@ -23,6 +23,8 @@ import ( "github.com/prometheus/client_golang/prometheus" corev1 "k8s.io/api/core/v1" ctrl "sigs.k8s.io/controller-runtime" + + readinessv1alpha1 "sigs.k8s.io/node-readiness-controller/api/v1alpha1" ) // collectTimeout limits how long a scrape can wait for cached data. @@ -33,6 +35,11 @@ type NodeLister interface { ListNodes(ctx context.Context) ([]corev1.Node, error) } +// RuleLister lists NodeReadinessRules for the collector. +type RuleLister interface { + ListRules(ctx context.Context) ([]*readinessv1alpha1.NodeReadinessRule, error) +} + // RuleNodeCounts holds the number of held and released nodes for a rule. type RuleNodeCounts struct { Held float64 @@ -41,7 +48,7 @@ type RuleNodeCounts struct { // RuleNodeStateLister lists held and released nodes for each rule. type RuleNodeStateLister interface { - ListRuleNodeStates(ctx context.Context, nodes []corev1.Node) (map[string]RuleNodeCounts, error) + ListRuleNodeStates(ctx context.Context, nodes []corev1.Node, rules []*readinessv1alpha1.NodeReadinessRule) (map[string]RuleNodeCounts, error) } // RuleBlockedConditions holds blocked node counts by condition. @@ -49,12 +56,13 @@ type RuleBlockedConditions map[string]float64 // BlockedNodesLister lists blocked node counts for each rule and condition. type BlockedNodesLister interface { - ListBlockedNodes(ctx context.Context, nodes []corev1.Node) (map[string]RuleBlockedConditions, error) + ListBlockedNodes(ctx context.Context, nodes []corev1.Node, rules []*readinessv1alpha1.NodeReadinessRule) (map[string]RuleBlockedConditions, error) } // ReadinessLister aggregates the scrape-time lookups the collector needs. type ReadinessLister interface { NodeLister + RuleLister RuleNodeStateLister BlockedNodesLister } @@ -99,17 +107,23 @@ func (c *ReadinessCollector) Collect(ch chan<- prometheus.Metric) { return } - counts, err := c.lister.ListRuleNodeStates(ctx, nodes) + rules, err := c.lister.ListRules(ctx) + if err != nil { + ctrl.Log.V(2).Info("Failed to list rules", "error", err) + return + } + + nodeStatesByRule, err := c.lister.ListRuleNodeStates(ctx, nodes, rules) if err != nil { ctrl.Log.V(2).Info("Failed to list rule node states", "error", err) } else { - for rule, rc := range counts { + for rule, rc := range nodeStatesByRule { ch <- prometheus.MustNewConstMetric(ruleNodesDesc, prometheus.GaugeValue, rc.Held, rule, string(RuleNodeStateHeld)) ch <- prometheus.MustNewConstMetric(ruleNodesDesc, prometheus.GaugeValue, rc.Released, rule, string(RuleNodeStateReleased)) } } - blocked, err := c.lister.ListBlockedNodes(ctx, nodes) + blocked, err := c.lister.ListBlockedNodes(ctx, nodes, rules) if err != nil { ctrl.Log.V(2).Info("Failed to list blocked nodes", "error", err) } else { diff --git a/internal/metrics/collector_test.go b/internal/metrics/collector_test.go index d2824c5a..f7c6f295 100644 --- a/internal/metrics/collector_test.go +++ b/internal/metrics/collector_test.go @@ -28,6 +28,8 @@ import ( dto "github.com/prometheus/client_model/go" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + readinessv1alpha1 "sigs.k8s.io/node-readiness-controller/api/v1alpha1" ) // stubLister is a test double for RuleNodeStateLister. @@ -35,6 +37,9 @@ type stubLister struct { nodes []corev1.Node nodesErr error + rules []*readinessv1alpha1.NodeReadinessRule + rulesErr error + counts map[string]RuleNodeCounts err error @@ -44,6 +49,8 @@ type stubLister struct { mu sync.Mutex gotNodesForRuleStates []corev1.Node gotNodesForBlocked []corev1.Node + gotRulesForRuleStates []*readinessv1alpha1.NodeReadinessRule + gotRulesForBlocked []*readinessv1alpha1.NodeReadinessRule } func (s *stubLister) ListNodes(_ context.Context) ([]corev1.Node, error) { @@ -53,9 +60,17 @@ func (s *stubLister) ListNodes(_ context.Context) ([]corev1.Node, error) { return s.nodes, nil } -func (s *stubLister) ListRuleNodeStates(_ context.Context, nodes []corev1.Node) (map[string]RuleNodeCounts, error) { +func (s *stubLister) ListRules(_ context.Context) ([]*readinessv1alpha1.NodeReadinessRule, error) { + if s.rulesErr != nil { + return nil, s.rulesErr + } + return s.rules, nil +} + +func (s *stubLister) ListRuleNodeStates(_ context.Context, nodes []corev1.Node, rules []*readinessv1alpha1.NodeReadinessRule) (map[string]RuleNodeCounts, error) { s.mu.Lock() s.gotNodesForRuleStates = nodes + s.gotRulesForRuleStates = rules s.mu.Unlock() if s.err != nil { return nil, s.err @@ -63,9 +78,10 @@ func (s *stubLister) ListRuleNodeStates(_ context.Context, nodes []corev1.Node) return s.counts, nil } -func (s *stubLister) ListBlockedNodes(_ context.Context, nodes []corev1.Node) (map[string]RuleBlockedConditions, error) { +func (s *stubLister) ListBlockedNodes(_ context.Context, nodes []corev1.Node, rules []*readinessv1alpha1.NodeReadinessRule) (map[string]RuleBlockedConditions, error) { s.mu.Lock() s.gotNodesForBlocked = nodes + s.gotRulesForBlocked = rules s.mu.Unlock() if s.blockedErr != nil { return nil, s.blockedErr @@ -287,6 +303,52 @@ func TestReadinessCollector_NodesSharedBetweenBothListers(t *testing.T) { } } +func TestReadinessCollector_RuleListErrorSkipsBothMetrics(t *testing.T) { + stub := &stubLister{ + rulesErr: errors.New("rule cache not synced"), + counts: map[string]RuleNodeCounts{"gpu-ready": {Held: 1, Released: 1}}, + blocked: map[string]RuleBlockedConditions{"gpu-ready": {"GPUDriverReady": 1}}, + } + c := NewReadinessCollector(stub) + + got := collectAll(t, c) + + if len(got["node_readiness_rule_nodes"]) != 0 { + t.Fatalf("expected no node_readiness_rule_nodes metrics when ListRules fails, got %v", got["node_readiness_rule_nodes"]) + } + if len(got["node_readiness_blocked_nodes"]) != 0 { + t.Fatalf("expected no node_readiness_blocked_nodes metrics when ListRules fails, got %v", got["node_readiness_blocked_nodes"]) + } + + if stub.gotRulesForRuleStates != nil || stub.gotRulesForBlocked != nil { + t.Fatalf("expected Collect to short-circuit before calling either counting method, but ListRuleNodeStates got %v, ListBlockedNodes got %v", + stub.gotRulesForRuleStates, stub.gotRulesForBlocked) + } +} + +func TestReadinessCollector_RulesSharedBetweenBothListers(t *testing.T) { + rules := []*readinessv1alpha1.NodeReadinessRule{{ObjectMeta: metav1.ObjectMeta{Name: "gpu-ready"}}} + stub := &stubLister{ + rules: rules, + counts: map[string]RuleNodeCounts{}, + blocked: map[string]RuleBlockedConditions{}, + } + c := NewReadinessCollector(stub) + + ch := make(chan prometheus.Metric, 4) + c.Collect(ch) + close(ch) + for range ch { + } + + if len(stub.gotRulesForRuleStates) != 1 || stub.gotRulesForRuleStates[0].Name != "gpu-ready" { + t.Fatalf("ListRuleNodeStates did not receive the shared rule snapshot: %v", stub.gotRulesForRuleStates) + } + if len(stub.gotRulesForBlocked) != 1 || stub.gotRulesForBlocked[0].Name != "gpu-ready" { + t.Fatalf("ListBlockedNodes did not receive the shared rule snapshot: %v", stub.gotRulesForBlocked) + } +} + func TestReadinessCollector_CollectAndLint(t *testing.T) { c := NewReadinessCollector(&stubLister{ nodes: []corev1.Node{{}},