Repository: koordinator-sh/koordinator

Issue:
[proposal] Support dominant resource weight for scoring in load aware scheduling.
<!-- Please only use this template for submitting proposal -->

**What is your proposal**:

We can add new scoring weight arg for dominant resource in load aware scheduling.

The node's dominant resource is defined as the most used resource on node:

> dominant resource score = min(allocatable - usedOfExistingPods - estimatedOfIncomingPod) / allocatable)

This idea comes from the concept of Dominant Resource Fairness, known as DRF. Paper is here: [www.cs.berkeley.edu/~matei/papers/2011/nsdi_drf.pdf](https://people.eecs.berkeley.edu/~matei/papers/2011/nsdi_drf.pdf)


**Why is this needed**:

Currently, We can only set resource weight base on resource name. The weight is hard to set when resources usage are already unbalanced in a multi-resource environment because of multiple node type, long run pods' affinity specs or other reasons.

For example, if node A uses cpu heavily and memory are available, node B uses memory heavily and cpu are available. We should prefer to schedule cpu-heavy pods on B and memory-heavy pods on A. We can only have fixed weight on resource now and all pods will prefer to A if memory has higher weight or B if cpu has higher weight.

**Is there a suggested solution, if so, please add it**:

Provided in proposal.


Hints from issue comments:
@zheng-weihao: @saintube @ZiMengSheng Hi, any thought on this issue? PR is ready after #2582 is merged.@ZiMengSheng: Will we introduce a new plugin similar to BalancedResourceAllocation of NodeResourcesFit?@zheng-weihao: > Will we introduce a new plugin similar to BalancedResourceAllocation of NodeResourcesFit?

I prefer to add args on load aware scheduling plugin, since the main goal for load aware is to balance resource usage.

Solution patch:
diff --git a/pkg/scheduler/apis/config/types.go b/pkg/scheduler/apis/config/types.go
index 449c735b..5392698b 100644
--- a/pkg/scheduler/apis/config/types.go
+++ b/pkg/scheduler/apis/config/types.go
@@ -49,6 +49,9 @@ type LoadAwareSchedulingArgs struct {
 	// ProdUsageThresholds indicates the resource utilization threshold of Prod Pods compared to the whole machine.
 	// Not enabled by default
 	ProdUsageThresholds map[corev1.ResourceName]int64
+	// ProdUsageIncludeSys indicates whether to include system usage (not used by pods)
+	// when summing up current usage for prod pods.
+	ProdUsageIncludeSys bool
 	// ScoreAccordingProdUsage controls whether to score according to the utilization of Prod Pod
 	ScoreAccordingProdUsage bool
 	// Estimator indicates the expected Estimator to use
@@ -66,6 +69,14 @@ type LoadAwareSchedulingArgs struct {
 	AllowCustomizeEstimation bool
 	// Aggregated supports resource utilization filtering and scoring based on percentile statistics
 	Aggregated *LoadAwareSchedulingAggregatedArgs
+	// SupportedResourceNames is the list of extra resource names that can be used in load-aware scheduling.
+	// cpu, memory and all other resources that show up in args are supported by default.
+	//
+	// If more resource are added in collection, don't show up as
+	// filter thresholds or score weights in plugin args
+	// and only set up in custom node annotations,
+	// we should pass these resource names in plugin args explicitly.
+	SupportedResources []corev1.ResourceName
 }
 
 type LoadAwareSchedulingAggregatedArgs struct {
diff --git a/pkg/scheduler/apis/config/v1/types.go b/pkg/scheduler/apis/config/v1/types.go
index cdc27680..1cf9561f 100644
--- a/pkg/scheduler/apis/config/v1/types.go
+++ b/pkg/scheduler/apis/config/v1/types.go
@@ -48,6 +48,9 @@ type LoadAwareSchedulingArgs struct {
 	// ProdUsageThresholds indicates the resource utilization threshold of Prod Pods compared to the whole machine.
 	// Not enabled by default
 	ProdUsageThresholds map[corev1.ResourceName]int64 `json:"prodUsageThresholds,omitempty"`
+	// ProdUsageIncludeSys indicates whether to include system usage (not used by pods)
+	// when summing up current usage for prod pods.
+	ProdUsageIncludeSys bool `json:"prodUsageIncludeSys,omitempty"`
 	// ScoreAccordingProdUsage controls whether to score according to the utilization of Prod Pod
 	ScoreAccordingProdUsage *bool `json:"scoreAccordingProdUsage,omitempty"`
 	// Estimator indicates the expected Estimator to use
@@ -65,6 +68,14 @@ type LoadAwareSchedulingArgs struct {
 	AllowCustomizeEstimation bool `json:"allowCustomizeEstimation,omitempty"`
 	// Aggregated supports resource utilization filtering and scoring based on percentile statistics
 	Aggregated *LoadAwareSchedulingAggregatedArgs `json:"aggregated,omitempty"`
+	// SupportedResourceNames is the list of extra resource names that can be used in load-aware scheduling.
+	// cpu, memory and all other resources that show up in args are supported by default.
+	//
+	// If more resource are added in collection, don't show up as
+	// filter thresholds or score weights in plugin args
+	// and only set up in custom node annotations,
+	// we should pass these resource names in plugin args explicitly.
+	SupportedResources []corev1.ResourceName `json:"supportedResources,omitempty"`
 }
 
 type LoadAwareSchedulingAggregatedArgs struct {
diff --git a/pkg/scheduler/apis/config/v1/zz_generated.conversion.go b/pkg/scheduler/apis/config/v1/zz_generated.conversion.go
index ce419caa..66ee75a0 100644
--- a/pkg/scheduler/apis/config/v1/zz_generated.conversion.go
+++ b/pkg/scheduler/apis/config/v1/zz_generated.conversion.go
@@ -395,6 +395,7 @@ func autoConvert_v1_LoadAwareSchedulingArgs_To_config_LoadAwareSchedulingArgs(in
 	out.ResourceWeights = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.ResourceWeights))
 	out.UsageThresholds = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.UsageThresholds))
 	out.ProdUsageThresholds = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.ProdUsageThresholds))
+	out.ProdUsageIncludeSys = in.ProdUsageIncludeSys
 	if err := metav1.Convert_Pointer_bool_To_bool(&in.ScoreAccordingProdUsage, &out.ScoreAccordingProdUsage, s); err != nil {
 		return err
 	}
@@ -412,6 +413,7 @@ func autoConvert_v1_LoadAwareSchedulingArgs_To_config_LoadAwareSchedulingArgs(in
 	} else {
 		out.Aggregated = nil
 	}
+	out.SupportedResources = *(*[]corev1.ResourceName)(unsafe.Pointer(&in.SupportedResources))
 	return nil
 }
 
@@ -427,6 +429,7 @@ func autoConvert_config_LoadAwareSchedulingArgs_To_v1_LoadAwareSchedulingArgs(in
 	out.ResourceWeights = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.ResourceWeights))
 	out.UsageThresholds = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.UsageThresholds))
 	out.ProdUsageThresholds = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.ProdUsageThresholds))
+	out.ProdUsageIncludeSys = in.ProdUsageIncludeSys
 	if err := metav1.Convert_bool_To_Pointer_bool(&in.ScoreAccordingProdUsage, &out.ScoreAccordingProdUsage, s); err != nil {
 		return err
 	}
@@ -444,6 +447,7 @@ func autoConvert_config_LoadAwareSchedulingArgs_To_v1_LoadAwareSchedulingArgs(in
 	} else {
 		out.Aggregated = nil
 	}
+	out.SupportedResources = *(*[]corev1.ResourceName)(unsafe.Pointer(&in.SupportedResources))
 	return nil
 }
 
diff --git a/pkg/scheduler/apis/config/v1/zz_generated.deepcopy.go b/pkg/scheduler/apis/config/v1/zz_generated.deepcopy.go
index 24d02eae..4e3fcb31 100644
--- a/pkg/scheduler/apis/config/v1/zz_generated.deepcopy.go
+++ b/pkg/scheduler/apis/config/v1/zz_generated.deepcopy.go
@@ -319,6 +319,11 @@ func (in *LoadAwareSchedulingArgs) DeepCopyInto(out *LoadAwareSchedulingArgs) {
 		*out = new(LoadAwareSchedulingAggregatedArgs)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.SupportedResources != nil {
+		in, out := &in.SupportedResources, &out.SupportedResources
+		*out = make([]corev1.ResourceName, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 
diff --git a/pkg/scheduler/apis/config/v1beta3/types.go b/pkg/scheduler/apis/config/v1beta3/types.go
index 15a133fe..ef33e46a 100644
--- a/pkg/scheduler/apis/config/v1beta3/types.go
+++ b/pkg/scheduler/apis/config/v1beta3/types.go
@@ -49,6 +49,9 @@ type LoadAwareSchedulingArgs struct {
 	// ProdUsageThresholds indicates the resource utilization threshold of Prod Pods compared to the whole machine.
 	// Not enabled by default
 	ProdUsageThresholds map[corev1.ResourceName]int64 `json:"prodUsageThresholds,omitempty"`
+	// ProdUsageIncludeSys indicates whether to include system usage (not used by pods)
+	// when summing up current usage for prod pods.
+	ProdUsageIncludeSys bool `json:"prodUsageIncludeSys,omitempty"`
 	// ScoreAccordingProdUsage controls whether to score according to the utilization of Prod Pod
 	ScoreAccordingProdUsage *bool `json:"scoreAccordingProdUsage,omitempty"`
 	// Estimator indicates the expected Estimator to use
@@ -66,6 +69,14 @@ type LoadAwareSchedulingArgs struct {
 	AllowCustomizeEstimation bool `json:"allowCustomizeEstimation,omitempty"`
 	// Aggregated supports resource utilization filtering and scoring based on percentile statistics
 	Aggregated *LoadAwareSchedulingAggregatedArgs `json:"aggregated,omitempty"`
+	// SupportedResourceNames is the list of extra resource names that can be used in load-aware scheduling.
+	// cpu, memory and all other resources that show up in args are supported by default.
+	//
+	// If more resource are added in collection, don't show up as
+	// filter thresholds or score weights in plugin args
+	// and only set up in custom node annotations,
+	// we should pass these resource names in plugin args explicitly.
+	SupportedResources []corev1.ResourceName `json:"supportedResources,omitempty"`
 }
 
 type LoadAwareSchedulingAggregatedArgs struct {
diff --git a/pkg/scheduler/apis/config/v1beta3/zz_generated.conversion.go b/pkg/scheduler/apis/config/v1beta3/zz_generated.conversion.go
index 27e4450b..fb176ba7 100644
--- a/pkg/scheduler/apis/config/v1beta3/zz_generated.conversion.go
+++ b/pkg/scheduler/apis/config/v1beta3/zz_generated.conversion.go
@@ -395,6 +395,7 @@ func autoConvert_v1beta3_LoadAwareSchedulingArgs_To_config_LoadAwareSchedulingAr
 	out.ResourceWeights = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.ResourceWeights))
 	out.UsageThresholds = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.UsageThresholds))
 	out.ProdUsageThresholds = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.ProdUsageThresholds))
+	out.ProdUsageIncludeSys = in.ProdUsageIncludeSys
 	if err := v1.Convert_Pointer_bool_To_bool(&in.ScoreAccordingProdUsage, &out.ScoreAccordingProdUsage, s); err != nil {
 		return err
 	}
@@ -412,6 +413,7 @@ func autoConvert_v1beta3_LoadAwareSchedulingArgs_To_config_LoadAwareSchedulingAr
 	} else {
 		out.Aggregated = nil
 	}
+	out.SupportedResources = *(*[]corev1.ResourceName)(unsafe.Pointer(&in.SupportedResources))
 	return nil
 }
 
@@ -422,6 +424,7 @@ func autoConvert_config_LoadAwareSchedulingArgs_To_v1beta3_LoadAwareSchedulingAr
 	out.ResourceWeights = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.ResourceWeights))
 	out.UsageThresholds = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.UsageThresholds))
 	out.ProdUsageThresholds = *(*map[corev1.ResourceName]int64)(unsafe.Pointer(&in.ProdUsageThresholds))
+	out.ProdUsageIncludeSys = in.ProdUsageIncludeSys
 	if err := v1.Convert_bool_To_Pointer_bool(&in.ScoreAccordingProdUsage, &out.ScoreAccordingProdUsage, s); err != nil {
 		return err
 	}
@@ -439,6 +442,7 @@ func autoConvert_config_LoadAwareSchedulingArgs_To_v1beta3_LoadAwareSchedulingAr
 	} else {
 		out.Aggregated = nil
 	}
+	out.SupportedResources = *(*[]corev1.ResourceName)(unsafe.Pointer(&in.SupportedResources))
 	return nil
 }
 
