Merge pull request #45 from DataDog/alberto.mardomingo/describe-instances-pagination

Add pagination to describeInstances in aws

Author: amardomingo

.github/workflows/dd-build.yml

@@ -15,25 +15,9 @@ jobs:
       matrix:
         platform: ["linux/arm64","linux/amd64"]
     steps:
-    - name: Maximize build space
-      uses: easimon/maximize-build-space@master
-      with:
-        root-reserve-mb: 512
-        swap-size-mb: 1024
-        remove-dotnet: 'true'
-        remove-android: 'true'
-        remove-haskell: 'true'
-    - name: FreeSpace
-      run: |
-        echo "Free space:"
-        df -h
     - uses: actions/checkout@v2
       with:
         fetch-depth: 0
-    - name: Set up Go
-      uses: actions/setup-go@v2
-      with:
-        go-version: 1.15
     - name: Build recent
       env:
         KUBE_BUILD_PLATFORMS: ${{ matrix.platform }}
@@ -93,9 +77,9 @@ jobs:
     strategy:
       matrix:
         assets: [
-           "kubernetes-client",
-           "kubernetes-node",
-           "kubernetes-server"
+          "kubernetes-client",
+          "kubernetes-node",
+          "kubernetes-server"
         ]
         platform: ["linux-arm64","linux-amd64"]
     steps:

pkg/cloudprovider/providers/aws/aws.go

@@ -46,7 +46,7 @@ import (
 	"gopkg.in/gcfg.v1"
 	"k8s.io/klog"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -280,6 +280,9 @@ type EC2 interface {
 	// Query EC2 for instances matching the filter
 	DescribeInstances(request *ec2.DescribeInstancesInput) ([]*ec2.Instance, error)
 
+	// Query EC2 for instances matching the filter with pagination
+	DescribeInstancesPages(input *ec2.DescribeInstancesInput, fn func(*ec2.DescribeInstancesOutput, bool) bool) error
+
 	// Attach a volume to an instance
 	AttachVolume(*ec2.AttachVolumeInput) (*ec2.VolumeAttachment, error)
 	// Detach a volume from an instance it is attached to
@@ -902,6 +905,11 @@ func (s *awsSdkEC2) DescribeInstances(request *ec2.DescribeInstancesInput) ([]*e
 	return results, nil
 }
 
+// Implementation of EC2.Instances with pagination
+func (s *awsSdkEC2) DescribeInstancesPages(input *ec2.DescribeInstancesInput, fn func(*ec2.DescribeInstancesOutput, bool) bool) error {
+	return s.ec2.DescribeInstancesPages(input, fn)
+}
+
 // Implements EC2.DescribeSecurityGroups
 func (s *awsSdkEC2) DescribeSecurityGroups(request *ec2.DescribeSecurityGroupsInput) ([]*ec2.SecurityGroup, error) {
 	// Security groups are paged
@@ -4499,21 +4507,29 @@ func (c *Cloud) getInstancesByNodeNames(nodeNames []string, states ...string) ([
 
 // TODO: Move to instanceCache
 func (c *Cloud) describeInstances(filters []*ec2.Filter) ([]*ec2.Instance, error) {
+	// Set the Maxresults to the maximum allowed per the AWS doc
+	// https://docs.aws.amazon.com/sdk-for-go/api/service/ec2/#DescribeInstancesInput
+	maxResults := int64(1000)
 	request := &ec2.DescribeInstancesInput{
-		Filters: filters,
+		Filters:    filters,
+		MaxResults: &maxResults,
 	}
 
-	response, err := c.ec2.DescribeInstances(request)
+	var matches []*ec2.Instance
+	err := c.ec2.DescribeInstancesPages(request, func(page *ec2.DescribeInstancesOutput, lastPage bool) bool {
+		for _, reservations := range page.Reservations {
+			for _, instance := range reservations.Instances {
+				if c.tagging.hasClusterTag(instance.Tags) {
+					matches = append(matches, instance)
+				}
+			}
+		}
+		return !lastPage
+	})
 	if err != nil {
 		return nil, err
 	}
 
-	var matches []*ec2.Instance
-	for _, instance := range response {
-		if c.tagging.hasClusterTag(instance.Tags) {
-			matches = append(matches, instance)
-		}
-	}
 	return matches, nil
 }
 

pkg/cloudprovider/providers/aws/aws_fakes.go

@@ -174,6 +174,11 @@ func (ec2i *FakeEC2Impl) DescribeInstances(request *ec2.DescribeInstancesInput)
 	return matches, nil
 }
 
+// DescribeInstancesPages is not implemented but is required for interface conformance
+func (ec2i *FakeEC2Impl) DescribeInstancesPages(input *ec2.DescribeInstancesInput, fn func(*ec2.DescribeInstancesOutput, bool) bool) error {
+	panic("Not implemented")
+}
+
 // 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/cloudprovider/providers/aws/instances.go

@@ -26,9 +26,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/klog"
-
-	"k8s.io/api/core/v1"
 )
 
 // awsInstanceRegMatch represents Regex Match for AWS instance.