E2E test cases for waypoint management

.codespellrc

@@ -1,3 +1,3 @@
 [codespell]
 skip = .git,go.sum,go.mod,*.png,*.svg
-ignore-words-list = kmesh,outter,nd,Donot,donot,doesnot
+ignore-words-list = kmesh,outter,nd,Donot,donot,doesnot,Failer

test/e2e/baseline_test.go

@@ -24,19 +24,24 @@
 package kmesh
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"strings"
 	"testing"
 	"time"
 
+	"istio.io/istio/pkg/config/constants"
 	echot "istio.io/istio/pkg/test/echo"
 	"istio.io/istio/pkg/test/echo/common/scheme"
 	"istio.io/istio/pkg/test/framework"
 	"istio.io/istio/pkg/test/framework/components/echo"
 	"istio.io/istio/pkg/test/framework/components/echo/check"
 	"istio.io/istio/pkg/test/framework/components/echo/common/ports"
+	"istio.io/istio/pkg/test/scopes"
 	"istio.io/istio/pkg/util/sets"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 func IsL7() echo.Checker {
@@ -443,6 +448,132 @@ spec:
 	})
 }
 
+// Test add/remove waypoint at pod granularity.
+func TestAddRemovePodWaypoint(t *testing.T) {
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+		waypoint := "pod-waypoint"
+		newWaypointProxyOrFail(t, t, apps.Namespace, waypoint, constants.WorkloadTraffic)
+
+		t.Cleanup(func() {
+			deleteWaypointProxyOrFail(t, t, apps.Namespace, waypoint)
+		})
+
+		dst := apps.EnrolledToKmesh
+		t.NewSubTest("before").Run(func(t framework.TestContext) {
+			for _, src := range apps.All {
+				if src.Config().IsUncaptured() {
+					continue
+				}
+				for _, dstWl := range dst.WorkloadsOrFail(t) {
+					t.NewSubTestf("from %v", src.Config().Service).Run(func(t framework.TestContext) {
+						c := IsL4()
+						opt := echo.CallOptions{
+							Address: dstWl.Address(),
+							Port:    echo.Port{ServicePort: ports.All().MustForName("http").WorkloadPort},
+							Scheme:  scheme.HTTP,
+							Count:   10,
+							Check:   check.And(check.OK(), c),
+						}
+						src.CallOrFail(t, opt)
+					})
+				}
+
+			}
+		})
+
+		// Configure pods to use waypoint.
+		for _, dstWl := range dst.WorkloadsOrFail(t) {
+			SetWaypoint(t, apps.Namespace.Name(), dstWl.PodName(), waypoint, Workload)
+		}
+
+		// Now should always be L7.
+		t.NewSubTest("after").Run(func(t framework.TestContext) {
+			for _, src := range apps.All {
+				if src.Config().IsUncaptured() {
+					continue
+				}
+				for _, dstWl := range dst.WorkloadsOrFail(t) {
+					t.NewSubTestf("from %v", src.Config().Service).Run(func(t framework.TestContext) {
+						c := IsL4()
+						opt := echo.CallOptions{
+							Address: dstWl.Address(),
+							Port:    echo.Port{ServicePort: ports.All().MustForName("http").WorkloadPort},
+							Scheme:  scheme.HTTP,
+							Count:   10,
+							Check:   check.And(check.OK(), c),
+						}
+						src.CallOrFail(t, opt)
+					})
+				}
+
+			}
+		})
+	})
+}
+
+// Test add/remove waypoint at ns or service granularity.
+func TestRemoveAddNsOrServiceWaypoint(t *testing.T) {
+	for _, granularity := range []Granularity{Service /*,Namespace*/} {
+		framework.NewTest(t).Run(func(t framework.TestContext) {
+			var waypoint string
+			switch granularity {
+			case Namespace:
+				waypoint = "namespace-waypoint"
+			case Service:
+				waypoint = "service-waypoint"
+			}
+
+			newWaypointProxyOrFail(t, t, apps.Namespace, waypoint, constants.ServiceTraffic)
+
+			t.Cleanup(func() {
+				deleteWaypointProxyOrFail(t, t, apps.Namespace, waypoint)
+			})
+
+			t.NewSubTest("before").Run(func(t framework.TestContext) {
+				dst := apps.EnrolledToKmesh
+				for _, src := range apps.All {
+					if src.Config().IsUncaptured() {
+						continue
+					}
+					t.NewSubTestf("from %v", src.Config().Service).Run(func(t framework.TestContext) {
+						c := IsL4()
+						opt := echo.CallOptions{
+							To:     dst,
+							Port:   echo.Port{Name: "http"},
+							Scheme: scheme.HTTP,
+							Count:  10,
+							Check:  check.And(check.OK(), c),
+						}
+						src.CallOrFail(t, opt)
+					})
+				}
+			})
+
+			SetWaypoint(t, apps.Namespace.Name(), EnrolledToKmesh, waypoint, granularity)
+
+			// Now should always be L7
+			t.NewSubTest("after").Run(func(t framework.TestContext) {
+				dst := apps.EnrolledToKmesh
+				for _, src := range apps.All {
+					if src.Config().IsUncaptured() {
+						continue
+					}
+					t.NewSubTestf("from %v", src.Config().Service).Run(func(t framework.TestContext) {
+						opt := echo.CallOptions{
+							To:     dst,
+							Port:   echo.Port{Name: "http"},
+							Scheme: scheme.HTTP,
+							Count:  10,
+							Check:  check.And(check.OK(), IsL7()),
+						}
+						src.CallOrFail(t, opt)
+					})
+				}
+			})
+		})
+	}
+}
+
 func runTest(t *testing.T, f func(t framework.TestContext, src echo.Instance, dst echo.Instance, opt echo.CallOptions)) {
 	framework.NewTest(t).Run(func(t framework.TestContext) {
 		runTestContext(t, f)
@@ -487,3 +618,47 @@ func runTestContext(t framework.TestContext, f func(t framework.TestContext, src
 		})
 	}
 }
+
+type Granularity int
+
+const (
+	Namespace Granularity = iota
+	Service
+	Workload
+)
+
+func SetWaypoint(t framework.TestContext, ns string, name string, waypoint string, granularity Granularity) {
+	for _, c := range t.Clusters() {
+		setWaypoint := func(waypoint string) error {
+			if waypoint == "" {
+				waypoint = "null"
+			} else {
+				waypoint = fmt.Sprintf("%q", waypoint)
+			}
+			label := []byte(fmt.Sprintf(`{"metadata":{"labels":{"%s":%s}}}`, constants.AmbientUseWaypointLabel, waypoint))
+
+			switch granularity {
+			case Namespace:
+				_, err := c.Kube().CoreV1().Namespaces().Patch(context.TODO(), ns, types.MergePatchType, label, metav1.PatchOptions{})
+				return err
+			case Service:
+				_, err := c.Kube().CoreV1().Services(ns).Patch(context.TODO(), name, types.MergePatchType, label, metav1.PatchOptions{})
+				return err
+			case Workload:
+				_, err := c.Kube().CoreV1().Pods(ns).Patch(context.TODO(), name, types.MergePatchType, label, metav1.PatchOptions{})
+				return err
+			}
+
+			return nil
+		}
+
+		if err := setWaypoint(waypoint); err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() {
+			if err := setWaypoint(""); err != nil {
+				scopes.Framework.Errorf("failed resetting waypoint for %s/%s", ns, name)
+			}
+		})
+	}
+}

