connect/proxy: add a full proxy test, parallel

connect/proxy/config_test.go

@@ -15,6 +15,8 @@ import (
 )
 
 func TestParseConfigFile(t *testing.T) {
+	t.Parallel()
+
 	cfg, err := ParseConfigFile("testdata/config-kitchensink.hcl")
 	require.Nil(t, err)
 
@@ -54,6 +56,8 @@ func TestParseConfigFile(t *testing.T) {
 }
 
 func TestUpstreamResolverFromClient(t *testing.T) {
+	t.Parallel()
+
 	tests := []struct {
 		name string
 		cfg  UpstreamConfig
@@ -115,6 +119,8 @@ func TestUpstreamResolverFromClient(t *testing.T) {
 }
 
 func TestAgentConfigWatcher(t *testing.T) {
+	t.Parallel()
+
 	a := agent.NewTestAgent("agent_smith", "")
 	defer a.Shutdown()
 

connect/proxy/conn_test.go

@@ -64,6 +64,8 @@ func testConnPipelineSetup(t *testing.T) (net.Conn, net.Conn, *Conn, func()) {
 }
 
 func TestConn(t *testing.T) {
+	t.Parallel()
+
 	src, dst, c, stop := testConnPipelineSetup(t)
 	defer stop()
 
@@ -117,6 +119,8 @@ func TestConn(t *testing.T) {
 }
 
 func TestConnSrcClosing(t *testing.T) {
+	t.Parallel()
+
 	src, dst, c, stop := testConnPipelineSetup(t)
 	defer stop()
 
@@ -155,6 +159,8 @@ func TestConnSrcClosing(t *testing.T) {
 }
 
 func TestConnDstClosing(t *testing.T) {
+	t.Parallel()
+
 	src, dst, c, stop := testConnPipelineSetup(t)
 	defer stop()
 

connect/proxy/listener_test.go

@@ -15,23 +15,23 @@ import (
 )
 
 func TestPublicListener(t *testing.T) {
+	t.Parallel()
+
 	ca := agConnect.TestCA(t, nil)
-	ports := freeport.GetT(t, 2)
+	ports := freeport.GetT(t, 1)
+
+	testApp := NewTestTCPServer(t)
+	defer testApp.Close()
 
 	cfg := PublicListenerConfig{
 		BindAddress:           "127.0.0.1",
 		BindPort:              ports[0],
-		LocalServiceAddress:   TestLocalAddr(ports[1]),
+		LocalServiceAddress:   testApp.Addr().String(),
 		HandshakeTimeoutMs:    100,
 		LocalConnectTimeoutMs: 100,
 	}
 
-	testApp, err := NewTestTCPServer(t, cfg.LocalServiceAddress)
-	require.NoError(t, err)
-	defer testApp.Close()
-
 	svc := connect.TestService(t, "db", ca)
-
 	l := NewPublicListener(svc, cfg, log.New(os.Stderr, "", log.LstdFlags))
 
 	// Run proxy
@@ -53,6 +53,8 @@ func TestPublicListener(t *testing.T) {
 }
 
 func TestUpstreamListener(t *testing.T) {
+	t.Parallel()
+
 	ca := agConnect.TestCA(t, nil)
 	ports := freeport.GetT(t, 1)
 

connect/proxy/proxy_test.go

@@ -0,0 +1,79 @@
+package proxy
+
+import (
+	"context"
+	"log"
+	"net"
+	"os"
+	"testing"
+
+	"github.com/hashicorp/consul/agent"
+	agConnect "github.com/hashicorp/consul/agent/connect"
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/connect"
+	"github.com/hashicorp/consul/lib/freeport"
+	"github.com/hashicorp/consul/testutil/retry"
+	"github.com/stretchr/testify/require"
+)
+
+func TestProxy_public(t *testing.T) {
+	t.Parallel()
+
+	require := require.New(t)
+	ports := freeport.GetT(t, 1)
+
+	a := agent.NewTestAgent(t.Name(), "")
+	defer a.Shutdown()
+	client := a.Client()
+
+	// Register the service so we can get a leaf cert
+	_, err := client.Catalog().Register(&api.CatalogRegistration{
+		Datacenter: "dc1",
+		Node:       "local",
+		Address:    "127.0.0.1",
+		Service: &api.AgentService{
+			Service: "echo",
+		},
+	}, nil)
+	require.NoError(err)
+
+	// Start the backend service that is being proxied
+	testApp := NewTestTCPServer(t)
+	defer testApp.Close()
+
+	// Start the proxy
+	p, err := New(client, NewStaticConfigWatcher(&Config{
+		ProxiedServiceName: "echo",
+		PublicListener: PublicListenerConfig{
+			BindAddress:         "127.0.0.1",
+			BindPort:            ports[0],
+			LocalServiceAddress: testApp.Addr().String(),
+		},
+	}), testLogger(t))
+	require.NoError(err)
+	defer p.Close()
+	go p.Serve()
+
+	// Create a test connection to the proxy. We retry here a few times
+	// since this is dependent on the agent actually starting up and setting
+	// up the CA.
+	var conn net.Conn
+	svc, err := connect.NewService("echo", client)
+	require.NoError(err)
+	retry.Run(t, func(r *retry.R) {
+		conn, err = svc.Dial(context.Background(), &connect.StaticResolver{
+			Addr:    TestLocalAddr(ports[0]),
+			CertURI: agConnect.TestSpiffeIDService(t, "echo"),
+		})
+		if err != nil {
+			r.Fatalf("err: %s", err)
+		}
+	})
+
+	// Connection works, test it is the right one
+	TestEchoConn(t, conn, "")
+}
+
+func testLogger(t *testing.T) *log.Logger {
+	return log.New(os.Stderr, "", log.LstdFlags)
+}

connect/proxy/testing.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"sync/atomic"
 
+	"github.com/hashicorp/consul/lib/freeport"
 	"github.com/mitchellh/go-testing-interface"
 	"github.com/stretchr/testify/require"
 )
@@ -26,17 +27,20 @@ type TestTCPServer struct {
 // NewTestTCPServer opens as a listening socket on the given address and returns
 // a TestTCPServer serving requests to it. The server is already started and can
 // be stopped by calling Close().
-func NewTestTCPServer(t testing.T, addr string) (*TestTCPServer, error) {
+func NewTestTCPServer(t testing.T) *TestTCPServer {
+	port := freeport.GetT(t, 1)
+	addr := TestLocalAddr(port[0])
+
 	l, err := net.Listen("tcp", addr)
-	if err != nil {
+	require.NoError(t, err)
-		return nil, err
+
-	}
 	log.Printf("test tcp server listening on %s", addr)
 	s := &TestTCPServer{
 		l: l,
 	}
 	go s.accept()
-	return s, nil
+
+	return s
 }
 
 // Close stops the server
@@ -47,6 +51,11 @@ func (s *TestTCPServer) Close() {
 	}
 }
 
+// Addr returns the address that this server is listening on.
+func (s *TestTCPServer) Addr() net.Addr {
+	return s.l.Addr()
+}
+
 func (s *TestTCPServer) accept() error {
 	for {
 		conn, err := s.l.Accept()