Pass flags, add tests, cleanup and polish

Author: jshahriddhi

cmd/consul-dataplane/main.go

@@ -32,6 +32,9 @@ var (
 	adminBindPort int
 	readyBindAddr string
 	readyBindPort int
+
+	xdsBindAddr string
+	xdsBindPort int
 )
 
 func init() {
@@ -63,6 +66,9 @@ func init() {
 	flag.IntVar(&adminBindPort, "envoy-admin-bind-port", 19000, "The port on which the Envoy admin server will be available.")
 	flag.StringVar(&readyBindAddr, "envoy-ready-bind-address", "", "The address on which Envoy's readiness probe will be available.")
 	flag.IntVar(&readyBindPort, "envoy-ready-bind-port", 0, "The port on which Envoy's readiness probe will be available.")
+
+	flag.StringVar(&xdsBindAddr, "xds-bind-addr", "127.0.0.1", "The address on which the envoy xDS server will be available.")
+	flag.IntVar(&xdsBindPort, "xds-bind-port", 0, "The port on which the envoy xDS server will be available.")
 }
 
 // validateFlags performs semantic validation of the flag values
@@ -111,6 +117,10 @@ func main() {
 			ReadyBindAddress: readyBindAddr,
 			ReadyBindPort:    readyBindPort,
 		},
+		XDSServer: &consuldp.XDSServer{
+			BindAddress: xdsBindAddr,
+			BindPort:    xdsBindPort,
+		},
 	}
 	consuldpInstance, err := consuldp.NewConsulDP(consuldpCfg)
 	if err != nil {

pkg/consuldp/bootstrap.go

@@ -25,9 +25,6 @@ const (
 
 // bootstrapConfig generates the Envoy bootstrap config in JSON format.
 func (cdp *ConsulDataplane) bootstrapConfig(ctx context.Context) ([]byte, error) {
-	cdpGRPCFullAddr := strings.Split(cdp.gRPCServer.listener.Addr().String(), ":")
-	cdpGRPCAddr := cdpGRPCFullAddr[0]
-	cdpGRPCPort := cdpGRPCFullAddr[1]
 	svc := cdp.cfg.Service
 	envoy := cdp.cfg.Envoy
 
@@ -54,8 +51,8 @@ func (cdp *ConsulDataplane) bootstrapConfig(ctx context.Context) ([]byte, error)
 
 	args := &bootstrap.BootstrapTplArgs{
 		GRPC: bootstrap.GRPC{
-			AgentAddress: cdpGRPCAddr,
-			AgentPort:    cdpGRPCPort,
+			AgentAddress: cdp.cfg.XDSServer.BindAddress,
+			AgentPort:    strconv.Itoa(cdp.cfg.XDSServer.BindPort),
 			AgentTLS:     false,
 		},
 		ProxyCluster:          rsp.Service,
@@ -66,11 +63,19 @@ func (cdp *ConsulDataplane) bootstrapConfig(ctx context.Context) ([]byte, error)
 		AdminBindAddress:      envoy.AdminBindAddress,
 		AdminBindPort:         strconv.Itoa(envoy.AdminBindPort),
 		LocalAgentClusterName: localClusterName,
-		// TODO(NET-148): Support login via an ACL auth-method.
-		Token:      cdp.cfg.Consul.Credentials.Static.Token,
-		Namespace:  rsp.Namespace,
-		Partition:  rsp.Partition,
-		Datacenter: rsp.Datacenter,
+		Namespace:             rsp.Namespace,
+		Partition:             rsp.Partition,
+		Datacenter:            rsp.Datacenter,
+	}
+
+	if cdp.localXDSServer != nil && cdp.localXDSServer.enabled {
+		if cdp.localXDSServer.listenerNetwork == "unix" {
+			args.GRPC.AgentSocket = cdp.localXDSServer.listenerAddress
+		} else {
+			xdsServerFullAddr := strings.Split(cdp.localXDSServer.listenerAddress, ":")
+			args.GRPC.AgentAddress = xdsServerFullAddr[0]
+			args.GRPC.AgentPort = xdsServerFullAddr[1]
+		}
 	}
 
 	var bootstrapConfig bootstrap.BootstrapConfig

pkg/consuldp/bootstrap_test.go

@@ -4,10 +4,11 @@ import (
 	"bytes"
 	"context"
 	"flag"
-	"net"
+	"fmt"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
@@ -24,8 +25,9 @@ var (
 
 func TestBootstrapConfig(t *testing.T) {
 	const (
-		serverAddr = "1.2.3.4"
-		nodeName   = "agentless-node"
+		nodeName    = "agentless-node"
+		xdsBindPort = 1234
+		socketPath  = "/var/run/xds.sock"
 	)
 
 	makeStruct := func(kv map[string]any) *structpb.Struct {
@@ -42,11 +44,6 @@ func TestBootstrapConfig(t *testing.T) {
 			&Config{
 				Consul: &ConsulConfig{
 					GRPCPort: 1234,
-					Credentials: &CredentialsConfig{
-						Static: &StaticCredentialsConfig{
-							Token: "some-acl-token",
-						},
-					},
 				},
 				Service: &ServiceConfig{
 					ServiceID: "web-proxy",
@@ -59,6 +56,7 @@ func TestBootstrapConfig(t *testing.T) {
 				Telemetry: &TelemetryConfig{
 					UseCentralConfig: false,
 				},
+				XDSServer: &XDSServer{BindAddress: "1.2.3.4", BindPort: xdsBindPort},
 			},
 			&pbdataplane.GetEnvoyBootstrapParamsResponse{
 				Service:  "web",
@@ -72,11 +70,6 @@ func TestBootstrapConfig(t *testing.T) {
 			&Config{
 				Consul: &ConsulConfig{
 					GRPCPort: 1234,
-					Credentials: &CredentialsConfig{
-						Static: &StaticCredentialsConfig{
-							Token: "some-acl-token",
-						},
-					},
 				},
 				Service: &ServiceConfig{
 					ServiceID: "web-proxy",
@@ -89,6 +82,7 @@ func TestBootstrapConfig(t *testing.T) {
 				Telemetry: &TelemetryConfig{
 					UseCentralConfig: true,
 				},
+				XDSServer: &XDSServer{BindAddress: "1.2.3.4", BindPort: xdsBindPort},
 			},
 			&pbdataplane.GetEnvoyBootstrapParamsResponse{
 				Service:  "web",
@@ -102,11 +96,6 @@ func TestBootstrapConfig(t *testing.T) {
 			&Config{
 				Consul: &ConsulConfig{
 					GRPCPort: 1234,
-					Credentials: &CredentialsConfig{
-						Static: &StaticCredentialsConfig{
-							Token: "some-acl-token",
-						},
-					},
 				},
 				Service: &ServiceConfig{
 					ServiceID: "web-proxy",
@@ -121,12 +110,65 @@ func TestBootstrapConfig(t *testing.T) {
 				Telemetry: &TelemetryConfig{
 					UseCentralConfig: false,
 				},
+				XDSServer: &XDSServer{BindAddress: "1.2.3.4", BindPort: xdsBindPort},
 			},
 			&pbdataplane.GetEnvoyBootstrapParamsResponse{
 				Service:  "web",
 				NodeName: nodeName,
 			},
 		},
+		"local-xds-server": {
+			&Config{
+				Consul: &ConsulConfig{
+					GRPCPort: 1234,
+				},
+				Service: &ServiceConfig{
+					ServiceID: "web-proxy",
+					NodeName:  nodeName,
+				},
+				Envoy: &EnvoyConfig{
+					AdminBindAddress: "127.0.0.1",
+					AdminBindPort:    19000,
+				},
+				Telemetry: &TelemetryConfig{
+					UseCentralConfig: false,
+				},
+				XDSServer: &XDSServer{BindAddress: "127.0.0.1", BindPort: xdsBindPort},
+			},
+			&pbdataplane.GetEnvoyBootstrapParamsResponse{
+				Service:  "web",
+				NodeName: nodeName,
+				Config: makeStruct(map[string]any{
+					"envoy_dogstatsd_url": "this-should-not-appear-in-generated-config",
+				}),
+			},
+		},
+		"local-unix-socket-xds-server": {
+			&Config{
+				Consul: &ConsulConfig{
+					GRPCPort: 1234,
+				},
+				Service: &ServiceConfig{
+					ServiceID: "web-proxy",
+					NodeName:  nodeName,
+				},
+				Envoy: &EnvoyConfig{
+					AdminBindAddress: "127.0.0.1",
+					AdminBindPort:    19000,
+				},
+				Telemetry: &TelemetryConfig{
+					UseCentralConfig: false,
+				},
+				XDSServer: &XDSServer{BindAddress: fmt.Sprintf("unix://%s", socketPath)},
+			},
+			&pbdataplane.GetEnvoyBootstrapParamsResponse{
+				Service:  "web",
+				NodeName: nodeName,
+				Config: makeStruct(map[string]any{
+					"envoy_dogstatsd_url": "this-should-not-appear-in-generated-config",
+				}),
+			},
+		},
 	}
 	for desc, tc := range testCases {
 		t.Run(desc, func(t *testing.T) {
@@ -144,7 +186,14 @@ func TestBootstrapConfig(t *testing.T) {
 			dp := &ConsulDataplane{
 				cfg:             tc.cfg,
 				dpServiceClient: client,
-				consulServer:    &consulServer{address: net.IPAddr{IP: net.ParseIP(serverAddr)}},
+			}
+
+			if checkLocalXDSServer(tc.cfg.XDSServer.BindAddress) {
+				if strings.HasPrefix(tc.cfg.XDSServer.BindAddress, "unix://") {
+					dp.localXDSServer = &localXDSServer{enabled: true, listenerAddress: socketPath, listenerNetwork: "unix"}
+				} else {
+					dp.localXDSServer = &localXDSServer{enabled: true, listenerAddress: fmt.Sprintf("127.0.0.1:%d", xdsBindPort)}
+				}
 			}
 
 			bsCfg, err := dp.bootstrapConfig(ctx)

pkg/consuldp/config.go

@@ -76,6 +76,14 @@ type EnvoyConfig struct {
 	ReadyBindPort int
 }
 
+//
+type XDSServer struct {
+	// BindAddress is the address on which the Envoy xDS server will be available.
+	BindAddress string
+	// BindPort is the address on which the Envoy xDS port will be available.
+	BindPort int
+}
+
 // Config is the configuration used by consul-dataplane, consolidated
 // from various sources - CLI flags, env vars, config file settings.
 type Config struct {
@@ -84,4 +92,5 @@ type Config struct {
 	Logging   *LoggingConfig
 	Telemetry *TelemetryConfig
 	Envoy     *EnvoyConfig
+	XDSServer *XDSServer
 }

pkg/consuldp/consul_dataplane.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"math/rand"
 	"net"
+	"strings"
 	"time"
 
 	"github.com/hashicorp/go-hclog"
@@ -29,10 +30,13 @@ type consulServer struct {
 	grpcClientConn *grpc.ClientConn
 }
 
-type gRPCServer struct {
-	listener net.Listener
-	server   *grpc.Server
-	exitedCh chan struct{}
+type localXDSServer struct {
+	enabled         bool
+	listener        net.Listener
+	listenerAddress string
+	listenerNetwork string
+	gRPCServer      *grpc.Server
+	exitedCh        chan struct{}
 }
 
 // ConsulDataplane represents the consul-dataplane process
@@ -41,7 +45,7 @@ type ConsulDataplane struct {
 	cfg             *Config
 	consulServer    *consulServer
 	dpServiceClient pbdataplane.DataplaneServiceClient
-	gRPCServer      *gRPCServer
+	localXDSServer  *localXDSServer
 }
 
 // NewConsulDP creates a new instance of ConsulDataplane
@@ -61,8 +65,9 @@ func NewConsulDP(cfg *Config) (*ConsulDataplane, error) {
 	})
 
 	return &ConsulDataplane{
-		logger: logger,
-		cfg:    cfg,
+		logger:         logger,
+		cfg:            cfg,
+		localXDSServer: &localXDSServer{enabled: false},
 	}, nil
 }
 
@@ -86,6 +91,10 @@ func validateConfig(cfg *Config) error {
 		return errors.New("envoy admin bind port not specified")
 	case cfg.Logging == nil:
 		return errors.New("logging settings not specified")
+	case cfg.XDSServer.BindAddress == "":
+		return errors.New("envoy xDS bind address not specified")
+	case cfg.XDSServer.BindPort == 0 && !checkLocalXDSServer(cfg.XDSServer.BindAddress):
+		return errors.New("envoy xDS bind port not specified")
 	}
 	return nil
 }
@@ -124,6 +133,28 @@ func (cdp *ConsulDataplane) setConsulServerSupportedFeatures(ctx context.Context
 	return nil
 }
 
+// checkLocalXDSServer checks if the specified xds bind address is local.
+func checkLocalXDSServer(xdsBindAddr string) bool {
+	if strings.HasPrefix(xdsBindAddr, "unix://") || xdsBindAddr == "127.0.0.1" || xdsBindAddr == "localhost" {
+		return true
+	}
+	return false
+}
+
+// checkAndEnableLocalXDSServer checks if the specified xds bind address is local.
+// If local, it enables the flag to configure consul-dataplane (later on in the setup process)
+// with a gRPC server to serve envoy xDS requests.
+// If not local, the flag remains turned off and it is assumed the xDS requests will be served
+// by a remote gRPC server.
+// Potential TODO: Explicitly allow specifying an option (xds.disable?) to disable configuring
+// consul-dataplane as the xDS server in case xDS requests can be served on another port locally
+// (example: consul server process on localhost).
+func (cdp *ConsulDataplane) checkAndEnableLocalXDSServer() {
+	if checkLocalXDSServer(cdp.cfg.XDSServer.BindAddress) {
+		cdp.localXDSServer.enabled = true
+	}
+}
+
 func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 	cdp.logger.Info("started consul-dataplane process")
 
@@ -156,11 +187,16 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 		return fmt.Errorf("failed to set supported features: %w", err)
 	}
 
-	err = cdp.setupGRPCServer()
-	if err != nil {
-		return err
+	cdp.checkAndEnableLocalXDSServer()
+
+	if cdp.localXDSServer.enabled {
+		err = cdp.setupXDSServer()
+		if err != nil {
+			return err
+		}
+		go cdp.startXDSServer()
+		defer cdp.stopXDSServer()
 	}
-	go cdp.startGRPCServer()
 
 	cfg, err := cdp.bootstrapConfig(ctx)
 	if err != nil {
@@ -190,16 +226,14 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 			if err := proxy.Stop(); err != nil {
 				cdp.logger.Error("failed to stop proxy", "error", err)
 			}
-			cdp.stopGRPCServer()
 			doneCh <- nil
 		case <-proxy.Exited():
-			cdp.stopGRPCServer()
 			doneCh <- errors.New("envoy proxy exited unexpectedly")
-		case <-cdp.gRPCServerExited():
+		case <-cdp.xdsServerExited():
 			if err := proxy.Stop(); err != nil {
 				cdp.logger.Error("failed to stop proxy", "error", err)
 			}
-			doneCh <- errors.New("gRPC server exited unexpectedly")
+			doneCh <- errors.New("xDS server exited unexpectedly")
 		}
 	}()
 	return <-doneCh