Add p2p integration test (#11270)

Add an integration test that exercises the direct pod-to-pod multicluster mode.

Signed-off-by: Alex Leong <alex@buoyant.io>
Co-authored-by: Alejandro Pedraza <alejandro@buoyant.io>

justfile

@@ -508,7 +508,7 @@ mc-test-build:
     go build --mod=readonly \
         ./test/integration/multicluster/...
 
-mc-test-load: _mc-load _mc-target-load
+mc-test-load: _mc-load _mc-target-load mc-flat-network-init
 
 k3d-source-server := "k3d-" + k3d-name + "-server-0"
 k3d-target-server := "k3d-" + k3d-name + "-target-server-0"
@@ -519,8 +519,8 @@ _mc-print-target-route := "kubectl --context=k3d-" + k3d-name + "-target "+ "get
 
 # Allow two k3d server nodes to participate in a flat network
 mc-flat-network-init:
-	@docker exec -it k3d-{{k3d-name}}-server-0 `{{_mc-print-target-route}}`
-	@docker exec -it k3d-{{k3d-name}}-target-server-0 `{{_mc-print-source-route}}`
+	@docker exec k3d-{{k3d-name}}-server-0 `{{_mc-print-target-route}}`
+	@docker exec k3d-{{k3d-name}}-target-server-0 `{{_mc-print-source-route}}`
 
 
 # Run the multicluster tests without any setup

test/integration/multicluster/install_test.go

@@ -165,7 +165,10 @@ func TestMulticlusterResourcesPostInstall(t *testing.T) {
 		{Namespace: "linkerd-multicluster", Name: "linkerd-gateway"},
 	}
 
-	testutil.TestResourcesPostInstall(TestHelper.GetMulticlusterNamespace(), multiclusterSvcs, testutil.MulticlusterDeployReplicas, TestHelper, t)
+	for _, ctx := range contexts {
+		TestHelper.SwitchContext(ctx)
+		testutil.TestResourcesPostInstall(TestHelper.GetMulticlusterNamespace(), multiclusterSvcs, testutil.MulticlusterDeployReplicas, TestHelper, t)
+	}
 }
 
 func TestLinkClusters(t *testing.T) {

test/integration/multicluster/multicluster-traffic/mc_traffic_test.go

@@ -163,16 +163,28 @@ func TestTargetTraffic(t *testing.T) {
 	TestHelper.WithDataPlaneNamespace(ctx, "emojivoto", annotations, t, func(t *testing.T, ns string) {
 		t.Run("Deploy resources in source and target clusters", func(t *testing.T) {
 			// Deploy vote-bot client in source-cluster
-			o, err := TestHelper.KubectlApplyWithContext("", contexts[testutil.SourceContextKey], "-f", "testdata/vote-bot.yml")
+			o, err := TestHelper.KubectlWithContext("", contexts[testutil.SourceContextKey], "create", "ns", ns)
+			if err != nil {
+				testutil.AnnotatedFatalf(t, "failed to create ns", "failed to create ns: %s\n%s", err, o)
+			}
+			o, err = TestHelper.KubectlApplyWithContext("", contexts[testutil.SourceContextKey], "--namespace", ns, "-f", "testdata/vote-bot.yml")
 			if err != nil {
 				testutil.AnnotatedFatalf(t, "failed to install vote-bot", "failed to install vote-bot: %s\n%s", err, o)
 			}
 
-			out, err := TestHelper.KubectlApplyWithContext("", contexts[testutil.TargetContextKey], "-f", "testdata/emojivoto-no-bot.yml")
+			out, err := TestHelper.KubectlApplyWithContext("", contexts[testutil.TargetContextKey], "--namespace", ns, "-f", "testdata/emojivoto-no-bot.yml")
 			if err != nil {
 				testutil.AnnotatedFatalf(t, "failed to install emojivoto", "failed to install emojivoto: %s\n%s", err, out)
 			}
 
+			timeout := time.Minute
+			err = testutil.RetryFor(timeout, func() error {
+				out, err = TestHelper.KubectlWithContext("", contexts[testutil.TargetContextKey], "--namespace", ns, "label", "service/web-svc", "mirror.linkerd.io/exported=true")
+				return err
+			})
+			if err != nil {
+				testutil.AnnotatedFatalf(t, "failed to label web-svc", "%s\n%s", err, out)
+			}
 		})
 
 		t.Run("Wait until target workloads are ready", func(t *testing.T) {

test/integration/multicluster/multicluster-traffic/pod_to_pod_test.go

@@ -0,0 +1,138 @@
+package multiclustertraffic
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/linkerd/linkerd2/pkg/k8s"
+	"github.com/linkerd/linkerd2/testutil"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+)
+
+// TestPodToPodTraffic inspects the target cluster's web-svc pod to see if the
+// source cluster's vote-bot has been able to hit it with requests. If it has
+// successfully issued requests, then we'll see log messages indicating that the
+// web-svc can't reach the voting-svc (because it's not running).
+//
+// We verify that the service has been mirrored in remote discovery mode by
+// checking that it had no endpoints in the source cluster.
+func TestPodToPodTraffic(t *testing.T) {
+	if err := TestHelper.SwitchContext(contexts[testutil.TargetContextKey]); err != nil {
+		testutil.AnnotatedFatalf(t,
+			"failed to rebuild helper clientset with new context",
+			"failed to rebuild helper clientset with new context [%s]: %v",
+			contexts[testutil.TargetContextKey], err)
+	}
+
+	ctx := context.Background()
+	// Create emojivoto in target cluster, to be deleted at the end of the test.
+	annotations := map[string]string{
+		// "config.linkerd.io/proxy-log-level": "linkerd=debug,info",
+	}
+	TestHelper.WithDataPlaneNamespace(ctx, "emojivoto-p2p", annotations, t, func(t *testing.T, ns string) {
+		t.Run("Deploy resources in source and target clusters", func(t *testing.T) {
+			// Deploy vote-bot client in source-cluster
+			o, err := TestHelper.KubectlWithContext("", contexts[testutil.SourceContextKey], "create", "ns", ns)
+			if err != nil {
+				testutil.AnnotatedFatalf(t, "failed to create ns", "failed to create ns: %s\n%s", err, o)
+			}
+			o, err = TestHelper.KubectlApplyWithContext("", contexts[testutil.SourceContextKey], "--namespace", ns, "-f", "testdata/vote-bot.yml")
+			if err != nil {
+				testutil.AnnotatedFatalf(t, "failed to install vote-bot", "failed to install vote-bot: %s\n%s", err, o)
+			}
+
+			out, err := TestHelper.KubectlApplyWithContext("", contexts[testutil.TargetContextKey], "--namespace", ns, "-f", "testdata/emojivoto-no-bot.yml")
+			if err != nil {
+				testutil.AnnotatedFatalf(t, "failed to install emojivoto", "failed to install emojivoto: %s\n%s", err, out)
+			}
+
+			timeout := time.Minute
+			err = testutil.RetryFor(timeout, func() error {
+				out, err = TestHelper.KubectlWithContext("", contexts[testutil.TargetContextKey], "--namespace", ns, "label", "service/web-svc", "mirror.linkerd.io/exported=remote-discovery")
+				return err
+			})
+			if err != nil {
+				testutil.AnnotatedFatalf(t, "failed to label web-svc", "%s\n%s", err, out)
+			}
+		})
+
+		t.Run("Wait until target workloads are ready", func(t *testing.T) {
+			// Wait until client is up and running in source cluster
+			voteBotDeployReplica := map[string]testutil.DeploySpec{"vote-bot": {Namespace: ns, Replicas: 1}}
+			TestHelper.WaitRolloutWithContext(t, voteBotDeployReplica, contexts[testutil.SourceContextKey])
+
+			// Wait until "target" services and replicas are up and running.
+			emojiDeployReplicas := map[string]testutil.DeploySpec{
+				"web":    {Namespace: ns, Replicas: 1},
+				"emoji":  {Namespace: ns, Replicas: 1},
+				"voting": {Namespace: ns, Replicas: 1},
+			}
+			TestHelper.WaitRolloutWithContext(t, emojiDeployReplicas, targetCtx)
+		})
+
+		timeout := time.Minute
+		t.Run("Ensure mirror service exists and has no endpoints", func(t *testing.T) {
+			err := TestHelper.SwitchContext(contexts[testutil.SourceContextKey])
+			if err != nil {
+				testutil.AnnotatedFatal(t, "failed to switch contexts", err)
+			}
+			err = testutil.RetryFor(timeout, func() error {
+				svc, err := TestHelper.GetService(ctx, ns, "web-svc-target")
+				if err != nil {
+					return err
+				}
+				remoteDiscovery, found := svc.Labels[k8s.RemoteDiscoveryLabel]
+				if !found {
+					testutil.AnnotatedFatal(t, "mirror service missing label", "mirror service missing label: "+k8s.RemoteDiscoveryLabel)
+				}
+				if remoteDiscovery != "target" {
+					testutil.AnnotatedFatal(t, "mirror service has incorrect remote discovery", fmt.Sprintf("mirror service remote discovery was %s, expected %s", remoteDiscovery, "target"))
+				}
+				remoteService, found := svc.Labels[k8s.RemoteServiceLabel]
+				if !found {
+					testutil.AnnotatedFatal(t, "mirror service missing label", "mirror service missing label: "+k8s.RemoteServiceLabel)
+				}
+				if remoteService != "web-svc" {
+					testutil.AnnotatedFatal(t, "mirror service has incorrect remote service", fmt.Sprintf("mirror service remote service was %s, expected %s", remoteService, "web-svc"))
+				}
+				_, err = TestHelper.GetEndpoints(ctx, ns, "web-svc-target")
+				if err == nil {
+					testutil.AnnotatedFatal(t, "mirror service should not have endpoints", "mirror service should not have endpoints")
+				}
+				if !kerrors.IsNotFound(err) {
+					testutil.AnnotatedFatalf(t, "failed to retrieve mirror service endpoints", err.Error())
+				}
+				return nil
+			})
+			if err != nil {
+				testutil.AnnotatedFatal(t, "timed-out verifying mirror service", err)
+			}
+		})
+
+		err := testutil.RetryFor(timeout, func() error {
+			out, err := TestHelper.KubectlWithContext("",
+				targetCtx,
+				"--namespace", ns,
+				"logs",
+				"--selector", "app=web-svc",
+				"--container", "web-svc",
+			)
+			if err != nil {
+				return fmt.Errorf("%w\n%s", err, out)
+			}
+			// Check for expected error messages
+			for _, row := range strings.Split(out, "\n") {
+				if strings.Contains(row, " /api/vote?choice=:doughnut: ") {
+					return nil
+				}
+			}
+			return fmt.Errorf("web-svc logs in target cluster do not include voting errors\n%s", out)
+		})
+		if err != nil {
+			testutil.AnnotatedFatal(t, fmt.Sprintf("timed-out waiting for traffic (%s)", timeout), err)
+		}
+	})
+}

test/integration/multicluster/multicluster-traffic/testdata/emojivoto-no-bot.yml

@@ -2,25 +2,21 @@ apiVersion: v1
 kind: ServiceAccount
 metadata:
   name: emoji
-  namespace: linkerd-emojivoto
 ---
 apiVersion: v1
 kind: ServiceAccount
 metadata:
   name: voting
-  namespace: linkerd-emojivoto
 ---
 apiVersion: v1
 kind: ServiceAccount
 metadata:
   name: web
-  namespace: linkerd-emojivoto
 ---
 apiVersion: v1
 kind: Service
 metadata:
   name: emoji-svc
-  namespace: linkerd-emojivoto
 spec:
   ports:
   - name: grpc
@@ -36,7 +32,6 @@ apiVersion: v1
 kind: Service
 metadata:
   name: voting-svc
-  namespace: linkerd-emojivoto
 spec:
   ports:
   - name: grpc
@@ -52,9 +47,6 @@ apiVersion: v1
 kind: Service
 metadata:
   name: web-svc
-  namespace: linkerd-emojivoto
-  labels:
-    mirror.linkerd.io/exported: "true"
 spec:
   ports:
   - name: http
@@ -72,7 +64,6 @@ metadata:
     app.kubernetes.io/part-of: emojivoto
     app.kubernetes.io/version: v10
   name: emoji
-  namespace: linkerd-emojivoto
 spec:
   replicas: 1
   selector:
@@ -113,7 +104,6 @@ metadata:
     app.kubernetes.io/part-of: emojivoto
     app.kubernetes.io/version: v10
   name: voting
-  namespace: linkerd-emojivoto
 spec:
   replicas: 1
   selector:
@@ -154,7 +144,6 @@ metadata:
     app.kubernetes.io/part-of: emojivoto
     app.kubernetes.io/version: v10
   name: web
-  namespace: linkerd-emojivoto
 spec:
   replicas: 1
   selector:

test/integration/multicluster/multicluster-traffic/testdata/vote-bot.yml

@@ -1,10 +1,3 @@
-apiVersion: v1
-kind: Namespace
-metadata:
-  labels:
-    test.linkerd.io/is-test-data-plane: "true"
-  name: linkerd-emojivoto
----
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -13,7 +6,6 @@ metadata:
     app.kubernetes.io/part-of: emojivoto
     app.kubernetes.io/version: v10
   name: vote-bot
-  namespace: linkerd-emojivoto
 spec:
   replicas: 1
   selector:
@@ -39,4 +31,3 @@ spec:
         resources:
           requests:
             cpu: 10m
----

testutil/kubernetes_helper.go

@@ -288,6 +288,15 @@ func (h *KubernetesHelper) GetService(ctx context.Context, namespace string, ser
 	return service, nil
 }
 
+// GetEndpoints gets endpoints that exist in a namespace.
+func (h *KubernetesHelper) GetEndpoints(ctx context.Context, namespace string, serviceName string) (*corev1.Endpoints, error) {
+	ep, err := h.clientset.CoreV1().Endpoints(namespace).Get(ctx, serviceName, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+	return ep, nil
+}
+
 // GetPods returns all pods with the given labels
 func (h *KubernetesHelper) GetPods(ctx context.Context, namespace string, podLabels map[string]string) ([]corev1.Pod, error) {
 	podList, err := h.clientset.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{