Review

cluster-autoscaler/core/static_autoscaler.go

@@ -341,15 +341,7 @@ func (a *StaticAutoscaler) RunOnce(currentTime time.Time) caerrors.AutoscalerErr
 		klog.Errorf("Failed to refresh cloud provider config: %v", err)
 		return caerrors.ToAutoscalerError(caerrors.CloudProviderError, err)
 	}
-	if a.loopStartNotifier != nil {
-		stop, err := a.loopStartNotifier.Refresh()
-		if stop {
-			return caerrors.ToAutoscalerError(caerrors.TransientError, err)
-		}
-		if err != nil {
-			klog.Errorf("Failed to refresh loopstart observers: %v", err)
-		}
-	}
+	a.loopStartNotifier.Refresh()
 
 	// Update node groups min/max and maximum number of nodes being set for all node groups after cloud provider refresh
 	maxNodesCount := 0

cluster-autoscaler/core/static_autoscaler_test.go

@@ -42,6 +42,7 @@ import (
 	. "k8s.io/autoscaler/cluster-autoscaler/core/test"
 	core_utils "k8s.io/autoscaler/cluster-autoscaler/core/utils"
 	"k8s.io/autoscaler/cluster-autoscaler/estimator"
+	"k8s.io/autoscaler/cluster-autoscaler/observers/loopstart"
 	ca_processors "k8s.io/autoscaler/cluster-autoscaler/processors"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/callbacks"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupconfig"
@@ -281,6 +282,7 @@ func setupAutoscaler(config *autoscalerSetupConfig) (*StaticAutoscaler, error) {
 		scaleDownActuator:    sdActuator,
 		scaleUpOrchestrator:  suOrchestrator,
 		processors:           processors,
+		loopStartNotifier:    loopstart.NewObserversList(nil),
 		processorCallbacks:   processorCallbacks,
 	}
 
@@ -374,6 +376,7 @@ func TestStaticAutoscalerRunOnce(t *testing.T) {
 		scaleDownActuator:     sdActuator,
 		scaleUpOrchestrator:   suOrchestrator,
 		processors:            processors,
+		loopStartNotifier:     loopstart.NewObserversList(nil),
 		processorCallbacks:    processorCallbacks,
 		initialized:           true,
 	}
@@ -573,6 +576,7 @@ func TestStaticAutoscalerRunOnceWithScaleDownDelayPerNG(t *testing.T) {
 		scaleDownActuator:     sdActuator,
 		scaleUpOrchestrator:   suOrchestrator,
 		processors:            processors,
+		loopStartNotifier:     loopstart.NewObserversList(nil),
 		processorCallbacks:    processorCallbacks,
 		initialized:           true,
 	}
@@ -798,6 +802,7 @@ func TestStaticAutoscalerRunOnceWithAutoprovisionedEnabled(t *testing.T) {
 		scaleDownActuator:     sdActuator,
 		scaleUpOrchestrator:   suOrchestrator,
 		processors:            processors,
+		loopStartNotifier:     loopstart.NewObserversList(nil),
 		processorCallbacks:    processorCallbacks,
 		initialized:           true,
 	}
@@ -948,6 +953,7 @@ func TestStaticAutoscalerRunOnceWithALongUnregisteredNode(t *testing.T) {
 		scaleDownActuator:     sdActuator,
 		scaleUpOrchestrator:   suOrchestrator,
 		processors:            processors,
+		loopStartNotifier:     loopstart.NewObserversList(nil),
 		processorCallbacks:    processorCallbacks,
 	}
 
@@ -1096,6 +1102,7 @@ func TestStaticAutoscalerRunOncePodsWithPriorities(t *testing.T) {
 		scaleDownActuator:     sdActuator,
 		scaleUpOrchestrator:   suOrchestrator,
 		processors:            processors,
+		loopStartNotifier:     loopstart.NewObserversList(nil),
 		processorCallbacks:    processorCallbacks,
 	}
 
@@ -1224,6 +1231,7 @@ func TestStaticAutoscalerRunOnceWithFilteringOnBinPackingEstimator(t *testing.T)
 		scaleDownPlanner:      sdPlanner,
 		scaleDownActuator:     sdActuator,
 		processors:            processors,
+		loopStartNotifier:     loopstart.NewObserversList(nil),
 		processorCallbacks:    processorCallbacks,
 	}
 
@@ -1322,6 +1330,7 @@ func TestStaticAutoscalerRunOnceWithFilteringOnUpcomingNodesEnabledNoScaleUp(t *
 		scaleDownPlanner:      sdPlanner,
 		scaleDownActuator:     sdActuator,
 		processors:            processors,
+		loopStartNotifier:     loopstart.NewObserversList(nil),
 		processorCallbacks:    processorCallbacks,
 	}
 
@@ -1427,6 +1436,7 @@ func TestStaticAutoscalerRunOnceWithUnselectedNodeGroups(t *testing.T) {
 				scaleDownPlanner:     sdPlanner,
 				scaleDownActuator:    sdActuator,
 				processors:           NewTestProcessors(&context),
+				loopStartNotifier:    loopstart.NewObserversList(nil),
 				processorCallbacks:   processorCallbacks,
 			}
 
