Merge pull request #4466 from zac-nixon/AGAController

Merge AGAController branch into main

Author: k8s-ci-robot

apis/aga/v1beta1/globalaccelerator_types.go

@@ -48,6 +48,7 @@ const (
 )
 
 // PortRange defines the port range for Global Accelerator listeners.
+// +kubebuilder:validation:XValidation:rule="self.fromPort <= self.toPort",message="FromPort must be less than or equal to ToPort"
 type PortRange struct {
 	// FromPort is the first port in the range of ports, inclusive.
 	// +kubebuilder:validation:Minimum=1

config/crd/aga/aga-crds.yaml

@@ -264,6 +264,9 @@ spec:
                         - fromPort
                         - toPort
                         type: object
+                        x-kubernetes-validations:
+                        - message: FromPort must be less than or equal to ToPort
+                          rule: self.fromPort <= self.toPort
                       maxItems: 10
                       minItems: 1
                       type: array

config/crd/aga/aga.k8s.aws_globalaccelerators.yaml

@@ -264,6 +264,9 @@ spec:
                         - fromPort
                         - toPort
                         type: object
+                        x-kubernetes-validations:
+                        - message: FromPort must be less than or equal to ToPort
+                          rule: self.fromPort <= self.toPort
                       maxItems: 10
                       minItems: 1
                       type: array

config/webhook/globalaccelerator_validator_patch.yaml

@@ -0,0 +1,18 @@
+# This patch adds the GlobalAccelerator validator webhook configuration to the webhook configurations
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingWebhookConfiguration
+metadata:
+  name: webhook-configuration
+webhooks:
+  - name: vglobalaccelerator.aga.k8s.aws
+    rules:
+      - apiGroups:
+          - "aga.k8s.aws"
+        apiVersions:
+          - v1beta1
+        operations:
+          - CREATE
+          - UPDATE
+        resources:
+          - globalaccelerators
+        scope: "Namespaced"

config/webhook/kustomization.yaml

@@ -9,3 +9,4 @@ patchesStrategicMerge:
   - pod_mutator_patch.yaml
   - service_mutator_patch.yaml
   - ingressclassparams_validator_patch.yaml
+  - globalaccelerator_validator_patch.yaml

config/webhook/manifests.yaml

@@ -125,6 +125,27 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: webhook
 webhooks:
+  - admissionReviewVersions:
+      - v1beta1
+    clientConfig:
+      service:
+        name: webhook-service
+        namespace: system
+        path: /validate-aga-k8s-aws-v1beta1-globalaccelerator
+    failurePolicy: Fail
+    matchPolicy: Equivalent
+    name: vglobalaccelerator.aga.k8s.aws
+    rules:
+      - apiGroups:
+          - aga.k8s.aws
+        apiVersions:
+          - v1beta1
+        operations:
+          - CREATE
+          - UPDATE
+        resources:
+          - globalaccelerators
+    sideEffects: None
   - admissionReviewVersions:
       - v1beta1
     clientConfig:

controllers/aga/globalaccelerator_controller.go

@@ -19,11 +19,18 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/service/globalaccelerator/types"
 	"github.com/go-logr/logr"
+	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/types"
+	ktypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/services"
+	agadeploy "sigs.k8s.io/aws-load-balancer-controller/pkg/deploy/aga"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/shared_constants"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -42,6 +49,7 @@ import (
 	agamodel "sigs.k8s.io/aws-load-balancer-controller/pkg/model/aga"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/model/core"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/runtime"
+	agastatus "sigs.k8s.io/aws-load-balancer-controller/pkg/status/aga"
 )
 
 const (
@@ -52,24 +60,30 @@ const (
 	agaResourcesGroupVersion = "aga.k8s.aws/v1beta1"
 	globalAcceleratorKind    = "GlobalAccelerator"
 
+	// Requeue constants for provisioning state monitoring
+	requeueMessage          = "Monitoring provisioning state"
+	statusUpdateRequeueTime = 1 * time.Minute
+
 	// Metric stage constants
 	MetricStageFetchGlobalAccelerator     = "fetch_globalAccelerator"
 	MetricStageAddFinalizers              = "add_finalizers"
 	MetricStageBuildModel                 = "build_model"
+	MetricStageDeployStack                = "deploy_stack"
 	MetricStageReconcileGlobalAccelerator = "reconcile_globalaccelerator"
 
 	// Metric error constants
 	MetricErrorAddFinalizers              = "add_finalizers_error"
 	MetricErrorRemoveFinalizers           = "remove_finalizers_error"
 	MetricErrorBuildModel                 = "build_model_error"
+	MetricErrorDeployStack                = "deploy_stack_error"
 	MetricErrorReconcileGlobalAccelerator = "reconcile_globalaccelerator_error"
 )
 
 // NewGlobalAcceleratorReconciler constructs new globalAcceleratorReconciler
-func NewGlobalAcceleratorReconciler(k8sClient client.Client, eventRecorder record.EventRecorder, finalizerManager k8s.FinalizerManager, config config.ControllerConfig, logger logr.Logger, metricsCollector lbcmetrics.MetricCollector, reconcileCounters *metricsutil.ReconcileCounters) *globalAcceleratorReconciler {
+func NewGlobalAcceleratorReconciler(k8sClient client.Client, eventRecorder record.EventRecorder, finalizerManager k8s.FinalizerManager, config config.ControllerConfig, cloud services.Cloud, logger logr.Logger, metricsCollector lbcmetrics.MetricCollector, reconcileCounters *metricsutil.ReconcileCounters) *globalAcceleratorReconciler {
 
 	// Create tracking provider
-	trackingProvider := tracking.NewDefaultProvider(agaTagPrefix, config.ClusterName)
+	trackingProvider := tracking.NewDefaultProvider(agaTagPrefix, config.ClusterName, tracking.WithRegion(config.AWSConfig.Region))
 
 	// Create model builder
 	agaModelBuilder := aga.NewDefaultModelBuilder(
@@ -78,6 +92,7 @@ func NewGlobalAcceleratorReconciler(k8sClient client.Client, eventRecorder recor
 		trackingProvider,
 		config.FeatureGates,
 		config.ClusterName,
+		config.AWSConfig.Region,
 		config.DefaultTags,
 		config.ExternalManagedTags,
 		logger.WithName("aga-model-builder"),
@@ -87,17 +102,26 @@ func NewGlobalAcceleratorReconciler(k8sClient client.Client, eventRecorder recor
 	// Create stack marshaller
 	stackMarshaller := deploy.NewDefaultStackMarshaller()
 
+	// Create AGA stack deployer
+	stackDeployer := agadeploy.NewDefaultStackDeployer(cloud, config, trackingProvider, logger.WithName("aga-stack-deployer"), metricsCollector, controllerName)
+
+	// Create status updater
+	statusUpdater := agastatus.NewStatusUpdater(k8sClient, logger)
+
 	return &globalAcceleratorReconciler{
 		k8sClient:        k8sClient,
 		eventRecorder:    eventRecorder,
 		finalizerManager: finalizerManager,
 		logger:           logger,
 		modelBuilder:     agaModelBuilder,
 		stackMarshaller:  stackMarshaller,
+		stackDeployer:    stackDeployer,
+		statusUpdater:    statusUpdater,
 		metricsCollector: metricsCollector,
 		reconcileTracker: reconcileCounters.IncrementAGA,
 
-		maxConcurrentReconciles: config.GlobalAcceleratorMaxConcurrentReconciles,
+		maxConcurrentReconciles:    config.GlobalAcceleratorMaxConcurrentReconciles,
+		maxExponentialBackoffDelay: config.GlobalAcceleratorMaxExponentialBackoffDelay,
 	}
 }
 
@@ -108,11 +132,14 @@ type globalAcceleratorReconciler struct {
 	finalizerManager k8s.FinalizerManager
 	modelBuilder     aga.ModelBuilder
 	stackMarshaller  deploy.StackMarshaller
+	stackDeployer    agadeploy.StackDeployer
+	statusUpdater    agastatus.StatusUpdater
 	logger           logr.Logger
 	metricsCollector lbcmetrics.MetricCollector
-	reconcileTracker func(namespaceName types.NamespacedName)
+	reconcileTracker func(namespaceName ktypes.NamespacedName)
 
-	maxConcurrentReconciles int
+	maxConcurrentReconciles    int
+	maxExponentialBackoffDelay time.Duration
 }
 
 // +kubebuilder:rbac:groups=aga.k8s.aws,resources=globalaccelerators,verbs=get;list;watch;patch
@@ -155,26 +182,19 @@ func (r *globalAcceleratorReconciler) reconcileGlobalAccelerator(ctx context.Con
 		return ctrlerrors.NewErrorWithMetrics(controllerName, MetricErrorAddFinalizers, err, r.metricsCollector)
 	}
 
-	// TODO: Implement GlobalAccelerator resource management
-	// This would include:
-	// 1. Creating/updating AWS Global Accelerator
-	// 2. Managing listeners and endpoint groups
-	// 3. Handling endpoint discovery from Services/Ingresses/Gateways
 	reconcileResourceFn := func() {
 		err = r.reconcileGlobalAcceleratorResources(ctx, ga)
 	}
 	r.metricsCollector.ObserveControllerReconcileLatency(controllerName, MetricStageReconcileGlobalAccelerator, reconcileResourceFn)
 	if err != nil {
 		return ctrlerrors.NewErrorWithMetrics(controllerName, MetricErrorReconcileGlobalAccelerator, err, r.metricsCollector)
 	}
-
-	r.eventRecorder.Event(ga, corev1.EventTypeNormal, k8s.GlobalAcceleratorEventReasonSuccessfullyReconciled, "Successfully reconciled")
 	return nil
 }
 
 func (r *globalAcceleratorReconciler) cleanupGlobalAccelerator(ctx context.Context, ga *agaapi.GlobalAccelerator) error {
 	if k8s.HasFinalizer(ga, shared_constants.GlobalAcceleratorFinalizer) {
-		// TODO: Implement cleanup logic for AWS Global Accelerator resources
+		// TODO: Implement cleanup logic for AWS Global Accelerator resources (Only cleaning up accelerator for now)
 		if err := r.cleanupGlobalAcceleratorResources(ctx, ga); err != nil {
 			r.eventRecorder.Event(ga, corev1.EventTypeWarning, k8s.GlobalAcceleratorEventReasonFailedCleanup, fmt.Sprintf("Failed cleanup due to %v", err))
 			return err
@@ -203,7 +223,7 @@ func (r *globalAcceleratorReconciler) buildModel(ctx context.Context, ga *agaapi
 }
 
 func (r *globalAcceleratorReconciler) reconcileGlobalAcceleratorResources(ctx context.Context, ga *agaapi.GlobalAccelerator) error {
-	r.logger.Info("Reconciling GlobalAccelerator resources", "name", ga.Name, "namespace", ga.Namespace)
+	r.logger.Info("Reconciling GlobalAccelerator resources", "globalAccelerator", k8s.NamespacedName(ga))
 	var stack core.Stack
 	var accelerator *agamodel.Accelerator
 	var err error
@@ -212,25 +232,93 @@ func (r *globalAcceleratorReconciler) reconcileGlobalAcceleratorResources(ctx co
 	}
 	r.metricsCollector.ObserveControllerReconcileLatency(controllerName, MetricStageBuildModel, buildModelFn)
 	if err != nil {
+		// Update status to indicate model building failure
+		if statusErr := r.statusUpdater.UpdateStatusFailure(ctx, ga, agadeploy.ModelBuildFailed, fmt.Sprintf("Failed to build model: %v", err)); statusErr != nil {
+			r.logger.Error(statusErr, "Failed to update GlobalAccelerator status after model build failure")
+		}
 		return ctrlerrors.NewErrorWithMetrics(controllerName, MetricErrorBuildModel, err, r.metricsCollector)
 	}
 
-	// Log the built model for debugging
-	r.logger.Info("Built model successfully", "accelerator", accelerator.ID(), "stackID", stack.StackID())
+	// Deploy the stack to create/update AWS Global Accelerator resources
+	deployStackFn := func() {
+		err = r.stackDeployer.Deploy(ctx, stack, r.metricsCollector, controllerName)
+	}
+	r.metricsCollector.ObserveControllerReconcileLatency(controllerName, MetricStageDeployStack, deployStackFn)
+	if err != nil {
+		r.eventRecorder.Event(ga, corev1.EventTypeWarning, k8s.GlobalAcceleratorEventReasonFailedDeploy, fmt.Sprintf("Failed to deploy stack due to %v", err))
 
-	// TODO: Implement the deploy phase
-	// This would include:
-	// 1. Deploy the stack to create/update AWS Global Accelerator resources
-	// 2. Update the GlobalAccelerator status with the created resources
-	// 3. Handle any deployment errors and update status accordingly
+		// Update status to indicate deployment failure
+		if statusErr := r.statusUpdater.UpdateStatusFailure(ctx, ga, agadeploy.DeploymentFailed, fmt.Sprintf("Failed to deploy stack: %v", err)); statusErr != nil {
+			r.logger.Error(statusErr, "Failed to update GlobalAccelerator status after deployment failure")
+		}
+
+		return ctrlerrors.NewErrorWithMetrics(controllerName, MetricErrorDeployStack, err, r.metricsCollector)
+	}
+
+	r.logger.Info("Successfully deployed GlobalAccelerator stack", "stackID", stack.StackID())
+
+	// Update GlobalAccelerator status after successful deployment
+	requeueNeeded, err := r.statusUpdater.UpdateStatusSuccess(ctx, ga, accelerator)
+	if err != nil {
+		r.eventRecorder.Event(ga, corev1.EventTypeWarning, k8s.GlobalAcceleratorEventReasonFailedUpdateStatus, fmt.Sprintf("Failed update status due to %v", err))
+		return err
+	}
+	if requeueNeeded {
+		return ctrlerrors.NewRequeueNeededAfter(requeueMessage, statusUpdateRequeueTime)
+	}
+
+	r.eventRecorder.Event(ga, corev1.EventTypeNormal, k8s.GlobalAcceleratorEventReasonSuccessfullyReconciled, "Successfully reconciled")
 
 	return nil
 }
 
 func (r *globalAcceleratorReconciler) cleanupGlobalAcceleratorResources(ctx context.Context, ga *agaapi.GlobalAccelerator) error {
-	// TODO: Implement the actual AWS Global Accelerator resource cleanup
-	// This is a placeholder implementation
-	r.logger.Info("Cleaning up GlobalAccelerator resources", "name", ga.Name, "namespace", ga.Namespace)
+	r.logger.Info("Cleaning up GlobalAccelerator resources", "globalAccelerator", k8s.NamespacedName(ga))
+
+	// Our enhanced AcceleratorManager now handles deletion of listeners before accelerator.
+	// TODO: This will be enhanced to delete endpoint groups and endpoints
+	// before deleting listeners and accelerator (when those features are implemented)
+	// 1. Find the accelerator ARN from the CRD status
+	if ga.Status.AcceleratorARN == nil {
+		r.logger.Info("No accelerator ARN found in status, nothing to clean up", "globalAccelerator", k8s.NamespacedName(ga))
+		return nil
+	}
+
+	acceleratorARN := *ga.Status.AcceleratorARN
+	if acceleratorARN == "" {
+		r.logger.Info("Empty accelerator ARN in status, nothing to clean up", "globalAccelerator", k8s.NamespacedName(ga))
+		return nil
+	}
+
+	// 2. Delete the accelerator using accelerator delete manager
+	acceleratorManager := r.stackDeployer.GetAcceleratorManager()
+	r.logger.Info("Deleting accelerator", "acceleratorARN", acceleratorARN, "globalAccelerator", k8s.NamespacedName(ga))
+
+	// Initialize reference to existing accelerator for deletion
+	acceleratorWithTags := agadeploy.AcceleratorWithTags{
+		Accelerator: &types.Accelerator{
+			AcceleratorArn: &acceleratorARN,
+		},
+		Tags: nil,
+	}
+
+	if err := acceleratorManager.Delete(ctx, acceleratorWithTags); err != nil {
+		// Check if it's an AcceleratorNotDisabledError
+		var notDisabledErr *agadeploy.AcceleratorNotDisabledError
+		if errors.As(err, &notDisabledErr) {
+			// Update status to indicate we're waiting for the accelerator to be disabled
+			if updateErr := r.statusUpdater.UpdateStatusDeletion(ctx, ga); updateErr != nil {
+				r.logger.Error(updateErr, "Failed to update status during accelerator deletion")
+			}
+			return ctrlerrors.NewRequeueNeeded("Waiting for accelerator to be disabled")
+		}
+
+		// Any other error
+		r.logger.Error(err, "Failed to delete accelerator", "acceleratorARN", acceleratorARN, "globalAccelerator", k8s.NamespacedName(ga))
+		return fmt.Errorf("failed to delete accelerator %s: %w", acceleratorARN, err)
+	}
+
+	r.logger.Info("Successfully cleaned up all GlobalAccelerator resources", "globalAccelerator", k8s.NamespacedName(ga))
 	return nil
 }
 
@@ -259,6 +347,7 @@ func (r *globalAcceleratorReconciler) SetupWithManager(ctx context.Context, mgr
 		Named(controllerName).
 		WithOptions(controller.Options{
 			MaxConcurrentReconciles: r.maxConcurrentReconciles,
+			RateLimiter:             workqueue.NewTypedItemExponentialFailureRateLimiter[reconcile.Request](5*time.Second, r.maxExponentialBackoffDelay),
 		}).
 		Complete(r)
 }