Refactoring AutoscalerBuilder and Extracting NodeInfoComparator

cluster-autoscaler/builder/autoscaler.go

@@ -38,7 +38,6 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/metrics"
 	ca_processors "k8s.io/autoscaler/cluster-autoscaler/processors"
 	cbprocessor "k8s.io/autoscaler/cluster-autoscaler/processors/capacitybuffer"
-	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupset"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodeinfosprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/podinjection"
 	podinjectionbackoff "k8s.io/autoscaler/cluster-autoscaler/processors/podinjection/backoff"
@@ -256,31 +255,6 @@ func (b *AutoscalerBuilder) Build(ctx context.Context) (core.Autoscaler, *loop.L
 	}
 	opts.Processors.ScaleDownNodeProcessor = cp
 
-	var nodeInfoComparator nodegroupset.NodeInfoComparator
-	if len(autoscalingOptions.BalancingLabels) > 0 {
-		nodeInfoComparator = nodegroupset.CreateLabelNodeInfoComparator(autoscalingOptions.BalancingLabels)
-	} else {
-		// TODO elmiko - now that we are passing the AutoscalerOptions in to the
-		// NewCloudProvider function, we should migrate these cloud provider specific
-		// configurations to the NewCloudProvider method so that we remove more provider
-		// code from the core.
-		nodeInfoComparatorBuilder := nodegroupset.CreateGenericNodeInfoComparator
-		if autoscalingOptions.CloudProviderName == cloudprovider.AzureProviderName {
-			nodeInfoComparatorBuilder = nodegroupset.CreateAzureNodeInfoComparator
-		} else if autoscalingOptions.CloudProviderName == cloudprovider.AwsProviderName {
-			nodeInfoComparatorBuilder = nodegroupset.CreateAwsNodeInfoComparator
-			opts.Processors.TemplateNodeInfoProvider = nodeinfosprovider.NewAsgTagResourceNodeInfoProvider(&autoscalingOptions.NodeInfoCacheExpireTime, autoscalingOptions.ForceDaemonSets)
-		} else if autoscalingOptions.CloudProviderName == cloudprovider.GceProviderName {
-			nodeInfoComparatorBuilder = nodegroupset.CreateGceNodeInfoComparator
-			opts.Processors.TemplateNodeInfoProvider = nodeinfosprovider.NewAnnotationNodeInfoProvider(&autoscalingOptions.NodeInfoCacheExpireTime, autoscalingOptions.ForceDaemonSets)
-		}
-		nodeInfoComparator = nodeInfoComparatorBuilder(autoscalingOptions.BalancingExtraIgnoredLabels, autoscalingOptions.NodeGroupSetRatios)
-	}
-
-	opts.Processors.NodeGroupSetProcessor = &nodegroupset.BalancingNodeGroupSetProcessor{
-		Comparator: nodeInfoComparator,
-	}
-
 	// These metrics should be published only once.
 	metrics.UpdateCPULimitsCores(autoscalingOptions.MinCoresTotal, autoscalingOptions.MaxCoresTotal)
 	metrics.UpdateMemoryLimitsBytes(autoscalingOptions.MinMemoryTotal, autoscalingOptions.MaxMemoryTotal)
@@ -294,7 +268,7 @@ func (b *AutoscalerBuilder) Build(ctx context.Context) (core.Autoscaler, *loop.L
 	}
 
 	// Create autoscaler.
-	autoscaler, err := core.NewAutoscaler(ctx, opts, b.informerFactory)
+	autoscaler, err := NewAutoscaler(ctx, opts, b.informerFactory)
 	if err != nil {
 		return nil, nil, err
 	}

cluster-autoscaler/builder/autoscaler_factory.go

