From a83819eaa760e78a2f060e3ee7722c2f7ee43d16 Mon Sep 17 00:00:00 2001
From: kytsukam <kytsukam@lycorp.co.jp>
Date: Thu, 13 Aug 2026 17:08:49 +0900
Subject: [PATCH] fix: support deleting prometheus metric series

Implement metric deletion for counters, gauges, summaries, and histograms so Benthos stream cleanup removes stale Prometheus series.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .../impl/prometheus/metrics_prometheus.go     | 41 ++++++++++++++++---
 .../prometheus/metrics_prometheus_test.go     | 38 +++++++++++++++++
 2 files changed, 73 insertions(+), 6 deletions(-)

diff --git a/internal/impl/prometheus/metrics_prometheus.go b/internal/impl/prometheus/metrics_prometheus.go
index 085cbd1652..c3c7509089 100644
--- a/internal/impl/prometheus/metrics_prometheus.go
+++ b/internal/impl/prometheus/metrics_prometheus.go
@@ -140,7 +140,9 @@ func init() {
 //------------------------------------------------------------------------------
 
 type promGauge struct {
-	ctr prometheus.Gauge
+	ctr         prometheus.Gauge
+	vec         *prometheus.GaugeVec
+	labelValues []string
 }
 
 func (p *promGauge) Incr(count int64) {
@@ -167,8 +169,14 @@ func (p *promGauge) SetFloat64(value float64) {
 	p.ctr.Set(value)
 }
 
+func (p *promGauge) Delete() {
+	p.vec.DeleteLabelValues(p.labelValues...)
+}
+
 type promCounter struct {
-	ctr prometheus.Counter
+	ctr         prometheus.Counter
+	vec         *prometheus.CounterVec
+	labelValues []string
 }
 
 func (p *promCounter) Incr(count int64) {
@@ -179,9 +187,14 @@ func (p *promCounter) IncrFloat64(count float64) {
 	p.ctr.Add(count)
 }
 
+func (p *promCounter) Delete() {
+	p.vec.DeleteLabelValues(p.labelValues...)
+}
+
 type promTiming struct {
 	sum       prometheus.Observer
 	asSeconds bool
+	delete    func()
 }
 
 func (p *promTiming) Timing(val int64) {
@@ -192,6 +205,10 @@ func (p *promTiming) Timing(val int64) {
 	p.sum.Observe(vFloat)
 }
 
+func (p *promTiming) Delete() {
+	p.delete()
+}
+
 //------------------------------------------------------------------------------
 
 type promCounterVec struct {
@@ -201,7 +218,9 @@ type promCounterVec struct {
 
 func (p *promCounterVec) With(labelValues ...string) service.MetricsExporterCounter {
 	return &promCounter{
-		ctr: p.ctr.WithLabelValues(labelValues...),
+		ctr:         p.ctr.WithLabelValues(labelValues...),
+		vec:         p.ctr,
+		labelValues: append([]string(nil), labelValues...),
 	}
 }
 
@@ -211,8 +230,12 @@ type promTimingVec struct {
 }
 
 func (p *promTimingVec) With(labelValues ...string) service.MetricsExporterTimer {
+	values := append([]string(nil), labelValues...)
 	return &promTiming{
-		sum: p.sum.WithLabelValues(labelValues...),
+		sum: p.sum.WithLabelValues(values...),
+		delete: func() {
+			p.sum.DeleteLabelValues(values...)
+		},
 	}
 }
 
@@ -222,9 +245,13 @@ type promTimingHistVec struct {
 }
 
 func (p *promTimingHistVec) With(labelValues ...string) service.MetricsExporterTimer {
+	values := append([]string(nil), labelValues...)
 	return &promTiming{
 		asSeconds: true,
-		sum:       p.sum.WithLabelValues(labelValues...),
+		sum:       p.sum.WithLabelValues(values...),
+		delete: func() {
+			p.sum.DeleteLabelValues(values...)
+		},
 	}
 }
 
@@ -235,7 +262,9 @@ type promGaugeVec struct {
 
 func (p *promGaugeVec) With(labelValues ...string) service.MetricsExporterGauge {
 	return &promGauge{
-		ctr: p.ctr.WithLabelValues(labelValues...),
+		ctr:         p.ctr.WithLabelValues(labelValues...),
+		vec:         p.ctr,
+		labelValues: append([]string(nil), labelValues...),
 	}
 }
 
diff --git a/internal/impl/prometheus/metrics_prometheus_test.go b/internal/impl/prometheus/metrics_prometheus_test.go
index ef296ee51a..a1f8253779 100644
--- a/internal/impl/prometheus/metrics_prometheus_test.go
+++ b/internal/impl/prometheus/metrics_prometheus_test.go
@@ -172,6 +172,44 @@ func TestPrometheusMetrics(t *testing.T) {
 	assert.Contains(t, body, "\ngaugethree 10.452")
 }
 
+func TestPrometheusMetricDelete(t *testing.T) {
+	nm, handler := getTestProm(t)
+
+	counter := nm.NewCounterCtor("counter", "stream")("deleted")
+	counter.Incr(1)
+	gauge := nm.NewGaugeCtor("gauge", "stream")("deleted")
+	gauge.Set(1)
+	timer := nm.NewTimerCtor("timer", "stream")("deleted")
+	timer.Timing(1)
+
+	body := getPage(t, handler)
+	assert.Contains(t, body, `counter{stream="deleted"}`)
+	assert.Contains(t, body, `gauge{stream="deleted"}`)
+	assert.Contains(t, body, `timer_sum{stream="deleted"}`)
+
+	counter.(interface{ Delete() }).Delete()
+	gauge.(interface{ Delete() }).Delete()
+	timer.(interface{ Delete() }).Delete()
+
+	body = getPage(t, handler)
+	assert.NotContains(t, body, `stream="deleted"`)
+}
+
+func TestPrometheusHistogramMetricDelete(t *testing.T) {
+	nm := promFromYAML(t, `
+use_histogram_timing: true
+`)
+	timer := nm.NewTimerCtor("timer", "stream")("deleted")
+	timer.Timing(1)
+
+	handler := nm.HandlerFunc()
+	assert.Contains(t, getPage(t, handler), `timer_sum{stream="deleted"}`)
+
+	timer.(interface{ Delete() }).Delete()
+
+	assert.NotContains(t, getPage(t, handler), `stream="deleted"`)
+}
+
 func TestPrometheusHistMetrics(t *testing.T) {
 	nm := promFromYAML(t, `
 use_histogram_timing: true