Add dynamic labels to pipeline step event metrics

helm/bundles/cortex-nova/templates/alerts.yaml

@@ -342,7 +342,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
@@ -356,11 +356,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

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)
 }

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,106 @@ 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),
+	}
+}
+
+// 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()
+	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:    cleaned,
+		}
+	}
+	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")
+}