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 466d094 commit 0022ac0
Show file tree
Hide file tree
Showing 5 changed files with 74 additions and 988 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 @@ -73,6 +73,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 @@ -143,25 +146,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 @@ -363,7 +348,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.mu.RLock()
if _, exists = handler.config.StoreIDWithRanges[id]; !exists {
Expand All @@ -372,46 +356,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.mu.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.mu.Lock()
handler.config.cluster.ResumeLeaderTransfer(id)
handler.config.mu.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 @@ -267,18 +267,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.mu.Lock()
handler.config.cluster.ResumeLeaderTransfer(id)
handler.config.mu.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
59 changes: 59 additions & 0 deletions tests/pdctl/scheduler/scheduler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"testing"
"time"

"github.com/pingcap/failpoint"
"github.com/pingcap/kvproto/pkg/metapb"
"github.com/stretchr/testify/require"
"github.com/tikv/pd/pkg/utils/testutil"
Expand Down Expand Up @@ -482,3 +483,61 @@ func TestScheduler(t *testing.T) {
re.NoError(err)
checkSchedulerWithStatusCommand(nil, "disabled", nil)
}

func TestEvictLeaderScheduler(t *testing.T) {
re := require.New(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cluster, err := tests.NewTestCluster(ctx, 1)
re.NoError(err)
defer cluster.Destroy()
err = cluster.RunInitialServers()
re.NoError(err)
cluster.WaitLeader()
pdAddr := cluster.GetConfig().GetClientURL()
cmd := pdctlCmd.GetRootCmd()

stores := []*metapb.Store{
{
Id: 1,
State: metapb.StoreState_Up,
LastHeartbeat: time.Now().UnixNano(),
},
{
Id: 2,
State: metapb.StoreState_Up,
LastHeartbeat: time.Now().UnixNano(),
},
{
Id: 3,
State: metapb.StoreState_Up,
LastHeartbeat: time.Now().UnixNano(),
},
{
Id: 4,
State: metapb.StoreState_Up,
LastHeartbeat: time.Now().UnixNano(),
},
}
leaderServer := cluster.GetServer(cluster.GetLeader())
re.NoError(leaderServer.BootstrapCluster())
for _, store := range stores {
pdctl.MustPutStore(re, leaderServer.GetServer(), store)
}

pdctl.MustPutRegion(re, cluster, 1, 1, []byte("a"), []byte("b"))
output, err := pdctl.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 = pdctl.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "add", "evict-leader-scheduler", "1"}...)
re.NoError(err)
re.Contains(string(output), "fail to build with args")
failpoint.Disable("github.com/tikv/pd/pkg/schedule/schedulers/buildWithArgsErr")
output, err = pdctl.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "remove", "evict-leader-scheduler"}...)
re.NoError(err)
re.Contains(string(output), "Success!")
output, err = pdctl.ExecuteCommand(cmd, []string{"-u", pdAddr, "scheduler", "add", "evict-leader-scheduler", "1"}...)
re.NoError(err)
re.Contains(string(output), "Success!")
}
Loading

0 comments on commit 0022ac0

Please sign in to comment.