tso: Fix Local tso alloactor become alloactor leader without pd leader knowing its dc-location

go.mod

@@ -28,7 +28,7 @@ require (
 	github.com/pingcap/errcode v0.3.0
 	github.com/pingcap/errors v0.11.5-0.20200917111840-a15ef68f753d
 	github.com/pingcap/failpoint v0.0.0-20200702092429-9f69995143ce
-	github.com/pingcap/kvproto v0.0.0-20201027123903-c4791e779a8c
+	github.com/pingcap/kvproto v0.0.0-20201104042953-62eb316d5182
 	github.com/pingcap/log v0.0.0-20200511115504-543df19646ad
 	github.com/pingcap/sysutil v0.0.0-20201021075216-f93ced2829e2
 	github.com/pingcap/tiup v1.2.2

go.sum

@@ -636,8 +636,8 @@ github.com/pingcap/kvproto v0.0.0-20200411081810-b85805c9476c/go.mod h1:IOdRDPLy
 github.com/pingcap/kvproto v0.0.0-20200417092353-efbe03bcffbd/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20200420075417-e0c6e8842f22/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20200810113304-6157337686b1/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
-github.com/pingcap/kvproto v0.0.0-20201027123903-c4791e779a8c h1:O3tnG2+7F0yvTli099KS6yMAv6YtHx5EFW9tn46MpY8=
-github.com/pingcap/kvproto v0.0.0-20201027123903-c4791e779a8c/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
+github.com/pingcap/kvproto v0.0.0-20201104042953-62eb316d5182 h1:werchMhPPSWn+MCbBGQdh9VpW1CJEZQ3+lrQO8zSb+4=
+github.com/pingcap/kvproto v0.0.0-20201104042953-62eb316d5182/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20200117041106-d28c14d3b1cd/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20200511115504-543df19646ad h1:SveG82rmu/GFxYanffxsSF503SiQV+2JLnWEiGiF+Tc=

server/grpc_service.go

@@ -33,6 +33,7 @@ import (
 	"github.com/tikv/pd/server/cluster"
 	"github.com/tikv/pd/server/config"
 	"github.com/tikv/pd/server/core"
+	"github.com/tikv/pd/server/tso"
 	"github.com/tikv/pd/server/versioninfo"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
@@ -994,6 +995,32 @@ func (s *Server) SyncMaxTS(ctx context.Context, request *pdpb.SyncMaxTSRequest)
 	}, nil
 }
 
