factor, router: routing fails when there's only one backend with a different label (#820) (#821)

Signed-off-by: ti-chi-bot <[email protected]>
Co-authored-by: djshow832 <[email protected]>

Author: ti-chi-bot

pkg/balance/factor/factor_balance.go

@@ -177,25 +177,29 @@ func (fbb *FactorBasedBalance) updateScore(backends []policy.BackendCtx) []score
 
 // BackendToRoute returns one backend to route a new connection to.
 func (fbb *FactorBasedBalance) BackendToRoute(backends []policy.BackendCtx) policy.BackendCtx {
+	fields := []zap.Field{zap.Int("backend_num", len(backends))}
+	defer func() {
+		fbb.lg.Debug("route", fields...)
+	}()
+
 	if len(backends) == 0 {
 		return nil
 	}
 
 	fbb.Lock()
 	defer fbb.Unlock()
 	scoredBackends := fbb.updateScore(backends)
+	for _, backend := range scoredBackends {
+		fields = append(fields, zap.String(backend.Addr(), strconv.FormatUint(backend.scoreBits, 16)))
+	}
 	if !fbb.canBeRouted(scoredBackends[0].scoreBits) {
 		return nil
 	}
 	if len(backends) == 1 {
+		fields = append(fields, zap.String("target", backends[0].Addr()))
 		return backends[0]
 	}
 
-	var fields []zap.Field
-	for _, backend := range scoredBackends {
-		fields = append(fields, zap.String(backend.Addr(), strconv.FormatUint(backend.scoreBits, 16)))
-	}
-
 	// Evict the backends that are so busy that it should migrate connections to another, and then randomly choose one.
 	// Always choosing the idlest one works badly for short connections because even a little jitter may cause all the connections
 	// in the next second route to the same backend.
@@ -234,7 +238,6 @@ func (fbb *FactorBasedBalance) BackendToRoute(backends []policy.BackendCtx) poli
 		idx = idxes[int(time.Now().UnixMicro()%int64(len(idxes)))]
 	}
 	fields = append(fields, zap.String("target", scoredBackends[idx].Addr()), zap.Ints("rand", idxes))
-	fbb.lg.Debug("route", fields...)
 	return scoredBackends[idx].BackendCtx
 }
 

pkg/balance/factor/factor_label.go

