Merge pull request #4502 from wweiwei-li/aga-fix

k8s-ci-robot · web-flow · commit d4419fe02e26 · 2025-12-15T19:28:11.000-08:00
Add check for duplicate endpoints
diff --git a/apis/aga/v1beta1/globalaccelerator_types.go b/apis/aga/v1beta1/globalaccelerator_types.go
@@ -279,7 +279,7 @@ type IPSet struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
 // +kubebuilder:storageversion
-// +kubebuilder:printcolumn:name="NAME",type="string",JSONPath=".spec.name",description="The Global Accelerator name"
+// +kubebuilder:printcolumn:name="ACCELERATOR-NAME",type="string",JSONPath=".spec.name",description="The Global Accelerator name"
 // +kubebuilder:printcolumn:name="DNS-NAME",type="string",JSONPath=".status.dnsName",description="The Global Accelerator DNS name"
 // +kubebuilder:printcolumn:name="TYPE",type="string",JSONPath=".spec.type",description="The Global Accelerator type"
 // +kubebuilder:printcolumn:name="STATUS",type="string",JSONPath=".status.status",description="The Global Accelerator status"
diff --git a/config/crd/aga/aga-crds.yaml b/config/crd/aga/aga-crds.yaml
@@ -17,7 +17,7 @@ spec:
   - additionalPrinterColumns:
     - description: The Global Accelerator name
       jsonPath: .spec.name
-      name: NAME
+      name: ACCELERATOR-NAME
       type: string
     - description: The Global Accelerator DNS name
       jsonPath: .status.dnsName
diff --git a/config/crd/aga/aga.k8s.aws_globalaccelerators.yaml b/config/crd/aga/aga.k8s.aws_globalaccelerators.yaml
@@ -17,7 +17,7 @@ spec:
   - additionalPrinterColumns:
     - description: The Global Accelerator name
       jsonPath: .spec.name
-      name: NAME
+      name: ACCELERATOR-NAME
       type: string
     - description: The Global Accelerator DNS name
       jsonPath: .status.dnsName
diff --git a/helm/aws-load-balancer-controller/crds/aga-crds.yaml b/helm/aws-load-balancer-controller/crds/aga-crds.yaml
@@ -17,7 +17,7 @@ spec:
   - additionalPrinterColumns:
     - description: The Global Accelerator name
       jsonPath: .spec.name
-      name: NAME
+      name: ACCELERATOR-NAME
       type: string
     - description: The Global Accelerator DNS name
       jsonPath: .status.dnsName
diff --git a/helm/aws-load-balancer-controller/templates/webhook.yaml b/helm/aws-load-balancer-controller/templates/webhook.yaml
@@ -326,6 +326,36 @@ webhooks:
     - ingresses
   sideEffects: None
 {{- end }}
+{{- if .Values.controllerConfig.featureGates.GlobalAcceleratorController }}
+- clientConfig:
+    {{- if not $.Values.enableCertManager }}
+    caBundle: {{ $tls.caCert }}
+    {{- end }}
+    service:
+      name: {{ template "aws-load-balancer-controller.webhookService" . }}
+      namespace: {{ $.Release.Namespace }}
+      path: /validate-aga-k8s-aws-v1beta1-globalaccelerator
+      port: 443
+  failurePolicy: Fail
+  matchPolicy: Equivalent
+  name: vglobalaccelerator.aga.k8s.aws
+  admissionReviewVersions:
+  - v1beta1
+  namespaceSelector: {}
+  objectSelector: {}
+  rules:
+  - apiGroups:
+    - aga.k8s.aws
+    apiVersions:
+    - v1beta1
+    operations:
+    - CREATE
+    - UPDATE
+    resources:
+    - globalaccelerators
+  sideEffects: None
+  timeoutSeconds: 10
+{{- end }}
 ---
 {{- if not $.Values.enableCertManager }}
 apiVersion: v1
