connectivity: add endpointslice clustermesh sync test

This adds endpointslice synchronization testing inside a clustermesh.
It does that by checking that we can get a DNS answer with a global
headless service with endpointslice synchronization enabled across two clusters.

Testing it via DNS ensures that the created EndpointSlice are correctly
linked to the headless service.

Signed-off-by: Arthur Outhenin-Chalandre <[email protected]>

connectivity/builder/builder.go

@@ -116,6 +116,7 @@ var (
 		clientEgressToEchoServiceAccountDeny{},
 		clientEgressToCidrDeny{},
 		clientEgressToCidrDenyDefault{},
+		endpointSliceClusterMeshSync{},
 		health{},
 		northSouthLoadbalancing{},
 		podToPodEncryption{},

connectivity/builder/endpointslice-clustermesh-sync.go

@@ -0,0 +1,18 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Cilium
+
+package builder
+
+import (
+	"github.com/cilium/cilium-cli/connectivity/check"
+	"github.com/cilium/cilium-cli/connectivity/tests"
+	"github.com/cilium/cilium-cli/utils/features"
+)
+
+type endpointSliceClusterMeshSync struct{}
+
+func (t endpointSliceClusterMeshSync) build(ct *check.ConnectivityTest, _ map[string]string) {
+	newTest("endpointslice-clustermesh-sync", ct).
+		WithFeatureRequirements(features.RequireEnabled(features.ClusterMeshEnableEndpointSync)).
+		WithScenarios(tests.EndpointSliceClusterMeshSync())
+}

connectivity/check/context.go

@@ -1040,6 +1040,12 @@ func (ct *ConnectivityTest) DigCommand(peer TestPeer, ipFam features.IPFamily) [
 	return cmd
 }
 
+func (ct *ConnectivityTest) DigCommandService(peer TestPeer) []string {
+	cmd := []string{"dig", "+time=2", peer.Name()}
+
+	return cmd
+}
+
 func (ct *ConnectivityTest) RandomClientPod() *Pod {
 	for _, p := range ct.clientPods {
 		return &p

connectivity/check/deployment.go

@@ -43,15 +43,16 @@ const (
 
 	DNSTestServerContainerName = "dns-test-server"
 
-	echoSameNodeDeploymentName     = "echo-same-node"
-	echoOtherNodeDeploymentName    = "echo-other-node"
-	echoExternalNodeDeploymentName = "echo-external-node"
-	corednsConfigMapName           = "coredns-configmap"
-	corednsConfigVolumeName        = "coredns-config-volume"
-	kindEchoName                   = "echo"
-	kindEchoExternalNodeName       = "echo-external-node"
-	kindClientName                 = "client"
-	kindPerfName                   = "perf"
+	echoSameNodeDeploymentName                 = "echo-same-node"
+	echoOtherNodeDeploymentName                = "echo-other-node"
+	echoOtherNodeDeploymentHeadlessServiceName = "echo-other-node-headless"
+	echoExternalNodeDeploymentName             = "echo-external-node"
+	corednsConfigMapName                       = "coredns-configmap"
+	corednsConfigVolumeName                    = "coredns-config-volume"
+	kindEchoName                               = "echo"
+	kindEchoExternalNodeName                   = "echo-external-node"
+	kindClientName                             = "client"
+	kindPerfName                               = "perf"
 
 	hostNetNSDeploymentName          = "host-netns"
 	hostNetNSDeploymentNameNonCilium = "host-netns-non-cilium" // runs on non-Cilium test nodes
@@ -531,6 +532,16 @@ func (ct *ConnectivityTest) deploy(ctx context.Context) error {
 			if err != nil {
 				return err
 			}
+
+			svcHeadless := svc.DeepCopy()
+			svcHeadless.Name = echoOtherNodeDeploymentHeadlessServiceName
+			svcHeadless.Spec.ClusterIP = corev1.ClusterIPNone
+			svcHeadless.Spec.Type = corev1.ServiceTypeClusterIP
+			svcHeadless.ObjectMeta.Annotations["service.cilium.io/global-sync-endpoint-slices"] = "true"
+			_, err = ct.clients.src.CreateService(ctx, ct.params.TestNamespace, svcHeadless, metav1.CreateOptions{})
+			if err != nil {
+				return err
+			}
 		}
 	}
 
@@ -744,6 +755,16 @@ func (ct *ConnectivityTest) deploy(ctx context.Context) error {
 				svc.ObjectMeta.Annotations = map[string]string{}
 				svc.ObjectMeta.Annotations["service.cilium.io/global"] = "true"
 				svc.ObjectMeta.Annotations["io.cilium/global-service"] = "true"
+
+				svcHeadless := svc.DeepCopy()
+				svcHeadless.Name = echoOtherNodeDeploymentHeadlessServiceName
+				svcHeadless.Spec.ClusterIP = corev1.ClusterIPNone
+				svcHeadless.Spec.Type = corev1.ServiceTypeClusterIP
+				svcHeadless.ObjectMeta.Annotations["service.cilium.io/global-sync-endpoint-slices"] = "true"
+				_, err = ct.clients.dst.CreateService(ctx, ct.params.TestNamespace, svcHeadless, metav1.CreateOptions{})
+				if err != nil {
+					return err
+				}
 			}
 
 			_, err = ct.clients.dst.CreateService(ctx, ct.params.TestNamespace, svc, metav1.CreateOptions{})
@@ -1074,6 +1095,7 @@ func (ct *ConnectivityTest) deleteDeployments(ctx context.Context, client *k8s.C
 	_ = client.DeleteServiceAccount(ctx, ct.params.TestNamespace, client3DeploymentName, metav1.DeleteOptions{})
 	_ = client.DeleteService(ctx, ct.params.TestNamespace, echoSameNodeDeploymentName, metav1.DeleteOptions{})
 	_ = client.DeleteService(ctx, ct.params.TestNamespace, echoOtherNodeDeploymentName, metav1.DeleteOptions{})
+	_ = client.DeleteService(ctx, ct.params.TestNamespace, echoOtherNodeDeploymentHeadlessServiceName, metav1.DeleteOptions{})
 	_ = client.DeleteConfigMap(ctx, ct.params.TestNamespace, corednsConfigMapName, metav1.DeleteOptions{})
 	_ = client.DeleteNamespace(ctx, ct.params.TestNamespace, metav1.DeleteOptions{})
 

connectivity/check/wait.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/cilium/cilium/api/v1/models"
 	"golang.org/x/exp/slices"
+	corev1 "k8s.io/api/core/v1"
 
 	"github.com/cilium/cilium-cli/defaults"
 	"github.com/cilium/cilium-cli/k8s"
@@ -141,6 +142,11 @@ func WaitForService(ctx context.Context, log Logger, client Pod, service Service
 	ctx, cancel := context.WithTimeout(ctx, ShortTimeout)
 	defer cancel()
 
+	if service.Service.Spec.ClusterIP == corev1.ClusterIPNone {
+		// If the cluster is headless there is nothing to wait for here
+		return nil
+	}
+
 	for {
 		stdout, err := client.K8sClient.ExecInPod(ctx,
 			client.Namespace(), client.NameWithoutNamespace(), "",
@@ -224,6 +230,9 @@ func checkServiceEndpoints(ctx context.Context, agent Pod, service Service, back
 	}
 
 	for _, ip := range service.Service.Spec.ClusterIPs {
+		if ip == corev1.ClusterIPNone {
+			continue
+		}
 		addr, err := netip.ParseAddr(ip)
 		if err != nil {
 			return fmt.Errorf("failed to parse ClusterIP %q: %w", ip, err)

connectivity/tests/endpointslice-clustermesh-sync.go

@@ -0,0 +1,44 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Cilium
+
+package tests
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cilium/cilium-cli/connectivity/check"
+	"github.com/cilium/cilium-cli/utils/features"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func EndpointSliceClusterMeshSync() check.Scenario {
+	return &endpointSliceClusterMeshSync{}
+}
+
+type endpointSliceClusterMeshSync struct{}
+
+func (s *endpointSliceClusterMeshSync) Name() string {
+	return "endpointslice-clustermesh-sync"
+}
+
+func (s *endpointSliceMesh) Run(ctx context.Context, t *check.Test) {
+	ct := t.Context()
+
+	i := 0
+	for _, client := range ct.ClientPods() {
+		client := client // copy to avoid memory aliasing when using reference
+
+		kubeService, err := ct.K8sClient().GetService(ctx, ct.Params().TestNamespace, "echo-other-node-headless", metav1.GetOptions{})
+		if err != nil {
+			t.Fatal("Cannot get echo-other-node-headless service")
+		}
+		service := check.Service{Service: kubeService}
+
+		t.NewAction(s, fmt.Sprintf("dig-%d", i), &client, service, features.IPFamilyV4).Run(func(a *check.Action) {
+			a.ExecInPod(ctx, ct.DigCommandService(service))
+		})
+		i++
+	}
+}

utils/features/features.go

@@ -64,6 +64,8 @@ const (
 	EnableEnvoyConfig Feature = "enable-envoy-config"
 
 	WireguardEncapsulate Feature = "wireguard-encapsulate"
+
+	ClusterMeshEnableEndpointSync Feature = "clustermesh-enable-endpoint-sync"
 )
 
 // Feature is the name of a Cilium Feature (e.g. l7-proxy, cni chaining mode etc)
@@ -272,6 +274,10 @@ func (fs Set) ExtractFromConfigMap(cm *v1.ConfigMap) {
 	fs[WireguardEncapsulate] = Status{
 		Enabled: cm.Data[string(WireguardEncapsulate)] == "true",
 	}
+
+	fs[ClusterMeshEnableEndpointSync] = Status{
+		Enabled: cm.Data[string(ClusterMeshEnableEndpointSync)] == "true",
+	}
 }
 
 func (fs Set) ExtractFromNodes(nodesWithoutCilium map[string]struct{}) {