diff --git a/cluster-autoscaler/cloudprovider/aws/CA_with_AWS_IAM_OIDC.md b/cluster-autoscaler/cloudprovider/aws/CA_with_AWS_IAM_OIDC.md
index cf251787c234465c786aa0fa475b8ed4126c8558..f638a9808d6b5fcff76ce448595c1ddb02bf06f5 100644
--- a/cluster-autoscaler/cloudprovider/aws/CA_with_AWS_IAM_OIDC.md
+++ b/cluster-autoscaler/cloudprovider/aws/CA_with_AWS_IAM_OIDC.md
@@ -67,6 +67,7 @@ __NOTE:__ Please see [the README](README.md#IAM-Policy) for more information on
"autoscaling:DescribeAutoScalingGroups",
"autoscaling:DescribeAutoScalingInstances",
"autoscaling:DescribeLaunchConfigurations",
+ "autoscaling:DescribeScalingActivities",
"autoscaling:DescribeTags",
"ec2:DescribeInstanceTypes",
"ec2:DescribeLaunchTemplateVersions"
diff --git a/cluster-autoscaler/cloudprovider/aws/README.md b/cluster-autoscaler/cloudprovider/aws/README.md
index c4064ad7e9312129fe3dde4fc64f088ff90afc59..956f797ade9bb230fc3765a736131604718a0a8b 100644
--- a/cluster-autoscaler/cloudprovider/aws/README.md
+++ b/cluster-autoscaler/cloudprovider/aws/README.md
@@ -93,6 +93,7 @@ binary, replacing min and max node counts and the ASG:
"autoscaling:DescribeAutoScalingGroups",
"autoscaling:DescribeAutoScalingInstances",
"autoscaling:DescribeLaunchConfigurations",
+ "autoscaling:DescribeScalingActivities",
"autoscaling:SetDesiredCapacity",
"autoscaling:TerminateInstanceInAutoScalingGroup"
],
diff --git a/cluster-autoscaler/cloudprovider/aws/auto_scaling_groups.go b/cluster-autoscaler/cloudprovider/aws/auto_scaling_groups.go
index 1f85b0906780ecf879e4d5c6740d4b7233437dcf..f15c778d9eee38942b008ac52466b8df211d4371 100644
--- a/cluster-autoscaler/cloudprovider/aws/auto_scaling_groups.go
+++ b/cluster-autoscaler/cloudprovider/aws/auto_scaling_groups.go
@@ -31,14 +31,16 @@ import (
)
const (
- scaleToZeroSupported = true
- placeholderInstanceNamePrefix = "i-placeholder"
+ scaleToZeroSupported = true
+ placeholderInstanceNamePrefix = "i-placeholder"
+ placeholderUnfulfillableStatus = "placeholder-cannot-be-fulfilled"
)
type asgCache struct {
- registeredAsgs []*asg
+ registeredAsgs map[AwsRef]*asg
asgToInstances map[AwsRef][]AwsInstanceRef
instanceToAsg map[AwsInstanceRef]*asg
+ instanceStatus map[AwsInstanceRef]*string
asgInstanceTypeCache *instanceTypeExpirationStore
mutex sync.Mutex
awsService *awsWrapper
@@ -62,9 +64,10 @@ type mixedInstancesPolicy struct {
type asg struct {
AwsRef
- minSize int
- maxSize int
- curSize int
+ minSize int
+ maxSize int
+ curSize int
+ lastUpdateTime time.Time
AvailabilityZones []string
LaunchConfigurationName string
@@ -75,10 +78,11 @@ type asg struct {
func newASGCache(awsService *awsWrapper, explicitSpecs []string, autoDiscoverySpecs []asgAutoDiscoveryConfig) (*asgCache, error) {
registry := &asgCache{
- registeredAsgs: make([]*asg, 0),
+ registeredAsgs: make(map[AwsRef]*asg, 0),
awsService: awsService,
asgToInstances: make(map[AwsRef][]AwsInstanceRef),
instanceToAsg: make(map[AwsInstanceRef]*asg),
+ instanceStatus: make(map[AwsInstanceRef]*string),
asgInstanceTypeCache: newAsgInstanceTypeCache(awsService),
interrupt: make(chan struct{}),
asgAutoDiscoverySpecs: autoDiscoverySpecs,
@@ -121,53 +125,44 @@ func (m *asgCache) parseExplicitAsgs(specs []string) error {
// Register ASG. Returns the registered ASG.
func (m *asgCache) register(asg *asg) *asg {
- for i := range m.registeredAsgs {
- if existing := m.registeredAsgs[i]; existing.AwsRef == asg.AwsRef {
- if reflect.DeepEqual(existing, asg) {
- return existing
- }
+ if existing, asgExists := m.registeredAsgs[asg.AwsRef]; asgExists {
+ if reflect.DeepEqual(existing, asg) {
+ return existing
+ }
- klog.V(4).Infof("Updating ASG %s", asg.AwsRef.Name)
+ klog.V(4).Infof("Updating ASG %s", asg.AwsRef.Name)
- // Explicit registered groups should always use the manually provided min/max
- // values and the not the ones returned by the API
- if !m.explicitlyConfigured[asg.AwsRef] {
- existing.minSize = asg.minSize
- existing.maxSize = asg.maxSize
- }
+ // Explicit registered groups should always use the manually provided min/max
+ // values and the not the ones returned by the API
+ if !m.explicitlyConfigured[asg.AwsRef] {
+ existing.minSize = asg.minSize
+ existing.maxSize = asg.maxSize
+ }
- existing.curSize = asg.curSize
+ existing.curSize = asg.curSize
- // Those information are mainly required to create templates when scaling
- // from zero
- existing.AvailabilityZones = asg.AvailabilityZones
- existing.LaunchConfigurationName = asg.LaunchConfigurationName
- existing.LaunchTemplate = asg.LaunchTemplate
- existing.MixedInstancesPolicy = asg.MixedInstancesPolicy
- existing.Tags = asg.Tags
+ // Those information are mainly required to create templates when scaling
+ // from zero
+ existing.AvailabilityZones = asg.AvailabilityZones
+ existing.LaunchConfigurationName = asg.LaunchConfigurationName
+ existing.LaunchTemplate = asg.LaunchTemplate
+ existing.MixedInstancesPolicy = asg.MixedInstancesPolicy
+ existing.Tags = asg.Tags
- return existing
- }
+ return existing
}
klog.V(1).Infof("Registering ASG %s", asg.AwsRef.Name)
- m.registeredAsgs = append(m.registeredAsgs, asg)
+ m.registeredAsgs[asg.AwsRef] = asg
return asg
}
// Unregister ASG. Returns the unregistered ASG.
func (m *asgCache) unregister(a *asg) *asg {
- updated := make([]*asg, 0, len(m.registeredAsgs))
- var changed *asg
- for _, existing := range m.registeredAsgs {
- if existing.AwsRef == a.AwsRef {
- klog.V(1).Infof("Unregistered ASG %s", a.AwsRef.Name)
- changed = a
- continue
- }
- updated = append(updated, existing)
+ if _, asgExists := m.registeredAsgs[a.AwsRef]; asgExists {
+ klog.V(1).Infof("Unregistered ASG %s", a.AwsRef.Name)
+ delete(m.registeredAsgs, a.AwsRef)
}
- m.registeredAsgs = updated
- return changed
+ return a
}
func (m *asgCache) buildAsgFromSpec(spec string) (*asg, error) {
@@ -184,7 +179,7 @@ func (m *asgCache) buildAsgFromSpec(spec string) (*asg, error) {
}
// Get returns the currently registered ASGs
-func (m *asgCache) Get() []*asg {
+func (m *asgCache) Get() map[AwsRef]*asg {
m.mutex.Lock()
defer m.mutex.Unlock()
@@ -226,6 +221,17 @@ func (m *asgCache) InstancesByAsg(ref AwsRef) ([]AwsInstanceRef, error) {
return nil, fmt.Errorf("error while looking for instances of ASG: %s", ref)
}
+func (m *asgCache) InstanceStatus(ref AwsInstanceRef) (*string, error) {
+ m.mutex.Lock()
+ defer m.mutex.Unlock()
+
+ if status, found := m.instanceStatus[ref]; found {
+ return status, nil
+ }
+
+ return nil, fmt.Errorf("could not find instance %v", ref)
+}
+
func (m *asgCache) SetAsgSize(asg *asg, size int) error {
m.mutex.Lock()
defer m.mutex.Unlock()
@@ -248,6 +254,7 @@ func (m *asgCache) setAsgSizeNoLock(asg *asg, size int) error {
}
// Proactively set the ASG size so autoscaler makes better decisions
+ asg.lastUpdateTime = start
asg.curSize = size
return nil
@@ -367,6 +374,7 @@ func (m *asgCache) regenerate() error {
newInstanceToAsgCache := make(map[AwsInstanceRef]*asg)
newAsgToInstancesCache := make(map[AwsRef][]AwsInstanceRef)
+ newInstanceStatusMap := make(map[AwsInstanceRef]*string)
// Build list of known ASG names
refreshNames, err := m.buildAsgNames()
@@ -403,6 +411,7 @@ func (m *asgCache) regenerate() error {
ref := m.buildInstanceRefFromAWS(instance)
newInstanceToAsgCache[ref] = asg
newAsgToInstancesCache[asg.AwsRef][i] = ref
+ newInstanceStatusMap[ref] = instance.HealthStatus
}
}
@@ -431,6 +440,7 @@ func (m *asgCache) regenerate() error {
m.asgToInstances = newAsgToInstancesCache
m.instanceToAsg = newInstanceToAsgCache
m.autoscalingOptions = newAutoscalingOptions
+ m.instanceStatus = newInstanceStatusMap
return nil
}
@@ -444,17 +454,57 @@ func (m *asgCache) createPlaceholdersForDesiredNonStartedInstances(groups []*aut
klog.V(4).Infof("Instance group %s has only %d instances created while requested count is %d. "+
"Creating placeholder instances.", *g.AutoScalingGroupName, realInstances, desired)
+
+ healthStatus := ""
+ isAvailable, err := m.isNodeGroupAvailable(g)
+ if err != nil {
+ klog.V(4).Infof("Could not check instance availability, creating placeholder node anyways: %v", err)
+ } else if !isAvailable {
+ klog.Warningf("Instance group %s cannot provision any more nodes!", *g.AutoScalingGroupName)
+ healthStatus = placeholderUnfulfillableStatus
+ }
+
for i := realInstances; i < desired; i++ {
id := fmt.Sprintf("%s-%s-%d", placeholderInstanceNamePrefix, *g.AutoScalingGroupName, i)
g.Instances = append(g.Instances, &autoscaling.Instance{
InstanceId: &id,
AvailabilityZone: g.AvailabilityZones[0],
+ HealthStatus: &healthStatus,
})
}
}
return groups
}
+func (m *asgCache) isNodeGroupAvailable(group *autoscaling.Group) (bool, error) {
+ input := &autoscaling.DescribeScalingActivitiesInput{
+ AutoScalingGroupName: group.AutoScalingGroupName,
+ }
+
+ start := time.Now()
+ response, err := m.awsService.DescribeScalingActivities(input)
+ observeAWSRequest("DescribeScalingActivities", err, start)
+ if err != nil {
+ return true, err // If we can't describe the scaling activities we assume the node group is available
+ }
+
+ for _, activity := range response.Activities {
+ asgRef := AwsRef{Name: *group.AutoScalingGroupName}
+ if a, ok := m.registeredAsgs[asgRef]; ok {
+ lut := a.lastUpdateTime
+ if activity.StartTime.Before(lut) {
+ break
+ } else if *activity.StatusCode == "Failed" {
+ klog.Warningf("ASG %s scaling failed with %s", asgRef.Name, *activity)
+ return false, nil
+ }
+ } else {
+ klog.V(4).Infof("asg %v is not registered yet, skipping DescribeScalingActivities check", asgRef.Name)
+ }
+ }
+ return true, nil
+}
+
func (m *asgCache) buildAsgFromAWS(g *autoscaling.Group) (*asg, error) {
spec := dynamic.NodeGroupSpec{
Name: aws.StringValue(g.AutoScalingGroupName),
diff --git a/cluster-autoscaler/cloudprovider/aws/auto_scaling_groups_test.go b/cluster-autoscaler/cloudprovider/aws/auto_scaling_groups_test.go
index e9f9ad1ab79eaa43a2d7e8fd06fbc2ab36890069..c5078b03b5b72b15ddbd42a9229a61345bbfe62f 100644
--- a/cluster-autoscaler/cloudprovider/aws/auto_scaling_groups_test.go
+++ b/cluster-autoscaler/cloudprovider/aws/auto_scaling_groups_test.go
@@ -17,8 +17,12 @@ limitations under the License.
package aws
import (
+ "errors"
"testing"
+ "time"
+ "github.com/aws/aws-sdk-go/aws"
+ "github.com/aws/aws-sdk-go/service/autoscaling"
"github.com/stretchr/testify/assert"
)
@@ -46,3 +50,119 @@ func validateAsg(t *testing.T, asg *asg, name string, minSize int, maxSize int)
assert.Equal(t, minSize, asg.minSize)
assert.Equal(t, maxSize, asg.maxSize)
}
+
+func TestCreatePlaceholders(t *testing.T) {
+ registeredAsgName := aws.String("test-asg")
+ registeredAsgRef := AwsRef{Name: *registeredAsgName}
+
+ cases := []struct {
+ name string
+ desiredCapacity *int64
+ activities []*autoscaling.Activity
+ groupLastUpdateTime time.Time
+ describeErr error
+ asgToCheck *string
+ }{
+ {
+ name: "add placeholders successful",
+ desiredCapacity: aws.Int64(10),
+ },
+ {
+ name: "no placeholders needed",
+ desiredCapacity: aws.Int64(0),
+ },
+ {
+ name: "DescribeScalingActivities failed",
+ desiredCapacity: aws.Int64(1),
+ describeErr: errors.New("timeout"),
+ },
+ {
+ name: "early abort if AWS scaling up fails",
+ desiredCapacity: aws.Int64(1),
+ activities: []*autoscaling.Activity{
+ {
+ StatusCode: aws.String("Failed"),
+ StartTime: aws.Time(time.Unix(10, 0)),
+ },
+ },
+ groupLastUpdateTime: time.Unix(9, 0),
+ },
+ {
+ name: "AWS scaling failed event before CA scale_up",
+ desiredCapacity: aws.Int64(1),
+ activities: []*autoscaling.Activity{
+ {
+ StatusCode: aws.String("Failed"),
+ StartTime: aws.Time(time.Unix(9, 0)),
+ },
+ },
+ groupLastUpdateTime: time.Unix(10, 0),
+ },
+ {
+ name: "asg not registered",
+ desiredCapacity: aws.Int64(10),
+ activities: []*autoscaling.Activity{
+ {
+ StatusCode: aws.String("Failed"),
+ StartTime: aws.Time(time.Unix(10, 0)),
+ },
+ },
+ groupLastUpdateTime: time.Unix(9, 0),
+ asgToCheck: aws.String("unregisteredAsgName"),
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ shouldCallDescribeScalingActivities := true
+ if *tc.desiredCapacity == int64(0) {
+ shouldCallDescribeScalingActivities = false
+ }
+
+ asgName := registeredAsgName
+ if tc.asgToCheck != nil {
+ asgName = tc.asgToCheck
+ }
+
+ a := &autoScalingMock{}
+ if shouldCallDescribeScalingActivities {
+ a.On("DescribeScalingActivities", &autoscaling.DescribeScalingActivitiesInput{
+ AutoScalingGroupName: asgName,
+ }).Return(
+ &autoscaling.DescribeScalingActivitiesOutput{Activities: tc.activities},
+ tc.describeErr,
+ ).Once()
+ }
+
+ asgCache := &asgCache{
+ awsService: &awsWrapper{
+ autoScalingI: a,
+ ec2I: nil,
+ },
+ registeredAsgs: map[AwsRef]*asg{
+ registeredAsgRef: {
+ AwsRef: registeredAsgRef,
+ lastUpdateTime: tc.groupLastUpdateTime,
+ },
+ },
+ }
+
+ groups := []*autoscaling.Group{
+ {
+ AutoScalingGroupName: asgName,
+ AvailabilityZones: []*string{aws.String("westeros-1a")},
+ DesiredCapacity: tc.desiredCapacity,
+ Instances: []*autoscaling.Instance{},
+ },
+ }
+ asgCache.createPlaceholdersForDesiredNonStartedInstances(groups)
+ assert.Equal(t, int64(len(groups[0].Instances)), *tc.desiredCapacity)
+ if tc.activities != nil && *tc.activities[0].StatusCode == "Failed" && tc.activities[0].StartTime.After(tc.groupLastUpdateTime) && asgName == registeredAsgName {
+ assert.Equal(t, *groups[0].Instances[0].HealthStatus, placeholderUnfulfillableStatus)
+ } else if len(groups[0].Instances) > 0 {
+ assert.Equal(t, *groups[0].Instances[0].HealthStatus, "")
+ }
+ a.AssertExpectations(t)
+ })
+ }
+}
diff --git a/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go b/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go
index 8357966b59cf08b09a1f465e376c15dc8500c13f..4bf11960209d39071a47cfea27f4a93fa4590a14 100644
--- a/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go
@@ -86,12 +86,12 @@ func (aws *awsCloudProvider) GetAvailableGPUTypes() map[string]struct{} {
// NodeGroups returns all node groups configured for this cloud provider.
func (aws *awsCloudProvider) NodeGroups() []cloudprovider.NodeGroup {
asgs := aws.awsManager.getAsgs()
- ngs := make([]cloudprovider.NodeGroup, len(asgs))
- for i, asg := range asgs {
- ngs[i] = &AwsNodeGroup{
+ ngs := make([]cloudprovider.NodeGroup, 0, len(asgs))
+ for _, asg := range asgs {
+ ngs = append(ngs, &AwsNodeGroup{
asg: asg,
awsManager: aws.awsManager,
- }
+ })
}
return ngs
@@ -320,7 +320,24 @@ func (ng *AwsNodeGroup) Nodes() ([]cloudprovider.Instance, error) {
instances := make([]cloudprovider.Instance, len(asgNodes))
for i, asgNode := range asgNodes {
- instances[i] = cloudprovider.Instance{Id: asgNode.ProviderID}
+ var status *cloudprovider.InstanceStatus
+ instanceStatusString, err := ng.awsManager.GetInstanceStatus(asgNode)
+ if err != nil {
+ klog.V(4).Infof("Could not get instance status, continuing anyways: %v", err)
+ } else if instanceStatusString != nil && *instanceStatusString == placeholderUnfulfillableStatus {
+ status = &cloudprovider.InstanceStatus{
+ State: cloudprovider.InstanceCreating,
+ ErrorInfo: &cloudprovider.InstanceErrorInfo{
+ ErrorClass: cloudprovider.OutOfResourcesErrorClass,
+ ErrorCode: placeholderUnfulfillableStatus,
+ ErrorMessage: "AWS cannot provision any more instances for this node group",
+ },
+ }
+ }
+ instances[i] = cloudprovider.Instance{
+ Id: asgNode.ProviderID,
+ Status: status,
+ }
}
return instances, nil
}
diff --git a/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider_test.go b/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider_test.go
index 7c61340ec0e81617b51209944889d7665a1c4e5c..e1d6a92a1ef93852933c8337b408ba8d66a21ce7 100644
--- a/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider_test.go
+++ b/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider_test.go
@@ -29,7 +29,7 @@ import (
var testAwsManager = &AwsManager{
asgCache: &asgCache{
- registeredAsgs: make([]*asg, 0),
+ registeredAsgs: make(map[AwsRef]*asg, 0),
asgToInstances: make(map[AwsRef][]AwsInstanceRef),
instanceToAsg: make(map[AwsInstanceRef]*asg),
interrupt: make(chan struct{}),
@@ -43,7 +43,7 @@ func newTestAwsManagerWithMockServices(mockAutoScaling autoScalingI, mockEC2 ec2
return &AwsManager{
awsService: awsService,
asgCache: &asgCache{
- registeredAsgs: make([]*asg, 0),
+ registeredAsgs: make(map[AwsRef]*asg, 0),
asgToInstances: make(map[AwsRef][]AwsInstanceRef),
instanceToAsg: make(map[AwsInstanceRef]*asg),
asgInstanceTypeCache: newAsgInstanceTypeCache(&awsService),
@@ -463,6 +463,12 @@ func TestDeleteNodesWithPlaceholder(t *testing.T) {
expectedInstancesCount = 1
}).Return(nil)
+ a.On("DescribeScalingActivities",
+ &autoscaling.DescribeScalingActivitiesInput{
+ AutoScalingGroupName: aws.String("test-asg"),
+ },
+ ).Return(&autoscaling.DescribeScalingActivitiesOutput{}, nil)
+
provider.Refresh()
initialSize, err := asgs[0].TargetSize()
diff --git a/cluster-autoscaler/cloudprovider/aws/aws_manager.go b/cluster-autoscaler/cloudprovider/aws/aws_manager.go
index 8648a37a1819005830a6b6932625b82ce8e1470d..17d29e8ff7076af3b9e66c43110629b8ad6d0cc2 100644
--- a/cluster-autoscaler/cloudprovider/aws/aws_manager.go
+++ b/cluster-autoscaler/cloudprovider/aws/aws_manager.go
@@ -273,7 +273,7 @@ func (m *AwsManager) Cleanup() {
m.asgCache.Cleanup()
}
-func (m *AwsManager) getAsgs() []*asg {
+func (m *AwsManager) getAsgs() map[AwsRef]*asg {
return m.asgCache.Get()
}
@@ -301,6 +301,11 @@ func (m *AwsManager) GetAsgNodes(ref AwsRef) ([]AwsInstanceRef, error) {
return m.asgCache.InstancesByAsg(ref)
}
+// GetInstanceStatus returns the status of ASG nodes
+func (m *AwsManager) GetInstanceStatus(ref AwsInstanceRef) (*string, error) {
+ return m.asgCache.InstanceStatus(ref)
+}
+
func (m *AwsManager) getAsgTemplate(asg *asg) (*asgTemplate, error) {
if len(asg.AvailabilityZones) < 1 {
return nil, fmt.Errorf("unable to get first AvailabilityZone for ASG %q", asg.Name)
diff --git a/cluster-autoscaler/cloudprovider/aws/aws_manager_test.go b/cluster-autoscaler/cloudprovider/aws/aws_manager_test.go
index d18efa2439e6ff700958ca5b7129f51deced1ad0..50df508f8df639e03c4457618f72d9d09701b599 100644
--- a/cluster-autoscaler/cloudprovider/aws/aws_manager_test.go
+++ b/cluster-autoscaler/cloudprovider/aws/aws_manager_test.go
@@ -358,6 +358,7 @@ func makeTaintSet(taints []apiv1.Taint) map[apiv1.Taint]bool {
func TestFetchExplicitAsgs(t *testing.T) {
min, max, groupname := 1, 10, "coolasg"
+ asgRef := AwsRef{Name: groupname}
a := &autoScalingMock{}
a.On("DescribeAutoScalingGroups", &autoscaling.DescribeAutoScalingGroupsInput{
@@ -390,6 +391,12 @@ func TestFetchExplicitAsgs(t *testing.T) {
}}, false)
}).Return(nil)
+ a.On("DescribeScalingActivities",
+ &autoscaling.DescribeScalingActivitiesInput{
+ AutoScalingGroupName: aws.String("coolasg"),
+ },
+ ).Return(&autoscaling.DescribeScalingActivitiesOutput{}, nil)
+
do := cloudprovider.NodeGroupDiscoveryOptions{
// Register the same node group twice with different max nodes.
// The intention is to test that the asgs.Register method will update
@@ -408,7 +415,7 @@ func TestFetchExplicitAsgs(t *testing.T) {
asgs := m.asgCache.Get()
assert.Equal(t, 1, len(asgs))
- validateAsg(t, asgs[0], groupname, min, max)
+ validateAsg(t, asgs[asgRef], groupname, min, max)
}
func TestGetASGTemplate(t *testing.T) {
@@ -504,6 +511,7 @@ func TestGetASGTemplate(t *testing.T) {
func TestFetchAutoAsgs(t *testing.T) {
min, max := 1, 10
groupname, tags := "coolasg", []string{"tag", "anothertag"}
+ asgRef := AwsRef{Name: groupname}
a := &autoScalingMock{}
// Lookup groups associated with tags
@@ -547,6 +555,12 @@ func TestFetchAutoAsgs(t *testing.T) {
}}}, false)
}).Return(nil).Twice()
+ a.On("DescribeScalingActivities",
+ &autoscaling.DescribeScalingActivitiesInput{
+ AutoScalingGroupName: aws.String("coolasg"),
+ },
+ ).Return(&autoscaling.DescribeScalingActivitiesOutput{}, nil)
+
do := cloudprovider.NodeGroupDiscoveryOptions{
NodeGroupAutoDiscoverySpecs: []string{fmt.Sprintf("asg:tag=%s", strings.Join(tags, ","))},
}
@@ -561,7 +575,7 @@ func TestFetchAutoAsgs(t *testing.T) {
asgs := m.asgCache.Get()
assert.Equal(t, 1, len(asgs))
- validateAsg(t, asgs[0], groupname, min, max)
+ validateAsg(t, asgs[asgRef], groupname, min, max)
// Simulate the previously discovered ASG disappearing
a.On("DescribeTagsPages", mock.MatchedBy(tagsMatcher(expectedTagsInput)),
diff --git a/cluster-autoscaler/cloudprovider/aws/aws_wrapper.go b/cluster-autoscaler/cloudprovider/aws/aws_wrapper.go
index 86453300b757ca56ea5d275a4e4c219d93c6044e..1c7ef4eaa1cea4f3edacaf5b04781d0834151f63 100644
--- a/cluster-autoscaler/cloudprovider/aws/aws_wrapper.go
+++ b/cluster-autoscaler/cloudprovider/aws/aws_wrapper.go
@@ -33,6 +33,7 @@ import (
type autoScalingI interface {
DescribeAutoScalingGroupsPages(input *autoscaling.DescribeAutoScalingGroupsInput, fn func(*autoscaling.DescribeAutoScalingGroupsOutput, bool) bool) error
DescribeLaunchConfigurations(*autoscaling.DescribeLaunchConfigurationsInput) (*autoscaling.DescribeLaunchConfigurationsOutput, error)
+ DescribeScalingActivities(*autoscaling.DescribeScalingActivitiesInput) (*autoscaling.DescribeScalingActivitiesOutput, error)
DescribeTagsPages(input *autoscaling.DescribeTagsInput, fn func(*autoscaling.DescribeTagsOutput, bool) bool) error
SetDesiredCapacity(input *autoscaling.SetDesiredCapacityInput) (*autoscaling.SetDesiredCapacityOutput, error)
TerminateInstanceInAutoScalingGroup(input *autoscaling.TerminateInstanceInAutoScalingGroupInput) (*autoscaling.TerminateInstanceInAutoScalingGroupOutput, error)
diff --git a/cluster-autoscaler/cloudprovider/aws/aws_wrapper_test.go b/cluster-autoscaler/cloudprovider/aws/aws_wrapper_test.go
index 4fc2778db03e7cd1c68f01404fefde114d342fe7..e360f341c8b61dee57028231b0a198db2fd73986 100644
--- a/cluster-autoscaler/cloudprovider/aws/aws_wrapper_test.go
+++ b/cluster-autoscaler/cloudprovider/aws/aws_wrapper_test.go
@@ -45,6 +45,11 @@ func (a *autoScalingMock) DescribeLaunchConfigurations(i *autoscaling.DescribeLa
return args.Get(0).(*autoscaling.DescribeLaunchConfigurationsOutput), nil
}
+func (a *autoScalingMock) DescribeScalingActivities(i *autoscaling.DescribeScalingActivitiesInput) (*autoscaling.DescribeScalingActivitiesOutput, error) {
+ args := a.Called(i)
+ return args.Get(0).(*autoscaling.DescribeScalingActivitiesOutput), args.Error(1)
+}
+
func (a *autoScalingMock) DescribeTagsPages(i *autoscaling.DescribeTagsInput, fn func(*autoscaling.DescribeTagsOutput, bool) bool) error {
args := a.Called(i, fn)
return args.Error(0)
diff --git a/cluster-autoscaler/cloudprovider/aws/instance_type_cache.go b/cluster-autoscaler/cloudprovider/aws/instance_type_cache.go
index 2b2ce74c8044319560df0e25782b0bac410df747..90c7f2fd2c4d6ed077e78a0fd9ff7631e52811a7 100644
--- a/cluster-autoscaler/cloudprovider/aws/instance_type_cache.go
+++ b/cluster-autoscaler/cloudprovider/aws/instance_type_cache.go
@@ -84,7 +84,7 @@ func (c *jitterClock) Since(ts time.Time) time.Duration {
return since
}
-func (es instanceTypeExpirationStore) populate(autoscalingGroups []*asg) error {
+func (es instanceTypeExpirationStore) populate(autoscalingGroups map[AwsRef]*asg) error {
asgsToQuery := []*asg{}
if c, ok := es.jitterClock.(*jitterClock); ok {
diff --git a/cluster-autoscaler/cloudprovider/aws/instance_type_cache_test.go b/cluster-autoscaler/cloudprovider/aws/instance_type_cache_test.go
index 94d0018e20ccde5218ab7e42d76a39ed2e05eb22..88cc75f087a8468ce3451113468403e64cee2ebb 100644
--- a/cluster-autoscaler/cloudprovider/aws/instance_type_cache_test.go
+++ b/cluster-autoscaler/cloudprovider/aws/instance_type_cache_test.go
@@ -79,9 +79,10 @@ func TestLTVersionChange(t *testing.T) {
m := newAsgInstanceTypeCacheWithClock(&awsWrapper{a, e, nil}, fakeClock, fakeStore)
for i := 0; i < 2; i++ {
- err := m.populate([]*asg{
- {
- AwsRef: AwsRef{Name: asgName},
+ asgRef := AwsRef{Name: asgName}
+ err := m.populate(map[AwsRef]*asg{
+ asgRef: {
+ AwsRef: asgRef,
LaunchTemplate: &launchTemplate{
name: ltName,
version: aws.StringValue(ltVersions[i]),