add custom aws endpoint configuration (#2179)

Co-authored-by: Kishor Joshi <[email protected]>

Author: papigers

docs/deploy/configurations.md

@@ -71,6 +71,7 @@ Currently, you can set only 1 namespace to watch in this flag. See [this Kuberne
 |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 |
 |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 |

helm/aws-load-balancer-controller/templates/deployment.yaml

@@ -59,6 +59,9 @@ spec:
         {{- if .Values.vpcId }}
         - --aws-vpc-id={{ .Values.vpcId }}
         {{- end }}
+        {{- if .Values.awsApiEndpoints }}
+        - --aws-api-endpoints={{ .Values.awsApiEndpoints }}
+        {{- end }}
         {{- if .Values.awsMaxRetries }}
         - --aws-max-retries={{ .Values.awsMaxRetries }}
         {{- end }}

helm/aws-load-balancer-controller/values.yaml

@@ -93,6 +93,9 @@ region:
 # The VPC ID for the Kubernetes cluster. Set this manually when your pods are unable to use the metadata service to determine this automatically
 vpcId:
 
+# Custom AWS API Endpoints (serviceID1=URL1,serviceID2=URL2)
+awsApiEndpoints:
+
 # Maximum retries for AWS APIs (default 10)
 awsMaxRetries:
 

main.go

@@ -31,6 +31,7 @@ import (
 	"sigs.k8s.io/aws-load-balancer-controller/controllers/ingress"
 	"sigs.k8s.io/aws-load-balancer-controller/controllers/service"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws"
+	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/endpoints"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/throttle"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/config"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/inject"
@@ -170,7 +171,10 @@ func main() {
 func loadControllerConfig() (config.ControllerConfig, error) {
 	defaultAWSThrottleCFG := throttle.NewDefaultServiceOperationsThrottleConfig()
 	controllerCFG := config.ControllerConfig{
-		AWSConfig: aws.CloudConfig{ThrottleConfig: defaultAWSThrottleCFG},
+		AWSConfig: aws.CloudConfig{
+			ThrottleConfig: defaultAWSThrottleCFG,
+			AWSEndpointResolver: &endpoints.AWSEndpointResolver{},
+		},
 	}
 
 	fs := pflag.NewFlagSet("", pflag.ExitOnError)

pkg/aws/cloud.go

@@ -43,7 +43,8 @@ type Cloud interface {
 
 // NewCloud constructs new Cloud implementation.
 func NewCloud(cfg CloudConfig, metricsRegisterer prometheus.Registerer) (Cloud, error) {
-	metadataSess := session.Must(session.NewSession(aws.NewConfig()))
+	metadataCFG := aws.NewConfig().WithEndpointResolver(cfg.AWSEndpointResolver)
+	metadataSess := session.Must(session.NewSession(metadataCFG))
 	metadata := services.NewEC2Metadata(metadataSess)
 	if len(cfg.VpcID) == 0 {
 		vpcId, err := metadata.VpcID()
@@ -68,8 +69,7 @@ func NewCloud(cfg CloudConfig, metricsRegisterer prometheus.Registerer) (Cloud,
 		}
 		cfg.Region = region
 	}
-
-	awsCFG := aws.NewConfig().WithRegion(cfg.Region).WithSTSRegionalEndpoint(endpoints.RegionalSTSEndpoint).WithMaxRetries(cfg.MaxRetries)
+	awsCFG := aws.NewConfig().WithRegion(cfg.Region).WithSTSRegionalEndpoint(endpoints.RegionalSTSEndpoint).WithMaxRetries(cfg.MaxRetries).WithEndpointResolver(cfg.AWSEndpointResolver)
 	sess := session.Must(session.NewSession(awsCFG))
 	injectUserAgent(&sess.Handlers)
 

pkg/aws/cloud_config.go

@@ -2,6 +2,7 @@ package aws
 
 import (
 	"github.com/spf13/pflag"
+	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/endpoints"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/throttle"
 )
 
@@ -10,6 +11,7 @@ const (
 	flagAWSAPIThrottle   = "aws-api-throttle"
 	flagAWSVpcID         = "aws-vpc-id"
 	flagAWSMaxRetries    = "aws-max-retries"
+	flagAWSAPIEndpoints  = "aws-api-endpoints"
 	defaultVpcID         = ""
 	defaultRegion        = ""
 	defaultAPIMaxRetries = 10
@@ -27,11 +29,15 @@ type CloudConfig struct {
 
 	// Max retries configuration for AWS APIs
 	MaxRetries int
+
+	// AWS endpoint configuration
+	AWSEndpointResolver *endpoints.AWSEndpointResolver
 }
 
 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.IntVar(&cfg.MaxRetries, flagAWSMaxRetries, defaultAPIMaxRetries, "Maximum retries for AWS APIs")
+	fs.Var(cfg.AWSEndpointResolver, flagAWSAPIEndpoints, "Custom AWS endpoint configuration, format: serviceID1=URL1,serviceID2=URL2")
 }

pkg/aws/endpoints/resolver.go

@@ -0,0 +1,84 @@
+package endpoints
+
+import (
+	"fmt"
+	"net/url"
+	"sort"
+	"strings"
+
+	awsendpoints "github.com/aws/aws-sdk-go/aws/endpoints"
+	"github.com/pkg/errors"
+	"github.com/spf13/pflag"
+)
+
+var _ pflag.Value = &AWSEndpointResolver{}
+
+// AWSEndpointResolver is an AWS endpoints.Resolver that allows to customize AWS API endpoints.
+// It can be configured using the following format "${AWSServiceID}=${URL}"
+// e.g. "ec2=https://ec2.domain.com,elasticloadbalancing=https://elbv2.domain.com"
+type AWSEndpointResolver struct {
+	configuration map[string]string
+}
+
+func (c *AWSEndpointResolver) String() string {
+	if c == nil {
+		return ""
+	}
+
+	var configs []string
+	var serviceIDs []string
+	for serviceID := range c.configuration {
+		serviceIDs = append(serviceIDs, serviceID)
+	}
+	sort.Strings(serviceIDs)
+	for _, serviceID := range serviceIDs {
+		configs = append(configs, fmt.Sprintf("%s=%s", serviceID, c.configuration[serviceID]))
+	}
+	return strings.Join(configs, ",")
+}
+
+func (c *AWSEndpointResolver) Set(val string) error {
+	configurationOverride := make(map[string]string)
+
+	if val != "" {
+		configPairs := strings.Split(val, ",")
+		for _, pair := range configPairs {
+			kv := strings.Split(pair, "=")
+			if len(kv) != 2 {
+				return errors.Errorf("%s must be formatted as serviceID=URL", pair)
+			}
+			serviceID := kv[0]
+			urlStr := kv[1]
+			url, err := url.Parse(urlStr)
+			if err != nil {
+				return errors.Errorf("%s must be a valid url", urlStr)
+			}
+			if !url.IsAbs() {
+				return errors.Errorf("%s must be an absolute url", urlStr)
+			}
+			configurationOverride[serviceID] = url.String()
+		}
+	}
+
+	if c.configuration == nil {
+		c.configuration = make(map[string]string)
+	}
+	for k, v := range configurationOverride {
+		c.configuration[k] = v
+	}
+	return nil
+}
+
+func (c *AWSEndpointResolver) Type() string {
+	return "awsEndpointResolver"
+}
+
+func (c *AWSEndpointResolver) EndpointFor(service, region string, opts ...func(*awsendpoints.Options)) (awsendpoints.ResolvedEndpoint, error) {
+	customEndpoint := c.configuration[service]
+	if len(customEndpoint) != 0 {
+		return awsendpoints.ResolvedEndpoint{
+			URL: customEndpoint,
+		}, nil
+	}
+	return awsendpoints.DefaultResolver().EndpointFor(service, region, opts...)
+}

pkg/aws/endpoints/resolver_test.go

@@ -0,0 +1,206 @@
+package endpoints
+
+import (
+	"testing"
+
+	awsendpoints "github.com/aws/aws-sdk-go/aws/endpoints"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAWSEndpointResolver_String(t *testing.T) {
+	type fields struct {
+		configuration map[string]string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   string
+	}{
+		{
+			name: "non-empty configuration",
+			fields: fields{
+				configuration: map[string]string{
+					awsendpoints.Ec2ServiceID: "https://ec2.domain.com",
+					awsendpoints.ElasticloadbalancingServiceID: "https://elbv2.domain.com",
+				},
+			},
+			want: "ec2=https://ec2.domain.com,elasticloadbalancing=https://elbv2.domain.com",
+		},
+		{
+			name: "nil configuration",
+			fields: fields{
+				configuration: nil,
+			},
+			want: "",
+		},
+		{
+			name: "empty configuration",
+			fields: fields{
+				configuration: nil,
+			},
+			want: "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &AWSEndpointResolver{
+				configuration: tt.fields.configuration,
+			}
+			got := c.String()
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestAWSEndpointResolver_Set(t *testing.T) {
+	type fields struct {
+		configuration map[string]string
+	}
+	type args struct {
+		val string
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		want    AWSEndpointResolver
+		wantErr error
+	}{
+		{
+			name: "when default value is nil",
+			fields: fields{
+				configuration: nil,
+			},
+			args: args{
+				val: "ec2=https://ec2.domain.com,elasticloadbalancing=https://elbv2.domain.com",
+			},
+			want: AWSEndpointResolver{
+				configuration: map[string]string{
+					awsendpoints.Ec2ServiceID: "https://ec2.domain.com",
+					awsendpoints.ElasticloadbalancingServiceID: "https://elbv2.domain.com",
+				},
+			},
+		},
+		{
+			name: "when val is empty",
+			fields: fields{
+				configuration: map[string]string{},
+			},
+			args: args{
+				val: "",
+			},
+			want: AWSEndpointResolver{
+				configuration: map[string]string{},
+			},
+		},
+		{
+			name: "when val is not valid format - case 1",
+			fields: fields{
+				configuration: map[string]string{},
+			},
+			args: args{
+				val: "a=b=c",
+			},
+			wantErr: errors.Errorf("a=b=c must be formatted as serviceID=URL"),
+		},
+		{
+			name: "when url is not absolute",
+			fields: fields{
+				configuration: map[string]string{},
+			},
+			args: args{
+				val: "a=/relative/url",
+			},
+			wantErr: errors.Errorf("/relative/url must be an absolute url"),
+		},
+		{
+			name: "when url is invalid",
+			fields: fields{
+				configuration: map[string]string{},
+			},
+			args: args{
+				val: "a=invalid\turl",
+			},
+			wantErr: errors.Errorf("invalid\turl must be a valid url"),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &AWSEndpointResolver{
+				configuration: tt.fields.configuration,
+			}
+			err := c.Set(tt.args.val)
+			if tt.wantErr != nil {
+				assert.EqualError(t, err, tt.wantErr.Error())
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.want, *c)
+			}
+		})
+	}
+}
+
+func TestAWSEndpointResolver_Type(t *testing.T) {
+	c := &AWSEndpointResolver{}
+	got := c.Type()
+	assert.Equal(t, "awsEndpointResolver", got)
+}
+
+func TestAWSEndpointResolver_EndpointFor(t *testing.T) {
+	configuration := map[string]string{
+		awsendpoints.Ec2ServiceID: "https://ec2.domain.com",
+		awsendpoints.ElasticloadbalancingServiceID: "https://elbv2.domain.com",
+	}
+	c := &AWSEndpointResolver{
+		configuration: configuration,
+	}
+
+	testRegion := "region"
+
+	type args struct {
+		val string
+	}
+
+	tests := []struct {
+		name    string
+		args    args
+		want    *awsendpoints.ResolvedEndpoint
+		wantErr error
+	}{
+		{
+			name: "when custom endpoint is configured",
+			args: args{
+				val: awsendpoints.Ec2ServiceID,
+			},
+			want: &awsendpoints.ResolvedEndpoint{
+				URL: configuration[awsendpoints.Ec2ServiceID],
+			},
+		},
+		{
+			name: "when custom endpoint is unconfigured",
+			args: args{
+				val: awsendpoints.WafServiceID,
+			},
+			want: nil,
+		},
+	}
+	
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			res, err := c.EndpointFor(tt.args.val, testRegion)
+			if tt.wantErr != nil {
+				assert.EqualError(t, err, tt.wantErr.Error())
+			} else {
+				assert.NoError(t, err)
+				if tt.want != nil {
+					assert.Equal(t, *tt.want, res)
+				} else {
+					defaultEndpoint, err := awsendpoints.DefaultResolver().EndpointFor(tt.args.val, testRegion)
+					assert.NoError(t, err)
+					assert.Equal(t, defaultEndpoint, res)
+				}
+			}
+		})
+	}
+}