Skip to content

Commit

Permalink
resolve conflicts
Browse files Browse the repository at this point in the history
Signed-off-by: Ryan Leung <[email protected]>
  • Loading branch information
rleungx committed Sep 25, 2024
1 parent 84049b0 commit 7b95065
Show file tree
Hide file tree
Showing 4 changed files with 56 additions and 86 deletions.
53 changes: 9 additions & 44 deletions pkg/schedule/schedulers/evict_leader.go
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,9 @@ func (conf *evictLeaderSchedulerConfig) getStores() []uint64 {
}

func (conf *evictLeaderSchedulerConfig) BuildWithArgs(args []string) error {
failpoint.Inject("buildWithArgsErr", func() {
failpoint.Return(errors.New("fail to build with args"))
})
if len(args) != 1 {
return errs.ErrSchedulerConfig.FastGenByArgs("id")
}
Expand Down Expand Up @@ -144,25 +147,7 @@ func (conf *evictLeaderSchedulerConfig) removeStore(id uint64) (succ bool, last
succ = true
last = len(conf.StoreIDWithRanges) == 0
}
<<<<<<< HEAD
return succ, last
=======
return false, errs.ErrScheduleConfigNotExist.FastGenByArgs()
}

func (conf *evictLeaderSchedulerConfig) removeStore(id uint64) {
conf.Lock()
defer conf.Unlock()
// if the store is not existed, no need to resume leader transfer
_, _ = conf.removeStoreLocked(id)
}

