From 45e9ecd36264c45704812e35506b3d9626b3c455 Mon Sep 17 00:00:00 2001 From: Zexi Li Date: Thu, 18 Oct 2018 17:01:46 +0800 Subject: [PATCH] scheduler: filter host when cloudprovider status not connected --- .../predicates/guest/status_predicate.go | 22 +++- pkg/scheduler/cache/candidate/baremetals.go | 56 ++------- pkg/scheduler/cache/candidate/desc.go | 106 +++++++++++++++++- pkg/scheduler/cache/candidate/hosts.go | 94 ++-------------- pkg/scheduler/core/types.go | 4 + pkg/scheduler/db/models/cloudprovider.go | 43 +++++++ pkg/scheduler/db/models/default.go | 11 +- pkg/scheduler/db/models/zone.go | 41 +++++++ pkg/scheduler/manager/task_queue.go | 4 +- 9 files changed, 240 insertions(+), 141 deletions(-) create mode 100644 pkg/scheduler/db/models/cloudprovider.go create mode 100644 pkg/scheduler/db/models/zone.go diff --git a/pkg/scheduler/algorithm/predicates/guest/status_predicate.go b/pkg/scheduler/algorithm/predicates/guest/status_predicate.go index e6f295ca1b..30a7709a96 100644 --- a/pkg/scheduler/algorithm/predicates/guest/status_predicate.go +++ b/pkg/scheduler/algorithm/predicates/guest/status_predicate.go @@ -3,6 +3,8 @@ package guest import ( "yunion.io/x/onecloud/pkg/scheduler/algorithm/predicates" "yunion.io/x/onecloud/pkg/scheduler/core" + + "yunion.io/x/onecloud/pkg/compute/models" ) const ( @@ -27,10 +29,14 @@ 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 := h.Get("Status").(string) - curHostStatus := h.Get("HostStatus").(string) - curEnableStatus := h.Get("EnableStatus").(string) + curStatus := hc.Status + curHostStatus := hc.HostStatus + curEnableStatus := hc.Enabled if curStatus != ExpectedStatus { h.Exclude2("status", curStatus, ExpectedStatus) @@ -40,8 +46,14 @@ func (p *StatusPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core h.Exclude2("host_status", curHostStatus, ExpectedHostStatus) } - if curEnableStatus != ExpectedEnableStatus { - h.Exclude2("enable_status", curEnableStatus, ExpectedEnableStatus) + if !curEnableStatus { + h.Exclude2("enable_status", curEnableStatus, true) + } + + if hc.Cloudprovider != nil { + if hc.Cloudprovider.Status != models.CLOUD_PROVIDER_CONNECTED { + h.Exclude2("cloud_provider_status", hc.Cloudprovider.Status, models.CLOUD_PROVIDER_CONNECTED) + } } return h.GetResult() diff --git a/pkg/scheduler/cache/candidate/baremetals.go b/pkg/scheduler/cache/candidate/baremetals.go index 56c7c99931..88910cb4a0 100644 --- a/pkg/scheduler/cache/candidate/baremetals.go +++ b/pkg/scheduler/cache/candidate/baremetals.go @@ -16,7 +16,7 @@ import ( ) type BaremetalDesc struct { - baseHostDesc + *baseHostDesc Storages []*baremetal.BaremetalStorage `json:"storages"` StorageType string `json:"storage_type"` @@ -212,30 +212,21 @@ func (bb *BaremetalBuilder) build() ([]interface{}, error) { } func (bb *BaremetalBuilder) buildOne(bm *models.Host) (interface{}, error) { - desc := new(BaremetalDesc) - desc.ID = bm.ID - desc.Name = bm.Name - desc.UpdatedAt = bm.UpdatedAt - desc.Status = bm.Status - desc.CPUCount = bm.CPUCount - desc.NodeCount = bm.NodeCount - desc.MemSize = int64(bm.MemSize) + baseDesc, err := newBaseHostDesc(bm) + if err != nil { + return nil, err + } + desc := &BaremetalDesc{ + baseHostDesc: baseDesc, + } + desc.StorageDriver = bm.StorageDriver desc.StorageType = bm.StorageType desc.StorageSize = int64(bm.StorageSize) desc.StorageInfo = bm.StorageInfo - desc.PoolID = bm.PoolID - desc.ZoneID = bb.getZoneID(bm) - desc.Enabled = bm.Enabled - desc.ClusterID = bm.ClusterID - - desc.HostStatus = bm.HostStatus - desc.Enabled = bm.Enabled - desc.HostType = bm.HostType - desc.IsBaremetal = bm.IsBaremetal var baremetalStorages []*baremetal.BaremetalStorage - err := fjson.Unmarshal([]byte(bm.StorageInfo), &baremetalStorages) + err = fjson.Unmarshal([]byte(bm.StorageInfo), &baremetalStorages) if err != nil { // StorageInfo maybe is NULL if bm.StorageInfo != "" { @@ -250,22 +241,6 @@ func (bb *BaremetalBuilder) buildOne(bm *models.Host) (interface{}, error) { return nil, err } - err = bb.fillResidentTenants(desc, bm) - if err != nil { - return nil, err - } - - // data from db - err = bb.fillNetworks(desc, bm) - if err != nil { - return nil, err - } - - err = desc.fillAggregates() - if err != nil { - return nil, err - } - return desc, nil } @@ -293,17 +268,6 @@ func (bb *BaremetalBuilder) fillNetworks(desc *BaremetalDesc, b *models.Host) er return desc.fillNetworks(b.ID) } -func (b *BaremetalBuilder) fillResidentTenants(desc *BaremetalDesc, host *models.Host) error { - rets, err := HostResidentTenantCount(host.ID) - if err != nil { - return err - } - - desc.Tenants = rets - - return nil -} - func (b *BaremetalBuilder) getZoneID(bm *models.Host) string { return bm.ZoneID } diff --git a/pkg/scheduler/cache/candidate/desc.go b/pkg/scheduler/cache/candidate/desc.go index 968b14a05a..5aaeb1506b 100644 --- a/pkg/scheduler/cache/candidate/desc.go +++ b/pkg/scheduler/cache/candidate/desc.go @@ -1,12 +1,15 @@ package candidate import ( + "fmt" "time" + "yunion.io/x/jsonutils" "yunion.io/x/log" + "yunion.io/x/pkg/utils" + "yunion.io/x/onecloud/pkg/scheduler/api" "yunion.io/x/onecloud/pkg/scheduler/db/models" - "yunion.io/x/pkg/utils" ) type baseDesc struct { @@ -22,6 +25,7 @@ func (b *baseDesc) UUID() string { type baseHostDesc struct { baseDesc + ManagerID *string `json:"manager_id"` Status string `json:"status"` CPUCount int64 `json:"cpu_count"` MemSize int64 `json:"mem_size"` @@ -30,13 +34,113 @@ type baseHostDesc struct { Enabled bool `json:"enabled"` HostType string `json:"host_type"` IsBaremetal bool `json:"is_baremetal"` + IsMaintenance bool `json:"is_maintenance"` NodeCount int64 `json:"node_count"` Tenants map[string]int64 `json:"tenants"` ZoneID string `json:"zone_id"` + Zone string `json:"zone"` PoolID string `json:"pool_id"` ClusterID string `json:"cluster_id"` Aggregates []*models.Aggregate `json:"aggregates"` HostAggregates []*models.Aggregate `json:"host_aggregates"` + Cloudprovider *models.Cloudprovider `json:"cloudprovider"` +} + +func newBaseHostDesc(host *models.Host) (*baseHostDesc, error) { + desc := &baseHostDesc{ + baseDesc: baseDesc{ + ID: host.ID, + Name: host.Name, + UpdatedAt: host.UpdatedAt, + }, + ManagerID: host.ManagerID, + Status: host.Status, + CPUCount: host.CPUCount, + MemSize: host.MemSize, + HostStatus: host.HostStatus, + Enabled: host.Enabled, + HostType: host.HostType, + IsBaremetal: host.IsBaremetal, + NodeCount: host.NodeCount, + IsMaintenance: host.IsMaintenance, + } + + if err := desc.fillCloudProvider(host); err != nil { + return nil, fmt.Errorf("Fill cloudprovider info error: %v", err) + } + + if err := desc.fillNetworks(desc.ID); err != nil { + return nil, fmt.Errorf("Fill networks error: %v", err) + } + + if err := desc.fillZone(host); err != nil { + return nil, fmt.Errorf("Fill zone error: %v", err) + } + + if err := desc.fillResidentTenants(host); err != nil { + return nil, fmt.Errorf("Fill resident tenants error: %v", err) + } + + if err := desc.fillAggregates(); err != nil { + return nil, fmt.Errorf("Fill schetags error: %v", err) + } + return desc, nil +} + +func (b baseHostDesc) GetSchedDesc() *jsonutils.JSONDict { + desc := jsonutils.NewDict() + + desc.Add(jsonutils.NewString(b.ID), "id") + desc.Add(jsonutils.NewString(b.Name), "name") + desc.Add(jsonutils.NewInt(b.CPUCount), "cpu_count") + desc.Add(jsonutils.NewInt(b.MemSize), "mem_size") + desc.Add(jsonutils.NewString(b.HostType), "host_type") + desc.Add(jsonutils.NewString(b.ZoneID), "zone_id") + desc.Add(jsonutils.NewString(b.Zone), "zone") + + if b.Cloudprovider != nil { + p := b.Cloudprovider + cloudproviderDesc := jsonutils.NewDict() + cloudproviderDesc.Add(jsonutils.NewString(p.ProjectId), "tenant_id") + cloudproviderDesc.Add(jsonutils.NewString(p.Provider), "provider") + desc.Add(cloudproviderDesc, "cloudprovider") + } + + return desc +} + +func (b *baseHostDesc) fillCloudProvider(host *models.Host) error { + if host.ManagerID == nil { + log.Debugf("Host %q manager id is empty, no cloud provider", host.Name) + return nil + } + provider, err := models.FetchCloudproviderById(*(host.ManagerID)) + if err != nil { + return err + } + b.Cloudprovider = provider + return nil +} + +func (b *baseHostDesc) fillZone(host *models.Host) error { + zone, err := models.FetchZoneByID(host.ZoneID) + if err != nil { + return err + } + b.ZoneID = zone.ID + b.Zone = zone.Name + return nil +} + +func (b *baseHostDesc) fillResidentTenants(host *models.Host) error { + rets, err := HostResidentTenantCount(host.ID) + if err != nil { + return err + } + + b.Tenants = rets + + return nil } func (b *baseHostDesc) fillAggregates() error { diff --git a/pkg/scheduler/cache/candidate/hosts.go b/pkg/scheduler/cache/candidate/hosts.go index 0a17ce34db..7eda20c3ba 100644 --- a/pkg/scheduler/cache/candidate/hosts.go +++ b/pkg/scheduler/cache/candidate/hosts.go @@ -22,7 +22,7 @@ import ( ) type HostDesc struct { - baseHostDesc + *baseHostDesc // cpu CPUMHZ int64 `json:"cpu_mhz"` @@ -65,7 +65,6 @@ type HostDesc struct { Groups *GroupCounts `json:"groups"` Metadata map[string]string `json:"metadata"` IsolatedDevices []*IsolatedDeviceDesc `json:"isolated_devices"` - ManagerID *string `json:"manager_id"` IsMaintenance bool `json:"is_maintenance"` GuestReservedResource *ReservedResource `json:"guest_reserved_resource"` GuestReservedResourceUsed *ReservedResource `json:"guest_reserved_used"` @@ -551,7 +550,6 @@ func (b *HostBuilder) init(ids []string, dbCache DBGroupCacher, syncCache SyncGr b.setStorages(ids, errMessageChannel) b.setDiskStats(errMessageChannel) }, - func() { b.setResidentTenantStats(ids, errMessageChannel) }, func() { b.setMetadataInfo(ids, errMessageChannel) }, func() { b.setIsolatedDevs(ids, errMessageChannel) }, func() { b.setCPUIOLoadInfo(errMessageChannel) }, @@ -701,16 +699,6 @@ func (b *HostBuilder) setGuests(ids []string, errMessageChannel chan error) { return } -func (b *HostBuilder) setResidentTenantStats(hostIDs []string, errMessageChannel chan error) { - residentTenantDict, err := HostsResidentTenantStats(hostIDs) - if err != nil { - errMessageChannel <- err - return - } - b.residentTenantDict = residentTenantDict - return -} - func (b *HostBuilder) setGroupInfo(errMessageChannel chan error) { groupGuests, err := models.FetchByGuestIDs(models.GroupGuests, b.guestIDs) if err != nil { @@ -957,24 +945,17 @@ func (b *HostBuilder) build() ([]interface{}, error) { } func (b *HostBuilder) buildOne(host *models.Host) (interface{}, error) { - desc := new(HostDesc) - desc.ID = host.ID - desc.UpdatedAt = host.UpdatedAt - desc.Name = host.Name - desc.Status = host.Status + baseDesc, err := newBaseHostDesc(host) + if err != nil { + return nil, err + } + desc := &HostDesc{ + baseHostDesc: baseDesc, + } + desc.Metadata = make(map[string]string) - desc.HostStatus = host.HostStatus - desc.Enabled = host.Enabled - desc.HostType = host.HostType - desc.IsBaremetal = host.IsBaremetal desc.ManagerID = host.ManagerID - desc.IsMaintenance = host.IsMaintenance - // TODO: purge PoolID & ClusterID not used - desc.PoolID = host.PoolID - desc.ClusterID = host.ClusterID - - desc.CPUCount = host.CPUCount desc.CPUCmtbound = host.CPUOverCommitBound() desc.CPUDesc = host.CPUDesc desc.CPUCache = host.CPUCache @@ -982,7 +963,6 @@ func (b *HostBuilder) buildOne(host *models.Host) (interface{}, error) { desc.NodeCount = host.NodeCount desc.CPUMHZ = host.CPUMHZ - desc.MemSize = host.MemSize desc.MemCmtbound = host.MemOverCommitBound() desc.MemReserved = host.MemReserved @@ -995,14 +975,11 @@ func (b *HostBuilder) buildOne(host *models.Host) (interface{}, error) { fillFuncs := []func(*HostDesc, *models.Host) error{ b.fillGuestsResourceInfo, - b.fillZoneID, b.fillStorages, - b.fillResidentTenants, b.fillResidentGroups, b.fillMetadata, b.fillIsolatedDevices, b.fillCPUIOLoads, - b.fillNetworks, } for _, f := range fillFuncs { @@ -1012,18 +989,9 @@ func (b *HostBuilder) buildOne(host *models.Host) (interface{}, error) { } } - err = desc.fillAggregates() - if err != nil { - return nil, err - } - return desc, nil } -func (b *HostBuilder) fillNetworks(desc *HostDesc, host *models.Host) error { - return desc.fillNetworks(host.ID) -} - func _in(s string, ss []string) bool { for _, str := range ss { if s == str { @@ -1155,33 +1123,6 @@ func (b *HostBuilder) guestAppTags(guest *models.Guest) []string { return []string{} } -func (b *HostBuilder) fillZoneID(desc *HostDesc, host *models.Host) error { - if host.ZoneID != "" { - desc.ZoneID = host.ZoneID - return nil - } - - zoneID, err := b.zoneID(host.ClusterID) - if err != nil { - panic(fmt.Errorf("host: %v, fillZoneID error: %v", host.ID, err)) - return err - } - desc.ZoneID = zoneID - return nil -} - -func (b *HostBuilder) zoneID(clusterID string) (string, error) { - obj, err := b.clusters.Get(clusterID) - if err != nil { - return "", err - } - cluster, ok := obj.(*models.Cluster) - if !ok { - return "", utils.ConvertError(obj, "*models.Cluster") - } - return cluster.ZoneID, nil -} - func (b *HostBuilder) fillStorages(desc *HostDesc, host *models.Host) error { objs, ok := b.hostStoragesDict[host.ID] if !ok { @@ -1249,23 +1190,6 @@ func (b *HostBuilder) storageUsedCapacity(storage *models.Storage, ready bool) i return total } -func (b *HostBuilder) fillResidentTenants(desc *HostDesc, host *models.Host) error { - - tenantMap, ok := b.residentTenantDict[host.ID] - if !ok { - log.V(10).Infof("Not found host ID: %s when fill resident tenants, may be no guests on it.", host.ID) - return nil - } - - rets := make(map[string]int64, len(tenantMap)) - for tenantID, countObj := range tenantMap { - rets[tenantID] = countObj.(int64) - } - desc.Tenants = rets - - return nil -} - func (b *HostBuilder) fillResidentGroups(desc *HostDesc, host *models.Host) error { groups, ok := b.hostGroupCountDict[host.ID] if !ok { diff --git a/pkg/scheduler/core/types.go b/pkg/scheduler/core/types.go index 0a6bd07fb9..e2ca3ab950 100644 --- a/pkg/scheduler/core/types.go +++ b/pkg/scheduler/core/types.go @@ -3,6 +3,8 @@ package core import ( //"sync" //"yunion.io/x/onecloud/pkg/scheduler/cache/candidate" + "yunion.io/x/jsonutils" + "yunion.io/x/onecloud/pkg/scheduler/db/models" ) @@ -45,6 +47,8 @@ type Candidater interface { GetAggregates() []*models.Aggregate GetHostAggregates() []*models.Aggregate + + GetSchedDesc() *jsonutils.JSONDict } // HostPriority represents the priority of scheduling to particular host, higher priority is better. diff --git a/pkg/scheduler/db/models/cloudprovider.go b/pkg/scheduler/db/models/cloudprovider.go new file mode 100644 index 0000000000..a33c5311d3 --- /dev/null +++ b/pkg/scheduler/db/models/cloudprovider.go @@ -0,0 +1,43 @@ +package models + +import ( + "github.com/jinzhu/gorm" +) + +type Cloudprovider struct { + StandaloneModel + Status string `json:"status" gorm:"column:status;not null"` + Enabled bool `json:"enabled" gorm:"column:enabled;not null"` + AccessUrl string `json:"access_url" gorm:"column:access_url"` + Provider string `json:"provider" gorm:"column:provider"` + CloudaccountId string `json:"cloudaccount_id" gorm:"column:cloudaccount_id"` + ProjectId string `json:"tenant_id" gorm:"column:tenant_id"` +} + +func (c Cloudprovider) TableName() string { + return cloudproviderTable +} + +func (c Cloudprovider) String() string { + s, _ := JsonString(c) + return s +} + +func NewCloudproviderResource(db *gorm.DB) (Resourcer, error) { + return newResource(db, cloudproviderTable, + func() interface{} { + return &Cloudprovider{} + }, + func() interface{} { + cs := []Cloudprovider{} + return &cs + }) +} + +func FetchCloudproviderById(id string) (*Cloudprovider, error) { + obj, err := FetchByID(CloudProviders, id) + if err != nil { + return nil, err + } + return obj.(*Cloudprovider), nil +} diff --git a/pkg/scheduler/db/models/default.go b/pkg/scheduler/db/models/default.go index 7866de9dc0..24fe2904b0 100644 --- a/pkg/scheduler/db/models/default.go +++ b/pkg/scheduler/db/models/default.go @@ -7,6 +7,7 @@ import ( const ( hostsTable = "hosts_tbl" clustersTable = "clusters_tbl" + zonesTable = "zones_tbl" guestsTable = "guests_tbl" baremetalsTable = "baremetals_tbl" @@ -38,14 +39,18 @@ const ( hostWiresTable = "hostwires_tbl" reserveDipsTable = "reservedips_tbl" + + cloudproviderTable = "cloudproviders_tbl" ) var ( Hosts Resourcer HostWires Resourcer - Clusters Resourcer - Guests Resourcer + Clusters Resourcer + Zones Resourcer + Guests Resourcer + CloudProviders Resourcer Baremetals Resourcer BaremetalAgents Resourcer @@ -84,7 +89,9 @@ func Init(dialect string, args ...interface{}) error { } Hosts, _ = NewHostResource(db.DB) Clusters, _ = NewClusterResource(db.DB) + Zones, _ = NewZoneResource(db.DB) Guests, _ = NewGuestResource(db.DB) + CloudProviders, _ = NewCloudproviderResource(db.DB) Baremetals, _ = NewBaremetalResource(db.DB) BaremetalAgents, _ = NewBaremetalAgentResource(db.DB) diff --git a/pkg/scheduler/db/models/zone.go b/pkg/scheduler/db/models/zone.go new file mode 100644 index 0000000000..b55a4f7737 --- /dev/null +++ b/pkg/scheduler/db/models/zone.go @@ -0,0 +1,41 @@ +package models + +import ( + "github.com/jinzhu/gorm" +) + +type Zone struct { + StandaloneModel + Location string `gorm:"column:location"` + ManagerUri string `gorm:"column:manager_uri"` + CloudregionId string `gorm:"column:cloudregion_id"` +} + +func (z Zone) TableName() string { + return zonesTable +} + +func (z Zone) String() string { + s, _ := JsonString(z) + return s +} + +func NewZoneResource(db *gorm.DB) (Resourcer, error) { + return newResource(db, zonesTable, + func() interface{} { + return &Zone{} + }, + func() interface{} { + zones := []Zone{} + return &zones + }, + ) +} + +func FetchZoneByID(id string) (*Zone, error) { + zone, err := FetchByID(Zones, id) + if err != nil { + return nil, err + } + return zone.(*Zone), nil +} diff --git a/pkg/scheduler/manager/task_queue.go b/pkg/scheduler/manager/task_queue.go index 0fffd0268d..0a413c8703 100644 --- a/pkg/scheduler/manager/task_queue.go +++ b/pkg/scheduler/manager/task_queue.go @@ -200,8 +200,8 @@ type TaskManager struct { func NewTaskManager(stopCh <-chan struct{}) *TaskManager { return &TaskManager{ taskExecutorQueueManager: NewTaskExecutorQueueManager(stopCh), - stopCh: stopCh, - lock: sync.Mutex{}, + stopCh: stopCh, + lock: sync.Mutex{}, } }