test: add a v2 container integration test of xRoute splits

test-integ/catalogv2/explicit_destinations_test.go

@@ -167,8 +167,8 @@ func (c testBasicL4ExplicitDestinationsCreator) topologyConfigAddNodes(
 				newServiceID("single-client"),
 				topology.NodeVersionV2,
 				func(svc *topology.Service) {
-					delete(svc.Ports, "grpc")     // v2 mode turns this on, so turn it off
-					delete(svc.Ports, "http-alt") // v2 mode turns this on, so turn it off
+					delete(svc.Ports, "grpc")  // v2 mode turns this on, so turn it off
+					delete(svc.Ports, "http2") // v2 mode turns this on, so turn it off
 					svc.Upstreams = []*topology.Upstream{{
 						ID:           newServiceID("single-server"),
 						PortName:     "http",
@@ -232,7 +232,7 @@ func (c testBasicL4ExplicitDestinationsCreator) topologyConfigAddNodes(
 						},
 						{
 							ID:           newServiceID("multi-server"),
-							PortName:     "http-alt",
+							PortName:     "http2",
 							LocalAddress: "0.0.0.0", // needed for an assertion
 							LocalPort:    5001,
 						},

test-integ/catalogv2/helpers_test.go

@@ -11,12 +11,16 @@ import (
 
 func clusterPrefixForUpstream(u *topology.Upstream) string {
 	if u.Peer == "" {
-		if u.ID.PartitionOrDefault() == "default" {
-			return strings.Join([]string{u.PortName, u.ID.Name, u.ID.Namespace, u.Cluster, "internal"}, ".")
-		} else {
-			return strings.Join([]string{u.PortName, u.ID.Name, u.ID.Namespace, u.ID.Partition, u.Cluster, "internal-v1"}, ".")
-		}
+		return clusterPrefix(u.PortName, u.ID, u.Cluster)
 	} else {
 		return strings.Join([]string{u.ID.Name, u.ID.Namespace, u.Peer, "external"}, ".")
 	}
 }
+
+func clusterPrefix(port string, svcID topology.ServiceID, cluster string) string {
+	if svcID.PartitionOrDefault() == "default" {
+		return strings.Join([]string{port, svcID.Name, svcID.Namespace, cluster, "internal"}, ".")
+	} else {
+		return strings.Join([]string{port, svcID.Name, svcID.Namespace, svcID.Partition, cluster, "internal-v1"}, ".")
+	}
+}

test-integ/catalogv2/implicit_destinations_test.go

@@ -177,7 +177,7 @@ func (c testBasicL4ImplicitDestinationsCreator) topologyConfigAddNodes(
 						},
 						{
 							ID:       newServiceID("static-server"),
-							PortName: "http-alt",
+							PortName: "http2",
 						},
 					}
 				},

test-integ/go.mod

@@ -11,7 +11,10 @@ require (
 	github.com/hashicorp/go-cleanhttp v0.5.2
 	github.com/itchyny/gojq v0.12.13
 	github.com/mitchellh/copystructure v1.2.0
+	github.com/rboyer/blankspace v0.1.0
 	github.com/stretchr/testify v1.8.4
+	golang.org/x/net v0.17.0
+	google.golang.org/grpc v1.58.3
 )
 
 require (
@@ -97,12 +100,10 @@ require (
 	golang.org/x/crypto v0.14.0 // indirect
 	golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63 // indirect
 	golang.org/x/mod v0.12.0 // indirect
-	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/sys v0.13.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	golang.org/x/tools v0.12.1-0.20230815132531-74c255bcf846 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230803162519-f966b187b2e5 // indirect
-	google.golang.org/grpc v1.57.2 // indirect
 	google.golang.org/protobuf v1.31.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

test-integ/go.sum

@@ -259,6 +259,8 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
 github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
 github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5mo=
 github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4=
+github.com/rboyer/blankspace v0.1.0 h1:6AeQoyKXaKbWwaTrJ4jZ6mj0yaA4Bt5Pl7rte7S+pIY=
+github.com/rboyer/blankspace v0.1.0/go.mod h1:KjXEbJwg8BwZ0i6gVLQ4HZ0jQ8/85X3jd3dI29r2OJ4=
 github.com/rboyer/safeio v0.2.3 h1:gUybicx1kp8nuM4vO0GA5xTBX58/OBd8MQuErBfDxP8=
 github.com/rboyer/safeio v0.2.3/go.mod h1:d7RMmt7utQBJZ4B7f0H/cU/EdZibQAU1Y8NWepK2dS8=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
@@ -380,8 +382,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20230803162519-f966b187b2e5 h1:eSaPbMR4T7WfH9FvABk36NBMacoTUKdWCvV0dx+KfOg=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20230803162519-f966b187b2e5/go.mod h1:zBEcrKX2ZOcEkHWxBPAIvYUWOKKMIhYcmNiUIu2ji3I=
-google.golang.org/grpc v1.57.2 h1:uw37EN34aMFFXB2QPW7Tq6tdTbind1GpRxw5aOX3a5k=
-google.golang.org/grpc v1.57.2/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo=
+google.golang.org/grpc v1.58.3 h1:BjnpXut1btbtgN/6sp+brB2Kbm2LjNXnidYujAVbSoQ=
+google.golang.org/grpc v1.58.3/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=

test-integ/topoutil/asserter.go

@@ -260,7 +260,7 @@ func (a *Asserter) FortioFetch2HeaderEcho(t *testing.T, fortioSvc *topology.Serv
 
 	var (
 		node   = fortioSvc.Node
-		addr   = fmt.Sprintf("%s:%d", node.LocalAddress(), fortioSvc.PortOrDefault("http"))
+		addr   = fmt.Sprintf("%s:%d", node.LocalAddress(), fortioSvc.PortOrDefault(upstream.PortName))
 		client = a.mustGetHTTPClient(t, node.Cluster)
 	)
 
@@ -286,7 +286,7 @@ func (a *Asserter) FortioFetch2FortioName(
 
 	var (
 		node   = fortioSvc.Node
-		addr   = fmt.Sprintf("%s:%d", node.LocalAddress(), fortioSvc.PortOrDefault("http"))
+		addr   = fmt.Sprintf("%s:%d", node.LocalAddress(), fortioSvc.PortOrDefault(upstream.PortName))
 		client = a.mustGetHTTPClient(t, node.Cluster)
 	)
 

test-integ/topoutil/asserter_blankspace.go

@@ -0,0 +1,237 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package topoutil
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/consul/sdk/testutil/retry"
+	"github.com/hashicorp/consul/testing/deployer/topology"
+	"github.com/stretchr/testify/require"
+)
+
+func (a *Asserter) CheckBlankspaceNameViaHTTP(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	useHTTP2 bool,
+	path string,
+	clusterName string,
+	sid topology.ServiceID,
+) {
+	t.Helper()
+
+	a.checkBlankspaceNameViaHTTPWithCallback(t, service, upstream, useHTTP2, path, 1, func(r *retry.R, remoteName string) {
+		require.Equal(r, fmt.Sprintf("%s::%s", clusterName, sid.String()), remoteName)
+	}, func(r *retry.R) {})
+}
+
+func (a *Asserter) CheckBlankspaceNameTrafficSplitViaHTTP(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	useHTTP2 bool,
+	path string,
+	expect map[string]int,
+	epsilon int,
+) {
+	got := make(map[string]int)
+	a.checkBlankspaceNameViaHTTPWithCallback(t, service, upstream, useHTTP2, path, 100, func(_ *retry.R, name string) {
+		got[name]++
+	}, func(r *retry.R) {
+		assertTrafficSplit(r, got, expect, epsilon)
+	})
+}
+
+func (a *Asserter) checkBlankspaceNameViaHTTPWithCallback(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	useHTTP2 bool,
+	path string,
+	count int,
+	attemptFn func(r *retry.R, remoteName string),
+	checkFn func(r *retry.R),
+) {
+	t.Helper()
+
+	var (
+		node   = service.Node
+		addr   = fmt.Sprintf("%s:%d", node.LocalAddress(), service.PortOrDefault(upstream.PortName))
+		client = a.mustGetHTTPClient(t, node.Cluster)
+	)
+
+	if useHTTP2 {
+		client = EnableHTTP2(client)
+	}
+
+	var actualURL string
+	if upstream.Implied {
+		actualURL = fmt.Sprintf("http://%s--%s--%s.virtual.consul:%d/%s",
+			upstream.ID.Name,
+			upstream.ID.Namespace,
+			upstream.ID.Partition,
+			upstream.VirtualPort,
+			path,
+		)
+	} else {
+		actualURL = fmt.Sprintf("http://localhost:%d/%s", upstream.LocalPort, path)
+	}
+
+	multiassert(t, count, func(r *retry.R) {
+		name, err := GetBlankspaceNameViaHTTP(context.Background(), client, addr, actualURL)
+		require.NoError(r, err)
+		attemptFn(r, name)
+	}, func(r *retry.R) {
+		checkFn(r)
+	})
+}
+
+func (a *Asserter) CheckBlankspaceNameViaTCP(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	clusterName string,
+	sid topology.ServiceID,
+) {
+	t.Helper()
+
+	a.checkBlankspaceNameViaTCPWithCallback(t, service, upstream, 1, func(r *retry.R, remoteName string) {
+		require.Equal(r, fmt.Sprintf("%s::%s", clusterName, sid.String()), remoteName)
+	}, func(r *retry.R) {})
+}
+
+func (a *Asserter) CheckBlankspaceNameTrafficSplitViaTCP(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	expect map[string]int,
+	epsilon int,
+) {
+	got := make(map[string]int)
+	a.checkBlankspaceNameViaTCPWithCallback(t, service, upstream, 100, func(_ *retry.R, name string) {
+		got[name]++
+	}, func(r *retry.R) {
+		assertTrafficSplit(r, got, expect, epsilon)
+	})
+}
+
+func (a *Asserter) checkBlankspaceNameViaTCPWithCallback(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	count int,
+	attemptFn func(r *retry.R, remoteName string),
+	checkFn func(r *retry.R),
+) {
+	t.Helper()
+
+	require.False(t, upstream.Implied, "helper does not support tproxy yet")
+	port := upstream.LocalPort
+	require.True(t, port > 0)
+
+	node := service.Node
+
+	// We can't use the forward proxy for TCP yet, so use the exposed port on localhost instead.
+	exposedPort := node.ExposedPort(port)
+	require.True(t, exposedPort > 0)
+
+	addr := fmt.Sprintf("%s:%d", "127.0.0.1", exposedPort)
+
+	multiassert(t, count, func(r *retry.R) {
+		name, err := GetBlankspaceNameViaTCP(context.Background(), addr)
+		require.NoError(r, err)
+		attemptFn(r, name)
+	}, func(r *retry.R) {
+		checkFn(r)
+	})
+}
+
+func (a *Asserter) CheckBlankspaceNameViaGRPC(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	clusterName string,
+	sid topology.ServiceID,
+) {
+	t.Helper()
+
+	a.checkBlankspaceNameViaGRPCWithCallback(t, service, upstream, 1, func(r *retry.R, remoteName string) {
+		require.Equal(r, fmt.Sprintf("%s::%s", clusterName, sid.String()), remoteName)
+	}, func(r *retry.R) {})
+}
+
+func (a *Asserter) CheckBlankspaceNameTrafficSplitViaGRPC(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	expect map[string]int,
+	epsilon int,
+) {
+	got := make(map[string]int)
+	a.checkBlankspaceNameViaGRPCWithCallback(t, service, upstream, 100, func(_ *retry.R, name string) {
+		got[name]++
+	}, func(r *retry.R) {
+		assertTrafficSplit(r, got, expect, epsilon)
+	})
+}
+
+func (a *Asserter) checkBlankspaceNameViaGRPCWithCallback(
+	t *testing.T,
+	service *topology.Service,
+	upstream *topology.Upstream,
+	count int,
+	attemptFn func(r *retry.R, remoteName string),
+	checkFn func(r *retry.R),
+) {
+	t.Helper()
+
+	require.False(t, upstream.Implied, "helper does not support tproxy yet")
+	port := upstream.LocalPort
+	require.True(t, port > 0)
+
+	node := service.Node
+
+	// We can't use the forward proxy for gRPC yet, so use the exposed port on localhost instead.
+	exposedPort := node.ExposedPort(port)
+	require.True(t, exposedPort > 0)
+
+	addr := fmt.Sprintf("%s:%d", "127.0.0.1", exposedPort)
+
+	multiassert(t, count, func(r *retry.R) {
+		name, err := GetBlankspaceNameViaGRPC(context.Background(), addr)
+		require.NoError(r, err)
+		attemptFn(r, name)
+	}, func(r *retry.R) {
+		checkFn(r)
+	})
+}
+
+func assertTrafficSplit(t require.TestingT, nameCounts map[string]int, expect map[string]int, epsilon int) {
+	require.Len(t, nameCounts, len(expect))
+	for name, expectCount := range expect {
+		gotCount, ok := nameCounts[name]
+		require.True(t, ok)
+		if len(expect) == 1 {
+			require.Equal(t, expectCount, gotCount)
+		} else {
+			require.InEpsilon(t, expectCount, gotCount, float64(epsilon),
+				"expected %q side of split to have %d requests not %d (e=%d)",
+				name, expectCount, gotCount, epsilon,
+			)
+		}
+	}
+}
+
+func multiassert(t *testing.T, count int, attemptFn, checkFn func(r *retry.R)) {
+	retry.RunWith(&retry.Timer{Timeout: 30 * time.Second, Wait: 500 * time.Millisecond}, t, func(r *retry.R) {
+		for i := 0; i < count; i++ {
+			attemptFn(r)
+		}
+		checkFn(r)
+	})
+}