Merge pull request #1041 from mmerkes/top-sdk-upgrade

Migrates InstanceTopologyManager to use sdk v2

go.mod

@@ -35,6 +35,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.21 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.21 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ec2 v1.186.0
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.0 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.2 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.24.2 // indirect

go.sum

@@ -22,6 +22,8 @@ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.21 h1:6jZVETqmYCadGFvrYE
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.21/go.mod h1:1SR0GbLlnN3QUmYaflZNiH1ql+1qrSiB2vwcJ+4UM60=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 h1:VaRN3TlFdd6KxX1x3ILT5ynH6HvKgqdiXoTxAF4HQcQ=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1/go.mod h1:FbtygfRFze9usAadmnGJNc8KsP346kEe+y2/oyhGAGc=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.186.0 h1:n2l2WeV+lEABrGwG/4MsE0WFEbd3j7yKsmZzbnEm5CY=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.186.0/go.mod h1:kYXaB4FzyhEJjvrJ84oPnMElLiEAjGxxUunVW2tBSng=
 github.com/aws/aws-sdk-go-v2/service/ecr v1.36.2 h1:VDQaVwGOokbd3VUbHF+wupiffdrbAZPdQnr5XZMJqrs=
 github.com/aws/aws-sdk-go-v2/service/ecr v1.36.2/go.mod h1:lvUlMghKYmSxSfv0vU7pdU/8jSY+s0zpG8xXhaGKCw0=
 github.com/aws/aws-sdk-go-v2/service/ecrpublic v1.27.2 h1:Zru9Iy2JPM5+uRnFnoqeOZzi8JIVIHJ0ua6JdeDHcyg=

pkg/providers/v1/aws.go

@@ -61,6 +61,8 @@ import (
 	"k8s.io/cloud-provider-aws/pkg/providers/v1/iface"
 	"k8s.io/cloud-provider-aws/pkg/providers/v1/variant"
 	_ "k8s.io/cloud-provider-aws/pkg/providers/v1/variant/fargate" // ensure the fargate variant gets registered
+	"k8s.io/cloud-provider-aws/pkg/resourcemanagers"
+	"k8s.io/cloud-provider-aws/pkg/services"
 )
 
 // NLBHealthCheckRuleDescription is the comment used on a security group rule to
