From 106036c88694d54221349c68a5ee9e103cbed69f Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Fri, 31 Jul 2026 13:48:20 +0200 Subject: [PATCH 1/2] Add dynamic labels to pipeline step event metrics Signed-off-by: Philipp Matthes --- .../bundles/cortex-nova/templates/alerts.yaml | 13 +- .../lib/filter_weigher_pipeline_monitor.go | 22 ++-- .../filter_weigher_pipeline_step_monitor.go | 100 +++++++++++++-- ...lter_weigher_pipeline_step_monitor_test.go | 114 ++++++++++++++++-- .../filter_weigher_pipeline_step_result.go | 18 ++- .../filters/filter_image_properties.go | 18 ++- .../filters/filter_image_properties_test.go | 37 +++++- 7 files changed, 280 insertions(+), 42 deletions(-) diff --git a/helm/bundles/cortex-nova/templates/alerts.yaml b/helm/bundles/cortex-nova/templates/alerts.yaml index e6e08e8e3..d59894cc2 100644 --- a/helm/bundles/cortex-nova/templates/alerts.yaml +++ b/helm/bundles/cortex-nova/templates/alerts.yaml @@ -305,7 +305,7 @@ spec: - alert: CortexNovaImagePropertiesHypervisorTypeUndetermined expr: | - sum by (pipeline, step) (rate(cortex_filter_weigher_pipeline_step_events_total{service="cortex-nova-metrics", event="image_properties_hv_type_undetermined"}[5m])) > 0.1 + sum by (pipeline, step, intent) (rate(cortex_filter_weigher_pipeline_step_events_total{service="cortex-nova-metrics", event="image_properties_hv_type_undetermined"}[5m])) > 0.1 for: 15m labels: context: scheduling @@ -319,11 +319,12 @@ spec: description: > The `filter_image_properties` step in pipeline `{{ "{{" }} $labels.pipeline {{ "}}" }}` is frequently unable to determine the hypervisor type from the image properties - of incoming scheduling requests. In this case the filter is skipped and all hosts - are returned, so kvm-only images may be placed on non-kvm hypervisors (or vice - versa). This may indicate that images are missing the expected hypervisor - type property, or that the property format has changed. Investigate the - image metadata of the affected requests. + of incoming scheduling requests for intent `{{ "{{" }} $labels.intent {{ "}}" }}`. + In this case the filter is skipped and all hosts are returned, so kvm-only + images may be placed on non-kvm hypervisors (or vice versa). This may indicate + that images are missing the expected hypervisor type property, or that the + property format has changed. Investigate the image metadata of the affected + requests. {{- if .Values.kvm.enabled }} - alert: CortexNovaDoesntFindValidKVMHosts diff --git a/internal/scheduling/lib/filter_weigher_pipeline_monitor.go b/internal/scheduling/lib/filter_weigher_pipeline_monitor.go index 341152322..785e18953 100644 --- a/internal/scheduling/lib/filter_weigher_pipeline_monitor.go +++ b/internal/scheduling/lib/filter_weigher_pipeline_monitor.go @@ -22,8 +22,6 @@ type FilterWeigherPipelineMonitor struct { stepReorderingsObserver *prometheus.HistogramVec // A histogram to observe the impact of the step on the hosts. stepImpactObserver *prometheus.HistogramVec - // A counter for named events reported by a step during its run. - stepEventCounter *prometheus.CounterVec // A histogram to measure how long the pipeline takes to run in total. pipelineRunTimer *prometheus.HistogramVec // A histogram to observe the number of hosts going into the scheduler pipeline. @@ -32,6 +30,10 @@ type FilterWeigherPipelineMonitor struct { hostNumberOutObserver *prometheus.HistogramVec // Counter for the number of requests processed by the scheduler. requestCounter *prometheus.CounterVec + + // stepEventCollector is a collector for named events reported by steps + // during their run. The collector supports a dynamic set of labels per event. + stepEventCollector *pipelineStepEventCollector } // Create a new scheduler monitor and register the necessary Prometheus metrics. @@ -66,10 +68,6 @@ func NewPipelineMonitor() FilterWeigherPipelineMonitor { Help: "Impact of the step on the hosts", Buckets: prometheus.ExponentialBucketsRange(0.01, 1000, 20), }, []string{"pipeline", "step", "stat", "unit"}), - stepEventCounter: prometheus.NewCounterVec(prometheus.CounterOpts{ - Name: "cortex_filter_weigher_pipeline_step_events_total", - Help: "Number of named events reported by a scheduler pipeline step", - }, []string{"pipeline", "step", "event"}), pipelineRunTimer: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "cortex_filter_weigher_pipeline_run_duration_seconds", Help: "Duration of scheduler pipeline run", @@ -89,6 +87,8 @@ func NewPipelineMonitor() FilterWeigherPipelineMonitor { Name: "cortex_filter_weigher_pipeline_requests_total", Help: "Total number of requests processed by the scheduler.", }, []string{"pipeline"}), + // Additional collectors. + stepEventCollector: newPipelineStepEventCollector(), } } @@ -122,12 +122,17 @@ func (m *FilterWeigherPipelineMonitor) observePipelineResult(request FilterWeigh } func (m *FilterWeigherPipelineMonitor) Describe(ch chan<- *prometheus.Desc) { + // stepEventCollector is intentionally not described here: its label names + // are dynamic and only known when events are recorded, so a fixed + // descriptor cannot be provided upfront. The remaining metrics are + // described normally so the registry can validate them at registration + // time. In practice the current prometheus client allows Collect to emit + // event metrics with label sets that differ from the described metrics. m.stepRunTimer.Describe(ch) m.stepHostWeight.Describe(ch) m.stepRemovedHostsObserver.Describe(ch) m.stepReorderingsObserver.Describe(ch) m.stepImpactObserver.Describe(ch) - m.stepEventCounter.Describe(ch) m.pipelineRunTimer.Describe(ch) m.hostNumberInObserver.Describe(ch) m.hostNumberOutObserver.Describe(ch) @@ -140,9 +145,10 @@ func (m *FilterWeigherPipelineMonitor) Collect(ch chan<- prometheus.Metric) { m.stepRemovedHostsObserver.Collect(ch) m.stepReorderingsObserver.Collect(ch) m.stepImpactObserver.Collect(ch) - m.stepEventCounter.Collect(ch) m.pipelineRunTimer.Collect(ch) m.hostNumberInObserver.Collect(ch) m.hostNumberOutObserver.Collect(ch) m.requestCounter.Collect(ch) + + m.stepEventCollector.Collect(ch) } diff --git a/internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go b/internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go index 2ad84e9fd..74dbbd7a1 100644 --- a/internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go +++ b/internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go @@ -12,6 +12,7 @@ import ( "sort" "strconv" "strings" + "sync" "github.com/prometheus/client_golang/prometheus" ) @@ -36,8 +37,10 @@ type FilterWeigherPipelineStepMonitor[RequestType FilterWeigherPipelineRequest] stepReorderingsObserver *prometheus.HistogramVec // A metric measuring the impact of the step on the hosts. stepImpactObserver *prometheus.HistogramVec - // A counter for named events reported by the step during its run. - stepEventCounter *prometheus.CounterVec + + // stepEventCollector is a collector for named events reported by the + // step during its run. + stepEventCollector *pipelineStepEventCollector } // Schedule using the wrapped step and measure the time it takes. @@ -60,7 +63,7 @@ func monitorStep[RequestType FilterWeigherPipelineRequest](stepName string, m Fi removedHostsObserver: removedHostsObserver, stepReorderingsObserver: m.stepReorderingsObserver, stepImpactObserver: m.stepImpactObserver, - stepEventCounter: m.stepEventCounter, + stepEventCollector: m.stepEventCollector, } } @@ -87,11 +90,9 @@ func (s *FilterWeigherPipelineStepMonitor[RequestType]) RunWrapped( ) // Count named events reported by the step during its run. - if s.stepEventCounter != nil { + if s.stepEventCollector != nil { for _, event := range stepResult.Events { - s.stepEventCounter. - WithLabelValues(s.pipelineName, s.stepName, event). - Inc() + s.stepEventCollector.Record(s.pipelineName, s.stepName, event.Name, event.Labels) } } @@ -272,3 +273,88 @@ func impact(before, after []string, stats map[string]float64, topK int) (float64 return impact, nil } + +// pipelineStepEventCollector is a prometheus.Collector that counts named events +// reported by scheduler pipeline steps. It supports a dynamic set of labels per +// event: the label names are not fixed upfront but derived from the labels +// attached to each event. +type pipelineStepEventCollector struct { + sync.Mutex + // counts holds the event counter values keyed by a deterministic signature. + counts map[string]float64 + // events maps the signature to the full event context (pipeline, step, + // event name and dynamic labels). + events map[string]recordedStepEvent +} + +type recordedStepEvent struct { + pipeline string + step string + eventName string + labels map[string]string +} + +func newPipelineStepEventCollector() *pipelineStepEventCollector { + return &pipelineStepEventCollector{ + counts: make(map[string]float64), + events: make(map[string]recordedStepEvent), + } +} + +// Record increments the counter for the given event. The same event can be +// recorded multiple times; each call increments the matching counter. +func (c *pipelineStepEventCollector) Record(pipeline, step, eventName string, labels map[string]string) { + c.Lock() + defer c.Unlock() + key := stepEventKey(pipeline, step, eventName, labels) + if _, ok := c.counts[key]; !ok { + c.counts[key] = 0 + c.events[key] = recordedStepEvent{ + pipeline: pipeline, + step: step, + eventName: eventName, + labels: maps.Clone(labels), + } + } + c.counts[key]++ +} + +// Describe is intentionally a no-op. Because the label names are dynamic and +// only known at collect time, we cannot describe the metric upfront. +func (c *pipelineStepEventCollector) Describe(ch chan<- *prometheus.Desc) { +} + +// Collect emits one metric per recorded event. Each metric shares the same +// fully-qualified name but may have a different set of labels depending on the +// labels provided by the step that reported the event. +func (c *pipelineStepEventCollector) Collect(ch chan<- prometheus.Metric) { + c.Lock() + defer c.Unlock() + for key, count := range c.counts { + ev := c.events[key] + labelNames := slices.Sorted(maps.Keys(ev.labels)) + labelValues := make([]string, len(labelNames)) + for i, name := range labelNames { + labelValues[i] = ev.labels[name] + } + desc := prometheus.NewDesc( + "cortex_filter_weigher_pipeline_step_events_total", + "Number of named events reported by a scheduler pipeline step", + append([]string{"pipeline", "step", "event"}, labelNames...), + nil, + ) + values := append([]string{ev.pipeline, ev.step, ev.eventName}, labelValues...) + ch <- prometheus.MustNewConstMetric(desc, prometheus.CounterValue, count, values...) + } +} + +// stepEventKey returns a deterministic signature for an event record. +func stepEventKey(pipeline, step, eventName string, labels map[string]string) string { + keys := slices.Sorted(maps.Keys(labels)) + parts := make([]string, 0, 3+2*len(keys)) + parts = append(parts, pipeline, step, eventName) + for _, k := range keys { + parts = append(parts, k, labels[k]) + } + return strings.Join(parts, "\x00") +} diff --git a/internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go b/internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go index 5ae402111..0c32c5641 100644 --- a/internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go +++ b/internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go @@ -6,6 +6,7 @@ package lib import ( "log/slog" "os" + "strings" "testing" "github.com/prometheus/client_golang/prometheus" @@ -59,20 +60,19 @@ func TestStepMonitorRun(t *testing.T) { } func TestStepMonitorRunEvents(t *testing.T) { - stepEventCounter := prometheus.NewCounterVec(prometheus.CounterOpts{ - Name: "cortex_filter_weigher_pipeline_step_events_total", - Help: "Number of named events reported by a scheduler pipeline step", - }, []string{"pipeline", "step", "event"}) + stepEventCollector := newPipelineStepEventCollector() monitor := &FilterWeigherPipelineStepMonitor[mockFilterWeigherPipelineRequest]{ - stepName: "mock_step", - pipelineName: "mock_pipeline", - stepEventCounter: stepEventCounter, + stepName: "mock_step", + pipelineName: "mock_pipeline", + stepEventCollector: stepEventCollector, } step := &mockWeigher[mockFilterWeigherPipelineRequest]{ RunFunc: func(traceLog *slog.Logger, request mockFilterWeigherPipelineRequest) (*FilterWeigherPipelineStepResult, error) { return &FilterWeigherPipelineStepResult{ Activations: map[string]float64{"host1": 0.0}, - Events: []string{"hypervisor_type_undetermined"}, + Events: []FilterWeigherPipelineStepEvent{ + {Name: "hypervisor_type_undetermined"}, + }, }, nil }, } @@ -83,9 +83,52 @@ func TestStepMonitorRunEvents(t *testing.T) { if _, err := monitor.RunWrapped(slog.Default(), request, step); err != nil { t.Fatalf("Run() error = %v, want nil", err) } - got := testutil.ToFloat64(stepEventCounter.WithLabelValues("mock_pipeline", "mock_step", "hypervisor_type_undetermined")) - if got != 1 { - t.Errorf("stepEventCounter = %v, want 1", got) + if got := len(stepEventCollector.counts); got != 1 { + t.Fatalf("stepEventCollector.counts = %v, want 1", got) + } + for _, count := range stepEventCollector.counts { + if count != 1 { + t.Errorf("stepEventCollector count = %v, want 1", count) + } + } +} + +func TestStepMonitorRunEventsWithDynamicLabels(t *testing.T) { + stepEventCollector := newPipelineStepEventCollector() + monitor := &FilterWeigherPipelineStepMonitor[mockFilterWeigherPipelineRequest]{ + stepName: "mock_step", + pipelineName: "mock_pipeline", + stepEventCollector: stepEventCollector, + } + step := &mockWeigher[mockFilterWeigherPipelineRequest]{ + RunFunc: func(traceLog *slog.Logger, request mockFilterWeigherPipelineRequest) (*FilterWeigherPipelineStepResult, error) { + return &FilterWeigherPipelineStepResult{ + Activations: map[string]float64{"host1": 0.0}, + Events: []FilterWeigherPipelineStepEvent{ + { + Name: "hypervisor_type_undetermined", + Labels: map[string]string{"intent": "create"}, + }, + }, + }, nil + }, + } + request := mockFilterWeigherPipelineRequest{ + Hosts: []string{"host1"}, + Weights: map[string]float64{"host1": 0.0}, + } + if _, err := monitor.RunWrapped(slog.Default(), request, step); err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + registry := prometheus.NewRegistry() + registry.MustRegister(stepEventCollector) + expected := ` + # HELP cortex_filter_weigher_pipeline_step_events_total Number of named events reported by a scheduler pipeline step + # TYPE cortex_filter_weigher_pipeline_step_events_total counter + cortex_filter_weigher_pipeline_step_events_total{event="hypervisor_type_undetermined",intent="create",pipeline="mock_pipeline",step="mock_step"} 1 + ` + if err := testutil.GatherAndCompare(registry, strings.NewReader(expected)); err != nil { + t.Fatalf("GatherAndCompare() error = %v", err) } } @@ -180,3 +223,52 @@ func TestImpact(t *testing.T) { }) } } + +func TestPipelineStepEventCollector_DynamicLabels(t *testing.T) { + collector := newPipelineStepEventCollector() + collector.Record("pipeline", "step", "event_a", map[string]string{"intent": "create"}) + collector.Record("pipeline", "step", "event_a", map[string]string{"intent": "create"}) + collector.Record("pipeline", "step", "event_a", map[string]string{"intent": "resize"}) + collector.Record("pipeline", "step", "event_b", map[string]string{"foo": "bar"}) + + registry := prometheus.NewRegistry() + registry.MustRegister(collector) + + expected := ` + # HELP cortex_filter_weigher_pipeline_step_events_total Number of named events reported by a scheduler pipeline step + # TYPE cortex_filter_weigher_pipeline_step_events_total counter + cortex_filter_weigher_pipeline_step_events_total{event="event_a",intent="create",pipeline="pipeline",step="step"} 2 + cortex_filter_weigher_pipeline_step_events_total{event="event_a",intent="resize",pipeline="pipeline",step="step"} 1 + cortex_filter_weigher_pipeline_step_events_total{event="event_b",foo="bar",pipeline="pipeline",step="step"} 1 + ` + if err := testutil.GatherAndCompare(registry, strings.NewReader(expected)); err != nil { + t.Fatalf("GatherAndCompare() error = %v", err) + } +} + +func TestPipelineStepEventCollector_NoLabels(t *testing.T) { + collector := newPipelineStepEventCollector() + collector.Record("pipeline", "step", "event_c", nil) + + registry := prometheus.NewRegistry() + registry.MustRegister(collector) + + expected := ` + # HELP cortex_filter_weigher_pipeline_step_events_total Number of named events reported by a scheduler pipeline step + # TYPE cortex_filter_weigher_pipeline_step_events_total counter + cortex_filter_weigher_pipeline_step_events_total{event="event_c",pipeline="pipeline",step="step"} 1 + ` + if err := testutil.GatherAndCompare(registry, strings.NewReader(expected)); err != nil { + t.Fatalf("GatherAndCompare() error = %v", err) + } +} + +func TestPipelineStepEventCollector_DescribeIsNoOp(t *testing.T) { + collector := newPipelineStepEventCollector() + ch := make(chan *prometheus.Desc, 1) + collector.Describe(ch) + close(ch) + if len(ch) != 0 { + t.Errorf("expected Describe to send no descriptors, got %d", len(ch)) + } +} diff --git a/internal/scheduling/lib/filter_weigher_pipeline_step_result.go b/internal/scheduling/lib/filter_weigher_pipeline_step_result.go index 8ccee7cf3..f7595cadd 100644 --- a/internal/scheduling/lib/filter_weigher_pipeline_step_result.go +++ b/internal/scheduling/lib/filter_weigher_pipeline_step_result.go @@ -27,9 +27,21 @@ type FilterWeigherPipelineStepResult struct { // Named events reported by the step during its run, e.g. // "hypervisor_type_undetermined". Each event is counted by the pipeline // monitor as cortex_filter_weigher_pipeline_step_events_total, labeled by - // pipeline, step and event. Use this to expose noteworthy step conditions - // (skipped filtering, missing data, ...) as Prometheus metrics. - Events []string + // pipeline, step, event and any additional labels provided by the step. + // Use this to expose noteworthy step conditions (skipped filtering, missing + // data, ...) as Prometheus metrics. + Events []FilterWeigherPipelineStepEvent +} + +// FilterWeigherPipelineStepEvent is a named event reported by a scheduling +// pipeline step. Additional labels can be attached to the event to enrich the +// exported Prometheus metric. +type FilterWeigherPipelineStepEvent struct { + // Name of the event, e.g. "image_properties_hv_type_undetermined". + Name string + // Additional labels to attach to the exported metric. Label names must be + // valid Prometheus label names. + Labels map[string]string } type FilterWeigherPipelineStepStatistics struct { diff --git a/internal/scheduling/nova/plugins/filters/filter_image_properties.go b/internal/scheduling/nova/plugins/filters/filter_image_properties.go index 2fe1ada44..dfc0d6446 100644 --- a/internal/scheduling/nova/plugins/filters/filter_image_properties.go +++ b/internal/scheduling/nova/plugins/filters/filter_image_properties.go @@ -22,9 +22,17 @@ type FilterImagePropertiesStep struct { func (s *FilterImagePropertiesStep) Run(traceLog *slog.Logger, request api.ExternalSchedulerRequest) (*lib.FilterWeigherPipelineStepResult, error) { result := s.IncludeAllHostsFromRequest(request) + // Resolve the scheduling intent once and fall back to a readable default + // when it cannot be determined. + intent, intentErr := request.GetIntent() + intentLabel := "unknown" + if intentErr == nil { + intentLabel = string(intent) + } + // Apply this filter to all requests, unless we know from the request's // intent that image metadata is expected to not be set. - if intent, err := request.GetIntent(); err == nil { + if intentErr == nil { intentsExpectedToNotHaveImageMeta := []v1alpha1.SchedulingIntent{ // Cortex-internal intents in which scheduling requests are sent // mainly based on flavor-related metadata, independent of the @@ -50,8 +58,12 @@ func (s *FilterImagePropertiesStep) Run(traceLog *slog.Logger, request api.Exter traceLog.Warn("could not determine hypervisor type from image properties", "error", err) // Expose this event through the step monitor to alert on high-frequency - // occurrences of this situation. - result.Events = append(result.Events, "image_properties_hv_type_undetermined") + // occurrences of this situation. Include the request intent as a dynamic + // label so the alert can pinpoint which scheduling intents are affected. + result.Events = append(result.Events, lib.FilterWeigherPipelineStepEvent{ + Name: "image_properties_hv_type_undetermined", + Labels: map[string]string{"intent": intentLabel}, + }) return result, nil } diff --git a/internal/scheduling/nova/plugins/filters/filter_image_properties_test.go b/internal/scheduling/nova/plugins/filters/filter_image_properties_test.go index 64dd4bb26..8490b2b76 100644 --- a/internal/scheduling/nova/plugins/filters/filter_image_properties_test.go +++ b/internal/scheduling/nova/plugins/filters/filter_image_properties_test.go @@ -8,6 +8,7 @@ import ( "testing" api "github.com/cobaltcore-dev/cortex/api/external/nova" + "github.com/cobaltcore-dev/cortex/internal/scheduling/lib" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -49,10 +50,11 @@ func TestFilterImagePropertiesStep_Run(t *testing.T) { } tests := []struct { - name string - request api.ExternalSchedulerRequest - expectedHosts []string - filteredHosts []string + name string + request api.ExternalSchedulerRequest + expectedHosts []string + filteredHosts []string + expectedEvents []lib.FilterWeigherPipelineStepEvent }{ { name: "kvm image keeps all hosts", @@ -77,12 +79,18 @@ func TestFilterImagePropertiesStep_Run(t *testing.T) { request: requestWith([]string{"host1", "host2", "host3"}, nil), expectedHosts: []string{"host1", "host2", "host3"}, filteredHosts: []string{}, + expectedEvents: []lib.FilterWeigherPipelineStepEvent{ + {Name: "image_properties_hv_type_undetermined", Labels: map[string]string{"intent": "unknown"}}, + }, }, { name: "unsupported hypervisor type keeps all hosts", request: requestWith([]string{"host1", "host2"}, map[string]any{"img_hv_type": "xen"}), expectedHosts: []string{"host1", "host2"}, filteredHosts: []string{}, + expectedEvents: []lib.FilterWeigherPipelineStepEvent{ + {Name: "image_properties_hv_type_undetermined", Labels: map[string]string{"intent": "unknown"}}, + }, }, { name: "empty host list", @@ -149,6 +157,27 @@ func TestFilterImagePropertiesStep_Run(t *testing.T) { if len(result.Activations) != len(tt.expectedHosts) { t.Errorf("expected %d hosts, got %d", len(tt.expectedHosts), len(result.Activations)) } + + if len(result.Events) != len(tt.expectedEvents) { + t.Errorf("expected %d events, got %d", len(tt.expectedEvents), len(result.Events)) + } + for i, expectedEvent := range tt.expectedEvents { + if i >= len(result.Events) { + break + } + gotEvent := result.Events[i] + if gotEvent.Name != expectedEvent.Name { + t.Errorf("expected event name %q, got %q", expectedEvent.Name, gotEvent.Name) + } + if len(gotEvent.Labels) != len(expectedEvent.Labels) { + t.Errorf("expected event labels %v, got %v", expectedEvent.Labels, gotEvent.Labels) + } + for k, v := range expectedEvent.Labels { + if gotEvent.Labels[k] != v { + t.Errorf("expected event label %q=%q, got %q", k, v, gotEvent.Labels[k]) + } + } + } }) } } From 492a1aaf42968e850a680d74fdc5e2e1a353a9e6 Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Fri, 31 Jul 2026 14:12:57 +0200 Subject: [PATCH 2/2] Drop reserved label names from pipeline step events Signed-off-by: Philipp Matthes --- .../filter_weigher_pipeline_step_monitor.go | 22 +++++++++++++++-- ...lter_weigher_pipeline_step_monitor_test.go | 24 +++++++++++++++++++ 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go b/internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go index 74dbbd7a1..3fcdeebcc 100644 --- a/internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go +++ b/internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go @@ -301,19 +301,37 @@ func newPipelineStepEventCollector() *pipelineStepEventCollector { } } +// reservedStepEventLabels are the fixed label names used by the event metric. +// Dynamic labels with these names are dropped because prometheus.Desc does not +// allow duplicate label names. +var reservedStepEventLabels = map[string]struct{}{ + "pipeline": {}, + "step": {}, + "event": {}, +} + // Record increments the counter for the given event. The same event can be // recorded multiple times; each call increments the matching counter. +// Dynamic labels that collide with the fixed label names (pipeline, step, +// event) are silently dropped to avoid panicking prometheus.MustNewConstMetric. func (c *pipelineStepEventCollector) Record(pipeline, step, eventName string, labels map[string]string) { c.Lock() defer c.Unlock() - key := stepEventKey(pipeline, step, eventName, labels) + cleaned := make(map[string]string, len(labels)) + for k, v := range labels { + if _, reserved := reservedStepEventLabels[k]; reserved { + continue + } + cleaned[k] = v + } + key := stepEventKey(pipeline, step, eventName, cleaned) if _, ok := c.counts[key]; !ok { c.counts[key] = 0 c.events[key] = recordedStepEvent{ pipeline: pipeline, step: step, eventName: eventName, - labels: maps.Clone(labels), + labels: cleaned, } } c.counts[key]++ diff --git a/internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go b/internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go index 0c32c5641..f17252d42 100644 --- a/internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go +++ b/internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go @@ -272,3 +272,27 @@ func TestPipelineStepEventCollector_DescribeIsNoOp(t *testing.T) { t.Errorf("expected Describe to send no descriptors, got %d", len(ch)) } } + +func TestPipelineStepEventCollector_DropsReservedLabels(t *testing.T) { + collector := newPipelineStepEventCollector() + // Reserved label names must be dropped so the emitted metric does not + // contain duplicate label names. + collector.Record("pipeline", "step", "event_d", map[string]string{ + "intent": "create", + "pipeline": "ignored", + "step": "ignored", + "event": "ignored", + }) + + registry := prometheus.NewRegistry() + registry.MustRegister(collector) + + expected := ` + # HELP cortex_filter_weigher_pipeline_step_events_total Number of named events reported by a scheduler pipeline step + # TYPE cortex_filter_weigher_pipeline_step_events_total counter + cortex_filter_weigher_pipeline_step_events_total{event="event_d",intent="create",pipeline="pipeline",step="step"} 1 + ` + if err := testutil.GatherAndCompare(registry, strings.NewReader(expected)); err != nil { + t.Fatalf("GatherAndCompare() error = %v", err) + } +}