Add measurement to check memory usage in benchmark list job

Author: arshadd-b

clusterloader2/go.mod

@@ -60,6 +60,7 @@ require (
 	k8s.io/gengo v0.0.0-20230829151522-9cce18d56c01
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/kubelet v0.33.0
+	k8s.io/metrics v0.0.0-00010101000000-000000000000
 )
 
 require (

clusterloader2/go.sum

@@ -1588,6 +1588,8 @@ k8s.io/kube-openapi v0.0.0-20250318190949-c8a335a9a2ff h1:/usPimJzUKKu+m+TE36gUy
 k8s.io/kube-openapi v0.0.0-20250318190949-c8a335a9a2ff/go.mod h1:5jIi+8yX4RIb8wk3XwBo5Pq2ccx4FP10ohkbSKCZoK8=
 k8s.io/kubelet v0.33.0 h1:4pJA2Ge6Rp0kDNV76KH7pTBiaV2T1a1874QHMcubuSU=
 k8s.io/kubelet v0.33.0/go.mod h1:iDnxbJQMy9DUNaML5L/WUlt3uJtNLWh7ZAe0JSp4Yi0=
+k8s.io/metrics v0.33.0 h1:sKe5sC9qb1RakMhs8LWYNuN2ne6OTCWexj8Jos3rO2Y=
+k8s.io/metrics v0.33.0/go.mod h1:XewckTFXmE2AJiP7PT3EXaY7hi7bler3t2ZLyOdQYzU=
 k8s.io/utils v0.0.0-20240711033017-18e509b52bc8/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 h1:M3sRQVHv7vB20Xc2ybTt7ODCeFj6JSWYFzOFnYeS6Ro=
 k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=

clusterloader2/pkg/measurement/common/container_memory_usage.go

@@ -0,0 +1,211 @@
+/*
+Copyright 2025 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 common
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/client-go/kubernetes"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/klog/v2"
+	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
+	"k8s.io/perf-tests/clusterloader2/pkg/errors"
+	"k8s.io/perf-tests/clusterloader2/pkg/measurement"
+	"k8s.io/perf-tests/clusterloader2/pkg/util"
+)
+
+const (
+	containerMemoryUsageName = "ContainerMemoryUsage"
+)
+
+func init() {
+	if err := measurement.Register(containerMemoryUsageName, createContainerMemoryUsageMeasurement); err != nil {
+		klog.Fatalf("Cannot register %s: %v", containerMemoryUsageName, err)
+	}
+	klog.Infof("[ContainerMemoryUsage] Starting measurement with label selector")
+}
+
+func createContainerMemoryUsageMeasurement() measurement.Measurement {
+	return &containerMemoryUsageMeasurement{}
+}
+
+type containerMemoryUsageMeasurement struct {
+	isRunning     bool
+	summaries     []measurement.Summary
+	clientset     *kubernetes.Clientset
+	pollFrequency time.Duration
+	threshold     float64
+
+	stopCh  chan struct{}
+	mu      sync.Mutex
+	samples []float64
+	lock    sync.Mutex
+	wg      sync.WaitGroup
+}
+
+func (c *containerMemoryUsageMeasurement) Execute(config *measurement.Config) ([]measurement.Summary, error) {
+	action, err := util.GetString(config.Params, "action")
+	if err != nil {
+		return nil, err
+	}
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	switch action {
+	case "start":
+		return nil, c.start(config)
+	case "gather":
+		return c.gather()
+	default:
+		return nil, fmt.Errorf("unknown action %v", action)
+	}
+
+}
+
+// Configure parses configuration and initializes clientset
+func (c *containerMemoryUsageMeasurement) initFields(config *measurement.Config) error {
+	c.isRunning = true
+	c.stopCh = make(chan struct{})
+
+	frequency, err := util.GetDuration(config.Params, "pollFrequency")
+	if err != nil {
+		return err
+	}
+	c.pollFrequency = frequency
+
+	threshold, err := util.GetFloat64OrDefault(config.Params, "threshold", 0)
+	if err != nil {
+		return err
+	}
+	c.threshold = threshold
+
+	return nil
+}
+
+func (c *containerMemoryUsageMeasurement) start(config *measurement.Config) error {
+	klog.V(2).Infof("%v: measurement already running", c)
+	if c.isRunning {
+		klog.V(2).Infof("%v: measurement already running", c)
+		return nil
+	}
+	if err := c.initFields(config); err != nil {
+		return err
+	}
+	k8sClient := config.ClusterFramework.GetClientSets().GetClient()
+
+	c.wg.Add(1)
+	go func() {
+		defer c.wg.Done()
+
+		for {
+			select {
+			case <-c.stopCh:
+				return
+			case <-time.After(c.pollFrequency):
+				memoryUsage, err := c.getAPIServerMemoryUsage(k8sClient)
+				if err != nil {
+					klog.Warningf("ContainerMemoryUsage: read error: %v", err)
+					continue
+				}
+
+				// convert Memory bytes to GB
+				memoryInGB := memoryUsage / (1024 * 1024 * 1024)
+
+				c.samples = append(c.samples, memoryInGB)
+			}
+		}
+	}()
+	return nil
+}
+
+func (c *containerMemoryUsageMeasurement) gather() ([]measurement.Summary, error) {
+	klog.V(2).Infof("%v: measurement already running", c)
+	if !c.isRunning {
+		return nil, nil
+	}
+
+	// Close this channel to stop the execution
+	close(c.stopCh)
+	// Wait for execution goroutine to finish
+	c.wg.Wait()
+	c.isRunning = false
+
+	klog.V(2).Infof("%s: gathering summaries", containerMemoryUsageName)
+	maxMemoryUsage := slices.Max(c.samples)
+
+	content := fmt.Sprintf(`{"maxMemoryUsage": %d, "threshold": %d}`, maxMemoryUsage, c.threshold)
+
+	summary := measurement.CreateSummary(containerMemoryUsageName, "json", content)
+	klog.V(2).Infof("%d: maxMemoryUsage", maxMemoryUsage)
+	klog.V(2).Infof("%d: threshold", c.threshold)
+	if maxMemoryUsage > c.threshold {
+		err := errors.NewMetricViolationError("ContainerMemoryUsage", fmt.Sprintf("SLO not fulfilled (expected >= %f, got: %f)", c.threshold, maxMemoryUsage))
+		return nil, err
+	}
+
+	return []measurement.Summary{summary}, nil
+
+}
+
+func (c *containerMemoryUsageMeasurement) getAPIServerMemoryUsage(cl clientset.Interface) (float64, error) {
+
+	result, err := cl.CoreV1().RESTClient().
+		Get().
+		AbsPath("apis/metrics.k8s.io/v1beta1/namespaces/kube-system/pods").
+		Param("labelSelector", "component=kube-apiserver").
+		DoRaw(context.Background())
+
+	scheme := runtime.NewScheme()
+	_ = metricsapi.AddToScheme(scheme) // register metrics API types into scheme
+
+	codef := serializer.NewCodecFactory(scheme).UniversalDeserializer()
+
+	if err != nil {
+		return 0, err
+	}
+
+	podMetricsList := &metricsapi.PodMetricsList{}
+	_, _, err = codef.Decode(result, nil, podMetricsList)
+	if err != nil {
+		return 0, fmt.Errorf("failed to decode metrics: %v", err)
+	}
+
+	var memoryUsage float64
+
+	for _, podMetric := range podMetricsList.Items {
+		for _, container := range podMetric.Containers {
+			if container.Name == "kube-apiserver" {
+				memoryUsage = container.Usage.Memory().AsApproximateFloat64()
+				break
+			}
+		}
+	}
+	return memoryUsage, nil
+}
+
+func (c *containerMemoryUsageMeasurement) Dispose() {}
+
+func (c *containerMemoryUsageMeasurement) String() string {
+	return containerMemoryUsageName
+}

clusterloader2/testing/list/modules/measurements.yaml

@@ -3,6 +3,8 @@
 ## Input params
 # Valid actions: "start", "gather"
 {{$action := .action}}
+{{$ENABLE_CONTAINER_MEMORY_USAGE_MEASUREMENT := DefaultParam .CL2_ENABLE_CONTAINER_MEMORY_USAGE_MEASUREMENT true}}
+{{$CONTAINER_MEMORY_THRESHOLD := DefaultParam .CL2_CONTAINER_MEMORY_THRESHOLD 1}}
 
 
 steps:
@@ -17,4 +19,12 @@ steps:
   - Identifier: TestMetrics
     Method: TestMetrics
     Params:
-      action: {{$action}}
+      action: {{$action}}
+  {{if $ENABLE_CONTAINER_MEMORY_USAGE_MEASUREMENT}}
+  - Identifier: ContainerMemoryUsage
+    Method: ContainerMemoryUsage
+    Params:
+      action: {{$action}}
+      threshold: {{$CONTAINER_MEMORY_THRESHOLD}}
+      pollFrequency: 15s
+  {{end}}