@@ -35,7 +35,8 @@ func (fl *FactorLabel) Name() string {
 }
 
 func (fl *FactorLabel) UpdateScore(backends []scoredBackend) {
-	if len(fl.labelName) == 0 || len(fl.selfLabelVal) == 0 || len(backends) <= 1 {
+	// The score will be used for CanBeRouted so don't skip updating even when only one backend.
+	if len(fl.labelName) == 0 || len(fl.selfLabelVal) == 0 {
 		return
 	}
 	for i := 0; i < len(backends); i++ {
@@ -53,7 +54,12 @@ func (fl *FactorLabel) ScoreBitNum() int {
 }
 
 func (fl *FactorLabel) BalanceCount(from, to scoredBackend) (float64, []zap.Field) {
-	return balanceCount4Label, nil
+	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
 }
 
 func (fl *FactorLabel) SetConfig(cfg *config.Config) {

pkg/balance/factor/factor_label_test.go

@@ -53,14 +53,10 @@ func TestFactorLabelOneBackend(t *testing.T) {
 		backendCtx := &mockBackend{
 			BackendInfo: observer.BackendInfo{Labels: backendLabels},
 		}
-		// Create 2 backends so that UpdateScore won't skip calculating scores.
 		backends := []scoredBackend{
 			{
 				BackendCtx: backendCtx,
 			},
-			{
-				BackendCtx: backendCtx,
-			},
 		}
 		selfLabels := make(map[string]string)
 		if test.labelName != "" && test.selfLabelVal != "" {
@@ -75,6 +71,7 @@ func TestFactorLabelOneBackend(t *testing.T) {
 		factor.UpdateScore(backends)
 		for _, backend := range backends {
 			require.Equal(t, test.expectedScore, backend.score(), "test idx: %d", i)
+			require.Equal(t, test.expectedScore == 0, factor.CanBeRouted(backend.score()), "test idx: %d", i)
 		}
 	}
 }
@@ -133,5 +130,6 @@ func TestFactorLabelMultiBackends(t *testing.T) {
 	factor.UpdateScore(backends)
 	for i, test := range tests {
 		require.Equal(t, test.expectedScore, backends[i].score(), "test idx: %d", i)
+		require.Equal(t, test.expectedScore == 0, factor.CanBeRouted(backends[i].score()), "test idx: %d", i)
 	}
 }

pkg/balance/observer/backend_health.go

@@ -5,6 +5,8 @@ package observer
 
 import (
 	"fmt"
+	"maps"
+	"strings"
 
 	"github.com/pingcap/tiproxy/lib/config"
 )
@@ -37,19 +39,30 @@ func (bh *BackendHealth) setLocal(cfg *config.Config) {
 	bh.Local = false
 }
 
-func (bh *BackendHealth) Equals(health BackendHealth) bool {
-	return bh.Healthy == health.Healthy && bh.ServerVersion == health.ServerVersion && bh.Local == health.Local
+func (bh *BackendHealth) Equals(other BackendHealth) bool {
+	return bh.BackendInfo.Equals(other.BackendInfo) &&
+		bh.Healthy == other.Healthy &&
+		bh.ServerVersion == other.ServerVersion
 }
 
 func (bh *BackendHealth) String() string {
-	str := "down"
+	var sb strings.Builder
 	if bh.Healthy {
-		str = "healthy"
+		_, _ = sb.WriteString("healthy")
+	} else {
+		_, _ = sb.WriteString("down")
 	}
 	if bh.PingErr != nil {
-		str += fmt.Sprintf(", err: %s", bh.PingErr.Error())
+		_, _ = sb.WriteString(fmt.Sprintf(", err: %s", bh.PingErr.Error()))
 	}
-	return str
+	if len(bh.ServerVersion) > 0 {
+		_, _ = sb.WriteString(", version: ")
+		_, _ = sb.WriteString(bh.ServerVersion)
+	}
+	if bh.Labels != nil {
+		_, _ = sb.WriteString(fmt.Sprintf(", labels: %v", bh.Labels))
+	}
+	return sb.String()
 }
 
 // BackendInfo stores the status info of each backend.
@@ -59,6 +72,12 @@ type BackendInfo struct {
 	StatusPort uint
 }
 
+func (bi BackendInfo) Equals(other BackendInfo) bool {
+	return bi.IP == other.IP &&
+		bi.StatusPort == other.StatusPort &&
+		maps.Equal(bi.Labels, other.Labels)
+}
+
 // HealthResult contains the health check results and is used to notify the routers.
 // It's read-only for subscribers.
 type HealthResult struct {

pkg/balance/observer/backend_health_test.go

@@ -0,0 +1,109 @@
+// Copyright 2025 PingCAP, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package observer
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBackendHealthToString(t *testing.T) {
+	tests := []BackendHealth{
+		{},
+		{
+			BackendInfo: BackendInfo{
+				IP:         "127.0.0.1",
+				StatusPort: 1,
+				Labels:     map[string]string{"k1": "v1", "k2": "v2"},
+			},
+			Healthy:       true,
+			PingErr:       errors.New("mock error"),
+			ServerVersion: "v1.0.0",
+			Local:         true,
+		},
+	}
+	// Just test no error happens
+	for _, test := range tests {
+		_ = test.String()
+	}
+}
+
+func TestBackendHealthEquals(t *testing.T) {
+	tests := []struct {
+		a, b  BackendHealth
+		equal bool
+	}{
+		{
+			a:     BackendHealth{},
+			b:     BackendHealth{},
+			equal: true,
+		},
+		{
+			a: BackendHealth{
+				BackendInfo: BackendInfo{
+					IP:         "127.0.0.1",
+					StatusPort: 1,
+					Labels:     map[string]string{"k1": "v1", "k2": "v2"},
+				},
+			},
+			b: BackendHealth{
+				BackendInfo: BackendInfo{
+					IP:         "127.0.0.1",
+					StatusPort: 1,
+				},
+			},
+			equal: false,
+		},
+		{
+			a: BackendHealth{
+				BackendInfo: BackendInfo{
+					IP:         "127.0.0.1",
+					StatusPort: 1,
+					Labels:     map[string]string{"k1": "v1", "k2": "v2"},
+				},
+			},
+			b: BackendHealth{
+				BackendInfo: BackendInfo{
+					IP:         "127.0.0.1",
+					StatusPort: 1,
+					Labels:     map[string]string{"k1": "v1", "k2": "v2"},
+				},
+			},
+			equal: true,
+		},
+		{
+			a: BackendHealth{
+				BackendInfo: BackendInfo{
+					IP:         "127.0.0.1",
+					StatusPort: 1,
+					Labels:     map[string]string{"k1": "v1", "k2": "v2"},
+				},
+				Healthy:       true,
+				PingErr:       errors.New("mock error"),
+				ServerVersion: "v1.0.0",
+				Local:         true,
+			},
+			b:     BackendHealth{},
+			equal: false,
+		},
+		{
+			a: BackendHealth{
+				ServerVersion: "v1.0.0",
+			},
+			b: BackendHealth{
+				ServerVersion: "v1.1.0",
+			},
+			equal: false,
+		},
+	}
+	// Just test no error happens
+	for i, test := range tests {
+		require.True(t, test.a.Equals(test.a), "test %d", i)
+		require.True(t, test.b.Equals(test.b), "test %d", i)
+		require.Equal(t, test.equal, test.a.Equals(test.b), "test %d", i)
+		require.Equal(t, test.equal, test.b.Equals(test.a), "test %d", i)
+	}
+}

pkg/balance/observer/backend_observer.go

@@ -5,6 +5,7 @@ package observer
 
 import (
 	"context"
+	"maps"
 	"sync"
 	"time"
 
@@ -193,27 +194,22 @@ func (bo *DefaultBackendObserver) updateHealthResult(result HealthResult) {
 			continue
 		}
 		if oldHealth, ok := bo.curBackends[addr]; !ok || !oldHealth.Healthy {
-			prev := "none"
-			if oldHealth != nil {
-				prev = oldHealth.String()
-			}
-			bo.logger.Info("update backend", zap.String("backend_addr", addr),
-				zap.String("prev", prev), zap.String("cur", newHealth.String()))
+			bo.logger.Info("update backend", zap.String("addr", addr), zap.Stringer("prev", oldHealth), zap.Stringer("cur", newHealth),
+				zap.Int("total", len(bo.curBackends)))
 			updateBackendStatusMetrics(addr, true)
 			delete(bo.downBackends, addr)
+		} else if ok && !maps.Equal(oldHealth.Labels, newHealth.Labels) {
+			bo.logger.Info("update backend labels", zap.String("addr", addr),
+				zap.Any("prev", oldHealth.Labels), zap.Any("cur", newHealth.Labels))
 		}
 	}
 	for addr, oldHealth := range bo.curBackends {
 		if !oldHealth.Healthy {
 			continue
 		}
 		if newHealth, ok := result.backends[addr]; !ok || !newHealth.Healthy {
-			cur := "not in list"
-			if newHealth != nil {
-				cur = newHealth.String()
-			}
-			bo.logger.Info("update backend", zap.String("backend_addr", addr),
-				zap.String("prev", oldHealth.String()), zap.String("cur", cur))
+			bo.logger.Info("update backend", zap.String("addr", addr), zap.Stringer("prev", oldHealth), zap.Stringer("cur", newHealth),
+				zap.Int("total", len(bo.curBackends)))
 			updateBackendStatusMetrics(addr, false)
 			bo.downBackends[addr] = time.Now()
 		}

pkg/balance/router/router.go

@@ -110,6 +110,13 @@ func (b *backendWrapper) setHealth(health observer.BackendHealth) {
 	b.mu.Unlock()
 }
 
+func (b *backendWrapper) getHealth() observer.BackendHealth {
+	b.mu.RLock()
+	health := b.mu.BackendHealth
+	b.mu.RUnlock()
+	return health
+}
+
 func (b *backendWrapper) ConnScore() int {
 	return b.connScore
 }

pkg/balance/router/router_score.go

@@ -282,6 +282,8 @@ func (router *ScoreBasedRouter) updateBackendHealth(healthResults observer.Healt
 	// If some backends are removed from the list, add them to `backends`.
 	for addr, backend := range router.backends {
 		if _, ok := backends[addr]; !ok {
+			health := backend.getHealth()
+			router.logger.Debug("backend is removed from the list, add it back to router", zap.String("addr", addr), zap.Stringer("health", &health))
 			backends[addr] = &observer.BackendHealth{
 				BackendInfo: backend.GetBackendInfo(),
 				Healthy:     false,
@@ -293,16 +295,20 @@ func (router *ScoreBasedRouter) updateBackendHealth(healthResults observer.Healt
 	for addr, health := range backends {
 		backend, ok := router.backends[addr]
 		if !ok && health.Healthy {
+			router.logger.Debug("add new backend to router", zap.String("addr", addr), zap.Stringer("health", health))
 			router.backends[addr] = newBackendWrapper(addr, *health)
 			serverVersion = health.ServerVersion
 		} else if ok {
-			if !backend.Equals(*health) {
-				backend.setHealth(*health)
-				router.removeBackendIfEmpty(backend)
-				if health.Healthy {
-					serverVersion = health.ServerVersion
-				}
+			if !health.Equals(backend.getHealth()) {
+				router.logger.Debug("update backend in router", zap.String("addr", addr), zap.Stringer("health", health))
+			}
+			backend.setHealth(*health)
+			router.removeBackendIfEmpty(backend)
+			if health.Healthy {
+				serverVersion = health.ServerVersion
 			}
+		} else {
+			router.logger.Debug("unhealthy backend is not in router", zap.String("addr", addr), zap.Stringer("health", health))
 		}
 	}
 	if len(serverVersion) > 0 {