nodeutilization: NodeUtilization: make pod utilization extraction configurable

pkg/descheduler/node/node.go

@@ -218,7 +218,12 @@ func fitsRequest(nodeIndexer podutil.GetPodsAssignedToNodeFunc, pod *v1.Pod, nod
 		resourceNames = append(resourceNames, name)
 	}
 
-	availableResources, err := nodeAvailableResources(nodeIndexer, node, resourceNames)
+	availableResources, err := nodeAvailableResources(nodeIndexer, node, resourceNames,
+		func(pod *v1.Pod) (v1.ResourceList, error) {
+			req, _ := utils.PodRequestsAndLimits(pod)
+			return req, nil
+		},
+	)
 	if err != nil {
 		return false, err
 	}
@@ -239,12 +244,15 @@ func fitsRequest(nodeIndexer podutil.GetPodsAssignedToNodeFunc, pod *v1.Pod, nod
 }
 
 // nodeAvailableResources returns resources mapped to the quanitity available on the node.
-func nodeAvailableResources(nodeIndexer podutil.GetPodsAssignedToNodeFunc, node *v1.Node, resourceNames []v1.ResourceName) (map[v1.ResourceName]*resource.Quantity, error) {
+func nodeAvailableResources(nodeIndexer podutil.GetPodsAssignedToNodeFunc, node *v1.Node, resourceNames []v1.ResourceName, podUtilization podutil.PodUtilizationFnc) (map[v1.ResourceName]*resource.Quantity, error) {
 	podsOnNode, err := podutil.ListPodsOnANode(node.Name, nodeIndexer, nil)
 	if err != nil {
 		return nil, err
 	}
-	nodeUtilization := NodeUtilization(podsOnNode, resourceNames)
+	nodeUtilization, err := NodeUtilization(podsOnNode, resourceNames, podUtilization)
+	if err != nil {
+		return nil, err
+	}
 	remainingResources := map[v1.ResourceName]*resource.Quantity{
 		v1.ResourceCPU:    resource.NewMilliQuantity(node.Status.Allocatable.Cpu().MilliValue()-nodeUtilization[v1.ResourceCPU].MilliValue(), resource.DecimalSI),
 		v1.ResourceMemory: resource.NewQuantity(node.Status.Allocatable.Memory().Value()-nodeUtilization[v1.ResourceMemory].Value(), resource.BinarySI),
@@ -265,31 +273,34 @@ func nodeAvailableResources(nodeIndexer podutil.GetPodsAssignedToNodeFunc, node
 }
 
 // NodeUtilization returns the resources requested by the given pods. Only resources supplied in the resourceNames parameter are calculated.
-func NodeUtilization(pods []*v1.Pod, resourceNames []v1.ResourceName) map[v1.ResourceName]*resource.Quantity {
+func NodeUtilization(pods []*v1.Pod, resourceNames []v1.ResourceName, podUtilization podutil.PodUtilizationFnc) (map[v1.ResourceName]*resource.Quantity, error) {
-	totalReqs := map[v1.ResourceName]*resource.Quantity{
+	totalUtilization := map[v1.ResourceName]*resource.Quantity{
 		v1.ResourceCPU:    resource.NewMilliQuantity(0, resource.DecimalSI),
 		v1.ResourceMemory: resource.NewQuantity(0, resource.BinarySI),
 		v1.ResourcePods:   resource.NewQuantity(int64(len(pods)), resource.DecimalSI),
 	}
 	for _, name := range resourceNames {
 		if !IsBasicResource(name) {
-			totalReqs[name] = resource.NewQuantity(0, resource.DecimalSI)
+			totalUtilization[name] = resource.NewQuantity(0, resource.DecimalSI)
 		}
 	}
 
 	for _, pod := range pods {
-		req, _ := utils.PodRequestsAndLimits(pod)
+		podUtil, err := podUtilization(pod)
+		if err != nil {
+			return nil, err
+		}
 		for _, name := range resourceNames {
-			quantity, ok := req[name]
+			quantity, ok := podUtil[name]
 			if ok && name != v1.ResourcePods {
 				// As Quantity.Add says: Add adds the provided y quantity to the current value. If the current value is zero,
 				// the format of the quantity will be updated to the format of y.
-				totalReqs[name].Add(quantity)
+				totalUtilization[name].Add(quantity)
 			}
 		}
 	}
 
-	return totalReqs
+	return totalUtilization, nil
 }
 
 // IsBasicResource checks if resource is basic native.

pkg/descheduler/pod/pods.go

@@ -39,6 +39,9 @@ type FilterFunc func(*v1.Pod) bool
 // as input and returns the pods that assigned to the node.
 type GetPodsAssignedToNodeFunc func(string, FilterFunc) ([]*v1.Pod, error)
 
+// PodUtilizationFnc is a function for getting pod's utilization. E.g. requested resources of utilization from metrics.
+type PodUtilizationFnc func(pod *v1.Pod) (v1.ResourceList, error)
+
 // WrapFilterFuncs wraps a set of FilterFunc in one.
 func WrapFilterFuncs(filters ...FilterFunc) FilterFunc {
 	return func(pod *v1.Pod) bool {

pkg/framework/plugins/nodeutilization/highnodeutilization.go

@@ -94,9 +94,22 @@ func (h *HighNodeUtilization) Name() string {
 
 // Balance extension point implementation for the plugin
 func (h *HighNodeUtilization) Balance(ctx context.Context, nodes []*v1.Node) *frameworktypes.Status {
+	nodeUsage, err := getNodeUsage(nodes, h.resourceNames, h.handle.GetPodsAssignedToNodeFunc())
+	if err != nil {
+		return &frameworktypes.Status{
+			Err: fmt.Errorf("error getting node usage: %v", err),
+		}
+	}
+	thresholds, err := getNodeThresholds(nodes, h.args.Thresholds, h.targetThresholds, h.resourceNames, h.handle.GetPodsAssignedToNodeFunc(), false)
+	if err != nil {
+		return &frameworktypes.Status{
+			Err: fmt.Errorf("error getting node thresholds: %v", err),
+		}
+	}
+
 	sourceNodes, highNodes := classifyNodes(
-		getNodeUsage(nodes, h.resourceNames, h.handle.GetPodsAssignedToNodeFunc()),
+		nodeUsage,
-		getNodeThresholds(nodes, h.args.Thresholds, h.targetThresholds, h.resourceNames, h.handle.GetPodsAssignedToNodeFunc(), false),
+		thresholds,
 		func(node *v1.Node, usage NodeUsage, threshold NodeThresholds) bool {
 			return isNodeWithLowUtilization(usage, threshold.lowResourceThreshold)
 		},

pkg/framework/plugins/nodeutilization/lownodeutilization.go

@@ -102,9 +102,22 @@ func (l *LowNodeUtilization) Name() string {
 
 // Balance extension point implementation for the plugin
 func (l *LowNodeUtilization) Balance(ctx context.Context, nodes []*v1.Node) *frameworktypes.Status {
+	nodeUsage, err := getNodeUsage(nodes, l.resourceNames, l.handle.GetPodsAssignedToNodeFunc())
+	if err != nil {
+		return &frameworktypes.Status{
+			Err: fmt.Errorf("error getting node usage: %v", err),
+		}
+	}
+	thresholds, err := getNodeThresholds(nodes, l.args.Thresholds, l.args.TargetThresholds, l.resourceNames, l.handle.GetPodsAssignedToNodeFunc(), l.args.UseDeviationThresholds)
+	if err != nil {
+		return &frameworktypes.Status{
+			Err: fmt.Errorf("error getting node thresholds: %v", err),
+		}
+	}
+
 	lowNodes, sourceNodes := classifyNodes(
-		getNodeUsage(nodes, l.resourceNames, l.handle.GetPodsAssignedToNodeFunc()),
+		nodeUsage,
-		getNodeThresholds(nodes, l.args.Thresholds, l.args.TargetThresholds, l.resourceNames, l.handle.GetPodsAssignedToNodeFunc(), l.args.UseDeviationThresholds),
+		thresholds,
 		// The node has to be schedulable (to be able to move workload there)
 		func(node *v1.Node, usage NodeUsage, threshold NodeThresholds) bool {
 			if nodeutil.IsNodeUnschedulable(node) {

pkg/framework/plugins/nodeutilization/nodeutilization.go

@@ -80,12 +80,16 @@ func getNodeThresholds(
 	resourceNames []v1.ResourceName,
 	getPodsAssignedToNode podutil.GetPodsAssignedToNodeFunc,
 	useDeviationThresholds bool,
-) map[string]NodeThresholds {
+) (map[string]NodeThresholds, error) {
 	nodeThresholdsMap := map[string]NodeThresholds{}
 
 	averageResourceUsagePercent := api.ResourceThresholds{}
 	if useDeviationThresholds {
-		averageResourceUsagePercent = averageNodeBasicresources(nodes, getPodsAssignedToNode, resourceNames)
+		usage, err := averageNodeBasicresources(nodes, getPodsAssignedToNode, resourceNames)
+		if err != nil {
+			return nil, err
+		}
+		averageResourceUsagePercent = usage
 	}
 
 	for _, node := range nodes {
@@ -116,14 +120,14 @@ func getNodeThresholds(
 		}
 
 	}
-	return nodeThresholdsMap
+	return nodeThresholdsMap, nil
 }
 
 func getNodeUsage(
 	nodes []*v1.Node,
 	resourceNames []v1.ResourceName,
 	getPodsAssignedToNode podutil.GetPodsAssignedToNodeFunc,
-) []NodeUsage {
+) ([]NodeUsage, error) {
 	var nodeUsageList []NodeUsage
 
 	for _, node := range nodes {
@@ -133,14 +137,22 @@ func getNodeUsage(
 			continue
 		}
 
+		nodeUsage, err := nodeutil.NodeUtilization(pods, resourceNames, func(pod *v1.Pod) (v1.ResourceList, error) {
+			req, _ := utils.PodRequestsAndLimits(pod)
+			return req, nil
+		})
+		if err != nil {
+			return nil, err
+		}
+
 		nodeUsageList = append(nodeUsageList, NodeUsage{
 			node:    node,
-			usage:   nodeutil.NodeUtilization(pods, resourceNames),
+			usage:   nodeUsage,
 			allPods: pods,
 		})
 	}
 
-	return nodeUsageList
+	return nodeUsageList, nil
 }
 
 func resourceThreshold(nodeCapacity v1.ResourceList, resourceName v1.ResourceName, threshold api.Percentage) *resource.Quantity {
@@ -451,7 +463,7 @@ func classifyPods(pods []*v1.Pod, filter func(pod *v1.Pod) bool) ([]*v1.Pod, []*
 	return nonRemovablePods, removablePods
 }
 
-func averageNodeBasicresources(nodes []*v1.Node, getPodsAssignedToNode podutil.GetPodsAssignedToNodeFunc, resourceNames []v1.ResourceName) api.ResourceThresholds {
+func averageNodeBasicresources(nodes []*v1.Node, getPodsAssignedToNode podutil.GetPodsAssignedToNodeFunc, resourceNames []v1.ResourceName) (api.ResourceThresholds, error) {
 	total := api.ResourceThresholds{}
 	average := api.ResourceThresholds{}
 	numberOfNodes := len(nodes)
@@ -461,7 +473,14 @@ func averageNodeBasicresources(nodes []*v1.Node, getPodsAssignedToNode podutil.G
 			numberOfNodes--
 			continue
 		}
-		usage := nodeutil.NodeUtilization(pods, resourceNames)
+		usage, err := nodeutil.NodeUtilization(pods, resourceNames, func(pod *v1.Pod) (v1.ResourceList, error) {
+			req, _ := utils.PodRequestsAndLimits(pod)
+			return req, nil
+		})
+		if err != nil {
+			return nil, err
+		}
+
 		nodeCapacity := node.Status.Capacity
 		if len(node.Status.Allocatable) > 0 {
 			nodeCapacity = node.Status.Allocatable
@@ -478,5 +497,5 @@ func averageNodeBasicresources(nodes []*v1.Node, getPodsAssignedToNode podutil.G
 	for resource, value := range total {
 		average[resource] = value / api.Percentage(numberOfNodes)
 	}
-	return average
+	return average, nil
 }