Repository: grpc/grpc-go

Issue:
Add a test for the initial channel state while waiting for the first name resolver update
I don't think this is currently tested, and I'm not clear whether we have a bug.  We should have a test that does the following:

1. Create a channel with a manual name resolver set to not produce any updates when constructed.
2. Confirm the channel's `State()` == `Idle`.
3. Two forks for this test:
   1. Call `ClientConn.Connect()` and confirm the channel's `State()` == `Connecting` immediately upon returning.  (It's possible this is racy but it should not be.)
   2. Perform an RPC, which will block since the name resolver never produced anything.  In a separate goroutine, confirm the state eventually changes to `Connecting` using `testutils.AwaitState`.
4. Call `internal.EnterIdleModeForTesting` and then perform (2) and (3) above.


Discussion:
@janardhanvissa: @arjan-bal Please assign me this issue.@dfawley: The test is done.  However it seems we have a bug, so the test is disabled.@easwars: Here is a section from a pending client channel specification that indicates that we should be transitioning to `Connecting` once we create the name resolver: https://github.com/grpc/grpc/blob/201ed14bd019d53961b486fe18a54f5a8660fd3f/doc/client_channel_specification.md#starting-name-resolution@easwars: The changes we need to make would include the following:
- Set the channel's connectivity state to `Connecting` right before attempting to create the resolver: https://github.com/grpc/grpc-go/blob/master/resolver_wrapper.go#L88-L92
- If building the resolver succeeds, an LB policy will be created when the resolver returns a valid update, and it will be the responsibility of the LB policy to determine the connectivity state of the channel going forward. This is how things work today as well.
- If building the resolver fails, we will put the channel in `TransientFailure`, and return the build error back to `Dial`. If `NewClient` was used, we would have to ensure that the build error is returned back to the user when RPCs are made.
- We also need to ensure in the case where `NewClient` is used, and the resolver build fails, that the channel transitions back to `Idle` after the idle timeout expires and the resolver is shut down. And a subsequent RPC or `Connect` should trigger the creation of the resolver again, and if this time the resolver build succeeds, the channel should be usable again.

https://github.com/grpc/grpc-go/pull/8643 is a previous attempt at doing a part of what is discussed here.@easwars: It looks like once resolver creation fails, and the channel enters `TransientFailure`, it doesn't seem to go back to `Idle` after idle timeout expires. 

Solution patch:
diff --git a/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go b/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
index 1ebf7cea..2fc0a71f 100644
--- a/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
+++ b/balancer/pickfirst/pickfirstleaf/pickfirstleaf.go
@@ -30,6 +30,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"net/netip"
 	"sync"
 	"time"
 
@@ -53,6 +54,10 @@ func init() {
 	balancer.Register(pickfirstBuilder{})
 }
 
+// enableHealthListenerKeyType is a unique key type used in resolver attributes
+// to indicate whether the health listener usage is enabled.
+type enableHealthListenerKeyType struct{}
+
 var (
 	logger = grpclog.Component("pick-first-leaf-lb")
 	// Name is the name of the pick_first_leaf balancer.
@@ -108,10 +113,8 @@ func (pickfirstBuilder) Build(cc balancer.ClientConn, bo balancer.BuildOptions)
 		target:          bo.Target.String(),
 		metricsRecorder: bo.MetricsRecorder, // ClientConn will always create a Metrics Recorder.
 
-		addressList:           addressList{},
 		subConns:              resolver.NewAddressMap(),
 		state:                 connectivity.Connecting,
-		mu:                    sync.Mutex{},
 		cancelConnectionTimer: func() {},
 	}
 	b.logger = internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf(logPrefix, b))
@@ -130,6 +133,13 @@ func (pickfirstBuilder) ParseConfig(js json.RawMessage) (serviceconfig.LoadBalan
 	return cfg, nil
 }
 
