Allow to prefix provisioningClassName to filter provisioning requests

Author: macsko

cluster-autoscaler/config/autoscaling_options.go

@@ -313,6 +313,10 @@ 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
 }
 
 // KubeClientOptions specify options for kube client

cluster-autoscaler/main.go

@@ -283,6 +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.")
 )
 
 func isFlagPassed(name string) bool {
@@ -464,6 +465,7 @@ func createAutoscalingOptions() config.AutoscalingOptions {
 		ForceDeleteLongUnregisteredNodes:             *forceDeleteLongUnregisteredNodes,
 		DynamicResourceAllocationEnabled:             *enableDynamicResourceAllocation,
 		ClusterSnapshotParallelism:                   *clusterSnapshotParallelism,
+		CheckCapacityProvisioningClassPrefix:         *checkCapacityProvisioningClassPrefix,
 	}
 }
 
@@ -539,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)
+		ProvisioningRequestInjector, err = provreq.NewProvisioningRequestPodsInjector(restConfig, opts.ProvisioningRequestInitialBackoffTime, opts.ProvisioningRequestMaxBackoffTime, opts.ProvisioningRequestMaxBackoffCacheSize, opts.CheckCapacityBatchProcessing, opts.CheckCapacityProvisioningClassPrefix)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -558,7 +560,7 @@ func buildAutoscaler(context ctx.Context, debuggingSnapshotter debuggingsnapshot
 
 		scaleUpOrchestrator := provreqorchestrator.NewWrapperOrchestrator(provreqOrchestrator)
 		opts.ScaleUpOrchestrator = scaleUpOrchestrator
-		provreqProcesor := provreq.NewProvReqProcessor(client)
+		provreqProcesor := provreq.NewProvReqProcessor(client, opts.CheckCapacityProvisioningClassPrefix)
 		opts.LoopStartNotifier = loopstart.NewObserversList([]loopstart.Observer{provreqProcesor})
 
 		podListProcessor.AddProcessor(provreqProcesor)

cluster-autoscaler/processors/provreq/injector.go

@@ -37,13 +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
+	initialRetryTime                     time.Duration
+	maxBackoffTime                       time.Duration
+	backoffDuration                      *lru.Cache
+	clock                                clock.PassiveClock
+	client                               *provreqclient.ProvisioningRequestClient
+	lastProvisioningRequestProcessTime   time.Time
+	checkCapacityBatchProcessing         bool
+	checkCapacityProvisioningClassPrefix string
 }
 
 // IsAvailableForProvisioning checks if the provisioning request is the correct state for processing and provisioning has not been attempted recently.
@@ -93,16 +94,24 @@ func (p *ProvisioningRequestPodsInjector) MarkAsFailed(pr *provreqwrapper.Provis
 	p.UpdateLastProcessTime()
 }
 
+func (p *ProvisioningRequestPodsInjector) isSupportedClass(pr *provreqwrapper.ProvisioningRequest) bool {
+	return provisioningrequest.SupportedProvisioningClass(pr.Spec.ProvisioningClassName, p.checkCapacityProvisioningClassPrefix)
+}
+
+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)
+}
+
 // GetPodsFromNextRequest picks one ProvisioningRequest meeting the condition passed using isSupportedClass function, marks it as accepted and returns pods from it.