@@ -2023,6 +2033,7 @@ func TestStaticAutoscalerUpcomingScaleDownCandidates(t *testing.T) {
 		scaleDownActuator:    actuator,
 		scaleDownPlanner:     planner,
 		processors:           NewTestProcessors(&ctx),
+		loopStartNotifier:    loopstart.NewObserversList(nil),
 		processorCallbacks:   processorCallbacks,
 	}
 

cluster-autoscaler/go.mod

@@ -1,6 +1,6 @@
 module k8s.io/autoscaler/cluster-autoscaler
 
-go 1.21
+go 1.21.3
 
 require (
 	cloud.google.com/go/compute/metadata v0.2.3

cluster-autoscaler/observers/loopstart/loopstart.go

@@ -16,12 +16,10 @@ limitations under the License.
 
 package loopstart
 
-import "errors"
-
 // Observer interface is used to store object that needed to be refreshed in each CA loop.
 // It returns error and a bool value whether the loop should be skipped.
 type Observer interface {
-	Refresh() (bool, error)
+	Refresh()
 }
 
 // ObserversList interface is used to store objects that needed to be refreshed in each CA loop.
@@ -30,19 +28,13 @@ type ObserversList struct {
 }
 
 // Refresh refresh observers each CA loop.
-func (l *ObserversList) Refresh() (stopLoop bool, err error) {
-	var resultErr error
+func (l *ObserversList) Refresh() {
 	if l.observers == nil {
-		return false, nil
+		return
 	}
 	for _, observer := range l.observers {
-		stopLoop, err := observer.Refresh()
-		if stopLoop {
-			return true, nil
-		}
-		resultErr = errors.Join(resultErr, err)
+		observer.Refresh()
 	}
-	return false, resultErr
 }
 
 // NewObserversList return new ObserversList.

cluster-autoscaler/processors/provreq/provsioning_request_processor.go → cluster-autoscaler/processors/provreq/provisioning_request_processor.go

@@ -21,11 +21,12 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/provreqclient"
 	"k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/provreqwrapper"
 	"k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
 )
 
-// ProvisioningRequestProcessor process ProvisionignRequests in the cluster.
+// ProvisioningRequestProcessor process ProvisioningRequests in the cluster.
 type ProvisioningRequestProcessor interface {
-	Process([]*provreqwrapper.ProvisioningRequest) error
+	Process([]*provreqwrapper.ProvisioningRequest)
 	CleanUp()
 }
 
@@ -34,7 +35,8 @@ type provisioningRequestClient interface {
 	ProvisioningRequest(namespace, name string) (*provreqwrapper.ProvisioningRequest, error)
 }
 
-// CombinedProvReqProcessor contains ProvisioningClassProcessor for each ProvisioningClass.
+// CombinedProvReqProcessor is responsible for processing ProvisioningRequest for each ProvisioningClass
+// every CA loop and updating conditions for expired ProvisioningRequests.
 type CombinedProvReqProcessor struct {
 	client     provisioningRequestClient
 	processors []ProvisioningRequestProcessor
@@ -50,17 +52,15 @@ func NewCombinedProvReqProcessor(kubeConfig *rest.Config, processors []Provision
 }
 
 // Refresh iterates over ProvisioningRequests and updates its conditions/state.
-func (cp *CombinedProvReqProcessor) Refresh() (bool, error) {
+func (cp *CombinedProvReqProcessor) Refresh() {
 	provReqs, err := cp.client.ProvisioningRequests()
 	if err != nil {
-		return true, err
+		klog.Errorf("Failed to get ProvisioningRequests list, err: %v", err)
+		return
 	}
 	for _, p := range cp.processors {
-		if err := p.Process(provReqs); err != nil {
-			return false, err
-		}
+		p.Process(provReqs)
 	}
-	return false, nil
 }
 
 // CleanUp cleans up internal state

cluster-autoscaler/provisioningrequest/checkcapacity/processor.go

@@ -46,7 +46,7 @@ func NewCheckCapacityProcessor() *checkCapacityProcessor {
 // -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 *checkCapacityProcessor) Process(provReqs []*provreqwrapper.ProvisioningRequest) error {
+func (p *checkCapacityProcessor) Process(provReqs []*provreqwrapper.ProvisioningRequest) {
 	expiredProvReq := []*provreqwrapper.ProvisioningRequest{}
 	failedProvReq := []*provreqwrapper.ProvisioningRequest{}
 	for _, provReq := range provReqs {
@@ -85,7 +85,6 @@ func (p *checkCapacityProcessor) Process(provReqs []*provreqwrapper.Provisioning
 		setCondition(provReq, v1beta1.Failed, metav1.ConditionTrue, ExpiredReason, ExpiredMsg, metav1.NewTime(p.now()))
 		updated++
 	}
-	return nil
 }
 
 // Cleanup cleans up internal state.