fix(metric_aggregation): Fix duplicate metrics registration (backport…

… k230) (#15144) Co-authored-by: Shantanu Alshi <[email protected]>
grafana · Nov 27, 2024 · 1057389 · 1057389
1 parent 9109ca0
commit 1057389
Show file tree

Hide file tree

Showing 4 changed files with 91 additions and 100 deletions.
diff --git a/pkg/pattern/aggregation/metrics.go b/pkg/pattern/aggregation/metrics.go
@@ -1,12 +1,19 @@
 package aggregation
 
 import (
+	"sync"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 
 	"github.com/grafana/loki/v3/pkg/util/constants"
 )
 
+var (
+	aggMetrics  *Metrics
+	metricsOnce sync.Once
+)
+
 type Metrics struct {
 	reg prometheus.Registerer
 
@@ -28,96 +35,80 @@ type Metrics struct {
 }
 
 func NewMetrics(r prometheus.Registerer) *Metrics {
-	var m Metrics
-	m.reg = r
-
-	m = Metrics{
-		chunks: promauto.With(r).NewGaugeVec(prometheus.GaugeOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "metric_chunks",
-			Help:      "The total number of chunks in memory.",
-		}, []string{"service_name"}),
-		samples: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "metric_samples",
-			Help:      "The total number of samples in memory.",
-		}, []string{"service_name"}),
-		pushErrors: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "push_errors_total",
-			Help:      "Total number of errors when pushing metrics to Loki.",
-		}, []string{"tenant_id", "error_type"}),
-
-		pushRetries: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "push_retries_total",
-			Help:      "Total number of retries when pushing metrics to Loki.",
-		}, []string{"tenant_id"}),
-
-		pushSuccesses: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "push_successes_total",
-			Help:      "Total number of successful pushes to Loki.",
-		}, []string{"tenant_id"}),
-
-		// Batch metrics
-		payloadSize: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "push_payload_bytes",
-			Help:      "Size of push payloads in bytes.",
-			Buckets:   []float64{1024, 4096, 16384, 65536, 262144, 1048576},
-		}, []string{"tenant_id"}),
-
-		streamsPerPush: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "streams_per_push",
-			Help:      "Number of streams in each push request.",
-			Buckets:   []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000},
-		}, []string{"tenant_id"}),
-
-		entriesPerPush: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "entries_per_push",
-			Help:      "Number of entries in each push request.",
-			Buckets:   []float64{10, 50, 100, 500, 1000, 5000, 10000},
-		}, []string{"tenant_id"}),
-
-		servicesTracked: promauto.With(r).NewGaugeVec(prometheus.GaugeOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "services_tracked",
-			Help:      "Number of unique services being tracked.",
-		}, []string{"tenant_id"}),
-		writeTimeout: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
-			Namespace: constants.Loki,
-			Subsystem: "pattern_ingester",
-			Name:      "write_timeouts_total",
-			Help:      "Total number of write timeouts.",
-		}, []string{"tenant_id"}),
-	}
-
-	if m.reg != nil {
-		m.reg.MustRegister(
-			m.chunks,
-			m.samples,
-			m.pushErrors,
-			m.pushRetries,
-			m.pushSuccesses,
-			m.payloadSize,
-			m.streamsPerPush,
-			m.entriesPerPush,
-			m.servicesTracked,
-			m.writeTimeout,
-		)
-	}
-
-	return &m
+	metricsOnce.Do(func() {
+		aggMetrics = &Metrics{
+			chunks: promauto.With(r).NewGaugeVec(prometheus.GaugeOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "metric_chunks",
+				Help:      "The total number of chunks in memory.",
+			}, []string{"service_name"}),
+			samples: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "metric_samples",
+				Help:      "The total number of samples in memory.",
+			}, []string{"service_name"}),
+			pushErrors: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "push_errors_total",
+				Help:      "Total number of errors when pushing metrics to Loki.",
+			}, []string{"tenant_id", "error_type"}),
+
+			pushRetries: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "push_retries_total",
+				Help:      "Total number of retries when pushing metrics to Loki.",
+			}, []string{"tenant_id"}),
+
+			pushSuccesses: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "push_successes_total",
+				Help:      "Total number of successful pushes to Loki.",
+			}, []string{"tenant_id"}),
+
+			// Batch metrics
+			payloadSize: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "push_payload_bytes",
+				Help:      "Size of push payloads in bytes.",
+				Buckets:   []float64{1024, 4096, 16384, 65536, 262144, 1048576},
+			}, []string{"tenant_id"}),
+
+			streamsPerPush: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "streams_per_push",
+				Help:      "Number of streams in each push request.",
+				Buckets:   []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000},
+			}, []string{"tenant_id"}),
+
+			entriesPerPush: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "entries_per_push",
+				Help:      "Number of entries in each push request.",
+				Buckets:   []float64{10, 50, 100, 500, 1000, 5000, 10000},
+			}, []string{"tenant_id"}),
+
+			servicesTracked: promauto.With(r).NewGaugeVec(prometheus.GaugeOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "services_tracked",
+				Help:      "Number of unique services being tracked.",
+			}, []string{"tenant_id"}),
+			writeTimeout: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
+				Namespace: constants.Loki,
+				Subsystem: "pattern_ingester",
+				Name:      "write_timeouts_total",
+				Help:      "Total number of write timeouts.",
+			}, []string{"tenant_id"}),
+		}
+	})
+
+	return aggMetrics
 }
