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

scheduler: add pending usage process

Browse Source
This commit is contained in:
Zexi
2019-05-27 17:34:06 +08:00
parent 4fb501859c
commit 64cca04a17
44 changed files with 1020 additions and 679 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/climc/shell/schedulers.go
+13
View File
@@ -107,6 +107,19 @@ func init() {
return nil
})
type SchedulerCleanCacheOptions struct {
HostId string `help:"ID of host" short-token:"h"`
SessionId string `help:"Session id" short-token:"s"`
}
R(&SchedulerCleanCacheOptions{}, "scheduler-clean-cache", "Clean scheduler hosts cache",
func(s *mcclient.ClientSession, args *SchedulerCleanCacheOptions) error {
err := modules.SchedManager.CleanCache(s, args.HostId, args.SessionId)
if err != nil {
return err
}
return nil
})
type SchedulerHistoryListOptions struct {
Limit int `default:"50" help:"Page limit"`
Offset int `default:"0" help:"Page offset"`
pkg/apis/scheduler/api.go
+5 -4
View File
@@ -90,10 +90,11 @@ type CandidateNet struct {
}
type CandidateResource struct {
HostId string `json:"host_id"`
Name string `json:"name"`
Disks []*CandidateDisk `json:"disks"`
Nets []*CandidateNet `json:"nets"`
SessionId string `json:"session_id"`
HostId string `json:"host_id"`
Name string `json:"name"`
Disks []*CandidateDisk `json:"disks"`
Nets []*CandidateNet `json:"nets"`
// used by backup schedule
BackupCandidate *CandidateResource `json:"backup_candidate"`
pkg/compute/models/hosts.go
+13 -6
View File
@@ -838,18 +838,25 @@ func (self *SHostManager) GetPropertyBmStartRegisterScript(ctx context.Context,
return res, nil
}
func (maanger *SHostManager) ClearAllSchedDescCache() error {
s := auth.GetAdminSession(context.Background(), options.Options.Region, "")
return modules.SchedManager.CleanCache(s, "")
func (self *SHostManager) ClearAllSchedDescCache() error {
return self.ClearSchedDescSessionCache("", "")
}
func (maanger *SHostManager) ClearSchedDescCache(hostId string) error {
func (self *SHostManager) ClearSchedDescCache(hostId string) error {
return self.ClearSchedDescSessionCache(hostId, "")
}
func (self *SHostManager) ClearSchedDescSessionCache(hostId, sessionId string) error {
s := auth.GetAdminSession(context.Background(), options.Options.Region, "")
return modules.SchedManager.CleanCache(s, hostId)
return modules.SchedManager.CleanCache(s, hostId, sessionId)
}
func (self *SHost) ClearSchedDescCache() error {
return HostManager.ClearSchedDescCache(self.Id)
return self.ClearSchedDescSessionCache("")
}
func (self *SHost) ClearSchedDescSessionCache(sessionId string) error {
return HostManager.ClearSchedDescSessionCache(self.Id, sessionId)
}
func (self *SHost) AllowGetDetailsSpec(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) bool {
pkg/compute/tasks/schedule.go
+3 -3
View File
@@ -245,8 +245,8 @@ func onMasterSlaveScheduleSucc(
lockman.LockObject(ctx, obj)
defer lockman.ReleaseObject(ctx, obj)
task.SaveScheduleResultWithBackup(ctx, obj, master, slave)
models.HostManager.ClearSchedDescCache(master.HostId)
models.HostManager.ClearSchedDescCache(slave.HostId)
models.HostManager.ClearSchedDescSessionCache(master.HostId, master.SessionId)
models.HostManager.ClearSchedDescSessionCache(slave.HostId, slave.SessionId)
}
func onScheduleSucc(
@@ -260,5 +260,5 @@ func onScheduleSucc(
defer lockman.ReleaseRawObject(ctx, models.HostManager.KeywordPlural(), hostId)
task.SaveScheduleResult(ctx, obj, candidate)
models.HostManager.ClearSchedDescCache(candidate.HostId)
models.HostManager.ClearSchedDescSessionCache(candidate.HostId, candidate.SessionId)
}
pkg/mcclient/modules/mod_scheduler.go
+4 -1
View File
@@ -185,11 +185,14 @@ func (this *SchedulerManager) HistoryShow(s *mcclient.ClientSession, id string,
return this._post(s, url, params, "history")
}
func (this *SchedulerManager) CleanCache(s *mcclient.ClientSession, hostId string) error {
func (this *SchedulerManager) CleanCache(s *mcclient.ClientSession, hostId, sessionId string) error {
url := newSchedURL("clean-cache")
if len(hostId) > 0 {
url = fmt.Sprintf("%s/%s", url, hostId)
}
if len(sessionId) > 0 {
url = fmt.Sprintf("%s?session=%s", url, sessionId)
}
resp, err := this.rawRequest(s, "POST", url, nil, nil)
if err != nil {
return err
pkg/scheduler/algorithm/helper.go
-38
View File
@@ -1,38 +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 algorithm
import (
"fmt"
"yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
"yunion.io/x/onecloud/pkg/scheduler/core"
)
func ToHostCandidate(c core.Candidater) (*candidate.HostDesc, error) {
d, ok := c.(*candidate.HostDesc)
if !ok {
return nil, fmt.Errorf("Can't convert %#v to '*candidate.HostDesc'", c)
}
return d, nil
}
func ToBaremetalCandidate(c core.Candidater) (*candidate.BaremetalDesc, error) {
d, ok := c.(*candidate.BaremetalDesc)
if !ok {
return nil, fmt.Errorf("Can't convert %#v to '*candidate.BaremetalDesc'", c)
}
return d, nil
}
pkg/scheduler/algorithm/predicates/baremetal/cpu_predicate.go
+4 -2
View File
@@ -35,10 +35,12 @@ func (p *CPUPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.Pr
h := predicates.NewPredicateHelper(p, u, c)
d := u.SchedData()
freeCPUCount := h.GetInt64("FreeCPUCount", 0)
useRsvd := h.UseReserved()
getter := c.Getter()
freeCPUCount := getter.FreeCPUCount(useRsvd)
reqCPUCount := int64(d.Ncpu)
if freeCPUCount < reqCPUCount {
totalCPUCount := h.GetInt64("CPUCount", 0)
totalCPUCount := getter.TotalCPUCount(useRsvd)
h.AppendInsufficientResourceError(reqCPUCount, totalCPUCount, freeCPUCount)
h.SetCapacity(0)
} else {
pkg/scheduler/algorithm/predicates/baremetal/memory_predicate.go
+4 -2
View File
@@ -35,10 +35,12 @@ func (p *MemoryPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core
h := predicates.NewPredicateHelper(p, u, c)
d := u.SchedData()
freeMemSize := h.GetInt64("FreeMemSize", 0)
useRsvd := h.UseReserved()
getter := c.Getter()
freeMemSize := getter.FreeMemorySize(useRsvd)
reqMemSize := int64(d.Memory)
if freeMemSize < reqMemSize {
totalMemSize := h.GetInt64("MemSize", 0)
totalMemSize := getter.TotalMemorySize(useRsvd)
h.AppendInsufficientResourceError(reqMemSize, totalMemSize, freeMemSize)
h.SetCapacity(0)
} else {
pkg/scheduler/algorithm/predicates/baremetal/network_predicate.go
+3 -6
View File
@@ -60,10 +60,7 @@ func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
h := predicates.NewPredicateHelper(p, u, c)
schedData := u.SchedData()
candidate, err := h.BaremetalCandidate()
if err != nil {
return false, nil, err
}
networks := c.Getter().Networks()
counters := core.NewCounters()
@@ -73,7 +70,7 @@ func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
isRandomNetworkAvailable := func(private bool, exit bool, wire string) string {
var errMsgs []string
for _, network := range candidate.Networks {
for _, network := range networks {
appendError := func(errMsg string) {
errMsgs = append(errMsgs, fmt.Sprintf("%s: %s", network.Id, errMsg))
}
@@ -118,7 +115,7 @@ func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
if network.Network == "" {
return isRandomNetworkAvailable(network.Private, network.Exit, network.Wire)
}
for _, net := range candidate.Networks {
for _, net := range networks {
if (network.Network == net.Id || network.Network == net.Name) && (net.IsPublic || net.ProjectId == schedData.Project) && (net.GetPorts() > 0 || isMigrate()) {
h.SetCapacity(1)
return ""
pkg/scheduler/algorithm/predicates/baremetal/status_predicate.go
+7 -8
View File
@@ -40,22 +40,21 @@ func (p *StatusPredicate) Clone() core.FitPredicate {
func (p *StatusPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) {
h := predicates.NewPredicateHelper(p, u, c)
bm, err := h.BaremetalCandidate()
if err != nil {
return false, nil, err
}
getter := c.Getter()
if !ExpectedStatus.Has(bm.Status) {
h.Exclude2("status", bm.Status, ExpectedStatus)
status := getter.Status()
enabled := getter.Enabled()
if !ExpectedStatus.Has(status) {
h.Exclude2("status", status, ExpectedStatus)
return h.GetResult()
}
if !bm.Enabled {
if !enabled {
h.Exclude2("enable_status", "disable", "enable")
return h.GetResult()
}
if bm.ServerID == "" {
if getter.IsEmpty() {
h.SetCapacity(1)
} else {
h.AppendPredicateFailMsg(predicates.ErrBaremetalHasAlreadyBeenOccupied)
pkg/scheduler/algorithm/predicates/baremetal/storage_predicate.go
+2 -5
View File
@@ -57,14 +57,11 @@ func (p *StoragePredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
h := predicates.NewPredicateHelper(p, u, c)
schedData := u.SchedData()
candidate, err := h.BaremetalCandidate()
if err != nil {
return false, nil, err
}
storageInfo := c.Getter().StorageInfo()
layouts, err := baremetal.CalculateLayout(
schedData.BaremetalDiskConfigs,
candidate.StorageInfo,
storageInfo,
)
if err == nil && baremetal.CheckDisksAllocable(layouts, toBaremetalDisks(schedData.Disks)) {
pkg/scheduler/algorithm/predicates/guest/cpu_predicate.go
+3 -6
View File
@@ -50,16 +50,13 @@ func (f *CPUPredicate) PreExecute(u *core.Unit, cs []core.Candidater) (bool, err
func (f *CPUPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) {
h := predicates.NewPredicateHelper(f, u, c)
d := u.SchedData()
hc, err := h.HostCandidate()
if err != nil {
return false, nil, err
}
useRsvd := h.UseReserved()
freeCPUCount := hc.GetFreeCPUCount(useRsvd)
getter := c.Getter()
freeCPUCount := getter.FreeCPUCount(useRsvd)
reqCPUCount := int64(d.Ncpu)
if freeCPUCount < reqCPUCount {
totalCPUCount := hc.GetTotalCPUCount(useRsvd)
totalCPUCount := getter.TotalCPUCount(useRsvd)
h.AppendInsufficientResourceError(reqCPUCount, totalCPUCount, freeCPUCount)
}
pkg/scheduler/algorithm/predicates/guest/group_predicate.go
+2 -2
View File
@@ -14,7 +14,7 @@
package guest
import (
/*import (
"fmt"
"yunion.io/x/onecloud/pkg/scheduler/algorithm/plugin"
@@ -110,4 +110,4 @@ func (p *GroupPredicate) OnPriorityEnd(u *core.Unit, c core.Candidater) {
p.Name()+":prefer",
))
}
}
}*/
pkg/scheduler/algorithm/predicates/guest/hypervisor_predicate.go
+3 -2
View File
@@ -51,7 +51,8 @@ func hostHasContainerTag(c core.Candidater) bool {
}
func hostAllowRunContainer(c core.Candidater) bool {
hostType := c.Get("HostType")
getter := c.Getter()
hostType := getter.HostType()
if hostType == api.HostTypeKubelet {
return true
}
@@ -65,7 +66,7 @@ func hostAllowRunContainer(c core.Candidater) bool {
func (f *HypervisorPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) {
h := predicates.NewPredicateHelper(f, u, c)
hostType := c.Get("HostType")
hostType := c.Getter().HostType()
guestNeedType := u.SchedData().Hypervisor
if guestNeedType != hostType {
pkg/scheduler/algorithm/predicates/guest/hypervisor_predicate_test.go
-135
View File
@@ -1,135 +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 guest
import (
"reflect"
"testing"
"github.com/stretchr/testify/mock"
"yunion.io/x/jsonutils"
computeapi "yunion.io/x/onecloud/pkg/apis/compute"
schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
"yunion.io/x/onecloud/pkg/scheduler/algorithm/predicates"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/core"
)
func newUnitByHypervisor(hypervisor string) *core.Unit {
info := &api.SchedInfo{
ScheduleInput: &schedapi.ScheduleInput{
ServerConfig: schedapi.ServerConfig{
ServerConfigs: &computeapi.ServerConfigs{
Hypervisor: hypervisor,
},
},
},
}
return core.NewScheduleUnit(info, nil)
}
type FakeCandidater struct {
mock.Mock
}
func (c *FakeCandidater) Getter() core.CandidatePropertyGetter {
return nil
}
func (c *FakeCandidater) IndexKey() string {
return "fake_id"
}
func (c *FakeCandidater) XGet(key string, kind core.Kind) interface{} {
args := c.Called()
return args.String(0)
}
func (c *FakeCandidater) Get(key string) interface{} {
args := c.Called(key)
return args.String(0)
}
func (c *FakeCandidater) Type() int {
return 0
}
func (c *FakeCandidater) GetSchedDesc() *jsonutils.JSONDict {
return nil
}
func (c *FakeCandidater) GetGuestCount() int64 {
return 0
}
func (c *FakeCandidater) GetResourceType() string {
return ""
}
func TestHypervisorPredicate_Execute(t *testing.T) {
type args struct {
u *core.Unit
c core.Candidater
}
aliHost := new(FakeCandidater)
aliHost.On("Get", "HostType").Return(computeapi.HOST_TYPE_ALIYUN)
tests := []struct {
name string
args args
want bool
want1 []core.PredicateFailureReason
wantErr bool
}{
{
name: "hypervisor equals host_type always fits",
args: args{
u: newUnitByHypervisor("aliyun"),
c: aliHost,
},
want: true,
want1: nil,
wantErr: false,
},
{
name: "hypervisor not equals host_type not fit",
args: args{
u: newUnitByHypervisor("kvm"),
c: aliHost,
},
want: false,
want1: []core.PredicateFailureReason{predicates.NewUnexceptedResourceError(`host_hypervisor_runtime is 'aliyun', expected 'kvm'`)},
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
f := &HypervisorPredicate{}
got, got1, err := f.Execute(tt.args.u, tt.args.c)
if (err != nil) != tt.wantErr {
t.Errorf("HypervisorPredicate.Execute() error = %v, wantErr %v", err, tt.wantErr)
return
}
if got != tt.want {
t.Errorf("HypervisorPredicate.Execute() got = %v, want %v", got, tt.want)
}
if !reflect.DeepEqual(got1, tt.want1) {
t.Errorf("HypervisorPredicate.Execute() got1 = %v, want %v", got1, tt.want1)
}
})
}
}
pkg/scheduler/algorithm/predicates/guest/isolated_device_predicate.go
+6 -4
View File
@@ -18,6 +18,7 @@ import (
"fmt"
"yunion.io/x/onecloud/pkg/scheduler/algorithm/predicates"
"yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
"yunion.io/x/onecloud/pkg/scheduler/core"
)
@@ -46,9 +47,10 @@ func (f *IsolatedDevicePredicate) PreExecute(u *core.Unit, cs []core.Candidater)
func (f *IsolatedDevicePredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) {
h := predicates.NewPredicateHelper(f, u, c)
reqIsoDevs := u.SchedData().IsolatedDevices
hc, err := h.HostCandidate()
if err != nil {
return false, nil, err
// TODO: use interface function
hc, ok := c.(*candidate.HostDesc)
if !ok {
return false, nil, fmt.Errorf("Candidater is not *candidate.HostDesc")
}
minCapacity := int64(0xFFFFFFFF)
@@ -71,7 +73,7 @@ func (f *IsolatedDevicePredicate) Execute(u *core.Unit, c core.Candidater) (bool
}
reqCount := len(reqIsoDevs)
freeCount := len(hc.UnusedIsolatedDevices())
freeCount := len(hc.UnusedIsolatedDevices()) - hc.GetPendingUsage().IsolatedDevice
totalCount := len(hc.IsolatedDevices)
// check host isolated device count
pkg/scheduler/algorithm/predicates/guest/memory_predicate.go
+3 -6
View File
@@ -51,16 +51,13 @@ func (p *MemoryPredicate) PreExecute(u *core.Unit, cs []core.Candidater) (bool,
func (p *MemoryPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) {
h := predicates.NewPredicateHelper(p, u, c)
d := u.SchedData()
hc, err := h.HostCandidate()
if err != nil {
return false, nil, err
}
useRsvd := h.UseReserved()
freeMemSize := hc.GetFreeMemSize(useRsvd)
getter := c.Getter()
freeMemSize := getter.FreeMemorySize(useRsvd)
reqMemSize := int64(d.Memory)
if freeMemSize < reqMemSize {
totalMemSize := hc.GetTotalMemSize(useRsvd)
totalMemSize := getter.TotalMemorySize(useRsvd)
h.AppendInsufficientResourceError(reqMemSize, totalMemSize, freeMemSize)
}
pkg/scheduler/algorithm/predicates/guest/nest_predicate.go
+2 -2
View File
@@ -14,7 +14,7 @@
package guest
import (
/*import (
"yunion.io/x/onecloud/pkg/scheduler/algorithm/predicates"
"yunion.io/x/onecloud/pkg/scheduler/core"
)
@@ -51,4 +51,4 @@ func (p *NestPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.P
}
return h.GetResult()
}
}*/
pkg/scheduler/algorithm/predicates/guest/network_predicate.go
+7 -9
View File
@@ -60,10 +60,8 @@ func (p *NetworkPredicate) PreExecute(u *core.Unit, cs []core.Candidater) (bool,
func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) {
h := predicates.NewPredicateHelper(p, u, c)
hc, err := h.HostCandidate()
if err != nil {
return false, nil, err
}
getter := c.Getter()
networks := getter.Networks()
d := u.SchedData()
@@ -78,7 +76,7 @@ func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
counterOfNetwork := func(u *core.Unit, n *models.SNetwork, r int) core.Counter {
counter := u.CounterManager.GetOrCreate("net:"+n.Id, func() core.Counter {
return core.NewNormalCounter(int64(n.GetPorts() - r))
return core.NewNormalCounter(int64(getter.GetFreePort(n.Id) - r))
})
u.SharedResourceManager.Add(n.GetId(), counter)
@@ -91,7 +89,7 @@ func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
var fullErrMsgs []string
found := false
for _, n := range hc.Networks {
for _, n := range networks {
errMsgs := []string{}
appendError := func(errMsg string) {
errMsgs = append(errMsgs, errMsg)
@@ -154,13 +152,13 @@ func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
counters.Add(counters0)
return ret_msg
}
if len(hc.Networks) == 0 {
if len(networks) == 0 {
return predicates.ErrNoAvailableNetwork
}
errMsgs := make([]string, 0)
for _, net := range hc.Networks {
for _, net := range networks {
/*if !isMatchServerType(net) {
errMsgs = append(errMsgs, fmt.Sprintf("%v(%v): server type not matched", net.Name, net.ID))
continue
@@ -197,7 +195,7 @@ func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
var errMsgs []string
for _, n := range d.Networks {
if err_msg := isNetworkAvaliable(n, counters, hc.Networks); err_msg != "" {
if err_msg := isNetworkAvaliable(n, counters, networks); err_msg != "" {
errMsgs = append(errMsgs, err_msg)
}
}
pkg/scheduler/algorithm/predicates/guest/status_predicate.go
+13 -14
View File
@@ -44,14 +44,11 @@ func (p *StatusPredicate) Clone() core.FitPredicate {
func (p *StatusPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) {
h := predicates.NewPredicateHelper(p, u, c)
hc, err := h.HostCandidate()
if err != nil {
return false, nil, err
}
curStatus := hc.Status
curHostStatus := hc.HostStatus
curEnableStatus := hc.Enabled
getter := c.Getter()
curStatus := getter.Status()
curHostStatus := getter.HostStatus()
curEnableStatus := getter.Enabled()
if curStatus != ExpectedStatus {
h.Exclude2("status", curStatus, ExpectedStatus)
@@ -65,16 +62,18 @@ func (p *StatusPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core
h.Exclude2("enable_status", curEnableStatus, true)
}
if hc.Zone.Status != ExpectedEnableStatus {
h.Exclude2("zone_status", hc.Zone.Status, ExpectedEnableStatus)
zone := getter.Zone()
if zone.Status != ExpectedEnableStatus {
h.Exclude2("zone_status", zone.Status, ExpectedEnableStatus)
}
if hc.Cloudprovider != nil {
if !utils.IsInStringArray(hc.Cloudprovider.Status, api.CLOUD_PROVIDER_VALID_STATUS) {
h.Exclude2("cloud_provider_status", hc.Cloudprovider.Status, api.CLOUD_PROVIDER_VALID_STATUS)
cloudprovider := getter.Cloudprovider()
if cloudprovider != nil {
if !utils.IsInStringArray(cloudprovider.Status, api.CLOUD_PROVIDER_VALID_STATUS) {
h.Exclude2("cloud_provider_status", cloudprovider.Status, api.CLOUD_PROVIDER_VALID_STATUS)
}
if hc.Cloudprovider.HealthStatus != api.CLOUD_PROVIDER_HEALTH_NORMAL {
h.Exclude2("cloud_provider_health_status", hc.Cloudprovider.HealthStatus, api.CLOUD_PROVIDER_HEALTH_NORMAL)
if cloudprovider.HealthStatus != api.CLOUD_PROVIDER_HEALTH_NORMAL {
h.Exclude2("cloud_provider_health_status", cloudprovider.HealthStatus, api.CLOUD_PROVIDER_HEALTH_NORMAL)
}
}
pkg/scheduler/algorithm/predicates/guest/storage_predicate.go
+5 -8
View File
@@ -50,12 +50,9 @@ func (p *StoragePredicate) PreExecute(u *core.Unit, cs []core.Candidater) (bool,
func (p *StoragePredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) {
h := predicates.NewPredicateHelper(p, u, c)
hc, err := h.HostCandidate()
if err != nil {
return false, nil, err
}
d := u.SchedData()
getter := c.Getter()
storages := getter.Storages()
isMigrate := func() bool {
return len(d.HostId) > 0
@@ -66,7 +63,7 @@ func (p *StoragePredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
}
isStorageAccessible := func(storage string) bool {
for _, s := range hc.Storages {
for _, s := range storages {
if storage == s.Id || storage == s.Name {
return true
}
@@ -76,7 +73,7 @@ func (p *StoragePredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
}
getStorageCapacity := func(backend string, reqMaxSize int64, reqTotalSize int64, useRsvd bool) (int64, int64) {
totalFree := hc.GetFreeStorageSizeOfType(backend, useRsvd)
totalFree := getter.GetFreeStorageSizeOfType(backend, useRsvd)
capacity := totalFree / utils.Max(reqTotalSize, 1)
return capacity, totalFree
@@ -95,7 +92,7 @@ func (p *StoragePredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor
getStorageFreeStr := func(backend string, useRsvd bool) string {
ss := []string{}
for _, s := range hc.Storages {
for _, s := range getter.Storages() {
if s.StorageType == backend {
total := int64(float32(s.Capacity) * s.Cmtbound)
used := s.GetUsedCapacity(tristate.True)
pkg/scheduler/algorithm/predicates/predicates.go
+1 -37
View File
@@ -26,12 +26,9 @@ import (
computeapi "yunion.io/x/onecloud/pkg/apis/compute"
"yunion.io/x/onecloud/pkg/compute/models"
"yunion.io/x/onecloud/pkg/scheduler/algorithm"
"yunion.io/x/onecloud/pkg/scheduler/algorithm/plugin"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
"yunion.io/x/onecloud/pkg/scheduler/core"
"yunion.io/x/onecloud/pkg/scheduler/data_manager"
)
// BasePredicate is a default struct for all the predicates that will
@@ -112,7 +109,7 @@ func (h *PredicateHelper) AppendPredicateFailMsg(reason string) {
func (h *PredicateHelper) AppendInsufficientResourceError(req, total, free int64) {
h.AppendPredicateFail(
NewInsufficientResourceError(h.Candidate.Get("Name").(string), req, total, free))
NewInsufficientResourceError(h.Candidate.Getter().Name(), req, total, free))
}
// SetCapacity returns the current resource capacity calculated by a filter.
@@ -144,39 +141,6 @@ func (h *PredicateHelper) Exclude2(predicateName string, current, expected inter
h.Exclude(fmt.Sprintf("%s is '%v', expected '%v'", predicateName, current, expected))
}
func (h *PredicateHelper) Get(key string) interface{} {
return h.Candidate.Get(key)
}
func (h *PredicateHelper) GetInt64(key string, def int64) int64 {
value := h.Get(key)
if value == nil {
return def
}
return value.(int64)
}
func (h *PredicateHelper) GetGroupCounts() (*data_manager.GroupResAlgorithmResult, error) {
value := h.Get("Groups")
if value == nil {
return nil, nil
}
if r, ok := value.(*data_manager.GroupResAlgorithmResult); ok {
return r, nil
}
return nil, fmt.Errorf("type error: not *data_manager.GroupResAlgorithmResult (GetGroupCounts)")
}
func (h *PredicateHelper) HostCandidate() (*candidate.HostDesc, error) {
return algorithm.ToHostCandidate(h.Candidate)
}
func (h *PredicateHelper) BaremetalCandidate() (*candidate.BaremetalDesc, error) {
return algorithm.ToBaremetalCandidate(h.Candidate)
}
// UseReserved check whether the unit can use guest reserved resource
func (h *PredicateHelper) UseReserved() bool {
usable := false
pkg/scheduler/algorithm/priorities/guest/avoid_same_host.go
+1 -6
View File
@@ -34,13 +34,8 @@ func (p *AvoidSameHostPriority) Clone() core.Priority {
func (p *AvoidSameHostPriority) Map(u *core.Unit, c core.Candidater) (core.HostPriority, error) {
h := priorities.NewPriorityHelper(p, u, c)
hc, err := p.HostCandidate(c)
if err != nil {
return core.HostPriority{}, err
}
ownerTenantID := u.SchedData().Project
if count, ok := hc.Tenants[ownerTenantID]; ok && count > 0 {
if count, ok := c.Getter().ProjectGuests()[ownerTenantID]; ok && count > 0 {
h.SetFrontRawScore(-1 * int(count))
}
pkg/scheduler/algorithm/priorities/guest/creating.go
+3 -8
View File
@@ -32,16 +32,11 @@ func (p *CreatingPriority) Clone() core.Priority {
}
func (p *CreatingPriority) Map(u *core.Unit, c core.Candidater) (core.HostPriority, error) {
h := priorities.NewPriorityHelper(p, u, c)
hc, err := p.HostCandidate(c)
if err != nil {
return core.HostPriority{}, err
}
if hc.CreatingGuestCount > 0 {
score := -int(hc.CreatingGuestCount)
creatingGuestCount := c.Getter().CreatingGuestCount()
if creatingGuestCount > 0 {
score := -int(creatingGuestCount)
h.SetFrontScore(score)
}
pkg/scheduler/algorithm/priorities/guest/lowload.go
+3 -7
View File
@@ -35,13 +35,9 @@ func (p *LowLoadPriority) Clone() core.Priority {
func (p *LowLoadPriority) Map(u *core.Unit, c core.Candidater) (core.HostPriority, error) {
h := priorities.NewPriorityHelper(p, u, c)
hc, err := p.HostCandidate(c)
if err != nil {
return core.HostPriority{}, err
}
cpuCommitRate := float64(hc.RunningCPUCount) / float64(hc.TotalCPUCount)
memCommitRate := float64(hc.RunningMemSize) / float64(hc.TotalMemSize)
getter := c.Getter()
cpuCommitRate := float64(getter.RunningCPUCount()) / float64(getter.TotalCPUCount(false))
memCommitRate := float64(getter.RunningMemorySize()) / float64(getter.TotalMemorySize(false))
if cpuCommitRate < 0.5 && memCommitRate < 0.5 {
score := 10 * (1 - cpuCommitRate - memCommitRate)
h.SetScore(int(score))
pkg/scheduler/algorithm/priorities/priorities.go
-6
View File
@@ -15,8 +15,6 @@
package priorities
import (
"yunion.io/x/onecloud/pkg/scheduler/algorithm"
"yunion.io/x/onecloud/pkg/scheduler/cache/candidate"
"yunion.io/x/onecloud/pkg/scheduler/core"
"yunion.io/x/onecloud/pkg/scheduler/core/score"
)
@@ -105,10 +103,6 @@ func (b *BasePriority) Name() string {
return "base_priorites_should_not_be_called"
}
func (b *BasePriority) HostCandidate(c core.Candidater) (*candidate.HostDesc, error) {
return algorithm.ToHostCandidate(c)
}
func (b *BasePriority) ScoreIntervals() score.Intervals {
return score.NewIntervals(0, 1, 2)
}
pkg/scheduler/algorithmprovider/defaults.go
+2 -2
View File
@@ -33,8 +33,8 @@ func defaultPredicates() sets.String {
factory.RegisterFitPredicate("b-GuestHypervisorFilter", &predicateguest.HypervisorPredicate{}),
factory.RegisterFitPredicate("c-GuestAggregateFilter", &predicates.AggregatePredicate{}),
factory.RegisterFitPredicate("d-GuestMigrateFilter", &predicateguest.MigratePredicate{}),
factory.RegisterFitPredicate("e-GuestNestFilter", &predicateguest.NestPredicate{}),
factory.RegisterFitPredicate("f-GuestGroupFilter", &predicateguest.GroupPredicate{}),
//factory.RegisterFitPredicate("e-GuestNestFilter", &predicateguest.NestPredicate{}),
//factory.RegisterFitPredicate("f-GuestGroupFilter", &predicateguest.GroupPredicate{}),
factory.RegisterFitPredicate("g-GuestCPUFilter", &predicateguest.CPUPredicate{}),
factory.RegisterFitPredicate("h-GuestMemoryFilter", &predicateguest.MemoryPredicate{}),
factory.RegisterFitPredicate("i-GuestStorageFilter", &predicateguest.StoragePredicate{}),
pkg/scheduler/api/expire.go
+1
View File
@@ -17,6 +17,7 @@ package api
type ExpireArgs struct {
DirtyHosts []string
DirtyBaremetals []string
SessionId string
}
type ExpireResult struct {
pkg/scheduler/cache/candidate/baremetals.go
Vendored
+32 -63
View File
@@ -32,6 +32,34 @@ import (
computemodels "yunion.io/x/onecloud/pkg/compute/models"
)
type baremetalGetter struct {
*baseHostGetter
bm *BaremetalDesc
}
func newBaremetalGetter(bm *BaremetalDesc) *baremetalGetter {
return &baremetalGetter{
baseHostGetter: newBaseHostGetter(bm.BaseHostDesc),
bm: bm,
}
}
func (h baremetalGetter) FreeCPUCount(_ bool) int64 {
return h.bm.FreeCPUCount()
}
func (h baremetalGetter) FreeMemorySize(_ bool) int64 {
return h.bm.FreeMemSize()
}
func (h baremetalGetter) IsEmpty() bool {
return h.bm.ServerID == ""
}
func (h baremetalGetter) StorageInfo() []*baremetal.BaremetalStorage {
return h.bm.StorageInfo
}
type BaremetalDesc struct {
*BaseHostDesc
@@ -49,6 +77,10 @@ type BaremetalBuilder struct {
residentTenantDict map[string]map[string]interface{}
}
func (bd *BaremetalDesc) Getter() core.CandidatePropertyGetter {
return newBaremetalGetter(bd)
}
func (bd *BaremetalDesc) String() string {
s, _ := fjson.Marshal(bd)
return string(s)
@@ -59,65 +91,6 @@ func (bd *BaremetalDesc) Type() int {
return 1
}
// TODO: remove this ugly code
func (bd *BaremetalDesc) Get(key string) interface{} {
switch key {
case "ID":
return bd.Id
case "Name":
return bd.Name
case "Status":
return bd.Status
case "ZoneID":
return bd.ZoneId
case "ServerID":
return bd.ServerID
case "CPUCount":
return int64(bd.CpuCount)
case "FreeCPUCount":
return bd.FreeCPUCount()
case "NodeCount":
return int64(bd.NodeCount)
case "MemSize":
return bd.MemSize
case "FreeMemSize":
return bd.FreeMemSize()
case "Storages":
return bd.StorageType
case "StorageSize":
return bd.StorageSize
case "StorageType":
return bd.StorageType
case "StorageInfo":
return bd.StorageInfo
case "StorageDriver":
return bd.StorageDriver
case "FreeStorageSize":
return bd.FreeStorageSize()
case "HostStatus":
return bd.HostStatus
default:
return nil
}
}
func (bd *BaremetalDesc) GetGuestCount() int64 {
if bd.ServerID == "" {
return 0
@@ -125,10 +98,6 @@ func (bd *BaremetalDesc) GetGuestCount() int64 {
return 1
}
func (bd *BaremetalDesc) XGet(key string, kind core.Kind) interface{} {
return core.XGetCalculator(bd, key, kind)
}
func (bd *BaremetalDesc) IndexKey() string {
return bd.Id
}
pkg/scheduler/cache/candidate/base.go
Vendored
+78 -5
View File
@@ -23,12 +23,12 @@ import (
"yunion.io/x/sqlchemy"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/core"
"yunion.io/x/onecloud/pkg/scheduler/db/models"
computeapi "yunion.io/x/onecloud/pkg/apis/compute"
computedb "yunion.io/x/onecloud/pkg/cloudcommon/db"
computemodels "yunion.io/x/onecloud/pkg/compute/models"
schedmodels "yunion.io/x/onecloud/pkg/scheduler/models"
)
type BaseHostDesc struct {
@@ -52,10 +52,6 @@ func newBaseHostGetter(h *BaseHostDesc) *baseHostGetter {
return &baseHostGetter{h}
}
func (b *BaseHostDesc) Getter() core.CandidatePropertyGetter {
return newBaseHostGetter(b)
}
func (b baseHostGetter) Id() string {
return b.h.GetId()
}
@@ -68,6 +64,10 @@ func (b baseHostGetter) Zone() *computemodels.SZone {
return b.h.Zone
}
func (b baseHostGetter) Cloudprovider() *computemodels.SCloudprovider {
return b.h.Cloudprovider
}
func (b baseHostGetter) Region() *computemodels.SCloudregion {
return b.h.Region
}
@@ -96,6 +96,56 @@ func (b baseHostGetter) NetInterfaces() map[string][]computemodels.SNetInterface
return b.h.NetInterfaces
}
func (b baseHostGetter) Status() string {
return b.h.Status
}
func (b baseHostGetter) HostStatus() string {
return b.h.HostStatus
}
func (b baseHostGetter) Enabled() bool {
return b.h.Enabled
}
func (b baseHostGetter) ProjectGuests() map[string]int64 {
return b.h.Tenants
}
func (b baseHostGetter) CreatingGuestCount() int {
return 0
}
func (b baseHostGetter) RunningCPUCount() int64 {
return 0
}
func (b baseHostGetter) TotalCPUCount(_ bool) int64 {
return int64(b.h.CpuCount)
}
func (b baseHostGetter) RunningMemorySize() int64 {
return 0
}
func (b baseHostGetter) TotalMemorySize(_ bool) int64 {
return int64(b.h.MemSize)
}
func (b baseHostGetter) GetFreeStorageSizeOfType(storageType string, useRsvd bool) int64 {
var size int64
for _, s := range b.Storages() {
if s.StorageType == storageType {
size += int64(float32(s.Capacity) * s.Cmtbound)
}
}
return size
}
func (b baseHostGetter) GetFreePort(netId string) int {
return b.h.GetFreePort(netId)
}
func reviseResourceType(resType string) string {
if resType == "" {
return computeapi.HostResourceTypeDefault
@@ -154,6 +204,29 @@ func (b BaseHostDesc) GetSchedDesc() *jsonutils.JSONDict {
return desc
}
func (b *BaseHostDesc) GetPendingUsage() *schedmodels.SPendingUsage {
usage, err := schedmodels.HostPendingUsageManager.GetPendingUsage(b.GetId())
if err != nil {
return schedmodels.NewPendingUsageBySchedInfo(b.GetId(), nil)
}
return usage
}
func (b *BaseHostDesc) GetFreePort(netId string) int {
var selNet *api.CandidateNetwork = nil
for _, n := range b.Networks {
if n.GetId() == netId {
selNet = n
break
}
}
if selNet == nil {
return 0
}
freeCount, _ := selNet.GetFreeAddressCount()
return freeCount
}
func (b BaseHostDesc) GetResourceType() string {
return b.ResourceType
}
pkg/scheduler/cache/candidate/hosts.go
Vendored
+70 -122
View File
@@ -31,12 +31,69 @@ import (
computeapi "yunion.io/x/onecloud/pkg/apis/compute"
computedb "yunion.io/x/onecloud/pkg/cloudcommon/db"
"yunion.io/x/onecloud/pkg/compute/baremetal"
computemodels "yunion.io/x/onecloud/pkg/compute/models"
"yunion.io/x/onecloud/pkg/scheduler/core"
"yunion.io/x/onecloud/pkg/scheduler/db/models"
o "yunion.io/x/onecloud/pkg/scheduler/options"
)
type hostGetter struct {
*baseHostGetter
h *HostDesc
}
func newHostGetter(h *HostDesc) *hostGetter {
return &hostGetter{
baseHostGetter: newBaseHostGetter(h.BaseHostDesc),
h: h,
}
}
func (h *hostGetter) CreatingGuestCount() int {
return int(h.h.CreatingGuestCount)
}
func (h *hostGetter) RunningCPUCount() int64 {
return h.h.RunningCPUCount
}
func (h *hostGetter) TotalCPUCount(useRsvd bool) int64 {
return h.h.GetTotalCPUCount(useRsvd)
}
func (h *hostGetter) FreeCPUCount(useRsvd bool) int64 {
return h.h.GetFreeCPUCount(useRsvd)
}
func (h *hostGetter) FreeMemorySize(useRsvd bool) int64 {
return h.h.GetFreeMemSize(useRsvd)
}
func (h *hostGetter) RunningMemorySize() int64 {
return h.h.RunningMemSize
}
func (h *hostGetter) TotalMemorySize(useRsvd bool) int64 {
return h.h.GetTotalMemSize(useRsvd)
}
func (h *hostGetter) IsEmpty() bool {
return h.h.GuestCount == 0
}
func (h *hostGetter) StorageInfo() []*baremetal.BaremetalStorage {
return nil
}
func (h *hostGetter) GetFreeStorageSizeOfType(storageType string, useRsvd bool) int64 {
return h.h.GetFreeStorageSizeOfType(storageType, useRsvd)
}
func (h *hostGetter) GetFreePort(netId string) int {
return h.h.GetFreePort(netId)
}
type HostDesc struct {
*BaseHostDesc
@@ -161,38 +218,6 @@ func NewGuestReservedResourceUsedByBuilder(b *HostBuilder, host *computemodels.S
return
}
//type Storage struct {
//ID string `json:"id"`
//Name string `json:"name"`
//Capacity int64 `json:"capacity"`
//StorageType string `json:"type"`
//UsedCapacity int64 `json:"used"`
//WasteCapacity int64 `json:"waste"`
//FreeCapacity int64 `json:"free"`
//VCapacity int64 `json:"vcapacity"`
//Cmtbound float64 `json:"cmtbound"`
//StorageDriver string `json:"driver"`
//Adapter string `json:"adapter"`
//Splits []string `json:"splits"`
//Range string `json:"range"`
//Conf string `json:"conf"`
//MinStripSize int `json:"min_strip_size"`
//MaxStripSize int `json:"max_strip_size"`
//Size int `json:"size"`
//}
//func (storage *Storage) GetFreeSize() int64 {
//return storage.GetTotalSize() - storage.UsedCapacity - storage.WasteCapacity
//}
//func (storage *Storage) GetTotalSize() int64 {
//return int64(float64(storage.Capacity) * storage.Cmtbound)
//}
//func (storage *Storage) IsLocal() bool {
//return utils.IsLocalStorage(storage.StorageType)
//}
type HostBuilder struct {
residentTenantDict map[string]map[string]interface{}
@@ -240,96 +265,14 @@ func (h *HostDesc) Type() int {
return 0
}
func (h *HostDesc) Getter() core.CandidatePropertyGetter {
return newHostGetter(h)
}
func (h *HostDesc) GetGuestCount() int64 {
return h.GuestCount
}
// TODO: remove this ugly code
func (h *HostDesc) Get(key string) interface{} {
switch key {
case "ID":
return h.Id
case "Name":
return h.Name
case "CPUCount":
return h.CpuCount
case "MemSize":
return h.MemSize
case "ZoneID":
return h.ZoneId
case "TotalCPUCount":
return h.GetTotalCPUCount(true)
case "FreeCPUCount":
return h.GetFreeCPUCount(false)
case "TotalMemSize":
return h.GetTotalMemSize(true)
case "FreeMemSize":
return h.GetFreeMemSize(false)
case "Groups":
return h.Groups
case "IsolatedDevices":
return h.IsolatedDevices
case "Status":
return h.Status
case "TotalStorageSize":
return h.totalStorageSize(false, true)
case "TotalLocalStorageSize":
return h.totalStorageSize(true, true)
case "FreeStorageSize":
return h.freeStorageSize(false, false)
case "FreeLocalStorageSize":
return h.freeStorageSize(true, false)
case "StorageTypes":
return h.StorageTypes
case "HostStatus":
return h.HostStatus
case "EnableStatus":
return h.GetEnableStatus()
case "HostType":
return h.HostType
case "IsBaremetal":
return h.IsBaremetal
default:
index := strings.Index(key, ":")
if index >= 0 {
masterKey := key[0:index]
slaveKey := key[index+1:]
switch masterKey {
case "FreeStorageSize":
storageType := slaveKey
return h.freeStorageSizeOfType(storageType, false)
}
}
return nil
}
}
func (h *HostDesc) XGet(key string, kind core.Kind) interface{} {
return core.XGetCalculator(h, key, kind)
}
func (h *HostDesc) GetTotalLocalStorageSize(useRsvd bool) int64 {
return h.totalStorageSize(true, useRsvd)
}
@@ -378,7 +321,7 @@ func (h *HostDesc) GetFreeStorageSizeOfType(sType string, useRsvd bool) int64 {
}
func (h *HostDesc) freeStorageSizeOfType(storageType string, useRsvd bool) int64 {
total := int64(0)
var total int64
for _, storage := range h.Storages {
if storage.StorageType == storageType {
total += int64(storage.GetFreeCapacity())
@@ -395,7 +338,12 @@ func (h *HostDesc) freeStorageSizeOfType(storageType string, useRsvd bool) int64
return reservedResourceAddCal(total, h.GuestReservedStorageSizeFree(), useRsvd)
}
return total
return total - int64(h.GetPendingUsage().DiskUsage.Get(storageType))
}
func (h *HostDesc) GetFreePort(netId string) int {
freeCnt := h.BaseHostDesc.GetFreePort(netId)
return freeCnt - h.GetPendingUsage().NetUsage.Get(netId)
}
func reservedResourceCal(
@@ -426,7 +374,7 @@ func (h *HostDesc) GetTotalMemSize(useRsvd bool) int64 {
}
func (h *HostDesc) GetFreeMemSize(useRsvd bool) int64 {
return reservedResourceAddCal(h.FreeMemSize, h.GuestReservedMemSizeFree(), useRsvd)
return reservedResourceAddCal(h.FreeMemSize, h.GuestReservedMemSizeFree(), useRsvd) - int64(h.GetPendingUsage().Memory)
}
func (h *HostDesc) GuestReservedMemSizeFree() int64 {
@@ -458,7 +406,7 @@ func (h *HostDesc) GetTotalCPUCount(useRsvd bool) int64 {
}
func (h *HostDesc) GetFreeCPUCount(useRsvd bool) int64 {
return reservedResourceAddCal(h.FreeCPUCount, h.GuestReservedCPUCountFree(), useRsvd)
return reservedResourceAddCal(h.FreeCPUCount, h.GuestReservedCPUCountFree(), useRsvd) - int64(h.GetPendingUsage().Cpu)
}
func (h *HostDesc) IndexKey() string {
pkg/scheduler/core/generic_scheduler.go
+3 -8
View File
@@ -77,7 +77,7 @@ type Scheduler interface {
// mark already selected candidates dirty that
// can't be use again until cleanup them
DirtySelectedCandidates([]*SelectedCandidate)
//DirtySelectedCandidates([]*SelectedCandidate)
}
type GenericScheduler struct {
@@ -169,11 +169,6 @@ func (g *GenericScheduler) Schedule(unit *Unit, candidates []Candidater) (*Sched
return nil, err
}
// sync schedule candidates dirty mark
if !isSuggestion && !unit.SkipDirtyMarkHost() {
g.DirtySelectedCandidates(selectedCandidates)
}
return &SchedResultItemList{Unit: unit, Data: resultItems}, nil
}
@@ -184,7 +179,7 @@ func newSchedResultByCtx(u *Unit, count int64, c Candidater) *SchedResultItem {
ID: id,
Count: count,
Capacity: u.GetCapacity(id),
Name: fmt.Sprintf("%v", c.Get("Name")),
Name: c.Getter().Name(),
Score: u.GetScore(id).DigitString(),
Data: u.GetFiltedData(id, count),
Candidater: c,
@@ -534,7 +529,7 @@ func unitFitsOnCandidate(
}
}
candidateLogIndex := fmt.Sprintf("%v:%s", candidate.Get("Name"), candidate.IndexKey())
candidateLogIndex := fmt.Sprintf("%v:%s", candidate.Getter().Name(), candidate.IndexKey())
return NewSchedLog(candidateLogIndex, stage, fmt.Sprintf("%v %v", sFit, message), !fit)
}
pkg/scheduler/core/resource.go
-15
View File
@@ -75,18 +75,3 @@ func ReservedSub(key string, value value_t, reserved value_t) value_t {
}
return value
}
func XGetCalculator(c Candidater, key string, kind Kind) value_t {
value := c.Get(key)
switch kind {
case KindFree:
// TODO: reserved not impl by now
return ReservedSub(key, value, nil)
case KindRaw:
return value
case KindReserved:
return nil
}
return nil
}
pkg/scheduler/core/types.go
+21 -4
View File
@@ -18,6 +18,7 @@ import (
"yunion.io/x/jsonutils"
schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
"yunion.io/x/onecloud/pkg/compute/baremetal"
computemodels "yunion.io/x/onecloud/pkg/compute/models"
"yunion.io/x/onecloud/pkg/scheduler/api"
"yunion.io/x/onecloud/pkg/scheduler/core/score"
@@ -55,13 +56,33 @@ type CandidatePropertyGetter interface {
Id() string
Name() string
Zone() *computemodels.SZone
Cloudprovider() *computemodels.SCloudprovider
Region() *computemodels.SCloudregion
HostType() string
HostSchedtags() []computemodels.SSchedtag
Storages() []*api.CandidateStorage
Networks() []*api.CandidateNetwork
Status() string
HostStatus() string
Enabled() bool
IsEmpty() bool
ResourceType() string
NetInterfaces() map[string][]computemodels.SNetInterface
ProjectGuests() map[string]int64
CreatingGuestCount() int
RunningCPUCount() int64
TotalCPUCount(useRsvd bool) int64
FreeCPUCount(useRsvd bool) int64
RunningMemorySize() int64
TotalMemorySize(useRsvd bool) int64
FreeMemorySize(useRsvd bool) int64
StorageInfo() []*baremetal.BaremetalStorage
GetFreeStorageSizeOfType(storageType string, useRsvd bool) int64
GetFreePort(netId string) int
}
// Candidater replace host Candidate resource info
@@ -69,10 +90,6 @@ type Candidater interface {
Getter() CandidatePropertyGetter
// IndexKey return candidate cache item's ident, usually host ID
IndexKey() string
// Get return candidate cache item's value by key
Get(key string) interface{}
// XGet return candidate cache item's value by key and kind
XGet(key string, kind Kind) interface{}
Type() int
GetSchedDesc() *jsonutils.JSONDict
pkg/scheduler/data_manager/candidate_manager.go
+16 -35
View File
@@ -18,7 +18,7 @@ import (
"fmt"
"time"
"yunion.io/x/pkg/util/ttlpool"
//"yunion.io/x/pkg/util/ttlpool"
"yunion.io/x/pkg/utils"
"yunion.io/x/onecloud/pkg/scheduler/cache"
@@ -196,13 +196,6 @@ type CandidateManager struct {
stopCh <-chan struct{}
dataManager *DataManager
impls map[string]*CandidateManagerImpl
dirtyPool *ttlpool.CountPool
}
func (cm *CandidateManager) DirtyPoolHas(id string) bool {
ok, _ := cm.dirtyPool.HasByKey(id)
return ok
}
func (cm *CandidateManager) GetCandidates(args CandidateGetArgs) ([]core.Candidater, error) {
@@ -248,10 +241,6 @@ func (cm *CandidateManager) GetCandidates(args CandidateGetArgs) ([]core.Candida
for _, c := range candidates {
r := c.(core.Candidater)
if cm.DirtyPoolHas(r.IndexKey()) {
continue
}
if !matchRegion(r, args.RegionID) {
continue
}
@@ -278,10 +267,6 @@ func (cm *CandidateManager) GetCandidatesByIds(resType string, ids []string) ([]
candidates := []core.Candidater{}
for _, id := range ids {
if cm.DirtyPoolHas(id) {
continue
}
c, err2 := impl.GetCandidate(id)
if err2 != nil {
return nil, err2
@@ -298,10 +283,6 @@ func (cm *CandidateManager) GetCandidate(id string, resType string) (interface{}
return nil, err
}
if cm.DirtyPoolHas(id) {
return nil, fmt.Errorf("%s in dirtyPool", id)
}
c, err := impl.GetCandidate(id)
if err != nil {
return nil, err
@@ -332,7 +313,7 @@ func NewCandidateManager(dataManager *DataManager, stopCh <-chan struct{}) *Cand
stopCh: stopCh,
impls: make(map[string]*CandidateManagerImpl),
dataManager: dataManager,
dirtyPool: ttlpool.NewCountPool(),
//dirtyPool: ttlpool.NewCountPool(),
}
candidateManager.AddImpl("host", NewCandidateManagerImpl(
@@ -385,22 +366,22 @@ func (cm *CandidateManager) ReloadAll(resType string) ([]interface{}, error) {
return impl.ReloadAll()
}
type IDirtyPoolItem interface {
ttlpool.Item
GetCount() uint64
}
//type IDirtyPoolItem interface {
//ttlpool.Item
//GetCount() uint64
//}
func (cm *CandidateManager) SetCandidateDirty(item IDirtyPoolItem) {
cm.dirtyPool.Add(item, item.GetCount())
}
//func (cm *CandidateManager) SetCandidateDirty(item IDirtyPoolItem) {
//cm.dirtyPool.Add(item, item.GetCount())
//}
func (cm *CandidateManager) CleanDirtyCandidatesOnce(keys []string) {
for _, key := range keys {
cm.dirtyPool.DeleteByKey(key)
}
}
//func (cm *CandidateManager) CleanDirtyCandidatesOnce(keys []string, sessionId string) {
//for _, key := range keys {
//cm.dirtyPool.DeleteByKey(key)
//}
//}
func ToHostCandidate(c interface{}) (*candidatecache.HostDesc, error) {
/*func ToHostCandidate(c interface{}) (*candidatecache.HostDesc, error) {
h, ok := c.(*candidatecache.HostDesc)
if !ok {
return nil, fmt.Errorf("can't convert %#v to *candidatecache.HostDesc", c)
@@ -418,4 +399,4 @@ func ToHostCandidates(cs []core.Candidater) ([]*candidatecache.HostDesc, error)
hs = append(hs, h)
}
return hs, nil
}
}*/
pkg/scheduler/handler/backup_helper.go
+11 -16
View File
@@ -22,16 +22,16 @@ import (
schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
"yunion.io/x/onecloud/pkg/scheduler/core"
schedman "yunion.io/x/onecloud/pkg/scheduler/manager"
//schedman "yunion.io/x/onecloud/pkg/scheduler/manager"
)
func transToBackupSchedResult(result *core.SchedResultItemList, preferMasterHost, preferBackupHost string, count int64, setDirty bool) *schedapi.ScheduleOutput {
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, setDirty)
apiResults := newBackupSchedResult(result, preferMasterHost, preferBackupHost, count, sid)
return apiResults
}
@@ -39,13 +39,13 @@ func newBackupSchedResult(
result *core.SchedResultItemList,
preferMasterHost, preferBackupHost string,
count int64,
setDirty bool,
sid string,
) *schedapi.ScheduleOutput {
ret := new(schedapi.ScheduleOutput)
apiResults := make([]*schedapi.CandidateResource, 0)
for i := 0; i < int(count); i++ {
log.V(10).Debugf("Select backup host from result: %s", result)
target, err := getSchedBackupResult(result, preferMasterHost, preferBackupHost, setDirty)
target, err := getSchedBackupResult(result, preferMasterHost, preferBackupHost, sid)
if err != nil {
er := &schedapi.CandidateResource{Error: err.Error()}
apiResults = append(apiResults, er)
@@ -60,7 +60,7 @@ func newBackupSchedResult(
func getSchedBackupResult(
result *core.SchedResultItemList,
preferMasterHost, preferBackupHost string,
setDirty bool,
sid string,
) (*schedapi.CandidateResource, error) {
masterHost := selectMasterHost(result.Data, preferMasterHost, preferBackupHost)
if masterHost == nil {
@@ -71,21 +71,20 @@ func getSchedBackupResult(
return nil, fmt.Errorf("Can't find backup host %q by master %q", preferBackupHost, masterHost.ID)
}
markHostUsed(masterHost, setDirty)
markHostUsed(backupHost, setDirty)
markHostUsed(masterHost)
markHostUsed(backupHost)
sort.Sort(sort.Reverse(result))
ret := masterHost.ToCandidateResource()
ret.BackupCandidate = backupHost.ToCandidateResource()
ret.SessionId = sid
ret.BackupCandidate.SessionId = sid
return ret, nil
}
func markHostUsed(host *core.SchedResultItem, setDirty bool) {
func markHostUsed(host *core.SchedResultItem) {
host.Count++
host.Capacity--
if setDirty {
setHostDirty(host)
}
}
// selectMasterID find master host id run VM
@@ -160,7 +159,3 @@ func (a *dirtyItemAdapter) Index() (string, error) {
func (a *dirtyItemAdapter) GetCount() uint64 {
return uint64(a.Count)
}
func setHostDirty(host *core.SchedResultItem) {
schedman.GetCandidateManager().SetCandidateDirty(&dirtyItemAdapter{SchedResultItem: host})
}
pkg/scheduler/handler/forecast_helper.go
+8 -5
View File
@@ -26,7 +26,6 @@ func transToSchedForecastResult(result *core.SchedResultItemList) interface{} {
unit := result.Unit
schedData := unit.SchedData()
reqCount := int64(schedData.Count)
var readyCount int64
filters := make([]*api.ForecastFilter, 0)
filtersMap := make(map[string]*api.ForecastFilter)
@@ -45,7 +44,10 @@ func transToSchedForecastResult(result *core.SchedResultItemList) interface{} {
}
logIndex := func(item *core.SchedResultItem) string {
return fmt.Sprintf("%s:%s", item.Candidater.Get("Name"), item.Candidater.Get("ID"))
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 {
@@ -77,12 +79,14 @@ func transToSchedForecastResult(result *core.SchedResultItemList) interface{} {
}
var output *schedapi.ScheduleOutput
sid := schedData.SessionId
if schedData.Backup {
output = transToBackupSchedResult(result, schedData.PreferHost, schedData.PreferBackupHost, int64(schedData.Count), false)
output = transToBackupSchedResult(result, schedData.PreferHost, schedData.PreferBackupHost, int64(schedData.Count), sid)
} else {
output = transToRegionSchedResult(result.Data, int64(schedData.Count))
output = transToRegionSchedResult(result.Data, int64(schedData.Count), sid)
}
var readyCount int64
for _, candi := range output.Candidates {
if len(candi.Error) != 0 {
info, exist := getOrNewFilter("select_candidate")
@@ -91,7 +95,6 @@ func transToSchedForecastResult(result *core.SchedResultItemList) interface{} {
if !exist {
filters = append(filters, info)
}
readyCount--
} else {
readyCount++
}
pkg/scheduler/handler/handler.go
+51 -27
View File
@@ -17,22 +17,23 @@ package handler
import (
"fmt"
"net/http"
"net/http/httputil"
"time"
simplejson "github.com/bitly/go-simplejson"
gin "gopkg.in/gin-gonic/gin.v1"
"yunion.io/x/jsonutils"
"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"
"yunion.io/x/onecloud/pkg/scheduler/db/models"
schedman "yunion.io/x/onecloud/pkg/scheduler/manager"
schedmodels "yunion.io/x/onecloud/pkg/scheduler/models"
)
// InstallHandler is an interface that registes route and
@@ -49,12 +50,6 @@ func InstallHandler(r *gin.Engine) {
func timer(f gin.HandlerFunc) gin.HandlerFunc {
return func(c *gin.Context) {
startTime := time.Now()
bytes, _ := httputil.DumpRequest(c.Request, true)
log.V(10).Debugf(`
>>>>>>>>>>>>>
HTTP Request:
%s
>>>>>>>>>>>>>`, string(bytes))
f(c)
log.Infof("Handler %q cost: %v", c.Request.URL.Path, time.Since(startTime))
}
@@ -289,17 +284,32 @@ func doSyncSchedule(c *gin.Context) {
}
count := int64(schedInfo.Count)
var resp interface{}
var resp *schedapi.ScheduleOutput
sid := schedInfo.SessionId
if schedInfo.Backup {
resp = transToBackupSchedResult(result, schedInfo.PreferHost, schedInfo.PreferBackupHost, count, true)
resp = transToBackupSchedResult(result, schedInfo.PreferHost, schedInfo.PreferBackupHost, count, sid)
} else {
resp = transToRegionSchedResult(result.Data, count)
resp = transToRegionSchedResult(result.Data, count, sid)
}
if err := setSchedPendingUsage(schedInfo, resp); err != nil {
c.AbortWithError(http.StatusInternalServerError, err)
return
}
c.JSON(http.StatusOK, resp)
}
func transToRegionSchedResult(result []*core.SchedResultItem, count int64) *schedapi.ScheduleOutput {
func setSchedPendingUsage(req *api.SchedInfo, resp *schedapi.ScheduleOutput) error {
if req.IsSuggestion || req.SkipDirtyMarkHost() {
return nil
}
for _, item := range resp.Candidates {
schedmodels.HostPendingUsageManager.SetPendingUsage(req, item)
}
return nil
}
func transToRegionSchedResult(result []*core.SchedResultItem, count int64, sid string) *schedapi.ScheduleOutput {
apiResults := make([]*schedapi.CandidateResource, 0)
succCount := 0
for _, nr := range result {
@@ -308,6 +318,7 @@ func transToRegionSchedResult(result []*core.SchedResultItem, count int64) *sche
break
}
tr := nr.ToCandidateResource()
tr.SessionId = sid
apiResults = append(apiResults, tr)
nr.Count--
succCount++
@@ -334,37 +345,40 @@ func regionResponse(v interface{}) interface{} {
}{Result: v}
}
func newExpireArgsByHostIDs(ids []string) (*api.ExpireArgs, error) {
hs, err := models.FetchHostByIDs(ids)
if err != nil {
return nil, err
}
if len(hs) == 0 {
return nil, fmt.Errorf("Hostscache %v not found", ids)
func newExpireArgsByHostIDs(ids []string, sid string) (*api.ExpireArgs, error) {
hs := []computemodels.SHost{}
q := computemodels.HostManager.Query().In("id", ids)
if err := db.FetchModelObjects(computemodels.HostManager, q, &hs); err != nil {
return nil, fmt.Errorf("Fetch hosts by ids %v: %v", ids, err)
}
expireArgs := &api.ExpireArgs{
DirtyBaremetals: []string{},
DirtyHosts: []string{},
SessionId: sid,
}
for _, obj := range hs {
host := obj.(*models.Host)
if !host.IsHypervisor() {
expireArgs.DirtyBaremetals = append(expireArgs.DirtyBaremetals, host.ID)
for _, host := range hs {
if host.HostType == computeapi.HOST_TYPE_BAREMETAL {
expireArgs.DirtyBaremetals = append(expireArgs.DirtyBaremetals, host.GetId())
} else {
expireArgs.DirtyHosts = append(expireArgs.DirtyHosts, host.ID)
expireArgs.DirtyHosts = append(expireArgs.DirtyHosts, host.GetId())
}
}
return expireArgs, nil
}
func doCleanAllHostCache(c *gin.Context) {
ids, err := models.AllIDs(models.Hosts)
idMap, err := computemodels.HostManager.Query("id").AllStringMap()
if err != nil {
c.AbortWithError(http.StatusInternalServerError, err)
return
}
args, err := newExpireArgsByHostIDs(ids)
ids := []string{}
for _, obj := range idMap {
ids = append(ids, obj["id"])
}
sid := getSessionId(c)
args, err := newExpireArgsByHostIDs(ids, sid)
if err != nil {
c.AbortWithError(http.StatusBadRequest, err)
return
@@ -390,8 +404,18 @@ func doSyncSku(c *gin.Context) {
c.JSON(http.StatusOK, nil)
}
func getSessionId(c *gin.Context) string {
query, err := jsonutils.ParseQueryString(c.Request.URL.RawQuery)
if err != nil {
log.Warningf("not found session id in query")
return ""
}
return jsonutils.GetAnyString(query, []string{"session", "session_id"})
}
func doCleanHostCache(c *gin.Context, hostID string) {
args, err := newExpireArgsByHostIDs([]string{hostID})
sid := getSessionId(c)
args, err := newExpireArgsByHostIDs([]string{hostID}, sid)
if err != nil {
c.AbortWithError(http.StatusBadRequest, err)
return
pkg/scheduler/manager/expire_queue.go
+39 -32
View File
@@ -19,6 +19,7 @@ import (
"time"
"yunion.io/x/log"
"yunion.io/x/pkg/util/sets"
u "yunion.io/x/pkg/utils"
"yunion.io/x/onecloud/pkg/scheduler/api"
@@ -41,19 +42,31 @@ func (e *ExpireManager) Add(expireArgs *api.ExpireArgs) {
e.expireChannel <- expireArgs
}
type expireHost struct {
Id string
SessionId string
}
func newExpireHost(id string, sid string) *expireHost {
return &expireHost{
Id: id,
SessionId: sid,
}
}
func (e *ExpireManager) Run() {
t := time.Tick(u.ToDuration(o.GetOptions().ExpireQueueConsumptionPeriod))
notInSession := func(ids []string, resType string) []string {
//var newIds []string
//for _, id := range ids {
//if !schedManager.ReservedPoolManager.InSession(resType, id) {
//newIds = append(newIds, id)
//}
//}
//return newIds
return ids
}
//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{})
@@ -75,25 +88,21 @@ func (e *ExpireManager) Run() {
if expireRequestNumber <= 0 {
return
}
dirtyHostMap := make(map[string]int, expireRequestNumber)
dirtyBaremetalMap := make(map[string]int, expireRequestNumber)
dirtyHosts := make([]string, 0)
dirtyBaremetals := make([]string, 0)
dirtyHostSets := sets.NewString()
dirtyBaremetalSets := sets.NewString()
dirtyHosts := make([]*expireHost, 0)
dirtyBaremetals := make([]*expireHost, 0)
// Merge all same host.
for i := 0; i < expireRequestNumber; i++ {
expireArgs := <-e.expireChannel
log.V(4).Infof("Get expireArgs from channel: %#v", expireArgs)
dirtyHostSets.Insert(expireArgs.DirtyHosts...)
for _, host := range expireArgs.DirtyHosts {
if _, ok := dirtyHostMap[host]; !ok {
dirtyHostMap[host] = len(dirtyHosts)
dirtyHosts = append(dirtyHosts, host)
}
dirtyHosts = append(dirtyHosts, newExpireHost(host, expireArgs.SessionId))
}
dirtyBaremetalSets.Insert(expireArgs.DirtyBaremetals...)
for _, baremetal := range expireArgs.DirtyBaremetals {
if _, ok := dirtyBaremetalMap[baremetal]; !ok {
dirtyBaremetalMap[baremetal] = len(dirtyBaremetals)
dirtyBaremetals = append(dirtyBaremetals, baremetal)
}
dirtyBaremetals = append(dirtyBaremetals, newExpireHost(baremetal, expireArgs.SessionId))
}
}
log.V(4).Infof("batchMergeExpire dirtyHosts: %v, dirtyBaremetals: %v", dirtyHosts, dirtyBaremetals)
@@ -101,27 +110,25 @@ func (e *ExpireManager) Run() {
wg.Add(2)
go func() {
defer wg.Done()
dirtyHosts = notInSession(dirtyHosts, "host")
//dirtyHosts = notInSession(dirtyHosts, "host")
if len(dirtyHosts) > 0 {
log.V(10).Debugf("CleanDirty Hosts: %v\n", dirtyHosts)
_, err := schedManager.CandidateManager.Reload("host", dirtyHosts)
schedManager.CandidateManager.CleanDirtyCandidatesOnce(dirtyHosts)
if err != nil {
log.Errorf("%v", err)
if _, err := schedManager.CandidateManager.Reload("host", dirtyHostSets.List()); err != nil {
log.Errorf("Clean dirty hosts %v: %v", dirtyHosts, err)
}
schedManager.HistoryManager.CancelCandidatesPendingUsage(dirtyHosts)
}
}()
go func() {
defer wg.Done()
dirtyBaremetals = notInSession(dirtyBaremetals, "baremetal")
//dirtyBaremetals = notInSession(dirtyBaremetals, "baremetal")
if len(dirtyBaremetals) > 0 {
log.V(10).Debugf("CleanDirty Baremetals: %v\n", dirtyBaremetals)
_, err := schedManager.CandidateManager.Reload("baremetal", dirtyBaremetals)
schedManager.CandidateManager.CleanDirtyCandidatesOnce(dirtyBaremetals)
if err != nil {
log.Errorf("%v", err)
if _, err := schedManager.CandidateManager.Reload("baremetal", dirtyBaremetalSets.List()); err != nil {
log.Errorf("Clean dirty baremetals %v: %v", dirtyBaremetals, err)
}
schedManager.HistoryManager.CancelCandidatesPendingUsage(dirtyBaremetals)
}
}()
if ok := waitTimeOut(wg, u.ToDuration(o.GetOptions().ExpireQueueConsumptionTimeout)); !ok {
pkg/scheduler/manager/manager.go
+4 -12
View File
@@ -482,16 +482,8 @@ func GetHistoryDetail(historyDetailArgs *api.HistoryDetailArgs) (*api.HistoryDet
}, nil
}
func GetCandidateHostsDesc() ([]*candidate.HostDesc, error) {
cs, err := GetCandidateManager().GetCandidates(data_manager.CandidateGetArgs{ResType: "host"})
if err != nil {
return nil, err
}
hosts, err := data_manager.ToHostCandidates(cs)
if err != nil {
return nil, err
}
return hosts, nil
func GetCandidateHostsDesc() ([]core.Candidater, error) {
return GetCandidateManager().GetCandidates(data_manager.CandidateGetArgs{ResType: "host"})
}
func GetK8sCandidateHosts(nodesName ...string) ([]*candidatecache.HostDesc, error) {
@@ -501,8 +493,8 @@ func GetK8sCandidateHosts(nodesName ...string) ([]*candidatecache.HostDesc, erro
}
findHost := func(nodeName string) *candidatecache.HostDesc {
for _, host := range hosts {
if host.Name == nodeName {
return host
if host.Getter().Name() == nodeName {
return host.(*candidatecache.HostDesc)
}
}
return nil
pkg/scheduler/manager/scheduler.go
+6 -6
View File
@@ -115,7 +115,7 @@ type Scheduler interface {
Unit() *core.Unit
Candidates() ([]core.Candidater, error)
DirtySelectedCandidates([]*core.SelectedCandidate)
//DirtySelectedCandidates([]*core.SelectedCandidate)
}
type BaseScheduler struct {
@@ -151,11 +151,11 @@ func (s *BaseScheduler) BeforePredicate() error {
return nil
}
func (s *BaseScheduler) DirtySelectedCandidates(scs []*core.SelectedCandidate) {
for _, sc := range scs {
s.CandidateManager().SetCandidateDirty(sc)
}
}
//func (s *BaseScheduler) DirtySelectedCandidates(scs []*core.SelectedCandidate) {
//for _, sc := range scs {
//s.CandidateManager().SetCandidateDirty(sc)
//}
//}
// GuestScheduler for guest type schedule
type GuestScheduler struct {
pkg/scheduler/manager/task_history.go
+28
View File
@@ -19,10 +19,13 @@ import (
"sync"
"time"
"yunion.io/x/log"
"yunion.io/x/pkg/util/wait"
u "yunion.io/x/pkg/utils"
o "yunion.io/x/onecloud/pkg/scheduler/options"
"yunion.io/x/onecloud/pkg/scheduler/models"
)
type HistoryItem struct {
@@ -158,3 +161,28 @@ func (m *HistoryManager) GetHistory(sessionId string) *HistoryItem {
return nil
}
func (m *HistoryManager) GetCancelUsage(sessionId string, hostId string) *models.SessionPendingUsage {
item := m.GetHistory(sessionId)
if item == nil {
return nil
}
usage, _ := models.HostPendingUsageManager.GetSessionUsage(sessionId, hostId)
return usage
}
func (m *HistoryManager) CancelCandidatesPendingUsage(hosts []*expireHost) {
for _, h := range hosts {
hostId := h.Id
sid := h.SessionId
if len(sid) == 0 {
continue
}
cancelUsage := m.GetCancelUsage(sid, hostId)
if err := models.HostPendingUsageManager.CancelPendingUsage(hostId, cancelUsage); err != nil {
log.Errorf("Cancel host %s usage %#v: %v", hostId, cancelUsage, err)
} else {
cancelUsage.StopTimer()
}
}
}
pkg/scheduler/models/pending_usage.go
+397
View File
@@ -0,0 +1,397 @@
// 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 models
import (
"context"
"fmt"
"sync"
"time"
"github.com/pkg/errors"
"yunion.io/x/jsonutils"
"yunion.io/x/log"
schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
"yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
"yunion.io/x/onecloud/pkg/cloudcommon/db/quotas"
"yunion.io/x/onecloud/pkg/scheduler/api"
)
var HostPendingUsageManager *SHostPendingUsageManager
type SHostPendingUsageManager struct {
store *SHostMemoryPendingUsageStore
}
func init() {
pendingStore := NewHostMemoryPendingUsageStore()
HostPendingUsageManager = &SHostPendingUsageManager{pendingStore}
}
func (m *SHostPendingUsageManager) Keyword() string {
return "pending_usage_manager"
}
func (m *SHostPendingUsageManager) newSessionUsage(req *api.SchedInfo, hostId string) *SessionPendingUsage {
su := NewSessionUsage(req.SessionId)
su.Usage = NewPendingUsageBySchedInfo(hostId, req)
return su
}
func (m *SHostPendingUsageManager) newPendingUsage(hostId string) *SPendingUsage {
return NewPendingUsageBySchedInfo(hostId, nil)
}
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
}
func (m *SHostPendingUsageManager) GetSessionUsage(sessionId, hostId string) (*SessionPendingUsage, error) {
return m.store.GetSessionUsage(sessionId, hostId)
}
func (m *SHostPendingUsageManager) SetPendingUsage(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 {
sessionUsage = m.newSessionUsage(req, hostId)
sessionUsage.StartTimer()
}
m.addSessionUsage(candidate.HostId, sessionUsage)
}
func (m *SHostPendingUsageManager) addSessionUsage(hostId string, usage *SessionPendingUsage) {
pendingUsage, _ := m.GetPendingUsage(hostId)
if pendingUsage == nil {
pendingUsage = m.newPendingUsage(hostId)
}
pendingUsage.Add(usage.Usage)
usage.AddCount()
m.store.SetSessionUsage(usage.SessionId, hostId, usage)
m.store.SetPendingUsage(hostId, pendingUsage)
}
func (m *SHostPendingUsageManager) CancelPendingUsage(hostId string, su *SessionPendingUsage) error {
ctx := context.Background()
lockman.LockClass(ctx, m, hostId)
defer lockman.ReleaseClass(ctx, m, 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()
return nil
}
func (m *SHostPendingUsageManager) DeleteSessionUsage(usage *SessionPendingUsage) {
m.store.DeleteSessionUsage(usage)
}
type SHostMemoryPendingUsageStore struct {
store *sync.Map
}
func NewHostMemoryPendingUsageStore() *SHostMemoryPendingUsageStore {
return &SHostMemoryPendingUsageStore{
store: new(sync.Map),
}
}
func (self *SHostMemoryPendingUsageStore) sessionUsageKey(sid, hostId string) string {
return fmt.Sprintf("%s-%s", sid, hostId)
}
func (self *SHostMemoryPendingUsageStore) GetSessionUsage(sessionId string, hostId string) (*SessionPendingUsage, error) {
key := self.sessionUsageKey(sessionId, hostId)
ret, ok := self.store.Load(key)
if !ok {
return nil, errors.Errorf("Not fond session pending usage by %s", key)
}
return ret.(*SessionPendingUsage), nil
}
func (self *SHostMemoryPendingUsageStore) SetSessionUsage(sessionId, hostId string, usage *SessionPendingUsage) {
key := self.sessionUsageKey(sessionId, hostId)
self.store.Store(key, usage)
}
func (self *SHostMemoryPendingUsageStore) GetPendingUsage(hostId string) (*SPendingUsage, error) {
ret, ok := self.store.Load(hostId)
if !ok {
return nil, errors.Errorf("Not fond pending usage by %s", hostId)
}
usage := ret.(*SPendingUsage)
return usage, nil
}
func (self *SHostMemoryPendingUsageStore) SetPendingUsage(hostId string, usage *SPendingUsage) {
if usage.IsEmpty() {
self.store.Delete(hostId)
return
}
self.store.Store(hostId, usage)
}
func (self *SHostMemoryPendingUsageStore) DeleteSessionUsage(usage *SessionPendingUsage) {
self.store.Delete(self.sessionUsageKey(usage.SessionId, usage.Usage.HostId))
}
type SessionPendingUsage struct {
SessionId string
Usage *SPendingUsage
countLock *sync.Mutex
count int
cancelCh chan string
}
func NewSessionUsage(sid string) *SessionPendingUsage {
su := &SessionPendingUsage{
SessionId: sid,
Usage: NewPendingUsageBySchedInfo("", nil),
count: 0,
countLock: new(sync.Mutex),
cancelCh: make(chan string),
}
return su
}
func (su *SessionPendingUsage) AddCount() {
su.countLock.Lock()
defer su.countLock.Unlock()
su.count++
}
func (su *SessionPendingUsage) SubCount() {
su.countLock.Lock()
defer su.countLock.Unlock()
su.count--
}
type SResourcePendingUsage struct {
store *sync.Map
}
func NewResourcePendingUsage(vals map[string]int) *SResourcePendingUsage {
u := &SResourcePendingUsage{
store: new(sync.Map),
}
for key, val := range vals {
u.Set(key, val)
}
return u
}
func (u *SResourcePendingUsage) ToMap() map[string]int {
ret := make(map[string]int)
u.Range(func(key string, val int) bool {
ret[key] = val
return true
})
return ret
}
func (u *SResourcePendingUsage) Get(key string) int {
val, ok := u.store.Load(key)
if !ok {
return 0
}
return val.(int)
}
func (u *SResourcePendingUsage) Set(key string, size int) {
u.store.Store(key, size)
}
func (u *SResourcePendingUsage) Range(f func(key string, size int) bool) {
u.store.Range(func(key, val interface{}) bool {
return f(key.(string), val.(int))
})
}
func (u *SResourcePendingUsage) Add(su *SResourcePendingUsage) {
u.Range(func(key string, size int) bool {
size2 := su.Get(key)
u.Set(key, size+size2)
return true
})
su.Range(func(key string, size int) bool {
if _, ok := u.store.Load(key); !ok {
u.Set(key, size)
}
return true
})
}
func (u *SResourcePendingUsage) Sub(su *SResourcePendingUsage) {
u.Range(func(key string, size int) bool {
size2 := su.Get(key)
u.Set(key, quotas.NonNegative(size-size2))
return true
})
}
func (u *SResourcePendingUsage) IsEmpty() bool {
empty := true
u.Range(func(_ string, size int) bool {
if size != 0 {
empty = false
return false
}
return true
})
return empty
}
type SPendingUsage struct {
HostId string
Cpu int
Memory int
IsolatedDevice int
DiskUsage *SResourcePendingUsage
NetUsage *SResourcePendingUsage
}
func NewPendingUsageBySchedInfo(hostId string, req *api.SchedInfo) *SPendingUsage {
u := &SPendingUsage{
HostId: hostId,
DiskUsage: NewResourcePendingUsage(nil),
NetUsage: NewResourcePendingUsage(nil),
}
if req == nil {
return u
}
u.Cpu = req.Ncpu
u.Memory = req.Memory
u.IsolatedDevice = len(req.IsolatedDevices)
for _, disk := range req.Disks {
backend := disk.Backend
size := disk.SizeMb
osize := u.DiskUsage.Get(backend)
u.DiskUsage.Set(backend, osize+size)
}
for _, net := range req.Networks {
id := net.Network
if id == "" {
continue
}
ocount := u.NetUsage.Get(id)
u.NetUsage.Set(id, ocount+1)
}
return u
}
func (self *SPendingUsage) ToMap() map[string]interface{} {
return map[string]interface{}{
"cpu": self.Cpu,
"memory": self.Memory,
"isolated_device": self.IsolatedDevice,
"disk": self.DiskUsage.ToMap(),
"net": self.NetUsage.ToMap(),
}
}
func (self *SPendingUsage) Add(sUsage *SPendingUsage) {
self.Cpu = self.Cpu + sUsage.Cpu
self.Memory = self.Memory + sUsage.Memory
self.IsolatedDevice = self.IsolatedDevice + sUsage.IsolatedDevice
self.DiskUsage.Add(sUsage.DiskUsage)
self.NetUsage.Add(sUsage.NetUsage)
}
func (self *SPendingUsage) Sub(sUsage *SPendingUsage) {
self.Cpu = quotas.NonNegative(self.Cpu - sUsage.Cpu)
self.Memory = quotas.NonNegative(self.Memory - sUsage.Memory)
self.IsolatedDevice = quotas.NonNegative(self.IsolatedDevice - sUsage.IsolatedDevice)
self.DiskUsage.Sub(sUsage.DiskUsage)
self.NetUsage.Sub(sUsage.NetUsage)
}
func (self *SPendingUsage) IsEmpty() bool {
if self.Cpu > 0 {
return false
}
if self.Memory > 0 {
return false
}
if self.IsolatedDevice > 0 {
return false
}
if !self.DiskUsage.IsEmpty() {
return false
}
if !self.NetUsage.IsEmpty() {
return false
}
return true
}
func (self *SessionPendingUsage) cancelSelf() {
hostId := self.Usage.HostId
count := self.count
for i := 0; i <= count; i++ {
HostPendingUsageManager.CancelPendingUsage(hostId, self)
}
}
func (self *SessionPendingUsage) StartTimer() {
timeout := 1 * time.Minute
go func() {
for {
select {
case <-time.After(timeout):
log.Infof("timeout cancel session usage %#v", self)
self.cancelSelf()
goto ForEnd
case sid := <-self.cancelCh:
log.Infof("Cancel session %s usage, count: %d", sid, self.count)
if self.count <= 0 {
goto ForEnd
} else {
log.Infof("continue waiting next cancel...")
}
}
}
ForEnd:
log.Infof("delete session usage %#v", self)
HostPendingUsageManager.DeleteSessionUsage(self)
}()
}
func (self *SessionPendingUsage) StopTimer() {
self.cancelCh <- self.SessionId
}
pkg/scheduler/models/pending_usage_test.go
+143
View File
@@ -0,0 +1,143 @@
// 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 models
import (
"reflect"
"testing"
)
func TestNewResourcePendingUsage(t *testing.T) {
tests := []struct {
name string
args map[string]int
want map[string]int
}{
{
name: "map init",
args: map[string]int{"s1": 1, "s2": 2},
want: map[string]int{"s1": 1, "s2": 2},
},
{
name: "map nil init",
args: nil,
want: map[string]int{},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := NewResourcePendingUsage(tt.args); !reflect.DeepEqual(got.ToMap(), tt.want) {
t.Errorf("NewResourcePendingUsage() = %v, want %v", got, tt.want)
}
})
}
}
func TestSResourcePendingUsage_Add(t *testing.T) {
tests := []struct {
name string
ou *SResourcePendingUsage
su *SResourcePendingUsage
want map[string]int
}{
{
name: "same add",
ou: NewResourcePendingUsage(map[string]int{"k1": 1, "k2": 2}),
su: NewResourcePendingUsage(map[string]int{"k1": 3, "k2": 1}),
want: map[string]int{"k1": 4, "k2": 3},
},
{
name: "diff add",
ou: NewResourcePendingUsage(map[string]int{"k1": 1, "k2": 2}),
su: NewResourcePendingUsage(map[string]int{"k1": 3, "k2": 1, "k3": 4}),
want: map[string]int{"k1": 4, "k2": 3, "k3": 4},
},
{
name: "add nil",
ou: NewResourcePendingUsage(map[string]int{"k1": 1, "k2": 2}),
su: NewResourcePendingUsage(nil),
want: map[string]int{"k1": 1, "k2": 2},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
tt.ou.Add(tt.su)
if got := tt.ou.ToMap(); !reflect.DeepEqual(got, tt.want) {
t.Errorf("NewResourcePendingUsage_add() = %v, want %v", got, tt.want)
}
})
}
}
func TestSResourcePendingUsage_Sub(t *testing.T) {
tests := []struct {
name string
ou *SResourcePendingUsage
su *SResourcePendingUsage
want map[string]int
}{
{
name: "same sub",
ou: NewResourcePendingUsage(map[string]int{"k1": 1, "k2": 2}),
su: NewResourcePendingUsage(map[string]int{"k1": 3, "k2": 1}),
want: map[string]int{"k1": 0, "k2": 1},
},
{
name: "sub nil",
ou: NewResourcePendingUsage(map[string]int{"k1": 1, "k2": 2}),
su: NewResourcePendingUsage(nil),
want: map[string]int{"k1": 1, "k2": 2},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
tt.ou.Sub(tt.su)
if got := tt.ou.ToMap(); !reflect.DeepEqual(got, tt.want) {
t.Errorf("NewResourcePendingUsage_add() = %v, want %v", got, tt.want)
}
})
}
}
func TestSResourcePendingUsage_IsEmpty(t *testing.T) {
tests := []struct {
name string
u *SResourcePendingUsage
want bool
}{
{
name: "nil is empty",
u: NewResourcePendingUsage(nil),
want: true,
},
{
name: "item zeros is empty",
u: NewResourcePendingUsage(map[string]int{"k1": 0, "k2": 0}),
want: true,
},
{
name: "item no zeros not empty",
u: NewResourcePendingUsage(map[string]int{"k1": 1, "k2": 0}),
want: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := tt.u.IsEmpty(); got != tt.want {
t.Errorf("SResourcePendingUsage.IsEmpty() = %v, want %v", got, tt.want)
}
})
}
}