enable server/*.go execpt server/api/*

Signed-off-by: okJiang <[email protected]>

.golangci.yml

@@ -207,6 +207,6 @@ issues:
     - path: (pd-analysis|pd-api-bench|pd-backup|pd-ctl|pd-heartbeat-bench|pd-recover|pd-simulator|pd-tso-bench|pd-ut|regions-dump|stores-dump)
       linters:
         - errcheck
-    - path: (server/api/.*\.go|server/.*\.go|pkg/schedule/schedulers/.*\.go)
+    - path: (server/api/.*\.go|pkg/schedule/schedulers/.*\.go)
       linters:
         - errcheck

cmd/pd-server/main.go

@@ -196,8 +196,12 @@ func start(cmd *cobra.Command, args []string, services ...string) {
 	schedulers.Register()
 	cfg := config.NewConfig()
 	flagSet := cmd.Flags()
-	flagSet.Parse(args)
-	err := cfg.Parse(flagSet)
+	err := flagSet.Parse(args)
+	if err != nil {
+		cmd.Println(err)
+		return
+	}
+	err = cfg.Parse(flagSet)
 	defer logutil.LogPanic()
 
 	if err != nil {
@@ -231,7 +235,7 @@ func start(cmd *cobra.Command, args []string, services ...string) {
 		log.Fatal("initialize logger error", errs.ZapError(err))
 	}
 	// Flushing any buffered log entries
-	defer log.Sync()
+	defer log.Sync() //nolint:errcheck
 	memory.InitMemoryHook()
 	if len(services) != 0 {
 		versioninfo.Log(server.APIServiceMode)
@@ -295,6 +299,6 @@ func start(cmd *cobra.Command, args []string, services ...string) {
 }
 
 func exit(code int) {
-	log.Sync()
+	log.Sync() // nolint:errcheck
 	os.Exit(code)
 }

pkg/mcs/metastorage/server/grpc_service.go

@@ -47,7 +47,7 @@ type dummyRestService struct{}
 
 func (dummyRestService) ServeHTTP(w http.ResponseWriter, _ *http.Request) {
 	w.WriteHeader(http.StatusNotImplemented)
-	w.Write([]byte("not implemented"))
+	w.Write([]byte("not implemented")) // nolint:errcheck
 }
 
 // Service is the gRPC service for meta storage.
@@ -71,9 +71,9 @@ func (s *Service) RegisterGRPCService(g *grpc.Server) {
 }
 
 // RegisterRESTHandler registers the service to REST server.
-func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler) {
+func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler) error {
 	handler, group := SetUpRestHandler(s)
-	apiutil.RegisterUserDefinedHandlers(userDefineHandlers, &group, handler)
+	return apiutil.RegisterUserDefinedHandlers(userDefineHandlers, &group, handler)
 }
 
 func (s *Service) checkServing() error {

pkg/mcs/registry/registry.go

@@ -37,7 +37,7 @@ type ServiceBuilder func(bs.Server) RegistrableService
 // RegistrableService is the interface that should wraps the RegisterService method.
 type RegistrableService interface {
 	RegisterGRPCService(g *grpc.Server)
-	RegisterRESTHandler(userDefineHandlers map[string]http.Handler)
+	RegisterRESTHandler(userDefineHandlers map[string]http.Handler) error
 }
 
 // ServiceRegistry is a map that stores all registered grpc services.
@@ -82,14 +82,20 @@ func (r *ServiceRegistry) InstallAllRESTHandler(srv bs.Server, h map[string]http
 	for name, builder := range r.builders {
 		serviceName := createServiceName(prefix, name)
 		if l, ok := r.services[serviceName]; ok {
-			l.RegisterRESTHandler(h)
-			log.Info("restful API service already registered", zap.String("prefix", prefix), zap.String("service-name", name))
+			if err := l.RegisterRESTHandler(h); err != nil {
+				log.Error("register REST handler failed", zap.String("prefix", prefix), zap.String("service-name", name), zap.Error(err))
+			} else {
+				log.Info("restful API service already registered", zap.String("prefix", prefix), zap.String("service-name", name))
+			}
 			continue
 		}
 		l := builder(srv)
 		r.services[serviceName] = l
-		l.RegisterRESTHandler(h)
-		log.Info("restful API service registered successfully", zap.String("prefix", prefix), zap.String("service-name", name))
+		if err := l.RegisterRESTHandler(h); err != nil {
+			log.Error("register REST handler failed", zap.String("prefix", prefix), zap.String("service-name", name), zap.Error(err))
+		} else {
+			log.Info("restful API service registered successfully", zap.String("prefix", prefix), zap.String("service-name", name))
+		}
 	}
 }
 

pkg/mcs/resourcemanager/server/config.go

@@ -237,7 +237,9 @@ func (c *Config) Adjust(meta *toml.MetaData) error {
 	}
 
 	c.adjustLog(configMetaData.Child("log"))
-	c.Security.Encryption.Adjust()
+	if err := c.Security.Encryption.Adjust(); err != nil {
+		return err
+	}
 
 	c.Controller.Adjust(configMetaData.Child("controller"))
 	configutil.AdjustInt64(&c.LeaderLease, utils.DefaultLeaderLease)

pkg/mcs/resourcemanager/server/grpc_service.go

@@ -49,7 +49,7 @@ type dummyRestService struct{}
 
 func (dummyRestService) ServeHTTP(w http.ResponseWriter, _ *http.Request) {
 	w.WriteHeader(http.StatusNotImplemented)
-	w.Write([]byte("not implemented"))
+	w.Write([]byte("not implemented")) // nolint:errcheck
 }
 
 // Service is the gRPC service for resource manager.
@@ -76,9 +76,9 @@ func (s *Service) RegisterGRPCService(g *grpc.Server) {
 }
 
 // RegisterRESTHandler registers the service to REST server.
-func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler) {
+func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler) error {
 	handler, group := SetUpRestHandler(s)
-	apiutil.RegisterUserDefinedHandlers(userDefineHandlers, &group, handler)
+	return apiutil.RegisterUserDefinedHandlers(userDefineHandlers, &group, handler)
 }
 
 // GetManager returns the resource manager.
@@ -228,6 +228,8 @@ func (s *Service) AcquireTokenBuckets(stream rmpb.ResourceManager_AcquireTokenBu
 			log.Debug("finish token request from", zap.String("resource-group", resourceGroupName))
 			resps.Responses = append(resps.Responses, resp)
 		}
-		stream.Send(resps)
+		if err := stream.Send(resps); err != nil {
+			return errors.WithStack(err)
+		}
 	}
 }

pkg/mcs/resourcemanager/server/manager.go

@@ -349,7 +349,9 @@ func (m *Manager) persistResourceGroupRunningState() {
 		m.RLock()
 		group, ok := m.groups[keys[idx]]
 		if ok {
-			group.persistStates(m.storage)
+			if err := group.persistStates(m.storage); err != nil {
+				log.Error("persist resource group state failed", zap.Error(err))
+			}
 		}
 		m.RUnlock()
 	}

pkg/mcs/resourcemanager/server/server.go

@@ -180,7 +180,9 @@ func (s *Server) campaignLeader() {
 
 	log.Info("triggering the primary callback functions")
 	for _, cb := range s.primaryCallbacks {
-		cb(ctx)
+		if err := cb(ctx); err != nil {
+			log.Error("failed to trigger the primary callback function", errs.ZapError(err))
+		}
 	}
 
 	s.participant.EnableLeader()
@@ -213,7 +215,9 @@ func (s *Server) Close() {
 	}
 
 	log.Info("closing resource manager server ...")
-	s.serviceRegister.Deregister()
+	if err := s.serviceRegister.Deregister(); err != nil {
+		log.Error("failed to deregister the service", errs.ZapError(err))
+	}
 	utils.StopHTTPServer(s)
 	utils.StopGRPCServer(s)
 	s.GetListener().Close()
@@ -362,10 +366,14 @@ func CreateServer(ctx context.Context, cfg *Config) *Server {
 
 // CreateServerWrapper encapsulates the configuration/log/metrics initialization and create the server
 func CreateServerWrapper(cmd *cobra.Command, args []string) {
-	cmd.Flags().Parse(args)
+	err := cmd.Flags().Parse(args)
+	if err != nil {
+		cmd.Println(err)
+		return
+	}
 	cfg := NewConfig()
 	flagSet := cmd.Flags()
-	err := cfg.Parse(flagSet)
+	err = cfg.Parse(flagSet)
 	defer logutil.LogPanic()
 
 	if err != nil {
@@ -389,7 +397,7 @@ func CreateServerWrapper(cmd *cobra.Command, args []string) {
 		log.Fatal("initialize logger error", errs.ZapError(err))
 	}
 	// Flushing any buffered log entries
-	defer log.Sync()
+	defer log.Sync() // nolint:errcheck
 
 	versioninfo.Log(serviceName)
 	log.Info("resource manager config", zap.Reflect("config", cfg))

pkg/mcs/resourcemanager/server/testutil.go

@@ -32,7 +32,7 @@ func NewTestServer(ctx context.Context, re *require.Assertions, cfg *Config) (*S
 	re.NoError(err)
 	log.ReplaceGlobals(cfg.Logger, cfg.LogProps)
 	// Flushing any buffered log entries
-	defer log.Sync()
+	defer log.Sync() // nolint:errcheck
 
 	s := CreateServer(ctx, cfg)
 	if err = s.Run(); err != nil {

pkg/mcs/scheduling/server/config/config.go

@@ -146,7 +146,9 @@ func (c *Config) adjust(meta *toml.MetaData) error {
 	}
 
 	c.adjustLog(configMetaData.Child("log"))
-	c.Security.Encryption.Adjust()
+	if err := c.Security.Encryption.Adjust(); err != nil {
+		return err
+	}
 
 	configutil.AdjustInt64(&c.LeaderLease, utils.DefaultLeaderLease)
 

pkg/mcs/scheduling/server/grpc_service.go

@@ -53,7 +53,7 @@ type dummyRestService struct{}
 
 func (dummyRestService) ServeHTTP(w http.ResponseWriter, _ *http.Request) {
 	w.WriteHeader(http.StatusNotImplemented)
-	w.Write([]byte("not implemented"))
+	w.Write([]byte("not implemented")) // nolint:errcheck
 }
 
 // ConfigProvider is used to get scheduling config from the given
@@ -336,9 +336,9 @@ func (s *Service) RegisterGRPCService(g *grpc.Server) {
 }
 
 // RegisterRESTHandler registers the service to REST server.
-func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler) {
+func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler) error {
 	handler, group := SetUpRestHandler(s)
-	apiutil.RegisterUserDefinedHandlers(userDefineHandlers, &group, handler)
+	return apiutil.RegisterUserDefinedHandlers(userDefineHandlers, &group, handler)
 }
 
 func (s *Service) errorHeader(err *schedulingpb.Error) *schedulingpb.ResponseHeader {

pkg/mcs/scheduling/server/server.go

@@ -316,7 +316,9 @@ func (s *Server) Close() {
 	}
 
 	log.Info("closing scheduling server ...")
-	s.serviceRegister.Deregister()
+	if err := s.serviceRegister.Deregister(); err != nil {
+		log.Error("failed to deregister the service", errs.ZapError(err))
+	}
 	utils.StopHTTPServer(s)
 	utils.StopGRPCServer(s)
 	s.GetListener().Close()
@@ -563,10 +565,14 @@ func CreateServer(ctx context.Context, cfg *config.Config) *Server {
 // CreateServerWrapper encapsulates the configuration/log/metrics initialization and create the server
 func CreateServerWrapper(cmd *cobra.Command, args []string) {
 	schedulers.Register()
-	cmd.Flags().Parse(args)
+	err := cmd.Flags().Parse(args)
+	if err != nil {
+		cmd.Println(err)
+		return
+	}
 	cfg := config.NewConfig()
 	flagSet := cmd.Flags()
-	err := cfg.Parse(flagSet)
+	err = cfg.Parse(flagSet)
 	defer logutil.LogPanic()
 
 	if err != nil {
@@ -590,7 +596,7 @@ func CreateServerWrapper(cmd *cobra.Command, args []string) {
 		log.Fatal("initialize logger error", errs.ZapError(err))
 	}
 	// Flushing any buffered log entries
-	defer log.Sync()
+	defer log.Sync() // nolint:errcheck
 
 	versioninfo.Log(serviceName)
 	log.Info("scheduling service config", zap.Reflect("config", cfg))

pkg/mcs/scheduling/server/testutil.go

@@ -33,7 +33,7 @@ func NewTestServer(ctx context.Context, re *require.Assertions, cfg *config.Conf
 	re.NoError(err)
 	log.ReplaceGlobals(cfg.Logger, cfg.LogProps)
 	// Flushing any buffered log entries
-	defer log.Sync()
+	defer log.Sync() // nolint:errcheck
 
 	s := CreateServer(ctx, cfg)
 	if err = s.Run(); err != nil {

pkg/mcs/tso/server/config.go

@@ -224,9 +224,7 @@ func (c *Config) Adjust(meta *toml.MetaData) error {
 	}
 
 	c.adjustLog(configMetaData.Child("log"))
-	c.Security.Encryption.Adjust()
-
-	return nil
+	return c.Security.Encryption.Adjust()
 }
 
 func (c *Config) adjustLog(meta *configutil.ConfigMetaData) {

pkg/mcs/tso/server/grpc_service.go

@@ -50,7 +50,7 @@ type dummyRestService struct{}
 
 func (dummyRestService) ServeHTTP(w http.ResponseWriter, _ *http.Request) {
 	w.WriteHeader(http.StatusNotImplemented)
-	w.Write([]byte("not implemented"))
+	w.Write([]byte("not implemented")) // nolint:errcheck
 }
 
 // ConfigProvider is used to get tso config from the given
@@ -79,9 +79,9 @@ func (s *Service) RegisterGRPCService(g *grpc.Server) {
 }
 
 // RegisterRESTHandler registers the service to REST server.
-func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler) {
+func (s *Service) RegisterRESTHandler(userDefineHandlers map[string]http.Handler) error {
 	handler, group := SetUpRestHandler(s)
-	apiutil.RegisterUserDefinedHandlers(userDefineHandlers, &group, handler)
+	return apiutil.RegisterUserDefinedHandlers(userDefineHandlers, &group, handler)
 }
 
 // Tso returns a stream of timestamps

pkg/mcs/tso/server/server.go

@@ -174,7 +174,9 @@ func (s *Server) Close() {
 	log.Info("closing tso server ...")
 	// close tso service loops in the keyspace group manager
 	s.keyspaceGroupManager.Close()
-	s.serviceRegister.Deregister()
+	if err := s.serviceRegister.Deregister(); err != nil {
+		log.Error("failed to deregister the service", errs.ZapError(err))
+	}
 	utils.StopHTTPServer(s)
 	utils.StopGRPCServer(s)
 	s.GetListener().Close()
@@ -435,10 +437,14 @@ func CreateServer(ctx context.Context, cfg *Config) *Server {
 
 // CreateServerWrapper encapsulates the configuration/log/metrics initialization and create the server
 func CreateServerWrapper(cmd *cobra.Command, args []string) {
-	cmd.Flags().Parse(args)
+	err := cmd.Flags().Parse(args)
+	if err != nil {
+		cmd.Println(err)
+		return
+	}
 	cfg := NewConfig()
 	flagSet := cmd.Flags()
-	err := cfg.Parse(flagSet)
+	err = cfg.Parse(flagSet)
 	defer logutil.LogPanic()
 
 	if err != nil {
@@ -462,7 +468,7 @@ func CreateServerWrapper(cmd *cobra.Command, args []string) {
 		log.Fatal("initialize logger error", errs.ZapError(err))
 	}
 	// Flushing any buffered log entries
-	defer log.Sync()
+	defer log.Sync() // nolint:errcheck
 
 	versioninfo.Log(serviceName)
 	log.Info("TSO service config", zap.Reflect("config", cfg))

server/cluster/cluster.go

@@ -2172,7 +2172,9 @@ func (c *RaftCluster) runMinResolvedTSJob() {
 			interval = c.opt.GetMinResolvedTSPersistenceInterval()
 			if interval != 0 {
 				if current, needPersist := c.CheckAndUpdateMinResolvedTS(); needPersist {
-					c.storage.SaveMinResolvedTS(current)
+					if err := c.storage.SaveMinResolvedTS(current); err != nil {
+						log.Error("persist min resolved ts meet error", errs.ZapError(err))
+					}
 				}
 			} else {
 				// If interval in config is zero, it means not to persist resolved ts and check config with this interval
@@ -2249,8 +2251,7 @@ func (c *RaftCluster) SetExternalTS(timestamp uint64) error {
 	c.Lock()
 	defer c.Unlock()
 	c.externalTS = timestamp
-	c.storage.SaveExternalTS(timestamp)
-	return nil
+	return c.storage.SaveExternalTS(timestamp)
 }
 
 // SetStoreLimit sets a store limit for a given type and rate.
@@ -2286,8 +2287,8 @@ func (c *RaftCluster) SetAllStoresLimit(typ storelimit.Type, ratePerMin float64)
 }
 
 // SetAllStoresLimitTTL sets all store limit for a given type and rate with ttl.
-func (c *RaftCluster) SetAllStoresLimitTTL(typ storelimit.Type, ratePerMin float64, ttl time.Duration) {
-	c.opt.SetAllStoresLimitTTL(c.ctx, c.etcdClient, typ, ratePerMin, ttl)
+func (c *RaftCluster) SetAllStoresLimitTTL(typ storelimit.Type, ratePerMin float64, ttl time.Duration) error {
+	return c.opt.SetAllStoresLimitTTL(c.ctx, c.etcdClient, typ, ratePerMin, ttl)
 }
 
 // GetClusterVersion returns the current cluster version.