@@ -0,0 +1,165 @@
+/*
+Copyright 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 builder
+
+import (
+	"context"
+	"strings"
+
+	cqv1alpha1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacityquota/autoscaling.x-k8s.io/v1alpha1"
+	cloudBuilder "k8s.io/autoscaler/cluster-autoscaler/cloudprovider/builder"
+	ca_context "k8s.io/autoscaler/cluster-autoscaler/context"
+	"k8s.io/autoscaler/cluster-autoscaler/core"
+	coreoptions "k8s.io/autoscaler/cluster-autoscaler/core/options"
+	"k8s.io/autoscaler/cluster-autoscaler/core/scaledown/pdb"
+	"k8s.io/autoscaler/cluster-autoscaler/core/utils"
+	"k8s.io/autoscaler/cluster-autoscaler/estimator"
+	"k8s.io/autoscaler/cluster-autoscaler/expander/factory"
+	"k8s.io/autoscaler/cluster-autoscaler/observers/loopstart"
+	ca_processors "k8s.io/autoscaler/cluster-autoscaler/processors"
+	"k8s.io/autoscaler/cluster-autoscaler/resourcequotas"
+	"k8s.io/autoscaler/cluster-autoscaler/resourcequotas/capacityquota"
+	"k8s.io/autoscaler/cluster-autoscaler/simulator/clustersnapshot/predicate"
+	"k8s.io/autoscaler/cluster-autoscaler/simulator/clustersnapshot/store"
+	csinodeprovider "k8s.io/autoscaler/cluster-autoscaler/simulator/csi/provider"
+	"k8s.io/autoscaler/cluster-autoscaler/simulator/drainability/rules"
+	draprovider "k8s.io/autoscaler/cluster-autoscaler/simulator/dynamicresources/provider"
+	"k8s.io/autoscaler/cluster-autoscaler/simulator/framework"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/backoff"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
+	"k8s.io/client-go/informers"
+)
+
+// NewAutoscaler creates an autoscaler of an appropriate type according to the parameters
+func NewAutoscaler(ctx context.Context, opts coreoptions.AutoscalerOptions, informerFactory informers.SharedInformerFactory) (core.Autoscaler, errors.AutoscalerError) {
+	err := initializeDefaultOptions(ctx, &opts, informerFactory)
+	if err != nil {
+		return nil, errors.ToAutoscalerError(errors.InternalError, err)
+	}
+	return core.NewStaticAutoscaler(
+		opts.AutoscalingOptions,
+		opts.FrameworkHandle,
+		opts.ClusterSnapshot,
+		opts.AutoscalingKubeClients,
+		opts.Processors,
+		opts.LoopStartNotifier,
+		opts.CloudProvider,
+		opts.ExpanderStrategy,
+		opts.EstimatorBuilder,
+		opts.Backoff,
+		opts.DebuggingSnapshotter,
+		opts.RemainingPdbTracker,
+		opts.ScaleUpOrchestrator,
+		opts.DeleteOptions,
+		opts.DrainabilityRules,
+		opts.DraProvider,
+		opts.QuotasTrackerOptions,
+		opts.CSIProvider,
+	), nil
+}
+
+// Initialize default options if not provided.
+func initializeDefaultOptions(ctx context.Context, opts *coreoptions.AutoscalerOptions, informerFactory informers.SharedInformerFactory) error {
+	if opts.Processors == nil {
+		opts.Processors = ca_processors.DefaultProcessors(opts.AutoscalingOptions)
+	}
+	if opts.LoopStartNotifier == nil {
+		opts.LoopStartNotifier = loopstart.NewObserversList(nil)
+	}
+	if opts.AutoscalingKubeClients == nil {
+		opts.AutoscalingKubeClients = ca_context.NewAutoscalingKubeClients(ctx, opts.AutoscalingOptions, opts.KubeClient, opts.InformerFactory)
+	}
+	if opts.FrameworkHandle == nil {
+		fwHandle, err := framework.NewHandle(ctx, opts.InformerFactory, opts.SchedulerConfig, opts.DynamicResourceAllocationEnabled, opts.CSINodeAwareSchedulingEnabled)
+		if err != nil {
+			return err
+		}
+		opts.FrameworkHandle = fwHandle
+	}
+	if opts.ClusterSnapshot == nil {
+		opts.ClusterSnapshot = predicate.NewPredicateSnapshot(store.NewBasicSnapshotStore(), opts.FrameworkHandle, opts.DynamicResourceAllocationEnabled, opts.PredicateParallelism, opts.CSINodeAwareSchedulingEnabled)
+	}
+	if opts.RemainingPdbTracker == nil {
+		opts.RemainingPdbTracker = pdb.NewBasicRemainingPdbTracker()
+	}
+	if opts.EstimatorBuilder == nil {
+		thresholds := []estimator.Threshold{
+			estimator.NewStaticThreshold(opts.MaxNodesPerScaleUp, opts.MaxNodeGroupBinpackingDuration),
+			estimator.NewSngCapacityThreshold(),
+			estimator.NewClusterCapacityThreshold(),
+		}
+		estimatorBuilder, err := estimator.NewEstimatorBuilder(
+			opts.EstimatorName,
+			estimator.NewThresholdBasedEstimationLimiter(thresholds),
+			estimator.NewDecreasingPodOrderer(),
+			/* EstimationAnalyserFunc */ nil,
+			opts.FastpathBinpackingEnabled,
+		)
+		if err != nil {
+			return err
+		}
+		opts.EstimatorBuilder = estimatorBuilder
+	}
+	if opts.Backoff == nil {
+		opts.Backoff =
+			backoff.NewIdBasedExponentialBackoff(opts.InitialNodeGroupBackoffDuration, opts.MaxNodeGroupBackoffDuration, opts.NodeGroupBackoffResetTimeout)
+	}
+	if opts.DrainabilityRules == nil {
+		opts.DrainabilityRules = rules.Default(opts.DeleteOptions)
+	}
+	if opts.DraProvider == nil && opts.DynamicResourceAllocationEnabled {
+		opts.DraProvider = draprovider.NewProviderFromInformers(informerFactory)
+	}
+	if opts.CloudProvider == nil {
+		opts.CloudProvider = cloudBuilder.NewCloudProvider(opts, informerFactory)
+	}
+
+	if opts.ExpanderStrategy == nil {
+		expanderFactory := factory.NewFactory()
+		expanderFactory.RegisterDefaultExpanders(opts.CloudProvider, opts.AutoscalingKubeClients, opts.KubeClient, opts.ConfigNamespace, opts.GRPCExpanderCert, opts.GRPCExpanderURL)
+		expanderStrategy, err := expanderFactory.Build(strings.Split(opts.ExpanderNames, ","))
+		if err != nil {
+			return err
+		}
+		opts.ExpanderStrategy = expanderStrategy
+	}
+	if opts.QuotasTrackerOptions.QuotaProvider == nil {
+		providers := []resourcequotas.Provider{resourcequotas.NewCloudQuotasProvider(opts.CloudProvider)}
+
+		if opts.CapacityQuotasEnabled {
+			// register informer here to disable lazy initialization
+			if _, err := opts.KubeCache.GetInformer(context.TODO(), &cqv1alpha1.CapacityQuota{}); err != nil {
+				return err
+			}
+			providers = append(providers, capacityquota.NewCapacityQuotasProvider(opts.KubeClientNew))
+		}
+		opts.QuotasTrackerOptions.QuotaProvider = resourcequotas.NewCombinedQuotasProvider(providers)
+	}
+	if opts.QuotasTrackerOptions.CustomResourcesProcessor == nil {
+		opts.QuotasTrackerOptions.CustomResourcesProcessor = opts.Processors.CustomResourcesProcessor
+	}
+	if opts.QuotasTrackerOptions.NodeFilter == nil {
+		virtualKubeletNodeFilter := utils.VirtualKubeletNodeFilter{}
+		opts.QuotasTrackerOptions.NodeFilter = resourcequotas.NewCombinedNodeFilter([]resourcequotas.NodeFilter{virtualKubeletNodeFilter})
+	}
+
+	if opts.CSINodeAwareSchedulingEnabled {
+		csiProvider := csinodeprovider.NewCSINodeProviderFromInformers(informerFactory)
+		opts.CSIProvider = csiProvider
+	}
+	return nil
+}

cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go

@@ -28,10 +28,12 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/config"
 	coreoptions "k8s.io/autoscaler/cluster-autoscaler/core/options"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupset"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodeinfosprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/framework"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/gpu"
-	klog "k8s.io/klog/v2"
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -474,5 +476,14 @@ func BuildAWS(opts *coreoptions.AutoscalerOptions, do cloudprovider.NodeGroupDis
 		klog.Fatalf("Failed to create AWS cloud provider: %v", err)
 	}
 	RegisterMetrics()
+
+	// Configure AWS specific processors
+	if opts.Processors != nil {
+		opts.Processors.TemplateNodeInfoProvider = nodeinfosprovider.NewAsgTagResourceNodeInfoProvider(&opts.AutoscalingOptions.NodeInfoCacheExpireTime, opts.AutoscalingOptions.ForceDaemonSets)
+		opts.Processors.NodeGroupSetProcessor = &nodegroupset.BalancingNodeGroupSetProcessor{
+			Comparator: nodegroupset.CreateAwsNodeInfoComparator(opts.AutoscalingOptions.BalancingExtraIgnoredLabels, opts.AutoscalingOptions.NodeGroupSetRatios),
+		}
+	}
+
 	return provider
 }

cluster-autoscaler/cloudprovider/azure/azure_cloud_provider.go

