Repository: grpc/grpc-go

Issue:
xds/xdsclient: race in SetWatchExpiryTimeoutForTesting

There is a race in [`SetWatchExpiryTimeoutForTesting`](https://github.com/grpc/grpc-go/blob/fa0d6583208033fe4f69d359f80286736fd121d0/internal/xds/clients/xdsclient/xdsclient.go#L121) which is used to override the watch expiry timeout of XDSClient for testing. Currently it just sets the `watchExpiryTimeout` of the XDSClient to the provided value without a mutex each time we call [`NewClientForTesting`](https://github.com/grpc/grpc-go/blob/fa0d6583208033fe4f69d359f80286736fd121d0/internal/xds/xdsclient/pool.go#L116C16-L116C35) which might of might not create a new XDSClient if one is already there.


Hints from issue comments:

Solution patch:
diff --git a/internal/envconfig/xds.go b/internal/envconfig/xds.go
index e8755155..b1f883bc 100644
--- a/internal/envconfig/xds.go
+++ b/internal/envconfig/xds.go
@@ -68,4 +68,10 @@ var (
 	// trust.  For more details, see:
 	// https://github.com/grpc/proposal/blob/master/A87-mtls-spiffe-support.md
 	XDSSPIFFEEnabled = boolFromEnv("GRPC_EXPERIMENTAL_XDS_MTLS_SPIFFE", false)
+
+	// XDSHTTPConnectEnabled is true if gRPC should parse custom Metadata
+	// configuring use of an HTTP CONNECT proxy via xDS from cluster resources.
+	// For more details, see:
+	// https://github.com/grpc/proposal/blob/master/A86-xds-http-connect.md
+	XDSHTTPConnectEnabled = boolFromEnv("GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT", false)
 )
diff --git a/internal/xds/clients/xdsclient/internal/internal.go b/internal/xds/clients/xdsclient/internal/internal.go
index 7adb6719..38d2473e 100644
--- a/internal/xds/clients/xdsclient/internal/internal.go
+++ b/internal/xds/clients/xdsclient/internal/internal.go
@@ -21,10 +21,6 @@ package internal
 import "time"
 
 var (
-	// WatchExpiryTimeout is the watch expiry timeout for xDS client. It can be
-	// overridden by tests to change the default watch expiry timeout.
-	WatchExpiryTimeout time.Duration
-
 	// StreamBackoff is the stream backoff for xDS client. It can be overridden
 	// by tests to change the default backoff strategy.
 	StreamBackoff func(int) time.Duration
diff --git a/internal/xds/clients/xdsclient/xdsclient.go b/internal/xds/clients/xdsclient/xdsclient.go
index a1949cfa..cc7d5c4e 100644
--- a/internal/xds/clients/xdsclient/xdsclient.go
+++ b/internal/xds/clients/xdsclient/xdsclient.go
@@ -61,7 +61,6 @@ var (
 )
 
 func init() {
-	xdsclientinternal.WatchExpiryTimeout = defaultWatchExpiryTimeout
 	xdsclientinternal.StreamBackoff = defaultExponentialBackoff
 	xdsclientinternal.ResourceWatchStateForTesting = resourceWatchStateForTesting
 }
@@ -108,7 +107,9 @@ func New(config Config) (*XDSClient, error) {
 	case config.Authorities == nil && config.Servers == nil:
 		return nil, errors.New("xdsclient: no servers or authorities specified")
 	}
-
+	if config.WatchExpiryTimeout == 0 {
+		config.WatchExpiryTimeout = defaultWatchExpiryTimeout
+	}
 	client, err := newClient(&config, name)
 	if err != nil {
 		return nil, err
@@ -116,12 +117,6 @@ func New(config Config) (*XDSClient, error) {
 	return client, nil
 }
 
-// SetWatchExpiryTimeoutForTesting override the default watch expiry timeout
-// with provided timeout value.
-func (c *XDSClient) SetWatchExpiryTimeoutForTesting(watchExpiryTimeout time.Duration) {
-	c.watchExpiryTimeout = watchExpiryTimeout
-}
-
 // newClient returns a new XDSClient with the given config.
 func newClient(config *Config, target string) (*XDSClient, error) {
 	ctx, cancel := context.WithCancel(context.Background())
@@ -130,7 +125,7 @@ func newClient(config *Config, target string) (*XDSClient, error) {
 		done:               syncutil.NewEvent(),
 		authorities:        make(map[string]*authority),
 		config:             config,
-		watchExpiryTimeout: xdsclientinternal.WatchExpiryTimeout,
+		watchExpiryTimeout: config.WatchExpiryTimeout,
 		backoff:            xdsclientinternal.StreamBackoff,
 		serializer:         syncutil.NewCallbackSerializer(ctx),
 		serializerClose:    cancel,
diff --git a/internal/xds/clients/xdsclient/xdsconfig.go b/internal/xds/clients/xdsclient/xdsconfig.go
index 8d3c101e..9d376e50 100644
--- a/internal/xds/clients/xdsclient/xdsconfig.go
+++ b/internal/xds/clients/xdsclient/xdsconfig.go
@@ -19,6 +19,8 @@
 package xdsclient
 
 import (
+	"time"
+
 	"google.golang.org/grpc/internal/xds/clients"
 )
 
@@ -60,6 +62,13 @@ type Config struct {
 	// MetricsReporter is used to report registered metrics. If unset, no
 	// metrics will be reported.
 	MetricsReporter clients.MetricsReporter
+
+	// WatchExpiryTimeout is the duration after which a resource watch expires
+	// if the requested resource is not received from the management server.
+	// Most users will not need to set this. If zero, a default value of 15
+	// seconds is used as specified here:
+	// envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol#knowing-when-a-requested-resource-does-not-exist
+	WatchExpiryTimeout time.Duration
 }
 
 // ServerConfig contains configuration for an xDS management server.
diff --git a/internal/xds/xdsclient/clientimpl.go b/internal/xds/xdsclient/clientimpl.go
index dab08620..b1f79799 100644
--- a/internal/xds/xdsclient/clientimpl.go
+++ b/internal/xds/xdsclient/clientimpl.go
@@ -120,8 +120,8 @@ func (mr *metricsReporter) ReportMetric(metric any) {
 	}
 }
 
-func newClientImpl(config *bootstrap.Config, metricsRecorder estats.MetricsRecorder, target string) (*clientImpl, error) {
-	gConfig, err := buildXDSClientConfig(config, metricsRecorder, target)
+func newClientImpl(config *bootstrap.Config, metricsRecorder estats.MetricsRecorder, target string, watchExpiryTimeout time.Duration) (*clientImpl, error) {
+	gConfig, err := buildXDSClientConfig(config, metricsRecorder, target, watchExpiryTimeout)
 	if err != nil {
 		return nil, err
 	}
@@ -163,7 +163,7 @@ func (c *clientImpl) decrRef() int32 {
 }
 
 // buildXDSClientConfig builds the xdsclient.Config from the bootstrap.Config.
-func buildXDSClientConfig(config *bootstrap.Config, metricsRecorder estats.MetricsRecorder, target string) (xdsclient.Config, error) {
+func buildXDSClientConfig(config *bootstrap.Config, metricsRecorder estats.MetricsRecorder, target string, watchExpiryTimeout time.Duration) (xdsclient.Config, error) {
 	grpcTransportConfigs := make(map[string]grpctransport.Config)
 	gServerCfgMap := make(map[xdsclient.ServerConfig]*bootstrap.ServerConfig)
 
@@ -218,12 +218,13 @@ func buildXDSClientConfig(config *bootstrap.Config, metricsRecorder estats.Metri
 	}
 
 	return xdsclient.Config{
-		Authorities:      gAuthorities,
-		Servers:          gServerCfgs,
-		Node:             gNode,
-		TransportBuilder: grpctransport.NewBuilder(grpcTransportConfigs),
-		ResourceTypes:    supportedResourceTypes(config, gServerCfgMap),
-		MetricsReporter:  &metricsReporter{recorder: metricsRecorder, target: target},
+		Authorities:        gAuthorities,
+		Servers:            gServerCfgs,
+		Node:               gNode,
+		TransportBuilder:   grpctransport.NewBuilder(grpcTransportConfigs),
+		ResourceTypes:      supportedResourceTypes(config, gServerCfgMap),
+		MetricsReporter:    &metricsReporter{recorder: metricsRecorder, target: target},
+		WatchExpiryTimeout: watchExpiryTimeout,
 	}, nil
 }
 
diff --git a/internal/xds/xdsclient/pool.go b/internal/xds/xdsclient/pool.go
index d1cc8476..eb0197e0 100644
--- a/internal/xds/xdsclient/pool.go
+++ b/internal/xds/xdsclient/pool.go
@@ -99,7 +99,7 @@ func NewPool(config *bootstrap.Config) *Pool {
 // expected to invoke once they are done using the client.  It is safe for the
 // caller to invoke this close function multiple times.
 func (p *Pool) NewClient(name string, metricsRecorder estats.MetricsRecorder) (XDSClient, func(), error) {
-	return p.newRefCounted(name, metricsRecorder)
+	return p.newRefCounted(name, metricsRecorder, defaultWatchExpiryTimeout)
 }
 
 // NewClientForTesting returns an xDS client configured with the provided
@@ -126,11 +126,10 @@ func (p *Pool) NewClientForTesting(opts OptionsForTesting) (XDSClient, func(), e
 	if opts.MetricsRecorder == nil {
 		opts.MetricsRecorder = istats.NewMetricsRecorderList(nil)
 	}
-	c, cancel, err := p.newRefCounted(opts.Name, opts.MetricsRecorder)
+	c, cancel, err := p.newRefCounted(opts.Name, opts.MetricsRecorder, opts.WatchExpiryTimeout)
 	if err != nil {
 		return nil, nil, err
 	}
-	c.SetWatchExpiryTimeoutForTesting(opts.WatchExpiryTimeout)
 	return c, cancel, nil
 }
 
@@ -252,7 +251,7 @@ func (p *Pool) clientRefCountedClose(name string) {
 // newRefCounted creates a new reference counted xDS client implementation for
 // name, if one does not exist already. If an xDS client for the given name
 // exists, it gets a reference to it and returns it.
-func (p *Pool) newRefCounted(name string, metricsRecorder estats.MetricsRecorder) (*clientImpl, func(), error) {
+func (p *Pool) newRefCounted(name string, metricsRecorder estats.MetricsRecorder, watchExpiryTimeout time.Duration) (*clientImpl, func(), error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
@@ -276,7 +275,7 @@ func (p *Pool) newRefCounted(name string, metricsRecorder estats.MetricsRecorder
 		return c, sync.OnceFunc(func() { p.clientRefCountedClose(name) }), nil
 	}
 
-	c, err := newClientImpl(config, metricsRecorder, name)
+	c, err := newClientImpl(config, metricsRecorder, name, watchExpiryTimeout)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/internal/xds/xdsclient/xdsresource/metadata.go b/internal/xds/xdsclient/xdsresource/metadata.go
new file mode 100644
index 00000000..b56c309d
--- /dev/null
+++ b/internal/xds/xdsclient/xdsresource/metadata.go
@@ -0,0 +1,93 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package xdsresource
+
+import (
+	"fmt"
+	"net/netip"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+func init() {
+	registerMetadataConverter("type.googleapis.com/envoy.config.core.v3.Address", proxyAddressConvertor{})
+}
+
+var (
+	// metdataRegistry is a map from proto type to metadataConverter.
+	metdataRegistry = make(map[string]metadataConverter)
+)
+
+// metadataConverter converts xds metadata entries in
+// Metadata.typed_filter_metadata into an internal form with the fields relevant
+// to gRPC.
+type metadataConverter interface {
+	// convert parses the Any proto into a concrete struct.
+	convert(*anypb.Any) (any, error)
+}
+
+// registerMetadataConverter registers the converter to the map keyed on a proto
+// type_url. Must be called at init time. Not thread safe.
+func registerMetadataConverter(protoType string, c metadataConverter) {
+	metdataRegistry[protoType] = c
+}
+
+// metadataConverterForType retrieves a converter based on key given.
+func metadataConverterForType(typeURL string) metadataConverter {
+	return metdataRegistry[typeURL]
+}
+
+// StructMetadataValue stores the values in a google.protobuf.Struct from
+// FilterMetadata.
+type StructMetadataValue struct {
+	// Data stores the parsed JSON representation of a google.protobuf.Struct.
+	Data map[string]any
+}
+
+// ProxyAddressMetadataValue holds the address parsed from the
+// envoy.config.core.v3.Address proto message, as specified in gRFC A86.
+type ProxyAddressMetadataValue struct {
+	// Address stores the proxy address configured (A86). It will be in the form
+	// of host:port. It has to be either IPv6 or IPv4.
+	Address string
+}
+
+// proxyAddressConvertor implements the metadataConverter interface to handle
+// the conversion of envoy.config.core.v3.Address protobuf messages into an
+// internal representation.
+type proxyAddressConvertor struct{}
+
+func (proxyAddressConvertor) convert(anyProto *anypb.Any) (any, error) {
+	addressProto := &v3corepb.Address{}
+	if err := anyProto.UnmarshalTo(addressProto); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal resource from Any proto: %v", err)
+	}
+	socketaddress := addressProto.GetSocketAddress()
+	if socketaddress == nil {
+		return nil, fmt.Errorf("no socket_address field in metadata")
+	}
+	if _, err := netip.ParseAddr(socketaddress.GetAddress()); err != nil {
+		return nil, fmt.Errorf("address field is not a valid IPv4 or IPv6 address: %q", socketaddress.GetAddress())
+	}
+	portvalue := socketaddress.GetPortValue()
+	if portvalue == 0 {
+		return nil, fmt.Errorf("port value not set in socket_address")
+	}
+	return ProxyAddressMetadataValue{Address: parseAddress(socketaddress)}, nil
+}
diff --git a/internal/xds/xdsclient/xdsresource/type_eds.go b/internal/xds/xdsclient/xdsresource/type_eds.go
index 42eea009..a8d568e1 100644
--- a/internal/xds/xdsclient/xdsresource/type_eds.go
+++ b/internal/xds/xdsclient/xdsresource/type_eds.go
@@ -53,6 +53,7 @@ type Endpoint struct {
 	HealthStatus EndpointHealthStatus
 	Weight       uint32
 	HashKey      string
+	Metadata     map[string]any
 }
 
 // Locality contains information of a locality.
@@ -61,6 +62,7 @@ type Locality struct {
 	ID        clients.Locality
 	Priority  uint32
 	Weight    uint32
+	Metadata  map[string]any
 }
 
 // EndpointsUpdate contains an EDS update.
diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
index 3e1e59cd..f0e73acd 100644
--- a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
+++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
@@ -108,11 +108,16 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs
 			}
 			uniqueEndpointAddrs[a] = true
 		}
+		endpointMetadata, err := validateAndConstructMetadata(lbEndpoint.GetMetadata())
+		if err != nil {
+			return nil, err
+		}
 		endpoints = append(endpoints, Endpoint{
 			HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()),
 			Addresses:    addrs,
 			Weight:       weight,
 			HashKey:      hashKey(lbEndpoint),
+			Metadata:     endpointMetadata,
 		})
 	}
 	return endpoints, nil
@@ -190,11 +195,17 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate,
 		if err != nil {
 			return EndpointsUpdate{}, err
 		}
+		localityMetadata, err := validateAndConstructMetadata(locality.GetMetadata())
+		if err != nil {
+			return EndpointsUpdate{}, err
+		}
+
 		ret.Localities = append(ret.Localities, Locality{
 			ID:        lid,
 			Endpoints: endpoints,
 			Weight:    weight,
 			Priority:  priority,
+			Metadata:  localityMetadata,
 		})
 	}
 	for i := 0; i < len(priorities); i++ {
@@ -204,3 +215,36 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate,
 	}
 	return ret, nil
 }
+
+func validateAndConstructMetadata(metadataProto *v3corepb.Metadata) (map[string]any, error) {
+	// TODO(easwars): Find a better place for the environment variable check
+	// once A83 is implemented.
+	if !envconfig.XDSHTTPConnectEnabled || metadataProto == nil {
+		return nil, nil
+	}
+	metadata := make(map[string]any)
+	// First go through TypedFilterMetadata.
+	for key, anyProto := range metadataProto.GetTypedFilterMetadata() {
+		converter := metadataConverterForType(anyProto.GetTypeUrl())
+		// Ignore types we don't have a converter for.
+		if converter == nil {
+			continue
+		}
+		val, err := converter.convert(anyProto)
+		if err != nil {
+			// If the converter fails, nack the whole resource.
+			return nil, fmt.Errorf("metadata conversion for key %q and type %q failed: %v", key, anyProto.GetTypeUrl(), err)
+		}
+		metadata[key] = val
+	}
+
+	// Process FilterMetadata for any keys not already handled.
+	for key, structProto := range metadataProto.GetFilterMetadata() {
+		// Skip keys already added from TyperFilterMetadata.
+		if metadata[key] != nil {
+			continue
+		}
+		metadata[key] = StructMetadataValue{Data: structProto.AsMap()}
+	}
+	return metadata, nil
+}



Test patch:
diff --git a/internal/xds/clients/xdsclient/test/ads_stream_watch_test.go b/internal/xds/clients/xdsclient/test/ads_stream_watch_test.go
index 2875d537..40ca8eaa 100644
--- a/internal/xds/clients/xdsclient/test/ads_stream_watch_test.go
+++ b/internal/xds/clients/xdsclient/test/ads_stream_watch_test.go
@@ -28,6 +28,7 @@ import (
 	"github.com/google/uuid"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
+	"google.golang.org/grpc/internal/xds/clients"
 	"google.golang.org/grpc/internal/xds/clients/grpctransport"
 	"google.golang.org/grpc/internal/xds/clients/internal/testutils"
 	"google.golang.org/grpc/internal/xds/clients/xdsclient"
@@ -175,9 +176,34 @@ func (s) TestADS_WatchState_TimerFires(t *testing.T) {
 	// short resource expiry timeout.
 	nodeID := uuid.New().String()
 	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
-	overrideWatchExpiryTimeout(t, defaultTestWatchExpiryTimeout)
-	client := createXDSClient(t, mgmtServer.Address, nodeID, grpctransport.NewBuilder(configs))
-
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
+		TransportBuilder: grpctransport.NewBuilder(configs),
+		ResourceTypes:    resourceTypes,
+		// Xdstp resource names used in this test do not specify an
+		// authority. These will end up looking up an entry with the
+		// empty key in the authorities map. Having an entry with an
+		// empty key and empty configuration, results in these
+		// resources also using the top-level configuration.
+		Authorities: map[string]xdsclient.Authority{
+			"": {XDSServers: []xdsclient.ServerConfig{}},
+		},
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
+	}
+
+	// Create an xDS client with the above config.
+	client, err := xdsclient.New(xdsClientConfig)
+	if err != nil {
+		t.Fatalf("Failed to create xDS client: %v", err)
+	}
+	t.Cleanup(func() { client.Close() })
 	// Create a watch for the first listener resource and verify that the timer
 	// is running and the watch state is `requested`.
 	const listenerName = "listener"
diff --git a/internal/xds/clients/xdsclient/test/authority_test.go b/internal/xds/clients/xdsclient/test/authority_test.go
index 424ae15c..9941b989 100644
--- a/internal/xds/clients/xdsclient/test/authority_test.go
+++ b/internal/xds/clients/xdsclient/test/authority_test.go
@@ -105,10 +105,10 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.Liste
 			testAuthority2: {XDSServers: []xdsclient.ServerConfig{}},
 			testAuthority3: {XDSServers: []xdsclient.ServerConfig{{ServerIdentifier: siNonDefault}}},
 		},
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
 	}
 
 	// Create an xDS client with the above config.
-	overrideWatchExpiryTimeout(t, defaultTestWatchExpiryTimeout)
 	client, err := xdsclient.New(xdsClientConfig)
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)
diff --git a/internal/xds/clients/xdsclient/test/lds_watchers_test.go b/internal/xds/clients/xdsclient/test/lds_watchers_test.go
index e4137559..ae478392 100644
--- a/internal/xds/clients/xdsclient/test/lds_watchers_test.go
+++ b/internal/xds/clients/xdsclient/test/lds_watchers_test.go
@@ -35,7 +35,6 @@ import (
 	"google.golang.org/grpc/internal/xds/clients/internal/testutils"
 	"google.golang.org/grpc/internal/xds/clients/internal/testutils/e2e"
 	"google.golang.org/grpc/internal/xds/clients/xdsclient"
-	xdsclientinternal "google.golang.org/grpc/internal/xds/clients/xdsclient/internal"
 	"google.golang.org/grpc/internal/xds/clients/xdsclient/internal/xdsresource"
 
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
@@ -115,12 +114,6 @@ func badListenerResource(t *testing.T, name string) *v3listenerpb.Listener {
 	}
 }
 
-func overrideWatchExpiryTimeout(t *testing.T, watchExpiryTimeout time.Duration) {
-	originalWatchExpiryTimeout := xdsclientinternal.WatchExpiryTimeout
-	xdsclientinternal.WatchExpiryTimeout = watchExpiryTimeout
-	t.Cleanup(func() { xdsclientinternal.WatchExpiryTimeout = originalWatchExpiryTimeout })
-}
-
 // verifyNoListenerUpdate verifies that no listener update is received on the
 // provided update channel, and returns an error if an update is received.
 //
@@ -726,11 +719,10 @@ func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 		Node:             clients.Node{ID: nodeID},
 		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
+		// Override the default watch expiry timeout.
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
 	}
 
-	// Create an xDS client with the above config and override the default
-	// watch expiry timeout.
-	overrideWatchExpiryTimeout(t, defaultTestWatchExpiryTimeout)
 	client, err := xdsclient.New(xdsClientConfig)
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)
@@ -777,11 +769,11 @@ func (s) TestLDSWatch_ValidResponseCancelsExpiryTimerBehavior(t *testing.T) {
 		Node:             clients.Node{ID: nodeID},
 		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
+		// Override the default watch expiry timeout.
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
 	}
 
-	// Create an xDS client with the above config and override the default
-	// watch expiry timeout.
-	overrideWatchExpiryTimeout(t, defaultTestWatchExpiryTimeout)
+	// Create an xDS client with the above config.
 	client, err := xdsclient.New(xdsClientConfig)
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)
diff --git a/internal/xds/xdsclient/clientimpl_test.go b/internal/xds/xdsclient/clientimpl_test.go
index 85792a3d..d2fc8f7f 100644
--- a/internal/xds/xdsclient/clientimpl_test.go
+++ b/internal/xds/xdsclient/clientimpl_test.go
@@ -208,7 +208,7 @@ func (s) TestBuildXDSClientConfig_Success(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Failed to create bootstrap config: %v", err)
 			}
