feat: add support for global accelerator endpoint groups

Signed-off-by: Tobias Kässer <[email protected]>

Author: kaessert

docs/examples/globalaccelerator.yaml

@@ -0,0 +1,19 @@
+kind: Ingress
+metadata:
+    name: echoserver
+    annotations:
+        kubernetes.io/ingress.class: alb
+        alb.ingress.kubernetes.io/scheme: internet-facing
+        alb.ingress.kubernetes.io/ga-epg-arn: arn:aws:globalaccelerator::12345678912:accelerator/d60128f1-4134-4e03-bed9-edd00f77b3e6/listener/a309af4a/endpoint-group/ed7bf648f700
+        alb.ingress.kubernetes.io/ga-ep-create: "true"
+spec:
+    rules:
+    - http:
+        paths:
+          - backend:
+              service:
+                name: my-release-nginx
+                port:
+                  number: 80
+            path: /
+            pathType: Prefix

docs/guide/globalaccelerator/endpointgroup.md

@@ -0,0 +1,12 @@
+# Create Endpoint on exisitng Endpointgroup
+
+In order to create an endpoint for the ingress-group, the user
+needs to specify two annotations:
+
+`alb.ingress.kubernetes.io/ga-epg-arn: arn:aws:globalaccelerator::12345678912:accelerator/d60128f1-4134-4e03-bed9-edd00f77b3e6/listener/a309af4a/endpoint-group/ed7bf648f700`
+`alb.ingress.kubernetes.io/ga-ep-create: "true"`
+
+This second annotation exists because of the fact that endpoints don't support tags
+and with the current stateless logic it is not possible to identify the correct
+endpoint for deletion. This means that:
+*Deletion is only supported by setting `ga-ep-create: "false"`*

pkg/annotations/constants.go