test/e2e/main_test.go

@@ -37,12 +37,14 @@ import (
 	"istio.io/istio/pkg/config/protocol"
 	"istio.io/istio/pkg/config/schema/gvk"
 	istioKube "istio.io/istio/pkg/kube"
+	"istio.io/istio/pkg/test"
 	"istio.io/istio/pkg/test/framework"
 	"istio.io/istio/pkg/test/framework/components/ambient"
 	"istio.io/istio/pkg/test/framework/components/crd"
 	"istio.io/istio/pkg/test/framework/components/echo"
 	"istio.io/istio/pkg/test/framework/components/echo/common/ports"
 	"istio.io/istio/pkg/test/framework/components/echo/deployment"
+	"istio.io/istio/pkg/test/framework/components/echo/match"
 	"istio.io/istio/pkg/test/framework/components/istio"
 	"istio.io/istio/pkg/test/framework/components/namespace"
 	"istio.io/istio/pkg/test/framework/resource"
@@ -69,6 +71,9 @@ type EchoDeployments struct {
 	// All echo services
 	All echo.Instances
 
+	// The echo service which is enrolled to Kmesh without waypoint.
+	EnrolledToKmesh echo.Instances
+
 	// WaypointProxies by
 	WaypointProxies map[string]ambient.WaypointProxy
 }
@@ -170,6 +175,7 @@ func SetupApps(t resource.Context, i istio.Instance, apps *EchoDeployments) erro
 		scopes.Framework.Infof("built %v", b.Config().Service)
 	}
 	apps.All = echos