+// EnableHealthListener updates the state to configure pickfirst for using a
+// generic health listener.
+func EnableHealthListener(state resolver.State) resolver.State {
+	state.Attributes = state.Attributes.WithValue(enableHealthListenerKeyType{}, true)
+	return state
+}
+
 type pfConfig struct {
 	serviceconfig.LoadBalancingConfig `json:"-"`
 
@@ -147,15 +157,19 @@ type scData struct {
 	subConn balancer.SubConn
 	addr    resolver.Address
 
-	state                       connectivity.State
+	rawConnectivityState connectivity.State
+	// The effective connectivity state based on raw connectivity, health state
+	// and after following sticky TransientFailure behaviour defined in A62.
+	effectiveState              connectivity.State
 	lastErr                     error
 	connectionFailedInFirstPass bool
 }
 
 func (b *pickfirstBalancer) newSCData(addr resolver.Address) (*scData, error) {
 	sd := &scData{
-		state: connectivity.Idle,
-		addr:  addr,
+		rawConnectivityState: connectivity.Idle,
+		effectiveState:       connectivity.Idle,
+		addr:                 addr,
 	}
 	sc, err := b.cc.NewSubConn([]resolver.Address{addr}, balancer.NewSubConnOptions{
 		StateListener: func(state balancer.SubConnState) {
@@ -180,7 +194,9 @@ type pickfirstBalancer struct {
 	// The mutex is used to ensure synchronization of updates triggered
 	// from the idle picker and the already serialized resolver,
 	// SubConn state updates.
-	mu    sync.Mutex
+	mu sync.Mutex
+	// State reported to the channel based on SubConn states and resolver
+	// updates.
 	state connectivity.State
 	// scData for active subonns mapped by address.
 	subConns              *resolver.AddressMap
@@ -188,6 +204,7 @@ type pickfirstBalancer struct {
 	firstPass             bool
 	numTF                 int
 	cancelConnectionTimer func()
+	healthCheckingEnabled bool
 }
 
 // ResolverError is called by the ClientConn when the name resolver produces
@@ -213,7 +230,7 @@ func (b *pickfirstBalancer) resolverErrorLocked(err error) {
 		return
 	}
 
-	b.cc.UpdateState(balancer.State{
+	b.updateBalancerState(balancer.State{
 		ConnectivityState: connectivity.TransientFailure,
 		Picker:            &picker{err: fmt.Errorf("name resolver error: %v", err)},
 	})
@@ -226,12 +243,12 @@ func (b *pickfirstBalancer) UpdateClientConnState(state balancer.ClientConnState
 	if len(state.ResolverState.Addresses) == 0 && len(state.ResolverState.Endpoints) == 0 {
 		// Cleanup state pertaining to the previous resolver state.
 		// Treat an empty address list like an error by calling b.ResolverError.
-		b.state = connectivity.TransientFailure
 		b.closeSubConnsLocked()
 		b.addressList.updateAddrs(nil)
 		b.resolverErrorLocked(errors.New("produced zero addresses"))
 		return balancer.ErrBadResolverState
 	}
+	b.healthCheckingEnabled = state.ResolverState.Attributes.Value(enableHealthListenerKeyType{}) != nil
 	cfg, ok := state.BalancerConfig.(pfConfig)
 	if state.BalancerConfig != nil && !ok {
 		return fmt.Errorf("pickfirst: received illegal BalancerConfig (type %T): %v: %w", state.BalancerConfig, state.BalancerConfig, balancer.ErrBadResolverState)
@@ -278,12 +295,15 @@ func (b *pickfirstBalancer) UpdateClientConnState(state balancer.ClientConnState
 	newAddrs = deDupAddresses(newAddrs)
 	newAddrs = interleaveAddresses(newAddrs)
 
-	// If the previous ready SubConn exists in new address list,
-	// keep this connection and don't create new SubConns.
 	prevAddr := b.addressList.currentAddress()
+	prevSCData, found := b.subConns.Get(prevAddr)
 	prevAddrsCount := b.addressList.size()
+	isPrevRawConnectivityStateReady := found && prevSCData.(*scData).rawConnectivityState == connectivity.Ready
 	b.addressList.updateAddrs(newAddrs)
-	if b.state == connectivity.Ready && b.addressList.seekTo(prevAddr) {
+
+	// If the previous ready SubConn exists in new address list,
+	// keep this connection and don't create new SubConns.
+	if isPrevRawConnectivityStateReady && b.addressList.seekTo(prevAddr) {
 		return nil
 	}
 
@@ -295,10 +315,9 @@ func (b *pickfirstBalancer) UpdateClientConnState(state balancer.ClientConnState
 	// we should still enter CONNECTING because the sticky TF behaviour
 	//  mentioned in A62 applies only when the TRANSIENT_FAILURE is reported
 	// due to connectivity failures.
-	if b.state == connectivity.Ready || b.state == connectivity.Connecting || prevAddrsCount == 0 {
+	if isPrevRawConnectivityStateReady || b.state == connectivity.Connecting || prevAddrsCount == 0 {
 		// Start connection attempt at first address.
-		b.state = connectivity.Connecting
-		b.cc.UpdateState(balancer.State{
+		b.forceUpdateConcludedStateLocked(balancer.State{
 			ConnectivityState: connectivity.Connecting,
 			Picker:            &picker{err: balancer.ErrNoSubConnAvailable},
 		})
@@ -417,11 +436,14 @@ func addressFamily(address string) ipAddrFamily {
 	if err != nil {
 		return ipAddrFamilyUnknown
 	}
-	ip := net.ParseIP(host)
+	ip, err := netip.ParseAddr(host)
+	if err != nil {
+		return ipAddrFamilyUnknown
+	}
 	switch {
-	case ip.To4() != nil:
+	case ip.Is4() || ip.Is4In6():
 		return ipAddrFamilyV4
-	case ip.To16() != nil:
+	case ip.Is6():
 		return ipAddrFamilyV6
 	default:
 		return ipAddrFamilyUnknown
@@ -497,7 +519,7 @@ func (b *pickfirstBalancer) requestConnectionLocked() {
 		}
 
 		scd := sd.(*scData)
-		switch scd.state {
+		switch scd.rawConnectivityState {
 		case connectivity.Idle:
 			scd.subConn.Connect()
 			b.scheduleNextConnectionLocked()
@@ -515,7 +537,7 @@ func (b *pickfirstBalancer) requestConnectionLocked() {
 			b.scheduleNextConnectionLocked()
 			return
 		default:
-			b.logger.Errorf("SubConn with unexpected state %v present in SubConns map.", scd.state)
+			b.logger.Errorf("SubConn with unexpected state %v present in SubConns map.", scd.rawConnectivityState)
 			return
 
 		}
@@ -558,16 +580,17 @@ func (b *pickfirstBalancer) scheduleNextConnectionLocked() {
 func (b *pickfirstBalancer) updateSubConnState(sd *scData, newState balancer.SubConnState) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	oldState := sd.state
-	sd.state = newState.ConnectivityState
+	oldState := sd.rawConnectivityState
+	sd.rawConnectivityState = newState.ConnectivityState
 	// Previously relevant SubConns can still callback with state updates.
 	// To prevent pickers from returning these obsolete SubConns, this logic
 	// is included to check if the current list of active SubConns includes this
 	// SubConn.
-	if activeSD, found := b.subConns.Get(sd.addr); !found || activeSD != sd {
+	if !b.isActiveSCData(sd) {
 		return
 	}
 	if newState.ConnectivityState == connectivity.Shutdown {
+		sd.effectiveState = connectivity.Shutdown
 		return
 	}
 
@@ -586,10 +609,30 @@ func (b *pickfirstBalancer) updateSubConnState(sd *scData, newState balancer.Sub
 			b.logger.Errorf("Address %q not found address list in  %v", sd.addr, b.addressList.addresses)
 			return
 		}
-		b.state = connectivity.Ready
-		b.cc.UpdateState(balancer.State{
-			ConnectivityState: connectivity.Ready,
-			Picker:            &picker{result: balancer.PickResult{SubConn: sd.subConn}},
+		if !b.healthCheckingEnabled {
+			if b.logger.V(2) {
+				b.logger.Infof("SubConn %p reported connectivity state READY and the health listener is disabled. Transitioning SubConn to READY.", sd.subConn)
+			}
+
+			sd.effectiveState = connectivity.Ready
+			b.updateBalancerState(balancer.State{
+				ConnectivityState: connectivity.Ready,
+				Picker:            &picker{result: balancer.PickResult{SubConn: sd.subConn}},
+			})
+			return
+		}
+		if b.logger.V(2) {
+			b.logger.Infof("SubConn %p reported connectivity state READY. Registering health listener.", sd.subConn)
+		}
+		// Send a CONNECTING update to take the SubConn out of sticky-TF if
+		// required.
+		sd.effectiveState = connectivity.Connecting
+		b.updateBalancerState(balancer.State{
+			ConnectivityState: connectivity.Connecting,
+			Picker:            &picker{err: balancer.ErrNoSubConnAvailable},
+		})
+		sd.subConn.RegisterHealthListener(func(scs balancer.SubConnState) {
+			b.updateSubConnHealthState(sd, scs)
 		})
 		return
 	}
@@ -600,11 +643,13 @@ func (b *pickfirstBalancer) updateSubConnState(sd *scData, newState balancer.Sub
 	// a transport is successfully created, but the connection fails
 	// before the SubConn can send the notification for READY. We treat
 	// this as a successful connection and transition to IDLE.
-	if (b.state == connectivity.Ready && newState.ConnectivityState != connectivity.Ready) || (oldState == connectivity.Connecting && newState.ConnectivityState == connectivity.Idle) {
+	// TODO: https://github.com/grpc/grpc-go/issues/7862 - Remove the second
+	// part of the if condition below once the issue is fixed.
+	if oldState == connectivity.Ready || (oldState == connectivity.Connecting && newState.ConnectivityState == connectivity.Idle) {
 		// Once a transport fails, the balancer enters IDLE and starts from
 		// the first address when the picker is used.
 		b.shutdownRemainingLocked(sd)
-		b.state = connectivity.Idle
+		sd.effectiveState = newState.ConnectivityState
 		// READY SubConn interspliced in between CONNECTING and IDLE, need to
 		// account for that.
 		if oldState == connectivity.Connecting {
@@ -615,7 +660,7 @@ func (b *pickfirstBalancer) updateSubConnState(sd *scData, newState balancer.Sub
 		}
 		disconnectionsMetric.Record(b.metricsRecorder, 1, b.target)
 		b.addressList.reset()
-		b.cc.UpdateState(balancer.State{
+		b.updateBalancerState(balancer.State{
 			ConnectivityState: connectivity.Idle,
 			Picker:            &idlePicker{exitIdle: sync.OnceFunc(b.ExitIdle)},
 		})
@@ -625,19 +670,19 @@ func (b *pickfirstBalancer) updateSubConnState(sd *scData, newState balancer.Sub
 	if b.firstPass {
 		switch newState.ConnectivityState {
 		case connectivity.Connecting:
-			// The balancer can be in either IDLE, CONNECTING or
-			// TRANSIENT_FAILURE. If it's in TRANSIENT_FAILURE, stay in
+			// The effective state can be in either IDLE, CONNECTING or
+			// TRANSIENT_FAILURE. If it's  TRANSIENT_FAILURE, stay in
 			// TRANSIENT_FAILURE until it's READY. See A62.
-			// If the balancer is already in CONNECTING, no update is needed.
-			if b.state == connectivity.Idle {
-				b.state = connectivity.Connecting
-				b.cc.UpdateState(balancer.State{
+			if sd.effectiveState != connectivity.TransientFailure {
+				sd.effectiveState = connectivity.Connecting
+				b.updateBalancerState(balancer.State{
 					ConnectivityState: connectivity.Connecting,
 					Picker:            &picker{err: balancer.ErrNoSubConnAvailable},
 				})
 			}
 		case connectivity.TransientFailure:
 			sd.lastErr = newState.ConnectionError
+			sd.effectiveState = connectivity.TransientFailure
 			// Since we're re-using common SubConns while handling resolver
 			// updates, we could receive an out of turn TRANSIENT_FAILURE from
 			// a pass over the previous address list. Happy Eyeballs will also
@@ -664,7 +709,7 @@ func (b *pickfirstBalancer) updateSubConnState(sd *scData, newState balancer.Sub
 		b.numTF = (b.numTF + 1) % b.subConns.Len()
 		sd.lastErr = newState.ConnectionError
 		if b.numTF%b.subConns.Len() == 0 {
-			b.cc.UpdateState(balancer.State{
+			b.updateBalancerState(balancer.State{
 				ConnectivityState: connectivity.TransientFailure,
 				Picker:            &picker{err: newState.ConnectionError},
 			})
@@ -694,21 +739,79 @@ func (b *pickfirstBalancer) endFirstPassIfPossibleLocked(lastErr error) {
 		}
 	}
 	b.firstPass = false
-	b.state = connectivity.TransientFailure
-
-	b.cc.UpdateState(balancer.State{
+	b.updateBalancerState(balancer.State{
 		ConnectivityState: connectivity.TransientFailure,
 		Picker:            &picker{err: lastErr},
 	})
 	// Start re-connecting all the SubConns that are already in IDLE.
 	for _, v := range b.subConns.Values() {
 		sd := v.(*scData)
-		if sd.state == connectivity.Idle {
+		if sd.rawConnectivityState == connectivity.Idle {
 			sd.subConn.Connect()
 		}
 	}
 }
 
+func (b *pickfirstBalancer) isActiveSCData(sd *scData) bool {
+	activeSD, found := b.subConns.Get(sd.addr)
+	return found && activeSD == sd
+}
+
+func (b *pickfirstBalancer) updateSubConnHealthState(sd *scData, state balancer.SubConnState) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	// Previously relevant SubConns can still callback with state updates.
+	// To prevent pickers from returning these obsolete SubConns, this logic
+	// is included to check if the current list of active SubConns includes
+	// this SubConn.
+	if !b.isActiveSCData(sd) {
+		return
+	}
+	sd.effectiveState = state.ConnectivityState
+	switch state.ConnectivityState {
+	case connectivity.Ready:
+		b.updateBalancerState(balancer.State{
+			ConnectivityState: connectivity.Ready,
+			Picker:            &picker{result: balancer.PickResult{SubConn: sd.subConn}},
+		})
+	case connectivity.TransientFailure:
+		b.updateBalancerState(balancer.State{
+			ConnectivityState: connectivity.TransientFailure,
+			Picker:            &picker{err: fmt.Errorf("pickfirst: health check failure: %v", state.ConnectionError)},
+		})
+	case connectivity.Connecting:
+		b.updateBalancerState(balancer.State{
+			ConnectivityState: connectivity.Connecting,
+			Picker:            &picker{err: balancer.ErrNoSubConnAvailable},
+		})
+	default:
+		b.logger.Errorf("Got unexpected health update for SubConn %p: %v", state)
+	}
+}
+
+// updateBalancerState stores the state reported to the channel and calls
+// ClientConn.UpdateState(). As an optimization, it avoids sending duplicate
+// updates to the channel.
+func (b *pickfirstBalancer) updateBalancerState(newState balancer.State) {
+	// In case of TransientFailures allow the picker to be updated to update
+	// the connectivity error, in all other cases don't send duplicate state
+	// updates.
+	if newState.ConnectivityState == b.state && b.state != connectivity.TransientFailure {
+		return
+	}
+	b.forceUpdateConcludedStateLocked(newState)
+}
+
+// forceUpdateConcludedStateLocked stores the state reported to the channel and
+// calls ClientConn.UpdateState().
+// A separate function is defined to force update the ClientConn state since the
+// channel doesn't correctly assume that LB policies start in CONNECTING and
+// relies on LB policy to send an initial CONNECTING update.
+func (b *pickfirstBalancer) forceUpdateConcludedStateLocked(newState balancer.State) {
+	b.state = newState.ConnectivityState
+	b.cc.UpdateState(newState)
+}
+
 type picker struct {
 	result balancer.PickResult
 	err    error
diff --git a/credentials/google/google.go b/credentials/google/google.go
index d18e370d..5a9c9461 100644
--- a/credentials/google/google.go
+++ b/credentials/google/google.go
@@ -30,6 +30,8 @@ import (
 	"google.golang.org/grpc/internal"
 )
 
+const defaultCloudPlatformScope = "https://www.googleapis.com/auth/cloud-platform"
+
 var logger = grpclog.Component("credentials")
 
 // DefaultCredentialsOptions constructs options to build DefaultCredentials.
@@ -120,7 +122,7 @@ var (
 		return alts.NewClientCreds(alts.DefaultClientOptions())
 	}
 	newADC = func(ctx context.Context) (credentials.PerRPCCredentials, error) {
-		return oauth.NewApplicationDefault(ctx)
+		return oauth.NewApplicationDefault(ctx, defaultCloudPlatformScope)
 	}
 )
 
diff --git a/examples/features/csm_observability/client/main.go b/examples/features/csm_observability/client/main.go
index 20b357c2..cf33e6cb 100644
--- a/examples/features/csm_observability/client/main.go
+++ b/examples/features/csm_observability/client/main.go
@@ -29,6 +29,7 @@ import (
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	xdscreds "google.golang.org/grpc/credentials/xds"
 	"google.golang.org/grpc/examples/features/proto/echo"
 	"google.golang.org/grpc/stats/opentelemetry"
 	"google.golang.org/grpc/stats/opentelemetry/csm"
@@ -56,7 +57,13 @@ func main() {
 	cleanup := csm.EnableObservability(context.Background(), opentelemetry.Options{MetricsOptions: opentelemetry.MetricsOptions{MeterProvider: provider}})
 	defer cleanup()
 
-	cc, err := grpc.NewClient(*target, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	// Set up xds credentials that fall back to insecure as described in:
+	// https://cloud.google.com/service-mesh/docs/service-routing/security-proxyless-setup#workloads_are_unable_to_communicate_in_the_security_setup.
+	creds, err := xdscreds.NewClientCredentials(xdscreds.ClientOptions{FallbackCreds: insecure.NewCredentials()})
+	if err != nil {
+		log.Fatalf("Failed to create xDS credentials: %v", err)
+	}
+	cc, err := grpc.NewClient(*target, grpc.WithTransportCredentials(creds))
 	if err != nil {
 		log.Fatalf("Failed to start NewClient: %v", err)
 	}
diff --git a/examples/features/csm_observability/server/main.go b/examples/features/csm_observability/server/main.go
index b87f859a..3cafe233 100644
--- a/examples/features/csm_observability/server/main.go
+++ b/examples/features/csm_observability/server/main.go
@@ -28,9 +28,12 @@ import (
 	"net/http"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	xdscreds "google.golang.org/grpc/credentials/xds"
 	pb "google.golang.org/grpc/examples/features/proto/echo"
 	"google.golang.org/grpc/stats/opentelemetry"
 	"google.golang.org/grpc/stats/opentelemetry/csm"
+	"google.golang.org/grpc/xds"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.opentelemetry.io/otel/exporters/prometheus"
@@ -67,7 +70,16 @@ func main() {
 	if err != nil {
 		log.Fatalf("Failed to listen: %v", err)
 	}
-	s := grpc.NewServer()
+	// Set up xds credentials that fall back to insecure as described in:
+	// https://cloud.google.com/service-mesh/docs/service-routing/security-proxyless-setup#workloads_are_unable_to_communicate_in_the_security_setup.
+	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{FallbackCreds: insecure.NewCredentials()})
+	if err != nil {
+		log.Fatalf("Failed to create xDS credentials: %v", err)
+	}
+	s, err := xds.NewGRPCServer(grpc.Creds(creds))
+	if err != nil {
+		log.Fatalf("Failed to start xDS Server: %v", err)
+	}
 	pb.RegisterEchoServer(s, &echoServer{addr: ":" + *port})
 
 	log.Printf("Serving on %s\n", *port)
diff --git a/experimental/stats/metricregistry.go b/experimental/stats/metricregistry.go
index 1d827dd5..ad75313a 100644
--- a/experimental/stats/metricregistry.go
+++ b/experimental/stats/metricregistry.go
@@ -23,6 +23,7 @@ import (
 
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/stats"
 )
 
 func init() {
@@ -34,7 +35,7 @@ var logger = grpclog.Component("metrics-registry")
 // DefaultMetrics are the default metrics registered through global metrics
 // registry. This is written to at initialization time only, and is read only
 // after initialization.
-var DefaultMetrics = NewMetrics()
+var DefaultMetrics = stats.NewMetricSet()
 
 // MetricDescriptor is the data for a registered metric.
 type MetricDescriptor struct {
@@ -42,7 +43,7 @@ type MetricDescriptor struct {
 	// (including any per call metrics). See
 	// https://github.com/grpc/proposal/blob/master/A79-non-per-call-metrics-architecture.md#metric-instrument-naming-conventions
 	// for metric naming conventions.
-	Name Metric
+	Name string
 	// The description of this metric.
 	Description string
 	// The unit (e.g. entries, seconds) of this metric.
@@ -154,27 +155,27 @@ func (h *Int64GaugeHandle) Record(recorder MetricsRecorder, incr int64, labels .
 }
 
 // registeredMetrics are the registered metric descriptor names.
-var registeredMetrics = make(map[Metric]bool)
+var registeredMetrics = make(map[string]bool)
 
 // metricsRegistry contains all of the registered metrics.
 //
 // This is written to only at init time, and read only after that.
-var metricsRegistry = make(map[Metric]*MetricDescriptor)
+var metricsRegistry = make(map[string]*MetricDescriptor)
 
 // DescriptorForMetric returns the MetricDescriptor from the global registry.
 //
 // Returns nil if MetricDescriptor not present.
-func DescriptorForMetric(metric Metric) *MetricDescriptor {
-	return metricsRegistry[metric]
+func DescriptorForMetric(metricName string) *MetricDescriptor {
+	return metricsRegistry[metricName]
 }
 
-func registerMetric(name Metric, def bool) {
-	if registeredMetrics[name] {
-		logger.Fatalf("metric %v already registered", name)
+func registerMetric(metricName string, def bool) {
+	if registeredMetrics[metricName] {
+		logger.Fatalf("metric %v already registered", metricName)
 	}
-	registeredMetrics[name] = true
+	registeredMetrics[metricName] = true
 	if def {
-		DefaultMetrics = DefaultMetrics.Add(name)
+		DefaultMetrics = DefaultMetrics.Add(metricName)
 	}
 }
 
@@ -256,8 +257,8 @@ func snapshotMetricsRegistryForTesting() func() {
 	oldRegisteredMetrics := registeredMetrics
 	oldMetricsRegistry := metricsRegistry
 
-	registeredMetrics = make(map[Metric]bool)
-	metricsRegistry = make(map[Metric]*MetricDescriptor)
+	registeredMetrics = make(map[string]bool)
+	metricsRegistry = make(map[string]*MetricDescriptor)
 	maps.Copy(registeredMetrics, registeredMetrics)
 	maps.Copy(metricsRegistry, metricsRegistry)
 
diff --git a/experimental/stats/metrics.go b/experimental/stats/metrics.go
index 3221f7a6..bf9e7f98 100644
--- a/experimental/stats/metrics.go
+++ b/experimental/stats/metrics.go
@@ -19,8 +19,6 @@
 // Package stats contains experimental metrics/stats API's.
 package stats
 
-import "maps"
-
 // MetricsRecorder records on metrics derived from metric registry.
 type MetricsRecorder interface {
 	// RecordInt64Count records the measurement alongside labels on the int
@@ -39,76 +37,3 @@ type MetricsRecorder interface {
 	// gauge associated with the provided handle.
 	RecordInt64Gauge(handle *Int64GaugeHandle, incr int64, labels ...string)
 }
-
-// Metric is an identifier for a metric.
-type Metric string
-
-// Metrics is a set of metrics to record. Once created, Metrics is immutable,
-// however Add and Remove can make copies with specific metrics added or
-// removed, respectively.
-//
-// Do not construct directly; use NewMetrics instead.
-type Metrics struct {
-	// metrics are the set of metrics to initialize.
-	metrics map[Metric]bool
-}
-
-// NewMetrics returns a Metrics containing Metrics.
-func NewMetrics(metrics ...Metric) *Metrics {
-	newMetrics := make(map[Metric]bool)
-	for _, metric := range metrics {
-		newMetrics[metric] = true
-	}
-	return &Metrics{
-		metrics: newMetrics,
-	}
-}
-
-// Metrics returns the metrics set. The returned map is read-only and must not
-// be modified.
-func (m *Metrics) Metrics() map[Metric]bool {
-	return m.metrics
-}
-
-// Add adds the metrics to the metrics set and returns a new copy with the
-// additional metrics.
-func (m *Metrics) Add(metrics ...Metric) *Metrics {
-	newMetrics := make(map[Metric]bool)
-	for metric := range m.metrics {
-		newMetrics[metric] = true
-	}
-
-	for _, metric := range metrics {
-		newMetrics[metric] = true
-	}
-	return &Metrics{
-		metrics: newMetrics,
-	}
-}
-
-// Join joins the metrics passed in with the metrics set, and returns a new copy
-// with the merged metrics.
-func (m *Metrics) Join(metrics *Metrics) *Metrics {
-	newMetrics := make(map[Metric]bool)
-	maps.Copy(newMetrics, m.metrics)
-	maps.Copy(newMetrics, metrics.metrics)
-	return &Metrics{
-		metrics: newMetrics,
-	}
-}
-
-// Remove removes the metrics from the metrics set and returns a new copy with
-// the metrics removed.
-func (m *Metrics) Remove(metrics ...Metric) *Metrics {
-	newMetrics := make(map[Metric]bool)
-	for metric := range m.metrics {
-		newMetrics[metric] = true
-	}
-
-	for _, metric := range metrics {
-		delete(newMetrics, metric)
-	}
-	return &Metrics{
-		metrics: newMetrics,
-	}
-}
diff --git a/internal/resolver/dns/dns_resolver.go b/internal/resolver/dns/dns_resolver.go
index b080ae30..ba5c5a95 100644
--- a/internal/resolver/dns/dns_resolver.go
+++ b/internal/resolver/dns/dns_resolver.go
@@ -26,6 +26,7 @@ import (
 	"fmt"
 	rand "math/rand/v2"
 	"net"
+	"net/netip"
 	"os"
 	"strconv"
 	"strings"
@@ -122,7 +123,7 @@ func (b *dnsBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts
 	}
 
 	// IP address.
-	if ipAddr, ok := formatIP(host); ok {
+	if ipAddr, err := formatIP(host); err == nil {
 		addr := []resolver.Address{{Addr: ipAddr + ":" + port}}
 		cc.UpdateState(resolver.State{Addresses: addr})
 		return deadResolver{}, nil
@@ -260,9 +261,9 @@ func (d *dnsResolver) lookupSRV(ctx context.Context) ([]resolver.Address, error)
 			return nil, err
 		}
 		for _, a := range lbAddrs {
-			ip, ok := formatIP(a)
-			if !ok {
-				return nil, fmt.Errorf("dns: error parsing A record IP address %v", a)
+			ip, err := formatIP(a)
+			if err != nil {
+				return nil, fmt.Errorf("dns: error parsing A record IP address %v: %v", a, err)
 			}
 			addr := ip + ":" + strconv.Itoa(int(s.Port))
 			newAddrs = append(newAddrs, resolver.Address{Addr: addr, ServerName: s.Target})
@@ -322,9 +323,9 @@ func (d *dnsResolver) lookupHost(ctx context.Context) ([]resolver.Address, error
 	}
 	newAddrs := make([]resolver.Address, 0, len(addrs))
 	for _, a := range addrs {
-		ip, ok := formatIP(a)
-		if !ok {
-			return nil, fmt.Errorf("dns: error parsing A record IP address %v", a)
+		ip, err := formatIP(a)
+		if err != nil {
+			return nil, fmt.Errorf("dns: error parsing A record IP address %v: %v", a, err)
 		}
 		addr := ip + ":" + d.port
 		newAddrs = append(newAddrs, resolver.Address{Addr: addr})
@@ -351,19 +352,19 @@ func (d *dnsResolver) lookup() (*resolver.State, error) {
 	return &state, nil
 }
 
-// formatIP returns ok = false if addr is not a valid textual representation of
-// an IP address. If addr is an IPv4 address, return the addr and ok = true.
+// formatIP returns an error if addr is not a valid textual representation of
+// an IP address. If addr is an IPv4 address, return the addr and error = nil.
 // If addr is an IPv6 address, return the addr enclosed in square brackets and
-// ok = true.
-func formatIP(addr string) (addrIP string, ok bool) {
-	ip := net.ParseIP(addr)
-	if ip == nil {
-		return "", false
+// error = nil.
+func formatIP(addr string) (string, error) {
+	ip, err := netip.ParseAddr(addr)
+	if err != nil {
+		return "", err
 	}
-	if ip.To4() != nil {
-		return addr, true
+	if ip.Is4() {
+		return addr, nil
 	}
-	return "[" + addr + "]", true
+	return "[" + addr + "]", nil
 }
 
 // parseTarget takes the user input target string and default port, returns
@@ -379,7 +380,7 @@ func parseTarget(target, defaultPort string) (host, port string, err error) {
 	if target == "" {
 		return "", "", internal.ErrMissingAddr
 	}
-	if ip := net.ParseIP(target); ip != nil {
+	if _, err := netip.ParseAddr(target); err == nil {
 		// target is an IPv4 or IPv6(without brackets) address
 		return target, defaultPort, nil
 	}
diff --git a/internal/xds/rbac/matchers.go b/internal/xds/rbac/matchers.go
index e1c15018..07bb59ce 100644
--- a/internal/xds/rbac/matchers.go
+++ b/internal/xds/rbac/matchers.go
@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"net/netip"
 	"regexp"
 
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@@ -344,7 +345,8 @@ func newRemoteIPMatcher(cidrRange *v3corepb.CidrRange) (*remoteIPMatcher, error)
 }
 
 func (sim *remoteIPMatcher) match(data *rpcData) bool {
-	return sim.ipNet.Contains(net.IP(net.ParseIP(data.peerInfo.Addr.String())))
+	ip, _ := netip.ParseAddr(data.peerInfo.Addr.String())
+	return sim.ipNet.Contains(net.IP(ip.AsSlice()))
 }
 
 type localIPMatcher struct {
@@ -361,7 +363,8 @@ func newLocalIPMatcher(cidrRange *v3corepb.CidrRange) (*localIPMatcher, error) {
 }
 
 func (dim *localIPMatcher) match(data *rpcData) bool {
-	return dim.ipNet.Contains(net.IP(net.ParseIP(data.localAddr.String())))
+	ip, _ := netip.ParseAddr(data.localAddr.String())
+	return dim.ipNet.Contains(net.IP(ip.AsSlice()))
 }
 
 // portMatcher matches on whether the destination port of the RPC matches the
diff --git a/scripts/common.sh b/scripts/common.sh
index dede0789..3e641bae 100755
--- a/scripts/common.sh
+++ b/scripts/common.sh
@@ -11,6 +11,13 @@ not() {
   ! "$@"
 }
 
+# noret_grep will return 0 if zero or more lines were selected, and >1 if an
+# error occurred. Suppresses grep's return code of 1 when there are no matches
+# (for eg, empty file).
+noret_grep() {
+  grep "$@" || [[ $? == 1 ]]
+}
+
 die() {
   echo "$@" >&2
   exit 1
diff --git a/scripts/gen-deps.sh b/scripts/gen-deps.sh
index bc647c4d..f4d5ec63 100755
--- a/scripts/gen-deps.sh
+++ b/scripts/gen-deps.sh
@@ -3,6 +3,8 @@
 set -e          # Exit on error
 set -o pipefail # Fail a pipe if any sub-command fails.
 
+source "$(dirname $0)/common.sh"
+
 if [[ "$#" -ne 1 || ! -d "$1" ]]; then
     echo "Specify a valid output directory as the first parameter."
     exit 1
@@ -16,6 +18,6 @@ cd "${SCRIPTS_DIR}/.."
 git ls-files -- '*.go' | grep -v '\(^\|/\)\(internal\|examples\|benchmark\|interop\|test\|testdata\)\(/\|$\)' | xargs dirname | sort -u | while read d; do
   pushd "$d" > /dev/null
   pkg="$(echo "$d" | sed 's;\.;grpc;' | sed 's;/;_;g')"
-  go list -deps . | sort >| "${OUTPUT_DIR}/$pkg"
+  go list -deps . | sort | noret_grep -v 'google.golang.org/grpc' >| "${OUTPUT_DIR}/$pkg"
   popd > /dev/null
 done
diff --git a/scripts/vet.sh b/scripts/vet.sh
index 8db7e19c..4867e8c6 100755
--- a/scripts/vet.sh
+++ b/scripts/vet.sh
@@ -8,13 +8,6 @@ source "$(dirname $0)/common.sh"
 # Check to make sure it's safe to modify the user's git repo.
 git status --porcelain | fail_on_output
 
-# noret_grep will return 0 if zero or more lines were selected, and >1 if an
-# error occurred. Suppresses grep's return code of 1 when there are no matches
-# (for eg, empty file).
-noret_grep() {
-  grep "$@" || [[ $? == 1 ]]
-}
-
 # Undo any edits made by this script.
 cleanup() {
   git reset --hard HEAD
@@ -89,6 +82,10 @@ git grep '"github.com/envoyproxy/go-control-plane/envoy' -- '*.go' ':(exclude)*.
 git grep -e 'context.Background()' --or -e 'context.TODO()' -- "*_test.go" | grep -v "benchmark/primitives/context_test.go" | grep -v "credential
 s/google" | grep -v "internal/transport/" | grep -v "xds/internal/" | grep -v "security/advancedtls" | grep -v 'context.WithTimeout(' | not grep -v 'context.WithCancel('
 
+# Disallow usage of net.ParseIP in favour of netip.ParseAddr as the former
+# can't parse link local IPv6 addresses.
+not git grep 'net.ParseIP' -- '*.go'
+
 misspell -error .
 
 # - gofmt, goimports, go vet, go mod tidy.
diff --git a/stats/metrics.go b/stats/metrics.go
new file mode 100644
index 00000000..641c8e97
--- /dev/null
+++ b/stats/metrics.go
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2024 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 stats
+
+import "maps"
+
+// MetricSet is a set of metrics to record. Once created, MetricSet is immutable,
+// however Add and Remove can make copies with specific metrics added or
+// removed, respectively.
+//
+// Do not construct directly; use NewMetricSet instead.
+type MetricSet struct {
+	// metrics are the set of metrics to initialize.
+	metrics map[string]bool
+}
+
+// NewMetricSet returns a MetricSet containing metricNames.
+func NewMetricSet(metricNames ...string) *MetricSet {
+	newMetrics := make(map[string]bool)
+	for _, metric := range metricNames {
+		newMetrics[metric] = true
+	}
+	return &MetricSet{metrics: newMetrics}
+}
+
+// Metrics returns the metrics set. The returned map is read-only and must not
+// be modified.
+func (m *MetricSet) Metrics() map[string]bool {
+	return m.metrics
+}
+
+// Add adds the metricNames to the metrics set and returns a new copy with the
+// additional metrics.
+func (m *MetricSet) Add(metricNames ...string) *MetricSet {
+	newMetrics := make(map[string]bool)
+	for metric := range m.metrics {
+		newMetrics[metric] = true
+	}
+
+	for _, metric := range metricNames {
+		newMetrics[metric] = true
+	}
+	return &MetricSet{metrics: newMetrics}
+}
+
+// Join joins the metrics passed in with the metrics set, and returns a new copy
+// with the merged metrics.
+func (m *MetricSet) Join(metrics *MetricSet) *MetricSet {
+	newMetrics := make(map[string]bool)
+	maps.Copy(newMetrics, m.metrics)
+	maps.Copy(newMetrics, metrics.metrics)
+	return &MetricSet{metrics: newMetrics}
+}
+
+// Remove removes the metricNames from the metrics set and returns a new copy
+// with the metrics removed.
+func (m *MetricSet) Remove(metricNames ...string) *MetricSet {
+	newMetrics := make(map[string]bool)
+	for metric := range m.metrics {
+		newMetrics[metric] = true
+	}
+
+	for _, metric := range metricNames {
+		delete(newMetrics, metric)
+	}
+	return &MetricSet{metrics: newMetrics}
+}
diff --git a/stats/opentelemetry/client_metrics.go b/stats/opentelemetry/client_metrics.go
index 4af7f933..265791e5 100644
--- a/stats/opentelemetry/client_metrics.go
+++ b/stats/opentelemetry/client_metrics.go
@@ -260,18 +260,19 @@ func (h *clientStatsHandler) processRPCEnd(ctx context.Context, ai *attemptInfo,
 }
 
 const (
-	// ClientAttemptStarted is the number of client call attempts started.
-	ClientAttemptStarted estats.Metric = "grpc.client.attempt.started"
-	// ClientAttemptDuration is the end-to-end time taken to complete a client
-	// call attempt.
-	ClientAttemptDuration estats.Metric = "grpc.client.attempt.duration"
-	// ClientAttemptSentCompressedTotalMessageSize is the compressed message
-	// bytes sent per client call attempt.
-	ClientAttemptSentCompressedTotalMessageSize estats.Metric = "grpc.client.attempt.sent_total_compressed_message_size"
-	// ClientAttemptRcvdCompressedTotalMessageSize is the compressed message
-	// bytes received per call attempt.
-	ClientAttemptRcvdCompressedTotalMessageSize estats.Metric = "grpc.client.attempt.rcvd_total_compressed_message_size"
-	// ClientCallDuration is the time taken by gRPC to complete an RPC from
-	// application's perspective.
-	ClientCallDuration estats.Metric = "grpc.client.call.duration"
+	// ClientAttemptStartedMetricName is the number of client call attempts
+	// started.
+	ClientAttemptStartedMetricName string = "grpc.client.attempt.started"
+	// ClientAttemptDurationMetricName is the end-to-end time taken to complete
+	// a client call attempt.
+	ClientAttemptDurationMetricName string = "grpc.client.attempt.duration"
+	// ClientAttemptSentCompressedTotalMessageSizeMetricName is the compressed
+	// message bytes sent per client call attempt.
+	ClientAttemptSentCompressedTotalMessageSizeMetricName string = "grpc.client.attempt.sent_total_compressed_message_size"
+	// ClientAttemptRcvdCompressedTotalMessageSizeMetricName is the compressed
+	// message bytes received per call attempt.
+	ClientAttemptRcvdCompressedTotalMessageSizeMetricName string = "grpc.client.attempt.rcvd_total_compressed_message_size"
+	// ClientCallDurationMetricName is the time taken by gRPC to complete an RPC
+	// from application's perspective.
+	ClientCallDurationMetricName string = "grpc.client.call.duration"
 )
diff --git a/stats/opentelemetry/opentelemetry.go b/stats/opentelemetry/opentelemetry.go
index cc5ad387..dcc42477 100644
--- a/stats/opentelemetry/opentelemetry.go
+++ b/stats/opentelemetry/opentelemetry.go
@@ -16,6 +16,10 @@
 
 // Package opentelemetry implements opentelemetry instrumentation code for
 // gRPC-Go clients and servers.
+//
+// For details on configuring opentelemetry and various instruments that this
+// package creates, see
+// [gRPC OpenTelemetry Metrics](https://grpc.io/docs/guides/opentelemetry-metrics/).
 package opentelemetry
 
 import (
@@ -28,6 +32,7 @@ import (
 	estats "google.golang.org/grpc/experimental/stats"
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/stats"
 	otelinternal "google.golang.org/grpc/stats/opentelemetry/internal"
 
 	otelattribute "go.opentelemetry.io/otel/attribute"
@@ -57,27 +62,28 @@ type Options struct {
 type MetricsOptions struct {
 	// MeterProvider is the MeterProvider instance that will be used to create
 	// instruments. To enable metrics collection, set a meter provider. If
-	// unset, no metrics will be recorded. Any implementation knobs (i.e. views,
-	// bounds) set in the MeterProvider take precedence over the API calls from
-	// this interface. (i.e. it will create default views for unset views).
+	// unset, no metrics will be recorded.
 	MeterProvider otelmetric.MeterProvider
 
 	// Metrics are the metrics to instrument. Will create instrument and record telemetry
 	// for corresponding metric supported by the client and server
 	// instrumentation components if applicable. If not set, the default metrics
 	// will be recorded.
-	Metrics *estats.Metrics
-
-	// MethodAttributeFilter is to record the method name of RPCs handled by
-	// grpc.UnknownServiceHandler, but take care to limit the values allowed, as
-	// allowing too many will increase cardinality and could cause severe memory
-	// or performance problems. On Client Side, pass a
-	// grpc.StaticMethodCallOption as a call option into Invoke or NewStream.
-	// This only applies for server side metrics.
+	Metrics *stats.MetricSet
+
+	// MethodAttributeFilter is a function that determines whether to record the
+	// method name of RPCs as an attribute, or to bucket into "other". Take care
+	// to limit the values allowed, as allowing too many will increase
+	// cardinality and could cause severe memory or performance problems.
+	//
+	// This only applies for server-side metrics.  For clients, to record the
+	// method name in the attributes, pass grpc.StaticMethodCallOption to Invoke
+	// or NewStream. Note that when using protobuf generated clients, this
+	// CallOption is included automatically.
 	MethodAttributeFilter func(string) bool
 
-	// OptionalLabels are labels received from LB Policies that this component
-	// should add to metrics that record after receiving incoming metadata.
+	// OptionalLabels specifies a list of optional labels to enable on any
+	// metrics that support them.
 	OptionalLabels []string
 
 	// pluginOption is used to get labels to attach to certain metrics, if set.
@@ -207,7 +213,7 @@ type serverMetrics struct {
 	callDuration otelmetric.Float64Histogram
 }
 
-func createInt64Counter(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Int64CounterOption) otelmetric.Int64Counter {
+func createInt64Counter(setOfMetrics map[string]bool, metricName string, meter otelmetric.Meter, options ...otelmetric.Int64CounterOption) otelmetric.Int64Counter {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Int64Counter{}
 	}
@@ -219,7 +225,7 @@ func createInt64Counter(setOfMetrics map[estats.Metric]bool, metricName estats.M
 	return ret
 }
 
-func createFloat64Counter(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Float64CounterOption) otelmetric.Float64Counter {
+func createFloat64Counter(setOfMetrics map[string]bool, metricName string, meter otelmetric.Meter, options ...otelmetric.Float64CounterOption) otelmetric.Float64Counter {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Float64Counter{}
 	}
@@ -231,7 +237,7 @@ func createFloat64Counter(setOfMetrics map[estats.Metric]bool, metricName estats
 	return ret
 }
 
-func createInt64Histogram(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Int64HistogramOption) otelmetric.Int64Histogram {
+func createInt64Histogram(setOfMetrics map[string]bool, metricName string, meter otelmetric.Meter, options ...otelmetric.Int64HistogramOption) otelmetric.Int64Histogram {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Int64Histogram{}
 	}
@@ -243,7 +249,7 @@ func createInt64Histogram(setOfMetrics map[estats.Metric]bool, metricName estats
 	return ret
 }
 
-func createFloat64Histogram(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Float64HistogramOption) otelmetric.Float64Histogram {
+func createFloat64Histogram(setOfMetrics map[string]bool, metricName string, meter otelmetric.Meter, options ...otelmetric.Float64HistogramOption) otelmetric.Float64Histogram {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Float64Histogram{}
 	}
@@ -255,7 +261,7 @@ func createFloat64Histogram(setOfMetrics map[estats.Metric]bool, metricName esta
 	return ret
 }
 
-func createInt64Gauge(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Int64GaugeOption) otelmetric.Int64Gauge {
+func createInt64Gauge(setOfMetrics map[string]bool, metricName string, meter otelmetric.Meter, options ...otelmetric.Int64GaugeOption) otelmetric.Int64Gauge {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Int64Gauge{}
 	}
@@ -298,7 +304,7 @@ type registryMetrics struct {
 	optionalLabels []string
 }
 
-func (rm *registryMetrics) registerMetrics(metrics *estats.Metrics, meter otelmetric.Meter) {
+func (rm *registryMetrics) registerMetrics(metrics *stats.MetricSet, meter otelmetric.Meter) {
 	rm.intCounts = make(map[*estats.MetricDescriptor]otelmetric.Int64Counter)
 	rm.floatCounts = make(map[*estats.MetricDescriptor]otelmetric.Float64Counter)
 	rm.intHistos = make(map[*estats.MetricDescriptor]otelmetric.Int64Histogram)
@@ -379,12 +385,12 @@ var (
 	// DefaultSizeBounds are the default bounds for metrics which record size.
 	DefaultSizeBounds = []float64{0, 1024, 2048, 4096, 16384, 65536, 262144, 1048576, 4194304, 16777216, 67108864, 268435456, 1073741824, 4294967296}
 	// defaultPerCallMetrics are the default metrics provided by this module.
-	defaultPerCallMetrics = estats.NewMetrics(ClientAttemptStarted, ClientAttemptDuration, ClientAttemptSentCompressedTotalMessageSize, ClientAttemptRcvdCompressedTotalMessageSize, ClientCallDuration, ServerCallStarted, ServerCallSentCompressedTotalMessageSize, ServerCallRcvdCompressedTotalMessageSize, ServerCallDuration)
+	defaultPerCallMetrics = stats.NewMetricSet(ClientAttemptStartedMetricName, ClientAttemptDurationMetricName, ClientAttemptSentCompressedTotalMessageSizeMetricName, ClientAttemptRcvdCompressedTotalMessageSizeMetricName, ClientCallDurationMetricName, ServerCallStartedMetricName, ServerCallSentCompressedTotalMessageSizeMetricName, ServerCallRcvdCompressedTotalMessageSizeMetricName, ServerCallDurationMetricName)
 )
 
 // DefaultMetrics returns a set of default OpenTelemetry metrics.
 //
 // This should only be invoked after init time.
-func DefaultMetrics() *estats.Metrics {
+func DefaultMetrics() *stats.MetricSet {
 	return defaultPerCallMetrics.Join(estats.DefaultMetrics)
 }
diff --git a/stats/opentelemetry/server_metrics.go b/stats/opentelemetry/server_metrics.go
index eaea559b..4765afa8 100644
--- a/stats/opentelemetry/server_metrics.go
+++ b/stats/opentelemetry/server_metrics.go
@@ -264,15 +264,15 @@ func (h *serverStatsHandler) processRPCEnd(ctx context.Context, ai *attemptInfo,
 }
 
 const (
-	// ServerCallStarted is the number of server calls started.
-	ServerCallStarted estats.Metric = "grpc.server.call.started"
-	// ServerCallSentCompressedTotalMessageSize is the compressed message bytes
-	// sent per server call.
-	ServerCallSentCompressedTotalMessageSize estats.Metric = "grpc.server.call.sent_total_compressed_message_size"
-	// ServerCallRcvdCompressedTotalMessageSize is the compressed message bytes
-	// received per server call.
-	ServerCallRcvdCompressedTotalMessageSize estats.Metric = "grpc.server.call.rcvd_total_compressed_message_size"
-	// ServerCallDuration is the end-to-end time taken to complete a call from
-	// server transport's perspective.
-	ServerCallDuration estats.Metric = "grpc.server.call.duration"
+	// ServerCallStartedMetricName is the number of server calls started.
+	ServerCallStartedMetricName string = "grpc.server.call.started"
+	// ServerCallSentCompressedTotalMessageSizeMetricName is the compressed
+	// message bytes sent per server call.
+	ServerCallSentCompressedTotalMessageSizeMetricName string = "grpc.server.call.sent_total_compressed_message_size"
+	// ServerCallRcvdCompressedTotalMessageSizeMetricName is the compressed
+	// message bytes received per server call.
+	ServerCallRcvdCompressedTotalMessageSizeMetricName string = "grpc.server.call.rcvd_total_compressed_message_size"
+	// ServerCallDurationMetricName is the end-to-end time taken to complete a
+	// call from server transport's perspective.
+	ServerCallDurationMetricName string = "grpc.server.call.duration"
 )



Test patch:
diff --git a/balancer/pickfirst/pickfirst_ext_test.go b/balancer/pickfirst/pickfirst_ext_test.go
index faa6e7dd..4354009f 100644
--- a/balancer/pickfirst/pickfirst_ext_test.go
+++ b/balancer/pickfirst/pickfirst_ext_test.go
@@ -376,12 +376,12 @@ func (s) TestPickFirst_StickyTransientFailure(t *testing.T) {
 			},
 		}),
 	}
-	cc, err := grpc.Dial(lis.Addr().String(), dopts...)
+	cc, err := grpc.NewClient(lis.Addr().String(), dopts...)
 	if err != nil {
-		t.Fatalf("Failed to dial server at %q: %v", lis.Addr(), err)
+		t.Fatalf("Failed to create new client: %v", err)
 	}
 	t.Cleanup(func() { cc.Close() })
-
+	cc.Connect()
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
 
 	// Spawn a goroutine to ensure that the channel stays in TransientFailure.
@@ -837,12 +837,12 @@ func (s) TestPickFirst_ResolverError_WithPreviousUpdate_Connecting(t *testing.T)
 		grpc.WithResolvers(r),
 		grpc.WithDefaultServiceConfig(pickFirstServiceConfig),
 	}
-	cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
+	cc, err := grpc.NewClient(r.Scheme()+":///test.server", dopts...)
 	if err != nil {
-		t.Fatalf("grpc.Dial() failed: %v", err)
+		t.Fatalf("grpc.NewClient() failed: %v", err)
 	}
 	t.Cleanup(func() { cc.Close() })
-
+	cc.Connect()
 	addrs := []resolver.Address{{Addr: lis.Addr().String()}}
 	r.UpdateState(resolver.State{Addresses: addrs})
 	testutils.AwaitState(ctx, t, cc, connectivity.Connecting)
@@ -892,12 +892,12 @@ func (s) TestPickFirst_ResolverError_WithPreviousUpdate_TransientFailure(t *test
 		grpc.WithResolvers(r),
 		grpc.WithDefaultServiceConfig(pickFirstServiceConfig),
 	}
-	cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
+	cc, err := grpc.NewClient(r.Scheme()+":///test.server", dopts...)
 	if err != nil {
-		t.Fatalf("grpc.Dial() failed: %v", err)
+		t.Fatalf("grpc.NewClient() failed: %v", err)
 	}
 	t.Cleanup(func() { cc.Close() })
-
+	cc.Connect()
 	addrs := []resolver.Address{{Addr: lis.Addr().String()}}
 	r.UpdateState(resolver.State{Addresses: addrs})
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
diff --git a/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go b/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
index 00715724..9667c2b3 100644
--- a/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
+++ b/balancer/pickfirst/pickfirstleaf/pickfirstleaf_ext_test.go
@@ -34,6 +34,7 @@ import (
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/balancer/stub"
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/stubserver"
@@ -67,20 +68,54 @@ func Test(t *testing.T) {
 	grpctest.RunSubTests(t, s{})
 }
 
+// testServer is a server than can be stopped and resumed without closing
+// the listener. This guarantees the same port number (and address) is used
+// after restart. When a server is stopped, it accepts and closes all tcp
+// connections from clients.
+type testServer struct {
+	stubserver.StubServer
+	lis *testutils.RestartableListener
+}
+
+func (s *testServer) stop() {
+	s.lis.Stop()
+}
+
+func (s *testServer) resume() {
+	s.lis.Restart()
+}
+
+func newTestServer(t *testing.T) *testServer {
+	l, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to create listener: %v", err)
+	}
+	rl := testutils.NewRestartableListener(l)
+	ss := stubserver.StubServer{
+		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) { return &testpb.Empty{}, nil },
+		Listener:   rl,
+	}
+	return &testServer{
+		StubServer: ss,
+		lis:        rl,
+	}
+}
+
 // setupPickFirstLeaf performs steps required for pick_first tests. It starts a
 // bunch of backends exporting the TestService, and creates a ClientConn to them.
 func setupPickFirstLeaf(t *testing.T, backendCount int, opts ...grpc.DialOption) (*grpc.ClientConn, *manual.Resolver, *backendManager) {
 	t.Helper()
 	r := manual.NewBuilderWithScheme("whatever")
-	backends := make([]*stubserver.StubServer, backendCount)
+	backends := make([]*testServer, backendCount)
 	addrs := make([]resolver.Address, backendCount)
 
 	for i := 0; i < backendCount; i++ {
-		backend := stubserver.StartTestService(t, nil)
+		server := newTestServer(t)
+		backend := stubserver.StartTestService(t, &server.StubServer)
 		t.Cleanup(func() {
 			backend.Stop()
 		})
-		backends[i] = backend
+		backends[i] = server
 		addrs[i] = resolver.Address{Addr: backend.Address}
 	}
 
@@ -264,8 +299,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_DisjointLists(t *testing.T) {
 	stateSubscriber := &ccStateSubscriber{}
 	internal.SubscribeToConnectivityStateChanges.(func(cc *grpc.ClientConn, s grpcsync.Subscriber) func())(cc, stateSubscriber)
 
-	bm.backends[0].S.Stop()
-	bm.backends[0].S = nil
+	bm.backends[0].stop()
 	r.UpdateState(resolver.State{Addresses: []resolver.Address{addrs[0], addrs[1]}})
 	var bal *stateStoringBalancer
 	select {
@@ -287,8 +321,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_DisjointLists(t *testing.T) {
 		t.Errorf("SubConn states mismatch (-want +got):\n%s", diff)
 	}
 
-	bm.backends[2].S.Stop()
-	bm.backends[2].S = nil
+	bm.backends[2].stop()
 	r.UpdateState(resolver.State{Addresses: []resolver.Address{addrs[2], addrs[3]}})
 
 	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[3]); err != nil {
@@ -327,8 +360,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_ActiveBackendInUpdatedList(t *testing
 	stateSubscriber := &ccStateSubscriber{}
 	internal.SubscribeToConnectivityStateChanges.(func(cc *grpc.ClientConn, s grpcsync.Subscriber) func())(cc, stateSubscriber)
 
-	bm.backends[0].S.Stop()
-	bm.backends[0].S = nil
+	bm.backends[0].stop()
 	r.UpdateState(resolver.State{Addresses: []resolver.Address{addrs[0], addrs[1]}})
 	var bal *stateStoringBalancer
 	select {
@@ -350,8 +382,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_ActiveBackendInUpdatedList(t *testing
 		t.Errorf("SubConn states mismatch (-want +got):\n%s", diff)
 	}
 
-	bm.backends[2].S.Stop()
-	bm.backends[2].S = nil
+	bm.backends[2].stop()
 	r.UpdateState(resolver.State{Addresses: []resolver.Address{addrs[2], addrs[1]}})
 
 	// Verify that the ClientConn stays in READY.
@@ -391,8 +422,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_InActiveBackendInUpdatedList(t *testi
 	stateSubscriber := &ccStateSubscriber{}
 	internal.SubscribeToConnectivityStateChanges.(func(cc *grpc.ClientConn, s grpcsync.Subscriber) func())(cc, stateSubscriber)
 
-	bm.backends[0].S.Stop()
-	bm.backends[0].S = nil
+	bm.backends[0].stop()
 	r.UpdateState(resolver.State{Addresses: []resolver.Address{addrs[0], addrs[1]}})
 	var bal *stateStoringBalancer
 	select {
@@ -414,11 +444,9 @@ func (s) TestPickFirstLeaf_ResolverUpdates_InActiveBackendInUpdatedList(t *testi
 		t.Errorf("SubConn states mismatch (-want +got):\n%s", diff)
 	}
 
-	bm.backends[2].S.Stop()
-	bm.backends[2].S = nil
-	if err := bm.backends[0].StartServer(); err != nil {
-		t.Fatalf("Failed to re-start test backend: %v", err)
-	}
+	bm.backends[2].stop()
+	bm.backends[0].resume()
+
 	r.UpdateState(resolver.State{Addresses: []resolver.Address{addrs[0], addrs[2]}})
 
 	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[0]); err != nil {
@@ -456,8 +484,7 @@ func (s) TestPickFirstLeaf_ResolverUpdates_IdenticalLists(t *testing.T) {
 	stateSubscriber := &ccStateSubscriber{}
 	internal.SubscribeToConnectivityStateChanges.(func(cc *grpc.ClientConn, s grpcsync.Subscriber) func())(cc, stateSubscriber)
 
-	bm.backends[0].S.Stop()
-	bm.backends[0].S = nil
+	bm.backends[0].stop()
 	r.UpdateState(resolver.State{Addresses: []resolver.Address{addrs[0], addrs[1]}})
 	var bal *stateStoringBalancer
 	select {
@@ -554,14 +581,11 @@ func (s) TestPickFirstLeaf_StopConnectedServer_FirstServerRestart(t *testing.T)
 	}
 
 	// Shut down the connected server.
-	bm.backends[0].S.Stop()
-	bm.backends[0].S = nil
+	bm.backends[0].stop()
 	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
 
 	// Start the new target server.
-	if err := bm.backends[0].StartServer(); err != nil {
-		t.Fatalf("Failed to start server: %v", err)
-	}
+	bm.backends[0].resume()
 
 	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[0]); err != nil {
 		t.Fatal(err)
@@ -620,14 +644,11 @@ func (s) TestPickFirstLeaf_StopConnectedServer_SecondServerRestart(t *testing.T)
 	}
 
 	// Shut down the connected server.
-	bm.backends[1].S.Stop()
-	bm.backends[1].S = nil
+	bm.backends[1].stop()
 	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
 
 	// Start the new target server.
-	if err := bm.backends[1].StartServer(); err != nil {
-		t.Fatalf("Failed to start server: %v", err)
-	}
+	bm.backends[1].resume()
 
 	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[1]); err != nil {
 		t.Fatal(err)
@@ -692,14 +713,11 @@ func (s) TestPickFirstLeaf_StopConnectedServer_SecondServerToFirst(t *testing.T)
 	}
 
 	// Shut down the connected server.
-	bm.backends[1].S.Stop()
-	bm.backends[1].S = nil
+	bm.backends[1].stop()
 	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
 
 	// Start the new target server.
-	if err := bm.backends[0].StartServer(); err != nil {
-		t.Fatalf("Failed to start server: %v", err)
-	}
+	bm.backends[0].resume()
 
 	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[0]); err != nil {
 		t.Fatal(err)
@@ -763,14 +781,11 @@ func (s) TestPickFirstLeaf_StopConnectedServer_FirstServerToSecond(t *testing.T)
 	}
 
 	// Shut down the connected server.
-	bm.backends[0].S.Stop()
-	bm.backends[0].S = nil
+	bm.backends[0].stop()
 	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
 
 	// Start the new target server.
-	if err := bm.backends[1].StartServer(); err != nil {
-		t.Fatalf("Failed to start server: %v", err)
-	}
+	bm.backends[1].resume()
 
 	if err := pickfirst.CheckRPCsToBackend(ctx, cc, addrs[1]); err != nil {
 		t.Fatal(err)
@@ -1091,7 +1106,7 @@ func (s) TestPickFirstLeaf_InterleavingIPV4Preffered(t *testing.T) {
 				{Addresses: []resolver.Address{{Addr: "[::FFFF:192.168.0.1]:2222"}}},
 				{Addresses: []resolver.Address{{Addr: "[0001:0001:0001:0001:0001:0001:0001:0001]:8080"}}},
 				{Addresses: []resolver.Address{{Addr: "[0002:0002:0002:0002:0002:0002:0002:0002]:8080"}}},
-				{Addresses: []resolver.Address{{Addr: "[0003:0003:0003:0003:0003:0003:0003:0003]:3333"}}},
+				{Addresses: []resolver.Address{{Addr: "[fe80::1%eth0]:3333"}}},
 				{Addresses: []resolver.Address{{Addr: "grpc.io:80"}}}, // not an IP.
 			},
 		},
@@ -1107,7 +1122,7 @@ func (s) TestPickFirstLeaf_InterleavingIPV4Preffered(t *testing.T) {
 		{Addr: "2.2.2.2:2"},
 		{Addr: "[0002:0002:0002:0002:0002:0002:0002:0002]:8080"},
 		{Addr: "3.3.3.3:3"},
-		{Addr: "[0003:0003:0003:0003:0003:0003:0003:0003]:3333"},
+		{Addr: "[fe80::1%eth0]:3333"},
 		{Addr: "[::FFFF:192.168.0.1]:2222"},
 	}
 
@@ -1135,7 +1150,7 @@ func (s) TestPickFirstLeaf_InterleavingIPv6Preffered(t *testing.T) {
 				{Addresses: []resolver.Address{{Addr: "3.3.3.3:3"}}},
 				{Addresses: []resolver.Address{{Addr: "[::FFFF:192.168.0.1]:2222"}}},
 				{Addresses: []resolver.Address{{Addr: "[0002:0002:0002:0002:0002:0002:0002:0002]:2222"}}},
-				{Addresses: []resolver.Address{{Addr: "[0003:0003:0003:0003:0003:0003:0003:0003]:3333"}}},
+				{Addresses: []resolver.Address{{Addr: "[fe80::1%eth0]:3333"}}},
 				{Addresses: []resolver.Address{{Addr: "grpc.io:80"}}}, // not an IP.
 			},
 		},
@@ -1150,7 +1165,7 @@ func (s) TestPickFirstLeaf_InterleavingIPv6Preffered(t *testing.T) {
 		{Addr: "grpc.io:80"},
 		{Addr: "[0002:0002:0002:0002:0002:0002:0002:0002]:2222"},
 		{Addr: "2.2.2.2:2"},
-		{Addr: "[0003:0003:0003:0003:0003:0003:0003:0003]:3333"},
+		{Addr: "[fe80::1%eth0]:3333"},
 		{Addr: "3.3.3.3:3"},
 		{Addr: "[::FFFF:192.168.0.1]:2222"},
 	}
@@ -1180,7 +1195,7 @@ func (s) TestPickFirstLeaf_InterleavingUnknownPreffered(t *testing.T) {
 				{Addresses: []resolver.Address{{Addr: "[::FFFF:192.168.0.1]:2222"}}},
 				{Addresses: []resolver.Address{{Addr: "[0001:0001:0001:0001:0001:0001:0001:0001]:8080"}}},
 				{Addresses: []resolver.Address{{Addr: "[0002:0002:0002:0002:0002:0002:0002:0002]:8080"}}},
-				{Addresses: []resolver.Address{{Addr: "[0003:0003:0003:0003:0003:0003:0003:0003]:3333"}}},
+				{Addresses: []resolver.Address{{Addr: "[fe80::1%eth0]:3333"}}},
 				{Addresses: []resolver.Address{{Addr: "example.com:80"}}}, // not an IP.
 			},
 		},
@@ -1197,7 +1212,7 @@ func (s) TestPickFirstLeaf_InterleavingUnknownPreffered(t *testing.T) {
 		{Addr: "2.2.2.2:2"},
 		{Addr: "[0002:0002:0002:0002:0002:0002:0002:0002]:8080"},
 		{Addr: "3.3.3.3:3"},
-		{Addr: "[0003:0003:0003:0003:0003:0003:0003:0003]:3333"},
+		{Addr: "[fe80::1%eth0]:3333"},
 		{Addr: "[::FFFF:192.168.0.1]:2222"},
 	}
 
@@ -1210,6 +1225,253 @@ func (s) TestPickFirstLeaf_InterleavingUnknownPreffered(t *testing.T) {
 	}
 }
 
+// Test verifies that pickfirst balancer transitions to READY when the health
+// listener is enabled. Since client side health checking is not enabled in
+// the service config, the health listener will send a health update for READY
+// after registering the listener.
+func (s) TestPickFirstLeaf_HealthListenerEnabled(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	bf := stub.BalancerFuncs{
+		Init: func(bd *stub.BalancerData) {
+			bd.Data = balancer.Get(pickfirstleaf.Name).Build(bd.ClientConn, bd.BuildOptions)
+		},
+		Close: func(bd *stub.BalancerData) {
+			bd.Data.(balancer.Balancer).Close()
+		},
+		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
+			ccs.ResolverState = pickfirstleaf.EnableHealthListener(ccs.ResolverState)
+			return bd.Data.(balancer.Balancer).UpdateClientConnState(ccs)
+		},
+	}
+
+	stub.Register(t.Name(), bf)
+	svcCfg := fmt.Sprintf(`{ "loadBalancingConfig": [{%q: {}}] }`, t.Name())
+	backend := stubserver.StartTestService(t, nil)
+	defer backend.Stop()
+	opts := []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultServiceConfig(svcCfg),
+	}
+	cc, err := grpc.NewClient(backend.Address, opts...)
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", backend.Address, err)
+
+	}
+	defer cc.Close()
+
+	if err := pickfirst.CheckRPCsToBackend(ctx, cc, resolver.Address{Addr: backend.Address}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Test verifies that a health listener is not registered when pickfirst is not
+// under a petiole policy.
+func (s) TestPickFirstLeaf_HealthListenerNotEnabled(t *testing.T) {
+	// Wrap the clientconn to intercept NewSubConn.
+	// Capture the health list by wrapping the SC.
+	// Wrap the picker to unwrap the SC.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	healthListenerCh := make(chan func(balancer.SubConnState))
+
+	bf := stub.BalancerFuncs{
+		Init: func(bd *stub.BalancerData) {
+			ccw := &healthListenerCapturingCCWrapper{
+				ClientConn:       bd.ClientConn,
+				healthListenerCh: healthListenerCh,
+				subConnStateCh:   make(chan balancer.SubConnState, 5),
+			}
+			bd.Data = balancer.Get(pickfirstleaf.Name).Build(ccw, bd.BuildOptions)
+		},
+		Close: func(bd *stub.BalancerData) {
+			bd.Data.(balancer.Balancer).Close()
+		},
+		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
+			// Functions like a non-petiole policy by not configuring the use
+			// of health listeners.
+			return bd.Data.(balancer.Balancer).UpdateClientConnState(ccs)
+		},
+	}
+
+	stub.Register(t.Name(), bf)
+	svcCfg := fmt.Sprintf(`{ "loadBalancingConfig": [{%q: {}}] }`, t.Name())
+	backend := stubserver.StartTestService(t, nil)
+	defer backend.Stop()
+	opts := []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultServiceConfig(svcCfg),
+	}
+	cc, err := grpc.NewClient(backend.Address, opts...)
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", backend.Address, err)
+
+	}
+	defer cc.Close()
+	cc.Connect()
+
+	select {
+	case <-healthListenerCh:
+		t.Fatal("Health listener registered when not enabled.")
+	case <-time.After(defaultTestShortTimeout):
+	}
+
+	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
+}
+
+// Test mocks the updates sent to the health listener and verifies that the
+// balancer correctly reports the health state once the SubConn's connectivity
+// state becomes READY.
+func (s) TestPickFirstLeaf_HealthUpdates(t *testing.T) {
+	// Wrap the clientconn to intercept NewSubConn.
+	// Capture the health list by wrapping the SC.
+	// Wrap the picker to unwrap the SC.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	healthListenerCh := make(chan func(balancer.SubConnState))
+	scConnectivityStateCh := make(chan balancer.SubConnState, 5)
+
+	bf := stub.BalancerFuncs{
+		Init: func(bd *stub.BalancerData) {
+			ccw := &healthListenerCapturingCCWrapper{
+				ClientConn:       bd.ClientConn,
+				healthListenerCh: healthListenerCh,
+				subConnStateCh:   scConnectivityStateCh,
+			}
+			bd.Data = balancer.Get(pickfirstleaf.Name).Build(ccw, bd.BuildOptions)
+		},
+		Close: func(bd *stub.BalancerData) {
+			bd.Data.(balancer.Balancer).Close()
+		},
+		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
+			ccs.ResolverState = pickfirstleaf.EnableHealthListener(ccs.ResolverState)
+			return bd.Data.(balancer.Balancer).UpdateClientConnState(ccs)
+		},
+	}
+
+	stub.Register(t.Name(), bf)
+	svcCfg := fmt.Sprintf(`{ "loadBalancingConfig": [{%q: {}}] }`, t.Name())
+	backend := stubserver.StartTestService(t, nil)
+	defer backend.Stop()
+	opts := []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultServiceConfig(svcCfg),
+	}
+	cc, err := grpc.NewClient(backend.Address, opts...)
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed: %v", backend.Address, err)
+
+	}
+	defer cc.Close()
+	cc.Connect()
+
+	var healthListener func(balancer.SubConnState)
+	select {
+	case healthListener = <-healthListenerCh:
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for health listener to be registered.")
+	}
+
+	// Wait for the raw connectivity state to become READY. The LB policy should
+	// wait for the health updates before transitioning the channel to READY.
+	for {
+		var scs balancer.SubConnState
+		select {
+		case scs = <-scConnectivityStateCh:
+		case <-ctx.Done():
+			t.Fatal("Context timed out waiting for the SubConn connectivity state to become READY.")
+		}
+		if scs.ConnectivityState == connectivity.Ready {
+			break
+		}
+	}
+
+	shortCtx, cancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer cancel()
+	testutils.AwaitNoStateChange(shortCtx, t, cc, connectivity.Connecting)
+
+	// The LB policy should update the channel state based on the health state.
+	healthListener(balancer.SubConnState{
+		ConnectivityState: connectivity.TransientFailure,
+		ConnectionError:   fmt.Errorf("test health check failure"),
+	})
+	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
+
+	healthListener(balancer.SubConnState{
+		ConnectivityState: connectivity.Connecting,
+		ConnectionError:   balancer.ErrNoSubConnAvailable,
+	})
+	testutils.AwaitState(ctx, t, cc, connectivity.Connecting)
+
+	healthListener(balancer.SubConnState{
+		ConnectivityState: connectivity.Ready,
+	})
+	if err := pickfirst.CheckRPCsToBackend(ctx, cc, resolver.Address{Addr: backend.Address}); err != nil {
+		t.Fatal(err)
+	}
+
+	// When the health check fails, the channel should transition to TF.
+	healthListener(balancer.SubConnState{
+		ConnectivityState: connectivity.TransientFailure,
+		ConnectionError:   fmt.Errorf("test health check failure"),
+	})
+	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
+}
+
+// healthListenerCapturingCCWrapper is used to capture the health listener so
+// that health updates can be mocked for testing.
+type healthListenerCapturingCCWrapper struct {
+	balancer.ClientConn
+	healthListenerCh chan func(balancer.SubConnState)
+	subConnStateCh   chan balancer.SubConnState
+}
+
+func (ccw *healthListenerCapturingCCWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
+	oldListener := opts.StateListener
+	opts.StateListener = func(scs balancer.SubConnState) {
+		ccw.subConnStateCh <- scs
+		if oldListener != nil {
+			oldListener(scs)
+		}
+	}
+	sc, err := ccw.ClientConn.NewSubConn(addrs, opts)
+	if err != nil {
+		return nil, err
+	}
+	return &healthListenerCapturingSCWrapper{
+		SubConn:    sc,
+		listenerCh: ccw.healthListenerCh,
+	}, nil
+}
+
+func (ccw *healthListenerCapturingCCWrapper) UpdateState(state balancer.State) {
+	state.Picker = &unwrappingPicker{state.Picker}
+	ccw.ClientConn.UpdateState(state)
+}
+
+type healthListenerCapturingSCWrapper struct {
+	balancer.SubConn
+	listenerCh chan func(balancer.SubConnState)
+}
+
+func (scw *healthListenerCapturingSCWrapper) RegisterHealthListener(listener func(balancer.SubConnState)) {
+	scw.listenerCh <- listener
+}
+
+// unwrappingPicker unwraps SubConns because the channel expects SubConns to be
+// addrConns.
+type unwrappingPicker struct {
+	balancer.Picker
+}
+
+func (pw *unwrappingPicker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
+	pr, err := pw.Picker.Pick(info)
+	if pr.SubConn != nil {
+		pr.SubConn = pr.SubConn.(*healthListenerCapturingSCWrapper).SubConn
+	}
+	return pr, err
+}
+
 // subConnAddresses makes the pickfirst balancer create the requested number of
 // SubConns by triggering transient failures. The function returns the
 // addresses of the created SubConns.
@@ -1308,14 +1570,13 @@ type scState struct {
 }
 
 type backendManager struct {
-	backends []*stubserver.StubServer
+	backends []*testServer
 }
 
 func (b *backendManager) stopAllExcept(index int) {
 	for idx, b := range b.backends {
 		if idx != index {
-			b.S.Stop()
-			b.S = nil
+			b.stop()
 		}
 	}
 }
diff --git a/channelz/service/service_test.go b/channelz/service/service_test.go
index 78a5f2f8..111e5d17 100644
--- a/channelz/service/service_test.go
+++ b/channelz/service/service_test.go
@@ -22,6 +22,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"net/netip"
 	"strconv"
 	"strings"
 	"testing"
@@ -605,8 +606,8 @@ func (s) TestGetSocket(t *testing.T) {
 		lastMessageReceivedTimestamp:     time.Unix(3, 0),
 		localFlowControlWindow:           65536,
 		remoteFlowControlWindow:          1024,
-		localAddr:                        &net.TCPAddr{IP: net.ParseIP("1.0.0.1"), Port: 10001},
-		remoteAddr:                       &net.TCPAddr{IP: net.ParseIP("12.0.0.1"), Port: 10002},
+		localAddr:                        &net.TCPAddr{IP: netip.MustParseAddr("1.0.0.1").AsSlice(), Port: 10001},
+		remoteAddr:                       &net.TCPAddr{IP: netip.MustParseAddr("12.0.0.1").AsSlice(), Port: 10002},
 		remoteName:                       "remote.remote",
 	}), newSocket(czSocket{
 		streamsStarted:                   10,
@@ -637,11 +638,11 @@ func (s) TestGetSocket(t *testing.T) {
 		lastMessageReceivedTimestamp:     time.Unix(0, 0),
 		localFlowControlWindow:           65536,
 		remoteFlowControlWindow:          10240,
-		localAddr:                        &net.IPAddr{IP: net.ParseIP("1.0.0.1")},
-		remoteAddr:                       &net.IPAddr{IP: net.ParseIP("9.0.0.1")},
+		localAddr:                        &net.IPAddr{IP: netip.MustParseAddr("1.0.0.1").AsSlice()},
+		remoteAddr:                       &net.IPAddr{IP: netip.MustParseAddr("9.0.0.1").AsSlice()},
 		remoteName:                       "",
 	}), newSocket(czSocket{
-		localAddr: &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 10001},
+		localAddr: &net.TCPAddr{IP: netip.MustParseAddr("127.0.0.1").AsSlice(), Port: 10001},
 	}), newSocket(czSocket{
 		security: &credentials.TLSChannelzSecurityValue{
 			StandardName:      "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
diff --git a/internal/credentials/xds/handshake_info_test.go b/internal/credentials/xds/handshake_info_test.go
index a6802405..a3e598be 100644
--- a/internal/credentials/xds/handshake_info_test.go
+++ b/internal/credentials/xds/handshake_info_test.go
@@ -21,6 +21,7 @@ package xds
 import (
 	"crypto/x509"
 	"net"
+	"net/netip"
 	"net/url"
 	"regexp"
 	"testing"
@@ -142,8 +143,11 @@ func TestMatchingSANExists_FailureCases(t *testing.T) {
 	inputCert := &x509.Certificate{
 		DNSNames:       []string{"foo.bar.example.com", "bar.baz.test.com", "*.example.com"},
 		EmailAddresses: []string{"foobar@example.com", "barbaz@test.com"},
-		IPAddresses:    []net.IP{net.ParseIP("192.0.0.1"), net.ParseIP("2001:db8::68")},
-		URIs:           []*url.URL{url1, url2},
+		IPAddresses: []net.IP{
+			netip.MustParseAddr("192.0.0.1").AsSlice(),
+			netip.MustParseAddr("2001:db8::68").AsSlice(),
+		},
+		URIs: []*url.URL{url1, url2},
 	}
 
 	tests := []struct {
@@ -214,8 +218,11 @@ func TestMatchingSANExists_Success(t *testing.T) {
 	inputCert := &x509.Certificate{
 		DNSNames:       []string{"baz.test.com", "*.example.com"},
 		EmailAddresses: []string{"foobar@example.com", "barbaz@test.com"},
-		IPAddresses:    []net.IP{net.ParseIP("192.0.0.1"), net.ParseIP("2001:db8::68")},
-		URIs:           []*url.URL{url1, url2},
+		IPAddresses: []net.IP{
+			netip.MustParseAddr("192.0.0.1").AsSlice(),
+			netip.MustParseAddr("2001:db8::68").AsSlice(),
+		},
+		URIs: []*url.URL{url1, url2},
 	}
 
 	tests := []struct {
diff --git a/internal/resolver/dns/dns_resolver_test.go b/internal/resolver/dns/dns_resolver_test.go
index 4e6a4d54..47285052 100644
--- a/internal/resolver/dns/dns_resolver_test.go
+++ b/internal/resolver/dns/dns_resolver_test.go
@@ -755,7 +755,16 @@ func (s) TestIPResolver(t *testing.T) {
 			target:   "[2001:db8::1]:http",
 			wantAddr: []resolver.Address{{Addr: "[2001:db8::1]:http"}},
 		},
-		// TODO(yuxuanli): zone support?
+		{
+			name:     "ipv6 with zone and port",
+			target:   "[fe80::1%25eth0]:1234",
+			wantAddr: []resolver.Address{{Addr: "[fe80::1%eth0]:1234"}},
+		},
+		{
+			name:     "ipv6 with zone and default port",
+			target:   "fe80::1%25eth0",
+			wantAddr: []resolver.Address{{Addr: "[fe80::1%eth0]:443"}},
+		},
 	}
 
 	for _, test := range tests {
diff --git a/internal/testutils/fakegrpclb/server.go b/internal/testutils/fakegrpclb/server.go
index 82be2c1a..b19e863f 100644
--- a/internal/testutils/fakegrpclb/server.go
+++ b/internal/testutils/fakegrpclb/server.go
@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"net/netip"
 	"strconv"
 	"sync"
 	"time"
@@ -85,9 +86,9 @@ func NewServer(params ServerParams) (*Server, error) {
 		if err != nil {
 			return nil, fmt.Errorf("failed to parse list of backend address %q: %v", addr, err)
 		}
-		ip := net.ParseIP(ipStr)
-		if ip == nil {
-			return nil, fmt.Errorf("failed to parse ip: %q", ipStr)
+		ip, err := netip.ParseAddr(ipStr)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse ip %q: %v", ipStr, err)
 		}
 		port, err := strconv.Atoi(portStr)
 		if err != nil {
@@ -95,7 +96,7 @@ func NewServer(params ServerParams) (*Server, error) {
 		}
 		logger.Infof("Adding backend ip: %q, port: %d to server list", ip.String(), port)
 		servers = append(servers, &lbpb.Server{
-			IpAddress: ip,
+			IpAddress: ip.AsSlice(),
 			Port:      int32(port),
 		})
 	}
diff --git a/internal/testutils/stats/test_metrics_recorder.go b/internal/testutils/stats/test_metrics_recorder.go
index e13013e3..e1a03b8d 100644
--- a/internal/testutils/stats/test_metrics_recorder.go
+++ b/internal/testutils/stats/test_metrics_recorder.go
@@ -44,7 +44,7 @@ type TestMetricsRecorder struct {
 	// mu protects data.
 	mu sync.Mutex
 	// data is the most recent update for each metric name.
-	data map[estats.Metric]float64
+	data map[string]float64
 }
 
 // NewTestMetricsRecorder returns a new TestMetricsRecorder.
@@ -56,7 +56,7 @@ func NewTestMetricsRecorder() *TestMetricsRecorder {
 		floatHistoCh: testutils.NewChannelWithSize(10),
 		intGaugeCh:   testutils.NewChannelWithSize(10),
 
-		data: make(map[estats.Metric]float64),
+		data: make(map[string]float64),
 	}
 }
 
@@ -65,7 +65,7 @@ func NewTestMetricsRecorder() *TestMetricsRecorder {
 func (r *TestMetricsRecorder) Metric(name string) (float64, bool) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	data, ok := r.data[estats.Metric(name)]
+	data, ok := r.data[name]
 	return data, ok
 }
 
@@ -73,7 +73,7 @@ func (r *TestMetricsRecorder) Metric(name string) (float64, bool) {
 func (r *TestMetricsRecorder) ClearMetrics() {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	r.data = make(map[estats.Metric]float64)
+	r.data = make(map[string]float64)
 }
 
 // MetricsData represents data associated with a metric.
diff --git a/security/advancedtls/crl_test.go b/security/advancedtls/crl_test.go
index e24a8f9b..450ab561 100644
--- a/security/advancedtls/crl_test.go
+++ b/security/advancedtls/crl_test.go
@@ -30,6 +30,7 @@ import (
 	"fmt"
 	"math/big"
 	"net"
+	"net/netip"
 	"os"
 	"path"
 	"strings"
@@ -463,7 +464,7 @@ func setupTLSConn(t *testing.T) (net.Listener, *x509.Certificate, *ecdsa.Private
 		Subject:               pkix.Name{CommonName: "test-cert"},
 		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageCRLSign,
 		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
-		IPAddresses:           []net.IP{net.ParseIP("::1")},
+		IPAddresses:           []net.IP{netip.MustParseAddr("::1").AsSlice()},
 		CRLDistributionPoints: []string{"http://static.corp.google.com/crl/campus-sln/borg"},
 	}
 
diff --git a/stats/opentelemetry/example_test.go b/stats/opentelemetry/example_test.go
index 66676f3c..e87e4ebb 100644
--- a/stats/opentelemetry/example_test.go
+++ b/stats/opentelemetry/example_test.go
@@ -19,7 +19,7 @@ package opentelemetry_test
 import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
-	estats "google.golang.org/grpc/experimental/stats"
+	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/stats/opentelemetry"
 
 	"go.opentelemetry.io/otel/sdk/metric"
@@ -88,7 +88,7 @@ func ExampleMetrics_excludeSome() {
 	// To exclude specific metrics, initialize Options as follows:
 	opts := opentelemetry.Options{
 		MetricsOptions: opentelemetry.MetricsOptions{
-			Metrics: opentelemetry.DefaultMetrics().Remove(opentelemetry.ClientAttemptDuration, opentelemetry.ClientAttemptRcvdCompressedTotalMessageSize),
+			Metrics: opentelemetry.DefaultMetrics().Remove(opentelemetry.ClientAttemptDurationMetricName, opentelemetry.ClientAttemptRcvdCompressedTotalMessageSizeMetricName),
 		},
 	}
 	do := opentelemetry.DialOption(opts)
@@ -103,7 +103,7 @@ func ExampleMetrics_disableAll() {
 	// To disable all metrics, initialize Options as follows:
 	opts := opentelemetry.Options{
 		MetricsOptions: opentelemetry.MetricsOptions{
-			Metrics: estats.NewMetrics(), // Distinct to nil, which creates default metrics. This empty set creates no metrics.
+			Metrics: stats.NewMetricSet(), // Distinct to nil, which creates default metrics. This empty set creates no metrics.
 		},
 	}
 	do := opentelemetry.DialOption(opts)
@@ -118,7 +118,7 @@ func ExampleMetrics_enableSome() {
 	// To only create specific metrics, initialize Options as follows:
 	opts := opentelemetry.Options{
 		MetricsOptions: opentelemetry.MetricsOptions{
-			Metrics: estats.NewMetrics(opentelemetry.ClientAttemptDuration, opentelemetry.ClientAttemptRcvdCompressedTotalMessageSize), // only create these metrics
+			Metrics: stats.NewMetricSet(opentelemetry.ClientAttemptDurationMetricName, opentelemetry.ClientAttemptRcvdCompressedTotalMessageSizeMetricName), // only create these metrics
 		},
 	}
 	do := opentelemetry.DialOption(opts)
diff --git a/test/clientconn_state_transition_test.go b/test/clientconn_state_transition_test.go
index a0a61099..547b5ffb 100644
--- a/test/clientconn_state_transition_test.go
+++ b/test/clientconn_state_transition_test.go
@@ -39,6 +39,7 @@ import (
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
+	testpb "google.golang.org/grpc/interop/grpc_testing"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 )
@@ -678,7 +679,7 @@ func (s) TestChannelStateTransitionWithRPC(t *testing.T) {
 
 	// Make an RPC call to transition the channel to CONNECTING.
 	go func() {
-		_, err := testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testgrpc.Empty{})
+		_, err := testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{})
 		if err == nil {
 			t.Errorf("Expected RPC to fail, but it succeeded")
 		}
diff --git a/test/local_creds_test.go b/test/local_creds_test.go
index ac32ab52..241f6306 100644
--- a/test/local_creds_test.go
+++ b/test/local_creds_test.go
@@ -22,6 +22,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"net/netip"
 	"strings"
 	"testing"
 	"time"
@@ -181,11 +182,11 @@ func testLocalCredsE2EFail(dopts []grpc.DialOption) error {
 
 	var fakeClientAddr, fakeServerAddr net.Addr
 	fakeClientAddr = &net.IPAddr{
-		IP:   net.ParseIP("10.8.9.10"),
+		IP:   netip.MustParseAddr("10.8.9.10").AsSlice(),
 		Zone: "",
 	}
 	fakeServerAddr = &net.IPAddr{
-		IP:   net.ParseIP("10.8.9.11"),
+		IP:   netip.MustParseAddr("10.8.9.11").AsSlice(),
 		Zone: "",
 	}
 
diff --git a/xds/internal/xdsclient/xdsresource/filter_chain_test.go b/xds/internal/xdsclient/xdsresource/filter_chain_test.go
index 1897e387..cdcc8660 100644
--- a/xds/internal/xdsclient/xdsresource/filter_chain_test.go
+++ b/xds/internal/xdsclient/xdsresource/filter_chain_test.go
@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"net/netip"
 	"strings"
 	"testing"
 
@@ -2438,7 +2439,7 @@ func (s) TestLookup_Successes(t *testing.T) {
 			lis:  lisWithoutDefaultChain,
 			params: FilterChainLookupParams{
 				IsUnspecifiedListener: true,
-				DestAddr:              net.ParseIP("2001:68::db8"),
+				DestAddr:              netip.MustParseAddr("2001:68::db8").AsSlice(),
 				SourceAddr:            net.IPv4(10, 1, 1, 1),
 				SourcePort:            1,
 			},
@@ -2468,8 +2469,8 @@ func (s) TestLookup_Successes(t *testing.T) {
 			lis:  lisWithoutDefaultChain,
 			params: FilterChainLookupParams{
 				IsUnspecifiedListener: true,
-				DestAddr:              net.ParseIP("2001:68::1"),
-				SourceAddr:            net.ParseIP("2001:68::2"),
+				DestAddr:              netip.MustParseAddr("2001:68::1").AsSlice(),
+				SourceAddr:            netip.MustParseAddr("2001:68::2").AsSlice(),
 				SourcePort:            1,
 			},
 			wantFC: &FilterChain{