Addressing review comments

pkg/api/v1alpha2/types.go

@@ -74,5 +74,5 @@ type PluginSet struct {
 type MetricsCollector struct {
 	// Enabled metrics collection from kubernetes metrics.
 	// Later, the collection can be extended to other providers.
-	Enabled bool `json:"enabled"`
+	Enabled bool `json:"enabled,omitempty"`
 }

pkg/descheduler/descheduler.go

@@ -162,7 +162,7 @@ func newDescheduler(rs *options.DeschedulerServer, deschedulerPolicy *api.Desche
 		if deschedulerPolicy.NodeSelector != nil {
 			nodeSelector = *deschedulerPolicy.NodeSelector
 		}
-		metricsCollector = metricscollector.NewMetricsCollector(rs.Client, rs.MetricsClient, nodeSelector)
+		metricsCollector = metricscollector.NewMetricsCollector(rs.Client.CoreV1().Nodes(), rs.MetricsClient, nodeSelector)
 	}
 
 	return &descheduler{

pkg/descheduler/metricscollector/metricscollector.go

@@ -27,7 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/klog/v2"
 	metricsclient "k8s.io/metrics/pkg/client/clientset/versioned"
 	utilptr "k8s.io/utils/ptr"
@@ -38,7 +38,7 @@ const (
 )
 
 type MetricsCollector struct {
-	clientset        kubernetes.Interface
+	nodeLister       corev1.NodeInterface
 	metricsClientset metricsclient.Interface
 	nodeSelector     string
 
@@ -49,9 +49,9 @@ type MetricsCollector struct {
 	hasSynced bool
 }
 
-func NewMetricsCollector(clientset kubernetes.Interface, metricsClientset metricsclient.Interface, nodeSelector string) *MetricsCollector {
+func NewMetricsCollector(nodeLister corev1.NodeInterface, metricsClientset metricsclient.Interface, nodeSelector string) *MetricsCollector {
 	return &MetricsCollector{
-		clientset:        clientset,
+		nodeLister:       nodeLister,
 		metricsClientset: metricsClientset,
 		nodeSelector:     nodeSelector,
 		nodes:            make(map[string]map[v1.ResourceName]*resource.Quantity),
@@ -65,7 +65,7 @@ func (mc *MetricsCollector) Run(ctx context.Context) {
 }
 
 func weightedAverage(prevValue, value int64) int64 {
-	return int64(math.Floor(beta*float64(prevValue) + (1-beta)*float64(value)))
+	return int64(math.Round(beta*float64(prevValue) + (1-beta)*float64(value)))
 }
 
 func (mc *MetricsCollector) AllNodesUsage() (map[string]map[v1.ResourceName]*resource.Quantity, error) {
@@ -88,7 +88,7 @@ func (mc *MetricsCollector) NodeUsage(node *v1.Node) (map[v1.ResourceName]*resou
 	defer mc.mu.RUnlock()
 
 	if _, exists := mc.nodes[node.Name]; !exists {
-		klog.V(4).Infof("unable to find node %q in the collected metrics", node.Name)
+		klog.V(4).InfoS("unable to find node in the collected metrics", "node", node.Name)
 		return nil, fmt.Errorf("unable to find node %q in the collected metrics", node.Name)
 	}
 	return map[v1.ResourceName]*resource.Quantity{
@@ -108,7 +108,7 @@ func (mc *MetricsCollector) MetricsClient() metricsclient.Interface {
 func (mc *MetricsCollector) Collect(ctx context.Context) error {
 	mc.mu.Lock()
 	defer mc.mu.Unlock()
-	nodes, err := mc.clientset.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{LabelSelector: mc.nodeSelector})
+	nodes, err := mc.nodeLister.List(context.TODO(), metav1.ListOptions{LabelSelector: mc.nodeSelector})
 	if err != nil {
 		return fmt.Errorf("unable to list nodes: %v", err)
 	}
@@ -128,11 +128,13 @@ func (mc *MetricsCollector) Collect(ctx context.Context) error {
 			}
 		} else {
 			// get MilliValue to reduce loss of precision
-			mc.nodes[node.Name][v1.ResourceCPU].SetMilli(
-				weightedAverage(mc.nodes[node.Name][v1.ResourceCPU].MilliValue(), metrics.Usage.Cpu().MilliValue()),
+			mc.nodes[node.Name][v1.ResourceCPU].SetScaled(
+				weightedAverage(mc.nodes[node.Name][v1.ResourceCPU].ScaledValue(resource.Micro), metrics.Usage.Cpu().ScaledValue(resource.Micro)),
+				resource.Micro,
 			)
-			mc.nodes[node.Name][v1.ResourceMemory].SetMilli(
-				weightedAverage(mc.nodes[node.Name][v1.ResourceMemory].MilliValue(), metrics.Usage.Memory().MilliValue()),
+			mc.nodes[node.Name][v1.ResourceMemory].SetScaled(
+				weightedAverage(mc.nodes[node.Name][v1.ResourceMemory].ScaledValue(resource.Micro), metrics.Usage.Memory().ScaledValue(resource.Micro)),
+				resource.Micro,
 			)
 		}
 	}

pkg/descheduler/metricscollector/metricscollector_test.go

@@ -18,6 +18,7 @@ package metricscollector
 
 import (
 	"context"
+	"math"
 	"testing"
 
 	v1 "k8s.io/api/core/v1"
@@ -54,7 +55,7 @@ func TestMetricsCollector(t *testing.T) {
 	metricsClientset.Tracker().Create(gvr, n3metrics, "")
 
 	t.Logf("Set initial node cpu usage to 1400")
-	collector := NewMetricsCollector(clientset, metricsClientset, "")
+	collector := NewMetricsCollector(clientset.CoreV1().Nodes(), metricsClientset, "")
 	collector.Collect(context.TODO())
 	nodesUsage, _ := collector.NodeUsage(n2)
 	checkCpuNodeUsage(t, nodesUsage, 1400)
@@ -70,7 +71,7 @@ func TestMetricsCollector(t *testing.T) {
 	allnodesUsage, _ = collector.AllNodesUsage()
 	checkCpuNodeUsage(t, allnodesUsage[n2.Name], 1310)
 
-	t.Logf("Set current node cpu usage to 500")
+	t.Logf("Set current node cpu usage to 900")
 	n2metrics.Usage[v1.ResourceCPU] = *resource.NewMilliQuantity(900, resource.DecimalSI)
 	metricsClientset.Tracker().Update(gvr, n2metrics, "")
 	collector.Collect(context.TODO())
@@ -79,3 +80,47 @@ func TestMetricsCollector(t *testing.T) {
 	allnodesUsage, _ = collector.AllNodesUsage()
 	checkCpuNodeUsage(t, allnodesUsage[n2.Name], 1269)
 }
+
+func TestMetricsCollectorConvergence(t *testing.T) {
+	gvr := schema.GroupVersionResource{Group: "metrics.k8s.io", Version: "v1beta1", Resource: "nodes"}
+
+	n1 := test.BuildTestNode("n1", 2000, 3000, 10, nil)
+	n2 := test.BuildTestNode("n2", 2000, 3000, 10, nil)
+	n3 := test.BuildTestNode("n3", 2000, 3000, 10, nil)
+
+	n1metrics := test.BuildNodeMetrics("n1", 400, 1714978816)
+	n2metrics := test.BuildNodeMetrics("n2", 1400, 1714978816)
+	n3metrics := test.BuildNodeMetrics("n3", 300, 1714978816)
+
+	clientset := fakeclientset.NewSimpleClientset(n1, n2, n3)
+	metricsClientset := fakemetricsclient.NewSimpleClientset()
+	metricsClientset.Tracker().Create(gvr, n1metrics, "")
+	metricsClientset.Tracker().Create(gvr, n2metrics, "")
+	metricsClientset.Tracker().Create(gvr, n3metrics, "")
+
+	t.Logf("Set initial node cpu usage to 1400")
+	collector := NewMetricsCollector(clientset.CoreV1().Nodes(), metricsClientset, "")
+	collector.Collect(context.TODO())
+	nodesUsage, _ := collector.NodeUsage(n2)
+	checkCpuNodeUsage(t, nodesUsage, 1400)
+	allnodesUsage, _ := collector.AllNodesUsage()
+	checkCpuNodeUsage(t, allnodesUsage[n2.Name], 1400)
+
+	t.Logf("Set current node cpu usage to 900 and wait until it converges to it")
+	n2metrics.Usage[v1.ResourceCPU] = *resource.NewMilliQuantity(900, resource.DecimalSI)
+	metricsClientset.Tracker().Update(gvr, n2metrics, "")
+	converged := false
+	for i := 0; i < 100; i++ {
+		collector.Collect(context.TODO())
+		nodesUsage, _ = collector.NodeUsage(n2)
+		if math.Abs(float64(900-nodesUsage[v1.ResourceCPU].MilliValue())) < 2 {
+			t.Logf("Node usage converged to 900+-1")
+			converged = true
+			break
+		}
+		t.Logf("The current node usage: %v", nodesUsage[v1.ResourceCPU].MilliValue())
+	}
+	if !converged {
+		t.Fatalf("The node usage did not converged to 900+-1")
+	}
+}

pkg/framework/plugins/nodeutilization/lownodeutilization_test.go

@@ -1136,20 +1136,23 @@ func TestLowNodeUtilization(t *testing.T) {
 					objs = append(objs, pod)
 				}
 
-				metricsClientset := fakemetricsclient.NewSimpleClientset()
-				for _, nodemetrics := range tc.nodemetricses {
-					metricsClientset.Tracker().Create(nodesgvr, nodemetrics, "")
-				}
-				for _, podmetrics := range tc.podmetricses {
-					metricsClientset.Tracker().Create(podsgvr, podmetrics, podmetrics.Namespace)
-				}
-
 				fakeClient := fake.NewSimpleClientset(objs...)
 
-				collector := metricscollector.NewMetricsCollector(fakeClient, metricsClientset, "")
-				err := collector.Collect(ctx)
-				if err != nil {
-					t.Fatalf("unable to collect metrics: %v", err)
+				var collector *metricscollector.MetricsCollector
+				if len(tc.nodemetricses) > 0 || len(tc.podmetricses) > 0 {
+					metricsClientset := fakemetricsclient.NewSimpleClientset()
+					for _, nodemetrics := range tc.nodemetricses {
+						metricsClientset.Tracker().Create(nodesgvr, nodemetrics, "")
+					}
+					for _, podmetrics := range tc.podmetricses {
+						metricsClientset.Tracker().Create(podsgvr, podmetrics, podmetrics.Namespace)
+					}
+
+					collector = metricscollector.NewMetricsCollector(fakeClient.CoreV1().Nodes(), metricsClientset, "")
+					err := collector.Collect(ctx)
+					if err != nil {
+						t.Fatalf("unable to collect metrics: %v", err)
+					}
 				}
 
 				podsForEviction := make(map[string]struct{})

pkg/framework/plugins/nodeutilization/usageclients.go

@@ -181,6 +181,12 @@ func (client *actualUsageClient) sync(nodes []*v1.Node) error {
 		if !ok {
 			return fmt.Errorf("unable to find node %q in the collected metrics", node.Name)
 		}
+		if _, exists := nodeUsage[v1.ResourceCPU]; !exists {
+			return fmt.Errorf("unable to find %q resource for collected %q node metric", v1.ResourceCPU, node.Name)
+		}
+		if _, exists := nodeUsage[v1.ResourceMemory]; !exists {
+			return fmt.Errorf("unable to find %q resource for collected %q node metric", v1.ResourceMemory, node.Name)
+		}
 		nodeUsage[v1.ResourcePods] = resource.NewQuantity(int64(len(pods)), resource.DecimalSI)
 
 		// store the snapshot of pods from the same (or the closest) node utilization computation

pkg/framework/plugins/nodeutilization/usageclients_test.go

@@ -112,7 +112,7 @@ func TestActualUsageClient(t *testing.T) {
 	sharedInformerFactory.Start(ctx.Done())
 	sharedInformerFactory.WaitForCacheSync(ctx.Done())
 
-	collector := metricscollector.NewMetricsCollector(clientset, metricsClientset, "")
+	collector := metricscollector.NewMetricsCollector(clientset.CoreV1().Nodes(), metricsClientset, "")
 
 	usageClient := newActualUsageClient(
 		resourceNames,