fix: remove metrics when streams are deleted

[bookcherry]

Summary

• add cleanup scopes that track metric series created by each stream
• remove tracked series when streams are deleted, updated, stopped, or fail during creation
• expose optional stat deletion support to metrics exporters
• cover deletion, update, same-ID recreation, and stop lifecycle behavior

Companion PR

• fix: delete Prometheus series for removed streams connect#4694 adds Prometheus DeleteLabelValues support

Manual verification

The Benthos and Redpanda Connect branches were combined in a temporary Go workspace so the Connect process used the modified Benthos module. Redpanda Connect was then started in streams mode with file watching enabled, the HTTP API listening on port 4195, and the Prometheus metrics exporter enabled.

The root configuration used for verification was:

http:
  enabled: true
  address: 127.0.0.1:4195

metrics:
  prometheus: {}

logger:
  level: INFO

Each test stream continuously generated messages and emitted a custom counter:

input:
  generate:
    interval: 50ms
    mapping: "root = {}"

pipeline:
  processors:
    - metric:
        type: counter
        name: stream_lifecycle_test

output:
  drop: {}

Pre-fix reproduction

The same test was first run against the commits immediately before these changes:

• Benthos: 70c0dfc70
• Connect: f5203c9a5

After creating stream trigger-g, the Prometheus endpoint exposed 22 series carrying the stream label:

curl -s localhost:4195/metrics | grep -c "stream=\"trigger-g\""
# 22

The stream configuration was then deleted. The file watcher confirmed successful stream removal:

Stream trigger-g config deleted, attempting to remove stream.
Removed stream trigger-g.

However, all 22 series remained visible after waiting for the removal to complete:

curl -s localhost:4195/metrics | grep -c "stream=\"trigger-g\""
# 22

The procedure was repeated with four unique stream IDs: trigger-g, verify-a, verify-b, and verify-c. Each deleted stream retained 22 series, resulting in 88 stale series in total.

Patched behavior

The identical test was run with both patched branches. Stream trigger-g initially exposed 22 series. After deleting the stream configuration and observing the removal logs, no corresponding series remained:

curl -s localhost:4195/metrics | grep -c "stream=\"trigger-g\""
# 0

The three additional stream IDs were also created and deleted. Each exposed 22 series while active, and the combined stale-series count after deletion was 0.

Update lifecycle

A stream named foo was created with a custom metric named stream_old_metric. Its configuration was then updated in place to emit stream_new_metric instead.

After the watcher completed the update:

stream_old_metric series: 0
stream_new_metric series: 1

This confirms that the previous stream instance is cleaned up before the replacement remains active.

Same-ID recreation

Stream foo was deleted and the metrics endpoint was checked until no series labelled stream="foo" remained. A new configuration using the same stream ID was then created with a different custom metric.

The recreated stream emitted its new series normally. Deleting the recreated stream removed all of its series again, confirming that cleanup from the original instance does not delete metrics belonging to the replacement instance.

Manager shutdown

The stream manager test creates multiple active streams, calls Stop(), and immediately inspects the metrics exporter after Stop() returns. No series for either stream remain, confirming that cleanup completes synchronously before shutdown returns.

Automated verification

go test -race ./internal/component/metrics ./internal/stream/manager ./public/service

The tests cover:

• stream deletion
• stream update
• same-ID recreation
• manager shutdown
• cleanup idempotency
• counter, gauge, and timer cleanup
• concurrent access under the race detector

Related to #275.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}