fix(remote): Unregister metrics emitted by remote.WriteStorage when closed (#16868)

* Unregister metrics emitted by `remote.WriteStorage` when closed

Signed-off-by: Charles Korn <charles.korn@grafana.com>

* Address PR feedback: add test

Signed-off-by: Charles Korn <charles.korn@grafana.com>

---------

Signed-off-by: Charles Korn <charles.korn@grafana.com>

storage/remote/write.go

@@ -268,6 +268,14 @@ func (rws *WriteStorage) Close() error {
 		q.Stop()
 	}
 	close(rws.quit)
+
+	rws.watcherMetrics.Unregister()
+	rws.liveReaderMetrics.Unregister()
+
+	if rws.reg != nil {
+		rws.reg.Unregister(rws.highestTimestamp.Gauge)
+	}
+
 	return nil
 }
 

storage/remote/write_test.go

@@ -980,3 +980,12 @@ func (s syncAppender) AppendHistogram(ref storage.SeriesRef, l labels.Labels, t
 	defer s.lock.Unlock()
 	return s.Appender.AppendHistogram(ref, l, t, h, f)
 }
+
+func TestWriteStorage_CanRegisterMetricsAfterClosing(t *testing.T) {
+	dir := t.TempDir()
+	reg := prometheus.NewPedanticRegistry()
+
+	s := NewWriteStorage(nil, reg, dir, time.Millisecond, nil)
+	require.NoError(t, s.Close())
+	require.NotPanics(t, func() { NewWriteStorage(nil, reg, dir, time.Millisecond, nil) })
+}

tsdb/wlog/live_reader.go

@@ -29,6 +29,7 @@ import (
 
 // LiveReaderMetrics holds all metrics exposed by the LiveReader.
 type LiveReaderMetrics struct {
+	reg                    prometheus.Registerer
 	readerCorruptionErrors *prometheus.CounterVec
 }
 
@@ -36,6 +37,7 @@ type LiveReaderMetrics struct {
 // at LiveReader instantiation.
 func NewLiveReaderMetrics(reg prometheus.Registerer) *LiveReaderMetrics {
 	m := &LiveReaderMetrics{
+		reg: reg,
 		readerCorruptionErrors: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Name: "prometheus_tsdb_wal_reader_corruption_errors_total",
 			Help: "Errors encountered when reading the WAL.",
@@ -49,6 +51,15 @@ func NewLiveReaderMetrics(reg prometheus.Registerer) *LiveReaderMetrics {
 	return m
 }
 
+// Unregister unregisters metrics emitted by this instance.
+func (m *LiveReaderMetrics) Unregister() {
+	if m.reg == nil {
+		return
+	}
+
+	m.reg.Unregister(m.readerCorruptionErrors)
+}
+
 // NewLiveReader returns a new live reader.
 func NewLiveReader(logger *slog.Logger, metrics *LiveReaderMetrics, r io.Reader) *LiveReader {
 	lr := &LiveReader{

tsdb/wlog/watcher.go

@@ -73,6 +73,7 @@ type WriteNotified interface {
 }
 
 type WatcherMetrics struct {
+	reg                   prometheus.Registerer
 	recordsRead           *prometheus.CounterVec
 	recordDecodeFails     *prometheus.CounterVec
 	samplesSentPreTailing *prometheus.CounterVec
@@ -113,6 +114,7 @@ type Watcher struct {
 
 func NewWatcherMetrics(reg prometheus.Registerer) *WatcherMetrics {
 	m := &WatcherMetrics{
+		reg: reg,
 		recordsRead: prometheus.NewCounterVec(
 			prometheus.CounterOpts{
 				Namespace: "prometheus",
@@ -171,6 +173,19 @@ func NewWatcherMetrics(reg prometheus.Registerer) *WatcherMetrics {
 	return m
 }
 
+// Unregister unregisters metrics emitted by this instance.
+func (m *WatcherMetrics) Unregister() {
+	if m.reg == nil {
+		return
+	}
+
+	m.reg.Unregister(m.recordsRead)
+	m.reg.Unregister(m.recordDecodeFails)
+	m.reg.Unregister(m.samplesSentPreTailing)
+	m.reg.Unregister(m.currentSegment)
+	m.reg.Unregister(m.notificationsSkipped)
+}
+
 // NewWatcher creates a new WAL watcher for a given WriteTo.
 func NewWatcher(metrics *WatcherMetrics, readerMetrics *LiveReaderMetrics, logger *slog.Logger, name string, writer WriteTo, dir string, sendExemplars, sendHistograms, sendMetadata bool) *Watcher {
 	if logger == nil {