From b35ad4e9b01d3a396d952794106b8413b2f0cd32 Mon Sep 17 00:00:00 2001 From: Dan Winship Date: Mon, 2 Jun 2025 21:13:08 -0400 Subject: [PATCH] Add e2eendpointslice.WaitForEndpointPods, use in various test These tests were using validateEndpointsPortsOrFail() not because they cared about ports, but just because it was there, or in some cases because they needed to wait for one pod to exit and a different pod to start, which can't be done with framework.WaitForServiceEndpointsNum() (or e2eendpointslice.WaitForEndpointCount) without racing. Update these tests using the new e2eendpointslice.WaitForEndpointPods, which can wait for specific expected pods. (This also means these tests now only watch EndpointSlices, rather than watching both Endpoints and EndpointSlices, which is fine, because none of them are doing tricky things that actually require making assertions about the exact contents of the Endpoints/EndpointSlices. They just want to know when the controller has updated things to point to the expected pods.) --- test/e2e/framework/endpointslice/wait.go | 31 +++++++++++++++++++++++ test/e2e/network/conntrack.go | 32 +++++++++++++++--------- test/e2e/network/loadbalancer.go | 17 ++++++++----- test/e2e/network/service.go | 19 ++++++++++---- 4 files changed, 76 insertions(+), 23 deletions(-) diff --git a/test/e2e/framework/endpointslice/wait.go b/test/e2e/framework/endpointslice/wait.go index c57704eb675..fa655bfa983 100644 --- a/test/e2e/framework/endpointslice/wait.go +++ b/test/e2e/framework/endpointslice/wait.go @@ -85,3 +85,34 @@ func WaitForEndpointCount(ctx context.Context, cs clientset.Interface, namespace return true, nil }) } + +// WaitForEndpointPods waits (up to ControllerUpdateTimeout) for the named service to have +// at least one EndpointSlice, with the set of `Endpoints` across all slices covering +// exactly the pods named in expectPods. (Note that if called on a dual-stack service, it +// may return before both EndpointSlices have been written, since the first EndpointSlice +// alone will normally target all pods.) +func WaitForEndpointPods(ctx context.Context, cs clientset.Interface, namespace, serviceName string, expectPods ...string) error { + framework.Logf("Waiting for service %s/%s to have endpoints pointing to %v", namespace, serviceName, expectPods) + expectPodSet := sets.New(expectPods...) + return WaitForEndpointSlices(ctx, cs, namespace, serviceName, time.Second, ControllerUpdateTimeout, func(ctx context.Context, endpointSlices []discoveryv1.EndpointSlice) (bool, error) { + if len(endpointSlices) == 0 { + framework.Logf("Waiting for at least 1 EndpointSlice to exist") + return false, nil + } + + podSet := sets.Set[string]{} + for _, epSlice := range endpointSlices { + for _, ep := range epSlice.Endpoints { + if ep.TargetRef != nil && ep.TargetRef.Kind == "Pod" && ep.TargetRef.Namespace == namespace { + podSet.Insert(ep.TargetRef.Name) + } + } + } + + if !podSet.Equal(expectPodSet) { + framework.Logf("Unexpected endpoints on slices, missing: %v, extra: %v", sets.List(expectPodSet.Difference(podSet)), sets.List(podSet.Difference(expectPodSet))) + return false, nil + } + return true, nil + }) +} diff --git a/test/e2e/network/conntrack.go b/test/e2e/network/conntrack.go index 84e14e9b43a..b2903d0d487 100644 --- a/test/e2e/network/conntrack.go +++ b/test/e2e/network/conntrack.go @@ -31,6 +31,7 @@ import ( clientset "k8s.io/client-go/kubernetes" "k8s.io/kubernetes/test/e2e/framework" e2edaemonset "k8s.io/kubernetes/test/e2e/framework/daemonset" + e2eendpointslice "k8s.io/kubernetes/test/e2e/framework/endpointslice" e2enode "k8s.io/kubernetes/test/e2e/framework/node" e2epod "k8s.io/kubernetes/test/e2e/framework/pod" e2eservice "k8s.io/kubernetes/test/e2e/framework/service" @@ -172,9 +173,10 @@ var _ = common.SIGDescribe("Conntrack", func() { e2epod.SetNodeSelection(&serverPod1.Spec, nodeSelection) e2epod.NewPodClient(fr).CreateSync(ctx, serverPod1) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend1: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend1) + framework.ExpectNoError(err) - // Note that the fact that Endpoints object already exists, does NOT mean + // Note that the fact that EndpointSlice object already exists, does NOT mean // that iptables (or whatever else is used) was already programmed. // Additionally take into account that UDP conntract entries timeout is // 30 seconds by default. @@ -199,7 +201,8 @@ var _ = common.SIGDescribe("Conntrack", func() { framework.Logf("Cleaning up %s pod", podBackend1) e2epod.NewPodClient(fr).DeleteSync(ctx, podBackend1, metav1.DeleteOptions{}, fr.Timeouts.PodDelete) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend2: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend2) + framework.ExpectNoError(err) // Check that the second pod keeps receiving traffic // UDP conntrack entries timeout is 30 sec by default @@ -248,9 +251,10 @@ var _ = common.SIGDescribe("Conntrack", func() { e2epod.SetNodeSelection(&serverPod1.Spec, nodeSelection) e2epod.NewPodClient(fr).CreateSync(ctx, serverPod1) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend1: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend1) + framework.ExpectNoError(err) - // Note that the fact that Endpoints object already exists, does NOT mean + // Note that the fact that EndpointSlice object already exists, does NOT mean // that iptables (or whatever else is used) was already programmed. // Additionally take into account that UDP conntract entries timeout is // 30 seconds by default. @@ -275,7 +279,8 @@ var _ = common.SIGDescribe("Conntrack", func() { framework.Logf("Cleaning up %s pod", podBackend1) e2epod.NewPodClient(fr).DeleteSync(ctx, podBackend1, metav1.DeleteOptions{}, fr.Timeouts.PodDelete) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend2: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend2) + framework.ExpectNoError(err) // Check that the second pod keeps receiving traffic // UDP conntrack entries timeout is 30 sec by default @@ -343,7 +348,7 @@ var _ = common.SIGDescribe("Conntrack", func() { framework.Failf("expected 1 pod, got %d pods", len(pods.Items)) } podBackend1 := pods.Items[0].Name - // Note that the fact that Endpoints object already exists, does NOT mean + // Note that the fact that EndpointSlice object already exists, does NOT mean // that iptables (or whatever else is used) was already programmed. // Additionally take into account that UDP conntract entries timeout is // 30 seconds by default. @@ -420,9 +425,10 @@ var _ = common.SIGDescribe("Conntrack", func() { e2epod.SetNodeSelection(&serverPod1.Spec, nodeSelection) e2epod.NewPodClient(fr).CreateSync(ctx, serverPod1) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend1: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend1) + framework.ExpectNoError(err) - // Note that the fact that Endpoints object already exists, does NOT mean + // Note that the fact that EndpointSlice object already exists, does NOT mean // that iptables (or whatever else is used) was already programmed. // Additionally take into account that UDP conntract entries timeout is // 30 seconds by default. @@ -447,7 +453,8 @@ var _ = common.SIGDescribe("Conntrack", func() { framework.Logf("Cleaning up %s pod", podBackend1) e2epod.NewPodClient(fr).DeleteSync(ctx, podBackend1, metav1.DeleteOptions{}, fr.Timeouts.PodDelete) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend2: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend2) + framework.ExpectNoError(err) // Check that the second pod keeps receiving traffic // UDP conntrack entries timeout is 30 sec by default @@ -516,9 +523,10 @@ var _ = common.SIGDescribe("Conntrack", func() { e2epod.NewPodClient(fr).CreateSync(ctx, serverPod1) // wait until the endpoints are ready - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend1: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend1) + framework.ExpectNoError(err) - // Note that the fact that Endpoints object already exists, does NOT mean + // Note that the fact that EndpointSlice object already exists, does NOT mean // that iptables (or whatever else is used) was already programmed. // Additionally take into account that UDP conntract entries timeout is // 30 seconds by default. diff --git a/test/e2e/network/loadbalancer.go b/test/e2e/network/loadbalancer.go index 5d9b632b225..90249117e2e 100644 --- a/test/e2e/network/loadbalancer.go +++ b/test/e2e/network/loadbalancer.go @@ -43,6 +43,7 @@ import ( "k8s.io/kubernetes/test/e2e/framework" e2edaemonset "k8s.io/kubernetes/test/e2e/framework/daemonset" e2edeployment "k8s.io/kubernetes/test/e2e/framework/deployment" + e2eendpointslice "k8s.io/kubernetes/test/e2e/framework/endpointslice" e2enetwork "k8s.io/kubernetes/test/e2e/framework/network" e2enode "k8s.io/kubernetes/test/e2e/framework/node" e2epod "k8s.io/kubernetes/test/e2e/framework/pod" @@ -792,9 +793,10 @@ var _ = common.SIGDescribe("LoadBalancers", feature.LoadBalancer, func() { e2epod.SetNodeSelection(&serverPod1.Spec, nodeSelection) e2epod.NewPodClient(f).CreateSync(ctx, serverPod1) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend1: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend1) + framework.ExpectNoError(err) - // Note that the fact that Endpoints object already exists, does NOT mean + // Note that the fact that EndpointSlice object already exists, does NOT mean // that iptables (or whatever else is used) was already programmed. // Additionally take into account that UDP conntract entries timeout is // 30 seconds by default. @@ -821,7 +823,8 @@ var _ = common.SIGDescribe("LoadBalancers", feature.LoadBalancer, func() { framework.Logf("Cleaning up %s pod", podBackend1) e2epod.NewPodClient(f).DeleteSync(ctx, podBackend1, metav1.DeleteOptions{}, f.Timeouts.PodDelete) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend2: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend2) + framework.ExpectNoError(err) // Check that the second pod keeps receiving traffic // UDP conntrack entries timeout is 30 sec by default @@ -924,9 +927,10 @@ var _ = common.SIGDescribe("LoadBalancers", feature.LoadBalancer, func() { e2epod.SetNodeSelection(&serverPod1.Spec, nodeSelection) e2epod.NewPodClient(f).CreateSync(ctx, serverPod1) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend1: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend1) + framework.ExpectNoError(err) - // Note that the fact that Endpoints object already exists, does NOT mean + // Note that the fact that EndpointSlice object already exists, does NOT mean // that iptables (or whatever else is used) was already programmed. // Additionally take into account that UDP conntract entries timeout is // 30 seconds by default. @@ -953,7 +957,8 @@ var _ = common.SIGDescribe("LoadBalancers", feature.LoadBalancer, func() { framework.Logf("Cleaning up %s pod", podBackend1) e2epod.NewPodClient(f).DeleteSync(ctx, podBackend1, metav1.DeleteOptions{}, f.Timeouts.PodDelete) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{podBackend2: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, podBackend2) + framework.ExpectNoError(err) // Check that the second pod keeps receiving traffic // UDP conntrack entries timeout is 30 sec by default diff --git a/test/e2e/network/service.go b/test/e2e/network/service.go index 11a45ab1a0a..7df5e1658d6 100644 --- a/test/e2e/network/service.go +++ b/test/e2e/network/service.go @@ -763,7 +763,8 @@ var _ = common.SIGDescribe("Services", func() { svc, err := jig.CreateTCPServiceWithPort(ctx, nil, 80) framework.ExpectNoError(err) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{}) + err = e2eendpointslice.WaitForEndpointCount(ctx, cs, ns, serviceName, 0) + framework.ExpectNoError(err) names := map[string]bool{} ginkgo.DeferCleanup(func(ctx context.Context) { @@ -776,34 +777,42 @@ var _ = common.SIGDescribe("Services", func() { name1 := "pod1" name2 := "pod2" + ginkgo.By("creating first endpoint pod for service") createPodOrFail(ctx, f, ns, name1, jig.Labels, []v1.ContainerPort{{ContainerPort: 80}}, "netexec", "--http-port", "80") names[name1] = true - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{name1: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, name1) + framework.ExpectNoError(err) ginkgo.By("Checking if the Service forwards traffic to pod1") execPod := e2epod.CreateExecPodOrFail(ctx, cs, ns, "execpod", nil) err = jig.CheckServiceReachability(ctx, svc, execPod) framework.ExpectNoError(err) + ginkgo.By("creating second endpoint pod for service") createPodOrFail(ctx, f, ns, name2, jig.Labels, []v1.ContainerPort{{ContainerPort: 80}}, "netexec", "--http-port", "80") names[name2] = true - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{name1: {80}, name2: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, name1, name2) + framework.ExpectNoError(err) ginkgo.By("Checking if the Service forwards traffic to pod1 and pod2") err = jig.CheckServiceReachability(ctx, svc, execPod) framework.ExpectNoError(err) + ginkgo.By("deleting first endpoint pod") e2epod.DeletePodOrFail(ctx, cs, ns, name1) delete(names, name1) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{name2: {80}}) + err = e2eendpointslice.WaitForEndpointPods(ctx, cs, ns, serviceName, name2) + framework.ExpectNoError(err) ginkgo.By("Checking if the Service forwards traffic to pod2") err = jig.CheckServiceReachability(ctx, svc, execPod) framework.ExpectNoError(err) + ginkgo.By("deleting second endpoint pod") e2epod.DeletePodOrFail(ctx, cs, ns, name2) delete(names, name2) - validateEndpointsPortsOrFail(ctx, cs, ns, serviceName, portsByPodName{}) + err = e2eendpointslice.WaitForEndpointCount(ctx, cs, ns, serviceName, 0) + framework.ExpectNoError(err) }) /*