Reduce useless migrations based on connection count in short connection workload (#802) (#803)

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

Author: ti-chi-bot

pkg/balance/factor/factor_conn.go

@@ -12,13 +12,18 @@ const (
 	// connBalancedRatio is the threshold of ratio of the most connection count and least count.
 	// If the ratio exceeds the threshold, we migrate connections.
 	connBalancedRatio = 1.2
-	// balanceCount4Conn indicates how many connections to balance per second.
-	// Migrate slowly because we don't know the CPU usage here and we may migrate too many connections.
-	balanceCount4Conn = 1
+	// Narrow it to 20% in 120s, but the speed is slower and slower because the difference is getting smaller.
+	// The original difference is 30%: after 120s, the difference is 23.5%.
+	// The original difference is 100%: after 120s, the difference is 43.9%.
+	balanceSeconds4Conn = 120
 )
 
 var _ Factor = (*FactorConnCount)(nil)
 
+// FactorConnCount balances by connection count and must be used in following cases:
+// - The policy is set to `connection`
+// - The backend CPU usages are unavailable
+// - The workload just starts and the backend CPU usages are low
 type FactorConnCount struct {
 	bitNum int
 }
@@ -50,10 +55,15 @@ func (fcc *FactorConnCount) ScoreBitNum() int {
 }
 
 func (fcc *FactorConnCount) BalanceCount(from, to scoredBackend) (float64, []zap.Field) {
-	if float64(from.ConnScore()) > float64(to.ConnScore()+1)*connBalancedRatio {
-		return balanceCount4Conn, nil
+	if float64(from.ConnScore()) <= float64(to.ConnScore()+1)*connBalancedRatio {
+		return 0, nil
 	}
-	return 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
 }
 
 func (fcc *FactorConnCount) SetConfig(cfg *config.Config) {

pkg/balance/factor/factor_conn_test.go

@@ -38,3 +38,59 @@ func TestFactorConnCount(t *testing.T) {
 		require.Equal(t, test.expectedScore, backends[i].score(), "test idx: %d", i)
 	}
 }
+
+func TestFactorConnSpeed(t *testing.T) {
+	tests := []struct {
+		score1      int
+		score2      int
+		targetRange [2]int
+	}{
+		{
+			score1:      120,
+			score2:      100,
+			targetRange: [2]int{100, 100},
+		},
+		{
+			score1:      150,
+			score2:      100,
+			targetRange: [2]int{110, 114},
+		},
+		{
+			score1:      10000,
+			score2:      0,
+			targetRange: [2]int{3500, 4550},
+		},
+	}
+
+	factor := NewFactorConnCount()
+	backend1 := newMockBackend(true, 0)
+	backend2 := newMockBackend(true, 0)
+	scoredBackend1 := newScoredBackend(backend1, zap.NewNop())
+	scoredBackend2 := newScoredBackend(backend2, zap.NewNop())
+	for i, test := range tests {
+		backend1.connScore = test.score1
+		backend2.connScore = test.score2
+		lastRedirectTime := 0
+		// Simulate rebalance for 5 minutes.
+		for j := 0; j < 30000; j++ {
+			balanceCount, _ := factor.BalanceCount(scoredBackend1, scoredBackend2)
+			if balanceCount < 0.0001 {
+				break
+			}
+			migrationInterval := 100 / balanceCount
+			count := 0
+			if migrationInterval < 2 {
+				count = int(1 / migrationInterval)
+			} else if float64(j-lastRedirectTime) >= migrationInterval {
+				count = 1
+			}
+			if count > 0 {
+				lastRedirectTime = j
+				backend1.connScore -= count
+				backend2.connScore += count
+			}
+		}
+		require.GreaterOrEqual(t, backend2.connScore, test.targetRange[0], "case id: %d", i)
+		require.LessOrEqual(t, backend2.connScore, test.targetRange[1], "case id: %d", i)
+	}
+}