@@ -26,9 +26,10 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	coreoptions "k8s.io/autoscaler/cluster-autoscaler/core/options"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupset"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/gpu"
-	klog "k8s.io/klog/v2"
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -211,5 +212,13 @@ func BuildAzure(opts *coreoptions.AutoscalerOptions, do cloudprovider.NodeGroupD
 	if err != nil {
 		klog.Fatalf("Failed to create Azure cloud provider: %v", err)
 	}
+
+	// Configure Azure specific NodeInfoComparator
+	if opts.Processors != nil {
+		opts.Processors.NodeGroupSetProcessor = &nodegroupset.BalancingNodeGroupSetProcessor{
+			Comparator: nodegroupset.CreateAzureNodeInfoComparator(opts.AutoscalingOptions.BalancingExtraIgnoredLabels, opts.AutoscalingOptions.NodeGroupSetRatios),
+		}
+	}
+
 	return provider
 }

cluster-autoscaler/cloudprovider/builder/cloud_provider_builder.go

@@ -20,9 +20,10 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	ca_context "k8s.io/autoscaler/cluster-autoscaler/context"
 	coreoptions "k8s.io/autoscaler/cluster-autoscaler/core/options"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupset"
 	"k8s.io/client-go/informers"
 
-	klog "k8s.io/klog/v2"
+	"k8s.io/klog/v2"
 )
 
 // NewCloudProvider builds a cloud provider from provided parameters.
@@ -45,6 +46,11 @@ func NewCloudProvider(opts *coreoptions.AutoscalerOptions, informerFactory infor
 
 	provider := buildCloudProvider(opts, do, rl, informerFactory)
 	if provider != nil {
+		if len(opts.AutoscalingOptions.BalancingLabels) > 0 {
+			opts.Processors.NodeGroupSetProcessor = &nodegroupset.BalancingNodeGroupSetProcessor{
+				Comparator: nodegroupset.CreateLabelNodeInfoComparator(opts.AutoscalingOptions.BalancingLabels),
+			}
+		}
 		return provider
 	}
 

cluster-autoscaler/cloudprovider/gce/gce_cloud_provider.go

@@ -27,10 +27,12 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/config"
 	coreoptions "k8s.io/autoscaler/cluster-autoscaler/core/options"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupset"
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodeinfosprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/framework"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/gpu"
-	klog "k8s.io/klog/v2"
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -411,5 +413,14 @@ func BuildGCE(opts *coreoptions.AutoscalerOptions, do cloudprovider.NodeGroupDis
 	}
 	// Register GCE API usage metrics.
 	RegisterMetrics()
+
+	// Configure GCE specific processors
+	if opts.Processors != nil {
+		opts.Processors.TemplateNodeInfoProvider = nodeinfosprovider.NewAnnotationNodeInfoProvider(&opts.AutoscalingOptions.NodeInfoCacheExpireTime, opts.AutoscalingOptions.ForceDaemonSets)
+		opts.Processors.NodeGroupSetProcessor = &nodegroupset.BalancingNodeGroupSetProcessor{
+			Comparator: nodegroupset.CreateGceNodeInfoComparator(opts.AutoscalingOptions.BalancingExtraIgnoredLabels, opts.AutoscalingOptions.NodeGroupSetRatios),
+		}
+	}
+
 	return provider
 }