diff --git a/pkg/dashboard/adapter/manager.go b/pkg/dashboard/adapter/manager.go index 228203af035..c1fb5cc8f69 100644 --- a/pkg/dashboard/adapter/manager.go +++ b/pkg/dashboard/adapter/manager.go @@ -185,9 +185,7 @@ func (m *Manager) setNewAddress() { } // set new dashboard address - cfg := m.srv.GetPersistOptions().GetPDServerConfig().Clone() - cfg.DashboardAddress = m.members[minMemberIdx].GetClientUrls()[0] - if err := m.srv.SetPDServerConfig(*cfg); err != nil { + if err := m.srv.SetDashboardAddress(m.members[minMemberIdx].GetClientUrls()[0]); err != nil { log.Warn("failed to set persist options") } } diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go index b4d5902a35c..8dced9a370f 100644 --- a/pkg/mcs/scheduling/server/cluster.go +++ b/pkg/mcs/scheduling/server/cluster.go @@ -107,6 +107,7 @@ type Cluster struct { const ( regionLabelGCInterval = time.Hour requestTimeout = 3 * time.Second + pdLeaderRetryInterval = 50 * time.Millisecond requestInterval = 10 * time.Second // heartbeat relative const @@ -382,12 +383,12 @@ func (c *Cluster) GetStoreConfig() sc.StoreConfigProvider { return c.persistConf // AllocID allocates new IDs. func (c *Cluster) AllocID(count uint32) (uint64, uint32, error) { - client, err := c.getPDLeaderClient() + ctx, cancel := context.WithTimeout(c.ctx, requestTimeout) + defer cancel() + client, err := c.getPDLeaderClient(ctx) if err != nil { return 0, 0, err } - ctx, cancel := context.WithTimeout(c.ctx, requestTimeout) - defer cancel() req := &pdpb.AllocIDRequest{Header: &pdpb.RequestHeader{ClusterId: keypath.ClusterID()}, Count: count} failpoint.Inject("allocIDNonBatch", func() { @@ -401,13 +402,24 @@ func (c *Cluster) AllocID(count uint32) (uint64, uint32, error) { return resp.GetId(), resp.GetCount(), nil } -func (c *Cluster) getPDLeaderClient() (pdpb.PDClient, error) { - cli := c.pdLeader.Load() - if cli == nil { +func (c *Cluster) getPDLeaderClient(ctx context.Context) (pdpb.PDClient, error) { + if cli := c.pdLeader.Load(); cli != nil { + return cli.(pdpb.PDClient), nil + } + c.triggerMembershipCheck() + ticker := time.NewTicker(pdLeaderRetryInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil, errors.New("pd leader is not found") + case <-ticker.C: + } + if cli := c.pdLeader.Load(); cli != nil { + return cli.(pdpb.PDClient), nil + } c.triggerMembershipCheck() - return nil, errors.New("PD leader is not found") } - return cli.(pdpb.PDClient), nil } func (c *Cluster) triggerMembershipCheck() { diff --git a/pkg/mcs/scheduling/server/config/config.go b/pkg/mcs/scheduling/server/config/config.go index e04bea49e2e..43a7e1e17f0 100644 --- a/pkg/mcs/scheduling/server/config/config.go +++ b/pkg/mcs/scheduling/server/config/config.go @@ -19,6 +19,7 @@ import ( "os" "reflect" "strconv" + "sync" "sync/atomic" "time" "unsafe" @@ -190,7 +191,8 @@ func (c *Config) Clone() *Config { // PersistConfig wraps all configurations that need to persist to storage and // allows to access them safely. type PersistConfig struct { - ttl *cache.TTLString + ttl *cache.TTLString + configMu sync.Mutex // Store the global configuration that is related to the scheduling. clusterVersion unsafe.Pointer schedule atomic.Value @@ -232,7 +234,10 @@ func (o *PersistConfig) tryNotifySchedulersUpdating() { if notifier == nil { return } - notifier <- struct{}{} + select { + case notifier <- struct{}{}: + default: + } } // GetClusterVersion returns the cluster version. @@ -252,6 +257,12 @@ func (o *PersistConfig) GetScheduleConfig() *sc.ScheduleConfig { // SetScheduleConfig sets the scheduling configuration dynamically. func (o *PersistConfig) SetScheduleConfig(cfg *sc.ScheduleConfig) { + o.configMu.Lock() + defer o.configMu.Unlock() + o.storeScheduleConfig(cfg) +} + +func (o *PersistConfig) storeScheduleConfig(cfg *sc.ScheduleConfig) { old := o.GetScheduleConfig() o.schedule.Store(cfg) // The coordinator is not aware of the underlying scheduler config changes, @@ -261,6 +272,24 @@ func (o *PersistConfig) SetScheduleConfig(cfg *sc.ScheduleConfig) { } } +// UpdateScheduleConfig updates the latest scheduling configuration. +func (o *PersistConfig) UpdateScheduleConfig(storage endpoint.ConfigStorage, update func(*sc.ScheduleConfig) error) error { + o.configMu.Lock() + defer o.configMu.Unlock() + + old := o.GetScheduleConfig() + cfg := old.Clone() + if err := update(cfg); err != nil { + return err + } + o.storeScheduleConfig(cfg) + if err := o.Persist(storage); err != nil { + o.storeScheduleConfig(old) + return err + } + return nil +} + // AdjustScheduleCfg adjusts the schedule config during the initialization. func AdjustScheduleCfg(scheduleCfg *sc.ScheduleConfig) { // In case we add new default schedulers. diff --git a/pkg/mcs/scheduling/server/server.go b/pkg/mcs/scheduling/server/server.go index b778cc1c67a..86763f0b13d 100644 --- a/pkg/mcs/scheduling/server/server.go +++ b/pkg/mcs/scheduling/server/server.go @@ -190,47 +190,57 @@ func (s *Server) updatePDMemberLoop() { case <-ticker.C: case <-s.checkMembershipCh: } - if !s.IsServing() { + curLeader = s.updatePDLeader(ctx, curLeader) + } +} + +func (s *Server) updatePDLeader(ctx context.Context, curLeader uint64) uint64 { + if !s.IsServing() { + return curLeader + } + members, err := etcdutil.ListEtcdMembers(ctx, s.GetClient()) + if err != nil { + log.Warn("failed to list members", errs.ZapError(err)) + return curLeader + } + for _, ep := range members.Members { + if len(ep.GetClientURLs()) == 0 { // This member is not started yet. + log.Info("member is not started yet", zap.String("member-id", strconv.FormatUint(ep.GetID(), 16)), errs.ZapError(err)) continue } - members, err := etcdutil.ListEtcdMembers(ctx, s.GetClient()) + probeCtx, cancel := context.WithTimeout(ctx, requestTimeout) + status, err := s.GetClient().Status(probeCtx, ep.ClientURLs[0]) + cancel() if err != nil { - log.Warn("failed to list members", errs.ZapError(err)) + log.Info("failed to get status of member", zap.String("member-id", strconv.FormatUint(ep.ID, 16)), zap.String("endpoint", ep.ClientURLs[0]), errs.ZapError(err)) continue } - for _, ep := range members.Members { - if len(ep.GetClientURLs()) == 0 { // This member is not started yet. - log.Info("member is not started yet", zap.String("member-id", strconv.FormatUint(ep.GetID(), 16)), errs.ZapError(err)) - continue - } - status, err := s.GetClient().Status(ctx, ep.ClientURLs[0]) - if err != nil { - log.Info("failed to get status of member", zap.String("member-id", strconv.FormatUint(ep.ID, 16)), zap.String("endpoint", ep.ClientURLs[0]), errs.ZapError(err)) - continue - } - if status.Leader == ep.ID { - cc, err := s.GetDelegateClient(ctx, s.GetTLSConfig(), ep.ClientURLs[0]) - if err != nil { - log.Info("failed to get delegate client", errs.ZapError(err)) - continue - } - if !s.IsServing() { - // double check - break - } - cluster := s.GetCluster() - if cluster != nil { - if cluster.SwitchPDLeader(pdpb.NewPDClient(cc)) { - if status.Leader != curLeader { - log.Info("switch PD leader", zap.String("leader-id", strconv.FormatUint(ep.ID, 16)), zap.String("endpoint", ep.ClientURLs[0])) - } - curLeader = ep.ID - break - } - } + if status.Leader != ep.ID { + continue + } + probeCtx, cancel = context.WithTimeout(ctx, requestTimeout) + cc, err := s.GetDelegateClient(probeCtx, s.GetTLSConfig(), ep.ClientURLs[0]) + cancel() + if err != nil { + log.Info("failed to get delegate client", errs.ZapError(err)) + continue + } + if !s.IsServing() { + // double check + return curLeader + } + cluster := s.GetCluster() + if cluster == nil { + return curLeader + } + if cluster.SwitchPDLeader(pdpb.NewPDClient(cc)) { + if status.Leader != curLeader { + log.Info("switch PD leader", zap.String("leader-id", strconv.FormatUint(ep.ID, 16)), zap.String("endpoint", ep.ClientURLs[0])) } + return ep.ID } } + return curLeader } func (s *Server) primaryElectionLoop() { diff --git a/pkg/schedule/config/config_provider.go b/pkg/schedule/config/config_provider.go index 37566a48327..c6ea9d98d6f 100644 --- a/pkg/schedule/config/config_provider.go +++ b/pkg/schedule/config/config_provider.go @@ -69,6 +69,7 @@ type SchedulerConfigProvider interface { GetScheduleConfig() *ScheduleConfig SetScheduleConfig(*ScheduleConfig) + UpdateScheduleConfig(endpoint.ConfigStorage, func(*ScheduleConfig) error) error } // CheckerConfigProvider is the interface for checker configurations. diff --git a/pkg/schedule/coordinator.go b/pkg/schedule/coordinator.go index bc2b3761790..05e38ee5c66 100644 --- a/pkg/schedule/coordinator.go +++ b/pkg/schedule/coordinator.go @@ -16,6 +16,7 @@ package schedule import ( "context" + "reflect" "strconv" "sync" "time" @@ -38,6 +39,7 @@ import ( "github.com/tikv/pd/pkg/schedule/schedulers" "github.com/tikv/pd/pkg/schedule/splitter" "github.com/tikv/pd/pkg/schedule/types" + "github.com/tikv/pd/pkg/slice" "github.com/tikv/pd/pkg/statistics" "github.com/tikv/pd/pkg/statistics/utils" "github.com/tikv/pd/pkg/utils/logutil" @@ -283,6 +285,7 @@ func (c *Coordinator) InitSchedulers(needRun bool) { log.Fatal("cannot load schedulers' config", errs.ZapError(err)) } scheduleCfg := c.cluster.GetSchedulerConfig().GetScheduleConfig().Clone() + originalSchedulerCfgs := append(sc.SchedulerConfigs(nil), scheduleCfg.Schedulers...) // The new way to create scheduler with the independent configuration. for i, name := range scheduleNames { select { @@ -318,13 +321,13 @@ func (c *Coordinator) InitSchedulers(needRun bool) { } if needRun { log.Info("create scheduler with independent configuration", zap.String("scheduler-name", s.GetName())) - if err = c.schedulers.AddScheduler(s); err != nil { + if err = c.schedulers.AddScheduler(s, cfg.Args...); err != nil { log.Error("can not add scheduler with independent configuration", zap.String("scheduler-name", s.GetName()), zap.Strings("scheduler-args", cfg.Args), errs.ZapError(err)) } } else { log.Info("create scheduler handler with independent configuration", zap.String("scheduler-name", s.GetName())) - if err = c.schedulers.AddSchedulerHandler(s); err != nil { + if err = c.schedulers.AddSchedulerHandler(s, cfg.Args...); err != nil { log.Error("can not add scheduler handler with independent configuration", zap.String("scheduler-name", s.GetName()), zap.Strings("scheduler-args", cfg.Args), errs.ZapError(err)) } @@ -382,16 +385,46 @@ func (c *Coordinator) InitSchedulers(needRun bool) { } // Removes the invalid scheduler config and persist. - scheduleCfg.Schedulers = scheduleCfg.Schedulers[:k] - c.cluster.GetSchedulerConfig().SetScheduleConfig(scheduleCfg) - if err := c.cluster.GetSchedulerConfig().Persist(c.cluster.GetStorage()); err != nil { + validSchedulerCfgs := scheduleCfg.Schedulers[:k] + invalidSchedulerCfgs := make(sc.SchedulerConfigs, 0, len(originalSchedulerCfgs)-len(validSchedulerCfgs)) + for _, schedulerCfg := range originalSchedulerCfgs { + if !containsSchedulerConfig(validSchedulerCfgs, schedulerCfg) { + invalidSchedulerCfgs = append(invalidSchedulerCfgs, schedulerCfg) + } + } + var updatedSchedulers sc.SchedulerConfigs + if err := c.cluster.GetSchedulerConfig().UpdateScheduleConfig(c.cluster.GetStorage(), func(latest *sc.ScheduleConfig) error { + if len(invalidSchedulerCfgs) == 0 { + updatedSchedulers = append(sc.SchedulerConfigs(nil), latest.Schedulers...) + return nil + } + latest.Schedulers = removeSchedulerConfigs(latest.Schedulers, invalidSchedulerCfgs) + updatedSchedulers = append(sc.SchedulerConfigs(nil), latest.Schedulers...) + return nil + }); err != nil { log.Error("cannot persist schedule config", errs.ZapError(err)) } - log.Info("scheduler config is updated", zap.Reflect("scheduler-config", scheduleCfg.Schedulers)) + log.Info("scheduler config is updated", zap.Reflect("scheduler-config", updatedSchedulers)) c.markSchedulersInitialized() } +func containsSchedulerConfig(schedulers sc.SchedulerConfigs, target sc.SchedulerConfig) bool { + return slice.AnyOf(schedulers, func(i int) bool { + return reflect.DeepEqual(schedulers[i], target) + }) +} + +func removeSchedulerConfigs(schedulers, removed sc.SchedulerConfigs) sc.SchedulerConfigs { + retained := schedulers[:0] + for _, schedulerCfg := range schedulers { + if !containsSchedulerConfig(removed, schedulerCfg) { + retained = append(retained, schedulerCfg) + } + } + return retained +} + // LoadPlugin load user plugin func (c *Coordinator) LoadPlugin(pluginPath string, ch chan string) { log.Info("load plugin", zap.String("plugin-path", pluginPath)) diff --git a/pkg/schedule/coordinator_test.go b/pkg/schedule/coordinator_test.go new file mode 100644 index 00000000000..74ab255ee66 --- /dev/null +++ b/pkg/schedule/coordinator_test.go @@ -0,0 +1,58 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package schedule + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.uber.org/goleak" + + sc "github.com/tikv/pd/pkg/schedule/config" + "github.com/tikv/pd/pkg/schedule/types" + "github.com/tikv/pd/pkg/utils/testutil" +) + +func TestMain(m *testing.M) { + goleak.VerifyTestMain(m, testutil.LeakOptions...) +} + +func TestRemoveSchedulerConfigsKeepsConcurrentUpdates(t *testing.T) { + re := require.New(t) + + invalidOldCfg := sc.SchedulerConfig{ + Type: types.SchedulerTypeCompatibleMap[types.GrantLeaderScheduler], + Args: []string{"1"}, + } + concurrentUpdatedCfg := sc.SchedulerConfig{ + Type: types.SchedulerTypeCompatibleMap[types.GrantLeaderScheduler], + Args: []string{"2"}, + } + concurrentAddedCfg := sc.SchedulerConfig{ + Type: types.SchedulerTypeCompatibleMap[types.ShuffleRegionScheduler], + } + latest := sc.SchedulerConfigs{ + sc.DefaultSchedulers[0], + invalidOldCfg, + concurrentUpdatedCfg, + concurrentAddedCfg, + } + + re.Equal(sc.SchedulerConfigs{ + sc.DefaultSchedulers[0], + concurrentUpdatedCfg, + concurrentAddedCfg, + }, removeSchedulerConfigs(latest, sc.SchedulerConfigs{invalidOldCfg})) +} diff --git a/server/api/config.go b/server/api/config.go index 165569f1a1b..cd5790ea535 100644 --- a/server/api/config.go +++ b/server/api/config.go @@ -48,6 +48,7 @@ import ( var ( _ *sc.ScheduleConfig = nil resourceManagerControllerConfigURL = "/resource-manager/api/v1/config/controller" + errConfigNotUpdated = errors.New("config is not updated") ) type confHandler struct { @@ -259,20 +260,21 @@ func (h *confHandler) updateMicroserviceConfig(config *config.Config, key string return err } -func (h *confHandler) updateSchedule(config *config.Config, key string, value any) error { - updated, found, err := jsonutil.AddKeyValue(&config.Schedule, key, value) - if err != nil { - return err - } - - if !found { - return errors.Errorf("config item %s not found", key) - } +func (h *confHandler) updateSchedule(_ *config.Config, key string, value any) error { + return ignoreConfigNotUpdated(h.svr.UpdateScheduleConfig(func(config *sc.ScheduleConfig) error { + updated, found, err := jsonutil.AddKeyValue(config, key, value) + if err != nil { + return err + } - if updated { - err = h.svr.SetScheduleConfig(config.Schedule) - } - return err + if !found { + return errors.Errorf("config item %s not found", key) + } + if !updated { + return errConfigNotUpdated + } + return nil + })) } func (h *confHandler) updateReplication(config *config.Config, key string, value any) error { @@ -313,18 +315,26 @@ func (h *confHandler) updateReplicationModeConfig(config *config.Config, key []s return err } -func (h *confHandler) updatePDServerConfig(config *config.Config, key string, value any) error { - updated, found, err := jsonutil.AddKeyValue(&config.PDServerCfg, key, value) - if err != nil { - return err - } +func (h *confHandler) updatePDServerConfig(_ *config.Config, key string, value any) error { + return ignoreConfigNotUpdated(h.svr.UpdatePDServerConfig(func(config *config.PDServerConfig) error { + updated, found, err := jsonutil.AddKeyValue(config, key, value) + if err != nil { + return err + } - if !found { - return errors.Errorf("config item %s not found", key) - } + if !found { + return errors.Errorf("config item %s not found", key) + } + if !updated { + return errConfigNotUpdated + } + return nil + })) +} - if updated { - err = h.svr.SetPDServerConfig(config.PDServerCfg) +func ignoreConfigNotUpdated(err error) error { + if errors.Cause(err) == errConfigNotUpdated { + return nil } return err } @@ -422,21 +432,23 @@ func (h *confHandler) SetScheduleConfig(w http.ResponseWriter, r *http.Request) } } - config := h.svr.GetScheduleConfig() - err = json.Unmarshal(data, &config) - if err != nil { + var unmarshalErr error + err = h.svr.UpdateScheduleConfig(func(config *sc.ScheduleConfig) error { + unmarshalErr = json.Unmarshal(data, config) + return unmarshalErr + }) + if unmarshalErr != nil { var errCode errcode.ErrorCode - err = apiutil.TagJSONError(err) - if jsonErr, ok := errors.Cause(err).(apiutil.JSONError); ok { + unmarshalErr = apiutil.TagJSONError(unmarshalErr) + if jsonErr, ok := errors.Cause(unmarshalErr).(apiutil.JSONError); ok { errCode = errcode.NewInvalidInputErr(jsonErr.Err) } else { - errCode = errcode.NewInternalErr(err) + errCode = errcode.NewInternalErr(unmarshalErr) } apiutil.ErrorResp(h.rd, w, errCode) return } - - if err := h.svr.SetScheduleConfig(*config); err != nil { + if err != nil { h.rd.JSON(w, http.StatusInternalServerError, err.Error()) return } diff --git a/server/config/config_test.go b/server/config/config_test.go index a7ef66e07a6..23142506871 100644 --- a/server/config/config_test.go +++ b/server/config/config_test.go @@ -20,6 +20,7 @@ import ( "math" "os" "path/filepath" + "sync" "testing" "time" @@ -411,6 +412,159 @@ dashboard-address = "foo" } } +func TestUpdatePDServerConfigKeepsLatestFields(t *testing.T) { + re := require.New(t) + opt, err := newTestScheduleOption() + re.NoError(err) + storage := storage.NewStorageWithMemoryBackend() + + latest := opt.GetPDServerConfig().Clone() + latest.BlockSafePointV1 = true + opt.SetPDServerConfig(latest) + + re.NoError(opt.UpdatePDServerConfig(storage, func(cfg *PDServerConfig) error { + cfg.DashboardAddress = "none" + return nil + })) + + re.True(opt.GetPDServerConfig().BlockSafePointV1) + re.Equal("none", opt.GetPDServerConfig().DashboardAddress) + + reloaded, err := newTestScheduleOption() + re.NoError(err) + re.NoError(reloaded.Reload(storage)) + re.True(reloaded.GetPDServerConfig().BlockSafePointV1) + re.Equal("none", reloaded.GetPDServerConfig().DashboardAddress) +} + +func TestPersistSerializesFullConfigSaves(t *testing.T) { + re := require.New(t) + opt, err := newTestScheduleOption() + re.NoError(err) + + storage := newBlockingConfigStorage() + scheduleErrCh := make(chan error, 1) + go func() { + scheduleErrCh <- opt.UpdateScheduleConfig(storage, func(cfg *sc.ScheduleConfig) error { + cfg.MaxSnapshotCount++ + return nil + }) + }() + + releaseFirst := sync.OnceFunc(func() { + close(storage.releaseFirst) + }) + defer releaseFirst() + + select { + case <-storage.firstStarted: + case <-time.After(5 * time.Second): + re.FailNow("first config save did not start") + } + + replication := opt.GetReplicationConfig().Clone() + replication.MaxReplicas = 5 + opt.SetReplicationConfig(replication) + persistErrCh := make(chan error, 1) + go func() { + persistErrCh <- opt.Persist(storage) + }() + + select { + case <-storage.secondStarted: + re.FailNow("second config save started before the first save was released") + case <-time.After(100 * time.Millisecond): + } + releaseFirst() + + re.NoError(<-scheduleErrCh) + re.NoError(<-persistErrCh) + re.NoError(storage.err) + saved := storage.savedConfig() + re.NotNil(saved) + re.Equal(uint64(5), saved.Replication.MaxReplicas) +} + +type blockingConfigStorage struct { + mu sync.Mutex + + firstStarted chan struct{} + releaseFirst chan struct{} + secondStarted chan struct{} + + saveCount int + saved *persistedConfig + err error +} + +func newBlockingConfigStorage() *blockingConfigStorage { + return &blockingConfigStorage{ + firstStarted: make(chan struct{}), + releaseFirst: make(chan struct{}), + secondStarted: make(chan struct{}), + } +} + +func (*blockingConfigStorage) LoadConfig(any) (bool, error) { + return false, nil +} + +func (s *blockingConfigStorage) SaveConfig(cfg any) error { + s.mu.Lock() + s.saveCount++ + saveCount := s.saveCount + switch saveCount { + case 1: + close(s.firstStarted) + case 2: + close(s.secondStarted) + } + s.mu.Unlock() + + if saveCount == 1 { + <-s.releaseFirst + } + + data, err := json.Marshal(cfg) + if err == nil { + saved := &persistedConfig{Config: &Config{}} + err = json.Unmarshal(data, saved) + if err == nil { + s.mu.Lock() + s.saved = saved + s.mu.Unlock() + } + } + if err != nil { + s.mu.Lock() + s.err = err + s.mu.Unlock() + } + return err +} + +func (*blockingConfigStorage) LoadAllSchedulerConfigs() (names, configs []string, err error) { + return nil, nil, nil +} + +func (*blockingConfigStorage) LoadSchedulerConfig(string) (string, error) { + return "", nil +} + +func (*blockingConfigStorage) SaveSchedulerConfig(string, []byte) error { + return nil +} + +func (*blockingConfigStorage) RemoveSchedulerConfig(string) error { + return nil +} + +func (s *blockingConfigStorage) savedConfig() *persistedConfig { + s.mu.Lock() + defer s.mu.Unlock() + return s.saved +} + func TestDashboardConfig(t *testing.T) { re := require.New(t) cfgData := ` diff --git a/server/config/persist_options.go b/server/config/persist_options.go index 15cda11b139..cd4a31c3fbf 100644 --- a/server/config/persist_options.go +++ b/server/config/persist_options.go @@ -18,6 +18,7 @@ import ( "context" "reflect" "strconv" + "sync" "sync/atomic" "time" "unsafe" @@ -47,6 +48,7 @@ import ( type PersistOptions struct { // configuration -> ttl value ttl *cache.TTLString + configMu sync.Mutex schedule atomic.Value replication atomic.Value pdServerConfig atomic.Value @@ -83,9 +85,33 @@ func (o *PersistOptions) GetScheduleConfig() *sc.ScheduleConfig { // SetScheduleConfig sets the PD scheduling configuration. func (o *PersistOptions) SetScheduleConfig(cfg *sc.ScheduleConfig) { + o.configMu.Lock() + defer o.configMu.Unlock() + o.storeScheduleConfig(cfg) +} + +func (o *PersistOptions) storeScheduleConfig(cfg *sc.ScheduleConfig) { o.schedule.Store(cfg) } +// UpdateScheduleConfig updates and persists the latest scheduling configuration. +func (o *PersistOptions) UpdateScheduleConfig(storage endpoint.ConfigStorage, update func(*sc.ScheduleConfig) error) error { + o.configMu.Lock() + defer o.configMu.Unlock() + + old := o.GetScheduleConfig() + cfg := old.Clone() + if err := update(cfg); err != nil { + return err + } + o.storeScheduleConfig(cfg) + if err := o.persistLocked(storage); err != nil { + o.storeScheduleConfig(old) + return err + } + return nil +} + // GetReplicationConfig returns replication configurations. func (o *PersistOptions) GetReplicationConfig() *sc.ReplicationConfig { return o.replication.Load().(*sc.ReplicationConfig) @@ -103,9 +129,33 @@ func (o *PersistOptions) GetPDServerConfig() *PDServerConfig { // SetPDServerConfig sets the PD configuration. func (o *PersistOptions) SetPDServerConfig(cfg *PDServerConfig) { + o.configMu.Lock() + defer o.configMu.Unlock() + o.storePDServerConfig(cfg) +} + +func (o *PersistOptions) storePDServerConfig(cfg *PDServerConfig) { o.pdServerConfig.Store(cfg) } +// UpdatePDServerConfig updates and persists the latest PD server configuration. +func (o *PersistOptions) UpdatePDServerConfig(storage endpoint.ConfigStorage, update func(*PDServerConfig) error) error { + o.configMu.Lock() + defer o.configMu.Unlock() + + old := o.GetPDServerConfig() + cfg := old.Clone() + if err := update(cfg); err != nil { + return err + } + o.storePDServerConfig(cfg) + if err := o.persistLocked(storage); err != nil { + o.storePDServerConfig(old) + return err + } + return nil +} + // GetReplicationModeConfig returns the replication mode config. func (o *PersistOptions) GetReplicationModeConfig() *ReplicationModeConfig { return o.replicationMode.Load().(*ReplicationModeConfig) @@ -706,6 +756,9 @@ func (o *PersistOptions) GetHotRegionsReservedDays() uint64 { // AddSchedulerCfg adds the scheduler configurations. func (o *PersistOptions) AddSchedulerCfg(tp types.CheckerSchedulerType, args []string) { + o.configMu.Lock() + defer o.configMu.Unlock() + oldType := types.SchedulerTypeCompatibleMap[tp] v := o.GetScheduleConfig().Clone() for i, schedulerCfg := range v.Schedulers { @@ -719,16 +772,19 @@ func (o *PersistOptions) AddSchedulerCfg(tp types.CheckerSchedulerType, args []s if reflect.DeepEqual(schedulerCfg, sc.SchedulerConfig{Type: oldType, Args: args, Disable: true}) { schedulerCfg.Disable = false v.Schedulers[i] = schedulerCfg - o.SetScheduleConfig(v) + o.storeScheduleConfig(v) return } } v.Schedulers = append(v.Schedulers, sc.SchedulerConfig{Type: oldType, Args: args, Disable: false}) - o.SetScheduleConfig(v) + o.storeScheduleConfig(v) } // RemoveSchedulerCfg removes the scheduler configurations. func (o *PersistOptions) RemoveSchedulerCfg(tp types.CheckerSchedulerType) { + o.configMu.Lock() + defer o.configMu.Unlock() + oldType := types.SchedulerTypeCompatibleMap[tp] v := o.GetScheduleConfig().Clone() for i, schedulerCfg := range v.Schedulers { @@ -739,7 +795,7 @@ func (o *PersistOptions) RemoveSchedulerCfg(tp types.CheckerSchedulerType) { } else { v.Schedulers = append(v.Schedulers[:i], v.Schedulers[i+1:]...) } - o.SetScheduleConfig(v) + o.storeScheduleConfig(v) return } } @@ -783,12 +839,20 @@ type persistedConfig struct { // SwitchRaftV2 update some config if tikv raft engine switch into partition raft v2 func (o *PersistOptions) SwitchRaftV2(storage endpoint.ConfigStorage) error { - o.GetScheduleConfig().StoreLimitVersion = "v2" - return o.Persist(storage) + return o.UpdateScheduleConfig(storage, func(cfg *sc.ScheduleConfig) error { + cfg.StoreLimitVersion = "v2" + return nil + }) } // Persist saves the configuration to the storage. func (o *PersistOptions) Persist(storage endpoint.ConfigStorage) error { + o.configMu.Lock() + defer o.configMu.Unlock() + return o.persistLocked(storage) +} + +func (o *PersistOptions) persistLocked(storage endpoint.ConfigStorage) error { cfg := &persistedConfig{ Config: &Config{ Schedule: *o.GetScheduleConfig(), @@ -985,6 +1049,16 @@ func (*PersistOptions) SetSchedulingAllowanceStatus(halt bool, source string) { } } +// SetSchedulingAllowanceStatusIfCurrent sets the scheduling allowance status if the halt value is still current. +func (o *PersistOptions) SetSchedulingAllowanceStatusIfCurrent(halt bool, source string) { + o.configMu.Lock() + defer o.configMu.Unlock() + if o.GetScheduleConfig().HaltScheduling != halt { + return + } + o.SetSchedulingAllowanceStatus(halt, source) +} + // SetHaltScheduling set HaltScheduling. func (o *PersistOptions) SetHaltScheduling(halt bool, source string) { v := o.GetScheduleConfig().Clone() diff --git a/server/server.go b/server/server.go index 7572fa1c776..7adebb4d546 100644 --- a/server/server.go +++ b/server/server.go @@ -1264,25 +1264,46 @@ func (s *Server) GetScheduleConfig() *sc.ScheduleConfig { // SetScheduleConfig sets the balance config information. // This function is exported to be used by the API. func (s *Server) SetScheduleConfig(cfg sc.ScheduleConfig) error { - if err := cfg.Validate(); err != nil { - return err - } - if err := cfg.Deprecated(); err != nil { - return err - } - old := s.persistOptions.GetScheduleConfig() - s.persistOptions.SetScheduleConfig(&cfg) - if err := s.persistOptions.Persist(s.storage); err != nil { - s.persistOptions.SetScheduleConfig(old) - log.Error("failed to update schedule config", - zap.Reflect("new", cfg), - zap.Reflect("old", old), - errs.ZapError(err)) + return s.UpdateScheduleConfig(func(current *sc.ScheduleConfig) error { + *current = *cfg.Clone() + return nil + }) +} + +// UpdateScheduleConfig updates the latest schedule config and persists it atomically. +func (s *Server) UpdateScheduleConfig(update func(*sc.ScheduleConfig) error) error { + var ( + old *sc.ScheduleConfig + updated *sc.ScheduleConfig + persisting bool + ) + err := s.persistOptions.UpdateScheduleConfig(s.storage, func(cfg *sc.ScheduleConfig) error { + old = s.persistOptions.GetScheduleConfig() + if err := update(cfg); err != nil { + return err + } + if err := cfg.Validate(); err != nil { + return err + } + if err := cfg.Deprecated(); err != nil { + return err + } + updated = cfg.Clone() + persisting = true + return nil + }) + if err != nil { + if persisting { + log.Error("failed to update schedule config", + zap.Reflect("new", updated), + zap.Reflect("old", old), + errs.ZapError(err)) + } return err } // Update the scheduling halt status at the same time. - s.persistOptions.SetSchedulingAllowanceStatus(cfg.HaltScheduling, "manually") - log.Info("schedule config is updated", zap.Reflect("new", cfg), zap.Reflect("old", old)) + s.persistOptions.SetSchedulingAllowanceStatusIfCurrent(updated.HaltScheduling, "manually") + log.Info("schedule config is updated", zap.Reflect("new", updated), zap.Reflect("old", old)) return nil } @@ -1511,34 +1532,75 @@ func (s *Server) GetPDServerConfig() *config.PDServerConfig { // SetPDServerConfig sets the server config. func (s *Server) SetPDServerConfig(cfg config.PDServerConfig) error { + return s.UpdatePDServerConfig(func(current *config.PDServerConfig) error { + *current = *cfg.Clone() + return nil + }) +} + +// SetDashboardAddress updates the dashboard address without overwriting other PD server config fields. +func (s *Server) SetDashboardAddress(addr string) error { + return s.UpdatePDServerConfig(func(cfg *config.PDServerConfig) error { + cfg.DashboardAddress = addr + return nil + }) +} + +// UpdatePDServerConfig updates the latest PD server config and persists it atomically. +func (s *Server) UpdatePDServerConfig(update func(*config.PDServerConfig) error) error { + var ( + old *config.PDServerConfig + updated *config.PDServerConfig + persisting bool + ) + err := s.persistOptions.UpdatePDServerConfig(s.storage, func(cfg *config.PDServerConfig) error { + old = s.persistOptions.GetPDServerConfig() + if err := update(cfg); err != nil { + return err + } + if err := s.preparePDServerConfig(old, cfg); err != nil { + return err + } + updated = cfg.Clone() + persisting = true + return nil + }) + if err != nil { + if persisting { + log.Error("failed to update PDServer config", + zap.Reflect("new", updated), + zap.Reflect("old", old), + errs.ZapError(err)) + } + return err + } + log.Info("PD server config is updated", zap.Reflect("new", updated), zap.Reflect("old", old)) + return nil +} + +func (s *Server) preparePDServerConfig(old, cfg *config.PDServerConfig) error { + switch cfg.DashboardAddress { + case "auto": + case "none": + default: + if !strings.HasPrefix(cfg.DashboardAddress, "http") { + cfg.DashboardAddress = fmt.Sprintf("%s://%s", s.GetClientScheme(), cfg.DashboardAddress) + } + } if err := cfg.Validate(); err != nil { return err } - old := s.persistOptions.GetPDServerConfig() - // See https://github.com/tikv/pd/issues/10114 for more details + // See https://github.com/tikv/pd/issues/10114 for more details. if old.DashboardAddress != cfg.DashboardAddress { switch cfg.DashboardAddress { case "auto": case "none": default: - if !strings.HasPrefix(cfg.DashboardAddress, "http") { - cfg.DashboardAddress = fmt.Sprintf("%s://%s", s.GetClientScheme(), cfg.DashboardAddress) - } if !cluster.IsClientURL(cfg.DashboardAddress, s.client) { return errors.Errorf("dashboard address %s is not the client url of any member", cfg.DashboardAddress) } } } - s.persistOptions.SetPDServerConfig(&cfg) - if err := s.persistOptions.Persist(s.storage); err != nil { - s.persistOptions.SetPDServerConfig(old) - log.Error("failed to update PDServer config", - zap.Reflect("new", cfg), - zap.Reflect("old", old), - errs.ZapError(err)) - return err - } - log.Info("PD server config is updated", zap.Reflect("new", cfg), zap.Reflect("old", old)) return nil } diff --git a/tests/server/config/config_test.go b/tests/server/config/config_test.go index 86e57026a06..19076caf97c 100644 --- a/tests/server/config/config_test.go +++ b/tests/server/config/config_test.go @@ -28,6 +28,8 @@ import ( "github.com/stretchr/testify/suite" "go.uber.org/goleak" + "github.com/pingcap/failpoint" + cfg "github.com/tikv/pd/pkg/mcs/scheduling/server/config" "github.com/tikv/pd/pkg/ratelimit" sc "github.com/tikv/pd/pkg/schedule/config" @@ -106,6 +108,33 @@ func (suite *configTestSuite) TestKeyspaceConfigUpdate() { suite.env.RunTest(suite.checkKeyspaceConfigUpdate) } +func (suite *configTestSuite) TestNoopConfigUpdateSkipsPersist() { + suite.env.RunTest(suite.checkNoopConfigUpdateSkipsPersist) +} + +func (suite *configTestSuite) checkNoopConfigUpdateSkipsPersist(cluster *tests.TestCluster) { + re := suite.Require() + leaderServer := cluster.GetLeaderServer() + addr := fmt.Sprintf("%s/pd/api/v1/config", leaderServer.GetAddr()) + + re.NoError(failpoint.Enable("github.com/tikv/pd/server/config/persistFail", "return(true)")) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/server/config/persistFail")) + }() + + postData, err := json.Marshal(map[string]any{ + "schedule.region-schedule-limit": leaderServer.GetServer().GetScheduleConfig().RegionScheduleLimit, + }) + re.NoError(err) + re.NoError(testutil.CheckPostJSON(tests.TestDialClient, addr, postData, testutil.StatusOK(re))) + + postData, err = json.Marshal(map[string]any{ + "pd-server.dashboard-address": leaderServer.GetServer().GetPDServerConfig().DashboardAddress, + }) + re.NoError(err) + re.NoError(testutil.CheckPostJSON(tests.TestDialClient, addr, postData, testutil.StatusOK(re))) +} + func (suite *configTestSuite) checkKeyspaceConfigUpdate(cluster *tests.TestCluster) { re := suite.Require() leaderServer := cluster.GetLeaderServer() diff --git a/tests/server/server_test.go b/tests/server/server_test.go index a579531e156..7c19e687ccd 100644 --- a/tests/server/server_test.go +++ b/tests/server/server_test.go @@ -21,6 +21,7 @@ import ( "io" "net/http" "path/filepath" + "strings" "sync" "testing" @@ -570,7 +571,15 @@ func TestSetPDServerConfigWithDashboard(t *testing.T) { re.Equal(originalDashboard, newCfg.DashboardAddress) re.NotEqual(originalUseRegionStorage, newCfg.UseRegionStorage) + // The dynamic config API normalizes a bare dashboard address before validation. + cfg = svr.GetPDServerConfig() + cfg.DashboardAddress = strings.TrimPrefix(svr.GetAddr(), svr.GetClientScheme()+"://") + err = svr.SetPDServerConfig(*cfg) + re.NoError(err) + re.Equal(svr.GetAddr(), svr.GetPDServerConfig().DashboardAddress) + // Change both other field and dashboard + cfg = svr.GetPDServerConfig() cfg.UseRegionStorage = !cfg.UseRegionStorage cfg.DashboardAddress = "https://new-dashboard-address:1234" err = svr.SetPDServerConfig(*cfg)