Skip to content

Commit

Permalink
operator: check store status for running operators (#4223) (#4368)
Browse files Browse the repository at this point in the history 
* operator: check store status for running operators

close #3353

Signed-off-by: disksing <i@disksing.com>

* add test

Signed-off-by: disksing <i@disksing.com>

* add tests

Signed-off-by: disksing <i@disksing.com>

* address comment

Signed-off-by: disksing <i@disksing.com>

* fix build

Signed-off-by: disksing <i@disksing.com>

* fix ci (try)

Signed-off-by: disksing <i@disksing.com>

Co-authored-by: disksing <i@disksing.com>
  • Loading branch information
@ti-chi-bot @disksing
ti-chi-bot and disksing authored Dec 1, 2021
1 parent 9615a4a commit 8f64708
Show file tree
Hide file tree
Showing 4 changed files with 218 additions and 30 deletions.
76 changes: 50 additions & 26 deletions server/schedule/operator/step.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import (
"github.com/tikv/pd/pkg/typeutil"
"github.com/tikv/pd/server/core"
"github.com/tikv/pd/server/core/storelimit"
"github.com/tikv/pd/server/schedule/opt"
"go.uber.org/zap"
)

Expand All @@ -34,7 +35,7 @@ type OpStep interface {
fmt.Stringer
ConfVerChanged(region *core.RegionInfo) uint64
IsFinish(region *core.RegionInfo) bool
CheckSafety(region *core.RegionInfo) error
CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error
Influence(opInfluence OpInfluence, region *core.RegionInfo)
}

Expand All @@ -57,16 +58,16 @@ func (tl TransferLeader) IsFinish(region *core.RegionInfo) bool {
return region.GetLeader().GetStoreId() == tl.ToStore
}

// CheckSafety checks if the step meets the safety properties.
func (tl TransferLeader) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (tl TransferLeader) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
peer := region.GetStorePeer(tl.ToStore)
if peer == nil {
return errors.New("peer does not existed")
}
if core.IsLearner(peer) {
return errors.New("peer already is a learner")
}
return nil
return validateStore(cluster, tl.ToStore)
}

// Influence calculates the store difference that current step makes.
Expand Down Expand Up @@ -117,8 +118,11 @@ func (ap AddPeer) Influence(opInfluence OpInfluence, region *core.RegionInfo) {
to.AdjustStepCost(storelimit.AddPeer, regionSize)
}

// CheckSafety checks if the step meets the safety properties.
func (ap AddPeer) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (ap AddPeer) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
if err := validateStore(cluster, ap.ToStore); err != nil {
return err
}
peer := region.GetStorePeer(ap.ToStore)
if peer != nil && peer.GetId() != ap.PeerID {
return errors.Errorf("peer %d has already existed in store %d, the operator is trying to add peer %d on the same store", peer.GetId(), ap.ToStore, ap.PeerID)
Expand Down Expand Up @@ -153,8 +157,11 @@ func (al AddLearner) IsFinish(region *core.RegionInfo) bool {
return false
}

// CheckSafety checks if the step meets the safety properties.
func (al AddLearner) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (al AddLearner) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
if err := validateStore(cluster, al.ToStore); err != nil {
return err
}
peer := region.GetStorePeer(al.ToStore)
if peer == nil {
return nil
Expand Down Expand Up @@ -204,8 +211,8 @@ func (pl PromoteLearner) IsFinish(region *core.RegionInfo) bool {
return false
}

// CheckSafety checks if the step meets the safety properties.
func (pl PromoteLearner) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (pl PromoteLearner) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
peer := region.GetStorePeer(pl.ToStore)
if peer.GetId() != pl.PeerID {
return errors.New("peer does not exist")
Expand Down Expand Up @@ -245,8 +252,8 @@ func (rp RemovePeer) IsFinish(region *core.RegionInfo) bool {
return region.GetStorePeer(rp.FromStore) == nil
}

// CheckSafety checks if the step meets the safety properties.
func (rp RemovePeer) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (rp RemovePeer) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
if rp.FromStore == region.GetLeader().GetStoreId() {
return errors.New("cannot remove leader peer")
}
Expand Down Expand Up @@ -297,8 +304,8 @@ func (mr MergeRegion) IsFinish(region *core.RegionInfo) bool {
return false
}

// CheckSafety checks if the step meets the safety properties.
func (mr MergeRegion) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (mr MergeRegion) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
return nil
}

Expand Down Expand Up @@ -347,8 +354,8 @@ func (sr SplitRegion) Influence(opInfluence OpInfluence, region *core.RegionInfo
}
}

// CheckSafety checks if the step meets the safety properties.
func (sr SplitRegion) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (sr SplitRegion) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
return nil
}

