Merge pull request #5511 from zexi/bugfix/scheduler-disk-assign

scheduler: fix disk allocated not balanced

pkg/apis/scheduler/api.go

@@ -93,6 +93,17 @@ type CandidateDisk struct {
 	StorageIds []string `json:"storage_ids"`
 }
 
+type CandidateDiskV2 struct {
+	Index    int                 `json:"index"`
+	Storages []*CandidateStorage `json:"storages"`
+}
+
+type CandidateStorage struct {
+	Id           string
+	Name         string
+	FreeCapacity int64
+}
+
 type CandidateNet struct {
 	Index      int      `json:"index"`
 	NetworkIds []string `json:"network_ids"`

pkg/scheduler/algorithm/predicates/disk_schedtag_predicate.go

@@ -17,7 +17,6 @@ package predicates
 import (
 	"fmt"
 
-	"yunion.io/x/log"
 	"yunion.io/x/pkg/utils"
 
 	computeapi "yunion.io/x/onecloud/pkg/apis/compute"
@@ -28,6 +27,7 @@ import (
 
 type DiskSchedtagPredicate struct {
 	*BaseSchedtagPredicate
+	storageUsed map[string]int64
 }
 
 func (p *DiskSchedtagPredicate) Name() string {
@@ -37,6 +37,7 @@ func (p *DiskSchedtagPredicate) Name() string {
 func (p *DiskSchedtagPredicate) Clone() core.FitPredicate {
 	return &DiskSchedtagPredicate{
 		BaseSchedtagPredicate: NewBaseSchedtagPredicate(),
+		storageUsed:           make(map[string]int64),
 	}
 }
 
@@ -145,19 +146,24 @@ func (p *DiskSchedtagPredicate) DoSelect(
 }
 
 func (p *DiskSchedtagPredicate) GetCandidateResourceSortScore(selectRes ISchedtagCandidateResource) int64 {
-	return selectRes.(*api.CandidateStorage).GetFreeCapacity()
+	s := selectRes.(*api.CandidateStorage)
+	return s.GetFreeCapacity()
 }
 
-func (p *DiskSchedtagPredicate) AddSelectResult(index int, selectRes []ISchedtagCandidateResource, output *core.AllocatedResource) {
-	storageIds := []string{}
+func (p *DiskSchedtagPredicate) AddSelectResult(index int, input ISchedtagCustomer, selectRes []ISchedtagCandidateResource, output *core.AllocatedResource) {
+	storages := []*schedapi.CandidateStorage{}
 	for _, res := range selectRes {
-		storageIds = append(storageIds, res.GetId())
+		cs := res.(*api.CandidateStorage)
+		storages = append(storages, &schedapi.CandidateStorage{
+			Id:           cs.GetId(),
+			Name:         cs.GetName(),
+			FreeCapacity: cs.GetFreeCapacity(),
+		})
 	}
-	ret := &schedapi.CandidateDisk{
-		Index:      index,
-		StorageIds: storageIds,
+	ret := &schedapi.CandidateDiskV2{
+		Index:    index,
+		Storages: storages,
 	}
-	log.Debugf("Suggestion storages %v for disk%d", storageIds, index)
 	output.Disks = append(output.Disks, ret)
 }
 

pkg/scheduler/algorithm/predicates/network_schedtag_predicate.go

@@ -266,7 +266,7 @@ func (p *NetworkSchedtagPredicate) DoSelect(
 	return sNets.Results()
 }
 
-func (p *NetworkSchedtagPredicate) AddSelectResult(index int, selectRes []ISchedtagCandidateResource, output *core.AllocatedResource) {
+func (p *NetworkSchedtagPredicate) AddSelectResult(index int, input ISchedtagCustomer, selectRes []ISchedtagCandidateResource, output *core.AllocatedResource) {
 	networkIds := []string{}
 	for _, res := range selectRes {
 		networkIds = append(networkIds, res.GetId())

pkg/scheduler/algorithm/predicates/predicates.go

@@ -256,7 +256,7 @@ type ISchedtagPredicateInstance interface {
 	IsResourceFitInput(unit *core.Unit, c core.Candidater, res ISchedtagCandidateResource, input ISchedtagCustomer) core.PredicateFailureReason
 
 	DoSelect(c core.Candidater, input ISchedtagCustomer, res []ISchedtagCandidateResource) []ISchedtagCandidateResource
-	AddSelectResult(index int, selectRes []ISchedtagCandidateResource, output *core.AllocatedResource)
+	AddSelectResult(index int, input ISchedtagCustomer, selectRes []ISchedtagCandidateResource, output *core.AllocatedResource)
 	GetCandidateResourceSortScore(candidate ISchedtagCandidateResource) int64
 }
 
@@ -478,7 +478,7 @@ func (p *BaseSchedtagPredicate) OnSelectEnd(sp ISchedtagPredicateInstance, u *co
 		sortRes := newSortCandidateResource(sp, selRes)
 		sort.Sort(sortRes)
 		//log.Debugf("sort result: %s", sortRes.DebugString())
-		sp.AddSelectResult(idx, sortRes.res, output)
+		sp.AddSelectResult(idx, inputs[idx], sortRes.res, output)
 	}
 }
 

pkg/scheduler/core/generic_scheduler.go

@@ -29,7 +29,9 @@ import (
 	utiltrace "yunion.io/x/pkg/util/trace"
 	"yunion.io/x/pkg/util/workqueue"
 
+	"yunion.io/x/onecloud/pkg/apis/compute"
 	schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
+	"yunion.io/x/onecloud/pkg/scheduler/api"
 	o "yunion.io/x/onecloud/pkg/scheduler/options"
 )
 
@@ -184,6 +186,7 @@ func newSchedResultByCtx(u *Unit, count int64, c Candidater) *SchedResultItem {
 		Data:              u.GetFiltedData(id, count),
 		Candidater:        c,
 		AllocatedResource: u.GetAllocatedResource(id),
+		SchedData:         u.SchedData(),
 	}
 
 	if showDetails {
@@ -249,17 +252,114 @@ type SchedResultItem struct {
 	Candidater Candidater `json:"-"`
 
 	*AllocatedResource
+
+	SchedData *api.SchedInfo
 }
 
-func (item *SchedResultItem) ToCandidateResource() *schedapi.CandidateResource {
+type StorageUsed struct {
+	used map[string]int64
+}
+
+func NewStorageUsed() *StorageUsed {
+	return &StorageUsed{
+		used: make(map[string]int64),
+	}
+}
+
+func (s *StorageUsed) Get(storageId string) int64 {
+	if used, ok := s.used[storageId]; ok {
+		return used
+	}
+	return 0
+}
+
+func (s *StorageUsed) Add(storageId string, used int64) {
+	if s.used == nil {
+		s.used = make(map[string]int64)
+	}
+	oUsed, ok := s.used[storageId]
+	if ok {
+		s.used[storageId] = oUsed + used
+	} else {
+		s.used[storageId] = used
+	}
+}
+
+func (item *SchedResultItem) ToCandidateResource(storageUsed *StorageUsed) *schedapi.CandidateResource {
 	return &schedapi.CandidateResource{
 		HostId: item.ID,
 		Name:   item.Name,
-		Disks:  item.Disks,
+		Disks:  item.getDisks(storageUsed),
 		Nets:   item.Nets,
 	}
 }
 
+func (item *SchedResultItem) getDisks(used *StorageUsed) []*schedapi.CandidateDisk {
+	inputs := item.SchedData.Disks
+	ret := make([]*schedapi.CandidateDisk, 0)
+	for idx, disk := range item.Disks {
+		ret = append(ret, &schedapi.CandidateDisk{
+			Index:      idx,
+			StorageIds: item.getSortStorageIds(used, inputs[idx], disk.Storages),
+		})
+	}
+	return ret
+}
+
+type sortStorage struct {
+	Id      string
+	FeeSize int64
+}
+
+type sortStorages []sortStorage
+
+func (s sortStorages) Len() int {
+	return len(s)
+}
+
+func (s sortStorages) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}
+
+func (s sortStorages) Less(i, j int) bool {
+	s1 := s[i]
+	s2 := s[j]
+	return s1.FeeSize > s2.FeeSize
+}
+
+func (s sortStorages) getIds() []string {
+	ret := make([]string, 0)
+	for _, obj := range s {
+		ret = append(ret, obj.Id)
+	}
+	return ret
+}
+
+func (item *SchedResultItem) getSortStorageIds(
+	used *StorageUsed,
+	disk *compute.DiskConfig,
+	storages []*schedapi.CandidateStorage) []string {
+	reqSize := disk.SizeMb
+	ss := make([]sortStorage, 0)
+	for _, s := range storages {
+		ss = append(ss, sortStorage{
+			Id:      s.Id,
+			FeeSize: s.FreeCapacity - used.Get(s.Id),
+		})
+	}
+	toSort := sortStorages(ss)
+	sort.Sort(toSort)
+	sortedStorages := toSort.getIds()
+	ret := make([]string, 0)
+	for idx, id := range sortedStorages {
+		if idx == 0 {
+			used.Add(id, int64(reqSize))
+		}
+		ret = append(ret, id)
+	}
+	return ret
+}
+
 func GetCapacities(u *Unit, id string) (res map[string]int64) {
 	res = make(map[string]int64)
 	capacities := u.GetCapacities(id)

pkg/scheduler/core/types.go

@@ -195,13 +195,13 @@ type Priority interface {
 }
 
 type AllocatedResource struct {
-	Disks []*schedapi.CandidateDisk `json:"disks"`
-	Nets  []*schedapi.CandidateNet  `json:"nets"`
+	Disks []*schedapi.CandidateDiskV2 `json:"disks"`
+	Nets  []*schedapi.CandidateNet    `json:"nets"`
 }
 
 func NewAllocatedResource() *AllocatedResource {
 	return &AllocatedResource{
-		Disks: make([]*schedapi.CandidateDisk, 0),
+		Disks: make([]*schedapi.CandidateDiskV2, 0),
 		Nets:  make([]*schedapi.CandidateNet, 0),
 	}
 }

pkg/scheduler/handler/backup_helper.go

@@ -44,10 +44,11 @@ func newBackupSchedResult(
 ) *schedapi.ScheduleOutput {
 	ret := new(schedapi.ScheduleOutput)
 	apiResults := make([]*schedapi.CandidateResource, 0)
+	storageUsed := core.NewStorageUsed()
 	var wireHostMap map[string]core.SchedResultItems
 	for i := 0; i < int(count); i++ {
 		log.V(10).Debugf("Select backup host from result: %s", result)
-		target, err := getSchedBackupResult(result, preferMasterHost, preferBackupHost, sid, wireHostMap)
+		target, err := getSchedBackupResult(result, preferMasterHost, preferBackupHost, sid, wireHostMap, storageUsed)
 		if err != nil {
 			er := &schedapi.CandidateResource{Error: err.Error()}
 			apiResults = append(apiResults, er)
@@ -63,6 +64,7 @@ func getSchedBackupResult(
 	result *core.SchedResultItemList,
 	preferMasterHost, preferBackupHost string,
 	sid string, wireHostMap map[string]core.SchedResultItems,
+	storageUsed *core.StorageUsed,
 ) (*schedapi.CandidateResource, error) {
 	if wireHostMap == nil {
 		wireHostMap = buildWireHostMap(result)
@@ -81,8 +83,8 @@ func getSchedBackupResult(
 	markHostUsed(masterHost)
 	markHostUsed(backupHost)
 
-	ret := masterHost.ToCandidateResource()
-	ret.BackupCandidate = backupHost.ToCandidateResource()
+	ret := masterHost.ToCandidateResource(storageUsed)
+	ret.BackupCandidate = backupHost.ToCandidateResource(storageUsed)
 	ret.SessionId = sid
 	ret.BackupCandidate.SessionId = sid
 	return ret, nil

pkg/scheduler/handler/handler.go

@@ -318,12 +318,13 @@ func transToSchedResult(result *core.SchedResultItemList, schedInfo *api.SchedIn
 func transToRegionSchedResult(result core.SchedResultItems, count int64, sid string) *schedapi.ScheduleOutput {
 	apiResults := make([]*schedapi.CandidateResource, 0)
 	succCount := 0
+	storageUsed := core.NewStorageUsed()
 	for _, nr := range result {
 		for {
 			if nr.Count <= 0 {
 				break
 			}
-			tr := nr.ToCandidateResource()
+			tr := nr.ToCandidateResource(storageUsed)
 			tr.SessionId = sid
 			apiResults = append(apiResults, tr)
 			nr.Count--