feat: invalidate SSM cache upon AMI deprecation (#7301)

Co-authored-by: Jonathan Innis <[email protected]>

cmd/controller/main.go

@@ -53,6 +53,7 @@ func main() {
 			op.GetClient(),
 			op.EventRecorder,
 			op.UnavailableOfferingsCache,
+			op.SSMCache,
 			cloudProvider,
 			op.SubnetProvider,
 			op.SecurityGroupProvider,

pkg/apis/v1/ec2nodeclass.go

@@ -484,16 +484,40 @@ func (in *EC2NodeClass) AMIFamily() string {
 	if in.Spec.AMIFamily != nil {
 		return *in.Spec.AMIFamily
 	}
-	if term, ok := lo.Find(in.Spec.AMISelectorTerms, func(t AMISelectorTerm) bool {
-		return t.Alias != ""
-	}); ok {
-		return AMIFamilyFromAlias(term.Alias)
+	if alias := in.Alias(); alias != nil {
+		return alias.Family
 	}
 	// Unreachable: validation enforces that one of the above conditions must be met
 	return AMIFamilyCustom
 }
 
-func AMIFamilyFromAlias(alias string) string {
+type Alias struct {
+	Family  string
+	Version string
+}
+
+const (
+	AliasVersionLatest = "latest"
+)
+
+func (a *Alias) String() string {
+	return fmt.Sprintf("%s@%s", a.Family, a.Version)
+}
+
+func (in *EC2NodeClass) Alias() *Alias {
+	term, ok := lo.Find(in.Spec.AMISelectorTerms, func(term AMISelectorTerm) bool {
+		return term.Alias != ""
+	})
+	if !ok {
+		return nil
+	}
+	return &Alias{
+		Family:  amiFamilyFromAlias(term.Alias),
+		Version: amiVersionFromAlias(term.Alias),
+	}
+}
+
+func amiFamilyFromAlias(alias string) string {
 	components := strings.Split(alias, "@")
 	if len(components) != 2 {
 		log.Fatalf("failed to parse AMI alias %q, invalid format", alias)
@@ -513,7 +537,7 @@ func AMIFamilyFromAlias(alias string) string {
 	return family
 }
 
-func AMIVersionFromAlias(alias string) string {
+func amiVersionFromAlias(alias string) string {
 	components := strings.Split(alias, "@")
 	if len(components) != 2 {
 		log.Fatalf("failed to parse AMI alias %q, invalid format", alias)

pkg/cache/cache.go

@@ -37,7 +37,7 @@ const (
 	AssociatePublicIPAddressTTL = 5 * time.Minute
 	// SSMGetParametersByPathTTL is the time to drop SSM Parameters by path data. This only queries EKS Optimized AMI
 	// releases, so we should expect this to be updated relatively infrequently.
-	SSMGetParametersByPathTTL = 24 * time.Hour
+	SSMCacheTTL = 24 * time.Hour
 	// DiscoveredCapacityCacheTTL is the time to drop discovered resource capacity data per-instance type
 	// if it is not updated by a node creation event or refreshed during controller reconciliation
 	DiscoveredCapacityCacheTTL = 60 * 24 * time.Hour

pkg/controllers/controllers.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/awslabs/operatorpkg/controller"
 	"github.com/awslabs/operatorpkg/status"
+	"github.com/patrickmn/go-cache"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/karpenter/pkg/cloudprovider"
 
@@ -29,6 +30,7 @@ import (
 	controllersinstancetype "github.com/aws/karpenter-provider-aws/pkg/controllers/providers/instancetype"
 	controllersinstancetypecapacity "github.com/aws/karpenter-provider-aws/pkg/controllers/providers/instancetype/capacity"
 	controllerspricing "github.com/aws/karpenter-provider-aws/pkg/controllers/providers/pricing"
+	ssminvalidation "github.com/aws/karpenter-provider-aws/pkg/controllers/providers/ssm/invalidation"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/launchtemplate"
 
 	servicesqs "github.com/aws/aws-sdk-go-v2/service/sqs"
@@ -39,7 +41,7 @@ import (
 
 	"sigs.k8s.io/karpenter/pkg/events"
 
-	"github.com/aws/karpenter-provider-aws/pkg/cache"
+	awscache "github.com/aws/karpenter-provider-aws/pkg/cache"
 	"github.com/aws/karpenter-provider-aws/pkg/controllers/interruption"
 	nodeclaimgarbagecollection "github.com/aws/karpenter-provider-aws/pkg/controllers/nodeclaim/garbagecollection"
 	nodeclaimtagging "github.com/aws/karpenter-provider-aws/pkg/controllers/nodeclaim/tagging"
@@ -56,11 +58,25 @@ import (
 	config "github.com/aws/aws-sdk-go-v2/config"
 )
 
-func NewControllers(ctx context.Context, mgr manager.Manager, sess *session.Session, clk clock.Clock, kubeClient client.Client, recorder events.Recorder,
-	unavailableOfferings *cache.UnavailableOfferings, cloudProvider cloudprovider.CloudProvider, subnetProvider subnet.Provider,
-	securityGroupProvider securitygroup.Provider, instanceProfileProvider instanceprofile.Provider, instanceProvider instance.Provider,
-	pricingProvider pricing.Provider, amiProvider amifamily.Provider, launchTemplateProvider launchtemplate.Provider, instanceTypeProvider *instancetype.DefaultProvider) []controller.Controller {
-
+func NewControllers(
+	ctx context.Context,
+	mgr manager.Manager,
+	sess *session.Session,
+	clk clock.Clock,
+	kubeClient client.Client,
+	recorder events.Recorder,
+	unavailableOfferings *awscache.UnavailableOfferings,
+	ssmCache *cache.Cache,
+	cloudProvider cloudprovider.CloudProvider,
+	subnetProvider subnet.Provider,
+	securityGroupProvider securitygroup.Provider,
+	instanceProfileProvider instanceprofile.Provider,
+	instanceProvider instance.Provider,
+	pricingProvider pricing.Provider,
+	amiProvider amifamily.Provider,
+	launchTemplateProvider launchtemplate.Provider,
+	instanceTypeProvider *instancetype.DefaultProvider,
+) []controller.Controller {
 	controllers := []controller.Controller{
 		nodeclasshash.NewController(kubeClient),
 		nodeclassstatus.NewController(kubeClient, subnetProvider, securityGroupProvider, amiProvider, instanceProfileProvider, launchTemplateProvider),
@@ -70,6 +86,7 @@ func NewControllers(ctx context.Context, mgr manager.Manager, sess *session.Sess
 		controllerspricing.NewController(pricingProvider),
 		controllersinstancetype.NewController(instanceTypeProvider),
 		controllersinstancetypecapacity.NewController(kubeClient, instanceTypeProvider),
+		ssminvalidation.NewController(ssmCache, amiProvider),
 		status.NewController[*v1.EC2NodeClass](kubeClient, mgr.GetEventRecorderFor("karpenter")),
 	}
 	if options.FromContext(ctx).InterruptionQueue != "" {

pkg/controllers/providers/ssm/invalidation/controller.go

@@ -0,0 +1,95 @@
+/*
+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 invalidation
+
+import (
+	"context"
+	"time"
+
+	"github.com/awslabs/operatorpkg/singleton"
+	"github.com/patrickmn/go-cache"
+	"github.com/samber/lo"
+	controllerruntime "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/karpenter/pkg/operator/injection"
+
+	v1 "github.com/aws/karpenter-provider-aws/pkg/apis/v1"
+	"github.com/aws/karpenter-provider-aws/pkg/providers/amifamily"
+	"github.com/aws/karpenter-provider-aws/pkg/providers/ssm"
+)
+
+// The SSM Invalidation controller is responsible for invalidating "latest" SSM parameters when they point to deprecated
+// AMIs. This can occur when an EKS-optimized AMI with a regression is released, and the AMI team chooses to deprecate
+// the AMI. Normally, SSM parameter cache entries expire after 24 hours to prevent a thundering herd upon a new AMI
+// release, however Karpenter should react faster when an AMI is deprecated. This controller will ensure Karpenter
+// reacts to AMI deprecations within it's polling period (30m).
+type Controller struct {
+	cache       *cache.Cache
+	amiProvider amifamily.Provider
+}
+
+func NewController(ssmCache *cache.Cache, amiProvider amifamily.Provider) *Controller {
+	return &Controller{
+		cache:       ssmCache,
+		amiProvider: amiProvider,
+	}
+}
+
+func (c *Controller) Name() string {
+	return "providers.ssm.invalidation"
+}
+
+func (c *Controller) Reconcile(ctx context.Context) (reconcile.Result, error) {
+	ctx = injection.WithControllerName(ctx, c.Name())
+
+	amiIDsToParameters := map[string]ssm.Parameter{}
+	for _, item := range c.cache.Items() {
+		entry := item.Object.(ssm.CacheEntry)
+		if !entry.Parameter.IsMutable {
+			continue
+		}
+		amiIDsToParameters[entry.Value] = entry.Parameter
+	}
+	amis := []amifamily.AMI{}
+	for _, nodeClass := range lo.Map(lo.Keys(amiIDsToParameters), func(amiID string, _ int) *v1.EC2NodeClass {
+		return &v1.EC2NodeClass{
+			Spec: v1.EC2NodeClassSpec{
+				AMISelectorTerms: []v1.AMISelectorTerm{{ID: amiID}},
+			},
+		}
+	}) {
+		resolvedAMIs, err := c.amiProvider.List(ctx, nodeClass)
+		if err != nil {
+			return reconcile.Result{}, err
+		}
+		amis = append(amis, resolvedAMIs...)
+	}
+	for _, ami := range amis {
+		if !ami.Deprecated {
+			continue
+		}
+		parameter := amiIDsToParameters[ami.AmiID]
+		c.cache.Delete(parameter.CacheKey())
+	}
+	return reconcile.Result{RequeueAfter: 30 * time.Minute}, nil
+}
+
+func (c *Controller) Register(_ context.Context, m manager.Manager) error {
+	return controllerruntime.NewControllerManagedBy(m).
+		Named(c.Name()).
+		WatchesRawSource(singleton.Source()).
+		Complete(singleton.AsReconciler(c))
+}