diff --git a/.changelog/16000.txt b/.changelog/16000.txt new file mode 100644 index 000000000..aeea16f5e --- /dev/null +++ b/.changelog/16000.txt @@ -0,0 +1,3 @@ +```release-note:breaking-change +connect: Fix configuration merging for transparent proxy upstreams. Proxy-defaults and service-defaults config entries were not correctly merged for implicit upstreams in transparent proxy mode and would result in some configuration not being applied. To avoid issues when upgrading, ensure that any proxy-defaults or service-defaults have correct configuration for upstreams, since all fields will now be properly used to configure proxies. +``` diff --git a/agent/configentry/resolve.go b/agent/configentry/resolve.go index ee108ee70..d5e5b7f0a 100644 --- a/agent/configentry/resolve.go +++ b/agent/configentry/resolve.go @@ -4,6 +4,7 @@ import ( "fmt" "github.com/hashicorp/go-hclog" + "github.com/imdario/mergo" "github.com/mitchellh/copystructure" "github.com/hashicorp/consul/agent/structs" @@ -20,10 +21,18 @@ func ComputeResolvedServiceConfig( thisReply.MeshGateway.Mode = structs.MeshGatewayModeDefault + // Store the upstream defaults under a wildcard key so that they can be applied to + // upstreams that are inferred from intentions and do not have explicit upstream configuration. + wildcard := structs.NewServiceID(structs.WildcardSpecifier, args.WithWildcardNamespace()) + wildcardUpstreamDefaults := make(map[string]interface{}) + // resolvedConfigs stores the opaque config map for each upstream and is keyed on the upstream's ID. + resolvedConfigs := make(map[structs.ServiceID]map[string]interface{}) + // TODO(freddy) Refactor this into smaller set of state store functions // Pass the WatchSet to both the service and proxy config lookups. If either is updated during the // blocking query, this function will be rerun and these state store lookups will both be current. // We use the default enterprise meta to look up the global proxy defaults because they are not namespaced. + var proxyConfGlobalProtocol string proxyConf := entries.GetProxyDefaults(args.PartitionOrDefault()) if proxyConf != nil { @@ -32,6 +41,7 @@ func ComputeResolvedServiceConfig( if err != nil { return nil, fmt.Errorf("failed to copy global proxy-defaults: %v", err) } + thisReply.ProxyConfig = mapCopy.(map[string]interface{}) thisReply.Mode = proxyConf.Mode thisReply.TransparentProxy = proxyConf.TransparentProxy @@ -40,14 +50,20 @@ func ComputeResolvedServiceConfig( thisReply.EnvoyExtensions = proxyConf.EnvoyExtensions thisReply.AccessLogs = proxyConf.AccessLogs - // Extract the global protocol from proxyConf for upstream configs. - rawProtocol := proxyConf.Config["protocol"] - if rawProtocol != nil { - var ok bool - proxyConfGlobalProtocol, ok = rawProtocol.(string) - if !ok { - return nil, fmt.Errorf("invalid protocol type %T", rawProtocol) - } + parsed, err := structs.ParseUpstreamConfigNoDefaults(mapCopy.(map[string]interface{})) + if err != nil { + return nil, fmt.Errorf("failed to parse upstream config map for proxy-defaults: %v", err) + } + proxyConfGlobalProtocol = parsed.Protocol + + // MeshGateway is strange. It's marshaled into UpstreamConfigs via the arbitrary map, but it + // uses concrete fields everywhere else. We always take the explicit definition here for + // wildcard upstreams and discard the user setting it via arbitrary map in proxy-defaults. + if err := mergo.MergeWithOverwrite(&wildcardUpstreamDefaults, mapCopy); err != nil { + return nil, fmt.Errorf("failed to merge upstream config map for proxy-defaults: %v", err) + } + if !proxyConf.MeshGateway.IsZero() { + wildcardUpstreamDefaults["mesh_gateway"] = proxyConf.MeshGateway } } @@ -64,6 +80,7 @@ func ComputeResolvedServiceConfig( } if serviceConf.MeshGateway.Mode != structs.MeshGatewayModeDefault { thisReply.MeshGateway.Mode = serviceConf.MeshGateway.Mode + wildcardUpstreamDefaults["mesh_gateway"] = serviceConf.MeshGateway } if serviceConf.TransparentProxy.OutboundListenerPort != 0 { thisReply.TransparentProxy.OutboundListenerPort = serviceConf.TransparentProxy.OutboundListenerPort @@ -139,10 +156,8 @@ func ComputeResolvedServiceConfig( // Then store upstreams inferred from service-defaults and mapify the overrides. var ( - upstreamOverrides = make(map[structs.ServiceID]*structs.UpstreamConfig) upstreamDefaults *structs.UpstreamConfig - // resolvedConfigs stores the opaque config map for each upstream and is keyed on the upstream's ID. - resolvedConfigs = make(map[structs.ServiceID]map[string]interface{}) + upstreamOverrides = make(map[structs.ServiceID]*structs.UpstreamConfig) ) if serviceConf != nil && serviceConf.UpstreamConfig != nil { for i, override := range serviceConf.UpstreamConfig.Overrides { @@ -161,25 +176,26 @@ func ComputeResolvedServiceConfig( } if serviceConf.UpstreamConfig.Defaults != nil { upstreamDefaults = serviceConf.UpstreamConfig.Defaults - if upstreamDefaults.MeshGateway.Mode == structs.MeshGatewayModeDefault { upstreamDefaults.MeshGateway.Mode = thisReply.MeshGateway.Mode } - - // Store the upstream defaults under a wildcard key so that they can be applied to - // upstreams that are inferred from intentions and do not have explicit upstream configuration. - cfgMap := make(map[string]interface{}) - upstreamDefaults.MergeInto(cfgMap) - - if !args.MeshGateway.IsZero() { - cfgMap["mesh_gateway"] = args.MeshGateway - } - - wildcard := structs.NewServiceID(structs.WildcardSpecifier, args.WithWildcardNamespace()) - resolvedConfigs[wildcard] = cfgMap + upstreamDefaults.MergeInto(wildcardUpstreamDefaults) + // Always add the wildcard upstream if a service-defaults default-upstream was configured. + resolvedConfigs[wildcard] = wildcardUpstreamDefaults } } + if !args.MeshGateway.IsZero() { + wildcardUpstreamDefaults["mesh_gateway"] = args.MeshGateway + } + + // Add the wildcard upstream if any fields were populated (it may have been already + // added if a service-defaults exists). We likely could always add it without issues, + // but this has been existing behavior, and many unit tests would break. + if len(wildcardUpstreamDefaults) > 0 { + resolvedConfigs[wildcard] = wildcardUpstreamDefaults + } + for upstream := range seenUpstreams { resolvedCfg := make(map[string]interface{}) diff --git a/agent/configentry/resolve_test.go b/agent/configentry/resolve_test.go index 151cf0636..cc07712b4 100644 --- a/agent/configentry/resolve_test.go +++ b/agent/configentry/resolve_test.go @@ -120,6 +120,14 @@ func Test_ComputeResolvedServiceConfig(t *testing.T) { want: &structs.ServiceConfigResponse{ MeshGateway: remoteMeshGW, UpstreamIDConfigs: structs.OpaqueUpstreamConfigs{ + { + Upstream: wildcard, + Config: map[string]interface{}{ + "mesh_gateway": structs.MeshGatewayConfig{ + Mode: structs.MeshGatewayModeRemote, + }, + }, + }, { Upstream: uid, Config: map[string]interface{}{ @@ -183,6 +191,15 @@ func Test_ComputeResolvedServiceConfig(t *testing.T) { Path: "/tmp/accesslog.txt", JSONFormat: "{ \"custom_start_time\": \"%START_TIME%\" }", }, + UpstreamIDConfigs: structs.OpaqueUpstreamConfigs{ + { + Upstream: wildcard, + Config: map[string]interface{}{ + "foo": "bar", + "mesh_gateway": remoteMeshGW, + }, + }, + }, }, }, { @@ -209,6 +226,12 @@ func Test_ComputeResolvedServiceConfig(t *testing.T) { want: &structs.ServiceConfigResponse{ MeshGateway: noneMeshGW, // service-defaults has a higher precedence. UpstreamIDConfigs: structs.OpaqueUpstreamConfigs{ + { + Upstream: wildcard, + Config: map[string]interface{}{ + "mesh_gateway": noneMeshGW, + }, + }, { Upstream: uid, Config: map[string]interface{}{ diff --git a/agent/consul/config_endpoint_test.go b/agent/consul/config_endpoint_test.go index fa06cf9b9..34e9ce72b 100644 --- a/agent/consul/config_endpoint_test.go +++ b/agent/consul/config_endpoint_test.go @@ -1057,6 +1057,9 @@ func TestConfigEntry_ResolveServiceConfig(t *testing.T) { "protocol": "http", }, UpstreamConfigs: map[string]map[string]interface{}{ + "*": { + "foo": int64(1), + }, "bar": { "protocol": "grpc", }, @@ -1270,6 +1273,9 @@ func TestConfigEntry_ResolveServiceConfig_Upstreams(t *testing.T) { "protocol": "grpc", }, UpstreamConfigs: map[string]map[string]interface{}{ + "*": { + "protocol": "grpc", + }, "mysql": { "protocol": "http", }, @@ -1314,6 +1320,12 @@ func TestConfigEntry_ResolveServiceConfig_Upstreams(t *testing.T) { "protocol": "grpc", }, UpstreamIDConfigs: structs.OpaqueUpstreamConfigs{ + { + Upstream: wildcard, + Config: map[string]interface{}{ + "protocol": "grpc", + }, + }, { Upstream: cache, Config: map[string]interface{}{ @@ -2052,6 +2064,9 @@ func TestConfigEntry_ResolveServiceConfig_UpstreamProxyDefaultsProtocol(t *testi "protocol": "http", }, UpstreamConfigs: map[string]map[string]interface{}{ + "*": { + "protocol": "http", + }, "bar": { "protocol": "http", }, @@ -2107,6 +2122,9 @@ func TestConfigEntry_ResolveServiceConfig_ProxyDefaultsProtocol_UsedForAllUpstre "protocol": "http", }, UpstreamConfigs: map[string]map[string]interface{}{ + "*": { + "protocol": "http", + }, "bar": { "protocol": "http", }, diff --git a/agent/proxycfg/connect_proxy.go b/agent/proxycfg/connect_proxy.go index f544ef701..81b41db69 100644 --- a/agent/proxycfg/connect_proxy.go +++ b/agent/proxycfg/connect_proxy.go @@ -197,7 +197,7 @@ func (s *handlerConnectProxy) initialize(ctx context.Context) (ConfigSnapshot, e case "": if u.DestinationPeer != "" { - err := s.setupWatchesForPeeredUpstream(ctx, snap.ConnectProxy, NewUpstreamID(&u), u.MeshGateway.Mode, dc) + err := s.setupWatchesForPeeredUpstream(ctx, snap.ConnectProxy, NewUpstreamID(&u), dc) if err != nil { return snap, fmt.Errorf("failed to setup watches for peered upstream %q: %w", uid.String(), err) } @@ -231,7 +231,6 @@ func (s *handlerConnectProxy) setupWatchesForPeeredUpstream( ctx context.Context, snapConnectProxy configSnapshotConnectProxy, uid UpstreamID, - mgwMode structs.MeshGatewayMode, dc string, ) error { s.logger.Trace("initializing watch of peered upstream", "upstream", uid) @@ -272,14 +271,10 @@ func (s *handlerConnectProxy) setupWatchesForPeeredUpstream( snapConnectProxy.UpstreamPeerTrustBundles.InitWatch(uid.Peer, cancel) } - // If a peered upstream is set to local mesh gw mode, - // set up a watch for them. - if mgwMode == structs.MeshGatewayModeLocal { - up := &handlerUpstreams{handlerState: s.handlerState} - up.setupWatchForLocalGWEndpoints(ctx, &snapConnectProxy.ConfigSnapshotUpstreams) - } else if mgwMode == structs.MeshGatewayModeNone { - s.logger.Warn(fmt.Sprintf("invalid mesh gateway mode 'none', defaulting to 'remote' for %q", uid)) - } + // Always watch local GW endpoints for peer upstreams so that we don't have to worry about + // the timing on whether the wildcard upstream config was fetched yet. + up := &handlerUpstreams{handlerState: s.handlerState} + up.setupWatchForLocalGWEndpoints(ctx, &snapConnectProxy.ConfigSnapshotUpstreams) return nil } @@ -329,7 +324,7 @@ func (s *handlerConnectProxy) handleUpdate(ctx context.Context, u UpdateEvent, s } seenUpstreams[uid] = struct{}{} - err := s.setupWatchesForPeeredUpstream(ctx, snap.ConnectProxy, uid, s.proxyCfg.MeshGateway.Mode, s.source.Datacenter) + err := s.setupWatchesForPeeredUpstream(ctx, snap.ConnectProxy, uid, s.source.Datacenter) if err != nil { return fmt.Errorf("failed to setup watches for peered upstream %q: %w", uid.String(), err) } @@ -402,8 +397,7 @@ func (s *handlerConnectProxy) handleUpdate(ctx context.Context, u UpdateEvent, s // Use the centralized upstream defaults if they exist and there isn't specific configuration for this upstream // This is only relevant to upstreams from intentions because for explicit upstreams the defaulting is handled // by the ResolveServiceConfig endpoint. - wildcardSID := structs.NewServiceID(structs.WildcardSpecifier, s.proxyID.WithWildcardNamespace()) - wildcardUID := NewUpstreamIDFromServiceID(wildcardSID) + wildcardUID := NewWildcardUID(&s.proxyID.EnterpriseMeta) defaults, ok := snap.ConnectProxy.UpstreamConfig[wildcardUID] if ok { u = defaults diff --git a/agent/proxycfg/naming.go b/agent/proxycfg/naming.go index 08ff216ed..0ab355160 100644 --- a/agent/proxycfg/naming.go +++ b/agent/proxycfg/naming.go @@ -180,3 +180,11 @@ func UpstreamsToMap(us structs.Upstreams) map[UpstreamID]*structs.Upstream { } return upstreamMap } + +func NewWildcardUID(entMeta *acl.EnterpriseMeta) UpstreamID { + wildcardSID := structs.NewServiceID( + structs.WildcardSpecifier, + entMeta.WithWildcardNamespace(), + ) + return NewUpstreamIDFromServiceID(wildcardSID) +} diff --git a/agent/proxycfg/snapshot.go b/agent/proxycfg/snapshot.go index 57be8bf7f..15d9aef70 100644 --- a/agent/proxycfg/snapshot.go +++ b/agent/proxycfg/snapshot.go @@ -189,6 +189,20 @@ func (c *configSnapshotConnectProxy) IsImplicitUpstream(uid UpstreamID) bool { return intentionImplicit || peeringImplicit } +func (c *configSnapshotConnectProxy) GetUpstream(uid UpstreamID, entMeta *acl.EnterpriseMeta) (*structs.Upstream, bool) { + upstream, found := c.UpstreamConfig[uid] + // We should fallback to the wildcard defaults generated from service-defaults + proxy-defaults + // whenever we don't find the upstream config. + if !found { + wildcardUID := NewWildcardUID(entMeta) + upstream = c.UpstreamConfig[wildcardUID] + } + + explicit := upstream != nil && upstream.HasLocalPortOrSocket() + implicit := c.IsImplicitUpstream(uid) + return upstream, !implicit && !explicit +} + type configSnapshotTerminatingGateway struct { MeshConfig *structs.MeshConfigEntry MeshConfigSet bool diff --git a/agent/service_manager_test.go b/agent/service_manager_test.go index c34626830..6760627e0 100644 --- a/agent/service_manager_test.go +++ b/agent/service_manager_test.go @@ -6,10 +6,12 @@ import ( "fmt" "os" "path/filepath" + "sort" "testing" "github.com/stretchr/testify/require" + "github.com/hashicorp/consul/acl" "github.com/hashicorp/consul/agent/structs" "github.com/hashicorp/consul/sdk/testutil/retry" "github.com/hashicorp/consul/testrpc" @@ -419,7 +421,13 @@ func TestServiceManager_PersistService_API(t *testing.T) { "protocol": "http", }, UpstreamIDConfigs: structs.OpaqueUpstreamConfigs{ - structs.OpaqueUpstreamConfig{ + { + Upstream: structs.NewServiceID(structs.WildcardSpecifier, acl.DefaultEnterpriseMeta().WithWildcardNamespace()), + Config: map[string]interface{}{ + "foo": int64(1), + }, + }, + { Upstream: structs.NewServiceID("redis", nil), Config: map[string]interface{}{ "protocol": "tcp", @@ -429,7 +437,7 @@ func TestServiceManager_PersistService_API(t *testing.T) { }, EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(), } - expectJSONFile(t, configFile, pcfg, resetDefaultsQueryMeta) + expectJSONFile(t, configFile, pcfg, fixPersistedServiceConfigForTest) // Verify in memory state. { @@ -467,7 +475,13 @@ func TestServiceManager_PersistService_API(t *testing.T) { "protocol": "http", }, UpstreamIDConfigs: structs.OpaqueUpstreamConfigs{ - structs.OpaqueUpstreamConfig{ + { + Upstream: structs.NewServiceID(structs.WildcardSpecifier, acl.DefaultEnterpriseMeta().WithWildcardNamespace()), + Config: map[string]interface{}{ + "foo": int64(1), + }, + }, + { Upstream: structs.NewServiceID("redis", nil), Config: map[string]interface{}{ "protocol": "tcp", @@ -477,7 +491,7 @@ func TestServiceManager_PersistService_API(t *testing.T) { }, EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(), } - expectJSONFile(t, configFile, pcfg, resetDefaultsQueryMeta) + expectJSONFile(t, configFile, pcfg, fixPersistedServiceConfigForTest) // Verify in memory state. expectState.Proxy.LocalServicePort = 8001 @@ -646,7 +660,13 @@ func TestServiceManager_PersistService_ConfigFiles(t *testing.T) { "protocol": "http", }, UpstreamIDConfigs: structs.OpaqueUpstreamConfigs{ - structs.OpaqueUpstreamConfig{ + { + Upstream: structs.NewServiceID(structs.WildcardSpecifier, acl.DefaultEnterpriseMeta().WithWildcardNamespace()), + Config: map[string]interface{}{ + "foo": int64(1), + }, + }, + { Upstream: structs.NewServiceID("redis", nil), Config: map[string]interface{}{ "protocol": "tcp", @@ -655,7 +675,7 @@ func TestServiceManager_PersistService_ConfigFiles(t *testing.T) { }, }, EnterpriseMeta: *structs.DefaultEnterpriseMetaInDefaultPartition(), - }, resetDefaultsQueryMeta) + }, fixPersistedServiceConfigForTest) // Verify in memory state. { @@ -813,6 +833,23 @@ func expectJSONFile(t *testing.T, file string, expect interface{}, fixupContentB require.JSONEq(t, string(expected), string(content)) } +func fixPersistedServiceConfigForTest(content []byte) ([]byte, error) { + var parsed persistedServiceConfig + if err := json.Unmarshal(content, &parsed); err != nil { + return nil, err + } + // Sort the output, since it's randomized and causes flaky tests otherwise. + sort.Slice(parsed.Defaults.UpstreamIDConfigs, func(i, j int) bool { + return parsed.Defaults.UpstreamIDConfigs[i].Upstream.ID < parsed.Defaults.UpstreamIDConfigs[j].Upstream.ID + }) + out, err := json.Marshal(parsed) + if err != nil { + return nil, err + } + // Clean the query meta + return resetDefaultsQueryMeta(out) +} + // resetDefaultsQueryMeta will reset the embedded fields from structs.QueryMeta // to their zero values in the json object keyed under 'Defaults'. func resetDefaultsQueryMeta(content []byte) ([]byte, error) { diff --git a/agent/xds/clusters.go b/agent/xds/clusters.go index 5aecd37dc..69d3cc7f4 100644 --- a/agent/xds/clusters.go +++ b/agent/xds/clusters.go @@ -87,18 +87,10 @@ func (s *ResourceGenerator) clustersFromSnapshotConnectProxy(cfgSnap *proxycfg.C clusters = append(clusters, passthroughs...) } - getUpstream := func(uid proxycfg.UpstreamID) (*structs.Upstream, bool) { - upstream := cfgSnap.ConnectProxy.UpstreamConfig[uid] - - explicit := upstream.HasLocalPortOrSocket() - implicit := cfgSnap.ConnectProxy.IsImplicitUpstream(uid) - return upstream, !implicit && !explicit - } - // NOTE: Any time we skip a chain below we MUST also skip that discovery chain in endpoints.go // so that the sets of endpoints generated matches the sets of clusters. for uid, chain := range cfgSnap.ConnectProxy.DiscoveryChain { - upstream, skip := getUpstream(uid) + upstream, skip := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta) if skip { continue } @@ -123,7 +115,7 @@ func (s *ResourceGenerator) clustersFromSnapshotConnectProxy(cfgSnap *proxycfg.C // upstream in endpoints.go so that the sets of endpoints generated matches // the sets of clusters. for _, uid := range cfgSnap.ConnectProxy.PeeredUpstreamIDs() { - upstream, skip := getUpstream(uid) + upstream, skip := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta) if skip { continue } diff --git a/agent/xds/endpoints.go b/agent/xds/endpoints.go index 1f3966de4..9c76f43f8 100644 --- a/agent/xds/endpoints.go +++ b/agent/xds/endpoints.go @@ -50,18 +50,10 @@ func (s *ResourceGenerator) endpointsFromSnapshotConnectProxy(cfgSnap *proxycfg. cfgSnap.ConnectProxy.PeerUpstreamEndpoints.Len()+ len(cfgSnap.ConnectProxy.WatchedUpstreamEndpoints)) - getUpstream := func(uid proxycfg.UpstreamID) (*structs.Upstream, bool) { - upstream := cfgSnap.ConnectProxy.UpstreamConfig[uid] - - explicit := upstream.HasLocalPortOrSocket() - implicit := cfgSnap.ConnectProxy.IsImplicitUpstream(uid) - return upstream, !implicit && !explicit - } - // NOTE: Any time we skip a chain below we MUST also skip that discovery chain in clusters.go // so that the sets of endpoints generated matches the sets of clusters. for uid, chain := range cfgSnap.ConnectProxy.DiscoveryChain { - upstream, skip := getUpstream(uid) + upstream, skip := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta) if skip { // Discovery chain is not associated with a known explicit or implicit upstream so it is skipped. continue @@ -92,7 +84,7 @@ func (s *ResourceGenerator) endpointsFromSnapshotConnectProxy(cfgSnap *proxycfg. // upstream in clusters.go so that the sets of endpoints generated matches // the sets of clusters. for _, uid := range cfgSnap.ConnectProxy.PeeredUpstreamIDs() { - upstream, skip := getUpstream(uid) + upstream, skip := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta) if skip { // Discovery chain is not associated with a known explicit or implicit upstream so it is skipped. continue @@ -555,6 +547,10 @@ func (s *ResourceGenerator) makeUpstreamLoadAssignmentForPeerService( return la, err } + if upstreamGatewayMode == structs.MeshGatewayModeNone { + s.Logger.Warn(fmt.Sprintf("invalid mesh gateway mode 'none', defaulting to 'remote' for %q", uid)) + } + // If an upstream is configured with local mesh gw mode, we make a load assignment // from the gateway endpoints instead of those of the upstreams. if upstreamGatewayMode == structs.MeshGatewayModeLocal { @@ -651,7 +647,7 @@ func (s *ResourceGenerator) endpointsFromDiscoveryChain( } mgwMode := structs.MeshGatewayModeDefault - if upstream := cfgSnap.ConnectProxy.UpstreamConfig[uid]; upstream != nil { + if upstream, _ := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta); upstream != nil { mgwMode = upstream.MeshGateway.Mode } diff --git a/agent/xds/listeners.go b/agent/xds/listeners.go index ed8889a29..2a94a050c 100644 --- a/agent/xds/listeners.go +++ b/agent/xds/listeners.go @@ -134,17 +134,8 @@ func (s *ResourceGenerator) listenersFromSnapshotConnectProxy(cfgSnap *proxycfg. return nil, err } - getUpstream := func(uid proxycfg.UpstreamID) (*structs.Upstream, bool) { - upstream := cfgSnap.ConnectProxy.UpstreamConfig[uid] - - explicit := upstream.HasLocalPortOrSocket() - implicit := cfgSnap.ConnectProxy.IsImplicitUpstream(uid) - return upstream, !implicit && !explicit - } - for uid, chain := range cfgSnap.ConnectProxy.DiscoveryChain { - upstreamCfg, skip := getUpstream(uid) - + upstreamCfg, skip := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta) if skip { // Discovery chain is not associated with a known explicit or implicit upstream so it is skipped. continue @@ -355,8 +346,7 @@ func (s *ResourceGenerator) listenersFromSnapshotConnectProxy(cfgSnap *proxycfg. // Looping over explicit and implicit upstreams is only needed for cross-peer // because they do not have discovery chains. for _, uid := range cfgSnap.ConnectProxy.PeeredUpstreamIDs() { - upstreamCfg, skip := getUpstream(uid) - + upstreamCfg, skip := cfgSnap.ConnectProxy.GetUpstream(uid, &cfgSnap.ProxyID.EnterpriseMeta) if skip { // Not associated with a known explicit or implicit upstream so it is skipped. continue @@ -2260,10 +2250,9 @@ func (s *ResourceGenerator) getAndModifyUpstreamConfigForPeeredListener( s.Logger.Warn("failed to parse", "upstream", uid, "error", err) } - protocol := cfg.Protocol - if protocol == "" { - protocol = peerMeta.Protocol - } + // Ignore the configured protocol for peer upstreams, since it is defined by the remote + // cluster, which we cannot control. + protocol := peerMeta.Protocol if protocol == "" { protocol = "tcp" }