Merge pull request #6873 from aleksandra-malinowska/provreq-processor

Use the same processors for all currently supported provisioning classes

cluster-autoscaler/main.go

@@ -508,7 +508,7 @@ func buildAutoscaler(debuggingSnapshotter debuggingsnapshot.DebuggingSnapshotter
 		scaleUpOrchestrator := provreqorchestrator.NewWrapperOrchestrator(provreqOrchestrator)
 
 		opts.ScaleUpOrchestrator = scaleUpOrchestrator
-		provreqProcesor := provreq.NewCombinedProvReqProcessor(client, []provreq.ProvisioningRequestProcessor{checkcapacity.NewCheckCapacityProcessor(client)})
+		provreqProcesor := provreq.NewProvReqProcessor(client)
 		if err != nil {
 			return nil, err
 		}

cluster-autoscaler/processors/provreq/provisioning_request_injector.go → cluster-autoscaler/processors/provreq/injector.go

@@ -37,9 +37,6 @@ const (
 	defaultRetryTime = 10 * time.Minute
 )
 
-// SupportedProvisioningClasses is a list of supported ProvisioningClasses in ClusterAutoscaler.
-var SupportedProvisioningClasses = []string{v1beta1.ProvisioningClassCheckCapacity}
-
 // ProvisioningRequestPodsInjector creates in-memory pods from ProvisioningRequest and inject them to unscheduled pods list.
 type ProvisioningRequestPodsInjector struct {
 	client *provreqclient.ProvisioningRequestClient

cluster-autoscaler/provisioningrequest/checkcapacity/processor_test.go → cluster-autoscaler/processors/provreq/processor_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package checkcapacity
+package provreq
 
 import (
 	"testing"
@@ -146,7 +146,7 @@ func TestProcess(t *testing.T) {
 		additionalPr := provreqclient.ProvisioningRequestWrapperForTesting("namespace", "additional")
 		additionalPr.CreationTimestamp = metav1.NewTime(weekAgo)
 		additionalPr.Spec.ProvisioningClassName = v1beta1.ProvisioningClassCheckCapacity
-		processor := checkCapacityProcessor{func() time.Time { return now }, 1, provreqclient.NewFakeProvisioningRequestClient(nil, t, pr, additionalPr)}
+		processor := provReqProcessor{func() time.Time { return now }, 1, provreqclient.NewFakeProvisioningRequestClient(nil, t, pr, additionalPr)}
 		processor.Process([]*provreqwrapper.ProvisioningRequest{pr, additionalPr})
 		assert.ElementsMatch(t, test.wantConditions, pr.Status.Conditions)
 		if len(test.conditions) == len(test.wantConditions) {

cluster-autoscaler/processors/provreq/provisioning_request_processor.go

@@ -17,41 +17,96 @@ limitations under the License.
 package provreq
 
 import (
-	"k8s.io/autoscaler/cluster-autoscaler/observers/loopstart"
+	"time"
+
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/autoscaler/cluster-autoscaler/apis/provisioningrequest/autoscaling.x-k8s.io/v1beta1"
+	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest"
+	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/conditions"
 	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/provreqclient"
 	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/provreqwrapper"
 	"k8s.io/klog/v2"
 )
 
-// ProvisioningRequestProcessor process ProvisioningRequests in the cluster.
-type ProvisioningRequestProcessor interface {
-	Process([]*provreqwrapper.ProvisioningRequest)
-	CleanUp()
-}
+const (
+	defaultReservationTime = 10 * time.Minute
+	defaultExpirationTime  = 7 * 24 * time.Hour // 7 days
+	// defaultMaxUpdated is a limit for ProvisioningRequest to update conditions in one ClusterAutoscaler loop.
+	defaultMaxUpdated = 20
+)
 
-// CombinedProvReqProcessor is responsible for processing ProvisioningRequest for each ProvisioningClass
-// every CA loop and updating conditions for expired ProvisioningRequests.
-type CombinedProvReqProcessor struct {
+type provReqProcessor struct {
+	now        func() time.Time
+	maxUpdated int
 	client     *provreqclient.ProvisioningRequestClient
-	processors []ProvisioningRequestProcessor
 }
 
-// NewCombinedProvReqProcessor return new CombinedProvReqProcessor.
-func NewCombinedProvReqProcessor(client *provreqclient.ProvisioningRequestClient, processors []ProvisioningRequestProcessor) loopstart.Observer {
-	return &CombinedProvReqProcessor{client: client, processors: processors}
+// NewProvReqProcessor return ProvisioningRequestProcessor.
+func NewProvReqProcessor(client *provreqclient.ProvisioningRequestClient) *provReqProcessor {
+	return &provReqProcessor{now: time.Now, maxUpdated: defaultMaxUpdated, client: client}
 }
 
-// Refresh iterates over ProvisioningRequests and updates its conditions/state.
-func (cp *CombinedProvReqProcessor) Refresh() {
-	provReqs, err := cp.client.ProvisioningRequests()
+// Refresh implements loop.Observer interface and will be run at the start
+// of every iteration of the main loop. It tries to fetch current
+// ProvisioningRequests and processes up to p.maxUpdated of them.
+func (p *provReqProcessor) Refresh() {
+	provReqs, err := p.client.ProvisioningRequests()
 	if err != nil {
 		klog.Errorf("Failed to get ProvisioningRequests list, err: %v", err)
 		return
 	}
-	for _, p := range cp.processors {
-		p.Process(provReqs)
+
+	p.Process(provReqs)
+}
+
+// Process iterates over ProvisioningRequests and apply:
+// -BookingExpired condition for Provisioned ProvisioningRequest if capacity reservation time is expired.
+// -Failed condition for ProvisioningRequest that were not provisioned during defaultExpirationTime.
+// TODO(yaroslava): fetch reservation and expiration time from ProvisioningRequest
+func (p *provReqProcessor) Process(provReqs []*provreqwrapper.ProvisioningRequest) {
+	expiredProvReq := []*provreqwrapper.ProvisioningRequest{}
+	failedProvReq := []*provreqwrapper.ProvisioningRequest{}
+	for _, provReq := range provReqs {
+		if len(expiredProvReq) >= p.maxUpdated {
+			break
+		}
+		if ok, found := provisioningrequest.SupportedProvisioningClasses[provReq.Spec.ProvisioningClassName]; !ok || !found {
+			continue
+		}
+		conditions := provReq.Status.Conditions
+		if apimeta.IsStatusConditionTrue(conditions, v1beta1.BookingExpired) || apimeta.IsStatusConditionTrue(conditions, v1beta1.Failed) {
+			continue
+		}
+		provisioned := apimeta.FindStatusCondition(conditions, v1beta1.Provisioned)
+		if provisioned != nil && provisioned.Status == metav1.ConditionTrue {
+			if provisioned.LastTransitionTime.Add(defaultReservationTime).Before(p.now()) {
+				expiredProvReq = append(expiredProvReq, provReq)
+			}
+		} else if len(failedProvReq) < p.maxUpdated-len(expiredProvReq) {
+			created := provReq.CreationTimestamp
+			if created.Add(defaultExpirationTime).Before(p.now()) {
+				failedProvReq = append(failedProvReq, provReq)
+			}
+		}
+	}
+	for _, provReq := range expiredProvReq {
+		conditions.AddOrUpdateCondition(provReq, v1beta1.BookingExpired, metav1.ConditionTrue, conditions.CapacityReservationTimeExpiredReason, conditions.CapacityReservationTimeExpiredMsg, metav1.NewTime(p.now()))
+		_, updErr := p.client.UpdateProvisioningRequest(provReq.ProvisioningRequest)
+		if updErr != nil {
+			klog.Errorf("failed to add BookingExpired condition to ProvReq %s/%s, err: %v", provReq.Namespace, provReq.Name, updErr)
+			continue
+		}
+	}
+	for _, provReq := range failedProvReq {
+		conditions.AddOrUpdateCondition(provReq, v1beta1.Failed, metav1.ConditionTrue, conditions.ExpiredReason, conditions.ExpiredMsg, metav1.NewTime(p.now()))
+		_, updErr := p.client.UpdateProvisioningRequest(provReq.ProvisioningRequest)
+		if updErr != nil {
+			klog.Errorf("failed to add Failed condition to ProvReq %s/%s, err: %v", provReq.Namespace, provReq.Name, updErr)
+			continue
+		}
 	}
 }
 
-// CleanUp cleans up internal state
-func (cp *CombinedProvReqProcessor) CleanUp() {}
+// Cleanup cleans up internal state.
+func (p *provReqProcessor) CleanUp() {}

cluster-autoscaler/provisioningrequest/conditions/condition_test.go

@@ -21,6 +21,7 @@ import (
 
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/apis/provisioningrequest/autoscaling.x-k8s.io/v1beta1"
+	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest"
 	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/provreqwrapper"
 )
 
@@ -89,18 +90,20 @@ func TestBookCapacity(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			pr := provreqwrapper.NewProvisioningRequest(
-				&v1beta1.ProvisioningRequest{
-					Spec: v1beta1.ProvisioningRequestSpec{
-						ProvisioningClassName: v1beta1.ProvisioningClassCheckCapacity,
-					},
-					Status: v1beta1.ProvisioningRequestStatus{
-						Conditions: test.prConditions,
-					},
-				}, nil)
-			got := ShouldCapacityBeBooked(pr)
-			if got != test.want {
-				t.Errorf("Want: %v, got: %v", test.want, got)
+			for class := range provisioningrequest.SupportedProvisioningClasses {
+				pr := provreqwrapper.NewProvisioningRequest(
+					&v1beta1.ProvisioningRequest{
+						Spec: v1beta1.ProvisioningRequestSpec{
+							ProvisioningClassName: class,
+						},
+						Status: v1beta1.ProvisioningRequestStatus{
+							Conditions: test.prConditions,
+						},
+					}, nil)
+				got := ShouldCapacityBeBooked(pr)
+				if got != test.want {
+					t.Errorf("Want: %v, got: %v", test.want, got)
+				}
 			}
 		})
 	}

cluster-autoscaler/provisioningrequest/conditions/conditions.go

@@ -20,6 +20,7 @@ import (
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/apis/provisioningrequest/autoscaling.x-k8s.io/v1beta1"
+	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest"
 	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/provreqwrapper"
 	"k8s.io/klog/v2"
 )
@@ -59,7 +60,7 @@ const (
 
 // ShouldCapacityBeBooked returns whether capacity should be booked.
 func ShouldCapacityBeBooked(pr *provreqwrapper.ProvisioningRequest) bool {
-	if pr.Spec.ProvisioningClassName != v1beta1.ProvisioningClassCheckCapacity && pr.Spec.ProvisioningClassName != v1beta1.ProvisioningClassBestEffortAtomicScaleUp {
+	if ok, found := provisioningrequest.SupportedProvisioningClasses[pr.Spec.ProvisioningClassName]; !ok || !found {
 		return false
 	}
 	conditions := pr.Status.Conditions

cluster-autoscaler/provisioningrequest/supported_classes.go

@@ -0,0 +1,28 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package provisioningrequest
+
+import (
+	"k8s.io/autoscaler/cluster-autoscaler/apis/provisioningrequest/autoscaling.x-k8s.io/v1beta1"
+)
+
+// SupportedProvisioningClasses is a set of ProvisioningRequest classes
+// supported by Cluster Autoscaler.
+var SupportedProvisioningClasses = map[string]bool{
+	v1beta1.ProvisioningClassCheckCapacity:           true,
+	v1beta1.ProvisioningClassBestEffortAtomicScaleUp: true,
+}