Define interfaces for per NodeGroup config.

This is the first step of implementing
https://github.com/kubernetes/autoscaler/issues/3583#issuecomment-743215343.
New method was added to cloudprovider interface. All existing providers
were updated with a no-op stub implementation that will result in no
behavior change.
The config values specified per NodeGroup are not yet applied.

cluster-autoscaler/cloudprovider/alicloud/alicloud_auto_scaling_group.go

@@ -21,6 +21,7 @@ import (
 
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	klog "k8s.io/klog/v2"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 )
@@ -211,3 +212,9 @@ func (asg *Asg) Autoprovisioned() bool {
 func (asg *Asg) Delete() error {
 	return cloudprovider.ErrNotImplemented
 }
+
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (asg *Asg) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}

cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go

@@ -215,6 +215,12 @@ func (ng *AwsNodeGroup) Delete() error {
 	return cloudprovider.ErrNotImplemented
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (ng *AwsNodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // IncreaseSize increases Asg size
 func (ng *AwsNodeGroup) IncreaseSize(delta int) error {
 	if delta <= 0 {

cluster-autoscaler/cloudprovider/azure/azure_agent_pool.go

@@ -32,6 +32,7 @@ import (
 	apiv1 "k8s.io/api/core/v1"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	"k8s.io/autoscaler/cluster-autoscaler/config/dynamic"
 	klog "k8s.io/klog/v2"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
@@ -119,6 +120,12 @@ func (as *AgentPool) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (as *AgentPool) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // MaxSize returns maximum size of the node group.
 func (as *AgentPool) MaxSize() int {
 	return as.maxSize

cluster-autoscaler/cloudprovider/azure/azure_kubernetes_service_pool.go

@@ -27,6 +27,7 @@ import (
 
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	"k8s.io/autoscaler/cluster-autoscaler/config/dynamic"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 )
@@ -433,3 +434,9 @@ func (agentPool *AKSAgentPool) Delete() error {
 func (agentPool *AKSAgentPool) Autoprovisioned() bool {
 	return false
 }
+
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (agentPool *AKSAgentPool) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}

cluster-autoscaler/cloudprovider/azure/azure_scale_set.go

@@ -25,6 +25,7 @@ import (
 
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	"k8s.io/autoscaler/cluster-autoscaler/config/dynamic"
 	klog "k8s.io/klog/v2"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
@@ -112,6 +113,12 @@ func (scaleSet *ScaleSet) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (scaleSet *ScaleSet) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // MaxSize returns maximum size of the node group.
 func (scaleSet *ScaleSet) MaxSize() int {
 	return scaleSet.maxSize

cluster-autoscaler/cloudprovider/baiducloud/baiducloud_cloud_provider.go

@@ -391,3 +391,9 @@ func (asg *Asg) Delete() error {
 func (asg *Asg) Autoprovisioned() bool {
 	return false
 }
+
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (asg *Asg) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}

cluster-autoscaler/cloudprovider/cloud_provider.go

@@ -22,6 +22,7 @@ import (
 
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 )
@@ -179,6 +180,11 @@ type NodeGroup interface {
 	// Autoprovisioned returns true if the node group is autoprovisioned. An autoprovisioned group
 	// was created by CA and can be deleted when scaled to 0.
 	Autoprovisioned() bool
+
+	// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+	// NodeGroup. Returning a nil will result in using default options.
+	// Implementation optional.
+	GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error)
 }
 
 // Instance represents a cloud-provider node. The node does not necessarily map to k8s node

cluster-autoscaler/cloudprovider/cloudstack/cloudstack_node_group.go

@@ -22,6 +22,7 @@ import (
 
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/cloudstack/service"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 
 	apiv1 "k8s.io/api/core/v1"
@@ -159,6 +160,12 @@ func (asg *asg) TemplateNodeInfo() (*schedulerframework.NodeInfo, error) {
 	return nil, cloudprovider.ErrNotImplemented
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (asg *asg) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 func (asg *asg) Copy(cluster *service.Cluster) {
 	asg.cluster = cluster
 }

cluster-autoscaler/cloudprovider/clusterapi/clusterapi_nodegroup.go

@@ -24,6 +24,7 @@ import (
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 )
 
 const (
@@ -264,6 +265,12 @@ func (ng *nodegroup) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (ng *nodegroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 func newNodeGroupFromScalableResource(controller *machineController, unstructuredScalableResource *unstructured.Unstructured) (*nodegroup, error) {
 	// Ensure that the resulting node group would be allowed based on the autodiscovery specs if defined
 	if !controller.allowedByAutoDiscoverySpecs(unstructuredScalableResource) {

cluster-autoscaler/cloudprovider/digitalocean/digitalocean_node_group.go

@@ -25,6 +25,7 @@ import (
 	apiv1 "k8s.io/api/core/v1"
 
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 )
 
@@ -231,6 +232,12 @@ func (n *NodeGroup) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (n *NodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // toInstances converts a slice of *godo.KubernetesNode to
 // cloudprovider.Instance
 func toInstances(nodes []*godo.KubernetesNode) []cloudprovider.Instance {

cluster-autoscaler/cloudprovider/exoscale/exoscale_node_group.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/exoscale/internal/github.com/exoscale/egoscale"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/exoscale/internal/k8s.io/klog"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 )
 
@@ -218,6 +219,12 @@ func (n *NodeGroup) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (n *NodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // toInstance converts the given egoscale.VirtualMachine to a
 // cloudprovider.Instance
 func toInstance(vm egoscale.VirtualMachine) cloudprovider.Instance {

cluster-autoscaler/cloudprovider/gce/gce_cloud_provider.go

@@ -327,6 +327,12 @@ func (mig *gceMig) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (mig *gceMig) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // TemplateNodeInfo returns a node template for this node group.
 func (mig *gceMig) TemplateNodeInfo() (*schedulerframework.NodeInfo, error) {
 	node, err := mig.gceManager.GetMigTemplateNode(mig)

cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud_auto_scaling_group.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	huaweicloudsdkasmodel "k8s.io/autoscaler/cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud-sdk-go-v3/services/as/v1/model"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
@@ -222,6 +223,12 @@ func (asg *AutoScalingGroup) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (asg *AutoScalingGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // String dumps current groups meta data.
 func (asg *AutoScalingGroup) String() string {
 	return fmt.Sprintf("group: %s min=%d max=%d", asg.groupID, asg.minInstanceNumber, asg.maxInstanceNumber)

cluster-autoscaler/cloudprovider/ionoscloud/ionoscloud_cloud_provider.go

@@ -172,6 +172,12 @@ func (n *nodePool) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (n *nodePool) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // IonosCloudCloudProvider implements cloudprovider.CloudProvider.
 type IonosCloudCloudProvider struct {
 	manager         IonosCloudManager

cluster-autoscaler/cloudprovider/kubemark/kubemark_linux.go

@@ -296,6 +296,12 @@ func (nodeGroup *NodeGroup) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (nodeGroup *NodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 func buildNodeGroup(value string, kubemarkController *kubemark.KubemarkController) (*NodeGroup, error) {
 	spec, err := dynamic.SpecFromString(value, true)
 	if err != nil {

cluster-autoscaler/cloudprovider/magnum/magnum_nodegroup.go

@@ -23,6 +23,7 @@ import (
 
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	klog "k8s.io/klog/v2"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 )
@@ -225,6 +226,12 @@ func (ng *magnumNodeGroup) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (ng *magnumNodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // MaxSize returns the maximum allowed size of the node group.
 func (ng *magnumNodeGroup) MaxSize() int {
 	return ng.maxSize

cluster-autoscaler/cloudprovider/mocks/NodeGroup.go

@@ -16,10 +16,16 @@ limitations under the License.
 
 package mocks
 
-import schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
-import cloudprovider "k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
-import mock "github.com/stretchr/testify/mock"
-import v1 "k8s.io/api/core/v1"
+import (
+	cloudprovider "k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	config "k8s.io/autoscaler/cluster-autoscaler/config"
+
+	framework "k8s.io/kubernetes/pkg/scheduler/framework"
+
+	mock "github.com/stretchr/testify/mock"
+
+	v1 "k8s.io/api/core/v1"
+)
 
 // NodeGroup is an autogenerated mock type for the NodeGroup type
 type NodeGroup struct {
@@ -133,6 +139,29 @@ func (_m *NodeGroup) Exist() bool {
 	return r0
 }
 
+// GetOptions provides a mock function with given fields: defaults
+func (_m *NodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	ret := _m.Called(defaults)
+
+	var r0 *config.NodeGroupAutoscalingOptions
+	if rf, ok := ret.Get(0).(func(config.NodeGroupAutoscalingOptions) *config.NodeGroupAutoscalingOptions); ok {
+		r0 = rf(defaults)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*config.NodeGroupAutoscalingOptions)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(config.NodeGroupAutoscalingOptions) error); ok {
+		r1 = rf(defaults)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // Id provides a mock function with given fields:
 func (_m *NodeGroup) Id() string {
 	ret := _m.Called()
@@ -234,15 +263,15 @@ func (_m *NodeGroup) TargetSize() (int, error) {
 }
 
 // TemplateNodeInfo provides a mock function with given fields:
-func (_m *NodeGroup) TemplateNodeInfo() (*schedulerframework.NodeInfo, error) {
+func (_m *NodeGroup) TemplateNodeInfo() (*framework.NodeInfo, error) {
 	ret := _m.Called()
 
-	var r0 *schedulerframework.NodeInfo
-	if rf, ok := ret.Get(0).(func() *schedulerframework.NodeInfo); ok {
+	var r0 *framework.NodeInfo
+	if rf, ok := ret.Get(0).(func() *framework.NodeInfo); ok {
 		r0 = rf()
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*schedulerframework.NodeInfo)
+			r0 = ret.Get(0).(*framework.NodeInfo)
 		}
 	}
 

cluster-autoscaler/cloudprovider/ovhcloud/ovh_cloud_node_group.go

@@ -29,6 +29,7 @@ import (
 
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/ovhcloud/sdk"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 )
 
 // instanceIdRegex defines the expression used for instance's ID
@@ -311,6 +312,12 @@ func (ng *NodeGroup) Autoprovisioned() bool {
 	return false
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (ng *NodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
+
 // extractNodeIds find in an array of node resource their cloud instances IDs
 func extractNodeIds(nodes []*apiv1.Node, instances []cloudprovider.Instance, groupLabel string) ([]string, error) {
 	nodeIds := make([]string, 0)

cluster-autoscaler/cloudprovider/packet/packet_node_group.go

@@ -23,6 +23,7 @@ import (
 
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	klog "k8s.io/klog/v2"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 )
@@ -293,3 +294,9 @@ func (ng *packetNodeGroup) MinSize() int {
 func (ng *packetNodeGroup) TargetSize() (int, error) {
 	return *ng.targetSize, nil
 }
+
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (ng *packetNodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}

cluster-autoscaler/cloudprovider/test/test_cloud_provider.go

@@ -23,6 +23,7 @@ import (
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
 	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
 )
@@ -451,6 +452,12 @@ func (tng *TestNodeGroup) TemplateNodeInfo() (*schedulerframework.NodeInfo, erro
 	return template, nil
 }
 
+// GetOptions returns NodeGroupAutoscalingOptions that should be used for this particular
+// NodeGroup. Returning a nil will result in using default options.
+func (tng *TestNodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, nil
+}
+
 // Labels returns labels passed to the test node group when it was created.
 func (tng *TestNodeGroup) Labels() map[string]string {
 	return tng.labels

cluster-autoscaler/config/autoscaling_options.go

@@ -30,8 +30,17 @@ type GpuLimits struct {
 	Max int64
 }
 
+// NodeGroupAutoscalingOptions contain various options to customize how autoscaling of
+// a given NodeGroup works. Different options can be used for each NodeGroup.
+type NodeGroupAutoscalingOptions struct {
+	ScaleDownUnneededTime time.Duration
+}
+
 // AutoscalingOptions contain various options to customize how autoscaling works
 type AutoscalingOptions struct {
+	// NodeGroupAutoscalingOptions are default values for per NodeGroup options.
+	// They will be used any time a specific value is not provided for a given NodeGroup.
+	NodeGroupAutoscalingOptions
 	// MaxEmptyBulkDelete is a number of empty nodes that can be removed at the same time.
 	MaxEmptyBulkDelete int
 	// ScaleDownUtilizationThreshold sets threshold for nodes to be considered for scale down if cpu or memory utilization is over threshold.
@@ -42,7 +51,7 @@ type AutoscalingOptions struct {
 	ScaleDownGpuUtilizationThreshold float64
 	// ScaleDownUnneededTime sets the duration CA expects a node to be unneeded/eligible for removal
 	// before scaling down the node.
-	ScaleDownUnneededTime time.Duration
+	// ScaleDownUnneededTime time.Duration
 	// ScaleDownUnreadyTime represents how long an unready node should be unneeded before it is eligible for scale down
 	ScaleDownUnreadyTime time.Duration
 	// MaxNodesTotal sets the maximum number of nodes in the whole cluster

cluster-autoscaler/core/scale_down_test.go

@@ -1012,8 +1012,10 @@ func TestScaleDown(t *testing.T) {
 	assert.NotNil(t, provider)
 
 	options := config.AutoscalingOptions{
-		ScaleDownUtilizationThreshold: 0.5,
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
 			ScaleDownUnneededTime: time.Minute,
+		},
+		ScaleDownUtilizationThreshold: 0.5,
 		MaxGracefulTerminationSec:     60,
 	}
 	jobLister, err := kube_util.NewTestJobLister([]*batchv1.Job{&job})
@@ -1069,9 +1071,11 @@ func assertSubset(t *testing.T, a []string, b []string) {
 }
 
 var defaultScaleDownOptions = config.AutoscalingOptions{
+	NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
+		ScaleDownUnneededTime: time.Minute,
+	},
 	ScaleDownUtilizationThreshold:    0.5,
 	ScaleDownGpuUtilizationThreshold: 0.5,
-	ScaleDownUnneededTime:            time.Minute,
 	MaxGracefulTerminationSec:        60,
 	MaxEmptyBulkDelete:               10,
 	MinCoresTotal:                    0,
@@ -1472,8 +1476,10 @@ func TestNoScaleDownUnready(t *testing.T) {
 	provider.AddNode("ng1", n2)
 
 	options := config.AutoscalingOptions{
-		ScaleDownUtilizationThreshold: 0.5,
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
 			ScaleDownUnneededTime: time.Minute,
+		},
+		ScaleDownUtilizationThreshold: 0.5,
 		ScaleDownUnreadyTime:          time.Hour,
 		MaxGracefulTerminationSec:     60,
 	}
@@ -1580,8 +1586,10 @@ func TestScaleDownNoMove(t *testing.T) {
 	assert.NotNil(t, provider)
 
 	options := config.AutoscalingOptions{
-		ScaleDownUtilizationThreshold: 0.5,
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
 			ScaleDownUnneededTime: time.Minute,
+		},
+		ScaleDownUtilizationThreshold: 0.5,
 		ScaleDownUnreadyTime:          time.Hour,
 		MaxGracefulTerminationSec:     60,
 	}
@@ -1828,8 +1836,10 @@ func TestSoftTaint(t *testing.T) {
 	assert.NotNil(t, provider)
 
 	options := config.AutoscalingOptions{
-		ScaleDownUtilizationThreshold: 0.5,
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
 			ScaleDownUnneededTime: 10 * time.Minute,
+		},
+		ScaleDownUtilizationThreshold: 0.5,
 		MaxGracefulTerminationSec:     60,
 		MaxBulkSoftTaintCount:         1,
 		MaxBulkSoftTaintTime:          3 * time.Second,
@@ -1947,8 +1957,10 @@ func TestSoftTaintTimeLimit(t *testing.T) {
 	assert.NotNil(t, provider)
 
 	options := config.AutoscalingOptions{
-		ScaleDownUtilizationThreshold: 0.5,
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
 			ScaleDownUnneededTime: 10 * time.Minute,
+		},
+		ScaleDownUtilizationThreshold: 0.5,
 		MaxGracefulTerminationSec:     60,
 		MaxBulkSoftTaintCount:         10,
 		MaxBulkSoftTaintTime:          maxSoftTaintDuration,

cluster-autoscaler/core/static_autoscaler_test.go

@@ -172,6 +172,9 @@ func TestStaticAutoscalerRunOnce(t *testing.T) {
 
 	// Create context with mocked lister registry.
 	options := config.AutoscalingOptions{
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
+			ScaleDownUnneededTime: time.Minute,
+		},
 		EstimatorName:                 estimator.BinpackingEstimatorName,
 		ScaleDownEnabled:              true,
 		ScaleDownUtilizationThreshold: 0.5,
@@ -179,7 +182,6 @@ func TestStaticAutoscalerRunOnce(t *testing.T) {
 		MaxCoresTotal:                 10,
 		MaxMemoryTotal:                100000,
 		ScaleDownUnreadyTime:          time.Minute,
-		ScaleDownUnneededTime:         time.Minute,
 	}
 	processorCallbacks := newStaticAutoscalerProcessorCallbacks()
 
@@ -358,6 +360,9 @@ func TestStaticAutoscalerRunOnceWithAutoprovisionedEnabled(t *testing.T) {
 
 	// Create context with mocked lister registry.
 	options := config.AutoscalingOptions{
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
+			ScaleDownUnneededTime: time.Minute,
+		},
 		EstimatorName:                    estimator.BinpackingEstimatorName,
 		ScaleDownEnabled:                 true,
 		ScaleDownUtilizationThreshold:    0.5,
@@ -365,7 +370,6 @@ func TestStaticAutoscalerRunOnceWithAutoprovisionedEnabled(t *testing.T) {
 		MaxCoresTotal:                    100,
 		MaxMemoryTotal:                   100000,
 		ScaleDownUnreadyTime:             time.Minute,
-		ScaleDownUnneededTime:            time.Minute,
 		NodeAutoprovisioningEnabled:      true,
 		MaxAutoprovisionedNodeGroupCount: 10,
 	}
@@ -492,6 +496,9 @@ func TestStaticAutoscalerRunOnceWithALongUnregisteredNode(t *testing.T) {
 
 	// Create context with mocked lister registry.
 	options := config.AutoscalingOptions{
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
+			ScaleDownUnneededTime: time.Minute,
+		},
 		EstimatorName:                 estimator.BinpackingEstimatorName,
 		ScaleDownEnabled:              true,
 		ScaleDownUtilizationThreshold: 0.5,
@@ -499,7 +506,6 @@ func TestStaticAutoscalerRunOnceWithALongUnregisteredNode(t *testing.T) {
 		MaxCoresTotal:                 10,
 		MaxMemoryTotal:                100000,
 		ScaleDownUnreadyTime:          time.Minute,
-		ScaleDownUnneededTime:         time.Minute,
 		MaxNodeProvisionTime:          10 * time.Second,
 	}
 	processorCallbacks := newStaticAutoscalerProcessorCallbacks()
@@ -635,6 +641,9 @@ func TestStaticAutoscalerRunOncePodsWithPriorities(t *testing.T) {
 
 	// Create context with mocked lister registry.
 	options := config.AutoscalingOptions{
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
+			ScaleDownUnneededTime: time.Minute,
+		},
 		EstimatorName:                 estimator.BinpackingEstimatorName,
 		ScaleDownEnabled:              true,
 		ScaleDownUtilizationThreshold: 0.5,
@@ -642,7 +651,6 @@ func TestStaticAutoscalerRunOncePodsWithPriorities(t *testing.T) {
 		MaxCoresTotal:                 10,
 		MaxMemoryTotal:                100000,
 		ScaleDownUnreadyTime:          time.Minute,
-		ScaleDownUnneededTime:         time.Minute,
 		ExpendablePodsPriorityCutoff:  10,
 	}
 	processorCallbacks := newStaticAutoscalerProcessorCallbacks()
@@ -912,6 +920,9 @@ func TestStaticAutoscalerInstaceCreationErrors(t *testing.T) {
 
 	// Create context with mocked lister registry.
 	options := config.AutoscalingOptions{
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
+			ScaleDownUnneededTime: time.Minute,
+		},
 		EstimatorName:                 estimator.BinpackingEstimatorName,
 		ScaleDownEnabled:              true,
 		ScaleDownUtilizationThreshold: 0.5,
@@ -919,7 +930,6 @@ func TestStaticAutoscalerInstaceCreationErrors(t *testing.T) {
 		MaxCoresTotal:                 10,
 		MaxMemoryTotal:                100000,
 		ScaleDownUnreadyTime:          time.Minute,
-		ScaleDownUnneededTime:         time.Minute,
 		ExpendablePodsPriorityCutoff:  10,
 	}
 	processorCallbacks := newStaticAutoscalerProcessorCallbacks()

cluster-autoscaler/expander/waste/waste_test.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
 	. "k8s.io/autoscaler/cluster-autoscaler/utils/test"
 
 	"github.com/stretchr/testify/assert"
@@ -54,6 +55,9 @@ func (f *FakeNodeGroup) Create() (cloudprovider.NodeGroup, error) {
 }
 func (f *FakeNodeGroup) Delete() error         { return cloudprovider.ErrNotImplemented }
 func (f *FakeNodeGroup) Autoprovisioned() bool { return false }
+func (f *FakeNodeGroup) GetOptions(defaults config.NodeGroupAutoscalingOptions) (*config.NodeGroupAutoscalingOptions, error) {
+	return nil, cloudprovider.ErrNotImplemented
+}
 
 func makeNodeInfo(cpu int64, memory int64, pods int64) *schedulerframework.NodeInfo {
 	node := &apiv1.Node{

cluster-autoscaler/main.go

@@ -197,6 +197,9 @@ func createAutoscalingOptions() config.AutoscalingOptions {
 		klog.Fatalf("Failed to parse flags: %v", err)
 	}
 	return config.AutoscalingOptions{
+		NodeGroupAutoscalingOptions: config.NodeGroupAutoscalingOptions{
+			ScaleDownUnneededTime: *scaleDownUnneededTime,
+		},
 		CloudConfig:                        *cloudConfig,
 		CloudProviderName:                  *cloudProviderFlag,
 		NodeGroupAutoDiscovery:             *nodeGroupAutoDiscoveryFlag,
@@ -223,7 +226,6 @@ func createAutoscalingOptions() config.AutoscalingOptions {
 		ScaleDownDelayAfterDelete:          *scaleDownDelayAfterDelete,
 		ScaleDownDelayAfterFailure:         *scaleDownDelayAfterFailure,
 		ScaleDownEnabled:                   *scaleDownEnabled,
-		ScaleDownUnneededTime:              *scaleDownUnneededTime,
 		ScaleDownUnreadyTime:               *scaleDownUnreadyTime,
 		ScaleDownUtilizationThreshold:      *scaleDownUtilizationThreshold,
 		ScaleDownGpuUtilizationThreshold:   *scaleDownGpuUtilizationThreshold,

cluster-autoscaler/processors/nodegroupconfig/node_group_config_processor.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2020 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 nodegroupconfig
+
+import (
+	"time"
+
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/context"
+)
+
+// NodeGroupConfigProcessor provides config values for a particular NodeGroup.
+type NodeGroupConfigProcessor interface {
+	// Process processes a map of nodeInfos for node groups.
+	GetScaleDownUnneededTime(context *context.AutoscalingContext, nodeGroup cloudprovider.NodeGroup) (time.Duration, error)
+	// CleanUp cleans up processor's internal structures.
+	CleanUp()
+}
+
+// DelegatingNodeGroupConfigProcessor calls NodeGroup.GetOptions to get config
+// for each NodeGroup. If NodeGroup doesn't return a value default config is
+// used instead.
+type DelegatingNodeGroupConfigProcessor struct {
+}
+
+// GetScaleDownUnneededTime returns ScaleDownUnneededTime value that should be used for a given NodeGroup.
+func (p *DelegatingNodeGroupConfigProcessor) GetScaleDownUnneededTime(context *context.AutoscalingContext, nodeGroup cloudprovider.NodeGroup) (time.Duration, error) {
+	ngConfig, err := nodeGroup.GetOptions(context.NodeGroupAutoscalingOptions)
+	if err != nil && err != cloudprovider.ErrNotImplemented {
+		return time.Duration(0), err
+	}
+	if ngConfig == nil || err == cloudprovider.ErrNotImplemented {
+		return context.ScaleDownUnneededTime, nil
+	}
+	return ngConfig.ScaleDownUnneededTime, nil
+}
+
+// CleanUp cleans up processor's internal structures.
+func (p *DelegatingNodeGroupConfigProcessor) CleanUp() {
+}
+
+// NewDefaultNodeGroupConfigProcessor returns a default instance of NodeGroupConfigProcessor.
+func NewDefaultNodeGroupConfigProcessor() NodeGroupConfigProcessor {
+	return &DelegatingNodeGroupConfigProcessor{}
+}

cluster-autoscaler/processors/nodegroupconfig/node_group_config_processor_test.go

@@ -0,0 +1,81 @@
+/*
+Copyright 2020 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 nodegroupconfig
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider/mocks"
+	"k8s.io/autoscaler/cluster-autoscaler/config"
+	"k8s.io/autoscaler/cluster-autoscaler/context"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestApplyingDefaults(t *testing.T) {
+	defaultOptions := config.NodeGroupAutoscalingOptions{
+		ScaleDownUnneededTime: 3 * time.Minute,
+	}
+
+	cases := map[string]struct {
+		globalOptions         config.NodeGroupAutoscalingOptions
+		ngOptions             *config.NodeGroupAutoscalingOptions
+		ngError               error
+		wantScaleDownUnneeded time.Duration
+		wantError             error
+	}{
+		"NodeGroup.GetOptions not implemented": {
+			globalOptions:         defaultOptions,
+			ngError:               cloudprovider.ErrNotImplemented,
+			wantScaleDownUnneeded: 3 * time.Minute,
+		},
+		"NodeGroup returns error leads to error": {
+			globalOptions: defaultOptions,
+			ngError:       errors.New("This sentence is false."),
+			wantError:     errors.New("This sentence is false."),
+		},
+		"NodeGroup returns no value fallbacks to default": {
+			globalOptions:         defaultOptions,
+			wantScaleDownUnneeded: 3 * time.Minute,
+		},
+		"NodeGroup option overrides global default": {
+			globalOptions: defaultOptions,
+			ngOptions: &config.NodeGroupAutoscalingOptions{
+				ScaleDownUnneededTime: 10 * time.Minute,
+			},
+			wantScaleDownUnneeded: 10 * time.Minute,
+		},
+	}
+	for tn, tc := range cases {
+		t.Run(tn, func(t *testing.T) {
+			context := &context.AutoscalingContext{
+				AutoscalingOptions: config.AutoscalingOptions{
+					NodeGroupAutoscalingOptions: tc.globalOptions,
+				},
+			}
+			ng := &mocks.NodeGroup{}
+			ng.On("GetOptions", tc.globalOptions).Return(tc.ngOptions, tc.ngError).Once()
+			p := NewDefaultNodeGroupConfigProcessor()
+			res, err := p.GetScaleDownUnneededTime(context, ng)
+			assert.Equal(t, res, tc.wantScaleDownUnneeded)
+			assert.Equal(t, err, tc.wantError)
+		})
+	}
+}

cluster-autoscaler/processors/processors.go

@@ -17,6 +17,7 @@ limitations under the License.
 package processors
 
 import (
+	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupconfig"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroups"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodegroupset"
 	"k8s.io/autoscaler/cluster-autoscaler/processors/nodeinfos"
@@ -46,6 +47,8 @@ type AutoscalingProcessors struct {
 	NodeGroupManager nodegroups.NodeGroupManager
 	// NodeInfoProcessor is used to process nodeInfos after they're created.
 	NodeInfoProcessor nodeinfos.NodeInfoProcessor
+	// NodeGroupConfigProcessor provides config option for each NodeGroup.
+	NodeGroupConfigProcessor nodegroupconfig.NodeGroupConfigProcessor
 }
 
 // DefaultProcessors returns default set of processors.
@@ -60,6 +63,7 @@ func DefaultProcessors() *AutoscalingProcessors {
 		AutoscalingStatusProcessor: status.NewDefaultAutoscalingStatusProcessor(),
 		NodeGroupManager:           nodegroups.NewDefaultNodeGroupManager(),
 		NodeInfoProcessor:          nodeinfos.NewDefaultNodeInfoProcessor(),
+		NodeGroupConfigProcessor:   nodegroupconfig.NewDefaultNodeGroupConfigProcessor(),
 	}
 }
 
@@ -74,4 +78,5 @@ func (ap *AutoscalingProcessors) CleanUp() {
 	ap.NodeGroupManager.CleanUp()
 	ap.ScaleDownNodeProcessor.CleanUp()
 	ap.NodeInfoProcessor.CleanUp()
+	ap.NodeGroupConfigProcessor.CleanUp()
 }