feat(scheduler): loadaware support nodemetric

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/evanphx/json-patch v4.12.0+incompatible
 	github.com/google/go-cmp v0.5.9
 	github.com/kubewharf/godel-scheduler-api v0.0.0-20231129083306-7282cca41429
-	github.com/kubewharf/katalyst-api v0.1.16
+	github.com/kubewharf/katalyst-api v0.4.1-0.20240222122824-be538f641f58
 	github.com/onsi/ginkgo v1.14.0
 	github.com/onsi/gomega v1.19.0
 	github.com/pkg/errors v0.9.1

go.sum

@@ -364,8 +364,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kubewharf/godel-scheduler-api v0.0.0-20231129083306-7282cca41429 h1:YpoBFCSUzafRlN5FVEGfGvopEWH3OSV/sAik3MAmlVo=
 github.com/kubewharf/godel-scheduler-api v0.0.0-20231129083306-7282cca41429/go.mod h1:ten2w2gEy8fdAjGiy7WJ5FrTBQ2PF2GQjSJK02oHvhU=
-github.com/kubewharf/katalyst-api v0.1.16 h1:oMjwYOb3WIikZsJjkWcb1ZfpcLJqGim8h05MsbXte3w=
-github.com/kubewharf/katalyst-api v0.1.16/go.mod h1:iVILS5UL5PRtkUPH2Iu1K/gFGTPMNItnth5fmQ80VGE=
+github.com/kubewharf/katalyst-api v0.4.1-0.20240222122824-be538f641f58 h1:D9dCR5EIR0k0Qil2A5biZjrubagRkEr7fyov6fb2ApY=
+github.com/kubewharf/katalyst-api v0.4.1-0.20240222122824-be538f641f58/go.mod h1:Y2IeIorxQamF2a3oa0+URztl5QCSty6Jj3zD83R8J9k=
 github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de/go.mod h1:zAbeS9B/r2mtpb6U+EI2rYA5OAXxsYw6wTamcNW+zcE=
 github.com/lithammer/dedent v1.1.0/go.mod h1:jrXYCQtgg0nJiN+StA2KgR7w6CiQNv9Fd/Z9BP0jIOc=
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=

pkg/framework/api/interface.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
 
+	podutil "github.com/kubewharf/godel-scheduler/pkg/util/pod"
 	"github.com/kubewharf/godel-scheduler/pkg/volume/scheduling"
 )
 
@@ -660,6 +661,8 @@ type SchedulerFrameworkHandle interface {
 	GetPDBItemListForOwner(ownerType, ownerKey string) (bool, bool, []string)
 	// Note: The function's underlying access is Snapshot, Snapshot operations are lock-free.
 	GetOwnerLabels(ownerType, ownerKey string) map[string]string
+	GetLoadAwareNodeMetricInfo(nodeName string, resourceType podutil.PodResourceType) *LoadAwareNodeMetricInfo
+	GetLoadAwareNodeUsage(nodeName string, resourceType podutil.PodResourceType) *LoadAwareNodeUsage
 
 	GetPreemptionFrameworkForPod(*v1.Pod) SchedulerPreemptionFramework
 	GetPreemptionPolicy(deployName string) string

pkg/framework/api/types.go

@@ -1149,3 +1149,19 @@ func (s *GenerationStringSetImpl) Strings() []string {
 	}
 	return strs
 }
+
+// In order to minimize computational overhead as much as possible, only the CPU/mem fields are used here.
+// In the future, field expansion needs to be carried out based on actual needs
+type LoadAwareNodeUsage struct {
+	RequestMilliCPU int64
+	RequestMEM      int64
+	ProfileMilliCPU int64
+	ProfileMEM      int64
+}
+
+type LoadAwareNodeMetricInfo struct {
+	Name                 string
+	UpdateTime           metav1.Time
+	ProfileMilliCPUUsage int64
+	ProfileMEMUsage      int64
+}

pkg/scheduler/apis/config/types_pluginargs.go

