Differentiate provisioning requests using Parameters field. Keep prefixing as not recommended approach

Author: macsko

cluster-autoscaler/FAQ.md

@@ -630,9 +630,17 @@ When using this class, Cluster Autoscaler performs following actions:
   Adds a BookingExpired=True condition when the 10-minute reservation period expires.
 
   Since Cluster Autoscaler version 1.33, it is possible to configure the autoscaler 
-  to process only those check capacity ProvisioningRequests, that have a prefix matching the `--check-capacity-provisioning-class-prefix=<prefix>` flag.
-  This allows to run two Cluster Autoscalers in the cluster, but instance with the configured prefix
+  to process only subset of check capacity ProvisioningRequests and ignore the rest.
+  It should be done with caution by specifying `--check-capacity-processor-instance=<name>` flag.
+  Then, ProvReq Parameters map should contain a key "processorInstance" with a value equal to the configured instance name.
+
+  This allows to run two Cluster Autoscalers in the cluster, but the second instance (likely this with configured instance name)
   **should only** handle check capacity ProvisioningRequests and not overlap node groups with the main instance.
+  It is responsibility of the user to ensure the capacity checks are not overlapping.
+  Best-effort atomic ProvisioningRequests processing is disabled in the instance that has this flag set.
+
+  For backwards compatibility, it is possible to differentiate the ProvReqs by prefixing provisioningClassName with the instance name,
+  but it is **not recommended** and will be removed in CA 1.35.
 
 * `best-effort-atomic-scale-up.autoscaling.x-k8s.io` (supported from Cluster Autoscaler version 1.30.2 or later).
 When using this class, Cluster Autoscaler performs following actions:
@@ -978,7 +986,7 @@ The following startup parameters are supported for cluster autoscaler:
 | `bulk-mig-instances-listing-enabled` | Fetch GCE mig instances in bulk instead of per mig |  |
 | `bypassed-scheduler-names` | Names of schedulers to bypass. If set to non-empty value, CA will not wait for pods to reach a certain age before triggering a scale-up. |  |
 | `check-capacity-batch-processing` | Whether to enable batch processing for check capacity requests. |  |
-| `check-capacity-provisioning-class-prefix` | Prefix of provisioningClassName that will be filtered by processors. Only ProvisioningRequests with this prefix in their class will be processed by this CA. It refers only to check capacity ProvisioningRequests. |  |
+| `check-capacity-processor-instance` | Name of the processor instance. Only ProvisioningRequests that define this name in their parameters with the key "processorInstance" will be processed by this CA instance. It only refers to check capacity ProvisioningRequests, but if not empty, best-effort atomic ProvisioningRequests processing is disabled in this instance. Not recommended: Until CA 1.35, ProvisioningRequests with this name as prefix in their class will be also processed. |  |
 | `check-capacity-provisioning-request-batch-timebox` | Maximum time to process a batch of provisioning requests. | 10s |
 | `check-capacity-provisioning-request-max-batch-size` | Maximum number of provisioning requests to process in a single batch. | 10 |
 | `cloud-config` | The path to the cloud provider configuration file. Empty string for no configuration file. |  |
@@ -1022,7 +1030,13 @@ The following startup parameters are supported for cluster autoscaler:
 | `kube-client-qps` | QPS value for kubernetes client. | 5 |
 | `kubeconfig` | Path to kubeconfig file with authorization and master location information. |  |
 | `kubernetes` | Kubernetes master location. Leave blank for default |  |
