<commit_before>// Copyright 2020 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 operator

import (
	"fmt"
	"sort"

	"github.com/pingcap/errors"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/tikv/pd/pkg/typeutil"
	"github.com/tikv/pd/server/core"
	"github.com/tikv/pd/server/schedule/filter"
	"github.com/tikv/pd/server/schedule/opt"
	"github.com/tikv/pd/server/schedule/placement"
	"github.com/tikv/pd/server/versioninfo"
)

// Builder is used to create operators. Usage:
//     op, err := NewBuilder(desc, cluster, region).
//                 RemovePeer(store1).
//                 AddPeer(peer1).
//                 SetLeader(store2).
//                 Build(kind)
// The generated Operator will choose the most appropriate execution order
// according to various constraints.
type Builder struct {
	// basic info
	desc          string
	cluster       opt.Cluster
	regionID      uint64
	regionEpoch   *metapb.RegionEpoch
	rules         []*placement.Rule
	expectedRoles map[uint64]placement.PeerRoleType

	// operation record
	originPeers          peersMap
	unhealthyPeers       peersMap
	originLeaderStoreID  uint64
	targetPeers          peersMap
	targetLeaderStoreID  uint64
	targetLeaderStoreIDs []uint64 // This field is only used during multi-target evict leader, and will not be filtered during `Build`.
	err                  error

	// skip origin check flags
	skipOriginJointStateCheck bool

	// build flags
	useJointConsensus bool
	lightWeight       bool
	forceTargetLeader bool

	// intermediate states
	currentPeers                         peersMap
	currentLeaderStoreID                 uint64
	toAdd, toRemove, toPromote, toDemote peersMap       // pending tasks.
	steps                                []OpStep       // generated steps.
	peerAddStep                          map[uint64]int // record at which step a peer is created.

	// comparison function
	stepPlanPreferFuncs []func(stepPlan) int // for buildStepsWithoutJointConsensus
}

// BuilderOption is used to create operator builder.
type BuilderOption func(*Builder)

// SkipOriginJointStateCheck lets the builder skip the joint state check for origin peers.
func SkipOriginJointStateCheck(b *Builder) {
	b.skipOriginJointStateCheck = true
}

// NewBuilder creates a Builder.
func NewBuilder(desc string, cluster opt.Cluster, region *core.RegionInfo, opts ...BuilderOption) *Builder {
	b := &Builder{
		desc:        desc,
		cluster:     cluster,
		regionID:    region.GetID(),
		regionEpoch: region.GetRegionEpoch(),
	}

	// options
	for _, option := range opts {
		option(b)
	}

	// origin peers
	err := b.err
	originPeers := newPeersMap()
	unhealthyPeers := newPeersMap()

	for _, p := range region.GetPeers() {
		if p == nil || p.GetStoreId() == 0 {
			err = errors.Errorf("cannot build operator for region with nil peer")
			break
		}
		originPeers.Set(p)
	}

	for _, p := range region.GetPendingPeers() {
		unhealthyPeers.Set(p)
	}

	for _, p := range region.GetDownPeers() {
		unhealthyPeers.Set(p.Peer)
	}

	// origin leader
	originLeaderStoreID := region.GetLeader().GetStoreId()
	if _, ok := originPeers[originLeaderStoreID]; err == nil && !ok {
		err = errors.Errorf("cannot build operator for region with no leader")
	}

	// placement rules
	var rules []*placement.Rule
	if err == nil && cluster.GetOpts().IsPlacementRulesEnabled() {
		fit := cluster.GetRuleManager().FitRegion(cluster, region)
		for _, rf := range fit.RuleFits {
			rules = append(rules, rf.Rule)
		}
		if len(rules) == 0 {
			err = errors.Errorf("cannot build operator for region match no placement rule")
		}
	}

	// joint state check
	if err == nil && !b.skipOriginJointStateCheck && core.IsInJointState(region.GetPeers()...) {
		err = errors.Errorf("cannot build operator for region which is in joint state")
	}

	// build flags
	supportJointConsensus := versioninfo.IsFeatureSupported(cluster.GetOpts().GetClusterVersion(), versioninfo.JointConsensus)

	b.rules = rules
	b.originPeers = originPeers
	b.unhealthyPeers = unhealthyPeers
	b.originLeaderStoreID = originLeaderStoreID
	b.targetPeers = originPeers.Copy()
	b.useJointConsensus = supportJointConsensus && cluster.GetOpts().IsUseJointConsensus()
	b.err = err
	return b
}