+// GetDCLocations will return the dcLocations which hold by the Global TSO Allocator.
+// If the receiving PD Member is not PD Leader, GetDCLocations will return error.
+func (s *Server) GetDCLocations(ctx context.Context, request *pdpb.GetDCLocationsRequest) (*pdpb.GetDCLocationsResponse, error) {
+	if err := s.validateInternalRequest(request.GetHeader()); err != nil {
+		return nil, err
+	}
+	if !s.member.IsLeader() {
+		return nil, fmt.Errorf("receiving pd member[%v] is not pd leader", s.member.ID())
+	}
+	allocator, err := s.GetTSOAllocatorManager().GetAllocator(config.GlobalDCLocation)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get tso allocator[%v]", config.GlobalDCLocation)
+	}
+	globalAllocator, ok := allocator.(*tso.GlobalTSOAllocator)
+	if !ok {
+		return nil, fmt.Errorf("tso allocator[%v] is not global tso allocator", config.GlobalDCLocation)
+	}
+	if !globalAllocator.IsInitialize() {
+		return nil, fmt.Errorf("global tso alloactor is not initialized")
+	}
+	return &pdpb.GetDCLocationsResponse{
+		Header:      s.header(),
+		DcLocations: globalAllocator.GetDcLocations(),
+	}, nil
+}
+
 // validateInternalRequest checks if server is closed, which is used to validate
 // the gRPC communication between PD servers internally.
 func (s *Server) validateInternalRequest(header *pdpb.RequestHeader) error {

server/tso/allocator_manager.go

@@ -34,6 +34,7 @@ import (
 	"github.com/tikv/pd/server/member"
 	"go.etcd.io/etcd/clientv3"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
 )
 
 const (
@@ -85,6 +86,11 @@ type AllocatorManager struct {
 	updatePhysicalInterval time.Duration
 	maxResetTSGap          func() time.Duration
 	securityConfig         *grpcutil.TLSConfig
+	// for gRPC use
+	localAllocatorConn struct {
+		sync.RWMutex
+		clientConns map[string]*grpc.ClientConn
+	}
 }
 
 // NewAllocatorManager creates a new TSO Allocator Manager.
@@ -105,6 +111,7 @@ func NewAllocatorManager(
 		securityConfig:         sc,
 	}
 	allocatorManager.mu.allocatorGroups = make(map[string]*allocatorGroup)
+	allocatorManager.localAllocatorConn.clientConns = make(map[string]*grpc.ClientConn)
 	return allocatorManager
 }
 
@@ -221,6 +228,22 @@ func (am *AllocatorManager) allocatorLeaderLoop(ctx context.Context, allocator *
 		default:
 		}
 
+		ok, err := am.isLeaderAwareOfDCLocation(ctx, allocator.dcLocation)
+		if err != nil {
+			log.Error("get dc-locations from pd leader failed",
+				zap.String("dc-location", allocator.dcLocation),
+				errs.ZapError(err))
+			time.Sleep(200 * time.Millisecond)
+			continue
+		}
+		if !ok {
+			log.Error("pd leader is not aware of dc-location during allocatorLeaderLoop, wait next round",
+				zap.String("dc-location", allocator.dcLocation),
+				zap.String("wait-duration", checkStep.String()))
+			time.Sleep(checkStep)
+			continue
+		}
+
 		allocatorLeader, rev, checkAgain := allocator.CheckAllocatorLeader()
 		if checkAgain {
 			continue
@@ -656,3 +679,84 @@ func (am *AllocatorManager) GetLocalAllocatorLeaders() (map[string]*pdpb.Member,
 	}
 	return localAllocatorLeaderMember, nil
 }
+
+func (am *AllocatorManager) getOrCreateGRPCConn(ctx context.Context, addr string) (*grpc.ClientConn, error) {
+	conn, ok := am.getGRPCConn(addr)
+	if ok {
+		return conn, nil
+	}
+	tlsCfg, err := am.securityConfig.ToTLSConfig()
+	if err != nil {
+		return nil, err
+	}
+	ctxWithTimeout, cancel := context.WithTimeout(ctx, dialTimeout)
+	defer cancel()
+	cc, err := grpcutil.GetClientConn(ctxWithTimeout, addr, tlsCfg)
+	if err != nil {
+		return nil, err
+	}
+	am.setGRPCConn(cc, addr)
+	conn, _ = am.getGRPCConn(addr)
+	return conn, nil
+}
+
+func (am *AllocatorManager) isLeaderAwareOfDCLocation(ctx context.Context, dcLocation string) (bool, error) {
+	dcLocations, err := am.getLeaderDCLocations(ctx)
+	if err != nil {
+		return false, err
+	}
+	for _, dc := range dcLocations {
+		if dcLocation == dc {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func (am *AllocatorManager) getLeaderDCLocations(ctx context.Context) ([]string, error) {
+	dcLocations := make([]string, 0)
+	if am.member.IsLeader() {
+		for dcLocation := range am.GetClusterDCLocations() {
+			dcLocations = append(dcLocations, dcLocation)
+		}
+		return dcLocations, nil
+	}
+
+	leaderAddrs := am.member.GetLeader().GetClientUrls()
+	if leaderAddrs == nil || len(leaderAddrs) < 1 {
+		return nil, fmt.Errorf("failed to get leader client url")
+	}
+	conn, err := am.getOrCreateGRPCConn(ctx, leaderAddrs[0])
+	if err != nil {
+		return nil, err
+	}
+	getCtx, cancel := context.WithTimeout(ctx, rpcTimeout)
+	defer cancel()
+	resp, err := pdpb.NewPDClient(conn).GetDCLocations(getCtx, &pdpb.GetDCLocationsRequest{
+		Header: &pdpb.RequestHeader{
+			SenderId: am.member.GetLeader().GetMemberId(),
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return resp.GetDcLocations(), nil
+}
+
+func (am *AllocatorManager) getGRPCConn(addr string) (*grpc.ClientConn, bool) {
+	am.localAllocatorConn.RLock()
+	defer am.localAllocatorConn.RUnlock()
+	conn, ok := am.localAllocatorConn.clientConns[addr]
+	return conn, ok
+}
+
+func (am *AllocatorManager) setGRPCConn(newConn *grpc.ClientConn, addr string) {
+	am.localAllocatorConn.Lock()
+	defer am.localAllocatorConn.Unlock()
+	if _, ok := am.localAllocatorConn.clientConns[addr]; ok {
+		newConn.Close()
+		log.Debug("use old connection", zap.String("target", newConn.Target()), zap.String("state", newConn.GetState().String()))
+		return
+	}
+	am.localAllocatorConn.clientConns[addr] = newConn
+}

server/tso/global_allocator.go

@@ -22,7 +22,6 @@ import (
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/errs"
-	"github.com/tikv/pd/pkg/grpcutil"
 	"github.com/tikv/pd/pkg/slice"
 	"github.com/tikv/pd/pkg/tsoutil"
 	"github.com/tikv/pd/pkg/typeutil"
@@ -59,11 +58,6 @@ type GlobalTSOAllocator struct {
 	timestampOracle *timestampOracle
 	// for global TSO synchronization
 	allocatorManager *AllocatorManager
-	// for gRPC use
-	localAllocatorConn struct {
-		sync.RWMutex
-		clientConns map[string]*grpc.ClientConn
-	}
 }
 
 // NewGlobalTSOAllocator creates a new global TSO allocator.
@@ -86,7 +80,6 @@ func NewGlobalTSOAllocator(
 		},
 		allocatorManager: am,
 	}
-	gta.localAllocatorConn.clientConns = make(map[string]*grpc.ClientConn)
 	return gta
 }
 
@@ -178,7 +171,7 @@ func (gta *GlobalTSOAllocator) syncMaxTS(ctx context.Context, dcLocationMap map[
 		var errList []error
 		wg := sync.WaitGroup{}
 		for _, leaderURL := range leaderURLs {
-			leaderConn, err := gta.getOrCreateGRPCConn(ctx, leaderURL)
+			leaderConn, err := gta.allocatorManager.getOrCreateGRPCConn(ctx, leaderURL)
 			if err != nil {
 				return err
 			}
@@ -255,34 +248,6 @@ func (gta *GlobalTSOAllocator) checkSyncedDCs(dcLocationMap map[string][]uint64,
 	return len(unsyncedDCs) == 0
 }
 
-func (gta *GlobalTSOAllocator) getOrCreateGRPCConn(ctx context.Context, addr string) (*grpc.ClientConn, error) {
-	gta.localAllocatorConn.RLock()
-	conn, ok := gta.localAllocatorConn.clientConns[addr]
-	gta.localAllocatorConn.RUnlock()
-	if ok {
-		return conn, nil
-	}
-	tlsCfg, err := gta.allocatorManager.securityConfig.ToTLSConfig()
-	if err != nil {
-		return nil, err
-	}
-	ctxWithTimeout, cancel := context.WithTimeout(ctx, dialTimeout)
-	defer cancel()
-	cc, err := grpcutil.GetClientConn(ctxWithTimeout, addr, tlsCfg)
-	if err != nil {
-		return nil, err
-	}
-	gta.localAllocatorConn.Lock()
-	defer gta.localAllocatorConn.Unlock()
-	if old, ok := gta.localAllocatorConn.clientConns[addr]; ok {
-		cc.Close()
-		log.Debug("use old connection", zap.String("target", cc.Target()), zap.String("state", cc.GetState().String()))
-		return old, nil
-	}
-	gta.localAllocatorConn.clientConns[addr] = cc
-	return cc, nil
-}
-
 func (gta *GlobalTSOAllocator) getCurrentTSO() (pdpb.Timestamp, error) {
 	currentPhysical, currentLogical := gta.timestampOracle.getTSO()
 	if currentPhysical == typeutil.ZeroTime {
@@ -295,3 +260,13 @@ func (gta *GlobalTSOAllocator) getCurrentTSO() (pdpb.Timestamp, error) {
 func (gta *GlobalTSOAllocator) Reset() {
 	gta.timestampOracle.ResetTimestamp()
 }
+
+// GetDcLocations return all the dcLocations the GlobalTSOAllocator will check
+func (gta *GlobalTSOAllocator) GetDcLocations() []string {
+	dcLocationsMap := gta.allocatorManager.GetClusterDCLocations()
+	dcLocations := make([]string, 0, len(dcLocationsMap))
+	for dc := range dcLocationsMap {
+		dcLocations = append(dcLocations, dc)
+	}
+	return dcLocations
+}

tests/server/tso/local_tso_test.go

@@ -15,6 +15,7 @@ package tso_test
 
 import (
 	"context"
+	"sort"
 	"sync"
 	"time"
 
@@ -75,6 +76,13 @@ func (s *testLocalTSOSuite) TestLocalTSO(c *C) {
 
 	cluster.WaitLeader()
 	leaderServer := cluster.GetServer(cluster.GetLeader())
+	s.testGetDcLocations(c, dcClientMap[leaderServer.GetConfig().LocalTSO.DCLocation],
+		&pdpb.GetDCLocationsRequest{
+			Header: &pdpb.RequestHeader{
+				SenderId: leaderServer.GetServer().GetMember().ID(),
+			},
+		},
+		[]string{"dc-1", "dc-2", "dc-3"})
 
 	var wg sync.WaitGroup
 	for i := 0; i < 10; i++ {
@@ -125,3 +133,13 @@ func (s *testLocalTSOSuite) testGetLocalTimestamp(c *C, pdCli pdpb.PDClient, req
 	c.Assert(res.GetLogical(), Greater, int64(0))
 	return res
 }
+
+func (s *testLocalTSOSuite) testGetDcLocations(c *C, pdCli pdpb.PDClient, req *pdpb.GetDCLocationsRequest, dcLocations []string) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	resp, err := pdCli.GetDCLocations(ctx, req)
+	c.Assert(err, IsNil)
+	sort.Strings(dcLocations)
+	sort.Strings(resp.DcLocations)
+	c.Assert(resp.DcLocations, DeepEquals, dcLocations)
+}