-| `lease-resource-name` | The lease resource to use in leader election. | "cluster-autoscaler" |
+| `leader-elect` | Start a leader election client and gain leadership before executing the main loop. Enable this when running replicated components for high availability. | true |
+| `leader-elect-lease-duration` | The duration that non-leader candidates will wait after observing a leadership renewal until attempting to acquire leadership of a led but unrenewed leader slot. This is effectively the maximum duration that a leader can be stopped before it is replaced by another candidate. This is only applicable if leader election is enabled. | 15s |
+| `leader-elect-renew-deadline` | The interval between attempts by the acting master to renew a leadership slot before it stops leading. This must be less than the lease duration. This is only applicable if leader election is enabled. | 10s |
+| `leader-elect-resource-lock` | The type of resource object that is used for locking during leader election. Supported options are 'leases'. | "leases" |
+| `leader-elect-resource-name` | The name of resource object that is used for locking during leader election. | "cluster-autoscaler" |
+| `leader-elect-resource-namespace` | The namespace of resource object that is used for locking during leader election. |  |
+| `leader-elect-retry-period` | The duration the clients should wait between attempting acquisition and renewal of a leadership. This is only applicable if leader election is enabled. | 2s |
 | `log-backtrace-at` | when logging hits line file:N, emit a stack trace | :0 |
 | `log-dir` | If non-empty, write log files in this directory (no effect when -logtostderr=true) |  |
 | `log-file` | If non-empty, use this log file (no effect when -logtostderr=true) |  |

cluster-autoscaler/config/autoscaling_options.go

@@ -313,10 +313,11 @@ type AutoscalingOptions struct {
 	DynamicResourceAllocationEnabled bool
 	// ClusterSnapshotParallelism is the maximum parallelism of cluster snapshot creation.
 	ClusterSnapshotParallelism int
-	// CheckCapacityProvisioningClassPrefix is the prefix of provisioningClassName that will be filtered by processors.
-	// Only ProvisioningRequests with this prefix in their class will be processed by this CA.
-	// It only refers to check capacity ProvisioningRequests.
-	CheckCapacityProvisioningClassPrefix string
+	// CheckCapacityProcessorInstance is the name of the processor instance.
+	// Only ProvisioningRequests that define this name in their parameters with the key "processorInstance" will be processed by this CA instance.
+	// It only refers to check capacity ProvisioningRequests, but if not empty, best-effort atomic ProvisioningRequests processing is disabled in this instance.
+	// Not recommended: Until CA 1.35, ProvisioningRequests with this name as prefix in their class will be also processed.
+	CheckCapacityProcessorInstance string
 }
 
 // KubeClientOptions specify options for kube client

cluster-autoscaler/main.go

@@ -283,7 +283,7 @@ var (
 	forceDeleteLongUnregisteredNodes             = flag.Bool("force-delete-unregistered-nodes", false, "Whether to enable force deletion of long unregistered nodes, regardless of the min size of the node group the belong to.")
 	enableDynamicResourceAllocation              = flag.Bool("enable-dynamic-resource-allocation", false, "Whether logic for handling DRA (Dynamic Resource Allocation) objects is enabled.")
 	clusterSnapshotParallelism                   = flag.Int("cluster-snapshot-parallelism", 16, "Maximum parallelism of cluster snapshot creation.")
-	checkCapacityProvisioningClassPrefix         = flag.String("check-capacity-provisioning-class-prefix", "", "Prefix of provisioningClassName that will be filtered by processors. Only ProvisioningRequests with this prefix in their class will be processed by this CA. It refers only to check capacity ProvisioningRequests.")
+	checkCapacityProcessorInstance               = flag.String("check-capacity-processor-instance", "", "Name of the processor instance. Only ProvisioningRequests that define this name in their parameters with the key \"processorInstance\" will be processed by this CA instance. It only refers to check capacity ProvisioningRequests, but if not empty, best-effort atomic ProvisioningRequests processing is disabled in this instance. Not recommended: Until CA 1.35, ProvisioningRequests with this name as prefix in their class will be also processed.")
 )
 
 func isFlagPassed(name string) bool {
@@ -465,7 +465,7 @@ func createAutoscalingOptions() config.AutoscalingOptions {
 		ForceDeleteLongUnregisteredNodes:             *forceDeleteLongUnregisteredNodes,
 		DynamicResourceAllocationEnabled:             *enableDynamicResourceAllocation,
 		ClusterSnapshotParallelism:                   *clusterSnapshotParallelism,
-		CheckCapacityProvisioningClassPrefix:         *checkCapacityProvisioningClassPrefix,
+		CheckCapacityProcessorInstance:               *checkCapacityProcessorInstance,
 	}
 }
 
