Add a gauge to track the number of installed publisher handlers

Signed-off-by: Marcelo E. Magallon <marcelo.magallon@grafana.com>

internal/pusher/metrics.go

@@ -25,12 +25,15 @@ type Metrics struct {
 	// For experimental publisher only
 	DroppedCounter  *prometheus.CounterVec
 	ResponseCounter *prometheus.CounterVec
+
+	InstalledHandlers *prometheus.GaugeVec
 }
 
 var (
-	labelsWithType       = []string{"regionID", "tenantID", "type"}
-	labelsWithTypeStatus = []string{"regionID", "tenantID", "type", "status"}
-	labelsWithTypeReason = []string{"regionID", "tenantID", "type", "reason"}
+	labelsWithType             = []string{"type"}
+	labelsWithTenantType       = []string{"regionID", "tenantID", "type"}
+	labelsWithTenantTypeStatus = []string{"regionID", "tenantID", "type", "status"}
+	labelsWithTenantTypeReason = []string{"regionID", "tenantID", "type", "reason"}
 )
 
 // NewMetrics returns a new set of publisher metrics registered in the given registerer.
@@ -42,7 +45,7 @@ func NewMetrics(promRegisterer prometheus.Registerer) (m Metrics) {
 			Name:      "push_total",
 			Help:      "Total number of push events by type.",
 		},
-		labelsWithType)
+		labelsWithTenantType)
 
 	promRegisterer.MustRegister(m.PushCounter)
 
@@ -53,7 +56,7 @@ func NewMetrics(promRegisterer prometheus.Registerer) (m Metrics) {
 			Name:      "push_errors_total",
 			Help:      "Total number of push errors by type and status.",
 		},
-		labelsWithTypeStatus)
+		labelsWithTenantTypeStatus)
 
 	promRegisterer.MustRegister(m.ErrorCounter)
 
@@ -64,7 +67,7 @@ func NewMetrics(promRegisterer prometheus.Registerer) (m Metrics) {
 			Name:      "push_failed_total",
 			Help:      "Total number of push failures by type.",
 		},
-		labelsWithTypeReason)
+		labelsWithTenantTypeReason)
 
 	promRegisterer.MustRegister(m.FailedCounter)
 
@@ -75,7 +78,7 @@ func NewMetrics(promRegisterer prometheus.Registerer) (m Metrics) {
 			Name:      "push_bytes",
 			Help:      "Total number of bytes pushed by type.",
 		},
-		labelsWithType)
+		labelsWithTenantType)
 
 	promRegisterer.MustRegister(m.BytesOut)
 
@@ -86,7 +89,7 @@ func NewMetrics(promRegisterer prometheus.Registerer) (m Metrics) {
 			Name:      "retries_total",
 			Help:      "Total number of retries performed by type.",
 		},
-		labelsWithType)
+		labelsWithTenantType)
 
 	promRegisterer.MustRegister(m.RetriesCounter)
 
@@ -97,7 +100,7 @@ func NewMetrics(promRegisterer prometheus.Registerer) (m Metrics) {
 			Name:      "drop_total",
 			Help:      "Total number of results dropped by type.",
 		},
-		labelsWithType)
+		labelsWithTenantType)
 
 	promRegisterer.MustRegister(m.DroppedCounter)
 
@@ -108,10 +111,22 @@ func NewMetrics(promRegisterer prometheus.Registerer) (m Metrics) {
 			Name:      "responses_total",
 			Help:      "Total number of responses received by type and status code.",
 		},
-		labelsWithTypeStatus)
+		labelsWithTenantTypeStatus)
 
 	promRegisterer.MustRegister(m.ResponseCounter)
 
+	m.InstalledHandlers = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "sm_agent",
+			Subsystem: "publisher",
+			Name:      "handlers_total",
+			Help:      "Total number of installed publisher handlers.",
+		},
+		labelsWithType,
+	)
+
+	promRegisterer.MustRegister(m.InstalledHandlers)
+
 	return m
 }
 
@@ -123,13 +138,14 @@ func (m Metrics) WithTenant(localID int64, regionID int) Metrics {
 		"tenantID": strconv.FormatInt(localID, 10),
 	}
 	return Metrics{
-		PushCounter:     m.PushCounter.MustCurryWith(labels),
-		ErrorCounter:    m.ErrorCounter.MustCurryWith(labels),
-		BytesOut:        m.BytesOut.MustCurryWith(labels),
-		FailedCounter:   m.FailedCounter.MustCurryWith(labels),
-		RetriesCounter:  m.RetriesCounter.MustCurryWith(labels),
-		DroppedCounter:  m.DroppedCounter.MustCurryWith(labels),
-		ResponseCounter: m.ResponseCounter.MustCurryWith(labels),
+		PushCounter:       m.PushCounter.MustCurryWith(labels),
+		ErrorCounter:      m.ErrorCounter.MustCurryWith(labels),
+		BytesOut:          m.BytesOut.MustCurryWith(labels),
+		FailedCounter:     m.FailedCounter.MustCurryWith(labels),
+		RetriesCounter:    m.RetriesCounter.MustCurryWith(labels),
+		DroppedCounter:    m.DroppedCounter.MustCurryWith(labels),
+		ResponseCounter:   m.ResponseCounter.MustCurryWith(labels),
+		InstalledHandlers: m.InstalledHandlers,
 	}
 }
 
