feat: cache SSM calls

pkg/cache/cache.go

@@ -31,6 +31,9 @@ const (
 	InstanceTypesAndZonesTTL = 5 * time.Minute
 	// InstanceProfileTTL is the time before we refresh checking instance profile existence at IAM
 	InstanceProfileTTL = 15 * time.Minute
+	// SSMProviderTTL is the time to drop SSM Provider data. This only queries EKS Optimized AMI
+	// releases, so we should expect this to be updated relatively infrequently.
+	SSMProviderTTL = 24 * time.Hour
 )
 
 const (

pkg/operator/operator.go

@@ -61,6 +61,7 @@ import (
 	"github.com/aws/karpenter-provider-aws/pkg/providers/launchtemplate"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/pricing"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/securitygroup"
+	ssmp "github.com/aws/karpenter-provider-aws/pkg/providers/ssm"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/subnet"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/version"
 )
@@ -87,6 +88,7 @@ type Operator struct {
 	VersionProvider           *version.Provider
 	InstanceTypesProvider     *instancetype.Provider
 	InstanceProvider          *instance.Provider
+	SSMProvider               ssmp.Provider
 }
 
 func NewOperator(ctx context.Context, operator *operator.Operator) (context.Context, *Operator) {
@@ -143,7 +145,8 @@ func NewOperator(ctx context.Context, operator *operator.Operator) (context.Cont
 		*sess.Config.Region,
 	)
 	versionProvider := version.NewProvider(operator.KubernetesInterface, cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval))
-	amiProvider := amifamily.NewProvider(versionProvider, ssm.New(sess), ec2api, cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval))
+	ssmProvider := ssmp.NewDefaultProvider(ssm.New(sess), cache.New(awscache.SSMProviderTTL, awscache.DefaultCleanupInterval))
+	amiProvider := amifamily.NewProvider(versionProvider, ssmProvider, ec2api, cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval))
 	amiResolver := amifamily.New(amiProvider)
 	launchTemplateProvider := launchtemplate.NewProvider(
 		ctx,

pkg/providers/amifamily/ami.go

@@ -25,15 +25,14 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/ec2/ec2iface"
-	"github.com/aws/aws-sdk-go/service/ssm"
-	"github.com/aws/aws-sdk-go/service/ssm/ssmiface"
 	"github.com/mitchellh/hashstructure/v2"
 	"github.com/patrickmn/go-cache"
 	"github.com/samber/lo"
 	v1 "k8s.io/api/core/v1"
 	"knative.dev/pkg/logging"
 
 	"github.com/aws/karpenter-provider-aws/pkg/apis/v1beta1"
+	"github.com/aws/karpenter-provider-aws/pkg/providers/ssm"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/version"
 
 	"sigs.k8s.io/karpenter/pkg/cloudprovider"
@@ -44,10 +43,10 @@ import (
 type Provider struct {
 	sync.Mutex
 	cache           *cache.Cache
-	ssm             ssmiface.SSMAPI
 	ec2api          ec2iface.EC2API
 	cm              *pretty.ChangeMonitor
 	versionProvider *version.Provider
+	ssmProvider     ssm.Provider
 }
 
 type AMI struct {
@@ -98,12 +97,12 @@ func (a AMIs) MapToInstanceTypes(instanceTypes []*cloudprovider.InstanceType) ma
 	return amiIDs
 }
 
-func NewProvider(versionProvider *version.Provider, ssm ssmiface.SSMAPI, ec2api ec2iface.EC2API, cache *cache.Cache) *Provider {
+func NewProvider(versionProvider *version.Provider, ssmProvider ssm.Provider, ec2api ec2iface.EC2API, cache *cache.Cache) *Provider {
 	return &Provider{
 		cache:           cache,
-		ssm:             ssm,
 		ec2api:          ec2api,
 		cm:              pretty.NewChangeMonitor(),
+		ssmProvider: ssmProvider,
 		versionProvider: versionProvider,
 	}
 }
@@ -174,12 +173,11 @@ func (p *Provider) getDefaultAMIs(ctx context.Context, nodeClass *v1beta1.EC2Nod
 }
 
 func (p *Provider) resolveSSMParameter(ctx context.Context, ssmQuery string) (string, error) {
-	output, err := p.ssm.GetParameterWithContext(ctx, &ssm.GetParameterInput{Name: aws.String(ssmQuery)})
+	imageID, err := p.ssmProvider.Get(ctx, ssmQuery)
 	if err != nil {
-		return "", fmt.Errorf("getting ssm parameter %q, %w", ssmQuery, err)
+		return "", err
 	}
-	ami := aws.StringValue(output.Parameter.Value)
-	return ami, nil
+	return imageID, nil
 }
 
 func (p *Provider) getAMIs(ctx context.Context, terms []v1beta1.AMISelectorTerm) (AMIs, error) {

pkg/providers/ssm/provider.go

@@ -0,0 +1,61 @@
+/*
+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 ssm
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/aws/aws-sdk-go/service/ssm"
+	"github.com/aws/aws-sdk-go/service/ssm/ssmiface"
+	"github.com/patrickmn/go-cache"
+	"github.com/samber/lo"
+	"knative.dev/pkg/logging"
+)
+
+type Provider interface {
+	Get(context.Context, string) (string, error)
+}
+
+type DefaultProvider struct {
+	sync.Mutex
+	cache  *cache.Cache
+	ssmapi ssmiface.SSMAPI
+}
+
+func NewDefaultProvider(ssmapi ssmiface.SSMAPI, cache *cache.Cache) *DefaultProvider {
+	return &DefaultProvider{
+		ssmapi: ssmapi,
+		cache:  cache,
+	}
+}
+
+func (p *DefaultProvider) Get(ctx context.Context, parameter string) (string, error) {
+	p.Lock()
+	defer p.Unlock()
+	if result, ok := p.cache.Get(parameter); ok {
+		return result.(string), nil
+	}
+	result, err := p.ssmapi.GetParameterWithContext(ctx, &ssm.GetParameterInput{
+		Name: lo.ToPtr(parameter),
+	})
+	if err != nil {
+		return "", fmt.Errorf("getting ssm parameter %q, %w", parameter, err)
+	}
+	p.cache.SetDefault(parameter, lo.FromPtr(result.Parameter.Value))
+	logging.FromContext(ctx).With("parameter", parameter, "value", lo.FromPtr(result.Parameter.Value)).Info("discovered ssm parameter")
+	return lo.FromPtr(result.Parameter.Value), nil
+}

pkg/test/environment.go

@@ -36,6 +36,7 @@ import (
 	"github.com/aws/karpenter-provider-aws/pkg/providers/launchtemplate"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/pricing"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/securitygroup"
+	"github.com/aws/karpenter-provider-aws/pkg/providers/ssm"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/subnet"
 	"github.com/aws/karpenter-provider-aws/pkg/providers/version"
 
@@ -65,6 +66,7 @@ type Environment struct {
 	SubnetCache               *cache.Cache
 	SecurityGroupCache        *cache.Cache
 	InstanceProfileCache      *cache.Cache
+	SSMProviderCache              *cache.Cache
 
 	// Providers
 	InstanceTypesProvider   *instancetype.Provider
@@ -94,6 +96,7 @@ func NewEnvironment(ctx context.Context, env *coretest.Environment) *Environment
 	subnetCache := cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval)
 	securityGroupCache := cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval)
 	instanceProfileCache := cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval)
+	ssmProviderCache := cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval)
 	fakePricingAPI := &fake.PricingAPI{}
 
 	// Providers
@@ -102,7 +105,8 @@ func NewEnvironment(ctx context.Context, env *coretest.Environment) *Environment
 	securityGroupProvider := securitygroup.NewProvider(ec2api, securityGroupCache)
 	versionProvider := version.NewProvider(env.KubernetesInterface, kubernetesVersionCache)
 	instanceProfileProvider := instanceprofile.NewProvider(fake.DefaultRegion, iamapi, instanceProfileCache)
-	amiProvider := amifamily.NewProvider(versionProvider, ssmapi, ec2api, ec2Cache)
+	ssmProvider := ssm.NewDefaultProvider(ssmapi, ssmProviderCache)
+	amiProvider := amifamily.NewProvider(versionProvider, ssmProvider, ec2api, ec2Cache)
 	amiResolver := amifamily.New(amiProvider)
 	instanceTypesProvider := instancetype.NewProvider(fake.DefaultRegion, instanceTypeCache, ec2api, subnetProvider, unavailableOfferingsCache, pricingProvider)
 	launchTemplateProvider :=
@@ -143,6 +147,7 @@ func NewEnvironment(ctx context.Context, env *coretest.Environment) *Environment
 		SecurityGroupCache:        securityGroupCache,
 		InstanceProfileCache:      instanceProfileCache,
 		UnavailableOfferingsCache: unavailableOfferingsCache,
+		SSMProviderCache:              ssmProviderCache,
 
 		InstanceTypesProvider:   instanceTypesProvider,
 		InstanceProvider:        instanceProvider,
@@ -172,6 +177,7 @@ func (env *Environment) Reset() {
 	env.SubnetCache.Flush()
 	env.SecurityGroupCache.Flush()
 	env.InstanceProfileCache.Flush()
+	env.SSMProviderCache.Flush()
 
 	mfs, err := crmetrics.Registry.Gather()
 	if err != nil {