backendcluster: add cluster manager and cluster-scoped topology runtime

Introduce the backend cluster manager, cluster-scoped InfoSync runtime, topology aggregation, and single-cluster compatibility hooks.

Author: YangKeao

AGENTS.md

@@ -47,6 +47,7 @@ When adding or modifying features, prefer extending existing packages before cre
 - `pkg/manager/config/` - Auto-reloads configuration files and provides interfaces to query them.
 - `pkg/manager/elect/` - Manages TiProxy owner elections (for example, metrics reader and VIP modules need an owner).
 - `pkg/manager/id/` - Generates global IDs.
+- `pkg/manager/backendcluster/` - Manages cluster-scoped backend runtimes and shared resources such as PD or etcd clients.
 - `pkg/manager/infosync/` - Queries the topology of TiDB and Prometheus from PD and updates TiProxy information to PD.
 - `pkg/manager/logger/` - Manages the logger service.
 - `pkg/manager/memory/` - Records heap and goroutine profiles when memory usage is high.

pkg/balance/observer/backend_fetcher.go

@@ -46,11 +46,13 @@ func NewPDFetcher(tpFetcher TopologyFetcher, logger *zap.Logger, config *config.
 func (pf *PDFetcher) GetBackendList(ctx context.Context) (map[string]*BackendInfo, error) {
 	backends := pf.fetchBackendList(ctx)
 	infos := make(map[string]*BackendInfo, len(backends))
-	for addr, backend := range backends {
-		infos[addr] = &BackendInfo{
-			Labels:     backend.Labels,
-			IP:         backend.IP,
-			StatusPort: backend.StatusPort,
+	for key, backend := range backends {
+		infos[key] = &BackendInfo{
+			Addr:        backend.Addr,
+			ClusterName: backend.ClusterName,
+			Labels:      backend.Labels,
+			IP:          backend.IP,
+			StatusPort:  backend.StatusPort,
 		}
 	}
 	return infos, nil
@@ -98,7 +100,7 @@ func (sf *StaticFetcher) GetBackendList(context.Context) (map[string]*BackendInf
 func backendListToMap(addrs []string) map[string]*BackendInfo {
 	backends := make(map[string]*BackendInfo, len(addrs))
 	for _, addr := range addrs {
-		backends[addr] = &BackendInfo{}
+		backends[addr] = &BackendInfo{Addr: addr}
 	}
 	return backends
 }

pkg/balance/observer/backend_fetcher_test.go

@@ -26,6 +26,7 @@ func TestPDFetcher(t *testing.T) {
 		{
 			infos: map[string]*infosync.TiDBTopologyInfo{
 				"1.1.1.1:4000": {
+					Addr:       "1.1.1.1:4000",
 					Labels:     map[string]string{"k1": "v1"},
 					IP:         "1.1.1.1",
 					StatusPort: 10080,
@@ -34,6 +35,7 @@ func TestPDFetcher(t *testing.T) {
 			check: func(m map[string]*BackendInfo) {
 				require.Len(t, m, 1)
 				require.NotNil(t, m["1.1.1.1:4000"])
+				require.Equal(t, "1.1.1.1:4000", m["1.1.1.1:4000"].Addr)
 				require.Equal(t, "1.1.1.1", m["1.1.1.1:4000"].IP)
 				require.Equal(t, uint(10080), m["1.1.1.1:4000"].StatusPort)
 				require.Equal(t, map[string]string{"k1": "v1"}, m["1.1.1.1:4000"].Labels)
@@ -42,24 +44,43 @@ func TestPDFetcher(t *testing.T) {
 		{
 			infos: map[string]*infosync.TiDBTopologyInfo{
 				"1.1.1.1:4000": {
+					Addr:       "1.1.1.1:4000",
 					IP:         "1.1.1.1",
 					StatusPort: 10080,
 				},
 				"2.2.2.2:4000": {
+					Addr:       "2.2.2.2:4000",
 					IP:         "2.2.2.2",
 					StatusPort: 10080,
 				},
 			},
 			check: func(m map[string]*BackendInfo) {
 				require.Len(t, m, 2)
 				require.NotNil(t, m["1.1.1.1:4000"])
+				require.Equal(t, "1.1.1.1:4000", m["1.1.1.1:4000"].Addr)
 				require.Equal(t, "1.1.1.1", m["1.1.1.1:4000"].IP)
 				require.Equal(t, uint(10080), m["1.1.1.1:4000"].StatusPort)
 				require.NotNil(t, m["2.2.2.2:4000"])
+				require.Equal(t, "2.2.2.2:4000", m["2.2.2.2:4000"].Addr)
 				require.Equal(t, "2.2.2.2", m["2.2.2.2:4000"].IP)
 				require.Equal(t, uint(10080), m["2.2.2.2:4000"].StatusPort)
 			},
 		},
+		{
+			infos: map[string]*infosync.TiDBTopologyInfo{
+				"cluster-a/shared.tidb:4000": {
+					Addr:       "shared.tidb:4000",
+					IP:         "10.0.0.1",
+					StatusPort: 10080,
+				},
+			},
+			check: func(m map[string]*BackendInfo) {
+				require.Len(t, m, 1)
+				require.NotNil(t, m["cluster-a/shared.tidb:4000"])
+				require.Equal(t, "shared.tidb:4000", m["cluster-a/shared.tidb:4000"].Addr)
+				require.Equal(t, "10.0.0.1", m["cluster-a/shared.tidb:4000"].IP)
+			},
+		},
 		{
 			ctx: func() context.Context {
 				ctx, cancel := context.WithCancel(context.Background())

pkg/balance/observer/backend_health.go

@@ -76,13 +76,17 @@ func (bh *BackendHealth) String() string {
 
 // BackendInfo stores the status info of each backend.
 type BackendInfo struct {
-	Labels     map[string]string
-	IP         string
-	StatusPort uint
+	Addr        string
+	ClusterName string
+	Labels      map[string]string
+	IP          string
+	StatusPort  uint
 }
 
 func (bi BackendInfo) Equals(other BackendInfo) bool {
-	return bi.IP == other.IP &&
+	return bi.Addr == other.Addr &&
+		bi.ClusterName == other.ClusterName &&
+		bi.IP == other.IP &&
 		bi.StatusPort == other.StatusPort &&
 		maps.Equal(bi.Labels, other.Labels)
 }

pkg/balance/observer/backend_health_test.go

@@ -15,6 +15,7 @@ func TestBackendHealthToString(t *testing.T) {
 		{},
 		{
 			BackendInfo: BackendInfo{
+				Addr:       "127.0.0.1:4000",
 				IP:         "127.0.0.1",
 				StatusPort: 1,
 				Labels:     map[string]string{"k1": "v1", "k2": "v2"},
@@ -45,13 +46,15 @@ func TestBackendHealthEquals(t *testing.T) {
 		{
 			a: BackendHealth{
 				BackendInfo: BackendInfo{
+					Addr:       "127.0.0.1:4000",
 					IP:         "127.0.0.1",
 					StatusPort: 1,
 					Labels:     map[string]string{"k1": "v1", "k2": "v2"},
 				},
 			},
 			b: BackendHealth{
 				BackendInfo: BackendInfo{
+					Addr:       "127.0.0.1:4000",
 					IP:         "127.0.0.1",
 					StatusPort: 1,
 				},
@@ -61,13 +64,15 @@ func TestBackendHealthEquals(t *testing.T) {
 		{
 			a: BackendHealth{
 				BackendInfo: BackendInfo{
+					Addr:       "127.0.0.1:4000",
 					IP:         "127.0.0.1",
 					StatusPort: 1,
 					Labels:     map[string]string{"k1": "v1", "k2": "v2"},
 				},
 			},
 			b: BackendHealth{
 				BackendInfo: BackendInfo{
+					Addr:       "127.0.0.1:4000",
 					IP:         "127.0.0.1",
 					StatusPort: 1,
 					Labels:     map[string]string{"k1": "v1", "k2": "v2"},
@@ -78,6 +83,7 @@ func TestBackendHealthEquals(t *testing.T) {
 		{
 			a: BackendHealth{
 				BackendInfo: BackendInfo{
+					Addr:       "127.0.0.1:4000",
 					IP:         "127.0.0.1",
 					StatusPort: 1,
 					Labels:     map[string]string{"k1": "v1", "k2": "v2"},

pkg/balance/observer/backend_observer_test.go

@@ -279,6 +279,7 @@ func (ts *observerTestSuite) addBackend() (string, BackendInfo) {
 	ts.backendIdx++
 	addr := fmt.Sprintf("%d", ts.backendIdx)
 	info := &BackendInfo{
+		Addr:       addr,
 		IP:         "127.0.0.1",
 		StatusPort: uint(ts.backendIdx),
 	}

pkg/balance/observer/health_check.go

@@ -21,7 +21,7 @@ import (
 
 // HealthCheck is used to check the backends of one backend. One can pass a customized health check function to the observer.
 type HealthCheck interface {
-	Check(ctx context.Context, addr string, info *BackendInfo, lastHealth *BackendHealth) *BackendHealth
+	Check(ctx context.Context, backendID string, info *BackendInfo, lastHealth *BackendHealth) *BackendHealth
 }
 
 const (
@@ -62,7 +62,7 @@ func NewDefaultHealthCheck(httpCli *http.Client, cfg *config.HealthCheck, logger
 	}
 }
 
-func (dhc *DefaultHealthCheck) Check(ctx context.Context, addr string, info *BackendInfo, lastBh *BackendHealth) *BackendHealth {
+func (dhc *DefaultHealthCheck) Check(ctx context.Context, _ string, info *BackendInfo, lastBh *BackendHealth) *BackendHealth {
 	bh := &BackendHealth{
 		BackendInfo: *info,
 		Healthy:     true,
@@ -80,16 +80,22 @@ func (dhc *DefaultHealthCheck) Check(ctx context.Context, addr string, info *Bac
 	if !bh.Healthy {
 		return bh
 	}
-	dhc.checkSqlPort(ctx, addr, bh)
+	dhc.checkSqlPort(ctx, info, bh)
 	if !bh.Healthy {
 		return bh
 	}
 	dhc.queryConfig(ctx, info, bh, lastBh)
 	return bh
 }
 
-func (dhc *DefaultHealthCheck) checkSqlPort(ctx context.Context, addr string, bh *BackendHealth) {
+func (dhc *DefaultHealthCheck) checkSqlPort(ctx context.Context, info *BackendInfo, bh *BackendHealth) {
 	// Also dial the SQL port just in case that the SQL port hangs.
+	if info == nil || info.Addr == "" {
+		bh.Healthy = false
+		bh.PingErr = errors.New("backend address is empty")
+		return
+	}
+	addr := info.Addr
 	b := backoff.WithContext(backoff.WithMaxRetries(backoff.NewConstantBackOff(dhc.cfg.RetryInterval), uint64(dhc.cfg.MaxRetries)), ctx)
 	err := http.ConnectWithRetry(func() error {
 		startTime := time.Now()

pkg/balance/observer/health_check_test.go

@@ -143,6 +143,7 @@ func newBackendServer(t *testing.T) (*backendServer, *BackendInfo) {
 	backend.setSqlResp(true)
 	backend.startSQLServer()
 	return backend, &BackendInfo{
+		Addr:       backend.sqlAddr,
 		IP:         backend.ip,
 		StatusPort: backend.statusPort,
 	}

pkg/balance/observer/mock_test.go

@@ -82,11 +82,11 @@ func newMockHealthCheck() *mockHealthCheck {
 	}
 }
 
-func (mhc *mockHealthCheck) Check(_ context.Context, addr string, info *BackendInfo, _ *BackendHealth) *BackendHealth {
+func (mhc *mockHealthCheck) Check(_ context.Context, backendID string, info *BackendInfo, _ *BackendHealth) *BackendHealth {
 	mhc.Lock()
 	defer mhc.Unlock()
-	mhc.backends[addr].BackendInfo = *info
-	return mhc.backends[addr]
+	mhc.backends[backendID].BackendInfo = *info
+	return mhc.backends[backendID]
 }
 
 func (mhc *mockHealthCheck) setBackend(addr string, health *BackendHealth) {

pkg/balance/router/group.go

@@ -158,17 +158,17 @@ func (g *Group) RefreshCidr() {
 	}
 }
 
-func (g *Group) AddBackend(addr string, backend *backendWrapper) {
+func (g *Group) AddBackend(backendID string, backend *backendWrapper) {
 	g.Lock()
 	defer g.Unlock()
-	g.backends[addr] = backend
+	g.backends[backendID] = backend
 	backend.group = g
 }
 
-func (g *Group) RemoveBackend(addr string) {
+func (g *Group) RemoveBackend(backendID string) {
 	g.Lock()
 	defer g.Unlock()
-	delete(g.backends, addr)
+	delete(g.backends, backendID)
 }
 
 func (g *Group) Empty() bool {
@@ -200,7 +200,7 @@ func (g *Group) Route(excluded []BackendInst) (policy.BackendCtx, error) {
 		// Exclude the backends that are already tried.
 		found := false
 		for _, e := range excluded {
-			if backend.Addr() == e.Addr() {
+			if backend.ID() == e.ID() {
 				found = true
 				break
 			}
@@ -273,7 +273,7 @@ func (g *Group) Balance(ctx context.Context) {
 func (g *Group) onCreateConn(backendInst BackendInst, conn RedirectableConn, succeed bool) {
 	g.Lock()
 	defer g.Unlock()
-	backend := g.ensureBackend(backendInst.Addr())
+	backend := g.ensureBackend(backendInst.ID())
 	if succeed {
 		connWrapper := &connWrapper{
 			RedirectableConn: conn,
@@ -319,23 +319,24 @@ func (g *Group) RedirectConnections() error {
 	return nil
 }
 
-func (g *Group) ensureBackend(addr string) *backendWrapper {
-	backend, ok := g.backends[addr]
+func (g *Group) ensureBackend(backendID string) *backendWrapper {
+	backend, ok := g.backends[backendID]
 	if ok {
 		return backend
 	}
 	// The backend should always exist if it will be needed. Add a warning and add it back.
-	g.lg.Warn("backend is not found in the router", zap.String("backend_addr", addr), zap.Stack("stack"))
-	ip, _, _ := net.SplitHostPort(addr)
-	backend = newBackendWrapper(addr, observer.BackendHealth{
+	g.lg.Warn("backend is not found in the router", zap.String("backend_id", backendID), zap.Stack("stack"))
+	ip, _, _ := net.SplitHostPort(backendID)
+	backend = newBackendWrapper(backendID, observer.BackendHealth{
 		BackendInfo: observer.BackendInfo{
+			Addr:       backendID,
 			IP:         ip,
 			StatusPort: 10080, // impossible anyway
 		},
 		SupportRedirection: true,
 		Healthy:            false,
 	})
-	g.backends[addr] = backend
+	g.backends[backendID] = backend
 	return backend
 }