Merge branch 'release-6.5' into cherry-pick-6919-to-release-6.5

server/election/lease.go

@@ -152,8 +152,11 @@ func (l *lease) keepAliveWorker(ctx context.Context, interval time.Duration) <-c
 					expire := start.Add(time.Duration(res.TTL) * time.Second)
 					select {
 					case ch <- expire:
-					case <-ctx1.Done():
+					// Here we don't use `ctx1.Done()` because we want to make sure if the keep alive success, we can update the expire time.
+					case <-ctx.Done():
 					}
+				} else {
+					log.Error("keep alive response ttl is zero", zap.String("purpose", l.Purpose))
 				}
 			}()
 

server/election/lease_test.go

@@ -101,3 +101,34 @@ func TestLease(t *testing.T) {
 	time.Sleep((defaultLeaseTimeout + 1) * time.Second)
 	re.True(lease1.IsExpired())
 }
+
+func TestLeaseKeepAlive(t *testing.T) {
+	re := require.New(t)
+	cfg := etcdutil.NewTestSingleConfig(t)
+	etcd, err := embed.StartEtcd(cfg)
+	defer func() {
+		etcd.Close()
+	}()
+	re.NoError(err)
+
+	ep := cfg.LCUrls[0].String()
+	client, err := clientv3.New(clientv3.Config{
+		Endpoints: []string{ep},
+	})
+	re.NoError(err)
+
+	<-etcd.Server.ReadyNotify()
+
+	// Create the lease.
+	lease := &lease{
+		Purpose: "test_lease",
+		client:  client,
+		lease:   clientv3.NewLease(client),
+	}
+
+	re.NoError(lease.Grant(defaultLeaseTimeout))
+	ch := lease.keepAliveWorker(context.Background(), 2*time.Second)
+	time.Sleep(2 * time.Second)
+	<-ch
+	re.NoError(lease.Close())
+}

server/grpc_service.go

@@ -63,6 +63,7 @@ var (
 	ErrNotLeader            = status.Errorf(codes.Unavailable, "not leader")
 	ErrNotStarted           = status.Errorf(codes.Unavailable, "server not started")
 	ErrSendHeartbeatTimeout = status.Errorf(codes.DeadlineExceeded, "send heartbeat timeout")
+	ErrEtcdNotStarted       = status.Errorf(codes.Unavailable, "server is started, but etcd not started")
 )
 
 // GrpcServer wraps Server to provide grpc service.
@@ -1896,6 +1897,9 @@ func checkStream(streamCtx context.Context, cancel context.CancelFunc, done chan
 
 // StoreGlobalConfig store global config into etcd by transaction
 func (s *GrpcServer) StoreGlobalConfig(_ context.Context, request *pdpb.StoreGlobalConfigRequest) (*pdpb.StoreGlobalConfigResponse, error) {
+	if s.client == nil {
+		return nil, ErrEtcdNotStarted
+	}
 	ops := make([]clientv3.Op, len(request.Changes))
 	for i, item := range request.Changes {
 		name := globalConfigPath + item.GetName()
@@ -1915,6 +1919,9 @@ func (s *GrpcServer) StoreGlobalConfig(_ context.Context, request *pdpb.StoreGlo
 
 // LoadGlobalConfig load global config from etcd
 func (s *GrpcServer) LoadGlobalConfig(ctx context.Context, request *pdpb.LoadGlobalConfigRequest) (*pdpb.LoadGlobalConfigResponse, error) {
+	if s.client == nil {
+		return nil, ErrEtcdNotStarted
+	}
 	names := request.Names
 	res := make([]*pdpb.GlobalConfigItem, len(names))
 	for i, name := range names {
@@ -1935,6 +1942,9 @@ func (s *GrpcServer) LoadGlobalConfig(ctx context.Context, request *pdpb.LoadGlo
 // or stoped by whatever reason
 // just reconnect to it.
 func (s *GrpcServer) WatchGlobalConfig(_ *pdpb.WatchGlobalConfigRequest, server pdpb.PD_WatchGlobalConfigServer) error {
+	if s.client == nil {
+		return ErrEtcdNotStarted
+	}
 	ctx, cancel := context.WithCancel(s.Context())
 	defer cancel()
 	err := s.sendAllGlobalConfig(ctx, server)

server/server.go

@@ -1770,3 +1770,9 @@ func (s *Server) SetExternalTS(externalTS uint64) error {
 	s.GetRaftCluster().SetExternalTS(externalTS)
 	return nil
 }
+
+// SetClient sets the etcd client.
+// Notes: it is only used for test.
+func (s *Server) SetClient(client *clientv3.Client) {
+	s.client = client
+}

tests/server/global_config/global_config_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -57,6 +58,7 @@ type globalConfigTestSuite struct {
 	server  *server.GrpcServer
 	client  *grpc.ClientConn
 	cleanup server.CleanupFunc
+	mu      sync.Mutex
 }
 
 func TestGlobalConfigTestSuite(t *testing.T) {
@@ -237,3 +239,27 @@ func (suite *globalConfigTestSuite) TestClientWatch() {
 		}
 	}
 }
+
+func (suite *globalConfigTestSuite) TestEtcdNotStart() {
+	cli := suite.server.GetClient()
+	defer func() {
+		suite.mu.Lock()
+		suite.server.SetClient(cli)
+		suite.mu.Unlock()
+	}()
+	suite.mu.Lock()
+	suite.server.SetClient(nil)
+	suite.mu.Unlock()
+	err := suite.server.WatchGlobalConfig(&pdpb.WatchGlobalConfigRequest{}, testReceiver{re: suite.Require()})
+	suite.Error(err)
+
+	_, err = suite.server.StoreGlobalConfig(suite.server.Context(), &pdpb.StoreGlobalConfigRequest{
+		Changes: []*pdpb.GlobalConfigItem{{Name: "0", Value: "0"}},
+	})
+	suite.Error(err)
+
+	_, err = suite.server.LoadGlobalConfig(suite.server.Context(), &pdpb.LoadGlobalConfigRequest{
+		Names: []string{"test_etcd"},
+	})
+	suite.Error(err)
+}