diff --git a/Makefile b/Makefile index 0ea4e9247aa9..aff6eb3f8d09 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,5 @@ CLUSTER_NAME ?= $(shell kubectl config view --minify -o jsonpath='{.clusters[].name}' | rev | cut -d"/" -f1 | rev | cut -d"." -f1) +EKS_CONTROL_PLANE ?= false ## Inject the app version into operator.Version LDFLAGS ?= -ldflags=-X=sigs.k8s.io/karpenter/pkg/operator.Version=$(shell git describe --tags --always | cut -d"v" -f2) diff --git a/pkg/fake/eksapi.go b/pkg/fake/eksapi.go index f3e1aff9452d..e052ec196968 100644 --- a/pkg/fake/eksapi.go +++ b/pkg/fake/eksapi.go @@ -54,7 +54,7 @@ func (s *EKSAPI) DescribeCluster(_ context.Context, input *eks.DescribeClusterIn KubernetesNetworkConfig: &ekstypes.KubernetesNetworkConfigResponse{ ServiceIpv4Cidr: lo.ToPtr("10.100.0.0/16"), }, - Version: lo.ToPtr("1.30"), + Version: lo.ToPtr("1.29"), }, }, nil }) diff --git a/pkg/providers/version/suite_test.go b/pkg/providers/version/suite_test.go index 66ba552d2f3f..d4f80c5da81a 100644 --- a/pkg/providers/version/suite_test.go +++ b/pkg/providers/version/suite_test.go @@ -16,6 +16,10 @@ package version_test import ( "context" + "fmt" + "net/http" + "os" + "strings" "testing" "sigs.k8s.io/karpenter/pkg/test/v1alpha1" @@ -23,11 +27,18 @@ import ( coreoptions "sigs.k8s.io/karpenter/pkg/operator/options" coretest "sigs.k8s.io/karpenter/pkg/test" + "github.com/patrickmn/go-cache" + "github.com/aws/karpenter-provider-aws/pkg/apis" + awscache "github.com/aws/karpenter-provider-aws/pkg/cache" "github.com/aws/karpenter-provider-aws/pkg/fake" "github.com/aws/karpenter-provider-aws/pkg/operator/options" + "github.com/aws/karpenter-provider-aws/pkg/providers/version" "github.com/aws/karpenter-provider-aws/pkg/test" + awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http" + smithyhttp "github.com/aws/smithy-go/transport/http" + . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" . "sigs.k8s.io/karpenter/pkg/test/expectations" @@ -39,6 +50,8 @@ var stop context.CancelFunc var env *coretest.Environment var awsEnv *test.Environment var fakeEKSAPI *fake.EKSAPI +var k8sVersion string +var versionProvider *version.DefaultProvider func TestAWS(t *testing.T) { ctx = TestContextWithLogger(t) @@ -53,6 +66,10 @@ var _ = BeforeSuite(func() { ctx, stop = context.WithCancel(ctx) awsEnv = test.NewEnvironment(ctx, env) + serverVersion, err := env.KubernetesInterface.Discovery().ServerVersion() + Expect(err).ToNot(HaveOccurred()) + k8sVersion = fmt.Sprintf("%s.%s", serverVersion.Major, strings.TrimSuffix(serverVersion.Minor, "+")) + fakeEKSAPI = &fake.EKSAPI{} }) @@ -70,9 +87,61 @@ var _ = AfterEach(func() { }) var _ = Describe("Operator", func() { - It("should resolve Kubernetes Version via Describe Cluster", func() { - endpoint, err := awsEnv.VersionProvider.Get(ctx) - Expect(err).ToNot(HaveOccurred()) - Expect(endpoint).To(Equal("1.30")) + + Context("with EKS_CONTROL_PLANE=true", func() { + BeforeEach(func() { + versionProvider = version.NewDefaultProvider(env.KubernetesInterface, cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval), fakeEKSAPI) + os.Setenv("EKS_CONTROL_PLANE", "true") + }) + + It("should resolve Kubernetes Version via Describe Cluster with no errors", func() { + endpoint, err := versionProvider.Get(ctx) + Expect(err).ToNot(HaveOccurred()) + Expect(endpoint).To(Equal("1.29")) + }) + + It("should handle EKS API errors and fallback to K8s API", func() { + fakeEKSAPI.DescribeClusterBehavior.Error.Set(fmt.Errorf("some error")) + _, err := versionProvider.Get(ctx) + Expect(err).To(HaveOccurred()) + }) + + It("should return error for access-denied EKS API errors", func() { + accessDeniedErr := &awshttp.ResponseError{ + ResponseError: &smithyhttp.ResponseError{ + Response: &smithyhttp.Response{ + Response: &http.Response{ + StatusCode: 403, + }, + }, + Err: fmt.Errorf("User is not authorized to perform this operation"), + }, + } + + fakeEKSAPI.DescribeClusterBehavior.Error.Set(accessDeniedErr) + endpoint, err := versionProvider.Get(ctx) + Expect(err).ToNot(HaveOccurred()) + Expect(endpoint).To(Equal(k8sVersion)) + }) + }) + + Context("with EKS_CONTROL_PLANE=false", func() { + It("should resolve Kubernetes Version via K8s API", func() { + versionProvider = version.NewDefaultProvider(env.KubernetesInterface, cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval), fakeEKSAPI) + os.Setenv("EKS_CONTROL_PLANE", "false") + endpoint, err := versionProvider.Get(ctx) + Expect(err).ToNot(HaveOccurred()) + Expect(endpoint).To(Equal(k8sVersion)) + }) + }) + + Context("with EKS_CONTROL_PLANE not set", func() { + It("should resolve Kubernetes Version via K8s API", func() { + versionProvider = version.NewDefaultProvider(env.KubernetesInterface, cache.New(awscache.DefaultTTL, awscache.DefaultCleanupInterval), fakeEKSAPI) + os.Unsetenv("EKS_CONTROL_PLANE") + endpoint, err := versionProvider.Get(ctx) + Expect(err).ToNot(HaveOccurred()) + Expect(endpoint).To(Equal(k8sVersion)) + }) }) }) diff --git a/pkg/providers/version/version.go b/pkg/providers/version/version.go index c9e124169632..e14d3b5626b6 100644 --- a/pkg/providers/version/version.go +++ b/pkg/providers/version/version.go @@ -17,6 +17,7 @@ package version import ( "context" "fmt" + "os" "strconv" "strings" @@ -69,31 +70,26 @@ func NewDefaultProvider(kubernetesInterface kubernetes.Interface, cache *cache.C } func (p *DefaultProvider) Get(ctx context.Context) (string, error) { - var version string - var versionSource string if version, ok := p.cache.Get(kubernetesVersionCacheKey); ok { return version.(string), nil } - output, err := p.eksapi.DescribeCluster(ctx, &eks.DescribeClusterInput{ - Name: lo.ToPtr(options.FromContext(ctx).ClusterName), - }) - if err != nil { - if !awserrors.IsAccessDenied(err) { - return "", err - } - output, err := p.kubernetesInterface.Discovery().ServerVersion() + + var version, versionSource string + var err error + + if os.Getenv("EKS_CONTROL_PLANE") == "true" { + version, versionSource, err = p.getEKSVersion(ctx) if err != nil { - return "", fmt.Errorf("getting kubernetes version from the kubernetes API") - } else if output != nil { - version = fmt.Sprintf("%s.%s", output.Major, strings.TrimSuffix(output.Minor, "+")) - versionSource = "Kubernetes API" + return "", err } - } else if lo.FromPtr(output.Cluster.Version) != "" { - version = *output.Cluster.Version - versionSource = "EKS DescribeCluster" - } else { - return "", fmt.Errorf("unable to retrieve Kubernetes version from EKS DescribeCluster") } + if version == "" { + version, versionSource, err = p.getK8sVersion() + } + if err != nil { + return "", err + } + p.cache.SetDefault(kubernetesVersionCacheKey, version) if p.cm.HasChanged("kubernetes-version", version) || p.cm.HasChanged("version-source", versionSource) { log.FromContext(ctx).WithValues("version", version, "source", versionSource).V(1).Info("discovered kubernetes version") @@ -129,3 +125,24 @@ func validateK8sVersion(v string) error { return nil } + +func (p *DefaultProvider) getEKSVersion(ctx context.Context) (string, string, error) { + output, err := p.eksapi.DescribeCluster(ctx, &eks.DescribeClusterInput{ + Name: lo.ToPtr(options.FromContext(ctx).ClusterName), + }) + if err == nil && lo.FromPtr(output.Cluster.Version) != "" { + return *output.Cluster.Version, "EKS DescribeCluster", err + } + if err != nil && !awserrors.IsAccessDenied(err) { + return "", "", err + } + return "", "", nil +} + +func (p *DefaultProvider) getK8sVersion() (string, string, error) { + output, err := p.kubernetesInterface.Discovery().ServerVersion() + if err != nil || output == nil { + return "", "", fmt.Errorf("getting kubernetes version from the kubernetes API") + } + return fmt.Sprintf("%s.%s", output.Major, strings.TrimSuffix(output.Minor, "+")), "Kubernetes API", err +}