balance, infosync: support reading metrics from both prometheus and backend (#606)

Author: djshow832

lib/config/health.go

@@ -3,6 +3,64 @@
 
 package config
 
+import "time"
+
 type HealthInfo struct {
 	ConfigChecksum uint32 `json:"config_checksum"`
 }
+
+const (
+	healthCheckInterval      = 3 * time.Second
+	healthCheckMaxRetries    = 3
+	healthCheckRetryInterval = 1 * time.Second
+	healthCheckTimeout       = 2 * time.Second
+	readMetricsInterval      = 5 * time.Second
+	readMetricsTimeout       = 3 * time.Second
+)
+
+// HealthCheck contains some configurations for health check.
+// Some general configurations of them may be exposed to users in the future.
+// We can use shorter durations to speed up unit tests.
+type HealthCheck struct {
+	Enable          bool          `yaml:"enable" json:"enable" toml:"enable"`
+	Interval        time.Duration `yaml:"interval" json:"interval" toml:"interval"`
+	MaxRetries      int           `yaml:"max-retries" json:"max-retries" toml:"max-retries"`
+	RetryInterval   time.Duration `yaml:"retry-interval" json:"retry-interval" toml:"retry-interval"`
+	DialTimeout     time.Duration `yaml:"dial-timeout" json:"dial-timeout" toml:"dial-timeout"`
+	MetricsInterval time.Duration `yaml:"metrics-interval" json:"metrics-interval" toml:"metrics-interval"`
+	MetricsTimeout  time.Duration `yaml:"metrics-timeout" json:"metrics-timeout" toml:"metrics-timeout"`
+}
+
+// NewDefaultHealthCheckConfig creates a default HealthCheck.
+func NewDefaultHealthCheckConfig() *HealthCheck {
+	return &HealthCheck{
+		Enable:          true,
+		Interval:        healthCheckInterval,
+		MaxRetries:      healthCheckMaxRetries,
+		RetryInterval:   healthCheckRetryInterval,
+		DialTimeout:     healthCheckTimeout,
+		MetricsInterval: readMetricsInterval,
+		MetricsTimeout:  readMetricsTimeout,
+	}
+}
+
+func (hc *HealthCheck) Check() {
+	if hc.Interval == 0 {
+		hc.Interval = healthCheckInterval
+	}
+	if hc.MaxRetries == 0 {
+		hc.MaxRetries = healthCheckMaxRetries
+	}
+	if hc.RetryInterval == 0 {
+		hc.RetryInterval = healthCheckRetryInterval
+	}
+	if hc.DialTimeout == 0 {
+		hc.DialTimeout = healthCheckTimeout
+	}
+	if hc.MetricsInterval == 0 {
+		hc.MetricsInterval = readMetricsInterval
+	}
+	if hc.MetricsTimeout == 0 {
+		hc.MetricsTimeout = readMetricsTimeout
+	}
+}

lib/config/namespace.go

@@ -5,7 +5,6 @@ package config
 
 import (
 	"bytes"
-	"time"
 
 	"github.com/BurntSushi/toml"
 )
@@ -24,63 +23,6 @@ type FrontendNamespace struct {
 type BackendNamespace struct {
 	Instances []string  `yaml:"instances" json:"instances" toml:"instances"`
 	Security  TLSConfig `yaml:"security" json:"security" toml:"security"`
-	//HealthCheck  HealthCheck `yaml:"health-check" json:"health-check" toml:"health-check"`
-}
-
-const (
-	healthCheckInterval      = 3 * time.Second
-	healthCheckMaxRetries    = 3
-	healthCheckRetryInterval = 1 * time.Second
-	healthCheckTimeout       = 2 * time.Second
-	readMetricsInterval      = 5 * time.Second
-	readMetricsTimeout       = 3 * time.Second
-)
-
-// HealthCheck contains some configurations for health check.
-// Some general configurations of them may be exposed to users in the future.
-// We can use shorter durations to speed up unit tests.
-type HealthCheck struct {
-	Enable          bool          `yaml:"enable" json:"enable" toml:"enable"`
-	Interval        time.Duration `yaml:"interval" json:"interval" toml:"interval"`
-	MaxRetries      int           `yaml:"max-retries" json:"max-retries" toml:"max-retries"`
-	RetryInterval   time.Duration `yaml:"retry-interval" json:"retry-interval" toml:"retry-interval"`
-	DialTimeout     time.Duration `yaml:"dial-timeout" json:"dial-timeout" toml:"dial-timeout"`
-	MetricsInterval time.Duration `yaml:"metrics-interval" json:"metrics-interval" toml:"metrics-interval"`
-	MetricsTimeout  time.Duration `yaml:"metrics-timeout" json:"metrics-timeout" toml:"metrics-timeout"`
-}
-
-// NewDefaultHealthCheckConfig creates a default HealthCheck.
-func NewDefaultHealthCheckConfig() *HealthCheck {
-	return &HealthCheck{
-		Enable:          true,
-		Interval:        healthCheckInterval,
-		MaxRetries:      healthCheckMaxRetries,
-		RetryInterval:   healthCheckRetryInterval,
-		DialTimeout:     healthCheckTimeout,
-		MetricsInterval: readMetricsInterval,
-		MetricsTimeout:  readMetricsTimeout,
-	}
-}
-
-func (hc *HealthCheck) Check() {
-	if hc.Interval == 0 {
-		hc.Interval = healthCheckInterval
-	}
-	if hc.MaxRetries == 0 {
-		hc.MaxRetries = healthCheckMaxRetries
-	}
-	if hc.RetryInterval == 0 {
-		hc.RetryInterval = healthCheckRetryInterval
-	}
-	if hc.DialTimeout == 0 {
-		hc.DialTimeout = healthCheckTimeout
-	}
-	if hc.MetricsInterval == 0 {
-		hc.MetricsInterval = readMetricsInterval
-	}
-	if hc.MetricsTimeout == 0 {
-		hc.MetricsTimeout = readMetricsTimeout
-	}
 }
 
 func NewNamespace(data []byte) (*Namespace, error) {

pkg/balance/factor/factor_cpu.go

@@ -56,6 +56,10 @@ var (
 			if timeDiff < 1e-4 {
 				return model.SampleValue(math.NaN())
 			}
+			// Maybe the backend just rebooted.
+			if pair1.Value > pair2.Value {
+				return model.SampleValue(math.NaN())
+			}
 			return (pair2.Value - pair1.Value) / model.SampleValue(timeDiff)
 		},
 		ResultType: model.ValMatrix,
@@ -80,17 +84,17 @@ type FactorCPU struct {
 	// The estimated average CPU usage used by one connection.
 	usagePerConn float64
 	mr           metricsreader.MetricsReader
-	queryID      uint64
 	bitNum       int
 }
 
 func NewFactorCPU(mr metricsreader.MetricsReader) *FactorCPU {
-	return &FactorCPU{
+	fc := &FactorCPU{
 		mr:       mr,
-		queryID:  mr.AddQueryExpr(cpuQueryExpr),
 		bitNum:   5,
 		snapshot: make(map[string]cpuBackendSnapshot),
 	}
+	mr.AddQueryExpr(fc.Name(), cpuQueryExpr, cpuQueryRule)
+	return fc
 }
 
 func (fc *FactorCPU) Name() string {
@@ -101,8 +105,8 @@ func (fc *FactorCPU) UpdateScore(backends []scoredBackend) {
 	if len(backends) <= 1 {
 		return
 	}
-	qr := fc.mr.GetQueryResult(fc.queryID)
-	if qr.Err != nil || qr.Empty() {
+	qr := fc.mr.GetQueryResult(fc.Name())
+	if qr.Empty() {
 		return
 	}
 
@@ -250,5 +254,5 @@ func (fc *FactorCPU) SetConfig(cfg *config.Config) {
 }
 
 func (fc *FactorCPU) Close() {
-	fc.mr.RemoveQueryExpr(fc.queryID)
+	fc.mr.RemoveQueryExpr(fc.Name())
 }

pkg/balance/factor/factor_cpu_test.go

@@ -83,8 +83,8 @@ func TestCPUBalanceOnce(t *testing.T) {
 			values = append(values, createSampleStream(test.cpus[j], j, model.Now()))
 		}
 		mmr := &mockMetricsReader{
-			qrs: map[uint64]metricsreader.QueryResult{
-				1: {
+			qrs: map[string]metricsreader.QueryResult{
+				"cpu": {
 					UpdateTime: monotime.Now(),
 					Value:      model.Matrix(values),
 				},
@@ -197,7 +197,7 @@ func TestCPUBalanceContinuously(t *testing.T) {
 			values = append(values, createSampleStream(test.cpus[j], j, curTime))
 		}
 		curTime = curTime.Add(time.Millisecond)
-		mmr.qrs[1] = metricsreader.QueryResult{
+		mmr.qrs["cpu"] = metricsreader.QueryResult{
 			UpdateTime: monotime.Now(),
 			Value:      model.Matrix(values),
 		}
@@ -251,7 +251,7 @@ func TestNoCPUMetric(t *testing.T) {
 			ss := createSampleStream(test.cpus[j], j, model.Time(test.updateTime/monotime.Time(time.Millisecond)))
 			values = append(values, ss)
 		}
-		mmr.qrs[1] = metricsreader.QueryResult{
+		mmr.qrs["cpu"] = metricsreader.QueryResult{
 			UpdateTime: test.updateTime,
 			Value:      model.Matrix(values),
 		}
@@ -290,7 +290,7 @@ func TestCPUResultNotUpdated(t *testing.T) {
 	for i, test := range tests {
 		array := []float64{test.cpu}
 		values := []*model.SampleStream{createSampleStream(array, 0, test.updateTime), createSampleStream(array, 1, test.updateTime)}
-		mmr.qrs[1] = metricsreader.QueryResult{
+		mmr.qrs["cpu"] = metricsreader.QueryResult{
 			UpdateTime: monotime.Now(),
 			Value:      model.Matrix(values),
 		}
@@ -330,6 +330,14 @@ tidb_server_maxprocs 2
 			curValue:   6,
 			finalValue: 1,
 		},
+		{
+			text: `process_cpu_seconds_total 2
+tidb_server_maxprocs 2
+`,
+			timestamp:  model.Time(3000),
+			curValue:   1,
+			finalValue: model.SampleValue(math.NaN()),
+		},
 	}
 
 	historyPair := make([]model.SamplePair, 0)

pkg/balance/factor/factor_health.go

@@ -34,6 +34,7 @@ const (
 type errDefinition struct {
 	queryRule        metricsreader.QueryRule
 	promQL           string
+	key              string
 	failThreshold    int
 	recoverThreshold int
 }
@@ -65,6 +66,7 @@ var (
 			promQL:           `sum(increase(tidb_tikvclient_backoff_seconds_count{type="pdRPC"}[2m])) by (instance)`,
 			failThreshold:    50,
 			recoverThreshold: 10,
+			key:              "health_pd",
 			queryRule: metricsreader.QueryRule{
 				Names:     []string{"tidb_tikvclient_backoff_seconds_count"},
 				Retention: 2 * time.Minute,
@@ -87,7 +89,12 @@ var (
 					if len(pairs) < 2 {
 						return model.SampleValue(math.NaN())
 					}
-					return pairs[len(pairs)-1].Value - pairs[0].Value
+					diff := pairs[len(pairs)-1].Value - pairs[0].Value
+					// Maybe the backend just rebooted.
+					if diff < 0 {
+						return model.SampleValue(math.NaN())
+					}
+					return diff
 				},
 				ResultType: model.ValVector,
 			},
@@ -99,6 +106,7 @@ var (
 			promQL:           `sum(increase(tidb_tikvclient_backoff_seconds_count{type=~"regionMiss|tikvRPC"}[2m])) by (instance)`,
 			failThreshold:    1000,
 			recoverThreshold: 100,
+			key:              "health_tikv",
 			queryRule: metricsreader.QueryRule{
 				Names:     []string{"tidb_tikvclient_backoff_seconds_count"},
 				Retention: 2 * time.Minute,
@@ -121,7 +129,12 @@ var (
 					if len(pairs) < 2 {
 						return model.SampleValue(math.NaN())
 					}
-					return pairs[len(pairs)-1].Value - pairs[0].Value
+					diff := pairs[len(pairs)-1].Value - pairs[0].Value
+					// Maybe the backend just rebooted.
+					if diff < 0 {
+						return model.SampleValue(math.NaN())
+					}
+					return diff
 				},
 				ResultType: model.ValVector,
 			},
@@ -141,8 +154,9 @@ type healthBackendSnapshot struct {
 
 type errIndicator struct {
 	queryExpr        metricsreader.QueryExpr
+	queryRule        metricsreader.QueryRule
 	queryResult      metricsreader.QueryResult
-	queryID          uint64
+	key              string
 	failThreshold    int
 	recoverThreshold int
 }
@@ -170,10 +184,12 @@ func initErrIndicator(mr metricsreader.MetricsReader) []errIndicator {
 			queryExpr: metricsreader.QueryExpr{
 				PromQL: def.promQL,
 			},
+			queryRule:        def.queryRule,
+			key:              def.key,
 			failThreshold:    def.failThreshold,
 			recoverThreshold: def.recoverThreshold,
 		}
-		indicator.queryID = mr.AddQueryExpr(indicator.queryExpr)
+		mr.AddQueryExpr(indicator.key, indicator.queryExpr, indicator.queryRule)
 		indicators = append(indicators, indicator)
 	}
 	return indicators
@@ -189,8 +205,8 @@ func (fh *FactorHealth) UpdateScore(backends []scoredBackend) {
 	}
 	needUpdateSnapshot, latestTime := false, monotime.Time(0)
 	for i := 0; i < len(fh.indicators); i++ {
-		qr := fh.mr.GetQueryResult(fh.indicators[i].queryID)
-		if qr.Err != nil || qr.Empty() {
+		qr := fh.mr.GetQueryResult(fh.indicators[i].key)
+		if qr.Empty() {
 			continue
 		}
 		if fh.indicators[i].queryResult.UpdateTime != qr.UpdateTime {
@@ -313,6 +329,6 @@ func (fh *FactorHealth) SetConfig(cfg *config.Config) {
 
 func (fh *FactorHealth) Close() {
 	for _, indicator := range fh.indicators {
-		fh.mr.RemoveQueryExpr(indicator.queryID)
+		fh.mr.RemoveQueryExpr(indicator.key)
 	}
 }