Allow TargetGroup endpoints outside the ELB VPC

To support the scenario where the ELB is deployed in a separate VPC to
the EKS cluster, we need to conditionally set the `AvailabilityZone`
field to `all` when the target is outside of the ELB VPC.

This commit addresses this by checking if the pod IP is found
in the ELB VPC's CIDR or a secondary CIDR; to improve performance and
reduce calls to the AWS API, VPC info is cached for a configurable
amount of time, defaulting to 5 minutes.

Signed-off-by: Stephen Hoekstra <[email protected]>

Author: shoekstra

docs/deploy/configurations.md

@@ -67,11 +67,12 @@ Currently, you can set only 1 namespace to watch in this flag. See [this Kuberne
 
 |Flag                                   | Type                            | Default         | Description |
 |---------------------------------------|---------------------------------|-----------------|-------------|
+|aws-api-endpoints                      | AWS API Endpoints Config        |                 | AWS API endpoints mapping, format: serviceID1=URL1,serviceID2=URL2 |
 |aws-api-throttle                       | AWS Throttle Config             | [default value](#default-throttle-config ) | throttle settings for AWS APIs, format: serviceID1:operationRegex1=rate:burst,serviceID2:operationRegex2=rate:burst |
 |aws-max-retries                        | int                             | 10              | Maximum retries for AWS APIs |
 |aws-region                             | string                          | [instance metadata](#instance-metadata)    | AWS Region for the kubernetes cluster |
-|aws-vpc-id                             | string                          | [instance metadata](#instance-metadata)    | AWS VPC ID for the Kubernetes cluster |
-|aws-api-endpoints                      | AWS API Endpoints Config        |                 | AWS API endpoints mapping, format: serviceID1=URL1,serviceID2=URL2 |
+|aws-vpc-cache-duration                 | string                          | 5               | Length of time in minutes to cache VPC info |
+|aws-vpc-id                             | string                          | [instance metadata](#instance-metadata)    | ID of the AWS VPC where Load Balancer resources will be created |
 |cluster-name                           | string                          |                 | Kubernetes cluster name|
 |default-tags                           | stringMap                       |                 | AWS Tags that will be applied to all AWS resources managed by this controller. Specified Tags takes highest priority |
 |default-ssl-policy                     | string                          | ELBSecurityPolicy-2016-08 | Default SSL Policy that will be applied to all Ingresses or Services that do not have the SSL Policy annotation |

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/google/go-cmp v0.5.6
 	github.com/onsi/ginkgo v1.16.4
 	github.com/onsi/gomega v1.16.0
+	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0
 	github.com/spf13/pflag v1.0.5

go.sum

@@ -643,6 +643,10 @@ github.com/openzipkin/zipkin-go v0.2.1/go.mod h1:NaW6tEwdmWMaCDZzg8sh+IBNOxHMPnh
 github.com/openzipkin/zipkin-go v0.2.2/go.mod h1:NaW6tEwdmWMaCDZzg8sh+IBNOxHMPnhQw8ySjnjRyN4=
 github.com/pact-foundation/pact-go v1.0.4/go.mod h1:uExwJY4kCzNPcHRj+hCR/HBbOOIwwtUjcrb0b5/5kLM=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
+github.com/patrickmn/go-cache v1.0.0 h1:3gD5McaYs9CxjyK5AXGcq8gdeCARtd/9gJDUvVeaZ0Y=
+github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
+github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
+github.com/pborman/uuid v1.2.0 h1:J7Q5mO4ysT1dv8hyrUGHb9+ooztCXu1D8MY8DZYsu3g=
 github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/performancecopilot/speed v3.0.0+incompatible/go.mod h1:/CLtqpZ5gBg1M9iaPbIdPPGyKcA8hKdoy6hAWba7Yac=

main.go

@@ -103,10 +103,11 @@ func main() {
 	sgManager := networking.NewDefaultSecurityGroupManager(cloud.EC2(), ctrl.Log)
 	sgReconciler := networking.NewDefaultSecurityGroupReconciler(sgManager, ctrl.Log)
 	azInfoProvider := networking.NewDefaultAZInfoProvider(cloud.EC2(), ctrl.Log.WithName("az-info-provider"))
+	vpcInfoProvider := networking.NewDefaultVPCInfoProvider(cloud.VpcCacheDuration(), cloud.EC2(), ctrl.Log.WithName("vpc-info-provider"))
 	subnetResolver := networking.NewDefaultSubnetsResolver(azInfoProvider, cloud.EC2(), cloud.VpcID(), controllerCFG.ClusterName, ctrl.Log.WithName("subnets-resolver"))
 	vpcResolver := networking.NewDefaultVPCResolver(cloud.EC2(), cloud.VpcID(), ctrl.Log.WithName("vpc-resolver"))
 	tgbResManager := targetgroupbinding.NewDefaultResourceManager(mgr.GetClient(), cloud.ELBV2(), cloud.EC2(),
-		podInfoRepo, sgManager, sgReconciler, cloud.VpcID(), controllerCFG.ClusterName, mgr.GetEventRecorderFor("targetGroupBinding"), ctrl.Log, controllerCFG.EnableEndpointSlices, controllerCFG.DisableRestrictedSGRules)
+		podInfoRepo, sgManager, sgReconciler, cloud.VpcID(), controllerCFG.ClusterName, mgr.GetEventRecorderFor("targetGroupBinding"), ctrl.Log, controllerCFG.EnableEndpointSlices, controllerCFG.DisableRestrictedSGRules, vpcInfoProvider)
 	backendSGProvider := networking.NewBackendSGProvider(controllerCFG.ClusterName, controllerCFG.BackendSecurityGroup,
 		cloud.VpcID(), cloud.EC2(), mgr.GetClient(), ctrl.Log.WithName("backend-sg-provider"))
 	ingGroupReconciler := ingress.NewGroupReconciler(cloud, mgr.GetClient(), mgr.GetEventRecorderFor("ingress"),

pkg/aws/cloud.go

@@ -38,8 +38,11 @@ type Cloud interface {
 	// Region for the kubernetes cluster
 	Region() string
 
-	// VPC ID for the the kubernetes cluster
+	// ID of VPC to create load balancers in
 	VpcID() string
+
+	// VPC cache duration in minutes
+	VpcCacheDuration() int
 }
 
 // NewCloud constructs new Cloud implementation.
@@ -149,3 +152,7 @@ func (c *defaultCloud) Region() string {
 func (c *defaultCloud) VpcID() string {
 	return c.cfg.VpcID
 }
+
+func (c *defaultCloud) VpcCacheDuration() int {
+	return c.cfg.VpcCacheDuration
+}

pkg/aws/cloud_config.go

@@ -6,14 +6,16 @@ import (
 )
 
 const (
-	flagAWSRegion        = "aws-region"
-	flagAWSAPIThrottle   = "aws-api-throttle"
-	flagAWSVpcID         = "aws-vpc-id"
-	flagAWSMaxRetries    = "aws-max-retries"
-	flagAWSAPIEndpoints  = "aws-api-endpoints"
-	defaultVpcID         = ""
-	defaultRegion        = ""
-	defaultAPIMaxRetries = 10
+	flagAWSRegion           = "aws-region"
+	flagAWSAPIEndpoints     = "aws-api-endpoints"
+	flagAWSAPIThrottle      = "aws-api-throttle"
+	flagAWSVpcID            = "aws-vpc-id"
+	flagAWSVpcCacheDuration = "aws-vpc-cache-duration"
+	flagAWSMaxRetries       = "aws-max-retries"
+	defaultVpcID            = ""
+	defaultRegion           = ""
+	defaultAPIMaxRetries    = 10
+	defaultVpcCacheDuration = 5
 )
 
 type CloudConfig struct {
@@ -23,9 +25,12 @@ type CloudConfig struct {
 	// Throttle settings for AWS APIs
 	ThrottleConfig *throttle.ServiceOperationsThrottleConfig
 
-	// VPC ID of the Kubernetes cluster
+	// ID of VPC to create load balancers in
 	VpcID string
 
+	// VPC cache duration in minutes
+	VpcCacheDuration int
+
 	// Max retries configuration for AWS APIs
 	MaxRetries int
 
@@ -36,7 +41,8 @@ type CloudConfig struct {
 func (cfg *CloudConfig) BindFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&cfg.Region, flagAWSRegion, defaultRegion, "AWS Region for the kubernetes cluster")
 	fs.Var(cfg.ThrottleConfig, flagAWSAPIThrottle, "throttle settings for AWS APIs, format: serviceID1:operationRegex1=rate:burst,serviceID2:operationRegex2=rate:burst")
-	fs.StringVar(&cfg.VpcID, flagAWSVpcID, defaultVpcID, "AWS VPC ID for the Kubernetes cluster")
+	fs.StringVar(&cfg.VpcID, flagAWSVpcID, defaultVpcID, "AWS ID of VPC to create load balancers in")
+	fs.IntVar(&cfg.VpcCacheDuration, flagAWSVpcCacheDuration, defaultVpcCacheDuration, "VPC cache duration in minutes")
 	fs.IntVar(&cfg.MaxRetries, flagAWSMaxRetries, defaultAPIMaxRetries, "Maximum retries for AWS APIs")
 	fs.StringToStringVar(&cfg.AWSEndpoints, flagAWSAPIEndpoints, nil, "Custom AWS endpoint configuration, format: serviceID1=URL1,serviceID2=URL2")
 }

pkg/networking/vpc_info_provider.go

@@ -0,0 +1,86 @@
+package networking
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	awssdk "github.com/aws/aws-sdk-go/aws"
+	ec2sdk "github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/go-logr/logr"
+	"github.com/patrickmn/go-cache"
+	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/services"
+)
+
+// VPCInfoProvider is responsible for providing VPC info.
+type VPCInfoProvider interface {
+	FetchVPCInfo(ctx context.Context, vpcID string) (*ec2sdk.Vpc, error)
+}
+
+// NewDefaultVPCInfoProvider constructs new defaultVPCInfoProvider.
+func NewDefaultVPCInfoProvider(cacheDuration int, ec2Client services.EC2, logger logr.Logger) *defaultVPCInfoProvider {
+	return &defaultVPCInfoProvider{
+		ec2Client:         ec2Client,
+		vpcInfoCache:      cache.New(time.Duration(cacheDuration)*time.Minute, 10*time.Minute),
+		vpcInfoCacheMutex: sync.RWMutex{},
+		logger:            logger,
+	}
+}
+
+var _ VPCInfoProvider = &defaultVPCInfoProvider{}
+
+// default implementation for VPCInfoProvider.
+type defaultVPCInfoProvider struct {
+	ec2Client         services.EC2
+	vpcInfoCache      *cache.Cache
+	vpcInfoCacheMutex sync.RWMutex
+
+	logger logr.Logger
+}
+
+func (p *defaultVPCInfoProvider) FetchVPCInfo(ctx context.Context, vpcID string) (*ec2sdk.Vpc, error) {
+	if vpcInfo := p.fetchVPCInfoFromCache(); vpcInfo != nil {
+		return vpcInfo, nil
+	}
+
+	// Fetch VPC info from the AWS API and cache response before returning.
+	vpcInfo, err := p.fetchVPCInfoFromAWS(ctx, vpcID)
+	if err != nil {
+		return nil, nil
+	}
+	p.saveVPCInfoToCache(vpcInfo)
+
+	return vpcInfo, nil
+}
+
+func (p *defaultVPCInfoProvider) fetchVPCInfoFromCache() *ec2sdk.Vpc {
+	p.vpcInfoCacheMutex.RLock()
+	defer p.vpcInfoCacheMutex.RUnlock()
+
+	vpcInfo, found := p.vpcInfoCache.Get("vpcInfo")
+	if !found {
+		return nil
+	}
+
+	return vpcInfo.(*ec2sdk.Vpc)
+}
+
+func (p *defaultVPCInfoProvider) saveVPCInfoToCache(vpcInfo *ec2sdk.Vpc) {
+	p.vpcInfoCacheMutex.Lock()
+	defer p.vpcInfoCacheMutex.Unlock()
+
+	p.vpcInfoCache.SetDefault("vpcInfo", vpcInfo)
+}
+
+// fetchVPCInfoFromAWS will fetch VPC info from the AWS API.
+func (p *defaultVPCInfoProvider) fetchVPCInfoFromAWS(ctx context.Context, vpcID string) (*ec2sdk.Vpc, error) {
+	req := &ec2sdk.DescribeVpcsInput{
+		VpcIds: []*string{awssdk.String(vpcID)},
+	}
+	resp, err := p.ec2Client.DescribeVpcsWithContext(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	return resp.Vpcs[0], nil
+}

pkg/networking/vpc_info_provider_mocks.go

@@ -0,0 +1,83 @@
+package networking
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	awssdk "github.com/aws/aws-sdk-go/aws"
+	ec2sdk "github.com/aws/aws-sdk-go/service/ec2"
+	gomock "github.com/golang/mock/gomock"
+	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/services"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+func Test_defaultVPCInfoProvider_FetchVPCInfo(t *testing.T) {
+	type describeVpcCall struct {
+		input  *ec2sdk.DescribeVpcsInput
+		output *ec2sdk.DescribeVpcsOutput
+		err    error
+	}
+
+	type fields struct {
+		describeVpcCalls []describeVpcCall
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		want    *ec2sdk.Vpc
+		wantErr bool
+	}{
+		{
+			name: "from AWS",
+			fields: fields{
+				describeVpcCalls: []describeVpcCall{
+					{
+						input: &ec2sdk.DescribeVpcsInput{
+							VpcIds: []*string{awssdk.String("vpc-2f09a348")},
+						},
+						output: &ec2sdk.DescribeVpcsOutput{
+							Vpcs: []*ec2sdk.Vpc{{VpcId: awssdk.String("vpc-2f09a348")}},
+						},
+						err: nil,
+					},
+				},
+			},
+			want: &ec2sdk.Vpc{
+				VpcId: awssdk.String("vpc-2f09a348"),
+			},
+			wantErr: false,
+		},
+		{
+			name:   "from cache",
+			fields: fields{},
+			want: &ec2sdk.Vpc{
+				VpcId: awssdk.String("vpc-2f09a348"),
+			},
+			wantErr: false,
+		},
+	}
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	ec2Client := services.NewMockEC2(ctrl)
+	p := NewDefaultVPCInfoProvider(5, ec2Client, &log.NullLogger{})
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			for _, call := range tt.fields.describeVpcCalls {
+				ec2Client.EXPECT().DescribeVpcsWithContext(gomock.Any(), call.input).Return(call.output, call.err)
+			}
+
+			got, err := p.FetchVPCInfo(context.Background(), "vpc-2f09a348")
+			if (err != nil) != tt.wantErr {
+				t.Errorf("defaultVPCInfoProvider.FetchVPCInfo() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("defaultVPCInfoProvider.FetchVPCInfo() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/networking/vpc_info_provider_test.go

@@ -3,6 +3,10 @@ package targetgroupbinding
 import (
 	"context"
 	"fmt"
+	"net"
+	"strings"
+	"sync"
+
 	awssdk "github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	ec2sdk "github.com/aws/aws-sdk-go/service/ec2"
@@ -13,14 +17,11 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"net"
 	elbv2api "sigs.k8s.io/aws-load-balancer-controller/apis/elbv2/v1beta1"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/backend"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/k8s"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/networking"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"strings"
-	"sync"
 )
 
 const (
@@ -40,6 +41,9 @@ type NetworkingManager interface {
 
 	// Cleanup reconcile network settings for TargetGroupBindings with zero endpoints.
 	Cleanup(ctx context.Context, tgb *elbv2api.TargetGroupBinding) error
+
+	// VpcID returns the ID of the VPC the ELB is attached to.
+	VpcID() string
 }
 
 // NewDefaultNetworkingManager constructs defaultNetworkingManager.
@@ -119,6 +123,10 @@ func (m *defaultNetworkingManager) Cleanup(ctx context.Context, tgb *elbv2api.Ta
 	return m.reconcileWithIngressPermissionsPerSG(ctx, tgb, nil)
 }
 
+func (m *defaultNetworkingManager) VpcID() string {
+	return m.vpcID
+}
+
 func (m *defaultNetworkingManager) computeIngressPermissionsPerSGWithPodEndpoints(ctx context.Context, tgbNetworking elbv2api.TargetGroupBindingNetworking, endpoints []backend.PodEndpoint) (map[string][]networking.IPPermissionInfo, error) {
 	pods := make([]k8s.PodInfo, 0, len(endpoints))
 	podByPodKey := make(map[types.NamespacedName]k8s.PodInfo, len(endpoints))