initial work on ambient mc integ tests (#57644)

* initial work on ambient mc integ tests

* lint

* skip mc for uncaptured

* add cleanup to globals

* unlabel at cleanup

* sanity check

* fix unlabel

* fmt

* add registry changes

* fmt?

* fix mc registry server

* clarify test name

* add remote secrets

* lint

* improve mc checks

* attempt to unbreak other tests

* don't label sidecar services

* skip sidecar tests for mc ambient

* improve or skip tests from ingress

* re-add commented out tests

* skip cni tests for mc

* skip more sidecar tests

* skip more sidecars, and always use global labels

* skip more sidecars

* check for race condition.

* label waypoints as global, skip known failures.

* skip known failures

* lint

* skip authn flakey tests

* remove skip of PeerAuthn in favor of targeted sleep.

* don't skip serversidelb, as fixed on master and 128.

* apply review comments

* remove debugging code

* add accidentally commented test

* skip traffic split, expected behavior

Author: therealmitchconnors

common/scripts/kind_provisioner.sh

@@ -196,7 +196,7 @@ function setup_kind_cluster() {
   if [[ -n "${DEVCONTAINER:-}" ]]; then
     # identify our docker container id using proc and regex
     containerid=$(grep 'resolv.conf' /proc/self/mountinfo | sed 's/.*\/docker\/containers\/\([0-9a-f]*\).*/\1/')
-    docker network connect kind "$containerid"
+    docker network connect kind "$containerid" 2>/dev/null || true
     kind export kubeconfig --name="${NAME}" --internal
   fi
 

pkg/test/framework/components/ambient/waypoint.go

@@ -43,10 +43,11 @@ var _ io.Closer = &kubeComponent{}
 type kubeComponent struct {
 	id resource.ID
 
-	ns       namespace.Instance
-	inbound  istioKube.PortForwarder
-	outbound istioKube.PortForwarder
-	pod      v1.Pod
+	ns          namespace.Instance
+	inbound     istioKube.PortForwarder
+	outbound    istioKube.PortForwarder
+	pod         v1.Pod
+	clusterName string
 }
 
 func (k kubeComponent) Namespace() namespace.Instance {
@@ -79,17 +80,36 @@ func (k kubeComponent) Close() error {
 	return nil
 }
 
+func (k kubeComponent) ClusterName() string {
+	return k.clusterName
+}
+
 // WaypointProxy describes a waypoint proxy deployment
 type WaypointProxy interface {
 	Namespace() namespace.Instance
 	Inbound() string
 	Outbound() string
 	PodIP() string
+	ClusterName() string
+}
+
+type Waypoints []WaypointProxy
+
+// ForCluster returns a list of instances that match the cluster name
+func (i Waypoints) ForCluster(name string) Waypoints {
+	out := make(Waypoints, 0, len(i))
+	for _, c := range i {
+		if c.ClusterName() == name {
+			out = append(out, c)
+		}
+	}
+	return out
 }
 
 func NewWaypointProxyForCluster(ctx resource.Context, ns namespace.Instance, name string, cls cluster.Cluster) (WaypointProxy, error) {
 	server := &kubeComponent{
-		ns: ns,
+		ns:          ns,
+		clusterName: cls.Name(),
 	}
 	server.id = ctx.TrackResource(server)
 	if err := crd.DeployGatewayAPI(ctx); err != nil {
@@ -147,11 +167,8 @@ func NewWaypointProxyForCluster(ctx resource.Context, ns namespace.Instance, nam
 }
 
 // NewWaypointProxy creates a new WaypointProxy.
-func NewWaypointProxy(ctx resource.Context, ns namespace.Instance, name string) (WaypointProxy, error) {
-	server := &kubeComponent{
-		ns: ns,
-	}
-	server.id = ctx.TrackResource(server)
+func NewWaypointProxy(ctx resource.Context, ns namespace.Instance, name string) (Waypoints, error) {
+	var servers Waypoints
 	if err := crd.DeployGatewayAPI(ctx); err != nil {
 		return nil, err
 	}
@@ -181,34 +198,41 @@ func NewWaypointProxy(ctx resource.Context, ns namespace.Instance, name string)
 		}
 	}
 
-	cls := ctx.Clusters().Default()
-	// Find the Waypoint pod and service, and start forwarding a local port.
-	fetchFn := testKube.NewSinglePodFetch(cls, ns.Name(), fmt.Sprintf("%s=%s", label.IoK8sNetworkingGatewayGatewayName.Name, name))
-	pods, err := testKube.WaitUntilPodsAreReady(fetchFn)
-	if err != nil {
-		return nil, err
-	}
-	pod := pods[0]
-	inbound, err := cls.NewPortForwarder(pod.Name, pod.Namespace, "", 0, 15008)
-	if err != nil {
-		return nil, err
-	}
+	for _, cls := range ctx.AllClusters() {
+		server := &kubeComponent{
+			ns:          ns,
+			clusterName: cls.Name(),
+		}
+		server.id = ctx.TrackResource(server)
+		// Find the Waypoint pod and service, and start forwarding a local port.
+		fetchFn := testKube.NewSinglePodFetch(cls, ns.Name(), fmt.Sprintf("%s=%s", label.IoK8sNetworkingGatewayGatewayName.Name, name))
+		pods, err := testKube.WaitUntilPodsAreReady(fetchFn)
+		if err != nil {
+			return nil, err
+		}
+		pod := pods[0]
+		inbound, err := cls.NewPortForwarder(pod.Name, pod.Namespace, "", 0, 15008)
+		if err != nil {
+			return nil, err
+		}
 
-	if err := inbound.Start(); err != nil {
-		return nil, err
-	}
-	outbound, err := cls.NewPortForwarder(pod.Name, pod.Namespace, "", 0, 15001)
-	if err != nil {
-		return nil, err
-	}
+		if err := inbound.Start(); err != nil {
+			return nil, err
+		}
+		outbound, err := cls.NewPortForwarder(pod.Name, pod.Namespace, "", 0, 15001)
+		if err != nil {
+			return nil, err
+		}
 
-	if err := outbound.Start(); err != nil {
-		return nil, err
+		if err := outbound.Start(); err != nil {
+			return nil, err
+		}
+		server.inbound = inbound
+		server.outbound = outbound
+		server.pod = pod
+		servers = append(servers, server)
 	}
-	server.inbound = inbound
-	server.outbound = outbound
-	server.pod = pod
-	return server, nil
+	return servers, nil
 }
 
 func NewWaypointProxyOrFailForCluster(t framework.TestContext, ns namespace.Instance, name string, cls cluster.Cluster) WaypointProxy {
@@ -221,7 +245,7 @@ func NewWaypointProxyOrFailForCluster(t framework.TestContext, ns namespace.Inst
 }
 
 // NewWaypointProxyOrFail calls NewWaypointProxy and fails if an error occurs.
-func NewWaypointProxyOrFail(t framework.TestContext, ns namespace.Instance, name string) WaypointProxy {
+func NewWaypointProxyOrFail(t framework.TestContext, ns namespace.Instance, name string) Waypoints {
 	t.Helper()
 	s, err := NewWaypointProxy(t, ns, name)
 	if err != nil {

pkg/test/framework/components/echo/common/deployment/echos.go

@@ -479,7 +479,7 @@ func New(ctx resource.Context, cfg Config) (*Echos, error) {
 
 	if ctx.Settings().Ambient {
 
-		waypointProxies := make(map[string]ambient.WaypointProxy)
+		waypointProxies := make(map[string]ambient.Waypoints)
 
 		for _, echo := range echos {
 			svcwp := echo.Config().ServiceWaypointProxy

pkg/test/framework/components/echo/config.go

@@ -377,6 +377,10 @@ func (c Config) HasProxyCapabilities() bool {
 	return !c.IsUncaptured() || c.HasSidecar() || c.IsProxylessGRPC()
 }
 
+func (c Config) IsAmbient() bool {
+	return c.HasProxyCapabilities() && !c.HasSidecar()
+}
+
 func (c Config) IsVM() bool {
 	return c.DeployAsVM
 }

pkg/test/framework/components/echo/echotest/run.go

@@ -220,6 +220,11 @@ func (t *T) RunViaIngress(testFn ingressTest) {
 			doTest(ctx, ctx.Clusters()[0], dstInstances)
 		} else {
 			t.fromEachCluster(ctx, func(ctx framework.TestContext, c cluster.Cluster) {
+				if ctx.Settings().AmbientMultiNetwork {
+					// Ambient multi-network does not yet support routing across clusters from ingress
+					// https://github.com/istio/istio/issues/57537
+					dstInstances = dstInstances.ForCluster(c.Name())
+				}
 				doTest(ctx, c, dstInstances)
 			})
 		}

pkg/test/framework/components/istio/ca.go

@@ -46,7 +46,7 @@ func CreateCertificateForCluster(t framework.TestContext, i Instance, serviceAcc
 		return Cert{}, fmt.Errorf("failed to fetch root cert: %v", err)
 	}
 
-	token, err := GetServiceAccountToken(c.Kube(), "istio-ca", namespace, serviceAccount)
+	token, err := GetServiceAccountToken(c, "istio-ca", namespace, serviceAccount)
 	if err != nil {
 		return Cert{}, err
 	}
@@ -96,12 +96,13 @@ func CreateCertificate(t framework.TestContext, i Instance, serviceAccount, name
 // 7 days
 var saTokenExpiration int64 = 60 * 60 * 24 * 7
 
-func GetServiceAccountToken(c kubernetes.Interface, aud, ns, sa string) (string, error) {
+func GetServiceAccountToken(cluster cluster.Cluster, aud, ns, sa string) (string, error) {
 	san := san(ns, sa)
-
-	if got, f := cachedTokens.Load(san); f {
+	c := cluster.Kube()
+	key := fmt.Sprintf("%s:%s", cluster.Name(), san)
+	if got, f := cachedTokens.Load(key); f {
 		t := got.(token)
-		if t.expiration.After(time.Now().Add(time.Minute)) {
+		if t.expiration.After(time.Now().Add(5 * time.Minute)) {
 			return t.token, nil
 		}
 		// Otherwise, its expired, load a new one
@@ -117,7 +118,7 @@ func GetServiceAccountToken(c kubernetes.Interface, aud, ns, sa string) (string,
 		return "", err
 	}
 	exp := rt.Status.ExpirationTimestamp.Time
-	cachedTokens.Store(san, token{rt.Status.Token, exp})
+	cachedTokens.Store(key, token{rt.Status.Token, exp})
 	return rt.Status.Token, nil
 }
 

pkg/test/framework/components/registryredirector/kube.go

@@ -54,8 +54,9 @@ type kubeComponent struct {
 }
 
 func newKube(ctx resource.Context, cfg Config) (Instance, error) {
+	cluster := ctx.Clusters().GetOrDefault(cfg.Cluster)
 	c := &kubeComponent{
-		cluster: ctx.Clusters().GetOrDefault(cfg.Cluster),
+		cluster: cluster,
 	}
 	c.id = ctx.TrackResource(c)
 	var err error
@@ -96,7 +97,7 @@ func newKube(ctx resource.Context, cfg Config) (Instance, error) {
 		return nil, fmt.Errorf("failed to apply rendered %s, err: %v", env.RegistryRedirectorServerInstallFilePath, err)
 	}
 
-	fetchFn := testKube.NewPodFetch(ctx.Clusters().Default(), c.ns.Name(), podSelector)
+	fetchFn := testKube.NewPodFetch(cluster, c.ns.Name(), podSelector)
 	pods, err := testKube.WaitUntilPodsAreReady(fetchFn)
 	if err != nil {
 		return nil, err