func (conf *evictLeaderSchedulerConfig) resetStoreLocked(id uint64, keyRange []core.KeyRange) {
if err := conf.cluster.PauseLeaderTransfer(id); err != nil {
log.Error("pause leader transfer failed", zap.Uint64("store-id", id), errs.ZapError(err))
}
conf.StoreIDWithRanges[id] = keyRange
>>>>>>> 6b927e117 (*: reset config if the input is invalid (#8632))
}

func (conf *evictLeaderSchedulerConfig) resetStore(id uint64, keyRange []core.KeyRange) {
Expand Down Expand Up @@ -383,7 +368,6 @@ func (handler *evictLeaderHandler) UpdateConfig(w http.ResponseWriter, r *http.R
var id uint64
idFloat, ok := input["store_id"].(float64)
if ok {
<<<<<<< HEAD
id = (uint64)(idFloat)
handler.config.RLock()
if _, exists = handler.config.StoreIDWithRanges[id]; !exists {
Expand All @@ -392,46 +376,27 @@ func (handler *evictLeaderHandler) UpdateConfig(w http.ResponseWriter, r *http.R
handler.rd.JSON(w, http.StatusInternalServerError, err.Error())
return
}
=======
if batchFloat < 1 || batchFloat > 10 {
handler.config.removeStore(id)
handler.rd.JSON(w, http.StatusBadRequest, "batch is invalid, it should be in [1, 10]")
return
>>>>>>> 6b927e117 (*: reset config if the input is invalid (#8632))
}
handler.config.RUnlock()
args = append(args, strconv.FormatUint(id, 10))
}

ranges, ok := (input["ranges"]).([]string)
if ok {
<<<<<<< HEAD
args = append(args, ranges...)
} else if exists {
args = append(args, handler.config.getRanges(id)...)
}

handler.config.BuildWithArgs(args)
err := handler.config.Persist()
=======
if !inputHasStoreID {
handler.config.removeStore(id)
handler.rd.JSON(w, http.StatusInternalServerError, errs.ErrSchedulerConfig.FastGenByArgs("id"))
return
}
} else if exist {
ranges = handler.config.getRanges(id)
}

newRanges, err = getKeyRanges(ranges)
err := handler.config.BuildWithArgs(args)
if err != nil {
handler.config.removeStore(id)
handler.rd.JSON(w, http.StatusInternalServerError, err.Error())
handler.config.Lock()
handler.config.cluster.ResumeLeaderTransfer(id)
handler.config.Unlock()
handler.rd.JSON(w, http.StatusBadRequest, err.Error())
return
}

err = handler.config.update(id, newRanges, batch)
>>>>>>> 6b927e117 (*: reset config if the input is invalid (#8632))
err = handler.config.Persist()
if err != nil {
handler.config.removeStore(id)
handler.rd.JSON(w, http.StatusInternalServerError, err.Error())
Expand Down
13 changes: 5 additions & 8 deletions pkg/schedule/schedulers/grant_leader.go
Original file line number Diff line number Diff line change
Expand Up @@ -296,18 +296,15 @@ func (handler *grantLeaderHandler) UpdateConfig(w http.ResponseWriter, r *http.R
args = append(args, handler.config.getRanges(id)...)
}

<<<<<<< HEAD
handler.config.BuildWithArgs(args)
err := handler.config.Persist()
=======
err := handler.config.buildWithArgs(args)
err := handler.config.BuildWithArgs(args)
if err != nil {
_, _ = handler.config.removeStore(id)
handler.config.Lock()
handler.config.cluster.ResumeLeaderTransfer(id)
handler.config.Unlock()
handler.rd.JSON(w, http.StatusBadRequest, err.Error())
return
}
err = handler.config.persist()
>>>>>>> 6b927e117 (*: reset config if the input is invalid (#8632))
err = handler.config.Persist()
if err != nil {
_, _ = handler.config.removeStore(id)
handler.rd.JSON(w, http.StatusInternalServerError, err.Error())
Expand Down
28 changes: 1 addition & 27 deletions plugin/scheduler_example/evict_leader.go
Original file line number Diff line number Diff line change
Expand Up @@ -275,10 +275,6 @@ func (handler *evictLeaderHandler) UpdateConfig(w http.ResponseWriter, r *http.R
args = append(args, handler.config.getRanges(id)...)
}

<<<<<<< HEAD
handler.config.BuildWithArgs(args)
err := handler.config.Persist()
=======
err := handler.config.BuildWithArgs(args)
if err != nil {
handler.config.mu.Lock()
Expand All @@ -287,8 +283,8 @@ func (handler *evictLeaderHandler) UpdateConfig(w http.ResponseWriter, r *http.R
handler.rd.JSON(w, http.StatusBadRequest, err.Error())
return
}

err = handler.config.Persist()
>>>>>>> 6b927e117 (*: reset config if the input is invalid (#8632))
if err != nil {
handler.config.mu.Lock()
delete(handler.config.StoreIDWitRanges, id)
Expand All @@ -314,7 +310,6 @@ func (handler *evictLeaderHandler) DeleteConfig(w http.ResponseWriter, r *http.R

handler.config.mu.Lock()
defer handler.config.mu.Unlock()
<<<<<<< HEAD
_, exists := handler.config.StoreIDWitRanges[id]
if exists {
delete(handler.config.StoreIDWitRanges, id)
Expand All @@ -333,27 +328,6 @@ func (handler *evictLeaderHandler) DeleteConfig(w http.ResponseWriter, r *http.R
}

handler.rd.JSON(w, http.StatusInternalServerError, errors.New("the config does not exist"))
=======
ranges, exists := handler.config.StoreIDWitRanges[id]
if !exists {
handler.rd.JSON(w, http.StatusInternalServerError, errors.New("the config does not exist"))
return
}
delete(handler.config.StoreIDWitRanges, id)
handler.config.cluster.ResumeLeaderTransfer(id)

if err := handler.config.Persist(); err != nil {
handler.config.StoreIDWitRanges[id] = ranges
_ = handler.config.cluster.PauseLeaderTransfer(id)
handler.rd.JSON(w, http.StatusInternalServerError, err.Error())
return
}
var resp any
if len(handler.config.StoreIDWitRanges) == 0 {
resp = noStoreInSchedulerInfo
}
handler.rd.JSON(w, http.StatusOK, resp)
>>>>>>> 6b927e117 (*: reset config if the input is invalid (#8632))
}

func newEvictLeaderHandler(config *evictLeaderSchedulerConfig) http.Handler {
Expand Down
48 changes: 41 additions & 7 deletions tools/pd-ctl/tests/scheduler/scheduler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ func (suite *schedulerTestSuite) TearDownTest() {
if slice.NoneOf(currentSchedulers, func(i int) bool {
return currentSchedulers[i] == scheduler
}) {
fmt.Println("xxxadd", scheduler)
echo := mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "add", scheduler}, nil)
re.Contains(echo, "Success!")
}
Expand Down Expand Up @@ -661,7 +662,7 @@ func (suite *schedulerTestSuite) checkScheduler(cluster *pdTests.TestCluster) {
}

func (suite *schedulerTestSuite) TestSchedulerDiagnostic() {
suite.env.RunTestInTwoModes(suite.checkSchedulerDiagnostic)
suite.env.RunTestInAPIMode(suite.checkSchedulerDiagnostic)
}

func (suite *schedulerTestSuite) checkSchedulerDiagnostic(cluster *pdTests.TestCluster) {
Expand Down Expand Up @@ -723,8 +724,7 @@ func (suite *schedulerTestSuite) checkSchedulerDiagnostic(cluster *pdTests.TestC
}

func (suite *schedulerTestSuite) TestEvictLeaderScheduler() {
// FIXME: API mode may have the problem
suite.env.RunTestInPDMode(suite.checkEvictLeaderScheduler)
suite.env.RunFuncInTwoModes(suite.checkEvictLeaderScheduler)
}

func (suite *schedulerTestSuite) checkEvictLeaderScheduler(cluster *pdTests.TestCluster) {
Expand Down Expand Up @@ -759,18 +759,23 @@ func (suite *schedulerTestSuite) checkEvictLeaderScheduler(cluster *pdTests.Test
}

pdTests.MustPutRegion(re, cluster, 1, 1, []byte("a"), []byte("b"))
suite.checkDefaultSchedulers(re, cmd, pdAddr)

output, err := tests.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "add", "evict-leader-scheduler", "2"}...)
re.NoError(err)
re.Contains(string(output), "Success!")
failpoint.Enable("github.com/tikv/pd/pkg/schedule/schedulers/buildWithArgsErr", "return(true)")
output, err = tests.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "add", "evict-leader-scheduler", "1"}...)
re.NoError(err)
re.Contains(string(output), "Success!")
re.Contains(string(output), "fail to build with args")
failpoint.Disable("github.com/tikv/pd/pkg/schedule/schedulers/buildWithArgsErr")
output, err = tests.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "remove", "evict-leader-scheduler"}...)
re.NoError(err)
re.Contains(string(output), "Success!")
output, err = tests.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "add", "evict-leader-scheduler", "1"}...)
re.NoError(err)
re.Contains(string(output), "Success!")
testutil.Eventually(re, func() bool {
output, err = tests.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "add", "evict-leader-scheduler", "1"}...)
return err == nil && strings.Contains(string(output), "Success!")
})
output, err = tests.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "remove", "evict-leader-scheduler-1"}...)
re.NoError(err)
re.Contains(string(output), "Success!")
Expand All @@ -794,3 +799,32 @@ func mightExec(re *require.Assertions, cmd *cobra.Command, args []string, v any)
}
json.Unmarshal(output, v)
}

func (suite *schedulerTestSuite) checkDefaultSchedulers(re *require.Assertions, cmd *cobra.Command, pdAddr string) {
// scheduler show command
expected := make(map[string]bool)
for _, scheduler := range suite.defaultSchedulers {
expected[scheduler] = true
}
checkSchedulerCommand(re, cmd, pdAddr, nil, expected)
}

func checkSchedulerCommand(re *require.Assertions, cmd *cobra.Command, pdAddr string, args []string, expected map[string]bool) {
if args != nil {
echo := mustExec(re, cmd, args, nil)
re.Contains(echo, "Success!")
}
testutil.Eventually(re, func() bool {
var schedulers []string
mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "show"}, &schedulers)
if len(schedulers) != len(expected) {
return false
}
for _, scheduler := range schedulers {
if _, ok := expected[scheduler]; !ok {
return false
}
}
return true
})
}

0 comments on commit 7b95065

Please sign in to comment.