server, etcd: integrate vip manager into the server (#592)

Author: djshow832

pkg/manager/elect/mock_test.go

@@ -8,10 +8,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pingcap/tiproxy/lib/config"
 	"github.com/pingcap/tiproxy/lib/util/logger"
 	"github.com/pingcap/tiproxy/pkg/manager/cert"
-	"github.com/pingcap/tiproxy/pkg/manager/infosync"
+	"github.com/pingcap/tiproxy/pkg/util/etcd"
 	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/server/v3/embed"
@@ -95,13 +94,13 @@ func newEtcdTestSuite(t *testing.T, elecCfg electionConfig, key string) *etcdTes
 
 	ts.startServer("0.0.0.0:0")
 	endpoint := ts.server.Clients[0].Addr().String()
-	cfg := newConfig(endpoint)
+	cfg := etcd.ConfigForEtcdTest(endpoint)
 
 	certMgr := cert.NewCertManager()
 	err := certMgr.Init(cfg, lg, nil)
 	require.NoError(t, err)
 
-	ts.client, err = infosync.InitEtcdClient(ts.lg, cfg, certMgr)
+	ts.client, err = etcd.InitEtcdClient(ts.lg, cfg, certMgr)
 	require.NoError(t, err)
 	ts.kv = clientv3.NewKV(ts.client)
 	return ts
@@ -175,7 +174,7 @@ func (ts *etcdTestSuite) close() {
 }
 
 func (ts *etcdTestSuite) startServer(addr string) {
-	etcd, err := infosync.CreateEtcdServer(addr, ts.t.TempDir(), ts.lg)
+	etcd, err := etcd.CreateEtcdServer(addr, ts.t.TempDir(), ts.lg)
 	require.NoError(ts.t, err)
 	ts.server = etcd
 }
@@ -188,18 +187,6 @@ func (ts *etcdTestSuite) shutdownServer() string {
 	return addr
 }
 
-func newConfig(endpoint string) *config.Config {
-	return &config.Config{
-		Proxy: config.ProxyServer{
-			Addr:    "0.0.0.0:6000",
-			PDAddrs: endpoint,
-		},
-		API: config.API{
-			Addr: "0.0.0.0:3080",
-		},
-	}
-}
-
 func electionConfigForTest(ttl int) electionConfig {
 	return electionConfig{
 		sessionTTL: ttl,

pkg/manager/infosync/info.go

@@ -18,7 +18,6 @@ import (
 	"github.com/pingcap/tiproxy/lib/util/retry"
 	"github.com/pingcap/tiproxy/lib/util/sys"
 	"github.com/pingcap/tiproxy/lib/util/waitgroup"
-	"github.com/pingcap/tiproxy/pkg/manager/cert"
 	"github.com/pingcap/tiproxy/pkg/util/versioninfo"
 	"github.com/siddontang/go/hack"
 	clientv3 "go.etcd.io/etcd/client/v3"
@@ -108,7 +107,7 @@ type PrometheusInfo struct {
 	Port       int    `json:"port"`
 }
 
-func NewInfoSyncer(lg *zap.Logger) *InfoSyncer {
+func NewInfoSyncer(lg *zap.Logger, etcdCli *clientv3.Client) *InfoSyncer {
 	return &InfoSyncer{
 		lg: lg,
 		syncConfig: syncConfig{
@@ -121,16 +120,11 @@ func NewInfoSyncer(lg *zap.Logger) *InfoSyncer {
 			getPromRetryCnt:   getPromRetryCnt,
 			getPromRetryIntvl: getPromRetryIntvl,
 		},
+		etcdCli: etcdCli,
 	}
 }
 
-func (is *InfoSyncer) Init(ctx context.Context, cfg *config.Config, certMgr *cert.CertManager) error {
-	etcdCli, err := InitEtcdClient(is.lg, cfg, certMgr)
-	if err != nil {
-		return err
-	}
-	is.etcdCli = etcdCli
-
+func (is *InfoSyncer) Init(ctx context.Context, cfg *config.Config) error {
 	topologyInfo, err := is.getTopologyInfo(cfg)
 	if err != nil {
 		is.lg.Error("get topology failed", zap.Error(err))
@@ -338,14 +332,5 @@ func (is *InfoSyncer) Close() error {
 		is.cancelFunc()
 	}
 	is.wg.Wait()
-	errs := make([]error, 0)
-	if err := is.removeTopology(context.Background()); err != nil {
-		errs = append(errs, err)
-	}
-	if is.etcdCli != nil {
-		if err := is.etcdCli.Close(); err != nil {
-			errs = append(errs, errors.WithStack(err))
-		}
-	}
-	return errors.Collect(errors.New("closing InfoSyncer"), errs...)
+	return is.removeTopology(context.Background())
 }

pkg/manager/infosync/info_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/pingcap/tiproxy/lib/util/sys"
 	"github.com/pingcap/tiproxy/lib/util/waitgroup"
 	"github.com/pingcap/tiproxy/pkg/manager/cert"
+	"github.com/pingcap/tiproxy/pkg/util/etcd"
 	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/server/v3/embed"
@@ -276,12 +277,16 @@ func newEtcdTestSuite(t *testing.T) *etcdTestSuite {
 
 	ts.startServer("0.0.0.0:0")
 	endpoint := ts.server.Clients[0].Addr().String()
-	cfg := newConfig(endpoint)
+	cfg := etcd.ConfigForEtcdTest(endpoint)
 
 	certMgr := cert.NewCertManager()
 	err := certMgr.Init(cfg, lg, nil)
 	require.NoError(t, err)
-	is := NewInfoSyncer(lg)
+	ts.client, err = etcd.InitEtcdClient(ts.lg, cfg, certMgr)
+	require.NoError(t, err)
+	ts.kv = clientv3.NewKV(ts.client)
+
+	is := NewInfoSyncer(lg, ts.client)
 	is.syncConfig = syncConfig{
 		sessionTTL:        1,
 		refreshIntvl:      50 * time.Millisecond,
@@ -293,14 +298,10 @@ func newEtcdTestSuite(t *testing.T) *etcdTestSuite {
 		getPromRetryCnt:   2,
 	}
 	ctx, cancel := context.WithCancel(context.Background())
-	err = is.Init(ctx, cfg, certMgr)
+	err = is.Init(ctx, cfg)
 	require.NoError(t, err)
 	ts.is = is
 	ts.cancel = cancel
-
-	ts.client, err = InitEtcdClient(ts.lg, cfg, certMgr)
-	require.NoError(t, err)
-	ts.kv = clientv3.NewKV(ts.client)
 	return ts
 }
 
@@ -386,19 +387,7 @@ func (ts *etcdTestSuite) setPromInfo(info *PrometheusInfo) {
 }
 
 func (ts *etcdTestSuite) createEtcdServer(addr string) {
-	etcd, err := CreateEtcdServer(addr, ts.t.TempDir(), ts.lg)
+	etcd, err := etcd.CreateEtcdServer(addr, ts.t.TempDir(), ts.lg)
 	require.NoError(ts.t, err)
 	ts.server = etcd
 }
-
-func newConfig(endpoint string) *config.Config {
-	return &config.Config{
-		Proxy: config.ProxyServer{
-			Addr:    "0.0.0.0:6000",
-			PDAddrs: endpoint,
-		},
-		API: config.API{
-			Addr: "0.0.0.0:3080",
-		},
-	}
-}

pkg/manager/vip/network_test.go

@@ -6,13 +6,18 @@
 package vip
 
 import (
+	"runtime"
 	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 )
 
 func TestAddDelIP(t *testing.T) {
+	if runtime.GOOS != "linux" {
+		t.Skipf("unsupported on %s", runtime.GOOS)
+	}
+
 	tests := []struct {
 		virtualIP string
 		link      string