balance: fix connection is repeatedly migrated by CPU and location (#823) (#824)

Co-authored-by: djshow832 <[email protected]>

Author: ti-chi-bot

pkg/balance/factor/factor.go

@@ -8,6 +8,17 @@ import (
 	"go.uber.org/zap"
 )
 
+type BalanceAdvice int
+
+const (
+	// AdviceNeutral indicates skipping this factor and continue to the next factor.
+	AdviceNeutral BalanceAdvice = iota
+	// AdviceNegtive indicates don't balance these 2 backends, even for the rest factors.
+	AdviceNegtive
+	// AdvicePositive indicates balancing these 2 backends now.
+	AdvicePositive
+)
+
 type Factor interface {
 	// Name returns the name of the factor.
 	Name() string
@@ -17,7 +28,7 @@ type Factor interface {
 	ScoreBitNum() int
 	// BalanceCount returns the count of connections to balance per second.
 	// 0 indicates the factor is already balanced.
-	BalanceCount(from, to scoredBackend) (float64, []zap.Field)
+	BalanceCount(from, to scoredBackend) (BalanceAdvice, float64, []zap.Field)
 	SetConfig(cfg *config.Config)
 	// CanBeRouted returns whether a connection can be routed or migrated to the backend with the score.
 	CanBeRouted(score uint64) bool

pkg/balance/factor/factor_balance.go

@@ -213,7 +213,9 @@ func (fbb *FactorBasedBalance) BackendToRoute(backends []policy.BackendCtx) poli
 			score2 := scoredBackends[0].scoreBits << (maxBitNum - leftBitNum) >> (maxBitNum - bitNum)
 			if score1 > score2 {
 				var balanceFields []zap.Field
-				if balanceCount, balanceFields = factor.BalanceCount(scoredBackends[i], scoredBackends[0]); balanceCount > 0.0001 {
+				var advice BalanceAdvice
+				advice, balanceCount, balanceFields = factor.BalanceCount(scoredBackends[i], scoredBackends[0])
+				if advice == AdvicePositive && balanceCount > 0.0001 {
 					// This backend is too busy. If it's routed, migration may happen.
 					fields = append(fields, zap.String(scoredBackends[i].Addr(), factor.Name()))
 					fields = append(fields, balanceFields...)
@@ -274,20 +276,28 @@ func (fbb *FactorBasedBalance) BackendsToBalance(backends []policy.BackendCtx) (
 			score1 := scoredBackends[i].scoreBits << (maxBitNum - leftBitNum) >> (maxBitNum - bitNum)
 			score2 := scoredBackends[0].scoreBits << (maxBitNum - leftBitNum) >> (maxBitNum - bitNum)
 			if score1 > score2 {
-				// The previous factors are ordered, so this factor won't violate them.
-				// E.g. the factor scores of 2 backends are [1, 1], [0, 0]
-				// Balancing the second factor won't make the first factor unbalanced.
+				// 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 fields []zap.Field
-				if balanceCount, fields = factor.BalanceCount(scoredBackends[i], scoredBackends[0]); balanceCount > 0.0001 {
-					from, to = scoredBackends[i].BackendCtx, scoredBackends[0].BackendCtx
-					reason = factor.Name()
-					logFields = append(fields, zap.String("factor", reason),
-						zap.String("from_total_score", strconv.FormatUint(scoredBackends[i].scoreBits, 16)),
-						zap.String("to_total_score", strconv.FormatUint(scoredBackends[0].scoreBits, 16)),
-						zap.Uint64("from_factor_score", score1),
-						zap.Uint64("to_factor_score", score2),
-						zap.Float64("balance_count", balanceCount))
-					return
+				var advice BalanceAdvice
+				advice, balanceCount, 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,
+					// don't migrate from B to A even if A is preferred in location.
+					break
+				} else if advice == AdvicePositive {
+					if balanceCount > 0.0001 {
+						from, to = scoredBackends[i].BackendCtx, scoredBackends[0].BackendCtx
+						reason = factor.Name()
+						logFields = append(fields, zap.String("factor", reason),
+							zap.String("from_total_score", strconv.FormatUint(scoredBackends[i].scoreBits, 16)),
+							zap.String("to_total_score", strconv.FormatUint(scoredBackends[0].scoreBits, 16)),
+							zap.Uint64("from_factor_score", score1),
+							zap.Uint64("to_factor_score", score2),
+							zap.Float64("balance_count", balanceCount))
+						return
+					}
 				}
 			} else if score1 < score2 {
 				// Stop it once a factor is in the opposite order, otherwise a subsequent factor may violate this one.

pkg/balance/factor/factor_balance_test.go

@@ -475,6 +475,51 @@ func TestCanBeRouted(t *testing.T) {
 	}
 }
 
+func TestCanBalance(t *testing.T) {
+	fm := NewFactorBasedBalance(zap.NewNop(), newMockMetricsReader())
+	factor1 := &mockFactor{bitNum: 2, balanceCount: 1, advice: AdviceNegtive}
+	factor2 := &mockFactor{bitNum: 2, balanceCount: 1}
+	fm.factors = []Factor{factor1, factor2}
+	require.NoError(t, fm.updateBitNum())
+
+	tests := []struct {
+		scores1  []int
+		scores2  []int
+		balanced bool
+	}{
+		{
+			scores1:  []int{1, 0},
+			scores2:  []int{0, 0},
+			balanced: false,
+		},
+		{
+			scores1:  []int{1, 1},
+			scores2:  []int{1, 0},
+			balanced: false,
+		},
+		{
+			scores1:  []int{1, 1},
+			scores2:  []int{1, 1},
+			balanced: false,
+		},
+	}
+	for tIdx, test := range tests {
+		factor1.updateScore = func(backends []scoredBackend) {
+			for i := 0; i < len(backends); i++ {
+				backends[i].addScore(test.scores1[i], factor1.bitNum)
+			}
+		}
+		factor2.updateScore = func(backends []scoredBackend) {
+			for i := 0; i < len(backends); i++ {
+				backends[i].addScore(test.scores2[i], factor2.bitNum)
+			}
+		}
+		backends := createBackends(len(test.scores1))
+		_, _, count, _, _ := fm.BackendsToBalance(backends)
+		require.Equal(t, test.balanced, count > 0, "test index %d", tIdx)
+	}
+}
+
 func TestSetFactorConcurrently(t *testing.T) {
 	fbb := NewFactorBasedBalance(zap.NewNop(), newMockMetricsReader())
 	var wg waitgroup.WaitGroup

pkg/balance/factor/factor_conn.go

@@ -54,16 +54,16 @@ func (fcc *FactorConnCount) ScoreBitNum() int {
 	return fcc.bitNum
 }
 
-func (fcc *FactorConnCount) BalanceCount(from, to scoredBackend) (float64, []zap.Field) {
+func (fcc *FactorConnCount) BalanceCount(from, to scoredBackend) (BalanceAdvice, float64, []zap.Field) {
 	if float64(from.ConnScore()) <= float64(to.ConnScore()+1)*connBalancedRatio {
-		return 0, nil
+		return AdviceNeutral, 0, nil
 	}
 	targetTo := float64(from.ConnScore()+to.ConnScore()+1) / (1 + connBalancedRatio)
 	count := (targetTo - float64(to.ConnScore()+1)) / balanceSeconds4Conn
 	if count < 0 {
 		count = 0
 	}
-	return count, nil
+	return AdvicePositive, count, nil
 }
 
 func (fcc *FactorConnCount) SetConfig(cfg *config.Config) {

pkg/balance/factor/factor_conn_test.go

@@ -73,8 +73,8 @@ func TestFactorConnSpeed(t *testing.T) {
 		lastRedirectTime := 0
 		// Simulate rebalance for 5 minutes.
 		for j := 0; j < 30000; j++ {
-			balanceCount, _ := factor.BalanceCount(scoredBackend1, scoredBackend2)
-			if balanceCount < 0.0001 {
+			advice, balanceCount, _ := factor.BalanceCount(scoredBackend1, scoredBackend2)
+			if advice != AdvicePositive || balanceCount < 0.0001 {
 				break
 			}
 			migrationInterval := 100 / balanceCount

pkg/balance/factor/factor_cpu.go

@@ -21,8 +21,9 @@ const (
 	cpuMetricExpDuration = 2 * time.Minute
 	cpuScoreStep         = 5
 	// 0.001 represents for 0.1%
-	minCpuPerConn    = 0.001
-	cpuBalancedRatio = 1.2
+	minCpuPerConn      = 0.001
+	cpuBalancedRatio   = 1.2
+	cpuUnbalancedRatio = 1.1
 	// If the CPU difference of 2 backends is 30% and we're narrowing it to 20% in 30 seconds,
 	// then in each round, we migrate ((30% - 20%) / 2) / usagePerConn / 30 = 1 / usagePerConn / 600 connections.
 	balanceRatio4Cpu = 600
@@ -246,7 +247,7 @@ func (fc *FactorCPU) ScoreBitNum() int {
 	return fc.bitNum
 }
 
-func (fc *FactorCPU) BalanceCount(from, to scoredBackend) (float64, []zap.Field) {
+func (fc *FactorCPU) BalanceCount(from, to scoredBackend) (BalanceAdvice, float64, []zap.Field) {
 	fromAvgUsage, fromLatestUsage := fc.getUsage(from)
 	toAvgUsage, toLatestUsage := fc.getUsage(to)
 	fields := []zap.Field{
@@ -260,13 +261,18 @@ func (fc *FactorCPU) BalanceCount(from, to scoredBackend) (float64, []zap.Field)
 		zap.Int("to_conn", to.ConnScore()),
 		zap.Float64("usage_per_conn", fc.usagePerConn),
 	}
+	// Reject migration if it will make the target backend even much busier than the source.
+	if (1.3-(toAvgUsage+fc.usagePerConn))*cpuUnbalancedRatio < 1.3-(fromAvgUsage-fc.usagePerConn) ||
+		(1.3-(toLatestUsage+fc.usagePerConn))*cpuUnbalancedRatio < 1.3-(fromLatestUsage-fc.usagePerConn) {
+		return AdviceNegtive, 0, fields
+	}
 	// The higher the CPU usage, the more sensitive the load balance should be.
 	// E.g. 10% vs 25% don't need rebalance, but 80% vs 95% need rebalance.
 	// Use the average usage to avoid thrash when CPU jitters too much and use the latest usage to avoid migrate too many connections.
 	if 1.3-toAvgUsage < (1.3-fromAvgUsage)*cpuBalancedRatio || 1.3-toLatestUsage < (1.3-fromLatestUsage)*cpuBalancedRatio {
-		return 0, nil
+		return AdviceNeutral, 0, nil
 	}
-	return 1 / fc.usagePerConn / balanceRatio4Cpu, fields
+	return AdvicePositive, 1 / fc.usagePerConn / balanceRatio4Cpu, fields
 }
 
 func (fc *FactorCPU) SetConfig(cfg *config.Config) {

pkg/balance/factor/factor_cpu_test.go

@@ -79,7 +79,7 @@ func TestCPUBalanceOnce(t *testing.T) {
 		backends := make([]scoredBackend, 0, len(test.cpus))
 		values := make([]*model.SampleStream, 0, len(test.cpus))
 		for j := 0; j < len(test.cpus); j++ {
-			backends = append(backends, createBackend(j, 0, 0))
+			backends = append(backends, createBackend(j, 100, 100))
 			values = append(values, createSampleStream(test.cpus[j], j, model.Now()))
 		}
 		mmr := &mockMetricsReader{
@@ -100,7 +100,8 @@ func TestCPUBalanceOnce(t *testing.T) {
 		}
 		require.Equal(t, test.scoreOrder, sortedIdx, "test index %d", i)
 		from, to := backends[len(backends)-1], backends[0]
-		balanceCount, _ := fc.BalanceCount(from, to)
+		advice, balanceCount, _ := fc.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)
 	}
 }
@@ -207,8 +208,8 @@ func TestCPUBalanceContinuously(t *testing.T) {
 				t.Fatal("balance doesn't stop")
 			}
 			updateScore(fc, backends)
-			balanceCount, _ := fc.BalanceCount(backends[len(backends)-1], backends[0])
-			if balanceCount == 0 {
+			advice, balanceCount, _ := fc.BalanceCount(backends[len(backends)-1], backends[0])
+			if advice != AdvicePositive || balanceCount == 0 {
 				break
 			}
 			count := int(balanceCount + 0.9999)
@@ -414,3 +415,53 @@ func TestCPUScore(t *testing.T) {
 		require.Equal(t, test.scores, scores, "test index %d", i)
 	}
 }
+
+func TestCPURejectBalance(t *testing.T) {
+	tests := []struct {
+		cpus   [][]float64
+		conns  []int
+		advice BalanceAdvice
+	}{
+		{
+			cpus:   [][]float64{{0.7}, {0.7}},
+			conns:  []int{10, 10},
+			advice: AdviceNegtive,
+		},
+		{
+			cpus:   [][]float64{{0.75}, {0.7}},
+			conns:  []int{10, 10},
+			advice: AdviceNegtive,
+		},
+		{
+			cpus:   [][]float64{{0.78}, {0.7}},
+			conns:  []int{20, 20},
+			advice: AdviceNeutral,
+		},
+		{
+			cpus:   [][]float64{{0.8}, {0.6}},
+			conns:  []int{10, 10},
+			advice: AdvicePositive,
+		},
+	}
+
+	for i, test := range tests {
+		backends := make([]scoredBackend, 0, len(test.cpus))
+		values := make([]*model.SampleStream, 0, len(test.cpus))
+		for j := 0; j < len(test.cpus); j++ {
+			backends = append(backends, createBackend(j, test.conns[j], test.conns[j]))
+			values = append(values, createSampleStream(test.cpus[j], j, model.Now()))
+		}
+		mmr := &mockMetricsReader{
+			qrs: map[string]metricsreader.QueryResult{
+				"cpu": {
+					UpdateTime: time.Now(),
+					Value:      model.Matrix(values),
+				},
+			},
+		}
+		fc := NewFactorCPU(mmr, zap.NewNop())
+		updateScore(fc, backends)
+		advice, _, _ := fc.BalanceCount(backends[1], backends[0])
+		require.Equal(t, test.advice, advice, "test index %d", i)
+	}
+}

pkg/balance/factor/factor_health.go

@@ -349,18 +349,18 @@ func (fh *FactorHealth) ScoreBitNum() int {
 	return fh.bitNum
 }
 
-func (fh *FactorHealth) BalanceCount(from, to scoredBackend) (float64, []zap.Field) {
+func (fh *FactorHealth) BalanceCount(from, to scoredBackend) (BalanceAdvice, float64, []zap.Field) {
 	// Only migrate connections when one is valueRangeNormal and the other is valueRangeAbnormal.
 	fromScore := fh.caclErrScore(from.Addr())
 	toScore := fh.caclErrScore(to.Addr())
 	if fromScore-toScore <= 1 {
-		return 0, nil
+		return AdviceNeutral, 0, nil
 	}
 	snapshot := fh.snapshot[from.Addr()]
 	fields := []zap.Field{
 		zap.String("indicator", snapshot.indicator),
 		zap.Int("value", snapshot.value)}
-	return snapshot.balanceCount, fields
+	return AdvicePositive, snapshot.balanceCount, fields
 }
 
 func (fh *FactorHealth) SetConfig(cfg *config.Config) {

pkg/balance/factor/factor_health_test.go

@@ -163,7 +163,8 @@ func TestHealthBalance(t *testing.T) {
 			return backends[i].score() < backends[j].score()
 		})
 		from, to := backends[len(backends)-1], backends[0]
-		balanceCount, _ := fh.BalanceCount(from, to)
+		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)
 	}
 }
@@ -286,7 +287,8 @@ func TestHealthBalanceCount(t *testing.T) {
 		if test.count == 0 {
 			continue
 		}
-		count, _ := fh.BalanceCount(backends[0], backends[1])
+		advice, count, _ := fh.BalanceCount(backends[0], backends[1])
+		require.Equal(t, AdvicePositive, advice)
 		require.Equal(t, test.count, count, "test idx: %d", i)
 	}
 }
@@ -373,14 +375,16 @@ func TestMissBackendInHealth(t *testing.T) {
 		Value:      model.Vector(values),
 	}
 	fh.UpdateScore(backends)
-	count, _ := fh.BalanceCount(backends[0], backends[1])
+	advice, count, _ := fh.BalanceCount(backends[0], backends[1])
+	require.Equal(t, AdvicePositive, advice)
 	require.Equal(t, 100/balanceSeconds4Health, count)
 
 	// Miss the first backend but the snapshot should be preserved.
 	fh.UpdateScore(backends[1:])
 	unhealthyBackend := backends[0].BackendCtx.(*mockBackend)
 	unhealthyBackend.connScore = 50
 	fh.UpdateScore(backends)
-	count, _ = fh.BalanceCount(backends[0], backends[1])
+	advice, count, _ = fh.BalanceCount(backends[0], backends[1])
+	require.Equal(t, AdvicePositive, advice)
 	require.Equal(t, 100/balanceSeconds4Health, count)
 }

pkg/balance/factor/factor_label.go

@@ -53,13 +53,13 @@ func (fl *FactorLabel) ScoreBitNum() int {
 	return fl.bitNum
 }
 
-func (fl *FactorLabel) BalanceCount(from, to scoredBackend) (float64, []zap.Field) {
+func (fl *FactorLabel) BalanceCount(from, to scoredBackend) (BalanceAdvice, float64, []zap.Field) {
 	fields := []zap.Field{
 		zap.String("label_key", fl.labelName),
 		zap.Any("from_labels", from.GetBackendInfo().Labels),
 		zap.String("self_label_value", fl.selfLabelVal),
 	}
-	return balanceCount4Label, fields
+	return AdvicePositive, balanceCount4Label, fields
 }
 
 func (fl *FactorLabel) SetConfig(cfg *config.Config) {