Apply CPU startup boost in admission controller if its set

Author: kamarabbas99

vertical-pod-autoscaler/docs/flags.md

@@ -24,6 +24,7 @@ This document is auto-generated from the flag definitions in the VPA admission-c
 | `log-file` | string |  | If non-empty, use this log file (no effect when -logtostderr=true) |
 | `log-file-max-size` | int |  1800 | uDefines the maximum size a log file can grow to (no effect when -logtostderr=true). Unit is megabytes. If the value is 0, the maximum file size is unlimited.  |
 | `logtostderr` |  |  true | log to standard error instead of files  |
+| `max-allowed-cpu-boost` | string |  | Maximum amount of CPU that will be applied for a container with boost. |
 | `min-tls-version` | string |  | The minimum TLS version to accept.  Must be set to either tls1_2  or tls1_3. (default "tls1_2") |
 | `one-output` | severity |  | If true, only write logs to their native level (vs also writing to each lower severity level; no effect when -logtostderr=true) |
 | `port` | int |  8000 | The port to listen on.  |

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

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/spf13/pflag"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/client-go/informers"
 	kube_client "k8s.io/client-go/kubernetes"
 	typedadmregv1 "k8s.io/client-go/kubernetes/typed/admissionregistration/v1"
@@ -78,6 +79,7 @@ var (
 	registerWebhook      = flag.Bool("register-webhook", true, "If set to true, admission webhook object will be created on start up to register with the API server.")
 	webhookLabels        = flag.String("webhook-labels", "", "Comma separated list of labels to add to the webhook object. Format: key1:value1,key2:value2")
 	registerByURL        = flag.Bool("register-by-url", false, "If set to true, admission webhook will be registered by URL (webhookAddress:webhookPort) instead of by service name")
+	maxAllowedCPUBoost   = flag.String("max-allowed-cpu-boost", "", "Maximum amount of CPU that will be applied for a container with boost.")
 )
 
 func main() {
@@ -93,6 +95,13 @@ func main() {
 		klog.FlushAndExit(klog.ExitFlushTimeout, 1)
 	}
 
+	if *maxAllowedCPUBoost != "" {
+		if _, err := resource.ParseQuantity(*maxAllowedCPUBoost); err != nil {
+			klog.ErrorS(err, "Failed to parse maxAllowedCPUBoost")
+			klog.FlushAndExit(klog.ExitFlushTimeout, 1)
+		}
+	}
+
 	healthCheck := metrics.NewHealthCheck(time.Minute)
 	metrics_admission.Register()
 	server.Initialize(&commonFlags.EnableProfiling, healthCheck, address)
@@ -145,7 +154,7 @@ func main() {
 		hostname,
 	)
 
-	calculators := []patch.Calculator{patch.NewResourceUpdatesCalculator(recommendationProvider), patch.NewObservedContainersCalculator()}
+	calculators := []patch.Calculator{patch.NewResourceUpdatesCalculator(recommendationProvider, *maxAllowedCPUBoost), patch.NewObservedContainersCalculator()}
 	as := logic.NewAdmissionServer(podPreprocessor, vpaPreprocessor, limitRangeCalculator, vpaMatcher, calculators)
 	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		as.Serve(w, r)

vertical-pod-autoscaler/pkg/admission-controller/resource/pod/patch/resource_updates.go

@@ -21,10 +21,13 @@ import (
 	"strings"
 
 	core "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 
 	resource_admission "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/admission-controller/resource"
 	"k8s.io/autoscaler/vertical-pod-autoscaler/pkg/admission-controller/resource/pod/recommendation"
 	vpa_types "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/apis/autoscaling.k8s.io/v1"
+	"k8s.io/autoscaler/vertical-pod-autoscaler/pkg/features"
+	"k8s.io/autoscaler/vertical-pod-autoscaler/pkg/utils/annotations"
 	resourcehelpers "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/utils/resources"
 	vpa_api_util "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/utils/vpa"
 )
@@ -37,13 +40,19 @@ const (
 
 type resourcesUpdatesPatchCalculator struct {
 	recommendationProvider recommendation.Provider
+	maxAllowedCPUBoost     resource.Quantity
 }
 
 // NewResourceUpdatesCalculator returns a calculator for
 // resource update patches.
-func NewResourceUpdatesCalculator(recommendationProvider recommendation.Provider) Calculator {
+func NewResourceUpdatesCalculator(recommendationProvider recommendation.Provider, maxAllowedCPUBoost string) Calculator {
+	var maxAllowedCPUBoostQuantity resource.Quantity
+	if maxAllowedCPUBoost != "" {
+		maxAllowedCPUBoostQuantity = resource.MustParse(maxAllowedCPUBoost)
+	}
 	return &resourcesUpdatesPatchCalculator{
 		recommendationProvider: recommendationProvider,
+		maxAllowedCPUBoost:     maxAllowedCPUBoostQuantity,
 	}
 }
 
@@ -59,15 +68,42 @@ func (c *resourcesUpdatesPatchCalculator) CalculatePatches(pod *core.Pod, vpa *v
 		return []resource_admission.PatchRecord{}, fmt.Errorf("failed to calculate resource patch for pod %s/%s: %v", pod.Namespace, pod.Name, err)
 	}
 
+	if vpa_api_util.GetUpdateMode(vpa) == vpa_types.UpdateModeOff {
+		// If update mode is "Off", we don't want to apply any recommendations,
+		// but we still want to apply startup boost.
+		for i := range containersResources {
+			containersResources[i].Requests = nil
+			containersResources[i].Limits = nil
+		}
+		annotationsPerContainer = vpa_api_util.ContainerToAnnotationsMap{}
+	}
+
 	if annotationsPerContainer == nil {
 		annotationsPerContainer = vpa_api_util.ContainerToAnnotationsMap{}
 	}
 
 	updatesAnnotation := []string{}
-	for i, containerResources := range containersResources {
-		newPatches, newUpdatesAnnotation := getContainerPatch(pod, i, annotationsPerContainer, containerResources)
-		result = append(result, newPatches...)
-		updatesAnnotation = append(updatesAnnotation, newUpdatesAnnotation)
+	for i := range containersResources {
+
+		// Apply startup boost if configured
+		if features.Enabled(features.CPUStartupBoost) {
+			// Get the container resource policy to check for scaling mode.
+			policy := vpa_api_util.GetContainerResourcePolicy(pod.Spec.Containers[i].Name, vpa.Spec.ResourcePolicy)
+			if policy != nil && policy.Mode != nil && *policy.Mode == vpa_types.ContainerScalingModeOff {
+				continue
+			}
+			boostPatches, err := c.applyCPUStartupBoost(i, &pod.Spec.Containers[i], vpa, &containersResources[i])
+			if err != nil {
+				return nil, err
+			}
+			result = append(result, boostPatches...)
+		}
+
+		newPatches, newUpdatesAnnotation := getContainerPatch(pod, i, annotationsPerContainer, &containersResources[i])
+		if len(newPatches) > 0 {
+			result = append(result, newPatches...)
+			updatesAnnotation = append(updatesAnnotation, newUpdatesAnnotation)
+		}
 	}
 
 	if len(updatesAnnotation) > 0 {
@@ -77,7 +113,7 @@ func (c *resourcesUpdatesPatchCalculator) CalculatePatches(pod *core.Pod, vpa *v
 	return result, nil
 }
 
-func getContainerPatch(pod *core.Pod, i int, annotationsPerContainer vpa_api_util.ContainerToAnnotationsMap, containerResources vpa_api_util.ContainerResources) ([]resource_admission.PatchRecord, string) {
+func getContainerPatch(pod *core.Pod, i int, annotationsPerContainer vpa_api_util.ContainerToAnnotationsMap, containerResources *vpa_api_util.ContainerResources) ([]resource_admission.PatchRecord, string) {
 	var patches []resource_admission.PatchRecord
 	// Add empty resources object if missing.
 	requests, limits := resourcehelpers.ContainerRequestsAndLimits(pod.Spec.Containers[i].Name, pod)
@@ -108,3 +144,114 @@ func appendPatchesAndAnnotations(patches []resource_admission.PatchRecord, annot
 	}
 	return patches, annotations
 }
+
+func (c *resourcesUpdatesPatchCalculator) applyCPUStartupBoost(containerIndex int, container *core.Container, vpa *vpa_types.VerticalPodAutoscaler, containerResources *vpa_api_util.ContainerResources) ([]resource_admission.PatchRecord, error) {
+	var patches []resource_admission.PatchRecord
+
+	startupBoostPolicy := getContainerStartupBoostPolicy(container, vpa)
+	if startupBoostPolicy == nil {
+		return nil, nil
+	}
+
+	originalRequest := container.Resources.Requests[core.ResourceCPU]
+	boostedRequest, err := calculateBoostedCPU(originalRequest, startupBoostPolicy)
+	if err != nil {
+		return nil, err
+	}
+
+	if !c.maxAllowedCPUBoost.IsZero() && boostedRequest.Cmp(c.maxAllowedCPUBoost) > 0 {
+		boostedRequest = &c.maxAllowedCPUBoost
+	}
+
+	controlledCPUResourcePatches, err := c.applyControlledCPUResources(containerIndex, container, vpa, containerResources, boostedRequest, startupBoostPolicy)
+	if err != nil {
+		return nil, err
+	}
+	patches = append(patches, controlledCPUResourcePatches...)
+
+	originalResources, err := annotations.GetOriginalResourcesAnnotationValue(container)
+	if err != nil {
+		return nil, err
+	}
+	patches = append(patches, GetAddAnnotationPatch(annotations.StartupCPUBoostAnnotation, originalResources))
+
+	return patches, nil
+}
+
+func getContainerStartupBoostPolicy(container *core.Container, vpa *vpa_types.VerticalPodAutoscaler) *vpa_types.StartupBoost {
+	policy := vpa_api_util.GetContainerResourcePolicy(container.Name, vpa.Spec.ResourcePolicy)
+	startupBoost := vpa.Spec.StartupBoost
+	if policy != nil && policy.StartupBoost != nil {
+		startupBoost = policy.StartupBoost
+	}
+	return startupBoost
+}
+
+func calculateBoostedCPU(baseCPU resource.Quantity, startupBoost *vpa_types.StartupBoost) (*resource.Quantity, error) {
+	if startupBoost == nil {
+		return &baseCPU, nil
+	}
+
+	boostType := startupBoost.CPU.Type
+	if boostType == "" {
+		boostType = vpa_types.FactorStartupBoostType
+	}
+
+	switch boostType {
+	case vpa_types.FactorStartupBoostType:
+		if startupBoost.CPU.Factor == nil {
+			return nil, fmt.Errorf("startupBoost.CPU.Factor is required when Type is Factor or not specified")
+		}
+		factor := *startupBoost.CPU.Factor
+		if factor < 1 {
+			return nil, fmt.Errorf("boost factor must be >= 1")
+		}
+		boostedCPU := baseCPU.MilliValue()
+		boostedCPU = int64(float64(boostedCPU) * float64(factor))
+		return resource.NewMilliQuantity(boostedCPU, resource.DecimalSI), nil
+	case vpa_types.QuantityStartupBoostType:
+		if startupBoost.CPU.Quantity == nil {
+			return nil, fmt.Errorf("startupBoost.CPU.Quantity is required when Type is Quantity")
+		}
+		quantity := *startupBoost.CPU.Quantity
+		boostedCPU := baseCPU.MilliValue() + quantity.MilliValue()
+		return resource.NewMilliQuantity(boostedCPU, resource.DecimalSI), nil
+	default:
+		return nil, fmt.Errorf("unsupported startup boost type: %s", startupBoost.CPU.Type)
+	}
+}
+
+func (c *resourcesUpdatesPatchCalculator) applyControlledCPUResources(containerIndex int, container *core.Container, vpa *vpa_types.VerticalPodAutoscaler, containerResources *vpa_api_util.ContainerResources, boostedRequest *resource.Quantity, startupBoostPolicy *vpa_types.StartupBoost) ([]resource_admission.PatchRecord, error) {
+	controlledValues := vpa_api_util.GetContainerControlledValues(container.Name, vpa.Spec.ResourcePolicy)
+
+	// Apply CPU Requests
+	if containerResources.Requests == nil {
+		containerResources.Requests = core.ResourceList{}
+	}
+	resourceList := core.ResourceList{core.ResourceCPU: *boostedRequest}
+	if controlledValues == vpa_types.ContainerControlledValuesRequestsOnly {
+		vpa_api_util.CapRecommendationToContainerLimit(resourceList, container.Resources.Limits)
+	}
+	containerResources.Requests[core.ResourceCPU] = resourceList[core.ResourceCPU]
+
+	// Apply CPU Limits
+	if controlledValues == vpa_types.ContainerControlledValuesRequestsAndLimits {
+		if containerResources.Limits == nil {
+			containerResources.Limits = core.ResourceList{}
+		}
+		originalLimit := container.Resources.Limits[core.ResourceCPU]
+		if originalLimit.IsZero() {
+			originalLimit = container.Resources.Requests[core.ResourceCPU]
+		}
+		boostedLimit, err := calculateBoostedCPU(originalLimit, startupBoostPolicy)
+		if err != nil {
+			return nil, err
+		}
+		if !c.maxAllowedCPUBoost.IsZero() && boostedLimit.Cmp(c.maxAllowedCPUBoost) > 0 {
+			boostedLimit = &c.maxAllowedCPUBoost
+		}
+		containerResources.Limits[core.ResourceCPU] = *boostedLimit
+	}
+
+	return nil, nil
+}