Ensure that upstream configuration is properly normalized. (#19076)

This PR fixes an issue where upstreams did not correctly inherit the proper
namespace / partition from the parent service when attempting to fetch the
upstream protocol due to inconsistent normalization.

Some of the merge-service-configuration logic would normalize to default, while
some of the proxycfg logic would normalize to match the parent service. Due to
this mismatch in logic, an incorrect service-defaults configuration entry would
be fetched and have its protocol applied to the upstream.

.changelog/_6074.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+connect: **(Enterprise only)** Fix bug where incorrect service-defaults entries were fetched to determine an upstream's protocol whenever the upstream did not explicitly define the namespace / partition. When this bug occurs, upstreams would use the protocol from a service-default entry in the default namespace / partition, rather than their own namespace / partition.
+```

agent/configentry/merge_service_config.go

@@ -26,32 +26,22 @@ type StateStore interface {
 func MergeNodeServiceWithCentralConfig(
 	ws memdb.WatchSet,
 	state StateStore,
-	ns *structs.NodeService,
+	unmergedNS *structs.NodeService,
 	logger hclog.Logger) (uint64, *structs.NodeService, error) {
 
+	ns := unmergedNS.WithNormalizedUpstreams()
 	serviceName := ns.Service
-	var upstreams []structs.PeeredServiceName
 	if ns.IsSidecarProxy() {
 		// This is a sidecar proxy, ignore the proxy service's config since we are
 		// managed by the target service config.
 		serviceName = ns.Proxy.DestinationServiceName
-
+	}
-		// Also if we have any upstreams defined, add them to the defaults lookup request
+	var upstreams []structs.PeeredServiceName
-		// so we can learn about their configs.
 	for _, us := range ns.Proxy.Upstreams {
 		if us.DestinationType == "" || us.DestinationType == structs.UpstreamDestTypeService {
-				psn := us.DestinationID()
+			upstreams = append(upstreams, us.DestinationID())
-				if psn.Peer == "" {
-					psn.ServiceName.EnterpriseMeta.Merge(&ns.EnterpriseMeta)
-				} else {
-					// Peer services should not have their namespace overwritten.
-					psn.ServiceName.EnterpriseMeta.OverridePartition(ns.EnterpriseMeta.PartitionOrDefault())
-				}
-				upstreams = append(upstreams, psn)
 		}
 	}
-	}
-
 	configReq := &structs.ServiceConfigRequest{
 		Name:                 serviceName,
 		MeshGateway:          ns.Proxy.MeshGateway,

agent/service_manager.go

@@ -148,7 +148,8 @@ func (w *serviceConfigWatch) register(ctx context.Context) error {
 
 	// Merge the local registration with the central defaults and update this service
 	// in the local state.
-	merged, err := configentry.MergeServiceConfig(serviceDefaults, w.registration.Service)
+	ns := w.registration.Service.WithNormalizedUpstreams()
+	merged, err := configentry.MergeServiceConfig(serviceDefaults, ns)
 	if err != nil {
 		return err
 	}
@@ -278,7 +279,8 @@ func (w *serviceConfigWatch) handleUpdate(ctx context.Context, event cache.Updat
 
 	// Merge the local registration with the central defaults and update this service
 	// in the local state.
-	merged, err := configentry.MergeServiceConfig(serviceDefaults, w.registration.Service)
+	ns := w.registration.Service.WithNormalizedUpstreams()
+	merged, err := configentry.MergeServiceConfig(serviceDefaults, ns)
 	if err != nil {
 		return err
 	}

agent/structs/config_entry.go

@@ -867,19 +867,6 @@ func (s *ServiceConfigRequest) RequestDatacenter() string {
 	return s.Datacenter
 }
 
-// GetLocalUpstreamIDs returns the list of non-peer service ids for upstreams defined on this request.
-// This is often used for fetching service-defaults config entries.
-func (s *ServiceConfigRequest) GetLocalUpstreamIDs() []ServiceID {
-	var upstreams []ServiceID
-	for i := range s.UpstreamServiceNames {
-		u := &s.UpstreamServiceNames[i]
-		if u.Peer == "" {
-			upstreams = append(upstreams, u.ServiceName.ToServiceID())
-		}
-	}
-	return upstreams
-}
-
 func (r *ServiceConfigRequest) CacheInfo() cache.RequestInfo {
 	info := cache.RequestInfo{
 		Token:          r.Token,

agent/structs/config_entry_ce.go

@@ -62,3 +62,16 @@ func validateRatelimit(rl *RateLimits) error {
 }
 
 func (rl RateLimits) ToEnvoyExtension() *EnvoyExtension { return nil }
+
+// GetLocalUpstreamIDs returns the list of non-peer service ids for upstreams defined on this request.
+// This is often used for fetching service-defaults config entries.
+func (s *ServiceConfigRequest) GetLocalUpstreamIDs() []ServiceID {
+	var upstreams []ServiceID
+	for _, u := range s.UpstreamServiceNames {
+		if u.Peer != "" {
+			continue
+		}
+		upstreams = append(upstreams, u.ServiceName.ToServiceID())
+	}
+	return upstreams
+}

agent/structs/config_entry_ce_test.go

@@ -102,3 +102,60 @@ func TestDecodeConfigEntry_CE(t *testing.T) {
 		})
 	}
 }
+
+func Test_GetLocalUpstreamIDs(t *testing.T) {
+	cases := map[string]struct {
+		input  *ServiceConfigRequest
+		expect []ServiceID
+	}{
+		"no_upstreams": {
+			input: &ServiceConfigRequest{
+				Name: "svc",
+			},
+			expect: nil,
+		},
+		"upstreams": {
+			input: &ServiceConfigRequest{
+				Name: "svc",
+				UpstreamServiceNames: []PeeredServiceName{
+					{ServiceName: NewServiceName("a", nil)},
+					{ServiceName: NewServiceName("b", nil)},
+					{ServiceName: NewServiceName("c", nil)},
+				},
+			},
+			expect: []ServiceID{
+				{ID: "a"},
+				{ID: "b"},
+				{ID: "c"},
+			},
+		},
+		"peer_upstream": {
+			input: &ServiceConfigRequest{
+				Name: "svc",
+				UpstreamServiceNames: []PeeredServiceName{
+					{Peer: "p", ServiceName: NewServiceName("a", nil)},
+				},
+			},
+			expect: nil,
+		},
+		"mixed_upstreams": {
+			input: &ServiceConfigRequest{
+				Name: "svc",
+				UpstreamServiceNames: []PeeredServiceName{
+					{ServiceName: NewServiceName("a", nil)},
+					{Peer: "p", ServiceName: NewServiceName("b", nil)},
+					{ServiceName: NewServiceName("c", nil)},
+				},
+			},
+			expect: []ServiceID{
+				{ID: "a"},
+				{ID: "c"},
+			},
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			require.Equal(t, tc.expect, tc.input.GetLocalUpstreamIDs())
+		})
+	}
+}

agent/structs/structs_ce.go

@@ -174,3 +174,13 @@ func (s *ServiceNode) NodeIdentity() Identity {
 }
 
 type EnterpriseServiceUsage struct{}
+
+// WithNormalizedUpstreams returns a deep copy of the NodeService with no modifications to
+// data for CE versions.
+func (ns *NodeService) WithNormalizedUpstreams() *NodeService {
+	// Simply return a copy for CE, since it doesn't have partitions or namespaces.
+	if ns == nil {
+		return nil
+	}
+	return ns.DeepCopy()
+}

website/content/docs/connect/proxies/proxy-config-reference.mdx

@@ -164,9 +164,9 @@ You can configure the service mesh proxy to create listeners for upstream servic
 | Parameter | Description | Required | Default |
 | ---       | ---         | ---      | ---     |
 |`destination_name` | String value that specifies the name of the service or prepared query to route the service mesh to. The prepared query should be the name or the ID of the prepared query. | Required | None |
-| `destination_namespace` | String value that specifies the namespace containing the upstream service. <EnterpriseAlert inline /> | Optional | `default` |
+| `destination_namespace` | String value that specifies the namespace containing the upstream service. <EnterpriseAlert inline /> | Optional | Defaults to the local namespace |
 | `destination_peer` | String value that specifies the name of the peer cluster containing the upstream service. | Optional | None |
-| `destination_partition` | String value that specifies the name of the admin partition containing the upstream service. If `destination_peer` is set, `destination_partition` refers to the local admin partition in which the peering was established. <EnterpriseAlert inline /> | Optional | `default` |
+| `destination_partition` | String value that specifies the name of the admin partition containing the upstream service. If `destination_peer` is set, `destination_partition` refers to the local admin partition in which the peering was established. <EnterpriseAlert inline /> | Optional | Defaults to the local partition |
 | `local_bind_port` | Integer value that specifies the port to bind a local listener to. The application will make outbound connections to the upstream from the local port. | Required | None |
 | `local_bind_address` | String value that specifies the address to bind a local listener to. The application will make outbound connections to the upstream service from the local bind address. | Optional | `127.0.0.1` |
 | `local_bind_socket_path` | String value that specifies the path at which to bind a Unix domain socket listener. The application will make outbound connections to the upstream from the local bind socket path. <br/>This parameter conflicts with the `local_bind_port` or `local_bind_address` parameters. <br/>Supported when using Envoy as a proxy. | Optional | None|