utils: add metric server client (#1881)

Signed-off-by: 佑祎 <zzw261520@alibaba-inc.com>

.licenseignore

@@ -32,3 +32,4 @@ test/e2e/e2e_test.go
 test/e2e/suites.go
 test/utils/image
 pkg/util/histogram
+pkg/util/metricserver

go.mod

@@ -55,6 +55,7 @@ require (
 	k8s.io/kubectl v0.22.6
 	k8s.io/kubelet v0.22.6
 	k8s.io/kubernetes v1.24.15
+	k8s.io/metrics v0.24.15
 	k8s.io/utils v0.0.0-20221128185143-99ec85e7a448
 	sigs.k8s.io/controller-runtime v0.12.3
 	sigs.k8s.io/controller-runtime/tools/setup-envtest v0.0.0-20231005234617-5771399a8ce5

go.sum

@@ -2000,6 +2000,7 @@ k8s.io/kubernetes v1.24.15 h1:FupxlSyYgbz22yjGVZ7dxH+azhuO0OU8MnACdydrBzQ=
 k8s.io/kubernetes v1.24.15/go.mod h1:MlcoxAWSYrfeOwlfRNne7zYyZsHmlT3dlw7v3xzDnDM=
 k8s.io/legacy-cloud-providers v0.24.15 h1:0yLKue7fbat+rXdZoJRVP+iQ/y0Oxy0tHIwU0WKIAQg=
 k8s.io/legacy-cloud-providers v0.24.15/go.mod h1:sj/vZmVN9070GMNU9cuqSTupFI7ErHjT+bMXSc0rvac=
+k8s.io/metrics v0.24.15 h1:DdQZ6/7/yxkg856MlNlyS3bYWeRoi764kElIQuZb5bs=
 k8s.io/metrics v0.24.15/go.mod h1:0ZqaLxkIiopW4h1QfW5qaUZeajmLVxrwJJvWEljRYSM=
 k8s.io/mount-utils v0.24.15 h1:q3sm4Gcp00iWXUInIEi5x8CqAmy2chmUTedIZdUxRkg=
 k8s.io/mount-utils v0.24.15/go.mod h1:Xjtb0dquC5PG63kOD8shViqRczdkdQqW5Pc/rlmbsiU=

pkg/util/metricserver/metrics_client.go

@@ -0,0 +1,114 @@
+/*
+Copyright 2023 The Koordinator Authors.
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"context"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/metrics/pkg/apis/metrics/v1beta1"
+	resourceclient "k8s.io/metrics/pkg/client/clientset/versioned/typed/metrics/v1beta1"
+)
+
+// ContainerMetricsSnapshot contains information about usage of certain container within defined time window.
+type ContainerMetricsSnapshot struct {
+	// identifies a specific container those metrics are coming from.
+	Namespace     string
+	PodName       string
+	ContainerName string
+	// End time of the measurement interval.
+	SnapshotTime time.Time
+	// Duration of the measurement interval, which is [SnapshotTime - SnapshotWindow, SnapshotTime].
+	SnapshotWindow time.Duration
+	// Actual usage of the resources over the measurement interval.
+	Usage corev1.ResourceList
+}
+
+// MetricsClient provides simple metrics on resources usage on containter level.
+type MetricsClient interface {
+	// GetContainersMetrics returns an array of ContainerMetricsSnapshots,
+	// representing resource usage for every running container in the cluster
+	GetContainersMetrics() ([]*ContainerMetricsSnapshot, error)
+	// GetContainersMetricsByPod returns an array of ContainerMetricsSnapshots,
+	// representing resource usage for every running container in the given pod
+	GetContainersMetricsByPod(podNs, podName string) ([]*ContainerMetricsSnapshot, error)
+}
+
+type metricsClient struct {
+	metricsGetter resourceclient.PodMetricsesGetter
+	namespace     string
+}
+
+// NewMetricsClient creates new instance of MetricsClient, which is used by recommender.
+// It requires an instance of PodMetricsesGetter, which is used for underlying communication with metrics server.
+// namespace limits queries to particular namespace, use core.NamespaceAll to select all namespaces.
+func NewMetricsClient(metricsGetter resourceclient.PodMetricsesGetter, namespace string) MetricsClient {
+	return &metricsClient{
+		metricsGetter: metricsGetter,
+		namespace:     namespace,
+	}
+}
+
+func (c *metricsClient) GetContainersMetricsByPod(podNs, podName string) ([]*ContainerMetricsSnapshot, error) {
+	var metricsSnapshots []*ContainerMetricsSnapshot
+	podMetricsInterface := c.metricsGetter.PodMetricses(podNs)
+	podMetric, err := podMetricsInterface.Get(context.TODO(), podName, metav1.GetOptions{})
+	if err != nil || podMetric == nil {
+		return nil, err
+	}
+	metricsSnapshotsForPod := createContainerMetricsSnapshots(*podMetric)
+	metricsSnapshots = append(metricsSnapshots, metricsSnapshotsForPod...)
+	return metricsSnapshots, nil
+}
+
+func (c *metricsClient) GetContainersMetrics() ([]*ContainerMetricsSnapshot, error) {
+	var metricsSnapshots []*ContainerMetricsSnapshot
+
+	podMetricsInterface := c.metricsGetter.PodMetricses(c.namespace)
+	podMetricsList, err := podMetricsInterface.List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+	for _, podMetrics := range podMetricsList.Items {
+		metricsSnapshotsForPod := createContainerMetricsSnapshots(podMetrics)
+		metricsSnapshots = append(metricsSnapshots, metricsSnapshotsForPod...)
+	}
+
+	return metricsSnapshots, nil
+}
+
+func createContainerMetricsSnapshots(podMetrics v1beta1.PodMetrics) []*ContainerMetricsSnapshot {
+	snapshots := make([]*ContainerMetricsSnapshot, len(podMetrics.Containers))
+	for i, containerMetrics := range podMetrics.Containers {
+		snapshots[i] = newContainerMetricsSnapshot(containerMetrics, podMetrics)
+	}
+	return snapshots
+}
+
+func newContainerMetricsSnapshot(containerMetrics v1beta1.ContainerMetrics, podMetrics v1beta1.PodMetrics) *ContainerMetricsSnapshot {
+	return &ContainerMetricsSnapshot{
+		Namespace:      podMetrics.Namespace,
+		PodName:        podMetrics.Name,
+		ContainerName:  containerMetrics.Name,
+		Usage:          containerMetrics.Usage,
+		SnapshotTime:   podMetrics.Timestamp.Time,
+		SnapshotWindow: podMetrics.Window.Duration,
+	}
+}

pkg/util/metricserver/metrics_client_test.go

@@ -0,0 +1,70 @@
+/*
+Copyright 2023 The Koordinator Authors.
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetContainersMetricsReturnsEmptyList(t *testing.T) {
+	tc := newEmptyMetricsClientTestCase()
+	emptyMetricsClient := tc.createFakeMetricsClient()
+
+	containerMetricsSnapshots, err := emptyMetricsClient.GetContainersMetrics()
+
+	assert.NoError(t, err)
+	assert.Empty(t, containerMetricsSnapshots, "should be empty for empty MetricsGetter")
+}
+
+func TestGetContainersMetricsReturnsResults(t *testing.T) {
+	tc := newMetricsClientTestCase()
+	fakeMetricsClient := tc.createFakeMetricsClient()
+
+	snapshots, err := fakeMetricsClient.GetContainersMetrics()
+
+	assert.NoError(t, err)
+	assert.Len(t, snapshots, len(tc.getAllSnaps()), "It should return right number of snapshots")
+	for _, snap := range snapshots {
+		assert.Contains(t, tc.getAllSnaps(), snap, "One of returned ContainerMetricsSnapshot is different then expected ")
+	}
+}
+
+func TestGetContainersMetricsByPodReturnsEmptyList(t *testing.T) {
+	tc := newEmptyMetricsClientTestCase()
+	emptyMetricsClient := tc.createFakeMetricsClient()
+
+	containerMetricsSnapshots, err := emptyMetricsClient.GetContainersMetricsByPod("test-namespace", "Pod1")
+
+	assert.NoError(t, err)
+	assert.Empty(t, containerMetricsSnapshots, "should be empty for empty MetricsGetter")
+}
+
+func Test_metricsClient_GetContainersMetricsByPod(t *testing.T) {
+	tc := newMetricsClientTestCase()
+	fakeMetricsClient := tc.createFakeMetricsClient()
+
+	snapshots, err := fakeMetricsClient.GetContainersMetricsByPod("test-namespace", "Pod1")
+
+	assert.NoError(t, err)
+	assert.Len(t, snapshots, len(tc.pod1Snaps), "It should return right number of snapshots")
+	for _, snap := range snapshots {
+		assert.Contains(t, tc.pod1Snaps, snap, "One of returned ContainerMetricsSnapshot is different then expected ")
+	}
+}

pkg/util/metricserver/metrics_client_test_util.go

@@ -0,0 +1,137 @@
+/*
+Copyright 2023 The Koordinator Authors.
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"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/runtime"
+	core "k8s.io/client-go/testing"
+	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
+	"k8s.io/metrics/pkg/client/clientset/versioned/fake"
+)
+
+type metricsClientTestCase struct {
+	snapshotTimestamp    time.Time
+	snapshotWindow       time.Duration
+	namespace            *corev1.Namespace
+	pod1Snaps, pod2Snaps []*ContainerMetricsSnapshot
+}
+
+type containerID struct {
+	namespace     string
+	podname       string
+	containerName string
+}
+
+func newMetricsClientTestCase() *metricsClientTestCase {
+	namespaceName := "test-namespace"
+
+	testCase := &metricsClientTestCase{
+		snapshotTimestamp: time.Now(),
+		snapshotWindow:    time.Duration(1234),
+		namespace:         &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespaceName}},
+	}
+
+	id1 := containerID{namespace: namespaceName, podname: "Pod1", containerName: "Name1"}
+	id2 := containerID{namespace: namespaceName, podname: "Pod1", containerName: "Name2"}
+	id3 := containerID{namespace: namespaceName, podname: "Pod2", containerName: "Name1"}
+	id4 := containerID{namespace: namespaceName, podname: "Pod2", containerName: "Name2"}
+
+	testCase.pod1Snaps = append(testCase.pod1Snaps, testCase.newContainerMetricsSnapshot(id1, 400, 333))
+	testCase.pod1Snaps = append(testCase.pod1Snaps, testCase.newContainerMetricsSnapshot(id2, 800, 666))
+	testCase.pod2Snaps = append(testCase.pod2Snaps, testCase.newContainerMetricsSnapshot(id3, 401, 334))
+	testCase.pod2Snaps = append(testCase.pod2Snaps, testCase.newContainerMetricsSnapshot(id4, 801, 667))
+
+	return testCase
+}
+
+func newEmptyMetricsClientTestCase() *metricsClientTestCase {
+	return &metricsClientTestCase{}
+}
+
+func (tc *metricsClientTestCase) newContainerMetricsSnapshot(id containerID, cpuUsage int64, memUsage int64) *ContainerMetricsSnapshot {
+	return &ContainerMetricsSnapshot{
+		Namespace:      id.namespace,
+		PodName:        id.podname,
+		ContainerName:  id.containerName,
+		SnapshotTime:   tc.snapshotTimestamp,
+		SnapshotWindow: tc.snapshotWindow,
+		Usage: corev1.ResourceList{
+			corev1.ResourceCPU:    *resource.NewQuantity(cpuUsage, resource.DecimalSI),
+			corev1.ResourceMemory: *resource.NewQuantity(memUsage, resource.BinarySI),
+		},
+	}
+}
+
+func (tc *metricsClientTestCase) createFakeMetricsClient() MetricsClient {
+	fakeMetricsGetter := &fake.Clientset{}
+	fakeMetricsGetter.AddReactor("list", "pods", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		return true, tc.getFakePodMetricsList(), nil
+	})
+	fakeMetricsGetter.AddReactor("get", "pods", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		return true, tc.getFakePodMetric(), nil
+	})
+	return NewMetricsClient(fakeMetricsGetter.MetricsV1beta1(), "")
+}
+
+func (tc *metricsClientTestCase) getFakePodMetricsList() *metricsapi.PodMetricsList {
+	metrics := &metricsapi.PodMetricsList{}
+	if tc.pod1Snaps != nil && tc.pod2Snaps != nil {
+		metrics.Items = append(metrics.Items, makePodMetrics(tc.pod1Snaps))
+		metrics.Items = append(metrics.Items, makePodMetrics(tc.pod2Snaps))
+	}
+	return metrics
+}
+
+func (tc *metricsClientTestCase) getFakePodMetric() *metricsapi.PodMetrics {
+	metrics := &metricsapi.PodMetrics{}
+	if tc.pod1Snaps != nil {
+		*metrics = makePodMetrics(tc.pod1Snaps)
+	}
+
+	return metrics
+}
+
+func makePodMetrics(snaps []*ContainerMetricsSnapshot) metricsapi.PodMetrics {
+	firstSnap := snaps[0]
+	podMetrics := metricsapi.PodMetrics{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: firstSnap.Namespace,
+			Name:      firstSnap.PodName,
+		},
+		Timestamp:  metav1.Time{Time: firstSnap.SnapshotTime},
+		Window:     metav1.Duration{Duration: firstSnap.SnapshotWindow},
+		Containers: make([]metricsapi.ContainerMetrics, len(snaps)),
+	}
+
+	for i, snap := range snaps {
+		podMetrics.Containers[i] = metricsapi.ContainerMetrics{
+			Name:  snap.ContainerName,
+			Usage: snap.Usage,
+		}
+	}
+	return podMetrics
+}
+
+func (tc *metricsClientTestCase) getAllSnaps() []*ContainerMetricsSnapshot {
+	return append(tc.pod1Snaps, tc.pod2Snaps...)
+}