xdsclient/xdsresource: add AutoHostRewrite and Endpoint Hostname support

internal/envconfig/envconfig.go

@@ -77,6 +77,11 @@ var (
 	//   - Target resolution is disabled.
 	//   - The DNS resolver is being used.
 	EnableDefaultPortForProxyTarget = boolFromEnv("GRPC_EXPERIMENTAL_ENABLE_DEFAULT_PORT_FOR_PROXY_TARGET", true)
+
+	// XDSAuthorityRewrite is set if xDS authority rewriting is enabled,
+	// according to gRFC A81. It can be enabled by setting
+	// GRPC_EXPERIMENTAL_XDS_AUTHORITY_REWRITE to true.
+	XDSAuthorityRewrite = boolFromEnv("GRPC_EXPERIMENTAL_XDS_AUTHORITY_REWRITE", false)
 )
 
 func boolFromEnv(envVar string, def bool) bool {

internal/xds/xdsclient/xdsresource/filter_chain.go

@@ -680,7 +680,7 @@ func processNetworkFilters(filters []*v3listenerpb.Filter) (*FilterChain, error)
 					// server-side." - A36
 					// Can specify v3 here, as will never get to this function
 					// if v2.
-					routeU, err := generateRDSUpdateFromRouteConfiguration(hcm.GetRouteConfig())
+					routeU, err := generateRDSUpdateFromRouteConfiguration(nil, hcm.GetRouteConfig())
 					if err != nil {
 						return nil, fmt.Errorf("failed to parse inline RDS resp: %v", err)
 					}

internal/xds/xdsclient/xdsresource/route_config_resource_type.go

@@ -37,8 +37,8 @@ type routeConfigResourceDecoder struct {
 	bootstrapConfig *bootstrap.Config
 }
 
-func (d *routeConfigResourceDecoder) Decode(resource *xdsclient.AnyProto, _ xdsclient.DecodeOptions) (*xdsclient.DecodeResult, error) {
-	name, rc, err := unmarshalRouteConfigResource(resource.ToAny())
+func (d *routeConfigResourceDecoder) Decode(resource *xdsclient.AnyProto, opts xdsclient.DecodeOptions) (*xdsclient.DecodeResult, error) {
+	name, rc, err := unmarshalRouteConfigResource(&opts, resource.ToAny())
 	if name == "" {
 		// Name is unset only when protobuf deserialization fails.
 		return nil, err

internal/xds/xdsclient/xdsresource/type_eds.go

@@ -54,6 +54,7 @@ type Endpoint struct {
 	Weight       uint32
 	HashKey      string
 	Metadata     map[string]any
+	Hostname     string
 }
 
 // Locality contains information of a locality.

internal/xds/xdsclient/xdsresource/type_rds.go

@@ -150,6 +150,9 @@ type Route struct {
 	// ClusterSpecifierPlugin is the name of the Cluster Specifier Plugin that
 	// this Route is linked to, if specified by xDS.
 	ClusterSpecifierPlugin string
+	// AutoHostRewrite indicates that the ":authority" header can be rewritten
+	// to the hostname of the upstream endpoint.
+	AutoHostRewrite bool
 }
 
 // WeightedCluster contains settings for an xds ActionType.WeightedCluster.

internal/xds/xdsclient/xdsresource/unmarshal_eds.go

@@ -132,6 +132,7 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs
 			Weight:       weight,
 			HashKey:      hashKey,
 			Metadata:     endpointMetadata,
+			Hostname:     lbEndpoint.GetEndpoint().GetHostname(),
 		})
 	}
 	return endpoints, nil

internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go

@@ -130,11 +130,11 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "good",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_UNHEALTHY},
 					Weight: []uint32{271},
 				})
-				clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr2:159"}}, &addLocalityOptions{
+				clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr2:159", hostname: "addr2"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_DRAINING},
 					Weight: []uint32{828},
 				})
@@ -148,6 +148,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 							Addresses:    []string{"addr1:314"},
 							HealthStatus: EndpointHealthStatusUnhealthy,
 							Weight:       271,
+							Hostname:     "addr1",
 						}},
 						ID:       clients.Locality{SubZone: "locality-1"},
 						Priority: 1,
@@ -158,6 +159,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 							Addresses:    []string{"addr2:159"},
 							HealthStatus: EndpointHealthStatusDraining,
 							Weight:       828,
+							Hostname:     "addr2",
 						}},
 						ID:       clients.Locality{SubZone: "locality-2"},
 						Priority: 0,
@@ -171,12 +173,12 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			name: "good duplicate locality with different priority",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
 				clab0 := newClaBuilder("test", nil)
