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

scheduler: fix memory over committed

Browse Source
This commit is contained in:
Zexi Li
2020-06-15 09:55:49 +08:00
parent c0659c1453
commit 4ac3f8e4fa
15 changed files with 500 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/scheduler/api/candidate.go
+10 -9
View File
@@ -66,15 +66,16 @@ func NewResultResourceInt64(f, r, t int64) *ResultResource {
}
type CandidateListResultItem struct {
ID string `json:"id"`
Name string `json:"name"`
Cpu ResultResource `json:"cpu"`
Mem ResultResource `json:"mem"`
Storage ResultResource `json:"storage"`
Status string `json:"status"`
HostStatus string `json:"host_status"`
EnableStatus string `json:"enable_status"`
HostType string `json:"host_type"`
ID string `json:"id"`
Name string `json:"name"`
Cpu ResultResource `json:"cpu"`
Mem ResultResource `json:"mem"`
Storage ResultResource `json:"storage"`
Status string `json:"status"`
HostStatus string `json:"host_status"`
EnableStatus string `json:"enable_status"`
HostType string `json:"host_type"`
PendingUsage map[string]interface{} `json:"pending_usage"`
}
type CandidateListResult struct {
pkg/scheduler/api/history.go
+4 -3
View File
@@ -85,9 +85,10 @@ type HistoryTask struct {
Time string `json:"time"`
Consuming string `json:"consuming"`
//Result []SchedResultItem `json:"result"`
Result interface{} `json:"result"`
Error string `json:"error"`
Logs []string `json:"logs"`
Result interface{} `json:"result"`
Error string `json:"error"`
Logs []string `json:"logs"`
CapacityMap interface{} `json:"capacity_map"`
}
type HistoryDetail struct {
pkg/scheduler/api/sched.go
+4
View File
@@ -82,6 +82,10 @@ func FetchSchedInfo(req *http.Request) (*SchedInfo, error) {
details[groups[i].Id] = &groups[i]
}
data.InstanceGroupsDetail = details
if data.Count == 0 {
log.Warningf("schedule info data count is 0, set to 1")
data.Count = 1
}
return data, nil
}
pkg/scheduler/cache/candidate/base.go
Vendored
+3
View File
@@ -46,6 +46,7 @@ type BaseHostDesc struct {
InstanceGroups map[string]*api.CandidateGroup `json:"instance_groups"`
IpmiInfo types.SIPMIInfo `json:"ipmi_info"`
PendingUsage map[string]interface{} `json:"pending_usage"`
}
type baseHostGetter struct {
@@ -259,6 +260,8 @@ func newBaseHostDesc(host *computemodels.SHost) (*BaseHostDesc, error) {
return nil, fmt.Errorf("Fill ipmi info error: %v", err)
}
desc.PendingUsage = desc.GetPendingUsage().ToMap()
return desc, nil
}
pkg/scheduler/cache/candidate/common.go
Vendored
+1
View File
@@ -36,6 +36,7 @@ var (
)
VMCreatingStatus = sets.NewString(
computeapi.VM_SCHEDULE,
computeapi.VM_CREATE_NETWORK,
computeapi.VM_CREATE_DISK,
computeapi.VM_START_DEPLOY,
pkg/scheduler/core/context.go
+1 -1
View File
@@ -437,7 +437,7 @@ func NewScheduleUnit(info *api.SchedInfo, schedManager interface{}) *Unit {
}
func (u *Unit) Info() string {
return fmt.Sprintf("%#v", u.SchedInfo)
return u.SchedInfo.JSON(u.SchedInfo).String()
}
func (u *Unit) SessionID() string {
pkg/scheduler/core/generic_scheduler.go
+1 -1
View File
@@ -118,7 +118,7 @@ func (g *GenericScheduler) Schedule(unit *Unit, candidates []Candidater) (*Sched
trace := utiltrace.New(fmt.Sprintf("SessionID: %s, schedule info: %s",
schedInfo.SessionId, unit.Info()))
defer trace.LogIfLong(100 * time.Millisecond)
defer trace.LogIfLong(1 * time.Second)
if len(candidates) == 0 {
return nil, &NoResourceError{
sessionID: schedInfo.SessionId,
pkg/scheduler/handler/forecast_helper.go
-118
View File
@@ -1,118 +0,0 @@
// Copyright 2019 Yunion
//
// 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 handler
import (
"fmt"
"yunion.io/x/log"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/core"
)
func transToSchedForecastResult(result *core.SchedResultItemList) interface{} {
unit := result.Unit
schedData := unit.SchedData()
reqCount := int64(schedData.Count)
filters := make([]*api.ForecastFilter, 0)
filtersMap := make(map[string]*api.ForecastFilter)
getOrNewFilter := func(preName string) (*api.ForecastFilter, bool) {
if info, ok := filtersMap[preName]; !ok {
i := &api.ForecastFilter{
Filter: preName,
Count: 0,
Messages: make([]string, 0),
}
filtersMap[preName] = i
return i, false
} else {
return info, true
}
}
logIndex := func(item *core.SchedResultItem) string {
getter := item.Candidater.Getter()
name := getter.Name()
id := getter.Id()
return fmt.Sprintf("%s:%s", name, id)
}
addInfos := func(logs core.SchedLogList, item *core.SchedResultItem) {
for preName, cnt := range item.CapacityDetails {
if cnt > 0 {
continue
}
failedLog := logs.Get(logIndex(item))
if failedLog == nil {
log.Errorf("predicate %q count is 0, but not found failed log", preName)
continue
}
for _, msg := range failedLog.Messages {
info, exist := getOrNewFilter(msg.Type)
info.Count++
info.Messages = append(info.Messages, msg.Info)
if !exist {
filters = append(filters, info)
}
}
}
}
items := make(core.SchedResultItems, 0)
for _, item := range result.Data {
hostType := item.Candidater.Getter().HostType()
if schedData.Hypervisor == hostType {
items = append(items, item)
}
}
for _, item := range items {
addInfos(result.Unit.LogManager.FailedLogs(), item)
}
var (
output = transToSchedResult(result, schedData)
readyCount int64
)
for _, candi := range output.Candidates {
if len(candi.Error) != 0 {
info, exist := getOrNewFilter("select_candidate")
msg := candi.Error
info.Messages = append(info.Messages, msg)
if !exist {
filters = append(filters, info)
}
} else {
readyCount++
}
}
canCreate := true
if readyCount < reqCount {
canCreate = false
filters = append(filters, &api.ForecastFilter{
Messages: []string{
fmt.Sprintf("No enough resources: %d/%d(free/request)", readyCount, reqCount),
},
})
}
return &api.SchedForecastResult{
CanCreate: canCreate,
Filters: filters,
Results: output.Candidates,
}
}
pkg/scheduler/handler/handler.go
+9 -76
View File
@@ -26,14 +26,11 @@ import (
"yunion.io/x/log"
computeapi "yunion.io/x/onecloud/pkg/apis/compute"
schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
"yunion.io/x/onecloud/pkg/cloudcommon/db"
computemodels "yunion.io/x/onecloud/pkg/compute/models"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/core"
skuman "yunion.io/x/onecloud/pkg/scheduler/data_manager/sku"
schedman "yunion.io/x/onecloud/pkg/scheduler/manager"
schedmodels "yunion.io/x/onecloud/pkg/scheduler/models"
)
// InstallHandler is an interface that registes route and
@@ -51,7 +48,12 @@ func timer(f gin.HandlerFunc) gin.HandlerFunc {
return func(c *gin.Context) {
startTime := time.Now()
f(c)
log.Infof("Handler %q cost: %v", c.Request.URL.Path, time.Since(startTime))
path := c.Request.URL.Path
raw := c.Request.URL.RawQuery
if raw != "" {
path = path + "?" + raw
}
log.Infof("Handler %q cost: %v", path, time.Since(startTime))
}
}
@@ -118,16 +120,7 @@ func doSchedulerTest(c *gin.Context) {
c.AbortWithError(http.StatusBadRequest, err)
return
}
c.JSON(http.StatusOK, transToSchedTestResult(result, schedInfo.SuggestionLimit))
}
func transToSchedTestResult(result *core.SchedResultItemList, limit int64) interface{} {
return &api.SchedTestResult{
Data: result.Data,
Total: int64(result.Data.Len()),
Limit: limit,
Offset: 0,
}
c.JSON(http.StatusOK, result.TestResult)
}
func doSchedulerForecast(c *gin.Context) {
@@ -150,7 +143,7 @@ func doSchedulerForecast(c *gin.Context) {
c.AbortWithError(http.StatusBadRequest, err)
return
}
c.JSON(http.StatusOK, transToSchedForecastResult(result))
c.JSON(http.StatusOK, result.ForecastResult)
}
func doCandidateList(c *gin.Context) {
@@ -283,67 +276,7 @@ func doSyncSchedule(c *gin.Context) {
return
}
resp := transToSchedResult(result, schedInfo)
driver := result.Unit.GetHypervisorDriver()
if err := setSchedPendingUsage(driver, schedInfo, resp); err != nil {
c.AbortWithError(http.StatusInternalServerError, err)
return
}
c.JSON(http.StatusOK, resp)
}
func IsDriverSkipScheduleDirtyMark(driver computemodels.IGuestDriver) bool {
return !(driver.DoScheduleCPUFilter() && driver.DoScheduleMemoryFilter() && driver.DoScheduleStorageFilter())
}
func setSchedPendingUsage(driver computemodels.IGuestDriver, req *api.SchedInfo, resp *schedapi.ScheduleOutput) error {
if req.IsSuggestion || IsDriverSkipScheduleDirtyMark(driver) || req.SkipDirtyMarkHost() {
return nil
}
for _, item := range resp.Candidates {
schedmodels.HostPendingUsageManager.SetPendingUsage(req, item)
}
return nil
}
func transToSchedResult(result *core.SchedResultItemList, schedInfo *api.SchedInfo) *schedapi.ScheduleOutput {
if schedInfo.Backup {
return transToBackupSchedResult(result,
schedInfo.PreferHost, schedInfo.PreferBackupHost, int64(schedInfo.Count), schedInfo.SessionId)
} else {
return transToRegionSchedResult(result.Data, int64(schedInfo.Count), schedInfo.SessionId)
}
}
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(storageUsed)
tr.SessionId = sid
apiResults = append(apiResults, tr)
nr.Count--
succCount++
}
}
for {
if int64(succCount) >= count {
break
}
er := &schedapi.CandidateResource{Error: "Out of resource"}
apiResults = append(apiResults, er)
succCount++
}
return &schedapi.ScheduleOutput{
Candidates: apiResults,
}
c.JSON(http.StatusOK, result.Result)
}
func regionResponse(v interface{}) interface{} {
pkg/scheduler/manager/expire_queue.go
-11
View File
@@ -57,17 +57,6 @@ func newExpireHost(id string, sid string) *expireHost {
func (e *ExpireManager) Run() {
t := time.Tick(u.ToDuration(o.GetOptions().ExpireQueueConsumptionPeriod))
//notInSession := func(hosts []*expireHost, resType string) []*expireHost {
////var newIds []string
////for _, id := range ids {
////if !schedManager.ReservedPoolManager.InSession(resType, id) {
////newIds = append(newIds, id)
////}
////}
////return newIds
//return ids
//}
waitTimeOut := func(wg *sync.WaitGroup, timeout time.Duration) bool {
ch := make(chan struct{})
go func() {
pkg/scheduler/manager/manager.go
+12 -11
View File
@@ -30,6 +30,7 @@ import (
candidatecache "yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
"yunion.io/x/onecloud/pkg/scheduler/core"
"yunion.io/x/onecloud/pkg/scheduler/data_manager"
schedmodels "yunion.io/x/onecloud/pkg/scheduler/models"
o "yunion.io/x/onecloud/pkg/scheduler/options"
"yunion.io/x/onecloud/pkg/util/k8s"
)
@@ -44,10 +45,8 @@ type SchedulerManager struct {
HistoryManager *HistoryManager
TaskManager *TaskManager
DataManager *data_manager.DataManager
CandidateManager *data_manager.CandidateManager
//ReservedPoolManager *data_manager.ReservedPoolManager
//NetworkManager *data_manager.NetworkManager
DataManager *data_manager.DataManager
CandidateManager *data_manager.CandidateManager
KubeClusterManager *k8s.SKubeClusterManager
}
@@ -59,8 +58,6 @@ func NewSchedulerManager(stopCh <-chan struct{}) *SchedulerManager {
sm.CompletedManager = NewCompletedManager(stopCh)
sm.HistoryManager = NewHistoryManager(stopCh)
sm.TaskManager = NewTaskManager(stopCh)
//sm.ReservedPoolManager = data_manager.NewReservedPoolManager(stopCh)
//sm.NetworkManager = data_manager.NewNetworkManager(sm.DataManager, sm.ReservedPoolManager)
sm.KubeClusterManager = k8s.NewKubeClusterManager(o.GetOptions().Region, 30*time.Second)
return sm
@@ -92,16 +89,14 @@ func (sm *SchedulerManager) start() {
sm.TaskManager.Run,
sm.DataManager.Run,
sm.CandidateManager.Run,
//sm.ReservedPoolManager.Run,
//sm.NetworkManager.Run,
sm.KubeClusterManager.Start,
//sm.KubeClusterManager.Start,
}
for _, f := range startFuncs {
go f()
}
}
func (sm *SchedulerManager) schedule(info *api.SchedInfo) (*core.SchedResultItemList, error) {
func (sm *SchedulerManager) schedule(info *api.SchedInfo) (*ScheduleResult, error) {
log.V(10).Infof("SchedulerManager do schedule, input: %#v", info)
task, err := sm.TaskManager.AddTask(sm, info)
if err != nil {
@@ -126,7 +121,7 @@ func NewSessionID() string {
// Schedule process the request data that is scheduled for dispatch and complements
// the session information.
func Schedule(info *api.SchedInfo) (*core.SchedResultItemList, error) {
func Schedule(info *api.SchedInfo) (*ScheduleResult, error) {
if len(info.SessionId) == 0 {
info.SessionId = NewSessionID()
}
@@ -274,6 +269,11 @@ func GetCandidateHostList(
HostType: c.GetHostType(),
EnableStatus: c.GetEnableStatus(),
}
pendingUsage, _ := schedmodels.HostPendingUsageManager.GetPendingUsage(c.GetId())
if pendingUsage != nil {
item.PendingUsage = pendingUsage.ToMap()
}
r.Data = append(r.Data, item)
}
return r, nil
@@ -447,6 +447,7 @@ func GetHistoryDetail(historyDetailArgs *api.HistoryDetailArgs) (*api.HistoryDet
if historyDetailArgs.Log {
historyTask.Logs = taskExecutor.GetLogs()
historyTask.CapacityMap = taskExecutor.GetCapacityMap()
}
}
pkg/scheduler/manager/result_helper.go
+358
View File
@@ -0,0 +1,358 @@
// Copyright 2019 Yunion
//
// 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 manager
import (
"fmt"
"sort"
"yunion.io/x/log"
schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
computemodels "yunion.io/x/onecloud/pkg/compute/models"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/core"
schedmodels "yunion.io/x/onecloud/pkg/scheduler/models"
)
func transToSchedResult(result *core.SchedResultItemList, schedInfo *api.SchedInfo) *schedapi.ScheduleOutput {
if schedInfo.Backup {
return transToBackupSchedResult(result,
schedInfo.PreferHost, schedInfo.PreferBackupHost, int64(schedInfo.Count), schedInfo.SessionId)
} else {
return transToRegionSchedResult(result.Data, int64(schedInfo.Count), schedInfo.SessionId)
}
}
func setSchedPendingUsage(driver computemodels.IGuestDriver, req *api.SchedInfo, resp *schedapi.ScheduleOutput) error {
if req.IsSuggestion || IsDriverSkipScheduleDirtyMark(driver) || req.SkipDirtyMarkHost() {
return nil
}
for _, item := range resp.Candidates {
schedmodels.HostPendingUsageManager.AddPendingUsage(req, item)
}
return nil
}
func IsDriverSkipScheduleDirtyMark(driver computemodels.IGuestDriver) bool {
return !(driver.DoScheduleCPUFilter() && driver.DoScheduleMemoryFilter() && driver.DoScheduleStorageFilter())
}
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(storageUsed)
tr.SessionId = sid
apiResults = append(apiResults, tr)
nr.Count--
succCount++
}
}
for {
if int64(succCount) >= count {
break
}
er := &schedapi.CandidateResource{Error: "Out of resource"}
apiResults = append(apiResults, er)
succCount++
}
return &schedapi.ScheduleOutput{
Candidates: apiResults,
}
}
func transToBackupSchedResult(
result *core.SchedResultItemList, preferMasterHost, preferBackupHost string, count int64, sid string,
) *schedapi.ScheduleOutput {
// clean each result sched result item's count
for _, item := range result.Data {
item.Count = 0
}
apiResults := newBackupSchedResult(result, preferMasterHost, preferBackupHost, count, sid)
return apiResults
}
func newBackupSchedResult(
result *core.SchedResultItemList,
preferMasterHost, preferBackupHost string,
count int64,
sid string,
) *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, storageUsed)
if err != nil {
er := &schedapi.CandidateResource{Error: err.Error()}
apiResults = append(apiResults, er)
continue
}
apiResults = append(apiResults, target)
}
ret.Candidates = apiResults
return ret
}
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)
} else {
reviseWireHostMap(wireHostMap)
}
masterHost, backupHost := selectHosts(wireHostMap, preferMasterHost, preferBackupHost)
if masterHost == nil {
return nil, fmt.Errorf("Can't find master host %q", preferMasterHost)
}
if backupHost == nil {
return nil, fmt.Errorf("Can't find backup host %q by master %q", preferBackupHost, masterHost.ID)
}
markHostUsed(masterHost)
markHostUsed(backupHost)
ret := masterHost.ToCandidateResource(storageUsed)
ret.BackupCandidate = backupHost.ToCandidateResource(storageUsed)
ret.SessionId = sid
ret.BackupCandidate.SessionId = sid
return ret, nil
}
func buildWireHostMap(result *core.SchedResultItemList) map[string]core.SchedResultItems {
sort.Sort(sort.Reverse(result.Data))
wireHostMap := make(map[string]core.SchedResultItems)
for i := 0; i < len(result.Data); i++ {
networks := result.Data[i].Candidater.Getter().Networks()
for j := 0; j < len(networks); j++ {
if hosts, ok := wireHostMap[networks[j].WireId]; ok {
if hostInResultItemsIndex(result.Data[i].ID, hosts) < 0 {
wireHostMap[networks[j].WireId] = append(hosts, result.Data[i])
}
} else {
wireHostMap[networks[j].WireId] = core.SchedResultItems{result.Data[i]}
}
}
}
return wireHostMap
}
func reviseWireHostMap(wireHostMap map[string]core.SchedResultItems) {
for _, hosts := range wireHostMap {
sort.Sort(sort.Reverse(hosts))
}
}
func markHostUsed(host *core.SchedResultItem) {
host.Count++
host.Capacity--
}
func hostInResultItemsIndex(hostId string, hosts core.SchedResultItems) int {
for i := 0; i < len(hosts); i++ {
if hosts[i].ID == hostId {
return i
}
}
return -1
}
func selectHosts(
wireHostMap map[string]core.SchedResultItems, preferMasterHost, preferBackupHost string,
) (*core.SchedResultItem, *core.SchedResultItem) {
var scroe int64
var masterIdx, backupIdx int
var selectedWireId string
for wireId, hosts := range wireHostMap {
masterIdx, backupIdx = -1, -1
if len(hosts) < 2 {
continue
}
if len(preferMasterHost) > 0 {
if masterIdx = hostInResultItemsIndex(preferMasterHost, hosts); masterIdx < 0 {
continue
}
}
if len(preferBackupHost) > 0 {
if backupIdx = hostInResultItemsIndex(preferBackupHost, hosts); backupIdx < 0 {
continue
}
}
// select master host index
if masterIdx < 0 {
for i := 0; i < len(hosts); i++ {
if hosts[i].ID != preferBackupHost {
masterIdx = i
}
}
}
if hosts[masterIdx].Capacity <= 0 {
if len(preferMasterHost) > 0 {
// in case prefer master host capacity isn't enough
break
} else {
continue
}
}
// select backup host index
if backupIdx < 0 {
for i := 0; i < len(hosts); i++ {
if i != masterIdx {
backupIdx = i
}
}
}
if hosts[backupIdx].Capacity <= 0 {
if len(preferBackupHost) > 0 {
// in case perfer backup host capacity isn't enough
break
} else {
continue
}
}
// the highest total score wins
curScore := hosts[masterIdx].Capacity + hosts[backupIdx].Capacity
if curScore > scroe {
selectedWireId = wireId
scroe = curScore
}
}
if len(selectedWireId) == 0 {
return nil, nil
}
return wireHostMap[selectedWireId][masterIdx], wireHostMap[selectedWireId][backupIdx]
}
func transToSchedTestResult(result *core.SchedResultItemList, limit int64) interface{} {
return &api.SchedTestResult{
Data: result.Data,
Total: int64(result.Data.Len()),
Limit: limit,
Offset: 0,
}
}
func transToSchedForecastResult(result *core.SchedResultItemList) interface{} {
unit := result.Unit
schedData := unit.SchedData()
reqCount := int64(schedData.Count)
filters := make([]*api.ForecastFilter, 0)
filtersMap := make(map[string]*api.ForecastFilter)
getOrNewFilter := func(preName string) (*api.ForecastFilter, bool) {
if info, ok := filtersMap[preName]; !ok {
i := &api.ForecastFilter{
Filter: preName,
Count: 0,
Messages: make([]string, 0),
}
filtersMap[preName] = i
return i, false
} else {
return info, true
}
}
logIndex := func(item *core.SchedResultItem) string {
getter := item.Candidater.Getter()
name := getter.Name()
id := getter.Id()
return fmt.Sprintf("%s:%s", name, id)
}
addInfos := func(logs core.SchedLogList, item *core.SchedResultItem) {
for preName, cnt := range item.CapacityDetails {
if cnt > 0 {
continue
}
failedLog := logs.Get(logIndex(item))
if failedLog == nil {
log.Errorf("predicate %q count is 0, but not found failed log", preName)
continue
}
for _, msg := range failedLog.Messages {
info, exist := getOrNewFilter(msg.Type)
info.Count++
info.Messages = append(info.Messages, msg.Info)
if !exist {
filters = append(filters, info)
}
}
}
}
items := make(core.SchedResultItems, 0)
for _, item := range result.Data {
hostType := item.Candidater.Getter().HostType()
if schedData.Hypervisor == hostType {
items = append(items, item)
}
}
for _, item := range items {
addInfos(result.Unit.LogManager.FailedLogs(), item)
}
var (
output = transToSchedResult(result, schedData)
readyCount int64
)
for _, candi := range output.Candidates {
if len(candi.Error) != 0 {
info, exist := getOrNewFilter("select_candidate")
msg := candi.Error
info.Messages = append(info.Messages, msg)
if !exist {
filters = append(filters, info)
}
} else {
readyCount++
}
}
canCreate := true
if readyCount < reqCount {
canCreate = false
filters = append(filters, &api.ForecastFilter{
Messages: []string{
fmt.Sprintf("No enough resources: %d/%d(free/request)", readyCount, reqCount),
},
})
}
return &api.SchedForecastResult{
CanCreate: canCreate,
Filters: filters,
Results: output.Candidates,
}
}
pkg/scheduler/manager/scheduler.go
-6
View File
@@ -152,12 +152,6 @@ func (s *BaseScheduler) BeforePredicate() error {
return nil
}
//func (s *BaseScheduler) DirtySelectedCandidates(scs []*core.SelectedCandidate) {
//for _, sc := range scs {
//s.CandidateManager().SetCandidateDirty(sc)
//}
//}
// GuestScheduler for guest type schedule
type GuestScheduler struct {
*BaseScheduler
pkg/scheduler/manager/task_queue.go
+72 -42
View File
@@ -21,7 +21,9 @@ import (
"time"
"yunion.io/x/log"
"yunion.io/x/pkg/errors"
schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/core"
)
@@ -45,9 +47,10 @@ type TaskExecutor struct {
callback TaskExecuteCallback
unit *core.Unit
resultItems *core.SchedResultItemList
resultItems *ScheduleResult
resultError error
logs []string
capacityMap interface{}
completed bool
}
@@ -78,7 +81,16 @@ func (te *TaskExecutor) Execute() {
}
}
func (te *TaskExecutor) execute() (*core.SchedResultItemList, error) {
type ScheduleResult struct {
// Result is sync schedule result
Result *schedapi.ScheduleOutput
// ForecastResult is forecast schedule result
ForecastResult interface{}
// TestResult is test schedule result
TestResult interface{}
}
func (te *TaskExecutor) execute() (*ScheduleResult, error) {
scheduler := te.scheduler
genericScheduler, err := core.NewGenericScheduler(scheduler.(core.Scheduler))
if err != nil {
@@ -92,7 +104,26 @@ func (te *TaskExecutor) execute() (*core.SchedResultItemList, error) {
}
te.unit = scheduler.Unit()
return genericScheduler.Schedule(te.unit, candidates)
schedInfo := te.unit.SchedInfo
result, err := genericScheduler.Schedule(te.unit, candidates)
if err != nil {
return nil, errors.Wrap(err, "genericScheduler.Schedule")
}
out := new(ScheduleResult)
if schedInfo.IsSuggestion {
if schedInfo.ShowSuggestionDetails && schedInfo.SuggestionAll {
out.ForecastResult = transToSchedForecastResult(result)
} else {
out.TestResult = transToSchedTestResult(result, schedInfo.SuggestionLimit)
}
} else {
out.Result = transToSchedResult(result, schedInfo)
driver := te.unit.GetHypervisorDriver()
if err := setSchedPendingUsage(driver, schedInfo, out.Result); err != nil {
return nil, errors.Wrap(err, "setSchedPendingUsage")
}
}
return out, nil
}
func (te *TaskExecutor) cleanup() {
@@ -107,7 +138,7 @@ func (te *TaskExecutor) Kill() {
}
}
func (te *TaskExecutor) GetResult() (*core.SchedResultItemList, error) {
func (te *TaskExecutor) GetResult() (*ScheduleResult, error) {
return te.resultItems, te.resultError
}
@@ -115,9 +146,12 @@ func (te *TaskExecutor) GetLogs() []string {
return te.logs
}
func (te *TaskExecutor) GetCapacityMap() interface{} {
return te.capacityMap
}
type TaskExecutorQueue struct {
schedType string
poolId string
queue chan *TaskExecutor
running bool
}
@@ -129,39 +163,38 @@ func (teq *TaskExecutorQueue) AddTaskExecutor(scheduler Scheduler,
return taskExecutor
}
func NewTaskExecutorQueue(schedType string, poolId string, stopCh <-chan struct{}) *TaskExecutorQueue {
taskExecutorQueue := &TaskExecutorQueue{
func NewTaskExecutorQueue(schedType string, stopCh <-chan struct{}) *TaskExecutorQueue {
teq := &TaskExecutorQueue{
schedType: schedType,
poolId: poolId,
running: false,
}
taskExecutorQueue.Start(stopCh)
return taskExecutorQueue
teq.Start(stopCh)
return teq
}
func (teq *TaskExecutorQueue) Start(stopCh <-chan struct{}) {
if !teq.running {
teq.running = true
teq.queue = make(chan *TaskExecutor, 5000)
go func() {
defer close(teq.queue)
var taskExecutor *TaskExecutor
for taskExecutor = <-teq.queue; teq.running; taskExecutor = <-teq.queue {
if taskExecutor.Status == TaskExecutorStatusWaiting {
taskExecutor.Execute()
}
}
}()
go func() {
<-stopCh
teq.running = false
teq.queue <- nil
}()
if teq.running {
return
}
teq.running = true
teq.queue = make(chan *TaskExecutor, 5000)
go func() {
defer close(teq.queue)
var taskExecutor *TaskExecutor
for taskExecutor = <-teq.queue; teq.running; taskExecutor = <-teq.queue {
if taskExecutor.Status == TaskExecutorStatusWaiting {
taskExecutor.Execute()
}
}
}()
go func() {
<-stopCh
teq.running = false
teq.queue <- nil
}()
}
type TaskExecutorQueueManager struct {
@@ -178,8 +211,7 @@ func NewTaskExecutorQueueManager(stopCh <-chan struct{}) *TaskExecutorQueueManag
}
}
func (teqm *TaskExecutorQueueManager) GetQueue(schedType string, poolId string,
) *TaskExecutorQueue {
func (teqm *TaskExecutorQueueManager) GetQueue(schedType string) *TaskExecutorQueue {
teqm.lock.Lock()
defer teqm.lock.Unlock()
@@ -189,9 +221,8 @@ func (teqm *TaskExecutorQueueManager) GetQueue(schedType string, poolId string,
ok bool
)
key = fmt.Sprintf("%v:%v", schedType, poolId)
if taskExecutorQueue, ok = teqm.taskExecutorMap[key]; !ok {
taskExecutorQueue = NewTaskExecutorQueue(schedType, poolId, teqm.stopCh)
taskExecutorQueue = NewTaskExecutorQueue(schedType, teqm.stopCh)
teqm.taskExecutorMap[key] = taskExecutorQueue
}
@@ -201,8 +232,7 @@ func (teqm *TaskExecutorQueueManager) GetQueue(schedType string, poolId string,
func (teqm *TaskExecutorQueueManager) AddTaskExecutor(
scheduler Scheduler, callback TaskExecuteCallback) *TaskExecutor {
schedData := scheduler.SchedData()
log.V(10).Infof("AddTaskExecutor schedData: %#v", schedData)
taskQueue := teqm.GetQueue(schedData.Type, "")
taskQueue := teqm.GetQueue(schedData.Hypervisor)
return taskQueue.AddTaskExecutor(scheduler, callback)
}
@@ -269,7 +299,7 @@ type Task struct {
waitCh chan struct{}
completedCount int
resultItems *core.SchedResultItemList
resultItems *ScheduleResult
resultError error
}
@@ -344,6 +374,7 @@ func (t *Task) readLog(taskExecutor *TaskExecutor) {
if u != nil {
logs := u.LogManager.Read()
taskExecutor.logs = logs
taskExecutor.capacityMap = u.CapacityMap
}
}
@@ -352,8 +383,7 @@ func (t *Task) onError() {
taskExecutor.Kill()
}
log.Errorf("Remove Session on error: %v\n", t.SchedInfo.SessionId)
//t.manager.ReservedPoolManager.RemoveSession(t.SchedInfo.SessionID)
log.Errorf("Remove Session on error: %v", t.SchedInfo.SessionId)
close(t.waitCh)
}
@@ -363,12 +393,12 @@ func (t *Task) onCompleted() {
close(t.waitCh)
}
func (t *Task) Wait() (*core.SchedResultItemList, error) {
func (t *Task) Wait() (*ScheduleResult, error) {
log.V(10).Infof("Task wait...")
<-t.waitCh
return t.GetResult()
}
func (t *Task) GetResult() (*core.SchedResultItemList, error) {
func (t *Task) GetResult() (*ScheduleResult, error) {
return t.resultItems, t.resultError
}
pkg/scheduler/models/pending_usage.go
+25 -14
View File
@@ -23,7 +23,6 @@ import (
"github.com/pkg/errors"
"yunion.io/x/jsonutils"
"yunion.io/x/log"
schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
@@ -42,7 +41,9 @@ type SHostPendingUsageManager struct {
func init() {
pendingStore := NewHostMemoryPendingUsageStore()
HostPendingUsageManager = &SHostPendingUsageManager{pendingStore}
HostPendingUsageManager = &SHostPendingUsageManager{
store: pendingStore,
}
}
func (m *SHostPendingUsageManager) Keyword() string {
@@ -50,7 +51,7 @@ func (m *SHostPendingUsageManager) Keyword() string {
}
func (m *SHostPendingUsageManager) newSessionUsage(req *api.SchedInfo, hostId string) *SessionPendingUsage {
su := NewSessionUsage(req.SessionId)
su := NewSessionUsage(req.SessionId, hostId)
su.Usage = NewPendingUsageBySchedInfo(hostId, req)
return su
}
@@ -60,11 +61,14 @@ func (m *SHostPendingUsageManager) newPendingUsage(hostId string) *SPendingUsage
}
func (m *SHostPendingUsageManager) GetPendingUsage(hostId string) (*SPendingUsage, error) {
return m.getPendingUsage(hostId)
}
func (m *SHostPendingUsageManager) getPendingUsage(hostId string) (*SPendingUsage, error) {
pending, err := m.store.GetPendingUsage(hostId)
if err != nil {
return nil, err
}
log.Debugf("Get host %s pending usage: %s", hostId, jsonutils.Marshal(pending.ToMap()).PrettyString())
return pending, nil
}
@@ -72,11 +76,8 @@ func (m *SHostPendingUsageManager) GetSessionUsage(sessionId, hostId string) (*S
return m.store.GetSessionUsage(sessionId, hostId)
}
func (m *SHostPendingUsageManager) SetPendingUsage(req *api.SchedInfo, candidate *schedapi.CandidateResource) {
func (m *SHostPendingUsageManager) AddPendingUsage(req *api.SchedInfo, candidate *schedapi.CandidateResource) {
hostId := candidate.HostId
ctx := context.Background()
lockman.LockClass(ctx, m, hostId)
defer lockman.ReleaseClass(ctx, m, hostId)
sessionUsage, _ := m.GetSessionUsage(req.SessionId, hostId)
if sessionUsage == nil {
@@ -85,12 +86,17 @@ func (m *SHostPendingUsageManager) SetPendingUsage(req *api.SchedInfo, candidate
}
m.addSessionUsage(candidate.HostId, sessionUsage)
if candidate.BackupCandidate != nil {
m.SetPendingUsage(req, candidate.BackupCandidate)
m.AddPendingUsage(req, candidate.BackupCandidate)
}
}
// addSessionUsage add pending usage and session usage
func (m *SHostPendingUsageManager) addSessionUsage(hostId string, usage *SessionPendingUsage) {
pendingUsage, _ := m.GetPendingUsage(hostId)
ctx := context.Background()
lockman.LockClass(ctx, m, hostId)
defer lockman.ReleaseClass(ctx, m, hostId)
pendingUsage, _ := m.getPendingUsage(hostId)
if pendingUsage == nil {
pendingUsage = m.newPendingUsage(hostId)
}
@@ -105,14 +111,13 @@ func (m *SHostPendingUsageManager) CancelPendingUsage(hostId string, su *Session
lockman.LockClass(ctx, m, hostId)
defer lockman.ReleaseClass(ctx, m, hostId)
pendingUsage, _ := m.GetPendingUsage(hostId)
pendingUsage, _ := m.getPendingUsage(hostId)
if pendingUsage == nil {
return nil
}
if su == nil {
return nil
}
log.Debugf("Cancel pendingUsage %#v - %#v", pendingUsage.ToMap(), su.Usage.ToMap())
pendingUsage.Sub(su.Usage)
m.store.SetPendingUsage(hostId, pendingUsage)
su.SubCount()
@@ -173,6 +178,7 @@ func (self *SHostMemoryPendingUsageStore) DeleteSessionUsage(usage *SessionPendi
}
type SessionPendingUsage struct {
HostId string
SessionId string
Usage *SPendingUsage
countLock *sync.Mutex
@@ -180,10 +186,11 @@ type SessionPendingUsage struct {
cancelCh chan string
}
func NewSessionUsage(sid string) *SessionPendingUsage {
func NewSessionUsage(sid, hostId string) *SessionPendingUsage {
su := &SessionPendingUsage{
HostId: hostId,
SessionId: sid,
Usage: NewPendingUsageBySchedInfo("", nil),
Usage: NewPendingUsageBySchedInfo(hostId, nil),
count: 0,
countLock: new(sync.Mutex),
cancelCh: make(chan string),
@@ -191,6 +198,10 @@ func NewSessionUsage(sid string) *SessionPendingUsage {
return su
}
func (su *SessionPendingUsage) GetHostId() string {
return su.Usage.HostId
}
func (su *SessionPendingUsage) AddCount() {
su.countLock.Lock()
defer su.countLock.Unlock()