streams mode: purge deleted stream metric series from exporters

[squiidz]

Ref: CON-555

Problem

In streams mode, DELETE /streams/{id} stops the stream and removes it from the API, but its metric series (labeled stream="<id>") remain registered in the shared metrics exporter forever. Pull-based exporters such as prometheus keep exposing every deleted stream's counters/gauges/timers with frozen values until the process restarts, growing cardinality without bound for workloads that churn streams:

==> Metrics AFTER delete:
$ curl http://localhost:4195/metrics | grep 'stream="foo"'
input_connection_up{label="",path="root.input",stream="foo"} 1
input_latency_ns{label="",path="root.input",stream="foo",quantile="0.5"} 56417
... (21 series total)

Per-stream metrics are label children of process-wide shared vecs (stats.WithLabels("stream", id)), so there is no per-stream object whose lifetime could clean them up, and no deletion concept existed anywhere in the metrics plumbing.

Fix

• Adds service.MetricsExporterSeriesDeleter, an optional interface for metrics exporter plugins that can delete all series matching a set of label values (DeleteSeriesPartialMatch(labels map[string]string)).
• Adds metrics.LabelPurger, the internal equivalent, implemented by Namespaced (forwarding to its child when supported) and by the air-gapped plugin wrapper (forwarding to the exporter when it implements the public interface).
• The stream manager now purges series matching {stream: <id>} after a stream is stopped and removed by Delete.

Exporters that do not implement the optional interface are unaffected (the purge is a no-op).

The prometheus exporter in redpanda-connect implements the new interface via DeletePartialMatch in a follow-up PR; with both in place, the reproduction above reports 21 series before delete and 0 after.

Notes

• Update is implemented as Delete + Create, so updating a stream now also resets that stream's series. For prometheus this is an ordinary counter reset, which scrapers handle; it also matches the semantics of the stream restarting.
• The purge is best effort: if a stream is force-deleted while components are still shutting down, a straggling metric write can re-create a series.

Testing

• Unit tests for the Namespaced forwarding (supported and unsupported child), the air-gap forwarding (supported and unsupported exporter), and the stream manager purging exactly {stream: <id>} on delete and not purging on failed deletes.
• Full test suite passes.
• Verified end to end against a redpanda-connect build carrying the exporter-side implementation, using the reproduction script from the ticket.

[claude]

Commits

1. streams mode: purge deleted stream metric series from exporters — the scope prefix contains a space. The enforced format is system: message (single lowercase system token) or system(subsystem): message; this should be something like streams: purge deleted stream metric series from exporters. Everything else about the commit is fine: single logical change, imperative lowercase subject, and a body that accurately explains the problem and the fix.

Review

Reviewed the added optional series-deletion plumbing: metrics.LabelPurger in internal/component/metrics/type.go, forwarding in Namespaced.DeleteSeriesPartialMatch and airGapMetrics.DeleteSeriesPartialMatch, the public service.MetricsExporterSeriesDeleter interface, and the purge of {stream: <id>} in stream/manager.Type.Delete.

The forwarding chain checks out end to end: manager.Type.Metrics() returns the *metrics.Namespaced root, which now always satisfies LabelPurger and delegates to its child only when the child supports it, so exporters that do not implement the interface are a genuine no-op rather than a panic. Update is Delete + Create in that order, so the purge cannot race the recreated stream's registration, and the counter-reset consequence is documented in both the changelog and the PR body. Test coverage matches the project patterns — supported/unsupported child for Namespaced, supported/unsupported exporter for the air gap, and a stream-manager test asserting the purge happens exactly once with {stream: "foo"} on a successful delete and not at all on ErrStreamDoesNotExist.

LGTM

[claude]

The purge runs after m.lock has been released, which opens a window where it can wipe the series of a newly created stream that reuses the same id.

Interleaving:

1. Delete("foo") removes foo from m.streams and releases m.lock.
2. A concurrent Create("foo", ...) acquires m.lock, builds the stream and calls strm.TriggerStartConsuming() — the new stream begins emitting series labelled stream="foo" — and returns.
3. Delete's DeleteSeriesPartialMatch({stream: "foo"}) then runs and deletes those series.

This is worse than the "straggling write re-creates a series" case noted in the PR description: for an exporter like Prometheus, DeletePartialMatch unregisters the children, but the running components already hold references to the metric objects they were handed, so their increments no longer surface in the registry. The live stream silently loses its metrics for its whole lifetime rather than just resetting.

Since Create holds m.lock across the entire construct-and-start sequence, performing the purge inside the same critical section that does delete(m.streams, id) closes the window — a concurrent Create for that id then cannot have started emitting before the purge completes.

The sequential Update path (Delete then Create) is unaffected; this only shows up with concurrent API requests for the same stream id.

[claude]

Commits
LGTM

Review
Focused, well-scoped change: a new optional DeleteSeriesPartialMatch interface at both the public (service.MetricsExporterSeriesDeleter) and internal (metrics.LabelPurger) boundaries, forwarded through Namespaced and the air-gap wrapper, and invoked by the stream manager on delete. Both forwarding layers correctly degrade to a no-op, and unit tests cover the supported/unsupported child and exporter cases plus the manager purging exactly {stream: <id>} (and not purging on a failed delete). CHANGELOG entry is present and accurate.

One issue:

1. internal/stream/manager/type.go — the purge is performed after m.lock is released, so a concurrent Create reusing the same stream id can start emitting series before the purge runs, causing the purge to wipe the new stream's series. For Prometheus this is not a recoverable reset: the running components hold references to already-unregistered children, so the live stream loses its metrics for its whole lifetime. Moving the purge into the same critical section as delete(m.streams, id) closes the window, since Create holds m.lock across construct-and-start.