[feat aga] Implement endpoint group deployer with port override conflict resolution

Author: shraddhabang

controllers/aga/globalaccelerator_controller.go

@@ -282,8 +282,8 @@ func (r *globalAcceleratorReconciler) reconcileGlobalAcceleratorResources(ctx co
 	_, fatalErrors := r.endpointLoader.LoadEndpoints(ctx, ga, endpoints)
 	if len(fatalErrors) > 0 {
 		err := fmt.Errorf("failed to load endpoints: %v", fatalErrors[0])
-		r.logger.Error(err, "Fatal error loading endpoints")
-
+		r.eventRecorder.Event(ga, corev1.EventTypeWarning, k8s.GlobalAcceleratorEventReasonFailedEndpointLoad, fmt.Sprintf("Failed to reconcile due to %v", err))
+		r.logger.Error(err, fmt.Sprintf("fatal error loading endpoints for %v", k8s.NamespacedName(ga)))
 		// Handle other endpoint loading errors
 		if statusErr := r.statusUpdater.UpdateStatusFailure(ctx, ga, agadeploy.EndpointLoadFailed, err.Error()); statusErr != nil {
 			r.logger.Error(statusErr, "Failed to update GlobalAccelerator status after endpoint load failure")
@@ -315,7 +315,7 @@ func (r *globalAcceleratorReconciler) reconcileGlobalAcceleratorResources(ctx co
 	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))
-
+		r.logger.Error(err, fmt.Sprintf("Failed to deploy stack for: %v", k8s.NamespacedName(ga)))
 		// 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")

pkg/aws/services/globalaccelerator.go

@@ -41,6 +41,21 @@ type GlobalAccelerator interface {
 	// ListListenersForAccelerator lists all listeners for an accelerator.
 	ListListenersForAcceleratorWithContext(ctx context.Context, input *globalaccelerator.ListListenersInput) (*globalaccelerator.ListListenersOutput, error)
 
+	// CreateEndpointGroup creates a new endpoint group.
+	CreateEndpointGroupWithContext(ctx context.Context, input *globalaccelerator.CreateEndpointGroupInput) (*globalaccelerator.CreateEndpointGroupOutput, error)
+
+	// DescribeEndpointGroup describes an endpoint group.
+	DescribeEndpointGroupWithContext(ctx context.Context, input *globalaccelerator.DescribeEndpointGroupInput) (*globalaccelerator.DescribeEndpointGroupOutput, error)
+
+	// UpdateEndpointGroup updates an endpoint group.
+	UpdateEndpointGroupWithContext(ctx context.Context, input *globalaccelerator.UpdateEndpointGroupInput) (*globalaccelerator.UpdateEndpointGroupOutput, error)
+
+	// DeleteEndpointGroup deletes an endpoint group.
+	DeleteEndpointGroupWithContext(ctx context.Context, input *globalaccelerator.DeleteEndpointGroupInput) (*globalaccelerator.DeleteEndpointGroupOutput, error)
+
+	// wrapper to ListEndpointGroups API, which aggregates paged results into list.
+	ListEndpointGroupsAsList(ctx context.Context, input *globalaccelerator.ListEndpointGroupsInput) ([]types.EndpointGroup, error)
+
 	// TagResource tags a resource.
 	TagResourceWithContext(ctx context.Context, input *globalaccelerator.TagResourceInput) (*globalaccelerator.TagResourceOutput, error)
 
@@ -192,3 +207,52 @@ func (c *defaultGlobalAccelerator) ListListenersAsList(ctx context.Context, inpu
 	}
 	return result, nil
 }
+
+func (c *defaultGlobalAccelerator) CreateEndpointGroupWithContext(ctx context.Context, input *globalaccelerator.CreateEndpointGroupInput) (*globalaccelerator.CreateEndpointGroupOutput, error) {
+	client, err := c.awsClientsProvider.GetGlobalAcceleratorClient(ctx, "CreateEndpointGroup")
+	if err != nil {
+		return nil, err
+	}
+	return client.CreateEndpointGroup(ctx, input)
+}
+
+func (c *defaultGlobalAccelerator) DescribeEndpointGroupWithContext(ctx context.Context, input *globalaccelerator.DescribeEndpointGroupInput) (*globalaccelerator.DescribeEndpointGroupOutput, error) {
+	client, err := c.awsClientsProvider.GetGlobalAcceleratorClient(ctx, "DescribeEndpointGroup")
+	if err != nil {
+		return nil, err
+	}
+	return client.DescribeEndpointGroup(ctx, input)
+}
+
+func (c *defaultGlobalAccelerator) UpdateEndpointGroupWithContext(ctx context.Context, input *globalaccelerator.UpdateEndpointGroupInput) (*globalaccelerator.UpdateEndpointGroupOutput, error) {
+	client, err := c.awsClientsProvider.GetGlobalAcceleratorClient(ctx, "UpdateEndpointGroup")
+	if err != nil {
+		return nil, err
+	}
+	return client.UpdateEndpointGroup(ctx, input)
+}
+
+func (c *defaultGlobalAccelerator) DeleteEndpointGroupWithContext(ctx context.Context, input *globalaccelerator.DeleteEndpointGroupInput) (*globalaccelerator.DeleteEndpointGroupOutput, error) {
+	client, err := c.awsClientsProvider.GetGlobalAcceleratorClient(ctx, "DeleteEndpointGroup")
+	if err != nil {
+		return nil, err
+	}
+	return client.DeleteEndpointGroup(ctx, input)
+}
+
+func (c *defaultGlobalAccelerator) ListEndpointGroupsAsList(ctx context.Context, input *globalaccelerator.ListEndpointGroupsInput) ([]types.EndpointGroup, error) {
+	var result []types.EndpointGroup
+	client, err := c.awsClientsProvider.GetGlobalAcceleratorClient(ctx, "ListEndpointGroups")
+	if err != nil {
+		return nil, err
+	}
+	paginator := globalaccelerator.NewListEndpointGroupsPaginator(client, input)
+	for paginator.HasMorePages() {
+		output, err := paginator.NextPage(ctx)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, output.EndpointGroups...)
+	}
+	return result, nil
+}