Merge pull request #7900 from omerap12/ctx-propagation-recommender

Refactor recommender to use context

Author: k8s-ci-robot

vertical-pod-autoscaler/pkg/recommender/input/cluster_feeder.go

@@ -61,7 +61,7 @@ type ClusterStateFeeder interface {
 	InitFromHistoryProvider(historyProvider history.HistoryProvider)
 
 	// InitFromCheckpoints loads historical checkpoints into clusterState.
-	InitFromCheckpoints()
+	InitFromCheckpoints(ctx context.Context)
 
 	// LoadVPAs updates clusterState with current state of VPAs.
 	LoadVPAs(ctx context.Context)
@@ -70,10 +70,10 @@ type ClusterStateFeeder interface {
 	LoadPods()
 
 	// LoadRealTimeMetrics updates clusterState with current usage metrics of containers.
-	LoadRealTimeMetrics()
+	LoadRealTimeMetrics(ctx context.Context)
 
 	// GarbageCollectCheckpoints removes historical checkpoints that don't have a matching VPA.
-	GarbageCollectCheckpoints()
+	GarbageCollectCheckpoints(ctx context.Context)
 }
 
 // ClusterStateFeederFactory makes instances of ClusterStateFeeder.
@@ -113,26 +113,32 @@ func (m ClusterStateFeederFactory) Make() *clusterStateFeeder {
 }
 
 // WatchEvictionEventsWithRetries watches new Events with reason=Evicted and passes them to the observer.
-func WatchEvictionEventsWithRetries(kubeClient kube_client.Interface, observer oom.Observer, namespace string) {
+func WatchEvictionEventsWithRetries(ctx context.Context, kubeClient kube_client.Interface, observer oom.Observer, namespace string) {
 	go func() {
 		options := metav1.ListOptions{
 			FieldSelector: "reason=Evicted",
 		}
 
 		watchEvictionEventsOnce := func() {
-			watchInterface, err := kubeClient.CoreV1().Events(namespace).Watch(context.TODO(), options)
+			watchInterface, err := kubeClient.CoreV1().Events(namespace).Watch(ctx, options)
 			if err != nil {
 				klog.ErrorS(err, "Cannot initialize watching events")
 				return
 			}
+			defer watchInterface.Stop()
 			watchEvictionEvents(watchInterface.ResultChan(), observer)
 		}
 		for {
-			watchEvictionEventsOnce()
-			// Wait between attempts, retrying too often breaks API server.
-			waitTime := wait.Jitter(evictionWatchRetryWait, evictionWatchJitterFactor)
-			klog.V(1).InfoS("An attempt to watch eviction events finished", "waitTime", waitTime)
-			time.Sleep(waitTime)
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				watchEvictionEventsOnce()
+				// Wait between attempts, retrying too often breaks API server.
+				waitTime := wait.Jitter(evictionWatchRetryWait, evictionWatchJitterFactor)
+				klog.V(1).InfoS("An attempt to watch eviction events finished", "waitTime", waitTime)
+				time.Sleep(waitTime)
+			}
 		}
 	}()
 }
@@ -188,10 +194,10 @@ func newPodClients(kubeClient kube_client.Interface, resourceEventHandler cache.
 }
 
 // NewPodListerAndOOMObserver creates pair of pod lister and OOM observer.
-func NewPodListerAndOOMObserver(kubeClient kube_client.Interface, namespace string, stopCh <-chan struct{}) (v1lister.PodLister, oom.Observer) {
+func NewPodListerAndOOMObserver(ctx context.Context, kubeClient kube_client.Interface, namespace string, stopCh <-chan struct{}) (v1lister.PodLister, oom.Observer) {
 	oomObserver := oom.NewObserver()
 	podLister := newPodClients(kubeClient, oomObserver, namespace, stopCh)
-	WatchEvictionEventsWithRetries(kubeClient, oomObserver, namespace)
+	WatchEvictionEventsWithRetries(ctx, kubeClient, oomObserver, namespace)
 	return podLister, oomObserver
 }
 
@@ -258,9 +264,9 @@ func (feeder *clusterStateFeeder) setVpaCheckpoint(checkpoint *vpa_types.Vertica
 	return nil
 }
 
-func (feeder *clusterStateFeeder) InitFromCheckpoints() {
+func (feeder *clusterStateFeeder) InitFromCheckpoints(ctx context.Context) {
 	klog.V(3).InfoS("Initializing VPA from checkpoints")
-	feeder.LoadVPAs(context.TODO())
+	feeder.LoadVPAs(ctx)
 
 	namespaces := make(map[string]bool)
 	for _, v := range feeder.clusterState.VPAs() {
@@ -269,7 +275,7 @@ func (feeder *clusterStateFeeder) InitFromCheckpoints() {
 
 	for namespace := range namespaces {
 		klog.V(3).InfoS("Fetching checkpoints", "namespace", namespace)
-		checkpointList, err := feeder.vpaCheckpointClient.VerticalPodAutoscalerCheckpoints(namespace).List(context.TODO(), metav1.ListOptions{})
+		checkpointList, err := feeder.vpaCheckpointClient.VerticalPodAutoscalerCheckpoints(namespace).List(ctx, metav1.ListOptions{})
 		if err != nil {
 			klog.ErrorS(err, "Cannot list VPA checkpoints", "namespace", namespace)
 		}
@@ -285,7 +291,7 @@ func (feeder *clusterStateFeeder) InitFromCheckpoints() {
 	}
 }
 
-func (feeder *clusterStateFeeder) GarbageCollectCheckpoints() {
+func (feeder *clusterStateFeeder) GarbageCollectCheckpoints(ctx context.Context) {
 	klog.V(3).InfoS("Starting garbage collection of checkpoints")
 
 	allVPAKeys := map[model.VpaID]bool{}
@@ -303,7 +309,7 @@ func (feeder *clusterStateFeeder) GarbageCollectCheckpoints() {
 		allVPAKeys[vpaID] = true
 	}
 
-	namespaceList, err := feeder.coreClient.Namespaces().List(context.TODO(), metav1.ListOptions{})
+	namespaceList, err := feeder.coreClient.Namespaces().List(ctx, metav1.ListOptions{})
 	if err != nil {
 		klog.ErrorS(err, "Cannot list namespaces")
 		return
@@ -319,7 +325,7 @@ func (feeder *clusterStateFeeder) GarbageCollectCheckpoints() {
 			klog.V(3).InfoS("Skipping namespace; it does not meet cleanup criteria", "namespace", namespace, "vpaObjectNamespace", feeder.vpaObjectNamespace, "ignoredNamespaces", feeder.ignoredNamespaces)
 			continue
 		}
-		err := feeder.cleanupCheckpointsForNamespace(namespace, allVPAKeys)
+		err := feeder.cleanupCheckpointsForNamespace(ctx, namespace, allVPAKeys)
 		if err != nil {
 			klog.ErrorS(err, "error cleanining checkpoints")
 		}
@@ -338,16 +344,16 @@ func (feeder *clusterStateFeeder) shouldIgnoreNamespace(namespace string) bool {
 	return false
 }
 
-func (feeder *clusterStateFeeder) cleanupCheckpointsForNamespace(namespace string, allVPAKeys map[model.VpaID]bool) error {
+func (feeder *clusterStateFeeder) cleanupCheckpointsForNamespace(ctx context.Context, namespace string, allVPAKeys map[model.VpaID]bool) error {
 	var err error
-	checkpointList, err := feeder.vpaCheckpointClient.VerticalPodAutoscalerCheckpoints(namespace).List(context.TODO(), metav1.ListOptions{})
+	checkpointList, err := feeder.vpaCheckpointClient.VerticalPodAutoscalerCheckpoints(namespace).List(ctx, metav1.ListOptions{})
 	if err != nil {
 		return err
 	}
 	for _, checkpoint := range checkpointList.Items {
 		vpaID := model.VpaID{Namespace: checkpoint.Namespace, VpaName: checkpoint.Spec.VPAObjectName}
 		if !allVPAKeys[vpaID] {
-			if errFeeder := feeder.vpaCheckpointClient.VerticalPodAutoscalerCheckpoints(namespace).Delete(context.TODO(), checkpoint.Name, metav1.DeleteOptions{}); errFeeder != nil {
+			if errFeeder := feeder.vpaCheckpointClient.VerticalPodAutoscalerCheckpoints(namespace).Delete(ctx, checkpoint.Name, metav1.DeleteOptions{}); errFeeder != nil {
 				err = fmt.Errorf("failed to delete orphaned checkpoint %s: %w", klog.KRef(namespace, checkpoint.Name), err)
 				continue
 			}
@@ -484,8 +490,8 @@ func (feeder *clusterStateFeeder) LoadPods() {
 	}
 }
 
-func (feeder *clusterStateFeeder) LoadRealTimeMetrics() {
-	containersMetrics, err := feeder.metricsClient.GetContainersMetrics()
+func (feeder *clusterStateFeeder) LoadRealTimeMetrics(ctx context.Context) {
+	containersMetrics, err := feeder.metricsClient.GetContainersMetrics(ctx)
 	if err != nil {
 		klog.ErrorS(err, "Cannot get ContainerMetricsSnapshot from MetricsClient")
 	}

vertical-pod-autoscaler/pkg/recommender/input/cluster_feeder_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	core "k8s.io/client-go/testing"
+	"k8s.io/klog/v2/ktesting"
 
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	v1 "k8s.io/api/core/v1"
@@ -831,9 +832,10 @@ func TestFilterVPAsIgnoreNamespaces(t *testing.T) {
 }
 
 func TestCanCleanupCheckpoints(t *testing.T) {
+	_, tctx := ktesting.NewTestContext(t)
 	client := fake.NewSimpleClientset()
 
-	_, err := client.CoreV1().Namespaces().Create(context.TODO(), &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "testNamespace"}}, metav1.CreateOptions{})
+	_, err := client.CoreV1().Namespaces().Create(tctx, &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "testNamespace"}}, metav1.CreateOptions{})
 	assert.NoError(t, err)
 
 	vpaBuilder := test.VerticalPodAutoscaler().WithContainer("container").WithNamespace("testNamespace").WithTargetRef(&autoscalingv1.CrossVersionObjectReference{
@@ -898,7 +900,7 @@ func TestCanCleanupCheckpoints(t *testing.T) {
 		recommenderName:     "default",
 	}
 
-	feeder.GarbageCollectCheckpoints()
+	feeder.GarbageCollectCheckpoints(tctx)
 
 	assert.Contains(t, deletedCheckpoints, "nonExistentVPA")
 

vertical-pod-autoscaler/pkg/recommender/input/metrics/metrics_client.go

@@ -45,7 +45,7 @@ type ContainerMetricsSnapshot struct {
 type MetricsClient interface {
 	// GetContainersMetrics returns an array of ContainerMetricsSnapshots,
 	// representing resource usage for every running container in the cluster
-	GetContainersMetrics() ([]*ContainerMetricsSnapshot, error)
+	GetContainersMetrics(ctx context.Context) ([]*ContainerMetricsSnapshot, error)
 }
 
 type metricsClient struct {
@@ -64,10 +64,10 @@ func NewMetricsClient(source PodMetricsLister, namespace, clientName string) Met
 	}
 }
 
-func (c *metricsClient) GetContainersMetrics() ([]*ContainerMetricsSnapshot, error) {
+func (c *metricsClient) GetContainersMetrics(ctx context.Context) ([]*ContainerMetricsSnapshot, error) {
 	var metricsSnapshots []*ContainerMetricsSnapshot
 
-	podMetricsList, err := c.source.List(context.TODO(), c.namespace, metav1.ListOptions{})
+	podMetricsList, err := c.source.List(ctx, c.namespace, metav1.ListOptions{})
 	recommender_metrics.RecordMetricsServerResponse(err, c.clientName)
 	if err != nil {
 		return nil, err

vertical-pod-autoscaler/pkg/recommender/input/metrics/metrics_client_test.go

@@ -20,23 +20,26 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"k8s.io/klog/v2/ktesting"
 )
 
 func TestGetContainersMetricsReturnsEmptyList(t *testing.T) {
+	_, tctx := ktesting.NewTestContext(t)
 	tc := newEmptyMetricsClientTestCase()
 	emptyMetricsClient := tc.createFakeMetricsClient()
 
-	containerMetricsSnapshots, err := emptyMetricsClient.GetContainersMetrics()
+	containerMetricsSnapshots, err := emptyMetricsClient.GetContainersMetrics(tctx)
 
 	assert.NoError(t, err)
 	assert.Empty(t, containerMetricsSnapshots, "should be empty for empty MetricsGetter")
 }
 
 func TestGetContainersMetricsReturnsResults(t *testing.T) {
+	_, tctx := ktesting.NewTestContext(t)
 	tc := newMetricsClientTestCase()
 	fakeMetricsClient := tc.createFakeMetricsClient()
 
-	snapshots, err := fakeMetricsClient.GetContainersMetrics()
+	snapshots, err := fakeMetricsClient.GetContainersMetrics(tctx)
 
 	assert.NoError(t, err)
 	assert.Len(t, snapshots, len(tc.getAllSnaps()), "It should return right number of snapshots")

vertical-pod-autoscaler/pkg/recommender/main.go

@@ -138,14 +138,15 @@ func main() {
 		klog.ErrorS(nil, "--vpa-object-namespace and --ignored-vpa-object-namespaces are mutually exclusive and can't be set together.")
 		os.Exit(255)
 	}
+	ctx := context.Background()
 
 	healthCheck := metrics.NewHealthCheck(*metricsFetcherInterval * 5)
 	metrics_recommender.Register()
 	metrics_quality.Register()
 	server.Initialize(&commonFlags.EnableProfiling, healthCheck, address)
 
 	if !leaderElection.LeaderElect {
-		run(healthCheck, commonFlags)
+		run(ctx, healthCheck, commonFlags)
 	} else {
 		id, err := os.Hostname()
 		if err != nil {
@@ -172,15 +173,15 @@ func main() {
 			os.Exit(255)
 		}
 
-		leaderelection.RunOrDie(context.TODO(), leaderelection.LeaderElectionConfig{
+		leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
 			Lock:            lock,
 			LeaseDuration:   leaderElection.LeaseDuration.Duration,
 			RenewDeadline:   leaderElection.RenewDeadline.Duration,
 			RetryPeriod:     leaderElection.RetryPeriod.Duration,
 			ReleaseOnCancel: true,
 			Callbacks: leaderelection.LeaderCallbacks{
 				OnStartedLeading: func(_ context.Context) {
-					run(healthCheck, commonFlags)
+					run(ctx, healthCheck, commonFlags)
 				},
 				OnStoppedLeading: func() {
 					klog.Fatal("lost master")
@@ -209,7 +210,7 @@ func defaultLeaderElectionConfiguration() componentbaseconfig.LeaderElectionConf
 	}
 }
 
-func run(healthCheck *metrics.HealthCheck, commonFlag *common.CommonFlags) {
+func run(ctx context.Context, healthCheck *metrics.HealthCheck, commonFlag *common.CommonFlags) {
 	// Create a stop channel that will be used to signal shutdown
 	stopCh := make(chan struct{})
 	defer close(stopCh)
@@ -218,7 +219,7 @@ func run(healthCheck *metrics.HealthCheck, commonFlag *common.CommonFlags) {
 	clusterState := model.NewClusterState(aggregateContainerStateGCInterval)
 	factory := informers.NewSharedInformerFactoryWithOptions(kubeClient, defaultResyncPeriod, informers.WithNamespace(commonFlag.VpaObjectNamespace))
 	controllerFetcher := controllerfetcher.NewControllerFetcher(config, kubeClient, factory, scaleCacheEntryFreshnessTime, scaleCacheEntryLifetime, scaleCacheEntryJitterFactor)
-	podLister, oomObserver := input.NewPodListerAndOOMObserver(kubeClient, commonFlag.VpaObjectNamespace, stopCh)
+	podLister, oomObserver := input.NewPodListerAndOOMObserver(ctx, kubeClient, commonFlag.VpaObjectNamespace, stopCh)
 
 	model.InitializeAggregationsConfig(model.NewAggregationsConfig(*memoryAggregationInterval, *memoryAggregationIntervalCount, *memoryHistogramDecayHalfLife, *cpuHistogramDecayHalfLife, *oomBumpUpRatio, *oomMinBumpUp))
 
@@ -266,7 +267,7 @@ func run(healthCheck *metrics.HealthCheck, commonFlag *common.CommonFlags) {
 		IgnoredNamespaces:   ignoredNamespaces,
 		VpaObjectNamespace:  commonFlag.VpaObjectNamespace,
 	}.Make()
-	controllerFetcher.Start(context.Background(), scaleCacheLoopPeriod)
+	controllerFetcher.Start(ctx, scaleCacheLoopPeriod)
 
 	recommender := routines.RecommenderFactory{
 		ClusterState:                 clusterState,
@@ -287,7 +288,7 @@ func run(healthCheck *metrics.HealthCheck, commonFlag *common.CommonFlags) {
 	}
 
 	if useCheckpoints {
-		recommender.GetClusterStateFeeder().InitFromCheckpoints()
+		recommender.GetClusterStateFeeder().InitFromCheckpoints(ctx)
 	} else {
 		config := history.PrometheusHistoryProviderConfig{
 			Address:                *prometheusAddress,

vertical-pod-autoscaler/pkg/recommender/routines/recommender.go

@@ -134,7 +134,7 @@ func (r *recommender) MaintainCheckpoints(ctx context.Context, minCheckpointsPer
 		}
 		if time.Since(r.lastCheckpointGC) > r.checkpointsGCInterval {
 			r.lastCheckpointGC = now
-			r.clusterStateFeeder.GarbageCollectCheckpoints()
+			r.clusterStateFeeder.GarbageCollectCheckpoints(ctx)
 		}
 	}
 }
@@ -153,7 +153,7 @@ func (r *recommender) RunOnce() {
 	r.clusterStateFeeder.LoadPods()
 	timer.ObserveStep("LoadPods")
 
-	r.clusterStateFeeder.LoadRealTimeMetrics()
+	r.clusterStateFeeder.LoadRealTimeMetrics(ctx)
 	timer.ObserveStep("LoadMetrics")
 	klog.V(3).InfoS("ClusterState is tracking", "pods", len(r.clusterState.Pods()), "vpas", len(r.clusterState.VPAs()))