@@ -250,4 +250,12 @@ type LoadAwareArgs struct {
 
 	// Estimator indicates the expected Estimator to use
 	Estimator string `json:"estimator,omitempty"`
+
+	FilterExpiredNodeMetrics    bool                      `json:"filterExpiredNodeMetrics,omitempty"`
+	NodeMetricExpirationSeconds int64                     `json:"nodeMetricExpirationSeconds,omitempty"`
+	UsageThresholds             map[v1.ResourceName]int64 `json:"usageThresholds,omitempty"`
+
+	// EstimatedScalingFactors indicates the factor when estimating resource usage.
+	// Is CPU scaling factor is 80, estimated CPU = 80 / 100 * request.cpu
+	EstimatedScalingFactors map[v1.ResourceName]int64 `json:"estimatedScalingFactors,omitempty"`
 }

pkg/scheduler/cache/commonstores/load_aware_store/internal.go

@@ -0,0 +1,196 @@
+/*
+Copyright 2023 The Godel Scheduler Authors.
+
+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 loadawarestore
+
+import (
+	"github.com/kubewharf/godel-scheduler/pkg/util/generationstore"
+	podutil "github.com/kubewharf/godel-scheduler/pkg/util/pod"
+	katalystv1alpha1 "github.com/kubewharf/katalyst-api/pkg/apis/node/v1alpha1"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+type podBasicInfo struct {
+	Key              string
+	MilliCPU, Memory int64
+	PodResourceType  podutil.PodResourceType
+}
+
+func newPodBasicInfo(pod *v1.Pod) (*podBasicInfo, error) {
+	resourceType, err := podutil.GetPodResourceType(pod)
+	if err != nil {
+		return nil, err
+	}
+	return &podBasicInfo{
+		Key:             podutil.GetPodKey(pod),
+		MilliCPU:        podutil.GetPodRequest(pod, v1.ResourceCPU, resource.DecimalSI).MilliValue(),
+		Memory:          podutil.GetPodRequest(pod, v1.ResourceMemory, resource.DecimalSI).Value(),
+		PodResourceType: resourceType,
+	}, nil
+}
+
+// ----------------------------------- PodMetricInfos -----------------------------------
+
+type PodMetricInfos struct {
+	ProfileMilliCPU, ProfileMEM int64
+	RequestMilliCPU, RequestMEM int64
+	ProfilePods                 sets.String
+}
+
+func newPodMetricInfos(cnr *katalystv1alpha1.CustomNodeResource, existingPods map[string]podBasicInfo, resourceType podutil.PodResourceType) *PodMetricInfos {
+	var profileMilliCPU, profileMEM int64
+	var requestMilliCPU, requestMEM int64
+	pods := sets.NewString()
+
+	if cnr != nil && cnr.Status.NodeMetricStatus != nil {
+		groupMetric := cnr.Status.NodeMetricStatus.GroupMetric
+		for i := range groupMetric {
+			if podutil.GetResourceTypeFromQoS(groupMetric[i].QoSLevel) == resourceType {
+				u := groupMetric[i].GenericUsage
+				profileMilliCPU += u.CPU.MilliValue()
+				profileMEM += u.Memory.Value()
+				pods.Insert(groupMetric[i].PodList...)
+			}
+		}
+	}
+
+	for key, pInfo := range existingPods {
+		if pInfo.PodResourceType == resourceType && !pods.Has(key) {
+			requestMilliCPU += pInfo.MilliCPU
+			requestMEM += pInfo.Memory
+		}
+	}
+
+	return &PodMetricInfos{
+		ProfileMilliCPU: profileMilliCPU,
+		ProfileMEM:      profileMEM,
+		RequestMilliCPU: requestMilliCPU,
+		RequestMEM:      requestMEM,
+		ProfilePods:     pods,
+	}
+}
+
+func (i *PodMetricInfos) Clone() *PodMetricInfos {
+	clone := *i // No need to DeepCopy sets for Snapshot
+	return &clone
+}
+
+// ----------------------------------- NodeMetricInfo -----------------------------------
+
+// ATTENTION: the NodeMetricInfo's lifecycle is accompanied by CNR, not CNR.Status.NodeMetricStatus
+type NodeMetricInfo struct {
+	name             string
+	cnrExist         bool
+	updateTime       metav1.Time
+	gtPodMetricInfos *PodMetricInfos
+	bePodMetricInfos *PodMetricInfos
+	allPods          map[string]podBasicInfo
+	generation       int64
+}
+
+var _ generationstore.StoredObj = &NodeMetricInfo{}
+
+func NewNodeMetricInfo(nodeName string, cnr *katalystv1alpha1.CustomNodeResource) *NodeMetricInfo {
+	return &NodeMetricInfo{
+		name:             nodeName,
+		cnrExist:         cnr != nil,
+		updateTime:       parseUpdateTimeFromCNR(cnr),
+		gtPodMetricInfos: newPodMetricInfos(cnr, nil, podutil.GuaranteedPod),
+		bePodMetricInfos: newPodMetricInfos(cnr, nil, podutil.BestEffortPod),
+		allPods:          make(map[string]podBasicInfo),
+	}
+}
+
+func (i *NodeMetricInfo) Reset(cnr *katalystv1alpha1.CustomNodeResource) {
+	i.cnrExist = cnr != nil
+	i.updateTime = parseUpdateTimeFromCNR(cnr)
+	i.gtPodMetricInfos = newPodMetricInfos(cnr, i.allPods, podutil.GuaranteedPod)
+	i.bePodMetricInfos = newPodMetricInfos(cnr, i.allPods, podutil.BestEffortPod)
+}
+
+func (i *NodeMetricInfo) PodOp(pInfo *podBasicInfo, isCacheStore, isAdd bool) {
+	if isAdd {
+		i.allPods[pInfo.Key] = *pInfo
+	} else {
+		delete(i.allPods, pInfo.Key)
+	}
+
+	var infos *PodMetricInfos
+	if pInfo.PodResourceType == podutil.GuaranteedPod {
+		infos = i.gtPodMetricInfos
+	} else {
+		infos = i.bePodMetricInfos
+	}
+	if isCacheStore && infos.ProfilePods.Has(pInfo.Key) {
+		// 1. We won't read ProfilePods in Snapshot, because we believe that 'unscheduled Pod should not appear in ProfilePods'.
+		// 2. We only care about pods that have not been counted by NodeMetric.
+		return
+	}
+
+	if isAdd {
+		infos.RequestMilliCPU += pInfo.MilliCPU
+		infos.RequestMEM += pInfo.Memory
+	} else {
+		infos.RequestMilliCPU -= pInfo.MilliCPU
+		infos.RequestMEM -= pInfo.Memory
+	}
+}
+
+func (i *NodeMetricInfo) GetGeneration() int64 {
+	return i.generation
+}
+
+func (i *NodeMetricInfo) SetGeneration(generation int64) {
+	i.generation = generation
+}
+
+func (i *NodeMetricInfo) CanBeRecycle() bool {
+	if i == nil {
+		return true
+	}
+	return !i.cnrExist && len(i.allPods) == 0
+}
+
+func (i *NodeMetricInfo) Clone() *NodeMetricInfo {
+	return &NodeMetricInfo{
+		name:             i.name,
+		cnrExist:         i.cnrExist,
+		updateTime:       i.updateTime,
+		gtPodMetricInfos: i.gtPodMetricInfos.Clone(),
+		bePodMetricInfos: i.bePodMetricInfos.Clone(),
+		allPods:          cloneAllPods(i.allPods),
+		generation:       i.generation,
+	}
+}
+
+func parseUpdateTimeFromCNR(cnr *katalystv1alpha1.CustomNodeResource) metav1.Time {
+	if cnr == nil || cnr.Status.NodeMetricStatus == nil {
+		return metav1.Time{}
+	}
+	return cnr.Status.NodeMetricStatus.UpdateTime
+}
+
+func cloneAllPods(allPods map[string]podBasicInfo) map[string]podBasicInfo {
+	ret := make(map[string]podBasicInfo, len(allPods))
+	for key, info := range allPods {
+		ret[key] = info
+	}
+	return ret
+}