+	apps.EnrolledToKmesh = match.ServiceName(echo.NamespacedName{Name: EnrolledToKmesh, Namespace: apps.Namespace}).GetMatches(echos)
 
 	if apps.WaypointProxies == nil {
 		apps.WaypointProxies = make(map[string]ambient.WaypointProxy)
@@ -180,15 +186,15 @@ func SetupApps(t resource.Context, i istio.Instance, apps *EchoDeployments) erro
 		wlwp := echo.Config().WorkloadWaypointProxy
 		if svcwp != "" {
 			if _, found := apps.WaypointProxies[svcwp]; !found {
-				apps.WaypointProxies[svcwp], err = newWaypointProxy(t, apps.Namespace, svcwp)
+				apps.WaypointProxies[svcwp], err = newWaypointProxy(t, apps.Namespace, svcwp, constants.ServiceTraffic)
 				if err != nil {
 					return err
 				}
 			}
 		}
 		if wlwp != "" {
 			if _, found := apps.WaypointProxies[wlwp]; !found {
-				apps.WaypointProxies[wlwp], err = newWaypointProxy(t, apps.Namespace, wlwp)
+				apps.WaypointProxies[wlwp], err = newWaypointProxy(t, apps.Namespace, wlwp, constants.WorkloadTraffic)
 				if err != nil {
 					return err
 				}
@@ -240,7 +246,13 @@ func (k kubeComponent) Close() error {
 	return nil
 }
 
-func newWaypointProxy(ctx resource.Context, ns namespace.Instance, name string) (ambient.WaypointProxy, error) {
+func newWaypointProxyOrFail(t test.Failer, ctx resource.Context, ns namespace.Instance, name string, trafficType string) {
+	if _, err := newWaypointProxy(ctx, ns, name, trafficType); err != nil {
+		t.Fatal("create new waypoint proxy failed: %v", err)
+	}
+}
+
+func newWaypointProxy(ctx resource.Context, ns namespace.Instance, name string, trafficType string) (ambient.WaypointProxy, error) {
 	err := crd.DeployGatewayAPI(ctx)
 	if err != nil {
 		return nil, err
@@ -255,6 +267,9 @@ func newWaypointProxy(ctx resource.Context, ns namespace.Instance, name string)
 			Name:        name,
 			Namespace:   ns.Name(),
 			Annotations: make(map[string]string, 0),
+			Labels: map[string]string{
+				constants.AmbientWaypointForTrafficTypeLabel: trafficType,
+			},
 		},
 		Spec: gateway.GatewaySpec{
 			GatewayClassName: constants.WaypointGatewayClassName,
@@ -314,3 +329,15 @@ func newWaypointProxy(ctx resource.Context, ns namespace.Instance, name string)
 
 	return server, nil
 }
+
+func deleteWaypointProxyOrFail(t test.Failer, ctx resource.Context, ns namespace.Instance, name string) {
+	if err := deleteWaypointProxy(ctx, ns, name); err != nil {
+		t.Fatal("delete waypoint proxy failed: %v", err)
+	}
+}
+
+func deleteWaypointProxy(ctx resource.Context, ns namespace.Instance, name string) error {
+	cls := ctx.Clusters().Default()
+
+	return cls.GatewayAPI().GatewayV1().Gateways(ns.Name()).Delete(context.Background(), name, metav1.DeleteOptions{})
+}