diff --git a/pkg/scheduler/apis/config/v1beta3/zz_generated.deepcopy.go b/pkg/scheduler/apis/config/v1beta3/zz_generated.deepcopy.go
index b4bb2875..8e2638df 100644
--- a/pkg/scheduler/apis/config/v1beta3/zz_generated.deepcopy.go
+++ b/pkg/scheduler/apis/config/v1beta3/zz_generated.deepcopy.go
@@ -319,6 +319,11 @@ func (in *LoadAwareSchedulingArgs) DeepCopyInto(out *LoadAwareSchedulingArgs) {
 		*out = new(LoadAwareSchedulingAggregatedArgs)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.SupportedResources != nil {
+		in, out := &in.SupportedResources, &out.SupportedResources
+		*out = make([]corev1.ResourceName, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 
diff --git a/pkg/scheduler/apis/config/zz_generated.deepcopy.go b/pkg/scheduler/apis/config/zz_generated.deepcopy.go
index ef839b91..1de975db 100644
--- a/pkg/scheduler/apis/config/zz_generated.deepcopy.go
+++ b/pkg/scheduler/apis/config/zz_generated.deepcopy.go
@@ -258,6 +258,11 @@ func (in *LoadAwareSchedulingArgs) DeepCopyInto(out *LoadAwareSchedulingArgs) {
 		*out = new(LoadAwareSchedulingAggregatedArgs)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.SupportedResources != nil {
+		in, out := &in.SupportedResources, &out.SupportedResources
+		*out = make([]v1.ResourceName, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 
diff --git a/pkg/scheduler/plugins/loadaware/estimator/default_estimator.go b/pkg/scheduler/plugins/loadaware/estimator/default_estimator.go
index cf012b3b..b0b526a2 100644
--- a/pkg/scheduler/plugins/loadaware/estimator/default_estimator.go
+++ b/pkg/scheduler/plugins/loadaware/estimator/default_estimator.go
@@ -39,16 +39,14 @@ const (
 )
 
 type DefaultEstimator struct {
-	resourceWeights map[corev1.ResourceName]int64
-	scalingFactors  map[corev1.ResourceName]int64
-	allowCustomize  bool
+	scalingFactors map[corev1.ResourceName]int64
+	allowCustomize bool
 }
 
 func NewDefaultEstimator(args *config.LoadAwareSchedulingArgs, handle framework.Handle) (Estimator, error) {
 	return &DefaultEstimator{
-		resourceWeights: args.ResourceWeights,
-		scalingFactors:  args.EstimatedScalingFactors,
-		allowCustomize:  args.AllowCustomizeEstimation,
+		scalingFactors: args.EstimatedScalingFactors,
+		allowCustomize: args.AllowCustomizeEstimation,
 	}, nil
 }
 
@@ -70,16 +68,16 @@ func (e *DefaultEstimator) EstimatePod(pod *corev1.Pod) (map[corev1.ResourceName
 			}
 		}
 	}
-	return estimatedPodUsed(pod, e.resourceWeights, factors), nil
+	return estimatedPodUsed(pod, factors), nil
 }
 
-func estimatedPodUsed(pod *corev1.Pod, resourceWeights map[corev1.ResourceName]int64, scalingFactors map[corev1.ResourceName]int64) map[corev1.ResourceName]int64 {
+func estimatedPodUsed(pod *corev1.Pod, scalingFactors map[corev1.ResourceName]int64) map[corev1.ResourceName]int64 {
 	requests, limits := resourceapi.PodRequests(pod, resourceapi.PodResourcesOptions{}), resourceapi.PodLimits(pod, resourceapi.PodResourcesOptions{})
 	estimatedUsed := make(map[corev1.ResourceName]int64)
 	priorityClass := extension.GetPodPriorityClassWithDefault(pod)
-	for resourceName := range resourceWeights {
+	for resourceName, factor := range scalingFactors {
 		realResourceName := extension.TranslateResourceNameByPriorityClass(priorityClass, resourceName)
-		estimatedUsed[resourceName] = estimatedUsedByResource(requests, limits, realResourceName, scalingFactors[resourceName])
+		estimatedUsed[resourceName] = estimatedUsedByResource(requests, limits, realResourceName, factor)
 	}
 	return estimatedUsed
 }
diff --git a/pkg/scheduler/plugins/loadaware/helper.go b/pkg/scheduler/plugins/loadaware/helper.go
index 1619cb55..7e65c675 100644
--- a/pkg/scheduler/plugins/loadaware/helper.go
+++ b/pkg/scheduler/plugins/loadaware/helper.go
@@ -17,19 +17,19 @@ limitations under the License.
 package loadaware
 
 import (
+	"sort"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 
 	"github.com/koordinator-sh/koordinator/apis/extension"
 	slov1alpha1 "github.com/koordinator-sh/koordinator/apis/slo/v1alpha1"
-	schedulingconfig "github.com/koordinator-sh/koordinator/pkg/scheduler/apis/config"
-	"github.com/koordinator-sh/koordinator/pkg/util"
+	"github.com/koordinator-sh/koordinator/pkg/scheduler/apis/config"
 )
 
 func isNodeMetricExpired(nodeMetric *slov1alpha1.NodeMetric, nodeMetricExpirationSeconds int64) bool {
@@ -46,102 +46,78 @@ func getNodeMetricReportInterval(nodeMetric *slov1alpha1.NodeMetric) time.Durati
 	return time.Duration(*nodeMetric.Spec.CollectPolicy.ReportIntervalSeconds) * time.Second
 }
 
-func missedLatestUpdateTime(assignedTime, updateTime time.Time) bool {
-	return assignedTime.After(updateTime)
+type usageThresholdsFilterProfile struct {
+	UsageThresholds     ResourceVector
+	ProdUsageThresholds ResourceVector
+	AggregatedUsage     *aggregatedUsageFilterProfile
 }
 
-func stillInTheReportInterval(assignedTime, updateTime time.Time, reportInterval time.Duration) bool {
-	return assignedTime.Before(updateTime) && updateTime.Sub(assignedTime) < reportInterval
+type aggregatedUsageFilterProfile struct {
+	UsageThresholds         ResourceVector
+	UsageAggregationType    extension.AggregationType
+	UsageAggregatedDuration metav1.Duration
 }
 
-func getTargetAggregatedUsage(nodeMetric *slov1alpha1.NodeMetric, aggregatedDuration *metav1.Duration, aggregationType extension.AggregationType) *slov1alpha1.ResourceMap {
-	if nodeMetric.Status.NodeMetric == nil || len(nodeMetric.Status.NodeMetric.AggregatedNodeUsages) == 0 {
-		return nil
+func NewUsageThresholdsFilterProfile(args *config.LoadAwareSchedulingArgs, vectorizer ResourceVectorizer) *usageThresholdsFilterProfile {
+	p := &usageThresholdsFilterProfile{}
+	if len(args.UsageThresholds) > 0 {
+		p.UsageThresholds = vectorizer.ToFactorVec(args.UsageThresholds)
 	}
-
-	// If no specific period is set, the non-empty maximum period recorded by NodeMetrics will be used by default.
-	// This is a default policy.
-	if aggregatedDuration == nil || aggregatedDuration.Duration == 0 {
-		var maxDuration time.Duration
-		var maxIndex int = -1
-		for i, v := range nodeMetric.Status.NodeMetric.AggregatedNodeUsages {
-			if len(v.Usage[aggregationType].ResourceList) > 0 && v.Duration.Duration > maxDuration {
-				maxDuration = v.Duration.Duration
-				maxIndex = i
-			}
-		}
-
-		if maxIndex == -1 {
-			// All values in aggregatedDuration are empty, downgrade to use the values in NodeUsage
-			usage := nodeMetric.Status.NodeMetric.NodeUsage
-			if len(usage.ResourceList) > 0 {
-				return &usage
-			}
-		} else {
-			usage := nodeMetric.Status.NodeMetric.AggregatedNodeUsages[maxIndex].Usage[aggregationType]
-			return &usage
-		}
-	} else if aggregatedDuration != nil {
-		for _, v := range nodeMetric.Status.NodeMetric.AggregatedNodeUsages {
-			if v.Duration.Duration == aggregatedDuration.Duration {
-				usage := v.Usage[aggregationType]
-				if len(usage.ResourceList) > 0 {
-					return &usage
-				}
-			}
+	if len(args.ProdUsageThresholds) > 0 {
+		p.ProdUsageThresholds = vectorizer.ToFactorVec(args.ProdUsageThresholds)
+	}
+	if aggArgs := args.Aggregated; aggArgs != nil && len(aggArgs.UsageThresholds) > 0 && aggArgs.UsageAggregationType != "" {
+		p.AggregatedUsage = &aggregatedUsageFilterProfile{
+			UsageThresholds:         vectorizer.ToFactorVec(args.Aggregated.UsageThresholds),
+			UsageAggregationType:    args.Aggregated.UsageAggregationType,
+			UsageAggregatedDuration: args.Aggregated.UsageAggregatedDuration,
 		}
 	}
-	return nil
-}
-
-func filterWithAggregation(args *schedulingconfig.LoadAwareSchedulingAggregatedArgs) bool {
-	return args != nil && len(args.UsageThresholds) > 0 && args.UsageAggregationType != ""
-}
-
-func scoreWithAggregation(args *schedulingconfig.LoadAwareSchedulingAggregatedArgs) bool {
-	return args != nil && args.ScoreAggregationType != ""
+	return p
 }
 
-type usageThresholdsFilterProfile = extension.CustomUsageThresholds
-
-func generateUsageThresholdsFilterProfile(node *corev1.Node, args *schedulingconfig.LoadAwareSchedulingArgs) *usageThresholdsFilterProfile {
-	usageThresholds, prodUsageThresholds := args.UsageThresholds, args.ProdUsageThresholds
-	customUsageThresholds, err := extension.GetCustomUsageThresholds(node)
+// NOTICE: unknown resource name in custom usage thresholds for vectorizer will be skipped in calculation.
+//
+// Currently, we add all supported resources (cpu, memory) collected by koordlet with hard code
+// that can be used in load aware plugin for compatibility.
+func (tfp *usageThresholdsFilterProfile) generateUsageThresholdsFilterProfile(node *corev1.Node, vectorizer ResourceVectorizer) *usageThresholdsFilterProfile {
+	c, err := extension.GetCustomUsageThresholds(node)
 	if err != nil {
 		klog.V(5).ErrorS(err, "failed to GetCustomUsageThresholds from", "node", node.Name)
-		customUsageThresholds = &extension.CustomUsageThresholds{
-			UsageThresholds:     usageThresholds,
-			ProdUsageThresholds: prodUsageThresholds,
-		}
-		if filterWithAggregation(args.Aggregated) {
-			customUsageThresholds.AggregatedUsage = &extension.CustomAggregatedUsage{
-				UsageThresholds:         args.Aggregated.UsageThresholds,
-				UsageAggregationType:    args.Aggregated.UsageAggregationType,
-				UsageAggregatedDuration: &args.Aggregated.UsageAggregatedDuration,
-			}
-		}
+		return tfp
+	}
+	if c == nil {
+		return tfp
+	}
+	if aggArgs := c.AggregatedUsage; aggArgs != nil && !(len(aggArgs.UsageThresholds) > 0 && aggArgs.UsageAggregationType != "") {
+		c.AggregatedUsage = nil
+	}
+	if len(c.UsageThresholds) == 0 && len(c.ProdUsageThresholds) == 0 && c.AggregatedUsage == nil {
+		return tfp
+	}
+	p := &usageThresholdsFilterProfile{}
+	if len(c.UsageThresholds) == 0 {
+		p.UsageThresholds = tfp.UsageThresholds
 	} else {
-		if len(customUsageThresholds.UsageThresholds) == 0 {
-			customUsageThresholds.UsageThresholds = usageThresholds
-		}
-		if len(customUsageThresholds.ProdUsageThresholds) == 0 {
-			customUsageThresholds.ProdUsageThresholds = prodUsageThresholds
-		}
-		if customUsageThresholds.AggregatedUsage != nil {
-			if len(customUsageThresholds.AggregatedUsage.UsageThresholds) == 0 ||
-				customUsageThresholds.AggregatedUsage.UsageAggregationType == "" {
-				customUsageThresholds.AggregatedUsage = nil
-			}
+		p.UsageThresholds = vectorizer.ToFactorVec(c.UsageThresholds)
+	}
+	if len(c.ProdUsageThresholds) == 0 {
+		p.ProdUsageThresholds = tfp.ProdUsageThresholds
+	} else {
+		p.ProdUsageThresholds = vectorizer.ToFactorVec(c.ProdUsageThresholds)
+	}
+	if c.AggregatedUsage == nil {
+		p.AggregatedUsage = tfp.AggregatedUsage
+	} else {
+		p.AggregatedUsage = &aggregatedUsageFilterProfile{
+			UsageThresholds:      vectorizer.ToFactorVec(c.AggregatedUsage.UsageThresholds),
+			UsageAggregationType: c.AggregatedUsage.UsageAggregationType,
 		}
-		if customUsageThresholds.AggregatedUsage == nil && filterWithAggregation(args.Aggregated) {
-			customUsageThresholds.AggregatedUsage = &extension.CustomAggregatedUsage{
-				UsageThresholds:         args.Aggregated.UsageThresholds,
-				UsageAggregationType:    args.Aggregated.UsageAggregationType,
-				UsageAggregatedDuration: &args.Aggregated.UsageAggregatedDuration,
-			}
+		if d := c.AggregatedUsage.UsageAggregatedDuration; d != nil {
+			p.AggregatedUsage.UsageAggregatedDuration = *d
 		}
 	}
-	return customUsageThresholds
+	return p
 }
 
 func getResourceValue(resourceName corev1.ResourceName, quantity resource.Quantity) int64 {
@@ -151,46 +127,147 @@ func getResourceValue(resourceName corev1.ResourceName, quantity resource.Quanti
 	return quantity.Value()
 }
 
-func buildPodMetricMap(nodeMetric *slov1alpha1.NodeMetric, filterProdPod bool) map[types.NamespacedName]corev1.ResourceList {
-	if len(nodeMetric.Status.PodsMetric) == 0 {
-		return nil
+func getResourceQuantity(resourceName corev1.ResourceName, value int64) resource.Quantity {
+	switch {
+	case resourceName == corev1.ResourceCPU:
+		return *resource.NewMilliQuantity(value, resource.DecimalSI)
+	case resourceName == corev1.ResourceMemory && value%1024 == 0:
+		return *resource.NewQuantity(value, resource.BinarySI)
+	default:
+		return *resource.NewQuantity(value, resource.DecimalSI)
+	}
+}
+
+// isDaemonSetPod returns true if the pod is a IsDaemonSetPod.
+func isDaemonSetPod(ownerRefList []metav1.OwnerReference) bool {
+	for _, ownerRef := range ownerRefList {
+		if ownerRef.Kind == "DaemonSet" {
+			return true
+		}
+	}
+	return false
+}
+
+type ResourceVectorizer []corev1.ResourceName
+type ResourceVector []int64
+
+func NewResourceVectorizer(names ...corev1.ResourceName) ResourceVectorizer {
+	sort.Slice(names, func(i, j int) bool {
+		return names[i] < names[j]
+	})
+	return ResourceVectorizer(names)
+}
+
+// cpu and memory are added by default for custom usage thresholds compatibility.
+func NewResourceVectorizerFromArgs(args *config.LoadAwareSchedulingArgs) ResourceVectorizer {
+	resourceNames := sets.New(corev1.ResourceCPU, corev1.ResourceMemory)
+	resourceNames = resourceNames.Union(sets.KeySet(args.UsageThresholds))
+	resourceNames = resourceNames.Union(sets.KeySet(args.ProdUsageThresholds))
+	if aggArgs := args.Aggregated; aggArgs != nil {
+		resourceNames = resourceNames.Union(sets.KeySet(aggArgs.UsageThresholds))
 	}
-	podMetrics := make(map[types.NamespacedName]corev1.ResourceList)
-	for _, podMetric := range nodeMetric.Status.PodsMetric {
-		if filterProdPod && podMetric.Priority != extension.PriorityProd {
-			continue
+	resourceNames = resourceNames.Union(sets.KeySet(args.ResourceWeights))
+	resourceNames = resourceNames.Union(sets.KeySet(args.EstimatedScalingFactors))
+	resourceNames.Insert(args.SupportedResources...)
+	return NewResourceVectorizer(resourceNames.UnsortedList()...)
+}
+
+// NOTICE: unknown resource name will be ignored in vectorization
+func (rv ResourceVectorizer) ToVec(list corev1.ResourceList) ResourceVector {
+	vec := make(ResourceVector, len(rv))
+	for i, name := range rv {
+		if q, ok := list[name]; ok {
+			vec[i] = getResourceValue(name, q)
 		}
-		name := types.NamespacedName{
-			Namespace: podMetric.Namespace,
-			Name:      podMetric.Name,
+	}
+	return vec
+}
+
+// NOTICE: unknown resource name will be ignored in vectorization
+func (rv ResourceVectorizer) ToFactorVec(list map[corev1.ResourceName]int64) ResourceVector {
+	vec := make(ResourceVector, len(rv))
+	for i, name := range rv {
+		vec[i] = list[name]
+	}
+	return vec
+}
+
+func (rv ResourceVectorizer) ToList(vec ResourceVector) corev1.ResourceList {
+	list := make(corev1.ResourceList, len(rv))
+	for i, name := range rv {
+		list[name] = getResourceQuantity(name, vec[i])
+	}
+	return list
+}
+
+func (rv ResourceVectorizer) ToFactorList(vec ResourceVector) map[corev1.ResourceName]int64 {
+	list := make(map[corev1.ResourceName]int64, len(rv))
+	for i, name := range rv {
+		list[name] = vec[i]
+	}
+	return list
+}
+
+func (rv ResourceVectorizer) EmptyVec() ResourceVector {
+	return make(ResourceVector, len(rv))
+}
+
+func (v ResourceVector) Empty() bool {
+	for i := range v {
+		if v[i] != 0 {
+			return false
 		}
-		podMetrics[name] = podMetric.PodUsage.ResourceList
 	}
-	return podMetrics
+	return true
 }
 
-func sumPodUsages(podMetrics map[types.NamespacedName]corev1.ResourceList, estimatedPods sets.Set[types.NamespacedName]) (podUsages, estimatedPodsUsages corev1.ResourceList) {
-	if len(podMetrics) == 0 {
-		return nil, nil
+func (v ResourceVector) Add(y ResourceVector) {
+	for i := range v {
+		v[i] += y[i]
 	}
-	podUsages = make(corev1.ResourceList)
-	estimatedPodsUsages = make(corev1.ResourceList)
-	for podName, usage := range podMetrics {
-		if estimatedPods.Has(podName) {
-			util.AddResourceList(estimatedPodsUsages, usage)
-			continue
+}
+
+// v = v + max(0, x - y)
+func (v ResourceVector) AddDelta(x, y ResourceVector) (changed bool) {
+	for i, value := range x {
+		if y != nil {
+			value -= y[i]
+		}
+		if value > 0 {
+			v[i] += value
+			changed = true
 		}
-		util.AddResourceList(podUsages, usage)
 	}
-	return podUsages, estimatedPodsUsages
+	return
 }
 
-// isDaemonSetPod returns true if the pod is a IsDaemonSetPod.
-func isDaemonSetPod(ownerRefList []metav1.OwnerReference) bool {
-	for _, ownerRef := range ownerRefList {
-		if ownerRef.Kind == "DaemonSet" {
-			return true
+func (v ResourceVector) Sub(y ResourceVector) {
+	for i := range v {
+		v[i] -= y[i]
+	}
+}
+
+// v = v - max(0, x - y)
+func (v ResourceVector) SubDelta(x, y ResourceVector) (changed bool) {
+	for i, value := range x {
+		if y != nil {
+			value -= y[i]
+		}
+		if value > 0 {
+			v[i] -= value
+			changed = true
 		}
 	}
-	return false
+	return
+}
+
+func (v ResourceVector) Clone() framework.StateData {
+	if v == nil {
+		return nil
+	}
+	copy := make(ResourceVector, len(v))
+	for i := range v {
+		copy[i] = v[i]
+	}
+	return copy
 }
diff --git a/pkg/scheduler/plugins/loadaware/load_aware.go b/pkg/scheduler/plugins/loadaware/load_aware.go
index ed471e32..557ab935 100644
--- a/pkg/scheduler/plugins/loadaware/load_aware.go
+++ b/pkg/scheduler/plugins/loadaware/load_aware.go
@@ -24,15 +24,13 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 
 	"github.com/koordinator-sh/koordinator/apis/extension"
 	slov1alpha1 "github.com/koordinator-sh/koordinator/apis/slo/v1alpha1"
-	slolisters "github.com/koordinator-sh/koordinator/pkg/client/listers/slo/v1alpha1"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/apis/config"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/apis/config/validation"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/frameworkext"
@@ -41,7 +39,12 @@ import (
 )
 
 const (
-	Name                                    = "LoadAwareScheduling"
+	Name = "LoadAwareScheduling"
+
+	// incomingPodEstimatedStateKey is the key in CycleState to LoadAwareScheduling estimated resources for incoming pod.
+	// Using the name of the plugin will likely help us avoid collisions with other plugins.
+	incomingPodEstimatedStateKey = "IncomingPodEstimated" + Name
+
 	ErrReasonNodeMetricExpired              = "node(s) nodeMetric expired"
 	ErrReasonUsageExceedThreshold           = "node(s) %s usage exceed threshold"
 	ErrReasonAggregatedUsageExceedThreshold = "node(s) %s aggregated usage exceed threshold"
@@ -60,17 +63,20 @@ const (
 var (
 	_ framework.EnqueueExtensions = &Plugin{}
 
-	_ framework.FilterPlugin  = &Plugin{}
-	_ framework.ScorePlugin   = &Plugin{}
-	_ framework.ReservePlugin = &Plugin{}
+	_ framework.PreFilterPlugin = &Plugin{}
+	_ framework.FilterPlugin    = &Plugin{}
+	_ framework.ScorePlugin     = &Plugin{}
+	_ framework.ReservePlugin   = &Plugin{}
 )
 
 type Plugin struct {
-	handle           framework.Handle
-	args             *config.LoadAwareSchedulingArgs
-	nodeMetricLister slolisters.NodeMetricLister
-	estimator        estimator.Estimator
-	podAssignCache   *podAssignCache
+	handle         framework.Handle
+	args           *config.LoadAwareSchedulingArgs
+	vectorizer     ResourceVectorizer
+	filterProfile  *usageThresholdsFilterProfile
+	scoreWeights   ResourceVector
+	estimator      estimator.Estimator
+	podAssignCache *podAssignCache
 }
 
 func New(args runtime.Object, handle framework.Handle) (framework.Plugin, error) {
@@ -92,20 +98,29 @@ func New(args runtime.Object, handle framework.Handle) (framework.Plugin, error)
 	if err != nil {
 		return nil, err
 	}
-	assignCache := newPodAssignCache(estimator, pluginArgs)
+
+	vectorizer := NewResourceVectorizerFromArgs(pluginArgs)
+	assignCache := newPodAssignCache(estimator, vectorizer, pluginArgs)
 	podInformer := frameworkExtender.SharedInformerFactory().Core().V1().Pods()
 	frameworkexthelper.ForceSyncFromInformer(context.TODO().Done(), frameworkExtender.SharedInformerFactory(), podInformer.Informer(), assignCache)
 	frameworkExtender.RegisterForgetPodHandler(func(pod *corev1.Pod) {
 		assignCache.unAssign(pod.Spec.NodeName, pod)
 	})
-	nodeMetricLister := frameworkExtender.KoordinatorSharedInformerFactory().Slo().V1alpha1().NodeMetrics().Lister()
-
+	koordInformers := frameworkExtender.KoordinatorSharedInformerFactory()
+	nodeMetricInformer := koordInformers.Slo().V1alpha1().NodeMetrics()
+	frameworkexthelper.ForceSyncFromInformer(context.TODO().Done(), koordInformers, nodeMetricInformer.Informer(), assignCache.NodeMetricHandler())
+	scoreWeights := vectorizer.ToFactorVec(pluginArgs.ResourceWeights)
+	if scoreWeights.Empty() {
+		scoreWeights = nil
+	}
 	return &Plugin{
-		handle:           handle,
-		args:             pluginArgs,
-		nodeMetricLister: nodeMetricLister,
-		estimator:        estimator,
-		podAssignCache:   assignCache,
+		handle:         handle,
+		args:           pluginArgs,
+		vectorizer:     vectorizer,
+		filterProfile:  NewUsageThresholdsFilterProfile(pluginArgs, vectorizer),
+		scoreWeights:   scoreWeights,
+		estimator:      estimator,
+		podAssignCache: assignCache,
 	}, nil
 }
 
@@ -121,6 +136,17 @@ func (p *Plugin) EventsToRegister() []framework.ClusterEventWithHint {
 	}
 }
 
+func (p *Plugin) PreFilter(ctx context.Context, state *framework.CycleState, pod *corev1.Pod) (*framework.PreFilterResult, *framework.Status) {
+	// add estimated resources for incoming pod to cache
+	_ = p.addEstimatedOfIncoming(p.vectorizer.EmptyVec(), state, pod)
+	return nil, nil
+}
+
+// PreFilterExtensions returns a PreFilterExtensions interface if the plugin implements one.
+func (p *Plugin) PreFilterExtensions() framework.PreFilterExtensions {
+	return nil
+}
+
 func (p *Plugin) Filter(ctx context.Context, state *framework.CycleState, pod *corev1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
 	node := nodeInfo.Node()
 	if node == nil {
@@ -131,7 +157,37 @@ func (p *Plugin) Filter(ctx context.Context, state *framework.CycleState, pod *c
 		return nil
 	}
 
-	nodeMetric, err := p.nodeMetricLister.Get(node.Name)
+	filterProfile := p.filterProfile.generateUsageThresholdsFilterProfile(node, p.vectorizer)
+	prodPod := !filterProfile.ProdUsageThresholds.Empty() && extension.GetPodPriorityClassWithDefault(pod) == extension.PriorityProd
+	var usageThresholds ResourceVector
+	isAgg := false
+	if prodPod {
+		usageThresholds = filterProfile.ProdUsageThresholds
+	} else if agg := filterProfile.AggregatedUsage; agg != nil {
+		usageThresholds = agg.UsageThresholds
+		isAgg = true
+	} else {
+		usageThresholds = filterProfile.UsageThresholds
+	}
+	if usageThresholds.Empty() {
+		return nil // skip if filter thresholds are disabled
+	}
+
+	allocatableList, err := p.estimator.EstimateNode(node)
+	if err != nil {
+		klog.ErrorS(err, "Estimated node allocatable failed!", "node", node.Name)
+		return nil
+	}
+	allocatable := p.vectorizer.ToVec(allocatableList)
+
+	var aggDuration metav1.Duration
+	var aggType extension.AggregationType
+	if !prodPod && isAgg {
+		agg := filterProfile.AggregatedUsage
+		aggDuration, aggType = agg.UsageAggregatedDuration, agg.UsageAggregationType
+	}
+
+	nodeMetric, estimated, estimatedPods, err := p.podAssignCache.GetNodeMetricAndEstimatedOfExisting(node.Name, prodPod, aggDuration, aggType, klog.V(6).Enabled())
 	if err != nil {
 		// For nodes that lack load information, fall back to the situation where there is no load-aware scheduling.
 		// Some nodes in the cluster do not install the koordlet, but users newly created Pod use koord-scheduler to schedule,
@@ -141,7 +197,6 @@ func (p *Plugin) Filter(ctx context.Context, state *framework.CycleState, pod *c
 		}
 		return framework.NewStatus(framework.Error, err.Error())
 	}
-
 	if p.args.FilterExpiredNodeMetrics != nil && *p.args.FilterExpiredNodeMetrics &&
 		p.args.NodeMetricExpirationSeconds != nil && isNodeMetricExpired(nodeMetric, *p.args.NodeMetricExpirationSeconds) {
 		if p.args.EnableScheduleWhenNodeMetricsExpired != nil && !*p.args.EnableScheduleWhenNodeMetricsExpired {
@@ -154,37 +209,14 @@ func (p *Plugin) Filter(ctx context.Context, state *framework.CycleState, pod *c
 		return nil
 	}
 
-	allocatable, err := p.estimator.EstimateNode(node)
-	if err != nil {
-		klog.ErrorS(err, "Estimated node allocatable failed!", "node", node.Name)
-		return nil
-	}
-	filterProfile := generateUsageThresholdsFilterProfile(node, p.args)
-	prodPod := len(filterProfile.ProdUsageThresholds) > 0 && extension.GetPodPriorityClassWithDefault(pod) == extension.PriorityProd
-
-	var nodeUsage *slov1alpha1.ResourceMap
-	var usageThresholds map[corev1.ResourceName]int64
-	if prodPod {
-		usageThresholds = filterProfile.ProdUsageThresholds
-	} else {
-		if filterProfile.AggregatedUsage != nil {
-			nodeUsage = getTargetAggregatedUsage(
-				nodeMetric,
-				filterProfile.AggregatedUsage.UsageAggregatedDuration,
-				filterProfile.AggregatedUsage.UsageAggregationType,
-			)
-			usageThresholds = filterProfile.AggregatedUsage.UsageThresholds
-		} else {
-			nodeUsage = &nodeMetric.Status.NodeMetric.NodeUsage
-			usageThresholds = filterProfile.UsageThresholds
-		}
+	if err = p.addEstimatedOfIncoming(estimated, state, pod); err != nil {
+		klog.ErrorS(err, "Failed to estimate incoming pod usage", "pod", klog.KObj(pod))
+	} else if klog.V(6).Enabled() {
+		klog.InfoS("Estimate node usage for filtering", "pod", klog.KObj(pod), "node", nodeMetric.Name,
+			"estimated", klog.Format(p.vectorizer.ToList(estimated)),
+			"estimatedExistingPods", klog.KObjSlice(estimatedPods))
 	}
-	estimatedUsed, err := p.GetEstimatedUsed(node.Name, nodeMetric, pod, nodeUsage, prodPod)
-	if err != nil {
-		klog.ErrorS(err, "GetEstimatedUsed failed!", "node", node.Name)
-		return nil
-	}
-	return filterNodeUsage(node.Name, pod, usageThresholds, estimatedUsed, allocatable, prodPod, filterProfile)
+	return p.filterNodeUsage(node.Name, pod, usageThresholds, estimated, allocatable, isAgg)
 }
 
 func (p *Plugin) ScoreExtensions() framework.ScoreExtensions {
@@ -201,6 +233,10 @@ func (p *Plugin) Unreserve(ctx context.Context, state *framework.CycleState, pod
 }
 
 func (p *Plugin) Score(ctx context.Context, state *framework.CycleState, pod *corev1.Pod, nodeName string) (int64, *framework.Status) {
+	if p.scoreWeights == nil {
+		return 0, nil // skip if score weights are disabled
+	}
+
 	nodeInfo, err := p.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
 	if err != nil {
 		return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
@@ -209,7 +245,22 @@ func (p *Plugin) Score(ctx context.Context, state *framework.CycleState, pod *co
 	if node == nil {
 		return 0, framework.NewStatus(framework.Error, "node not found")
 	}
-	nodeMetric, err := p.nodeMetricLister.Get(nodeName)
+
+	allocatableList, err := p.estimator.EstimateNode(node)
+	if err != nil {
+		klog.ErrorS(err, "Estimated node allocatable failed!", "node", node.Name)
+		return 0, nil
+	}
+	allocatable := p.vectorizer.ToVec(allocatableList)
+
+	prodPod := p.args.ScoreAccordingProdUsage && extension.GetPodPriorityClassWithDefault(pod) == extension.PriorityProd
+	var aggDuration metav1.Duration
+	var aggType extension.AggregationType
+	if agg := p.args.Aggregated; !prodPod && agg != nil && agg.ScoreAggregationType != "" {
+		aggDuration, aggType = agg.ScoreAggregatedDuration, agg.ScoreAggregationType
+	}
+
+	nodeMetric, estimated, estimatedPods, err := p.podAssignCache.GetNodeMetricAndEstimatedOfExisting(nodeName, prodPod, aggDuration, aggType, klog.V(6).Enabled())
 	if err != nil {
 		// caused by load-aware scheduling itself is an optimization,
 		// so we should skip the node and score the node 0
@@ -226,146 +277,81 @@ func (p *Plugin) Score(ctx context.Context, state *framework.CycleState, pod *co
 		return 0, nil
 	}
 
-	prodPod := extension.GetPodPriorityClassWithDefault(pod) == extension.PriorityProd && p.args.ScoreAccordingProdUsage
-	var nodeUsage *slov1alpha1.ResourceMap
-	if !prodPod {
-		if scoreWithAggregation(p.args.Aggregated) {
-			nodeUsage = getTargetAggregatedUsage(nodeMetric, &p.args.Aggregated.ScoreAggregatedDuration, p.args.Aggregated.ScoreAggregationType)
-		} else {
-			nodeUsage = &nodeMetric.Status.NodeMetric.NodeUsage
-		}
-	}
-	estimatedUsed, err := p.GetEstimatedUsed(nodeName, nodeMetric, pod, nodeUsage, prodPod)
-	if err != nil {
-		klog.ErrorS(err, "GetEstimatedUsed failed!", "node", node.Name)
+	if err = p.addEstimatedOfIncoming(estimated, state, pod); err != nil {
+		klog.ErrorS(err, "Failed to estimate incoming pod usage", "pod", klog.KObj(pod))
 		return 0, nil
 	}
-
-	allocatable, err := p.estimator.EstimateNode(node)
-	if err != nil {
-		klog.ErrorS(err, "Estimated node allocatable failed!", "node", node.Name)
-		return 0, nil
+	if klog.V(6).Enabled() {
+		klog.InfoS("Estimate node usage for scoring", "pod", klog.KObj(pod), "node", nodeMetric.Name,
+			"estimated", klog.Format(p.vectorizer.ToList(estimated)),
+			"estimatedExistingPods", klog.KObjSlice(estimatedPods))
 	}
-	score := loadAwareSchedulingScorer(p.args.ResourceWeights, estimatedUsed, allocatable)
+	score := loadAwareSchedulingScorer(p.scoreWeights, estimated, allocatable)
 	return score, nil
 }
 
-func (p *Plugin) GetEstimatedUsed(nodeName string, nodeMetric *slov1alpha1.NodeMetric, pod *corev1.Pod, nodeUsage *slov1alpha1.ResourceMap, prodPod bool) (map[corev1.ResourceName]int64, error) {
-	if nodeMetric == nil {
-		return nil, nil
+// try to use state cache before EstimatePod
+func (p *Plugin) addEstimatedOfIncoming(estimated ResourceVector, cycleState *framework.CycleState, pod *corev1.Pod) error {
+	var podEstimated ResourceVector
+	if c, err := cycleState.Read(incomingPodEstimatedStateKey); err == nil {
+		podEstimated, _ = c.(ResourceVector)
 	}
-	podMetrics := buildPodMetricMap(nodeMetric, prodPod)
-
-	estimatedUsed, err := p.estimator.EstimatePod(pod)
-	if err != nil {
-		return nil, err
-	}
-	assignedPodEstimatedUsed, estimatedPods := p.estimatedAssignedPodUsed(nodeName, nodeMetric, podMetrics, prodPod)
-	for resourceName, value := range assignedPodEstimatedUsed {
-		estimatedUsed[resourceName] += value
-	}
-	podActualUsages, estimatedPodActualUsages := sumPodUsages(podMetrics, estimatedPods)
-	if prodPod {
-		for resourceName, quantity := range podActualUsages {
-			estimatedUsed[resourceName] += getResourceValue(resourceName, quantity)
-		}
-	} else {
-		if nodeMetric.Status.NodeMetric != nil {
-			if nodeUsage != nil {
-				for resourceName, quantity := range nodeUsage.ResourceList {
-					if q := estimatedPodActualUsages[resourceName]; !q.IsZero() {
-						quantity = quantity.DeepCopy()
-						if quantity.Cmp(q) >= 0 {
-							quantity.Sub(q)
-						}
-					}
-					estimatedUsed[resourceName] += getResourceValue(resourceName, quantity)
-				}
-			}
+	if podEstimated == nil {
+		list, err := p.estimator.EstimatePod(pod)
+		if err != nil {
+			// use len=0 but not empty vector to indicate error occurred
+			cycleState.Write(incomingPodEstimatedStateKey, ResourceVector{})
+			return err
 		}
+		podEstimated = p.vectorizer.ToFactorVec(list)
+		cycleState.Write(incomingPodEstimatedStateKey, podEstimated)
+	} else if len(podEstimated) == 0 {
+		return fmt.Errorf("error occurred in former estimation from cycleState with key %q", incomingPodEstimatedStateKey)
 	}
-	klog.V(6).Infof("GetEstimatedUsed: node %s, pod %s, estimatedUsed %+v, assignedPodEstimatedUsed %+v, estimatedPods: %+v",
-		nodeName, klog.KObj(pod), estimatedUsed, assignedPodEstimatedUsed, estimatedPods)
-	return estimatedUsed, nil
+	estimated.Add(podEstimated)
+	return nil
 }
 
-func filterNodeUsage(nodeName string, pod *corev1.Pod, usageThresholds, estimatedUsed map[corev1.ResourceName]int64, allocatable corev1.ResourceList, prodPod bool, filterProfile *usageThresholdsFilterProfile) *framework.Status {
-	for resourceName, value := range usageThresholds {
+func (p *Plugin) filterNodeUsage(nodeName string, pod *corev1.Pod, usageThresholds, estimatedUsed, allocatable ResourceVector, isAgg bool) *framework.Status {
+	for i, value := range usageThresholds {
 		if value == 0 {
 			continue
 		}
-		total := getResourceValue(resourceName, allocatable[resourceName])
+		total := allocatable[i]
 		if total == 0 {
 			continue
 		}
-		usage := int64(math.Round(float64(estimatedUsed[resourceName]) / float64(total) * 100))
+		estimated := estimatedUsed[i]
+		usage := int64(math.Round(float64(estimated) / float64(total) * 100))
 		if usage <= value {
 			continue
 		}
 
 		reason := ErrReasonUsageExceedThreshold
-		if !prodPod && filterProfile.AggregatedUsage != nil {
+		if isAgg {
 			reason = ErrReasonAggregatedUsageExceedThreshold
 		}
-		klog.V(5).InfoS("failed to filter node usage for pod", "pod", klog.KObj(pod), "node", nodeName,
-			"resource", resourceName, "total", total, "usage", usage, "threshold", value)
+		resourceName := p.vectorizer[i]
+		if klog.V(5).Enabled() {
+			klog.InfoS("Node is unschedulable since usage exceeds threshold", "pod", klog.KObj(pod), "node", nodeName,
+				"resource", resourceName, "usage", usage, "threshold", value,
+				"estimated", getResourceQuantity(resourceName, estimated),
+				"total", getResourceQuantity(resourceName, total))
+		}
 		return framework.NewStatus(framework.Unschedulable, fmt.Sprintf(reason, resourceName))
 	}
 	return nil
 }
 
-func (p *Plugin) estimatedAssignedPodUsed(nodeName string, nodeMetric *slov1alpha1.NodeMetric, podMetrics map[types.NamespacedName]corev1.ResourceList, filterProdPod bool) (map[corev1.ResourceName]int64, sets.Set[types.NamespacedName]) {
-	estimatedUsed := make(map[corev1.ResourceName]int64)
-	estimatedPods := make(sets.Set[types.NamespacedName])
-	var nodeMetricUpdateTime time.Time
-	if nodeMetric.Status.UpdateTime != nil {
-		nodeMetricUpdateTime = nodeMetric.Status.UpdateTime.Time
-	}
-	nodeMetricReportInterval := getNodeMetricReportInterval(nodeMetric)
-
-	assignedPodsOnNode := p.podAssignCache.getPodsAssignInfoOnNode(nodeName)
-	now := time.Now()
-	for _, assignInfo := range assignedPodsOnNode {
-		if filterProdPod && extension.GetPodPriorityClassWithDefault(assignInfo.pod) != extension.PriorityProd {
-			continue
-		}
-		podName := types.NamespacedName{
-			Namespace: assignInfo.pod.Namespace,
-			Name:      assignInfo.pod.Name,
-		}
-		podUsage := podMetrics[podName]
-		if len(podUsage) == 0 ||
-			missedLatestUpdateTime(assignInfo.timestamp, nodeMetricUpdateTime) ||
-			stillInTheReportInterval(assignInfo.timestamp, nodeMetricUpdateTime, nodeMetricReportInterval) ||
-			(scoreWithAggregation(p.args.Aggregated) &&
-				getTargetAggregatedUsage(nodeMetric, &p.args.Aggregated.ScoreAggregatedDuration, p.args.Aggregated.ScoreAggregationType) == nil) ||
-			(!assignInfo.estimatedDeadline.IsZero() && assignInfo.estimatedDeadline.After(now)) {
-			estimated := assignInfo.estimated
-			if estimated == nil {
-				continue
-			}
-			for resourceName, value := range estimated {
-				if quantity, ok := podUsage[resourceName]; ok {
-					usage := getResourceValue(resourceName, quantity)
-					if usage > value {
-						value = usage
-					}
-				}
-				estimatedUsed[resourceName] += value
-			}
-			estimatedPods.Insert(podName)
-		}
-	}
-	return estimatedUsed, estimatedPods
-}
-
-func loadAwareSchedulingScorer(resToWeightMap, used map[corev1.ResourceName]int64, allocatable corev1.ResourceList) int64 {
+func loadAwareSchedulingScorer(resToWeightMap, used, allocatable ResourceVector) int64 {
 	var nodeScore, weightSum int64
-	for resourceName, weight := range resToWeightMap {
-		resourceScore := leastUsedScore(used[resourceName], getResourceValue(resourceName, allocatable[resourceName]))
-		nodeScore += resourceScore * weight
+	for i, weight := range resToWeightMap {
+		nodeScore += leastUsedScore(used[i], allocatable[i]) * weight
 		weightSum += weight
 	}
+	if weightSum <= 0 {
+		return 0
+	}
 	return nodeScore / weightSum
 }
 
diff --git a/pkg/scheduler/plugins/loadaware/pod_assign_cache.go b/pkg/scheduler/plugins/loadaware/pod_assign_cache.go
index e7397d79..01196fcb 100644
--- a/pkg/scheduler/plugins/loadaware/pod_assign_cache.go
+++ b/pkg/scheduler/plugins/loadaware/pod_assign_cache.go
@@ -22,82 +22,278 @@ import (
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
+	"k8s.io/utils/clock"
 
 	"github.com/koordinator-sh/koordinator/apis/extension"
+	slov1alpha1 "github.com/koordinator-sh/koordinator/apis/slo/v1alpha1"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/apis/config"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/plugins/loadaware/estimator"
 	"github.com/koordinator-sh/koordinator/pkg/util"
 )
 
-var (
-	timeNowFn = time.Now
-)
-
-// podAssignCache stores the Pod information that has been successfully scheduled or is about to be bound
+// podAssignCache stores the NodeMetric and Pod information that has been successfully scheduled or is about to be bound.
+//
+// The cache must handle multiple goroutines' object delta events concurrently.
+//  1. When NodeMetric or Pod add or update event on a new node is received, a new nodeInfo should be created and stored in the cache.
+//  2. When NodeMetric or Pod event on an existing node in cache is received, the corresponding nodeInfo should be updated.
+//  3. When NodeMetric and Pod on a node are all deleted, the corresponding nodeInfo will be empty and should be deleted.
+//
+// # Implementation
+//
+// We implement the cache with nodeInfo items indexed by nodeName and maintain them in following rules:
+//  1. To delete empty nodeInfo from cache correctly, handlers should hold both the cache lock and this nodeInfo lock when deleting.
+//  2. To minimize event handlers' lock occupation on the entire cache, especially write lock, we choose to lock the nodeInfo
+//     first and the cache then for deletion. We don't need to hold read lock the cache once for update,
+//     release it and write lock the cache again for deletion if events makes nodeInfo empty.
+//  3. To prevent deadlock risk with Rule 2, we MUST not lock any nodeInfo when holding cache lock already,
+//     which means we have to unlock the cache before locking the nodeInfo.
+//  4. Goroutine might get a write locked nodeInfo from cache because of Rule 3, which another goroutine is adding
+//     or deleting object on it. When it grabs the lock, this nodeInfo might be empty and removed from cache.
+//     We add a deleted flag in nodeInfo, which indicates this case. For adding or updating object event, we MUST not use
+//     the deleted nodeInfo, instead, we should retry to get or create a new one from cache to prevent object event missing.
+//     For deleting object event or reading nodeInfo data, we can ignore this flag.
+//  5. When add or update object, we will try to get a loaded nodeInfo from cache, if not exist, we will create and store a new one.
+//     When new nodeInfo is stored, it should be locked already to avoid race conditions on reading brought by Rule 3.
+//
+// # Concurrent Scenarios
+//
+// Adding on existing nodeInfo
+//   - Alice: <get nodeInfo> - (nodeInfo Lock) - (nodeInfo Update) - (nodeInfo Unlock)
+//   - Bob: <get nodeInfo> - (nodeInfo Lock after Alice Unlock) - (nodeInfo Update) - (nodeInfo Unlock)
+//
+// Adding for not existing nodeInfo
+//   - Alice: <create locked nodeInfo, store, return> - (nodeInfo Update) - (nodeInfo Unlock)
+//   - Bob: <get nodeInfo after Alice Unlock cache> - (nodeInfo Lock after Alice Unlock) - (nodeInfo Update) - (nodeInfo Unlock)
+//
+// Deleting and adding nodeInfo
+//   - Alice: <get nodeInfo> - (nodeInfo Lock) - (nodeInfo Update) - (nodeInfo tag deleted) - <delete nodeInfo> - (nodeInfo Unlock)
+//   - Bob: <get nodeInfo when cache is not locked> - (nodeInfo Lock after Alice Unlock) - (nodeInfo find deleted) - (nodeInfo Unlock) - return failed and retry adding
+//
+// Reading and writing on existing nodeInfo
+//   - Alice: <get nodeInfo> - (nodeInfo Lock) - (nodeInfo Update) - (nodeInfo Unlock)
+//   - Bob: <get nodeInfo> - (nodeInfo RLock after Alice Unlock)- (nodeInfo Read) - (nodeInfo RUnlock)
+//
+// Reading and writing on not existing nodeInfo
+//   - Alice: <create locked nodeInfo, store, return> - (nodeInfo Update) - (nodeInfo Unlock)
+//   - Bob: <get nodeInfo after Alice Unlock cache> - (nodeInfo RLock after Alice Unlock)- (nodeInfo Read) - (nodeInfo RUnlock)
+//
+// <...> represents cache is locked before action and unlocked after action
 type podAssignCache struct {
-	lock sync.RWMutex
-	// podInfoItems stores podAssignInfo according to each node.
+	items      sync.Map // items stores nodeInfo, using sync.Map because nodes are not frequently added or deleted.
+	estimator  estimator.Estimator
+	vectorizer ResourceVectorizer
+	clock      clock.Clock
+	args       *config.LoadAwareSchedulingArgs
+}
+
+// nodeInfo stores
+//  1. assigned pods on this node
+//  2. this node's metric collected by koordlet
+//  3. cached calculation result for node usage and estimation.
+type nodeInfo struct {
+	sync.RWMutex
+	// nodeInfo is marked as deleted when podInfos and nodeMetric is empty.
+	// Deleted info should not be updated anymore.
+	deleted bool
 	// podAssignInfo is indexed using the Pod's types.UID
-	podInfoItems map[string]map[types.UID]*podAssignInfo
-	estimator    estimator.Estimator
-	args         *config.LoadAwareSchedulingArgs
+	podInfos map[types.UID]*podAssignInfo
+
+	nodeMetric     *slov1alpha1.NodeMetric
+	updateTime     time.Time
+	reportInterval time.Duration
+
+	podUsages     map[NamespacedName]ResourceVector
+	prodPods      sets.Set[NamespacedName]
+	nodeUsage     ResourceVector
+	prodUsage     ResourceVector
+	aggUsages     map[aggUsageKey]ResourceVector
+	nodeDelta     ResourceVector // delta estimated resources of existing pods
+	prodDelta     ResourceVector // delta estimated resources of existing prod pods
+	nodeEstimated ResourceVector // sum of full estimated resources of all existing pods
+
+	nodeDeltaPods     sets.Set[NamespacedName] // pods that is part of delta estimated, used in logging only
+	prodDeltaPods     sets.Set[NamespacedName] // pods that is part of delta estimated for prod, used in logging only
+	nodeEstimatedPods sets.Set[NamespacedName] // pods that is part of full estimated, used in logging only
 }
 
 type podAssignInfo struct {
 	timestamp         time.Time
 	pod               *corev1.Pod
-	estimated         map[corev1.ResourceName]int64
+	estimated         ResourceVector
 	estimatedDeadline time.Time
 }
 
-func newPodAssignCache(estimator estimator.Estimator, args *config.LoadAwareSchedulingArgs) *podAssignCache {
+// implements klog.KMetadata
+type NamespacedName struct {
+	Namespace string
+	Name      string
+}
+
+func (n NamespacedName) GetName() string {
+	return n.Name
+}
+
+func (n NamespacedName) GetNamespace() string {
+	return n.Namespace
+}
+
+type aggUsageKey struct {
+	Type extension.AggregationType
+	// aggDuration == 0 means the non-empty maximum period recorded
+	Duration time.Duration
+}
+
+func newPodAssignCache(estimator estimator.Estimator, vectorizer ResourceVectorizer, args *config.LoadAwareSchedulingArgs) *podAssignCache {
 	return &podAssignCache{
-		podInfoItems: map[string]map[types.UID]*podAssignInfo{},
-		estimator:    estimator,
-		args:         args,
+		estimator:  estimator,
+		vectorizer: vectorizer,
+		clock:      clock.RealClock{},
+		args:       args,
 	}
 }
 
+func (p *podAssignCache) GetNodeMetricAndEstimatedOfExisting(name string, prodPod bool,
+	aggregatedDuration metav1.Duration, aggregationType extension.AggregationType, logEnabled bool) (
+	nodeMetric *slov1alpha1.NodeMetric, estimated ResourceVector, estimatedPods []NamespacedName, _ error) {
+	n, exists := p.getNodeInfo(name)
+	if !exists || n == nil {
+		return nil, nil, nil, errors.NewNotFound(slov1alpha1.Resource("nodemetric"), name)
+	}
+	n.RLock()
+	defer n.RUnlock()
+	if nodeMetric = n.nodeMetric; nodeMetric == nil {
+		return nil, nil, nil, errors.NewNotFound(slov1alpha1.Resource("nodemetric"), name)
+	}
+	estimated = p.vectorizer.EmptyVec()
+	var pods sets.Set[NamespacedName]
+	if prodPod {
+		estimated.Add(n.prodUsage)
+		estimated.Add(n.prodDelta)
+		pods = n.prodDeltaPods
+	} else {
+		var nodeUsage ResourceVector
+		if aggregationType != "" {
+			nodeUsage = n.getTargetAggregatedUsage(aggregatedDuration, aggregationType)
+		} else {
+			nodeUsage = n.nodeUsage
+		}
+		if nodeUsage != nil {
+			estimated.Add(nodeUsage)
+			estimated.Add(n.nodeDelta)
+			pods = n.nodeDeltaPods
+		} else {
+			estimated.Add(n.nodeEstimated)
+			pods = n.nodeEstimatedPods
+		}
+	}
+	if logEnabled {
+		estimatedPods = pods.UnsortedList()
+	}
+	return
+}
+func (n *nodeInfo) getTargetAggregatedUsage(aggregatedDuration metav1.Duration, aggregationType extension.AggregationType) ResourceVector {
+	// If no specific period is set, the non-empty maximum period recorded by NodeMetrics will be used by default.
+	// This is a default policy.
+	d := aggregatedDuration.Duration
+	vec := n.aggUsages[aggUsageKey{Type: aggregationType, Duration: d}]
+	if vec == nil && d == 0 {
+		// All values in aggregatedDuration are empty, downgrade to use the values in NodeUsage
+		vec = n.nodeUsage
+	}
+	return vec
+}
 func (p *podAssignCache) getPodAssignInfo(nodeName string, pod *corev1.Pod) *podAssignInfo {
 	if nodeName == "" {
 		return nil
 	}
-	p.lock.RLock()
-	defer p.lock.RUnlock()
-	m := p.podInfoItems[nodeName]
-	if m == nil {
+	n, exists := p.getNodeInfo(nodeName)
+	if !exists || n == nil {
 		return nil
 	}
-	if info, ok := m[pod.UID]; ok {
-		return info
+	n.RLock()
+	defer n.RUnlock()
+	return n.podInfos[pod.UID]
+}
+
+func (p *podAssignCache) getClonedNodeInfo(nodeName string) *nodeInfo {
+	ret := &nodeInfo{}
+	if nodeName == "" {
+		return ret
+	}
+	n, exists := p.getNodeInfo(nodeName)
+	if !exists || n == nil {
+		return ret
 	}
-	return nil
+	n.RLock()
+	defer n.RUnlock()
+	*ret = nodeInfo{
+		podInfos:          make(map[types.UID]*podAssignInfo, len(n.podInfos)),
+		nodeMetric:        n.nodeMetric,
+		updateTime:        n.updateTime,
+		reportInterval:    n.reportInterval,
+		prodUsage:         n.prodUsage.Clone().(ResourceVector),
+		nodeDelta:         n.nodeDelta.Clone().(ResourceVector),
+		prodDelta:         n.prodDelta.Clone().(ResourceVector),
+		nodeEstimated:     n.nodeEstimated.Clone().(ResourceVector),
+		nodeDeltaPods:     n.nodeDeltaPods.Clone(),
+		prodDeltaPods:     n.prodDeltaPods.Clone(),
+		nodeEstimatedPods: n.nodeEstimatedPods.Clone(),
+	}
+	for uid, pod := range n.podInfos {
+		ret.podInfos[uid] = pod
+	}
+	return ret
 }
 
-func (p *podAssignCache) getPodsAssignInfoOnNode(nodeName string) []*podAssignInfo {
-	p.lock.RLock()
-	defer p.lock.RUnlock()
-	m := p.podInfoItems[nodeName]
-	if m == nil {
-		return nil
+func (p *podAssignCache) getNodeInfo(nodeName string) (*nodeInfo, bool) {
+	v, ok := p.items.Load(nodeName)
+	if !ok {
+		return nil, ok
 	}
-	podInfos := make([]*podAssignInfo, 0, len(m))
-	for _, info := range m {
-		podInfos = append(podInfos, info)
+	return v.(*nodeInfo), ok
+}
+
+// getOrCreateNodeInfo returns the nodeInfo for the given nodeName.
+// If the nodeInfo does not exist, it will be created with a locked mutex
+// which prevent the empty nodeInfo is read and used by plugin with RLock before objects' updating.
+//
+// NOTICE: it should only be called in objects' add or update methods.
+func (p *podAssignCache) getOrCreateNodeInfo(nodeName string) (_ *nodeInfo, created bool) {
+	n := &nodeInfo{}
+	n.Lock()
+	v, loaded := p.items.LoadOrStore(nodeName, n)
+	return v.(*nodeInfo), !loaded
+}
+
+// tryCleanup cleans up the nodeInfo from cache.items if it's empty.
+//
+// NOTICE: nodeInfo should be locked before calling this method.
+func (p *podAssignCache) tryCleanup(name string, n *nodeInfo) {
+	if n.nodeMetric == nil && len(n.podInfos) == 0 {
+		n.deleted = true
+		// only delete action has the chance that goroutine holds two locks,
+		// and the order always will be nodeInfo lock first, then podAssignCache.items lock
+		p.items.CompareAndDelete(name, n)
 	}
-	return podInfos
 }
 
+// add or update pod with node name provided
 func (p *podAssignCache) assign(nodeName string, pod *corev1.Pod) {
 	if nodeName == "" || util.IsPodTerminated(pod) {
 		return
 	}
-	estimated, err := p.estimator.EstimatePod(pod)
-	if err != nil || len(estimated) == 0 {
-		estimated = nil
+	var estimated ResourceVector
+	if list, err := p.estimator.EstimatePod(pod); err == nil && len(list) != 0 {
+		if vec := p.vectorizer.ToFactorVec(list); !vec.Empty() {
+			estimated = vec
+		}
 	}
 	var timestamp time.Time
 	// try to use time from PodScheduled condition first
@@ -105,30 +301,26 @@ func (p *podAssignCache) assign(nodeName string, pod *corev1.Pod) {
 		timestamp = c.LastTransitionTime.Time
 	} else {
 		// if PodScheduled condition not found, fallback to use assign timestamp from scheduler internal, which cannot be zero.
-		timestamp = timeNowFn()
+		timestamp = p.clock.Now()
 	}
 	estimatedDeadline := p.shouldEstimatePodDeadline(pod, timestamp)
-	p.lock.Lock()
-	defer p.lock.Unlock()
-	m := p.podInfoItems[nodeName]
-	if m == nil {
-		m = make(map[types.UID]*podAssignInfo)
-		p.podInfoItems[nodeName] = m
-	}
-
-	if info, ok := m[pod.UID]; ok {
-		info.timestamp = timestamp
-		info.pod = pod
-		info.estimated = estimated
-		info.estimatedDeadline = estimatedDeadline
-	} else {
-		m[pod.UID] = &podAssignInfo{
-			timestamp:         timestamp,
-			pod:               pod,
-			estimated:         estimated,
-			estimatedDeadline: estimatedDeadline,
+	newPod := &podAssignInfo{
+		timestamp:         timestamp,
+		pod:               pod,
+		estimated:         estimated,
+		estimatedDeadline: estimatedDeadline,
+	}
+
+	// We only try 2 times here, to avoid any bug in AddOrUpdate that keeps returning false and retrying forever,
+	// which will block the whole informer event handling procedure.
+	for i := 0; i < 2; i++ {
+		n, created := p.getOrCreateNodeInfo(nodeName)
+		// if nodeInfo is created in getOrCreate, it is locked already
+		if n.AddOrUpdatePod(newPod, created) {
+			return
 		}
 	}
+	klog.ErrorS(nil, "Failed to assign pod in cache after retried", "pod", klog.KObj(pod), "node", nodeName)
 }
 
 func (p *podAssignCache) shouldEstimatePodDeadline(pod *corev1.Pod, timestamp time.Time) time.Time {
@@ -162,11 +354,8 @@ func (p *podAssignCache) unAssign(nodeName string, pod *corev1.Pod) {
 	if nodeName == "" {
 		return
 	}
-	p.lock.Lock()
-	defer p.lock.Unlock()
-	delete(p.podInfoItems[nodeName], pod.UID)
-	if len(p.podInfoItems[nodeName]) == 0 {
-		delete(p.podInfoItems, nodeName)
+	if n, ok := p.getNodeInfo(nodeName); ok {
+		n.DeletePod(nodeName, pod.UID, p)
 	}
 }
 
@@ -211,3 +400,297 @@ func (p *podAssignCache) OnDelete(obj interface{}) {
 	}
 	p.unAssign(pod.Spec.NodeName, pod)
 }
+
+// AddOrUpdatePod add or update pod to nodeInfo.
+// It returns false is nodeInfo is already deleted and caller should get a new nodeInfo and retry.
+// Unlock is called whether nodeInfo is locked before calling or not.
+func (n *nodeInfo) AddOrUpdatePod(pod *podAssignInfo, locked bool) bool {
+	if n.deleted {
+		if locked {
+			n.Unlock()
+		}
+		return false
+	}
+	if !locked {
+		n.Lock()
+	}
+	defer n.Unlock()
+	if n.deleted {
+		return false
+	}
+	var oldPod *podAssignInfo
+	if n.podInfos == nil {
+		n.podInfos = map[types.UID]*podAssignInfo{}
+	} else {
+		oldPod = n.podInfos[pod.pod.UID]
+	}
+	n.podInfos[pod.pod.UID] = pod
+	if n.nodeMetric != nil {
+		if oldPod == nil {
+			n.addPod(pod)
+		} else {
+			n.updatePod(oldPod, pod)
+		}
+	}
+	return true
+}
+
+func (n *nodeInfo) DeletePod(name string, uid types.UID, p *podAssignCache) {
+	if n.deleted {
+		return
+	}
+	n.Lock()
+	defer n.Unlock()
+	if n.deleted {
+		return
+	}
+	oldPod := n.podInfos[uid]
+	if oldPod != nil {
+		delete(n.podInfos, uid)
+	}
+	if n.nodeMetric != nil && oldPod != nil {
+		n.deletePod(oldPod)
+	}
+	p.tryCleanup(name, n)
+}
+
+func (p *podAssignCache) NodeMetricHandler() cache.ResourceEventHandler {
+	return cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj any) {
+			if m, ok := obj.(*slov1alpha1.NodeMetric); ok && m != nil {
+				p.AddOrUpdateNodeMetric(m)
+			}
+		},
+		UpdateFunc: func(_ any, obj any) {
+			if m, ok := obj.(*slov1alpha1.NodeMetric); ok && m != nil {
+				p.AddOrUpdateNodeMetric(m)
+			}
+		},
+		DeleteFunc: func(obj any) {
+			var m *slov1alpha1.NodeMetric
+			switch o := obj.(type) {
+			case *slov1alpha1.NodeMetric:
+				m = o
+			case cache.DeletedFinalStateUnknown:
+				var ok bool
+				if m, ok = o.Obj.(*slov1alpha1.NodeMetric); !ok {
+					return
+				}
+			default:
+				return
+			}
+			p.DeleteNodeMetric(m.Name)
+		},
+	}
+}
+
+func (p *podAssignCache) AddOrUpdateNodeMetric(metric *slov1alpha1.NodeMetric) {
+	// We only try 2 times here, to avoid any bug in AddOrUpdate that keeps returning false and retrying forever,
+	// which will block the whole informer event handling procedure.
+	for i := 0; i < 2; i++ {
+		n, created := p.getOrCreateNodeInfo(metric.Name)
+		// if nodeInfo is created in getOrCreate, it is locked already
+		if n.AddOrUpdateNodeMetric(metric, p, created) {
+			return
+		}
+	}
+	klog.ErrorS(nil, "Failed to add or update nodemetric in cache after retried", "node", klog.KObj(metric))
+}
+
+func (p *podAssignCache) DeleteNodeMetric(name string) {
+	if n, ok := p.getNodeInfo(name); ok {
+		n.DeleteNodeMetric(name, p)
+	}
+}
+
+// AddOrUpdateNodeMetric add or update node metric to nodeInfo.
+// It returns false is nodeInfo is already deleted and caller should get a new nodeInfo and retry.
+// Unlock is called whether nodeInfo is locked before calling or not.
+func (n *nodeInfo) AddOrUpdateNodeMetric(metric *slov1alpha1.NodeMetric, p *podAssignCache, locked bool) bool {
+	if n.deleted {
+		if locked {
+			n.Unlock()
+		}
+		return false
+	}
+	var podUsages map[NamespacedName]ResourceVector
+	var prodPods sets.Set[NamespacedName]
+	var nodeUsage, prodUsage ResourceVector = nil, p.vectorizer.EmptyVec()
+	var aggUsages map[aggUsageKey]ResourceVector
+	if info := metric.Status.NodeMetric; info != nil {
+		nodeUsage = p.vectorizer.ToVec(info.NodeUsage.ResourceList)
+		if aggLen := len(info.AggregatedNodeUsages); aggLen > 0 {
+			typeLen := len(info.AggregatedNodeUsages[0].Usage)
+			aggUsages = make(map[aggUsageKey]ResourceVector, (aggLen+1)*typeLen)
+			maxDurations := make(map[extension.AggregationType]time.Duration, typeLen)
+			for _, aggInfo := range info.AggregatedNodeUsages {
+				d := aggInfo.Duration.Duration
+				for t, u := range aggInfo.Usage {
+					if len(u.ResourceList) == 0 {
+						continue
+					}
+					key := aggUsageKey{Type: t, Duration: d}
+					vec := p.vectorizer.ToVec(u.ResourceList)
+					aggUsages[key] = vec
+					if md := maxDurations[t]; d > md {
+						maxDurations[t] = d
+					}
+				}
+			}
+			for t, d := range maxDurations {
+				aggUsages[aggUsageKey{Type: t}] = aggUsages[aggUsageKey{Type: t, Duration: d}]
+			}
+		}
+		if p.args.ProdUsageIncludeSys {
+			prodUsage.Add(p.vectorizer.ToVec(info.SystemUsage.ResourceList))
+		}
+	}
+	if infos := metric.Status.PodsMetric; len(infos) != 0 {
+		podUsages = make(map[NamespacedName]ResourceVector, len(infos))
+		prodPods = sets.New[NamespacedName]()
+		for _, info := range infos {
+			if info == nil {
+				continue
+			}
+			if len(info.PodUsage.ResourceList) == 0 {
+				continue
+			}
+			key := NamespacedName{Namespace: info.Namespace, Name: info.Name}
+			vec := p.vectorizer.ToVec(info.PodUsage.ResourceList)
+			podUsages[key] = vec
+			if info.Priority == extension.PriorityProd {
+				prodPods.Insert(key)
+			}
+		}
+	}
+	if !locked {
+		n.Lock()
+	}
+	defer n.Unlock()
+	if n.deleted {
+		return false
+	}
+	n.nodeMetric = metric
+	n.reportInterval = getNodeMetricReportInterval(metric)
+	if metric.Status.UpdateTime != nil {
+		n.updateTime = metric.Status.UpdateTime.Time
+	}
+	n.podUsages, n.prodPods = podUsages, prodPods
+	n.nodeUsage = nodeUsage
+	n.prodUsage = prodUsage
+	n.aggUsages = aggUsages
+	n.nodeDelta = p.vectorizer.EmptyVec()
+	n.prodDelta = p.vectorizer.EmptyVec()
+	n.nodeEstimated = p.vectorizer.EmptyVec()
+	n.nodeDeltaPods = sets.New[NamespacedName]()
+	n.prodDeltaPods = sets.New[NamespacedName]()
+	n.nodeEstimatedPods = sets.New[NamespacedName]()
+	for _, pod := range n.podInfos {
+		n.addPod(pod)
+	}
+	return true
+}
+
+func (n *nodeInfo) DeleteNodeMetric(name string, p *podAssignCache) {
+	if n.deleted {
+		return
+	}
+	n.Lock()
+	defer n.Unlock()
+	if n.deleted {
+		return
+	}
+	n.nodeMetric = nil
+	p.tryCleanup(name, n)
+}
+
+func (n *nodeInfo) addPod(pod *podAssignInfo) {
+	key := NamespacedName{Namespace: pod.pod.Namespace, Name: pod.pod.Name}
+	u := n.podUsages[key]
+	prod := extension.GetPodPriorityClassWithDefault(pod.pod) == extension.PriorityProd
+	// Only use prod pod's usage when both pod claims and node metric reports it as prod.
+	// 1. pod priority class are updated dynamically
+	// 2. prod / non prod pod is wrongly reported or terminated pod is leaked in node metrics status
+	activeProd := prod && n.prodPods.Has(key)
+	if activeProd {
+		n.prodUsage.Add(u)
+	}
+
+	e := pod.estimated
+	if e == nil {
+		return
+	}
+	// 1. when usage is not collected
+	// 2. when pod miss lastest metrics update
+	// 3. when pod metrics is still in the report interval
+	// 4. when pod is configured in estimation
+	should := u == nil ||
+		n.updateTime.Add(-n.reportInterval).Before(pod.timestamp) ||
+		(!pod.estimatedDeadline.IsZero() && pod.estimatedDeadline.After(n.updateTime))
+	if should {
+		if n.nodeDelta.AddDelta(e, u) && n.nodeDeltaPods != nil {
+			n.nodeDeltaPods.Insert(key)
+		}
+	}
+	n.nodeEstimated.Add(e)
+	if n.nodeEstimatedPods != nil {
+		n.nodeEstimatedPods.Insert(key)
+	}
+
+	if !prod {
+		return
+	}
+	if !activeProd && u != nil {
+		u, should = nil, true
+	}
+	if should {
+		if n.prodDelta.AddDelta(e, u) && n.prodDeltaPods != nil {
+			n.prodDeltaPods.Insert(key)
+		}
+	}
+}
+
+func (n *nodeInfo) updatePod(oldPod, newPod *podAssignInfo) {
+	n.deletePod(oldPod)
+	n.addPod(newPod)
+}
+
+// reverse procedure of addPod
+func (n *nodeInfo) deletePod(pod *podAssignInfo) {
+	key := NamespacedName{Namespace: pod.pod.Namespace, Name: pod.pod.Name}
+	u := n.podUsages[key]
+	prod := extension.GetPodPriorityClassWithDefault(pod.pod) == extension.PriorityProd
+	activeProd := prod && n.prodPods.Has(key)
+	if activeProd {
+		n.prodUsage.Sub(u)
+	}
+
+	e := pod.estimated
+	if e == nil {
+		return
+	}
+	should := u == nil ||
+		n.updateTime.Add(-n.reportInterval).Before(pod.timestamp) ||
+		(!pod.estimatedDeadline.IsZero() && pod.estimatedDeadline.After(n.updateTime))
+	if should {
+		if n.nodeDelta.SubDelta(e, u) && n.nodeDeltaPods != nil {
+			n.nodeDeltaPods.Delete(key)
+		}
+	}
+	n.nodeEstimated.Sub(e)
+	if n.nodeEstimatedPods != nil {
+		n.nodeEstimatedPods.Delete(key)
+	}
+
+	if !prod {
+		return
+	}
+	if !activeProd && u != nil {
+		u, should = nil, true
+	}
+	if should {
+		if n.prodDelta.SubDelta(e, u) && n.prodDeltaPods != nil {
+			n.prodDeltaPods.Delete(key)
+		}
+	}
+}
diff --git a/pkg/scheduler/plugins/loadaware/service.go b/pkg/scheduler/plugins/loadaware/service.go
index 5ea19bee..8cecf7a9 100644
--- a/pkg/scheduler/plugins/loadaware/service.go
+++ b/pkg/scheduler/plugins/loadaware/service.go
@@ -18,10 +18,12 @@ package loadaware
 
 import (
 	"net/http"
+	"sort"
 	"time"
 
 	"github.com/gin-gonic/gin"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/klog/v2"
 
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/frameworkext/services"
 )
@@ -30,6 +32,14 @@ var _ services.APIServiceProvider = &Plugin{}
 
 type NodeAssignInfoData struct {
 	Pods []PodAssignInfoData `json:"pods,omitempty"`
+
+	ProdUsage         corev1.ResourceList `json:"prodUsage,omitempty"`
+	NodeDelta         corev1.ResourceList `json:"nodeDelta,omitempty"`
+	ProdDelta         corev1.ResourceList `json:"prodDelta,omitempty"`
+	NodeEstimated     corev1.ResourceList `json:"nodeEstimated,omitempty"`
+	NodeDeltaPods     []string            `json:"nodeDeltaPods,omitempty"`
+	ProdDeltaPods     []string            `json:"prodDeltaPods,omitempty"`
+	NodeEstimatedPods []string            `json:"nodeEstimatedPods,omitempty"`
 }
 
 type PodAssignInfoData struct {
@@ -40,19 +50,46 @@ type PodAssignInfoData struct {
 func (p *Plugin) RegisterEndpoints(group *gin.RouterGroup) {
 	group.GET("/node/:nodeName", func(c *gin.Context) {
 		nodeName := c.Param("nodeName")
-		assignInfo := p.podAssignCache.getPodsAssignInfoOnNode(nodeName)
-		if len(assignInfo) == 0 {
+		n := p.podAssignCache.getClonedNodeInfo(nodeName)
+		if n == nil || (len(n.podInfos) == 0 && n.nodeMetric == nil) {
 			c.JSON(http.StatusOK, &NodeAssignInfoData{})
 			return
 		}
 
 		resp := &NodeAssignInfoData{
-			Pods: make([]PodAssignInfoData, 0, len(assignInfo)),
+			Pods: make([]PodAssignInfoData, 0, len(n.podInfos)),
 		}
-		for i := range assignInfo {
+		for i := range n.podInfos {
 			resp.Pods = append(resp.Pods, PodAssignInfoData{
-				Timestamp: assignInfo[i].timestamp,
-				Pod:       assignInfo[i].pod,
+				Timestamp: n.podInfos[i].timestamp,
+				Pod:       n.podInfos[i].pod,
+			})
+		}
+		sort.Slice(resp.Pods, func(i, j int) bool {
+			return resp.Pods[i].Timestamp.Before(resp.Pods[j].Timestamp)
+		})
+		if n.nodeMetric != nil {
+			resp.ProdUsage = p.vectorizer.ToList(n.prodUsage)
+			resp.NodeDelta = p.vectorizer.ToList(n.nodeDelta)
+			resp.ProdDelta = p.vectorizer.ToList(n.prodDelta)
+			resp.NodeEstimated = p.vectorizer.ToList(n.nodeEstimated)
+			for _, pod := range n.nodeDeltaPods.UnsortedList() {
+				resp.NodeDeltaPods = append(resp.NodeDeltaPods, klog.KObj(pod).String())
+			}
+			sort.Slice(resp.NodeDeltaPods, func(i, j int) bool {
+				return resp.NodeDeltaPods[i] < resp.NodeDeltaPods[j]
+			})
+			for _, pod := range n.prodDeltaPods.UnsortedList() {
+				resp.ProdDeltaPods = append(resp.ProdDeltaPods, klog.KObj(pod).String())
+			}
+			sort.Slice(resp.ProdDeltaPods, func(i, j int) bool {
+				return resp.ProdDeltaPods[i] < resp.ProdDeltaPods[j]
+			})
+			for _, pod := range n.nodeEstimatedPods.UnsortedList() {
+				resp.NodeEstimatedPods = append(resp.NodeEstimatedPods, klog.KObj(pod).String())
+			}
+			sort.Slice(resp.NodeEstimatedPods, func(i, j int) bool {
+				return resp.NodeEstimatedPods[i] < resp.NodeEstimatedPods[j]
 			})
 		}
 



Test patch:
diff --git a/pkg/scheduler/plugins/loadaware/helper_test.go b/pkg/scheduler/plugins/loadaware/helper_test.go
index 52de3899..711941f8 100644
--- a/pkg/scheduler/plugins/loadaware/helper_test.go
+++ b/pkg/scheduler/plugins/loadaware/helper_test.go
@@ -27,6 +27,8 @@ import (
 
 	"github.com/koordinator-sh/koordinator/apis/extension"
 	slov1alpha1 "github.com/koordinator-sh/koordinator/apis/slo/v1alpha1"
+	"github.com/koordinator-sh/koordinator/pkg/scheduler/apis/config"
+	"github.com/koordinator-sh/koordinator/pkg/scheduler/plugins/loadaware/estimator"
 )
 
 // Test cases description:
@@ -39,7 +41,7 @@ func TestGetTargetAggregatedUsage(t *testing.T) {
 	tests := []struct {
 		name               string
 		nodeMetric         *slov1alpha1.NodeMetric
-		aggregatedDuration *metav1.Duration
+		aggregatedDuration metav1.Duration
 
 		expectedResult *slov1alpha1.ResourceMap
 	}{
@@ -81,7 +83,6 @@ func TestGetTargetAggregatedUsage(t *testing.T) {
 					},
 				},
 			},
-			aggregatedDuration: nil,
 			expectedResult: &slov1alpha1.ResourceMap{
 				ResourceList: corev1.ResourceList{
 					corev1.ResourceCPU: resource.MustParse("50"),
@@ -128,7 +129,7 @@ func TestGetTargetAggregatedUsage(t *testing.T) {
 					},
 				},
 			},
-			aggregatedDuration: &metav1.Duration{Duration: 5 * time.Minute},
+			aggregatedDuration: metav1.Duration{Duration: 5 * time.Minute},
 			expectedResult: &slov1alpha1.ResourceMap{
 				ResourceList: corev1.ResourceList{
 					corev1.ResourceCPU: resource.MustParse("30"),
@@ -158,7 +159,6 @@ func TestGetTargetAggregatedUsage(t *testing.T) {
 					},
 				},
 			},
-			aggregatedDuration: nil,
 			expectedResult: &slov1alpha1.ResourceMap{
 				ResourceList: corev1.ResourceList{
 					corev1.ResourceCPU: resource.MustParse("30"),
@@ -169,8 +169,13 @@ func TestGetTargetAggregatedUsage(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := getTargetAggregatedUsage(tt.nodeMetric, tt.aggregatedDuration, aggregationType)
-			assert.Equal(t, tt.expectedResult, result)
+			e, _ := estimator.NewDefaultEstimator(&config.LoadAwareSchedulingArgs{}, nil)
+			v := NewResourceVectorizer(corev1.ResourceCPU, corev1.ResourceMemory)
+			c := newPodAssignCache(e, v, &config.LoadAwareSchedulingArgs{})
+			c.AddOrUpdateNodeMetric(tt.nodeMetric)
+			n, _ := c.getNodeInfo(tt.nodeMetric.Name)
+			result := n.getTargetAggregatedUsage(tt.aggregatedDuration, aggregationType)
+			assert.Equal(t, v.ToVec(tt.expectedResult.ResourceList), result)
 		})
 	}
 }
diff --git a/pkg/scheduler/plugins/loadaware/load_aware_test.go b/pkg/scheduler/plugins/loadaware/load_aware_test.go
index 993626fa..0e7ece76 100644
--- a/pkg/scheduler/plugins/loadaware/load_aware_test.go
+++ b/pkg/scheduler/plugins/loadaware/load_aware_test.go
@@ -223,6 +223,9 @@ func TestFilterExpiredNodeMetric(t *testing.T) {
 			cs := kubefake.NewSimpleClientset()
 			informerFactory := informers.NewSharedInformerFactory(cs, 0)
 
+			_, err = koordClientSet.SloV1alpha1().NodeMetrics().Create(context.TODO(), tt.nodeMetric, metav1.CreateOptions{})
+			assert.NoError(t, err)
+
 			nodes := []*corev1.Node{
 				{
 					ObjectMeta: metav1.ObjectMeta{
@@ -247,9 +250,6 @@ func TestFilterExpiredNodeMetric(t *testing.T) {
 			assert.NotNil(t, p)
 			assert.Nil(t, err)
 
-			_, err = koordClientSet.SloV1alpha1().NodeMetrics().Create(context.TODO(), tt.nodeMetric, metav1.CreateOptions{})
-			assert.NoError(t, err)
-
 			koordSharedInformerFactory.Start(context.TODO().Done())
 			koordSharedInformerFactory.WaitForCacheSync(context.TODO().Done())
 
@@ -382,6 +382,9 @@ func TestEnableScheduleWhenNodeMetricsExpired(t *testing.T) {
 			cs := kubefake.NewSimpleClientset()
 			informerFactory := informers.NewSharedInformerFactory(cs, 0)
 
+			_, err = koordClientSet.SloV1alpha1().NodeMetrics().Create(context.TODO(), tt.nodeMetric, metav1.CreateOptions{})
+			assert.NoError(t, err)
+
 			nodes := []*corev1.Node{
 				{
 					ObjectMeta: metav1.ObjectMeta{
@@ -406,9 +409,6 @@ func TestEnableScheduleWhenNodeMetricsExpired(t *testing.T) {
 			assert.NotNil(t, p)
 			assert.Nil(t, err)
 
-			_, err = koordClientSet.SloV1alpha1().NodeMetrics().Create(context.TODO(), tt.nodeMetric, metav1.CreateOptions{})
-			assert.NoError(t, err)
-
 			koordSharedInformerFactory.Start(context.TODO().Done())
 			koordSharedInformerFactory.WaitForCacheSync(context.TODO().Done())
 
@@ -1371,6 +1371,11 @@ func TestFilterUsage(t *testing.T) {
 			cs := kubefake.NewSimpleClientset()
 			informerFactory := informers.NewSharedInformerFactory(cs, 0)
 
+			if tt.nodeMetric != nil {
+				_, err = koordClientSet.SloV1alpha1().NodeMetrics().Create(context.TODO(), tt.nodeMetric, metav1.CreateOptions{})
+				assert.NoError(t, err)
+			}
+
 			for _, v := range tt.pods {
 				v.Spec.NodeName = tt.nodeName
 				v.UID = uuid.NewUUID()
@@ -1428,11 +1433,6 @@ func TestFilterUsage(t *testing.T) {
 			assert.NotNil(t, p)
 			assert.Nil(t, err)
 
-			if tt.nodeMetric != nil {
-				_, err = koordClientSet.SloV1alpha1().NodeMetrics().Create(context.TODO(), tt.nodeMetric, metav1.CreateOptions{})
-				assert.NoError(t, err)
-			}
-
 			informerFactory.Start(context.TODO().Done())
 			informerFactory.WaitForCacheSync(context.TODO().Done())
 
diff --git a/pkg/scheduler/plugins/loadaware/pod_assign_cache_test.go b/pkg/scheduler/plugins/loadaware/pod_assign_cache_test.go
index e0ad2187..de7c5046 100644
--- a/pkg/scheduler/plugins/loadaware/pod_assign_cache_test.go
+++ b/pkg/scheduler/plugins/loadaware/pod_assign_cache_test.go
@@ -22,363 +22,437 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	schedulertesting "k8s.io/kubernetes/pkg/scheduler/testing"
+	clock "k8s.io/utils/clock/testing"
 	"k8s.io/utils/ptr"
 
 	"github.com/koordinator-sh/koordinator/apis/extension"
+	slov1alpha1 "github.com/koordinator-sh/koordinator/apis/slo/v1alpha1"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/apis/config"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/plugins/loadaware/estimator"
 )
 
-var fakeTimeNowFn = func() time.Time {
-	t := time.Time{}
-	_ = t.Add(100 * time.Second)
-	return t
-}
-
 func TestPodAssignCache_OnAdd(t *testing.T) {
+	vectorizer := NewResourceVectorizer(corev1.ResourceCPU, corev1.ResourceMemory)
+	node := "test-node"
+	now := metav1.Now().Rfc3339Copy().Time
+	m := &slov1alpha1.NodeMetric{ObjectMeta: metav1.ObjectMeta{Name: node}}
 	tests := []struct {
-		name      string
-		pod       *corev1.Pod
-		wantCache map[string]map[types.UID]*podAssignInfo
+		name string
+		pod  *corev1.Pod
+		want func(*testing.T, *nodeInfo)
 	}{
 		{
-			name:      "update pending pod",
-			pod:       &corev1.Pod{},
-			wantCache: map[string]map[types.UID]*podAssignInfo{},
+			name: "add pending pod",
+			pod:  schedulertesting.MakePod().Obj(),
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, 0, len(n.podInfos))
+			},
 		},
 		{
-			name: "update terminated pod",
-			pod: &corev1.Pod{
-				Spec: corev1.PodSpec{
-					NodeName: "test-node",
-				},
-				Status: corev1.PodStatus{
-					Phase: corev1.PodFailed,
-				},
+			name: "add terminated pod",
+			pod:  schedulertesting.MakePod().Node(node).Phase(corev1.PodFailed).Obj(),
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, 0, len(n.podInfos))
 			},
-			wantCache: map[string]map[types.UID]*podAssignInfo{},
 		},
 		{
-			name: "update scheduled running pod",
-			pod: &corev1.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					UID:       "123456789",
-					Namespace: "default",
-					Name:      "test",
-				},
-				Spec: corev1.PodSpec{
-					NodeName: "test-node",
-				},
-				Status: corev1.PodStatus{
-					Phase: corev1.PodRunning,
-				},
-			},
-			wantCache: map[string]map[types.UID]*podAssignInfo{
-				"test-node": {
-					"123456789": &podAssignInfo{
-						pod: &corev1.Pod{
-							ObjectMeta: metav1.ObjectMeta{
-								UID:       "123456789",
-								Namespace: "default",
-								Name:      "test",
-							},
-							Spec: corev1.PodSpec{
-								NodeName: "test-node",
-							},
-							Status: corev1.PodStatus{
-								Phase: corev1.PodRunning,
-							},
-						},
-						timestamp: fakeTimeNowFn(),
+			name: "add scheduled running pod without resources",
+			pod:  schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).Obj(),
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, map[types.UID]*podAssignInfo{
+					"123456789": {
+						pod:       schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).Obj(),
+						timestamp: now,
+						estimated: vectorizer.ToFactorVec(map[corev1.ResourceName]int64{
+							corev1.ResourceCPU:    estimator.DefaultMilliCPURequest,
+							corev1.ResourceMemory: estimator.DefaultMemoryRequest,
+						}),
 					},
-				},
+				}, n.podInfos)
+				v := vectorizer.ToFactorVec(map[corev1.ResourceName]int64{
+					corev1.ResourceCPU:    estimator.DefaultMilliCPURequest,
+					corev1.ResourceMemory: estimator.DefaultMemoryRequest,
+				})
+				assert.Equal(t, v, n.nodeDelta)
+				assert.Equal(t, v, n.nodeEstimated)
+				s := sets.New(NamespacedName{Namespace: "default", Name: "test"})
+				assert.Equal(t, s, n.nodeDeltaPods)
+				assert.Equal(t, s, n.nodeEstimatedPods)
+			},
+		},
+		{
+			name: "add prod pod",
+			pod: schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).
+				Req(map[corev1.ResourceName]string{corev1.ResourceCPU: "1", corev1.ResourceMemory: "4Gi"}).Obj(),
+			want: func(t *testing.T, n *nodeInfo) {
+				v := vectorizer.ToVec(corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("1"),
+					corev1.ResourceMemory: resource.MustParse("4Gi"),
+				})
+				assert.Equal(t, v, n.nodeDelta)
+				assert.Equal(t, v, n.prodDelta)
+				assert.Equal(t, v, n.nodeEstimated)
+				s := sets.New(NamespacedName{Namespace: "default", Name: "test"})
+				assert.Equal(t, s, n.nodeDeltaPods)
+				assert.Equal(t, s, n.prodDeltaPods)
+				assert.Equal(t, s, n.nodeEstimatedPods)
+			},
+		},
+		{
+			name: "add non prod pod",
+			pod: schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Priority(extension.PriorityMidValueDefault).
+				Req(map[corev1.ResourceName]string{extension.MidCPU: "1k", extension.MidMemory: "4Gi"}).Obj(),
+			want: func(t *testing.T, n *nodeInfo) {
+				v := vectorizer.ToVec(corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("1"),
+					corev1.ResourceMemory: resource.MustParse("4Gi"),
+				})
+				assert.Equal(t, v, n.nodeDelta)
+				assert.Equal(t, v, n.nodeEstimated)
+				s := sets.New(NamespacedName{Namespace: "default", Name: "test"})
+				assert.Equal(t, s, n.nodeDeltaPods)
+				assert.Equal(t, s, n.nodeEstimatedPods)
 			},
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			preTimeNowFn := timeNowFn
-			defer func() {
-				timeNowFn = preTimeNowFn
-			}()
-			timeNowFn = fakeTimeNowFn
-			e, _ := estimator.NewDefaultEstimator(&config.LoadAwareSchedulingArgs{}, nil)
-			assignCache := newPodAssignCache(e, &config.LoadAwareSchedulingArgs{})
+			args := &config.LoadAwareSchedulingArgs{EstimatedScalingFactors: map[corev1.ResourceName]int64{
+				corev1.ResourceCPU:    100,
+				corev1.ResourceMemory: 100,
+			}}
+			e, _ := estimator.NewDefaultEstimator(args, nil)
+			assignCache := newPodAssignCache(e, vectorizer, args)
+			assignCache.clock = clock.NewFakeClock(now)
+			assignCache.AddOrUpdateNodeMetric(m)
 			assignCache.OnAdd(tt.pod, true)
-			assert.Equal(t, tt.wantCache, assignCache.podInfoItems)
+			actual, _ := assignCache.getNodeInfo(node)
+			tt.want(t, actual)
 		})
 	}
 }
 
 func TestPodAssignCache_OnUpdate(t *testing.T) {
+	vectorizer := NewResourceVectorizer(corev1.ResourceCPU, corev1.ResourceMemory)
+	node := "test-node"
+	now := metav1.Now().Rfc3339Copy().Time
+	m := &slov1alpha1.NodeMetric{ObjectMeta: metav1.ObjectMeta{Name: node}}
 	tests := []struct {
 		name         string
 		pod          *corev1.Pod
-		podInfoItems map[string]map[types.UID]*podAssignInfo
-		wantCache    map[string]map[types.UID]*podAssignInfo
+		existingPods []*corev1.Pod
+		want         func(*testing.T, *nodeInfo)
 	}{
 		{
-			name:      "update pending pod",
-			pod:       &corev1.Pod{},
-			wantCache: map[string]map[types.UID]*podAssignInfo{},
+			name: "update pending pod",
+			pod:  schedulertesting.MakePod().Obj(),
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, 0, len(n.podInfos))
+			},
 		},
 		{
 			name: "update terminated pod",
-			pod: &corev1.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					UID:       "123456789",
-					Namespace: "default",
-					Name:      "test",
-				},
-				Spec: corev1.PodSpec{
-					NodeName: "test-node",
-				},
-				Status: corev1.PodStatus{
-					Phase: corev1.PodFailed,
-				},
+			pod:  schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodFailed).Obj(),
+			existingPods: []*corev1.Pod{
+				schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).Obj(),
 			},
-			podInfoItems: map[string]map[types.UID]*podAssignInfo{
-				"test-node": {
-					"123456789": &podAssignInfo{
-						pod: &corev1.Pod{
-							ObjectMeta: metav1.ObjectMeta{
-								UID:       "123456789",
-								Namespace: "default",
-								Name:      "test",
-							},
-							Spec: corev1.PodSpec{
-								NodeName: "test-node",
-							},
-							Status: corev1.PodStatus{
-								Phase: corev1.PodRunning,
-							},
-						},
-						timestamp: fakeTimeNowFn(),
-					},
-				},
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, 0, len(n.podInfos))
 			},
-			wantCache: map[string]map[types.UID]*podAssignInfo{},
 		},
 		{
-			name: "update scheduled running pod",
-			pod: &corev1.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					UID:       "123456789",
-					Namespace: "default",
-					Name:      "test",
-				},
-				Spec: corev1.PodSpec{
-					NodeName: "test-node",
-				},
-				Status: corev1.PodStatus{
-					Phase: corev1.PodRunning,
-				},
-			},
-			wantCache: map[string]map[types.UID]*podAssignInfo{
-				"test-node": {
-					"123456789": &podAssignInfo{
-						pod: &corev1.Pod{
-							ObjectMeta: metav1.ObjectMeta{
-								UID:       "123456789",
-								Namespace: "default",
-								Name:      "test",
-							},
-							Spec: corev1.PodSpec{
-								NodeName: "test-node",
-							},
-							Status: corev1.PodStatus{
-								Phase: corev1.PodRunning,
-							},
-						},
-						timestamp: fakeTimeNowFn(),
+			name: "update scheduled running pod without resources",
+			pod:  schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).Obj(),
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, map[types.UID]*podAssignInfo{
+					"123456789": {
+						pod:       schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).Obj(),
+						timestamp: now,
+						estimated: vectorizer.ToFactorVec(map[corev1.ResourceName]int64{
+							corev1.ResourceCPU:    estimator.DefaultMilliCPURequest,
+							corev1.ResourceMemory: estimator.DefaultMemoryRequest,
+						}),
 					},
-				},
+				}, n.podInfos)
+				v := vectorizer.ToFactorVec(map[corev1.ResourceName]int64{
+					corev1.ResourceCPU:    estimator.DefaultMilliCPURequest,
+					corev1.ResourceMemory: estimator.DefaultMemoryRequest,
+				})
+				assert.Equal(t, v, n.nodeDelta)
+				assert.Equal(t, v, n.nodeEstimated)
+				s := sets.New(NamespacedName{Namespace: "default", Name: "test"})
+				assert.Equal(t, s, n.nodeDeltaPods)
+				assert.Equal(t, s, n.nodeEstimatedPods)
 			},
 		},
 		{
 			name: "update pod metadata only, cache won't be updated",
-			pod: &corev1.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					UID:       "123456789",
-					Namespace: "default",
-					Name:      "test",
-				},
-				Spec: corev1.PodSpec{
-					NodeName: "test-node",
-				},
-				Status: corev1.PodStatus{
-					Phase: corev1.PodRunning,
-				},
+			pod: schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).
+				Annotation("foo", "bar").Label("foo", "bar").Obj(),
+			existingPods: []*corev1.Pod{
+				schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).Obj(),
 			},
-			podInfoItems: map[string]map[types.UID]*podAssignInfo{
-				"test-node": {
-					"123456789": &podAssignInfo{
-						pod: &corev1.Pod{
-							ObjectMeta: metav1.ObjectMeta{
-								UID:       "123456789",
-								Namespace: "default",
-								Name:      "test",
-							},
-							Spec: corev1.PodSpec{
-								NodeName: "test-node",
-							},
-							Status: corev1.PodStatus{
-								Phase: corev1.PodRunning,
-							},
-						},
-						timestamp: fakeTimeNowFn().Add(1000),
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, map[types.UID]*podAssignInfo{
+					"123456789": {
+						pod:       schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).Obj(),
+						timestamp: now,
+						estimated: vectorizer.ToFactorVec(map[corev1.ResourceName]int64{
+							corev1.ResourceCPU:    estimator.DefaultMilliCPURequest,
+							corev1.ResourceMemory: estimator.DefaultMemoryRequest,
+						}),
 					},
-				},
-			},
-			wantCache: map[string]map[types.UID]*podAssignInfo{
-				"test-node": {
-					"123456789": &podAssignInfo{
-						pod: &corev1.Pod{
-							ObjectMeta: metav1.ObjectMeta{
-								UID:       "123456789",
-								Namespace: "default",
-								Name:      "test",
-							},
-							Spec: corev1.PodSpec{
-								NodeName: "test-node",
-							},
-							Status: corev1.PodStatus{
-								Phase: corev1.PodRunning,
-							},
-						},
-						timestamp: fakeTimeNowFn().Add(1000),
-					},
-				},
+				}, n.podInfos)
 			},
 		},
 		{
 			name: "update pod conditions, cache will be updated",
-			pod: &corev1.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					UID:       "123456789",
-					Namespace: "default",
-					Name:      "test",
-				},
-				Spec: corev1.PodSpec{
-					NodeName: "test-node",
-				},
-				Status: corev1.PodStatus{
-					Phase: corev1.PodRunning,
-					Conditions: []corev1.PodCondition{
-						{Type: corev1.PodScheduled, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(fakeTimeNowFn().Add(1000))},
-						{Type: corev1.PodInitialized, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(fakeTimeNowFn().Add(3000))},
-					},
-				},
+			pod: schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).
+				Conditions([]corev1.PodCondition{
+					{Type: corev1.PodScheduled, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(now.Add(1000))},
+					{Type: corev1.PodInitialized, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(now.Add(3000))},
+				}).Obj(),
+			existingPods: []*corev1.Pod{
+				schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).
+					Conditions([]corev1.PodCondition{
+						{Type: corev1.PodScheduled, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(now.Add(1000))},
+					}).Obj(),
 			},
-			podInfoItems: map[string]map[types.UID]*podAssignInfo{
-				"test-node": {
-					"123456789": &podAssignInfo{
-						pod: &corev1.Pod{
-							ObjectMeta: metav1.ObjectMeta{
-								UID:       "123456789",
-								Namespace: "default",
-								Name:      "test",
-							},
-							Spec: corev1.PodSpec{
-								NodeName: "test-node",
-							},
-							Status: corev1.PodStatus{
-								Phase: corev1.PodRunning,
-								Conditions: []corev1.PodCondition{
-									{Type: corev1.PodScheduled, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(fakeTimeNowFn().Add(1000))},
-								},
-							},
-						},
-						timestamp: fakeTimeNowFn().Add(1000),
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, map[types.UID]*podAssignInfo{
+					"123456789": {
+						pod: schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).
+							Conditions([]corev1.PodCondition{
+								{Type: corev1.PodScheduled, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(now.Add(1000))},
+								{Type: corev1.PodInitialized, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(now.Add(3000))},
+							}).Obj(),
+						timestamp: now.Add(1000),
+						estimated: vectorizer.ToFactorVec(map[corev1.ResourceName]int64{
+							corev1.ResourceCPU:    estimator.DefaultMilliCPURequest,
+							corev1.ResourceMemory: estimator.DefaultMemoryRequest,
+						}),
 					},
-				},
+				}, n.podInfos)
+				v := vectorizer.ToFactorVec(map[corev1.ResourceName]int64{
+					corev1.ResourceCPU:    estimator.DefaultMilliCPURequest,
+					corev1.ResourceMemory: estimator.DefaultMemoryRequest,
+				})
+				assert.Equal(t, v, n.nodeDelta)
+				assert.Equal(t, v, n.nodeEstimated)
+				s := sets.New(NamespacedName{Namespace: "default", Name: "test"})
+				assert.Equal(t, s, n.nodeDeltaPods)
+				assert.Equal(t, s, n.nodeEstimatedPods)
 			},
-			wantCache: map[string]map[types.UID]*podAssignInfo{
-				"test-node": {
-					"123456789": &podAssignInfo{
-						pod: &corev1.Pod{
-							ObjectMeta: metav1.ObjectMeta{
-								UID:       "123456789",
-								Namespace: "default",
-								Name:      "test",
-							},
-							Spec: corev1.PodSpec{
-								NodeName: "test-node",
-							},
-							Status: corev1.PodStatus{
-								Phase: corev1.PodRunning,
-								Conditions: []corev1.PodCondition{
-									{Type: corev1.PodScheduled, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(fakeTimeNowFn().Add(1000))},
-									{Type: corev1.PodInitialized, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(fakeTimeNowFn().Add(3000))},
-								},
-							},
-						},
-						timestamp: fakeTimeNowFn().Add(1000),
+		},
+		{
+			name: "update pod resources, cache will be updated",
+			pod: schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).
+				Req(map[corev1.ResourceName]string{corev1.ResourceCPU: "1", corev1.ResourceMemory: "4Gi"}).Obj(),
+
+			existingPods: []*corev1.Pod{
+				schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).
+					Conditions([]corev1.PodCondition{
+						{Type: corev1.PodScheduled, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(now.Add(1000))},
+					}).Obj(),
+			},
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, map[types.UID]*podAssignInfo{
+					"123456789": {
+						pod: schedulertesting.MakePod().UID("123456789").Namespace("default").Name("test").Node(node).Phase(corev1.PodRunning).
+							Req(map[corev1.ResourceName]string{corev1.ResourceCPU: "1", corev1.ResourceMemory: "4Gi"}).Obj(),
+						timestamp: now,
+						estimated: vectorizer.ToVec(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("1"),
+							corev1.ResourceMemory: resource.MustParse("4Gi"),
+						}),
 					},
-				},
+				}, n.podInfos)
+				v := vectorizer.ToVec(corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("1"),
+					corev1.ResourceMemory: resource.MustParse("4Gi"),
+				})
+				assert.Equal(t, v, n.nodeDelta)
+				assert.Equal(t, v, n.prodDelta)
+				assert.Equal(t, v, n.nodeEstimated)
+				s := sets.New(NamespacedName{Namespace: "default", Name: "test"})
+				assert.Equal(t, s, n.nodeDeltaPods)
+				assert.Equal(t, s, n.prodDeltaPods)
+				assert.Equal(t, s, n.nodeEstimatedPods)
 			},
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			preTimeNowFn := timeNowFn
-			defer func() {
-				timeNowFn = preTimeNowFn
-			}()
-			timeNowFn = fakeTimeNowFn
-			e, _ := estimator.NewDefaultEstimator(&config.LoadAwareSchedulingArgs{}, nil)
-			assignCache := newPodAssignCache(e, &config.LoadAwareSchedulingArgs{})
-			if len(tt.podInfoItems) > 0 {
-				assignCache.podInfoItems = tt.podInfoItems
+			args := &config.LoadAwareSchedulingArgs{EstimatedScalingFactors: map[corev1.ResourceName]int64{
+				corev1.ResourceCPU:    100,
+				corev1.ResourceMemory: 100,
+			}}
+			e, _ := estimator.NewDefaultEstimator(args, nil)
+			assignCache := newPodAssignCache(e, vectorizer, args)
+			assignCache.clock = clock.NewFakeClock(now)
+			assignCache.AddOrUpdateNodeMetric(m)
+			for _, pod := range tt.existingPods {
+				assignCache.OnAdd(pod, true)
 			}
 			assignCache.OnUpdate(nil, tt.pod)
-			assert.Equal(t, tt.wantCache, assignCache.podInfoItems)
+			actual, _ := assignCache.getNodeInfo(node)
+			tt.want(t, actual)
 		})
 	}
 }
 
 func TestPodAssignCache_OnDelete(t *testing.T) {
-	pod := &corev1.Pod{
-		ObjectMeta: metav1.ObjectMeta{
-			UID:       "123456789",
-			Namespace: "default",
-			Name:      "test",
-		},
-		Spec: corev1.PodSpec{
-			NodeName: "test-node",
-		},
-		Status: corev1.PodStatus{
-			Phase: corev1.PodFailed,
-		},
-	}
-	e, _ := estimator.NewDefaultEstimator(&config.LoadAwareSchedulingArgs{}, nil)
-	assignCache := newPodAssignCache(e, &config.LoadAwareSchedulingArgs{})
-	assignCache.podInfoItems = map[string]map[types.UID]*podAssignInfo{
-		"test-node": {
-			"123456789": &podAssignInfo{
-				pod: &corev1.Pod{
-					ObjectMeta: metav1.ObjectMeta{
-						UID:       "123456789",
-						Namespace: "default",
-						Name:      "test",
+	vectorizer := NewResourceVectorizer(corev1.ResourceCPU, corev1.ResourceMemory)
+	node := "test-node"
+	m := &slov1alpha1.NodeMetric{
+		ObjectMeta: metav1.ObjectMeta{Name: node},
+		Status: slov1alpha1.NodeMetricStatus{
+			NodeMetric: &slov1alpha1.NodeMetricInfo{
+				NodeUsage: slov1alpha1.ResourceMap{
+					ResourceList: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("70"),
+						corev1.ResourceMemory: resource.MustParse("280Gi"),
 					},
-					Spec: corev1.PodSpec{
-						NodeName: "test-node",
+				},
+			},
+			PodsMetric: []*slov1alpha1.PodMetricInfo{
+				{
+					Namespace: "default", Name: "prod-0", Priority: extension.PriorityProd,
+					PodUsage: slov1alpha1.ResourceMap{
+						ResourceList: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("50"),
+							corev1.ResourceMemory: resource.MustParse("200Gi"),
+						},
 					},
-					Status: corev1.PodStatus{
-						Phase: corev1.PodRunning,
+				},
+				{
+					Namespace: "default", Name: "prod-1", Priority: extension.PriorityProd,
+					PodUsage: slov1alpha1.ResourceMap{
+						ResourceList: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("1"),
+							corev1.ResourceMemory: resource.MustParse("6Gi"),
+						},
+					},
+				},
+				{
+					Namespace: "default", Name: "prod-2", Priority: extension.PriorityProd,
+					PodUsage: slov1alpha1.ResourceMap{
+						ResourceList: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("4"),
+							corev1.ResourceMemory: resource.MustParse("2Gi"),
+						},
+					},
+				},
+				{
+					Namespace: "default", Name: "prod-3", Priority: extension.PriorityMid,
+					PodUsage: slov1alpha1.ResourceMap{
+						ResourceList: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("3"),
+							corev1.ResourceMemory: resource.MustParse("6Gi"),
+						},
+					},
+				},
+				{
+					Namespace: "default", Name: "mid-1", Priority: extension.PriorityMid,
+					PodUsage: slov1alpha1.ResourceMap{
+						ResourceList: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("1"),
+							corev1.ResourceMemory: resource.MustParse("6Gi"),
+						},
+					},
+				},
+				{
+					Namespace: "default", Name: "mid-2", Priority: extension.PriorityMid,
+					PodUsage: slov1alpha1.ResourceMap{
+						ResourceList: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("4"),
+							corev1.ResourceMemory: resource.MustParse("2Gi"),
+						},
+					},
+				},
+				{
+					Namespace: "default", Name: "mid-3", Priority: extension.PriorityProd,
+					PodUsage: slov1alpha1.ResourceMap{
+						ResourceList: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("3"),
+							corev1.ResourceMemory: resource.MustParse("6Gi"),
+						},
 					},
 				},
-				timestamp: fakeTimeNowFn(),
 			},
 		},
 	}
-	assignCache.OnDelete(pod)
-	wantCache := map[string]map[types.UID]*podAssignInfo{}
-	assert.Equal(t, wantCache, assignCache.podInfoItems)
+	e, _ := estimator.NewDefaultEstimator(&config.LoadAwareSchedulingArgs{EstimatedScalingFactors: map[corev1.ResourceName]int64{
+		corev1.ResourceCPU:    50,
+		corev1.ResourceMemory: 50,
+	}}, nil)
+	assignCache := newPodAssignCache(e, vectorizer, &config.LoadAwareSchedulingArgs{})
+	assignCache.AddOrUpdateNodeMetric(m)
+	assignCache.OnAdd(schedulertesting.MakePod().UID("1").Namespace("default").Name("prod-1").Node(node).Phase(corev1.PodRunning).
+		Req(map[corev1.ResourceName]string{corev1.ResourceCPU: "4", corev1.ResourceMemory: "8Gi"}).Obj(), false)
+	assignCache.OnAdd(schedulertesting.MakePod().UID("2").Namespace("default").Name("prod-2").Node(node).Phase(corev1.PodRunning).
+		Req(map[corev1.ResourceName]string{corev1.ResourceCPU: "4", corev1.ResourceMemory: "8Gi"}).Obj(), false)
+	assignCache.OnAdd(schedulertesting.MakePod().UID("3").Namespace("default").Name("prod-3").Node(node).Phase(corev1.PodRunning).
+		Req(map[corev1.ResourceName]string{corev1.ResourceCPU: "4", corev1.ResourceMemory: "8Gi"}).Obj(), false)
+	assignCache.OnAdd(schedulertesting.MakePod().UID("4").Namespace("default").Name("mid-1").Node(node).Phase(corev1.PodRunning).Priority(extension.PriorityMidValueDefault).
+		Req(map[corev1.ResourceName]string{extension.MidCPU: "4k", extension.MidMemory: "8Gi"}).Obj(), false)
+	assignCache.OnAdd(schedulertesting.MakePod().UID("5").Namespace("default").Name("mid-2").Node(node).Phase(corev1.PodRunning).Priority(extension.PriorityMidValueDefault).
+		Req(map[corev1.ResourceName]string{extension.MidCPU: "4k", extension.MidMemory: "8Gi"}).Obj(), false)
+	assignCache.OnAdd(schedulertesting.MakePod().UID("6").Namespace("default").Name("mid-3").Node(node).Phase(corev1.PodRunning).Priority(extension.PriorityMidValueDefault).
+		Req(map[corev1.ResourceName]string{extension.MidCPU: "4k", extension.MidMemory: "8Gi"}).Obj(), false)
+	actual, _ := assignCache.getNodeInfo(node)
+	assert.Equal(t, vectorizer.ToVec(corev1.ResourceList{
+		corev1.ResourceCPU:    resource.MustParse("5"),
+		corev1.ResourceMemory: resource.MustParse("8Gi"),
+	}), actual.prodUsage)
+	assert.Equal(t, vectorizer.ToVec(corev1.ResourceList{
+		corev1.ResourceCPU:    resource.MustParse("2"),
+		corev1.ResourceMemory: resource.MustParse("4Gi"),
+	}), actual.nodeDelta)
+	assert.Equal(t, vectorizer.ToVec(corev1.ResourceList{
+		corev1.ResourceCPU:    resource.MustParse("3"),
+		corev1.ResourceMemory: resource.MustParse("6Gi"),
+	}), actual.prodDelta)
+	assert.Equal(t, vectorizer.ToVec(corev1.ResourceList{
+		corev1.ResourceCPU:    resource.MustParse("12"),
+		corev1.ResourceMemory: resource.MustParse("24Gi"),
+	}), actual.nodeEstimated)
+	assert.Equal(t, sets.New(
+		NamespacedName{Namespace: "default", Name: "prod-1"}, NamespacedName{Namespace: "default", Name: "prod-2"},
+		NamespacedName{Namespace: "default", Name: "mid-1"}, NamespacedName{Namespace: "default", Name: "mid-2"},
+	), actual.nodeDeltaPods)
+	assert.Equal(t, sets.New(
+		NamespacedName{Namespace: "default", Name: "prod-1"}, NamespacedName{Namespace: "default", Name: "prod-2"},
+		NamespacedName{Namespace: "default", Name: "prod-3"},
+	), actual.prodDeltaPods)
+	assert.Equal(t, sets.New(
+		NamespacedName{Namespace: "default", Name: "prod-1"}, NamespacedName{Namespace: "default", Name: "prod-2"},
+		NamespacedName{Namespace: "default", Name: "prod-3"}, NamespacedName{Namespace: "default", Name: "mid-1"},
+		NamespacedName{Namespace: "default", Name: "mid-2"}, NamespacedName{Namespace: "default", Name: "mid-3"},
+	), actual.nodeEstimatedPods)
+
+	assignCache.OnDelete(schedulertesting.MakePod().UID("1").Namespace("default").Name("prod-1").Node(node).Phase(corev1.PodFailed).
+		Req(map[corev1.ResourceName]string{corev1.ResourceCPU: "2", corev1.ResourceMemory: "8Gi"}).Obj())
+	assignCache.OnDelete(schedulertesting.MakePod().UID("2").Namespace("default").Name("prod-2").Node(node).Obj())
+	assignCache.OnDelete(schedulertesting.MakePod().UID("3").Namespace("default").Name("prod-3").Node(node).Obj())
+	assignCache.OnDelete(schedulertesting.MakePod().UID("4").Namespace("default").Name("mid-1").Node(node).Obj())
+	assignCache.OnDelete(schedulertesting.MakePod().UID("5").Namespace("default").Name("mid-2").Node(node).Obj())
+	assignCache.OnDelete(schedulertesting.MakePod().UID("6").Namespace("default").Name("mid-3").Node(node).Obj())
+	actual, _ = assignCache.getNodeInfo(node)
+	assert.Equal(t, 0, len(actual.podInfos))
+	assert.Equal(t, vectorizer.EmptyVec(), actual.nodeDelta)
+	assert.Equal(t, vectorizer.EmptyVec(), actual.prodDelta)
+	assert.Equal(t, vectorizer.EmptyVec(), actual.nodeEstimated)
+	s := sets.New[NamespacedName]()
+	assert.Equal(t, s, actual.nodeDeltaPods)
+	assert.Equal(t, s, actual.prodDeltaPods)
+	assert.Equal(t, s, actual.nodeEstimatedPods)
 }
 
 func TestShouldEstimatePodDeadline(t *testing.T) {
@@ -450,20 +524,211 @@ func TestShouldEstimatePodDeadline(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			preTimeNowFn := timeNowFn
-			defer func() {
-				timeNowFn = preTimeNowFn
-			}()
-			timeNowFn = fakeTimeNowFn
 			args := &config.LoadAwareSchedulingArgs{
 				EstimatedSecondsAfterPodScheduled: tt.estimatedSecondsAfterPodScheduled,
 				EstimatedSecondsAfterInitialized:  tt.estimatedSecondsAfterInitialized,
 				AllowCustomizeEstimation:          tt.allowCustomizeEstimation,
 			}
 			e, _ := estimator.NewDefaultEstimator(args, nil)
-			assignCache := newPodAssignCache(e, args)
+			assignCache := newPodAssignCache(e, NewResourceVectorizer(corev1.ResourceCPU, corev1.ResourceMemory), args)
 			actual := assignCache.shouldEstimatePodDeadline(tt.pod, now.Add(-time.Minute))
 			assert.Equal(t, tt.expected, actual)
 		})
 	}
 }
+
+func TestNodeMetric(t *testing.T) {
+	vectorizer := NewResourceVectorizer(corev1.ResourceCPU, corev1.ResourceMemory)
+	node := "test-node"
+	now := metav1.Now().Rfc3339Copy()
+	tests := []struct {
+		name         string
+		args         *config.LoadAwareSchedulingArgs
+		nodeMetric   *slov1alpha1.NodeMetric
+		existingPods []*corev1.Pod
+		want         func(*testing.T, *nodeInfo)
+	}{
+		{
+			name: "disable estimator",
+			args: &config.LoadAwareSchedulingArgs{},
+			nodeMetric: &slov1alpha1.NodeMetric{
+				Status: slov1alpha1.NodeMetricStatus{
+					NodeMetric: &slov1alpha1.NodeMetricInfo{
+						NodeUsage: slov1alpha1.ResourceMap{
+							ResourceList: corev1.ResourceList{
+								corev1.ResourceCPU:    resource.MustParse("72"),
+								corev1.ResourceMemory: resource.MustParse("280Gi"),
+							},
+						},
+					},
+					PodsMetric: []*slov1alpha1.PodMetricInfo{
+						nil,
+						{Name: "invalid"},
+						{
+							Namespace: "default", Name: "prod-1", Priority: extension.PriorityProd,
+							PodUsage: slov1alpha1.ResourceMap{
+								ResourceList: corev1.ResourceList{
+									corev1.ResourceCPU:    resource.MustParse("50"),
+									corev1.ResourceMemory: resource.MustParse("200Gi"),
+								},
+							},
+						},
+						{
+							Namespace: "default", Name: "mid-1", Priority: extension.PriorityMid,
+							PodUsage: slov1alpha1.ResourceMap{
+								ResourceList: corev1.ResourceList{
+									corev1.ResourceCPU:    resource.MustParse("20"),
+									corev1.ResourceMemory: resource.MustParse("75Gi"),
+								},
+							},
+						},
+					},
+				},
+			},
+			existingPods: []*corev1.Pod{
+				schedulertesting.MakePod().Namespace("default").UID("1").Name("prod-1").Node(node).Phase(corev1.PodRunning).
+					Req(map[corev1.ResourceName]string{corev1.ResourceCPU: "40", corev1.ResourceMemory: "160Gi"}).Obj(),
+				schedulertesting.MakePod().Namespace("default").UID("2").Name("mid-1").Node(node).Phase(corev1.PodRunning).Priority(extension.PriorityMidValueDefault).
+					Req(map[corev1.ResourceName]string{extension.MidCPU: "4k", extension.MidMemory: "8Gi"}).Obj(),
+			},
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, vectorizer.ToVec(corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("50"),
+					corev1.ResourceMemory: resource.MustParse("200Gi"),
+				}), n.prodUsage)
+			},
+		},
+		{
+			name: "enable prod usage include sys",
+			args: &config.LoadAwareSchedulingArgs{ProdUsageIncludeSys: true},
+			nodeMetric: &slov1alpha1.NodeMetric{
+				Spec: slov1alpha1.NodeMetricSpec{
+					CollectPolicy: &slov1alpha1.NodeMetricCollectPolicy{
+						ReportIntervalSeconds: ptr.To[int64](180),
+					},
+				},
+				Status: slov1alpha1.NodeMetricStatus{
+					UpdateTime: &now,
+					NodeMetric: &slov1alpha1.NodeMetricInfo{
+						NodeUsage: slov1alpha1.ResourceMap{
+							ResourceList: corev1.ResourceList{
+								corev1.ResourceCPU:    resource.MustParse("2"),
+								corev1.ResourceMemory: resource.MustParse("4Gi"),
+							},
+						},
+						SystemUsage: slov1alpha1.ResourceMap{
+							ResourceList: corev1.ResourceList{
+								corev1.ResourceCPU:    resource.MustParse("2"),
+								corev1.ResourceMemory: resource.MustParse("4Gi"),
+							},
+						},
+						AggregatedNodeUsages: []slov1alpha1.AggregatedUsage{
+							{
+								Duration: metav1.Duration{Duration: time.Minute},
+								Usage: map[extension.AggregationType]slov1alpha1.ResourceMap{
+									extension.AVG: {
+										ResourceList: corev1.ResourceList{
+											corev1.ResourceCPU:    resource.MustParse("1"),
+											corev1.ResourceMemory: resource.MustParse("2Gi"),
+										},
+									},
+									extension.P90: {
+										ResourceList: corev1.ResourceList{
+											corev1.ResourceCPU:    resource.MustParse("2"),
+											corev1.ResourceMemory: resource.MustParse("4Gi"),
+										},
+									},
+								},
+							},
+							{
+								Duration: metav1.Duration{Duration: time.Hour},
+								Usage: map[extension.AggregationType]slov1alpha1.ResourceMap{
+									extension.AVG: {
+										ResourceList: corev1.ResourceList{
+											corev1.ResourceCPU:    resource.MustParse("500m"),
+											corev1.ResourceMemory: resource.MustParse("1Gi"),
+										},
+									},
+									extension.P90: {
+										ResourceList: corev1.ResourceList{
+											corev1.ResourceCPU:    resource.MustParse("1500m"),
+											corev1.ResourceMemory: resource.MustParse("3Gi"),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			want: func(t *testing.T, n *nodeInfo) {
+				assert.Equal(t, now.Time, n.updateTime)
+				assert.Equal(t, 180*time.Second, n.reportInterval)
+				assert.Equal(t, vectorizer.ToVec(corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("2"),
+					corev1.ResourceMemory: resource.MustParse("4Gi"),
+				}), n.nodeUsage)
+				assert.Equal(t, vectorizer.ToVec(corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("2"),
+					corev1.ResourceMemory: resource.MustParse("4Gi"),
+				}), n.prodUsage)
+				assert.Equal(t, map[aggUsageKey]ResourceVector{
+					{Type: extension.AVG}: vectorizer.ToVec(corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("500m"),
+						corev1.ResourceMemory: resource.MustParse("1Gi"),
+					}),
+					{Type: extension.P90}: vectorizer.ToVec(corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("1500m"),
+						corev1.ResourceMemory: resource.MustParse("3Gi"),
+					}),
+					{Type: extension.AVG, Duration: time.Minute}: vectorizer.ToVec(corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("1"),
+						corev1.ResourceMemory: resource.MustParse("2Gi"),
+					}),
+					{Type: extension.P90, Duration: time.Minute}: vectorizer.ToVec(corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("2"),
+						corev1.ResourceMemory: resource.MustParse("4Gi"),
+					}),
+					{Type: extension.AVG, Duration: time.Hour}: vectorizer.ToVec(corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("500m"),
+						corev1.ResourceMemory: resource.MustParse("1Gi"),
+					}),
+					{Type: extension.P90, Duration: time.Hour}: vectorizer.ToVec(corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("1500m"),
+						corev1.ResourceMemory: resource.MustParse("3Gi"),
+					}),
+				}, n.aggUsages)
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.nodeMetric.Name = node
+			if tt.args == nil {
+				tt.args = &config.LoadAwareSchedulingArgs{EstimatedScalingFactors: map[corev1.ResourceName]int64{
+					corev1.ResourceCPU:    100,
+					corev1.ResourceMemory: 100,
+				}}
+			}
+			e, _ := estimator.NewDefaultEstimator(tt.args, nil)
+			assignCache := newPodAssignCache(e, vectorizer, tt.args)
+			for _, pod := range tt.existingPods {
+				assignCache.OnAdd(pod, true)
+			}
+			// test add node metrics
+			assignCache.AddOrUpdateNodeMetric(tt.nodeMetric)
+			actual, _ := assignCache.getNodeInfo(node)
+			tt.want(t, actual)
+			// test delete all pods
+			for _, pod := range tt.existingPods {
+				assignCache.OnDelete(pod)
+			}
+			// test delete node metrics
+			assignCache.DeleteNodeMetric(node)
+			_, ok := assignCache.getNodeInfo(node)
+			assert.Equal(t, false, ok)
+			_, _, _, err := assignCache.GetNodeMetricAndEstimatedOfExisting(node, false, metav1.Duration{}, "", false)
+			assert.True(t, errors.IsNotFound(err))
+		})
+	}
+}
diff --git a/pkg/scheduler/plugins/loadaware/service_test.go b/pkg/scheduler/plugins/loadaware/service_test.go
index cb22bda5..c4817603 100644
--- a/pkg/scheduler/plugins/loadaware/service_test.go
+++ b/pkg/scheduler/plugins/loadaware/service_test.go
@@ -28,23 +28,29 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/uuid"
+	clock "k8s.io/utils/clock/testing"
 
+	slov1alpha1 "github.com/koordinator-sh/koordinator/apis/slo/v1alpha1"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/apis/config"
 	"github.com/koordinator-sh/koordinator/pkg/scheduler/plugins/loadaware/estimator"
 )
 
 func TestQueryNode(t *testing.T) {
 	t.Run("test", func(t *testing.T) {
-
-		preTimeNowFn := timeNowFn
-		defer func() {
-			timeNowFn = preTimeNowFn
-		}()
-		timeNowFn = fakeTimeNowFn
-		e, _ := estimator.NewDefaultEstimator(&config.LoadAwareSchedulingArgs{}, nil)
+		now := metav1.Now().Rfc3339Copy().Time
+		args := &config.LoadAwareSchedulingArgs{
+			EstimatedScalingFactors: map[corev1.ResourceName]int64{
+				corev1.ResourceCPU:    100,
+				corev1.ResourceMemory: 100,
+			},
+		}
+		v := NewResourceVectorizerFromArgs(args)
+		e, _ := estimator.NewDefaultEstimator(args, nil)
 		pl := &Plugin{
-			podAssignCache: newPodAssignCache(e, &config.LoadAwareSchedulingArgs{}),
+			vectorizer:     v,
+			podAssignCache: newPodAssignCache(e, v, args),
 		}
+		pl.podAssignCache.clock = clock.NewFakeClock(now)
 		testNodeName := "test-node"
 		pendingPod := &corev1.Pod{
 			ObjectMeta: metav1.ObjectMeta{
@@ -117,6 +123,7 @@ func TestQueryNode(t *testing.T) {
 			},
 		}
 		pl.podAssignCache.unAssign(testNodeName, terminatedPod)
+		pl.podAssignCache.AddOrUpdateNodeMetric(&slov1alpha1.NodeMetric{ObjectMeta: metav1.ObjectMeta{Name: testNodeName}})
 
 		// got unknown node
 		engine := gin.Default()
@@ -142,10 +149,17 @@ func TestQueryNode(t *testing.T) {
 		expectedResp = &NodeAssignInfoData{
 			Pods: []PodAssignInfoData{
 				{
-					Timestamp: timeNowFn(),
+					Timestamp: now,
 					Pod:       assignedPod,
 				},
 			},
+			ProdUsage:         corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("0"), corev1.ResourceMemory: resource.MustParse("0")},
+			NodeDelta:         corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("4"), corev1.ResourceMemory: resource.MustParse("8Gi")},
+			ProdDelta:         corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("4"), corev1.ResourceMemory: resource.MustParse("8Gi")},
+			NodeEstimated:     corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("4"), corev1.ResourceMemory: resource.MustParse("8Gi")},
+			NodeDeltaPods:     []string{"test-pod-1"},
+			ProdDeltaPods:     []string{"test-pod-1"},
+			NodeEstimatedPods: []string{"test-pod-1"},
 		}
 		assert.Equal(t, expectedResp, got)
 
@@ -158,7 +172,12 @@ func TestQueryNode(t *testing.T) {
 		got = &NodeAssignInfoData{}
 		err = json.NewDecoder(w.Result().Body).Decode(got)
 		assert.NoError(t, err)
-		expectedResp = &NodeAssignInfoData{}
+		expectedResp = &NodeAssignInfoData{
+			ProdUsage:     corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("0"), corev1.ResourceMemory: resource.MustParse("0")},
+			NodeDelta:     corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("0"), corev1.ResourceMemory: resource.MustParse("0")},
+			ProdDelta:     corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("0"), corev1.ResourceMemory: resource.MustParse("0")},
+			NodeEstimated: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("0"), corev1.ResourceMemory: resource.MustParse("0")},
+		}
 		assert.Equal(t, expectedResp, got)
 	})
 }