Envoy config cluster (#5308)

* Start adding tests for cluster override

* Refactor tests for clusters

* Passing tests for custom upstream cluster override

* Added capability to customise local app cluster

* Rename config for local cluster override

agent/xds/clusters.go

@@ -1,15 +1,20 @@
 package xds
 
 import (
+	"encoding/json"
 	"errors"
+	"fmt"
 	"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/jsonpb"
 	"github.com/gogo/protobuf/proto"
+	"github.com/gogo/protobuf/types"
 
 	"github.com/hashicorp/consul/agent/proxycfg"
+	"github.com/hashicorp/consul/agent/structs"
 )
 
 // clustersFromSnapshot returns the xDS API representation of the "clusters"
@@ -21,59 +26,146 @@ func clustersFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]pro
 	// Include the "app" cluster for the public listener
 	clusters := make([]proto.Message, len(cfgSnap.Proxy.Upstreams)+1)
 
-	clusters[0] = makeAppCluster(cfgSnap)
+	var err error
+	clusters[0], err = makeAppCluster(cfgSnap)
+	if err != nil {
+		return nil, err
+	}
 
 	for idx, upstream := range cfgSnap.Proxy.Upstreams {
-		clusters[idx+1] = makeUpstreamCluster(upstream.Identifier(), cfgSnap)
+		clusters[idx+1], err = makeUpstreamCluster(upstream, cfgSnap)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	return clusters, nil
 }
 
-func makeAppCluster(cfgSnap *proxycfg.ConfigSnapshot) *envoy.Cluster {
-	addr := cfgSnap.Proxy.LocalServiceAddress
-	if addr == "" {
-		addr = "127.0.0.1"
+func makeAppCluster(cfgSnap *proxycfg.ConfigSnapshot) (*envoy.Cluster, error) {
+	var c *envoy.Cluster
+	var err error
+
+	// If we have overriden local cluster config try to parse it into an Envoy cluster
+	if clusterJSONRaw, ok := cfgSnap.Proxy.Config["envoy_local_cluster_json"]; ok {
+		if clusterJSON, ok := clusterJSONRaw.(string); ok {
+			c, err = makeClusterFromUserConfig(clusterJSON)
+			if err != nil {
+				return c, err
+			}
+		}
 	}
-	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),
-		//      },
-		//    },
-		//  },
-		// },
+
+	if c == nil {
+		addr := cfgSnap.Proxy.LocalServiceAddress
+		if addr == "" {
+			addr = "127.0.0.1"
+		}
+		c = &envoy.Cluster{
+			Name:           LocalAppClusterName,
+			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),
+			//      },
+			//    },
+			//  },
+			// },
+		}
 	}
+
+	return c, err
 }
 
-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{},
+func makeUpstreamCluster(upstream structs.Upstream, cfgSnap *proxycfg.ConfigSnapshot) (*envoy.Cluster, error) {
+	var c *envoy.Cluster
+	var err error
+
+	// If we have overriden cluster config attempt to parse it into an Envoy cluster
+	if clusterJSONRaw, ok := upstream.Config["envoy_cluster_json"]; ok {
+		if clusterJSON, ok := clusterJSONRaw.(string); ok {
+			c, err = makeClusterFromUserConfig(clusterJSON)
+			if err != nil {
+				return c, err
+			}
+		}
+	}
+
+	if c == nil {
+		c = &envoy.Cluster{
+			Name:           upstream.Identifier(),
+			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),
-		},
+		}
+	}
+
+	// Enable TLS upstream with the configured client certificate.
+	c.TlsContext = &envoyauth.UpstreamTlsContext{
+		CommonTlsContext: makeCommonTLSContext(cfgSnap),
 	}
+
+	return c, nil
+}
+
+// makeClusterFromUserConfig returns the listener config decoded from an
+// arbitrary proto3 json format string or an error if it's invalid.
+//
+// For now we only support embedding in JSON strings because of the hcl parsing
+// pain (see config.go comment above call to patchSliceOfMaps). Until we
+// refactor config parser a _lot_ user's opaque config that contains arrays will
+// be mangled. We could actually fix that up in mapstructure which knows the
+// type of the target so could resolve the slices to singletons unambiguously
+// and it would work for us here... but we still have the problem that the
+// config would render incorrectly in general in our HTTP API responses so we
+// really need to fix it "properly".
+//
+// When we do that we can support just nesting the config directly into the
+// JSON/hcl naturally but this is a stop-gap that gets us an escape hatch
+// immediately. It's also probably not a bad thing to support long-term since
+// any config generated by other systems will likely be in canonical protobuf
+// from rather than our slight variant in JSON/hcl.
+func makeClusterFromUserConfig(configJSON string) (*envoy.Cluster, error) {
+	var jsonFields map[string]*json.RawMessage
+	if err := json.Unmarshal([]byte(configJSON), &jsonFields); err != nil {
+		fmt.Println("Custom error", err, configJSON)
+		return nil, err
+	}
+
+	var c envoy.Cluster
+
+	if _, ok := jsonFields["@type"]; ok {
+		// Type field is present so decode it as a types.Any
+		var any types.Any
+		err := jsonpb.UnmarshalString(configJSON, &any)
+		if err != nil {
+			return nil, err
+		}
+		// And then unmarshal the listener again...
+		err = proto.Unmarshal(any.Value, &c)
+		if err != nil {
+			panic(err)
+			//return nil, err
+		}
+		return &c, err
+	}
+
+	// No @type so try decoding as a straight listener.
+	err := jsonpb.UnmarshalString(configJSON, &c)
+	return &c, err
 }