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) }) /*