✨ edge subnets/gateway: carrier gateway routing for Wavelength subnets

✨ edge subnets/cagw: carrier gateway for public subnets in Wavelength

Introduce Carrier Gateway resource reconciliator in the network service.

Carrier Gateway is the gateway responsible to route ingress and egress
traffic **in/out the Wavelength Zone**, located in the Carrier
Infrastructure - communications service providers’ (CSP) 5G networks.

Carrier Gateway is similar Internet Gatewat resource, responsible for
the network border groups in the Region and Local Zones for public
subnets.

✨ edge subnets/routes: supporting custom routes for Wavelength

For private and public subnets in edge zones, the following changes is
introduced according to each rule:

General:

- IPv6 subnets is not be supported in AWS Local Zones and Wavelength
  zone, consequently no ip6 routes will be created
- nat gateways is not supported, default gateway's route for private
  subnets will use nat gateways from the zones in the Region
(availability-zone's zone type)
- one route table by zone's role by zone (standard flow)

Private tables for Local Zones and Wavelength:
- default route's gateways is assigned using nat gateway created in
  the region (availability-zones).

Public tables for Wavelength zones:
- default route's gateways is assigned using Carrier Gateway, resource
  introduced in the edge zone's feature.

The changes in the standard flow (without edge subnets' support) was
isolated in the PR kubernetes-sigs#4900

api/v1beta2/conditions_consts.go

@@ -69,6 +69,14 @@ const (
 	EgressOnlyInternetGatewayFailedReason = "EgressOnlyInternetGatewayFailed"
 )
 
+const (
+	// CarrierGatewayReadyCondition reports on the successful reconciliation of carrier gateways.
+	// Only applicable to managed clusters.
+	CarrierGatewayReadyCondition clusterv1.ConditionType = "CarrierGatewayReady"
+	// CarrierGatewayFailedReason used when errors occur during carrier gateway reconciliation.
+	CarrierGatewayFailedReason = "CarrierGatewayFailed"
+)
+
 const (
 	// NatGatewaysReadyCondition reports successful reconciliation of NAT gateways.
 	// Only applicable to managed clusters.

pkg/cloud/awserrors/errors.go

@@ -33,6 +33,7 @@ const (
 	GatewayNotFound                   = "InvalidGatewayID.NotFound"
 	GroupNotFound                     = "InvalidGroup.NotFound"
 	InternetGatewayNotFound           = "InvalidInternetGatewayID.NotFound"
+	InvalidCarrierGatewayNotFound     = "InvalidCarrierGatewayID.NotFound"
 	EgressOnlyInternetGatewayNotFound = "InvalidEgressOnlyInternetGatewayID.NotFound"
 	InUseIPAddress                    = "InvalidIPAddress.InUse"
 	InvalidAccessKeyID                = "InvalidAccessKeyId"

pkg/cloud/services/network/carriergateways.go

@@ -0,0 +1,145 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package network
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/pkg/errors"
+
+	infrav1 "sigs.k8s.io/cluster-api-provider-aws/v2/api/v1beta2"
+	"sigs.k8s.io/cluster-api-provider-aws/v2/pkg/cloud/awserrors"
+	"sigs.k8s.io/cluster-api-provider-aws/v2/pkg/cloud/converters"
+	"sigs.k8s.io/cluster-api-provider-aws/v2/pkg/cloud/filter"
+	"sigs.k8s.io/cluster-api-provider-aws/v2/pkg/cloud/services"
+	"sigs.k8s.io/cluster-api-provider-aws/v2/pkg/cloud/services/wait"
+	"sigs.k8s.io/cluster-api-provider-aws/v2/pkg/cloud/tags"
+	"sigs.k8s.io/cluster-api-provider-aws/v2/pkg/record"
+	"sigs.k8s.io/cluster-api/util/conditions"
+)
+
+func (s *Service) reconcileCarrierGateway() error {
+	if s.scope.VPC().IsUnmanaged(s.scope.Name()) {
+		s.scope.Trace("Skipping carrier gateway reconcile in unmanaged mode")
+		return nil
+	}
+
+	if !s.scope.Subnets().HasPublicSubnetWavelength() {
+		s.scope.Trace("Skipping carrier gateway reconcile in VPC without subnets in zone type wavelength-zone")
+		return nil
+	}
+
+	s.scope.Debug("Reconciling carrier gateway")
+
+	cagw, err := s.describeVpcCarrierGateway()
+	if awserrors.IsNotFound(err) {
+		if s.scope.VPC().IsUnmanaged(s.scope.Name()) {
+			return errors.Errorf("failed to validate network: no carrier gateway found in VPC %q", s.scope.VPC().ID)
+		}
+
+		cg, err := s.createCarrierGateway()
+		if err != nil {
+			return err
+		}
+		cagw = cg
+	} else if err != nil {
+		return err
+	}
+
+	s.scope.VPC().CarrierGatewayID = cagw.CarrierGatewayId
+
+	// Make sure tags are up-to-date.
+	if err := wait.WaitForWithRetryable(wait.NewBackoff(), func() (bool, error) {
+		buildParams := s.getGatewayTagParams(*cagw.CarrierGatewayId)
+		tagsBuilder := tags.New(&buildParams, tags.WithEC2(s.EC2Client))
+		if err := tagsBuilder.Ensure(converters.TagsToMap(cagw.Tags)); err != nil {
+			return false, err
+		}
+		return true, nil
+	}, awserrors.InvalidCarrierGatewayNotFound); err != nil {
+		record.Warnf(s.scope.InfraCluster(), "FailedTagCarrierGateway", "Failed to tag managed Carrier Gateway %q: %v", cagw.CarrierGatewayId, err)
+		return errors.Wrapf(err, "failed to tag carrier gateway %q", *cagw.CarrierGatewayId)
+	}
+	conditions.MarkTrue(s.scope.InfraCluster(), infrav1.CarrierGatewayReadyCondition)
+	return nil
+}
+
+func (s *Service) deleteCarrierGateway() error {
+	if s.scope.VPC().IsUnmanaged(s.scope.Name()) {
+		s.scope.Trace("Skipping carrier gateway deletion in unmanaged mode")
+		return nil
+	}
+
+	cagw, err := s.describeVpcCarrierGateway()
+	if awserrors.IsNotFound(err) {
+		return nil