add integration tests for troubleshoot (#16223)

* draft

* expose internal admin port and add proxy test

* update tests

* move comment

* add failure case, fix lint issues

* cleanup

* handle error

* revert changes to service interface

* address review comments

* fix merge conflict

* merge the tests so cluster is created once

* fix other test

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

@@ -20,17 +20,33 @@ import (
 
 // ConnectContainer
 type ConnectContainer struct {
-	ctx              context.Context
-	container        testcontainers.Container
-	ip               string
-	appPort          int
-	adminPort        int
-	mappedPublicPort int
-	serviceName      string
+	ctx               context.Context
+	container         testcontainers.Container
+	ip                string
+	appPort           int
+	externalAdminPort int
+	internalAdminPort int
+	mappedPublicPort  int
+	serviceName       string
 }
 
 var _ Service = (*ConnectContainer)(nil)
 
+func (g ConnectContainer) Exec(ctx context.Context, cmd []string) (string, error) {
+	exitCode, reader, err := g.container.Exec(ctx, cmd)
+	if err != nil {
+		return "", fmt.Errorf("exec with error %s", err)
+	}
+	if exitCode != 0 {
+		return "", fmt.Errorf("exec with exit code %d", exitCode)
+	}
+	buf, err := io.ReadAll(reader)
+	if err != nil {
+		return "", fmt.Errorf("error reading from exec output: %w", err)
+	}
+	return string(buf), nil
+}
+
 func (g ConnectContainer) Export(partition, peer string, client *api.Client) error {
 	return fmt.Errorf("ConnectContainer export unimplemented")
 }
@@ -97,8 +113,13 @@ func (g ConnectContainer) Terminate() error {
 	return cluster.TerminateContainer(g.ctx, g.container, true)
 }
 
+func (g ConnectContainer) GetInternalAdminAddr() (string, int) {
+	return "localhost", g.internalAdminPort
+}
+
+// GetAdminAddr returns the external admin port
 func (g ConnectContainer) GetAdminAddr() (string, int) {
-	return "localhost", g.adminPort
+	return "localhost", g.externalAdminPort
 }
 
 func (g ConnectContainer) GetStatus() (string, error) {
@@ -133,7 +154,7 @@ func NewConnectService(ctx context.Context, sidecarServiceName string, serviceID
 	}
 	dockerfileCtx.BuildArgs = buildargs
 
-	adminPort, err := node.ClaimAdminPort()
+	internalAdminPort, err := node.ClaimAdminPort()
 	if err != nil {
 		return nil, err
 	}
@@ -146,7 +167,7 @@ func NewConnectService(ctx context.Context, sidecarServiceName string, serviceID
 		Cmd: []string{
 			"consul", "connect", "envoy",
 			"-sidecar-for", serviceID,
-			"-admin-bind", fmt.Sprintf("0.0.0.0:%d", adminPort),
+			"-admin-bind", fmt.Sprintf("0.0.0.0:%d", internalAdminPort),
 			"--",
 			"--log-level", envoyLogLevel,
 		},
@@ -182,7 +203,7 @@ func NewConnectService(ctx context.Context, sidecarServiceName string, serviceID
 
 	var (
 		appPortStr   = strconv.Itoa(serviceBindPort)
-		adminPortStr = strconv.Itoa(adminPort)
+		adminPortStr = strconv.Itoa(internalAdminPort)
 	)
 
 	info, err := cluster.LaunchContainerOnNode(ctx, node, req, []string{appPortStr, adminPortStr})
@@ -191,18 +212,19 @@ func NewConnectService(ctx context.Context, sidecarServiceName string, serviceID
 	}
 
 	out := &ConnectContainer{
-		ctx:         ctx,
-		container:   info.Container,
-		ip:          info.IP,
-		appPort:     info.MappedPorts[appPortStr].Int(),
-		adminPort:   info.MappedPorts[adminPortStr].Int(),
-		serviceName: sidecarServiceName,
+		ctx:               ctx,
+		container:         info.Container,
+		ip:                info.IP,
+		appPort:           info.MappedPorts[appPortStr].Int(),
+		externalAdminPort: info.MappedPorts[adminPortStr].Int(),
+		internalAdminPort: internalAdminPort,
+		serviceName:       sidecarServiceName,
 	}
 
 	fmt.Printf("NewConnectService: name %s, mapped App Port %d, service bind port %d\n",
 		serviceID, out.appPort, serviceBindPort)
 	fmt.Printf("NewConnectService sidecar: name %s, mapped admin port %d, admin port %d\n",
-		sidecarServiceName, out.adminPort, adminPort)
+		sidecarServiceName, out.externalAdminPort, internalAdminPort)
 
 	return out, nil
 }

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

@@ -29,6 +29,21 @@ type exampleContainer struct {
 
 var _ Service = (*exampleContainer)(nil)
 
+func (g exampleContainer) Exec(ctx context.Context, cmd []string) (string, error) {
+	exitCode, reader, err := g.container.Exec(ctx, cmd)
+	if err != nil {
+		return "", fmt.Errorf("exec with error %s", err)
+	}
+	if exitCode != 0 {
+		return "", fmt.Errorf("exec with exit code %d", exitCode)
+	}
+	buf, err := io.ReadAll(reader)
+	if err != nil {
+		return "", fmt.Errorf("error reading from exec output: %w", err)
+	}
+	return string(buf), nil
+}
+
 func (g exampleContainer) Export(partition, peerName string, client *api.Client) error {
 	config := &api.ExportedServicesConfigEntry{
 		Name: partition,

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

@@ -30,6 +30,21 @@ type gatewayContainer struct {
 
 var _ Service = (*gatewayContainer)(nil)
 
+func (g gatewayContainer) Exec(ctx context.Context, cmd []string) (string, error) {
+	exitCode, reader, err := g.container.Exec(ctx, cmd)
+	if err != nil {
+		return "", fmt.Errorf("exec with error %s", err)
+	}
+	if exitCode != 0 {
+		return "", fmt.Errorf("exec with exit code %d", exitCode)
+	}
+	buf, err := io.ReadAll(reader)
+	if err != nil {
+		return "", fmt.Errorf("error reading from exec output: %w", err)
+	}
+	return string(buf), nil
+}
+
 func (g gatewayContainer) Export(partition, peer string, client *api.Client) error {
 	return fmt.Errorf("gatewayContainer export unimplemented")
 }

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

@@ -3,9 +3,7 @@ package service
 import (
 	"context"
 	"fmt"
-
 	"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"
 )

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

@@ -1,13 +1,18 @@
 package service
 
-import "github.com/hashicorp/consul/api"
+import (
+	"context"
+	"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 {
+	Exec(ctx context.Context, cmd []string) (string, error)
 	// Export a service to the peering cluster
 	Export(partition, peer string, client *api.Client) error
 	GetAddr() (string, int)
+	// GetAdminAddr returns the external admin address
 	GetAdminAddr() (string, int)
 	GetLogs() (string, error)
 	GetName() string

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

@@ -0,0 +1,51 @@
+package topology
+
+import (
+	"fmt"
+	"testing"
+
+	"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/stretchr/testify/require"
+)
+
+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
+	serviceOpts := &libservice.ServiceOpts{
+		Name:     libservice.StaticServerServiceName,
+		ID:       "static-server",
+		HTTPPort: 8080,
+		GRPCPort: 8079,
+	}
+
+	// Create a service and proxy instance
+	_, serverConnectProxy, err := libservice.CreateAndRegisterStaticServerAndSidecar(node, serviceOpts)
+	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/access_logs_test.go

@@ -64,7 +64,7 @@ func TestAccessLogs(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, set)
 
-	serverService, clientService := createServices(t, cluster)
+	serverService, clientService := topology.CreateServices(t, cluster)
 	_, port := clientService.GetAddr()
 
 	// Validate Custom JSON
@@ -121,42 +121,3 @@ func TestAccessLogs(t *testing.T) {
 	// 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
-	serviceOpts := &libservice.ServiceOpts{
-		Name:     libservice.StaticServerServiceName,
-		ID:       "static-server",
-		HTTPPort: 8080,
-		GRPCPort: 8079,
-	}
-
-	// Create a service and proxy instance
-	_, serverConnectProxy, err := libservice.CreateAndRegisterStaticServerAndSidecar(node, serviceOpts)
-	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/troubleshoot/troubleshoot_test.go

@@ -0,0 +1,68 @@
+package troubleshoot
+
+import (
+	"context"
+	"fmt"
+	"github.com/stretchr/testify/assert"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	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"
+)
+
+func TestTroubleshootProxy(t *testing.T) {
+	t.Parallel()
+	cluster, _, _ := topology.NewPeeringCluster(t, 1, &libcluster.BuildOptions{
+		Datacenter:           "dc1",
+		InjectAutoEncryption: true,
+	})
+
+	serverService, clientService := topology.CreateServices(t, cluster)
+
+	clientSidecar, ok := clientService.(*libservice.ConnectContainer)
+	require.True(t, ok)
+	_, clientAdminPort := clientSidecar.GetInternalAdminAddr()
+
+	t.Run("upstream exists and is healthy", func(t *testing.T) {
+		require.Eventually(t, func() bool {
+			output, err := clientSidecar.Exec(context.Background(),
+				[]string{"consul", "troubleshoot", "upstreams",
+					"-envoy-admin-endpoint", fmt.Sprintf("localhost:%v", clientAdminPort)})
+			require.NoError(t, err)
+			upstreamExists := assert.Contains(t, output, libservice.StaticServerServiceName)
+
+			output, err = clientSidecar.Exec(context.Background(), []string{"consul", "troubleshoot", "proxy",
+				"-envoy-admin-endpoint", fmt.Sprintf("localhost:%v", clientAdminPort),
+				"-upstream-envoy-id", libservice.StaticServerServiceName})
+			require.NoError(t, err)
+			certsValid := strings.Contains(output, "certificates are valid")
+			listenersExist := strings.Contains(output, fmt.Sprintf("listener for upstream \"%s\" found", libservice.StaticServerServiceName))
+			routesExist := strings.Contains(output, fmt.Sprintf("route for upstream \"%s\" found", libservice.StaticServerServiceName))
+			healthyEndpoints := strings.Contains(output, "✓ healthy endpoints for cluster")
+			return upstreamExists && certsValid && listenersExist && routesExist && healthyEndpoints
+		}, 60*time.Second, 10*time.Second)
+	})
+
+	t.Run("terminate upstream and check if client sees it as unhealthy", func(t *testing.T) {
+		err := serverService.Terminate()
+		require.NoError(t, err)
+
+		require.Eventually(t, func() bool {
+			output, err := clientSidecar.Exec(context.Background(), []string{"consul", "troubleshoot", "proxy",
+				"-envoy-admin-endpoint", fmt.Sprintf("localhost:%v", clientAdminPort),
+				"-upstream-envoy-id", libservice.StaticServerServiceName})
+			require.NoError(t, err)
+
+			certsValid := strings.Contains(output, "certificates are valid")
+			listenersExist := strings.Contains(output, fmt.Sprintf("listener for upstream \"%s\" found", libservice.StaticServerServiceName))
+			routesExist := strings.Contains(output, fmt.Sprintf("route for upstream \"%s\" found", libservice.StaticServerServiceName))
+			endpointUnhealthy := strings.Contains(output, "no healthy endpoints for cluster")
+			return certsValid && listenersExist && routesExist && endpointUnhealthy
+		}, 60*time.Second, 10*time.Second)
+	})
+}