From 65cff99b98e69fb0087559b008bb9a012433babe Mon Sep 17 00:00:00 2001 From: kytsukam Date: Thu, 13 Aug 2026 17:08:44 +0900 Subject: [PATCH] fix: remove metrics when streams are deleted Track metric series created within each stream scope and delete them when the stream is removed, updated, or stopped. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- internal/bundle/package.go | 1 + internal/component/metrics/combine.go | 27 +++ internal/component/metrics/local.go | 49 +++++- internal/component/metrics/namespaced.go | 162 ++++++++++++++++-- internal/component/metrics/namespaced_test.go | 28 +++ internal/component/metrics/type.go | 6 + internal/manager/mock/manager.go | 3 + internal/manager/type.go | 8 + internal/stream/manager/type.go | 20 ++- internal/stream/manager/type_test.go | 110 ++++++++++++ public/service/metrics.go | 24 +++ 11 files changed, 416 insertions(+), 22 deletions(-) diff --git a/internal/bundle/package.go b/internal/bundle/package.go index c81c12442..f8842a8cd 100644 --- a/internal/bundle/package.go +++ b/internal/bundle/package.go @@ -45,6 +45,7 @@ type NewManagement interface { ForStream(id string) NewManagement IntoPath(segments ...string) NewManagement WithAddedMetrics(m metrics.Type) NewManagement + WithMetricsCleanup() NewManagement EngineVersion() string diff --git a/internal/component/metrics/combine.go b/internal/component/metrics/combine.go index eba025153..2f6a5170d 100644 --- a/internal/component/metrics/combine.go +++ b/internal/component/metrics/combine.go @@ -58,6 +58,15 @@ func (c *combinedCounter) IncrFloat64(count float64) { c.c2.IncrFloat64(count) } +func (c *combinedCounter) Delete() { + if d, ok := c.c1.(StatDeleter); ok { + d.Delete() + } + if d, ok := c.c2.(StatDeleter); ok { + d.Delete() + } +} + type combinedTimer struct { c1 StatTimer c2 StatTimer @@ -68,6 +77,15 @@ func (c *combinedTimer) Timing(delta int64) { c.c2.Timing(delta) } +func (c *combinedTimer) Delete() { + if d, ok := c.c1.(StatDeleter); ok { + d.Delete() + } + if d, ok := c.c2.(StatDeleter); ok { + d.Delete() + } +} + type combinedGauge struct { c1 StatGauge c2 StatGauge @@ -103,6 +121,15 @@ func (c *combinedGauge) DecrFloat64(count float64) { c.c2.DecrFloat64(count) } +func (c *combinedGauge) Delete() { + if d, ok := c.c1.(StatDeleter); ok { + d.Delete() + } + if d, ok := c.c2.(StatDeleter); ok { + d.Delete() + } +} + //------------------------------------------------------------------------------ type combinedCounterVec struct { diff --git a/internal/component/metrics/local.go b/internal/component/metrics/local.go index cedc26952..c3f9564f6 100644 --- a/internal/component/metrics/local.go +++ b/internal/component/metrics/local.go @@ -66,6 +66,37 @@ func (l *LocalTiming) Timing(delta int64) { l.lock.Unlock() } +type localStatRef struct { + *LocalStat + owner *Local + path string +} + +func (l *localStatRef) Delete() { + l.owner.mut.Lock() + if l.owner.flatCounters[l.path] == l.LocalStat { + delete(l.owner.flatCounters, l.path) + } + l.owner.mut.Unlock() +} + +type localTimingRef struct { + *LocalTiming + owner *Local + path string +} + +func (l *localTimingRef) Delete() { + l.owner.mut.Lock() + if l.owner.flatTimings[l.path] == l.LocalTiming { + delete(l.owner.flatTimings, l.path) + l.LocalTiming.lock.Lock() + l.LocalTiming.t.Stop() + l.LocalTiming.lock.Unlock() + } + l.owner.mut.Unlock() +} + //------------------------------------------------------------------------------ // Local is a metrics aggregator that stores metrics locally. @@ -231,7 +262,11 @@ func (l *Local) GetCounterVec(path string, k ...string) StatCounterVec { l.flatCounters[newPath] = st } l.mut.Unlock() - return st + return &localStatRef{ + LocalStat: st, + owner: l, + path: newPath, + } }) } @@ -247,7 +282,11 @@ func (l *Local) GetTimerVec(path string, k ...string) StatTimerVec { l.flatTimings[newPath] = st } l.mut.Unlock() - return st + return &localTimingRef{ + LocalTiming: st, + owner: l, + path: newPath, + } }) } @@ -264,7 +303,11 @@ func (l *Local) GetGaugeVec(path string, k ...string) StatGaugeVec { l.flatCounters[newPath] = st } l.mut.Unlock() - return st + return &localStatRef{ + LocalStat: st, + owner: l, + path: newPath, + } }) } diff --git a/internal/component/metrics/namespaced.go b/internal/component/metrics/namespaced.go index 13ebcb0e2..04a41d91c 100644 --- a/internal/component/metrics/namespaced.go +++ b/internal/component/metrics/namespaced.go @@ -6,6 +6,9 @@ import ( "maps" "net/http" "sort" + "strconv" + "strings" + "sync" ) // Namespaced wraps a child metrics exporter and exposes a Type API that @@ -14,6 +17,7 @@ type Namespaced struct { labels map[string]string mappings []*Mapping child Type + cleanup *namespacedCleanup } // NewNamespaced wraps a metrics exporter and adds prefixes and custom labels. @@ -38,6 +42,16 @@ func (n *Namespaced) WithStats(s Type) *Namespaced { return &newNs } +// WithCleanup returns a metrics exporter scope that removes all series created +// through it when closed, where supported by the underlying exporter. +func (n *Namespaced) WithCleanup() *Namespaced { + newNs := *n + newNs.cleanup = &namespacedCleanup{ + stats: map[string]StatDeleter{}, + } + return &newNs +} + // WithLabels returns a namespaced metrics exporter with a new set of labels, // which are added to any prior labels. func (n *Namespaced) WithLabels(labels ...string) *Namespaced { @@ -101,41 +115,119 @@ func (n *Namespaced) getPathAndLabels(path string) (newPath string, labelKeys, l type counterVecWithStatic struct { staticValues []string child StatCounterVec + path string + cleanup *namespacedCleanup } func (c *counterVecWithStatic) With(values ...string) StatCounter { newValues := make([]string, 0, len(c.staticValues)+len(values)) newValues = append(newValues, c.staticValues...) newValues = append(newValues, values...) - return c.child.With(newValues...) + stat := c.child.With(newValues...) + if c.cleanup != nil { + c.cleanup.track(cleanupKey("counter", c.path, newValues), stat) + } + return stat } type timerVecWithStatic struct { staticValues []string child StatTimerVec + path string + cleanup *namespacedCleanup } func (c *timerVecWithStatic) With(values ...string) StatTimer { newValues := make([]string, 0, len(c.staticValues)+len(values)) newValues = append(newValues, c.staticValues...) newValues = append(newValues, values...) - return c.child.With(newValues...) + stat := c.child.With(newValues...) + if c.cleanup != nil { + c.cleanup.track(cleanupKey("timer", c.path, newValues), stat) + } + return stat } type gaugeVecWithStatic struct { staticValues []string child StatGaugeVec + path string + cleanup *namespacedCleanup } func (c *gaugeVecWithStatic) With(values ...string) StatGauge { newValues := make([]string, 0, len(c.staticValues)+len(values)) newValues = append(newValues, c.staticValues...) newValues = append(newValues, values...) - return c.child.With(newValues...) + stat := c.child.With(newValues...) + if c.cleanup != nil { + c.cleanup.track(cleanupKey("gauge", c.path, newValues), stat) + } + return stat } //------------------------------------------------------------------------------ +type namespacedCleanup struct { + mut sync.Mutex + stats map[string]StatDeleter + closed bool +} + +func cleanupKey(kind, path string, labelValues []string) string { + var b strings.Builder + writePart := func(value string) { + b.WriteString(strconv.Itoa(len(value))) + b.WriteByte(':') + b.WriteString(value) + } + writePart(kind) + writePart(path) + for _, value := range labelValues { + writePart(value) + } + return b.String() +} + +func (c *namespacedCleanup) track(key string, stat any) { + deleter, ok := stat.(StatDeleter) + if !ok { + return + } + c.mut.Lock() + if c.closed { + c.mut.Unlock() + deleter.Delete() + return + } + if _, exists := c.stats[key]; !exists { + c.stats[key] = deleter + } + c.mut.Unlock() +} + +func (c *namespacedCleanup) close() { + c.mut.Lock() + if c.closed { + c.mut.Unlock() + return + } + c.closed = true + stats := c.stats + c.stats = nil + c.mut.Unlock() + + for _, stat := range stats { + stat.Delete() + } +} + +func (n *Namespaced) track(kind, path string, labelValues []string, stat any) { + if n.cleanup != nil { + n.cleanup.track(cleanupKey(kind, path, labelValues), stat) + } +} + // GetCounter returns an editable counter stat for a given path. func (n *Namespaced) GetCounter(path string) StatCounter { path, labelKeys, labelValues := n.getPathAndLabels(path) @@ -143,9 +235,13 @@ func (n *Namespaced) GetCounter(path string) StatCounter { return DudStat{} } if len(labelKeys) > 0 { - return n.child.GetCounterVec(path, labelKeys...).With(labelValues...) + stat := n.child.GetCounterVec(path, labelKeys...).With(labelValues...) + n.track("counter", path, labelValues, stat) + return stat } - return n.child.GetCounter(path) + stat := n.child.GetCounter(path) + n.track("counter", path, nil, stat) + return stat } // GetCounterVec returns an editable counter stat for a given path with labels, @@ -165,9 +261,19 @@ func (n *Namespaced) GetCounterVec(path string, labelNames ...string) StatCounte return &counterVecWithStatic{ staticValues: staticValues, child: n.child.GetCounterVec(path, newNames...), + path: path, + cleanup: n.cleanup, } } - return n.child.GetCounterVec(path, labelNames...) + child := n.child.GetCounterVec(path, labelNames...) + if n.cleanup == nil { + return child + } + return &counterVecWithStatic{ + child: child, + path: path, + cleanup: n.cleanup, + } } // GetTimer returns an editable timer stat for a given path. @@ -177,9 +283,13 @@ func (n *Namespaced) GetTimer(path string) StatTimer { return DudStat{} } if len(labelKeys) > 0 { - return n.child.GetTimerVec(path, labelKeys...).With(labelValues...) + stat := n.child.GetTimerVec(path, labelKeys...).With(labelValues...) + n.track("timer", path, labelValues, stat) + return stat } - return n.child.GetTimer(path) + stat := n.child.GetTimer(path) + n.track("timer", path, nil, stat) + return stat } // GetTimerVec returns an editable timer stat for a given path with labels, @@ -199,9 +309,19 @@ func (n *Namespaced) GetTimerVec(path string, labelNames ...string) StatTimerVec return &timerVecWithStatic{ staticValues: staticValues, child: n.child.GetTimerVec(path, newNames...), + path: path, + cleanup: n.cleanup, } } - return n.child.GetTimerVec(path, labelNames...) + child := n.child.GetTimerVec(path, labelNames...) + if n.cleanup == nil { + return child + } + return &timerVecWithStatic{ + child: child, + path: path, + cleanup: n.cleanup, + } } // GetGauge returns an editable gauge stat for a given path. @@ -211,9 +331,13 @@ func (n *Namespaced) GetGauge(path string) StatGauge { return DudStat{} } if len(labelKeys) > 0 { - return n.child.GetGaugeVec(path, labelKeys...).With(labelValues...) + stat := n.child.GetGaugeVec(path, labelKeys...).With(labelValues...) + n.track("gauge", path, labelValues, stat) + return stat } - return n.child.GetGauge(path) + stat := n.child.GetGauge(path) + n.track("gauge", path, nil, stat) + return stat } // GetGaugeVec returns an editable gauge stat for a given path with labels, @@ -233,12 +357,26 @@ func (n *Namespaced) GetGaugeVec(path string, labelNames ...string) StatGaugeVec return &gaugeVecWithStatic{ staticValues: staticValues, child: n.child.GetGaugeVec(path, newNames...), + path: path, + cleanup: n.cleanup, } } - return n.child.GetGaugeVec(path, labelNames...) + child := n.child.GetGaugeVec(path, labelNames...) + if n.cleanup == nil { + return child + } + return &gaugeVecWithStatic{ + child: child, + path: path, + cleanup: n.cleanup, + } } // Close stops aggregating stats and cleans up resources. func (n *Namespaced) Close() error { + if n.cleanup != nil { + n.cleanup.close() + return nil + } return n.child.Close() } diff --git a/internal/component/metrics/namespaced_test.go b/internal/component/metrics/namespaced_test.go index 66ed55e07..f07469af1 100644 --- a/internal/component/metrics/namespaced_test.go +++ b/internal/component/metrics/namespaced_test.go @@ -83,6 +83,34 @@ func TestNamespacedNothing(t *testing.T) { assert.Contains(t, body, `"timertwo{label3=\"value4\",label4=\"value5\"}":{"p50":13,"p90":13,"p99":13}`) } +func TestNamespacedCleanup(t *testing.T) { + prom, handler := getTestMetrics(t) + + persistent := metrics.NewNamespaced(prom) + persistent.GetCounterVec("counter", "stream").With("persistent").Incr(1) + + scoped := metrics.NewNamespaced(prom). + WithLabels("stream", "temporary"). + WithCleanup() + scoped.GetCounter("counter").Incr(1) + scoped.GetCounter("counter").Incr(1) + scoped.GetGaugeVec("gauge", "status").With("ready").Set(1) + scoped.GetTimer("timer").Timing(1) + + body := getPage(t, handler) + assert.Contains(t, body, `"counter{stream=\"persistent\"}":1`) + assert.Contains(t, body, `"counter{stream=\"temporary\"}":2`) + assert.Contains(t, body, `"gauge{status=\"ready\",stream=\"temporary\"}":1`) + assert.Contains(t, body, `"timer{stream=\"temporary\"}"`) + + require.NoError(t, scoped.Close()) + require.NoError(t, scoped.Close()) + + body = getPage(t, handler) + assert.Contains(t, body, `"counter{stream=\"persistent\"}":1`) + assert.NotContains(t, body, `stream=\"temporary\"`) +} + func TestNamespacedPrefix(t *testing.T) { prom, handler := getTestMetrics(t) diff --git a/internal/component/metrics/type.go b/internal/component/metrics/type.go index b77d8e898..ef131c6fc 100644 --- a/internal/component/metrics/type.go +++ b/internal/component/metrics/type.go @@ -46,6 +46,12 @@ type StatGauge interface { DecrFloat64(count float64) } +// StatDeleter is optionally implemented by stats that can remove themselves +// from their metrics exporter. +type StatDeleter interface { + Delete() +} + //------------------------------------------------------------------------------ // StatCounterVec creates StatCounters with dynamic labels. diff --git a/internal/manager/mock/manager.go b/internal/manager/mock/manager.go index 9fd18f071..290c1f2a0 100644 --- a/internal/manager/mock/manager.go +++ b/internal/manager/mock/manager.go @@ -92,6 +92,9 @@ func (m *Manager) IntoPath(segments ...string) bundle.NewManagement { return m } // WithAddedMetrics returns the same mock manager. func (m *Manager) WithAddedMetrics(m2 metrics.Type) bundle.NewManagement { return m } +// WithMetricsCleanup returns the same mock manager. +func (m *Manager) WithMetricsCleanup() bundle.NewManagement { return m } + // NewBuffer always errors on invalid type. func (m *Manager) NewBuffer(conf buffer.Config) (buffer.Streamed, error) { return nil, component.ErrInvalidType("buffer", conf.Type) diff --git a/internal/manager/type.go b/internal/manager/type.go index 46731bd6e..396b930fc 100644 --- a/internal/manager/type.go +++ b/internal/manager/type.go @@ -457,6 +457,14 @@ func (t *Type) WithAddedMetrics(m metrics.Type) bundle.NewManagement { return &newT } +// WithMetricsCleanup returns a modified manager whose metrics are removed when +// its metrics aggregator is closed. +func (t *Type) WithMetricsCleanup() bundle.NewManagement { + newT := *t + newT.stats = newT.stats.WithCleanup() + return &newT +} + //------------------------------------------------------------------------------ // RegisterEndpoint registers a server wide HTTP endpoint. diff --git a/internal/stream/manager/type.go b/internal/stream/manager/type.go index 0997f1dc4..aa37e908c 100644 --- a/internal/stream/manager/type.go +++ b/internal/stream/manager/type.go @@ -23,14 +23,16 @@ type StreamStatus struct { config stream.Config strm *stream.Type metrics *metrics.Local + metricsScope metrics.Type createdAt time.Time } -func newStreamStatus(conf stream.Config, stats *metrics.Local) *StreamStatus { +func newStreamStatus(conf stream.Config, stats *metrics.Local, metricsScope metrics.Type) *StreamStatus { return &StreamStatus{ - config: conf, - metrics: stats, - createdAt: time.Now(), + config: conf, + metrics: stats, + metricsScope: metricsScope, + createdAt: time.Now(), } } @@ -153,18 +155,19 @@ func (m *Type) Create(id string, conf stream.Config) error { } strmFlatMetrics := metrics.NewLocal() - sMgr := m.manager.ForStream(id).WithAddedMetrics(strmFlatMetrics) + sMgr := m.manager.ForStream(id).WithMetricsCleanup().WithAddedMetrics(strmFlatMetrics) // Note we initialise the status without a stream pointer, this is okay as // long as we do not add it to m.streams without one set. // // This seems a bit wonky but we can't rule out a race condition between // the stream terminating and setClosed and actually initialising a status. - wrapper := newStreamStatus(conf, strmFlatMetrics) + wrapper := newStreamStatus(conf, strmFlatMetrics, sMgr.Metrics()) strm, err := stream.New(conf, sMgr, stream.OptOnClose(func() { wrapper.setClosed() })) if err != nil { + _ = wrapper.metricsScope.Close() return err } @@ -235,6 +238,7 @@ func (m *Type) Delete(ctx context.Context, id string) error { return err } } + _ = wrapper.metricsScope.Close() m.lock.Lock() delete(m.streams, id) @@ -255,7 +259,9 @@ func (m *Type) Stop(ctx context.Context) error { for k, v := range m.streams { go func(id string, strm *StreamStatus) { - if err := strm.strm.Stop(ctx); err != nil { + err := strm.strm.Stop(ctx) + _ = strm.metricsScope.Close() + if err != nil { resultChan <- id } else { resultChan <- "" diff --git a/internal/stream/manager/type_test.go b/internal/stream/manager/type_test.go index fd3c182c5..f85d70f3d 100644 --- a/internal/stream/manager/type_test.go +++ b/internal/stream/manager/type_test.go @@ -4,13 +4,16 @@ package manager import ( "context" + "fmt" "reflect" + "strings" "testing" "time" "github.com/stretchr/testify/require" "github.com/redpanda-data/benthos/v4/internal/component" + "github.com/redpanda-data/benthos/v4/internal/component/metrics" "github.com/redpanda-data/benthos/v4/internal/component/testutil" bmanager "github.com/redpanda-data/benthos/v4/internal/manager" "github.com/redpanda-data/benthos/v4/internal/stream" @@ -31,6 +34,55 @@ output: return c } +func metricConf(t testing.TB, name string) stream.Config { + t.Helper() + + c, err := testutil.StreamFromYAML(fmt.Sprintf(` +input: + generate: + mapping: 'root = deleted()' +pipeline: + processors: + - metric: + type: counter + name: %s +output: + drop: {} +`, name)) + require.NoError(t, err) + return c +} + +func metricsManager(t testing.TB) (*Type, *metrics.Local) { + t.Helper() + + stats := metrics.NewLocal() + res, err := bmanager.New( + bmanager.NewResourceConfig(), + bmanager.OptSetMetrics(metrics.NewNamespaced(stats)), + ) + require.NoError(t, err) + return New(res), stats +} + +func hasStreamMetric(stats *metrics.Local, streamID, name string) bool { + matches := func(path string) bool { + return strings.Contains(path, `stream="`+streamID+`"`) && + (name == "" || strings.Contains(path, name)) + } + for path := range stats.GetCounters() { + if matches(path) { + return true + } + } + for path := range stats.GetTimings() { + if matches(path) { + return true + } + } + return false +} + func TestTypeBasicOperations(t *testing.T) { ctx, done := context.WithTimeout(t.Context(), time.Second*30) defer done() @@ -115,3 +167,61 @@ func TestTypeBasicClose(t *testing.T) { t.Errorf("Unexpected error: %v != %v", act, exp) } } + +func TestTypeDeleteRemovesStreamMetrics(t *testing.T) { + ctx, done := context.WithTimeout(t.Context(), time.Second*30) + defer done() + + mgr, stats := metricsManager(t) + require.NoError(t, mgr.Create("foo", metricConf(t, "stream_test_metric"))) + require.True(t, hasStreamMetric(stats, "foo", "")) + + require.NoError(t, mgr.Delete(ctx, "foo")) + require.False(t, hasStreamMetric(stats, "foo", "")) +} + +func TestTypeRecreateStreamMetrics(t *testing.T) { + ctx, done := context.WithTimeout(t.Context(), time.Second*30) + defer done() + + mgr, stats := metricsManager(t) + conf := metricConf(t, "stream_recreate_metric") + + require.NoError(t, mgr.Create("foo", conf)) + require.True(t, hasStreamMetric(stats, "foo", "stream_recreate_metric")) + require.NoError(t, mgr.Delete(ctx, "foo")) + require.False(t, hasStreamMetric(stats, "foo", "")) + + require.NoError(t, mgr.Create("foo", conf)) + require.True(t, hasStreamMetric(stats, "foo", "stream_recreate_metric")) + require.NoError(t, mgr.Delete(ctx, "foo")) + require.False(t, hasStreamMetric(stats, "foo", "")) +} + +func TestTypeUpdateReplacesStreamMetrics(t *testing.T) { + ctx, done := context.WithTimeout(t.Context(), time.Second*30) + defer done() + + mgr, stats := metricsManager(t) + require.NoError(t, mgr.Create("foo", metricConf(t, "stream_old_metric"))) + require.True(t, hasStreamMetric(stats, "foo", "stream_old_metric")) + + require.NoError(t, mgr.Update(ctx, "foo", metricConf(t, "stream_new_metric"))) + require.False(t, hasStreamMetric(stats, "foo", "stream_old_metric")) + require.True(t, hasStreamMetric(stats, "foo", "stream_new_metric")) +} + +func TestTypeStopRemovesStreamMetrics(t *testing.T) { + ctx, done := context.WithTimeout(t.Context(), time.Second*30) + defer done() + + mgr, stats := metricsManager(t) + require.NoError(t, mgr.Create("foo", metricConf(t, "stream_foo_metric"))) + require.NoError(t, mgr.Create("bar", metricConf(t, "stream_bar_metric"))) + require.True(t, hasStreamMetric(stats, "foo", "")) + require.True(t, hasStreamMetric(stats, "bar", "")) + + require.NoError(t, mgr.Stop(ctx)) + require.False(t, hasStreamMetric(stats, "foo", "")) + require.False(t, hasStreamMetric(stats, "bar", "")) +} diff --git a/public/service/metrics.go b/public/service/metrics.go index 6b95e30fd..34b82ca16 100644 --- a/public/service/metrics.go +++ b/public/service/metrics.go @@ -220,6 +220,12 @@ type MetricsExporterGauge interface { // SetFloat64(value float64) } +// MetricsExporterStatDeleter is optionally implemented by metrics exporter +// stats that support removing their series from the exporter. +type MetricsExporterStatDeleter interface { + Delete() +} + //------------------------------------------------------------------------------ // Implements internal metrics plugin interface. @@ -272,6 +278,12 @@ func (a *airGapGauge) Set(value int64) { a.airGapped.Set(value) } +func (a *airGapGauge) Delete() { + if d, ok := a.airGapped.(MetricsExporterStatDeleter); ok { + d.Delete() + } +} + type airGapCounter struct { airGapped MetricsExporterCounter } @@ -290,6 +302,12 @@ func (a *airGapCounter) IncrFloat64(count float64) { } } +func (a *airGapCounter) Delete() { + if d, ok := a.airGapped.(MetricsExporterStatDeleter); ok { + d.Delete() + } +} + type airGapTiming struct { airGapped MetricsExporterTimer } @@ -298,6 +316,12 @@ func (a *airGapTiming) Timing(val int64) { a.airGapped.Timing(val) } +func (a *airGapTiming) Delete() { + if d, ok := a.airGapped.(MetricsExporterStatDeleter); ok { + d.Delete() + } +} + type airGapCounterVec struct { ctor MetricsExporterCounterCtor }