fix(e2e): wait for endpoints and retry service trace in kubectl-ko test

test/e2e/kube-ovn/kubectl-ko/kubectl-ko.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 	"math/rand/v2"
 	"strings"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	netv1 "k8s.io/api/networking/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	clientset "k8s.io/client-go/kubernetes"
@@ -311,6 +313,23 @@ var _ = framework.Describe("[group:kubectl-ko]", func() {
 			return len(s.Spec.ClusterIPs) != 0, nil
 		}, "cluster ips are not empty")
 
+		ginkgo.By("Waiting for endpoints " + serviceName + " to be ready")
+		framework.WaitUntil(2*time.Second, time.Minute, func(_ context.Context) (bool, error) {
+			eps, err := cs.CoreV1().Endpoints(namespaceName).Get(context.TODO(), serviceName, metav1.GetOptions{})
+			if err == nil {
+				for _, subset := range eps.Subsets {
+					if len(subset.Addresses) > 0 {
+						return true, nil
+					}
+				}
+				return false, nil
+			}
+			if k8serrors.IsNotFound(err) {
+				return false, nil
+			}
+			return false, err
+		}, fmt.Sprintf("endpoints %s has at least one ready address", serviceName))
+
 		ginkgo.By("Creating pod " + podName)
 		pod := framework.MakePod(namespaceName, podName, nil, nil, "", nil, nil)
 		pod = podClient.CreateSync(pod)
@@ -323,6 +342,13 @@ var _ = framework.Describe("[group:kubectl-ko]", func() {
 		}
 		matchPod := fmt.Sprintf("output to %q", ovs.PodNameToPortName(pod2Name, pod2.Namespace, util.OvnProvider))
 		matchLocalnet := fmt.Sprintf("output to %q", "localnet."+util.DefaultSubnet)
+		checkOutput := func(output, match string) bool {
+			if subCmd == "ovn-trace" {
+				lines := strings.Split(strings.TrimSpace(output), "\n")
+				return strings.Contains(lines[len(lines)-1], match)
+			}
+			return strings.Contains(output, match)
+		}
 		checkFunc := func(output string) {
 			ginkgo.GinkgoHelper()
 			var match string
@@ -331,11 +357,8 @@ var _ = framework.Describe("[group:kubectl-ko]", func() {
 			} else {
 				match = matchPod
 			}
-			if subCmd == "ovn-trace" {
-				lines := strings.Split(strings.TrimSpace(output), "\n")
-				framework.ExpectContainSubstring(lines[len(lines)-1], match)
-			} else {
-				framework.ExpectContainSubstring(output, match)
+			if !checkOutput(output, match) {
+				framework.Failf("expected trace output to contain %q, but got %q", match, output)
 			}
 		}
 		for protocol, port := range map[corev1.Protocol]int32{corev1.ProtocolTCP: tcpPort, corev1.ProtocolUDP: udpPort} {
@@ -346,7 +369,19 @@ var _ = framework.Describe("[group:kubectl-ko]", func() {
 			}
 			traceService = true
 			for _, ip := range service.Spec.ClusterIPs {
-				execOrDie(fmt.Sprintf("ko %s %s/%s %s %s %d", subCmd, pod.Namespace, pod.Name, ip, proto, port), checkFunc)
+				cmd := fmt.Sprintf("ko %s %s/%s %s %s %d", subCmd, pod.Namespace, pod.Name, ip, proto, port)
+				var match string
+				if f.VersionPriorTo(1, 11) && f.IsUnderlay() {
+					match = matchLocalnet
+				} else {
+					match = matchPod
+				}
+				// Retry Service ClusterIP trace to allow OVN LB rules to be synced
+				framework.WaitUntil(2*time.Second, 30*time.Second, func(_ context.Context) (bool, error) {
+					ginkgo.By(fmt.Sprintf("Executing \"kubectl %s\"", cmd))
+					output := e2ekubectl.NewKubectlCommand("", strings.Fields(cmd)...).ExecOrDie("")
+					return checkOutput(output, match), nil
+				}, fmt.Sprintf("trace to service %s should reach target pod", ip))
 			}
 		}
 	})