[1.15.x] peering: ensure that merged central configs of peered upstreams for partitioned downstreams work (#17181)

Backport of #17179 into release/1.15.x

.changelog/17179.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+peering: ensure that merged central configs of peered upstreams for partitioned downstreams work
+```

agent/configentry/merge_service_config.go

@@ -154,6 +154,10 @@ func MergeServiceConfig(defaults *structs.ServiceConfigResponse, service *struct
 	// remoteUpstreams contains synthetic Upstreams generated from central config (service-defaults.UpstreamConfigs).
 	remoteUpstreams := make(map[structs.PeeredServiceName]structs.Upstream)
 
+	// If the arguments did not fully normalize tenancy stuff, take care of that now.
+	entMeta := ns.EnterpriseMeta
+	entMeta.Normalize()
+
 	if len(defaults.UpstreamIDConfigs) > 0 {
 		// Handle legacy upstreams. This should be removed in Consul 1.16.
 		for _, us := range defaults.UpstreamIDConfigs {
@@ -183,11 +187,21 @@ func MergeServiceConfig(defaults *structs.ServiceConfigResponse, service *struct
 				return nil, fmt.Errorf("failed to parse upstream config map for %s: %v", us.Upstream.String(), err)
 			}
 
+			// If the defaults did not fully normalize tenancy stuff, take care of
+			// that now too.
+			psn := us.Upstream // only normalize the copy
+			psn.ServiceName.EnterpriseMeta.Normalize()
+
+			// Normalize the partition field specially.
+			if psn.Peer != "" {
+				psn.ServiceName.OverridePartition(entMeta.PartitionOrDefault())
+			}
+
 			remoteUpstreams[us.Upstream] = structs.Upstream{
-				DestinationNamespace: us.Upstream.ServiceName.NamespaceOrDefault(),
+				DestinationNamespace: psn.ServiceName.NamespaceOrDefault(),
-				DestinationPartition: us.Upstream.ServiceName.PartitionOrDefault(),
+				DestinationPartition: psn.ServiceName.PartitionOrDefault(),
-				DestinationName:      us.Upstream.ServiceName.Name,
+				DestinationName:      psn.ServiceName.Name,
-				DestinationPeer:      us.Upstream.Peer,
+				DestinationPeer:      psn.Peer,
 				Config:               us.Config,
 				MeshGateway:          parsed.MeshGateway,
 				CentrallyConfigured:  true,
@@ -209,6 +223,12 @@ func MergeServiceConfig(defaults *structs.ServiceConfigResponse, service *struct
 		}
 
 		uid := us.DestinationID()
+
+		// Normalize the partition field specially.
+		if uid.Peer != "" {
+			uid.ServiceName.OverridePartition(entMeta.PartitionOrDefault())
+		}
+
 		localUpstreams[uid] = struct{}{}
 		remoteCfg, ok := remoteUpstreams[uid]
 		if !ok {

agent/configentry/merge_service_config_test.go

@@ -7,6 +7,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/structs"
 )
 
@@ -270,6 +271,214 @@ func Test_MergeServiceConfig_Extensions(t *testing.T) {
 	}
 }
 
+func isEnterprise() bool {
+	return acl.PartitionOrDefault("") == "default"
+}
+
+func Test_MergeServiceConfig_peeredCentralDefaultsMerging(t *testing.T) {
+	partitions := []string{"default"}
+	if isEnterprise() {
+		partitions = append(partitions, "part1")
+	}
+
+	const peerName = "my-peer"
+
+	newDefaults := func(partition string) *structs.ServiceConfigResponse {
+		// client agents
+		return &structs.ServiceConfigResponse{
+			ProxyConfig: map[string]any{
+				"protocol": "http",
+			},
+			UpstreamConfigs: []structs.OpaqueUpstreamConfig{
+				{
+					Upstream: structs.PeeredServiceName{
+						ServiceName: structs.ServiceName{
+							Name:           "*",
+							EnterpriseMeta: acl.NewEnterpriseMetaWithPartition(partition, "*"),
+						},
+					},
+					Config: map[string]any{
+						"mesh_gateway": map[string]any{
+							"Mode": "local",
+						},
+						"protocol": "http",
+					},
+				},
+				{
+					Upstream: structs.PeeredServiceName{
+						ServiceName: structs.ServiceName{
+							Name:           "static-server",
+							EnterpriseMeta: acl.NewEnterpriseMetaWithPartition(partition, "default"),
+						},
+						Peer: peerName,
+					},
+					Config: map[string]any{
+						"mesh_gateway": map[string]any{
+							"Mode": "local",
+						},
+						"protocol": "http",
+					},
+				},
+			},
+			MeshGateway: structs.MeshGatewayConfig{
+				Mode: "local",
+			},
+		}
+	}
+
+	for _, partition := range partitions {
+		t.Run("partition="+partition, func(t *testing.T) {
+			t.Run("clients", func(t *testing.T) {
+				defaults := newDefaults(partition)
+
+				service := &structs.NodeService{
+					Kind:    "connect-proxy",
+					ID:      "static-client-sidecar-proxy",
+					Service: "static-client-sidecar-proxy",
+					Address: "",
+					Port:    21000,
+					Proxy: structs.ConnectProxyConfig{
+						DestinationServiceName: "static-client",
+						DestinationServiceID:   "static-client",
+						LocalServiceAddress:    "127.0.0.1",
+						LocalServicePort:       8080,
+						Upstreams: []structs.Upstream{
+							{
+								DestinationType:      "service",
+								DestinationNamespace: "default",
+								DestinationPartition: partition,
+								DestinationPeer:      peerName,
+								DestinationName:      "static-server",
+								LocalBindAddress:     "0.0.0.0",
+								LocalBindPort:        5000,
+							},
+						},
+					},
+					EnterpriseMeta: acl.NewEnterpriseMetaWithPartition(partition, "default"),
+				}
+
+				expect := &structs.NodeService{
+					Kind:    "connect-proxy",
+					ID:      "static-client-sidecar-proxy",
+					Service: "static-client-sidecar-proxy",
+					Address: "",
+					Port:    21000,
+					Proxy: structs.ConnectProxyConfig{
+						DestinationServiceName: "static-client",
+						DestinationServiceID:   "static-client",
+						LocalServiceAddress:    "127.0.0.1",
+						LocalServicePort:       8080,
+						Config: map[string]any{
+							"protocol": "http",
+						},
+						Upstreams: []structs.Upstream{
+							{
+								DestinationType:      "service",
+								DestinationNamespace: "default",
+								DestinationPartition: partition,
+								DestinationPeer:      peerName,
+								DestinationName:      "static-server",
+								LocalBindAddress:     "0.0.0.0",
+								LocalBindPort:        5000,
+								MeshGateway: structs.MeshGatewayConfig{
+									Mode: "local",
+								},
+								Config: map[string]any{},
+							},
+						},
+						MeshGateway: structs.MeshGatewayConfig{
+							Mode: "local",
+						},
+					},
+					EnterpriseMeta: acl.NewEnterpriseMetaWithPartition(partition, "default"),
+				}
+
+				got, err := MergeServiceConfig(defaults, service)
+				require.NoError(t, err)
+				require.Equal(t, expect, got)
+			})
+
+			t.Run("dataplanes", func(t *testing.T) {
+				defaults := newDefaults(partition)
+
+				service := &structs.NodeService{
+					Kind:    "connect-proxy",
+					ID:      "static-client-sidecar-proxy",
+					Service: "static-client-sidecar-proxy",
+					Address: "10.61.57.9",
+					TaggedAddresses: map[string]structs.ServiceAddress{
+						"consul-virtual": {
+							Address: "240.0.0.2",
+							Port:    20000,
+						},
+					},
+					Port: 20000,
+					Proxy: structs.ConnectProxyConfig{
+						DestinationServiceName: "static-client",
+						DestinationServiceID:   "static-client",
+						LocalServicePort:       8080,
+						Upstreams: []structs.Upstream{
+							{
+								DestinationType:      "",
+								DestinationNamespace: "default",
+								DestinationPeer:      peerName,
+								DestinationName:      "static-server",
+								LocalBindAddress:     "0.0.0.0",
+								LocalBindPort:        5000,
+							},
+						},
+					},
+					EnterpriseMeta: acl.NewEnterpriseMetaWithPartition(partition, "default"),
+				}
+
+				expect := &structs.NodeService{
+					Kind:    "connect-proxy",
+					ID:      "static-client-sidecar-proxy",
+					Service: "static-client-sidecar-proxy",
+					Address: "10.61.57.9",
+					TaggedAddresses: map[string]structs.ServiceAddress{
+						"consul-virtual": {
+							Address: "240.0.0.2",
+							Port:    20000,
+						},
+					},
+					Port: 20000,
+					Proxy: structs.ConnectProxyConfig{
+						DestinationServiceName: "static-client",
+						DestinationServiceID:   "static-client",
+						LocalServicePort:       8080,
+						Config: map[string]any{
+							"protocol": "http",
+						},
+						Upstreams: []structs.Upstream{
+							{
+								DestinationType:      "",
+								DestinationNamespace: "default",
+								DestinationPeer:      peerName,
+								DestinationName:      "static-server",
+								LocalBindAddress:     "0.0.0.0",
+								LocalBindPort:        5000,
+								MeshGateway: structs.MeshGatewayConfig{
+									Mode: "local", // This field vanishes if the merging does not work for dataplanes.
+								},
+								Config: map[string]any{},
+							},
+						},
+						MeshGateway: structs.MeshGatewayConfig{
+							Mode: "local",
+						},
+					},
+					EnterpriseMeta: acl.NewEnterpriseMetaWithPartition(partition, "default"),
+				}
+
+				got, err := MergeServiceConfig(defaults, service)
+				require.NoError(t, err)
+				require.Equal(t, expect, got)
+			})
+		})
+	}
+}
+
 func Test_MergeServiceConfig_UpstreamOverrides(t *testing.T) {
 	type args struct {
 		defaults *structs.ServiceConfigResponse