-			gotCfg, err := buildXDSClientConfig(bootstrapConfig, stats.NewTestMetricsRecorder(), testTargetName)
+			gotCfg, err := buildXDSClientConfig(bootstrapConfig, stats.NewTestMetricsRecorder(), testTargetName, 0)
 			if err != nil {
 				t.Fatalf("Failed to build XDSClientConfig: %v", err)
 			}
diff --git a/internal/xds/xdsclient/tests/loadreport_test.go b/internal/xds/xdsclient/tests/loadreport_test.go
index e33f3799..95a9354d 100644
--- a/internal/xds/xdsclient/tests/loadreport_test.go
+++ b/internal/xds/xdsclient/tests/loadreport_test.go
@@ -481,8 +481,6 @@ func (s) TestConcurrentReportLoad(t *testing.T) {
 // concurrently with a shared XDSClient, each of which will create a new LRS
 // stream without any race.
 func (s) TestConcurrentChannels(t *testing.T) {
-	// TODO(emchandwani) : Unskip after https://github.com/grpc/grpc-go/pull/8526 gets merged.
-	t.Skip()
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
diff --git a/internal/xds/xdsclient/xdsresource/metadata_test.go b/internal/xds/xdsclient/xdsresource/metadata_test.go
new file mode 100644
index 00000000..28de8b51
--- /dev/null
+++ b/internal/xds/xdsclient/xdsresource/metadata_test.go
@@ -0,0 +1,199 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package xdsresource
+
+import (
+	"testing"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/grpc/internal/testutils"
+)
+
+const proxyAddressTypeURL = "type.googleapis.com/envoy.config.core.v3.Address"
+
+func (s) TestProxyAddressConverterSuccess(t *testing.T) {
+	converter := metadataConverterForType(proxyAddressTypeURL)
+	if converter == nil {
+		t.Fatalf("Converter for %q not found in registry", proxyAddressTypeURL)
+	}
+	tests := []struct {
+		name string
+		addr *v3corepb.Address
+		want ProxyAddressMetadataValue
+	}{
+		{
+			name: "valid IPv4 address and port",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "192.168.1.1",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 8080,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "192.168.1.1:8080",
+			},
+		},
+		{
+			name: "valid full IPv6 address and port",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "2001:0db8:85a3:0000:0000:8a2e:0370:7334",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 9090,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "[2001:0db8:85a3:0000:0000:8a2e:0370:7334]:9090",
+			},
+		},
+		{
+			name: "valid shortened IPv6 address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "2001:db8::1",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 9090,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "[2001:db8::1]:9090",
+			},
+		},
+		{
+			name: "valid link-local IPv6 address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "fe80::1ff:fe23:4567:890a",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 8888,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "[fe80::1ff:fe23:4567:890a]:8888",
+			},
+		},
+		{
+			name: "valid IPv4-mapped IPv6 address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "::ffff:192.0.2.128",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 1234,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "[::ffff:192.0.2.128]:1234",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			anyProto := testutils.MarshalAny(t, tt.addr)
+			got, err := converter.convert(anyProto)
+			if err != nil {
+				t.Fatalf("convert() failed with error: %v", err)
+			}
+			if diff := cmp.Diff(tt.want, got, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
+				t.Errorf("convert() returned unexpected value (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
+func (s) TestProxyAddressConverterFailure(t *testing.T) {
+	converter := metadataConverterForType(proxyAddressTypeURL)
+	if converter == nil {
+		t.Fatalf("Converter for %q not found in registry", proxyAddressTypeURL)
+	}
+	tests := []struct {
+		name    string
+		addr    *v3corepb.Address
+		wantErr string
+	}{
+		{
+			name: "invalid address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "invalid-ip",
+					},
+				},
+			},
+			wantErr: "address field is not a valid IPv4 or IPv6 address: \"invalid-ip\"",
+		},
+		{
+			name: "missing socket_address",
+			addr: &v3corepb.Address{
+				// No SocketAddress field set.
+			},
+			wantErr: "no socket_address field in metadata",
+		},
+		{
+			name: "address is not a socket address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_EnvoyInternalAddress{
+					EnvoyInternalAddress: &v3corepb.EnvoyInternalAddress{
+						AddressNameSpecifier: &v3corepb.EnvoyInternalAddress_ServerListenerName{
+							ServerListenerName: "some-internal-listener",
+						},
+					},
+				},
+			},
+			wantErr: "no socket_address field in metadata",
+		},
+		{
+			name: "port value not set",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address:       "127.0.0.1",
+						PortSpecifier: nil,
+					},
+				},
+			},
+			wantErr: "port value not set in socket_address",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			anyProto := testutils.MarshalAny(t, tt.addr)
+			_, err := converter.convert(anyProto)
+			if err == nil || err.Error() != tt.wantErr {
+				t.Errorf("convert() got error = %v, wantErr = %q", err, tt.wantErr)
+			}
+		})
+	}
+}
diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
index 4eb088dc..bad0798f 100644
--- a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
+++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
@@ -41,12 +41,124 @@ import (
 )
 
 func (s) TestEDSParseRespProto(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
 	tests := []struct {
 		name    string
 		m       *v3endpointpb.ClusterLoadAssignment
 		want    EndpointsUpdate
 		wantErr bool
 	}{
+		{
+			name: "with endpoint metadata from typed",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{
+					addrWithPort: "addr1:314",
+					metadata: &v3corepb.Metadata{
+						TypedFilterMetadata: map[string]*anypb.Any{
+							"typed.key": testutils.MarshalAny(t, &v3corepb.Address{
+								Address: &v3corepb.Address_SocketAddress{
+									SocketAddress: &v3corepb.SocketAddress{
+										Address: "1.2.3.4",
+										PortSpecifier: &v3corepb.SocketAddress_PortValue{
+											PortValue: 1111,
+										}},
+								},
+							}),
+						},
+						FilterMetadata: map[string]*structpb.Struct{
+							"some.key": {Fields: map[string]*structpb.Value{
+								"field": structpb.NewStringValue("untyped-value")}},
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			want: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+							Metadata: map[string]any{
+								"typed.key": ProxyAddressMetadataValue{
+									Address: "1.2.3.4:1111",
+								},
+								"some.key": StructMetadataValue{
+									Data: map[string]any{"field": "untyped-value"},
+								},
+							},
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+
+		{
+			name: "with endpoint metadata from filtered",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{
+					addrWithPort: "addr1:314",
+					metadata: &v3corepb.Metadata{
+						FilterMetadata: map[string]*structpb.Struct{
+							"test-key": {},
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			want: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+							Metadata: map[string]any{
+								"test-key": StructMetadataValue{Data: map[string]any{}},
+							},
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "with locality metadata from filtered",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+					Metadata: &v3corepb.Metadata{
+						FilterMetadata: map[string]*structpb.Struct{"test-key": {}},
+					},
+				})
+				return clab0.Build()
+			}(),
+			want: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+						Metadata: map[string]any{
+							"test-key": StructMetadataValue{Data: map[string]any{}},
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "missing-priority",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
@@ -214,8 +326,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			got, err := parseEDSRespProto(tt.m)
 			if (err != nil) != tt.wantErr {
-				t.Errorf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
-				return
+				t.Fatalf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
 			}
 			if d := cmp.Diff(got, tt.want, cmpopts.EquateEmpty()); d != "" {
 				t.Errorf("parseEDSRespProto() got = %v, want %v, diff: %v", got, tt.want, d)
@@ -325,8 +436,7 @@ func (s) TestEDSParseRespProtoAdditionalAddrs(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			got, err := parseEDSRespProto(tt.m)
 			if (err != nil) != tt.wantErr {
-				t.Errorf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
-				return
+				t.Fatalf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
 			}
 			if d := cmp.Diff(got, tt.want, cmpopts.EquateEmpty()); d != "" {
 				t.Errorf("parseEDSRespProto() got = %v, want %v, diff: %v", got, tt.want, d)
@@ -465,6 +575,7 @@ func (s) TestUnmarshalEndpointHashKey(t *testing.T) {
 }
 
 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{
@@ -588,6 +699,174 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 	}
 }
 
+func (s) TestValidateAndConstructMetadataWithXDSHTTPConnectEnabled(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
+	tests := []struct {
+		name          string
+		metadataProto *v3corepb.Metadata
+		want          map[string]any
+		wantErr       bool
+	}{
+		{
+			name: "typed filter metadata over filter metadata-xDSHTTPConnect-enabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"some.key": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "1.2.3.4",
+								PortSpecifier: &v3corepb.SocketAddress_PortValue{
+									PortValue: 1111,
+								},
+							},
+						},
+					}),
+				},
+				FilterMetadata: map[string]*structpb.Struct{
+					"some.key": {Fields: map[string]*structpb.Value{
+						"field": structpb.NewStringValue("untyped-value")}},
+				},
+			},
+
+			want: map[string]any{
+				"some.key": ProxyAddressMetadataValue{Address: "1.2.3.4:1111"},
+			},
+		},
+		{
+			name: "success-case-xDSHTTPConnect-enabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "1.2.3.4",
+								PortSpecifier: &v3corepb.SocketAddress_PortValue{
+									PortValue: 8080,
+								},
+							},
+						},
+					}),
+				},
+				FilterMetadata: map[string]*structpb.Struct{
+					"untyped-key": {Fields: map[string]*structpb.Value{"field": structpb.NewStringValue("value")}},
+				},
+			},
+			want: map[string]any{
+				"envoy.http11_proxy_transport_socket.proxy_address": ProxyAddressMetadataValue{Address: "1.2.3.4:8080"},
+				"untyped-key": StructMetadataValue{Data: map[string]any{"field": string("value")}},
+			},
+		},
+		{
+			name: "failure-case-converter-error-xDSHTTPConnect-enabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "invalid",
+							},
+						},
+					}),
+				},
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := validateAndConstructMetadata(tt.metadataProto)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("validateAndConstructMetadata() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if diff := cmp.Diff(tt.want, got, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
+				t.Errorf("validateAndConstructMetadata() returned unexpected diff (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
+func (s) TestValidateAndConstructMetadataWithXDSHTTPConnectDisabled(t *testing.T) {
+	tests := []struct {
+		name          string
+		metadataProto *v3corepb.Metadata
+		want          map[string]any
+		wantErr       bool
+	}{
+		{
+			name: "typed-filter-metadata-over-filter-metadata-disabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"some.key": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "1.2.3.4",
+								PortSpecifier: &v3corepb.SocketAddress_PortValue{
+									PortValue: 1111,
+								},
+							},
+						},
+					}),
+				},
+				FilterMetadata: map[string]*structpb.Struct{
+					"some.key": {Fields: map[string]*structpb.Value{
+						"field": structpb.NewStringValue("untyped-value")}},
+				},
+			},
+			want: nil,
+		},
+		{
+			name: "success-case-metadata-disabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "1.2.3.4",
+								PortSpecifier: &v3corepb.SocketAddress_PortValue{
+									PortValue: 8080,
+								},
+							},
+						},
+					}),
+				},
+				FilterMetadata: map[string]*structpb.Struct{
+					"untyped-key": {Fields: map[string]*structpb.Value{
+						"field": structpb.NewStringValue("value")}},
+				},
+			},
+			want: nil,
+		},
+		{
+			name: "failure-case-converter-error-metadata-disabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "invalid",
+							},
+						},
+					}),
+				},
+			},
+			want: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := validateAndConstructMetadata(tt.metadataProto)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("validateAndConstructMetadata() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if diff := cmp.Diff(tt.want, got, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
+				t.Errorf("validateAndConstructMetadata() returned unexpected diff (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
 // claBuilder builds a ClusterLoadAssignment, aka EDS
 // response.
 type claBuilder struct {
@@ -619,13 +898,15 @@ func newClaBuilder(clusterName string, dropPercents []uint32) *claBuilder {
 
 // addLocalityOptions contains options when adding locality to the builder.
 type addLocalityOptions struct {
-	Health []v3corepb.HealthStatus
-	Weight []uint32
+	Health   []v3corepb.HealthStatus
+	Weight   []uint32
+	Metadata *v3corepb.Metadata
 }
 
 type endpointOpts struct {
 	addrWithPort            string
 	additionalAddrWithPorts []string
+	metadata                *v3corepb.Metadata
 }
 
 func addressFromStr(addrWithPort string) *v3corepb.Address {
@@ -666,6 +947,7 @@ func (clab *claBuilder) addLocality(subzone string, weight uint32, priority uint
 					AdditionalAddresses: additionalAddrs,
 				},
 			},
+			Metadata: e.metadata,
 		}
 		if opts != nil {
 			if i < len(opts.Health) {
@@ -687,12 +969,16 @@ func (clab *claBuilder) addLocality(subzone string, weight uint32, priority uint
 		}
 	}
 
-	clab.v.Endpoints = append(clab.v.Endpoints, &v3endpointpb.LocalityLbEndpoints{
+	locality := &v3endpointpb.LocalityLbEndpoints{
 		Locality:            localityID,
 		LbEndpoints:         lbEndPoints,
 		LoadBalancingWeight: &wrapperspb.UInt32Value{Value: weight},
 		Priority:            priority,
-	})
+	}
+	if opts != nil {
+		locality.Metadata = opts.Metadata
+	}
+	clab.v.Endpoints = append(clab.v.Endpoints, locality)
 }
 
 // Build builds ClusterLoadAssignment.