router: refactor creation of backendFetcher (#141)

pkg/manager/namespace/manager.go

@@ -42,7 +42,13 @@ func NewNamespaceManager() *NamespaceManager {
 func (mgr *NamespaceManager) buildNamespace(cfg *config.Namespace) (*Namespace, error) {
 	logger := mgr.logger.With(zap.String("namespace", cfg.Namespace))
 
-	rt, err := router.NewScoreBasedRouter(logger.Named("router"), &cfg.Backend, mgr.client, mgr.httpCli)
+	var fetcher router.BackendFetcher
+	if mgr.client != nil {
+		fetcher = router.NewPDFetcher(mgr.client, logger.Named("be_fetcher"), router.NewDefaultHealthCheckConfig())
+	} else {
+		fetcher = router.NewStaticFetcher(cfg.Backend.Instances)
+	}
+	rt, err := router.NewScoreBasedRouter(logger.Named("router"), mgr.httpCli, fetcher)
 	if err != nil {
 		return nil, errors.Errorf("build router error: %w", err)
 	}

pkg/manager/router/backend_observer.go

@@ -24,7 +24,6 @@ import (
 
 	"github.com/pingcap/TiProxy/lib/util/waitgroup"
 	pnet "github.com/pingcap/TiProxy/pkg/proxy/net"
-	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/zap"
 )
 
@@ -88,7 +87,8 @@ type HealthCheckConfig struct {
 	tombstoneThreshold       time.Duration
 }
 
-func newDefaultHealthCheckConfig() *HealthCheckConfig {
+// NewDefaultHealthCheckConfig creates a default HealthCheckConfig.
+func NewDefaultHealthCheckConfig() *HealthCheckConfig {
 	return &HealthCheckConfig{
 		healthCheckInterval:      healthCheckInterval,
 		healthCheckMaxRetries:    healthCheckMaxRetries,
@@ -125,9 +125,9 @@ type BackendObserver struct {
 }
 
 // StartBackendObserver creates a BackendObserver and starts watching.
-func StartBackendObserver(logger *zap.Logger, eventReceiver BackendEventReceiver, client *clientv3.Client, httpCli *http.Client,
-	config *HealthCheckConfig, staticAddrs []string, customBackendFetcher BackendFetcher) (*BackendObserver, error) {
-	bo, err := NewBackendObserver(logger, eventReceiver, client, httpCli, config, staticAddrs, customBackendFetcher)
+func StartBackendObserver(logger *zap.Logger, eventReceiver BackendEventReceiver, httpCli *http.Client,
+	config *HealthCheckConfig, backendFetcher BackendFetcher) (*BackendObserver, error) {
+	bo, err := NewBackendObserver(logger, eventReceiver, httpCli, config, backendFetcher)
 	if err != nil {
 		return nil, err
 	}
@@ -136,8 +136,8 @@ func StartBackendObserver(logger *zap.Logger, eventReceiver BackendEventReceiver
 }
 
 // NewBackendObserver creates a BackendObserver.
-func NewBackendObserver(logger *zap.Logger, eventReceiver BackendEventReceiver, client *clientv3.Client, httpCli *http.Client,
-	config *HealthCheckConfig, staticAddrs []string, customBackendFetcher BackendFetcher) (*BackendObserver, error) {
+func NewBackendObserver(logger *zap.Logger, eventReceiver BackendEventReceiver, httpCli *http.Client,
+	config *HealthCheckConfig, backendFetcher BackendFetcher) (*BackendObserver, error) {
 	if httpCli == nil {
 		httpCli = http.DefaultClient
 	}
@@ -153,13 +153,7 @@ func NewBackendObserver(logger *zap.Logger, eventReceiver BackendEventReceiver,
 		httpTLS:        httpTLS,
 		eventReceiver:  eventReceiver,
 	}
-	if customBackendFetcher != nil {
-		bo.fetcher = customBackendFetcher
-	} else if client != nil {
-		bo.fetcher = NewPDFetcher(client, logger.Named("be_fetcher"), config)
-	} else {
-		bo.fetcher = NewStaticFetcher(staticAddrs)
-	}
+	bo.fetcher = backendFetcher
 	return bo, nil
 }
 

pkg/manager/router/backend_observer_test.go

@@ -156,7 +156,7 @@ func TestExternalFetcher(t *testing.T) {
 	backendChan := make(chan map[string]BackendStatus, 1)
 	mer := newMockEventReceiver(backendChan)
 	fetcher := NewExternalFetcher(backendGetter)
-	bo, err := NewBackendObserver(logger.CreateLoggerForTest(t), mer, nil, nil, newHealthCheckConfigForTest(), nil, fetcher)
+	bo, err := NewBackendObserver(logger.CreateLoggerForTest(t), mer, nil, newHealthCheckConfigForTest(), fetcher)
 	require.NoError(t, err)
 	bo.Start()
 	defer bo.Close()
@@ -183,7 +183,8 @@ func runETCDTest(t *testing.T, f func(etcd *embed.Etcd, kv clientv3.KV, bo *Back
 	kv := clientv3.NewKV(client)
 	backendChan := make(chan map[string]BackendStatus, 1)
 	mer := newMockEventReceiver(backendChan)
-	bo, err := NewBackendObserver(logger.CreateLoggerForTest(t), mer, client, nil, newHealthCheckConfigForTest(), nil, nil)
+	fetcher := NewPDFetcher(client, logger.CreateLoggerForTest(t), newHealthCheckConfigForTest())
+	bo, err := NewBackendObserver(logger.CreateLoggerForTest(t), mer, nil, newHealthCheckConfigForTest(), fetcher)
 	require.NoError(t, err)
 	f(etcd, kv, bo, backendChan)
 	bo.Close()

pkg/manager/router/router.go

@@ -21,10 +21,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/pingcap/TiProxy/lib/config"
 	"github.com/pingcap/TiProxy/lib/util/errors"
 	"github.com/pingcap/TiProxy/lib/util/waitgroup"
-	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/zap"
 )
 
@@ -117,14 +115,14 @@ type ScoreBasedRouter struct {
 }
 
 // NewScoreBasedRouter creates a ScoreBasedRouter.
-func NewScoreBasedRouter(logger *zap.Logger, cfg *config.BackendNamespace, client *clientv3.Client, httpCli *http.Client) (*ScoreBasedRouter, error) {
+func NewScoreBasedRouter(logger *zap.Logger, httpCli *http.Client, fetcher BackendFetcher) (*ScoreBasedRouter, error) {
 	router := &ScoreBasedRouter{
 		logger:   logger,
 		backends: list.New(),
 	}
 	router.Lock()
 	defer router.Unlock()
-	observer, err := StartBackendObserver(logger.Named("observer"), router, client, httpCli, newDefaultHealthCheckConfig(), cfg.Instances, nil)
+	observer, err := StartBackendObserver(logger.Named("observer"), router, httpCli, NewDefaultHealthCheckConfig(), fetcher)
 	if err != nil {
 		return nil, err
 	}

pkg/manager/router/router_test.go

@@ -24,7 +24,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pingcap/TiProxy/lib/config"
 	"github.com/pingcap/TiProxy/lib/util/logger"
 	"github.com/pingcap/TiProxy/lib/util/waitgroup"
 	"github.com/pingcap/TiProxy/pkg/metrics"
@@ -520,12 +519,12 @@ func TestRebalanceCornerCase(t *testing.T) {
 
 // Test all kinds of events occur concurrently.
 func TestConcurrency(t *testing.T) {
-	cfg := &config.BackendNamespace{}
 	// Router.observer doesn't work because the etcd is always empty.
 	// We create other goroutines to change backends easily.
 	etcd := createEtcdServer(t, "127.0.0.1:0")
 	client := createEtcdClient(t, etcd)
-	router, err := NewScoreBasedRouter(logger.CreateLoggerForTest(t), cfg, client, nil)
+	fetcher := NewPDFetcher(client, logger.CreateLoggerForTest(t), newHealthCheckConfigForTest())
+	router, err := NewScoreBasedRouter(logger.CreateLoggerForTest(t), nil, fetcher)
 	require.NoError(t, err)
 
 	var wg waitgroup.WaitGroup