@@ -19,6 +19,8 @@ const (
 	IngressSuffixWAFACLID                     = "waf-acl-id"
 	IngressSuffixWebACLID                     = "web-acl-id" // deprecated, use "waf-acl-id" instead.
 	IngressSuffixShieldAdvancedProtection     = "shield-advanced-protection"
+	IngressSuffixGAEndpointGroup              = "ga-epg-arn"
+	IngressSuffixGAEndpointCreate             = "ga-ep-create"
 	IngressSuffixSecurityGroups               = "security-groups"
 	IngressSuffixListenPorts                  = "listen-ports"
 	IngressSuffixSSLRedirect                  = "ssl-redirect"

pkg/aws/cloud.go

@@ -41,6 +41,9 @@ type Cloud interface {
 	// RGT provides API to AWS RGT
 	RGT() services.RGT
 
+	// GA provides API to AWS Global Accelerator
+	GA() services.GlobalAccelerator
+
 	// Region for the kubernetes cluster
 	Region() string
 
@@ -125,6 +128,7 @@ func NewCloud(cfg CloudConfig, metricsRegisterer prometheus.Registerer) (Cloud,
 		ec2:         ec2Service,
 		elbv2:       services.NewELBV2(sess),
 		acm:         services.NewACM(sess),
+		ga:          services.NewGlobalAccelerator(sess),
 		wafv2:       services.NewWAFv2(sess),
 		wafRegional: services.NewWAFRegional(sess, cfg.Region),
 		shield:      services.NewShield(sess),
@@ -177,6 +181,7 @@ type defaultCloud struct {
 	elbv2 services.ELBV2
 
 	acm         services.ACM
+	ga          services.GlobalAccelerator
 	wafv2       services.WAFv2
 	wafRegional services.WAFRegional
 	shield      services.Shield
@@ -211,6 +216,10 @@ func (c *defaultCloud) RGT() services.RGT {
 	return c.rgt
 }
 
+func (c *defaultCloud) GA() services.GlobalAccelerator {
+	return c.ga
+}
+
 func (c *defaultCloud) Region() string {
 	return c.cfg.Region
 }

pkg/aws/services/globalaccelerator.go

@@ -0,0 +1,25 @@
+package services
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/globalaccelerator"
+	"github.com/aws/aws-sdk-go/service/globalaccelerator/globalacceleratoriface"
+)
+
+type GlobalAccelerator interface {
+	globalacceleratoriface.GlobalAcceleratorAPI
+}
+
+// NewGlobalAccelerator constructs new GlobalAccelerator implementation.
+func NewGlobalAccelerator(session *session.Session) GlobalAccelerator {
+	return &defaultGlobalAcceleratorAPI{
+		// global accelerator always needs `us-west-2`-region
+		GlobalAcceleratorAPI: globalaccelerator.New(session, aws.NewConfig().WithRegion("us-west-2")),
+	}
+}
+
+// default implementation for Global Accelerator.
+type defaultGlobalAcceleratorAPI struct {
+	globalacceleratoriface.GlobalAcceleratorAPI
+}

pkg/deploy/globalaccelerator/endpoint_manager.go

@@ -0,0 +1,85 @@
+package globalaccelerator
+
+import (
+	"context"
+
+	awssdk "github.com/aws/aws-sdk-go/aws"
+	gasdk "github.com/aws/aws-sdk-go/service/globalaccelerator"
+
+	"github.com/go-logr/logr"
+	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/services"
+)
+
+type EndpointManager interface {
+	// AddEndpoint add an endpoint to a globalaccelerator endpoint group
+	AddEndpoint(ctx context.Context, endpointGroupARN string, lbArn string) error
+
+	// GetEndpoint gets the existing endpoint for an endpointgroup and load balancer
+	GetEndpoint(ctx context.Context, endpointGroupARN string, lbArn string) (string, error)
+
+	// DeleteEndpoints deletes an existing endpoint for an endpointgroup and load balancer
+	DeleteEndpoint(ctx context.Context, endpointGroupARN string, lbArn string) error
+}
+
+func NewDefaultEndpointManager(gaClient services.GlobalAccelerator, logger logr.Logger) *defaultEndpointManager {
+	return &defaultEndpointManager{
+		gaClient: gaClient,
+	}
+}
+
+var _ EndpointManager = &defaultEndpointManager{}
+
+type defaultEndpointManager struct {
+	gaClient services.GlobalAccelerator
+}
+
+func (m *defaultEndpointManager) AddEndpoint(ctx context.Context, endpointGroupARN string, lbArn string) error {
+	_, err := m.gaClient.AddEndpoints(&gasdk.AddEndpointsInput{
+		EndpointGroupArn: &endpointGroupARN,
+		EndpointConfigurations: []*gasdk.EndpointConfiguration{
+			{
+				EndpointId: awssdk.String(lbArn),
+			},
+		},
+	})
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m *defaultEndpointManager) GetEndpoint(ctx context.Context, endpointGroupARN string, lbArn string) (string, error) {
+	epResponse, err := m.gaClient.DescribeEndpointGroup(&gasdk.DescribeEndpointGroupInput{
+		EndpointGroupArn: &endpointGroupARN,
+	})
+
+	if err != nil {
+		return "", err
+	}
+
+	for _, endpoint := range epResponse.EndpointGroup.EndpointDescriptions {
+		if *endpoint.EndpointId == lbArn {
+			return lbArn, nil
+		}
+	}
+
+	return "", nil
+}
+
+func (m *defaultEndpointManager) DeleteEndpoint(ctx context.Context, endpointGroupARN string, lbArn string) error {
+	_, err := m.gaClient.RemoveEndpoints(&gasdk.RemoveEndpointsInput{
+		EndpointGroupArn: &endpointGroupARN,
+		EndpointIdentifiers: []*gasdk.EndpointIdentifier{
+			{
+				EndpointId: awssdk.String(lbArn),
+			},
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/deploy/globalaccelerator/endpoint_synthesizer.go

@@ -0,0 +1,103 @@
+package globalaccelerator
+
+import (
+	"context"
+	"github.com/go-logr/logr"
+	"github.com/pkg/errors"
+	"sigs.k8s.io/aws-load-balancer-controller/pkg/model/core"
+	elbv2model "sigs.k8s.io/aws-load-balancer-controller/pkg/model/elbv2"
+	gamodel "sigs.k8s.io/aws-load-balancer-controller/pkg/model/globalaccelerator"
+)
+
+// NewEndpointSynthesizer constructs new endpointSynthesizer
+func NewEndpointSynthesizer(epManager EndpointManager, logger logr.Logger, stack core.Stack) *endpointSynthesizer {
+	return &endpointSynthesizer{
+		endpointManager: epManager,
+		logger:          logger,
+		stack:           stack,
+	}
+}
+
+type endpointSynthesizer struct {
+	endpointManager EndpointManager
+	logger          logr.Logger
+	stack           core.Stack
+}
+
+func (s *endpointSynthesizer) Synthesize(ctx context.Context) error {
+	var resEndpoints []*gamodel.Endpoint
+	s.stack.ListResources(&resEndpoints)
+	resEndpointsByARN, err := mapResEndpointByResourceARN(resEndpoints)
+	if err != nil {
+		return err
+	}
+
+	var resLBs []*elbv2model.LoadBalancer
+	s.stack.ListResources(&resLBs)
+	for _, resLB := range resLBs {
+		// Global Accelerator can only be created for ALB for now.
+		if resLB.Spec.Type != elbv2model.LoadBalancerTypeApplication {
+			continue
+		}
+		lbARN, err := resLB.LoadBalancerARN().Resolve(ctx)
+		if err != nil {
+			return err
+		}
+		resEndpoints := resEndpointsByARN[lbARN]
+
+		if err := s.synthesizeGAEndpoints(ctx, lbARN, resEndpoints); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (*endpointSynthesizer) PostSynthesize(ctx context.Context) error {
+	// nothing to do here.
+	return nil
+}
+
+func (s *endpointSynthesizer) synthesizeGAEndpoints(ctx context.Context, lbARN string, resEndpoints []*gamodel.Endpoint) error {
+	if len(resEndpoints) > 1 {
+		return fmt.Errorf("[should never happen] multiple Global Accelerator Endpoints desired on LoadBalancer: %v", lbARN)
+	}
+
+	var desiredEndpointGroupARN string
+	var desiredEndpointCreate bool
+	if len(resEndpoints) == 1 {
+		desiredEndpointGroupARN = resEndpoints[0].Spec.EndpointGroupARN
+		desiredEndpointCreate = resEndpoints[0].Spec.Create
+	}
+
+	if desiredEndpointGroupARN != "" {
+		// no lbARN means delete
+		if !desiredEndpointCreate {
+			s.endpointManager.DeleteEndpoint(ctx, desiredEndpointGroupARN, lbARN)
+			return nil
+		}
+
+		existingEndpoint, err := s.endpointManager.GetEndpoint(ctx, desiredEndpointGroupARN, lbARN)
+		if err != nil {
+			return err
+		}
+		if existingEndpoint == "" {
+			s.endpointManager.AddEndpoint(ctx, desiredEndpointGroupARN, lbARN)
+		}
+	}
+
+	return nil
+}
+
+func mapResEndpointByResourceARN(resEndpoints []*gamodel.Endpoint) (map[string][]*gamodel.Endpoint, error) {
+	resEndpointsByARN := make(map[string][]*gamodel.Endpoint, len(resEndpoints))
+	ctx := context.Background()
+	for _, resEndpoint := range resEndpoints {
+		resARN, err := resEndpoint.Spec.ResourceARN.Resolve(ctx)
+		if err != nil {
+			return nil, err
+		}
+		resEndpointsByARN[resARN] = append(resEndpointsByARN[resARN], resEndpoint)
+	}
+	return resEndpointsByARN, nil
+}

pkg/deploy/stack_deployer.go

@@ -7,6 +7,7 @@ import (
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/config"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/deploy/ec2"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/deploy/elbv2"
+	"sigs.k8s.io/aws-load-balancer-controller/pkg/deploy/globalaccelerator"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/deploy/shield"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/deploy/tracking"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/deploy/wafregional"
@@ -47,6 +48,7 @@ func NewDefaultStackDeployer(cloud aws.Cloud, k8sClient client.Client,
 		wafv2WebACLAssociationManager:       wafv2.NewDefaultWebACLAssociationManager(cloud.WAFv2(), logger),
 		wafRegionalWebACLAssociationManager: wafregional.NewDefaultWebACLAssociationManager(cloud.WAFRegional(), logger),
 		shieldProtectionManager:             shield.NewDefaultProtectionManager(cloud.Shield(), logger),
+		gaEndpointManager:                   globalaccelerator.NewDefaultEndpointManager(cloud.GA(), logger),
 		featureGates:                        config.FeatureGates,
 		vpcID:                               cloud.VpcID(),
 		logger:                              logger,
@@ -69,6 +71,7 @@ type defaultStackDeployer struct {
 	elbv2LRManager                      elbv2.ListenerRuleManager
 	elbv2TGManager                      elbv2.TargetGroupManager
 	elbv2TGBManager                     elbv2.TargetGroupBindingManager
+	gaEndpointManager                   globalaccelerator.EndpointManager
 	wafv2WebACLAssociationManager       wafv2.WebACLAssociationManager
 	wafRegionalWebACLAssociationManager wafregional.WebACLAssociationManager
 	shieldProtectionManager             shield.ProtectionManager
@@ -92,6 +95,7 @@ func (d *defaultStackDeployer) Deploy(ctx context.Context, stack core.Stack) err
 		elbv2.NewListenerSynthesizer(d.cloud.ELBV2(), d.elbv2TaggingManager, d.elbv2LSManager, d.logger, stack),
 		elbv2.NewListenerRuleSynthesizer(d.cloud.ELBV2(), d.elbv2TaggingManager, d.elbv2LRManager, d.logger, stack),
 		elbv2.NewTargetGroupBindingSynthesizer(d.k8sClient, d.trackingProvider, d.elbv2TGBManager, d.logger, stack),
+		globalaccelerator.NewEndpointSynthesizer(d.gaEndpointManager, d.logger, stack),
 	}
 
 	if d.addonsConfig.WAFV2Enabled {

pkg/ingress/model_build_load_balancer_addons.go

@@ -6,12 +6,14 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/annotations"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/model/core"
+	gamodel "sigs.k8s.io/aws-load-balancer-controller/pkg/model/globalaccelerator"
 	shieldmodel "sigs.k8s.io/aws-load-balancer-controller/pkg/model/shield"
 	wafregionalmodel "sigs.k8s.io/aws-load-balancer-controller/pkg/model/wafregional"
 	wafv2model "sigs.k8s.io/aws-load-balancer-controller/pkg/model/wafv2"
 )
 
 func (t *defaultModelBuildTask) buildLoadBalancerAddOns(ctx context.Context, lbARN core.StringToken) error {
+
 	if _, err := t.buildWAFv2WebACLAssociation(ctx, lbARN); err != nil {
 		return err
 	}
@@ -21,6 +23,10 @@ func (t *defaultModelBuildTask) buildLoadBalancerAddOns(ctx context.Context, lbA
 	if _, err := t.buildShieldProtection(ctx, lbARN); err != nil {
 		return err
 	}
+	if _, err := t.buildGAEndpoint(ctx, lbARN); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -102,3 +108,42 @@ func (t *defaultModelBuildTask) buildShieldProtection(_ context.Context, lbARN c
 	}
 	return nil, nil
 }
+
+func (t *defaultModelBuildTask) buildGAEndpoint(_ context.Context, lbARN core.StringToken) (*gamodel.Endpoint, error) {
+	explicitEPGARNs := sets.NewString()
+	epCreateByARN := make(map[string]string)
+	rawEPGARN := ""
+
+	for _, member := range t.ingGroup.Members {
+		// Unfortunately we can't support deletion of an endpoint just by removing
+		// the `ga-epg-arn` annotation, because cleaning up afterwards would require
+		// us to scan all accelerators and endpointgroups for the desired endpoint.
+		// To still enable a way to delete the endpoint, users can set `ga-epg-create=false`
+		epCreate := "false"
+		if exists := t.annotationParser.ParseStringAnnotation(annotations.IngressSuffixGAEndpointGroup, &rawEPGARN, member.Ing.Annotations); exists {
+			explicitEPGARNs.Insert(rawEPGARN)
+			t.annotationParser.ParseStringAnnotation(annotations.IngressSuffixGAEndpointCreate, &epCreate, member.Ing.Annotations)
+			epCreateByARN[rawEPGARN] = epCreate
+		}
+	}
+	if len(explicitEPGARNs) > 1 {
+		return nil, errors.Errorf("conflicting Global Accelerator EndpointGroup ARNs: %v", explicitEPGARNs.List())
+	}
+	if len(explicitEPGARNs) == 1 {
+		epgARN, _ := explicitEPGARNs.PopAny()
+		if epgARN != "" {
+			create := false
+			if epCreateByARN[epgARN] == "true" {
+				create = true
+			}
+			endpoint := gamodel.NewEndpoint(t.stack, resourceIDLoadBalancer, gamodel.EndpointSpec{
+				EndpointGroupARN: epgARN,
+				ResourceARN:      lbARN,
+				Create:           create,
+			})
+			return endpoint, nil
+		}
+	}
+
+	return nil, nil
+}