@@ -391,7 +393,7 @@ type Cloud struct {
 
 	instanceCache           instanceCache
 	zoneCache               zoneCache
-	instanceTopologyManager *instanceTopologyManager
+	instanceTopologyManager resourcemanagers.InstanceTopologyManager
 
 	clientBuilder cloudprovider.ControllerClientBuilder
 	kubeClient    clientset.Interface
@@ -587,6 +589,11 @@ func newAWSCloud2(cfg config.CloudConfig, awsServices Services, provider config.
 		return nil, fmt.Errorf("error creating AWS EC2 client: %v", err)
 	}
 
+	ec2v2, err := services.NewEc2SdkV2(regionName)
+	if err != nil {
+		return nil, fmt.Errorf("error creating AWS EC2v2 client: %v", err)
+	}
+
 	elb, err := awsServices.LoadBalancing(regionName)
 	if err != nil {
 		return nil, fmt.Errorf("error creating AWS ELB client: %v", err)
@@ -619,7 +626,7 @@ func newAWSCloud2(cfg config.CloudConfig, awsServices Services, provider config.
 	}
 	awsCloud.instanceCache.cloud = awsCloud
 	awsCloud.zoneCache.cloud = awsCloud
-	awsCloud.instanceTopologyManager = newInstanceTopologyManager(awsCloud.ec2)
+	awsCloud.instanceTopologyManager = resourcemanagers.NewInstanceTopologyManager(ec2v2)
 
 	tagged := cfg.Global.KubernetesClusterTag != "" || cfg.Global.KubernetesClusterID != ""
 	if cfg.Global.VPC != "" && (cfg.Global.SubnetID != "" || cfg.Global.RoleARN != "") && tagged {

pkg/providers/v1/aws_ec2.go

@@ -30,19 +30,6 @@ type awsSdkEC2 struct {
 	ec2 ec2iface.EC2API
 }
 
-func (s *awsSdkEC2) DescribeInstanceTopology(request *ec2.DescribeInstanceTopologyInput) ([]*ec2.InstanceTopology, error) {
-	var topologies []*ec2.InstanceTopology
-
-	err := s.ec2.DescribeInstanceTopologyPages(request,
-		func(page *ec2.DescribeInstanceTopologyOutput, lastPage bool) bool {
-			topologies = append(topologies, page.Instances...)
-			// Don't short-circuit. Just go through all of the pages
-			return false
-		})
-
-	return topologies, err
-}
-
 // Implementation of EC2.Instances
 func (s *awsSdkEC2) DescribeInstances(request *ec2.DescribeInstancesInput) ([]*ec2.Instance, error) {
 	// Instances are paged

pkg/providers/v1/aws_fakes.go

@@ -240,22 +240,6 @@ func (ec2i *FakeEC2Impl) DescribeInstances(request *ec2.DescribeInstancesInput)
 	return matches, nil
 }
 
-// DescribeInstanceTopology returns fake instance descriptions
-func (ec2i *FakeEC2Impl) DescribeInstanceTopology(request *ec2.DescribeInstanceTopologyInput) ([]*ec2.InstanceTopology, error) {
-	return []*ec2.InstanceTopology{
-		{
-			AvailabilityZone: aws.String("us-west-2b"),
-			InstanceId:       aws.String("i-123456789"),
-			NetworkNodes: []*string{
-				aws.String("nn-123456789"),
-				aws.String("nn-234567890"),
-				aws.String("nn-345678901"),
-			},
-			ZoneId: aws.String("az2"),
-		},
-	}, nil
-}
-
 // AttachVolume is not implemented but is required for interface conformance
 func (ec2i *FakeEC2Impl) AttachVolume(request *ec2.AttachVolumeInput) (resp *ec2.VolumeAttachment, err error) {
 	panic("Not implemented")

pkg/providers/v1/aws_test.go

@@ -3725,14 +3725,6 @@ func (m *MockedEC2API) DescribeInstances(input *ec2.DescribeInstancesInput) (*ec
 	return args.Get(0).(*ec2.DescribeInstancesOutput), args.Error(1)
 }
 
-func (m *MockedEC2API) DescribeInstanceTopologyPages(input *ec2.DescribeInstanceTopologyInput, fn func(*ec2.DescribeInstanceTopologyOutput, bool) bool) error {
-	args := m.Called(input)
-	if args.Get(0) != nil {
-		fn(args.Get(0).(*ec2.DescribeInstanceTopologyOutput), true)
-	}
-	return args.Error(1)
-}
-
 func (m *MockedEC2API) DescribeAvailabilityZones(input *ec2.DescribeAvailabilityZonesInput) (*ec2.DescribeAvailabilityZonesOutput, error) {
 	args := m.Called(input)
 	return args.Get(0).(*ec2.DescribeAvailabilityZonesOutput), args.Error(1)

pkg/providers/v1/iface/types.go

@@ -1,14 +1,15 @@
 package iface
 
-import "github.com/aws/aws-sdk-go/service/ec2"
+import (
+	"github.com/aws/aws-sdk-go/service/ec2"
+)
 
 // EC2 is an abstraction over AWS', to allow mocking/other implementations
 // Note that the DescribeX functions return a list, so callers don't need to deal with paging
 // TODO: Should we rename this to AWS (EBS & ELB are not technically part of EC2)
 type EC2 interface {
 	// Query EC2 for instances matching the filter
 	DescribeInstances(request *ec2.DescribeInstancesInput) ([]*ec2.Instance, error)
-	DescribeInstanceTopology(request *ec2.DescribeInstanceTopologyInput) ([]*ec2.InstanceTopology, error)
 
 	DescribeSecurityGroups(request *ec2.DescribeSecurityGroupsInput) ([]*ec2.SecurityGroup, error)
 

pkg/providers/v1/instances_v2.go

@@ -24,7 +24,6 @@ import (
 	"context"
 	"strconv"
 
-	"github.com/aws/aws-sdk-go/aws"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	cloudprovider "k8s.io/cloud-provider"
@@ -65,7 +64,7 @@ func (c *Cloud) InstanceShutdown(ctx context.Context, node *v1.Node) (bool, erro
 	return c.InstanceShutdownByProviderID(ctx, providerID)
 }
 
-func (c *Cloud) getAdditionalLabels(zoneName string, instanceID string, instanceType string,
+func (c *Cloud) getAdditionalLabels(ctx context.Context, zoneName string, instanceID string, instanceType string,
 	region string, existingLabels map[string]string) (map[string]string, error) {
 	additionalLabels := map[string]string{}
 
@@ -82,14 +81,14 @@ func (c *Cloud) getAdditionalLabels(zoneName string, instanceID string, instance
 
 	// If topology labels are already set, skip.
 	if _, ok := existingLabels[LabelNetworkNodePrefix+"1"]; !ok {
-		nodeTopology, err := c.instanceTopologyManager.getNodeTopology(instanceType, region, instanceID)
+		nodeTopology, err := c.instanceTopologyManager.GetNodeTopology(ctx, instanceType, region, instanceID)
 		if err != nil {
 			return nil, err
 		} else if nodeTopology != nil {
 			for index, networkNode := range nodeTopology.NetworkNodes {
 				layer := index + 1
 				label := LabelNetworkNodePrefix + strconv.Itoa(layer)
-				additionalLabels[label] = aws.StringValue(networkNode)
+				additionalLabels[label] = networkNode
 			}
 		}
 	}
@@ -128,7 +127,7 @@ func (c *Cloud) InstanceMetadata(ctx context.Context, node *v1.Node) (*cloudprov
 		return nil, err
 	}
 
-	additionalLabels, err := c.getAdditionalLabels(zone.FailureDomain, string(instanceID), instanceType, zone.Region, node.Labels)
+	additionalLabels, err := c.getAdditionalLabels(ctx, zone.FailureDomain, string(instanceID), instanceType, zone.Region, node.Labels)
 	if err != nil {
 		return nil, err
 	}

pkg/providers/v1/instances_v2_test.go

@@ -21,12 +21,15 @@ import (
 	"fmt"
 	"testing"
 
+	awsv2 "github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/ec2/types"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/cloud-provider-aws/pkg/resourcemanagers"
 )
 
 func TestGetProviderId(t *testing.T) {
@@ -165,6 +168,16 @@ func TestInstanceMetadata(t *testing.T) {
 	t.Run("Should return populated InstanceMetadata", func(t *testing.T) {
 		instance := makeInstance("i-00000000000000000", "192.168.0.1", "1.2.3.4", "instance-same.ec2.internal", "instance-same.ec2.external", nil, true)
 		c, _ := mockInstancesResp(&instance, []*ec2.Instance{&instance})
+		var mockedTopologyManager resourcemanagers.MockedInstanceTopologyManager
+		c.instanceTopologyManager = &mockedTopologyManager
+		mockedTopologyManager.On("GetNodeTopology", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&types.InstanceTopology{
+			AvailabilityZone: awsv2.String("us-west-2b"),
+			GroupName:        new(string),
+			InstanceId:       awsv2.String("i-123456789"),
+			InstanceType:     new(string),
+			NetworkNodes:     []string{"nn-123456789", "nn-234567890", "nn-345678901"},
+			ZoneId:           awsv2.String("az2"),
+		}, nil)
 		node := &v1.Node{
 			Spec: v1.NodeSpec{
 				ProviderID: fmt.Sprintf("aws:///us-west-2c/1abc-2def/%s", *instance.InstanceId),
@@ -176,6 +189,7 @@ func TestInstanceMetadata(t *testing.T) {
 			t.Errorf("Should not error getting InstanceMetadata: %s", err)
 		}
 
+		mockedTopologyManager.AssertNumberOfCalls(t, "GetNodeTopology", 1)
 		assert.Equal(t, "aws:///us-west-2c/1abc-2def/i-00000000000000000", result.ProviderID)
 		assert.Equal(t, "c3.large", result.InstanceType)
 		assert.Equal(t, []v1.NodeAddress{
@@ -198,6 +212,8 @@ func TestInstanceMetadata(t *testing.T) {
 	t.Run("Should skip additional labels if already set", func(t *testing.T) {
 		instance := makeInstance("i-00000000000000000", "192.168.0.1", "1.2.3.4", "instance-same.ec2.internal", "instance-same.ec2.external", nil, true)
 		c, _ := mockInstancesResp(&instance, []*ec2.Instance{&instance})
+		var mockedTopologyManager resourcemanagers.MockedInstanceTopologyManager
+		c.instanceTopologyManager = &mockedTopologyManager
 		node := &v1.Node{
 			Spec: v1.NodeSpec{
 				ProviderID: fmt.Sprintf("aws:///us-west-2c/1abc-2def/%s", *instance.InstanceId),
@@ -216,6 +232,7 @@ func TestInstanceMetadata(t *testing.T) {
 			t.Errorf("Should not error getting InstanceMetadata: %s", err)
 		}
 
+		mockedTopologyManager.AssertNumberOfCalls(t, "GetNodeTopology", 0)
 		// Validate that labels are unchanged.
 		assert.Equal(t, map[string]string{}, result.AdditionalLabels)
 	})