consul/connect: support for proxy upstreams opaque config

This PR adds support for configuring `proxy.upstreams[].config` for
Consul Connect upstreams. This is an opaque config value to Nomad -
the data is passed directly to Consul and is unknown to Nomad.

.changelog/15761.txt

@@ -0,0 +1,3 @@
+```release-note:improvement
+consul/connect: Adds support for proxy upstream opaque config
+```

api/consul.go

@@ -2,6 +2,8 @@ package api
 
 import (
 	"time"
+
+	"golang.org/x/exp/maps"
 )
 
 // Consul represents configuration related to consul.
@@ -203,6 +205,7 @@ type ConsulUpstream struct {
 	Datacenter           string             `mapstructure:"datacenter" hcl:"datacenter,optional"`
 	LocalBindAddress     string             `mapstructure:"local_bind_address" hcl:"local_bind_address,optional"`
 	MeshGateway          *ConsulMeshGateway `mapstructure:"mesh_gateway" hcl:"mesh_gateway,block"`
+	Config               map[string]any     `mapstructure:"config" hcl:"config,optional"`
 }
 
 func (cu *ConsulUpstream) Copy() *ConsulUpstream {
@@ -216,6 +219,7 @@ func (cu *ConsulUpstream) Copy() *ConsulUpstream {
 		Datacenter:           cu.Datacenter,
 		LocalBindAddress:     cu.LocalBindAddress,
 		MeshGateway:          cu.MeshGateway.Copy(),
+		Config:               maps.Clone(cu.Config),
 	}
 }
 
@@ -224,6 +228,9 @@ func (cu *ConsulUpstream) Canonicalize() {
 		return
 	}
 	cu.MeshGateway.Canonicalize()
+	if len(cu.Config) == 0 {
+		cu.Config = nil
+	}
 }
 
 type ConsulExposeConfig struct {

api/consul_test.go

@@ -172,6 +172,7 @@ func TestConsulUpstream_Copy(t *testing.T) {
 			LocalBindPort:        2000,
 			LocalBindAddress:     "10.0.0.1",
 			MeshGateway:          &ConsulMeshGateway{Mode: "remote"},
+			Config:               map[string]any{"connect_timeout_ms": 5000},
 		}
 		result := cu.Copy()
 		must.Eq(t, cu, result)
@@ -195,6 +196,7 @@ func TestConsulUpstream_Canonicalize(t *testing.T) {
 			LocalBindPort:        2000,
 			LocalBindAddress:     "10.0.0.1",
 			MeshGateway:          &ConsulMeshGateway{Mode: ""},
+			Config:               make(map[string]any),
 		}
 		cu.Canonicalize()
 		must.Eq(t, &ConsulUpstream{
@@ -204,6 +206,7 @@ func TestConsulUpstream_Canonicalize(t *testing.T) {
 			LocalBindPort:        2000,
 			LocalBindAddress:     "10.0.0.1",
 			MeshGateway:          &ConsulMeshGateway{Mode: ""},
+			Config:               nil,
 		}, cu)
 	})
 }

