xDS Server Implementation (#4731)

* Vendor updates for gRPC and xDS server

* xDS server implementation for serving Envoy as a Connect proxy

* Address initial review comments

* consistent envoy package aliases; typos fixed; override TLS and authz for custom listeners

* Moar Typos

* Moar typos

agent/connect/uri.go

@@ -33,6 +33,17 @@ var (
 		`^/ns/([^/]+)/dc/([^/]+)/svc/([^/]+)$`)
 )
 
+// ParseCertURIFromString attempts to parse a string representation of a
+// certificate URI as a convenince helper around ParseCertURI.
+func ParseCertURIFromString(input string) (CertURI, error) {
+	// Parse the certificate URI from the string
+	uriRaw, err := url.Parse(input)
+	if err != nil {
+		return nil, err
+	}
+	return ParseCertURI(uriRaw)
+}
+
 // ParseCertURI parses a the URI value from a TLS certificate.
 func ParseCertURI(input *url.URL) (CertURI, error) {
 	if input.Scheme != "spiffe" {

agent/connect/uri_test.go

@@ -1,7 +1,6 @@
 package connect
 
 import (
-	"net/url"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -57,17 +56,13 @@ var testCertURICases = []struct {
 	},
 }
 
-func TestParseCertURI(t *testing.T) {
+func TestParseCertURIFromString(t *testing.T) {
 	for _, tc := range testCertURICases {
 		t.Run(tc.Name, func(t *testing.T) {
 			assert := assert.New(t)
 
-			// Parse the URI, should always be valid
-			uri, err := url.Parse(tc.URI)
-			assert.Nil(err)
-
 			// Parse the ID and check the error/return value
-			actual, err := ParseCertURI(uri)
+			actual, err := ParseCertURIFromString(tc.URI)
 			if err != nil {
 				t.Logf("parse error: %s", err.Error())
 			}

agent/xds/clusters.go

@@ -0,0 +1,79 @@
+package xds
+
+import (
+	"errors"
+	"time"
+
+	envoy "github.com/envoyproxy/go-control-plane/envoy/api/v2"
+	envoyauth "github.com/envoyproxy/go-control-plane/envoy/api/v2/auth"
+	envoycore "github.com/envoyproxy/go-control-plane/envoy/api/v2/core"
+	"github.com/gogo/protobuf/proto"
+
+	"github.com/hashicorp/consul/agent/proxycfg"
+)
+
+// clustersFromSnapshot returns the xDS API representation of the "clusters"
+// (upstreams) in the snapshot.
+func clustersFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
+	if cfgSnap == nil {
+		return nil, errors.New("nil config given")
+	}
+	// Include the "app" cluster for the public listener
+	clusters := make([]proto.Message, len(cfgSnap.Proxy.Upstreams)+1)
+
+	clusters[0] = makeAppCluster(cfgSnap)
+
+	for idx, upstream := range cfgSnap.Proxy.Upstreams {
+		clusters[idx+1] = makeUpstreamCluster(upstream.Identifier(), cfgSnap)
+	}
+
+	return clusters, nil
+}
+
+func makeAppCluster(cfgSnap *proxycfg.ConfigSnapshot) *envoy.Cluster {
+	addr := cfgSnap.Proxy.LocalServiceAddress
+	if addr == "" {
+		addr = "127.0.0.1"
+	}
+	return &envoy.Cluster{
+		Name: LocalAppClusterName,
+		// TODO(banks): make this configurable from the proxy config
+		ConnectTimeout: 5 * time.Second,
+		Type:           envoy.Cluster_STATIC,
+		// API v2 docs say hosts is deprecated and should use LoadAssignment as
+		// below.. but it doesn't work for tcp_proxy target for some reason.
+		Hosts: []*envoycore.Address{makeAddressPtr(addr, cfgSnap.Proxy.LocalServicePort)},
+		// LoadAssignment: &envoy.ClusterLoadAssignment{
+		//  ClusterName: LocalAppClusterName,
+		//  Endpoints: []endpoint.LocalityLbEndpoints{
+		//    {
+		//      LbEndpoints: []endpoint.LbEndpoint{
+		//        makeEndpoint(LocalAppClusterName,
+		//          addr,
+		//          cfgSnap.Proxy.LocalServicePort),
+		//      },
+		//    },
+		//  },
+		// },
+	}
+}
+
+func makeUpstreamCluster(name string, cfgSnap *proxycfg.ConfigSnapshot) *envoy.Cluster {
+	return &envoy.Cluster{
+		Name: name,
+		// TODO(banks): make this configurable from the upstream config
+		ConnectTimeout: 5 * time.Second,
+		Type:           envoy.Cluster_EDS,
+		EdsClusterConfig: &envoy.Cluster_EdsClusterConfig{
+			EdsConfig: &envoycore.ConfigSource{
+				ConfigSourceSpecifier: &envoycore.ConfigSource_Ads{
+					Ads: &envoycore.AggregatedConfigSource{},
+				},
+			},
+		},
+		// Enable TLS upstream with the configured client certificate.
+		TlsContext: &envoyauth.UpstreamTlsContext{
+			CommonTlsContext: makeCommonTLSContext(cfgSnap),
+		},
+	}
+}

agent/xds/endpoints.go

@@ -0,0 +1,58 @@
+package xds
+
+import (
+	"errors"
+
+	envoy "github.com/envoyproxy/go-control-plane/envoy/api/v2"
+	envoyendpoint "github.com/envoyproxy/go-control-plane/envoy/api/v2/endpoint"
+	"github.com/gogo/protobuf/proto"
+
+	"github.com/hashicorp/consul/agent/proxycfg"
+	"github.com/hashicorp/consul/agent/structs"
+)
+
+// endpointsFromSnapshot returns the xDS API representation of the "endpoints"
+// (upstream instances) in the snapshot.
+func endpointsFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
+	if cfgSnap == nil {
+		return nil, errors.New("nil config given")
+	}
+	resources := make([]proto.Message, 0, len(cfgSnap.UpstreamEndpoints))
+	for id, endpoints := range cfgSnap.UpstreamEndpoints {
+		if len(endpoints) < 1 {
+			continue
+		}
+		la := makeLoadAssignment(id, endpoints)
+		resources = append(resources, la)
+	}
+	return resources, nil
+}
+
+func makeEndpoint(clusterName, host string, port int) envoyendpoint.LbEndpoint {
+	return envoyendpoint.LbEndpoint{
+		Endpoint: &envoyendpoint.Endpoint{
+			Address: makeAddressPtr(host, port),
+		},
+	}
+}
+
+func makeLoadAssignment(clusterName string, endpoints structs.CheckServiceNodes) *envoy.ClusterLoadAssignment {
+	es := make([]envoyendpoint.LbEndpoint, 0, len(endpoints))
+	for _, ep := range endpoints {
+		addr := ep.Service.Address
+		if addr == "" {
+			addr = ep.Node.Address
+		}
+		es = append(es, envoyendpoint.LbEndpoint{
+			Endpoint: &envoyendpoint.Endpoint{
+				Address: makeAddressPtr(addr, ep.Service.Port),
+			},
+		})
+	}
+	return &envoy.ClusterLoadAssignment{
+		ClusterName: clusterName,
+		Endpoints: []envoyendpoint.LocalityLbEndpoints{{
+			LbEndpoints: es,
+		}},
+	}
+}