fix(proxyCfg): propogate meshGatewayConfig to upstreams of API gateway (#23369)

* fix(proxyCfg): propogate meshGatewayConfig to upstreams of API gateway

- Fixed handleRouteConfigUpdate to properly propagate meshGatewayConfig to API gateway upstreams, which is required during XDS endpoint and cluster config generation.
- Added TestStateChangedAPIGateway test cases in state_test.go to validate API gateway update handling.
- Added API gateway-specific logging prefix (similar to mesh gateway) to help in debugging.

* fix(xds): correct endpoint config generation for API gateway in peered setups (#23370)

* fix(xds): correct endpoint config generation for API gateway in peered setups

- Previously, API gateway XDS endpoint generation incorrectly relied on cfgSnap.ConnectProxy config (instead of cgfSnap.APIGateway), which caused wrong/no endpoint configuration for peered environments.
- Changes made:
- Updated makeUpstreamLoadAssignmentForPeerService to fetch localGatewayEndpoint based on cfgSnap kind instead of always using cfgSnap.ConnectProxy.
- Updated endpointsFromDiscoveryChain to derive meshGatewayMode based on cfgSnap kind instead of always using cfgSnap.ConnectProxy.
- Recompiled golden test file to reflect fix.

* removed comment

* fix(xds): correct cluster config generation for API gateway in peered setups (#23371)

* fix(xds): correct cluster config generation for API gateway in peered setups

- Updated makeUpstreamClustersForDiscoveryChain to generate cluster config based on upstream endpoint type. Before this fix, it always generated cluster configs without endpoints, which is incorrect when the upstream endpoint type is hostname and mesh-gateway mode is remote; in such cases, endpoints must also be included in the cluster config.

- Added recompiled golden test file to reflect the fix.

* fix lint error

Author: LordAbhishek

agent/proxycfg/api_gateway.go

@@ -360,6 +360,19 @@ func (h *handlerAPIGateway) handleRouteConfigUpdate(ctx context.Context, u Updat
 
 		for _, rule := range route.Rules {
 			for _, service := range rule.Services {
+
+				// Retrieving the meshGatewayConfig from handlerAPIGateway instance.
+				// `handlerAPIGateway` embeds `handlerState`, which exposes `serviceInstance.proxyCfg`.
+				// serviceInstance.proxyCfg.MeshGateway is replicated from NodeService during state setup/update.
+				// and NodeService populated for all gateway's during service resistration `AgentRegisterService`.
+				//
+				// So, Whenever any change happens in NodeService, proxyCfg manager will recreate
+				// the state where it copies NodeService to serviceInstance and
+				// then calls this api_gateway handleUpdates method.
+				// which will update the Mesh-Gateway config to api_gateway upstreams (below).
+				// h.service = <name of api-gateway>
+				meshGatewayConfig := h.proxyCfg.MeshGateway
+
 				for _, listener := range snap.APIGateway.Listeners {
 					shouldBind := false
 					for _, parent := range route.Parents {
@@ -382,6 +395,10 @@ func (h *handlerAPIGateway) handleRouteConfigUpdate(ctx context.Context, u Updat
 						Config: map[string]interface{}{
 							"protocol": "http",
 						},
+						// Propogate the meshGatewayConfig in api gateway upstreams
+						// so that meshGatewayMode can be used in XDS for
+						// endpoints and cluster config generation.
+						MeshGateway: meshGatewayConfig,
 					}
 
 					listenerKey := APIGatewayListenerKeyFromListener(listener)
@@ -410,6 +427,7 @@ func (h *handlerAPIGateway) handleRouteConfigUpdate(ctx context.Context, u Updat
 		snap.APIGateway.TCPRoutes.Set(ref, route)
 
 		for _, service := range route.Services {
+			meshGatewayConfig := h.proxyCfg.MeshGateway
 			upstreamID := NewUpstreamIDFromServiceName(service.ServiceName())
 			seenUpstreamIDs.add(upstreamID)
 
@@ -436,6 +454,7 @@ func (h *handlerAPIGateway) handleRouteConfigUpdate(ctx context.Context, u Updat
 					Config: map[string]interface{}{
 						"protocol": "tcp",
 					},
+					MeshGateway: meshGatewayConfig,
 				}
 
 				listenerKey := APIGatewayListenerKeyFromListener(listener)

agent/proxycfg/state.go

@@ -225,6 +225,7 @@ func newKindHandler(config stateConfig, s serviceInstance, ch chan UpdateEvent)
 	case structs.ServiceKindIngressGateway:
 		handler = &handlerIngressGateway{handlerState: h}
 	case structs.ServiceKindAPIGateway:
+		h.logger = config.logger.Named(logging.APIGateway)
 		handler = &handlerAPIGateway{handlerState: h}
 	default:
 		return nil, errors.New("not a connect-proxy, terminating-gateway, mesh-gateway, or ingress-gateway")

agent/proxycfg/state_test.go

@@ -26,7 +26,7 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
-func TestStateChanged(t *testing.T) {
+func TestStateChangedConnectProxy(t *testing.T) {
 	tests := []struct {
 		name   string
 		ns     *structs.NodeService
@@ -108,6 +108,131 @@ func TestStateChanged(t *testing.T) {
 			},
 			want: true,
 		},
+		{
+			name:  "different proxy mesh gateway mode",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				ns.Proxy.MeshGateway.Mode = structs.MeshGatewayModeLocal
+				return &ns, token
+			},
+			want: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			proxyID := ProxyID{ServiceID: tt.ns.CompoundServiceID()}
+			state, err := newState(proxyID, tt.ns, testSource, tt.token, stateConfig{logger: hclog.New(nil)}, rate.NewLimiter(rate.Inf, 1))
+			require.NoError(t, err)
+			otherNS, otherToken := tt.mutate(*tt.ns, tt.token)
+			require.Equal(t, tt.want, state.Changed(otherNS, otherToken))
+		})
+	}
+}
+
+func TestStateChangedAPIGateway(t *testing.T) {
+	tests := []struct {
+		name   string
+		ns     *structs.NodeService
+		token  string
+		mutate func(ns structs.NodeService, token string) (*structs.NodeService, string)
+		want   bool
+	}{
+		{
+			name: "nil node service",
+			ns:   structs.TestNodeServiceAPIGateway(t),
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				return nil, token
+			},
+			want: true,
+		},
+		{
+			name: "same service",
+			ns:   structs.TestNodeServiceAPIGateway(t),
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				return &ns, token
+			}, want: false,
+		},
+		{
+			name:  "same service, different token",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				return &ns, "bar"
+			},
+			want: true,
+		},
+		{
+			name:  "different address",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				ns.Address = "10.10.10.10"
+				return &ns, token
+			},
+			want: true,
+		},
+		{
+			name:  "different port",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				ns.Port = 12345
+				return &ns, token
+			},
+			want: true,
+		},
+		{
+			name:  "different service kind",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				ns.Kind = ""
+				return &ns, token
+			},
+			want: true,
+		},
+		{
+			name:  "different proxy target",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				ns.Proxy.DestinationServiceName = "badger"
+				return &ns, token
+			},
+			want: true,
+		},
+		{
+			name:  "different proxy upstreams",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				ns.Proxy.Upstreams = nil
+				return &ns, token
+			},
+			want: true,
+		},
+		{
+			name:  "different mesh gateway mode (local)",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				ns.Proxy.MeshGateway.Mode = structs.MeshGatewayModeLocal
+				return &ns, token
+			},
+			want: true,
+		},
+		{
+			name:  "different mesh gateway mode (remote)",
+			ns:    structs.TestNodeServiceAPIGateway(t),
+			token: "foo",
+			mutate: func(ns structs.NodeService, token string) (*structs.NodeService, string) {
+				ns.Proxy.MeshGateway.Mode = structs.MeshGatewayModeRemote
+				return &ns, token
+			},
+			want: true,
+		},
 	}
 
 	for _, tt := range tests {

agent/structs/testing_catalog.go

@@ -176,10 +176,35 @@ func TestNodeServiceMeshGateway(t testing.T) *NodeService {
 }
 
 func TestNodeServiceAPIGateway(t testing.T) *NodeService {
+	entMeta := DefaultEnterpriseMetaInPartition("")
 	return &NodeService{
 		Kind:    ServiceKindAPIGateway,
 		Service: "api-gateway",
 		Address: "1.1.1.1",
+
+		// ---------------------------------------
+		// Adding TestConnectProxyConfig to the proxy field here within TestNodeServiceAPIGateway
+		// to test whether APIGateway is able to handle state changes within ConnectProxyConfig (TestStateChangedAPIGateway).
+		// Please note:
+		// The naming may suggest that ConnectProxyConfig should only be used for ConnectProxy,
+		// but "APIGateway state" uses serviceInstance, which embeds ConnectProxyConfig as part of its state,
+		// so any changes to ConnectProxyConfig will also impact APIGateway, such as change in "mesh gateway mode".
+		//
+		// For example, let's say a user updates the mesh gateway mode of an API gateway,
+		// First NodeService.Proxy will be updated and then proxyCfg manager detects change in config
+		// and it recreates the state for api_gateway which would copy the
+		// NodeService.Proxy.MeshGateway to serviceInstance.proxyCfg.MeshGateway in `newServiceInstanceFromNodeService` (serviceInstance is part of state)
+		// and then proxyCfg manager calls the api_gateway handleUpdates method which would
+		// update the api_gateway upstreams with new meshGateway config.
+		//
+		// Now, this serviceInstance.proxyCfg and NodeService.Proxy
+		// refers to same proxy configuration, which is of type ConnectProxyConfig.
+
+		// So, we need to test it as well.
+		// ---------------------------------------
+
+		Proxy:          TestConnectProxyConfig(t),
+		EnterpriseMeta: *entMeta,
 	}
 }
 

agent/xds/clusters.go

@@ -1533,25 +1533,47 @@ func (s *ResourceGenerator) makeUpstreamClustersForDiscoveryChain(
 		for _, groupedTarget := range targetGroups {
 			s.Logger.Debug("generating cluster for", "cluster", groupedTarget.ClusterName)
 
+			// Now this makeUpstreamClusterForDiscoveryChain, is a generic method
+			// and used by connect proxy, ingress gateway and api gateway.
+			//
+			// Issue: This method always make cluster (without endpoints).
+			// `ClusterDiscoveryType: &envoy_cluster_v3.Cluster_Type{Type: envoy_cluster_v3.Cluster_EDS},`
+			//
+			// Envoy Exception:
+			// As we know that any service whose upstream endpoint is of type hostname,
+			// envoy cannot resolve hostname as EDS.
+			// So, we need to use CDS to send endpoints as well along with cluster configs.
+			//
+			// Context:
+			// When we have 2 consul DC peered with mesh gw ON AWS and
+			// we have API gw on DC1 which need to access a service X which exist on DC2.
+			// Also, API gateway is configured to in mesh-gw remote mode.
+			// In this case, API gateway upstream would be DC2's mesh-gateway and
+			// we configure API GW envoy upstream endpoint to it.
+			// The problem is envoy exception and AWS mesh-gw lb type.
+			// AWS generates hostname based endpoints for mesh-gw lb endpoint and
+			// when we have hostname based endpoints envoy cannot resolve it via EDS,
+			// So we configure that endpoint via CDS.
+			//
+			// If not fixed, whenever any service (gateway) whose upstream endpoint is of hostname type,
+			// cluster endpoints will be empty and envoy will fail to route traffic to that cluster.
+			//
+			// Fix: Add logic to check if we should create a cluster config
+			// without upstream endpoint or with upstream endpoint (hostnames)
+			// based on upstream endpoint type.
+			//
+			// You can refer makeUpstreamClusterForPeerService - used by connect proxy for similar logic.
+			// or makeGatewayCluster - used by mesh gw for peering services.
+
 			c := &envoy_cluster_v3.Cluster{
-				Name:                 groupedTarget.ClusterName,
-				AltStatName:          groupedTarget.ClusterName,
-				ConnectTimeout:       durationpb.New(node.Resolver.ConnectTimeout),
-				ClusterDiscoveryType: &envoy_cluster_v3.Cluster_Type{Type: envoy_cluster_v3.Cluster_EDS},
+				Name:           groupedTarget.ClusterName,
+				AltStatName:    groupedTarget.ClusterName,
+				ConnectTimeout: durationpb.New(node.Resolver.ConnectTimeout),
 				CommonLbConfig: &envoy_cluster_v3.Cluster_CommonLbConfig{
 					HealthyPanicThreshold: &envoy_type_v3.Percent{
 						Value: 0, // disable panic threshold
 					},
 				},
-				EdsClusterConfig: &envoy_cluster_v3.Cluster_EdsClusterConfig{
-					EdsConfig: &envoy_core_v3.ConfigSource{
-						InitialFetchTimeout: cfgSnap.GetXDSCommonConfig(s.Logger).GetXDSFetchTimeout(),
-						ResourceApiVersion:  envoy_core_v3.ApiVersion_V3,
-						ConfigSourceSpecifier: &envoy_core_v3.ConfigSource_Ads{
-							Ads: &envoy_core_v3.AggregatedConfigSource{},
-						},
-					},
-				},
 				// TODO(peering): make circuit breakers or outlier detection work?
 				CircuitBreakers: &envoy_cluster_v3.CircuitBreakers{
 					Thresholds: makeThresholdsIfNeeded(upstreamConfig.Limits),
@@ -1582,7 +1604,60 @@ func (s *ResourceGenerator) makeUpstreamClustersForDiscoveryChain(
 				return nil, fmt.Errorf("cannot have more than one target")
 			}
 
-			if targetInfo := groupedTarget.Targets[0]; targetInfo.TLSContext != nil {
+			targetInfo := groupedTarget.Targets[0]
+			targetUID := proxycfg.NewUpstreamIDFromTargetID(targetInfo.TargetID)
+
+			meshGatewayMode, err := s.getMeshGatewayMode(cfgSnap, upstream, targetUID, groupedTarget.ClusterName)
+			if err != nil {
+				s.Logger.Error(err.Error(), "cluster", groupedTarget.ClusterName)
+			}
+
+			// Check if cluster need to be configured with hostnames or not.
+			useEDS := true
+			if targetUID.Peer != "" {
+				if _, ok := upstreamsSnapshot.PeerUpstreamEndpointsUseHostnames[targetUID]; ok {
+					// If we're using local mesh gw, the fact that upstreams use hostnames doesn't matter.
+					// If we're not using local mesh gw, then resort to CDS/DNS.
+					if meshGatewayMode != structs.MeshGatewayModeLocal {
+						useEDS = false
+					}
+				}
+			}
+
+			if useEDS {
+				c.ClusterDiscoveryType = &envoy_cluster_v3.Cluster_Type{Type: envoy_cluster_v3.Cluster_EDS}
+				c.EdsClusterConfig = &envoy_cluster_v3.Cluster_EdsClusterConfig{
+					EdsConfig: &envoy_core_v3.ConfigSource{
+						InitialFetchTimeout: cfgSnap.GetXDSCommonConfig(s.Logger).GetXDSFetchTimeout(),
+						ResourceApiVersion:  envoy_core_v3.ApiVersion_V3,
+						ConfigSourceSpecifier: &envoy_core_v3.ConfigSource_Ads{
+							Ads: &envoy_core_v3.AggregatedConfigSource{},
+						},
+					},
+				}
+			} else {
+				hostnameEndpoints, ok := upstreamsSnapshot.PeerUpstreamEndpoints.Get(targetUID)
+				if !ok || len(hostnameEndpoints) == 0 {
+					// The upstream snapshot should deliver hostname endpoints soon; skip this cluster until then.
+					s.Logger.Debug("peer hostname endpoints not ready for discovery chain target",
+						"target", targetInfo.TargetID,
+						"upstream", targetUID,
+						"cluster", groupedTarget.ClusterName,
+					)
+					continue
+				}
+				c.EdsClusterConfig = nil
+				configureClusterWithHostnames(
+					s.Logger,
+					c,
+					"", /*TODO: should make configurable ? */
+					hostnameEndpoints,
+					true,  /*isRemote*/
+					false, /*onlyPassing*/
+				)
+			}
+
+			if targetInfo.TLSContext != nil {
 				transportSocket, err := makeUpstreamTLSTransportSocket(targetInfo.TLSContext)
 				if err != nil {
 					return nil, err
@@ -1609,6 +1684,32 @@ func (s *ResourceGenerator) makeUpstreamClustersForDiscoveryChain(
 	return out, nil
 }
 
+func (s *ResourceGenerator) getMeshGatewayMode(
+	cfgSnap *proxycfg.ConfigSnapshot, upstream *structs.Upstream,
+	targetUID proxycfg.UpstreamID, clusterName string,
+) (structs.MeshGatewayMode, error) {
+	defaultMode := structs.MeshGatewayModeDefault
+	switch cfgSnap.Kind {
+	case structs.ServiceKindConnectProxy:
+		upstreamConfig, _ := cfgSnap.ConnectProxy.
+			GetUpstream(targetUID, &cfgSnap.ProxyID.EnterpriseMeta)
+		if upstreamConfig != nil {
+			return upstreamConfig.MeshGateway.Mode, nil
+		}
+		return defaultMode, nil
+	case structs.ServiceKindAPIGateway,
+		structs.ServiceKindIngressGateway:
+		if upstream != nil {
+			return upstream.MeshGateway.Mode, nil
+		}
+		return defaultMode, nil
+	case structs.ServiceKindMeshGateway:
+		// Mesh Gateway mesh mode will always be remote.
+		return structs.MeshGatewayModeRemote, nil
+	}
+	return structs.MeshGatewayModeDefault, fmt.Errorf("unexpected service kind %q when determining mesh gateway mode for cluster %q", cfgSnap.Kind, clusterName)
+}
+
 func (s *ResourceGenerator) makeExportedUpstreamClustersForMeshGateway(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
 	// NOTE: Despite the mesh gateway already having one cluster per service
 	// (and subset) in the local datacenter we cannot reliably use those to