Repository: grpc/grpc-go

Issue:
grpc: add a call option to override the `:authority` header on a per-RPC basis
This is one of the items mentioned in https://github.com/grpc/grpc-go/issues/4717 which did not get done. The below description has been copied over verbatim from https://github.com/grpc/grpc-go/issues/4717. This is low priority and we want to implement this only if there is sufficient interest for it.

- [x] Add a call option to override the `:authority` header on a per-RPC basis. #3444
  - [x] An optional interface will be added, to be implemented by `AuthInfo` implementations, to validate this override.
    - Implementations will need to perform a hostname validation check on the peer certificate, received during the handshake, to validate this override.
    - RPCs will fail if this call option is set, but:
      - `AuthInfo` for the subChannel does not implement this interface
      - validation by `AuthInfo` fails
  - [x] Existing TLS based credentials implementations will be enhanced to support this.
  - [ ] Existing xDS credentials implementation will be enhanced to support this.
     - Use list of SANs provided by management server to validate.
  - [x] Insecure credentials will allow for any authority value.
  - [x] Local credentials will allow for any authority value.


Discussion:
@110y: @easwars 

I created a PR that adds a new CallOption for setting `:authority` per RPC: https://github.com/grpc/grpc-go/pull/5787
Could you please check this PR...?@JaydenTeoh: @easwars @ginayeh Happy to work on this issue if this is still of priority. If so, can I have some pointers on where to start next? I would also appreciate some clarity on what needs to be done. Thank you!@arvindbr8: Thanks for your interest @JaydenTeoh. 
IMO, this issue requires some more clarification and might be easier for someone internal to pick this up. Sorry for the mislabel. 

If you are still interested, please look for other issues marked "Status: Help Wanted" or "Hacktoberfest".@eshitachandwani: The subpoint 3 , that is : 
- Existing xDS credentials implementation will be enhanced to support this.
	- Use list of SANs provided by management server to validate.
