From 5741782ccab34d7c9963a97a627707ea2b41fc0d Mon Sep 17 00:00:00 2001 From: rainzm Date: Thu, 29 Apr 2021 11:47:45 +0800 Subject: [PATCH] feat(scheduler): optimize the speed of network predicate --- pkg/compute/models/networks.go | 57 +++++++++++++++++++ .../algorithm/predicates/network_predicate.go | 39 +++++++++++-- .../predicates/network_schedtag_predicate.go | 2 +- 3 files changed, 91 insertions(+), 7 deletions(-) diff --git a/pkg/compute/models/networks.go b/pkg/compute/models/networks.go index e4a24996bc..0f1e5626cd 100644 --- a/pkg/compute/models/networks.go +++ b/pkg/compute/models/networks.go @@ -173,6 +173,59 @@ func (self *SNetwork) ValidateDeleteCondition(ctx context.Context) error { return self.SSharableVirtualResourceBase.ValidateDeleteCondition(ctx) } +func (nm *SNetworkManager) jointNetworkCount(manager db.IModelManager, netIds []string, filter func(*sqlchemy.SQuery) *sqlchemy.SQuery) *sqlchemy.SQuery { + q := manager.Query("network_id").In("network_id", netIds) + if filter != nil { + q = filter(q) + } + q = q.AppendField(sqlchemy.COUNT("count")) + return q +} + +func (nm *SNetworkManager) GetTotalNicCount(netIds []string) (map[string]int, error) { + guestNetworkQ := nm.jointNetworkCount(GuestnetworkManager, netIds, func(q *sqlchemy.SQuery) *sqlchemy.SQuery { + return q.Filter(sqlchemy.IsFalse(q.Field("virtual"))) + }) + + groupNetworkQ := nm.jointNetworkCount(GroupnetworkManager, netIds, nil) + hostNetworkQ := nm.jointNetworkCount(HostnetworkManager, netIds, nil) + lbNetworkQ := nm.jointNetworkCount(LoadbalancernetworkManager, netIds, nil) + dbInstanceNetworkQ := nm.jointNetworkCount(DBInstanceNetworkManager, netIds, nil) + eipNetworkQ := nm.jointNetworkCount(ElasticipManager, netIds, nil) + natgatewayNetworkQ := nm.jointNetworkCount(NatGatewayManager, netIds, func(q *sqlchemy.SQuery) *sqlchemy.SQuery { + return q.Filter(sqlchemy.IsNotEmpty(q.Field("ip_addr"))) + }) + + reserverIpsQ := nm.jointNetworkCount(ReservedipManager, netIds, filterExpiredReservedIps) + + sq := NetworkinterfacenetworkManager.Query("networkinterface_id", "network_id").In("network_id", netIds).Distinct().SubQuery() + networkInterfaceQ := NetworkInterfaceManager.Query() + networkInterfaceQ = networkInterfaceQ.Join(sq, sqlchemy.Equals(networkInterfaceQ.Field("id"), sq.Field("networkinterface_id"))) + networkInterfaceQ.AppendField(sq.Field("network_id", "network_id")) + networkInterfaceQ.AppendField(sqlchemy.COUNT("count")) + networkInterfaceQ.GroupBy("network_id") + + union, err := sqlchemy.UnionWithError(guestNetworkQ, groupNetworkQ, hostNetworkQ, lbNetworkQ, dbInstanceNetworkQ, eipNetworkQ, natgatewayNetworkQ, reserverIpsQ, networkInterfaceQ) + if err != nil { + return nil, err + } + networkCounts := make([]struct { + NetworkId string + Count int + }, 0) + err = union.Query().All(&networkCounts) + if err != nil { + return nil, err + } + ret := map[string]int{} + for _, nc := range networkCounts { + if len(nc.NetworkId) > 0 { + ret[nc.NetworkId] += nc.Count + } + } + return ret, nil +} + func (self *SNetwork) GetTotalNicCount() (int, error) { count, err := self.GetAllocatedNicCount() if err != nil { @@ -1002,6 +1055,10 @@ func (self *SNetwork) GetFreeAddressCount() (int, error) { return self.getFreeAddressCount() } +func (self *SNetwork) GetTotalAddressCount() int { + return self.getIPRange().AddressCount() +} + func (self *SNetwork) getFreeAddressCount() (int, error) { used, err := self.GetTotalNicCount() if err != nil { diff --git a/pkg/scheduler/algorithm/predicates/network_predicate.go b/pkg/scheduler/algorithm/predicates/network_predicate.go index 76ee3f4e2c..30b030afe4 100644 --- a/pkg/scheduler/algorithm/predicates/network_predicate.go +++ b/pkg/scheduler/algorithm/predicates/network_predicate.go @@ -17,7 +17,9 @@ package predicates import ( "fmt" + "yunion.io/x/pkg/errors" "yunion.io/x/pkg/util/netutils" + "yunion.io/x/pkg/util/sets" "yunion.io/x/pkg/utils" computeapi "yunion.io/x/onecloud/pkg/apis/compute" @@ -35,6 +37,7 @@ import ( type NetworkPredicate struct { BasePredicate plugin.BasePlugin + networkFreePortCount map[string]int } func (p *NetworkPredicate) Name() string { @@ -50,11 +53,27 @@ func (p *NetworkPredicate) PreExecute(u *core.Unit, cs []core.Candidater) (bool, if len(data.Networks) == 0 { return false, nil } + networkIds := sets.NewString() + for i := range cs { + for _, net := range cs[i].Getter().Networks() { + networkIds.Insert(net.GetId()) + } + } + netCounts, err := models.NetworkManager.GetTotalNicCount(networkIds.UnsortedList()) + if err != nil { + return false, errors.Wrap(err, "unable to GetTotalNicCount") + } + p.networkFreePortCount = map[string]int{} + for i := range cs { + for _, net := range cs[i].Getter().Networks() { + p.networkFreePortCount[net.Id] = net.GetTotalAddressCount() - netCounts[net.Id] + } + } return true, nil } -func IsNetworksAvailable(c core.Candidater, data *api.SchedInfo, req *computeapi.NetworkConfig, networks []*api.CandidateNetwork, netTypes []string) (int, []core.PredicateFailureReason) { +func IsNetworksAvailable(c core.Candidater, data *api.SchedInfo, req *computeapi.NetworkConfig, networks []*api.CandidateNetwork, netTypes []string, getFreePort func(string) int) (int, []core.PredicateFailureReason) { var fullErrMsgs []core.PredicateFailureReason var freeCnt int @@ -74,10 +93,10 @@ func IsNetworksAvailable(c core.Candidater, data *api.SchedInfo, req *computeapi checkNets := func(tmpNets []*api.CandidateNetwork) { for _, n := range tmpNets { - if errMsg := IsNetworkAvailable(c, data, req, n, netTypes); errMsg != nil { + if errMsg := IsNetworkAvailable(c, data, req, n, netTypes, getFreePort); errMsg != nil { fullErrMsgs = append(fullErrMsgs, errMsg) } else { - freeCnt = freeCnt + c.Getter().GetFreePort(n.GetId()) + freeCnt = freeCnt + getFreePort(n.GetId()) } } } @@ -108,7 +127,7 @@ func IsNetworksAvailable(c core.Candidater, data *api.SchedInfo, req *computeapi func IsNetworkAvailable( c core.Candidater, data *api.SchedInfo, req *computeapi.NetworkConfig, n *api.CandidateNetwork, - netTypes []string, + netTypes []string, getFreePort func(string) int, ) core.PredicateFailureReason { address := req.Address private := req.Private @@ -129,7 +148,11 @@ func IsNetworkAvailable( } } - if !(c.Getter().GetFreePort(n.GetId()) > 0 || isMigrate()) { + if getFreePort == nil { + getFreePort = c.Getter().GetFreePort + } + + if !(getFreePort(n.GetId()) > 0 || isMigrate()) { return FailReason{ Reason: fmt.Sprintf("%v(%v): ports use up", n.Name, n.Id), Type: NetworkPort, @@ -226,9 +249,13 @@ func (p *NetworkPredicate) Execute(u *core.Unit, c core.Candidater) (bool, []cor networks := getter.Networks() d := u.SchedData() + getFreePort := func(id string) int { + return p.networkFreePortCount[id] - c.Getter().GetPendingUsage().NetUsage.Get(id) + } + for _, reqNet := range d.Networks { netTypes := p.GetNetworkTypes(u, reqNet.NetType) - freePortCnt, errs := IsNetworksAvailable(c, d, reqNet, networks, netTypes) + freePortCnt, errs := IsNetworksAvailable(c, d, reqNet, networks, netTypes, getFreePort) if len(errs) > 0 { h.ExcludeByErrors(errs) return h.GetResult() diff --git a/pkg/scheduler/algorithm/predicates/network_schedtag_predicate.go b/pkg/scheduler/algorithm/predicates/network_schedtag_predicate.go index 9ea5309281..4261cc2e2b 100644 --- a/pkg/scheduler/algorithm/predicates/network_schedtag_predicate.go +++ b/pkg/scheduler/algorithm/predicates/network_schedtag_predicate.go @@ -98,7 +98,7 @@ func (p *NetworkSchedtagPredicate) IsResourceMatchInput(input ISchedtagCustomer, func (p *NetworkSchedtagPredicate) IsResourceFitInput(u *core.Unit, c core.Candidater, res ISchedtagCandidateResource, input ISchedtagCustomer) core.PredicateFailureReason { network := res.(*api.CandidateNetwork) net := input.(*netW) - return IsNetworkAvailable(c, u.SchedData(), net.NetworkConfig, network, p.GetNetworkTypes(net.NetType)) + return IsNetworkAvailable(c, u.SchedData(), net.NetworkConfig, network, p.GetNetworkTypes(net.NetType), nil) } func (p *NetworkSchedtagPredicate) GetNetworkTypes(specifyType string) []string {