add service subcommand

Signed-off-by: Ryan Leung <rleungx@gmail.com>

cmd/pd-server/main.go

@@ -28,6 +28,7 @@ import (
 	bs "github.com/tikv/pd/pkg/basicserver"
 	"github.com/tikv/pd/pkg/dashboard"
 	"github.com/tikv/pd/pkg/errs"
+	tso "github.com/tikv/pd/pkg/mcs/tso/server"
 	"github.com/tikv/pd/pkg/swaggerserver"
 	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/metricutil"
@@ -40,8 +41,27 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	serviceMode    = "service"
+	tsoServiceMode = "tso-service"
+)
+
 func main() {
-	ctx, cancel, svr := createServerWrapper(os.Args[1:])
+	var (
+		ctx    context.Context
+		cancel context.CancelFunc
+		svr    bs.Server
+	)
+	if len(os.Args) >= 3 && os.Args[1] == serviceMode {
+		switch os.Args[2] {
+		case tsoServiceMode:
+			ctx, cancel, svr = tso.CreateServerWrapper(os.Args[3:])
+		default:
+			log.Fatal("service are not implemented")
+		}
+	} else {
+		ctx, cancel, svr = createServerWrapper(os.Args[1:])
+	}
 
 	sc := make(chan os.Signal, 1)
 	signal.Notify(sc,
@@ -98,9 +118,9 @@ func createServerWrapper(args []string) (context.Context, context.CancelFunc, bs
 	}
 
 	// New zap logger
-	err = cfg.SetupLogger()
+	err = logutil.SetupLogger(cfg.Log, &cfg.Logger, &cfg.LogProps, cfg.Security.RedactInfoLog)
 	if err == nil {
-		log.ReplaceGlobals(cfg.GetZapLogger(), cfg.GetZapLogProperties())
+		log.ReplaceGlobals(cfg.Logger, cfg.LogProps)
 	} else {
 		log.Fatal("initialize logger error", errs.ZapError(err))
 	}

pkg/mcs/tso/server/server.go

@@ -71,10 +71,11 @@ func (s *Server) GetHTTPClient() *http.Client {
 // CreateServerWrapper encapsulates the configuration/log/metrics initialization and create the server
 func CreateServerWrapper(args []string) (context.Context, context.CancelFunc, bs.Server) {
 	cfg := tso.NewConfig()
-	err := cfg.Parse(os.Args[1:])
+	err := cfg.Parse(args)
 
 	if cfg.Version {
-		printVersionInfo()
+		// TODO: support printing TSO server info
+		// server.PrintTSOInfo()
 		exit(0)
 	}
 
@@ -88,31 +89,28 @@ func CreateServerWrapper(args []string) (context.Context, context.CancelFunc, bs
 		log.Fatal("parse cmd flags error", errs.ZapError(err))
 	}
 
-	if cfg.ConfigCheck {
-		printConfigCheckMsg(cfg)
-		exit(0)
+	// New zap logger
+	err = logutil.SetupLogger(cfg.Log, &cfg.Logger, &cfg.LogProps, cfg.Security.RedactInfoLog)
+	if err == nil {
+		log.ReplaceGlobals(cfg.Logger, cfg.LogProps)
+	} else {
+		log.Fatal("initialize logger error", errs.ZapError(err))
 	}
+	// Flushing any buffered log entries
+	defer log.Sync()
 
-	// TODO: Initialize logger
+	// TODO: support printing TSO server info
+	// LogTSOInfo()
 
 	// TODO: Make it configurable if it has big impact on performance.
 	grpcprometheus.EnableHandlingTimeHistogram()
 
 	metricutil.Push(&cfg.Metric)
 
 	// TODO: Create the server
-
 	return nil, nil, nil
 }
 
-// TODO: implement it
-func printVersionInfo() {
-}
-
-// TODO: implement it
-func printConfigCheckMsg(cfg *tso.Config) {
-}
-
 func exit(code int) {
 	log.Sync()
 	os.Exit(code)

pkg/tso/config.go

@@ -19,8 +19,12 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+	"github.com/tikv/pd/pkg/encryption"
+	"github.com/tikv/pd/pkg/utils/grpcutil"
 	"github.com/tikv/pd/pkg/utils/metricutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
+	"go.uber.org/zap"
 )
 
 const (
@@ -34,8 +38,7 @@ type Config struct {
 
 	Version bool `json:"-"`
 
-	ConfigCheck bool `json:"-"`
-	configFile  string
+	configFile string
 
 	// EnableLocalTSO is used to enable the Local TSO Allocator feature,
 	// which allows the PD server to generate Local TSO for certain DC-level transactions.
@@ -57,6 +60,14 @@ type Config struct {
 	MaxResetTSGap typeutil.Duration `toml:"max-gap-reset-ts" json:"max-gap-reset-ts"`
 
 	Metric metricutil.MetricConfig `toml:"metric" json:"metric"`
+
+	// Log related config.
+	Log log.Config `toml:"log" json:"log"`
+
+	Logger   *zap.Logger
+	LogProps *log.ZapProperties
+
+	Security SecurityConfig `toml:"security" json:"security"`
 }
 
 // NewConfig creates a new config.
@@ -82,3 +93,11 @@ func (c *Config) Parse(arguments []string) error {
 
 	return nil
 }
+
+// SecurityConfig indicates the security configuration for pd server
+type SecurityConfig struct {
+	grpcutil.TLSConfig
+	// RedactInfoLog indicates that whether enabling redact log
+	RedactInfoLog bool              `toml:"redact-info-log" json:"redact-info-log"`
+	Encryption    encryption.Config `toml:"encryption" json:"encryption"`
+}

pkg/utils/logutil/log.go

@@ -20,6 +20,7 @@ import (
 	"sync/atomic"
 
 	"github.com/pingcap/log"
+	"github.com/tikv/pd/pkg/errs"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
@@ -65,6 +66,20 @@ func StringToZapLogLevel(level string) zapcore.Level {
 	return zapcore.InfoLevel
 }
 
+// SetupLogger setup the logger.
+func SetupLogger(logConfig log.Config, logger **zap.Logger, logProps **log.ZapProperties, enabled ...bool) error {
+	lg, p, err := log.InitLogger(&logConfig, zap.AddStacktrace(zapcore.FatalLevel))
+	if err != nil {
+		return errs.ErrInitLogger.Wrap(err).FastGenWithCause()
+	}
+	*logger = lg
+	*logProps = p
+	if len(enabled) > 0 {
+		SetRedactLog(enabled[0])
+	}
+	return nil
+}
+
 // LogPanic logs the panic reason and stack, then exit the process.
 // Commonly used with a `defer`.
 func LogPanic() {

server/api/server_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/suite"
 	"github.com/tikv/pd/pkg/utils/apiutil"
 	"github.com/tikv/pd/pkg/utils/assertutil"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/testutil"
 	"github.com/tikv/pd/server"
 	"github.com/tikv/pd/server/config"
@@ -84,10 +85,10 @@ func mustNewCluster(re *require.Assertions, num int, opts ...func(cfg *config.Co
 	ch := make(chan *server.Server, num)
 	for _, cfg := range cfgs {
 		go func(cfg *config.Config) {
-			err := cfg.SetupLogger()
+			err := logutil.SetupLogger(cfg.Log, &cfg.Logger, &cfg.LogProps, cfg.Security.RedactInfoLog)
 			re.NoError(err)
 			zapLogOnce.Do(func() {
-				log.ReplaceGlobals(cfg.GetZapLogger(), cfg.GetZapLogProperties())
+				log.ReplaceGlobals(cfg.Logger, cfg.LogProps)
 			})
 			for _, opt := range opts {
 				opt(cfg)

server/config/config.go

@@ -32,7 +32,6 @@ import (
 	"github.com/tikv/pd/pkg/errs"
 	"github.com/tikv/pd/pkg/utils/configutil"
 	"github.com/tikv/pd/pkg/utils/grpcutil"
-	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/metricutil"
 	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
@@ -46,7 +45,6 @@ import (
 	"go.etcd.io/etcd/embed"
 	"go.etcd.io/etcd/pkg/transport"
 	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 )
 
 // Config is the pd server configuration.
@@ -160,8 +158,8 @@ type Config struct {
 	HeartbeatStreamBindInterval typeutil.Duration
 	LeaderPriorityCheckInterval typeutil.Duration
 
-	logger   *zap.Logger
-	logProps *log.ZapProperties
+	Logger   *zap.Logger        `json:"-"`
+	LogProps *log.ZapProperties `json:"-"`
 
 	Dashboard DashboardConfig `toml:"dashboard" json:"dashboard"`
 
@@ -1305,28 +1303,6 @@ func (c LabelPropertyConfig) Clone() LabelPropertyConfig {
 	return m
 }
 
-// SetupLogger setup the logger.
-func (c *Config) SetupLogger() error {
-	lg, p, err := log.InitLogger(&c.Log, zap.AddStacktrace(zapcore.FatalLevel))
-	if err != nil {
-		return errs.ErrInitLogger.Wrap(err).FastGenWithCause()
-	}
-	c.logger = lg
-	c.logProps = p
-	logutil.SetRedactLog(c.Security.RedactInfoLog)
-	return nil
-}
-
-// GetZapLogger gets the created zap logger.
-func (c *Config) GetZapLogger() *zap.Logger {
-	return c.logger
-}
-
-// GetZapLogProperties gets properties of the zap logger.
-func (c *Config) GetZapLogProperties() *log.ZapProperties {
-	return c.logProps
-}
-
 // GetConfigFile gets the config file.
 func (c *Config) GetConfigFile() string {
 	return c.configFile
@@ -1386,7 +1362,7 @@ func (c *Config) GenEmbedEtcdConfig() (*embed.Config, error) {
 	cfg.PeerTLSInfo.KeyFile = c.Security.KeyPath
 	cfg.PeerTLSInfo.AllowedCN = allowedCN
 	cfg.ForceNewCluster = c.ForceNewCluster
-	cfg.ZapLoggerBuilder = embed.NewZapCoreLoggerBuilder(c.logger, c.logger.Core(), c.logProps.Syncer)
+	cfg.ZapLoggerBuilder = embed.NewZapCoreLoggerBuilder(c.Logger, c.Logger.Core(), c.LogProps.Syncer)
 	cfg.EnableGRPCGateway = c.EnableGRPCGateway
 	cfg.EnableV2 = true
 	cfg.Logger = "zap"

server/server.go

@@ -249,8 +249,8 @@ func CreateServer(ctx context.Context, cfg *config.Config, legacyServiceBuilders
 	}
 
 	s.etcdCfg = etcdCfg
-	s.lg = cfg.GetZapLogger()
-	s.logProps = cfg.GetZapLogProperties()
+	s.lg = cfg.Logger
+	s.logProps = cfg.LogProps
 	return s, nil
 }
 

server/testutil.go

@@ -27,6 +27,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/pd/pkg/utils/apiutil"
 	"github.com/tikv/pd/pkg/utils/assertutil"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/tempurl"
 	"github.com/tikv/pd/pkg/utils/testutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
@@ -89,10 +90,10 @@ func NewTestSingleConfig(c *assertutil.Checker) *config.Config {
 	cfg.TickInterval = typeutil.NewDuration(100 * time.Millisecond)
 	cfg.ElectionInterval = typeutil.NewDuration(3 * time.Second)
 	cfg.LeaderPriorityCheckInterval = typeutil.NewDuration(100 * time.Millisecond)
-	err := cfg.SetupLogger()
+	err := logutil.SetupLogger(cfg.Log, &cfg.Logger, &cfg.LogProps, cfg.Security.RedactInfoLog)
 	c.AssertNil(err)
 	zapLogOnce.Do(func() {
-		log.ReplaceGlobals(cfg.GetZapLogger(), cfg.GetZapLogProperties())
+		log.ReplaceGlobals(cfg.Logger, cfg.LogProps)
 	})
 
 	c.AssertNil(cfg.Adjust(nil, false))

tests/cluster.go

@@ -34,6 +34,7 @@ import (
 	"github.com/tikv/pd/pkg/id"
 	"github.com/tikv/pd/pkg/swaggerserver"
 	"github.com/tikv/pd/pkg/tso"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/testutil"
 	"github.com/tikv/pd/server"
 	"github.com/tikv/pd/server/api"
@@ -72,12 +73,12 @@ var zapLogOnce sync.Once
 
 // NewTestServer creates a new TestServer.
 func NewTestServer(ctx context.Context, cfg *config.Config) (*TestServer, error) {
-	err := cfg.SetupLogger()
+	err := logutil.SetupLogger(cfg.Log, &cfg.Logger, &cfg.LogProps, cfg.Security.RedactInfoLog)
 	if err != nil {
 		return nil, err
 	}
 	zapLogOnce.Do(func() {
-		log.ReplaceGlobals(cfg.GetZapLogger(), cfg.GetZapLogProperties())
+		log.ReplaceGlobals(cfg.Logger, cfg.LogProps)
 	})
 	err = join.PrepareJoinCluster(cfg)
 	if err != nil {

tools/pd-heartbeat-bench/config/config.go

@@ -6,7 +6,6 @@ import (
 	"github.com/pingcap/log"
 	flag "github.com/spf13/pflag"
 	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 )
 
 const (
@@ -32,8 +31,8 @@ type Config struct {
 	StatusAddr string
 
 	Log      log.Config `toml:"log" json:"log"`
-	logger   *zap.Logger
-	logProps *log.ZapProperties
+	Logger   *zap.Logger
+	LogProps *log.ZapProperties
 
 	StoreCount        int     `toml:"store-count" json:"store-count"`
 	RegionCount       int     `toml:"region-count" json:"region-count"`
@@ -138,27 +137,6 @@ func (c *Config) configFromFile(path string) (*toml.MetaData, error) {
 	return &meta, err
 }
 
-// SetupLogger setup the logger.
-func (c *Config) SetupLogger() error {
-	lg, p, err := log.InitLogger(&c.Log, zap.AddStacktrace(zapcore.FatalLevel))
-	if err != nil {
-		return err
-	}
-	c.logger = lg
-	c.logProps = p
-	return nil
-}
-
-// GetZapLogger gets the created zap logger.
-func (c *Config) GetZapLogger() *zap.Logger {
-	return c.logger
-}
-
-// GetZapLogProperties gets properties of the zap logger.
-func (c *Config) GetZapLogProperties() *log.ZapProperties {
-	return c.logProps
-}
-
 func adjustFloat64(v *float64, defValue float64) {
 	if *v == 0 {
 		*v = defValue

tools/pd-heartbeat-bench/main.go

@@ -408,9 +408,9 @@ func main() {
 	}
 
 	// New zap logger
-	err = cfg.SetupLogger()
+	err = logutil.SetupLogger(cfg.Log, &cfg.Logger, &cfg.LogProps)
 	if err == nil {
-		log.ReplaceGlobals(cfg.GetZapLogger(), cfg.GetZapLogProperties())
+		log.ReplaceGlobals(cfg.Logger, cfg.LogProps)
 	} else {
 		log.Fatal("initialize logger error", zap.Error(err))
 	}