api/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/mitchellh/go-testing-interface v1.14.1
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/shoenig/test v0.6.0
+	golang.org/x/exp v0.0.0-20230108222341-4b8118a2686a
 )
 
 require (

api/go.sum

@@ -30,6 +30,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+golang.org/x/exp v0.0.0-20230108222341-4b8118a2686a h1:tlXy25amD5A7gOfbXdqCGN5k8ESEed/Ee1E5RcrYnqU=
+golang.org/x/exp v0.0.0-20230108222341-4b8118a2686a/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

client/taskenv/services.go

@@ -74,12 +74,12 @@ func interpolateMapStringString(taskEnv *TaskEnv, orig map[string]string) map[st
 	return m
 }
 
-func interpolateMapStringInterface(taskEnv *TaskEnv, orig map[string]interface{}) map[string]interface{} {
+func interpolateMapStringInterface(taskEnv *TaskEnv, orig map[string]any) map[string]any {
 	if len(orig) == 0 {
 		return nil
 	}
 
-	m := make(map[string]interface{}, len(orig))
+	m := make(map[string]any, len(orig))
 	for k, v := range orig {
 		envK := taskEnv.ReplaceEnv(k)
 		if vStr, ok := v.(string); ok {
@@ -155,6 +155,7 @@ func interpolateConnectSidecarService(taskEnv *TaskEnv, sidecar *structs.ConsulS
 			sidecar.Proxy.Upstreams[i].Datacenter = taskEnv.ReplaceEnv(sidecar.Proxy.Upstreams[i].Datacenter)
 			sidecar.Proxy.Upstreams[i].DestinationName = taskEnv.ReplaceEnv(sidecar.Proxy.Upstreams[i].DestinationName)
 			sidecar.Proxy.Upstreams[i].LocalBindAddress = taskEnv.ReplaceEnv(sidecar.Proxy.Upstreams[i].LocalBindAddress)
+			sidecar.Proxy.Upstreams[i].Config = interpolateMapStringInterface(taskEnv, sidecar.Proxy.Upstreams[i].Config)
 		}
 		sidecar.Proxy.Config = interpolateMapStringInterface(taskEnv, sidecar.Proxy.Config)
 	}

client/taskenv/services_test.go

@@ -228,6 +228,7 @@ func TestInterpolate_interpolateConnect(t *testing.T) {
 					Datacenter:       "${datacenter1}",
 					LocalBindPort:    10001,
 					LocalBindAddress: "${localbindaddress1}",
+					Config:           map[string]any{"${config1}": 1},
 				}},
 				Expose: &structs.ConsulExposeConfig{
 					Paths: []structs.ConsulExposePath{{
@@ -337,6 +338,7 @@ func TestInterpolate_interpolateConnect(t *testing.T) {
 					Datacenter:       "_datacenter1",
 					LocalBindPort:    10001,
 					LocalBindAddress: "127.0.0.2",
+					Config:           map[string]any{"_config1": 1},
 				}},
 				Expose: &structs.ConsulExposeConfig{
 					Paths: []structs.ConsulExposePath{{

command/agent/consul/connect.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/nomad/nomad/structs"
+	"golang.org/x/exp/maps"
 	"golang.org/x/exp/slices"
 )
 
@@ -202,6 +203,7 @@ func connectUpstreams(in []structs.ConsulUpstream) []api.Upstream {
 			Datacenter:           upstream.Datacenter,
 			LocalBindAddress:     upstream.LocalBindAddress,
 			MeshGateway:          connectMeshGateway(upstream.MeshGateway),
+			Config:               maps.Clone(upstream.Config),
 		}
 	}
 	return upstreams

command/agent/consul/connect_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/nomad/helper/pointer"
 	"github.com/hashicorp/nomad/helper/uuid"
 	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/shoenig/test/must"
 	"github.com/stretchr/testify/require"
 )
 
@@ -360,11 +361,11 @@ func TestConnect_connectUpstreams(t *testing.T) {
 	ci.Parallel(t)
 
 	t.Run("nil", func(t *testing.T) {
-		require.Nil(t, connectUpstreams(nil))
+		must.Nil(t, connectUpstreams(nil))
 	})
 
 	t.Run("not empty", func(t *testing.T) {
-		require.Equal(t,
+		must.Eq(t,
 			[]api.Upstream{{
 				DestinationName: "foo",
 				LocalBindPort:   8000,
@@ -374,6 +375,7 @@ func TestConnect_connectUpstreams(t *testing.T) {
 				LocalBindPort:        9000,
 				Datacenter:           "dc2",
 				LocalBindAddress:     "127.0.0.2",
+				Config:               map[string]any{"connect_timeout_ms": 5000},
 			}},
 			connectUpstreams([]structs.ConsulUpstream{{
 				DestinationName: "foo",
@@ -384,6 +386,7 @@ func TestConnect_connectUpstreams(t *testing.T) {
 				LocalBindPort:        9000,
 				Datacenter:           "dc2",
 				LocalBindAddress:     "127.0.0.2",
+				Config:               map[string]any{"connect_timeout_ms": 5000},
 			}}),
 		)
 	})

command/agent/job_endpoint.go

@@ -1668,6 +1668,7 @@ func apiUpstreamsToStructs(in []*api.ConsulUpstream) []structs.ConsulUpstream {
 			Datacenter:           upstream.Datacenter,
 			LocalBindAddress:     upstream.LocalBindAddress,
 			MeshGateway:          apiMeshGatewayToStructs(upstream.MeshGateway),
+			Config:               maps.Clone(upstream.Config),
 		}
 	}
 	return upstreams

go.mod

@@ -120,7 +120,7 @@ require (
 	go.etcd.io/bbolt v1.3.6
 	go.uber.org/goleak v1.2.0
 	golang.org/x/crypto v0.1.0
-	golang.org/x/exp v0.0.0-20221215174704-0915cd710c24
+	golang.org/x/exp v0.0.0-20230108222341-4b8118a2686a
 	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
 	golang.org/x/sys v0.3.0
 	golang.org/x/time v0.0.0-20220224211638-0e9765cccd65

go.sum

@@ -1364,8 +1364,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0
 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
-golang.org/x/exp v0.0.0-20221215174704-0915cd710c24 h1:6w3iSY8IIkp5OQtbYj8NeuKG1jS9d+kYaubXqsoOiQ8=
-golang.org/x/exp v0.0.0-20221215174704-0915cd710c24/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
+golang.org/x/exp v0.0.0-20230108222341-4b8118a2686a h1:tlXy25amD5A7gOfbXdqCGN5k8ESEed/Ee1E5RcrYnqU=
+golang.org/x/exp v0.0.0-20230108222341-4b8118a2686a/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=

nomad/structs/services.go

@@ -854,6 +854,7 @@ func hashConnect(h hash.Hash, connect *ConsulConnect) {
 				hashString(h, strconv.Itoa(upstream.LocalBindPort))
 				hashStringIfNonEmpty(h, upstream.Datacenter)
 				hashStringIfNonEmpty(h, upstream.LocalBindAddress)
+				hashConfig(h, upstream.Config)
 			}
 		}
 	}
@@ -1481,18 +1482,46 @@ type ConsulUpstream struct {
 	// MeshGateway is the optional configuration of the mesh gateway for this
 	// upstream to use.
 	MeshGateway ConsulMeshGateway
+
+	// Config is an upstream configuration. It is opaque to Nomad and passed
+	// directly to Consul.
+	Config map[string]any
 }
 
 // Equal returns true if the structs are recursively equal.
 func (u *ConsulUpstream) Equal(o *ConsulUpstream) bool {
 	if u == nil || o == nil {
 		return u == o
 	}
-	return *u == *o
+	switch {
+	case u.DestinationName != o.DestinationName:
+		return false
+	case u.DestinationNamespace != o.DestinationNamespace:
+		return false
+	case u.LocalBindPort != o.LocalBindPort:
+		return false
+	case u.Datacenter != o.Datacenter:
+		return false
+	case u.LocalBindAddress != o.LocalBindAddress:
+		return false
+	case !u.MeshGateway.Equal(o.MeshGateway):
+		return false
+	case !opaqueMapsEqual(u.Config, o.Config):
+		return false
+	}
+	return true
+}
+
+// Hash implements a GoString based "hash" function for ConsulUpstream; because
+// this struct now contains an opaque map we cannot do much better than this.
+func (u ConsulUpstream) Hash() string {
+	return fmt.Sprintf("%#v", u)
 }
 
 func upstreamsEquals(a, b []ConsulUpstream) bool {
-	return set.From(a).Equal(set.From(b))
+	setA := set.HashSetFrom[ConsulUpstream, string](a)
+	setB := set.HashSetFrom[ConsulUpstream, string](b)
+	return setA.Equal(setB)
 }
 
 // ConsulExposeConfig represents a Consul Connect expose jobspec stanza.

nomad/structs/services_test.go

@@ -382,11 +382,12 @@ func TestService_Hash(t *testing.T) {
 				Proxy: &ConsulProxy{
 					LocalServiceAddress: "127.0.0.1",
 					LocalServicePort:    24000,
-					Config:              map[string]interface{}{"foo": "bar"},
+					Config:              map[string]any{"foo": "bar"},
 					Upstreams: []ConsulUpstream{{
 						DestinationName:      "upstream1",
 						DestinationNamespace: "ns2",
 						LocalBindPort:        29000,
+						Config:               map[string]any{"foo": "bar"},
 					}},
 				},
 			},
@@ -478,6 +479,10 @@ func TestService_Hash(t *testing.T) {
 	t.Run("mod connect sidecar proxy upstream destination local bind port", func(t *testing.T) {
 		try(t, func(s *svc) { s.Connect.SidecarService.Proxy.Upstreams[0].LocalBindPort = 29999 })
 	})
+
+	t.Run("mod connect sidecar proxy upstream config", func(t *testing.T) {
+		try(t, func(s *svc) { s.Connect.SidecarService.Proxy.Upstreams[0].Config = map[string]any{"foo": "baz"} })
+	})
 }
 
 func TestConsulConnect_Validate(t *testing.T) {
@@ -703,13 +708,13 @@ func TestConsulUpstream_upstreamEqual(t *testing.T) {
 	t.Run("size mismatch", func(t *testing.T) {
 		a := []ConsulUpstream{up("foo", 8000)}
 		b := []ConsulUpstream{up("foo", 8000), up("bar", 9000)}
-		require.False(t, upstreamsEquals(a, b))
+		must.False(t, upstreamsEquals(a, b))
 	})
 
 	t.Run("different", func(t *testing.T) {
 		a := []ConsulUpstream{up("bar", 9000)}
 		b := []ConsulUpstream{up("foo", 8000)}
-		require.False(t, upstreamsEquals(a, b))
+		must.False(t, upstreamsEquals(a, b))
 	})
 
 	t.Run("different namespace", func(t *testing.T) {
@@ -719,25 +724,31 @@ func TestConsulUpstream_upstreamEqual(t *testing.T) {
 		b := []ConsulUpstream{up("foo", 8000)}
 		b[0].DestinationNamespace = "ns2"
 
-		require.False(t, upstreamsEquals(a, b))
+		must.False(t, upstreamsEquals(a, b))
 	})
 
 	t.Run("different mesh_gateway", func(t *testing.T) {
 		a := []ConsulUpstream{{DestinationName: "foo", MeshGateway: ConsulMeshGateway{Mode: "local"}}}
 		b := []ConsulUpstream{{DestinationName: "foo", MeshGateway: ConsulMeshGateway{Mode: "remote"}}}
-		require.False(t, upstreamsEquals(a, b))
+		must.False(t, upstreamsEquals(a, b))
+	})
+
+	t.Run("different opaque config", func(t *testing.T) {
+		a := []ConsulUpstream{{Config: map[string]any{"foo": 1}}}
+		b := []ConsulUpstream{{Config: map[string]any{"foo": 2}}}
+		must.False(t, upstreamsEquals(a, b))
 	})
 
 	t.Run("identical", func(t *testing.T) {
 		a := []ConsulUpstream{up("foo", 8000), up("bar", 9000)}
 		b := []ConsulUpstream{up("foo", 8000), up("bar", 9000)}
-		require.True(t, upstreamsEquals(a, b))
+		must.True(t, upstreamsEquals(a, b))
 	})
 
 	t.Run("unsorted", func(t *testing.T) {
 		a := []ConsulUpstream{up("foo", 8000), up("bar", 9000)}
 		b := []ConsulUpstream{up("bar", 9000), up("foo", 8000)}
-		require.True(t, upstreamsEquals(a, b))
+		must.True(t, upstreamsEquals(a, b))
 	})
 }
 

website/content/docs/job-specification/proxy.mdx

@@ -60,9 +60,8 @@ job "countdash" {
 - `expose` <code>([expose]: nil)</code> - Used to configure expose path configuration for Envoy.
   See Consul's [Expose Paths Configuration Reference](https://developer.hashicorp.com/consul/docs/connect/registration/service-registration#expose-paths-configuration-reference)
   for more information.
-- `config` `(map: nil)` - Proxy configuration that's opaque to Nomad and
-  passed directly to Consul. See [Consul Connect's
-  documentation](https://developer.hashicorp.com/consul/docs/connect/proxies/envoy#dynamic-configuration)
+- `config` `(map: nil)` - Proxy configuration that is opaque to Nomad and
+  passed directly to Consul. See [Consul Connect documentation](https://developer.hashicorp.com/consul/docs/connect/proxies/envoy#dynamic-configuration)
   for details. Keys and values support [runtime variable interpolation][interpolation].
 
 ## `proxy` Examples

website/content/docs/job-specification/upstreams.mdx

@@ -92,6 +92,9 @@ job "countdash" {
   connections for the upstream on.
 - `mesh_gateway` <code>([mesh_gateway][mesh_gateway_param]: nil)</code> - Configures the mesh gateway
   behavior for connecting to this upstream.
+- `config` `(map: nil)` - Upstream configuration that is opaque to Nomad and passed
+  directly to Consul. See [Consul Connect documentation](https://developer.hashicorp.com/consul/docs/connect/registration/service-registration#upstream-configuration-reference)
+  for details. Keys and values support [runtime variable interpolation][interpolation].
 
 ### `mesh_gateway` Parameters