config, api: Add Service Config

server/api/config.go

@@ -151,6 +151,8 @@ func (h *confHandler) updateConfig(cfg *config.Config, key string, value interfa
 			return errors.Errorf("cannot update config prefix %s", kp[0])
 		}
 		return h.updateReplicationModeConfig(cfg, kp[1:], value)
+	case "service":
+		return h.updateServiceConfig(cfg, kp[len(kp)-1], value)
 	case "pd-server":
 		return h.updatePDServerConfig(cfg, kp[len(kp)-1], value)
 	case "log":
@@ -250,6 +252,27 @@ func (h *confHandler) updateReplicationModeConfig(config *config.Config, key []s
 	return err
 }
 
+func (h *confHandler) updateServiceConfig(config *config.Config, key string, value interface{}) error {
+	data, err := json.Marshal(map[string]interface{}{key: value})
+	if err != nil {
+		return err
+	}
+
+	updated, found, err := mergeConfig(&config.ServiceCfg, data)
+	if err != nil {
+		return err
+	}
+
+	if !found {
+		return errors.Errorf("config item %s not found", key)
+	}
+
+	if updated {
+		err = h.svr.SetServiceConfig(config.ServiceCfg)
+	}
+	return err
+}
+
 func (h *confHandler) updatePDServerConfig(config *config.Config, key string, value interface{}) error {
 	data, err := json.Marshal(map[string]interface{}{key: value})
 	if err != nil {
@@ -535,6 +558,15 @@ func (h *confHandler) SetReplicationModeConfig(w http.ResponseWriter, r *http.Re
 	h.rd.JSON(w, http.StatusOK, "The replication mode config is updated.")
 }
 
+// @Tags config
+// @Summary Get Service config.
+// @Produce json
+// @Success 200 {object} config.ServiceConfig
+// @Router /config/service [get]
+func (h *confHandler) GetServiceConfig(w http.ResponseWriter, r *http.Request) {
+	h.rd.JSON(w, http.StatusOK, h.svr.GetServiceConfig())
+}
+
 // @Tags config
 // @Summary Get PD server config.
 // @Produce json

server/api/config_test.go

@@ -279,6 +279,34 @@ func (s *testConfigSuite) TestConfigDefault(c *C) {
 	c.Assert(defaultCfg.PDServerCfg.MetricStorage, Equals, "")
 }
 
+func (s *testConfigSuite) TestConfigService(c *C) {
+	addrGet := fmt.Sprintf("%s/config/service", s.urlPrefix)
+	sc := &config.ServiceConfig{}
+	c.Assert(readJSON(testDialClient, addrGet, sc), IsNil)
+	c.Assert(sc.EnableAudit, Equals, false)
+
+	addrPost := fmt.Sprintf("%s/config", s.urlPrefix)
+	// test update enable-audit
+	ms := map[string]interface{}{
+		"enable-audit": "true",
+	}
+	postData, err := json.Marshal(ms)
+	c.Assert(err, IsNil)
+	c.Assert(postJSON(testDialClient, addrPost, postData), IsNil)
+	sc = &config.ServiceConfig{}
+	c.Assert(readJSON(testDialClient, addrGet, sc), IsNil)
+	c.Assert(sc.EnableAudit, Equals, true)
+	ms = map[string]interface{}{
+		"enable-audit": "false",
+	}
+	postData, err = json.Marshal(ms)
+	c.Assert(err, IsNil)
+	c.Assert(postJSON(testDialClient, addrPost, postData), IsNil)
+	sc = &config.ServiceConfig{}
+	c.Assert(readJSON(testDialClient, addrGet, sc), IsNil)
+	c.Assert(sc.EnableAudit, Equals, false)
+}
+
 func (s *testConfigSuite) TestConfigPDServer(c *C) {
 	addrPost := fmt.Sprintf("%s/config", s.urlPrefix)
 
@@ -301,27 +329,6 @@ func (s *testConfigSuite) TestConfigPDServer(c *C) {
 	c.Assert(sc.FlowRoundByDigit, Equals, int(3))
 	c.Assert(sc.MinResolvedTSPersistenceInterval, Equals, typeutil.NewDuration(0))
 	c.Assert(sc.MaxResetTSGap.Duration, Equals, 24*time.Hour)
-	c.Assert(sc.EnableAudit, Equals, false)
-
-	// test update enable-audit
-	ms = map[string]interface{}{
-		"enable-audit": true,
-	}
-	postData, err = json.Marshal(ms)
-	c.Assert(err, IsNil)
-	c.Assert(postJSON(testDialClient, addrPost, postData), IsNil)
-	sc = &config.PDServerConfig{}
-	c.Assert(readJSON(testDialClient, addrGet, sc), IsNil)
-	c.Assert(sc.EnableAudit, Equals, true)
-	ms = map[string]interface{}{
-		"enable-audit": false,
-	}
-	postData, err = json.Marshal(ms)
-	c.Assert(err, IsNil)
-	c.Assert(postJSON(testDialClient, addrPost, postData), IsNil)
-	sc = &config.PDServerConfig{}
-	c.Assert(readJSON(testDialClient, addrGet, sc), IsNil)
-	c.Assert(sc.EnableAudit, Equals, false)
 }
 
 var ttlConfig = map[string]interface{}{

server/api/router.go

@@ -148,6 +148,7 @@ func createRouter(prefix string, svr *server.Server) *mux.Router {
 	registerFunc(apiRouter, "/config/default", confHandler.GetDefaultConfig, setMethods("GET"))
 	registerFunc(apiRouter, "/config/schedule", confHandler.GetScheduleConfig, setMethods("GET"))
 	registerFunc(apiRouter, "/config/schedule", confHandler.SetScheduleConfig, setMethods("POST"), setAuditBackend(localLog))
+	registerFunc(apiRouter, "/config/service", confHandler.GetServiceConfig, setMethods("GET"))
 	registerFunc(apiRouter, "/config/pd-server", confHandler.GetPDServerConfig, setMethods("GET"))
 	registerFunc(apiRouter, "/config/replicate", confHandler.GetReplicationConfig, setMethods("GET"))
 	registerFunc(apiRouter, "/config/replicate", confHandler.SetReplicationConfig, setMethods("POST"), setAuditBackend(localLog))

server/config/config.go

@@ -107,6 +107,8 @@ type Config struct {
 
 	Replication ReplicationConfig `toml:"replication" json:"replication"`
 
+	ServiceCfg ServiceConfig `toml:"service" json:"service"`
+
 	PDServerCfg PDServerConfig `toml:"pd-server" json:"pd-server"`
 
 	ClusterVersion semver.Version `toml:"cluster-version" json:"cluster-version"`
@@ -574,6 +576,10 @@ func (c *Config) Adjust(meta *toml.MetaData, reloading bool) error {
 		return err
 	}
 
+	if err := c.ServiceCfg.adjust(configMetaData.Child("service")); err != nil {
+		return err
+	}
+
 	if err := c.PDServerCfg.adjust(configMetaData.Child("pd-server")); err != nil {
 		return err
 	}
@@ -1086,6 +1092,25 @@ func (c *ReplicationConfig) adjust(meta *configMetaData) error {
 	return c.Validate()
 }
 
+// ServiceConfig is the configuration for PD service such as HTTP API and gRPC.
+type ServiceConfig struct {
+	// EnableAudit controls the switch of the audit middleware
+	EnableAudit bool `toml:"enable-audit" json:"enable-audit,string"`
+}
+
+// Clone returns a cloned PD server config.
+func (c *ServiceConfig) Clone() *ServiceConfig {
+	cfg := *c
+	return &cfg
+}
+
+func (c *ServiceConfig) adjust(meta *configMetaData) error {
+	if !meta.IsDefined("enable-audit") {
+		c.EnableAudit = defaultEnableAuditMiddleware
+	}
+	return nil
+}
+
 // PDServerConfig is the configuration for pd server.
 // NOTE: This type is exported by HTTP API. Please pay more attention when modifying it.
 type PDServerConfig struct {
@@ -1110,8 +1135,6 @@ type PDServerConfig struct {
 	FlowRoundByDigit int `toml:"flow-round-by-digit" json:"flow-round-by-digit"`
 	// MinResolvedTSPersistenceInterval is the interval to save the min resolved ts.
 	MinResolvedTSPersistenceInterval typeutil.Duration `toml:"min-resolved-ts-persistence-interval" json:"min-resolved-ts-persistence-interval"`
-	// EnableAudit controls the switch of the audit middleware
-	EnableAudit bool `toml:"enable-audit" json:"enable-audit"`
 }
 
 func (c *PDServerConfig) adjust(meta *configMetaData) error {
@@ -1137,9 +1160,6 @@ func (c *PDServerConfig) adjust(meta *configMetaData) error {
 	if !meta.IsDefined("min-resolved-ts-persistence-interval") {
 		adjustDuration(&c.MinResolvedTSPersistenceInterval, defaultMinResolvedTSPersistenceInterval)
 	}
-	if !meta.IsDefined("enable-audit") {
-		c.EnableAudit = defaultEnableAuditMiddleware
-	}
 	c.migrateConfigurationFromFile(meta)
 	return c.Validate()
 }

server/config/config_test.go

@@ -502,6 +502,10 @@ func (s *testConfigSuite) TestConfigClone(c *C) {
 	replication.adjust(emptyConfigMetaData)
 	c.Assert(replication.Clone(), DeepEquals, replication)
 
+	service := &ServiceConfig{}
+	service.adjust(emptyConfigMetaData)
+	c.Assert(service.Clone(), DeepEquals, service)
+
 	pdServer := &PDServerConfig{}
 	pdServer.adjust(emptyConfigMetaData)
 	c.Assert(pdServer.Clone(), DeepEquals, pdServer)

server/config/persist_options.go

@@ -46,6 +46,7 @@ type PersistOptions struct {
 	ttl             *cache.TTLString
 	schedule        atomic.Value
 	replication     atomic.Value
+	serviceConfig   atomic.Value
 	pdServerConfig  atomic.Value
 	replicationMode atomic.Value
 	labelProperty   atomic.Value
@@ -57,6 +58,7 @@ func NewPersistOptions(cfg *Config) *PersistOptions {
 	o := &PersistOptions{}
 	o.schedule.Store(&cfg.Schedule)
 	o.replication.Store(&cfg.Replication)
+	o.serviceConfig.Store(&cfg.ServiceCfg)
 	o.pdServerConfig.Store(&cfg.PDServerCfg)
 	o.replicationMode.Store(&cfg.ReplicationMode)
 	o.labelProperty.Store(cfg.LabelProperty)
@@ -85,6 +87,16 @@ func (o *PersistOptions) SetReplicationConfig(cfg *ReplicationConfig) {
 	o.replication.Store(cfg)
 }
 
+// GetServiceConfig returns pd server configurations.
+func (o *PersistOptions) GetServiceConfig() *ServiceConfig {
+	return o.serviceConfig.Load().(*ServiceConfig)
+}
+
+// SetServiceConfig sets the PD configuration.
+func (o *PersistOptions) SetServiceConfig(cfg *ServiceConfig) {
+	o.serviceConfig.Store(cfg)
+}
+
 // GetPDServerConfig returns pd server configurations.
 func (o *PersistOptions) GetPDServerConfig() *PDServerConfig {
 	return o.pdServerConfig.Load().(*PDServerConfig)
@@ -446,7 +458,7 @@ func (o *PersistOptions) GetLeaderSchedulePolicy() core.SchedulePolicy {
 
 // IsAuditEnabled returns whether audit middleware is enabled
 func (o *PersistOptions) IsAuditEnabled() bool {
-	return o.GetPDServerConfig().EnableAudit
+	return o.GetServiceConfig().EnableAudit
 }
 
 // GetKeyType is to get key type.
@@ -598,6 +610,7 @@ func (o *PersistOptions) Persist(storage endpoint.ConfigStorage) error {
 	cfg := &Config{
 		Schedule:        *o.GetScheduleConfig(),
 		Replication:     *o.GetReplicationConfig(),
+		ServiceCfg:      *o.GetServiceConfig(),
 		PDServerCfg:     *o.GetPDServerConfig(),
 		ReplicationMode: *o.GetReplicationModeConfig(),
 		LabelProperty:   o.GetLabelPropertyConfig(),
@@ -625,6 +638,7 @@ func (o *PersistOptions) Reload(storage endpoint.ConfigStorage) error {
 	if isExist {
 		o.schedule.Store(&cfg.Schedule)
 		o.replication.Store(&cfg.Replication)
+		o.serviceConfig.Store(&cfg.ServiceCfg)
 		o.pdServerConfig.Store(&cfg.PDServerCfg)
 		o.replicationMode.Store(&cfg.ReplicationMode)
 		o.labelProperty.Store(cfg.LabelProperty)

server/server.go

@@ -813,6 +813,7 @@ func (s *Server) GetConfig() *config.Config {
 	cfg := s.cfg.Clone()
 	cfg.Schedule = *s.persistOptions.GetScheduleConfig().Clone()
 	cfg.Replication = *s.persistOptions.GetReplicationConfig().Clone()
+	cfg.ServiceCfg = *s.persistOptions.GetServiceConfig().Clone()
 	cfg.PDServerCfg = *s.persistOptions.GetPDServerConfig().Clone()
 	cfg.ReplicationMode = *s.persistOptions.GetReplicationModeConfig()
 	cfg.LabelProperty = s.persistOptions.GetLabelPropertyConfig().Clone()
@@ -955,6 +956,27 @@ func (s *Server) SetReplicationConfig(cfg config.ReplicationConfig) error {
 	return nil
 }
 
+// GetServiceConfig gets the service config information.
+func (s *Server) GetServiceConfig() *config.ServiceConfig {
+	return s.persistOptions.GetServiceConfig().Clone()
+}
+
+// SetServiceConfig sets the service config.
+func (s *Server) SetServiceConfig(cfg config.ServiceConfig) error {
+	old := s.persistOptions.GetServiceConfig()
+	s.persistOptions.SetServiceConfig(&cfg)
+	if err := s.persistOptions.Persist(s.storage); err != nil {
+		s.persistOptions.SetServiceConfig(old)
+		log.Error("failed to update Service config",
+			zap.Reflect("new", cfg),
+			zap.Reflect("old", old),
+			errs.ZapError(err))
+		return err
+	}
+	log.Info("Service config is updated", zap.Reflect("new", cfg), zap.Reflect("old", old))
+	return nil
+}
+
 // GetPDServerConfig gets the balance config information.
 func (s *Server) GetPDServerConfig() *config.PDServerConfig {
 	return s.persistOptions.GetPDServerConfig().Clone()

tests/server/api/api_test.go

@@ -153,7 +153,7 @@ func (s *testMiddlewareSuite) TestRequestInfoMiddleware(c *C) {
 	leader := s.cluster.GetServer(s.cluster.GetLeader())
 
 	input := map[string]interface{}{
-		"enable-audit": true,
+		"enable-audit": "true",
 	}
 	data, err := json.Marshal(input)
 	c.Assert(err, IsNil)
@@ -181,7 +181,7 @@ func (s *testMiddlewareSuite) TestRequestInfoMiddleware(c *C) {
 	c.Assert(resp.Header.Get("ip"), Equals, "127.0.0.1")
 
 	input = map[string]interface{}{
-		"enable-audit": false,
+		"enable-audit": "false",
 	}
 	data, err = json.Marshal(input)
 	c.Assert(err, IsNil)
@@ -206,7 +206,7 @@ func BenchmarkDoRequestWithServiceMiddleware(b *testing.B) {
 	cluster.WaitLeader()
 	leader := cluster.GetServer(cluster.GetLeader())
 	input := map[string]interface{}{
-		"enable-audit": true,
+		"enable-audit": "true",
 	}
 	data, _ := json.Marshal(input)
 	req, _ := http.NewRequest("POST", leader.GetAddr()+"/pd/api/v1/config", bytes.NewBuffer(data))
@@ -229,7 +229,7 @@ func BenchmarkDoRequestWithoutServiceMiddleware(b *testing.B) {
 	cluster.WaitLeader()
 	leader := cluster.GetServer(cluster.GetLeader())
 	input := map[string]interface{}{
-		"enable-audit": false,
+		"enable-audit": "false",
 	}
 	data, _ := json.Marshal(input)
 	req, _ := http.NewRequest("POST", leader.GetAddr()+"/pd/api/v1/config", bytes.NewBuffer(data))
@@ -254,7 +254,7 @@ func doTestRequest(srv *tests.TestServer) {
 func (s *testMiddlewareSuite) TestAuditPrometheusBackend(c *C) {
 	leader := s.cluster.GetServer(s.cluster.GetLeader())
 	input := map[string]interface{}{
-		"enable-audit": true,
+		"enable-audit": "true",
 	}
 	data, err := json.Marshal(input)
 	c.Assert(err, IsNil)
@@ -302,7 +302,7 @@ func (s *testMiddlewareSuite) TestAuditPrometheusBackend(c *C) {
 	c.Assert(strings.Contains(output, "pd_service_audit_handling_seconds_count{component=\"anonymous\",method=\"HTTP\",service=\"GetTrend\"} 2"), Equals, true)
 
 	input = map[string]interface{}{
-		"enable-audit": false,
+		"enable-audit": "false",
 	}
 	data, err = json.Marshal(input)
 	c.Assert(err, IsNil)
@@ -322,7 +322,7 @@ func (s *testMiddlewareSuite) TestAuditLocalLogBackend(c *C) {
 	log.ReplaceGlobals(lg, p)
 	leader := s.cluster.GetServer(s.cluster.GetLeader())
 	input := map[string]interface{}{
-		"enable-audit": true,
+		"enable-audit": "true",
 	}
 	data, err := json.Marshal(input)
 	c.Assert(err, IsNil)
@@ -354,7 +354,7 @@ func BenchmarkDoRequestWithLocalLogAudit(b *testing.B) {
 	cluster.WaitLeader()
 	leader := cluster.GetServer(cluster.GetLeader())
 	input := map[string]interface{}{
-		"enable-audit": true,
+		"enable-audit": "true",
 	}
 	data, _ := json.Marshal(input)
 	req, _ := http.NewRequest("POST", leader.GetAddr()+"/pd/api/v1/config", bytes.NewBuffer(data))
@@ -377,7 +377,7 @@ func BenchmarkDoRequestWithoutLocalLogAudit(b *testing.B) {
 	cluster.WaitLeader()
 	leader := cluster.GetServer(cluster.GetLeader())
 	input := map[string]interface{}{
-		"enable-audit": false,
+		"enable-audit": "false",
 	}
 	data, _ := json.Marshal(input)
 	req, _ := http.NewRequest("POST", leader.GetAddr()+"/pd/api/v1/config", bytes.NewBuffer(data))