feat: ICE AZs temporarily when subnets in that AZ run out of available ips (#8199)

Author: DerekFrank

pkg/cache/unavailableofferings.go

@@ -34,13 +34,15 @@ type UnavailableOfferings struct {
 	// key: <capacityType>:<instanceType>:<zone>, value: struct{}{}
 	offeringCache     *cache.Cache
 	capacityTypeCache *cache.Cache
+	azCache           *cache.Cache
 	SeqNum            uint64
 }
 
 func NewUnavailableOfferings() *UnavailableOfferings {
 	uo := &UnavailableOfferings{
 		offeringCache:     cache.New(UnavailableOfferingsTTL, UnavailableOfferingsCleanupInterval),
 		capacityTypeCache: cache.New(UnavailableOfferingsTTL, UnavailableOfferingsCleanupInterval),
+		azCache:           cache.New(UnavailableOfferingsTTL, UnavailableOfferingsCleanupInterval),
 		SeqNum:            0,
 	}
 	uo.offeringCache.OnEvicted(func(_ string, _ interface{}) {
@@ -56,7 +58,8 @@ func NewUnavailableOfferings() *UnavailableOfferings {
 func (u *UnavailableOfferings) IsUnavailable(instanceType ec2types.InstanceType, zone, capacityType string) bool {
 	_, offeringFound := u.offeringCache.Get(u.key(instanceType, zone, capacityType))
 	_, capacityTypeFound := u.capacityTypeCache.Get(capacityType)
-	return offeringFound || capacityTypeFound
+	_, azFound := u.azCache.Get(zone)
+	return offeringFound || capacityTypeFound || azFound
 }
 
 // MarkUnavailable communicates recently observed temporary capacity shortages in the provided offerings
@@ -83,13 +86,18 @@ func (u *UnavailableOfferings) MarkCapacityTypeUnavailable(capacityType string)
 	atomic.AddUint64(&u.SeqNum, 1)
 }
 
+func (u *UnavailableOfferings) MarkAZUnavailable(zone string) {
+	u.azCache.SetDefault(zone, struct{}{})
+}
+
 func (u *UnavailableOfferings) Delete(instanceType ec2types.InstanceType, zone string, capacityType string) {
 	u.offeringCache.Delete(u.key(instanceType, zone, capacityType))
 }
 
 func (u *UnavailableOfferings) Flush() {
 	u.offeringCache.Flush()
 	u.capacityTypeCache.Flush()
+	u.azCache.Flush()
 }
 
 // key returns the cache key for all offerings in the cache

pkg/errors/errors.go

@@ -30,6 +30,7 @@ const (
 	UnauthorizedOperationErrorCode                 = "UnauthorizedOperation"
 	RateLimitingErrorCode                          = "RequestLimitExceeded"
 	ServiceLinkedRoleCreationNotPermittedErrorCode = "AuthFailure.ServiceLinkedRoleCreationNotPermitted"
+	InsufficientFreeAddressesInSubnetErrorCode     = "InsufficientFreeAddressesInSubnet"
 )
 
 var (
@@ -159,6 +160,10 @@ func IsServiceLinkedRoleCreationNotPermitted(err ec2types.CreateFleetError) bool
 	return *err.ErrorCode == ServiceLinkedRoleCreationNotPermittedErrorCode
 }
 
+func IsInsufficientFreeAddressesInSubnet(err ec2types.CreateFleetError) bool {
+	return *err.ErrorCode == InsufficientFreeAddressesInSubnetErrorCode
+}
+
 // IsReservationCapacityExceeded returns true if the fleet error means there is no remaining capacity for the provided
 // capacity reservation.
 func IsReservationCapacityExceeded(err ec2types.CreateFleetError) bool {

pkg/providers/instance/instance.go

@@ -445,6 +445,13 @@ func (p *DefaultProvider) updateUnavailableOfferingsCache(
 	nodeClaim *karpv1.NodeClaim,
 	instanceTypes []*cloudprovider.InstanceType,
 ) {
+	for _, err := range errs {
+		zone := lo.FromPtr(err.LaunchTemplateAndOverrides.Overrides.AvailabilityZone)
+		if awserrors.IsInsufficientFreeAddressesInSubnet(err) && zone != "" {
+			p.unavailableOfferings.MarkAZUnavailable(zone)
+		}
+	}
+
 	if capacityType != karpv1.CapacityTypeReserved {
 		for _, err := range errs {
 			if awserrors.IsUnfulfillableCapacity(err) {

pkg/providers/instance/suite_test.go

@@ -351,4 +351,41 @@ var _ = Describe("InstanceProvider", func() {
 		retrievedIDs := sets.New[string](lo.Map(instances, func(i *instance.Instance, _ int) string { return i.ID })...)
 		Expect(ids.Equal(retrievedIDs)).To(BeTrue())
 	})
+	It("should mark subnets as unavailable when they run out of IPs", func() {
+		ExpectApplied(ctx, env.Client, nodeClaim, nodePool, nodeClass)
+		nodeClass = ExpectExists(ctx, env.Client, nodeClass)
+		awsEnv.EC2API.CreateFleetBehavior.Output.Set(&ec2.CreateFleetOutput{
+			Errors: []ec2types.CreateFleetError{
+				{
+					ErrorCode:    lo.ToPtr("InsufficientFreeAddressesInSubnet"),
+					ErrorMessage: lo.ToPtr("There are insufficient free addresses in that subnet to run instance"),
+					LaunchTemplateAndOverrides: &ec2types.LaunchTemplateAndOverridesResponse{
+						Overrides: &ec2types.FleetLaunchTemplateOverrides{
+							InstanceType:     "m5.xlarge",
+							AvailabilityZone: lo.ToPtr("test-zone-1a"),
+						},
+					},
+				},
+			},
+		})
+		instanceTypes, err := cloudProvider.GetInstanceTypes(ctx, nodePool)
+		Expect(err).ToNot(HaveOccurred())
+
+		// We expect to treat that error as an ICE
+		instance, err := awsEnv.InstanceProvider.Create(ctx, nodeClass, nodeClaim, nil, instanceTypes)
+		Expect(corecloudprovider.IsInsufficientCapacityError(err)).To(BeTrue())
+		Expect(instance).To(BeNil())
+
+		// We should have set the zone used in the request as unavailable for all instance types
+		for _, instance := range instanceTypes {
+			Expect(awsEnv.UnavailableOfferingsCache.IsUnavailable(ec2types.InstanceType(instance.Name), "test-zone-1a", "on-demand")).To(BeTrue())
+		}
+		// But we should not have set the other zones as unavailable
+		zones := []string{"test-zone-1b", "test-zone-1c"}
+		for _, zone := range zones {
+			for _, instance := range instanceTypes {
+				Expect(awsEnv.UnavailableOfferingsCache.IsUnavailable(ec2types.InstanceType(instance.Name), zone, "on-demand")).To(BeFalse())
+			}
+		}
+	})
 })