factor: update migration speed for factors (#573)

Author: djshow832

pkg/balance/factor/factor.go

@@ -14,7 +14,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) int
+	BalanceCount(from, to scoredBackend) float64
 	SetConfig(cfg *config.Config)
 	Close()
 }

pkg/balance/factor/factor_balance.go

@@ -176,7 +176,7 @@ func (fbb *FactorBasedBalance) BackendToRoute(backends []policy.BackendCtx) poli
 // BackendsToBalance returns the busiest/unhealthy backend and the idlest backend.
 // balanceCount: the count of connections to migrate in this round. 0 indicates no need to balance.
 // reason: the debug information to be logged.
-func (fbb *FactorBasedBalance) BackendsToBalance(backends []policy.BackendCtx) (from, to policy.BackendCtx, balanceCount int, reason string, logFields []zap.Field) {
+func (fbb *FactorBasedBalance) BackendsToBalance(backends []policy.BackendCtx) (from, to policy.BackendCtx, balanceCount float64, reason string, logFields []zap.Field) {
 	if len(backends) <= 1 {
 		return
 	}
@@ -217,7 +217,7 @@ func (fbb *FactorBasedBalance) BackendsToBalance(backends []policy.BackendCtx) (
 			// backend2 factor scores: 0, 0
 			// Balancing the second factor won't make the first factor unbalanced.
 			balanceCount = factor.BalanceCount(*busiestBackend, *idlestBackend)
-			if balanceCount > 0 {
+			if balanceCount > 0.0001 {
 				break
 			}
 		} else if score1 < score2 {

pkg/balance/factor/factor_balance_test.go

@@ -137,7 +137,7 @@ func TestBalanceWithOneFactor(t *testing.T) {
 		}
 		backends := createBackends(len(test.scores))
 		from, to, count, reason, _ := fm.BackendsToBalance(backends)
-		require.Equal(t, test.count, count, "test index %d", tIdx)
+		require.EqualValues(t, test.count, count, "test index %d", tIdx)
 		if test.count > 0 {
 			require.Equal(t, backends[test.fromIdx], from, "test index %d", tIdx)
 			require.Equal(t, backends[test.toIdx], to, "test index %d", tIdx)
@@ -217,7 +217,7 @@ func TestBalanceWith2Factors(t *testing.T) {
 		}
 		backends := createBackends(len(test.scores1))
 		from, to, count, _, _ := fm.BackendsToBalance(backends)
-		require.Equal(t, test.count, count, "test index %d", tIdx)
+		require.EqualValues(t, test.count, count, "test index %d", tIdx)
 		if test.count > 0 {
 			require.Equal(t, backends[test.fromIdx], from, "test index %d", tIdx)
 			require.Equal(t, backends[test.toIdx], to, "test index %d", tIdx)
@@ -258,7 +258,7 @@ func TestBalanceWith3Factors(t *testing.T) {
 	for tIdx, test := range tests {
 		for factorIdx, factor := range factors {
 			func(factorIdx int, factor *mockFactor) {
-				factor.balanceCount = test.balanceCounts[factorIdx]
+				factor.balanceCount = float64(test.balanceCounts[factorIdx])
 				factor.updateScore = func(backends []scoredBackend) {
 					for i := 0; i < len(backends); i++ {
 						backends[i].addScore(test.scores[i][factorIdx], factor.bitNum)
@@ -268,7 +268,7 @@ func TestBalanceWith3Factors(t *testing.T) {
 		}
 		backends := createBackends(len(test.scores))
 		from, to, count, _, _ := fm.BackendsToBalance(backends)
-		require.Equal(t, test.count, count, "test index %d", tIdx)
+		require.EqualValues(t, test.count, count, "test index %d", tIdx)
 		if test.count > 0 {
 			require.Equal(t, backends[test.fromIdx], from, "test index %d", tIdx)
 			require.Equal(t, backends[test.toIdx], to, "test index %d", tIdx)

pkg/balance/factor/factor_conn.go

@@ -40,7 +40,7 @@ func (fcc *FactorConnCount) ScoreBitNum() int {
 	return fcc.bitNum
 }
 
-func (fcc *FactorConnCount) BalanceCount(from, to scoredBackend) int {
+func (fcc *FactorConnCount) BalanceCount(from, to scoredBackend) float64 {
 	if float64(from.ConnScore()) > float64(to.ConnScore()+1)*connBalancedRatio {
 		return balanceCount4Conn
 	}

pkg/balance/factor/factor_cpu.go

@@ -208,17 +208,14 @@ func (fc *FactorCPU) ScoreBitNum() int {
 	return fc.bitNum
 }
 
-func (fc *FactorCPU) BalanceCount(from, to scoredBackend) int {
+func (fc *FactorCPU) BalanceCount(from, to scoredBackend) float64 {
 	fromAvgUsage, fromLatestUsage := fc.getUsage(from)
 	toAvgUsage, toLatestUsage := fc.getUsage(to)
 	// 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 {
-		if balanceCount := int(1 / fc.usagePerConn / balanceRatio4Cpu); balanceCount > 1 {
-			return balanceCount
-		}
-		return 1
+		return 1 / fc.usagePerConn / balanceRatio4Cpu
 	}
 	return 0
 }

pkg/balance/factor/factor_cpu_test.go

@@ -17,59 +17,59 @@ import (
 
 func TestCPUBalanceOnce(t *testing.T) {
 	tests := []struct {
-		cpus         [][]float64
-		scoreOrder   []int
-		balanceCount int
+		cpus       [][]float64
+		scoreOrder []int
+		balanced   bool
 	}{
 		{
-			cpus:         [][]float64{{0}, {0.2}},
-			scoreOrder:   []int{0, 1},
-			balanceCount: 0,
+			cpus:       [][]float64{{0}, {0.2}},
+			scoreOrder: []int{0, 1},
+			balanced:   true,
 		},
 		{
-			cpus:         [][]float64{{1, 0.5, 0.25}, {0, 0.25, 0.5}, {0, 0, 0}, {1, 1, 1}},
-			scoreOrder:   []int{2, 0, 1, 3},
-			balanceCount: 1,
+			cpus:       [][]float64{{1, 0.5, 0.25}, {0, 0.25, 0.5}, {0, 0, 0}, {1, 1, 1}},
+			scoreOrder: []int{2, 0, 1, 3},
+			balanced:   false,
 		},
 		{
-			cpus:         [][]float64{{0.25}, {}, {0.5}},
-			scoreOrder:   []int{0, 2, 1},
-			balanceCount: 1,
+			cpus:       [][]float64{{0.25}, {}, {0.5}},
+			scoreOrder: []int{0, 2, 1},
+			balanced:   false,
 		},
 		{
-			cpus:         [][]float64{{0.95, 0.92, 0.93, 0.94, 0.92, 0.94}, {0.81, 0.79, 0.82, 0.83, 0.76, 0.78}},
-			scoreOrder:   []int{1, 0},
-			balanceCount: 1,
+			cpus:       [][]float64{{0.95, 0.92, 0.93, 0.94, 0.92, 0.94}, {0.81, 0.79, 0.82, 0.83, 0.76, 0.78}},
+			scoreOrder: []int{1, 0},
+			balanced:   false,
 		},
 		{
-			cpus:         [][]float64{{0.35, 0.42, 0.37, 0.45, 0.42, 0.44}, {0.56, 0.62, 0.58, 0.57, 0.59, 0.63}},
-			scoreOrder:   []int{0, 1},
-			balanceCount: 1,
+			cpus:       [][]float64{{0.35, 0.42, 0.37, 0.45, 0.42, 0.44}, {0.56, 0.62, 0.58, 0.57, 0.59, 0.63}},
+			scoreOrder: []int{0, 1},
+			balanced:   false,
 		},
 		{
-			cpus:         [][]float64{{0, 0.1, 0, 0.1}, {0.15, 0.1, 0.15, 0.15}, {0.1, 0, 0.1, 0}},
-			scoreOrder:   []int{2, 0, 1},
-			balanceCount: 0,
+			cpus:       [][]float64{{0, 0.1, 0, 0.1}, {0.15, 0.1, 0.15, 0.15}, {0.1, 0, 0.1, 0}},
+			scoreOrder: []int{2, 0, 1},
+			balanced:   true,
 		},
 		{
-			cpus:         [][]float64{{0.5}, {}, {0.1, 0.3, 0.4}},
-			scoreOrder:   []int{2, 0, 1},
-			balanceCount: 1,
+			cpus:       [][]float64{{0.5}, {}, {0.1, 0.3, 0.4}},
+			scoreOrder: []int{2, 0, 1},
+			balanced:   false,
 		},
 		{
-			cpus:         [][]float64{{1.0}, {0.97}},
-			scoreOrder:   []int{1, 0},
-			balanceCount: 0,
+			cpus:       [][]float64{{1.0}, {0.97}},
+			scoreOrder: []int{1, 0},
+			balanced:   true,
 		},
 		{
-			cpus:         [][]float64{{0.8, 0.2, 0.8, 0.2}, {0.3, 0.5, 0.3, 0.5}},
-			scoreOrder:   []int{0, 1},
-			balanceCount: 0,
+			cpus:       [][]float64{{0.8, 0.2, 0.8, 0.2}, {0.3, 0.5, 0.3, 0.5}},
+			scoreOrder: []int{0, 1},
+			balanced:   true,
 		},
 		{
-			cpus:         [][]float64{{1.0}, {0.9}, {0.8}, {0.7}, {0.6}, {0.5}, {0.4}, {0.3}, {0.1}},
-			scoreOrder:   []int{8, 7, 6, 5, 4, 3, 2, 1, 0},
-			balanceCount: 1,
+			cpus:       [][]float64{{1.0}, {0.9}, {0.8}, {0.7}, {0.6}, {0.5}, {0.4}, {0.3}, {0.1}},
+			scoreOrder: []int{8, 7, 6, 5, 4, 3, 2, 1, 0},
+			balanced:   false,
 		},
 	}
 
@@ -99,7 +99,7 @@ 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)
-		require.Equal(t, test.balanceCount, balanceCount, "test index %d", i)
+		require.Equal(t, test.balanced, balanceCount < 0.0001, "test index %d", i)
 	}
 }
 
@@ -209,8 +209,9 @@ func TestCPUBalanceContinuously(t *testing.T) {
 			if balanceCount == 0 {
 				break
 			}
-			backends[len(backends)-1].BackendCtx.(*mockBackend).connScore -= balanceCount
-			backends[0].BackendCtx.(*mockBackend).connScore += balanceCount
+			count := int(balanceCount + 0.9999)
+			backends[len(backends)-1].BackendCtx.(*mockBackend).connScore -= count
+			backends[0].BackendCtx.(*mockBackend).connScore += count
 		}
 		connScores := make([]int, len(test.connScores))
 		for _, backend := range backends {

pkg/balance/factor/factor_health.go

@@ -16,7 +16,7 @@ import (
 const (
 	errMetricExpDuration = 1 * time.Minute
 	// balanceSeconds4Health indicates the time (in seconds) to migrate all the connections.
-	balanceSeconds4Health = 10
+	balanceSeconds4Health = 5.0
 )
 
 type valueRange int
@@ -79,6 +79,8 @@ var _ Factor = (*FactorHealth)(nil)
 type healthBackendSnapshot struct {
 	updatedTime monotime.Time
 	valueRange  valueRange
+	// 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 {
@@ -185,9 +187,20 @@ func (fh *FactorHealth) updateSnapshot(backends []scoredBackend) {
 			}
 			continue
 		}
+		// Set balance count if the backend is unhealthy, otherwise reset it to 0.
+		var balanceCount float64
+		if valueRange >= valueRangeAbnormal {
+			if existSnapshot && snapshot.balanceCount > 0.0001 {
+				balanceCount = snapshot.balanceCount
+			} else {
+				balanceCount = float64(backend.ConnScore()) / balanceSeconds4Health
+			}
+		}
+
 		snapshots[addr] = healthBackendSnapshot{
-			updatedTime: updatedTime,
-			valueRange:  valueRange,
+			updatedTime:  updatedTime,
+			valueRange:   valueRange,
+			balanceCount: balanceCount,
 		}
 	}
 	fh.snapshot = snapshots
@@ -221,39 +234,22 @@ func calcValueRange(sample *model.Sample, indicator errIndicator) valueRange {
 }
 
 func (fh *FactorHealth) caclErrScore(addr string) int {
-	snapshot, ok := fh.snapshot[addr]
-	if !ok {
-		return 2
-	}
-	switch snapshot.valueRange {
-	case valueRangeNormal:
-		return 0
-	case valueRangeMid:
-		return 1
-	default:
-		return 2
-	}
+	// If the backend has no metrics (not in snapshot), take it as healthy.
+	return int(fh.snapshot[addr].valueRange)
 }
 
 func (fh *FactorHealth) ScoreBitNum() int {
 	return fh.bitNum
 }
 
-func (fh *FactorHealth) BalanceCount(from, to scoredBackend) int {
+func (fh *FactorHealth) BalanceCount(from, to scoredBackend) float64 {
 	// 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 {
-		// Assuming that the source and target backends have similar connections at first.
-		// We wish the connections to be migrated in 10 seconds but only a few are migrated in each round.
-		// If we use from.ConnScore() / 10, the migration will be slower and slower.
-		conns := (from.ConnScore() + to.ConnScore()) / (balanceSeconds4Health * 2)
-		if conns > 0 {
-			return conns
-		}
-		return 1
+	if fromScore-toScore <= 1 {
+		return 0
 	}
-	return 0
+	return fh.snapshot[from.Addr()].balanceCount
 }
 
 func (fh *FactorHealth) SetConfig(cfg *config.Config) {