backport of commit d923fc554d09ceb51b530467a354860b25114fd3 (#18450)

Co-authored-by: James Rasell <jrasell@users.noreply.github.com>

.changelog/16745.txt

@@ -0,0 +1,3 @@
+```release-note:improvement
+consul/connect: Added support for `DestinationPeer`, `DestinationType`, `LocalBindSocketPath`, and `LocalBindSocketMode` in upstream block
+```

api/consul.go

@@ -210,9 +210,13 @@ func (c *ConsulMeshGateway) Copy() *ConsulMeshGateway {
 type ConsulUpstream struct {
 	DestinationName      string             `mapstructure:"destination_name" hcl:"destination_name,optional"`
 	DestinationNamespace string             `mapstructure:"destination_namespace" hcl:"destination_namespace,optional"`
+	DestinationPeer      string             `mapstructure:"destination_peer" hcl:"destination_peer,optional"`
+	DestinationType      string             `mapstructure:"destination_type" hcl:"destination_type,optional"`
 	LocalBindPort        int                `mapstructure:"local_bind_port" hcl:"local_bind_port,optional"`
 	Datacenter           string             `mapstructure:"datacenter" hcl:"datacenter,optional"`
 	LocalBindAddress     string             `mapstructure:"local_bind_address" hcl:"local_bind_address,optional"`
+	LocalBindSocketPath  string             `mapstructure:"local_bind_socket_path" hcl:"local_bind_socket_path,optional"`
+	LocalBindSocketMode  string             `mapstructure:"local_bind_socket_mode" hcl:"local_bind_socket_mode,optional"`
 	MeshGateway          *ConsulMeshGateway `mapstructure:"mesh_gateway" hcl:"mesh_gateway,block"`
 	Config               map[string]any     `mapstructure:"config" hcl:"config,block"`
 }
@@ -224,9 +228,13 @@ func (cu *ConsulUpstream) Copy() *ConsulUpstream {
 	return &ConsulUpstream{
 		DestinationName:      cu.DestinationName,
 		DestinationNamespace: cu.DestinationNamespace,
+		DestinationPeer:      cu.DestinationPeer,
+		DestinationType:      cu.DestinationType,
 		LocalBindPort:        cu.LocalBindPort,
 		Datacenter:           cu.Datacenter,
 		LocalBindAddress:     cu.LocalBindAddress,
+		LocalBindSocketPath:  cu.LocalBindSocketPath,
+		LocalBindSocketMode:  cu.LocalBindSocketMode,
 		MeshGateway:          cu.MeshGateway.Copy(),
 		Config:               maps.Clone(cu.Config),
 	}

api/consul_test.go

@@ -177,9 +177,13 @@ func TestConsulUpstream_Copy(t *testing.T) {
 		cu := &ConsulUpstream{
 			DestinationName:      "dest1",
 			DestinationNamespace: "ns2",
+			DestinationPeer:      "10.0.0.1:6379",
+			DestinationType:      "tcp",
 			Datacenter:           "dc2",
 			LocalBindPort:        2000,
 			LocalBindAddress:     "10.0.0.1",
+			LocalBindSocketPath:  "/var/run/testsocket.sock",
+			LocalBindSocketMode:  "0666",
 			MeshGateway:          &ConsulMeshGateway{Mode: "remote"},
 			Config:               map[string]any{"connect_timeout_ms": 5000},
 		}
@@ -201,9 +205,13 @@ func TestConsulUpstream_Canonicalize(t *testing.T) {
 		cu := &ConsulUpstream{
 			DestinationName:      "dest1",
 			DestinationNamespace: "ns2",
+			DestinationPeer:      "10.0.0.1:6379",
+			DestinationType:      "tcp",
 			Datacenter:           "dc2",
 			LocalBindPort:        2000,
 			LocalBindAddress:     "10.0.0.1",
+			LocalBindSocketPath:  "/var/run/testsocket.sock",
+			LocalBindSocketMode:  "0666",
 			MeshGateway:          &ConsulMeshGateway{Mode: ""},
 			Config:               make(map[string]any),
 		}
@@ -211,9 +219,13 @@ func TestConsulUpstream_Canonicalize(t *testing.T) {
 		must.Eq(t, &ConsulUpstream{
 			DestinationName:      "dest1",
 			DestinationNamespace: "ns2",
+			DestinationPeer:      "10.0.0.1:6379",
+			DestinationType:      "tcp",
 			Datacenter:           "dc2",
 			LocalBindPort:        2000,
 			LocalBindAddress:     "10.0.0.1",
+			LocalBindSocketPath:  "/var/run/testsocket.sock",
+			LocalBindSocketMode:  "0666",
 			MeshGateway:          &ConsulMeshGateway{Mode: ""},
 			Config:               nil,
 		}, cu)

command/agent/consul/connect.go

@@ -203,7 +203,11 @@ func connectUpstreams(in []structs.ConsulUpstream) []api.Upstream {
 		upstreams[i] = api.Upstream{
 			DestinationName:      upstream.DestinationName,
 			DestinationNamespace: upstream.DestinationNamespace,
+			DestinationType:      api.UpstreamDestType(upstream.DestinationType),
+			DestinationPeer:      upstream.DestinationPeer,
 			LocalBindPort:        upstream.LocalBindPort,
+			LocalBindSocketPath:  upstream.LocalBindSocketPath,
+			LocalBindSocketMode:  upstream.LocalBindSocketMode,
 			Datacenter:           upstream.Datacenter,
 			LocalBindAddress:     upstream.LocalBindAddress,
 			MeshGateway:          connectMeshGateway(upstream.MeshGateway),

command/agent/consul/connect_test.go

@@ -374,8 +374,12 @@ func TestConnect_connectUpstreams(t *testing.T) {
 				LocalBindPort:   8000,
 			}, {
 				DestinationName:      "bar",
+				DestinationPeer:      "10.0.0.1:6379",
+				DestinationType:      "tcp",
 				DestinationNamespace: "ns2",
 				LocalBindPort:        9000,
+				LocalBindSocketPath:  "/var/run/testsocket.sock",
+				LocalBindSocketMode:  "0666",
 				Datacenter:           "dc2",
 				LocalBindAddress:     "127.0.0.2",
 				Config:               map[string]any{"connect_timeout_ms": 5000},
@@ -386,7 +390,11 @@ func TestConnect_connectUpstreams(t *testing.T) {
 			}, {
 				DestinationName:      "bar",
 				DestinationNamespace: "ns2",
+				DestinationPeer:      "10.0.0.1:6379",
+				DestinationType:      "tcp",
 				LocalBindPort:        9000,
+				LocalBindSocketPath:  "/var/run/testsocket.sock",
+				LocalBindSocketMode:  "0666",
 				Datacenter:           "dc2",
 				LocalBindAddress:     "127.0.0.2",
 				Config:               map[string]any{"connect_timeout_ms": 5000},

command/agent/consul/service_client.go

@@ -343,6 +343,14 @@ func proxyUpstreamsDifferent(wanted *api.AgentServiceConnect, sidecar *api.Agent
 				return true
 			case A.MeshGateway.Mode != B.MeshGateway.Mode:
 				return true
+			case A.DestinationPeer != B.DestinationPeer:
+				return true
+			case A.DestinationType != B.DestinationType:
+				return true
+			case A.LocalBindSocketPath != B.LocalBindSocketPath:
+				return true
+			case A.LocalBindSocketMode != B.LocalBindSocketMode:
+				return true
 			case !reflect.DeepEqual(A.Config, B.Config):
 				return true
 			}

command/agent/consul/service_client_test.go

@@ -704,6 +704,42 @@ func TestSyncLogic_proxyUpstreamsDifferent(t *testing.T) {
 			upstream2(),
 		}
 	})
+
+	try(t, "different destination peer", func(p proxy) {
+		diff := upstream1()
+		diff.DestinationPeer = "foo"
+		p.Upstreams = []api.Upstream{
+			diff,
+			upstream2(),
+		}
+	})
+
+	try(t, "different destination type", func(p proxy) {
+		diff := upstream1()
+		diff.DestinationType = "service"
+		p.Upstreams = []api.Upstream{
+			diff,
+			upstream2(),
+		}
+	})
+
+	try(t, "different local bind socket path", func(p proxy) {
+		diff := upstream1()
+		diff.LocalBindSocketPath = "/var/run.sock"
+		p.Upstreams = []api.Upstream{
+			diff,
+			upstream2(),
+		}
+	})
+
+	try(t, "different local bind socket mode", func(p proxy) {
+		diff := upstream1()
+		diff.LocalBindSocketMode = "foo"
+		p.Upstreams = []api.Upstream{
+			diff,
+			upstream2(),
+		}
+	})
 }
 
 func TestSyncReason_String(t *testing.T) {

command/agent/job_endpoint.go

@@ -1733,7 +1733,11 @@ func apiUpstreamsToStructs(in []*api.ConsulUpstream) []structs.ConsulUpstream {
 		upstreams[i] = structs.ConsulUpstream{
 			DestinationName:      upstream.DestinationName,
 			DestinationNamespace: upstream.DestinationNamespace,
+			DestinationPeer:      upstream.DestinationPeer,
+			DestinationType:      upstream.DestinationType,
 			LocalBindPort:        upstream.LocalBindPort,
+			LocalBindSocketPath:  upstream.LocalBindSocketPath,
+			LocalBindSocketMode:  upstream.LocalBindSocketMode,
 			Datacenter:           upstream.Datacenter,
 			LocalBindAddress:     upstream.LocalBindAddress,
 			MeshGateway:          apiMeshGatewayToStructs(upstream.MeshGateway),

command/agent/job_endpoint_test.go

@@ -3842,14 +3842,22 @@ func TestConversion_apiUpstreamsToStructs(t *testing.T) {
 	require.Equal(t, []structs.ConsulUpstream{{
 		DestinationName:      "upstream",
 		DestinationNamespace: "ns2",
+		DestinationPeer:      "10.0.0.1:6379",
+		DestinationType:      "tcp",
 		LocalBindPort:        8000,
+		LocalBindSocketPath:  "/var/run/testsocket.sock",
+		LocalBindSocketMode:  "0666",
 		Datacenter:           "dc2",
 		LocalBindAddress:     "127.0.0.2",
 		MeshGateway:          structs.ConsulMeshGateway{Mode: "local"},
 	}}, apiUpstreamsToStructs([]*api.ConsulUpstream{{
 		DestinationName:      "upstream",
 		DestinationNamespace: "ns2",
+		DestinationPeer:      "10.0.0.1:6379",
+		DestinationType:      "tcp",
 		LocalBindPort:        8000,
+		LocalBindSocketPath:  "/var/run/testsocket.sock",
+		LocalBindSocketMode:  "0666",
 		Datacenter:           "dc2",
 		LocalBindAddress:     "127.0.0.2",
 		MeshGateway:          &api.ConsulMeshGateway{Mode: "local"},

jobspec/parse_service.go

@@ -922,8 +922,12 @@ func parseExposePath(epo *ast.ObjectItem) (*api.ConsulExposePath, error) {
 func parseUpstream(uo *ast.ObjectItem) (*api.ConsulUpstream, error) {
 	valid := []string{
 		"destination_name",
+		"destination_peer",
+		"destination_type",
 		"local_bind_port",
 		"local_bind_address",
+		"local_bind_socket_path",
+		"local_bind_socket_mode",
 		"datacenter",
 		"mesh_gateway",
 	}

jobspec/parse_test.go

@@ -1201,10 +1201,14 @@ func TestParse(t *testing.T) {
 											LocalServicePort: 8080,
 											Upstreams: []*api.ConsulUpstream{
 												{
-													DestinationName:  "other-service",
-													LocalBindPort:    4567,
-													LocalBindAddress: "0.0.0.0",
-													Datacenter:       "dc1",
+													DestinationName:     "other-service",
+													DestinationPeer:     "10.0.0.1:6379",
+													DestinationType:     "tcp",
+													LocalBindPort:       4567,
+													LocalBindAddress:    "0.0.0.0",
+													LocalBindSocketPath: "/var/run/testsocket.sock",
+													LocalBindSocketMode: "0666",
+													Datacenter:          "dc1",
 
 													MeshGateway: &api.ConsulMeshGateway{
 														Mode: "local",

jobspec/test-fixtures/tg-network.hcl

@@ -37,10 +37,14 @@ job "foo" {
             local_service_port = 8080
 
             upstreams {
-              destination_name   = "other-service"
-              local_bind_port    = 4567
-              local_bind_address = "0.0.0.0"
-              datacenter         = "dc1"
+              destination_name       = "other-service"
+              destination_peer       = "10.0.0.1:6379"
+              destination_type       = "tcp"
+              local_bind_port        = 4567
+              local_bind_address     = "0.0.0.0"
+              local_bind_socket_path = "/var/run/testsocket.sock"
+              local_bind_socket_mode = "0666"
+              datacenter             = "dc1"
 
               mesh_gateway {
                 mode = "local"

nomad/structs/diff_test.go

@@ -3548,6 +3548,18 @@ func TestTaskGroupDiff(t *testing.T) {
 																Old:  "",
 																New:  "ns2",
 															},
+															{
+																Type: DiffTypeNone,
+																Name: "DestinationPeer",
+																Old:  "",
+																New:  "",
+															},
+															{
+																Type: DiffTypeNone,
+																Name: "DestinationType",
+																Old:  "",
+																New:  "",
+															},
 															{
 																Type: DiffTypeAdded,
 																Name: "LocalBindAddress",
@@ -3560,6 +3572,18 @@ func TestTaskGroupDiff(t *testing.T) {
 																Old:  "",
 																New:  "8000",
 															},
+															{
+																Type: DiffTypeNone,
+																Name: "LocalBindSocketMode",
+																Old:  "",
+																New:  "",
+															},
+															{
+																Type: DiffTypeNone,
+																Name: "LocalBindSocketPath",
+																Old:  "",
+																New:  "",
+															},
 														},
 														Objects: []*ObjectDiff{
 															{
@@ -8948,7 +8972,132 @@ func TestServicesDiff(t *testing.T) {
 					},
 				},
 			},
-		}, {
+		},
+		{
+			Name:       "ConsulUpstream with different Upstreams",
+			Contextual: false,
+			Old: []*Service{
+				{
+					Name:      "webapp",
+					Provider:  "consul",
+					PortLabel: "http",
+					Connect: &ConsulConnect{
+						SidecarService: &ConsulSidecarService{
+							Port: "http",
+							Proxy: &ConsulProxy{
+								LocalServiceAddress: "127.0.0.1",
+								LocalServicePort:    8080,
+								Upstreams: []ConsulUpstream{
+									{
+										DestinationName:  "count-api",
+										LocalBindPort:    8080,
+										Datacenter:       "dc2",
+										LocalBindAddress: "127.0.0.1",
+										DestinationType:  "prepared_query",
+										MeshGateway: ConsulMeshGateway{
+											Mode: "remote",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			New: []*Service{
+				{
+					Name:      "webapp",
+					Provider:  "consul",
+					PortLabel: "http",
+					Connect: &ConsulConnect{
+						SidecarService: &ConsulSidecarService{
+							Port: "http",
+							Proxy: &ConsulProxy{
+								LocalServiceAddress: "127.0.0.1",
+								LocalServicePort:    8080,
+								Upstreams: []ConsulUpstream{
+									{
+										DestinationName:     "count-api",
+										LocalBindPort:       8080,
+										Datacenter:          "dc2",
+										LocalBindAddress:    "127.0.0.1",
+										LocalBindSocketMode: "0700",
+										LocalBindSocketPath: "/tmp/redis_5678.sock",
+										DestinationPeer:     "cloud-services",
+										DestinationType:     "service",
+										MeshGateway: ConsulMeshGateway{
+											Mode: "remote",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			Expected: []*ObjectDiff{
+				{
+					Type: DiffTypeEdited,
+					Name: "Service",
+					Objects: []*ObjectDiff{
+						{
+							Type: DiffTypeEdited,
+							Name: "ConsulConnect",
+							Objects: []*ObjectDiff{
+								{
+									Type: DiffTypeEdited,
+									Name: "SidecarService",
+									Objects: []*ObjectDiff{
+										{
+											Type: DiffTypeEdited,
+											Name: "ConsulProxy",
+											Objects: []*ObjectDiff{
+												{
+													Type: DiffTypeEdited,
+													Name: "ConsulUpstreams",
+													Fields: []*FieldDiff{
+														{
+															Type:        DiffTypeAdded,
+															Name:        "DestinationPeer",
+															Old:         "",
+															New:         "cloud-services",
+															Annotations: nil,
+														},
+														{
+															Type:        DiffTypeEdited,
+															Name:        "DestinationType",
+															Old:         "prepared_query",
+															New:         "service",
+															Annotations: nil,
+														},
+														{
+															Type:        DiffTypeAdded,
+															Name:        "LocalBindSocketMode",
+															Old:         "",
+															New:         "0700",
+															Annotations: nil,
+														},
+														{
+															Type:        DiffTypeAdded,
+															Name:        "LocalBindSocketPath",
+															Old:         "",
+															New:         "/tmp/redis_5678.sock",
+															Annotations: nil,
+														},
+													},
+													Objects: nil,
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
 			Name:       "SidecarService with different meta",
 			Contextual: false,
 			Old: []*Service{

nomad/structs/services.go

@@ -878,6 +878,10 @@ func hashConnect(h hash.Hash, connect *ConsulConnect) {
 				hashString(h, strconv.Itoa(upstream.LocalBindPort))
 				hashStringIfNonEmpty(h, upstream.Datacenter)
 				hashStringIfNonEmpty(h, upstream.LocalBindAddress)
+				hashString(h, upstream.DestinationPeer)
+				hashString(h, upstream.DestinationType)
+				hashString(h, upstream.LocalBindSocketPath)
+				hashString(h, upstream.LocalBindSocketMode)
 				hashConfig(h, upstream.Config)
 			}
 		}
@@ -1490,6 +1494,13 @@ type ConsulUpstream struct {
 	// DestinationNamespace is the namespace of the upstream service.
 	DestinationNamespace string
 
+	// DestinationPeer the destination service address
+	DestinationPeer string
+
+	// DestinationType is the type of destination. It can be an IP address,
+	// a DNS hostname, or a service name.
+	DestinationType string
+
 	// LocalBindPort is the port the proxy will receive connections for the
 	// upstream on.
 	LocalBindPort int
@@ -1501,6 +1512,13 @@ type ConsulUpstream struct {
 	// upstream on.
 	LocalBindAddress string
 
+	// LocalBindSocketPath is the path of the local socket file that will be used
+	// to connect to the destination service
+	LocalBindSocketPath string
+
+	// LocalBindSocketMode defines access permissions to the local socket file
+	LocalBindSocketMode string
+
 	// MeshGateway is the optional configuration of the mesh gateway for this
 	// upstream to use.
 	MeshGateway ConsulMeshGateway
@@ -1520,8 +1538,16 @@ func (u *ConsulUpstream) Equal(o *ConsulUpstream) bool {
 		return false
 	case u.DestinationNamespace != o.DestinationNamespace:
 		return false
+	case u.DestinationPeer != o.DestinationPeer:
+		return false
+	case u.DestinationType != o.DestinationType:
+		return false
 	case u.LocalBindPort != o.LocalBindPort:
 		return false
+	case u.LocalBindSocketPath != o.LocalBindSocketPath:
+		return false
+	case u.LocalBindSocketMode != o.LocalBindSocketMode:
+		return false
 	case u.Datacenter != o.Datacenter:
 		return false
 	case u.LocalBindAddress != o.LocalBindAddress:

nomad/structs/services_test.go

@@ -747,6 +747,46 @@ func TestConsulUpstream_upstreamEqual(t *testing.T) {
 		must.False(t, upstreamsEquals(a, b))
 	})
 
+	t.Run("different dest peer", func(t *testing.T) {
+		a := []ConsulUpstream{up("foo", 8000)}
+		a[0].DestinationPeer = "10.0.0.1:6379"
+
+		b := []ConsulUpstream{up("foo", 8000)}
+		b[0].DestinationPeer = "10.0.0.1:6375"
+
+		must.False(t, upstreamsEquals(a, b))
+	})
+
+	t.Run("different dest type", func(t *testing.T) {
+		a := []ConsulUpstream{up("foo", 8000)}
+		a[0].DestinationType = "tcp"
+
+		b := []ConsulUpstream{up("foo", 8000)}
+		b[0].DestinationType = "udp"
+
+		must.False(t, upstreamsEquals(a, b))
+	})
+
+	t.Run("different socket path", func(t *testing.T) {
+		a := []ConsulUpstream{up("foo", 8000)}
+		a[0].LocalBindSocketPath = "/var/run/mysocket.sock"
+
+		b := []ConsulUpstream{up("foo", 8000)}
+		b[0].LocalBindSocketPath = "/var/run/testsocket.sock"
+
+		must.False(t, upstreamsEquals(a, b))
+	})
+
+	t.Run("different socket mode", func(t *testing.T) {
+		a := []ConsulUpstream{up("foo", 8000)}
+		a[0].LocalBindSocketMode = "0666"
+
+		b := []ConsulUpstream{up("foo", 8000)}
+		b[0].LocalBindSocketMode = "0600"
+
+		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"}}}
@@ -762,6 +802,14 @@ func TestConsulUpstream_upstreamEqual(t *testing.T) {
 	t.Run("identical", func(t *testing.T) {
 		a := []ConsulUpstream{up("foo", 8000), up("bar", 9000)}
 		b := []ConsulUpstream{up("foo", 8000), up("bar", 9000)}
+		a[0].DestinationPeer = "10.0.0.1:6379"
+		a[0].DestinationType = "tcp"
+		a[0].LocalBindSocketPath = "/var/run/mysocket.sock"
+		a[0].LocalBindSocketMode = "0666"
+		b[0].DestinationPeer = "10.0.0.1:6379"
+		b[0].DestinationType = "tcp"
+		b[0].LocalBindSocketPath = "/var/run/mysocket.sock"
+		b[0].LocalBindSocketMode = "0666"
 		must.True(t, upstreamsEquals(a, b))
 	})
 

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

@@ -81,8 +81,13 @@ job "countdash" {
 
 ## `upstreams` Parameters
 
+- `config` `(map: nil)` - Upstream configuration that is opaque to Nomad and passed
+  directly to Consul. See [Consul Connect documentation](/consul/docs/connect/registration/service-registration#upstream-configuration-reference)
+  for details. Keys and values support [runtime variable interpolation][interpolation].
 - `destination_name` `(string: <required>)` - Name of the upstream service.
 - `destination_namespace` `(string: <required>)` - Name of the upstream Consul namespace.
+- `destination_peer` `(string: "")` - Name of the peer cluster containing the upstream service.
+- `destination_type` - `(string: "service")` - The type of discovery query the proxy should use for finding service mesh instances.
 - `local_bind_port` - `(int: <required>)` - The port the proxy will receive
   connections for the upstream on.
 - `datacenter` `(string: "")` - The Consul datacenter in which to issue the
@@ -90,6 +95,8 @@ job "countdash" {
   local Consul datacenter.
 - `local_bind_address` - `(string: "")` - The address the proxy will receive
   connections for the upstream on.
+- `local_bind_socket_mode` - `(string: "")` - Unix octal that configures file permissions for the socket.
+- `local_bind_socket_path` - `(string: "")` - The path at which to bind a Unix domain socket listener.
 - `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