Repository: pingcap/tiproxy (go)

Issue:
Use tikv/pd request backoff rate instead of backoff number for judging health
## Development Task

Currently, the TiDB health is judged by the backoff number of TiKV and PD. When one TiDB has more transactions, the backoff number is higher. If the connections are migrated to the other TiDB, the other TiDB may have a higher backoff number instead. This leads to frequent migrations.

A better way is to calculate the backoff rate.

Pull request description:
balance: fix TiDB regionMiss backoff causes TiProxy to incorrectly migrate connections
<!--

Thank you for contributing to TiProxy!

PR Title Format:
1. pkg [, pkg2, pkg3]: what's changed
2. *: what's changed

-->

### What problem does this PR solve?
<!--

Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close: #xxx" or "ref: #xxx".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.

-->

Issue Number: close #828

Problem Summary:
- When TiDB executes heavy DML, there may be many regionMiss, which causes TiProxy to incorrectly take TiDB as unhealthy.
- When connections are migrated to one TiDB, the QPS is higher and has more backoff. Judging by the backoff number is unfair.
- There may be panic when TiDB locations are the same.
- If the router panics, the panic reason is not logged.

What is changed and how it works:
- Remove regionMiss backoffs in the health judgment
- Change backoff number to backoff rate in the health judgment
- Fix possible panic
- Recover router panic and log the panic stack

### Check List

Tests <!-- At least one of them must be included. -->

- [x] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No code

Notable changes

- [ ] Has configuration change
- [ ] Has HTTP API interfaces change
- [ ] Has tiproxyctl change
- [ ] Other user behavior changes

### Release note

<!-- compatibility change, improvement, bugfix, and new feature need a release note -->

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

```release-note
- Fix TiDB regionMiss backoff causes TiProxy to incorrectly migrate connections
```


