fix: remove metrics when streams are deleted

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
 

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 {

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,
+		}
 	})
 }