opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-01 15:07:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

optimized(scheduler): reduce scheduling time (#22031)

Browse Source
This commit is contained in:
Zexi Li
2025-02-05 10:25:27 +08:00
committed by GitHub
parent 2143052cac
commit 8ca9fd5640
10 changed files with 151 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/scheduler/algorithm/predicates/cloudregion_schedtag_predicate.go
+5 -1
View File
@@ -50,11 +50,15 @@ type cloudregionSchedtagW struct {
func (p *CloudregionSchedtagPredicate) GetInputs(u *core.Unit) []ISchedtagCustomer {
data := u.SchedData()
tags := data.Schedtags
schedtags := GetInputSchedtagByType(tags, computemodels.CloudregionManager.KeywordPlural())
if len(schedtags) == 0 {
return nil
}
return []ISchedtagCustomer{
&cloudregionSchedtagW{
schedData: data,
cloudregion: data.PreferRegion,
schedtags: GetInputSchedtagByType(tags, computemodels.CloudregionManager.KeywordPlural()),
schedtags: schedtags,
}}
}
pkg/scheduler/algorithm/predicates/predicates.go
+44 -19
View File
@@ -33,6 +33,8 @@ import (
"strings"
"sync"
"golang.org/x/sync/errgroup"
"yunion.io/x/jsonutils"
"yunion.io/x/log"
"yunion.io/x/pkg/gotypes"
@@ -347,16 +349,12 @@ func (p *BaseSchedtagPredicate) GetHypervisorDriver() models.IGuestDriver {
return models.GetDriver(p.Hypervisor)
}
func (p *BaseSchedtagPredicate) check(input ISchedtagCustomer, candidate ISchedtagCandidateResource, u *core.Unit, c core.Candidater) (*PredicatedSchedtagResource, error) {
func (p *BaseSchedtagPredicate) check(input ISchedtagCustomer, candidate ISchedtagCandidateResource, u *core.Unit, c core.Candidater, allTags []schedtag.ISchedtag) (*PredicatedSchedtagResource, error) {
// allTags, err := GetAllSchedtags(getSchedtagResourceType(candidate))
// sMan, err := schedtag.GetSessionManager(u.SessionID())
// if err != nil {
// return nil, err
// }
allTags, err := schedtag.GetAllSchedtags(getSchedtagResourceType(candidate))
if err != nil {
return nil, err
}
tagPredicate := NewSchedtagPredicate(input.GetSchedtags(), allTags)
res := &PredicatedSchedtagResource{
ISchedtagCandidateResource: candidate,
@@ -376,22 +374,39 @@ func (p *BaseSchedtagPredicate) check(input ISchedtagCustomer, candidate ISchedt
return res, nil
}
func (p *BaseSchedtagPredicate) checkResources(input ISchedtagCustomer, ress []ISchedtagCandidateResource, u *core.Unit, c core.Candidater) ([]*PredicatedSchedtagResource, error) {
errs := make([]error, 0)
ret := make([]*PredicatedSchedtagResource, 0)
for _, res := range ress {
ps, err := p.check(input, res, u, c)
if err != nil {
// append err, resource not suit input customer
errs = append(errs, err)
continue
func (p *BaseSchedtagPredicate) checkResources(input ISchedtagCustomer, ress []ISchedtagCandidateResource, u *core.Unit, c core.Candidater, allTags []schedtag.ISchedtag) ([]*PredicatedSchedtagResource, error) {
errs := make([]error, len(ress))
ret := make([]*PredicatedSchedtagResource, len(ress))
errGrp := errgroup.Group{}
for i := range ress {
res := ress[i]
errGrp.Go(func() error {
ps, err := p.check(input, res, u, c, allTags)
if err != nil {
// append err, resource not suit input customer
errs[i] = err
} else {
ret[i] = ps
}
return nil
})
}
if err := errGrp.Wait(); err != nil {
return nil, fmt.Errorf("errGrp.Wait: %v", err)
}
newRet := make([]*PredicatedSchedtagResource, 0)
newErrs := make([]error, 0)
for i := range ress {
if ps := ret[i]; ps != nil {
newRet = append(newRet, ps)
} else {
newErrs = append(newErrs, errs[i])
}
ret = append(ret, ps)
}
if len(ret) == 0 {
return nil, errors.NewAggregate(errs)
if len(newRet) == 0 {
return nil, errors.NewAggregate(newErrs)
}
return ret, nil
return newRet, nil
}
func (p *BaseSchedtagPredicate) GetInputResourcesMap(candidateId string) SchedtagInputResourcesMap {
@@ -426,8 +441,10 @@ func (p *BaseSchedtagPredicate) Execute(
u *core.Unit,
c core.Candidater,
) (bool, []core.PredicateFailureReason, error) {
//inputTime := time.Now()
inputs := sp.GetInputs(u)
resources := sp.GetResources(c)
//log.Infof("=======%s get input time: %s, inputs: %s", sp.Name(), time.Since(inputTime), jsonutils.Marshal(inputs))
h := NewPredicateHelper(sp, u, c)
@@ -463,7 +480,14 @@ func (p *BaseSchedtagPredicate) Execute(
filterErrs = append(filterErrs, errs...)
}
matchedResources, err := p.checkResources(input, fitResources, u, c)
allTags, err := schedtag.GetAllSchedtags(getSchedtagResourceType(fitResources[0]))
if err != nil {
h.Exclude(fmt.Sprintf("get all schedtags"))
break
}
//checkTime := time.Now()
matchedResources, err := p.checkResources(input, fitResources, u, c, allTags)
//log.Infof("---%s checkResources time: %s", sp.Name(), time.Since(checkTime))
if err != nil {
if len(filterErrs) > 0 {
h.ExcludeByErrors(filterErrs)
@@ -474,6 +498,7 @@ func (p *BaseSchedtagPredicate) Execute(
inputRes[idx] = matchedResources
}
//log.Infof("=======%s get execute time: %s", sp.Name(), time.Since(inputTime))
return h.GetResult()
}
pkg/scheduler/algorithm/predicates/schedtag_helper.go
+39 -24
View File
@@ -25,6 +25,7 @@ import (
computeapi "yunion.io/x/onecloud/pkg/apis/compute"
"yunion.io/x/onecloud/pkg/compute/models"
"yunion.io/x/onecloud/pkg/scheduler/data_manager/schedtag"
"yunion.io/x/onecloud/pkg/scheduler/options"
"yunion.io/x/onecloud/pkg/util/conditionparser"
)
@@ -167,39 +168,46 @@ func GetRequestSchedtags(reqTags []*computeapi.SchedtagConfig, allTags []schedta
type SchedtagChecker struct {
}
type apiTags []computeapi.SchedtagConfig
type apiTags map[string]computeapi.SchedtagConfig
func newApiTags(tags []computeapi.SchedtagConfig) apiTags {
ret := make(map[string]computeapi.SchedtagConfig)
for _, tag := range tags {
ret[tag.Id] = tag
}
return ret
}
func (t apiTags) contains(objTag schedtag.ISchedtag) bool {
for _, tag := range t {
if tag.Id == objTag.GetId() || tag.Id == objTag.GetName() {
return true
}
if _, ok := t[objTag.GetId()]; ok {
return true
}
if _, ok := t[objTag.GetName()]; ok {
return true
}
return false
}
type objTags []schedtag.ISchedtag
type objTags map[string]schedtag.ISchedtag
func newObjTags(tags []schedtag.ISchedtag) objTags {
ret := make(map[string]schedtag.ISchedtag)
for _, tag := range tags {
ret[tag.GetId()] = tag
ret[tag.GetName()] = tag
}
return ret
}
func (t objTags) contains(atag computeapi.SchedtagConfig) bool {
for _, tag := range t {
if tag.GetId() == atag.Id || tag.GetName() == atag.Id {
return true
}
}
return false
}
func (c *SchedtagChecker) contains(tags []computeapi.SchedtagConfig, objTag models.SSchedtag) bool {
for _, tag := range tags {
if tag.Id == objTag.Id || tag.Id == objTag.Name {
return true
}
if _, ok := t[atag.Id]; ok {
return true
}
return false
}
func (c *SchedtagChecker) HasIntersection(tags []computeapi.SchedtagConfig, objTags []schedtag.ISchedtag) (bool, schedtag.ISchedtag) {
var atags apiTags = tags
atags := newApiTags(tags)
for _, objTag := range objTags {
if atags.contains(objTag) {
return true, objTag
@@ -209,7 +217,7 @@ func (c *SchedtagChecker) HasIntersection(tags []computeapi.SchedtagConfig, objT
}
func (c *SchedtagChecker) Contains(objectTags []schedtag.ISchedtag, tags []computeapi.SchedtagConfig) (bool, *computeapi.SchedtagConfig) {
var otags objTags = objectTags
otags := newObjTags(objectTags)
for _, tag := range tags {
if !otags.contains(tag) {
return false, &tag
@@ -268,18 +276,24 @@ func (c *SchedtagChecker) mergeSchedtags(candiate ISchedtagCandidate, staticTags
func (c *SchedtagChecker) GetCandidateSchedtags(candidate ISchedtagCandidate) ([]schedtag.ISchedtag, error) {
// staticTags := candidate.GetSchedtags()
staticTags := schedtag.GetCandidateSchedtags(candidate.ResourceType(), candidate.GetId())
dynamicTags, err := c.getDynamicSchedtags(candidate.ResourceType(), candidate.GetDynamicSchedDesc())
if err != nil {
return nil, err
dynamicTags := []schedtag.ISchedtag{}
var err error
if options.Options.EnableDynamicSchedtag {
dynamicTags, err = c.getDynamicSchedtags(candidate.ResourceType(), candidate.GetDynamicSchedDesc())
if err != nil {
return nil, err
}
}
return c.mergeSchedtags(candidate, staticTags, dynamicTags), nil
}
func (c *SchedtagChecker) Check(p ISchedtagPredicate, candidate ISchedtagCandidate) error {
//getT := time.Now()
candidateTags, err := c.GetCandidateSchedtags(candidate)
if err != nil {
return err
}
//log.Infof("=====%s getCandidateSchedtags %s =====", candidate.IndexKey(), time.Since(getT))
execludeTags := p.GetExcludeTags()
requireTags := p.GetRequireTags()
@@ -298,6 +312,7 @@ func (c *SchedtagChecker) Check(p ISchedtagPredicate, candidate ISchedtagCandida
return fmt.Errorf("%s need schedtag: %q", candiInfo, tag.Id)
}
}
//log.Infof("-------%s check time: %s", candidate.IndexKey(), time.Since(getT))
return nil
}
pkg/scheduler/algorithm/predicates/zone_schedtag_predicate.go
+5 -1
View File
@@ -50,11 +50,15 @@ type zoneSchedtagInputW struct {
func (p *ZoneSchedtagPredicate) GetInputs(u *core.Unit) []ISchedtagCustomer {
data := u.SchedData()
tags := data.Schedtags
schedtags := GetInputSchedtagByType(tags, computemodels.ZoneManager.KeywordPlural())
if len(schedtags) == 0 {
return nil
}
return []ISchedtagCustomer{
&zoneSchedtagInputW{
schedData: data,
zone: data.PreferZone,
schedtags: GetInputSchedtagByType(tags, computemodels.ZoneManager.KeywordPlural()),
schedtags: schedtags,
},
}
}
pkg/scheduler/core/analysor.go
+13
View File
@@ -20,9 +20,12 @@ import (
"time"
"yunion.io/x/log"
"yunion.io/x/onecloud/pkg/scheduler/options"
)
type predicateAnalysor struct {
enable bool
hint string
starts map[string]time.Time
elpased map[string]time.Duration
@@ -30,6 +33,7 @@ type predicateAnalysor struct {
func newPredicateAnalysor(hint string) *predicateAnalysor {
return &predicateAnalysor{
enable: options.Options.EnableAnalysis,
hint: hint,
starts: make(map[string]time.Time),
elpased: make(map[string]time.Duration),
@@ -37,11 +41,17 @@ func newPredicateAnalysor(hint string) *predicateAnalysor {
}
func (p *predicateAnalysor) Start(pName string) *predicateAnalysor {
if !p.enable {
return p
}
p.starts[pName] = time.Now()
return p
}
func (p *predicateAnalysor) End(pName string, end time.Time) *predicateAnalysor {
if !p.enable {
return p
}
start, ok := p.starts[pName]
if !ok {
panic(fmt.Sprintf("Not found start time of %q", pName))
@@ -70,6 +80,9 @@ func (p predicateDurations) Less(i, j int) bool {
}
func (p *predicateAnalysor) ShowResult() {
if !p.enable {
return
}
lists := make([]*predicateDuration, 0)
for name, d := range p.elpased {
lists = append(lists, &predicateDuration{
pkg/scheduler/core/generic_scheduler.go
+33 -17
View File
@@ -176,9 +176,24 @@ func (g *GenericScheduler) Schedule(ctx context.Context, unit *Unit, candidates
return helper.ResultHelp(itemList, unit.SchedInfo), nil
}
func newSchedResultByCtx(u *Unit, count int64, c Candidater) *SchedResultItem {
func doSelect(u *Unit, candidate Candidater, count int64) {
plugins := u.AllSelectPlugins()
analysor := newPredicateAnalysor("do select for " + candidate.IndexKey())
defer analysor.ShowResult()
for _, plugin := range plugins {
an := fmt.Sprintf("selected plugin: %s for %s", plugin.Name(), candidate.IndexKey())
analysor.Start(an)
plugin.OnSelectEnd(u, candidate, count)
analysor.End(an, time.Now())
}
}
func newSchedResultByCtx(u *Unit, count int64, c Candidater, useSelect bool) *SchedResultItem {
showDetails := u.SchedInfo.ShowSuggestionDetails
id := c.IndexKey()
if useSelect {
doSelect(u, c, count)
}
r := &SchedResultItem{
ID: id,
Count: count,
@@ -204,7 +219,7 @@ func generateScheduleResult(u *Unit, scs []*SelectedCandidate, fcs []Candidater)
for _, it := range scs {
cid := it.Candidate.IndexKey()
r := newSchedResultByCtx(u, it.Count, it.Candidate)
r := newSchedResultByCtx(u, it.Count, it.Candidate, false)
results = append(results, r)
itemMap[cid] = 1
}
@@ -217,7 +232,7 @@ func generateScheduleResult(u *Unit, scs []*SelectedCandidate, fcs []Candidater)
id := c.IndexKey()
if _, ok := itemMap[id]; !ok && u.GetCapacity(id) > 0 {
itemMap[id] = 1
r := newSchedResultByCtx(u, 0, c)
r := newSchedResultByCtx(u, 0, c, true)
results = append(results, r)
}
}
@@ -231,7 +246,7 @@ func generateScheduleResult(u *Unit, scs []*SelectedCandidate, fcs []Candidater)
id := c.IndexKey()
if _, ok := itemMap[id]; !ok {
itemMap[id] = 0
r := newSchedResultByCtx(u, 0, c)
r := newSchedResultByCtx(u, 0, c, true)
results = append(results, r)
}
}
@@ -341,8 +356,6 @@ func SelectHosts(unit *Unit, priorityList HostPriorityList) ([]*SelectedCandidat
bestEffort := unit.SchedInfo.BestEffort
selectedCandidates := []*SelectedCandidate{}
plugins := unit.AllSelectPlugins()
sort.Sort(sort.Reverse(priorityList))
completed:
@@ -378,18 +391,21 @@ completed:
//sort.Sort(sort.Reverse(priorityList))
}
analysor := newPredicateAnalysor("select Execute")
defer analysor.ShowResult()
for _, sc := range selectedMap {
for _, plugin := range plugins {
plugin.OnSelectEnd(unit, sc.Candidate, sc.Count)
}
doSelect(unit, sc.Candidate, sc.Count)
selectedCandidates = append(selectedCandidates, sc)
}
// hack: not selected host should also execute OnSelectEnd step to inject result of network and storage candiates
for _, nsc := range noSelectedMap {
// hack: not selected host should also execute OnSelectEnd step to inject result of network and storage candidates
/*for _, nsc := range noSelectedMap {
for _, plugin := range plugins {
an := fmt.Sprintf("not selected plugin: %s for %s", plugin.Name(), nsc.IndexKey())
analysor.Start(an)
plugin.OnSelectEnd(unit, nsc, 0)
analysor.End(an, time.Now())
}
}
}*/
if !isSuggestion && !bestEffort {
if count > 0 {
@@ -537,13 +553,13 @@ func unitFitsOnCandidate(
return NewSchedLog(candidateLogIndex, stage, messages, !fit)
}
// analysor := newPredicateAnalysor("predicate Execute")
// defer analysor.ShowResult()
analysor := newPredicateAnalysor("predicate Execute")
defer analysor.ShowResult()
for _, predicate := range predicates {
// n := fmt.Sprintf("%s for %s", predicate.Name(), candidate.Getter().Name())
// analysor.Start(n)
n := fmt.Sprintf("%s for %s", predicate.Name(), candidate.Getter().Name())
analysor.Start(n)
fit, reasons, err = predicate.Execute(ctx, unit, candidate)
// analysor.End(n, time.Now())
analysor.End(n, time.Now())
logs = append(logs, toLog(fit, reasons, err, predicate.Name()))
if err != nil {
return false, nil, err
pkg/scheduler/data_manager/schedtag/cache.go
+1 -7
View File
@@ -31,21 +31,15 @@ type iCache interface {
type cache struct {
sync.Map
mutex sync.Mutex
}
func newCache() iCache {
return &cache{
Map: sync.Map{},
mutex: sync.Mutex{},
Map: sync.Map{},
}
}
func (c *cache) get(key string, newFunc func() (interface{}, error)) (interface{}, error) {
c.mutex.Lock()
defer c.mutex.Unlock()
start := time.Now()
defer func() {
log.Errorf("+++get key %q elpased: %s", key, time.Since(start))
pkg/scheduler/data_manager/schedtag/sync.go
+4 -3
View File
@@ -133,13 +133,14 @@ func (m *dataManager) syncOnce() {
startTime := time.Now()
if err := func() error {
m.tagMan = newSchedtagManagerWithoutInit()
if err := m.tagMan.initAllSchedtags(); err != nil {
tagMan := newSchedtagManagerWithoutInit()
if err := tagMan.initAllSchedtags(); err != nil {
return errors.Wrap(err, "initAllSchedtags")
}
if err := m.tagMan.initDynamicschedtags(); err != nil {
if err := tagMan.initDynamicschedtags(); err != nil {
return errors.Wrap(err, "initResourceSchedtags")
}
m.tagMan = tagMan
return nil
}(); err != nil {
log.Errorf("Schedtag sync data error: %v", err)
pkg/scheduler/manager/manager.go
+1 -1
View File
@@ -99,7 +99,7 @@ func (sm *SchedulerManager) start() {
func (sm *SchedulerManager) schedule(info *api.SchedInfo) (*core.ScheduleResult, error) {
// force sync clean expire cache before do schedule
sm.ExpireManager.Trigger()
// sm.ExpireManager.Trigger()
log.V(10).Infof("SchedulerManager do schedule, input: %#v", info)
task, err := sm.TaskManager.AddTask(sm, info)
pkg/scheduler/options/options.go
+6 -3
View File
@@ -42,9 +42,9 @@ type SchedOptions struct {
SchedulerHistoryCleanPeriod string `help:"Scheduler history cleanup period" default:"60s"`
// parallelization options
HostBuildParallelizeSize int `help:"Number of host description build parallelization" default:"14"`
PredicateParallelizeSize int `help:"Number of execute predicates parallelization" default:"14"`
PriorityParallelizeSize int `help:"Number of execute priority parallelization" default:"14"`
HostBuildParallelizeSize int `help:"Number of host description build parallelization" default:"64"`
PredicateParallelizeSize int `help:"Number of execute predicates parallelization" default:"64"`
PriorityParallelizeSize int `help:"Number of execute priority parallelization" default:"64"`
// expire queue options
ExpireQueueConsumptionPeriod string `help:"Expire queue consumption period" default:"3s"`
@@ -90,6 +90,9 @@ type SchedOptions struct {
SkuRefreshInterval string `help:"Server SKU refresh interval" default:"12h"`
EnableDynamicSchedtag bool `help:"Enable dynamic schedtag feature" default:"false"`
EnableAnalysis bool `help:"Enable analysis feature" default:"false"`
OpenstackOptions
}