-				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+				clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_UNHEALTHY},
 					Weight: []uint32{271},
 				})
 				// Same locality name, but with different priority.
-				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr2:159"}}, &addLocalityOptions{
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr2:159", hostname: "addr2"}}, &addLocalityOptions{
 					Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_DRAINING},
 					Weight: []uint32{828},
 				})
@@ -190,6 +192,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 							Addresses:    []string{"addr1:314"},
 							HealthStatus: EndpointHealthStatusUnhealthy,
 							Weight:       271,
+							Hostname:     "addr1",
 						}},
 						ID:       clients.Locality{SubZone: "locality-1"},
 						Priority: 1,
@@ -200,6 +203,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 							Addresses:    []string{"addr2:159"},
 							HealthStatus: EndpointHealthStatusDraining,
 							Weight:       828,
+							Hostname:     "addr2",
 						}},
 						ID:       clients.Locality{SubZone: "locality-1"},
 						Priority: 0,
@@ -466,11 +470,11 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
 	var v3EndpointsAny = testutils.MarshalAny(t, func() *v3endpointpb.ClusterLoadAssignment {
 		clab0 := newClaBuilder("test", nil)
-		clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+		clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}}, &addLocalityOptions{
 			Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_UNHEALTHY},
 			Weight: []uint32{271},
 		})
-		clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr2:159"}}, &addLocalityOptions{
+		clab0.addLocality("locality-2", 1, 0, []endpointOpts{{addrWithPort: "addr2:159", hostname: "addr2"}}, &addLocalityOptions{
 			Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_DRAINING},
 			Weight: []uint32{828},
 		})
@@ -520,6 +524,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 							Addresses:    []string{"addr1:314"},
 							HealthStatus: EndpointHealthStatusUnhealthy,
 							Weight:       271,
+							Hostname:     "addr1",
 						}},
 						ID:       clients.Locality{SubZone: "locality-1"},
 						Priority: 1,
@@ -530,6 +535,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 							Addresses:    []string{"addr2:159"},
 							HealthStatus: EndpointHealthStatusDraining,
 							Weight:       828,
+							Hostname:     "addr2",
 						}},
 						ID:       clients.Locality{SubZone: "locality-2"},
 						Priority: 0,
@@ -551,6 +557,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 							Addresses:    []string{"addr1:314"},
 							HealthStatus: EndpointHealthStatusUnhealthy,
 							Weight:       271,
+							Hostname:     "addr1",
 						}},
 						ID:       clients.Locality{SubZone: "locality-1"},
 						Priority: 1,
@@ -561,6 +568,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 							Addresses:    []string{"addr2:159"},
 							HealthStatus: EndpointHealthStatusDraining,
 							Weight:       828,
+							Hostname:     "addr2",
 						}},
 						ID:       clients.Locality{SubZone: "locality-2"},
 						Priority: 0,