// AddPeer records an add Peer operation in Builder. If peer.Id is 0, the builder
// will allocate a new peer ID later.
func (b *Builder) AddPeer(peer *metapb.Peer) *Builder {
	if b.err != nil {
		return b
	}
	if peer == nil || peer.GetStoreId() == 0 {
		b.err = errors.Errorf("cannot add nil peer")
	} else if core.IsInJointState(peer) {
		b.err = errors.Errorf("cannot add peer %s: is in joint state", peer)
	} else if old, ok := b.targetPeers[peer.GetStoreId()]; ok {
		b.err = errors.Errorf("cannot add peer %s: already have peer %s", peer, old)
	} else {
		b.targetPeers.Set(peer)
	}
	return b
}

// RemovePeer records a remove peer operation in Builder.
func (b *Builder) RemovePeer(storeID uint64) *Builder {
	if b.err != nil {
		return b
	}
	if _, ok := b.targetPeers[storeID]; !ok {
		b.err = errors.Errorf("cannot remove peer from %d: not found", storeID)
	} else if b.targetLeaderStoreID == storeID {
		b.err = errors.Errorf("cannot remove peer from %d: is target leader", storeID)
	} else {
		delete(b.targetPeers, storeID)
	}
	return b
}

// PromoteLearner records a promote learner operation in Builder.
func (b *Builder) PromoteLearner(storeID uint64) *Builder {
	if b.err != nil {
		return b
	}
	if peer, ok := b.targetPeers[storeID]; !ok {
		b.err = errors.Errorf("cannot promote peer %d: not found", storeID)
	} else if !core.IsLearner(peer) {
		b.err = errors.Errorf("cannot promote peer %d: is not learner", storeID)
	} else if _, ok := b.unhealthyPeers[storeID]; ok {
		b.err = errors.Errorf("cannot promote peer %d: unhealthy", storeID)
	} else {
		b.targetPeers.Set(&metapb.Peer{
			Id:      peer.GetId(),
			StoreId: peer.GetStoreId(),
			Role:    metapb.PeerRole_Voter,
		})
	}
	return b
}

// DemoteVoter records a demote voter operation in Builder.
func (b *Builder) DemoteVoter(storeID uint64) *Builder {
	if b.err != nil {
		return b
	}
	if peer, ok := b.targetPeers[storeID]; !ok {
		b.err = errors.Errorf("cannot demote voter %d: not found", storeID)
	} else if core.IsLearner(peer) {
		b.err = errors.Errorf("cannot demote voter %d: is already learner", storeID)
	} else {
		b.targetPeers.Set(&metapb.Peer{
			Id:      peer.GetId(),
			StoreId: peer.GetStoreId(),
			Role:    metapb.PeerRole_Learner,
		})
	}
	return b
}

// SetLeader records the target leader in Builder.
func (b *Builder) SetLeader(storeID uint64) *Builder {
	if b.err != nil {
		return b
	}
	if peer, ok := b.targetPeers[storeID]; !ok {
		b.err = errors.Errorf("cannot transfer leader to %d: not found", storeID)
	} else if core.IsLearner(peer) {
		b.err = errors.Errorf("cannot transfer leader to %d: not voter", storeID)
	} else if _, ok := b.unhealthyPeers[storeID]; ok {
		b.err = errors.Errorf("cannot transfer leader to %d: unhealthy", storeID)
	} else {
		b.targetLeaderStoreID = storeID
	}
	return b
}