Patch:
diff --git a/pkg/balance/factor/factor_balance.go b/pkg/balance/factor/factor_balance.go
index 8a41bf7..76a9dc4 100644
--- a/pkg/balance/factor/factor_balance.go
+++ b/pkg/balance/factor/factor_balance.go
@@ -279,9 +279,7 @@ func (fbb *FactorBasedBalance) BackendsToBalance(backends []policy.BackendCtx) (
 			if score1 >= score2 {
 				// The factors with higher priorities are ordered, so this factor shouldn't violate them.
 				// E.g. if the CPU usage of A is higher than B, don't migrate from B to A even if A is preferred in location.
-				var advice BalanceAdvice
-				var fields []zap.Field
-				advice, balanceCount, fields = factor.BalanceCount(scoredBackends[i], scoredBackends[0])
+				advice, count, fields := factor.BalanceCount(scoredBackends[i], scoredBackends[0])
 				if advice == AdviceNegtive {
 					// If the factor will be unbalanced after migration, skip the rest factors.
 					// E.g. if the CPU usage of A will be much higher than B after migration,
@@ -289,8 +287,9 @@ func (fbb *FactorBasedBalance) BackendsToBalance(backends []policy.BackendCtx) (
 					break
 				}
 				logFields = append(logFields, fields...)
-				if score1 > score2 && advice == AdvicePositive && balanceCount > 0.0001 {
+				if score1 > score2 && advice == AdvicePositive && count > 0.0001 {
 					from, to = scoredBackends[i].BackendCtx, scoredBackends[0].BackendCtx
+					balanceCount = count
 					reason = factor.Name()
 					logFields = append(logFields, zap.String("factor", reason),
 						zap.String("from_total_score", strconv.FormatUint(scoredBackends[i].scoreBits, 16)),
diff --git a/pkg/balance/factor/factor_cpu.go b/pkg/balance/factor/factor_cpu.go
index adc0ede..ec46220 100644
--- a/pkg/balance/factor/factor_cpu.go
+++ b/pkg/balance/factor/factor_cpu.go
@@ -41,6 +41,12 @@ var (
 		Names:     []string{"process_cpu_seconds_total", "tidb_server_maxprocs"},
 		Retention: 1 * time.Minute,
 		Metric2Value: func(mfs map[string]*dto.MetricFamily) model.SampleValue {
+			if mfs["process_cpu_seconds_total"] == nil || len(mfs["process_cpu_seconds_total"].Metric) == 0 {
+				return model.SampleValue(math.NaN())
+			}
+			if mfs["tidb_server_maxprocs"] == nil || len(mfs["tidb_server_maxprocs"].Metric) == 0 {
+				return model.SampleValue(math.NaN())
+			}
 			cpuTotal := mfs["process_cpu_seconds_total"].Metric[0].Untyped
 			maxProcs := mfs["tidb_server_maxprocs"].Metric[0].Untyped
 			if cpuTotal == nil || maxProcs == nil {
diff --git a/pkg/balance/factor/factor_health.go b/pkg/balance/factor/factor_health.go
index f2637fa..8606fc2 100644
--- a/pkg/balance/factor/factor_health.go
+++ b/pkg/balance/factor/factor_health.go
@@ -35,11 +35,14 @@ const (
 )
 
 type errDefinition struct {
-	queryRule        metricsreader.QueryRule
-	promQL           string
-	key              string
-	failThreshold    int
-	recoverThreshold int
+	queryFailureRule metricsreader.QueryRule
+	queryTotalRule   metricsreader.QueryRule
+	failurePromQL    string
+	totalPromQL      string
+	failureKey       string
+	totalKey         string
+	failThreshold    float64
+	recoverThreshold float64
 }
 
 var (
@@ -64,109 +67,121 @@ var (
 	errDefinitions = []errDefinition{
 		{
 			// may be caused by disconnection to PD
-			// test with no connection in no network: around 80/m
-			// test with 100 connections in unstable network: [50, 135]/2m
-			// test with 50 TPS in unstable network: around 500/2m, the higher the TPS, the higher the metric value
-			promQL:           `sum(increase(tidb_tikvclient_backoff_seconds_count{type="pdRPC"}[2m])) by (instance)`,
-			failThreshold:    500,
-			recoverThreshold: 50,
-			key:              "health_pd",
-			queryRule: metricsreader.QueryRule{
-				Names:     []string{"tidb_tikvclient_backoff_seconds_count"},
+			failureKey:       "failure_pd",
+			totalKey:         "total_pd",
+			failThreshold:    0.5,
+			recoverThreshold: 0.1,
+			failurePromQL:    `sum(increase(pd_client_cmd_handle_failed_cmds_duration_seconds_count{type="tso"}[2m])) by (instance)`,
+			queryFailureRule: metricsreader.QueryRule{
+				Names:     []string{"pd_client_cmd_handle_failed_cmds_duration_seconds_count"},
 				Retention: 2 * time.Minute,
 				Metric2Value: func(mfs map[string]*dto.MetricFamily) model.SampleValue {
-					mt := mfs["tidb_tikvclient_backoff_seconds_count"].Metric
-					total := 0
-					for _, m := range mt {
-						for _, label := range m.Label {
-							if *label.Name == "type" {
-								if *label.Value == "pdRPC" && m.Untyped != nil {
-									total += int(*m.Untyped.Value)
-								}
-								break
-							}
-						}
-					}
-					return model.SampleValue(total)
+					return generalMetric2Value(mfs, "pd_client_cmd_handle_failed_cmds_duration_seconds_count", "tso")
 				},
-				Range2Value: func(pairs []model.SamplePair) model.SampleValue {
-					if len(pairs) < 2 {
-						return model.SampleValue(math.NaN())
-					}
-					diff := pairs[len(pairs)-1].Value - pairs[0].Value
-					// Maybe the backend just rebooted.
-					if diff < 0 {
-						return model.SampleValue(math.NaN())
-					}
-					return diff
+				Range2Value: generalRange2Value,
+				ResultType:  model.ValVector,
+			},
+			// pd_client_cmd_handle_cmds_duration_seconds_count only records successful commands
+			totalPromQL: `sum(increase(pd_client_cmd_handle_cmds_duration_seconds_count{type="tso"}[2m])) by (instance)`,
+			queryTotalRule: metricsreader.QueryRule{
+				Names:     []string{"pd_client_cmd_handle_cmds_duration_seconds_count"},
+				Retention: 2 * time.Minute,
+				Metric2Value: func(mfs map[string]*dto.MetricFamily) model.SampleValue {
+					return generalMetric2Value(mfs, "pd_client_cmd_handle_cmds_duration_seconds_count", "tso")
 				},
-				ResultType: model.ValVector,
+				Range2Value: generalRange2Value,
+				ResultType:  model.ValVector,
 			},
 		},
 		{
 			// may be caused by disconnection to TiKV
-			// test with no connection in no network: regionMiss is around 1300/m, tikvRPC is around 40/m
-			// test with 100 idle connections in unstable network: [1000, 3300]/2m
-			// test with 100 QPS in unstable network: around 1000/2m
-			// test with 50 TPS in unstable network: around 5000/2m, the higher the TPS, the higher the metric value
-			promQL:           `sum(increase(tidb_tikvclient_backoff_seconds_count{type=~"regionMiss|tikvRPC"}[2m])) by (instance)`,
-			failThreshold:    5000,
-			recoverThreshold: 500,
-			key:              "health_tikv",
-			queryRule: metricsreader.QueryRule{
+			failureKey:       "failure_tikv",
+			totalKey:         "total_tikv",
+			failThreshold:    0.3,
+			recoverThreshold: 0.1,
+			failurePromQL:    `sum(increase(tidb_tikvclient_backoff_seconds_count{type="tikvRPC"}[2m])) by (instance)`,
+			queryFailureRule: metricsreader.QueryRule{
 				Names:     []string{"tidb_tikvclient_backoff_seconds_count"},
 				Retention: 2 * time.Minute,
 				Metric2Value: func(mfs map[string]*dto.MetricFamily) model.SampleValue {
-					mt := mfs["tidb_tikvclient_backoff_seconds_count"].Metric
-					total := 0
-					for _, m := range mt {
-						for _, label := range m.Label {
-							if *label.Name == "type" {
-								if (*label.Value == "regionMiss" || *label.Value == "tikvRPC") && m.Untyped != nil {
-									total += int(*m.Untyped.Value)
-								}
-								break
-							}
-						}
-					}
-					return model.SampleValue(total)
+					return generalMetric2Value(mfs, "tidb_tikvclient_backoff_seconds_count", "tikvRPC")
 				},
-				Range2Value: func(pairs []model.SamplePair) model.SampleValue {
-					if len(pairs) < 2 {
-						return model.SampleValue(math.NaN())
-					}
-					diff := pairs[len(pairs)-1].Value - pairs[0].Value
-					// Maybe the backend just rebooted.
-					if diff < 0 {
-						return model.SampleValue(math.NaN())
-					}
-					return diff
+				Range2Value: generalRange2Value,
+				ResultType:  model.ValVector,
+			},
+			totalPromQL: `sum(increase(tidb_tikvclient_request_counter[2m])) by (instance)`,
+			queryTotalRule: metricsreader.QueryRule{
+				Names:     []string{"tidb_tikvclient_request_counter"},
+				Retention: 2 * time.Minute,
+				Metric2Value: func(mfs map[string]*dto.MetricFamily) model.SampleValue {
+					return generalMetric2Value(mfs, "tidb_tikvclient_request_counter", "")
 				},
-				ResultType: model.ValVector,
+				Range2Value: generalRange2Value,
+				ResultType:  model.ValVector,
 			},
 		},
 	}
 )
 
+func generalMetric2Value(mfs map[string]*dto.MetricFamily, metricName string, typeValue string) model.SampleValue {
+	fm, ok := mfs[metricName]
+	if !ok {
+		return model.SampleValue(math.NaN())
+	}
+	total := 0
+	for _, m := range fm.Metric {
+		for _, label := range m.Label {
+			if *label.Name == "type" {
+				if (typeValue == "" || *label.Value == typeValue) && m.Untyped != nil {
+					total += int(*m.Untyped.Value)
+				}
+				break
+			}
+		}
+	}
+	return model.SampleValue(total)
+}
+
+func generalRange2Value(pairs []model.SamplePair) model.SampleValue {
+	if len(pairs) < 2 {
+		return model.SampleValue(math.NaN())
+	}
+	if math.IsNaN(float64(pairs[0].Value)) || math.IsNaN(float64(pairs[len(pairs)-1].Value)) {
+		return model.SampleValue(math.NaN())
+	}
+	diff := pairs[len(pairs)-1].Value - pairs[0].Value
+	// Maybe the backend just rebooted.
+	if diff < 0 {
+		return model.SampleValue(math.NaN())
+	}
+	return diff
+}
+
 var _ Factor = (*FactorHealth)(nil)
 
 // The snapshot of backend statistics when the metric was updated.
 type healthBackendSnapshot struct {
 	updatedTime time.Time
 	valueRange  valueRange
-	indicator   string
-	value       int
+	// indicator, failureValue, and totalValue are the info of the failed indicator.
+	indicator    string
+	failureValue float64
+	totalValue   float64
 	// Record the balance count when the backend becomes unhealthy so that it won't be smaller in the next rounds.
 	balanceCount float64
 }
 
 type errIndicator struct {
-	queryExpr        metricsreader.QueryExpr
-	queryRule        metricsreader.QueryRule
-	queryResult      metricsreader.QueryResult
-	key              string
-	failThreshold    int
-	recoverThreshold int
+	queryFailureRule   metricsreader.QueryRule
+	queryTotalRule     metricsreader.QueryRule
+	queryFailureExpr   metricsreader.QueryExpr
+	queryTotalExpr     metricsreader.QueryExpr
+	queryFailureResult metricsreader.QueryResult
+	queryTotalResult   metricsreader.QueryResult
+	failureKey         string
+	totalKey           string
+	failThreshold      float64
+	recoverThreshold   float64
 }
 
 type FactorHealth struct {
@@ -191,15 +206,21 @@ func initErrIndicator(mr metricsreader.MetricsReader) []errIndicator {
 	indicators := make([]errIndicator, 0, len(errDefinitions))
 	for _, def := range errDefinitions {
 		indicator := errIndicator{
-			queryExpr: metricsreader.QueryExpr{
-				PromQL: def.promQL,
+			queryFailureExpr: metricsreader.QueryExpr{
+				PromQL: def.failurePromQL,
 			},
-			queryRule:        def.queryRule,
-			key:              def.key,
+			queryTotalExpr: metricsreader.QueryExpr{
+				PromQL: def.totalPromQL,
+			},
+			queryFailureRule: def.queryFailureRule,
+			queryTotalRule:   def.queryTotalRule,
+			failureKey:       def.failureKey,
+			totalKey:         def.totalKey,
 			failThreshold:    def.failThreshold,
 			recoverThreshold: def.recoverThreshold,
 		}
-		mr.AddQueryExpr(indicator.key, indicator.queryExpr, indicator.queryRule)
+		mr.AddQueryExpr(indicator.failureKey, indicator.queryFailureExpr, indicator.queryFailureRule)
+		mr.AddQueryExpr(indicator.totalKey, indicator.queryTotalExpr, indicator.queryTotalRule)
 		indicators = append(indicators, indicator)
 	}
 	return indicators
@@ -215,16 +236,27 @@ func (fh *FactorHealth) UpdateScore(backends []scoredBackend) {
 	}
 	needUpdateSnapshot, latestTime := false, time.Time{}
 	for i := 0; i < len(fh.indicators); i++ {
-		qr := fh.mr.GetQueryResult(fh.indicators[i].key)
-		if qr.Empty() {
+		failureQR := fh.mr.GetQueryResult(fh.indicators[i].failureKey)
+		if failureQR.Empty() {
+			continue
+		}
+		totalQR := fh.mr.GetQueryResult(fh.indicators[i].totalKey)
+		if totalQR.Empty() {
 			continue
 		}
-		if fh.indicators[i].queryResult.UpdateTime != qr.UpdateTime {
-			fh.indicators[i].queryResult = qr
+		if fh.indicators[i].queryFailureResult.UpdateTime != failureQR.UpdateTime {
+			fh.indicators[i].queryFailureResult = failureQR
+			needUpdateSnapshot = true
+		}
+		if failureQR.UpdateTime.After(latestTime) {
+			latestTime = failureQR.UpdateTime
+		}
+		if fh.indicators[i].queryTotalResult.UpdateTime != totalQR.UpdateTime {
+			fh.indicators[i].queryTotalResult = totalQR
 			needUpdateSnapshot = true
 		}
-		if qr.UpdateTime.After(latestTime) {
-			latestTime = qr.UpdateTime
+		if totalQR.UpdateTime.After(latestTime) {
+			latestTime = totalQR.UpdateTime
 		}
 	}
 	if time.Since(latestTime) > errMetricExpDuration {
@@ -249,9 +281,12 @@ func (fh *FactorHealth) updateSnapshot(backends []scoredBackend) {
 	for _, backend := range backends {
 		addr := backend.Addr()
 		// Get the current value range.
-		updatedTime, valueRange, indicator, value := time.Time{}, valueRangeNormal, "", 0
-		for i := 0; i < len(fh.indicators); i++ {
-			ts := fh.indicators[i].queryResult.UpdateTime
+		updatedTime, valueRange, indicator, failureValue, totalValue := time.Time{}, valueRangeNormal, "", 0.0, 0.0
+		for _, ind := range fh.indicators {
+			ts := ind.queryFailureResult.UpdateTime
+			if ind.queryTotalResult.UpdateTime.After(ts) {
+				ts = ind.queryTotalResult.UpdateTime
+			}
 			if ts.Add(errMetricExpDuration).Before(now) {
 				// The metrics have not been updated for a long time (maybe Prometheus is unavailable).
 				continue
@@ -259,16 +294,18 @@ func (fh *FactorHealth) updateSnapshot(backends []scoredBackend) {
 			if ts.After(updatedTime) {
 				updatedTime = ts
 			}
-			sample := fh.indicators[i].queryResult.GetSample4Backend(backend)
-			if sample == nil {
+			failureSample, totalSample := ind.queryFailureResult.GetSample4Backend(backend), ind.queryTotalResult.GetSample4Backend(backend)
+			if failureSample == nil || totalSample == nil {
 				continue
 			}
-			indicator = fh.indicators[i].key
-			metrics.BackendMetricGauge.WithLabelValues(addr, indicator).Set(float64(sample.Value))
-			vr := calcValueRange(sample, fh.indicators[i])
+			metrics.BackendMetricGauge.WithLabelValues(addr, ind.failureKey).Set(float64(failureSample.Value))
+			metrics.BackendMetricGauge.WithLabelValues(addr, ind.totalKey).Set(float64(totalSample.Value))
+			fv, tv, vr := calcValueRange(failureSample, totalSample, ind)
 			if vr > valueRange {
 				valueRange = vr
-				value = int(sample.Value)
+				failureValue = fv
+				totalValue = tv
+				indicator = ind.failureKey
 			}
 		}
 		// If the metric is unavailable, try to reuse the latest one.
@@ -291,7 +328,8 @@ func (fh *FactorHealth) updateSnapshot(backends []scoredBackend) {
 				zap.String("addr", addr),
 				zap.Int("risk", int(valueRange)),
 				zap.String("indicator", indicator),
-				zap.Int("value", value),
+				zap.Float64("failure_value", failureValue),
+				zap.Float64("total_value", totalValue),
 				zap.Int("last_risk", int(snapshot.valueRange)),
 				zap.Float64("balance_count", balanceCount),
 				zap.Int("conn_score", backend.ConnScore()))
@@ -300,7 +338,8 @@ func (fh *FactorHealth) updateSnapshot(backends []scoredBackend) {
 			updatedTime:  updatedTime,
 			valueRange:   valueRange,
 			indicator:    indicator,
-			value:        value,
+			failureValue: failureValue,
+			totalValue:   totalValue,
 			balanceCount: balanceCount,
 		}
 	}
@@ -313,31 +352,42 @@ func (fh *FactorHealth) updateSnapshot(backends []scoredBackend) {
 	}
 }
 
-func calcValueRange(sample *model.Sample, indicator errIndicator) valueRange {
+func calcValueRange(failureSample, totalSample *model.Sample, indicator errIndicator) (float64, float64, valueRange) {
 	// A backend is typically normal, so if its metric misses, take it as normal.
-	if sample == nil {
-		return valueRangeNormal
+	if failureSample == nil || totalSample == nil {
+		return 0, 0, valueRangeNormal
+	}
+
+	failureValue := float64(failureSample.Value)
+	totalValue := float64(totalSample.Value)
+	if math.IsNaN(failureValue) || math.IsNaN(totalValue) {
+		return failureValue, totalValue, valueRangeNormal
 	}
-	if math.IsNaN(float64(sample.Value)) {
-		return valueRangeNormal
+
+	if failureValue == 0 {
+		return failureValue, totalValue, valueRangeNormal
 	}
-	value := int(sample.Value)
+	if totalValue == 0 {
+		return failureValue, totalValue, valueRangeAbnormal
+	}
+
+	value := failureValue / totalValue
 	if indicator.failThreshold > indicator.recoverThreshold {
 		switch {
 		case value <= indicator.recoverThreshold:
-			return valueRangeNormal
+			return failureValue, totalValue, valueRangeNormal
 		case value >= indicator.failThreshold:
-			return valueRangeAbnormal
+			return failureValue, totalValue, valueRangeAbnormal
 		}
 	} else {
 		switch {
 		case value >= indicator.recoverThreshold:
-			return valueRangeNormal
+			return failureValue, totalValue, valueRangeNormal
 		case value <= indicator.failThreshold:
-			return valueRangeAbnormal
+			return failureValue, totalValue, valueRangeAbnormal
 		}
 	}
-	return valueRangeMid
+	return failureValue, totalValue, valueRangeMid
 }
 
 func (fh *FactorHealth) caclErrScore(addr string) int {
@@ -354,9 +404,14 @@ func (fh *FactorHealth) BalanceCount(from, to scoredBackend) (BalanceAdvice, flo
 	fromScore := fh.caclErrScore(from.Addr())
 	toScore := fh.caclErrScore(to.Addr())
 	snapshot := fh.snapshot[from.Addr()]
-	fields := []zap.Field{
-		zap.String("indicator", snapshot.indicator),
-		zap.Int("value", snapshot.value)}
+	var fields []zap.Field
+	if snapshot.indicator != "" {
+		fields = append(fields,
+			zap.String("indicator", snapshot.indicator),
+			zap.Float64("failure_value", snapshot.failureValue),
+			zap.Float64("total_value", snapshot.totalValue),
+		)
+	}
 	if fromScore-toScore <= 1 {
 		return AdviceNeutral, 0, fields
 	}
@@ -372,6 +427,7 @@ func (fh *FactorHealth) CanBeRouted(_ uint64) bool {
 
 func (fh *FactorHealth) Close() {
 	for _, indicator := range fh.indicators {
-		fh.mr.RemoveQueryExpr(indicator.key)
+		fh.mr.RemoveQueryExpr(indicator.failureKey)
+		fh.mr.RemoveQueryExpr(indicator.totalKey)
 	}
 }
diff --git a/pkg/balance/factor/factor_memory.go b/pkg/balance/factor/factor_memory.go
index 95257ea..792025a 100644
--- a/pkg/balance/factor/factor_memory.go
+++ b/pkg/balance/factor/factor_memory.go
@@ -38,6 +38,12 @@ var (
 		Names:     []string{"process_resident_memory_bytes", "tidb_server_memory_quota_bytes"},
 		Retention: 1 * time.Minute,
 		Metric2Value: func(mfs map[string]*dto.MetricFamily) model.SampleValue {
+			if mfs["process_resident_memory_bytes"] == nil || len(mfs["process_resident_memory_bytes"].Metric) == 0 {
+				return model.SampleValue(math.NaN())
+			}
+			if mfs["tidb_server_memory_quota_bytes"] == nil || len(mfs["tidb_server_memory_quota_bytes"].Metric) == 0 {
+				return model.SampleValue(math.NaN())
+			}
 			memoryUsage := mfs["process_resident_memory_bytes"].Metric[0].Untyped
 			memoryQuota := mfs["tidb_server_memory_quota_bytes"].Metric[0].Untyped
 			if memoryUsage == nil || memoryQuota == nil {
diff --git a/pkg/balance/metricsreader/metrics_reader.go b/pkg/balance/metricsreader/metrics_reader.go
index ffd6126..03e9077 100644
--- a/pkg/balance/metricsreader/metrics_reader.go
+++ b/pkg/balance/metricsreader/metrics_reader.go
@@ -101,7 +101,7 @@ func (dmr *DefaultMetricsReader) readMetrics(ctx context.Context) {
 	}
 	backendErr := dmr.backendReader.ReadMetrics(ctx)
 	if backendErr == nil {
-		dmr.setSource(sourceBackend, promErr)
+		dmr.setSource(sourceBackend, nil)
 		return
 	}
 	dmr.lg.Warn("read metrics failed", zap.NamedError("prometheus", promErr), zap.NamedError("backends", backendErr))
diff --git a/pkg/balance/router/router_score.go b/pkg/balance/router/router_score.go
index da23159..857eae9 100644
--- a/pkg/balance/router/router_score.go
+++ b/pkg/balance/router/router_score.go
@@ -64,10 +64,10 @@ func (r *ScoreBasedRouter) Init(ctx context.Context, ob observer.BackendObserver
 	childCtx, cancelFunc := context.WithCancel(ctx)
 	r.cancelFunc = cancelFunc
 	r.cfgCh = cfgCh
-	// Failing to rebalance backends may cause even more serious problems than TiProxy reboot, so we don't recover panics.
-	r.wg.Run(func() {
+	// Log the panic.
+	r.wg.RunWithRecover(func() {
 		r.rebalanceLoop(childCtx)
-	})
+	}, nil, r.logger)
 }
 
 // GetBackendSelector implements Router.GetBackendSelector interface.


Test patch:
diff --git a/pkg/balance/factor/factor_health_test.go b/pkg/balance/factor/factor_health_test.go
index edba62a..c233bd5 100644
--- a/pkg/balance/factor/factor_health_test.go
+++ b/pkg/balance/factor/factor_health_test.go
@@ -4,6 +4,7 @@
 package factor
 
 import (
+	"fmt"
 	"math"
 	"sort"
 	"strings"
@@ -23,58 +24,51 @@ func TestHealthScore(t *testing.T) {
 		score     uint64
 	}{
 		{
-			errCounts: []float64{math.NaN(), math.NaN()},
+			errCounts: []float64{math.NaN(), math.NaN(), math.NaN(), math.NaN()},
 			score:     0,
 		},
 		{
-			errCounts: []float64{math.NaN(), 0},
+			errCounts: []float64{math.NaN(), 0, 0, math.NaN()},
 			score:     0,
 		},
 		{
-			errCounts: []float64{float64(errDefinitions[0].failThreshold + 1), 0},
+			errCounts: []float64{float64(errDefinitions[0].failThreshold*100 + 1), 100, float64(errDefinitions[1].failThreshold - 1), 100},
 			score:     2,
 		},
 		{
-			errCounts: []float64{float64(errDefinitions[0].failThreshold + 1), float64(errDefinitions[1].failThreshold - 1)},
-			score:     2,
-		},
-		{
-			errCounts: []float64{float64(errDefinitions[0].recoverThreshold - 1), float64(errDefinitions[1].recoverThreshold - 1)},
+			errCounts: []float64{float64(errDefinitions[0].recoverThreshold*100 - 1), 100, float64(errDefinitions[1].recoverThreshold*100 - 1), 100},
 			score:     0,
 		},
 		{
-			errCounts: []float64{float64(errDefinitions[0].recoverThreshold + 1), float64(errDefinitions[1].recoverThreshold - 1)},
+			errCounts: []float64{float64(errDefinitions[0].recoverThreshold*100 + 1), 100, float64(errDefinitions[1].recoverThreshold*100 - 1), 100},
 			score:     1,
 		},
 		{
-			errCounts: []float64{float64(errDefinitions[0].failThreshold + 1), float64(errDefinitions[1].recoverThreshold - 1)},
+			errCounts: []float64{float64(errDefinitions[0].failThreshold*100 + 1), 100, float64(errDefinitions[1].recoverThreshold*100 - 1), 100},
 			score:     2,
 		},
 		{
-			errCounts: []float64{float64(errDefinitions[0].recoverThreshold + 1), float64(errDefinitions[1].recoverThreshold + 1)},
+			errCounts: []float64{float64(errDefinitions[0].recoverThreshold*100 + 1), 100, float64(errDefinitions[1].recoverThreshold*100 + 1), 100},
 			score:     1,
 		},
 	}
 
 	backends := make([]scoredBackend, 0, len(tests))
-	values1 := make([]*model.Sample, 0, len(tests))
-	values2 := make([]*model.Sample, 0, len(tests))
+	values := make([][]*model.Sample, 4)
 	for i, test := range tests {
 		backends = append(backends, createBackend(i, 0, 0))
-		values1 = append(values1, createSample(float64(test.errCounts[0]), i))
-		values2 = append(values2, createSample(float64(test.errCounts[1]), i))
+		for indIdx, fv := range test.errCounts {
+			values[indIdx] = append(values[indIdx], createSample(fv, i))
+		}
 	}
 	mmr := &mockMetricsReader{
-		qrs: map[string]metricsreader.QueryResult{
-			"health_pd": {
-				UpdateTime: time.Now(),
-				Value:      model.Vector(values1),
-			},
-			"health_tikv": {
-				UpdateTime: time.Now(),
-				Value:      model.Vector(values2),
-			},
-		},
+		qrs: make(map[string]metricsreader.QueryResult),
+	}
+	for i, indicator := range []string{"failure_pd", "total_pd", "failure_tikv", "total_tikv"} {
+		mmr.qrs[indicator] = metricsreader.QueryResult{
+			UpdateTime: time.Now(),
+			Value:      model.Vector(values[i]),
+		}
 	}
 	fh := NewFactorHealth(mmr, zap.NewNop())
 	fh.UpdateScore(backends)
@@ -94,78 +88,95 @@ func TestHealthBalance(t *testing.T) {
 			scores:    []uint64{0, 0},
 			balanced:  true,
 		},
+		{
+			errCounts: [][]float64{{math.NaN(), 0}, {0, math.NaN()}},
+			scores:    []uint64{0, 0},
+			balanced:  true,
+		},
+		{
+			errCounts: [][]float64{{0, 0}, {0, 0}},
+			scores:    []uint64{0, 0},
+			balanced:  true,
+		},
+		{
+			errCounts: [][]float64{{10, 0}, {0, 0}},
+			scores:    []uint64{2, 0},
+			balanced:  false,
+		},
 		{
 			errCounts: [][]float64{
-				{float64(errDefinitions[0].recoverThreshold + 1), float64(errDefinitions[1].recoverThreshold - 1)},
+				{float64(errDefinitions[0].recoverThreshold*100 + 1), 100},
 				{math.NaN(), math.NaN()}},
 			scores:   []uint64{1, 0},
 			balanced: true,
 		},
 		{
 			errCounts: [][]float64{
-				{float64(errDefinitions[0].recoverThreshold + 1), float64(errDefinitions[1].recoverThreshold - 1)},
-				{float64(errDefinitions[0].recoverThreshold + 1), float64(errDefinitions[1].recoverThreshold - 1)}},
+				{float64(errDefinitions[0].recoverThreshold*100 + 1), 100},
+				{float64(errDefinitions[0].recoverThreshold*100 + 1), 100}},
 			scores:   []uint64{1, 1},
 			balanced: true,
 		},
 		{
 			errCounts: [][]float64{
-				{float64(errDefinitions[0].recoverThreshold - 1), float64(errDefinitions[1].recoverThreshold - 1)},
-				{float64(errDefinitions[0].recoverThreshold + 1), float64(errDefinitions[1].recoverThreshold - 1)}},
+				{float64(errDefinitions[0].recoverThreshold*100 - 1), 100},
+				{float64(errDefinitions[0].recoverThreshold*100 + 1), 100}},
 			scores:   []uint64{0, 1},
 			balanced: true,
 		},
 		{
 			errCounts: [][]float64{
-				{float64(errDefinitions[0].failThreshold + 1), float64(errDefinitions[1].recoverThreshold + 1)},
-				{float64(errDefinitions[0].recoverThreshold + 1), float64(errDefinitions[1].recoverThreshold - 1)}},
+				{float64(errDefinitions[0].failThreshold*100 + 1), 100},
+				{float64(errDefinitions[0].recoverThreshold*100 + 1), 100}},
 			scores:   []uint64{2, 1},
 			balanced: true,
 		},
 		{
 			errCounts: [][]float64{
-				{float64(errDefinitions[0].failThreshold + 1), float64(errDefinitions[1].recoverThreshold + 1)},
-				{float64(errDefinitions[0].recoverThreshold - 1), float64(errDefinitions[1].recoverThreshold - 1)}},
+				{float64(errDefinitions[0].failThreshold*100 + 1), 100},
+				{float64(errDefinitions[0].recoverThreshold*100 - 1), 100}},
 			scores:   []uint64{2, 0},
 			balanced: false,
 		},
 	}
 
 	for i, test := range tests {
-		backends := make([]scoredBackend, 0, len(test.errCounts))
-		values1 := make([]*model.Sample, 0, len(test.errCounts))
-		values2 := make([]*model.Sample, 0, len(test.errCounts))
-		for j := 0; j < len(test.errCounts); j++ {
-			backends = append(backends, createBackend(j, 100, 100))
-			values1 = append(values1, createSample(test.errCounts[j][0], j))
-			values2 = append(values2, createSample(test.errCounts[j][1], j))
-		}
-		mmr := &mockMetricsReader{
-			qrs: map[string]metricsreader.QueryResult{
-				"health_pd": {
-					UpdateTime: time.Now(),
-					Value:      model.Vector(values1),
-				},
-				"health_tikv": {
-					UpdateTime: time.Now(),
-					Value:      model.Vector(values2),
+		for _, indicator := range []string{"pd", "tikv"} {
+			backends := make([]scoredBackend, 0, len(test.errCounts))
+			values1 := make([]*model.Sample, 0, len(test.errCounts))
+			values2 := make([]*model.Sample, 0, len(test.errCounts))
+			for beIndx := 0; beIndx < len(test.errCounts); beIndx++ {
+				backends = append(backends, createBackend(beIndx, 100, 100))
+				values1 = append(values1, createSample(test.errCounts[beIndx][0], beIndx))
+				values2 = append(values2, createSample(test.errCounts[beIndx][1], beIndx))
+			}
+			mmr := &mockMetricsReader{
+				qrs: map[string]metricsreader.QueryResult{
+					fmt.Sprintf("failure_%s", indicator): {
+						UpdateTime: time.Now(),
+						Value:      model.Vector(values1),
+					},
+					fmt.Sprintf("total_%s", indicator): {
+						UpdateTime: time.Now(),
+						Value:      model.Vector(values2),
+					},
 				},
-			},
-		}
-		fh := NewFactorHealth(mmr, zap.NewNop())
-		fh.UpdateScore(backends)
-		scores := make([]uint64, 0, len(backends))
-		for _, backend := range backends {
-			scores = append(scores, backend.score())
+			}
+			fh := NewFactorHealth(mmr, zap.NewNop())
+			fh.UpdateScore(backends)
+			scores := make([]uint64, 0, len(backends))
+			for _, backend := range backends {
+				scores = append(scores, backend.score())
+			}
+			require.Equal(t, test.scores, scores, "test index %d", i)
+			sort.Slice(backends, func(i, j int) bool {
+				return backends[i].score() < backends[j].score()
+			})
+			from, to := backends[len(backends)-1], backends[0]
+			advice, balanceCount, _ := fh.BalanceCount(from, to)
+			require.Equal(t, test.balanced, advice == AdviceNeutral, "test index %d, indicator %s", i, indicator)
+			require.Equal(t, test.balanced, balanceCount < 0.0001, "test index %d, indicator %s", i, indicator)
 		}
-		require.Equal(t, test.scores, scores, "test index %d", i)
-		sort.Slice(backends, func(i, j int) bool {
-			return backends[i].score() < backends[j].score()
-		})
-		from, to := backends[len(backends)-1], backends[0]
-		advice, balanceCount, _ := fh.BalanceCount(from, to)
-		require.Equal(t, test.balanced, advice == AdviceNeutral, "test index %d", i)
-		require.Equal(t, test.balanced, balanceCount < 0.0001, "test index %d", i)
 	}
 }
 
@@ -179,15 +190,15 @@ func TestNoHealthMetrics(t *testing.T) {
 			updateTime: time.Now(),
 		},
 		{
-			errCounts:  [][]float64{nil, {1, 1}},
+			errCounts:  [][]float64{nil, {1, 10000, 0, 10000}},
 			updateTime: time.Now(),
 		},
 		{
-			errCounts:  [][]float64{{1, 1}, {0, 0}},
+			errCounts:  [][]float64{{1, 1, 1, 1}, {0, 0, 0, 0}},
 			updateTime: time.Now().Add(-errMetricExpDuration * 2),
 		},
 		{
-			errCounts:  [][]float64{{math.NaN(), math.NaN()}, {math.NaN(), math.NaN()}},
+			errCounts:  [][]float64{{math.NaN(), math.NaN(), math.NaN(), math.NaN()}, {math.NaN(), math.NaN(), math.NaN(), math.NaN()}},
 			updateTime: time.Now(),
 		},
 	}
@@ -199,30 +210,26 @@ func TestNoHealthMetrics(t *testing.T) {
 		backends = append(backends, createBackend(i, i*100, i*100))
 	}
 	for i, test := range tests {
-		values1 := make([]*model.Sample, 0, len(test.errCounts))
-		values2 := make([]*model.Sample, 0, len(test.errCounts))
-		for j := 0; j < len(test.errCounts); j++ {
-			if test.errCounts[j] != nil {
-				values1 = append(values1, createSample(test.errCounts[j][0], j))
-				values2 = append(values2, createSample(test.errCounts[j][1], j))
+		values := make([][]*model.Sample, 4)
+		for beIdx := range test.errCounts {
+			if test.errCounts[beIdx] != nil {
+				for indIdx := 0; indIdx < len(test.errCounts[beIdx]); indIdx++ {
+					values[indIdx] = append(values[indIdx], createSample(test.errCounts[beIdx][indIdx], beIdx))
+				}
 			}
 		}
-		var v1, v2 model.Vector
-		if len(values1) > 0 {
-			v1 = model.Vector(values1)
-		}
-		if len(values2) > 0 {
-			v2 = model.Vector(values2)
+		vecs := make([]model.Vector, 4)
+		for indIdx := range vecs {
+			if len(values[indIdx]) > 0 {
+				vecs[indIdx] = model.Vector(values[indIdx])
+			}
 		}
-		mmr.qrs = map[string]metricsreader.QueryResult{
-			"health_pd": {
+		mmr.qrs = make(map[string]metricsreader.QueryResult)
+		for j, ind := range []string{"failure_pd", "total_pd", "failure_tikv", "total_tikv"} {
+			mmr.qrs[ind] = metricsreader.QueryResult{
 				UpdateTime: test.updateTime,
-				Value:      v1,
-			},
-			"health_tikv": {
-				UpdateTime: test.updateTime,
-				Value:      v2,
-			},
+				Value:      vecs[j],
+			}
 		}
 		updateScore(fh, backends)
 		require.Equal(t, backends[0].score(), backends[1].score(), "test index %d", i)
@@ -268,17 +275,26 @@ func TestHealthBalanceCount(t *testing.T) {
 	updateMmr := func(healthy bool) {
 		number := 0.0
 		if !healthy {
-			number = 99999999.0
+			number = 10000.0
 		}
-		values := []*model.Sample{
+		failureValues := []*model.Sample{
 			createSample(number, 0),
 			createSample(0, 1),
 		}
-		mmr.qrs["health_pd"] = metricsreader.QueryResult{
-			UpdateTime: time.Now(),
-			Value:      model.Vector(values),
+		totalValues := []*model.Sample{
+			createSample(10000, 0),
+			createSample(10000, 1),
+		}
+		mmr.qrs = map[string]metricsreader.QueryResult{
+			"failure_pd": {
+				UpdateTime: time.Now(),
+				Value:      model.Vector(failureValues),
+			},
+			"total_pd": {
+				UpdateTime: time.Now(),
+				Value:      model.Vector(totalValues),
+			},
 		}
-		mmr.qrs["health_tikv"] = mmr.qrs["health_pd"]
 	}
 	for i, test := range tests {
 		updateMmr(test.healthy)
@@ -301,56 +317,82 @@ func TestHealthQueryRule(t *testing.T) {
 	}{
 		{
 			text: `tidb_tikvclient_backoff_seconds_count{type=""} 0
-tidb_tikvclient_backoff_seconds_count{type="dataNotReady"} 0
 tidb_tikvclient_backoff_seconds_count{type="pdRPC"} 10
-tidb_tikvclient_backoff_seconds_count{type="regionMiss"} 10
-tidb_tikvclient_backoff_seconds_count{type="tikvRPC"} 0
+pd_client_cmd_handle_failed_cmds_duration_seconds_count{type="tso"} 10
+pd_client_cmd_handle_cmds_duration_seconds_count{type="tso"} 100
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="1",type="Get"} 20
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="2",type="Get"} 30
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="1",type="BatchGet"} 20
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="2",type="BatchGet"} 30
 `,
-			curValue:   []model.SampleValue{10, 10},
-			finalValue: []model.SampleValue{model.SampleValue(math.NaN()), model.SampleValue(math.NaN())},
+			curValue:   []model.SampleValue{10, 100, 0, 100},
+			finalValue: []model.SampleValue{model.SampleValue(math.NaN()), model.SampleValue(math.NaN()), model.SampleValue(math.NaN()), model.SampleValue(math.NaN())},
 		},
 		{
 			text: `tidb_tikvclient_backoff_seconds_count{type=""} 10
-tidb_tikvclient_backoff_seconds_count{type="dataNotReady"} 10
 tidb_tikvclient_backoff_seconds_count{type="pdRPC"} 20
-tidb_tikvclient_backoff_seconds_count{type="regionMiss"} 110
 tidb_tikvclient_backoff_seconds_count{type="tikvRPC"} 100
+pd_client_cmd_handle_failed_cmds_duration_seconds_count{type="tso"} 20
+pd_client_cmd_handle_cmds_duration_seconds_count{type="tso"} 200
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="1",type="Get"} 50
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="2",type="Get"} 50
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="1",type="BatchGet"} 50
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="2",type="BatchGet"} 50
 `,
-			curValue:   []model.SampleValue{20, 210},
-			finalValue: []model.SampleValue{10, 200},
+			curValue:   []model.SampleValue{20, 200, 100, 200},
+			finalValue: []model.SampleValue{10, 100, 100, 100},
 		},
 		{
 			text: `tidb_tikvclient_backoff_seconds_count{type=""} 10
-tidb_tikvclient_backoff_seconds_count{type="dataNotReady"} 10
 tidb_tikvclient_backoff_seconds_count{type="pdRPC"} 20
-tidb_tikvclient_backoff_seconds_count{type="regionMiss"} 110
-tidb_tikvclient_backoff_seconds_count{type="tikvRPC"} 100
+tidb_tikvclient_backoff_seconds_count{type="tikvRPC"} 150
+pd_client_cmd_handle_failed_cmds_duration_seconds_count{type="tso"} 20
+pd_client_cmd_handle_cmds_duration_seconds_count{type="tso"} 300
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="1",type="Get"} 100
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="2",type="Get"} 100
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="1",type="BatchGet"} 100
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="2",type="BatchGet"} 100
 `,
-			curValue:   []model.SampleValue{20, 210},
-			finalValue: []model.SampleValue{10, 200},
+			curValue:   []model.SampleValue{20, 300, 150, 400},
+			finalValue: []model.SampleValue{10, 200, 150, 300},
 		},
 		{
 			text: `tidb_tikvclient_backoff_seconds_count{type=""} 0
-tidb_tikvclient_backoff_seconds_count{type="dataNotReady"} 0
-tidb_tikvclient_backoff_seconds_count{type="pdRPC"} 0
-tidb_tikvclient_backoff_seconds_count{type="regionMiss"} 5
-tidb_tikvclient_backoff_seconds_count{type="tikvRPC"} 0
+tidb_tikvclient_backoff_seconds_count{type="tikvRPC"} 10
+pd_client_cmd_handle_failed_cmds_duration_seconds_count{type="tso"} 5
+pd_client_cmd_handle_cmds_duration_seconds_count{type="tso"} 50
+tidb_tikvclient_request_counter{scope="false",stale_read="false",store="1",type="Get"} 50
 `,
-			curValue:   []model.SampleValue{0, 5},
-			finalValue: []model.SampleValue{model.SampleValue(math.NaN()), model.SampleValue(math.NaN())},
+			curValue:   []model.SampleValue{5, 50, 10, 50},
+			finalValue: []model.SampleValue{model.SampleValue(math.NaN()), model.SampleValue(math.NaN()), 10, model.SampleValue(math.NaN())},
+		},
+		{
+			text:       ``,
+			curValue:   []model.SampleValue{model.SampleValue(math.NaN()), model.SampleValue(math.NaN()), model.SampleValue(math.NaN()), model.SampleValue(math.NaN())},
+			finalValue: []model.SampleValue{model.SampleValue(math.NaN()), model.SampleValue(math.NaN()), model.SampleValue(math.NaN()), model.SampleValue(math.NaN())},
 		},
 	}
 
-	historyPair := make([][]model.SamplePair, len(errDefinitions))
+	rules := []metricsreader.QueryRule{
+		errDefinitions[0].queryFailureRule,
+		errDefinitions[0].queryTotalRule,
+		errDefinitions[1].queryFailureRule,
+		errDefinitions[1].queryTotalRule,
+	}
+	historyPair := make([][]model.SamplePair, len(rules))
 	for i, test := range tests {
 		var parser expfmt.TextParser
 		mfs, err := parser.TextToMetricFamilies(strings.NewReader(test.text))
 		require.NoError(t, err, "case %d", i)
-		for j, ed := range errDefinitions {
-			value := ed.queryRule.Metric2Value(mfs)
-			require.Equal(t, test.curValue[j], value, "case %d %d", i, j)
+		for j, rule := range rules {
+			value := rule.Metric2Value(mfs)
+			if math.IsNaN(float64(test.curValue[j])) {
+				require.True(t, math.IsNaN(float64(value)), "case %d %d", i, j)
+			} else {
+				require.Equal(t, test.curValue[j], value, "case %d %d", i, j)
+			}
 			historyPair[j] = append(historyPair[j], model.SamplePair{Value: value})
-			value = ed.queryRule.Range2Value(historyPair[j])
+			value = rule.Range2Value(historyPair[j])
 			if math.IsNaN(float64(test.finalValue[j])) {
 				require.True(t, math.IsNaN(float64(value)), "case %d %d", i, j)
 			} else {
@@ -365,14 +407,19 @@ func TestMissBackendInHealth(t *testing.T) {
 	fh := NewFactorHealth(mmr, zap.NewNop())
 	backends := make([]scoredBackend, 0, 2)
 	errors := []float64{10000, 0}
-	var values []*model.Sample
+	var failureValues, totalValues []*model.Sample
 	for i := 0; i < 2; i++ {
 		backends = append(backends, createBackend(i, 100, 100))
-		values = append(values, createSample(errors[i], i))
+		failureValues = append(failureValues, createSample(errors[i], i))
+		totalValues = append(totalValues, createSample(10000, i))
+	}
+	mmr.qrs["failure_pd"] = metricsreader.QueryResult{
+		UpdateTime: time.Now(),
+		Value:      model.Vector(failureValues),
 	}
-	mmr.qrs["health_pd"] = metricsreader.QueryResult{
+	mmr.qrs["total_pd"] = metricsreader.QueryResult{
 		UpdateTime: time.Now(),
-		Value:      model.Vector(values),
+		Value:      model.Vector(totalValues),
 	}
 	fh.UpdateScore(backends)
 	advice, count, _ := fh.BalanceCount(backends[0], backends[1])