Address PR comments and add E2E test

Author: Doyoon Kim

pkg/deploy/lattice/targets_manager.go

@@ -92,7 +92,7 @@ func (s *defaultTargetsManager) findStaleTargets(
 			TargetIP: aws.StringValue(target.Id),
 			Port:     aws.Int64Value(target.Port),
 		}
-		if !modelSet.Contains(ipPort) {
+		if aws.StringValue(target.Status) != vpclattice.TargetStatusDraining && !modelSet.Contains(ipPort) {
 			staleTargets = append(staleTargets, ipPort)
 		}
 	}

pkg/deploy/lattice/targets_synthesizer.go

@@ -8,15 +8,15 @@ import (
 	model "github.com/aws/aws-application-networking-k8s/pkg/model/lattice"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
+	"github.com/aws/aws-application-networking-k8s/pkg/webhook"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/vpclattice"
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 const (
-	// TODO: use the constant on webhook side instead.
-	LatticeReadinessGateConditionType = "aws-application-networking-k8s/pod-readiness-gate"
+	LatticeReadinessGateConditionType = webhook.PodReadinessGateConditionType
 
 	ReadinessReasonHealthy                = "Healthy"
 	ReadinessReasonUnhealthy              = "Unhealthy"
@@ -125,66 +125,67 @@ func (t *targetsSynthesizer) syncStatus(ctx context.Context, modelTargets []mode
 
 	var requeue bool
 	for _, target := range modelTargets {
-		// Step 0: Check if the endpoint is not ready yet.
-		if !target.Ready && target.TargetRef.Name != "" {
-			pod := &corev1.Pod{}
-			t.client.Get(ctx, target.TargetRef, pod)
-
-			// Step 1: Check if the pod has the readiness gate spec.
-			if !utils.PodHasReadinessGate(pod, LatticeReadinessGateConditionType) {
-				continue
-			}
+		// Step 0: Check if the endpoint has a valid target, and is not ready yet.
+		if target.Ready || target.TargetRef.Name == "" {
+			continue
+		}
 
-			// Step 2: Check if the pod readiness condition exists with specific condition type.
-			// The condition is considered false when it does not exist.
-			cond := utils.FindPodStatusCondition(pod.Status.Conditions, LatticeReadinessGateConditionType)
-			if cond != nil && cond.Status == corev1.ConditionTrue {
-				continue
-			}
+		// Step 1: Check if the pod has the readiness gate spec.
+		pod := &corev1.Pod{}
+		t.client.Get(ctx, target.TargetRef, pod)
+		if !utils.PodHasReadinessGate(pod, LatticeReadinessGateConditionType) {
+			continue
+		}
 
-			// Step 3: Check if the Lattice target is healthy.
-			newCond := corev1.PodCondition{
-				Type:   LatticeReadinessGateConditionType,
-				Status: corev1.ConditionFalse,
-			}
-			targetIpPort := model.Target{
-				TargetIP: target.TargetIP,
-				Port:     target.Port,
-			}
-			// syncStatus is called at post synthesis, so we can assume:
-			// 1. Target for the pod (eventually) exists. If the target doesn't exist, we can simply requeue.
-			// 2. Target group will be always in use, except for ServiceExport TGs.
-			if latticeTarget, ok := latticeTargetMap[targetIpPort]; ok {
-				switch status := aws.StringValue(latticeTarget.Status); status {
-				case vpclattice.TargetStatusHealthy:
-					newCond.Status = corev1.ConditionTrue
-					newCond.Reason = ReadinessReasonHealthy
-				case vpclattice.TargetStatusUnavailable:
-					// Lattice HC not turned on. Readiness is designed to work only with HC but do not block deployment on this case.
-					newCond.Status = corev1.ConditionTrue
-					newCond.Reason = ReadinessReasonHealthCheckUnavailable
-				case vpclattice.TargetStatusUnused:
-					// Since this logic is called after HTTPRoute is wired, this only happens for ServiceExport TGs.
-					// In this case we do not have to evaluate them as Healthy, but we also do not have to requeue.
-					newCond.Reason = ReadinessReasonUnused
-				case vpclattice.TargetStatusInitial:
-					requeue = true
-					newCond.Reason = ReadinessReasonInitial
-				default:
-					requeue = true
-					newCond.Reason = ReadinessReasonUnhealthy
-					newCond.Message = fmt.Sprintf("Target health check status: %s", status)
-				}
-			} else {
+		// Step 2: Check if the pod readiness condition exists with specific condition type.
+		// The condition is considered false when it does not exist.
+		cond := utils.FindPodStatusCondition(pod.Status.Conditions, LatticeReadinessGateConditionType)
+		if cond != nil && cond.Status == corev1.ConditionTrue {
+			continue
+		}
+
+		// Step 3: Check if the Lattice target is healthy.
+		newCond := corev1.PodCondition{
+			Type:   LatticeReadinessGateConditionType,
+			Status: corev1.ConditionFalse,
+		}
+		targetIpPort := model.Target{
+			TargetIP: target.TargetIP,
+			Port:     target.Port,
+		}
+		// syncStatus is called at post synthesis, so we can assume:
+		// 1. Target for the pod (eventually) exists. If the target doesn't exist, we can simply requeue.
+		// 2. Target group will be always in use, except for ServiceExport TGs.
+		if latticeTarget, ok := latticeTargetMap[targetIpPort]; ok {
+			switch status := aws.StringValue(latticeTarget.Status); status {
+			case vpclattice.TargetStatusHealthy:
+				newCond.Status = corev1.ConditionTrue
+				newCond.Reason = ReadinessReasonHealthy
+			case vpclattice.TargetStatusUnavailable:
+				// Lattice HC not turned on. Readiness is designed to work only with HC but do not block deployment on this case.
+				newCond.Status = corev1.ConditionTrue
+				newCond.Reason = ReadinessReasonHealthCheckUnavailable
+			case vpclattice.TargetStatusUnused:
+				// Since this logic is called after HTTPRoute is wired, this only happens for ServiceExport TGs.
+				// In this case we do not have to evaluate them as Healthy, but we also do not have to requeue.
+				newCond.Reason = ReadinessReasonUnused
+			case vpclattice.TargetStatusInitial:
 				requeue = true
-				newCond.Reason = ReadinessReasonTargetNotFound
+				newCond.Reason = ReadinessReasonInitial
+			default:
+				requeue = true
+				newCond.Reason = ReadinessReasonUnhealthy
+				newCond.Message = fmt.Sprintf("Target health check status: %s", status)
 			}
+		} else {
+			requeue = true
+			newCond.Reason = ReadinessReasonTargetNotFound
+		}
 
-			// Step 4: Update status.
-			utils.SetPodStatusCondition(&pod.Status.Conditions, newCond)
-			if err := t.client.Status().Update(ctx, pod); err != nil {
-				return requeue, err
-			}
+		// Step 4: Update status.
+		utils.SetPodStatusCondition(&pod.Status.Conditions, newCond)
+		if err := t.client.Status().Update(ctx, pod); err != nil {
+			return requeue, err
 		}
 	}
 	return requeue, nil

test/pkg/test/nginxapp.go

@@ -1,6 +1,7 @@
 package test
 
 import (
+	"github.com/aws/aws-application-networking-k8s/pkg/deploy/lattice"
 	"github.com/samber/lo"
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
@@ -18,6 +19,7 @@ type ElasticSearchOptions struct {
 	TargetPort2         int
 	MergeFromDeployment []*appsv1.Deployment
 	MergeFromService    []*v1.Service
+	ReadinessGate       bool
 }
 
 func (env *Framework) NewNginxApp(options ElasticSearchOptions) (*appsv1.Deployment, *v1.Service) {
@@ -33,6 +35,12 @@ func (env *Framework) NewNginxApp(options ElasticSearchOptions) (*appsv1.Deploym
 	if options.TargetPort2 == 0 {
 		options.TargetPort2 = 9114
 	}
+	var readinessGates []v1.PodReadinessGate
+	if options.ReadinessGate {
+		readinessGates = append(readinessGates, v1.PodReadinessGate{
+			ConditionType: lattice.LatticeReadinessGateConditionType,
+		})
+	}
 	deployment := New(&appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: options.Namespace,
@@ -53,6 +61,7 @@ func (env *Framework) NewNginxApp(options ElasticSearchOptions) (*appsv1.Deploym
 					},
 				},
 				Spec: v1.PodSpec{
+					ReadinessGates: readinessGates,
 					Containers: []v1.Container{
 						{
 							Name:  options.Name,

test/suites/integration/readiness_gate_test.go

@@ -0,0 +1,66 @@
+package integration
+
+import (
+	"github.com/aws/aws-application-networking-k8s/pkg/deploy/lattice"
+	"github.com/aws/aws-application-networking-k8s/pkg/k8s"
+	"github.com/aws/aws-application-networking-k8s/pkg/utils"
+	"github.com/aws/aws-application-networking-k8s/test/pkg/test"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/samber/lo"
+	appsv1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	gwv1 "sigs.k8s.io/gateway-api/apis/v1"
+)
+
+var _ = Describe("Pod Readiness Gate", Ordered, func() {
+	var (
+		deployment *appsv1.Deployment
+		service    *v1.Service
+		route      *gwv1.HTTPRoute
+	)
+
+	BeforeAll(func() {
+		deployment, service = testFramework.NewNginxApp(test.ElasticSearchOptions{
+			Name:          "pod-test",
+			Namespace:     k8snamespace,
+			ReadinessGate: true,
+		})
+		route = testFramework.NewHttpRoute(testGateway, service, service.Kind)
+		testFramework.ExpectCreated(ctx, deployment, service, route)
+	})
+
+	It("updates condition when injected", func() {
+		Eventually(func(g Gomega) {
+			pods := testFramework.GetPodsByDeploymentName(deployment.Name, deployment.Namespace)
+			for _, pod := range pods {
+				cond := utils.FindPodStatusCondition(pod.Status.Conditions, lattice.LatticeReadinessGateConditionType)
+				g.Expect(cond).To(Not(BeNil()))
+				g.Expect(cond.Status).To(Equal(v1.ConditionTrue))
+			}
+		}).Should(Succeed())
+	})
+
+	It("updates condition when a new pod is added", func() {
+		testFramework.Get(ctx, k8s.NamespacedName(deployment), deployment)
+		deployment.Spec.Replicas = lo.ToPtr(int32(3))
+		testFramework.ExpectUpdated(ctx, deployment)
+
+		Eventually(func(g Gomega) {
+			pods := testFramework.GetPodsByDeploymentName(deployment.Name, deployment.Namespace)
+			for _, pod := range pods {
+				cond := utils.FindPodStatusCondition(pod.Status.Conditions, lattice.LatticeReadinessGateConditionType)
+				g.Expect(cond).To(Not(BeNil()))
+				g.Expect(cond.Status).To(Equal(v1.ConditionTrue))
+			}
+		}).Should(Succeed())
+	})
+
+	AfterAll(func() {
+		testFramework.ExpectDeletedThenNotFound(ctx,
+			deployment,
+			service,
+			route,
+		)
+	})
+})