From 99bca83a4b7e44f7b7a1f7616832bed645b4707c Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 31 Aug 2026 09:28:07 -0400 Subject: [PATCH 1/2] prometheus: purge deleted stream metric series in streams mode (CON-555) In streams mode, deleting a stream via DELETE /streams/{id} left its metric series (labeled stream="") registered in the prometheus exporter with frozen values until the process restarted, so /metrics accumulated series for every stream ever deleted. Implement the optional service.MetricsExporterSeriesDeleter interface on the prometheus exporter by calling DeletePartialMatch on every registered counter, gauge, summary and histogram vec. The benthos stream manager invokes it with {stream: } when a stream is deleted (or replaced by an update). The exporter-side implementation is inert until the benthos dependency is bumped to a release containing the stream manager hook; the unit tests exercise the exporter directly and pass against the currently pinned version. --- CHANGELOG.md | 6 ++ .../impl/prometheus/metrics_prometheus.go | 24 ++++++++ .../prometheus/metrics_prometheus_test.go | 60 +++++++++++++++++++ 3 files changed, 90 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d1a880659..64cc131227 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,12 @@ Changelog All notable changes to this project will be documented in this file. +## Unreleased + +### Fixed + +- prometheus: In streams mode, deleting a stream now purges its metric series (labeled `stream=""`) from the `/metrics` endpoint instead of exposing them with frozen values until the process restarts. ([@squiidz](https://github.com/squiidz), [#4742](https://github.com/redpanda-data/connect/pull/4742)) + ## 4.107.0 - 2026-08-27 ### Fixed 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..bc284f60b5 100644 --- a/internal/impl/prometheus/metrics_prometheus_test.go +++ b/internal/impl/prometheus/metrics_prometheus_test.go @@ -172,6 +172,66 @@ 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) + + 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") +} + +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 From dbba27bad24cd3de19d4951da37064da32529fd3 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Tue, 1 Sep 2026 09:58:52 -0400 Subject: [PATCH 2/2] prometheus: pin non-stream-label vec behavior in purge test, hold changelog entry for the benthos bump DeletePartialMatch on a vec that does not declare the matched label key is a non-match in client_golang (matchPartialLabels returns false for unknown keys), not a wildcard delete. Add a labeled vec without the stream label to the purge test to pin that behavior. The changelog entry moves to the PR that bumps benthos to a release containing the stream-manager purge hook, since the exporter side is inert until then and the note would otherwise ship in a release where the behavior is not observable. --- CHANGELOG.md | 6 ------ internal/impl/prometheus/metrics_prometheus_test.go | 6 ++++++ 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 64cc131227..1d1a880659 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,12 +3,6 @@ Changelog All notable changes to this project will be documented in this file. -## Unreleased - -### Fixed - -- prometheus: In streams mode, deleting a stream now purges its metric series (labeled `stream=""`) from the `/metrics` endpoint instead of exposing them with frozen values until the process restarts. ([@squiidz](https://github.com/squiidz), [#4742](https://github.com/redpanda-data/connect/pull/4742)) - ## 4.107.0 - 2026-08-27 ### Fixed diff --git a/internal/impl/prometheus/metrics_prometheus_test.go b/internal/impl/prometheus/metrics_prometheus_test.go index bc284f60b5..d28322ec2d 100644 --- a/internal/impl/prometheus/metrics_prometheus_test.go +++ b/internal/impl/prometheus/metrics_prometheus_test.go @@ -190,6 +190,11 @@ func TestPrometheusDeleteSeriesPartialMatch(t *testing.T) { 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") @@ -206,6 +211,7 @@ func TestPrometheusDeleteSeriesPartialMatch(t *testing.T) { 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) {