@@ -1382,6 +1390,7 @@ type endpointOpts struct {
 	addrWithPort            string
 	additionalAddrWithPorts []string
 	metadata                *v3corepb.Metadata
+	hostname                string
 }
 
 func addressFromStr(addrWithPort string) *v3corepb.Address {
@@ -1420,6 +1429,7 @@ func (clab *claBuilder) addLocality(subzone string, weight uint32, priority uint
 				Endpoint: &v3endpointpb.Endpoint{
 					Address:             addressFromStr(e.addrWithPort),
 					AdditionalAddresses: additionalAddrs,
+					Hostname:            e.hostname,
 				},
 			},
 			Metadata: e.metadata,

internal/xds/xdsclient/xdsresource/unmarshal_lds.go

@@ -95,7 +95,7 @@ func processClientSideListener(lis *v3listenerpb.Listener) (*ListenerUpdate, err
 		}
 		update.RouteConfigName = name
 	case *v3httppb.HttpConnectionManager_RouteConfig:
-		routeU, err := generateRDSUpdateFromRouteConfiguration(apiLis.GetRouteConfig())
+		routeU, err := generateRDSUpdateFromRouteConfiguration(nil, apiLis.GetRouteConfig())
 		if err != nil {
 			return nil, fmt.Errorf("failed to parse inline RDS resp: %v", err)
 		}

internal/xds/xdsclient/xdsresource/unmarshal_rds.go

@@ -25,6 +25,8 @@ import (
 	"time"
 
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/internal/envconfig"
+	"google.golang.org/grpc/internal/xds/clients/xdsclient"
 	"google.golang.org/grpc/internal/xds/clusterspecifier"
 	"google.golang.org/grpc/internal/xds/matcher"
 	"google.golang.org/protobuf/proto"
@@ -34,7 +36,7 @@ import (
 	v3typepb "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 )
 
-func unmarshalRouteConfigResource(r *anypb.Any) (string, RouteConfigUpdate, error) {
+func unmarshalRouteConfigResource(opts *xdsclient.DecodeOptions, r *anypb.Any) (string, RouteConfigUpdate, error) {
 	r, err := UnwrapResource(r)
 	if err != nil {
 		return "", RouteConfigUpdate{}, fmt.Errorf("failed to unwrap resource: %v", err)
@@ -48,7 +50,7 @@ func unmarshalRouteConfigResource(r *anypb.Any) (string, RouteConfigUpdate, erro
 		return "", RouteConfigUpdate{}, fmt.Errorf("failed to unmarshal resource: %v", err)
 	}
 
-	u, err := generateRDSUpdateFromRouteConfiguration(rc)
+	u, err := generateRDSUpdateFromRouteConfiguration(opts, rc)
 	if err != nil {
 		return rc.GetName(), RouteConfigUpdate{}, err
 	}
@@ -67,12 +69,12 @@ func unmarshalRouteConfigResource(r *anypb.Any) (string, RouteConfigUpdate, erro
 // The RouteConfiguration includes a list of virtualHosts, which may have zero
 // or more elements. We are interested in the element whose domains field
 // matches the server name specified in the "xds:" URI. The only field in the
-// VirtualHost proto that the we are interested in is the list of routes. We
+// VirtualHost proto that we are interested in is the list of routes. We
 // only look at the last route in the list (the default route), whose match
-// field must be empty and whose route field must be set.  Inside that route
+// field must be empty and whose route field must be set. Inside that route
 // message, the cluster field will contain the clusterName or weighted clusters
 // we are looking for.
-func generateRDSUpdateFromRouteConfiguration(rc *v3routepb.RouteConfiguration) (RouteConfigUpdate, error) {
+func generateRDSUpdateFromRouteConfiguration(opts *xdsclient.DecodeOptions, rc *v3routepb.RouteConfiguration) (RouteConfigUpdate, error) {
 	vhs := make([]*VirtualHost, 0, len(rc.GetVirtualHosts()))
 	csps, err := processClusterSpecifierPlugins(rc.ClusterSpecifierPlugins)
 	if err != nil {
@@ -83,7 +85,7 @@ func generateRDSUpdateFromRouteConfiguration(rc *v3routepb.RouteConfiguration) (
 	// ignored and not emitted by the xdsclient.
 	var cspNames = make(map[string]bool)
 	for _, vh := range rc.GetVirtualHosts() {
-		routes, cspNs, err := routesProtoToSlice(vh.Routes, csps)
+		routes, cspNs, err := routesProtoToSlice(opts, vh.Routes, csps)
 		if err != nil {
 			return RouteConfigUpdate{}, fmt.Errorf("received route is invalid: %v", err)
 		}
@@ -206,7 +208,7 @@ func generateRetryConfig(rp *v3routepb.RetryPolicy) (*RetryConfig, error) {
 	return cfg, nil
 }
 
-func routesProtoToSlice(routes []*v3routepb.Route, csps map[string]clusterspecifier.BalancerConfig) ([]*Route, map[string]bool, error) {
+func routesProtoToSlice(opts *xdsclient.DecodeOptions, routes []*v3routepb.Route, csps map[string]clusterspecifier.BalancerConfig) ([]*Route, map[string]bool, error) {
 	var routesRet []*Route
 	var cspNames = make(map[string]bool)
 	for _, r := range routes {
@@ -302,6 +304,12 @@ func routesProtoToSlice(routes []*v3routepb.Route, csps map[string]clusterspecif
 		case *v3routepb.Route_Route:
 			action := r.GetRoute()
 
+			if envconfig.XDSAuthorityRewrite {
+				if opts != nil && opts.ServerConfig != nil && opts.ServerConfig.SupportsServerFeature(xdsclient.ServerFeatureTrustedXDSServer) {
+					route.AutoHostRewrite = action.GetAutoHostRewrite().GetValue()
+				}
+			}
+
 			// Hash Policies are only applicable for a Ring Hash LB.
 			hp, err := hashPoliciesProtoToSlice(action.HashPolicy)
 			if err != nil {

internal/xds/xdsclient/xdsresource/unmarshal_rds_test.go

@@ -29,8 +29,10 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/pretty"
 	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/grpc/internal/xds/clients/xdsclient"
 	"google.golang.org/grpc/internal/xds/clusterspecifier"
 	"google.golang.org/grpc/internal/xds/httpfilter"
 	"google.golang.org/grpc/internal/xds/matcher"
@@ -710,7 +712,7 @@ func (s) TestRDSGenerateRDSUpdateFromRouteConfiguration(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			gotUpdate, gotError := generateRDSUpdateFromRouteConfiguration(test.rc)
+			gotUpdate, gotError := generateRDSUpdateFromRouteConfiguration(nil, test.rc)
 			if (gotError != nil) != test.wantError ||
 				!cmp.Equal(gotUpdate, test.wantUpdate, cmpopts.EquateEmpty(),
 					cmp.Transformer("FilterConfig", func(fc httpfilter.FilterConfig) string {
@@ -722,6 +724,95 @@ func (s) TestRDSGenerateRDSUpdateFromRouteConfiguration(t *testing.T) {
 	}
 }
 
+func (s) TestGenerateRDSUpdateFromRouteConfigurationWithAutoHostRewrite(t *testing.T) {
+	const (
+		clusterName = "clusterName"
+		ldsTarget   = "lds.target.good:1111"
+	)
+
+	buildRouteConfig := func() *v3routepb.RouteConfiguration {
+		return &v3routepb.RouteConfiguration{
+			Name: "routeName",
+			VirtualHosts: []*v3routepb.VirtualHost{{
+				Domains: []string{ldsTarget},
+				Routes: []*v3routepb.Route{{
+					Match: &v3routepb.RouteMatch{PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: "/"}},
+					Action: &v3routepb.Route_Route{
+						Route: &v3routepb.RouteAction{
+							ClusterSpecifier:     &v3routepb.RouteAction_Cluster{Cluster: clusterName},
+							HostRewriteSpecifier: &v3routepb.RouteAction_AutoHostRewrite{AutoHostRewrite: &wrapperspb.BoolValue{Value: true}},
+						},
+					},
+				}},
+			}},
+		}
+	}
+
+	tests := []struct {
+		name             string
+		isTrusted        bool // Corresponds to ServerConfig
+		envConfigRewrite bool // Corresponds to envconfig.XDSAuthorityRewrite
+		wantResult       bool
+	}{
+		{
+			name:             "envConfigOn_Trusted",
+			isTrusted:        true,
+			envConfigRewrite: true,
+			wantResult:       true,
+		},
+		{
+			name:             "envConfigOff_Trusted",
+			isTrusted:        true,
+			envConfigRewrite: false,
+			wantResult:       false,
+		},
+		{
+			name:             "envConfigOn_Untrusted",
+			isTrusted:        false,
+			envConfigRewrite: true,
+			wantResult:       false,
+		},
+		{
+			name:             "envConfigOff_Untrusted",
+			isTrusted:        false,
+			envConfigRewrite: false,
+			wantResult:       false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			testutils.SetEnvConfig(t, &envconfig.XDSAuthorityRewrite, test.envConfigRewrite)
+
+			var serverFeature xdsclient.ServerFeature
+			if test.isTrusted {
+				serverFeature = xdsclient.ServerFeatureTrustedXDSServer
+			}
+
+			opts := &xdsclient.DecodeOptions{
+				ServerConfig: &xdsclient.ServerConfig{
+					ServerFeature: serverFeature,
+				},
+			}
+
+			update, err := generateRDSUpdateFromRouteConfiguration(opts, buildRouteConfig())
+			if err != nil {
+				t.Errorf("generateRDSUpdateFromRouteConfiguration() failed, got : %v, want: <nil>", err)
+			}
+
+			if len(update.VirtualHosts) == 0 || len(update.VirtualHosts[0].Routes) == 0 {
+				t.Errorf("Unexpected parsed routes from generateRDSUpdateFromRouteConfiguration(), got : 0, want: 1")
+			}
+
+			got := update.VirtualHosts[0].Routes[0].AutoHostRewrite
+
+			if got != test.wantResult {
+				t.Errorf("AutoHostRewrite = %v, want %v", got, test.wantResult)
+			}
+		})
+	}
+}
+
 var configOfClusterSpecifierDoesntExist = &anypb.Any{
 	TypeUrl: "does.not.exist",
 	Value:   []byte{1, 2, 3},
@@ -874,7 +965,7 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			name, update, err := unmarshalRouteConfigResource(test.resource)
+			name, update, err := unmarshalRouteConfigResource(nil, test.resource)
 			if (err != nil) != test.wantErr {
 				t.Errorf("unmarshalRouteConfigResource(%s), got err: %v, wantErr: %v", pretty.ToJSON(test.resource), err, test.wantErr)
 			}
@@ -1505,7 +1596,7 @@ func (s) TestRoutesProtoToSlice(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, _, err := routesProtoToSlice(tt.routes, nil)
+			got, _, err := routesProtoToSlice(nil, tt.routes, nil)
 			if (err != nil) != tt.wantErr {
 				t.Fatalf("routesProtoToSlice() error = %v, wantErr %v", err, tt.wantErr)
 			}