test(integration): add access logging test (#16008)

test/integration/consul-container/libs/assert/service.go

@@ -7,9 +7,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/go-cleanhttp"
+	"github.com/stretchr/testify/require"
+
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/sdk/testutil/retry"
-	"github.com/hashicorp/go-cleanhttp"
+	libservice "github.com/hashicorp/consul/test/integration/consul-container/libs/service"
 )
 
 const (
@@ -17,9 +20,22 @@ const (
 	defaultHTTPWait    = defaultWait
 )
 
+// CatalogServiceExists verifies the service name exists in the Consul catalog
+func CatalogServiceExists(t *testing.T, c *api.Client, svc string) {
+	retry.Run(t, func(r *retry.R) {
+		services, _, err := c.Catalog().Service(svc, "", nil)
+		if err != nil {
+			r.Fatal("error reading peering data")
+		}
+		if len(services) == 0 {
+			r.Fatal("did not find catalog entry for ", svc)
+		}
+	})
+}
+
 // HTTPServiceEchoes verifies that a post to the given ip/port combination returns the data
-// in the response body
+// in the response body. Optional path can be provided to differentiate requests.
-func HTTPServiceEchoes(t *testing.T, ip string, port int) {
+func HTTPServiceEchoes(t *testing.T, ip string, port int, path string) {
 	const phrase = "hello"
 
 	failer := func() *retry.Timer {
@@ -29,6 +45,10 @@ func HTTPServiceEchoes(t *testing.T, ip string, port int) {
 	client := cleanhttp.DefaultClient()
 	url := fmt.Sprintf("http://%s:%d", ip, port)
 
+	if path != "" {
+		url += "/" + path
+	}
+
 	retry.RunWith(failer(), t, func(r *retry.R) {
 		t.Logf("making call to %s", url)
 		reader := strings.NewReader(phrase)
@@ -49,15 +69,9 @@ func HTTPServiceEchoes(t *testing.T, ip string, port int) {
 	})
 }
 
-// CatalogServiceExists verifies the service name exists in the Consul catalog
+// ServiceLogContains returns true if the service container has the target string in its logs
-func CatalogServiceExists(t *testing.T, c *api.Client, svc string) {
+func ServiceLogContains(t *testing.T, service libservice.Service, target string) bool {
-	retry.Run(t, func(r *retry.R) {
+	logs, err := service.GetLogs()
-		services, _, err := c.Catalog().Service(svc, "", nil)
+	require.NoError(t, err)
-		if err != nil {
+	return strings.Contains(logs, target)
-			r.Fatal("error reading peering data")
-		}
-		if len(services) == 0 {
-			r.Fatal("did not find catalog entry for ", svc)
-		}
-	})
 }

test/integration/consul-container/libs/cluster/agent.go

@@ -3,9 +3,10 @@ package cluster
 import (
 	"context"
 
-	"github.com/hashicorp/consul/api"
 	"github.com/testcontainers/testcontainers-go"
 
+	"github.com/hashicorp/consul/api"
+
 	"github.com/hashicorp/consul/test/integration/consul-container/libs/utils"
 )
 

test/integration/consul-container/libs/cluster/container.go

@@ -10,12 +10,13 @@ import (
 	"time"
 
 	dockercontainer "github.com/docker/docker/api/types/container"
-	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 	"github.com/testcontainers/testcontainers-go"
 	"github.com/testcontainers/testcontainers-go/wait"
 
+	"github.com/hashicorp/consul/api"
+
 	"github.com/hashicorp/consul/test/integration/consul-container/libs/utils"
 )
 
@@ -44,7 +45,8 @@ type consulContainerNode struct {
 	clientCACertFile string
 	ip               string
 
-	nextAdminPortOffset int
+	nextAdminPortOffset   int
+	nextConnectPortOffset int
 
 	info AgentInfo
 }
@@ -428,11 +430,11 @@ func newContainerRequest(config Config, opts containerOpts) (podRequest, consulR
 
 			"8443/tcp", // Envoy Gateway Listener
 
-			"5000/tcp", // Envoy Connect Listener
+			"5000/tcp", // Envoy App Listener
-			"8079/tcp", // Envoy Connect Listener
+			"8079/tcp", // Envoy App Listener
-			"8080/tcp", // Envoy Connect Listener
+			"8080/tcp", // Envoy App Listener
-			"9998/tcp", // Envoy Connect Listener
+			"9998/tcp", // Envoy App Listener
-			"9999/tcp", // Envoy Connect Listener
+			"9999/tcp", // Envoy App Listener
 
 			"19000/tcp", // Envoy Admin Port
 			"19001/tcp", // Envoy Admin Port

test/integration/consul-container/libs/service/connect.go

@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 	"fmt"
+	"io"
 	"path/filepath"
 	"strconv"
 	"time"
@@ -19,16 +20,39 @@ import (
 
 // ConnectContainer
 type ConnectContainer struct {
-	ctx         context.Context
+	ctx              context.Context
-	container   testcontainers.Container
+	container        testcontainers.Container
-	ip          string
+	ip               string
-	appPort     int
+	appPort          int
-	adminPort   int
+	adminPort        int
-	serviceName string
+	mappedPublicPort int
+	serviceName      string
 }
 
 var _ Service = (*ConnectContainer)(nil)
 
+func (g ConnectContainer) Export(partition, peer string, client *api.Client) error {
+	return fmt.Errorf("ConnectContainer export unimplemented")
+}
+
+func (g ConnectContainer) GetAddr() (string, int) {
+	return g.ip, g.appPort
+}
+
+func (g ConnectContainer) GetLogs() (string, error) {
+	rc, err := g.container.Logs(context.Background())
+	if err != nil {
+		return "", fmt.Errorf("could not get logs for connect service %s: %w", g.GetServiceName(), err)
+	}
+	defer rc.Close()
+
+	out, err := io.ReadAll(rc)
+	if err != nil {
+		return "", fmt.Errorf("could not read from logs for connect service %s: %w", g.GetServiceName(), err)
+	}
+	return string(out), nil
+}
+
 func (g ConnectContainer) GetName() string {
 	name, err := g.container.Name(g.ctx)
 	if err != nil {
@@ -37,8 +61,8 @@ func (g ConnectContainer) GetName() string {
 	return name
 }
 
-func (g ConnectContainer) GetAddr() (string, int) {
+func (g ConnectContainer) GetServiceName() string {
-	return g.ip, g.appPort
+	return g.serviceName
 }
 
 func (g ConnectContainer) Start() error {
@@ -48,20 +72,12 @@ func (g ConnectContainer) Start() error {
 	return g.container.Start(context.Background())
 }
 
-func (g ConnectContainer) GetAdminAddr() (string, int) {
-	return "localhost", g.adminPort
-}
-
 func (c ConnectContainer) Terminate() error {
 	return cluster.TerminateContainer(c.ctx, c.container, true)
 }
 
-func (g ConnectContainer) Export(partition, peer string, client *api.Client) error {
+func (g ConnectContainer) GetAdminAddr() (string, int) {
-	return fmt.Errorf("ConnectContainer export unimplemented")
+	return "localhost", g.adminPort
-}
-
-func (g ConnectContainer) GetServiceName() string {
-	return g.serviceName
 }
 
 // NewConnectService returns a container that runs envoy sidecar, launched by
@@ -154,7 +170,7 @@ func NewConnectService(ctx context.Context, name string, serviceName string, ser
 		serviceName: name,
 	}
 
-	fmt.Printf("NewConnectService: name %s, mappedAppPort %d, bind port %d\n",
+	fmt.Printf("NewConnectService: name %s, bind port %d, public listener port %d\\n\"",
 		serviceName, out.appPort, serviceBindPort)
 
 	return out, nil

test/integration/consul-container/libs/service/examples.go

@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 	"fmt"
+	"io"
 	"strconv"
 	"time"
 
@@ -28,29 +29,6 @@ type exampleContainer struct {
 
 var _ Service = (*exampleContainer)(nil)
 
-func (g exampleContainer) GetName() string {
-	name, err := g.container.Name(g.ctx)
-	if err != nil {
-		return ""
-	}
-	return name
-}
-
-func (g exampleContainer) GetAddr() (string, int) {
-	return g.ip, g.httpPort
-}
-
-func (g exampleContainer) Start() error {
-	if g.container == nil {
-		return fmt.Errorf("container has not been initialized")
-	}
-	return g.container.Start(context.Background())
-}
-
-func (c exampleContainer) Terminate() error {
-	return cluster.TerminateContainer(c.ctx, c.container, true)
-}
-
 func (g exampleContainer) Export(partition, peerName string, client *api.Client) error {
 	config := &api.ExportedServicesConfigEntry{
 		Name: partition,
@@ -67,10 +45,47 @@ func (g exampleContainer) Export(partition, peerName string, client *api.Client)
 	return err
 }
 
+func (g exampleContainer) GetAddr() (string, int) {
+	return g.ip, g.httpPort
+}
+
+func (g exampleContainer) GetLogs() (string, error) {
+	rc, err := g.container.Logs(context.Background())
+	if err != nil {
+		return "", fmt.Errorf("could not get logs for example service %s: %w", g.GetServiceName(), err)
+	}
+	defer rc.Close()
+
+	out, err := io.ReadAll(rc)
+	if err != nil {
+		return "", fmt.Errorf("could not read from logs for example service %s: %w", g.GetServiceName(), err)
+	}
+	return string(out), nil
+}
+
+func (g exampleContainer) GetName() string {
+	name, err := g.container.Name(g.ctx)
+	if err != nil {
+		return ""
+	}
+	return name
+}
+
 func (g exampleContainer) GetServiceName() string {
 	return g.serviceName
 }
 
+func (g exampleContainer) Start() error {
+	if g.container == nil {
+		return fmt.Errorf("container has not been initialized")
+	}
+	return g.container.Start(context.Background())
+}
+
+func (c exampleContainer) Terminate() error {
+	return cluster.TerminateContainer(c.ctx, c.container, true)
+}
+
 func NewExampleService(ctx context.Context, name string, httpPort int, grpcPort int, node libcluster.Agent) (Service, error) {
 	namePrefix := fmt.Sprintf("%s-service-example-%s", node.GetDatacenter(), name)
 	containerName := utils.RandName(namePrefix)

test/integration/consul-container/libs/service/gateway.go

@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 	"fmt"
+	"io"
 	"path/filepath"
 	"strconv"
 	"time"
@@ -29,6 +30,28 @@ type gatewayContainer struct {
 
 var _ Service = (*gatewayContainer)(nil)
 
+func (g gatewayContainer) Export(partition, peer string, client *api.Client) error {
+	return fmt.Errorf("gatewayContainer export unimplemented")
+}
+
+func (g gatewayContainer) GetAddr() (string, int) {
+	return g.ip, g.port
+}
+
+func (g gatewayContainer) GetLogs() (string, error) {
+	rc, err := g.container.Logs(context.Background())
+	if err != nil {
+		return "", fmt.Errorf("could not get logs for gateway service %s: %w", g.GetServiceName(), err)
+	}
+	defer rc.Close()
+
+	out, err := io.ReadAll(rc)
+	if err != nil {
+		return "", fmt.Errorf("could not read from logs for gateway service %s: %w", g.GetServiceName(), err)
+	}
+	return string(out), nil
+}
+
 func (g gatewayContainer) GetName() string {
 	name, err := g.container.Name(g.ctx)
 	if err != nil {
@@ -37,12 +60,8 @@ func (g gatewayContainer) GetName() string {
 	return name
 }
 
-func (g gatewayContainer) GetAddr() (string, int) {
+func (g gatewayContainer) GetServiceName() string {
-	return g.ip, g.port
+	return g.serviceName
-}
-
-func (g gatewayContainer) GetAdminAddr() (string, int) {
-	return "localhost", g.adminPort
 }
 
 func (g gatewayContainer) Start() error {
@@ -56,12 +75,8 @@ func (c gatewayContainer) Terminate() error {
 	return cluster.TerminateContainer(c.ctx, c.container, true)
 }
 
-func (g gatewayContainer) Export(partition, peer string, client *api.Client) error {
+func (g gatewayContainer) GetAdminAddr() (string, int) {
-	return fmt.Errorf("gatewayContainer export unimplemented")
+	return "localhost", g.adminPort
-}
-
-func (g gatewayContainer) GetServiceName() string {
-	return g.serviceName
 }
 
 func NewGatewayService(ctx context.Context, name string, kind string, node libcluster.Agent) (Service, error) {

test/integration/consul-container/libs/service/helpers.go

@@ -5,39 +5,29 @@ import (
 	"fmt"
 	"io"
 
-	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/go-cleanhttp"
 
+	"github.com/hashicorp/consul/api"
+
 	libcluster "github.com/hashicorp/consul/test/integration/consul-container/libs/cluster"
 	"github.com/hashicorp/consul/test/integration/consul-container/libs/utils"
 )
 
+const (
+	StaticServerServiceName = "static-server"
+	StaticClientServiceName = "static-client"
+)
+
 func CreateAndRegisterStaticServerAndSidecar(node libcluster.Agent) (Service, Service, error) {
 	// Do some trickery to ensure that partial completion is correctly torn
 	// down, but successful execution is not.
 	var deferClean utils.ResettableDefer
 	defer deferClean.Execute()
 
-	// Create a service and proxy instance
+	// Register the static-server service and sidecar first to prevent race with sidecar
-	serverService, err := NewExampleService(context.Background(), "static-server", 8080, 8079, node)
+	// trying to get xDS before it's ready
-	if err != nil {
-		return nil, nil, err
-	}
-	deferClean.Add(func() {
-		_ = serverService.Terminate()
-	})
-
-	serverConnectProxy, err := NewConnectService(context.Background(), "static-server-sidecar", "static-server", 8080, node) // bindPort not used
-	if err != nil {
-		return nil, nil, err
-	}
-	deferClean.Add(func() {
-		_ = serverConnectProxy.Terminate()
-	})
-
-	// Register the static-server service and sidecar
 	req := &api.AgentServiceRegistration{
-		Name: "static-server",
+		Name: StaticServerServiceName,
 		Port: 8080,
 		Connect: &api.AgentServiceConnect{
 			SidecarService: &api.AgentServiceRegistration{
@@ -52,11 +42,27 @@ func CreateAndRegisterStaticServerAndSidecar(node libcluster.Agent) (Service, Se
 		},
 	}
 
-	err = node.GetClient().Agent().ServiceRegister(req)
+	if err := node.GetClient().Agent().ServiceRegister(req); err != nil {
-	if err != nil {
+		return nil, nil, err
-		return serverService, serverConnectProxy, err
 	}
 
+	// Create a service and proxy instance
+	serverService, err := NewExampleService(context.Background(), StaticServerServiceName, 8080, 8079, node)
+	if err != nil {
+		return nil, nil, err
+	}
+	deferClean.Add(func() {
+		_ = serverService.Terminate()
+	})
+
+	serverConnectProxy, err := NewConnectService(context.Background(), fmt.Sprintf("%s-sidecar", StaticServerServiceName), StaticServerServiceName, 8080, node) // bindPort not used
+	if err != nil {
+		return nil, nil, err
+	}
+	deferClean.Add(func() {
+		_ = serverConnectProxy.Terminate()
+	})
+
 	// disable cleanup functions now that we have an object with a Terminate() function
 	deferClean.Reset()
 
@@ -73,29 +79,21 @@ func CreateAndRegisterStaticClientSidecar(
 	var deferClean utils.ResettableDefer
 	defer deferClean.Execute()
 
-	// Create a service and proxy instance
-	clientConnectProxy, err := NewConnectService(context.Background(), "static-client-sidecar", "static-client", 5000, node)
-	if err != nil {
-		return nil, err
-	}
-	deferClean.Add(func() {
-		_ = clientConnectProxy.Terminate()
-	})
-
 	mgwMode := api.MeshGatewayModeRemote
 	if localMeshGateway {
 		mgwMode = api.MeshGatewayModeLocal
 	}
 
-	// Register the static-client service and sidecar
+	// Register the static-client service and sidecar first to prevent race with sidecar
+	// trying to get xDS before it's ready
 	req := &api.AgentServiceRegistration{
-		Name: "static-client",
+		Name: StaticClientServiceName,
 		Port: 8080,
 		Connect: &api.AgentServiceConnect{
 			SidecarService: &api.AgentServiceRegistration{
 				Proxy: &api.AgentServiceConnectProxyConfig{
 					Upstreams: []api.Upstream{{
-						DestinationName:  "static-server",
+						DestinationName:  StaticServerServiceName,
 						DestinationPeer:  peerName,
 						LocalBindAddress: "0.0.0.0",
 						LocalBindPort:    5000,
@@ -108,11 +106,19 @@ func CreateAndRegisterStaticClientSidecar(
 		},
 	}
 
-	err = node.GetClient().Agent().ServiceRegister(req)
+	if err := node.GetClient().Agent().ServiceRegister(req); err != nil {
-	if err != nil {
+		return nil, err
-		return clientConnectProxy, err
 	}
 
+	// Create a service and proxy instance
+	clientConnectProxy, err := NewConnectService(context.Background(), fmt.Sprintf("%s-sidecar", StaticClientServiceName), StaticClientServiceName, 5000, node)
+	if err != nil {
+		return nil, err
+	}
+	deferClean.Add(func() {
+		_ = clientConnectProxy.Terminate()
+	})
+
 	// disable cleanup functions now that we have an object with a Terminate() function
 	deferClean.Reset()
 

test/integration/consul-container/libs/service/service.go

@@ -5,12 +5,13 @@ import "github.com/hashicorp/consul/api"
 // Service represents a process that will be registered with the
 // Consul catalog, including Consul components such as sidecars and gateways
 type Service interface {
-	Terminate() error
-	GetName() string
-	GetAddr() (string, int)
-	Start() (err error)
 	// Export a service to the peering cluster
 	Export(partition, peer string, client *api.Client) error
-	GetServiceName() string
+	GetAddr() (string, int)
 	GetAdminAddr() (string, int)
+	GetLogs() (string, error)
+	GetName() string
+	GetServiceName() string
+	Start() (err error)
+	Terminate() error
 }

test/integration/consul-container/libs/topology/peering_topology.go

@@ -5,9 +5,10 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/hashicorp/consul/api"
 	"github.com/stretchr/testify/require"
 
+	"github.com/hashicorp/consul/api"
+
 	libassert "github.com/hashicorp/consul/test/integration/consul-container/libs/assert"
 	libcluster "github.com/hashicorp/consul/test/integration/consul-container/libs/cluster"
 	libservice "github.com/hashicorp/consul/test/integration/consul-container/libs/service"
@@ -78,7 +79,7 @@ func BasicPeeringTwoClustersSetup(
 	}
 
 	_, port := clientSidecarService.GetAddr()
-	libassert.HTTPServiceEchoes(t, "localhost", port)
+	libassert.HTTPServiceEchoes(t, "localhost", port, "")
 
 	return &BuiltCluster{
 			Cluster:   acceptingCluster,

test/integration/consul-container/test/basic/connect_service_test.go

@@ -27,7 +27,7 @@ func TestBasicConnectService(t *testing.T) {
 	_, port := clientService.GetAddr()
 	_, adminPort := clientService.GetAdminAddr()
 
-	libassert.HTTPServiceEchoes(t, "localhost", port)
+	libassert.HTTPServiceEchoes(t, "localhost", port, "")
 	libassert.GetEnvoyListenerTCPFilters(t, adminPort)
 }
 

test/integration/consul-container/test/observability/access_logs_test.go

@@ -0,0 +1,149 @@
+package observability
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/go-cleanhttp"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/mod/semver"
+
+	"github.com/hashicorp/consul/api"
+	libassert "github.com/hashicorp/consul/test/integration/consul-container/libs/assert"
+	libcluster "github.com/hashicorp/consul/test/integration/consul-container/libs/cluster"
+	libservice "github.com/hashicorp/consul/test/integration/consul-container/libs/service"
+	"github.com/hashicorp/consul/test/integration/consul-container/libs/topology"
+	"github.com/hashicorp/consul/test/integration/consul-container/libs/utils"
+)
+
+// TestAccessLogs Summary
+// This test ensures that when enabled through `proxy-defaults`, Envoy will emit access logs.
+// Philosophically, we are trying to ensure the config options make their way to Envoy more than
+// trying to test Envoy's behavior. For this reason and simplicity file operations are not tested.
+//
+// Steps:
+//   - Create a single agent cluster.
+//   - Enable default access logs. We do this so Envoy's admin interface inherits the configuration on startup
+//   - Create the example static-server and sidecar containers, then register them both with Consul
+//   - Create an example static-client sidecar, then register both the service and sidecar with Consul
+//   - Make sure a call to the client sidecar emits an access log at the client-sidecar (outbound) and
+//     server-sidecar (inbound).
+//   - Make sure hitting the Envoy admin interface generates an access log
+//   - Change access log configuration to use custom text format and disable Listener logs
+//   - Make sure a call to the client sidecar emits an access log at the client-sidecar (outbound) and
+//     server-sidecar (inbound).
+//
+// Notes:
+//   - Does not test disabling listener logs. In practice, it's hard to get them to emit. The best chance would
+//     be running a service that throws a 404 on a random path or maybe use some path-based disco chains
+//   - JSON keys have no guaranteed ordering, so simple key-value pairs are tested
+//   - Because it takes a while for xDS updates to make it to envoy, it's not obvious when turning off access logs
+//     will actually cause the proxies to update. Testing this proved difficult.
+func TestAccessLogs(t *testing.T) {
+	if semver.IsValid(utils.TargetVersion) && semver.Compare(utils.TargetVersion, "v1.15") < 0 {
+		t.Skip()
+	}
+
+	cluster, _, _ := topology.NewPeeringCluster(t, "dc1", 1, "")
+
+	// Turn on access logs. Do this before starting the sidecars so that they inherit the configuration
+	// for their admin interface
+	proxyDefault := &api.ProxyConfigEntry{
+		Kind: api.ProxyDefaults,
+		Name: api.ProxyConfigGlobal,
+		AccessLogs: &api.AccessLogsConfig{
+			Enabled:    true,
+			JSONFormat: "{\"banana_path\":\"%REQ(X-ENVOY-ORIGINAL-PATH?:PATH)%\"}",
+		},
+	}
+	set, _, err := cluster.Agents[0].GetClient().ConfigEntries().Set(proxyDefault, nil)
+	require.NoError(t, err)
+	require.True(t, set)
+
+	serverService, clientService := createServices(t, cluster)
+	_, port := clientService.GetAddr()
+
+	// Validate Custom JSON
+	require.Eventually(t, func() bool {
+		libassert.HTTPServiceEchoes(t, "localhost", port, "banana")
+		client := libassert.ServiceLogContains(t, clientService, "\"banana_path\":\"/banana\"")
+		server := libassert.ServiceLogContains(t, serverService, "\"banana_path\":\"/banana\"")
+		return client && server
+	}, 60*time.Second, 1*time.Second)
+
+	// Validate Logs on the Admin Interface
+	serverSidecar, ok := serverService.(*libservice.ConnectContainer)
+	require.True(t, ok)
+	ip, port := serverSidecar.GetAdminAddr()
+
+	httpClient := cleanhttp.DefaultClient()
+	url := fmt.Sprintf("http://%s:%d/clusters?fruit=bananas", ip, port)
+	_, err = httpClient.Get(url)
+	require.NoError(t, err, "error making call to Envoy admin interface")
+
+	require.Eventually(t, func() bool {
+		return libassert.ServiceLogContains(t, serverService, "\"banana_path\":\"/clusters?fruit=bananas\"")
+	}, 15*time.Second, 1*time.Second)
+
+	// TODO: add a test to check that connections without a matching filter chain are logged
+
+	// Validate Listener Logs
+	proxyDefault = &api.ProxyConfigEntry{
+		Kind: api.ProxyDefaults,
+		Name: api.ProxyConfigGlobal,
+		AccessLogs: &api.AccessLogsConfig{
+			Enabled:             true,
+			DisableListenerLogs: true,
+			TextFormat:          "Orange you glad I didn't say banana: %REQ(X-ENVOY-ORIGINAL-PATH?:PATH)%, %RESPONSE_FLAGS%",
+		},
+	}
+
+	set, _, err = cluster.Agents[0].GetClient().ConfigEntries().Set(proxyDefault, nil)
+	require.NoError(t, err)
+	require.True(t, set)
+	time.Sleep(5 * time.Second) // time for xDS to propagate
+
+	// Validate Custom Text
+	_, port = clientService.GetAddr()
+	require.Eventually(t, func() bool {
+		libassert.HTTPServiceEchoes(t, "localhost", port, "orange")
+		client := libassert.ServiceLogContains(t, clientService, "Orange you glad I didn't say banana: /orange, -")
+		server := libassert.ServiceLogContains(t, serverService, "Orange you glad I didn't say banana: /orange, -")
+		return client && server
+	}, 60*time.Second, 500*time.Millisecond) // For some reason it takes a long time for the server sidecar to update
+
+	// TODO: add a test to check that connections without a matching filter chain are NOT logged
+
+}
+
+func createServices(t *testing.T, cluster *libcluster.Cluster) (libservice.Service, libservice.Service) {
+	node := cluster.Agents[0]
+	client := node.GetClient()
+
+	// Register service as HTTP
+	serviceDefault := &api.ServiceConfigEntry{
+		Kind:     api.ServiceDefaults,
+		Name:     libservice.StaticServerServiceName,
+		Protocol: "http",
+	}
+
+	ok, _, err := client.ConfigEntries().Set(serviceDefault, nil)
+	require.NoError(t, err, "error writing HTTP service-default")
+	require.True(t, ok, "did not write HTTP service-default")
+
+	// Create a service and proxy instance
+	_, serverConnectProxy, err := libservice.CreateAndRegisterStaticServerAndSidecar(node)
+	require.NoError(t, err)
+
+	libassert.CatalogServiceExists(t, client, fmt.Sprintf("%s-sidecar-proxy", libservice.StaticServerServiceName))
+	libassert.CatalogServiceExists(t, client, libservice.StaticServerServiceName)
+
+	// Create a client proxy instance with the server as an upstream
+	clientConnectProxy, err := libservice.CreateAndRegisterStaticClientSidecar(node, "", false)
+	require.NoError(t, err)
+
+	libassert.CatalogServiceExists(t, client, fmt.Sprintf("%s-sidecar-proxy", libservice.StaticClientServiceName))
+
+	return serverConnectProxy, clientConnectProxy
+}

test/integration/consul-container/test/observability/metrics_leader_test.go

@@ -1,13 +1,13 @@
-package metrics
+package observability
 
 import (
 	"strings"
 	"testing"
 
-	"github.com/hashicorp/consul/api"
-	"github.com/hashicorp/consul/sdk/testutil/retry"
 	"github.com/stretchr/testify/require"
 
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/sdk/testutil/retry"
 	libcluster "github.com/hashicorp/consul/test/integration/consul-container/libs/cluster"
 )
 

test/integration/consul-container/test/peering/rotate_server_and_ca_then_fail_test.go

@@ -6,10 +6,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/sdk/testutil"
 	"github.com/hashicorp/consul/sdk/testutil/retry"
-	"github.com/stretchr/testify/require"
 
 	libassert "github.com/hashicorp/consul/test/integration/consul-container/libs/assert"
 	libcluster "github.com/hashicorp/consul/test/integration/consul-container/libs/cluster"
@@ -89,7 +90,7 @@ func TestPeering_RotateServerAndCAThenFail_(t *testing.T) {
 		libassert.PeeringExports(t, client, peerName, 1)
 
 		_, port := clientSidecarService.GetAddr()
-		libassert.HTTPServiceEchoes(t, "localhost", port)
+		libassert.HTTPServiceEchoes(t, "localhost", port, "")
 	}
 
 	testutil.RunStep(t, "rotate exporting cluster's root CA", func(t *testing.T) {
@@ -138,7 +139,7 @@ func TestPeering_RotateServerAndCAThenFail_(t *testing.T) {
 
 		// Connectivity should still be contained
 		_, port := clientSidecarService.GetAddr()
-		libassert.HTTPServiceEchoes(t, "localhost", port)
+		libassert.HTTPServiceEchoes(t, "localhost", port, "")
 
 		verifySidecarHasTwoRootCAs(t, clientSidecarService)
 	})
@@ -159,7 +160,7 @@ func TestPeering_RotateServerAndCAThenFail_(t *testing.T) {
 		time.Sleep(30 * time.Second)
 
 		_, port := clientSidecarService.GetAddr()
-		libassert.HTTPServiceEchoes(t, "localhost", port)
+		libassert.HTTPServiceEchoes(t, "localhost", port, "")
 	})
 }