Expand Down Expand Up @@ -379,8 +386,11 @@ func (ap AddLightPeer) IsFinish(region *core.RegionInfo) bool {
return false
}

// CheckSafety checks if the step meets the safety properties.
func (ap AddLightPeer) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step meets the safety properties.
func (ap AddLightPeer) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
if err := validateStore(cluster, ap.ToStore); err != nil {
return err
}
peer := region.GetStorePeer(ap.ToStore)
if peer != nil && peer.GetId() != ap.PeerID {
return errors.Errorf("peer %d has already existed in store %d, the operator is trying to add peer %d on the same store", peer.GetId(), ap.ToStore, ap.PeerID)
Expand Down Expand Up @@ -423,8 +433,11 @@ func (al AddLightLearner) IsFinish(region *core.RegionInfo) bool {
return false
}

// CheckSafety checks if the step meets the safety properties.
func (al AddLightLearner) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step meets the safety properties.
func (al AddLightLearner) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
if err := validateStore(cluster, al.ToStore); err != nil {
return err
}
peer := region.GetStorePeer(al.ToStore)
if peer == nil {
return nil
Expand Down Expand Up @@ -472,8 +485,8 @@ func (df DemoteFollower) IsFinish(region *core.RegionInfo) bool {
return false
}

// CheckSafety checks if the step meets the safety properties.
func (df DemoteFollower) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (df DemoteFollower) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
peer := region.GetStorePeer(df.ToStore)
if peer.GetId() != df.PeerID {
return errors.New("peer does not exist")
Expand Down Expand Up @@ -572,8 +585,8 @@ func (cpe ChangePeerV2Enter) IsFinish(region *core.RegionInfo) bool {
return true
}

// CheckSafety checks if the step meets the safety properties.
func (cpe ChangePeerV2Enter) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (cpe ChangePeerV2Enter) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
inJointState, notInJointState := false, false
for _, pl := range cpe.PromoteLearners {
peer := region.GetStorePeer(pl.ToStore)
Expand Down Expand Up @@ -712,8 +725,8 @@ func (cpl ChangePeerV2Leave) IsFinish(region *core.RegionInfo) bool {
return true
}

// CheckSafety checks if the step meets the safety properties.
func (cpl ChangePeerV2Leave) CheckSafety(region *core.RegionInfo) error {
// CheckInProgress checks if the step is in the progress of advancing.
func (cpl ChangePeerV2Leave) CheckInProgress(cluster opt.Cluster, region *core.RegionInfo) error {
inJointState, notInJointState, demoteLeader := false, false, false
leaderStoreID := region.GetLeader().GetStoreId()

Expand Down Expand Up @@ -773,3 +786,14 @@ func (cpl ChangePeerV2Leave) CheckSafety(region *core.RegionInfo) error {

// Influence calculates the store difference that current step makes.
func (cpl ChangePeerV2Leave) Influence(opInfluence OpInfluence, region *core.RegionInfo) {}

func validateStore(cluster opt.Cluster, id uint64) error {
store := cluster.GetStore(id)
if store == nil {
return errors.New("target store does not exist")
}
if store.DownTime() > cluster.GetOpts().GetMaxStoreDownTime() {
return errors.New("target store is down")
}
return nil
}
148 changes: 145 additions & 3 deletions server/schedule/operator/step_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,20 +14,162 @@
package operator

import (
"context"

. "github.com/pingcap/check"
"github.com/pingcap/kvproto/pkg/metapb"
"github.com/tikv/pd/pkg/mock/mockcluster"
"github.com/tikv/pd/server/config"
"github.com/tikv/pd/server/core"
)

type testStepSuite struct{}
type testStepSuite struct {
cluster *mockcluster.Cluster
}

var _ = Suite(&testStepSuite{})

type testCase struct {
Peers []*metapb.Peer // first is leader
ConfVerChanged uint64
IsFinish bool
CheckSafety Checker
CheckInProgres Checker
}

func (s *testStepSuite) SetUpTest(c *C) {
s.cluster = mockcluster.NewCluster(context.Background(), config.NewTestOptions())
for i := 1; i <= 10; i++ {
s.cluster.PutStoreWithLabels(uint64(i))
}
s.cluster.SetStoreDown(8)
s.cluster.SetStoreDown(9)
s.cluster.SetStoreDown(10)
}

func (s *testStepSuite) TestTransferLeader(c *C) {
step := TransferLeader{FromStore: 1, ToStore: 2}
cases := []testCase{
{
[]*metapb.Peer{
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
{Id: 3, StoreId: 3, Role: metapb.PeerRole_Voter},
},
0,
false,
IsNil,
},
{
[]*metapb.Peer{
{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
{Id: 3, StoreId: 3, Role: metapb.PeerRole_Voter},
},
0,
true,
IsNil,
},
{
[]*metapb.Peer{
{Id: 3, StoreId: 3, Role: metapb.PeerRole_Voter},
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
},
0,
false,
IsNil,
},
}
s.check(c, step, "transfer leader from store 1 to store 2", cases)

step = TransferLeader{FromStore: 1, ToStore: 9} // 9 is down
cases = []testCase{
{
[]*metapb.Peer{
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
{Id: 9, StoreId: 9, Role: metapb.PeerRole_Voter},
},
0,
false,
NotNil,
},
}
s.check(c, step, "transfer leader from store 1 to store 9", cases)
}

func (s *testStepSuite) TestAddPeer(c *C) {
step := AddPeer{ToStore: 2, PeerID: 2}
cases := []testCase{
{
[]*metapb.Peer{
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
},
0,
false,
IsNil,
},
{
[]*metapb.Peer{
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
{Id: 2, StoreId: 2, Role: metapb.PeerRole_Voter},
},
1,
true,
IsNil,
},
}
s.check(c, step, "add peer 2 on store 2", cases)

step = AddPeer{ToStore: 9, PeerID: 9}
cases = []testCase{
{
[]*metapb.Peer{
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
},
0,
false,
NotNil,
},
}
s.check(c, step, "add peer 9 on store 9", cases)
}

func (s *testStepSuite) TestAddLearner(c *C) {
step := AddLearner{ToStore: 2, PeerID: 2}
cases := []testCase{
{
[]*metapb.Peer{
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
},
0,
false,
IsNil,
},
{
[]*metapb.Peer{
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
{Id: 2, StoreId: 2, Role: metapb.PeerRole_Learner},
},
1,
true,
IsNil,
},
}
s.check(c, step, "add learner peer 2 on store 2", cases)

step = AddLearner{ToStore: 9, PeerID: 9}
cases = []testCase{
{
[]*metapb.Peer{
{Id: 1, StoreId: 1, Role: metapb.PeerRole_Voter},
},
0,
false,
NotNil,
},
}
s.check(c, step, "add learner peer 9 on store 9", cases)
}

func (s *testStepSuite) TestDemoteFollower(c *C) {
Expand Down Expand Up @@ -344,6 +486,6 @@ func (s *testStepSuite) check(c *C, step OpStep, desc string, cases []testCase)
region := core.NewRegionInfo(&metapb.Region{Id: 1, Peers: tc.Peers}, tc.Peers[0])
c.Assert(step.ConfVerChanged(region), Equals, tc.ConfVerChanged)
c.Assert(step.IsFinish(region), Equals, tc.IsFinish)
c.Assert(step.CheckSafety(region), tc.CheckSafety)
c.Assert(step.CheckInProgress(s.cluster, region), tc.CheckInProgres)
}
}
2 changes: 1 addition & 1 deletion server/schedule/operator_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,7 @@ func (oc *OperatorController) Dispatch(region *core.RegionInfo, source string) {
}

func (oc *OperatorController) checkStaleOperator(op *operator.Operator, step operator.OpStep, region *core.RegionInfo) bool {
err := step.CheckSafety(region)
err := step.CheckInProgress(oc.cluster, region)
if err != nil {
if oc.RemoveOperator(op, zap.String("reason", err.Error())) {
operatorCounter.WithLabelValues(op.Desc(), "stale").Inc()
Expand Down
Loading

0 comments on commit 8f64708

Please sign in to comment.