fix: don't drift when launched into open ODCR (#7867)

Author: jmdeal

pkg/cloudprovider/cloudprovider.go

@@ -385,7 +385,12 @@ func (c *CloudProvider) instanceToNodeClaim(i *instance.Instance, instanceType *
 
 	if instanceType != nil {
 		for key, req := range instanceType.Requirements {
-			if req.Len() == 1 {
+			// We only want to add a label based on the instance type requirements if there is a single value for that
+			// requirement. For example, we can't add a label for zone based on this if the requirement is compatible with
+			// three. Capacity reservation IDs are a special case since we don't have a way to represent that the label may or
+			// may not exist. Since this requirement will be present regardless of the capacity type, we can't insert it here.
+			// Otherwise, you may end up with spot and on-demand NodeClaims with a reservation ID label.
+			if req.Len() == 1 && req.Key != cloudprovider.ReservationIDLabel {
 				labels[key] = req.Values()[0]
 			}
 		}

pkg/controllers/nodeclaim/capacityreservation/suite_test.go

@@ -42,6 +42,8 @@ import (
 	. "sigs.k8s.io/karpenter/pkg/test/expectations"
 	"sigs.k8s.io/karpenter/pkg/test/v1alpha1"
 	. "sigs.k8s.io/karpenter/pkg/utils/testing"
+
+	coreoptions "sigs.k8s.io/karpenter/pkg/operator/options"
 )
 
 var ctx context.Context
@@ -59,6 +61,7 @@ func TestAWS(t *testing.T) {
 var _ = BeforeSuite(func() {
 	env = coretest.NewEnvironment(coretest.WithCRDs(apis.CRDs...), coretest.WithCRDs(v1alpha1.CRDs...), coretest.WithFieldIndexers(coretest.NodeProviderIDFieldIndexer(ctx)))
 	ctx = options.ToContext(ctx, test.Options())
+	ctx = coreoptions.ToContext(ctx, coretest.Options(coretest.OptionsFields{FeatureGates: coretest.FeatureGates{ReservedCapacity: lo.ToPtr(true)}}))
 	ctx, stop = context.WithCancel(ctx)
 	awsEnv = test.NewEnvironment(ctx, env)
 

pkg/controllers/nodeclaim/garbagecollection/suite_test.go

@@ -46,6 +46,8 @@ import (
 	. "github.com/onsi/gomega"
 	. "sigs.k8s.io/karpenter/pkg/test/expectations"
 	. "sigs.k8s.io/karpenter/pkg/utils/testing"
+
+	coreoptions "sigs.k8s.io/karpenter/pkg/operator/options"
 )
 
 var ctx context.Context
@@ -63,6 +65,7 @@ func TestAPIs(t *testing.T) {
 var _ = BeforeSuite(func() {
 	ctx = options.ToContext(ctx, test.Options())
 	env = coretest.NewEnvironment(coretest.WithCRDs(apis.CRDs...), coretest.WithCRDs(v1alpha1.CRDs...))
+	ctx = coreoptions.ToContext(ctx, coretest.Options(coretest.OptionsFields{FeatureGates: coretest.FeatureGates{ReservedCapacity: lo.ToPtr(true)}}))
 	awsEnv = test.NewEnvironment(ctx, env)
 	cloudProvider = cloudprovider.New(awsEnv.InstanceTypesProvider, awsEnv.InstanceProvider, events.NewRecorder(&record.FakeRecorder{}),
 		env.Client, awsEnv.AMIProvider, awsEnv.SecurityGroupProvider, awsEnv.CapacityReservationProvider)

pkg/controllers/nodeclaim/tagging/suite_test.go

@@ -219,7 +219,7 @@ var _ = Describe("TaggingController", func() {
 				v1.EKSClusterNameTagKey: options.FromContext(ctx).ClusterName,
 			}
 			ec2Instance := lo.Must(awsEnv.EC2API.Instances.Load(*ec2Instance.InstanceId)).(ec2types.Instance)
-			instanceTags := instance.NewInstance(ec2Instance).Tags
+			instanceTags := instance.NewInstance(ctx, ec2Instance).Tags
 
 			for tag, value := range expectedTags {
 				if lo.Contains(customTags, tag) {

pkg/providers/instance/instance.go

@@ -164,7 +164,7 @@ func (p *DefaultProvider) Get(ctx context.Context, id string) (*Instance, error)
 	if err != nil {
 		return nil, fmt.Errorf("failed to describe ec2 instances, %w", err)
 	}
-	instances, err := instancesFromOutput(out)
+	instances, err := instancesFromOutput(ctx, out)
 	if err != nil {
 		return nil, fmt.Errorf("getting instances from output, %w", err)
 	}
@@ -202,7 +202,7 @@ func (p *DefaultProvider) List(ctx context.Context) ([]*Instance, error) {
 		}
 		out.Reservations = append(out.Reservations, page.Reservations...)
 	}
-	instances, err := instancesFromOutput(out)
+	instances, err := instancesFromOutput(ctx, out)
 	return instances, cloudprovider.IgnoreNodeClaimNotFoundError(err)
 }
 
@@ -612,7 +612,7 @@ func filterExoticInstanceTypes(instanceTypes []*cloudprovider.InstanceType) []*c
 	return instanceTypes
 }
 
-func instancesFromOutput(out *ec2.DescribeInstancesOutput) ([]*Instance, error) {
+func instancesFromOutput(ctx context.Context, out *ec2.DescribeInstancesOutput) ([]*Instance, error) {
 	if len(out.Reservations) == 0 {
 		return nil, cloudprovider.NewNodeClaimNotFoundError(fmt.Errorf("instance not found"))
 	}
@@ -626,7 +626,7 @@ func instancesFromOutput(out *ec2.DescribeInstancesOutput) ([]*Instance, error)
 	sort.Slice(instances, func(i, j int) bool {
 		return aws.ToString(instances[i].InstanceId) < aws.ToString(instances[j].InstanceId)
 	})
-	return lo.Map(instances, func(i ec2types.Instance, _ int) *Instance { return NewInstance(i) }), nil
+	return lo.Map(instances, func(i ec2types.Instance, _ int) *Instance { return NewInstance(ctx, i) }), nil
 }
 
 func combineFleetErrors(fleetErrs []ec2types.CreateFleetError) (errs error) {

pkg/providers/instance/suite_test.go

@@ -256,6 +256,30 @@ var _ = Describe("InstanceProvider", func() {
 		Expect(createFleetInput.LaunchTemplateConfigs).To(HaveLen(1))
 		Expect(createFleetInput.LaunchTemplateConfigs[0].Overrides).To(HaveLen(1))
 	})
+	It("should treat instances which launched into open ODCRs as on-demand when the ReservedCapacity gate is disabled", func() {
+		id := fake.InstanceID()
+		awsEnv.EC2API.DescribeInstancesBehavior.Output.Set(&ec2.DescribeInstancesOutput{
+			Reservations: []ec2types.Reservation{{
+				Instances: []ec2types.Instance{{
+					State:                 &ec2types.InstanceState{Name: ec2types.InstanceStateNameRunning},
+					PrivateDnsName:        lo.ToPtr(fake.PrivateDNSName()),
+					Placement:             &ec2types.Placement{AvailabilityZone: lo.ToPtr(fake.DefaultRegion)},
+					LaunchTime:            lo.ToPtr(time.Now().Add(-time.Minute)),
+					InstanceId:            &id,
+					InstanceType:          "m5.large",
+					CapacityReservationId: lo.ToPtr("cr-foo"),
+				}},
+			}},
+		})
+
+		coreoptions.FromContext(ctx).FeatureGates.ReservedCapacity = false
+		nodeClaims, err := cloudProvider.List(ctx)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(nodeClaims).To(HaveLen(1))
+		Expect(nodeClaims[0].Status.ProviderID).To(ContainSubstring(id))
+		Expect(nodeClaims[0].Labels).To(HaveKeyWithValue(karpv1.CapacityTypeLabelKey, karpv1.CapacityTypeOnDemand))
+		Expect(nodeClaims[0].Labels).ToNot(HaveKey(v1.LabelCapacityReservationID))
+	})
 	It("should return all NodePool-owned instances from List", func() {
 		ids := sets.New[string]()
 		// Provision instances that have the karpenter.sh/nodepool key

pkg/providers/instance/types.go

@@ -15,12 +15,14 @@ limitations under the License.
 package instance
 
 import (
+	"context"
 	"time"
 
 	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
 	"github.com/samber/lo"
 
 	karpv1 "sigs.k8s.io/karpenter/pkg/apis/v1"
+	"sigs.k8s.io/karpenter/pkg/operator/options"
 )
 
 // Instance is an internal data representation of either an ec2.Instance or an ec2.FleetInstance
@@ -40,18 +42,25 @@ type Instance struct {
 	EFAEnabled            bool
 }
 
-func NewInstance(out ec2types.Instance) *Instance {
+func NewInstance(ctx context.Context, out ec2types.Instance) *Instance {
 	return &Instance{
 		LaunchTime: lo.FromPtr(out.LaunchTime),
 		State:      out.State.Name,
 		ID:         lo.FromPtr(out.InstanceId),
 		ImageID:    lo.FromPtr(out.ImageId),
 		Type:       out.InstanceType,
 		Zone:       lo.FromPtr(out.Placement.AvailabilityZone),
+		// NOTE: Only set the capacity type to reserved and assign a reservation ID if the feature gate is enabled. It's
+		// possible for these to be set if the instance launched into an open ODCR, but treating it as reserved would induce
+		// drift.
 		CapacityType: lo.If(out.SpotInstanceRequestId != nil, karpv1.CapacityTypeSpot).
-			ElseIf(out.CapacityReservationId != nil, karpv1.CapacityTypeReserved).
+			ElseIf(out.CapacityReservationId != nil && options.FromContext(ctx).FeatureGates.ReservedCapacity, karpv1.CapacityTypeReserved).
 			Else(karpv1.CapacityTypeOnDemand),
-		CapacityReservationID: lo.FromPtr(out.CapacityReservationId),
+		CapacityReservationID: lo.Ternary(
+			options.FromContext(ctx).FeatureGates.ReservedCapacity,
+			lo.FromPtr(out.CapacityReservationId),
+			"",
+		),
 		SecurityGroupIDs: lo.Map(out.SecurityGroups, func(securitygroup ec2types.GroupIdentifier, _ int) string {
 			return lo.FromPtr(securitygroup.GroupId)
 		}),

test/suites/integration/tags_test.go

@@ -29,6 +29,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	karpv1 "sigs.k8s.io/karpenter/pkg/apis/v1"
+	"sigs.k8s.io/karpenter/pkg/operator/options"
 	coretest "sigs.k8s.io/karpenter/pkg/test"
 
 	v1 "github.com/aws/karpenter-provider-aws/pkg/apis/v1"
@@ -110,7 +111,8 @@ var _ = Describe("Tags", func() {
 				g.Expect(nodeClaim.Annotations).To(HaveKeyWithValue(v1.AnnotationClusterNameTaggedCompatability, "true"))
 			}, time.Minute).Should(Succeed())
 
-			nodeInstance := instance.NewInstance(env.GetInstance(node.Name))
+			ctx := options.ToContext(env.Context, &options.Options{})
+			nodeInstance := instance.NewInstance(ctx, env.GetInstance(node.Name))
 			Expect(nodeInstance.Tags).To(HaveKeyWithValue("Name", node.Name))
 			Expect(nodeInstance.Tags).To(HaveKeyWithValue("karpenter.sh/nodeclaim", nodeClaim.Name))
 			Expect(nodeInstance.Tags).To(HaveKeyWithValue("eks:eks-cluster-name", env.ClusterName))
@@ -147,7 +149,8 @@ var _ = Describe("Tags", func() {
 				g.Expect(nodeClaim.Annotations).To(HaveKeyWithValue(v1.AnnotationClusterNameTaggedCompatability, "true"))
 			}, time.Minute).Should(Succeed())
 
-			nodeInstance := instance.NewInstance(env.GetInstance(node.Name))
+			ctx := options.ToContext(env.Context, &options.Options{})
+			nodeInstance := instance.NewInstance(ctx, env.GetInstance(node.Name))
 			Expect(nodeInstance.Tags).To(HaveKeyWithValue("Name", "custom-name"))
 			Expect(nodeInstance.Tags).To(HaveKeyWithValue("karpenter.sh/nodeclaim", nodeClaim.Name))
 			Expect(nodeInstance.Tags).To(HaveKeyWithValue("eks:eks-cluster-name", env.ClusterName))