Merge pull request #3419 from alvaroaleman/limit-cardinality

🐛 Limit depthWithPriorityMetric cardinality to 25

Author: k8s-ci-robot

pkg/internal/metrics/workqueue.go

@@ -18,9 +18,11 @@ package metrics
 
 import (
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/metrics"
 )
@@ -154,17 +156,55 @@ type DepthMetricWithPriority interface {
 var _ MetricsProviderWithPriority = WorkqueueMetricsProvider{}
 
 func (WorkqueueMetricsProvider) NewDepthMetricWithPriority(name string) DepthMetricWithPriority {
-	return &depthWithPriorityMetric{lvs: []string{name, name}}
+	return &depthWithPriorityMetric{depth: depth, lvs: []string{name, name}, observedPriorities: sets.Set[int]{}}
 }
 
+type prometheusGaugeVec interface {
+	WithLabelValues(lvs ...string) prometheus.Gauge
+}
+
+const (
+	priorityCardinalityExceededPlaceholder = "exceeded_cardinality_limit"
+	// maxRecommendedUniquePriorities is not scientifically chosen, we assume
+	// that the 99% use-case is to only use the two priorities that c-r itself
+	// uses and then leave a bit of leeway for other use-cases.
+	// We may decide to update this value in the future if we find that a
+	// a different value is more appropriate.
+	maxRecommendedUniquePriorities = 25
+)
+
 type depthWithPriorityMetric struct {
-	lvs []string
+	depth prometheusGaugeVec
+	lvs   []string
+
+	observedPrioritiesLock          sync.Mutex
+	priorityCardinalityLimitReached bool
+	observedPriorities              sets.Set[int]
+}
+
+func (g *depthWithPriorityMetric) priorityLabel(priority int) string {
+	g.observedPrioritiesLock.Lock()
+	defer g.observedPrioritiesLock.Unlock()
+
+	if g.priorityCardinalityLimitReached {
+		return priorityCardinalityExceededPlaceholder
+	}
+
+	g.observedPriorities.Insert(priority)
+
+	if g.observedPriorities.Len() > maxRecommendedUniquePriorities {
+		g.observedPriorities = nil
+		g.priorityCardinalityLimitReached = true
+		return priorityCardinalityExceededPlaceholder
+	}
+
+	return strconv.Itoa(priority)
 }
 
 func (g *depthWithPriorityMetric) Inc(priority int) {
-	depth.WithLabelValues(append(g.lvs, strconv.Itoa(priority))...).Inc()
+	g.depth.WithLabelValues(append(g.lvs, g.priorityLabel(priority))...).Inc()
 }
 
 func (g *depthWithPriorityMetric) Dec(priority int) {
-	depth.WithLabelValues(append(g.lvs, strconv.Itoa(priority))...).Dec()
+	g.depth.WithLabelValues(append(g.lvs, g.priorityLabel(priority))...).Dec()
 }

pkg/internal/metrics/workqueue_suite_test.go

@@ -0,0 +1,29 @@
+/*
+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 metrics
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestWorkqueue(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Workqueue Metrics Suite")
+}

pkg/internal/metrics/workqueue_test.go

@@ -0,0 +1,134 @@
+/*
+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 metrics
+
+import (
+	"sync"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+type fakeGauge struct {
+	prometheus.Gauge
+	inc func()
+}
+
+func (f *fakeGauge) Inc() { f.inc() }
+
+type fakeGaugeVec struct {
+	gauges map[string]int
+	mu     sync.Mutex
+}
+
+func (f *fakeGaugeVec) WithLabelValues(lvs ...string) prometheus.Gauge {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	key := ""
+	for _, lv := range lvs {
+		key += lv + "|"
+	}
+	if _, ok := f.gauges[key]; !ok {
+		f.gauges[key] = 0
+	}
+	return &fakeGauge{inc: func() {
+		f.mu.Lock()
+		f.gauges[key]++
+		f.mu.Unlock()
+	}}
+}
+
+var _ = Describe("depthWithPriorityMetric", func() {
+	Describe("CardinalityLimit", func() {
+		type testCase struct {
+			numUniquePriorities            int
+			expectedKeyCount               int
+			expectedCardinalityExceededVal int
+		}
+
+		DescribeTable("should respect cardinality limits",
+			func(tc testCase) {
+				fakeVec := &fakeGaugeVec{gauges: make(map[string]int)}
+				m := &depthWithPriorityMetric{
+					depth:              fakeVec,
+					lvs:                []string{"test", "test"},
+					observedPriorities: sets.Set[int]{},
+				}
+
+				wg := &sync.WaitGroup{}
+				wg.Add(tc.numUniquePriorities)
+				for i := 1; i <= tc.numUniquePriorities; i++ {
+					go func() {
+						m.Inc(i)
+						wg.Done()
+					}()
+				}
+				wg.Wait()
+
+				Expect(fakeVec.gauges).To(HaveLen(tc.expectedKeyCount))
+
+				placeholderKey := "test|test|" + priorityCardinalityExceededPlaceholder + "|"
+				Expect(fakeVec.gauges[placeholderKey]).To(Equal(tc.expectedCardinalityExceededVal))
+			},
+			Entry("under limit does not use placeholder", testCase{
+				numUniquePriorities:            10,
+				expectedKeyCount:               10,
+				expectedCardinalityExceededVal: 0,
+			}),
+			Entry("at limit does not use placeholder", testCase{
+				numUniquePriorities:            25,
+				expectedKeyCount:               25,
+				expectedCardinalityExceededVal: 0,
+			}),
+			Entry("exceeding limit uses placeholder", testCase{
+				numUniquePriorities:            26,
+				expectedKeyCount:               26,
+				expectedCardinalityExceededVal: 1,
+			}),
+			Entry("well over limit uses placeholder for all excess", testCase{
+				numUniquePriorities:            30,
+				expectedKeyCount:               26,
+				expectedCardinalityExceededVal: 5,
+			}),
+		)
+	})
+
+	It("same priority many adds does not trigger cardinality limit", func() {
+		fakeVec := &fakeGaugeVec{gauges: make(map[string]int)}
+		m := &depthWithPriorityMetric{
+			depth:              fakeVec,
+			lvs:                []string{"test", "test"},
+			observedPriorities: sets.Set[int]{},
+		}
+
+		wg := &sync.WaitGroup{}
+		wg.Add(200)
+		for range 200 {
+			go func() {
+				m.Inc(1)
+				wg.Done()
+			}()
+		}
+		wg.Wait()
+
+		Expect(fakeVec.gauges).To(HaveLen(1))
+		Expect(fakeVec.gauges["test|test|1|"]).To(Equal(200))
+	})
+})