diff --git a/pkg/deploy/aga/accelerator_manager.go b/pkg/deploy/aga/accelerator_manager.go
@@ -118,7 +118,7 @@ func (m *defaultAcceleratorManager) Update(ctx context.Context, resAccelerator *
 
 	var updatedAccelerator *agatypes.Accelerator
 	if !m.isSDKAcceleratorSettingsDrifted(resAccelerator, sdkAccelerator) {
-		m.logger.Info("No drift detected in accelerator settings, skipping update",
+		m.logger.V(1).Info("No drift detected in accelerator settings, skipping update",
 			"stackID", resAccelerator.Stack().StackID(),
 			"resourceID", resAccelerator.ID(),
 			"acceleratorARN", *sdkAccelerator.Accelerator.AcceleratorArn)
diff --git a/pkg/deploy/aga/endpoint_group_manager.go b/pkg/deploy/aga/endpoint_group_manager.go
@@ -146,7 +146,7 @@ func (m *defaultEndpointGroupManager) buildSDKUpdateEndpointGroupInput(_ context
 func (m *defaultEndpointGroupManager) Update(ctx context.Context, resEndpointGroup *agamodel.EndpointGroup, sdkEndpointGroup *agatypes.EndpointGroup) (agamodel.EndpointGroupStatus, error) {
 	// Check if the endpoint group actually needs an update
 	if !m.isSDKEndpointGroupSettingsDrifted(resEndpointGroup, sdkEndpointGroup) {
-		m.logger.Info("No drift detected in endpoint group settings, skipping update",
+		m.logger.V(1).Info("No drift detected in endpoint group settings, skipping update",
 			"stackID", resEndpointGroup.Stack().StackID(),
 			"resourceID", resEndpointGroup.ID(),
 			"endpointGroupARN", *sdkEndpointGroup.EndpointGroupArn)
diff --git a/pkg/deploy/aga/listener_manager.go b/pkg/deploy/aga/listener_manager.go
@@ -131,7 +131,7 @@ func (m *defaultListenerManager) buildSDKUpdateListenerInput(ctx context.Context
 func (m *defaultListenerManager) Update(ctx context.Context, resListener *agamodel.Listener, sdkListener *ListenerResource) (agamodel.ListenerStatus, error) {
 	// Check if the listener actually needs an update
 	if !m.isSDKListenerSettingsDrifted(resListener, sdkListener) {
-		m.logger.Info("No drift detected in listener settings, skipping update",
+		m.logger.V(1).Info("No drift detected in listener settings, skipping update",
 			"stackID", resListener.Stack().StackID(),
 			"resourceID", resListener.ID(),
 			"listenerARN", *sdkListener.Listener.ListenerArn)
diff --git a/test/e2e/gateway/common_resource_stack.go b/test/e2e/gateway/common_resource_stack.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	elbv2gw "sigs.k8s.io/aws-load-balancer-controller/apis/gateway/v1beta1"
@@ -208,7 +209,14 @@ func deleteReferenceGrants(ctx context.Context, f *framework.Framework, refGrant
 
 func createGatewayClass(ctx context.Context, f *framework.Framework, gwc *gwv1.GatewayClass) error {
 	f.Logger.Info("creating gateway class", "gwc", k8s.NamespacedName(gwc))
-	return f.K8sClient.Create(ctx, gwc)
+	err := f.K8sClient.Create(ctx, gwc)
+	if err != nil && !apierrors.IsAlreadyExists(err) {
+		return err
+	}
+	if apierrors.IsAlreadyExists(err) {
+		f.Logger.Info("gateway class already exists", "gwc", k8s.NamespacedName(gwc))
+	}
+	return nil
 }
 
 func createLoadBalancerConfig(ctx context.Context, f *framework.Framework, lbc *elbv2gw.LoadBalancerConfiguration) error {
diff --git a/test/e2e/globalaccelerator/aga_verifier.go b/test/e2e/globalaccelerator/aga_verifier.go
@@ -47,6 +47,21 @@ type GlobalAcceleratorExpectation struct {
 	Listeners     []ListenerExpectation
 }
 
+func matchesListener(listener *types.Listener, expected ListenerExpectation) bool {
+	if expected.Protocol != "" && string(listener.Protocol) != expected.Protocol {
+		return false
+	}
+	if len(expected.PortRanges) > 0 {
+		if len(listener.PortRanges) == 0 {
+			return false
+		}
+		if awssdk.ToInt32(listener.PortRanges[0].FromPort) != expected.PortRanges[0].FromPort {
+			return false
+		}
+	}
+	return true
+}
+
 func verifyGlobalAcceleratorConfiguration(ctx context.Context, f *framework.Framework, acceleratorARN string, expected GlobalAcceleratorExpectation) error {
 	agaClient := f.Cloud.GlobalAccelerator()
 
@@ -79,8 +94,25 @@ func verifyGlobalAcceleratorConfiguration(ctx context.Context, f *framework.Fram
 			return fmt.Errorf("listener count mismatch: expected %d, got %d", len(expected.Listeners), len(listListenersResp.Listeners))
 		}
 
+		// Verify each expected listener exists (order-independent)
+		matched := make(map[int]bool)
 		for i, expectedListener := range expected.Listeners {
-			listener := listListenersResp.Listeners[i]
+			var listener *types.Listener
+			var matchedIdx int
+			for j := range listListenersResp.Listeners {
+				if !matched[j] && matchesListener(&listListenersResp.Listeners[j], expectedListener) {
+					listener = &listListenersResp.Listeners[j]
+					matchedIdx = j
+					break
+				}
+			}
+			if listener == nil {
+				if len(expectedListener.PortRanges) > 0 {
+					return fmt.Errorf("expected listener[%d] with port %d not found", i, expectedListener.PortRanges[0].FromPort)
+				}
+				return fmt.Errorf("expected listener[%d] not found", i)
+			}
+			matched[matchedIdx] = true
 
 			if expectedListener.Protocol != "" && string(listener.Protocol) != expectedListener.Protocol {
 				return fmt.Errorf("listener[%d] protocol mismatch: expected %s, got %s", i, expectedListener.Protocol, string(listener.Protocol))
diff --git a/test/e2e/globalaccelerator/gateway_endpoint_test.go b/test/e2e/globalaccelerator/gateway_endpoint_test.go
@@ -23,7 +23,7 @@ var _ = Describe("GlobalAccelerator with Gateway endpoint", func() {
 
 	BeforeEach(func() {
 		if !tf.Options.EnableAGATests || !tf.Options.EnableGatewayTests {
-			Skip("Skipping Global Accelerator Gateway tests (requires --enable-aga-tests and --enable-gateway-tests)")
+			Skip("Skipping Global Accelerator Gateway endpoint tests (requires --enable-aga-tests and --enable-gateway-tests)")
 		}
 		ctx = context.Background()
 	})
diff --git a/test/e2e/globalaccelerator/ingress_endpoint_test.go b/test/e2e/globalaccelerator/ingress_endpoint_test.go
@@ -11,6 +11,8 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	agav1beta1 "sigs.k8s.io/aws-load-balancer-controller/apis/aga/v1beta1"
 	"sigs.k8s.io/aws-load-balancer-controller/test/e2e/ingress"
 	"sigs.k8s.io/aws-load-balancer-controller/test/framework"
@@ -31,7 +33,7 @@ var _ = Describe("GlobalAccelerator with Ingress endpoint", func() {
 	BeforeEach(func() {
 		ctx = context.Background()
 		if !tf.Options.EnableAGATests {
-			Skip("Skipping Global Accelerator Gateway tests (requires --enable-aga-tests)")
+			Skip("Skipping Global Accelerator tests (requires --enable-aga-tests)")
 		}
 		ns, err := tf.NSManager.AllocateNamespace(ctx, "aga-ing-e2e")
 		Expect(err).NotTo(HaveOccurred())
@@ -60,6 +62,21 @@ var _ = Describe("GlobalAccelerator with Ingress endpoint", func() {
 			err := ingStack.Cleanup(ctx, tf)
 			Expect(err).NotTo(HaveOccurred())
 		}
+		if namespace != "" {
+			By("teardown namespace", func() {
+				ns := &corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{Name: namespace},
+				}
+				tf.Logger.Info("deleting namespace", "name", namespace)
+				err := tf.K8sClient.Delete(ctx, ns)
+				Expect(err).Should(SatisfyAny(BeNil(), Satisfy(apierrors.IsNotFound)))
+				tf.Logger.Info("deleted namespace", "name", namespace)
+				tf.Logger.Info("waiting namespace becomes deleted", "name", namespace)
+				err = tf.NSManager.WaitUntilNamespaceDeleted(ctx, ns)
+				Expect(err).NotTo(HaveOccurred())
+				tf.Logger.Info("namespace becomes deleted", "name", namespace)
+			})
+		}
 	})
 
 	Context("Basic Ingress endpoint with configuration verification", func() {
diff --git a/test/e2e/globalaccelerator/multi_endpoint_test.go b/test/e2e/globalaccelerator/multi_endpoint_test.go
@@ -31,7 +31,7 @@ var _ = Describe("GlobalAccelerator with multiple endpoint types", func() {
 
 	BeforeEach(func() {
 		if !tf.Options.EnableAGATests {
-			Skip("Skipping Global Accelerator Gateway tests (requires --enable-aga-tests)")
+			Skip("Skipping Global Accelerator tests (requires --enable-aga-tests)")
 		}
 		ctx = context.Background()
 		ns, err := tf.NSManager.AllocateNamespace(ctx, "aga-multi-e2e")
diff --git a/test/e2e/globalaccelerator/service_endpoint_test.go b/test/e2e/globalaccelerator/service_endpoint_test.go
@@ -24,7 +24,7 @@ var _ = Describe("GlobalAccelerator with Service endpoint", func() {
 
 	BeforeEach(func() {
 		if !tf.Options.EnableAGATests {
-			Skip("Skipping Global Accelerator Gateway tests (requires --enable-aga-tests)")
+			Skip("Skipping Global Accelerator tests (requires --enable-aga-tests)")
 		}
 		ctx = context.Background()
 	})
diff --git a/webhooks/aga/globalaccelerator_validator.go b/webhooks/aga/globalaccelerator_validator.go
@@ -39,6 +39,11 @@ func (v *globalAcceleratorValidator) Prototype(req admission.Request) (runtime.O
 func (v *globalAcceleratorValidator) ValidateCreate(_ context.Context, obj runtime.Object) error {
 	ga := obj.(*agaapi.GlobalAccelerator)
 
+	if err := v.checkForDuplicateEndpoints(ga); err != nil {
+		v.metricsCollector.ObserveWebhookValidationError(apiPathValidateAGAGlobalAccelerator, "checkForDuplicateEndpoints")
+		return err
+	}
+
 	if err := v.checkForOverlappingPortRanges(ga); err != nil {
 		v.metricsCollector.ObserveWebhookValidationError(apiPathValidateAGAGlobalAccelerator, "checkForOverlappingPortRanges")
 		return err
@@ -50,6 +55,11 @@ func (v *globalAcceleratorValidator) ValidateCreate(_ context.Context, obj runti
 func (v *globalAcceleratorValidator) ValidateUpdate(_ context.Context, obj runtime.Object, _ runtime.Object) error {
 	ga := obj.(*agaapi.GlobalAccelerator)
 
+	if err := v.checkForDuplicateEndpoints(ga); err != nil {
+		v.metricsCollector.ObserveWebhookValidationError(apiPathValidateAGAGlobalAccelerator, "checkForDuplicateEndpoints")
+		return err
+	}
+
 	if err := v.checkForOverlappingPortRanges(ga); err != nil {
 		v.metricsCollector.ObserveWebhookValidationError(apiPathValidateAGAGlobalAccelerator, "checkForOverlappingPortRanges")
 		return err
@@ -117,6 +127,59 @@ func portRangesOverlap(rangeA agaapi.PortRange, rangeB agaapi.PortRange) bool {
 	return rangeA.FromPort <= rangeB.ToPort && rangeA.ToPort >= rangeB.FromPort
 }
 
+// checkForDuplicateEndpoints validates that there are no duplicate endpoints within each endpoint group
+func (v *globalAcceleratorValidator) checkForDuplicateEndpoints(ga *agaapi.GlobalAccelerator) error {
+	if ga.Spec.Listeners == nil {
+		return nil
+	}
+
+	for listenerIdx, listener := range *ga.Spec.Listeners {
+		if listener.EndpointGroups == nil {
+			continue
+		}
+
+		for groupIdx, group := range *listener.EndpointGroups {
+			if group.Endpoints == nil || len(*group.Endpoints) == 0 {
+				continue
+			}
+
+			seen := make(map[string]bool)
+			for _, endpoint := range *group.Endpoints {
+				key := getEndpointKey(endpoint, ga.Namespace)
+				if seen[key] {
+					return errors.Errorf(
+						"duplicate endpoint detected in listener[%d].endpointGroups[%d]: %s",
+						listenerIdx, groupIdx, key)
+				}
+				seen[key] = true
+			}
+		}
+	}
+
+	return nil
+}
+
+// getEndpointKey generates a unique key for an endpoint
+func getEndpointKey(endpoint agaapi.GlobalAcceleratorEndpoint, defaultNamespace string) string {
+	namespace := defaultNamespace
+	if endpoint.Namespace != nil {
+		namespace = *endpoint.Namespace
+	}
+	name := ""
+	if endpoint.Name != nil {
+		name = *endpoint.Name
+	}
+
+	if endpoint.Type == agaapi.GlobalAcceleratorEndpointTypeEndpointID {
+		endpointID := ""
+		if endpoint.EndpointID != nil {
+			endpointID = *endpoint.EndpointID
+		}
+		return string(endpoint.Type) + "/" + endpointID
+	}
+	return string(endpoint.Type) + "/" + namespace + "/" + name
+}
+
 // +kubebuilder:webhook:path=/validate-aga-k8s-aws-v1beta1-globalaccelerator,mutating=false,failurePolicy=fail,groups=aga.k8s.aws,resources=globalaccelerators,verbs=create;update,versions=v1beta1,name=vglobalaccelerator.aga.k8s.aws,sideEffects=None,matchPolicy=Equivalent,webhookVersions=v1,admissionReviewVersions=v1beta1
 
 func (v *globalAcceleratorValidator) SetupWithManager(mgr ctrl.Manager) {
diff --git a/webhooks/aga/globalaccelerator_validator_test.go b/webhooks/aga/globalaccelerator_validator_test.go
