Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

schedule: fix operator kind bug. #975

Merged
merged 4 commits into from
Mar 5, 2018
Merged

schedule: fix operator kind bug. #975

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
099109a
schedule: fix operator kind bug.
disksing Mar 5, 2018
aab3a27
*: check operator kinds in tests.
disksing Mar 5, 2018
d229944
make test stable.
disksing Mar 5, 2018
030a565
Merge branch 'master' into operator-kind
disksing Mar 5, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions pd-client/leader_change_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,10 +229,10 @@ func (s *testLeaderChangeSuite) mustGetLeader(c *C, cli *client, urls []string)
}

func (s *testLeaderChangeSuite) verifyLeader(c *C, cli *client, leader string) {
cli.scheduleCheckLeader()
time.Sleep(time.Millisecond * 500)

cli.connMu.RLock()
defer cli.connMu.RUnlock()
c.Assert(cli.connMu.leader, Equals, leader)
testutil.WaitUntil(c, func(c *C) bool {
cli.scheduleCheckLeader()
cli.connMu.RLock()
defer cli.connMu.RUnlock()
return cli.connMu.leader == leader
})
}
6 changes: 3 additions & 3 deletions server/coordinator_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,10 +201,10 @@ func (s *testCoordinatorSuite) TestDispatch(c *C) {

// Wait for schedule and turn off balance.
waitOperator(c, co, 1)
schedulers.CheckTransferPeer(c, co.getOperator(1), 4, 1)
schedulers.CheckTransferPeer(c, co.getOperator(1), schedule.OpBalance, 4, 1)
c.Assert(co.removeScheduler("balance-region-scheduler"), IsNil)
waitOperator(c, co, 2)
schedulers.CheckTransferLeader(c, co.getOperator(2), 4, 2)
schedulers.CheckTransferLeader(c, co.getOperator(2), schedule.OpBalance, 4, 2)
c.Assert(co.removeScheduler("balance-leader-scheduler"), IsNil)

stream := newMockHeartbeatStream()
Expand Down Expand Up @@ -316,7 +316,7 @@ func (s *testCoordinatorSuite) TestPeerState(c *C) {

// Wait for schedule.
waitOperator(c, co, 1)
schedulers.CheckTransferPeer(c, co.getOperator(1), 4, 1)
schedulers.CheckTransferPeer(c, co.getOperator(1), schedule.OpBalance, 4, 1)

region := tc.GetRegion(1)

Expand Down
14 changes: 7 additions & 7 deletions server/namespace_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,10 +53,10 @@ func (s *testNamespaceSuite) TestReplica(c *C) {
s.classifier.setRegion(1, "ns1")
s.tc.addLeaderRegion(1, 1)
op := checker.Check(s.tc.GetRegion(1))
schedulers.CheckAddPeer(c, op, 2)
schedulers.CheckAddPeer(c, op, schedule.OpReplica, 2)
s.tc.addLeaderRegion(1, 3)
op = checker.Check(s.tc.GetRegion(1))
schedulers.CheckAddPeer(c, op, 1)
schedulers.CheckAddPeer(c, op, schedule.OpReplica, 1)

// Stop adding replica if no store in the same namespace.
s.tc.addLeaderRegion(1, 1, 2)
Expand All @@ -82,7 +82,7 @@ func (s *testNamespaceSuite) TestNamespaceChecker(c *C) {
s.classifier.setRegion(1, "ns2")
s.tc.addLeaderRegion(1, 1)
op := checker.Check(s.tc.GetRegion(1))
schedulers.CheckTransferPeer(c, op, 1, 3)
schedulers.CheckTransferPeer(c, op, schedule.OpReplica, 1, 3)

// Only move one region if the one was in the right store while the other was not.
s.classifier.setRegion(2, "ns1")
Expand All @@ -92,7 +92,7 @@ func (s *testNamespaceSuite) TestNamespaceChecker(c *C) {
op = checker.Check(s.tc.GetRegion(2))
c.Assert(op, IsNil)
op = checker.Check(s.tc.GetRegion(3))
schedulers.CheckTransferPeer(c, op, 2, 3)
schedulers.CheckTransferPeer(c, op, schedule.OpReplica, 2, 3)

// Do NOT move the region if it was in the right store.
s.classifier.setRegion(4, "ns2")
Expand All @@ -104,7 +104,7 @@ func (s *testNamespaceSuite) TestNamespaceChecker(c *C) {
s.classifier.setRegion(5, "ns1")
s.tc.addLeaderRegion(5, 1, 1, 3)
op = checker.Check(s.tc.GetRegion(5))
schedulers.CheckTransferPeer(c, op, 3, 2)
schedulers.CheckTransferPeer(c, op, schedule.OpReplica, 3, 2)
}

func (s *testNamespaceSuite) TestSchedulerBalanceRegion(c *C) {
Expand All @@ -125,7 +125,7 @@ func (s *testNamespaceSuite) TestSchedulerBalanceRegion(c *C) {
s.tc.addLeaderRegion(1, 2)
s.classifier.setRegion(1, "ns1")
op := scheduleByNamespace(s.tc, s.classifier, sched, schedule.NewOpInfluence(nil, s.tc))
schedulers.CheckTransferPeer(c, op, 2, 1)
schedulers.CheckTransferPeer(c, op, schedule.OpBalance, 2, 1)

// If no more store in the namespace, balance stops.
s.tc.addLeaderRegion(1, 3)
Expand Down Expand Up @@ -164,7 +164,7 @@ func (s *testNamespaceSuite) TestSchedulerBalanceLeader(c *C) {
s.tc.addLeaderRegion(1, 2, 1)
s.classifier.setRegion(1, "ns1")
op := scheduleByNamespace(s.tc, s.classifier, sched, schedule.NewOpInfluence(nil, s.tc))
schedulers.CheckTransferLeader(c, op, 2, 1)
schedulers.CheckTransferLeader(c, op, schedule.OpBalance, 2, 1)

// If region is not in the correct namespace, it will not be balanced.
s.tc.addLeaderRegion(1, 4, 1)
Expand Down
8 changes: 4 additions & 4 deletions server/schedule/operator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -263,16 +263,16 @@ func (o *Operator) History() []OperatorHistory {

// CreateRemovePeerOperator creates an Operator that removes a peer from region.
func CreateRemovePeerOperator(desc string, cluster Cluster, kind OperatorKind, region *core.RegionInfo, storeID uint64) *Operator {
kind, steps := removePeerSteps(cluster, region, storeID)
return NewOperator(desc, region.GetId(), kind, steps...)
removeKind, steps := removePeerSteps(cluster, region, storeID)
return NewOperator(desc, region.GetId(), removeKind|kind, steps...)
}

// CreateMovePeerOperator creates an Operator that replaces an old peer with a
// new peer
func CreateMovePeerOperator(desc string, cluster Cluster, region *core.RegionInfo, kind OperatorKind, oldStore, newStore uint64, peerID uint64) *Operator {
kind, steps := removePeerSteps(cluster, region, oldStore)
removeKind, steps := removePeerSteps(cluster, region, oldStore)
steps = append([]OperatorStep{AddPeer{ToStore: newStore, PeerID: peerID}}, steps...)
return NewOperator(desc, region.GetId(), kind|OpRegion, steps...)
return NewOperator(desc, region.GetId(), removeKind|kind|OpRegion, steps...)
}

// removePeerSteps returns the steps to safely remove a peer. It prevents removing leader by transfer its leadership first.
Expand Down
Loading