-func (p *ProvisioningRequestPodsInjector) GetPodsFromNextRequest(
-	isSupportedClass func(*provreqwrapper.ProvisioningRequest) bool,
-) ([]*apiv1.Pod, error) {
+func (p *ProvisioningRequestPodsInjector) GetPodsFromNextRequest() ([]*apiv1.Pod, error) {
 	provReqs, err := p.client.ProvisioningRequests()
 	if err != nil {
 		return nil, err
 	}
 	for _, pr := range provReqs {
-		if !isSupportedClass(pr) {
+		if !p.isSupportedClass(pr) {
 			continue
 		}
 
@@ -117,16 +126,13 @@ func (p *ProvisioningRequestPodsInjector) GetPodsFromNextRequest(
 			p.MarkAsFailed(pr, provreqconditions.FailedToCreatePodsReason, err.Error())
 			continue
 		}
-		// 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.
-		if pr.Spec.ProvisioningClassName == v1.ProvisioningClassCheckCapacity && p.checkCapacityBatchProcessing {
-			p.UpdateLastProcessTime()
+		if p.shouldMarkAsAccepted(pr) {
+			if err := p.MarkAsAccepted(pr); err != nil {
+				continue
+			}
 			return podsFromProvReq, nil
 		}
-		if err := p.MarkAsAccepted(pr); err != nil {
-			continue
-		}
-
+		p.UpdateLastProcessTime()
 		return podsFromProvReq, nil
 	}
 	return nil, nil
@@ -139,6 +145,10 @@ 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.
@@ -152,7 +162,7 @@ func (p *ProvisioningRequestPodsInjector) GetCheckCapacityBatch(maxPrs int) ([]P
 		if len(prsWithPods) >= maxPrs {
 			break
 		}
-		if pr.Spec.ProvisioningClassName != v1.ProvisioningClassCheckCapacity {
+		if !p.matchesCheckCapacityClass(pr.Spec.ProvisioningClassName) {
 			continue
 		}
 		if !p.IsAvailableForProvisioning(pr) {
@@ -175,15 +185,7 @@ func (p *ProvisioningRequestPodsInjector) Process(
 	_ *context.AutoscalingContext,
 	unschedulablePods []*apiv1.Pod,
 ) ([]*apiv1.Pod, error) {
-	podsFromProvReq, err := p.GetPodsFromNextRequest(
-		func(pr *provreqwrapper.ProvisioningRequest) bool {
-			_, found := provisioningrequest.SupportedProvisioningClasses[pr.Spec.ProvisioningClassName]
-			if !found {
-				klog.Warningf("Provisioning Class %s is not supported for ProvReq %s/%s", pr.Spec.ProvisioningClassName, pr.Namespace, pr.Name)
-			}
-			return found
-		})
-
+	podsFromProvReq, err := p.GetPodsFromNextRequest()
 	if err != nil {
 		return unschedulablePods, err
 	}
@@ -195,19 +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) (*ProvisioningRequestPodsInjector, error) {
+func NewProvisioningRequestPodsInjector(kubeConfig *rest.Config, initialBackoffTime, maxBackoffTime time.Duration, maxCacheSize int, checkCapacityBatchProcessing bool, checkCapacityProvisioningClassPrefix 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,
+		initialRetryTime:                     initialBackoffTime,
+		maxBackoffTime:                       maxBackoffTime,
+		backoffDuration:                      lru.New(maxCacheSize),
+		client:                               client,
+		clock:                                clock.RealClock{},
+		lastProvisioningRequestProcessTime:   time.Now(),
+		checkCapacityBatchProcessing:         checkCapacityBatchProcessing,
+		checkCapacityProvisioningClassPrefix: checkCapacityProvisioningClassPrefix,
 	}, nil
 }
 

cluster-autoscaler/processors/provreq/injector_test.go

@@ -69,6 +69,7 @@ func TestProvisioningRequestPodsInjector(t *testing.T) {
 	podsA := 10
 	newProvReqA := testProvisioningRequestWithCondition("new", podsA, v1.ProvisioningClassCheckCapacity)
 	newAcceptedProvReqA := testProvisioningRequestWithCondition("new-accepted", podsA, v1.ProvisioningClassCheckCapacity, accepted)
+	newProvReqAPrefixed := testProvisioningRequestWithCondition("new-prefixed", podsA, "test-prefix.check-capacity.autoscaling.x-k8s.io")
 
 	podsB := 20
 	notProvisionedAcceptedProvReqB := testProvisioningRequestWithCondition("provisioned-false-B", podsB, v1.ProvisioningClassBestEffortAtomicScaleUp, notProvisioned, accepted)
@@ -79,20 +80,20 @@ func TestProvisioningRequestPodsInjector(t *testing.T) {
 	unknownClass := testProvisioningRequestWithCondition("new-accepted", podsA, "unknown-class", accepted)
 
 	testCases := []struct {
-		name                             string
-		provReqs                         []*provreqwrapper.ProvisioningRequest
-		existingUnsUnschedulablePodCount int
-		checkCapacityBatchProcessing     bool
-		wantUnscheduledPodCount          int
-		wantUpdatedConditionName         string
+		name                                 string
+		provReqs                             []*provreqwrapper.ProvisioningRequest
+		existingUnsUnschedulablePodCount     int
+		checkCapacityBatchProcessing         bool
+		checkCapacityProvisioningClassPrefix string
+		wantUnscheduledPodCount              int
+		wantUpdatedConditionName             string
 	}{
 		{
 			name:                     "New ProvisioningRequest, pods are injected and Accepted condition is added",
 			provReqs:                 []*provreqwrapper.ProvisioningRequest{newProvReqA, provisionedAcceptedProvReqB},
 			wantUnscheduledPodCount:  podsA,
 			wantUpdatedConditionName: newProvReqA.Name,
 		},
-
 		{
 			name:                         "New check capacity ProvisioningRequest with batch processing, pods are injected and Accepted condition is not added",
 			provReqs:                     []*provreqwrapper.ProvisioningRequest{newProvReqA, provisionedAcceptedProvReqB},
@@ -106,6 +107,22 @@ func TestProvisioningRequestPodsInjector(t *testing.T) {
 			wantUnscheduledPodCount:  podsA,
 			wantUpdatedConditionName: newAcceptedProvReqA.Name,
 		},
+		{
+			name:     "New ProvisioningRequest with not matching custom prefix, no pods are injected",
+			provReqs: []*provreqwrapper.ProvisioningRequest{newProvReqAPrefixed},
+		},
+		{
+			name:                                 "New ProvisioningRequest with not matching prefix, no pods are injected",
+			provReqs:                             []*provreqwrapper.ProvisioningRequest{newProvReqA, provisionedAcceptedProvReqB},
+			checkCapacityProvisioningClassPrefix: "test-prefix.",
+		},
+		{
+			name:                                 "New check capacity ProvisioningRequest with matching prefix, pods are injected and Accepted condition is added",
+			provReqs:                             []*provreqwrapper.ProvisioningRequest{newProvReqAPrefixed, provisionedAcceptedProvReqB},
+			checkCapacityProvisioningClassPrefix: "test-prefix.",
+			wantUnscheduledPodCount:              podsA,
+			wantUpdatedConditionName:             newProvReqAPrefixed.Name,
+		},
 		{
 			name:                     "Provisioned=False, pods are injected",
 			provReqs:                 []*provreqwrapper.ProvisioningRequest{notProvisionedAcceptedProvReqB, failedProvReq},
@@ -140,7 +157,7 @@ func TestProvisioningRequestPodsInjector(t *testing.T) {
 		client := provreqclient.NewFakeProvisioningRequestClient(context.Background(), t, tc.provReqs...)
 		backoffTime := lru.New(100)
 		backoffTime.Add(key(notProvisionedRecentlyProvReqB), 2*time.Minute)
-		injector := ProvisioningRequestPodsInjector{1 * time.Minute, 10 * time.Minute, backoffTime, clock.NewFakePassiveClock(now), client, now, tc.checkCapacityBatchProcessing}
+		injector := ProvisioningRequestPodsInjector{1 * time.Minute, 10 * time.Minute, backoffTime, clock.NewFakePassiveClock(now), client, now, tc.checkCapacityBatchProcessing, tc.checkCapacityProvisioningClassPrefix}
 		getUnscheduledPods, err := injector.Process(nil, provreqwrapper.BuildTestPods("ns", "pod", tc.existingUnsUnschedulablePodCount))
 		if err != nil {
 			t.Errorf("%s failed: injector.Process return error %v", tc.name, err)

cluster-autoscaler/processors/provreq/processor.go

@@ -50,15 +50,16 @@ type injector interface {
 }
 
 type provReqProcessor struct {
-	now        func() time.Time
-	maxUpdated int
-	client     *provreqclient.ProvisioningRequestClient
-	injector   injector
+	now                                  func() time.Time
+	maxUpdated                           int
+	client                               *provreqclient.ProvisioningRequestClient
+	injector                             injector
+	checkCapacityProvisioningClassPrefix string
 }
 
 // NewProvReqProcessor return ProvisioningRequestProcessor.
-func NewProvReqProcessor(client *provreqclient.ProvisioningRequestClient) *provReqProcessor {
-	return &provReqProcessor{now: time.Now, maxUpdated: defaultMaxUpdated, client: client, injector: scheduling.NewHintingSimulator()}
+func NewProvReqProcessor(client *provreqclient.ProvisioningRequestClient, checkCapacityProvisioningClassPrefix string) *provReqProcessor {
+	return &provReqProcessor{now: time.Now, maxUpdated: defaultMaxUpdated, client: client, injector: scheduling.NewHintingSimulator(), checkCapacityProvisioningClassPrefix: checkCapacityProvisioningClassPrefix}
 }
 
 // Refresh implements loop.Observer interface and will be run at the start
@@ -84,7 +85,7 @@ func (p *provReqProcessor) refresh(provReqs []*provreqwrapper.ProvisioningReques
 		if len(expiredProvReq) >= p.maxUpdated {
 			break
 		}
-		if ok, found := provisioningrequest.SupportedProvisioningClasses[provReq.Spec.ProvisioningClassName]; !ok || !found {
+		if !provisioningrequest.SupportedProvisioningClass(provReq.Spec.ProvisioningClassName, p.checkCapacityProvisioningClassPrefix) {
 			continue
 		}
 		conditions := provReq.Status.Conditions
@@ -144,7 +145,7 @@ func (p *provReqProcessor) bookCapacity(ctx *context.AutoscalingContext) error {
 	}
 	podsToCreate := []*apiv1.Pod{}
 	for _, provReq := range provReqs {
-		if !conditions.ShouldCapacityBeBooked(provReq) {
+		if !conditions.ShouldCapacityBeBooked(provReq, p.checkCapacityProvisioningClassPrefix) {
 			continue
 		}
 		pods, err := provreq_pods.PodsForProvisioningRequest(provReq)

cluster-autoscaler/processors/provreq/processor_test.go

@@ -155,7 +155,7 @@ func TestRefresh(t *testing.T) {
 		additionalPr.CreationTimestamp = metav1.NewTime(weekAgo)
 		additionalPr.Spec.ProvisioningClassName = v1.ProvisioningClassCheckCapacity
 
-		processor := provReqProcessor{func() time.Time { return now }, 1, provreqclient.NewFakeProvisioningRequestClient(nil, t, pr, additionalPr), nil}
+		processor := provReqProcessor{func() time.Time { return now }, 1, provreqclient.NewFakeProvisioningRequestClient(nil, t, pr, additionalPr), nil, ""}
 		processor.refresh([]*provreqwrapper.ProvisioningRequest{pr, additionalPr})
 
 		assert.ElementsMatch(t, test.wantConditions, pr.Status.Conditions)
@@ -215,7 +215,7 @@ func TestDeleteOldProvReqs(t *testing.T) {
 
 	client := provreqclient.NewFakeProvisioningRequestClient(nil, t, pr, additionalPr, oldFailedPr, oldExpiredPr)
 
-	processor := provReqProcessor{func() time.Time { return now }, 1, client, nil}
+	processor := provReqProcessor{func() time.Time { return now }, 1, client, nil, ""}
 	processor.refresh([]*provreqwrapper.ProvisioningRequest{pr, additionalPr, oldFailedPr, oldExpiredPr})
 
 	_, err := client.ProvisioningRequestNoCache(oldFailedPr.Namespace, oldFailedPr.Name)

cluster-autoscaler/provisioningrequest/checkcapacity/provisioningclass.go

@@ -132,7 +132,7 @@ func (o *checkCapacityProvClass) getProvisioningRequestsAndPods(unschedulablePod
 	if !o.isBatchEnabled() {
 		klog.Info("Processing single provisioning request (non-batch)")
 		prs := provreqclient.ProvisioningRequestsForPods(o.client, unschedulablePods)
-		prs = provreqclient.FilterOutProvisioningClass(prs, v1.ProvisioningClassCheckCapacity)
+		prs = provreqclient.FilterOutProvisioningClass(prs, o.context.CheckCapacityProvisioningClassPrefix+v1.ProvisioningClassCheckCapacity)
 		if len(prs) == 0 {
 			return nil, nil
 		}