@@ -541,7 +541,7 @@ func buildAutoscaler(context ctx.Context, debuggingSnapshotter debuggingsnapshot
 			return nil, nil, err
 		}
 
-		ProvisioningRequestInjector, err = provreq.NewProvisioningRequestPodsInjector(restConfig, opts.ProvisioningRequestInitialBackoffTime, opts.ProvisioningRequestMaxBackoffTime, opts.ProvisioningRequestMaxBackoffCacheSize, opts.CheckCapacityBatchProcessing, opts.CheckCapacityProvisioningClassPrefix)
+		ProvisioningRequestInjector, err = provreq.NewProvisioningRequestPodsInjector(restConfig, opts.ProvisioningRequestInitialBackoffTime, opts.ProvisioningRequestMaxBackoffTime, opts.ProvisioningRequestMaxBackoffCacheSize, opts.CheckCapacityBatchProcessing, opts.CheckCapacityProcessorInstance)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -560,7 +560,7 @@ func buildAutoscaler(context ctx.Context, debuggingSnapshotter debuggingsnapshot
 
 		scaleUpOrchestrator := provreqorchestrator.NewWrapperOrchestrator(provreqOrchestrator)
 		opts.ScaleUpOrchestrator = scaleUpOrchestrator
-		provreqProcesor := provreq.NewProvReqProcessor(client, opts.CheckCapacityProvisioningClassPrefix)
+		provreqProcesor := provreq.NewProvReqProcessor(client, opts.CheckCapacityProcessorInstance)
 		opts.LoopStartNotifier = loopstart.NewObserversList([]loopstart.Observer{provreqProcesor})
 
 		podListProcessor.AddProcessor(provreqProcesor)

cluster-autoscaler/processors/provreq/injector.go

@@ -37,14 +37,14 @@ import (
 
 // ProvisioningRequestPodsInjector creates in-memory pods from ProvisioningRequest and inject them to unscheduled pods list.
 type ProvisioningRequestPodsInjector struct {
-	initialRetryTime                     time.Duration
-	maxBackoffTime                       time.Duration
-	backoffDuration                      *lru.Cache
-	clock                                clock.PassiveClock
-	client                               *provreqclient.ProvisioningRequestClient
-	lastProvisioningRequestProcessTime   time.Time
-	checkCapacityBatchProcessing         bool
-	checkCapacityProvisioningClassPrefix string
+	initialRetryTime                   time.Duration
+	maxBackoffTime                     time.Duration
+	backoffDuration                    *lru.Cache
+	clock                              clock.PassiveClock
+	client                             *provreqclient.ProvisioningRequestClient
+	lastProvisioningRequestProcessTime time.Time
+	checkCapacityBatchProcessing       bool
+	checkCapacityProcessorInstance     string
 }
 
 // IsAvailableForProvisioning checks if the provisioning request is the correct state for processing and provisioning has not been attempted recently.
@@ -95,13 +95,17 @@ func (p *ProvisioningRequestPodsInjector) MarkAsFailed(pr *provreqwrapper.Provis
 }
 
 func (p *ProvisioningRequestPodsInjector) isSupportedClass(pr *provreqwrapper.ProvisioningRequest) bool {
-	return provisioningrequest.SupportedProvisioningClass(pr.Spec.ProvisioningClassName, p.checkCapacityProvisioningClassPrefix)
+	return provisioningrequest.SupportedProvisioningClass(pr.ProvisioningRequest, p.checkCapacityProcessorInstance)
+}
+
+func (p *ProvisioningRequestPodsInjector) isSupportedCheckCapacityClass(pr *provreqwrapper.ProvisioningRequest) bool {
+	return provisioningrequest.SupportedCheckCapacityClass(pr.ProvisioningRequest, p.checkCapacityProcessorInstance)
 }
 
 func (p *ProvisioningRequestPodsInjector) shouldMarkAsAccepted(pr *provreqwrapper.ProvisioningRequest) bool {
 	// Don't mark as accepted the check capacity ProvReq when batch processing is enabled.
 	// It will be marked later, in parallel, during processing the requests.
-	return !p.checkCapacityBatchProcessing || !p.matchesCheckCapacityClass(pr.Spec.ProvisioningClassName)
+	return !p.checkCapacityBatchProcessing || !p.isSupportedCheckCapacityClass(pr)
 }
 
 // GetPodsFromNextRequest picks one ProvisioningRequest meeting the condition passed using isSupportedClass function, marks it as accepted and returns pods from it.
@@ -145,10 +149,6 @@ type ProvisioningRequestWithPods struct {
 	Pods      []*apiv1.Pod
 }
 
-func (p *ProvisioningRequestPodsInjector) matchesCheckCapacityClass(provisioningClassName string) bool {
-	return provisioningClassName == p.checkCapacityProvisioningClassPrefix+v1.ProvisioningClassCheckCapacity
-}
-
 // GetCheckCapacityBatch returns up to the requested number of ProvisioningRequestWithPods.
 // We do not mark the PRs as accepted here.
 // If we fail to get the pods for a PR, we mark the PR as failed and issue an update.
@@ -162,7 +162,7 @@ func (p *ProvisioningRequestPodsInjector) GetCheckCapacityBatch(maxPrs int) ([]P
 		if len(prsWithPods) >= maxPrs {
 			break
 		}
-		if !p.matchesCheckCapacityClass(pr.Spec.ProvisioningClassName) {
+		if !p.isSupportedCheckCapacityClass(pr) {
 			continue
 		}
 		if !p.IsAvailableForProvisioning(pr) {
@@ -197,20 +197,20 @@ func (p *ProvisioningRequestPodsInjector) Process(
 func (p *ProvisioningRequestPodsInjector) CleanUp() {}
 
 // NewProvisioningRequestPodsInjector creates a ProvisioningRequest filter processor.
-func NewProvisioningRequestPodsInjector(kubeConfig *rest.Config, initialBackoffTime, maxBackoffTime time.Duration, maxCacheSize int, checkCapacityBatchProcessing bool, checkCapacityProvisioningClassPrefix string) (*ProvisioningRequestPodsInjector, error) {
+func NewProvisioningRequestPodsInjector(kubeConfig *rest.Config, initialBackoffTime, maxBackoffTime time.Duration, maxCacheSize int, checkCapacityBatchProcessing bool, checkCapacityProcessorInstance string) (*ProvisioningRequestPodsInjector, error) {
 	client, err := provreqclient.NewProvisioningRequestClient(kubeConfig)
 	if err != nil {
 		return nil, err
 	}
 	return &ProvisioningRequestPodsInjector{
-		initialRetryTime:                     initialBackoffTime,
-		maxBackoffTime:                       maxBackoffTime,
-		backoffDuration:                      lru.New(maxCacheSize),
-		client:                               client,
-		clock:                                clock.RealClock{},
-		lastProvisioningRequestProcessTime:   time.Now(),
-		checkCapacityBatchProcessing:         checkCapacityBatchProcessing,
-		checkCapacityProvisioningClassPrefix: checkCapacityProvisioningClassPrefix,
+		initialRetryTime:                   initialBackoffTime,
+		maxBackoffTime:                     maxBackoffTime,
+		backoffDuration:                    lru.New(maxCacheSize),
+		client:                             client,
+		clock:                              clock.RealClock{},
+		lastProvisioningRequestProcessTime: time.Now(),
+		checkCapacityBatchProcessing:       checkCapacityBatchProcessing,
+		checkCapacityProcessorInstance:     checkCapacityProcessorInstance,
 	}, nil
 }