Replace DescribeNetworkInterfaces with paginated version

pkg/awsutils/awsutils.go

@@ -66,6 +66,9 @@ const (
 
 	// Http client timeout env for sessions
 	httpTimeoutEnv = "HTTP_TIMEOUT"
+
+	// the default page size when paginating the DescribeNetworkInterfaces call
+	describeENIPageSize = 1000
 )
 
 var (
@@ -336,7 +339,7 @@ func New(useCustomNetworking bool) (*EC2InstanceMetadataCache, error) {
 					Timeout: httpTimeoutValue,
 				},
 			},
-	))
+		))
 	ec2Metadata := ec2metadata.New(awsSession)
 
 	cache := &EC2InstanceMetadataCache{}
@@ -355,7 +358,7 @@ func New(useCustomNetworking bool) (*EC2InstanceMetadataCache, error) {
 
 	sess, err := session.NewSession(
 		&aws.Config{
-			Region: aws.String(cache.region), 
+			Region:     aws.String(cache.region),
 			MaxRetries: aws.Int(15),
 			HTTPClient: &http.Client{
 				Timeout: httpTimeoutValue,
@@ -1424,18 +1427,15 @@ func (cache *EC2InstanceMetadataCache) getFilteredListOfNetworkInterfaces() ([]*
 	}
 
 	input := &ec2.DescribeNetworkInterfacesInput{
-		Filters: []*ec2.Filter{tagFilter, statusFilter},
-	}
-	result, err := cache.ec2SVC.DescribeNetworkInterfacesWithContext(context.Background(), input, userAgent)
-	if err != nil {
-		return nil, errors.Wrap(err, "awsutils: unable to obtain filtered list of network interfaces")
+		Filters:    []*ec2.Filter{tagFilter, statusFilter},
+		MaxResults: aws.Int64(describeENIPageSize),
 	}
 
-	networkInterfaces := make([]*ec2.NetworkInterface, 0)
-	for _, networkInterface := range result.NetworkInterfaces {
+	var networkInterfaces []*ec2.NetworkInterface
+	filterFn := func(networkInterface *ec2.NetworkInterface) error {
 		// Verify the description starts with "aws-K8S-"
 		if !strings.HasPrefix(aws.StringValue(networkInterface.Description), eniDescriptionPrefix) {
-			continue
+			return nil
 		}
 		// Check that it's not a newly created ENI
 		tags := getTags(networkInterface, aws.StringValue(networkInterface.NetworkInterfaceId))
@@ -1448,17 +1448,24 @@ func (cache *EC2InstanceMetadataCache) getFilteredListOfNetworkInterfaces() ([]*
 			}
 			if time.Since(parsedTime) < eniDeleteCooldownTime {
 				log.Infof("Found an ENI created less than 5 minutes ago, so not cleaning it up")
-				continue
+				return nil
 			}
 			log.Debugf("%v", value)
 		} else {
 			/* Set a time if we didn't find one. This is to prevent accidentally deleting ENIs that are in the
 			 * process of being attached by CNI versions v1.5.x or earlier.
 			 */
 			cache.tagENIcreateTS(aws.StringValue(networkInterface.NetworkInterfaceId), maxENIBackoffDelay)
-			continue
+			return nil
 		}
 		networkInterfaces = append(networkInterfaces, networkInterface)
+		return nil
+	}
+
+	err := cache.getENIsFromPaginatedDescribeNetworkInterfaces(input, filterFn)
+
+	if err != nil {
+		return nil, errors.Wrap(err, "awsutils: unable to obtain filtered list of network interfaces")
 	}
 
 	if len(networkInterfaces) < 1 {
@@ -1515,3 +1522,26 @@ func (cache *EC2InstanceMetadataCache) IsUnmanagedENI(eniID string) bool {
 	}
 	return false
 }
+
+func (cache *EC2InstanceMetadataCache) getENIsFromPaginatedDescribeNetworkInterfaces(
+	input *ec2.DescribeNetworkInterfacesInput, filterFn func(networkInterface *ec2.NetworkInterface) error) error {
+	pageNum := 0
+	var innerErr error
+	pageFn := func(output *ec2.DescribeNetworkInterfacesOutput, lastPage bool) (nextPage bool) {
+		pageNum++
+		log.Debugf("EC2 DescribeNetworkInterfaces succeeded with %d results on page %d",
+			len(output.NetworkInterfaces), pageNum)
+		for _, eni := range output.NetworkInterfaces {
+			if err := filterFn(eni); err != nil {
+				innerErr = err
+				return false
+			}
+		}
+		return true
+	}
+
+	if err := cache.ec2SVC.DescribeNetworkInterfacesPagesWithContext(context.TODO(), input, pageFn, userAgent); err != nil {
+		return err
+	}
+	return innerErr
+}

pkg/awsutils/awsutils_test.go

@@ -24,6 +24,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -711,10 +712,8 @@ func TestEC2InstanceMetadataCache_getFilteredListOfNetworkInterfaces_OneResult(t
 	attachment := &ec2.NetworkInterfaceAttachment{AttachmentId: &attachmentID}
 	cureniID := eniID
 
-	result := &ec2.DescribeNetworkInterfacesOutput{
-		NetworkInterfaces: []*ec2.NetworkInterface{{Attachment: attachment, Status: &status, TagSet: tag, Description: &description, NetworkInterfaceId: &cureniID}}}
-	mockEC2.EXPECT().DescribeNetworkInterfacesWithContext(gomock.Any(), gomock.Any(), gomock.Any()).Return(result, nil)
-
+	interfaces := []*ec2.NetworkInterface{{Attachment: attachment, Status: &status, TagSet: tag, Description: &description, NetworkInterfaceId: &cureniID}}
+	setupDescribeNetworkInterfacesPagesWithContextMock(t, mockEC2, interfaces, nil, 1)
 	ins := &EC2InstanceMetadataCache{ec2SVC: mockEC2}
 	got, err := ins.getFilteredListOfNetworkInterfaces()
 	assert.NotNil(t, got)
@@ -725,10 +724,7 @@ func TestEC2InstanceMetadataCache_getFilteredListOfNetworkInterfaces_NoResult(t
 	ctrl, mockEC2 := setup(t)
 	defer ctrl.Finish()
 
-	result := &ec2.DescribeNetworkInterfacesOutput{
-		NetworkInterfaces: []*ec2.NetworkInterface{}}
-	mockEC2.EXPECT().DescribeNetworkInterfacesWithContext(gomock.Any(), gomock.Any(), gomock.Any()).Return(result, nil)
-
+	setupDescribeNetworkInterfacesPagesWithContextMock(t, mockEC2, []*ec2.NetworkInterface{}, nil, 1)
 	ins := &EC2InstanceMetadataCache{ec2SVC: mockEC2}
 	got, err := ins.getFilteredListOfNetworkInterfaces()
 	assert.Nil(t, got)
@@ -739,7 +735,12 @@ func TestEC2InstanceMetadataCache_getFilteredListOfNetworkInterfaces_Error(t *te
 	ctrl, mockEC2 := setup(t)
 	defer ctrl.Finish()
 
-	mockEC2.EXPECT().DescribeNetworkInterfacesWithContext(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, errors.New("dummy error"))
+	interfaces := []*ec2.NetworkInterface{{
+		TagSet: []*ec2.Tag{
+			{Key: aws.String("foo"), Value: aws.String("foo-value")},
+		},
+	}}
+	setupDescribeNetworkInterfacesPagesWithContextMock(t, mockEC2, interfaces, errors.New("dummy error"), 1)
 
 	ins := &EC2InstanceMetadataCache{ec2SVC: mockEC2}
 	got, err := ins.getFilteredListOfNetworkInterfaces()
@@ -857,19 +858,30 @@ func TestEC2InstanceMetadataCache_cleanUpLeakedENIsInternal(t *testing.T) {
 	defer ctrl.Finish()
 
 	description := eniDescriptionPrefix + "test"
-	result := &ec2.DescribeNetworkInterfacesOutput{
-		NetworkInterfaces: []*ec2.NetworkInterface{{
-			Description: &description,
-			TagSet: []*ec2.Tag{
-				{Key: aws.String(eniNodeTagKey), Value: aws.String("test-value")},
-			},
-		}},
-	}
+	interfaces := []*ec2.NetworkInterface{{
+		Description: &description,
+		TagSet: []*ec2.Tag{
+			{Key: aws.String(eniNodeTagKey), Value: aws.String("test-value")},
+		},
+	}}
 
-	mockEC2.EXPECT().DescribeNetworkInterfacesWithContext(gomock.Any(), gomock.Any(), gomock.Any()).Return(result, nil)
+	setupDescribeNetworkInterfacesPagesWithContextMock(t, mockEC2, interfaces, nil, 1)
 	mockEC2.EXPECT().CreateTagsWithContext(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
 
 	ins := &EC2InstanceMetadataCache{ec2SVC: mockEC2}
 	// Test checks that both mocks gets called.
 	ins.cleanUpLeakedENIsInternal(time.Millisecond)
 }
+
+func setupDescribeNetworkInterfacesPagesWithContextMock(
+	t *testing.T, mockEC2 *mock_ec2wrapper.MockEC2, interfaces []*ec2.NetworkInterface, err error, times int) {
+	mockEC2.EXPECT().
+		DescribeNetworkInterfacesPagesWithContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(times).
+		DoAndReturn(func(_ context.Context, _ *ec2.DescribeNetworkInterfacesInput,
+			fn func(*ec2.DescribeNetworkInterfacesOutput, bool) bool, userAgent request.Option) error {
+			assert.Equal(t, true, fn(&ec2.DescribeNetworkInterfacesOutput{
+				NetworkInterfaces: interfaces,
+			}, true))
+			return err
+		})
+}

pkg/ec2wrapper/client.go

@@ -33,6 +33,7 @@ type EC2 interface {
 	DescribeNetworkInterfacesWithContext(ctx aws.Context, input *ec2svc.DescribeNetworkInterfacesInput, opts ...request.Option) (*ec2svc.DescribeNetworkInterfacesOutput, error)
 	ModifyNetworkInterfaceAttributeWithContext(ctx aws.Context, input *ec2svc.ModifyNetworkInterfaceAttributeInput, opts ...request.Option) (*ec2svc.ModifyNetworkInterfaceAttributeOutput, error)
 	CreateTagsWithContext(ctx aws.Context, input *ec2svc.CreateTagsInput, opts ...request.Option) (*ec2svc.CreateTagsOutput, error)
+	DescribeNetworkInterfacesPagesWithContext(ctx aws.Context, input *ec2svc.DescribeNetworkInterfacesInput, fn func(*ec2svc.DescribeNetworkInterfacesOutput, bool) bool, opts ...request.Option) error
 }
 
 // New creates a new EC2 wrapper

pkg/ec2wrapper/ec2wrapper.go

@@ -2,6 +2,11 @@
 package ec2wrapper
 
 import (
+	"net/http"
+	"os"
+	"strconv"
+	"time"
+
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/ec2metadatawrapper"
 	"github.com/aws/amazon-vpc-cni-k8s/pkg/utils/logger"
 	"github.com/aws/aws-sdk-go/aws"
@@ -10,10 +15,6 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/ec2/ec2iface"
 	"github.com/pkg/errors"
-	"net/http"
-	"os"
-	"strconv"
-	"time"
 )
 
 const (

pkg/publisher/publisher.go

@@ -115,7 +115,7 @@ func New(ctx context.Context) (Publisher, error) {
 			MaxRetries: aws.Int(15),
 			HTTPClient: &http.Client{
 				Timeout: httpTimeoutValue,
-				},
+			},
 		},
 	))