@@ -140,12 +156,13 @@ func (m Metrics) WithType(t string) Metrics {
 	}
 
 	return Metrics{
-		PushCounter:     m.PushCounter.MustCurryWith(typeLabels),
-		ErrorCounter:    m.ErrorCounter.MustCurryWith(typeLabels),
-		BytesOut:        m.BytesOut.MustCurryWith(typeLabels),
-		FailedCounter:   m.FailedCounter.MustCurryWith(typeLabels),
-		RetriesCounter:  m.RetriesCounter.MustCurryWith(typeLabels),
-		DroppedCounter:  m.DroppedCounter.MustCurryWith(typeLabels),
-		ResponseCounter: m.ResponseCounter.MustCurryWith(typeLabels),
+		PushCounter:       m.PushCounter.MustCurryWith(typeLabels),
+		ErrorCounter:      m.ErrorCounter.MustCurryWith(typeLabels),
+		BytesOut:          m.BytesOut.MustCurryWith(typeLabels),
+		FailedCounter:     m.FailedCounter.MustCurryWith(typeLabels),
+		RetriesCounter:    m.RetriesCounter.MustCurryWith(typeLabels),
+		DroppedCounter:    m.DroppedCounter.MustCurryWith(typeLabels),
+		ResponseCounter:   m.ResponseCounter.MustCurryWith(typeLabels),
+		InstalledHandlers: m.InstalledHandlers.MustCurryWith(typeLabels),
 	}
 }

internal/pusher/metrics_test.go

@@ -2,6 +2,7 @@ package pusher
 
 import (
 	"reflect"
+	"sort"
 	"testing"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -16,8 +17,9 @@ func TestNewMetrics(t *testing.T) {
 		for i := 0; i < rVal.NumField(); i++ {
 			fType := rVal.Type().Field(i)
 			fVal := rVal.Field(i)
-			require.Equal(t, reflect.Pointer, fVal.Kind(), fType.Name)
-			require.NotZero(t, fVal.Pointer(), fType.Name)
+			if fVal.Kind() == reflect.Pointer {
+				require.NotZero(t, fVal.Pointer(), fType.Name)
+			}
 		}
 	})
 	t.Run("registered fields", func(t *testing.T) {
@@ -32,12 +34,14 @@ func TestNewMetrics(t *testing.T) {
 		m.BytesOut.WithLabelValues().Add(1200)
 		m.ErrorCounter.WithLabelValues("500").Inc()
 		m.ResponseCounter.WithLabelValues("200").Inc()
+		m.InstalledHandlers.WithLabelValues().Inc()
 
 		fam, err := reg.Gather()
 		require.NoError(t, err)
 		var (
 			expected = []string{
 				"sm_agent_publisher_drop_total",
+				"sm_agent_publisher_handlers_total",
 				"sm_agent_publisher_push_bytes",
 				"sm_agent_publisher_push_errors_total",
 				"sm_agent_publisher_push_failed_total",
@@ -50,6 +54,9 @@ func TestNewMetrics(t *testing.T) {
 			actual = append(actual, metric.GetName())
 		}
 
+		sort.Strings(expected)
+		sort.Strings(actual)
+
 		require.Equal(t, expected, actual)
 	})
 }

internal/pusher/v2/publisher.go

@@ -89,6 +89,17 @@ func (p *publisherImpl) replaceHandler(tenantID model.GlobalID, old, new payload
 	// Get the existing handler if any.
 	current := p.handlers[tenantID]
 
+	//  old     | current | new     | op
+	//  --------+---------+---------+---------------
+	//  nil     | nil     | nil     | delete (noop)
+	//  nil     | nil     | non-nil | add
+	// *nil     | non-nil | nil     | delete
+	// *nil     | non-nil | non-nil | replace
+	// *non-nil | nil     | nil     | delete (noop)
+	// *non-nil | nil     | non-nil | add
+	//  non-nil | non-nil | nil     | delete
+	//  non-nil | non-nil | non-nil | replace
+
 	// If old is nil, that means we are trying to add a handler. If current
 	// is not nil, that means there's an existing handler, and the addition
 	// is not necessary. If current is nil, we go ahead and add the new handler.
@@ -111,6 +122,8 @@ func (p *publisherImpl) replaceHandler(tenantID model.GlobalID, old, new payload
 		delete(p.handlers, tenantID)
 	}
 
+	p.options.metrics.InstalledHandlers.WithLabelValues().Set(float64(len(p.handlers)))
+
 	return new, true
 }