diff --git a/internal/impl/prometheus/metrics_prometheus.go b/internal/impl/prometheus/metrics_prometheus.go index 085cbd1652..7c168281c1 100644 --- a/internal/impl/prometheus/metrics_prometheus.go +++ b/internal/impl/prometheus/metrics_prometheus.go @@ -520,6 +520,30 @@ func (p *metrics) NewGaugeCtor(path string, labelNames ...string) service.Metric } } +// DeleteSeriesPartialMatch deletes all metric series containing labels +// matching all of the provided label key/value pairs, e.g. all series of a +// stream deleted in streams mode. Implements the optional +// service.MetricsExporterSeriesDeleter interface. +func (p *metrics) DeleteSeriesPartialMatch(labels map[string]string) { + promLabels := prometheus.Labels(labels) + + p.mut.Lock() + defer p.mut.Unlock() + + for _, pv := range p.counters { + pv.ctr.DeletePartialMatch(promLabels) + } + for _, pv := range p.gauges { + pv.ctr.DeletePartialMatch(promLabels) + } + for _, pv := range p.timers { + pv.sum.DeletePartialMatch(promLabels) + } + for _, pv := range p.timersHist { + pv.sum.DeletePartialMatch(promLabels) + } +} + func (p *metrics) Close(context.Context) error { if atomic.CompareAndSwapInt32(&p.running, 1, 0) { close(p.closedChan) diff --git a/internal/impl/prometheus/metrics_prometheus_test.go b/internal/impl/prometheus/metrics_prometheus_test.go index ef296ee51a..d28322ec2d 100644 --- a/internal/impl/prometheus/metrics_prometheus_test.go +++ b/internal/impl/prometheus/metrics_prometheus_test.go @@ -172,6 +172,72 @@ func TestPrometheusMetrics(t *testing.T) { assert.Contains(t, body, "\ngaugethree 10.452") } +func TestPrometheusDeleteSeriesPartialMatch(t *testing.T) { + nm, handler := getTestProm(t) + + ctr := nm.NewCounterCtor("input_received", "label", "stream") + ctr("in", "foo").Incr(3) + ctr("in", "bar").Incr(4) + + gge := nm.NewGaugeCtor("input_connection_up", "stream") + gge("foo").Set(1) + gge("bar").Set(1) + + tmr := nm.NewTimerCtor("input_latency_ns", "stream") + tmr("foo").Timing(100) + tmr("bar").Timing(200) + + unlabelled := nm.NewCounterCtor("uptime")() + unlabelled.Incr(9) + + // A vec with labels that do not include the matched key must be untouched: + // client_golang treats an unknown label key as a non-match, not a wildcard. + otherLabels := nm.NewCounterCtor("batch_created", "mechanism") + otherLabels("count").Incr(5) + + body := getPage(t, handler) + require.Contains(t, body, "\ninput_received{label=\"in\",stream=\"foo\"} 3") + require.Contains(t, body, "\ninput_latency_ns_sum{stream=\"foo\"} 100") + + purger, ok := any(nm).(interface { + DeleteSeriesPartialMatch(labels map[string]string) + }) + require.True(t, ok, "prometheus exporter should support deleting series by label match") + purger.DeleteSeriesPartialMatch(map[string]string{"stream": "foo"}) + + body = getPage(t, handler) + assert.NotContains(t, body, "stream=\"foo\"") + assert.Contains(t, body, "\ninput_received{label=\"in\",stream=\"bar\"} 4") + assert.Contains(t, body, "\ninput_connection_up{stream=\"bar\"} 1") + assert.Contains(t, body, "\ninput_latency_ns_sum{stream=\"bar\"} 200") + assert.Contains(t, body, "\nuptime 9") + assert.Contains(t, body, "\nbatch_created{mechanism=\"count\"} 5") +} + +func TestPrometheusDeleteSeriesPartialMatchHistogram(t *testing.T) { + nm := promFromYAML(t, ` +use_histogram_timing: true +`) + + tmr := nm.NewTimerCtor("input_latency_ns", "stream") + tmr("foo").Timing(100) + tmr("bar").Timing(200) + + handler := nm.HandlerFunc() + body := getPage(t, handler) + require.Contains(t, body, "stream=\"foo\"") + + purger, ok := any(nm).(interface { + DeleteSeriesPartialMatch(labels map[string]string) + }) + require.True(t, ok, "prometheus exporter should support deleting series by label match") + purger.DeleteSeriesPartialMatch(map[string]string{"stream": "foo"}) + + body = getPage(t, handler) + assert.NotContains(t, body, "stream=\"foo\"") + assert.Contains(t, body, "\ninput_latency_ns_count{stream=\"bar\"} 1") +} + func TestPrometheusHistMetrics(t *testing.T) { nm := promFromYAML(t, ` use_histogram_timing: true