[NET-5075] Implement mesh gateway mode for explicit destinations (#20361)

internal/mesh/internal/controllers/sidecarproxy/controller.go

@@ -200,7 +200,7 @@ func (r *reconciler) Reconcile(ctx context.Context, rt controller.Runtime, req c
 		BuildLocalApp(workloadDataWithInheritedPorts, ctp)
 
 	// Get all destinationsData.
-	destinationsData, err := dataFetcher.FetchComputedExplicitDestinationsData(ctx, req.ID)
+	destinationsData, err := dataFetcher.FetchComputedExplicitDestinationsData(ctx, req.ID, proxyCfg.GetData())
 	if err != nil {
 		rt.Logger.Error("error fetching explicit destinations for this proxy", "error", err)
 		return err

internal/mesh/internal/controllers/sidecarproxy/fetcher/data_fetcher.go

@@ -107,6 +107,7 @@ func (f *Fetcher) FetchComputedRoutes(ctx context.Context, id *pbresource.ID) (*
 func (f *Fetcher) FetchComputedExplicitDestinationsData(
 	ctx context.Context,
 	proxyID *pbresource.ID,
+	proxyCfg *pbmesh.ComputedProxyConfiguration,
 ) ([]*intermediateTypes.Destination, error) {
 	var destinations []*intermediateTypes.Destination
 
@@ -187,7 +188,38 @@ func (f *Fetcher) FetchComputedExplicitDestinationsData(
 			targetServiceID := resource.IDFromReference(routeTarget.BackendRef.Ref)
 
 			// Fetch ServiceEndpoints.
-			se, err := f.FetchServiceEndpoints(ctx, resource.ReplaceType(pbcatalog.ServiceEndpointsType, targetServiceID))
+			serviceEndpointsID := resource.ReplaceType(pbcatalog.ServiceEndpointsType, targetServiceID)
+
+			// If the target service is in a different partition and the mesh gateway mode is
+			// "local" or "remote", use the ServiceEndpoints for the corresponding MeshGateway
+			// instead of the ServiceEndpoints for the target service.
+			//
+			// TODO(nathancoleman) Consider cross-datacenter case as well
+			if routeTarget.BackendRef.Ref.Tenancy.Partition != proxyID.Tenancy.Partition {
+				mode := pbmesh.MeshGatewayMode_MESH_GATEWAY_MODE_NONE
+				if proxyCfg != nil && proxyCfg.DynamicConfig != nil {
+					mode = proxyCfg.GetDynamicConfig().GetMeshGatewayMode()
+				}
+
+				switch mode {
+				case pbmesh.MeshGatewayMode_MESH_GATEWAY_MODE_LOCAL:
+					// Use ServiceEndpoints for the MeshGateway in the source service's partition
+					serviceEndpointsID = &pbresource.ID{
+						Type:    pbcatalog.ServiceEndpointsType,
+						Name:    "mesh-gateway",
+						Tenancy: proxyID.Tenancy,
+					}
+				case pbmesh.MeshGatewayMode_MESH_GATEWAY_MODE_REMOTE:
+					// Use ServiceEndpoints for the MeshGateway in the target service's partition
+					serviceEndpointsID = &pbresource.ID{
+						Type:    pbcatalog.ServiceEndpointsType,
+						Name:    "mesh-gateway",
+						Tenancy: targetServiceID.Tenancy,
+					}
+				}
+			}
+
+			se, err := f.FetchServiceEndpoints(ctx, serviceEndpointsID)
 			if err != nil {
 				return nil, err
 			}

internal/mesh/internal/controllers/sidecarproxy/fetcher/data_fetcher_test.go

@@ -44,6 +44,7 @@ type dataFetcherSuite struct {
 	api1ServiceEndpointsData    *pbcatalog.ServiceEndpoints
 	api2ServiceEndpoints        *pbresource.Resource
 	api2ServiceEndpointsData    *pbcatalog.ServiceEndpoints
+	proxyCfg                    *pbmesh.ComputedProxyConfiguration
 	webComputedDestinationsData *pbmesh.ComputedExplicitDestinations
 	webProxy                    *pbresource.Resource
 	webWorkload                 *pbresource.Resource
@@ -123,6 +124,12 @@ func (suite *dataFetcherSuite) setupWithTenancy(tenancy *pbresource.Tenancy) {
 		WithData(suite.T(), suite.api2ServiceEndpointsData).
 		Write(suite.T(), suite.client)
 
+	suite.proxyCfg = &pbmesh.ComputedProxyConfiguration{
+		DynamicConfig: &pbmesh.DynamicConfig{
+			MeshGatewayMode: pbmesh.MeshGatewayMode_MESH_GATEWAY_MODE_NONE,
+		},
+	}
+
 	suite.webComputedDestinationsData = &pbmesh.ComputedExplicitDestinations{
 		Destinations: []*pbmesh.Destination{
 			{
@@ -250,7 +257,7 @@ func (suite *dataFetcherSuite) TestFetcher_FetchExplicitDestinationsData() {
 			c.TrackComputedDestinations(resourcetest.MustDecode[*pbmesh.ComputedExplicitDestinations](t, compDest))
 
 			// We will try to fetch explicit destinations for a proxy that doesn't have one.
-			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id)
+			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id, suite.proxyCfg)
 			require.NoError(t, err)
 			require.Nil(t, destinations)
 
@@ -275,7 +282,7 @@ func (suite *dataFetcherSuite) TestFetcher_FetchExplicitDestinationsData() {
 				WithTenancy(tenancy).
 				Write(t, suite.client)
 
-			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id)
+			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id, suite.proxyCfg)
 			require.NoError(t, err)
 			require.Nil(t, destinations)
 			cachedCompDestIDs := c.ComputedDestinationsByService(resource.IDFromReference(notFoundServiceRef))
@@ -305,7 +312,7 @@ func (suite *dataFetcherSuite) TestFetcher_FetchExplicitDestinationsData() {
 				WithTenancy(tenancy).
 				Write(t, suite.client)
 
-			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id)
+			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id, suite.proxyCfg)
 			require.NoError(t, err)
 			require.Nil(t, destinations)
 			cachedCompDestIDs := c.ComputedDestinationsByService(resource.IDFromReference(api1ServiceRef))
@@ -335,7 +342,7 @@ func (suite *dataFetcherSuite) TestFetcher_FetchExplicitDestinationsData() {
 				WithTenancy(tenancy).
 				Write(t, suite.client)
 
-			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id)
+			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id, suite.proxyCfg)
 			require.NoError(t, err)
 			require.Nil(t, destinations)
 			cachedCompDestIDs := c.ComputedDestinationsByService(resource.IDFromReference(api1ServiceRef))
@@ -367,7 +374,7 @@ func (suite *dataFetcherSuite) TestFetcher_FetchExplicitDestinationsData() {
 				WithTenancy(tenancy).
 				Write(t, suite.client)
 
-			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id)
+			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id, suite.proxyCfg)
 			require.NoError(t, err)
 			require.Empty(t, destinations)
 
@@ -400,7 +407,7 @@ func (suite *dataFetcherSuite) TestFetcher_FetchExplicitDestinationsData() {
 			require.NotNil(suite.T(), api1ComputedRoutes)
 
 			// This destination points to TCP, but the computed routes is stale and only knows about HTTP.
-			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id)
+			destinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id, suite.proxyCfg)
 			require.NoError(t, err)
 
 			// Check that we didn't return any destinations.
@@ -481,7 +488,7 @@ func (suite *dataFetcherSuite) TestFetcher_FetchExplicitDestinationsData() {
 				},
 			}
 
-			actualDestinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id)
+			actualDestinations, err := f.FetchComputedExplicitDestinationsData(suite.ctx, suite.webProxy.Id, suite.proxyCfg)
 			require.NoError(t, err)
 
 			// Check that we've computed expanded destinations correctly.

internal/mesh/internal/types/proxy_configuration.go

@@ -100,13 +100,6 @@ func validateDynamicProxyConfiguration(cfg *pbmesh.DynamicConfig) error {
 		})
 	}
 
-	if cfg.GetMeshGatewayMode() != pbmesh.MeshGatewayMode_MESH_GATEWAY_MODE_UNSPECIFIED {
-		err = multierror.Append(err, resource.ErrInvalidField{
-			Name:    "mesh_gateway_mode",
-			Wrapped: resource.ErrUnsupported,
-		})
-	}
-
 	if cfg.GetAccessLogs() != nil {
 		err = multierror.Append(err, resource.ErrInvalidField{
 			Name:    "access_logs",

internal/mesh/internal/types/proxy_configuration_test.go

@@ -130,7 +130,6 @@ func TestValidateProxyConfiguration_AllFieldsInvalid(t *testing.T) {
 		DynamicConfig: &pbmesh.DynamicConfig{
 			// Set unsupported fields.
 			MutualTlsMode:           pbmesh.MutualTLSMode_MUTUAL_TLS_MODE_PERMISSIVE,
-			MeshGatewayMode:         pbmesh.MeshGatewayMode_MESH_GATEWAY_MODE_LOCAL,
 			AccessLogs:              &pbmesh.AccessLogsConfig{},
 			PublicListenerJson:      "listener-json",
 			ListenerTracingJson:     "tracing-json",
@@ -167,7 +166,6 @@ func TestValidateProxyConfiguration_AllFieldsInvalid(t *testing.T) {
 	var dynamicCfgErr error
 	unsupportedFields := []string{
 		"mutual_tls_mode",
-		"mesh_gateway_mode",
 		"access_logs",
 		"public_listener_json",
 		"listener_tracing_json",
@@ -246,7 +244,7 @@ func TestValidateProxyConfiguration_AllFieldsValid(t *testing.T) {
 
 		DynamicConfig: &pbmesh.DynamicConfig{
 			MutualTlsMode:   pbmesh.MutualTLSMode_MUTUAL_TLS_MODE_DEFAULT,
-			MeshGatewayMode: pbmesh.MeshGatewayMode_MESH_GATEWAY_MODE_UNSPECIFIED,
+			MeshGatewayMode: pbmesh.MeshGatewayMode_MESH_GATEWAY_MODE_LOCAL,
 
 			TransparentProxy: &pbmesh.TransparentProxy{
 				DialedDirectly:       false,