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

server/grpc_service.go

@@ -61,6 +61,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.
@@ -1706,7 +1707,10 @@ func checkStream(streamCtx context.Context, cancel context.CancelFunc, done chan
 }
 
 // StoreGlobalConfig store global config into etcd by transaction
-func (s *GrpcServer) StoreGlobalConfig(ctx context.Context, request *pdpb.StoreGlobalConfigRequest) (*pdpb.StoreGlobalConfigResponse, error) {
+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()
@@ -1726,6 +1730,9 @@ func (s *GrpcServer) StoreGlobalConfig(ctx context.Context, request *pdpb.StoreG
 
 // 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 {
@@ -1743,9 +1750,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(request *pdpb.WatchGlobalConfigRequest, server pdpb.PD_WatchGlobalConfigServer) error {
+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

@@ -1414,3 +1414,9 @@ func (s *Server) SaveTTLConfig(data map[string]interface{}, ttl time.Duration) e
 func (s *Server) SplitAndScatterRegions(context context.Context, r *pdpb.SplitAndScatterRegionsRequest) (*pdpb.SplitAndScatterRegionsResponse, error) {
 	return nil, errors.New("no implemented")
 }
+
+// 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

@@ -17,6 +17,7 @@ package global_config_test
 import (
 	"context"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -47,6 +48,7 @@ type GlobalConfigTestSuite struct {
 	server  *server.GrpcServer
 	client  pd.Client
 	cleanup server.CleanupFunc
+	mu      sync.Mutex
 }
 
 type TestReceiver struct {
@@ -205,3 +207,27 @@ func (s *GlobalConfigTestSuite) TestClientWatchTimeout(c *C) {
 	_, opened := <-wc
 	c.Assert(opened, Equals, false)
 }
+
+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)
+}