From 89e231ddf0bdda70519202600f4541230f51d97d Mon Sep 17 00:00:00 2001
From: Sam Calder-Mason <sam@puritydev.io>
Date: Fri, 29 Sep 2023 17:16:41 +1000
Subject: [PATCH] fix(processor): Metrics fix

---
 pkg/processor/batch.go   | 20 ++++++++----------
 pkg/processor/metrics.go | 44 ++++++++++++++++------------------------
 2 files changed, 25 insertions(+), 39 deletions(-)

diff --git a/pkg/processor/batch.go b/pkg/processor/batch.go
index 924a7195..52d25e42 100644
--- a/pkg/processor/batch.go
+++ b/pkg/processor/batch.go
@@ -135,10 +135,7 @@ func NewBatchItemProcessor[T any](exporter ItemExporter[T], name string, log log
 		opt(&o)
 	}
 
-	metrics, err := NewMetrics("xatu", name)
-	if err != nil {
-		return nil, err
-	}
+	metrics := DefaultMetrics
 
 	bvp := BatchItemProcessor[T]{
 		e:       exporter,
@@ -165,10 +162,14 @@ func NewBatchItemProcessor[T any](exporter ItemExporter[T], name string, log log
 
 // OnEnd method enqueues a item for later processing.
 func (bvp *BatchItemProcessor[T]) Write(ctx context.Context, s []*T) error {
+	bvp.metrics.SetItemsQueued(bvp.name, float64(len(bvp.queue)))
+
 	if bvp.o.ShippingMethod == ShippingMethodSync {
 		return bvp.ImmediatelyExportItems(ctx, s)
 	}
 
+	bvp.metrics.SetItemsQueued(bvp.name, float64(len(bvp.queue)))
+
 	// Do not enqueue items if we are just going to drop them.
 	if bvp.e == nil {
 		return nil
@@ -205,7 +206,7 @@ func (bvp *BatchItemProcessor[T]) ImmediatelyExportItems(ctx context.Context, it
 
 			err := bvp.exportWithTimeout(ctx, itemsBatch)
 
-			bvp.metrics.IncItemsExportedBy(float64(len(itemsBatch)))
+			bvp.metrics.IncItemsExportedBy(bvp.name, float64(len(itemsBatch)))
 
 			if err != nil {
 				return err
@@ -329,8 +330,6 @@ func WithShippingMethod(method ShippingMethod) BatchItemProcessorOption {
 
 // exportItems is a subroutine of processing and draining the queue.
 func (bvp *BatchItemProcessor[T]) exportItems(ctx context.Context) error {
-	bvp.metrics.SetItemsQueued(float64(len(bvp.queue)))
-
 	bvp.timer.Reset(bvp.o.BatchTimeout)
 
 	bvp.batchMutex.Lock()
@@ -351,13 +350,9 @@ func (bvp *BatchItemProcessor[T]) exportItems(ctx context.Context) error {
 			"total_dropped": atomic.LoadUint32(&bvp.dropped),
 		}).Debug("exporting items")
 
-		defer func() {
-			bvp.metrics.SetItemsDropped(float64(atomic.LoadUint32(&bvp.dropped)))
-		}()
-
 		err := bvp.e.ExportItems(ctx, bvp.batch)
 
-		bvp.metrics.IncItemsExportedBy(float64(countItemsToExport))
+		bvp.metrics.IncItemsExportedBy(bvp.name, float64(countItemsToExport))
 
 		// A new batch is always created after exporting, even if the batch failed to be exported.
 		//
@@ -476,6 +471,7 @@ func (bvp *BatchItemProcessor[T]) enqueueDrop(ctx context.Context, sd *T) bool {
 		return true
 	default:
 		atomic.AddUint32(&bvp.dropped, 1)
+		bvp.metrics.IncItemsDroppedBy(bvp.name, float64(1))
 	}
 
 	return false
diff --git a/pkg/processor/metrics.go b/pkg/processor/metrics.go
index 796d0a19..6f20ab5c 100644
--- a/pkg/processor/metrics.go
+++ b/pkg/processor/metrics.go
@@ -2,15 +2,17 @@ package processor
 
 import "github.com/prometheus/client_golang/prometheus"
 
-type Metrics struct {
-	name string
+var (
+	DefaultMetrics = NewMetrics("xatu")
+)
 
+type Metrics struct {
 	itemsQueued   *prometheus.GaugeVec
-	itemsDropped  *prometheus.GaugeVec
+	itemsDropped  *prometheus.CounterVec
 	itemsExported *prometheus.CounterVec
 }
 
-func NewMetrics(namespace, name string) (*Metrics, error) {
+func NewMetrics(namespace string) *Metrics {
 	if namespace != "" {
 		namespace += "_"
 	}
@@ -23,7 +25,7 @@ func NewMetrics(namespace, name string) (*Metrics, error) {
 			Namespace: namespace,
 			Help:      "Number of items queued",
 		}, []string{"processor"}),
-		itemsDropped: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		itemsDropped: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Name:      "items_dropped_total",
 			Namespace: namespace,
 			Help:      "Number of items dropped",
@@ -35,33 +37,21 @@ func NewMetrics(namespace, name string) (*Metrics, error) {
 		}, []string{"processor"}),
 	}
 
-	dupMetricsCollectorError := "duplicate metrics collector registration attempted"
-
-	if err := prometheus.Register(m.itemsQueued); err != nil && err.Error() != dupMetricsCollectorError {
-		return nil, err
-	}
-
-	if err := prometheus.Register(m.itemsDropped); err != nil && err.Error() != dupMetricsCollectorError {
-		return nil, err
-	}
-
-	if err := prometheus.Register(m.itemsExported); err != nil && err.Error() != dupMetricsCollectorError {
-		return nil, err
-	}
-
-	m.name = name
+	prometheus.MustRegister(m.itemsQueued)
+	prometheus.MustRegister(m.itemsDropped)
+	prometheus.MustRegister(m.itemsExported)
 
-	return m, nil
+	return m
 }
 
-func (m *Metrics) SetItemsQueued(count float64) {
-	m.itemsQueued.WithLabelValues(m.name).Set(count)
+func (m *Metrics) SetItemsQueued(name string, count float64) {
+	m.itemsQueued.WithLabelValues(name).Set(count)
 }
 
-func (m *Metrics) SetItemsDropped(count float64) {
-	m.itemsDropped.WithLabelValues(m.name).Set(count)
+func (m *Metrics) IncItemsDroppedBy(name string, count float64) {
+	m.itemsDropped.WithLabelValues(name).Add(count)
 }
 
-func (m *Metrics) IncItemsExportedBy(count float64) {
-	m.itemsExported.WithLabelValues(m.name).Add(count)
+func (m *Metrics) IncItemsExportedBy(name string, count float64) {
+	m.itemsExported.WithLabelValues(name).Add(count)
 }