Add pagination to describeInstances in aws

For large clusters and AWS projects, the describeInstances api call
returns a 500 http error because the response exceed the limit in
the AWS api.

This means that, for example, for a classic loadbalancer, no new nodes
get added to it, which could lead to service disruption over time
due to node attrition.

This commit attempts to solve that by changing the call to use the
pagination method from the aws sdk

Author: amardomingo

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
@@ -4503,17 +4511,21 @@ func (c *Cloud) describeInstances(filters []*ec2.Filter) ([]*ec2.Instance, error
 		Filters: filters,
 	}
 
-	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,10 @@ 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.