app/promauto: support partial resets

app/metrics.go

@@ -75,7 +75,7 @@ var (
 		Help:      "Gauge set to the peer count of the upstream beacon node",
 	})
 
-	beaconNodeVersionGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
+	beaconNodeVersionGauge = promauto.NewResetGaugeVec(prometheus.GaugeOpts{
 		Namespace: "app",
 		Subsystem: "beacon_node",
 		Name:      "version",

app/monitoringapi.go

@@ -217,23 +217,15 @@ func beaconNodeSyncing(ctx context.Context, eth2Cl eth2client.NodeSyncingProvide
 func beaconNodeVersionMetric(ctx context.Context, eth2Cl eth2wrap.Client, clock clockwork.Clock) {
 	nodeVersionTicker := clock.NewTicker(10 * time.Minute)
 
-	// TODO(corver): Refactor to use ResetGauge.
-	var prevNodeVersion string
 	setNodeVersion := func() {
 		version, err := eth2Cl.NodeVersion(ctx)
 		if err != nil {
 			log.Error(ctx, "Failed to get beacon node version", err)
 			return
 		}
-		if version == prevNodeVersion {
-			return
-		}
 
-		if prevNodeVersion != "" {
-			beaconNodeVersionGauge.WithLabelValues(prevNodeVersion).Set(0)
-		}
+		beaconNodeVersionGauge.Reset()
 		beaconNodeVersionGauge.WithLabelValues(version).Set(1)
-		prevNodeVersion = version
 	}
 
 	go func() {

app/peerinfo/metrics.go

@@ -17,15 +17,15 @@ var (
 		ConstLabels: nil,
 	}, []string{"peer"})
 
-	peerVersion = promauto.NewGaugeVec(prometheus.GaugeOpts{
+	peerVersion = promauto.NewResetGaugeVec(prometheus.GaugeOpts{
 		Namespace:   "app",
 		Subsystem:   "peerinfo",
 		Name:        "version",
 		Help:        "Constant gauge with version label set to peer's charon version.",
 		ConstLabels: nil,
 	}, []string{"peer", "version"})
 
-	peerGitHash = promauto.NewGaugeVec(prometheus.GaugeOpts{
+	peerGitHash = promauto.NewResetGaugeVec(prometheus.GaugeOpts{
 		Namespace:   "app",
 		Subsystem:   "peerinfo",
 		Name:        "git_commit",

app/peerinfo/peerinfo.go

@@ -6,7 +6,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"sync"
 	"testing"
 	"time"
 
@@ -241,13 +240,6 @@ func supportedPeerVersion(peerVersion string, supported []version.SemVer) error
 
 // newMetricsSubmitter returns a prometheus metric submitter.
 func newMetricsSubmitter() metricSubmitter {
-	var (
-		mu sync.Mutex
-		// TODO(corver): Refactor to use ResetGauge.
-		prevVersions  = make(map[string]string)
-		prevGitHashes = make(map[string]string)
-	)
-
 	return func(peerID peer.ID, clockOffset time.Duration, version string, gitHash string,
 		startTime time.Time,
 	) {
@@ -274,20 +266,9 @@ func newMetricsSubmitter() metricSubmitter {
 		}
 		// TODO(corver): Validate version and githash with regex
 
+		peerVersion.Reset(peerName)
 		peerVersion.WithLabelValues(peerName, version).Set(1)
+		peerGitHash.Reset(peerName)
 		peerGitHash.WithLabelValues(peerName, gitHash).Set(1)
-
-		// Clear previous metrics if changed
-		mu.Lock()
-		defer mu.Unlock()
-
-		if prev, ok := prevVersions[peerName]; ok && version != prev {
-			peerVersion.WithLabelValues(peerName, prev).Set(0)
-		}
-		if prev, ok := prevGitHashes[peerName]; ok && gitHash != prev {
-			peerGitHash.WithLabelValues(peerName, prev).Set(0)
-		}
-		prevVersions[peerName] = version
-		prevGitHashes[peerName] = gitHash
 	}
 }

app/promauto/resetgauge.go

@@ -43,14 +43,26 @@ func (g *ResetGaugeVec) WithLabelValues(lvs ...string) prometheus.Gauge {
 	return g.inner.WithLabelValues(lvs...)
 }
 
-// Reset deletes all previously set labels.
-func (g *ResetGaugeVec) Reset() {
+// Reset deletes all previously set labels that match all the given label values.
+// An empty slice will delete all previously set labels.
+func (g *ResetGaugeVec) Reset(lvs ...string) {
 	g.mu.Lock()
 	defer g.mu.Unlock()
 
-	for lv := range g.labels {
-		g.inner.DeleteLabelValues(strings.Split(lv, separator)...)
-	}
+	for label := range g.labels {
+		match := true
+		for _, check := range lvs {
+			if !strings.Contains(label, check) {
+				match = false
+				break
+			}
+		}
 
-	g.labels = make(map[string]bool)
+		if !match {
+			continue
+		}
+
+		g.inner.DeleteLabelValues(strings.Split(label, separator)...)
+		delete(g.labels, label)
+	}
 }

app/promauto/resetgauge_test.go

@@ -16,22 +16,41 @@ const resetTest = "reset_test"
 var testResetGauge = promauto.NewResetGaugeVec(prometheus.GaugeOpts{
 	Name: resetTest,
 	Help: "",
-}, []string{"label"})
+}, []string{"label0", "label1"})
 
 func TestResetGaugeVec(t *testing.T) {
 	registry, err := promauto.NewRegistry(nil)
 	require.NoError(t, err)
 
-	testResetGauge.WithLabelValues("1").Set(1)
+	testResetGauge.WithLabelValues("1", "a").Set(0)
 	assertVecLen(t, registry, resetTest, 1)
 
-	testResetGauge.WithLabelValues("2").Set(2)
+	// Same labels, should not increase length
+	testResetGauge.WithLabelValues("1", "a").Set(1)
+	assertVecLen(t, registry, resetTest, 1)
+
+	testResetGauge.WithLabelValues("2", "b").Set(2)
 	assertVecLen(t, registry, resetTest, 2)
 
 	testResetGauge.Reset()
 	assertVecLen(t, registry, resetTest, 0)
 
-	testResetGauge.WithLabelValues("3").Set(3)
+	testResetGauge.WithLabelValues("3", "c").Set(3)
+	assertVecLen(t, registry, resetTest, 1)
+
+	testResetGauge.WithLabelValues("3", "d").Set(3)
+	assertVecLen(t, registry, resetTest, 2)
+
+	testResetGauge.WithLabelValues("3", "e").Set(3)
+	assertVecLen(t, registry, resetTest, 3)
+
+	testResetGauge.WithLabelValues("4", "z").Set(4)
+	assertVecLen(t, registry, resetTest, 4)
+
+	testResetGauge.Reset("3", "c")
+	assertVecLen(t, registry, resetTest, 3)
+
+	testResetGauge.Reset("3")
 	assertVecLen(t, registry, resetTest, 1)
 }
 

core/scheduler/metrics.go

@@ -49,7 +49,7 @@ var (
 		Help:      "Total balance of a validator by public key",
 	}, []string{"pubkey_full", "pubkey"})
 
-	statusGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
+	statusGauge = promauto.NewResetGaugeVec(prometheus.GaugeOpts{
 		Namespace: "core",
 		Subsystem: "scheduler",
 		Name:      "validator_status",
@@ -77,16 +77,10 @@ func instrumentDuty(duty core.Duty, defSet core.DutyDefinitionSet) {
 
 // newMetricSubmitter returns a function that sets validator balance and status metric.
 func newMetricSubmitter() func(pubkey core.PubKey, totalBal eth2p0.Gwei, status string) {
-	// TODO(corver): Refactor to use ResetGauge.
-	prevStatus := make(map[core.PubKey]string)
-
 	return func(pubkey core.PubKey, totalBal eth2p0.Gwei, status string) {
 		balanceGauge.WithLabelValues(string(pubkey), pubkey.String()).Set(float64(totalBal))
-		statusGauge.WithLabelValues(string(pubkey), pubkey.String(), status).Set(1)
 
-		if prev, ok := prevStatus[pubkey]; ok && prev != status { // Validator status changed
-			statusGauge.WithLabelValues(string(pubkey), pubkey.String(), prev).Set(0)
-		}
-		prevStatus[pubkey] = status
+		statusGauge.Reset(string(pubkey), pubkey.String())
+		statusGauge.WithLabelValues(string(pubkey), pubkey.String(), status).Set(1)
 	}
 }

p2p/p2p.go

@@ -307,7 +307,6 @@ func RegisterConnectionLogger(ctx context.Context, tcpNode host.Host, peerIDs []
 				return
 			case <-ticker.C:
 				// Instrument connection and stream counts.
-				peerStreamGauge.Reset() // Reset stream gauge to clear previously set protocols.
 				counts := make(map[connKey]int)
 				streams := make(map[streamKey]int)
 
@@ -328,6 +327,8 @@ func RegisterConnectionLogger(ctx context.Context, tcpNode host.Host, peerIDs []
 						streams[sKey]++
 					}
 				}
+
+				peerStreamGauge.Reset() // Reset stream gauge to clear previously set protocols.
 				for _, pID := range peerIDs {
 					for _, typ := range []string{addrTypeRelay, addrTypeDirect} {
 						cKey := connKey{PeerName: PeerName(pID), Type: typ}