feat(vpa): thread prediction store through disruption path

Author: jigisha620

kwok/charts/templates/clusterrole.yaml

@@ -44,6 +44,9 @@ rules:
   - apiGroups: ["apps"]
     resources: ["daemonsets", "deployments", "replicasets", "statefulsets"]
     verbs: ["get", "list", "watch"]
+  - apiGroups: ["batch"]
+    resources: ["jobs"]
+    verbs: ["get", "list", "watch"]
   - apiGroups: ["autoscaling.k8s.io"]
     resources: ["verticalpodautoscalers"]
     verbs: ["get", "list", "watch"]

pkg/controllers/controllers.go

@@ -73,15 +73,16 @@ import (
 )
 
 type ControllerOptions struct {
-	registrationHooks    []cloudprovider.NodeLifecycleHook
-	disableVPAPrediction bool
+	registrationHooks   []cloudprovider.NodeLifecycleHook
+	enableVPAPrediction bool
 }
 
-// WithoutVPAPrediction disables the VPA prediction controller. Use this when
-// a different prediction source is registered separately.
-func WithoutVPAPrediction() option.Function[ControllerOptions] {
+// WithVPAPrediction enables the VPA prediction controller as the prediction source.
+// When enabled alongside the PredictionEnabled feature gate, Karpenter uses VPA
+// recommendations to size nodes during provisioning and disruption.
+func WithVPAPrediction() option.Function[ControllerOptions] {
 	return func(o *ControllerOptions) {
-		o.disableVPAPrediction = true
+		o.enableVPAPrediction = true
 	}
 }
 
@@ -111,7 +112,7 @@ func NewControllers(
 	o := option.Resolve(opts...)
 	deviceAllocationController := deviceallocation.NewController(kubeClient)
 	virtualPodCache := virtualpods.NewVirtualPodCache(kubeClient)
-	p := provisioning.NewProvisioner(kubeClient, recorder, cloudProvider, cluster, clock, deviceAllocationController, virtualPodCache)
+	p := provisioning.NewProvisioner(kubeClient, recorder, cloudProvider, cluster, clock, deviceAllocationController, virtualPodCache, predictionStore)
 	evictionQueue := terminator.NewQueue(clock, kubeClient, recorder)
 	disruptionQueue := disruption.NewQueue(kubeClient, recorder, cluster, clock, p)
 	npState := nodepoolhealth.NewState()
@@ -206,7 +207,7 @@ func NewControllers(
 		}
 	}
 
-	if !o.disableVPAPrediction {
+	if o.enableVPAPrediction && options.FromContext(ctx).FeatureGates.PredictionEnabled {
 		controllers = append(controllers, informer.NewVPAController(kubeClient, mgr.GetAPIReader(), predictionStore))
 	}
 

pkg/controllers/disruption/controller.go

@@ -48,6 +48,7 @@ import (
 	"sigs.k8s.io/karpenter/pkg/events"
 	"sigs.k8s.io/karpenter/pkg/metrics"
 	"sigs.k8s.io/karpenter/pkg/operator/injection"
+	"sigs.k8s.io/karpenter/pkg/operator/options"
 	"sigs.k8s.io/karpenter/pkg/state/cost"
 	nodepoolutils "sigs.k8s.io/karpenter/pkg/utils/nodepool"
 	"sigs.k8s.io/karpenter/pkg/utils/pretty"
@@ -125,6 +126,7 @@ func (c *Controller) Register(_ context.Context, m manager.Manager) error {
 		Complete(singleton.AsReconciler(c))
 }
 
+//nolint:gocyclo
 func (c *Controller) Reconcile(ctx context.Context) (reconciler.Result, error) {
 	ctx = injection.WithControllerName(ctx, c.Name())
 
@@ -143,6 +145,9 @@ func (c *Controller) Reconcile(ctx context.Context) (reconciler.Result, error) {
 	if !c.cluster.Synced(ctx) {
 		return reconciler.Result{RequeueAfter: time.Second}, nil
 	}
+	if options.FromContext(ctx).FeatureGates.PredictionEnabled && !c.provisioner.PredictionStoreHydrated(ctx) {
+		return reconciler.Result{RequeueAfter: time.Second}, nil
+	}
 
 	// Karpenter taints nodes with a karpenter.sh/disruption taint as part of the disruption process while it progresses in memory.
 	// If Karpenter restarts or fails with an error during a disruption action, some nodes can be left tainted.