mcs: support advertise address

cmd/pd-server/main.go

@@ -81,7 +81,9 @@ func NewTSOServiceCommand() *cobra.Command {
 	cmd.Flags().BoolP("version", "V", false, "print version information and exit")
 	cmd.Flags().StringP("config", "", "", "config file")
 	cmd.Flags().StringP("backend-endpoints", "", "", "url for etcd client")
+	cmd.Flags().StringP("advertise-backend-endpoints", "", "", "advertise urls for backend endpoints (default '${backend-endpoints}')")
 	cmd.Flags().StringP("listen-addr", "", "", "listen address for tso service")
+	cmd.Flags().StringP("advertise-listen-addr", "", "", "advertise urls for listen address (default '${listen-addr}')")
 	cmd.Flags().StringP("cacert", "", "", "path of file that contains list of trusted TLS CAs")
 	cmd.Flags().StringP("cert", "", "", "path of file that contains X509 certificate in PEM format")
 	cmd.Flags().StringP("key", "", "", "path of file that contains X509 key in PEM format")
@@ -98,7 +100,9 @@ func NewResourceManagerServiceCommand() *cobra.Command {
 	cmd.Flags().BoolP("version", "V", false, "print version information and exit")
 	cmd.Flags().StringP("config", "", "", "config file")
 	cmd.Flags().StringP("backend-endpoints", "", "", "url for etcd client")
+	cmd.Flags().StringP("advertise-backend-endpoints", "", "", "advertise urls for backend endpoints (default '${backend-endpoints}')")
 	cmd.Flags().StringP("listen-addr", "", "", "listen address for resource management service")
+	cmd.Flags().StringP("advertise-listen-addr", "", "", "advertise urls for listen address (default '${listen-addr}')")
 	cmd.Flags().StringP("cacert", "", "", "path of file that contains list of trusted TLS CAs")
 	cmd.Flags().StringP("cert", "", "", "path of file that contains X509 certificate in PEM format")
 	cmd.Flags().StringP("key", "", "", "path of file that contains X509 key in PEM format")

pkg/mcs/resource_manager/server/config.go

@@ -55,11 +55,13 @@ const (
 
 // Config is the configuration for the resource manager.
 type Config struct {
-	BackendEndpoints  string `toml:"backend-endpoints" json:"backend-endpoints"`
-	ListenAddr        string `toml:"listen-addr" json:"listen-addr"`
-	Name              string `toml:"name" json:"name"`
-	DataDir           string `toml:"data-dir" json:"data-dir"` // TODO: remove this after refactoring
-	EnableGRPCGateway bool   `json:"enable-grpc-gateway"`      // TODO: use it
+	BackendEndpoints          string `toml:"backend-endpoints" json:"backend-endpoints"`
+	AdvertiseBackendEndpoints string `toml:"advertise-backend-endpoints" json:"advertise-backend-endpoints"`
+	ListenAddr                string `toml:"listen-addr" json:"listen-addr"`
+	AdvertiseListenAddr       string `toml:"advertise-listen-addr" json:"advertise-listen-addr"`
+	Name                      string `toml:"name" json:"name"`
+	DataDir                   string `toml:"data-dir" json:"data-dir"` // TODO: remove this after refactoring
+	EnableGRPCGateway         bool   `json:"enable-grpc-gateway"`      // TODO: use it
 
 	Metric metricutil.MetricConfig `toml:"metric" json:"metric"`
 
@@ -169,7 +171,9 @@ func (c *Config) Parse(flagSet *pflag.FlagSet) error {
 	configutil.AdjustCommandlineString(flagSet, &c.Security.CertPath, "cert")
 	configutil.AdjustCommandlineString(flagSet, &c.Security.KeyPath, "key")
 	configutil.AdjustCommandlineString(flagSet, &c.BackendEndpoints, "backend-endpoints")
+	configutil.AdjustCommandlineString(flagSet, &c.AdvertiseBackendEndpoints, "advertise-backend-endpoints")
 	configutil.AdjustCommandlineString(flagSet, &c.ListenAddr, "listen-addr")
+	configutil.AdjustCommandlineString(flagSet, &c.AdvertiseListenAddr, "advertise-listen-addr")
 
 	return c.Adjust(meta, false)
 }
@@ -198,7 +202,9 @@ func (c *Config) Adjust(meta *toml.MetaData, reloading bool) error {
 	}
 
 	configutil.AdjustString(&c.BackendEndpoints, defaultBackendEndpoints)
+	configutil.AdjustString(&c.AdvertiseBackendEndpoints, c.BackendEndpoints)
 	configutil.AdjustString(&c.ListenAddr, defaultListenAddr)
+	configutil.AdjustString(&c.AdvertiseListenAddr, c.ListenAddr)
 
 	if !configMetaData.IsDefined("enable-grpc-gateway") {
 		c.EnableGRPCGateway = utils.DefaultEnableGRPCGateway

pkg/mcs/resource_manager/server/server.go

@@ -365,7 +365,7 @@ func (s *Server) startServer() (err error) {
 	log.Info("joining primary election", zap.String("participant-name", uniqueName), zap.Uint64("participant-id", uniqueID))
 	resourceManagerPrimaryPrefix := fmt.Sprintf("/ms/%d/resource_manager", s.clusterID)
 	s.participant = member.NewParticipant(s.etcdClient)
-	s.participant.InitInfo(uniqueName, uniqueID, path.Join(resourceManagerPrimaryPrefix, fmt.Sprintf("%05d", 0)), "primary", "keyspace group primary election", s.cfg.ListenAddr)
+	s.participant.InitInfo(uniqueName, uniqueID, path.Join(resourceManagerPrimaryPrefix, fmt.Sprintf("%05d", 0)), "primary", "keyspace group primary election", s.cfg.AdvertiseListenAddr)
 
 	s.service = &Service{
 		ctx:     s.ctx,

pkg/mcs/tso/server/config.go

@@ -48,8 +48,10 @@ const (
 
 // Config is the configuration for the TSO.
 type Config struct {
-	BackendEndpoints string `toml:"backend-endpoints" json:"backend-endpoints"`
-	ListenAddr       string `toml:"listen-addr" json:"listen-addr"`
+	BackendEndpoints          string `toml:"backend-endpoints" json:"backend-endpoints"`
+	AdvertiseBackendEndpoints string `toml:"advertise-backend-endpoints" json:"advertise-backend-endpoints"`
+	ListenAddr                string `toml:"listen-addr" json:"listen-addr"`
+	AdvertiseListenAddr       string `toml:"advertise-listen-addr" json:"advertise-listen-addr"`
 
 	Name              string `toml:"name" json:"name"`
 	DataDir           string `toml:"data-dir" json:"data-dir"`
@@ -140,7 +142,9 @@ func (c *Config) Parse(flagSet *pflag.FlagSet) error {
 	configutil.AdjustCommandlineString(flagSet, &c.Security.CertPath, "cert")
 	configutil.AdjustCommandlineString(flagSet, &c.Security.KeyPath, "key")
 	configutil.AdjustCommandlineString(flagSet, &c.BackendEndpoints, "backend-endpoints")
+	configutil.AdjustCommandlineString(flagSet, &c.AdvertiseBackendEndpoints, "advertise-backend-endpoints")
 	configutil.AdjustCommandlineString(flagSet, &c.ListenAddr, "listen-addr")
+	configutil.AdjustCommandlineString(flagSet, &c.AdvertiseListenAddr, "advertise-listen-addr")
 
 	return c.Adjust(meta, false)
 }
@@ -166,7 +170,9 @@ func (c *Config) Adjust(meta *toml.MetaData, reloading bool) error {
 	}
 
 	configutil.AdjustString(&c.BackendEndpoints, defaultBackendEndpoints)
+	configutil.AdjustString(&c.AdvertiseBackendEndpoints, c.BackendEndpoints)
 	configutil.AdjustString(&c.ListenAddr, defaultListenAddr)
+	configutil.AdjustString(&c.AdvertiseListenAddr, c.ListenAddr)
 
 	configutil.AdjustDuration(&c.MaxResetTSGap, defaultMaxResetTSGap)
 	configutil.AdjustInt64(&c.LeaderLease, utils.DefaultLeaderLease)

pkg/mcs/tso/server/server.go

@@ -574,7 +574,7 @@ func (s *Server) startServer() (err error) {
 
 	s.participant = member.NewParticipant(s.etcdClient)
 	s.participant.InitInfo(uniqueName, uniqueID, fmt.Sprintf(tsoSvcDiscoveryPrefixFormat, s.clusterID, mcsutils.DefaultKeyspaceID),
-		"primary", "keyspace group primary election", s.cfg.ListenAddr)
+		"primary", "keyspace group primary election", s.cfg.AdvertiseListenAddr)
 
 	s.defaultGroupStorage = endpoint.NewStorageEndpoint(kv.NewEtcdKVBase(s.GetClient(), s.defaultGroupRootPath), nil)
 	s.tsoAllocatorManager = tso.NewAllocatorManager(

pkg/mcs/tso/server/testutil.go

@@ -21,7 +21,7 @@ import (
 
 	"github.com/pingcap/kvproto/pkg/tsopb"
 	"github.com/pingcap/log"
-	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/testutil"
@@ -49,21 +49,44 @@ func NewTSOTestServer(ctx context.Context, re *require.Assertions, cfg *Config)
 	return s, cleanup, nil
 }
 
-// NewTSOTestDefaultConfig is only for test to create one pd.
-// Because PD client also needs this, so export here.
-func NewTSOTestDefaultConfig() (*Config, error) {
-	cmd := &cobra.Command{
-		Use:   "tso",
-		Short: "Run the tso service",
-	}
-	cfg := NewConfig()
-	flagSet := cmd.Flags()
-	return cfg, cfg.Parse(flagSet)
-}
-
 // MustNewGrpcClient must create a new TSO grpc client.
 func MustNewGrpcClient(re *require.Assertions, addr string) (*grpc.ClientConn, tsopb.TSOClient) {
 	conn, err := grpc.Dial(strings.TrimPrefix(addr, "http://"), grpc.WithInsecure())
 	re.NoError(err)
 	return conn, tsopb.NewTSOClient(conn)
 }
+
+// ConfigOption is the type of the functional option for config.
+type ConfigOption func(conf *Config)
+
+// GenerateConfig generates a new config with the given options.
+func GenerateConfig(c *Config, opts ...ConfigOption) (*Config, error) {
+	arguments := []string{
+		"--listen-addr=" + c.ListenAddr,
+		"--advertise-listen-addr=" + c.AdvertiseListenAddr,
+		"--backend-endpoints=" + c.BackendEndpoints,
+		"--advertise-backend-endpoints=" + c.AdvertiseBackendEndpoints,
+	}
+
+	flagSet := pflag.NewFlagSet("test", pflag.ContinueOnError)
+	flagSet.BoolP("version", "V", false, "print version information and exit")
+	flagSet.StringP("config", "", "", "config file")
+	flagSet.StringP("backend-endpoints", "", "", "url for etcd client")
+	flagSet.StringP("advertise-backend-endpoints", "", "", "advertise urls for backend endpoints (default '${backend-endpoints}')")
+	flagSet.StringP("listen-addr", "", "", "listen address for tso service")
+	flagSet.StringP("advertise-listen-addr", "", "", "advertise urls for listen address (default '${listen-addr}')")
+	flagSet.StringP("cacert", "", "", "path of file that contains list of trusted TLS CAs")
+	flagSet.StringP("cert", "", "", "path of file that contains X509 certificate in PEM format")
+	flagSet.StringP("key", "", "", "path of file that contains X509 key in PEM format")
+	err := flagSet.Parse(arguments)
+	if err != nil {
+		return nil, err
+	}
+	cfg := NewConfig()
+	err = cfg.Parse(flagSet)
+	if err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
+}

pkg/member/participant.go

@@ -56,11 +56,11 @@ func NewParticipant(client *clientv3.Client) *Participant {
 }
 
 // InitInfo initializes the member info. The leader key is path.Join(rootPath, leaderName)
-func (m *Participant) InitInfo(name string, id uint64, rootPath string, leaderName string, purpose string, listenURL string) {
+func (m *Participant) InitInfo(name string, id uint64, rootPath string, leaderName string, purpose string, advertiseListenAddr string) {
 	leader := &tsopb.Participant{
 		Name:       name,
 		Id:         id, // id is unique among all participants
-		ListenUrls: []string{listenURL},
+		ListenUrls: []string{advertiseListenAddr},
 	}
 
 	data, err := leader.Marshal()

tests/mcs/discovery/register_test.go

@@ -22,6 +22,7 @@ import (
 	bs "github.com/tikv/pd/pkg/basicserver"
 	"github.com/tikv/pd/pkg/mcs/discovery"
 	"github.com/tikv/pd/pkg/mcs/utils"
+	"github.com/tikv/pd/pkg/utils/tempurl"
 	"github.com/tikv/pd/pkg/utils/testutil"
 	"github.com/tikv/pd/tests"
 	"github.com/tikv/pd/tests/mcs"
@@ -39,6 +40,7 @@ type serverRegisterTestSuite struct {
 	cluster          *tests.TestCluster
 	pdLeader         *tests.TestServer
 	backendEndpoints string
+	listenAddr       string
 }
 
 func TestServerRegisterTestSuite(t *testing.T) {
@@ -59,6 +61,7 @@ func (suite *serverRegisterTestSuite) SetupSuite() {
 	leaderName := suite.cluster.WaitLeader()
 	suite.pdLeader = suite.cluster.GetServer(leaderName)
 	suite.backendEndpoints = suite.pdLeader.GetAddr()
+	suite.listenAddr = tempurl.Alloc()
 }
 
 func (suite *serverRegisterTestSuite) TearDownSuite() {
@@ -147,7 +150,7 @@ func (suite *serverRegisterTestSuite) addServer(serviceName string) (bs.Server,
 	re := suite.Require()
 	switch serviceName {
 	case utils.TSOServiceName:
-		return mcs.StartSingleTSOTestServer(suite.ctx, re, suite.backendEndpoints)
+		return mcs.StartSingleTSOTestServer(suite.ctx, re, suite.backendEndpoints, suite.listenAddr)
 	case utils.ResourceManagerServiceName:
 		return mcs.StartSingleResourceManagerTestServer(suite.ctx, re, suite.backendEndpoints)
 	default:

tests/mcs/testutil.go

@@ -54,15 +54,12 @@ func StartSingleResourceManagerTestServer(ctx context.Context, re *require.Asser
 }
 
 // StartSingleTSOTestServer creates and starts a tso server with default config for testing.
-func StartSingleTSOTestServer(ctx context.Context, re *require.Assertions, backendEndpoints string, listenAddrs ...string) (*tso.Server, func()) {
-	cfg, err := tso.NewTSOTestDefaultConfig()
-	re.NoError(err)
+func StartSingleTSOTestServer(ctx context.Context, re *require.Assertions, backendEndpoints, listenAddrs string, opts ...tso.ConfigOption) (*tso.Server, func()) {
+	cfg := tso.NewConfig()
 	cfg.BackendEndpoints = backendEndpoints
-	if len(listenAddrs) > 0 {
-		cfg.ListenAddr = listenAddrs[0]
-	} else {
-		cfg.ListenAddr = tempurl.Alloc()
-	}
+	cfg.ListenAddr = listenAddrs
+	cfg, err := tso.GenerateConfig(cfg)
+	re.NoError(err)
 
 	s, cleanup, err := tso.NewTSOTestServer(ctx, re, cfg)
 	re.NoError(err)

tests/mcs/tso/server_test.go

@@ -32,6 +32,7 @@ import (
 	tsoapi "github.com/tikv/pd/pkg/mcs/tso/server/apis/v1"
 	"github.com/tikv/pd/pkg/mcs/utils"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
+	"github.com/tikv/pd/pkg/utils/tempurl"
 	"github.com/tikv/pd/pkg/utils/testutil"
 	"github.com/tikv/pd/pkg/utils/tsoutil"
 	"github.com/tikv/pd/tests"
@@ -52,6 +53,7 @@ type tsoServerTestSuite struct {
 	cluster          *tests.TestCluster
 	pdLeader         *tests.TestServer
 	backendEndpoints string
+	listenAddr       string
 }
 
 func TestTSOServerTestSuite(t *testing.T) {
@@ -72,6 +74,7 @@ func (suite *tsoServerTestSuite) SetupSuite() {
 	leaderName := suite.cluster.WaitLeader()
 	suite.pdLeader = suite.cluster.GetServer(leaderName)
 	suite.backendEndpoints = suite.pdLeader.GetAddr()
+	suite.listenAddr = tempurl.Alloc()
 }
 
 func (suite *tsoServerTestSuite) TearDownSuite() {
@@ -88,7 +91,7 @@ func (suite *tsoServerTestSuite) TestTSOServerStartAndStopNormally() {
 	}()
 
 	re := suite.Require()
-	s, cleanup := mcs.StartSingleTSOTestServer(suite.ctx, re, suite.backendEndpoints)
+	s, cleanup := mcs.StartSingleTSOTestServer(suite.ctx, re, suite.backendEndpoints, suite.listenAddr)
 
 	defer cleanup()
 	testutil.Eventually(re, func() bool {
@@ -150,7 +153,7 @@ func checkTSOPath(re *require.Assertions, isAPIServiceMode bool) {
 		re.Equal(1, getEtcdTimestampKeyNum(re, client))
 	}
 
-	_, cleanup := mcs.StartSingleTSOTestServer(ctx, re, backendEndpoints)
+	_, cleanup := mcs.StartSingleTSOTestServer(ctx, re, backendEndpoints, tempurl.Alloc())
 	defer cleanup()
 
 	cli := mcs.SetupTSOClient(ctx, re, []string{backendEndpoints})
@@ -183,6 +186,7 @@ type APIServerForwardTestSuite struct {
 	cluster          *tests.TestCluster
 	pdLeader         *tests.TestServer
 	backendEndpoints string
+	listenAddr       string
 	pdClient         pd.Client
 }
 
@@ -203,6 +207,7 @@ func (suite *APIServerForwardTestSuite) SetupSuite() {
 	leaderName := suite.cluster.WaitLeader()
 	suite.pdLeader = suite.cluster.GetServer(leaderName)
 	suite.backendEndpoints = suite.pdLeader.GetAddr()
+	suite.listenAddr = tempurl.Alloc()
 	suite.NoError(suite.pdLeader.BootstrapCluster())
 	suite.addRegions()
 
@@ -230,7 +235,7 @@ func (suite *APIServerForwardTestSuite) TestForwardTSORelated() {
 	// Unable to use the tso-related interface without tso server
 	suite.checkUnavailableTSO()
 	// can use the tso-related interface with tso server
-	s, cleanup := mcs.StartSingleTSOTestServer(suite.ctx, suite.Require(), suite.backendEndpoints)
+	s, cleanup := mcs.StartSingleTSOTestServer(suite.ctx, suite.Require(), suite.backendEndpoints, suite.listenAddr)
 	serverMap := make(map[string]bs.Server)
 	serverMap[s.GetAddr()] = s
 	mcs.WaitForPrimaryServing(suite.Require(), serverMap)
@@ -241,7 +246,7 @@ func (suite *APIServerForwardTestSuite) TestForwardTSORelated() {
 func (suite *APIServerForwardTestSuite) TestForwardTSOWhenPrimaryChanged() {
 	serverMap := make(map[string]bs.Server)
 	for i := 0; i < 3; i++ {
-		s, cleanup := mcs.StartSingleTSOTestServer(suite.ctx, suite.Require(), suite.backendEndpoints)
+		s, cleanup := mcs.StartSingleTSOTestServer(suite.ctx, suite.Require(), suite.backendEndpoints, suite.listenAddr)
 		defer cleanup()
 		serverMap[s.GetAddr()] = s
 	}
@@ -319,3 +324,27 @@ func (suite *APIServerForwardTestSuite) checkAvailableTSO() {
 	err = suite.pdClient.SetExternalTimestamp(suite.ctx, ts+1)
 	suite.NoError(err)
 }
+
+func TestAdvertiseAddr(t *testing.T) {
+	re := require.New(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	cluster, err := tests.NewTestAPICluster(ctx, 1)
+	defer cluster.Destroy()
+	re.NoError(err)
+
+	err = cluster.RunInitialServers()
+	re.NoError(err)
+
+	leaderName := cluster.WaitLeader()
+	leader := cluster.GetServer(leaderName)
+
+	u := tempurl.Alloc()
+	s, cleanup := mcs.StartSingleTSOTestServer(ctx, re, leader.GetAddr(), u)
+	defer cleanup()
+
+	tsoServerConf := s.GetConfig()
+	re.Equal(leader.GetAddr(), tsoServerConf.AdvertiseBackendEndpoints)
+	re.Equal(u, tsoServerConf.AdvertiseListenAddr)
+}

tests/tso/client_test.go

@@ -26,6 +26,7 @@ import (
 	pd "github.com/tikv/pd/client"
 	"github.com/tikv/pd/client/testutil"
 	tso "github.com/tikv/pd/pkg/mcs/tso/server"
+	"github.com/tikv/pd/pkg/utils/tempurl"
 	"github.com/tikv/pd/pkg/utils/tsoutil"
 	"github.com/tikv/pd/tests"
 	"github.com/tikv/pd/tests/mcs"
@@ -78,7 +79,7 @@ func (suite *tsoClientTestSuite) SetupSuite() {
 		suite.client, err = pd.NewClientWithContext(suite.ctx, strings.Split(backendEndpoints, ","), pd.SecurityOption{})
 		re.NoError(err)
 	} else {
-		suite.tsoServer, suite.tsoServerCleanup = mcs.StartSingleTSOTestServer(suite.ctx, re, backendEndpoints)
+		suite.tsoServer, suite.tsoServerCleanup = mcs.StartSingleTSOTestServer(suite.ctx, re, backendEndpoints, tempurl.Alloc())
 		suite.client = mcs.SetupTSOClient(suite.ctx, re, strings.Split(backendEndpoints, ","))
 	}
 }

tests/tso/consistency_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/stretchr/testify/suite"
 	tso "github.com/tikv/pd/pkg/mcs/tso/server"
 	tsopkg "github.com/tikv/pd/pkg/tso"
+	"github.com/tikv/pd/pkg/utils/tempurl"
 	pd "github.com/tikv/pd/pkg/utils/testutil"
 	"github.com/tikv/pd/pkg/utils/tsoutil"
 	"github.com/tikv/pd/tests"
@@ -84,7 +85,7 @@ func (suite *tsoConsistencyTestSuite) SetupSuite() {
 	if suite.legacy {
 		suite.pdClient = pd.MustNewGrpcClient(re, backendEndpoints)
 	} else {
-		suite.tsoServer, suite.tsoServerCleanup = mcs.StartSingleTSOTestServer(suite.ctx, re, backendEndpoints)
+		suite.tsoServer, suite.tsoServerCleanup = mcs.StartSingleTSOTestServer(suite.ctx, re, backendEndpoints, tempurl.Alloc())
 		suite.tsoClientConn, suite.tsoClient = tso.MustNewGrpcClient(re, suite.tsoServer.GetAddr())
 	}
 }

tests/tso/server_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/suite"
 	tso "github.com/tikv/pd/pkg/mcs/tso/server"
 	tsopkg "github.com/tikv/pd/pkg/tso"
+	"github.com/tikv/pd/pkg/utils/tempurl"
 	pd "github.com/tikv/pd/pkg/utils/testutil"
 	"github.com/tikv/pd/tests"
 	"github.com/tikv/pd/tests/mcs"
@@ -82,7 +83,7 @@ func (suite *tsoServerTestSuite) SetupSuite() {
 	if suite.legacy {
 		suite.pdClient = pd.MustNewGrpcClient(re, backendEndpoints)
 	} else {
-		suite.tsoServer, suite.tsoServerCleanup = mcs.StartSingleTSOTestServer(suite.ctx, re, backendEndpoints)
+		suite.tsoServer, suite.tsoServerCleanup = mcs.StartSingleTSOTestServer(suite.ctx, re, backendEndpoints, tempurl.Alloc())
 		suite.tsoClientConn, suite.tsoClient = tso.MustNewGrpcClient(re, suite.tsoServer.GetAddr())
 	}
 }