// SetLeaders records all valid target leaders in Builder.
func (b *Builder) SetLeaders(storeIDs []uint64) *Builder {
	if b.err != nil {
		return b
	}
	for _, storeID := range storeIDs {
		peer := b.targetPeers[storeID]
		if peer == nil || core.IsLearner(peer) || b.unhealthyPeers[storeID] != nil {
			continue<commit_msg>operator: sort multi target leader transfer store ids (#4563)<commit_after>// Copyright 2020 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 operator

import (
	"fmt"
	"sort"

	"github.com/pingcap/errors"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/tikv/pd/pkg/typeutil"
	"github.com/tikv/pd/server/core"
	"github.com/tikv/pd/server/schedule/filter"
	"github.com/tikv/pd/server/schedule/opt"
	"github.com/tikv/pd/server/schedule/placement"
	"github.com/tikv/pd/server/versioninfo"
)

// Builder is used to create operators. Usage:
//     op, err := NewBuilder(desc, cluster, region).
//                 RemovePeer(store1).
//                 AddPeer(peer1).
//                 SetLeader(store2).
//                 Build(kind)
// The generated Operator will choose the most appropriate execution order
// according to various constraints.
type Builder struct {
	// basic info
	desc          string
	cluster       opt.Cluster
	regionID      uint64
	regionEpoch   *metapb.RegionEpoch
	rules         []*placement.Rule
	expectedRoles map[uint64]placement.PeerRoleType

	// operation record
	originPeers          peersMap
	unhealthyPeers       peersMap
	originLeaderStoreID  uint64
	targetPeers          peersMap
	targetLeaderStoreID  uint64
	targetLeaderStoreIDs []uint64 // This field is only used during multi-target evict leader, and will not be filtered during `Build`.
	err                  error

	// skip origin check flags
	skipOriginJointStateCheck bool

	// build flags
	useJointConsensus bool
	lightWeight       bool
	forceTargetLeader bool

	// intermediate states
	currentPeers                         peersMap
	currentLeaderStoreID                 uint64
	toAdd, toRemove, toPromote, toDemote peersMap       // pending tasks.
	steps                                []OpStep       // generated steps.
	peerAddStep                          map[uint64]int // record at which step a peer is created.

	// comparison function
	stepPlanPreferFuncs []func(stepPlan) int // for buildStepsWithoutJointConsensus
}

// BuilderOption is used to create operator builder.
type BuilderOption func(*Builder)

// SkipOriginJointStateCheck lets the builder skip the joint state check for origin peers.
func SkipOriginJointStateCheck(b *Builder) {
	b.skipOriginJointStateCheck = true
}

// NewBuilder creates a Builder.
func NewBuilder(desc string, cluster opt.Cluster, region *core.RegionInfo, opts ...BuilderOption) *Builder {
	b := &Builder{
		desc:        desc,
		cluster:     cluster,
		regionID:    region.GetID(),
		regionEpoch: region.GetRegionEpoch(),
	}

	// options
	for _, option := range opts {
		option(b)
	}

	// origin peers
	err := b.err
	originPeers := newPeersMap()
	unhealthyPeers := newPeersMap()

	for _, p := range region.GetPeers() {
		if p == nil || p.GetStoreId() == 0 {
			err = errors.Errorf("cannot build operator for region with nil peer")
			break
		}
		originPeers.Set(p)
	}

	for _, p := range region.GetPendingPeers() {
		unhealthyPeers.Set(p)
	}

	for _, p := range region.GetDownPeers() {
		unhealthyPeers.Set(p.Peer)
	}

	// origin leader
	originLeaderStoreID := region.GetLeader().GetStoreId()
	if _, ok := originPeers[originLeaderStoreID]; err == nil && !ok {
		err = errors.Errorf("cannot build operator for region with no leader")
	}

	// placement rules
	var rules []*placement.Rule
	if err == nil && cluster.GetOpts().IsPlacementRulesEnabled() {
		fit := cluster.GetRuleManager().FitRegion(cluster, region)
		for _, rf := range fit.RuleFits {
			rules = append(rules, rf.Rule)
		}
		if len(rules) == 0 {
			err = errors.Errorf("cannot build operator for region match no placement rule")
		}
	}

	// joint state check
	if err == nil && !b.skipOriginJointStateCheck && core.IsInJointState(region.GetPeers()...) {
		err = errors.Errorf("cannot build operator for region which is in joint state")
	}

	// build flags
	supportJointConsensus := versioninfo.IsFeatureSupported(cluster.GetOpts().GetClusterVersion(), versioninfo.JointConsensus)

	b.rules = rules
	b.originPeers = originPeers
	b.unhealthyPeers = unhealthyPeers
	b.originLeaderStoreID = originLeaderStoreID
	b.targetPeers = originPeers.Copy()
	b.useJointConsensus = supportJointConsensus && cluster.GetOpts().IsUseJointConsensus()
	b.err = err
	return b
}

// AddPeer records an add Peer operation in Builder. If peer.Id is 0, the builder
// will allocate a new peer ID later.
func (b *Builder) AddPeer(peer *metapb.Peer) *Builder {
	if b.err != nil {
		return b
	}
	if peer == nil || peer.GetStoreId() == 0 {
		b.err = errors.Errorf("cannot add nil peer")
	} else if core.IsInJointState(peer) {
		b.err = errors.Errorf("cannot add peer %s: is in joint state", peer)
	} else if old, ok := b.targetPeers[peer.GetStoreId()]; ok {
		b.err = errors.Errorf("cannot add peer %s: already have peer %s", peer, old)
	} else {
		b.targetPeers.Set(peer)
	}
	return b
}

// RemovePeer records a remove peer operation in Builder.
func (b *Builder) RemovePeer(storeID uint64) *Builder {
	if b.err != nil {
		return b
	}
	if _, ok := b.targetPeers[storeID]; !ok {
		b.err = errors.Errorf("cannot remove peer from %d: not found", storeID)
	} else if b.targetLeaderStoreID == storeID {
		b.err = errors.Errorf("cannot remove peer from %d: is target leader", storeID)
	} else {
		delete(b.targetPeers, storeID)
	}
	return b
}

// PromoteLearner records a promote learner operation in Builder.
func (b *Builder) PromoteLearner(storeID uint64) *Builder {
	if b.err != nil {
		return b
	}
	if peer, ok := b.targetPeers[storeID]; !ok {
		b.err = errors.Errorf("cannot promote peer %d: not found", storeID)
	} else if !core.IsLearner(peer) {
		b.err = errors.Errorf("cannot promote peer %d: is not learner", storeID)
	} else if _, ok := b.unhealthyPeers[storeID]; ok {
		b.err = errors.Errorf("cannot promote peer %d: unhealthy", storeID)
	} else {
		b.targetPeers.Set(&metapb.Peer{
			Id:      peer.GetId(),
			StoreId: peer.GetStoreId(),
			Role:    metapb.PeerRole_Voter,
		})
	}
	return b
}

// DemoteVoter records a demote voter operation in Builder.
func (b *Builder) DemoteVoter(storeID uint64) *Builder {
	if b.err != nil {
		return b
	}
	if peer, ok := b.targetPeers[storeID]; !ok {
		b.err = errors.Errorf("cannot demote voter %d: not found", storeID)
	} else if core.IsLearner(peer) {
		b.err = errors.Errorf("cannot demote voter %d: is already learner", storeID)
	} else {
		b.targetPeers.Set(&metapb.Peer{
			Id:      peer.GetId(),
			StoreId: peer.GetStoreId(),
			Role:    metapb.PeerRole_Learner,
		})
	}
	return b
}

// SetLeader records the target leader in Builder.
func (b *Builder) SetLeader(storeID uint64) *Builder {
	if b.err != nil {
		return b
	}
	if peer, ok := b.targetPeers[storeID]; !ok {
		b.err = errors.Errorf("cannot transfer leader to %d: not found", storeID)
	} else if core.IsLearner(peer) {
		b.err = errors.Errorf("cannot transfer leader to %d: not voter", storeID)
	} else if _, ok := b.unhealthyPeers[storeID]; ok {
		b.err = errors.Errorf("cannot transfer leader to %d: unhealthy", storeID)
	} else {
		b.targetLeaderStoreID = storeID
	}
	return b
}

// SetLeaders records all valid target leaders in Builder.
func (b *Builder) SetLeaders(storeIDs []uint64) *Builder {
	if b.err != nil {
		return b
	}
	sort.Slice(storeIDs, func(i, j int) bool { return storeIDs[i] < storeIDs[j] })
	for _, storeID := range storeIDs {
		peer := b.targetPeers[storeID]
		if peer == nil || core.IsLearner(peer) || b.unhealthyPeers[storeID] != nil {
			continue
{'fraction_non_alphanumeric': 0.11540884465163159, 'fraction_numerical': 0.005165679727856873, 'mean_word_length': 3.3232570806100217, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2252908', 'n_tokens_mistral': 5403, 'n_tokens_neox': 5080, 'n_words': 1519}