Merge pull request #7275 from abdelrahman882/fix_scaleup_status_issue

Fix scale up status processor overriding default one with proactive s…

cluster-autoscaler/main.go

@@ -69,6 +69,7 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/processors/scaledowncandidates"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/scaledowncandidates/emptycandidates"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/scaledowncandidates/previouscandidates"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/status"
 	provreqorchestrator "k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/orchestrator"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/clustersnapshot"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/drainability/rules"
@@ -540,9 +541,9 @@ func buildAutoscaler(debuggingSnapshotter debuggingsnapshot.DebuggingSnapshotter
 
 		podListProcessor = pods.NewCombinedPodListProcessor([]pods.PodListProcessor{podInjectionPodListProcessor, podListProcessor, enforceInjectedPodsLimitProcessor})
 
-		// FakePodsScaleUpStatusProcessor processor needs to be the first processor in ScaleUpStatusProcessor as it filters out fake pods from
-		// Scale Up status so that we don't emit events.
-		opts.Processors.ScaleUpStatusProcessor = podinjection.NewFakePodsScaleUpStatusProcessor(podInjectionBackoffRegistry)
+		// FakePodsScaleUpStatusProcessor processor needs to be the first processor in ScaleUpStatusProcessor before the default processor
+		// As it filters out fake pods from Scale Up status so that we don't emit events.
+		opts.Processors.ScaleUpStatusProcessor = status.NewCombinedScaleUpStatusProcessor([]status.ScaleUpStatusProcessor{podinjection.NewFakePodsScaleUpStatusProcessor(podInjectionBackoffRegistry), opts.Processors.ScaleUpStatusProcessor})
 	}
 
 	opts.Processors.PodListProcessor = podListProcessor

cluster-autoscaler/processors/status/scale_up_status_processor.go

@@ -100,6 +100,43 @@ func (p *NoOpScaleUpStatusProcessor) Process(context *context.AutoscalingContext
 func (p *NoOpScaleUpStatusProcessor) CleanUp() {
 }
 
+// CombinedScaleUpStatusProcessor is a list of ScaleUpStatusProcessor
+type CombinedScaleUpStatusProcessor struct {
+	processors []ScaleUpStatusProcessor
+}
+
+// NewCombinedScaleUpStatusProcessor construct CombinedScaleUpStatusProcessor.
+func NewCombinedScaleUpStatusProcessor(processors []ScaleUpStatusProcessor) *CombinedScaleUpStatusProcessor {
+	var scaleUpProcessors []ScaleUpStatusProcessor
+	for _, processor := range processors {
+		if processor != nil {
+			scaleUpProcessors = append(scaleUpProcessors, processor)
+		}
+	}
+	return &CombinedScaleUpStatusProcessor{scaleUpProcessors}
+}
+
+// AddProcessor append processor to the list.
+func (p *CombinedScaleUpStatusProcessor) AddProcessor(processor ScaleUpStatusProcessor) {
+	if processor != nil {
+		p.processors = append(p.processors, processor)
+	}
+}
+
+// Process runs sub-processors sequentially in the same order of addition
+func (p *CombinedScaleUpStatusProcessor) Process(ctx *context.AutoscalingContext, status *ScaleUpStatus) {
+	for _, processor := range p.processors {
+		processor.Process(ctx, status)
+	}
+}
+
+// CleanUp cleans up the processor's internal structures.
+func (p *CombinedScaleUpStatusProcessor) CleanUp() {
+	for _, processor := range p.processors {
+		processor.CleanUp()
+	}
+}
+
 // UpdateScaleUpError updates ScaleUpStatus.
 func UpdateScaleUpError(s *ScaleUpStatus, err errors.AutoscalerError) (*ScaleUpStatus, errors.AutoscalerError) {
 	s.ScaleUpError = &err