Merge pull request #8259 from yalosev/fix/factory-start

[VPA] Use factory start to fill caches instead of separate informers

Author: k8s-ci-robot

vertical-pod-autoscaler/pkg/admission-controller/main.go

@@ -116,6 +116,17 @@ func main() {
 	recommendationProvider := recommendation.NewProvider(limitRangeCalculator, vpa_api_util.NewCappingRecommendationProcessor(limitRangeCalculator))
 	vpaMatcher := vpa.NewMatcher(vpaLister, targetSelectorFetcher, controllerFetcher)
 
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	factory.Start(stopCh)
+	informerMap := factory.WaitForCacheSync(stopCh)
+	for kind, synced := range informerMap {
+		if !synced {
+			klog.ErrorS(nil, fmt.Sprintf("Could not sync cache for the %s informer", kind.String()))
+			klog.FlushAndExit(klog.ExitFlushTimeout, 1)
+		}
+	}
+
 	hostname, err := os.Hostname()
 	if err != nil {
 		klog.ErrorS(err, "Unable to get hostname")
@@ -126,15 +137,13 @@ func main() {
 	if namespace != "" {
 		statusNamespace = namespace
 	}
-	stopCh := make(chan struct{})
 	statusUpdater := status.NewUpdater(
 		kubeClient,
 		status.AdmissionControllerStatusName,
 		statusNamespace,
 		statusUpdateInterval,
 		hostname,
 	)
-	defer close(stopCh)
 
 	calculators := []patch.Calculator{patch.NewResourceUpdatesCalculator(recommendationProvider), patch.NewObservedContainersCalculator()}
 	as := logic.NewAdmissionServer(podPreprocessor, vpaPreprocessor, limitRangeCalculator, vpaMatcher, calculators)

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

@@ -19,6 +19,7 @@ package main
 import (
 	"context"
 	"flag"
+	"fmt"
 	"os"
 	"strings"
 	"time"
@@ -232,6 +233,15 @@ func run(ctx context.Context, healthCheck *metrics.HealthCheck, commonFlag *comm
 	controllerFetcher := controllerfetcher.NewControllerFetcher(config, kubeClient, factory, scaleCacheEntryFreshnessTime, scaleCacheEntryLifetime, scaleCacheEntryJitterFactor)
 	podLister, oomObserver := input.NewPodListerAndOOMObserver(ctx, kubeClient, commonFlag.VpaObjectNamespace, stopCh)
 
+	factory.Start(stopCh)
+	informerMap := factory.WaitForCacheSync(stopCh)
+	for kind, synced := range informerMap {
+		if !synced {
+			klog.ErrorS(nil, fmt.Sprintf("Could not sync cache for the %s informer", kind.String()))
+			klog.FlushAndExit(klog.ExitFlushTimeout, 1)
+		}
+	}
+
 	model.InitializeAggregationsConfig(model.NewAggregationsConfig(*memoryAggregationInterval, *memoryAggregationIntervalCount, *memoryHistogramDecayHalfLife, *cpuHistogramDecayHalfLife, *oomBumpUpRatio, *oomMinBumpUp))
 
 	useCheckpoints := *storage != "prometheus"

vertical-pod-autoscaler/pkg/target/controller_fetcher/controller_fetcher.go

@@ -136,17 +136,6 @@ func NewControllerFetcher(config *rest.Config, kubeClient kube_client.Interface,
 		cronJob:               factory.Batch().V1().CronJobs().Informer(),
 	}
 
-	for kind, informer := range informersMap {
-		stopCh := make(chan struct{})
-		go informer.Run(stopCh)
-		synced := cache.WaitForCacheSync(stopCh, informer.HasSynced)
-		if !synced {
-			klog.V(0).InfoS("Initial sync failed", "kind", kind)
-		} else {
-			klog.InfoS("Initial sync completed", "kind", kind)
-		}
-	}
-
 	scaleNamespacer := scale.New(restClient, mapper, dynamic.LegacyAPIPathResolverFunc, resolver)
 	return &controllerFetcher{
 		scaleNamespacer:              scaleNamespacer,

vertical-pod-autoscaler/pkg/target/fetcher.go

@@ -91,18 +91,6 @@ func NewVpaTargetSelectorFetcher(config *rest.Config, kubeClient kube_client.Int
 		cronJob:               factory.Batch().V1().CronJobs().Informer(),
 	}
 
-	for kind, informer := range informersMap {
-		stopCh := make(chan struct{})
-		go informer.Run(stopCh)
-		synced := cache.WaitForCacheSync(stopCh, informer.HasSynced)
-		if !synced {
-			klog.ErrorS(nil, "Could not sync cache for "+string(kind))
-			klog.FlushAndExit(klog.ExitFlushTimeout, 1)
-		} else {
-			klog.InfoS("Initial sync completed", "kind", kind)
-		}
-	}
-
 	scaleNamespacer := scale.New(restClient, mapper, dynamic.LegacyAPIPathResolverFunc, resolver)
 	return &vpaTargetSelectorFetcher{
 		scaleNamespacer: scaleNamespacer,

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

@@ -19,6 +19,7 @@ package main
 import (
 	"context"
 	"flag"
+	"fmt"
 	"os"
 	"strings"
 	"time"
@@ -172,6 +173,8 @@ func defaultLeaderElectionConfiguration() componentbaseconfig.LeaderElectionConf
 }
 
 func run(healthCheck *metrics.HealthCheck, commonFlag *common.CommonFlags) {
+	stopCh := make(chan struct{})
+	defer close(stopCh)
 	config := common.CreateKubeConfigOrDie(commonFlag.KubeConfig, float32(commonFlag.KubeApiQps), int(commonFlag.KubeApiBurst))
 	kubeClient := kube_client.NewForConfigOrDie(config)
 	vpaClient := vpa_clientset.NewForConfigOrDie(config)
@@ -184,6 +187,16 @@ func run(healthCheck *metrics.HealthCheck, commonFlag *common.CommonFlags) {
 		klog.ErrorS(err, "Failed to create limitRangeCalculator, falling back to not checking limits")
 		limitRangeCalculator = limitrange.NewNoopLimitsCalculator()
 	}
+
+	factory.Start(stopCh)
+	informerMap := factory.WaitForCacheSync(stopCh)
+	for kind, synced := range informerMap {
+		if !synced {
+			klog.ErrorS(nil, fmt.Sprintf("Could not sync cache for the %s informer", kind.String()))
+			klog.FlushAndExit(klog.ExitFlushTimeout, 1)
+		}
+	}
+
 	admissionControllerStatusNamespace := status.AdmissionControllerStatusNamespace
 	if namespace != "" {
 		admissionControllerStatusNamespace = namespace

vertical-pod-autoscaler/pkg/utils/limitrange/limit_range_calculator.go

@@ -24,7 +24,6 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/informers"
 	listers "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/tools/cache"
 )
 
 // LimitRangeCalculator calculates limit range items that has the same effect as all limit range items present in the cluster.
@@ -55,13 +54,6 @@ func NewLimitsRangeCalculator(f informers.SharedInformerFactory) (*limitsChecker
 		return nil, fmt.Errorf("NewLimitsRangeCalculator requires a SharedInformerFactory but got nil")
 	}
 	limitRangeLister := f.Core().V1().LimitRanges().Lister()
-	stopCh := make(chan struct{})
-	informer := f.Core().V1().LimitRanges().Informer()
-	go informer.Run(stopCh)
-	ok := cache.WaitForCacheSync(stopCh, informer.HasSynced)
-	if !ok {
-		return nil, fmt.Errorf("informer did not sync")
-	}
 	return &limitsChecker{limitRangeLister}, nil
 }
 

vertical-pod-autoscaler/pkg/utils/limitrange/limit_range_calculator_test.go

@@ -138,6 +138,9 @@ func TestGetContainerLimitRangeItem(t *testing.T) {
 			cs := fake.NewSimpleClientset(tc.limitRanges...)
 			factory := informers.NewSharedInformerFactory(cs, 0)
 			lc, err := NewLimitsRangeCalculator(factory)
+
+			factory.Start(t.Context().Done())
+			_ = factory.WaitForCacheSync(t.Context().Done())
 			if assert.NoError(t, err) {
 				limitRange, err := lc.GetContainerLimitRangeItem(testNamespace)
 				if tc.expectedErr == nil {