feat(scheduler): optimize the speed of network predicate

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 {

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()

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 {