diff --git a/pkg/pattern/aggregation/push.go b/pkg/pattern/aggregation/push.go
@@ -17,7 +17,6 @@ import (
 	"github.com/go-kit/log/level"
 	"github.com/golang/snappy"
 	"github.com/opentracing/opentracing-go"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/config"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/labels"
@@ -112,7 +111,7 @@ func NewPush(
 	useTLS bool,
 	backoffCfg *backoff.Config,
 	logger log.Logger,
-	registrer prometheus.Registerer,
+	metrics *Metrics,
 ) (*Push, error) {
 	client, err := config.NewClientFromConfig(cfg, "pattern-ingester-push", config.WithHTTP2Disabled())
 	if err != nil {
@@ -147,7 +146,7 @@ func NewPush(
 		entries: entries{
 			entries: make([]entry, 0),
 		},
-		metrics: NewMetrics(registrer),
+		metrics: metrics,
 	}
 
 	go p.run(pushPeriod)
@@ -284,12 +283,12 @@ func (p *Push) run(pushPeriod time.Duration) {
 				}
 
 				if !backoff.Ongoing() {
-					level.Error(p.logger).Log("msg", "failed to send entry, retries exhausted, entry will be dropped", "entry", "status", status, "error", err)
+					level.Error(p.logger).Log("msg", "failed to send entry, retries exhausted, entry will be dropped", "status", status, "error", err)
 					pushTicker.Reset(pushPeriod)
 					break
 				}
 				level.Warn(p.logger).
-					Log("msg", "failed to send entry, retrying", "entry", "status", status, "error", err)
+					Log("msg", "failed to send entry, retrying", "status", status, "error", err)
 				backoff.Wait()
 			}
 

diff --git a/pkg/pattern/aggregation/push_test.go b/pkg/pattern/aggregation/push_test.go
@@ -58,7 +58,7 @@ func Test_Push(t *testing.T) {
 			false,
 			&backoff,
 			log.NewNopLogger(),
-			nil,
+			NewMetrics(nil),
 		)
 		require.NoError(t, err)
 		ts, payload := testPayload()
@@ -83,7 +83,7 @@ func Test_Push(t *testing.T) {
 			"user", "secret",
 			false,
 			&backoff,
-			log.NewNopLogger(), nil,
+			log.NewNopLogger(), NewMetrics(nil),
 		)
 		require.NoError(t, err)
 		ts, payload := testPayload()

diff --git a/pkg/pattern/ingester.go b/pkg/pattern/ingester.go
@@ -392,6 +392,7 @@ func (i *Ingester) GetOrCreateInstance(instanceID string) (*instance, error) { /
 
 		aggCfg := i.cfg.MetricAggregation
 		if i.limits.MetricAggregationEnabled(instanceID) {
+			metricAggregationMetrics := aggregation.NewMetrics(i.registerer)
 			writer, err = aggregation.NewPush(
 				aggCfg.LokiAddr,
 				instanceID,
@@ -403,7 +404,7 @@ func (i *Ingester) GetOrCreateInstance(instanceID string) (*instance, error) { /
 				aggCfg.UseTLS,
 				&aggCfg.BackoffConfig,
 				i.logger,
-				i.registerer,
+				metricAggregationMetrics,
 			)
 			if err != nil {
 				return nil, err