global config: fix etcd client not found (#6866)

server/grpc_service.go

@@ -64,6 +64,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.
@@ -1898,6 +1899,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()
@@ -1917,6 +1921,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 {
@@ -1934,9 +1941,12 @@ func (s *GrpcServer) LoadGlobalConfig(ctx context.Context, request *pdpb.LoadGlo
 }
 
 // WatchGlobalConfig if the connection of WatchGlobalConfig is end
-// or stoped by whatever reason
+// or stopped 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

@@ -1531,3 +1531,9 @@ func (s *Server) IsTTLConfigExist(key string) bool {
 	}
 	return false
 }
+
+// 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"
 
@@ -47,6 +48,7 @@ type GlobalConfigTestSuite struct {
 	server  *server.GrpcServer
 	client  *grpc.ClientConn
 	cleanup server.CleanupFunc
+	mu      sync.Mutex
 }
 
 type TestReceiver struct {
@@ -235,3 +237,27 @@ func (s *GlobalConfigTestSuite) TestClientWatch(c *C) {
 		}
 	}
 }
+
+func (s *GlobalConfigTestSuite) TestEtcdNotStart(c *C) {
+	cli := s.server.GetClient()
+	defer func() {
+		s.mu.Lock()
+		s.server.SetClient(cli)
+		s.mu.Unlock()
+	}()
+	s.mu.Lock()
+	s.server.SetClient(nil)
+	s.mu.Unlock()
+	err := s.server.WatchGlobalConfig(&pdpb.WatchGlobalConfigRequest{}, nil)
+	c.Assert(err, NotNil)
+
+	_, err = s.server.StoreGlobalConfig(s.server.Context(), &pdpb.StoreGlobalConfigRequest{
+		Changes: []*pdpb.GlobalConfigItem{{Name: "0", Value: "0"}},
+	})
+	c.Assert(err, NotNil)
+
+	_, err = s.server.LoadGlobalConfig(s.server.Context(), &pdpb.LoadGlobalConfigRequest{
+		Names: []string{"test_etcd"},
+	})
+	c.Assert(err, NotNil)
+}