metrics, observer: purge backend metrics when backend is down for too long

pkg/balance/observer/backend_health.go

@@ -48,7 +48,10 @@ func (bh *BackendHealth) Equals(health BackendHealth) bool {
 }
 
 func (bh *BackendHealth) String() string {
-	str := getHealthLabel(bh.Healthy)
+	str := "down"
+	if bh.Healthy {
+		str = "healthy"
+	}
 	if bh.PingErr != nil {
 		str += fmt.Sprintf(", err: %s", bh.PingErr.Error())
 	}

pkg/balance/observer/backend_observer.go

@@ -20,6 +20,8 @@ const (
 	// If the pool is full, it will still create new goroutines, but not return to the pool after use.
 	goPoolSize = 100
 	goMaxIdle  = time.Minute
+	// The backend metric is retained for 2 hours after it's down.
+	backendMetricRetention = 2 * time.Hour
 )
 
 var _ BackendObserver = (*DefaultBackendObserver)(nil)
@@ -37,6 +39,7 @@ type DefaultBackendObserver struct {
 	sync.Mutex
 	subscribers       map[string]chan HealthResult
 	curBackends       map[string]*BackendHealth
+	downBackends      map[string]time.Time
 	wg                waitgroup.WaitGroup
 	refreshChan       chan struct{}
 	fetcher           BackendFetcher
@@ -61,6 +64,7 @@ func NewDefaultBackendObserver(logger *zap.Logger, config *config.HealthCheck, b
 		fetcher:           backendFetcher,
 		subscribers:       make(map[string]chan HealthResult),
 		curBackends:       make(map[string]*BackendHealth),
+		downBackends:      make(map[string]time.Time),
 		cfgGetter:         cfgGetter,
 	}
 	return bo
@@ -97,6 +101,7 @@ func (bo *DefaultBackendObserver) observe(ctx context.Context) {
 			result.backends = bo.checkHealth(ctx, backendInfo)
 		}
 		bo.updateHealthResult(result)
+		bo.purgeBackendMetrics()
 		bo.notifySubscribers(ctx, result)
 
 		cost := monotime.Since(startTime)
@@ -147,6 +152,22 @@ func (bo *DefaultBackendObserver) checkHealth(ctx context.Context, backends map[
 	return curBackendHealth
 }
 
+// If a backend has been down for more than backendMetricRetention, remove it from the metrics.
+// For auto-scaling case, the metrics will occupy too much volume after weeks.
+func (bo *DefaultBackendObserver) purgeBackendMetrics() {
+	if len(bo.downBackends) == 0 {
+		return
+	}
+	now := time.Now()
+	for addr, ts := range bo.downBackends {
+		if ts.Add(backendMetricRetention).Before(now) {
+			bo.logger.Info("backend is down for too long, purging backend metrics", zap.String("backend", addr))
+			metrics.DelBackend(addr)
+			delete(bo.downBackends, addr)
+		}
+	}
+}
+
 func (bo *DefaultBackendObserver) Subscribe(name string) <-chan HealthResult {
 	ch := make(chan HealthResult, 1)
 	bo.Lock()
@@ -179,7 +200,8 @@ func (bo *DefaultBackendObserver) updateHealthResult(result HealthResult) {
 			}
 			bo.logger.Info("update backend", zap.String("backend_addr", addr),
 				zap.String("prev", prev), zap.String("cur", newHealth.String()))
-			updateBackendStatusMetrics(addr, false, true)
+			updateBackendStatusMetrics(addr, true)
+			delete(bo.downBackends, addr)
 		}
 	}
 	for addr, oldHealth := range bo.curBackends {
@@ -193,7 +215,8 @@ func (bo *DefaultBackendObserver) updateHealthResult(result HealthResult) {
 			}
 			bo.logger.Info("update backend", zap.String("backend_addr", addr),
 				zap.String("prev", oldHealth.String()), zap.String("cur", cur))
-			updateBackendStatusMetrics(addr, true, false)
+			updateBackendStatusMetrics(addr, false)
+			bo.downBackends[addr] = time.Now()
 		}
 	}
 	bo.curBackends = result.backends

pkg/balance/observer/metrics.go

@@ -10,29 +10,24 @@ import (
 	"github.com/pingcap/tiproxy/pkg/util/monotime"
 )
 
-const (
-	healthyLabel   = "healthy"
-	unhealthyLabel = "down"
-)
-
-func updateBackendStatusMetrics(addr string, prevHealthy, curHealthy bool) {
-	metrics.BackendStatusGauge.WithLabelValues(addr, getHealthLabel(prevHealthy)).Set(0)
-	metrics.BackendStatusGauge.WithLabelValues(addr, getHealthLabel(curHealthy)).Set(1)
+func updateBackendStatusMetrics(addr string, healthy bool) {
+	val := 0
+	if healthy {
+		val = 1
+	}
+	metrics.BackendStatusGauge.WithLabelValues(addr).Set(float64(val))
 }
 
 func checkBackendStatusMetrics(addr string, healthy bool) bool {
-	val, err := metrics.ReadGauge(metrics.BackendStatusGauge.WithLabelValues(addr, getHealthLabel(healthy)))
+	val, err := metrics.ReadGauge(metrics.BackendStatusGauge.WithLabelValues(addr))
 	if err != nil {
 		return false
 	}
-	return int(val) == 1
-}
-
-func getHealthLabel(healthy bool) string {
+	expected := 0
 	if healthy {
-		return healthyLabel
+		expected = 1
 	}
-	return unhealthyLabel
+	return int(val) == expected
 }
 
 func setPingBackendMetrics(addr string, startTime monotime.Time) {

pkg/metrics/backend.go

@@ -6,8 +6,7 @@ package metrics
 import "github.com/prometheus/client_golang/prometheus"
 
 const (
-	LblRes    = "res"
-	LblStatus = "status"
+	LblRes = "res"
 )
 
 var (
@@ -17,7 +16,7 @@ var (
 			Subsystem: LabelBackend,
 			Name:      "b_status",
 			Help:      "Gauge of backend status.",
-		}, []string{LblBackend, LblStatus})
+		}, []string{LblBackend})
 
 	GetBackendHistogram = prometheus.NewHistogram(
 		prometheus.HistogramOpts{

pkg/metrics/metrics.go

@@ -8,8 +8,8 @@ package metrics
 
 import (
 	"context"
+	"reflect"
 	"runtime"
-	"sync"
 	"time"
 
 	"github.com/pingcap/tiproxy/lib/util/systimemon"
@@ -46,12 +46,10 @@ func NewMetricsManager() *MetricsManager {
 	return &MetricsManager{}
 }
 
-var registerOnce = &sync.Once{}
-
-// Init registers metrics and pushes metrics to prometheus.
+// Init registers metrics and sets up a monitor.
 func (mm *MetricsManager) Init(ctx context.Context, logger *zap.Logger) {
 	mm.logger = logger
-	registerOnce.Do(registerProxyMetrics)
+	mm.registerProxyMetrics()
 	ctx, mm.cancel = context.WithCancel(ctx)
 	mm.setupMonitor(ctx)
 }
@@ -85,34 +83,62 @@ func (mm *MetricsManager) setupMonitor(ctx context.Context) {
 }
 
 // registerProxyMetrics registers metrics.
-func registerProxyMetrics() {
+func (mm *MetricsManager) registerProxyMetrics() {
 	prometheus.DefaultRegisterer.Unregister(collectors.NewGoCollector())
 	prometheus.MustRegister(collectors.NewGoCollector(collectors.WithGoCollections(collectors.GoRuntimeMetricsCollection | collectors.GoRuntimeMemStatsCollection)))
+	for _, c := range colls {
+		prometheus.MustRegister(c)
+	}
+}
+
+var colls []prometheus.Collector
+
+func init() {
+	colls = []prometheus.Collector{
+		ConnGauge,
+		CreateConnCounter,
+		DisConnCounter,
+		MaxProcsGauge,
+		ServerEventCounter,
+		ServerErrCounter,
+		TimeJumpBackCounter,
+		KeepAliveCounter,
+		QueryTotalCounter,
+		QueryDurationHistogram,
+		HandshakeDurationHistogram,
+		BackendStatusGauge,
+		GetBackendHistogram,
+		GetBackendCounter,
+		PingBackendGauge,
+		BackendConnGauge,
+		HealthCheckCycleGauge,
+		MigrateCounter,
+		MigrateDurationHistogram,
+		InboundBytesCounter,
+		InboundPacketsCounter,
+		OutboundBytesCounter,
+		OutboundPacketsCounter,
+		CrossLocationBytesCounter,
+	}
+}
 
-	prometheus.MustRegister(ConnGauge)
-	prometheus.MustRegister(CreateConnCounter)
-	prometheus.MustRegister(DisConnCounter)
-	prometheus.MustRegister(MaxProcsGauge)
-	prometheus.MustRegister(ServerEventCounter)
-	prometheus.MustRegister(ServerErrCounter)
-	prometheus.MustRegister(TimeJumpBackCounter)
-	prometheus.MustRegister(KeepAliveCounter)
-	prometheus.MustRegister(QueryTotalCounter)
-	prometheus.MustRegister(QueryDurationHistogram)
-	prometheus.MustRegister(HandshakeDurationHistogram)
-	prometheus.MustRegister(BackendStatusGauge)
-	prometheus.MustRegister(GetBackendHistogram)
-	prometheus.MustRegister(GetBackendCounter)
-	prometheus.MustRegister(PingBackendGauge)
-	prometheus.MustRegister(BackendConnGauge)
-	prometheus.MustRegister(HealthCheckCycleGauge)
-	prometheus.MustRegister(MigrateCounter)
-	prometheus.MustRegister(MigrateDurationHistogram)
-	prometheus.MustRegister(InboundBytesCounter)
-	prometheus.MustRegister(InboundPacketsCounter)
-	prometheus.MustRegister(OutboundBytesCounter)
-	prometheus.MustRegister(OutboundPacketsCounter)
-	prometheus.MustRegister(CrossLocationBytesCounter)
+func DelBackend(addr string) {
+	for _, name := range []string{LblBackend, LblFrom, LblTo} {
+		labels := prometheus.Labels{
+			name: addr,
+		}
+		delLabelValues(labels)
+	}
+}
+
+func delLabelValues(labels prometheus.Labels) {
+	for _, c := range colls {
+		field := reflect.Indirect(reflect.ValueOf(c)).FieldByName("MetricVec")
+		if field.IsValid() {
+			vec := field.Interface().(*prometheus.MetricVec)
+			vec.DeletePartialMatch(labels)
+		}
+	}
 }
 
 // ReadCounter reads the value from the counter. It is only used for testing.

pkg/metrics/metrics_test.go

@@ -5,9 +5,15 @@ package metrics
 
 import (
 	"context"
+	"slices"
+	"sort"
+	"strings"
 	"testing"
 
 	"github.com/pingcap/tiproxy/lib/util/logger"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
 )
 
 func TestStartMetricsManager(t *testing.T) {
@@ -16,3 +22,79 @@ func TestStartMetricsManager(t *testing.T) {
 	mm.Init(context.Background(), log)
 	mm.Close()
 }
+
+func TestDelLabelValues(t *testing.T) {
+	tests := []struct {
+		init     func()
+		coll     prometheus.Collector
+		allAddrs []string
+		delAddr  string
+	}{
+		{
+			// test LblBackend + Gauge
+			init: func() {
+				BackendConnGauge.WithLabelValues("addr0").Add(1)
+				BackendConnGauge.WithLabelValues("addr1").Add(1)
+			},
+			coll:     BackendConnGauge,
+			allAddrs: []string{"addr0", "addr1"},
+			delAddr:  "addr0",
+		},
+		{
+			// test LblFrom + LblTo + other labels + Histogram
+			init: func() {
+				MigrateDurationHistogram.WithLabelValues("addr0", "addr1", "succeed").Observe(1.0)
+				MigrateDurationHistogram.WithLabelValues("addr1", "addr2", "fail").Observe(1.0)
+				MigrateDurationHistogram.WithLabelValues("addr0", "addr2", "succeed").Observe(1.0)
+			},
+			coll:     MigrateDurationHistogram,
+			allAddrs: []string{"addr0", "addr1", "addr2"},
+			delAddr:  "addr1",
+		},
+		{
+			// test LblBackend + other labels + Counter
+			init: func() {
+				QueryTotalCounter.WithLabelValues("addr0", "query").Add(1)
+				QueryTotalCounter.WithLabelValues("addr0", "prepare").Add(1)
+				QueryTotalCounter.WithLabelValues("addr1", "query").Add(1)
+			},
+			coll:     QueryTotalCounter,
+			allAddrs: []string{"addr0", "addr1"},
+			delAddr:  "addr0",
+		},
+	}
+
+	getAddrs := func(coll prometheus.Collector) []string {
+		ch := make(chan prometheus.Metric)
+		go func() {
+			coll.Collect(ch)
+			close(ch)
+		}()
+		addrs := make([]string, 0, 3)
+		for m := range ch {
+			var metric dto.Metric
+			require.NoError(t, m.Write(&metric))
+			for _, l := range metric.Label {
+				if strings.HasPrefix(*l.Value, "addr") && !slices.Contains(addrs, *l.Value) {
+					addrs = append(addrs, *l.Value)
+				}
+			}
+		}
+		sort.Slice(addrs, func(i, j int) bool {
+			return addrs[i] < addrs[j]
+		})
+		return addrs
+	}
+	for i, test := range tests {
+		test.init()
+		require.Equal(t, test.allAddrs, getAddrs(test.coll), "%dth test failed", i)
+		DelBackend(test.delAddr)
+		newAddrs := make([]string, 0, 3)
+		for _, addr := range test.allAddrs {
+			if addr != test.delAddr {
+				newAddrs = append(newAddrs, addr)
+			}
+		}
+		require.Equal(t, newAddrs, getAddrs(test.coll), "%dth test failed", i)
+	}
+}