will be done as part of [gRFC A81](https://github.com/grpc/proposal/blob/master/A81-xds-authority-rewriting.md)

Solution patch:
diff --git a/balancer/weightedroundrobin/balancer.go b/balancer/weightedroundrobin/balancer.go
index 0ee70760..e6fe5a37 100644
--- a/balancer/weightedroundrobin/balancer.go
+++ b/balancer/weightedroundrobin/balancer.go
@@ -497,7 +497,6 @@ func (p *picker) start(stopPicker *grpcsync.Event) {
 // that listener.
 type endpointWeight struct {
 	// The following fields are immutable.
-	balancer.SubConn
 	logger          *grpclog.PrefixLogger
 	target          string
 	metricsRecorder estats.MetricsRecorder
@@ -527,7 +526,7 @@ type endpointWeight struct {
 
 func (w *endpointWeight) OnLoadReport(load *v3orcapb.OrcaLoadReport) {
 	if w.logger.V(2) {
-		w.logger.Infof("Received load report for subchannel %v: %v", w.SubConn, load)
+		w.logger.Infof("Received load report for subchannel %v: %v", w.pickedSC, load)
 	}
 	// Update weights of this endpoint according to the reported load.
 	utilization := load.ApplicationUtilization
@@ -536,7 +535,7 @@ func (w *endpointWeight) OnLoadReport(load *v3orcapb.OrcaLoadReport) {
 	}
 	if utilization == 0 || load.RpsFractional == 0 {
 		if w.logger.V(2) {
-			w.logger.Infof("Ignoring empty load report for subchannel %v", w.SubConn)
+			w.logger.Infof("Ignoring empty load report for subchannel %v", w.pickedSC)
 		}
 		return
 	}
@@ -547,7 +546,7 @@ func (w *endpointWeight) OnLoadReport(load *v3orcapb.OrcaLoadReport) {
 	errorRate := load.Eps / load.RpsFractional
 	w.weightVal = load.RpsFractional / (utilization + errorRate*w.cfg.ErrorUtilizationPenalty)
 	if w.logger.V(2) {
-		w.logger.Infof("New weight for subchannel %v: %v", w.SubConn, w.weightVal)
+		w.logger.Infof("New weight for subchannel %v: %v", w.pickedSC, w.weightVal)
 	}
 
 	w.lastUpdated = internal.TimeNow()
diff --git a/credentials/alts/internal/handshaker/service/service.go b/credentials/alts/internal/handshaker/service/service.go
index e0a1afc1..2580995d 100644
--- a/credentials/alts/internal/handshaker/service/service.go
+++ b/credentials/alts/internal/handshaker/service/service.go
@@ -22,9 +22,12 @@ package service
 
 import (
 	"sync"
+	"time"
 
 	grpc "google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/envconfig"
+	"google.golang.org/grpc/keepalive"
 )
 
 var (
@@ -50,7 +53,17 @@ func Dial(hsAddress string) (*grpc.ClientConn, error) {
 		// Disable the service config to avoid unnecessary TXT record lookups that
 		// cause timeouts with some versions of systemd-resolved.
 		var err error
-		hsConn, err = grpc.NewClient(hsAddress, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDisableServiceConfig())
+		opts := []grpc.DialOption{
+			grpc.WithTransportCredentials(insecure.NewCredentials()),
+			grpc.WithDisableServiceConfig(),
+		}
+		if envconfig.ALTSHandshakerKeepaliveParams {
+			opts = append(opts, grpc.WithKeepaliveParams(keepalive.ClientParameters{
+				Timeout: 10 * time.Second,
+				Time:    10 * time.Minute,
+			}))
+		}
+		hsConn, err = grpc.NewClient(hsAddress, opts...)
 		if err != nil {
 			return nil, err
 		}
diff --git a/credentials/local/local.go b/credentials/local/local.go
index 5991d0ad..2fe0215d 100644
--- a/credentials/local/local.go
+++ b/credentials/local/local.go
@@ -49,6 +49,12 @@ func (info) AuthType() string {
 	return "local"
 }
 
+// ValidateAuthority allows any value to be overridden for the :authority
+// header.
+func (info) ValidateAuthority(string) error {
+	return nil
+}
+
 // localTC is the credentials required to establish a local connection.
 type localTC struct {
 	info credentials.ProtocolInfo
diff --git a/internal/envconfig/envconfig.go b/internal/envconfig/envconfig.go
index f122a041..f5f2bdeb 100644
--- a/internal/envconfig/envconfig.go
+++ b/internal/envconfig/envconfig.go
@@ -69,6 +69,10 @@ var (
 	// to gRFC A76. It can be enabled by setting the environment variable
 	// "GRPC_EXPERIMENTAL_RING_HASH_SET_REQUEST_HASH_KEY" to "true".
 	RingHashSetRequestHashKey = boolFromEnv("GRPC_EXPERIMENTAL_RING_HASH_SET_REQUEST_HASH_KEY", false)
+
+	// ALTSHandshakerKeepaliveParams is set if we should add the
+	// KeepaliveParams when dial the ALTS handshaker service.
+	ALTSHandshakerKeepaliveParams = boolFromEnv("GRPC_EXPERIMENTAL_ALTS_HANDSHAKER_KEEPALIVE_PARAMS", false)
 )
 
 func boolFromEnv(envVar string, def bool) bool {
diff --git a/internal/resolver/delegatingresolver/delegatingresolver.go b/internal/resolver/delegatingresolver/delegatingresolver.go
index 7b93f692..c0e22757 100644
--- a/internal/resolver/delegatingresolver/delegatingresolver.go
+++ b/internal/resolver/delegatingresolver/delegatingresolver.go
@@ -28,6 +28,8 @@ import (
 
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal/proxyattributes"
+	"google.golang.org/grpc/internal/transport"
+	"google.golang.org/grpc/internal/transport/networktype"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/serviceconfig"
 )
@@ -40,7 +42,7 @@ var (
 
 // delegatingResolver manages both target URI and proxy address resolution by
 // delegating these tasks to separate child resolvers. Essentially, it acts as
-// a intermediary between the gRPC ClientConn and the child resolvers.
+// an intermediary between the gRPC ClientConn and the child resolvers.
 //
 // It implements the [resolver.Resolver] interface.
 type delegatingResolver struct {
@@ -48,6 +50,9 @@ type delegatingResolver struct {
 	cc       resolver.ClientConn // gRPC ClientConn
 	proxyURL *url.URL            // proxy URL, derived from proxy environment and target
 
+	// We do not hold both mu and childMu in the same goroutine. Avoid holding
+	// both locks when calling into the child, as the child resolver may
+	// synchronously callback into the channel.
 	mu                  sync.Mutex         // protects all the fields below
 	targetResolverState *resolver.State    // state of the target resolver
 	proxyAddrs          []resolver.Address // resolved proxy addresses; empty if no proxy is configured
@@ -66,8 +71,8 @@ func (nopResolver) ResolveNow(resolver.ResolveNowOptions) {}
 
 func (nopResolver) Close() {}
 
-// proxyURLForTarget determines the proxy URL for the given address based on
-// the environment. It can return the following:
+// proxyURLForTarget determines the proxy URL for the given address based on the
+// environment. It can return the following:
 //   - nil URL, nil error: No proxy is configured or the address is excluded
 //     using the `NO_PROXY` environment variable or if req.URL.Host is
 //     "localhost" (with or without // a port number)
@@ -86,7 +91,8 @@ func proxyURLForTarget(address string) (*url.URL, error) {
 // resolvers:
 //   - one to resolve the proxy address specified using the supported
 //     environment variables. This uses the registered resolver for the "dns"
-//     scheme.
+//     scheme. It is lazily built when a target resolver update contains at least
+//     one TCP address.
 //   - one to resolve the target URI using the resolver specified by the scheme
 //     in the target URI or specified by the user using the WithResolvers dial
 //     option. As a special case, if the target URI's scheme is "dns" and a
@@ -95,8 +101,10 @@ func proxyURLForTarget(address string) (*url.URL, error) {
 //     resolution is enabled using the dial option.
 func New(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions, targetResolverBuilder resolver.Builder, targetResolutionEnabled bool) (resolver.Resolver, error) {
 	r := &delegatingResolver{
-		target: target,
-		cc:     cc,
+		target:         target,
+		cc:             cc,
+		proxyResolver:  nopResolver{},
+		targetResolver: nopResolver{},
 	}
 
 	var err error
@@ -123,37 +131,26 @@ func New(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOpti
 	// resolution should be handled by the proxy, not the client. Therefore, we
 	// bypass the target resolver and store the unresolved target address.
 	if target.URL.Scheme == "dns" && !targetResolutionEnabled {
-		state := resolver.State{
+		r.targetResolverState = &resolver.State{
 			Addresses: []resolver.Address{{Addr: target.Endpoint()}},
 			Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: target.Endpoint()}}}},
 		}
-		r.targetResolverState = &state
-	} else {
-		wcc := &wrappingClientConn{
-			stateListener: r.updateTargetResolverState,
-			parent:        r,
-		}
-		if r.targetResolver, err = targetResolverBuilder.Build(target, wcc, opts); err != nil {
-			return nil, fmt.Errorf("delegating_resolver: unable to build the resolver for target %s: %v", target, err)
-		}
-	}
-
-	if r.proxyResolver, err = r.proxyURIResolver(opts); err != nil {
-		return nil, fmt.Errorf("delegating_resolver: failed to build resolver for proxy URL %q: %v", r.proxyURL, err)
+		r.updateTargetResolverState(*r.targetResolverState)
+		return r, nil
 	}
-
-	if r.targetResolver == nil {
-		r.targetResolver = nopResolver{}
+	wcc := &wrappingClientConn{
+		stateListener: r.updateTargetResolverState,
+		parent:        r,
 	}
-	if r.proxyResolver == nil {
-		r.proxyResolver = nopResolver{}
+	if r.targetResolver, err = targetResolverBuilder.Build(target, wcc, opts); err != nil {
+		return nil, fmt.Errorf("delegating_resolver: unable to build the resolver for target %s: %v", target, err)
 	}
 	return r, nil
 }
 
-// proxyURIResolver creates a resolver for resolving proxy URIs using the
-// "dns" scheme. It adjusts the proxyURL to conform to the "dns:///" format and
-// builds a resolver with a wrappingClientConn to capture resolved addresses.
+// proxyURIResolver creates a resolver for resolving proxy URIs using the "dns"
+// scheme. It adjusts the proxyURL to conform to the "dns:///" format and builds
+// a resolver with a wrappingClientConn to capture resolved addresses.
 func (r *delegatingResolver) proxyURIResolver(opts resolver.BuildOptions) (resolver.Resolver, error) {
 	proxyBuilder := resolver.Get("dns")
 	if proxyBuilder == nil {
@@ -189,18 +186,43 @@ func (r *delegatingResolver) Close() {
 	r.proxyResolver = nil
 }
 
-// updateClientConnStateLocked creates a list of combined addresses by
-// pairing each proxy address with every target address. For each pair, it
-// generates a new [resolver.Address] using the proxy address, and adding the
-// target address as the attribute along with user info. It returns nil if
-// either resolver has not sent update even once and returns the error from
-// ClientConn update once both resolvers have sent update atleast once.
+func networkTypeFromAddr(addr resolver.Address) string {
+	networkType, ok := networktype.Get(addr)
+	if !ok {
+		networkType, _ = transport.ParseDialTarget(addr.Addr)
+	}
+	return networkType
+}
+
+func isTCPAddressPresent(state *resolver.State) bool {
+	for _, addr := range state.Addresses {
+		if networkType := networkTypeFromAddr(addr); networkType == "tcp" {
+			return true
+		}
+	}
+	for _, endpoint := range state.Endpoints {
+		for _, addr := range endpoint.Addresses {
+			if networktype := networkTypeFromAddr(addr); networktype == "tcp" {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// updateClientConnStateLocked constructs a combined list of addresses by
+// pairing each proxy address with every target address of type TCP. For each
+// pair, it creates a new [resolver.Address] using the proxy address and
+// attaches the corresponding target address and user info as attributes. Target
+// addresses that are not of type TCP are appended to the list as-is. The
+// function returns nil if either resolver has not yet provided an update, and
+// returns the result of ClientConn.UpdateState once both resolvers have
+// provided at least one update.
 func (r *delegatingResolver) updateClientConnStateLocked() error {
 	if r.targetResolverState == nil || r.proxyAddrs == nil {
 		return nil
 	}
 
-	curState := *r.targetResolverState
 	// If multiple resolved proxy addresses are present, we send only the
 	// unresolved proxy host and let net.Dial handle the proxy host name
 	// resolution when creating the transport. Sending all resolved addresses
@@ -218,24 +240,30 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 	}
 	var addresses []resolver.Address
 	for _, targetAddr := range (*r.targetResolverState).Addresses {
+		// Avoid proxy when network is not tcp.
+		if networkType := networkTypeFromAddr(targetAddr); networkType != "tcp" {
+			addresses = append(addresses, targetAddr)
+			continue
+		}
 		addresses = append(addresses, proxyattributes.Set(proxyAddr, proxyattributes.Options{
 			User:        r.proxyURL.User,
 			ConnectAddr: targetAddr.Addr,
 		}))
 	}
 
-	// Create a list of combined endpoints by pairing all proxy endpoints
-	// with every target endpoint. Each time, it constructs a new
-	// [resolver.Endpoint] using the all addresses from all the proxy endpoint
-	// and the target addresses from one endpoint. The target address and user
-	// information from the proxy URL are added as attributes to the proxy
-	// address.The resulting list of addresses is then grouped into endpoints,
-	// covering all combinations of proxy and target endpoints.
+	// For each target endpoint, construct a new [resolver.Endpoint] that
+	// includes all addresses from all proxy endpoints and the addresses from
+	// that target endpoint, preserving the number of target endpoints.
 	var endpoints []resolver.Endpoint
 	for _, endpt := range (*r.targetResolverState).Endpoints {
 		var addrs []resolver.Address
-		for _, proxyAddr := range r.proxyAddrs {
-			for _, targetAddr := range endpt.Addresses {
+		for _, targetAddr := range endpt.Addresses {
+			// Avoid proxy when network is not tcp.
+			if networkType := networkTypeFromAddr(targetAddr); networkType != "tcp" {
+				addrs = append(addrs, targetAddr)
+				continue
+			}
+			for _, proxyAddr := range r.proxyAddrs {
 				addrs = append(addrs, proxyattributes.Set(proxyAddr, proxyattributes.Options{
 					User:        r.proxyURL.User,
 					ConnectAddr: targetAddr.Addr,
@@ -246,8 +274,9 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 	}
 	// Use the targetResolverState for its service config and attributes
 	// contents. The state update is only sent after both the target and proxy
-	// resolvers have sent their updates, and curState has been updated with
-	// the combined addresses.
+	// resolvers have sent their updates, and curState has been updated with the
+	// combined addresses.
+	curState := *r.targetResolverState
 	curState.Addresses = addresses
 	curState.Endpoints = endpoints
 	return r.cc.UpdateState(curState)
@@ -257,7 +286,8 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 // addresses and endpoints, marking the resolver as ready, and triggering a
 // state update if both proxy and target resolvers are ready. If the ClientConn
 // returns a non-nil error, it calls `ResolveNow()` on the target resolver.  It
-// is a StateListener function of wrappingClientConn passed to the proxy resolver.
+// is a StateListener function of wrappingClientConn passed to the proxy
+// resolver.
 func (r *delegatingResolver) updateProxyResolverState(state resolver.State) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -265,8 +295,8 @@ func (r *delegatingResolver) updateProxyResolverState(state resolver.State) erro
 		logger.Infof("Addresses received from proxy resolver: %s", state.Addresses)
 	}
 	if len(state.Endpoints) > 0 {
-		// We expect exactly one address per endpoint because the proxy
-		// resolver uses "dns" resolution.
+		// We expect exactly one address per endpoint because the proxy resolver
+		// uses "dns" resolution.
 		r.proxyAddrs = make([]resolver.Address, 0, len(state.Endpoints))
 		for _, endpoint := range state.Endpoints {
 			r.proxyAddrs = append(r.proxyAddrs, endpoint.Addresses...)
@@ -294,11 +324,14 @@ func (r *delegatingResolver) updateProxyResolverState(state resolver.State) erro
 	return err
 }
 
-// updateTargetResolverState updates the target resolver state by storing target
-// addresses, endpoints, and service config, marking the resolver as ready, and
-// triggering a state update if both resolvers are ready. If the ClientConn
-// returns a non-nil error, it calls `ResolveNow()` on the proxy resolver. It
-// is a StateListener function of wrappingClientConn passed to the target resolver.
+// updateTargetResolverState is the StateListener function provided to the
+// target resolver via wrappingClientConn. It updates the resolver state and
+// marks the target resolver as ready. If the update includes at least one TCP
+// address and the proxy resolver has not yet been constructed, it initializes
+// the proxy resolver. A combined state update is triggered once both resolvers
+// are ready. If all addresses are non-TCP, it proceeds without waiting for the
+// proxy resolver. If ClientConn.UpdateState returns a non-nil error,
+// ResolveNow() is called on the proxy resolver.
 func (r *delegatingResolver) updateTargetResolverState(state resolver.State) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -307,6 +340,31 @@ func (r *delegatingResolver) updateTargetResolverState(state resolver.State) err
 		logger.Infof("Addresses received from target resolver: %v", state.Addresses)
 	}
 	r.targetResolverState = &state
+	// If no addresses returned by resolver have network type as tcp , do not
+	// wait for proxy update.
+	if !isTCPAddressPresent(r.targetResolverState) {
+		return r.cc.UpdateState(*r.targetResolverState)
+	}
+
+	// The proxy resolver may be rebuilt multiple times, specifically each time
+	// the target resolver sends an update, even if the target resolver is built
+	// successfully but building the proxy resolver fails.
+	if len(r.proxyAddrs) == 0 {
+		go func() {
+			r.childMu.Lock()
+			defer r.childMu.Unlock()
+			if _, ok := r.proxyResolver.(nopResolver); !ok {
+				return
+			}
+			proxyResolver, err := r.proxyURIResolver(resolver.BuildOptions{})
+			if err != nil {
+				r.cc.ReportError(fmt.Errorf("delegating_resolver: unable to build the proxy resolver: %v", err))
+				return
+			}
+			r.proxyResolver = proxyResolver
+		}()
+	}
+
 	err := r.updateClientConnStateLocked()
 	if err != nil {
 		go func() {
@@ -335,7 +393,8 @@ func (wcc *wrappingClientConn) UpdateState(state resolver.State) error {
 	return wcc.stateListener(state)
 }
 
-// ReportError intercepts errors from the child resolvers and passes them to ClientConn.
+// ReportError intercepts errors from the child resolvers and passes them to
+// ClientConn.
 func (wcc *wrappingClientConn) ReportError(err error) {
 	wcc.parent.cc.ReportError(err)
 }
@@ -346,8 +405,8 @@ func (wcc *wrappingClientConn) NewAddress(addrs []resolver.Address) {
 	wcc.UpdateState(resolver.State{Addresses: addrs})
 }
 
-// ParseServiceConfig parses the provided service config and returns an
-// object that provides the parsed config.
+// ParseServiceConfig parses the provided service config and returns an object
+// that provides the parsed config.
 func (wcc *wrappingClientConn) ParseServiceConfig(serviceConfigJSON string) *serviceconfig.ParseResult {
 	return wcc.parent.cc.ParseServiceConfig(serviceConfigJSON)
 }
diff --git a/internal/status/status.go b/internal/status/status.go
index 1186f1e9..aad171cd 100644
--- a/internal/status/status.go
+++ b/internal/status/status.go
@@ -236,3 +236,11 @@ func IsRestrictedControlPlaneCode(s *Status) bool {
 	}
 	return false
 }
+
+// RawStatusProto returns the internal protobuf message for use by gRPC itself.
+func RawStatusProto(s *Status) *spb.Status {
+	if s == nil {
+		return nil
+	}
+	return s.s
+}
diff --git a/internal/transport/http2_client.go b/internal/transport/http2_client.go
index 32047128..ac0fd4b3 100644
--- a/internal/transport/http2_client.go
+++ b/internal/transport/http2_client.go
@@ -176,7 +176,7 @@ func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error
 		return fn(ctx, address)
 	}
 	if !ok {
-		networkType, address = parseDialTarget(address)
+		networkType, address = ParseDialTarget(address)
 	}
 	if opts, present := proxyattributes.Get(addr); present {
 		return proxyDial(ctx, addr, grpcUA, opts)
diff --git a/internal/transport/http2_server.go b/internal/transport/http2_server.go
index 7e53eb17..e4c3731b 100644
--- a/internal/transport/http2_server.go
+++ b/internal/transport/http2_server.go
@@ -39,6 +39,7 @@ import (
 	"google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/grpcutil"
 	"google.golang.org/grpc/internal/pretty"
+	istatus "google.golang.org/grpc/internal/status"
 	"google.golang.org/grpc/internal/syscall"
 	"google.golang.org/grpc/mem"
 	"google.golang.org/protobuf/proto"
@@ -1055,7 +1056,7 @@ func (t *http2Server) writeHeaderLocked(s *ServerStream) error {
 	return nil
 }
 
-// WriteStatus sends stream status to the client and terminates the stream.
+// writeStatus sends stream status to the client and terminates the stream.
 // There is no further I/O operations being able to perform on this stream.
 // TODO(zhaoq): Now it indicates the end of entire stream. Revisit if early
 // OK is adopted.
@@ -1083,7 +1084,7 @@ func (t *http2Server) writeStatus(s *ServerStream, st *status.Status) error {
 	headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-status", Value: strconv.Itoa(int(st.Code()))})
 	headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-message", Value: encodeGrpcMessage(st.Message())})
 
-	if p := st.Proto(); p != nil && len(p.Details) > 0 {
+	if p := istatus.RawStatusProto(st); len(p.GetDetails()) > 0 {
 		// Do not use the user's grpc-status-details-bin (if present) if we are
 		// even attempting to set our own.
 		delete(s.trailer, grpcStatusDetailsBinHeader)
diff --git a/internal/transport/http_util.go b/internal/transport/http_util.go
index 3613d7b6..f997f9fd 100644
--- a/internal/transport/http_util.go
+++ b/internal/transport/http_util.go
@@ -439,8 +439,8 @@ func getWriteBufferPool(size int) *sync.Pool {
 	return pool
 }
 
-// parseDialTarget returns the network and address to pass to dialer.
-func parseDialTarget(target string) (string, string) {
+// ParseDialTarget returns the network and address to pass to dialer.
+func ParseDialTarget(target string) (string, string) {
 	net := "tcp"
 	m1 := strings.Index(target, ":")
 	m2 := strings.Index(target, ":/")
diff --git a/stream.go b/stream.go
index bbbcbc72..d58bb647 100644
--- a/stream.go
+++ b/stream.go
@@ -101,9 +101,9 @@ type ClientStream interface {
 	// It must only be called after stream.CloseAndRecv has returned, or
 	// stream.Recv has returned a non-nil error (including io.EOF).
 	Trailer() metadata.MD
-	// CloseSend closes the send direction of the stream. It closes the stream
-	// when non-nil error is met. It is also not safe to call CloseSend
-	// concurrently with SendMsg.
+	// CloseSend closes the send direction of the stream. This method always
+	// returns a nil error. The status of the stream may be discovered using
+	// RecvMsg. It is also not safe to call CloseSend concurrently with SendMsg.
 	CloseSend() error
 	// Context returns the context for this stream.
 	//
@@ -993,7 +993,7 @@ func (cs *clientStream) RecvMsg(m any) error {
 
 func (cs *clientStream) CloseSend() error {
 	if cs.sentLast {
-		// TODO: return an error and finish the stream instead, due to API misuse?
+		// Return a nil error on repeated calls to this method.
 		return nil
 	}
 	cs.sentLast = true
@@ -1014,7 +1014,10 @@ func (cs *clientStream) CloseSend() error {
 			binlog.Log(cs.ctx, chc)
 		}
 	}
-	// We never returned an error here for reasons.
+	// We don't return an error here as we expect users to read all messages
+	// from the stream and get the RPC status from RecvMsg().  Note that
+	// SendMsg() must return an error when one occurs so the application
+	// knows to stop sending messages, but that does not apply here.
 	return nil
 }
 
@@ -1378,7 +1381,7 @@ func (as *addrConnStream) Trailer() metadata.MD {
 
 func (as *addrConnStream) CloseSend() error {
 	if as.sentLast {
-		// TODO: return an error and finish the stream instead, due to API misuse?
+		// Return a nil error on repeated calls to this method.
 		return nil
 	}
 	as.sentLast = true
diff --git a/xds/internal/clients/config.go b/xds/internal/clients/config.go
index 0ededfae..f106465f 100644
--- a/xds/internal/clients/config.go
+++ b/xds/internal/clients/config.go
@@ -100,3 +100,13 @@ type Locality struct {
 	// SubZone is the further subdivision within a zone.
 	SubZone string
 }
+
+// MetricsReporter is used by the XDSClient to report metrics.
+type MetricsReporter interface {
+	// ReportMetric reports a metric. The metric will be one of the predefined
+	// set of types depending on the client (XDSClient or LRSClient).
+	//
+	// Each client will produce different metrics. Please see the client's
+	// documentation for a list of possible metrics events.
+	ReportMetric(metric any)
+}
diff --git a/xds/internal/clients/grpctransport/grpc_transport.go b/xds/internal/clients/grpctransport/grpc_transport.go
index e13c982a..af39349e 100644
--- a/xds/internal/clients/grpctransport/grpc_transport.go
+++ b/xds/internal/clients/grpctransport/grpc_transport.go
@@ -43,31 +43,41 @@ var (
 // It must be set by value (not pointer) in the
 // clients.ServerIdentifier.Extensions field (See Example).
 type ServerIdentifierExtension struct {
-	// Credentials is name of the credentials to use for this transport to the
-	// server. It must be present in the map passed to NewBuilder.
-	Credentials string
+	// ConfigName is the name of the configuration to use for this transport.
+	// It must be present as a key in the map of configs passed to NewBuilder.
+	ConfigName string
 }
 
 // Builder creates gRPC-based Transports. It must be paired with ServerIdentifiers
 // that contain an Extension field of type ServerIdentifierExtension.
 type Builder struct {
-	// credentials is a map of credentials names to credentials.Bundle which
-	// can be used to connect to the server.
-	credentials map[string]credentials.Bundle
+	// configs is a map of configuration names to their respective Config.
+	configs map[string]Config
 
 	mu sync.Mutex
 	// connections is a map of clients.ServerIdentifiers in use by the Builder
 	// to connect to different servers.
 	connections map[clients.ServerIdentifier]*grpc.ClientConn
-	refs        map[clients.ServerIdentifier]int
+	// refs tracks the number of active references to each connection.
+	refs map[clients.ServerIdentifier]int
+}
+
+// Config defines the configuration for connecting to a gRPC server, including
+// credentials and an optional custom new client function.
+type Config struct {
+	// Credentials is the credentials bundle to be used for the connection.
+	Credentials credentials.Bundle
+	// GRPCNewClient is an optional custom function to establish a gRPC connection.
+	// If nil, grpc.NewClient will be used as the default.
+	GRPCNewClient func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error)
 }
 
 // NewBuilder provides a builder for creating gRPC-based Transports using
 // the credentials from provided map of credentials names to
 // credentials.Bundle.
-func NewBuilder(credentials map[string]credentials.Bundle) *Builder {
+func NewBuilder(configs map[string]Config) *Builder {
 	return &Builder{
-		credentials: credentials,
+		configs:     configs,
 		connections: make(map[clients.ServerIdentifier]*grpc.ClientConn),
 		refs:        make(map[clients.ServerIdentifier]int),
 	}
@@ -88,9 +98,12 @@ func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error)
 		return nil, fmt.Errorf("grpctransport: Extensions field is %T, but must be %T in ServerIdentifier", si.Extensions, ServerIdentifierExtension{})
 	}
 
-	creds, ok := b.credentials[sce.Credentials]
+	config, ok := b.configs[sce.ConfigName]
 	if !ok {
-		return nil, fmt.Errorf("grpctransport: unknown credentials type %q specified in extensions", sce.Credentials)
+		return nil, fmt.Errorf("grpctransport: unknown config name %q specified in ServerIdentifierExtension", sce.ConfigName)
+	}
+	if config.Credentials == nil {
+		return nil, fmt.Errorf("grpctransport: config %q has nil credentials bundle", sce.ConfigName)
 	}
 
 	b.mu.Lock()
@@ -114,7 +127,12 @@ func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error)
 		Time:    5 * time.Minute,
 		Timeout: 20 * time.Second,
 	})
-	cc, err := grpc.NewClient(si.ServerURI, kpCfg, grpc.WithCredentialsBundle(creds), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{})))
+	dopts := []grpc.DialOption{kpCfg, grpc.WithCredentialsBundle(config.Credentials), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{}))}
+	newClientFunc := grpc.NewClient
+	if config.GRPCNewClient != nil {
+		newClientFunc = config.GRPCNewClient
+	}
+	cc, err := newClientFunc(si.ServerURI, dopts...)
 	if err != nil {
 		return nil, fmt.Errorf("grpctransport: failed to create connection to server %q: %v", si.ServerURI, err)
 	}
diff --git a/xds/internal/clients/xdsclient/authority.go b/xds/internal/clients/xdsclient/authority.go
index 3ada1465..7a3a2969 100644
--- a/xds/internal/clients/xdsclient/authority.go
+++ b/xds/internal/clients/xdsclient/authority.go
@@ -26,8 +26,10 @@ import (
 
 	"google.golang.org/grpc/grpclog"
 	igrpclog "google.golang.org/grpc/internal/grpclog"
+	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/internal/syncutil"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient/metrics"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
@@ -87,6 +89,7 @@ type authority struct {
 	xdsClientSerializerClose  func()                       // Function to close the above serializer.
 	logger                    *igrpclog.PrefixLogger       // Logger for this authority.
 	target                    string                       // The gRPC Channel target.
+	metricsReporter           clients.MetricsReporter
 
 	// The below defined fields must only be accessed in the context of the
 	// serializer callback, owned by this authority.
@@ -121,6 +124,7 @@ type authorityBuildOptions struct {
 	getChannelForADS xdsChannelForADS             // Function to acquire a reference to an xdsChannel
 	logPrefix        string                       // Prefix for logging
 	target           string                       // Target for the gRPC Channel that owns xDS Client/Authority
+	metricsReporter  clients.MetricsReporter      // Metrics reporter for the authority
 }
 
 // newAuthority creates a new authority instance with the provided
@@ -145,6 +149,7 @@ func newAuthority(args authorityBuildOptions) *authority {
 		logger:                    igrpclog.NewPrefixLogger(l, logPrefix),
 		resources:                 make(map[ResourceType]map[string]*resourceState),
 		target:                    args.target,
+		metricsReporter:           args.metricsReporter,
 	}
 
 	// Create an ordered list of xdsChannels with their server configs. The
@@ -363,6 +368,11 @@ func (a *authority) handleADSResourceUpdate(serverConfig *ServerConfig, rType Re
 		// On error, keep previous version of the resource. But update status
 		// and error.
 		if uErr.Err != nil {
+			if a.metricsReporter != nil {
+				a.metricsReporter.ReportMetric(&metrics.ResourceUpdateInvalid{
+					ServerURI: serverConfig.ServerIdentifier.ServerURI, ResourceType: rType.TypeName,
+				})
+			}
 			state.md.ErrState = md.ErrState
 			state.md.Status = md.Status
 			for watcher := range state.watchers {
@@ -378,6 +388,12 @@ func (a *authority) handleADSResourceUpdate(serverConfig *ServerConfig, rType Re
 			continue
 		}
 
+		if a.metricsReporter != nil {
+			a.metricsReporter.ReportMetric(&metrics.ResourceUpdateValid{
+				ServerURI: serverConfig.ServerIdentifier.ServerURI, ResourceType: rType.TypeName,
+			})
+		}
+
 		if state.deletionIgnored {
 			state.deletionIgnored = false
 			a.logger.Infof("A valid update was received for resource %q of type %q after previously ignoring a deletion", name, rType.TypeName)
diff --git a/xds/internal/clients/xdsclient/metrics/metrics.go b/xds/internal/clients/xdsclient/metrics/metrics.go
new file mode 100644
index 00000000..2e14efb1
--- /dev/null
+++ b/xds/internal/clients/xdsclient/metrics/metrics.go
@@ -0,0 +1,42 @@
+/*
+ *
+ * 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 metrics defines all metrics that can be produced by an xDS client.
+// All calls to the MetricsRecorder by the xDS client will contain a struct
+// from this package passed by pointer.
+package metrics
+
+// ResourceUpdateValid is a metric to report a valid resource update from
+// the xDS management server for a given resource type.
+type ResourceUpdateValid struct {
+	ServerURI    string
+	ResourceType string
+}
+
+// ResourceUpdateInvalid is a metric to report an invalid resource update
+// from the xDS management server for a given resource type.
+type ResourceUpdateInvalid struct {
+	ServerURI    string
+	ResourceType string
+}
+
+// ServerFailure is a metric to report a server failure of the xDS
+// management server.
+type ServerFailure struct {
+	ServerURI string
+}
diff --git a/xds/internal/clients/xdsclient/xdsclient.go b/xds/internal/clients/xdsclient/xdsclient.go
index f0f4a678..415a2a1e 100644
--- a/xds/internal/clients/xdsclient/xdsclient.go
+++ b/xds/internal/clients/xdsclient/xdsclient.go
@@ -44,6 +44,7 @@ import (
 	"google.golang.org/grpc/xds/internal/clients/internal/backoff"
 	"google.golang.org/grpc/xds/internal/clients/internal/syncutil"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient/metrics"
 	"google.golang.org/protobuf/proto"
 
 	v3statuspb "github.com/envoyproxy/go-control-plane/envoy/service/status/v3"
@@ -82,8 +83,9 @@ type XDSClient struct {
 	resourceTypes      map[string]ResourceType      // Registry of resource types, for parsing incoming ADS responses.
 	serializer         *syncutil.CallbackSerializer // Serializer for invoking resource watcher callbacks.
 	serializerClose    func()                       // Function to close the serializer.
-	logger             *grpclog.PrefixLogger        // Logger for this client.
-	target             string                       // The gRPC target for this client.
+	logger             *grpclog.PrefixLogger
+	target             string
+	metricsReporter    clients.MetricsReporter
 
 	// The XDSClient owns a bunch of channels to individual xDS servers
 	// specified in the xDS client configuration. Authorities acquire references
@@ -138,6 +140,7 @@ func newClient(config *Config, watchExpiryTimeout time.Duration, streamBackoff f
 		transportBuilder:   config.TransportBuilder,
 		resourceTypes:      config.ResourceTypes,
 		xdsActiveChannels:  make(map[ServerConfig]*channelState),
+		metricsReporter:    config.MetricsReporter,
 	}
 
 	for name, cfg := range config.Authorities {
@@ -154,6 +157,7 @@ func newClient(config *Config, watchExpiryTimeout time.Duration, streamBackoff f
 			getChannelForADS: c.getChannelForADS,
 			logPrefix:        clientPrefix(c),
 			target:           target,
+			metricsReporter:  c.metricsReporter,
 		})
 	}
 	c.topLevelAuthority = newAuthority(authorityBuildOptions{
@@ -163,8 +167,10 @@ func newClient(config *Config, watchExpiryTimeout time.Duration, streamBackoff f
 		getChannelForADS: c.getChannelForADS,
 		logPrefix:        clientPrefix(c),
 		target:           target,
+		metricsReporter:  c.metricsReporter,
 	})
 	c.logger = prefixLogger(c)
+
 	return c, nil
 }
 
@@ -384,6 +390,12 @@ func (cs *channelState) adsStreamFailure(err error) {
 		return
 	}
 
+	if xdsresource.ErrType(err) != xdsresource.ErrTypeStreamFailedAfterRecv && cs.parent.metricsReporter != nil {
+		cs.parent.metricsReporter.ReportMetric(&metrics.ServerFailure{
+			ServerURI: cs.serverConfig.ServerIdentifier.ServerURI,
+		})
+	}
+
 	cs.parent.channelsMu.Lock()
 	defer cs.parent.channelsMu.Unlock()
 	for authority := range cs.interestedAuthorities {
diff --git a/xds/internal/clients/xdsclient/xdsconfig.go b/xds/internal/clients/xdsclient/xdsconfig.go
index 4c7d090d..a276309d 100644
--- a/xds/internal/clients/xdsclient/xdsconfig.go
+++ b/xds/internal/clients/xdsclient/xdsconfig.go
@@ -56,6 +56,10 @@ type Config struct {
 	//
 	// For example: "type.googleapis.com/envoy.config.listener.v3.Listener"
 	ResourceTypes map[string]ResourceType
+
+	// MetricsReporter is used to report registered metrics. If unset, no
+	// metrics will be reported.
+	MetricsReporter clients.MetricsReporter
 }
 
 // ServerConfig contains configuration for an xDS management server.



Test patch:
diff --git a/credentials/credentials_ext_test.go b/credentials/credentials_ext_test.go
index 27da3a26..f1733358 100644
--- a/credentials/credentials_ext_test.go
+++ b/credentials/credentials_ext_test.go
@@ -30,6 +30,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/credentials/local"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
@@ -57,45 +58,88 @@ func authorityChecker(ctx context.Context, wantAuthority string) error {
 	return nil
 }
 
-// Tests the `grpc.CallAuthority` option with TLS credentials. This test verifies
-// that the provided authority is correctly propagated to the server when a
-// correct authority is used.
-func (s) TestCorrectAuthorityWithTLSCreds(t *testing.T) {
+func loadTLSCreds(t *testing.T) (grpc.ServerOption, grpc.DialOption) {
+	t.Helper()
 	cert, err := tls.LoadX509KeyPair(testdata.Path("x509/server1_cert.pem"), testdata.Path("x509/server1_key.pem"))
 	if err != nil {
-		t.Fatalf("Failed to load key pair: %s", err)
+		t.Fatalf("Failed to load key pair: %v", err)
+		return nil, nil
 	}
-	creds, err := credentials.NewClientTLSFromFile(testdata.Path("x509/server_ca_cert.pem"), "x.test.example.com")
+	serverCreds := grpc.Creds(credentials.NewServerTLSFromCert(&cert))
+
+	clientCreds, err := credentials.NewClientTLSFromFile(testdata.Path("x509/server_ca_cert.pem"), "x.test.example.com")
 	if err != nil {
-		t.Fatalf("Failed to create credentials %v", err)
+		t.Fatalf("Failed to create client credentials: %v", err)
 	}
+	return serverCreds, grpc.WithTransportCredentials(clientCreds)
+}
+
+// Tests the scenario where the `grpc.CallAuthority` call option is used with
+// different transport credentials. The test verifies that the specified
+// authority is correctly propagated to the serve when a correct authority is
+// used.
+func (s) TestCorrectAuthorityWithCreds(t *testing.T) {
 	const authority = "auth.test.example.com"
-	ss := &stubserver.StubServer{
-		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
-			if err := authorityChecker(ctx, authority); err != nil {
-				return nil, err
-			}
-			return &testpb.Empty{}, nil
+
+	tests := []struct {
+		name         string
+		creds        func(t *testing.T) (grpc.ServerOption, grpc.DialOption)
+		expectedAuth string
+	}{
+		{
+			name: "Insecure",
+			creds: func(t *testing.T) (grpc.ServerOption, grpc.DialOption) {
+				c := insecure.NewCredentials()
+				return grpc.Creds(c), grpc.WithTransportCredentials(c)
+			},
+			expectedAuth: authority,
+		},
+		{
+			name: "Local",
+			creds: func(t *testing.T) (grpc.ServerOption, grpc.DialOption) {
+				c := local.NewCredentials()
+				return grpc.Creds(c), grpc.WithTransportCredentials(c)
+			},
+			expectedAuth: authority,
+		},
+		{
+			name: "TLS",
+			creds: func(t *testing.T) (grpc.ServerOption, grpc.DialOption) {
+				return loadTLSCreds(t)
+			},
+			expectedAuth: authority,
 		},
 	}
-	if err := ss.StartServer(grpc.Creds(credentials.NewServerTLSFromCert(&cert))); err != nil {
-		t.Fatalf("Error starting endpoint server: %v", err)
-	}
-	defer ss.Stop()
 
-	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(creds))
-	if err != nil {
-		t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
-	}
-	defer cc.Close()
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ss := &stubserver.StubServer{
+				EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+					if err := authorityChecker(ctx, tt.expectedAuth); err != nil {
+						return nil, err
+					}
+					return &testpb.Empty{}, nil
+				},
+			}
+			serverOpt, dialOpt := tt.creds(t)
+			if err := ss.StartServer(serverOpt); err != nil {
+				t.Fatalf("Error starting endpoint server: %v", err)
+			}
+			defer ss.Stop()
 
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
+			cc, err := grpc.NewClient(ss.Address, dialOpt)
+			if err != nil {
+				t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
+			}
+			defer cc.Close()
 
-	if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(authority)); status.Code(err) != codes.OK {
-		t.Fatalf("EmptyCall() returned status %v, want %v", status.Code(err), codes.OK)
+			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+			defer cancel()
+			if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(tt.expectedAuth)); err != nil {
+				t.Fatalf("EmptyCall() rpc failed: %v", err)
+			}
+		})
 	}
-
 }
 
 // Tests the `grpc.CallAuthority` option with TLS credentials. This test verifies
@@ -143,38 +187,6 @@ func (s) TestIncorrectAuthorityWithTLS(t *testing.T) {
 	}
 }
 
-// Tests the scenario where the `grpc.CallAuthority` call option is used with
-// insecure transport credentials. The test verifies that the specified
-// authority is correctly propagated to the server.
-func (s) TestAuthorityCallOptionWithInsecureCreds(t *testing.T) {
-	const authority = "test.server.name"
-
-	ss := &stubserver.StubServer{
-		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
-			if err := authorityChecker(ctx, authority); err != nil {
-				return nil, err
-			}
-			return &testpb.Empty{}, nil
-		},
-	}
-	if err := ss.Start(nil); err != nil {
-		t.Fatalf("Error starting endpoint server: %v", err)
-	}
-	defer ss.Stop()
-
-	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
-	}
-	defer cc.Close()
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(authority)); err != nil {
-		t.Fatalf("EmptyCall() rpc failed: %v", err)
-	}
-}
-
 // testAuthInfoNoValidator implements only credentials.AuthInfo and not
 // credentials.AuthorityValidator.
 type testAuthInfoNoValidator struct{}
diff --git a/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go b/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
index 148b4481..602c5372 100644
--- a/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
+++ b/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
@@ -31,6 +31,7 @@ import (
 	"google.golang.org/grpc/internal/proxyattributes"
 	"google.golang.org/grpc/internal/resolver/delegatingresolver"
 	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/grpc/internal/transport/networktype"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/serviceconfig"
@@ -187,8 +188,11 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithTargetResolution(t *testing.T)
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	proxyResolver.UpdateState(resolver.State{
-		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
+	targetResolver.UpdateState(resolver.State{
+		Addresses: []resolver.Address{
+			{Addr: resolvedTargetTestAddr1},
+			{Addr: resolvedTargetTestAddr2},
+		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
 
@@ -198,11 +202,8 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithTargetResolution(t *testing.T)
 	case <-time.After(defaultTestShortTimeout):
 	}
 
-	targetResolver.UpdateState(resolver.State{
-		Addresses: []resolver.Address{
-			{Addr: resolvedTargetTestAddr1},
-			{Addr: resolvedTargetTestAddr2},
-		},
+	proxyResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
 
@@ -325,8 +326,11 @@ func (s) TestDelegatingResolverwithCustomResolverAndProxy(t *testing.T) {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	proxyResolver.UpdateState(resolver.State{
-		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
+	targetResolver.UpdateState(resolver.State{
+		Addresses: []resolver.Address{
+			{Addr: resolvedTargetTestAddr1},
+			{Addr: resolvedTargetTestAddr2},
+		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
 
@@ -336,11 +340,8 @@ func (s) TestDelegatingResolverwithCustomResolverAndProxy(t *testing.T) {
 	case <-time.After(defaultTestShortTimeout):
 	}
 
-	targetResolver.UpdateState(resolver.State{
-		Addresses: []resolver.Address{
-			{Addr: resolvedTargetTestAddr1},
-			{Addr: resolvedTargetTestAddr2},
-		},
+	proxyResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
 
@@ -407,19 +408,6 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	proxyResolver.UpdateState(resolver.State{
-		Endpoints: []resolver.Endpoint{
-			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr1}}},
-			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr2}}},
-		},
-		ServiceConfig: &serviceconfig.ParseResult{},
-	})
-
-	select {
-	case <-stateCh:
-		t.Fatalf("Delegating resolver invoked UpdateState before both the proxy and target resolvers had updated their states.")
-	case <-time.After(defaultTestShortTimeout):
-	}
 	targetResolver.UpdateState(resolver.State{
 		Endpoints: []resolver.Endpoint{
 			{
@@ -435,22 +423,34 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
+	select {
+	case <-stateCh:
+		t.Fatalf("Delegating resolver invoked UpdateState before both the proxy and target resolvers had updated their states.")
+	case <-time.After(defaultTestShortTimeout):
+	}
 
+	proxyResolver.UpdateState(resolver.State{
+		Endpoints: []resolver.Endpoint{
+			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr1}}},
+			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr2}}},
+		},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	})
 	wantState := resolver.State{
 		Endpoints: []resolver.Endpoint{
 			{
 				Addresses: []resolver.Address{
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr1),
-					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr2),
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr2, resolvedTargetTestAddr1),
+					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr2),
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr2, resolvedTargetTestAddr2),
 				},
 			},
 			{
 				Addresses: []resolver.Address{
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr3),
-					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr4),
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr2, resolvedTargetTestAddr3),
+					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr4),
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr2, resolvedTargetTestAddr4),
 				},
 			},
@@ -474,7 +474,7 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 // The test verifies that the delegating resolver combines unresolved proxy
 // host and target addresses correctly, returning addresses with the proxy host
 // populated and the target address included as an attribute.
-func (s) TestDelegatingResolverForMutipleProxyAddress(t *testing.T) {
+func (s) TestDelegatingResolverForMultipleProxyAddress(t *testing.T) {
 	const (
 		targetTestAddr          = "test.com"
 		resolvedTargetTestAddr1 = "1.1.1.1:8080"
@@ -510,10 +510,10 @@ func (s) TestDelegatingResolverForMutipleProxyAddress(t *testing.T) {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	proxyResolver.UpdateState(resolver.State{
+	targetResolver.UpdateState(resolver.State{
 		Addresses: []resolver.Address{
-			{Addr: resolvedProxyTestAddr1},
-			{Addr: resolvedProxyTestAddr2},
+			{Addr: resolvedTargetTestAddr1},
+			{Addr: resolvedTargetTestAddr2},
 		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
@@ -524,10 +524,10 @@ func (s) TestDelegatingResolverForMutipleProxyAddress(t *testing.T) {
 	case <-time.After(defaultTestShortTimeout):
 	}
 
-	targetResolver.UpdateState(resolver.State{
+	proxyResolver.UpdateState(resolver.State{
 		Addresses: []resolver.Address{
-			{Addr: resolvedTargetTestAddr1},
-			{Addr: resolvedTargetTestAddr2},
+			{Addr: resolvedProxyTestAddr1},
+			{Addr: resolvedProxyTestAddr2},
 		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
@@ -588,7 +588,11 @@ func (s) TestDelegatingResolverUpdateStateDuringClose(t *testing.T) {
 	// Set up a manual DNS resolver to control the proxy address resolution.
 	proxyResolver := setupDNS(t)
 
-	unblockProxyResolverClose := make(chan struct{})
+	proxyResolverBuilt := make(chan struct{})
+	proxyResolver.BuildCallback = func(resolver.Target, resolver.ClientConn, resolver.BuildOptions) {
+		close(proxyResolverBuilt)
+	}
+	unblockProxyResolverClose := make(chan struct{}, 1)
 	proxyResolver.CloseCallback = func() {
 		<-unblockProxyResolverClose
 		t.Log("Proxy resolver is closed.")
@@ -607,11 +611,17 @@ func (s) TestDelegatingResolverUpdateStateDuringClose(t *testing.T) {
 		Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1"}}}},
 	})
 
+	// Wait for the proxy resolver to be built before calling Close.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	select {
+	case <-proxyResolverBuilt:
+	case <-ctx.Done():
+		t.Fatalf("Context timed out waiting for proxy resolver to be built.")
+	}
 	// Closing the delegating resolver will block until the test writes to the
 	// unblockProxyResolverClose channel.
 	go dr.Close()
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	select {
 	case <-targetResolverCloseCalled:
 	case <-ctx.Done():
@@ -716,19 +726,23 @@ func (s) TestDelegatingResolverResolveNow(t *testing.T) {
 
 	// Manual resolver to control the target resolution.
 	targetResolver := manual.NewBuilderWithScheme("test")
-	targetResolverCalled := make(chan struct{})
+	targetResolverCalled := make(chan struct{}, 1)
 	targetResolver.ResolveNowCallback = func(resolver.ResolveNowOptions) {
 		// Updating the resolver state should not deadlock.
 		targetResolver.CC().UpdateState(resolver.State{
 			Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1"}}}},
 		})
-		close(targetResolverCalled)
+		targetResolverCalled <- struct{}{}
 	}
 
 	target := targetResolver.Scheme() + ":///" + "ignored"
 	// Set up a manual DNS resolver to control the proxy address resolution.
 	proxyResolver := setupDNS(t)
 	proxyResolverCalled := make(chan struct{})
+	proxyResolverBuilt := make(chan struct{})
+	proxyResolver.BuildCallback = func(resolver.Target, resolver.ClientConn, resolver.BuildOptions) {
+		close(proxyResolverBuilt)
+	}
 	proxyResolver.ResolveNowCallback = func(resolver.ResolveNowOptions) {
 		// Updating the resolver state should not deadlock.
 		proxyResolver.CC().UpdateState(resolver.State{
@@ -743,12 +757,28 @@ func (s) TestDelegatingResolverResolveNow(t *testing.T) {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	// Call ResolveNow on the delegatingResolver and verify both children
-	// receive the ResolveNow call.
+	// ResolveNow of manual proxy resolver will not be called. Proxy resolver is
+	// only built when we get the first update from target resolver. Therefore
+	// in the first ResolveNow, proxy resolver will be a no-op resolver and only
+	// target resolver's ResolveNow will be called.
 	dr.ResolveNow(resolver.ResolveNowOptions{})
-
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
+	select {
+	case <-targetResolverCalled:
+	case <-ctx.Done():
+		t.Fatalf("context timed out waiting for targetResolver.ResolveNow() to be called.")
+	}
+
+	// Wait for proxy resolver to be built.
+	select {
+	case <-proxyResolverBuilt:
+	case <-ctx.Done():
+		t.Fatalf("Timeout when waiting for proxy resolver to be built")
+	}
+
+	dr.ResolveNow(resolver.ResolveNowOptions{})
+
 	select {
 	case <-targetResolverCalled:
 	case <-ctx.Done():
@@ -760,3 +790,154 @@ func (s) TestDelegatingResolverResolveNow(t *testing.T) {
 		t.Fatalf("context timed out waiting for proxyResolver.ResolveNow() to be called.")
 	}
 }
+
+// Tests the scenario where a proxy is configured, and the resolver returns a
+// network type other than tcp for all addresses. The test verifies that the
+// delegating resolver avoids the proxy build and directly sends the update
+// from target resolver to clientconn.
+func (s) TestDelegatingResolverForNonTCPTarget(t *testing.T) {
+	const (
+		targetTestAddr          = "test.target"
+		resolvedTargetTestAddr1 = "1.1.1.1:8080"
+		envProxyAddr            = "proxytest.com"
+	)
+	hpfe := func(req *http.Request) (*url.URL, error) {
+		if req.URL.Host == targetTestAddr {
+			return &url.URL{
+				Scheme: "https",
+				Host:   envProxyAddr,
+			}, nil
+		}
+		t.Errorf("Unexpected request host to proxy: %s want %s", req.URL.Host, targetTestAddr)
+		return nil, nil
+	}
+	originalhpfe := delegatingresolver.HTTPSProxyFromEnvironment
+	delegatingresolver.HTTPSProxyFromEnvironment = hpfe
+	defer func() {
+		delegatingresolver.HTTPSProxyFromEnvironment = originalhpfe
+	}()
+
+	// Manual resolver to control the target resolution.
+	targetResolver := manual.NewBuilderWithScheme("test")
+	target := targetResolver.Scheme() + ":///" + targetTestAddr
+	// Set up a manual DNS resolver to control the proxy address resolution.
+	proxyResolver := setupDNS(t)
+	proxyBuildCalled := make(chan struct{})
+	proxyResolver.BuildCallback = func(resolver.Target, resolver.ClientConn, resolver.BuildOptions) {
+		close(proxyBuildCalled)
+	}
+
+	tcc, stateCh, _ := createTestResolverClientConn(t)
+	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
+		t.Fatalf("Failed to create delegating resolver: %v", err)
+	}
+
+	// Set network to anything other than tcp.
+	nonTCPAddr := networktype.Set(resolver.Address{Addr: resolvedTargetTestAddr1}, "unix")
+	targetResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{nonTCPAddr},
+		Endpoints:     []resolver.Endpoint{{Addresses: []resolver.Address{nonTCPAddr}}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	})
+
+	var gotState resolver.State
+	select {
+	case gotState = <-stateCh:
+	case <-time.After(defaultTestTimeout):
+		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	}
+
+	// Verify that the delegating resolver doesn't call proxy resolver's
+	// UpdateState since we have no tcp address
+	select {
+	case <-proxyBuildCalled:
+		t.Fatal("Unexpected call to proxy resolver update state")
+	case <-time.After(defaultTestShortTimeout):
+	}
+
+	wantState := resolver.State{
+		Addresses:     []resolver.Address{nonTCPAddr},
+		Endpoints:     []resolver.Endpoint{{Addresses: []resolver.Address{nonTCPAddr}}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	}
+
+	// Verify that the state clientconn receives is same as updated by target
+	// resolver, since we want to avoid proxy for any network type apart from
+	// tcp.
+	if diff := cmp.Diff(gotState, wantState); diff != "" {
+		t.Fatalf("Unexpected state from delegating resolver. Diff (-got +want):\n%s", diff)
+	}
+}
+
+// Tests the scenario where a proxy is configured, and the resolver returns tcp
+// and non-tcp addresses. The test verifies that the delegating resolver doesn't
+// add proxyatrribute to adresses with network type other than tcp , but adds
+// the proxyattribute to addresses with network type tcp.
+func (s) TestDelegatingResolverForMixNetworkType(t *testing.T) {
+	const (
+		targetTestAddr          = "test.target"
+		resolvedTargetTestAddr1 = "1.1.1.1:8080"
+		resolvedTargetTestAddr2 = "2.2.2.2:8080"
+		envProxyAddr            = "proxytest.com"
+	)
+	hpfe := func(req *http.Request) (*url.URL, error) {
+		if req.URL.Host == targetTestAddr {
+			return &url.URL{
+				Scheme: "https",
+				Host:   envProxyAddr,
+			}, nil
+		}
+		t.Errorf("Unexpected request host to proxy: %s want %s", req.URL.Host, targetTestAddr)
+		return nil, nil
+	}
+	originalhpfe := delegatingresolver.HTTPSProxyFromEnvironment
+	delegatingresolver.HTTPSProxyFromEnvironment = hpfe
+	defer func() {
+		delegatingresolver.HTTPSProxyFromEnvironment = originalhpfe
+	}()
+
+	// Manual resolver to control the target resolution.
+	targetResolver := manual.NewBuilderWithScheme("test")
+	target := targetResolver.Scheme() + ":///" + targetTestAddr
+	// Set up a manual DNS resolver to control the proxy address resolution.
+	proxyResolver := setupDNS(t)
+
+	tcc, stateCh, _ := createTestResolverClientConn(t)
+	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
+		t.Fatalf("Failed to create delegating resolver: %v", err)
+	}
+	// Set network to anything other than tcp.
+	nonTCPAddr := networktype.Set(resolver.Address{Addr: resolvedTargetTestAddr1}, "unix")
+	targetResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{nonTCPAddr, {Addr: resolvedTargetTestAddr2}},
+		Endpoints:     []resolver.Endpoint{{Addresses: []resolver.Address{nonTCPAddr, {Addr: resolvedTargetTestAddr2}}}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	})
+
+	select {
+	case <-stateCh:
+		t.Fatalf("Delegating resolver invoked UpdateState before both the proxy and target resolvers had updated their states.")
+	case <-time.After(defaultTestShortTimeout):
+	}
+
+	proxyResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: envProxyAddr}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	})
+
+	var gotState resolver.State
+	select {
+	case gotState = <-stateCh:
+	case <-time.After(defaultTestTimeout):
+		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	}
+	wantState := resolver.State{
+		Addresses:     []resolver.Address{nonTCPAddr, proxyAddressWithTargetAttribute(envProxyAddr, resolvedTargetTestAddr2)},
+		Endpoints:     []resolver.Endpoint{{Addresses: []resolver.Address{nonTCPAddr, proxyAddressWithTargetAttribute(envProxyAddr, resolvedTargetTestAddr2)}}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	}
+
+	if diff := cmp.Diff(gotState, wantState); diff != "" {
+		t.Fatalf("Unexpected state from delegating resolver. Diff (-got +want):\n%v", diff)
+	}
+}
diff --git a/internal/status/status_test.go b/internal/status/status_test.go
new file mode 100644
index 00000000..1d3fc4c4
--- /dev/null
+++ b/internal/status/status_test.go
@@ -0,0 +1,40 @@
+/*
+ *
+ * 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 status_test
+
+import (
+	"testing"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/internal/status"
+)
+
+func TestRawStatusProto(t *testing.T) {
+	spb := status.RawStatusProto(nil)
+	if spb != nil {
+		t.Errorf("RawStatusProto(nil) = %v; must return nil", spb)
+	}
+	s := status.New(codes.Internal, "test internal error")
+	spb1 := status.RawStatusProto(s)
+	spb2 := status.RawStatusProto(s)
+	// spb1 and spb2 should be the same pointer: no copies
+	if spb1 != spb2 {
+		t.Errorf("RawStatusProto(s)=%p then %p; must return the same pointer", spb1, spb2)
+	}
+}
diff --git a/internal/transport/http_util_test.go b/internal/transport/http_util_test.go
index 5a259d43..5eb466f3 100644
--- a/internal/transport/http_util_test.go
+++ b/internal/transport/http_util_test.go
@@ -211,9 +211,9 @@ func (s) TestParseDialTarget(t *testing.T) {
 		{"dns:///google.com", "tcp", "dns:///google.com"},
 		{"/unix/socket/address", "tcp", "/unix/socket/address"},
 	} {
-		gotNet, gotAddr := parseDialTarget(test.target)
+		gotNet, gotAddr := ParseDialTarget(test.target)
 		if gotNet != test.wantNet || gotAddr != test.wantAddr {
-			t.Errorf("parseDialTarget(%q) = %s, %s want %s, %s", test.target, gotNet, gotAddr, test.wantNet, test.wantAddr)
+			t.Errorf("ParseDialTarget(%q) = %s, %s want %s, %s", test.target, gotNet, gotAddr, test.wantNet, test.wantAddr)
 		}
 	}
 }
diff --git a/status/status_test.go b/status/status_test.go
index 03085d97..b0f1f4e7 100644
--- a/status/status_test.go
+++ b/status/status_test.go
@@ -354,6 +354,9 @@ func (s) TestStatus_ErrorDetails(t *testing.T) {
 			t.Fatalf("(%v).WithDetails(%+v) failed: %v", str(s), tc.details, err)
 		}
 		details := s.Details()
+		if len(details) != len(tc.details) {
+			t.Fatalf("len(s.Details()) = %d, want = %d.", len(details), len(tc.details))
+		}
 		for i := range details {
 			if !proto.Equal(details[i].(protoreflect.ProtoMessage), tc.details[i].(protoreflect.ProtoMessage)) {
 				t.Fatalf("(%v).Details()[%d] = %+v, want %+v", str(s), i, details[i], tc.details[i])
@@ -420,7 +423,7 @@ func (s) TestStatus_ErrorDetails_Fail(t *testing.T) {
 	for _, tc := range tests {
 		details := tc.s.Details()
 		if len(details) != len(tc.want) {
-			t.Fatalf("len(s.Details()) = %v, want = %v.", len(details), len(tc.want))
+			t.Fatalf("len(s.Details()) = %d, want = %d.", len(details), len(tc.want))
 		}
 		for i, d := range details {
 			// s.Details can either contain an error or a proto message.  We
diff --git a/xds/internal/clients/grpctransport/examples_test.go b/xds/internal/clients/grpctransport/examples_test.go
index 40b0cb27..57191d1f 100644
--- a/xds/internal/clients/grpctransport/examples_test.go
+++ b/xds/internal/clients/grpctransport/examples_test.go
@@ -30,10 +30,10 @@ import (
 // its extensions.
 //
 // This example is creating clients.ServerIdentifier to connect to server at
-// localhost:5678 using the credentials named "local". Note that "local" must
-// exist as an entry in the provided credentials to grpctransport.Builder.
+// localhost:5678 using the config named "local". Note that "local" must
+// exist as an entry in the provided configs to grpctransport.Builder.
 func ExampleServerIdentifierExtension() {
 	// Note the Extensions field is set by value and not by pointer.
-	fmt.Printf("%+v", clients.ServerIdentifier{ServerURI: "localhost:5678", Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"}})
-	// Output: {ServerURI:localhost:5678 Extensions:{Credentials:local}}
+	fmt.Printf("%+v", clients.ServerIdentifier{ServerURI: "localhost:5678", Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"}})
+	// Output: {ServerURI:localhost:5678 Extensions:{ConfigName:local}}
 }
diff --git a/xds/internal/clients/grpctransport/grpc_transport_ext_test.go b/xds/internal/clients/grpctransport/grpc_transport_ext_test.go
index b34a2057..75ac5146 100644
--- a/xds/internal/clients/grpctransport/grpc_transport_ext_test.go
+++ b/xds/internal/clients/grpctransport/grpc_transport_ext_test.go
@@ -72,14 +72,14 @@ func (s) TestBuild_Single(t *testing.T) {
 
 	serverID := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local": &testCredentials{transportCredentials: local.NewCredentials()},
+	configs := map[string]grpctransport.Config{
+		"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
 
 	// Calling Build() first time should create new gRPC transport.
-	builder := grpctransport.NewBuilder(credentials)
+	builder := grpctransport.NewBuilder(configs)
 	tr, err := builder.Build(serverID)
 	if err != nil {
 		t.Fatalf("Failed to build transport: %v", err)
@@ -173,19 +173,19 @@ func (s) TestBuild_Multiple(t *testing.T) {
 
 	serverID1 := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"},
 	}
 	serverID2 := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local":    &testCredentials{transportCredentials: local.NewCredentials()},
-		"insecure": insecure.NewBundle(),
+	configs := map[string]grpctransport.Config{
+		"local":    {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
+		"insecure": {Credentials: insecure.NewBundle()},
 	}
 
 	// Create two gRPC transports.
-	builder := grpctransport.NewBuilder(credentials)
+	builder := grpctransport.NewBuilder(configs)
 
 	tr1, err := builder.Build(serverID1)
 	if err != nil {
diff --git a/xds/internal/clients/grpctransport/grpc_transport_test.go b/xds/internal/clients/grpctransport/grpc_transport_test.go
index ab2cd7c1..23e75257 100644
--- a/xds/internal/clients/grpctransport/grpc_transport_test.go
+++ b/xds/internal/clients/grpctransport/grpc_transport_test.go
@@ -137,15 +137,15 @@ func (tc *testCredentials) PerRPCCredentials() credentials.PerRPCCredentials {
 // Transport in both cases when provided clients.ServerIdentifer is same
 // one of the existing transport or a new one.
 func (s) TestBuild_Success(t *testing.T) {
-	credentials := map[string]credentials.Bundle{
-		"local":    &testCredentials{transportCredentials: local.NewCredentials()},
-		"insecure": insecure.NewBundle(),
+	configs := map[string]Config{
+		"local":    {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
+		"insecure": {Credentials: insecure.NewBundle()},
 	}
-	b := NewBuilder(credentials)
+	b := NewBuilder(configs)
 
 	serverID1 := clients.ServerIdentifier{
 		ServerURI:  "server-address",
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
 	tr1, err := b.Build(serverID1)
 	if err != nil {
@@ -155,7 +155,7 @@ func (s) TestBuild_Success(t *testing.T) {
 
 	serverID2 := clients.ServerIdentifier{
 		ServerURI:  "server-address",
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
 	tr2, err := b.Build(serverID2)
 	if err != nil {
@@ -165,7 +165,7 @@ func (s) TestBuild_Success(t *testing.T) {
 
 	serverID3 := clients.ServerIdentifier{
 		ServerURI:  "server-address",
-		Extensions: ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 	tr3, err := b.Build(serverID3)
 	if err != nil {
@@ -191,7 +191,7 @@ func (s) TestBuild_Failure(t *testing.T) {
 			name: "ServerURI is empty",
 			serverID: clients.ServerIdentifier{
 				ServerURI:  "",
-				Extensions: ServerIdentifierExtension{Credentials: "local"},
+				Extensions: ServerIdentifierExtension{ConfigName: "local"},
 			},
 		},
 		{
@@ -206,26 +206,41 @@ func (s) TestBuild_Failure(t *testing.T) {
 			},
 		},
 		{
-			name: "ServerIdentifierExtension Credentials is nil",
+			name: "ServerIdentifierExtension without ConfigName",
 			serverID: clients.ServerIdentifier{
 				ServerURI:  "server-address",
 				Extensions: ServerIdentifierExtension{},
 			},
 		},
+		{
+			name: "ServerIdentifierExtension ConfigName is not present",
+			serverID: clients.ServerIdentifier{
+				ServerURI:  "server-address",
+				Extensions: ServerIdentifierExtension{ConfigName: "unknown"},
+			},
+		},
+		{
+			name: "ServerIdentifierExtension ConfigName maps to nil credentials",
+			serverID: clients.ServerIdentifier{
+				ServerURI:  "server-address",
+				Extensions: ServerIdentifierExtension{ConfigName: "nil-credentials"},
+			},
+		},
 		{
 			name: "ServerIdentifierExtension is added as pointer",
 			serverID: clients.ServerIdentifier{
 				ServerURI:  "server-address",
-				Extensions: &ServerIdentifierExtension{Credentials: "local"},
+				Extensions: &ServerIdentifierExtension{ConfigName: "local"},
 			},
 		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			credentials := map[string]credentials.Bundle{
-				"local": &testCredentials{transportCredentials: local.NewCredentials()},
+			configs := map[string]Config{
+				"local":           {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
+				"nil-credentials": {Credentials: nil},
 			}
-			b := NewBuilder(credentials)
+			b := NewBuilder(configs)
 			tr, err := b.Build(test.serverID)
 			if err == nil {
 				t.Fatalf("Build() succeeded, want error")
@@ -238,18 +253,19 @@ func (s) TestBuild_Failure(t *testing.T) {
 }
 
 // TestNewStream_Success verifies that NewStream() successfully creates a new
-// client stream for the server when provided a valid server URI.
+// client stream for the server when provided a valid server URI and a config
+// with valid credentials.
 func (s) TestNewStream_Success(t *testing.T) {
 	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
 
 	serverCfg := clients.ServerIdentifier{
 		ServerURI:  ts.address,
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local": &testCredentials{transportCredentials: local.NewCredentials()},
+	configs := map[string]Config{
+		"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
-	builder := NewBuilder(credentials)
+	builder := NewBuilder(configs)
 	transport, err := builder.Build(serverCfg)
 	if err != nil {
 		t.Fatalf("Failed to build transport: %v", err)
@@ -263,17 +279,65 @@ func (s) TestNewStream_Success(t *testing.T) {
 	}
 }
 
+// TestNewStream_Success_WithCustomGRPCNewClient verifies that NewStream()
+// successfully creates a new client stream for the server when provided a
+// valid server URI and a config with valid credentials and a custom gRPC
+// NewClient function.
+func (s) TestNewStream_Success_WithCustomGRPCNewClient(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
+
+	// Create a custom dialer function that will be used by the gRPC client.
+	customDialerCalled := make(chan struct{}, 1)
+	customGRPCNewClient := func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+		customDialerCalled <- struct{}{}
+		return grpc.NewClient(target, opts...)
+	}
+
+	configs := map[string]Config{
+		"custom-dialer-config": {
+			Credentials:   &testCredentials{transportCredentials: local.NewCredentials()},
+			GRPCNewClient: customGRPCNewClient,
+		},
+	}
+	builder := NewBuilder(configs)
+
+	serverID := clients.ServerIdentifier{
+		ServerURI:  ts.address,
+		Extensions: ServerIdentifierExtension{ConfigName: "custom-dialer-config"},
+	}
+
+	transport, err := builder.Build(serverID)
+	if err != nil {
+		t.Fatalf("builder.Build(%+v) failed: %v", serverID, err)
+	}
+	defer transport.Close()
+
+	select {
+	case <-customDialerCalled:
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for custom dialer to be called: %v", ctx.Err())
+	}
+
+	// Verify that the transport works by creating a stream.
+	if _, err = transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources"); err != nil {
+		t.Fatalf("transport.NewStream() failed with custom dialer: %v", err)
+	}
+}
+
 // TestNewStream_Error verifies that NewStream() returns an error
 // when attempting to create a stream with an invalid server URI.
 func (s) TestNewStream_Error(t *testing.T) {
 	serverCfg := clients.ServerIdentifier{
 		ServerURI:  "invalid-server-uri",
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local": &testCredentials{transportCredentials: local.NewCredentials()},
+	configs := map[string]Config{
+		"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
-	builder := NewBuilder(credentials)
+	builder := NewBuilder(configs)
 	transport, err := builder.Build(serverCfg)
 	if err != nil {
 		t.Fatalf("Failed to build transport: %v", err)
@@ -304,12 +368,12 @@ func (s) TestStream_SendAndRecv(t *testing.T) {
 	// Build a grpc-based transport to the above server.
 	serverCfg := clients.ServerIdentifier{
 		ServerURI:  ts.address,
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local": &testCredentials{transportCredentials: local.NewCredentials()},
+	configs := map[string]Config{
+		"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
-	builder := NewBuilder(credentials)
+	builder := NewBuilder(configs)
 	transport, err := builder.Build(serverCfg)
 	if err != nil {
 		t.Fatalf("Failed to build transport: %v", err)
diff --git a/xds/internal/clients/lrsclient/loadreport_test.go b/xds/internal/clients/lrsclient/loadreport_test.go
index 1c5072ac..bcdec9c6 100644
--- a/xds/internal/clients/lrsclient/loadreport_test.go
+++ b/xds/internal/clients/lrsclient/loadreport_test.go
@@ -28,7 +28,6 @@ import (
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/google/uuid"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/status"
@@ -132,17 +131,17 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
 	// reporting is per-server and not per-authority.
 	nodeID := uuid.New().String()
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	config := lrsclient.Config{
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 	}
 	client, err := lrsclient.New(config)
 	if err != nil {
 		t.Fatalf("lrsclient.New() failed: %v", err)
 	}
 
-	serverIdentifier1 := clients.ServerIdentifier{ServerURI: mgmtServer1.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	serverIdentifier1 := clients.ServerIdentifier{ServerURI: mgmtServer1.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore1, err := client.ReportLoad(serverIdentifier1)
 	if err != nil {
 		t.Fatalf("client.ReportLoad() failed: %v", err)
@@ -162,7 +161,7 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
 
 	// Call the load reporting API to report load to the first management
 	// server, and ensure that a connection to the server is created.
-	serverIdentifier2 := clients.ServerIdentifier{ServerURI: mgmtServer2.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	serverIdentifier2 := clients.ServerIdentifier{ServerURI: mgmtServer2.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore2, err := client.ReportLoad(serverIdentifier2)
 	if err != nil {
 		t.Fatalf("client.ReportLoad() failed: %v", err)
@@ -276,10 +275,10 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 	// Create an LRS client with configuration pointing to the above server.
 	nodeID := uuid.New().String()
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	config := lrsclient.Config{
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 	}
 	client, err := lrsclient.New(config)
 	if err != nil {
@@ -287,7 +286,7 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 	}
 
 	// Call the load reporting API, and ensure that an LRS stream is created.
-	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore1, err := client.ReportLoad(serverIdentifier)
 	if err != nil {
 		t.Fatalf("client.ReportLoad() failed: %v", err)
@@ -471,10 +470,10 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
 	// Create an LRS client with configuration pointing to the above server.
 	nodeID := uuid.New().String()
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	config := lrsclient.Config{
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 	}
 	client, err := lrsclient.New(config)
 	if err != nil {
@@ -482,7 +481,7 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
 	}
 
 	// Call the load reporting API, and ensure that an LRS stream is created.
-	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore, err := client.ReportLoad(serverIdentifier)
 	if err != nil {
 		t.Fatalf("client.ReportLoad() failed: %v", err)
diff --git a/xds/internal/clients/xdsclient/channel_test.go b/xds/internal/clients/xdsclient/channel_test.go
index d7417f31..1a9f37f8 100644
--- a/xds/internal/clients/xdsclient/channel_test.go
+++ b/xds/internal/clients/xdsclient/channel_test.go
@@ -29,7 +29,6 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -55,10 +54,10 @@ func xdsChannelForTest(t *testing.T, serverURI, nodeID string, watchExpiryTimeou
 	// Create a grpc transport to the above management server.
 	si := clients.ServerIdentifier{
 		ServerURI:  serverURI,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
-	tr, err := (grpctransport.NewBuilder(credentials)).Build(si)
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	tr, err := (grpctransport.NewBuilder(configs)).Build(si)
 	if err != nil {
 		t.Fatalf("Failed to create a transport for server config %v: %v", si, err)
 	}
diff --git a/xds/internal/clients/xdsclient/test/authority_test.go b/xds/internal/clients/xdsclient/test/authority_test.go
index 27fd32ef..35d86921 100644
--- a/xds/internal/clients/xdsclient/test/authority_test.go
+++ b/xds/internal/clients/xdsclient/test/authority_test.go
@@ -24,7 +24,6 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -81,7 +80,7 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.Liste
 	nodeID := uuid.New().String()
 
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
-	ext := grpctransport.ServerIdentifierExtension{Credentials: "insecure"}
+	ext := grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}
 	siDefault := clients.ServerIdentifier{
 		ServerURI:  defaultAuthorityServer.Address,
 		Extensions: ext,
@@ -91,11 +90,11 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.Liste
 		Extensions: ext,
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: siDefault}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -242,20 +241,20 @@ func (s) TestAuthority_Fallback(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	psi := clients.ServerIdentifier{
 		ServerURI:  primaryMgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 	ssi := clients.ServerIdentifier{
 		ServerURI:  secondaryMgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
 	// Create config with the above primary and fallback management servers,
 	// and an xDS client with that configuration.
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: psi}, {ServerIdentifier: ssi}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		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
diff --git a/xds/internal/clients/xdsclient/test/dump_test.go b/xds/internal/clients/xdsclient/test/dump_test.go
index 9bf34e95..49f09453 100644
--- a/xds/internal/clients/xdsclient/test/dump_test.go
+++ b/xds/internal/clients/xdsclient/test/dump_test.go
@@ -28,7 +28,6 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -124,14 +123,14 @@ func (s) TestDumpResources_ManyToOne(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		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
@@ -308,18 +307,18 @@ func (s) TestDumpResources_ManyToMany(t *testing.T) {
 	resourceTypes[xdsresource.V3ListenerURL] = listenerType
 	si1 := clients.ServerIdentifier{
 		ServerURI:  mgmtServer1.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 	si2 := clients.ServerIdentifier{
 		ServerURI:  mgmtServer2.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si1}},
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
diff --git a/xds/internal/clients/xdsclient/test/helpers_test.go b/xds/internal/clients/xdsclient/test/helpers_test.go
index d296fee6..16453c80 100644
--- a/xds/internal/clients/xdsclient/test/helpers_test.go
+++ b/xds/internal/clients/xdsclient/test/helpers_test.go
@@ -20,6 +20,7 @@ package xdsclient_test
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"strconv"
@@ -29,12 +30,14 @@ import (
 
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/xds/internal/clients/internal/pretty"
+	"google.golang.org/grpc/xds/internal/clients/internal/testutils"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
 	"google.golang.org/protobuf/proto"
 
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
+	"github.com/google/go-cmp/cmp"
 )
 
 type s struct {
@@ -262,3 +265,36 @@ func buildResourceName(typeName, auth, id string, ctxParams map[string]string) s
 		ContextParams: ctxParams,
 	}).String()
 }
+
+// testMetricsReporter is a MetricsReporter to be used in tests. It sends
+// recording events on channels and provides helpers to check if certain events
+// have taken place.
+type testMetricsReporter struct {
+	metricsCh *testutils.Channel
+}
+
+// newTestMetricsReporter returns a new testMetricsReporter.
+func newTestMetricsReporter() *testMetricsReporter {
+	return &testMetricsReporter{
+		metricsCh: testutils.NewChannelWithSize(1),
+	}
+}
+
+// waitForMetric waits for a metric to be recorded and verifies that the
+// recorded metrics data matches the expected metricsDataWant. Returns
+// an error if failed to wait or received wrong data.
+func (r *testMetricsReporter) waitForMetric(ctx context.Context, metricsDataWant any) error {
+	got, err := r.metricsCh.Receive(ctx)
+	if err != nil {
+		return fmt.Errorf("timeout waiting for int64Count")
+	}
+	if diff := cmp.Diff(got, metricsDataWant); diff != "" {
+		return fmt.Errorf("received unexpected metrics value (-got, +want): %v", diff)
+	}
+	return nil
+}
+
+// ReportMetric sends the metrics data to the metricsCh channel.
+func (r *testMetricsReporter) ReportMetric(m any) {
+	r.metricsCh.Send(m)
+}
diff --git a/xds/internal/clients/xdsclient/test/lds_watchers_test.go b/xds/internal/clients/xdsclient/test/lds_watchers_test.go
index 6bf32e73..5fe60424 100644
--- a/xds/internal/clients/xdsclient/test/lds_watchers_test.go
+++ b/xds/internal/clients/xdsclient/test/lds_watchers_test.go
@@ -28,7 +28,6 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -252,14 +251,14 @@ func (s) TestLDSWatch(t *testing.T) {
 			resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 			si := clients.ServerIdentifier{
 				ServerURI:  mgmtServer.Address,
-				Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+				Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 			}
 
-			credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+			configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 			xdsClientConfig := xdsclient.Config{
 				Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 				Node:             clients.Node{ID: nodeID},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				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
@@ -398,14 +397,14 @@ func (s) TestLDSWatch_TwoWatchesForSameResourceName(t *testing.T) {
 			resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 			si := clients.ServerIdentifier{
 				ServerURI:  mgmtServer.Address,
-				Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+				Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 			}
 
-			credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+			configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 			xdsClientConfig := xdsclient.Config{
 				Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 				Node:             clients.Node{ID: nodeID},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				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
@@ -503,14 +502,14 @@ func (s) TestLDSWatch_ThreeWatchesForDifferentResourceNames(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -610,14 +609,14 @@ func (s) TestLDSWatch_ResourceCaching(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -684,7 +683,7 @@ func (s) TestLDSWatch_ResourceCaching(t *testing.T) {
 // does not receive an LDS response for the request that it sends. The test
 // verifies that the watch callback is invoked with an error once the
 // watchExpiryTimer fires.
-func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
+func TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
 
 	nodeID := uuid.New().String()
@@ -692,14 +691,14 @@ func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -743,14 +742,14 @@ func (s) TestLDSWatch_ValidResponseCancelsExpiryTimerBehavior(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -820,14 +819,14 @@ func (s) TestLDSWatch_ResourceRemoved(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -949,14 +948,14 @@ func (s) TestLDSWatch_NewWatcherForRemovedResource(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -1035,14 +1034,14 @@ func (s) TestLDSWatch_NACKError(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -1103,14 +1102,14 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -1196,14 +1195,14 @@ func (s) TestLDSWatch_PartialValid(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -1286,14 +1285,14 @@ func (s) TestLDSWatch_PartialResponse(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
diff --git a/xds/internal/clients/xdsclient/test/metrics_test.go b/xds/internal/clients/xdsclient/test/metrics_test.go
new file mode 100644
index 00000000..00857361
--- /dev/null
+++ b/xds/internal/clients/xdsclient/test/metrics_test.go
@@ -0,0 +1,284 @@
+/*
+ *
+ * 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 xdsclient_test
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/google/uuid"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/grpc/xds/internal/clients"
+	"google.golang.org/grpc/xds/internal/clients/grpctransport"
+	"google.golang.org/grpc/xds/internal/clients/internal/testutils/e2e"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient/metrics"
+
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+)
+
+// TestResourceUpdateMetrics configures an xDS client, and a management server
+// to send valid and invalid LDS updates, and verifies that the expected metrics
+// for both good and bad updates are emitted.
+func (s) TestResourceUpdateMetrics(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout*1000)
+	defer cancel()
+
+	tmr := newTestMetricsReporter()
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: l})
+	const listenerResourceName = "test-listener-resource"
+	const routeConfigurationName = "test-route-configuration-resource"
+	nodeID := uuid.New().String()
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerResourceName, routeConfigurationName)},
+		SkipValidation: true,
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
+	}
+
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+		Node:             clients.Node{ID: nodeID},
+		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{}},
+		},
+		MetricsReporter: tmr,
+	}
+	// 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)
+	}
+	defer client.Close()
+
+	// Watch the valid listener configured on the management server. This should
+	// cause a resource update valid metric to emit eventually.
+	client.WatchResource(listenerType.TypeURL, listenerResourceName, noopListenerWatcher{})
+	if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Update management server with a bad update. This should cause a resource
+	// update invalid metric to emit eventually.
+	resources = e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerResourceName, routeConfigurationName)},
+		SkipValidation: true,
+	}
+	resources.Listeners[0].ApiListener = nil
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
+	}
+	if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateInvalid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Resource update valid metric should have not emitted.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	if err := tmr.waitForMetric(sCtx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err == nil {
+		t.Fatal("tmr.WaitForInt64Count(ctx, mdWant) succeeded when expected to timeout.")
+	}
+}
+
+// TestServerFailureMetrics_BeforeResponseRecv configures an xDS client, and a
+// management server. It then register a watcher and stops the management
+// server before sending a resource update, and verifies that the expected
+// metric for server failure is emitted.
+func (s) TestServerFailureMetrics_BeforeResponseRecv(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	tmr := newTestMetricsReporter()
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+
+	lis := testutils.NewRestartableListener(l)
+	streamOpened := make(chan struct{}, 1)
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		Listener: lis,
+		OnStreamOpen: func(context.Context, int64, string) error {
+			select {
+			case streamOpened <- struct{}{}:
+			default:
+			}
+			return nil
+		},
+	})
+
+	nodeID := uuid.New().String()
+
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+		Node:             clients.Node{ID: nodeID},
+		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{}},
+		},
+		MetricsReporter: tmr,
+	}
+	// 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)
+	}
+	defer client.Close()
+
+	const listenerResourceName = "test-listener-resource"
+
+	// Watch for the listener on the above management server.
+	client.WatchResource(listenerType.TypeURL, listenerResourceName, noopListenerWatcher{})
+	// Verify that an ADS stream is opened and an LDS request with the above
+	// resource name is sent.
+	select {
+	case <-streamOpened:
+	case <-ctx.Done():
+		t.Fatal("Timeout when waiting for ADS stream to open")
+	}
+
+	// Close the listener and ensure that the ADS stream breaks. This should
+	// cause a server failure metric to emit eventually.
+	lis.Stop()
+
+	// Restart to prevent the attempt to create a new ADS stream after back off.
+	lis.Restart()
+
+	if err := tmr.waitForMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil {
+		t.Fatal(err.Error())
+	}
+}
+
+// TestServerFailureMetrics_AfterResponseRecv configures an xDS client, and a
+// management server to send a valid LDS updates, and verifies that the
+// resource update valid metric is emitted. It then closes the management server
+// listener to close the ADS stream and verifies that the server failure metric
+// is not emitted because the ADS stream was closed after having received a
+// response on the stream.
+func (s) TestServerFailureMetrics_AfterResponseRecv(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	tmr := newTestMetricsReporter()
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+
+	lis := testutils.NewRestartableListener(l)
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: lis})
+	const listenerResourceName = "test-listener-resource"
+	const routeConfigurationName = "test-route-configuration-resource"
+	nodeID := uuid.New().String()
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerResourceName, routeConfigurationName)},
+		SkipValidation: true,
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
+	}
+
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+		Node:             clients.Node{ID: nodeID},
+		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{}},
+		},
+		MetricsReporter: tmr,
+	}
+	// 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)
+	}
+	defer client.Close()
+
+	// Watch the valid listener configured on the management server. This should
+	// cause a resource updates valid count to emit eventually.
+	client.WatchResource(listenerType.TypeURL, listenerResourceName, noopListenerWatcher{})
+	if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Close the listener and ensure that the ADS stream breaks. This should
+	// cause a server failure metric to emit eventually.
+	lis.Stop()
+	if ctx.Err() != nil {
+		t.Fatalf("Timeout when waiting for ADS stream to close")
+	}
+	// Restart to prevent the attempt to create a new ADS stream after back off.
+	lis.Restart()
+
+	// Server failure should not have emitted.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	if err := tmr.waitForMetric(sCtx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err == nil {
+		t.Fatal("tmr.WaitForInt64Count(ctx, mdWant) succeeded when expected to timeout.")
+	}
+}
diff --git a/xds/internal/clients/xdsclient/test/misc_watchers_test.go b/xds/internal/clients/xdsclient/test/misc_watchers_test.go
index 39702272..b6d74a56 100644
--- a/xds/internal/clients/xdsclient/test/misc_watchers_test.go
+++ b/xds/internal/clients/xdsclient/test/misc_watchers_test.go
@@ -26,7 +26,6 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -109,14 +108,14 @@ func (s) TestWatchCallAnotherWatch(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -220,14 +219,14 @@ func (s) TestNodeProtoSentOnlyInFirstRequest(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		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
@@ -385,14 +384,14 @@ func (s) TestWatchErrorsContainNodeID(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		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
@@ -483,7 +482,7 @@ func (s) TestWatchErrorsContainNodeID_ChannelCreationFailure(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
 	xdsClientConfig := xdsclient.Config{
diff --git a/xds/internal/clients/xdsclient/xdsclient_test.go b/xds/internal/clients/xdsclient/xdsclient_test.go
index 6c5c7b59..69b64006 100644
--- a/xds/internal/clients/xdsclient/xdsclient_test.go
+++ b/xds/internal/clients/xdsclient/xdsclient_test.go
@@ -22,7 +22,6 @@ import (
 	"strings"
 	"testing"
 
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -30,7 +29,7 @@ import (
 )
 
 func (s) TestXDSClient_New(t *testing.T) {
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 
 	tests := []struct {
 		name    string
@@ -62,7 +61,7 @@ func (s) TestXDSClient_New(t *testing.T) {
 			config: Config{
 				Node:             clients.Node{ID: "node-id"},
 				ResourceTypes:    map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				TransportBuilder: grpctransport.NewBuilder(configs),
 			},
 			wantErr: "no servers or authorities specified",
 		},
@@ -71,7 +70,7 @@ func (s) TestXDSClient_New(t *testing.T) {
 			config: Config{
 				Node:             clients.Node{ID: "node-id"},
 				ResourceTypes:    map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				TransportBuilder: grpctransport.NewBuilder(configs),
 				Servers:          []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}},
 			},
 			wantErr: "",
@@ -81,7 +80,7 @@ func (s) TestXDSClient_New(t *testing.T) {
 			config: Config{
 				Node:             clients.Node{ID: "node-id"},
 				ResourceTypes:    map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				TransportBuilder: grpctransport.NewBuilder(configs),
 				Authorities:      map[string]Authority{"authority-name": {XDSServers: []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}}}},
 			},
 			wantErr: "",
@@ -107,11 +106,11 @@ func (s) TestXDSClient_New(t *testing.T) {
 }
 
 func (s) TestXDSClient_Close(t *testing.T) {
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	config := Config{
 		Node:             clients.Node{ID: "node-id"},
 		ResourceTypes:    map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		Servers:          []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}},
 	}
 	c, err := New(config)