Repository: grpc/grpc-go

Issue:
Flaky test package: test/xds
Alongside #6913 and #6912, I have ran the test/xds suite on master since I added tests to it for my xDS Server fix #6889. I have encountered numerous flakes on g3, particularly those outlined in custom lb tests for distribution #6601. However, I have encountered almost every client and server side xDS test flake with a context timeout for a RPC expected to proceed. Each has different logs/events preceeding it's timeout, but every test seems susceptible to timeout. The flakes are generally rare, but due to the number of tests in the test suite you can successfully trigger by running the full test suite enough times. My initial inkling tells me there's some synchronization needed or something gets stuck in the management server/testing xDS Client flow. This also manifests in rare flakes for my xDS Server fix, where I expect something like an err that represents Accept and Close, and I get a context timeout instead.

Discussion:
@arvindbr8: another one for  TestServerSideXDS_WithValidAndInvalidSecurityConfiguration: https://github.com/grpc/grpc-go/actions/runs/7480796959/job/20361025267?pr=6916@zasweq: https://github.com/grpc/grpc-go/actions/runs/7618526481/job/20749870802?pr=6933@arvindbr8: https://github.com/grpc/grpc-go/actions/runs/7716359885/job/21032969470?pr=6949@arvindbr8: https://github.com/grpc/grpc-go/actions/runs/7791646589/job/21248143511?pr=6965@zasweq: https://github.com/grpc/grpc-go/actions/runs/8546803299/job/23417838293?pr=7085@arjan-bal: https://github.com/grpc/grpc-go/actions/runs/9790522733/job/27032314481?pr=7390@arjan-bal: @zasweq I investigated this and the problem seems to be due to the xDS management server getting stuck while writing to this buffered channel
https://github.com/grpc/grpc-go/blob/d27ddb5eb5940c949f88bc2cb21eed9254f8be75/test/xds/xds_server_certificate_providers_test.go#L249

In the logs of failing runs for `TestServerSideXDS_WithValidAndInvalidSecurityConfiguration`, I noticed that the resource snapshot update request is sent to the xds management server before the xds client is able to connect to the xds server. This somehow results in more than 1 Listener requests being sent to the xds server which get stuck waiting to write to the buffered channel.

This seems to be a problem with the test and not the implementation. Adding a 50 millis sleep after starting both the servers did get rid of the flakiness in `TestServerSideXDS_WithValidAndInvalidSecurityConfiguration`.@zasweq: Ah nice thank you for figuring this out!@arvindbr8: https://github.com/grpc/grpc-go/actions/runs/9996172593/job/27629932247?pr=7397@zasweq: You mentioned this solved the test, but not the flakes in the full package. This was my flaky test in this PR so thanks for fixing this: https://github.com/grpc/grpc-go/actions/runs/10050840269/job/27779434995?pr=7434 :).@arjan-bal: Closing this issue as the root cause of all flaking tests seems to be different. There are individual issues open for tests that have flaked recently.

Solution patch:
diff --git a/experimental/stats/metricregistry.go b/experimental/stats/metricregistry.go
index 39225694..e82eda56 100644
--- a/experimental/stats/metricregistry.go
+++ b/experimental/stats/metricregistry.go
@@ -23,8 +23,13 @@ import (
 	"testing"
 
 	"google.golang.org/grpc/grpclog"
+	"google.golang.org/grpc/internal"
 )
 
+func init() {
+	internal.SnapshotMetricRegistryForTesting = snapshotMetricsRegistryForTesting
+}
+
 var logger = grpclog.Component("metrics-registry")
 
 // DefaultMetrics are the default metrics registered through global metrics
@@ -54,6 +59,10 @@ type MetricDescriptor struct {
 	// The type of metric. This is set by the metric registry, and not intended
 	// to be set by a component registering a metric.
 	Type MetricType
+	// Bounds are the bounds of this metric. This only applies to histogram
+	// metrics. If unset or set with length 0, stats handlers will fall back to
+	// default bounds.
+	Bounds []float64
 }
 
 // MetricType is the type of metric.
diff --git a/internal/internal.go b/internal/internal.go
index 46ed2576..e1e1422e 100644
--- a/internal/internal.go
+++ b/internal/internal.go
@@ -215,6 +215,12 @@ var (
 
 	// SetConnectedAddress sets the connected address for a SubConnState.
 	SetConnectedAddress any // func(scs *SubConnState, addr resolver.Address)
+
+	// SnapshotMetricRegistryForTesting snapshots the global data of the metric
+	// registry. Registers a cleanup function on the provided testing.T that
+	// sets the metric registry to its original state. Only called in testing
+	// functions.
+	SnapshotMetricRegistryForTesting any // func(t *testing.T)
 )
 
 // HealthChecker defines the signature of the client-side LB channel health
diff --git a/internal/transport/grpchttp2/errors.go b/internal/transport/grpchttp2/errors.go
new file mode 100644
index 00000000..c2aac6bd
--- /dev/null
+++ b/internal/transport/grpchttp2/errors.go
@@ -0,0 +1,72 @@
+/*
+ *
+ * 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 grpchttp2
+
+import "fmt"
+
+// ErrCode represents an HTTP/2 Error Code. Error codes are 32-bit fields
+// that are used in [RST_STREAM] and [GOAWAY] frames to convey the reasons for
+// the stream or connection error. See [HTTP/2 Error Code] for definitions of
+// each of the following error codes.
+//
+// [HTTP/2 Error Code]: https://httpwg.org/specs/rfc7540.html#ErrorCodes
+// [RST_STREAM]: https://httpwg.org/specs/rfc7540.html#RST_STREAM
+// [GOAWAY]: https://httpwg.org/specs/rfc7540.html#GOAWAY
+type ErrCode uint32
+
+const (
+	ErrCodeNoError            ErrCode = 0x0
+	ErrCodeProtocol           ErrCode = 0x1
+	ErrCodeInternal           ErrCode = 0x2
+	ErrCodeFlowControl        ErrCode = 0x3
+	ErrCodeSettingsTimeout    ErrCode = 0x4
+	ErrCodeStreamClosed       ErrCode = 0x5
+	ErrCodeFrameSize          ErrCode = 0x6
+	ErrCodeRefusedStream      ErrCode = 0x7
+	ErrCodeCancel             ErrCode = 0x8
+	ErrCodeCompression        ErrCode = 0x9
+	ErrCodeConnect            ErrCode = 0xa
+	ErrCodeEnhanceYourCalm    ErrCode = 0xb
+	ErrCodeIndaequateSecurity ErrCode = 0xc
+	ErrCodeHTTP11Required     ErrCode = 0xd
+)
+
+var errorCodeNames = map[ErrCode]string{
+	ErrCodeNoError:            "NO_ERROR",
+	ErrCodeProtocol:           "PROTOCOL_ERROR",
+	ErrCodeInternal:           "INTERNAL_ERROR",
+	ErrCodeFlowControl:        "FLOW_CONTROL_ERROR",
+	ErrCodeSettingsTimeout:    "SETTINGS_TIMEOUT",
+	ErrCodeStreamClosed:       "STREAM_CLOSED",
+	ErrCodeFrameSize:          "FRAME_SIZE_ERROR",
+	ErrCodeRefusedStream:      "REFUSED_STREAM",
+	ErrCodeCancel:             "CANCEL",
+	ErrCodeCompression:        "COMPRESSION_ERROR",
+	ErrCodeConnect:            "CONNECT_ERROR",
+	ErrCodeEnhanceYourCalm:    "ENHANCE_YOUR_CALM",
+	ErrCodeIndaequateSecurity: "INADEQUATE_SECURITY",
+	ErrCodeHTTP11Required:     "HTTP_1_1_REQUIRED",
+}
+
+func (err ErrCode) String() string {
+	if v, ok := errorCodeNames[err]; ok {
+		return v
+	}
+	return fmt.Sprintf("unknown error code %#x", uint32(err))
+}
diff --git a/internal/transport/grpchttp2/framer.go b/internal/transport/grpchttp2/framer.go
new file mode 100644
index 00000000..b12ae3c7
--- /dev/null
+++ b/internal/transport/grpchttp2/framer.go
@@ -0,0 +1,317 @@
+/*
+ *
+ * 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 grpchttp2 defines HTTP/2 types and a framer API and implementation.
+package grpchttp2
+
+import "golang.org/x/net/http2/hpack"
+
+// FrameType represents the type of an HTTP/2 Frame.
+// See [Frame Type].
+//
+// [Frame Type]: https://httpwg.org/specs/rfc7540.html#FrameType
+type FrameType uint8
+
+const (
+	FrameTypeData         FrameType = 0x0
+	FrameTypeHeaders      FrameType = 0x1
+	FrameTypeRSTStream    FrameType = 0x3
+	FrameTypeSettings     FrameType = 0x4
+	FrameTypePing         FrameType = 0x6
+	FrameTypeGoAway       FrameType = 0x7
+	FrameTypeWindowUpdate FrameType = 0x8
+	FrameTypeContinuation FrameType = 0x9
+)
+
+// Flag represents one or more flags set on an HTTP/2 Frame.
+type Flag uint8
+
+const (
+	FlagDataEndStream          Flag = 0x1
+	FlagDataPadded             Flag = 0x8
+	FlagHeadersEndStream       Flag = 0x1
+	FlagHeadersEndHeaders      Flag = 0x4
+	FlagHeadersPadded          Flag = 0x8
+	FlagHeadersPriority        Flag = 0x20
+	FlagSettingsAck            Flag = 0x1
+	FlagPingAck                Flag = 0x1
+	FlagContinuationEndHeaders Flag = 0x4
+)
+
+// Setting represents the id and value pair of an HTTP/2 setting.
+// See [Setting Format].
+//
+// [Setting Format]: https://httpwg.org/specs/rfc7540.html#SettingFormat
+type Setting struct {
+	ID    SettingID
+	Value uint32
+}
+
+// SettingID represents the id of an HTTP/2 setting.
+// See [Setting Values].
+//
+// [Setting Values]: https://httpwg.org/specs/rfc7540.html#SettingValues
+type SettingID uint16
+
+const (
+	SettingsHeaderTableSize      SettingID = 0x1
+	SettingsEnablePush           SettingID = 0x2
+	SettingsMaxConcurrentStreams SettingID = 0x3
+	SettingsInitialWindowSize    SettingID = 0x4
+	SettingsMaxFrameSize         SettingID = 0x5
+	SettingsMaxHeaderListSize    SettingID = 0x6
+)
+
+// FrameHeader is the 9 byte header of any HTTP/2 Frame.
+// See [Frame Header].
+//
+// [Frame Header]: https://httpwg.org/specs/rfc7540.html#FrameHeader
+type FrameHeader struct {
+	// Size is the size of the frame's payload without the 9 header bytes.
+	// As per the HTTP/2 spec, size can be up to 3 bytes, but only frames
+	// up to 16KB can be processed without agreement.
+	Size uint32
+	// Type is a byte that represents the Frame Type.
+	Type FrameType
+	// Flags is a byte representing the flags set on this Frame.
+	Flags Flag
+	// StreamID is the ID for the stream which this frame is for. If the
+	// frame is connection specific instead of stream specific, the
+	// streamID is 0.
+	StreamID uint32
+}
+
+// Frame represents an HTTP/2 Frame. This interface struct is only to be used
+// on the read path of the Framer. The writing path expects the data to be
+// passed individually, not using this type.
+//
+// Each concrete Frame type defined below implements the Frame interface.
+type Frame interface {
+	Header() *FrameHeader
+	// Free frees the underlying buffer if present so it can be reused by the
+	// framer.
+	//
+	// TODO: Remove method from the interface once the mem package gets merged.
+	// Free will be called on each mem.Buffer individually.
+	Free()
+}
+
+// DataFrame is the representation of a [DATA frame]. DATA frames convey
+// arbitrary, variable-length sequences of octets associated with a stream.
+//
+// [DATA frame]: https://httpwg.org/specs/rfc7540.html#DATA
+type DataFrame struct {
+	hdr  *FrameHeader
+	free func()
+	Data []byte
+}
+
+func (f *DataFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+func (f *DataFrame) Free() {
+	if f.free != nil {
+		f.free()
+	}
+}
+
+// HeadersFrame is the representation of a [HEADERS Frame]. The HEADERS frame
+// is used to open a stream, and additionally carries a header block fragment.
+//
+// [HEADERS Frame]: https://httpwg.org/specs/rfc7540.html#HEADERS
+type HeadersFrame struct {
+	hdr      *FrameHeader
+	free     func()
+	HdrBlock []byte
+}
+
+func (f *HeadersFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+func (f *HeadersFrame) Free() {
+	if f.free != nil {
+		f.free()
+	}
+}
+
+// RSTStreamFrame is the representation of a [RST_STREAM Frame]. There is no
+// underlying byte array in this frame, so Free() is a no-op. The RST_STREAM
+// frame allows for immediate termination of a stream
+//
+// [RST_STREAM Frame]: https://httpwg.org/specs/rfc7540.html#RST_STREAM
+type RSTStreamFrame struct {
+	hdr  *FrameHeader
+	Code ErrCode
+}
+
+func (f *RSTStreamFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+func (f *RSTStreamFrame) Free() {}
+
+// SettingsFrame is the representation of a [SETTINGS Frame]. There is no
+// underlying byte array in this frame, so Free() is a no-op.
+//
+// The SETTINGS frame conveys configuration parameters that affect how
+// endpoints communicate, such as preferences and constraints on peer behavior.
+//
+// [SETTINGS Frame]: https://httpwg.org/specs/rfc7540.html#SETTINGS
+type SettingsFrame struct {
+	hdr      *FrameHeader
+	Settings []Setting
+}
+
+func (f *SettingsFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+func (f *SettingsFrame) Free() {}
+
+// PingFrame is the representation of a [PING Frame]. The PING frame is a
+// mechanism for measuring a minimal round-trip time from the sender, as well
+// as determining whether an idle connection is still functional.
+//
+// [PING Frame]: https://httpwg.org/specs/rfc7540.html#PING
+type PingFrame struct {
+	hdr  *FrameHeader
+	free func()
+	Data []byte
+}
+
+func (f *PingFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+func (f *PingFrame) Free() {
+	if f.free != nil {
+		f.free()
+	}
+}
+
+// GoAwayFrame is the representation of a [GOAWAY Frame]. The GOAWAY frame is
+// used to initiate shutdown of a connection or to signal serious error
+// conditions.
+//
+// [GOAWAY Frame]: https://httpwg.org/specs/rfc7540.html#GOAWAY
+type GoAwayFrame struct {
+	hdr          *FrameHeader
+	free         func()
+	LastStreamID uint32
+	Code         ErrCode
+	DebugData    []byte
+}
+
+func (f *GoAwayFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+func (f *GoAwayFrame) Free() {
+	if f.free != nil {
+		f.free()
+	}
+}
+
+// WindowUpdateFrame is the representation of a [WINDOW_UPDATE Frame]. The
+// WINDOW_UPDATE frame is used to implement flow control.
+//
+// [WINDOW_UPDATE Frame]: https://httpwg.org/specs/rfc7540.html#WINDOW_UPDATE
+type WindowUpdateFrame struct {
+	hdr *FrameHeader
+	Inc uint32
+}
+
+func (f *WindowUpdateFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+// ContinuationFrame is the representation of a [CONTINUATION Frame]. The
+// CONTINUATION frame is used to continue a sequence of header block fragments.
+//
+// [CONTINUATION Frame]: https://httpwg.org/specs/rfc7540.html#CONTINUATION
+type ContinuationFrame struct {
+	hdr      *FrameHeader
+	free     func()
+	HdrBlock []byte
+}
+
+func (f *ContinuationFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+func (f *ContinuationFrame) Free() {
+	if f.free != nil {
+		f.free()
+	}
+}
+
+// MetaHeadersFrame is the representation of one HEADERS frame and zero or more
+// contiguous CONTINUATION frames and the decoding of their HPACK-encoded
+// contents.  This frame type is not transmitted over the network and is only
+// generated by the ReadFrame() function.
+//
+// Since there is no underlying buffer in this Frame, Free() is a no-op.
+type MetaHeadersFrame struct {
+	hdr    *FrameHeader
+	Fields []hpack.HeaderField
+	// Truncated indicates whether the MetaHeadersFrame has been truncated due
+	// to being longer than the MaxHeaderListSize.
+	Truncated bool
+}
+
+func (f *MetaHeadersFrame) Header() *FrameHeader {
+	return f.hdr
+}
+
+func (f *MetaHeadersFrame) Free() {}
+
+// Framer encapsulates the functionality to read and write HTTP/2 frames.
+type Framer interface {
+	// ReadFrame returns grpchttp2.Frame. It is the caller's responsibility to
+	// call Frame.Free() once it is done using it. Note that once the mem
+	// package gets merged, this API will change in favor of Buffer.Free().
+	ReadFrame() (Frame, error)
+	// WriteData writes an HTTP/2 DATA frame to the stream.
+	// TODO: Once the mem package gets merged, data will change type to
+	// mem.BufferSlice.
+	WriteData(streamID uint32, endStream bool, data ...[]byte) error
+	// WriteData writes an HTTP/2 HEADERS frame to the stream.
+	// TODO: Once the mem package gets merged, headerBlock will change type to
+	// mem.Buffer.
+	WriteHeaders(streamID uint32, endStream, endHeaders bool, headerBlocks []byte) error
+	// WriteData writes an HTTP/2 RST_STREAM frame to the stream.
+	WriteRSTStream(streamID uint32, code ErrCode) error
+	// WriteSettings writes an HTTP/2 SETTINGS frame to the connection.
+	WriteSettings(settings ...Setting) error
+	// WriteSettingsAck writes an HTTP/2 SETTINGS frame with the ACK flag set.
+	WriteSettingsAck() error
+	// WritePing writes an HTTP/2 PING frame to the connection.
+	WritePing(ack bool, data [8]byte) error
+	// WriteGoAway writes an HTTP/2 GOAWAY frame to the connection.
+	// TODO: Once the mem package gets merged, debugData will change type to
+	// mem.Buffer.
+	WriteGoAway(maxStreamID uint32, code ErrCode, debugData []byte) error
+	// WriteWindowUpdate writes an HTTP/2 WINDOW_UPDATE frame to the stream.
+	WriteWindowUpdate(streamID, inc uint32) error
+	// WriteContinuation writes an HTTP/2 CONTINUATION frame to the stream.
+	// TODO: Once the mem package gets merged, data will change type to
+	// mem.Buffer.
+	WriteContinuation(streamID uint32, endHeaders bool, headerBlock []byte) error
+}
diff --git a/scripts/install-protoc.sh b/scripts/install-protoc.sh
index e2424675..7c450ee5 100755
--- a/scripts/install-protoc.sh
+++ b/scripts/install-protoc.sh
@@ -1,5 +1,5 @@
 #!/bin/bash
-# 
+#
 # Copyright 2024 gRPC authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -14,21 +14,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-# install-protoc.sh 
-# 
-# This script installs the Protocol Buffers compiler (protoc) on the local
-# machine. It is used to generate code from .proto files for gRPC communication.
-# The script downloads the protoc binary from the official GitHub repository and
-# installs it in the system. 
-# 
-# Usage: ./install-protoc.sh [INSTALL_PATH]
+# install-protoc.sh
+#
+# This script installs the Protocol Buffers compiler (protoc) to the specified
+# directory.  It is used to generate code from .proto files for gRPC
+# communication. The script downloads the protoc binary from the official GitHub
+# repository and installs it in the system.
+#
+# Usage: ./install-protoc.sh INSTALL_PATH
 #
-# Arguments: 
-#   INSTALL_PATH: The path where the protoc binary will be installed (optional).
-#                 If not provided, the script will install the binary in the the
-#                 directory named by the GOBIN environment variable, which
-#                 defaults to $GOPATH/bin or $HOME/go/bin if the GOPATH
-#                 environment variable is not set. 
+# Arguments:
+#   INSTALL_PATH: The path where the protoc binary will be installed.
 #
 # Note: This script requires internet connectivity to download the protoc binary.
 
@@ -39,53 +35,60 @@ source "$(dirname $0)/common.sh"
 # The version of protoc that will be installed.
 PROTOC_VERSION="27.1"
 
-INSTALL_PATH="${1:-${GOBIN:-${GOPATH:-$HOME/go}}}"
+main() {
+  if [[ "$#" -ne 1 ]]; then
+    die "Usage: $0 INSTALL_PATH"
+  fi
 
-# downloads the pre-built binaries for Linux to $INSTALL_PATH.
-# Usage:
-#   download_binary ARCH OS
-# Arguments:
-#   ARCH:  The architecture of the system.  Accepted values: [x86_64, aarch_64]
-#   OS:    The operating system of the system.  Accepted values: [osx, linux]
-download_binary() {  
-  DOWNLOAD_URL="https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOC_VERSION}/protoc-${PROTOC_VERSION}-$2-$1.zip"
-  # -L follows redirects.
-  # -O writes output to a file.
-  curl -LO "${DOWNLOAD_URL}"
-  unzip "protoc-${PROTOC_VERSION}-$2-$1.zip" -d "${INSTALL_PATH}"
-  rm "protoc-${PROTOC_VERSION}-$2-$1.zip"
-  rm "${INSTALL_PATH}/readme.txt"
-}
+  INSTALL_PATH="${1}"
 
-main() {
-  # Check if protoc is already available.
-  if command -v protoc &> /dev/null; then
-    if INSTALL_VERSION=$(protoc --version | cut -d' ' -f2 2>/dev/null); then
-      if [ "$INSTALL_VERSION" = "$PROTOC_VERSION" ]; then
-        echo "protoc version $PROTOC_VERSION is already installed."
-        return
-      else
-        die "Existing protoc version ($INSTALL_VERSION) differs. Kindly make sure you have $PROTOC_VERSION installed."
-      fi
-    fi
+  if [[ ! -d "${INSTALL_PATH}" ]]; then
+    die "INSTALL_PATH (${INSTALL_PATH}) does not exist."
   fi
 
-  # Detect the architecture
+  echo "Installing protoc version $PROTOC_VERSION to ${INSTALL_PATH}..."
+
+  # Detect the hardware platform.
   case "$(uname -m)" in
-    "x86_64") ARCH="x86_64";;
-    "aarch64") ARCH="aarch_64";;
-    "arm64") ARCH="aarch_64";;
-  *) die "Unsupported architecture. Please consider manual installation from "\
-          "https://github.com/protocolbuffers/protobuf/releases/ and add to PATH."
+    "x86_64")   ARCH="x86_64";;
+    "aarch64")  ARCH="aarch_64";;
+    "arm64")    ARCH="aarch_64";;
+    *)          die "Install unsuccessful. Hardware platform not supported by installer: $1";;
   esac
 
-  # Detect the Operating System
+  # Detect the Operating System.
   case "$(uname -s)" in
-    "Darwin") download_binary $ARCH "osx";;
-    "Linux") download_binary $ARCH "linux";;
-  *) die "Unsupported OS. Please consider manual installation from "\
-          "https://github.com/protocolbuffers/protobuf/releases/ and add to PATH" ;;
+    "Darwin") OS="osx";;
+    "Linux")  OS="linux";;
+    *)        die "Install unsuccessful. OS not supported by installer: $2";;
   esac
+
+  # Check if the protoc binary with the right version is already installed.
+  if [[ -f "${INSTALL_PATH}/bin/protoc" ]]; then
+    if [[ "$("${INSTALL_PATH}/bin/protoc" --version)" == "libprotoc ${PROTOC_VERSION}" ]]; then
+      echo "protoc version ${PROTOC_VERSION} is already installed in ${INSTALL_PATH}"
+      return
+    fi
+  fi
+
+  DOWNLOAD_URL="https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOC_VERSION}/protoc-${PROTOC_VERSION}-${OS}-${ARCH}.zip"
+
+  # -L follows redirects.
+  # -O writes output to a file.
+  curl -LO "${DOWNLOAD_URL}"
+
+  # Unzip the downloaded file and except readme.txt.
+  # The file structure should look like:
+  # INSTALL_PATH
+  # ├── bin
+  # │   └── protoc
+  # └── include
+  #     └── other files...
+  unzip "protoc-${PROTOC_VERSION}-${OS}-${ARCH}.zip" -d "${INSTALL_PATH}" -x "readme.txt"
+  rm "protoc-${PROTOC_VERSION}-${OS}-${ARCH}.zip"
+
+  # Make the protoc binary executable. ¯\_(ツ)_/¯  crazy, right?
+  chmod +x "${INSTALL_PATH}/bin/protoc"
 }
 
 main "$@"
diff --git a/scripts/regenerate.sh b/scripts/regenerate.sh
index d63f88d8..282774b1 100755
--- a/scripts/regenerate.sh
+++ b/scripts/regenerate.sh
@@ -1,4 +1,5 @@
 #!/bin/bash
+#
 # Copyright 2020 gRPC authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -15,39 +16,43 @@
 
 set -eu -o pipefail
 
-WORKDIR=$(mktemp -d)
-
-function finish {
-  rm -rf "$WORKDIR"
-}
-trap finish EXIT
+WORKDIR="/tmp/grpc-go-tools"
+mkdir -p "${WORKDIR}"
 
-./scripts/install-protoc.sh "${WORKDIR}"
+"$(dirname "${0}")"/install-protoc.sh ${WORKDIR}
 
 export GOBIN="${WORKDIR}"/bin
 export PATH="${GOBIN}:${PATH}"
 mkdir -p "${GOBIN}"
 
-echo "remove existing generated files"
+echo "removing existing generated files..."
 # grpc_testing_not_regenerate/*.pb.go is not re-generated,
 # see grpc_testing_not_regenerate/README.md for details.
-rm -f $(find . -name '*.pb.go' | grep -v 'grpc_testing_not_regenerate')
+find . -name '*.pb.go' | grep -v 'grpc_testing_not_regenerate' | xargs rm -f || true
 
-echo "go install google.golang.org/protobuf/cmd/protoc-gen-go"
+echo "Executing: go install google.golang.org/protobuf/cmd/protoc-gen-go..."
 (cd test/tools && go install google.golang.org/protobuf/cmd/protoc-gen-go)
 
-echo "go install cmd/protoc-gen-go-grpc"
+echo "Executing: go install cmd/protoc-gen-go-grpc..."
 (cd cmd/protoc-gen-go-grpc && go install .)
 
-echo "git clone https://github.com/grpc/grpc-proto"
-git clone --quiet https://github.com/grpc/grpc-proto "${WORKDIR}/grpc-proto"
+echo "Pulling protos from https://github.com/grpc/grpc-proto..."
+if [ -d "${WORKDIR}/grpc-proto" ]; then
+  (cd "${WORKDIR}/grpc-proto" && git pull)
+else
+  git clone --quiet https://github.com/grpc/grpc-proto "${WORKDIR}/grpc-proto"
+fi
 
-echo "git clone https://github.com/protocolbuffers/protobuf"
-git clone --quiet https://github.com/protocolbuffers/protobuf "${WORKDIR}/protobuf"
+echo "Pulling protos from https://github.com/protocolbuffers/protobuf..."
+if [ -d "${WORKDIR}/protobuf" ]; then
+  (cd "${WORKDIR}/protobuf" && git pull)
+else
+  git clone --quiet https://github.com/protocolbuffers/protobuf "${WORKDIR}/protobuf"
+fi
 
 # Pull in code.proto as a proto dependency
 mkdir -p "${WORKDIR}/googleapis/google/rpc"
-echo "curl https://raw.githubusercontent.com/googleapis/googleapis/master/google/rpc/code.proto"
+echo "Pulling code.proto from https://raw.githubusercontent.com/googleapis/googleapis/master/google/rpc/code.proto..."
 curl --silent https://raw.githubusercontent.com/googleapis/googleapis/master/google/rpc/code.proto > "${WORKDIR}/googleapis/google/rpc/code.proto"
 
 
@@ -72,8 +77,8 @@ SOURCES=(
   "${WORKDIR}/grpc-proto/grpc/gcp/transport_security_common.proto"
   "${WORKDIR}/grpc-proto/grpc/lookup/v1/rls.proto"
   "${WORKDIR}/grpc-proto/grpc/lookup/v1/rls_config.proto"
-  "${WORKDIR}/grpc-proto/grpc/testing/*.proto"
-  "${WORKDIR}/grpc-proto/grpc/core/*.proto"
+  "${WORKDIR}"/grpc-proto/grpc/testing/*.proto
+  "${WORKDIR}"/grpc-proto/grpc/core/*.proto
 )
 
 # These options of the form 'Mfoo.proto=bar' instruct the codegen to use an
@@ -94,9 +99,9 @@ Mgrpc/testing/test.proto=google.golang.org/grpc/interop/grpc_testing,\
 Mgrpc/testing/payloads.proto=google.golang.org/grpc/interop/grpc_testing,\
 Mgrpc/testing/empty.proto=google.golang.org/grpc/interop/grpc_testing
 
-for src in ${SOURCES[@]}; do
+for src in "${SOURCES[@]}"; do
   echo "protoc ${src}"
-  protoc --go_out=${OPTS}:${WORKDIR}/out --go-grpc_out=${OPTS}:${WORKDIR}/out \
+  protoc --go_out="${OPTS}:${WORKDIR}/out" --go-grpc_out="${OPTS}:${WORKDIR}/out" \
     -I"." \
     -I"${WORKDIR}/grpc-proto" \
     -I"${WORKDIR}/googleapis" \
@@ -104,9 +109,9 @@ for src in ${SOURCES[@]}; do
     "${src}"
 done
 
-for src in ${LEGACY_SOURCES[@]}; do
+for src in "${LEGACY_SOURCES[@]}"; do
   echo "protoc ${src}"
-  protoc --go_out="${OPTS}:${WORKDIR}/out" --go-grpc_out="${OPTS}",require_unimplemented_servers=false:"${WORKDIR}/out" \
+  protoc --go_out="${OPTS}:${WORKDIR}/out" --go-grpc_out="${OPTS},require_unimplemented_servers=false:${WORKDIR}/out" \
     -I"." \
     -I"${WORKDIR}/grpc-proto" \
     -I"${WORKDIR}/googleapis" \
diff --git a/scripts/vet.sh b/scripts/vet.sh
index 9abb08ff..9db7b9ea 100755
--- a/scripts/vet.sh
+++ b/scripts/vet.sh
@@ -90,7 +90,7 @@ for MOD_FILE in $(find . -name 'go.mod'); do
   go mod tidy -compat=1.21
   git status --porcelain 2>&1 | fail_on_output || \
     (git status; git --no-pager diff; exit 1)
-  
+ 
   # - Collection of static analysis checks
   SC_OUT="$(mktemp)"
   staticcheck -go 1.21 -checks 'all' ./... >"${SC_OUT}" || true
@@ -121,7 +121,7 @@ XXXXX PleaseIgnoreUnused'
 
   # Ignore a false positive when operands have side affectes.
   # TODO(https://github.com/dominikh/go-tools/issues/54): Remove this once the issue is fixed in staticcheck.
-  noret_grep "(SA4000)" "${SC_OUT}" | not grep -ev "crl.go:\d*:\d*: identical expressions on the left and right side of the '||' operator (SA4000)"
+  noret_grep "(SA4000)" "${SC_OUT}" | not grep -v -e "crl.go:[0-9]\+:[0-9]\+: identical expressions on the left and right side of the '||' operator (SA4000)"
 
   # Only ignore the following deprecated types/fields/functions and exclude
   # generated code.
diff --git a/stats/opentelemetry/client_metrics.go b/stats/opentelemetry/client_metrics.go
index f9bcf466..3fbae43b 100644
--- a/stats/opentelemetry/client_metrics.go
+++ b/stats/opentelemetry/client_metrics.go
@@ -33,8 +33,8 @@ import (
 )
 
 type clientStatsHandler struct {
-	options Options
-
+	estats.MetricsRecorder
+	options       Options
 	clientMetrics clientMetrics
 }
 
@@ -52,7 +52,7 @@ func (h *clientStatsHandler) initializeMetrics() {
 
 	metrics := h.options.MetricsOptions.Metrics
 	if metrics == nil {
-		metrics = DefaultMetrics
+		metrics = DefaultMetrics()
 	}
 
 	h.clientMetrics.attemptStarted = createInt64Counter(metrics.Metrics(), "grpc.client.attempt.started", meter, otelmetric.WithUnit("attempt"), otelmetric.WithDescription("Number of client call attempts started."))
@@ -60,6 +60,12 @@ func (h *clientStatsHandler) initializeMetrics() {
 	h.clientMetrics.attemptSentTotalCompressedMessageSize = createInt64Histogram(metrics.Metrics(), "grpc.client.attempt.sent_total_compressed_message_size", meter, otelmetric.WithUnit("By"), otelmetric.WithDescription("Compressed message bytes sent per client call attempt."), otelmetric.WithExplicitBucketBoundaries(DefaultSizeBounds...))
 	h.clientMetrics.attemptRcvdTotalCompressedMessageSize = createInt64Histogram(metrics.Metrics(), "grpc.client.attempt.rcvd_total_compressed_message_size", meter, otelmetric.WithUnit("By"), otelmetric.WithDescription("Compressed message bytes received per call attempt."), otelmetric.WithExplicitBucketBoundaries(DefaultSizeBounds...))
 	h.clientMetrics.callDuration = createFloat64Histogram(metrics.Metrics(), "grpc.client.call.duration", meter, otelmetric.WithUnit("s"), otelmetric.WithDescription("Time taken by gRPC to complete an RPC from application's perspective."), otelmetric.WithExplicitBucketBoundaries(DefaultLatencyBounds...))
+
+	rm := &registryMetrics{
+		optionalLabels: h.options.MetricsOptions.OptionalLabels,
+	}
+	h.MetricsRecorder = rm
+	rm.registerMetrics(metrics, meter)
 }
 
 func (h *clientStatsHandler) unaryInterceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
diff --git a/stats/opentelemetry/opentelemetry.go b/stats/opentelemetry/opentelemetry.go
index aa5354d7..fbc1ec1c 100644
--- a/stats/opentelemetry/opentelemetry.go
+++ b/stats/opentelemetry/opentelemetry.go
@@ -30,7 +30,8 @@ import (
 	"google.golang.org/grpc/internal"
 	otelinternal "google.golang.org/grpc/stats/opentelemetry/internal"
 
-	"go.opentelemetry.io/otel/metric"
+	otelattribute "go.opentelemetry.io/otel/attribute"
+	otelmetric "go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/metric/noop"
 )
 
@@ -59,7 +60,7 @@ type MetricsOptions struct {
 	// 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).
-	MeterProvider metric.MeterProvider
+	MeterProvider otelmetric.MeterProvider
 
 	// Metrics are the metrics to instrument. Will create instrument and record telemetry
 	// for corresponding metric supported by the client and server
@@ -184,65 +185,190 @@ type attemptInfo struct {
 
 type clientMetrics struct {
 	// "grpc.client.attempt.started"
-	attemptStarted metric.Int64Counter
+	attemptStarted otelmetric.Int64Counter
 	// "grpc.client.attempt.duration"
-	attemptDuration metric.Float64Histogram
+	attemptDuration otelmetric.Float64Histogram
 	// "grpc.client.attempt.sent_total_compressed_message_size"
-	attemptSentTotalCompressedMessageSize metric.Int64Histogram
+	attemptSentTotalCompressedMessageSize otelmetric.Int64Histogram
 	// "grpc.client.attempt.rcvd_total_compressed_message_size"
-	attemptRcvdTotalCompressedMessageSize metric.Int64Histogram
-
+	attemptRcvdTotalCompressedMessageSize otelmetric.Int64Histogram
 	// "grpc.client.call.duration"
-	callDuration metric.Float64Histogram
+	callDuration otelmetric.Float64Histogram
 }
 
 type serverMetrics struct {
 	// "grpc.server.call.started"
-	callStarted metric.Int64Counter
+	callStarted otelmetric.Int64Counter
 	// "grpc.server.call.sent_total_compressed_message_size"
-	callSentTotalCompressedMessageSize metric.Int64Histogram
+	callSentTotalCompressedMessageSize otelmetric.Int64Histogram
 	// "grpc.server.call.rcvd_total_compressed_message_size"
-	callRcvdTotalCompressedMessageSize metric.Int64Histogram
+	callRcvdTotalCompressedMessageSize otelmetric.Int64Histogram
 	// "grpc.server.call.duration"
-	callDuration metric.Float64Histogram
+	callDuration otelmetric.Float64Histogram
 }
 
-func createInt64Counter(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter metric.Meter, options ...metric.Int64CounterOption) metric.Int64Counter {
+func createInt64Counter(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Int64CounterOption) otelmetric.Int64Counter {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Int64Counter{}
 	}
 	ret, err := meter.Int64Counter(string(metricName), options...)
 	if err != nil {
-		logger.Errorf("failed to register metric \"%v\", will not record", metricName)
+		logger.Errorf("failed to register metric \"%v\", will not record: %v", metricName, err)
 		return noop.Int64Counter{}
 	}
 	return ret
 }
 
-func createInt64Histogram(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter metric.Meter, options ...metric.Int64HistogramOption) metric.Int64Histogram {
+func createFloat64Counter(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Float64CounterOption) otelmetric.Float64Counter {
+	if _, ok := setOfMetrics[metricName]; !ok {
+		return noop.Float64Counter{}
+	}
+	ret, err := meter.Float64Counter(string(metricName), options...)
+	if err != nil {
+		logger.Errorf("failed to register metric \"%v\", will not record: %v", metricName, err)
+		return noop.Float64Counter{}
+	}
+	return ret
+}
+
+func createInt64Histogram(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Int64HistogramOption) otelmetric.Int64Histogram {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Int64Histogram{}
 	}
 	ret, err := meter.Int64Histogram(string(metricName), options...)
 	if err != nil {
-		logger.Errorf("failed to register metric \"%v\", will not record", metricName)
+		logger.Errorf("failed to register metric \"%v\", will not record: %v", metricName, err)
 		return noop.Int64Histogram{}
 	}
 	return ret
 }
 
-func createFloat64Histogram(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter metric.Meter, options ...metric.Float64HistogramOption) metric.Float64Histogram {
+func createFloat64Histogram(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Float64HistogramOption) otelmetric.Float64Histogram {
 	if _, ok := setOfMetrics[metricName]; !ok {
 		return noop.Float64Histogram{}
 	}
 	ret, err := meter.Float64Histogram(string(metricName), options...)
 	if err != nil {
-		logger.Errorf("failed to register metric \"%v\", will not record", metricName)
+		logger.Errorf("failed to register metric \"%v\", will not record: %v", metricName, err)
 		return noop.Float64Histogram{}
 	}
 	return ret
 }
 
+func createInt64Gauge(setOfMetrics map[estats.Metric]bool, metricName estats.Metric, meter otelmetric.Meter, options ...otelmetric.Int64GaugeOption) otelmetric.Int64Gauge {
+	if _, ok := setOfMetrics[metricName]; !ok {
+		return noop.Int64Gauge{}
+	}
+	ret, err := meter.Int64Gauge(string(metricName), options...)
+	if err != nil {
+		logger.Errorf("failed to register metric \"%v\", will not record: %v", metricName, err)
+		return noop.Int64Gauge{}
+	}
+	return ret
+}
+
+func optionFromLabels(labelKeys []string, optionalLabelKeys []string, optionalLabels []string, labelVals ...string) otelmetric.MeasurementOption {
+	var attributes []otelattribute.KeyValue
+
+	// Once it hits here lower level has guaranteed length of labelVals matches
+	// labelKeys + optionalLabelKeys.
+	for i, label := range labelKeys {
+		attributes = append(attributes, otelattribute.String(label, labelVals[i]))
+	}
+
+	for i, label := range optionalLabelKeys {
+		for _, optLabel := range optionalLabels { // o(n) could build out a set but n is currently capped at < 5
+			if label == optLabel {
+				attributes = append(attributes, otelattribute.String(label, labelVals[i+len(labelKeys)]))
+			}
+		}
+	}
+	return otelmetric.WithAttributes(attributes...)
+}
+
+// registryMetrics implements MetricsRecorder for the client and server stats
+// handlers.
+type registryMetrics struct {
+	intCounts   map[*estats.MetricDescriptor]otelmetric.Int64Counter
+	floatCounts map[*estats.MetricDescriptor]otelmetric.Float64Counter
+	intHistos   map[*estats.MetricDescriptor]otelmetric.Int64Histogram
+	floatHistos map[*estats.MetricDescriptor]otelmetric.Float64Histogram
+	intGauges   map[*estats.MetricDescriptor]otelmetric.Int64Gauge
+
+	optionalLabels []string
+}
+
+func (rm *registryMetrics) registerMetrics(metrics *estats.Metrics, 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)
+	rm.floatHistos = make(map[*estats.MetricDescriptor]otelmetric.Float64Histogram)
+	rm.intGauges = make(map[*estats.MetricDescriptor]otelmetric.Int64Gauge)
+
+	for metric := range metrics.Metrics() {
+		desc := estats.DescriptorForMetric(metric)
+		if desc == nil {
+			// Either the metric was per call or the metric is not registered.
+			// Thus, if this component ever receives the desc as a handle in
+			// record it will be a no-op.
+			continue
+		}
+		switch desc.Type {
+		case estats.MetricTypeIntCount:
+			rm.intCounts[desc] = createInt64Counter(metrics.Metrics(), desc.Name, meter, otelmetric.WithUnit(desc.Unit), otelmetric.WithDescription(desc.Description))
+		case estats.MetricTypeFloatCount:
+			rm.floatCounts[desc] = createFloat64Counter(metrics.Metrics(), desc.Name, meter, otelmetric.WithUnit(desc.Unit), otelmetric.WithDescription(desc.Description))
+		case estats.MetricTypeIntHisto:
+			rm.intHistos[desc] = createInt64Histogram(metrics.Metrics(), desc.Name, meter, otelmetric.WithUnit(desc.Unit), otelmetric.WithDescription(desc.Description), otelmetric.WithExplicitBucketBoundaries(desc.Bounds...))
+		case estats.MetricTypeFloatHisto:
+			rm.floatHistos[desc] = createFloat64Histogram(metrics.Metrics(), desc.Name, meter, otelmetric.WithUnit(desc.Unit), otelmetric.WithDescription(desc.Description), otelmetric.WithExplicitBucketBoundaries(desc.Bounds...))
+		case estats.MetricTypeIntGauge:
+			rm.intGauges[desc] = createInt64Gauge(metrics.Metrics(), desc.Name, meter, otelmetric.WithUnit(desc.Unit), otelmetric.WithDescription(desc.Description))
+		}
+	}
+}
+
+func (rm *registryMetrics) RecordInt64Count(handle *estats.Int64CountHandle, incr int64, labels ...string) {
+	desc := (*estats.MetricDescriptor)(handle)
+	ao := optionFromLabels(desc.Labels, desc.OptionalLabels, rm.optionalLabels, labels...)
+
+	if ic, ok := rm.intCounts[desc]; ok {
+		ic.Add(context.TODO(), incr, ao)
+	}
+}
+
+func (rm *registryMetrics) RecordFloat64Count(handle *estats.Float64CountHandle, incr float64, labels ...string) {
+	desc := (*estats.MetricDescriptor)(handle)
+	ao := optionFromLabels(desc.Labels, desc.OptionalLabels, rm.optionalLabels, labels...)
+	if fc, ok := rm.floatCounts[desc]; ok {
+		fc.Add(context.TODO(), incr, ao)
+	}
+}
+
+func (rm *registryMetrics) RecordInt64Histo(handle *estats.Int64HistoHandle, incr int64, labels ...string) {
+	desc := (*estats.MetricDescriptor)(handle)
+	ao := optionFromLabels(desc.Labels, desc.OptionalLabels, rm.optionalLabels, labels...)
+	if ih, ok := rm.intHistos[desc]; ok {
+		ih.Record(context.TODO(), incr, ao)
+	}
+}
+
+func (rm *registryMetrics) RecordFloat64Histo(handle *estats.Float64HistoHandle, incr float64, labels ...string) {
+	desc := (*estats.MetricDescriptor)(handle)
+	ao := optionFromLabels(desc.Labels, desc.OptionalLabels, rm.optionalLabels, labels...)
+	if fh, ok := rm.floatHistos[desc]; ok {
+		fh.Record(context.TODO(), incr, ao)
+	}
+}
+
+func (rm *registryMetrics) RecordInt64Gauge(handle *estats.Int64GaugeHandle, incr int64, labels ...string) {
+	desc := (*estats.MetricDescriptor)(handle)
+	ao := optionFromLabels(desc.Labels, desc.OptionalLabels, rm.optionalLabels, labels...)
+	if ig, ok := rm.intGauges[desc]; ok {
+		ig.Record(context.TODO(), incr, ao)
+	}
+}
+
 // Users of this component should use these bucket boundaries as part of their
 // SDK MeterProvider passed in. This component sends this as "advice" to the
 // API, which works, however this stability is not guaranteed, so for safety the
@@ -253,6 +379,13 @@ var (
 	DefaultLatencyBounds = []float64{0, 0.00001, 0.00005, 0.0001, 0.0003, 0.0006, 0.0008, 0.001, 0.002, 0.003, 0.004, 0.005, 0.006, 0.008, 0.01, 0.013, 0.016, 0.02, 0.025, 0.03, 0.04, 0.05, 0.065, 0.08, 0.1, 0.13, 0.16, 0.2, 0.25, 0.3, 0.4, 0.5, 0.65, 0.8, 1, 2, 5, 10, 20, 50, 100} // provide "advice" through API, SDK should set this too
 	// 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}
-	// DefaultMetrics are the default metrics provided by this module.
-	DefaultMetrics = estats.NewMetrics(ClientAttemptStarted, ClientAttemptDuration, ClientAttemptSentCompressedTotalMessageSize, ClientAttemptRcvdCompressedTotalMessageSize, ClientCallDuration, ServerCallStarted, ServerCallSentCompressedTotalMessageSize, ServerCallRcvdCompressedTotalMessageSize, ServerCallDuration)
+	// defaultPerCallMetrics are the default metrics provided by this module.
+	defaultPerCallMetrics = estats.NewMetrics(ClientAttemptStarted, ClientAttemptDuration, ClientAttemptSentCompressedTotalMessageSize, ClientAttemptRcvdCompressedTotalMessageSize, ClientCallDuration, ServerCallStarted, ServerCallSentCompressedTotalMessageSize, ServerCallRcvdCompressedTotalMessageSize, ServerCallDuration)
 )
+
+// DefaultMetrics returns a set of default OpenTelemetry metrics.
+//
+// This should only be invoked after init time.
+func DefaultMetrics() *estats.Metrics {
+	return defaultPerCallMetrics.Join(estats.DefaultMetrics)
+}
diff --git a/stats/opentelemetry/server_metrics.go b/stats/opentelemetry/server_metrics.go
index 6f9ad7a8..0b4246c1 100644
--- a/stats/opentelemetry/server_metrics.go
+++ b/stats/opentelemetry/server_metrics.go
@@ -33,8 +33,8 @@ import (
 )
 
 type serverStatsHandler struct {
-	options Options
-
+	estats.MetricsRecorder
+	options       Options
 	serverMetrics serverMetrics
 }
 
@@ -51,13 +51,19 @@ func (h *serverStatsHandler) initializeMetrics() {
 	}
 	metrics := h.options.MetricsOptions.Metrics
 	if metrics == nil {
-		metrics = DefaultMetrics
+		metrics = DefaultMetrics()
 	}
 
 	h.serverMetrics.callStarted = createInt64Counter(metrics.Metrics(), "grpc.server.call.started", meter, otelmetric.WithUnit("call"), otelmetric.WithDescription("Number of server calls started."))
 	h.serverMetrics.callSentTotalCompressedMessageSize = createInt64Histogram(metrics.Metrics(), "grpc.server.call.sent_total_compressed_message_size", meter, otelmetric.WithUnit("By"), otelmetric.WithDescription("Compressed message bytes sent per server call."), otelmetric.WithExplicitBucketBoundaries(DefaultSizeBounds...))
 	h.serverMetrics.callRcvdTotalCompressedMessageSize = createInt64Histogram(metrics.Metrics(), "grpc.server.call.rcvd_total_compressed_message_size", meter, otelmetric.WithUnit("By"), otelmetric.WithDescription("Compressed message bytes received per server call."), otelmetric.WithExplicitBucketBoundaries(DefaultSizeBounds...))
 	h.serverMetrics.callDuration = createFloat64Histogram(metrics.Metrics(), "grpc.server.call.duration", meter, otelmetric.WithUnit("s"), otelmetric.WithDescription("End-to-end time taken to complete a call from server transport's perspective."), otelmetric.WithExplicitBucketBoundaries(DefaultLatencyBounds...))
+
+	rm := &registryMetrics{
+		optionalLabels: h.options.MetricsOptions.OptionalLabels,
+	}
+	h.MetricsRecorder = rm
+	rm.registerMetrics(metrics, meter)
 }
 
 // attachLabelsTransportStream intercepts SetHeader and SendHeader calls of the



Test patch:
diff --git a/internal/testutils/blocking_context_dialer.go b/internal/testutils/blocking_context_dialer.go
index ea7a8519..32dfabf3 100644
--- a/internal/testutils/blocking_context_dialer.go
+++ b/internal/testutils/blocking_context_dialer.go
@@ -21,36 +21,109 @@ package testutils
 import (
 	"context"
 	"net"
+	"sync"
+
+	"google.golang.org/grpc/grpclog"
 )
 
+var logger = grpclog.Component("testutils")
+
 // BlockingDialer is a dialer that waits for Resume() to be called before
 // dialing.
 type BlockingDialer struct {
-	dialer  *net.Dialer
-	blockCh chan struct{}
+	// mu protects holds.
+	mu sync.Mutex
+	// holds maps network addresses to a list of holds for that address.
+	holds map[string][]*Hold
 }
 
 // NewBlockingDialer returns a dialer that waits for Resume() to be called
 // before dialing.
 func NewBlockingDialer() *BlockingDialer {
 	return &BlockingDialer{
-		dialer:  &net.Dialer{},
-		blockCh: make(chan struct{}),
+		holds: make(map[string][]*Hold),
 	}
 }
 
 // DialContext implements a context dialer for use with grpc.WithContextDialer
 // dial option for a BlockingDialer.
 func (d *BlockingDialer) DialContext(ctx context.Context, addr string) (net.Conn, error) {
+	d.mu.Lock()
+	holds := d.holds[addr]
+	if len(holds) == 0 {
+		// No hold for this addr.
+		d.mu.Unlock()
+		return (&net.Dialer{}).DialContext(ctx, "tcp", addr)
+	}
+	hold := holds[0]
+	d.holds[addr] = holds[1:]
+	d.mu.Unlock()
+
+	logger.Infof("Hold %p: Intercepted connection attempt to addr %q", hold, addr)
+	close(hold.waitCh)
 	select {
-	case <-d.blockCh:
+	case err := <-hold.blockCh:
+		if err != nil {
+			return nil, err
+		}
+		return (&net.Dialer{}).DialContext(ctx, "tcp", addr)
 	case <-ctx.Done():
+		logger.Infof("Hold %p: Connection attempt to addr %q timed out", hold, addr)
 		return nil, ctx.Err()
 	}
-	return d.dialer.DialContext(ctx, "tcp", addr)
 }
 
-// Resume unblocks the dialer. It panics if called more than once.
-func (d *BlockingDialer) Resume() {
-	close(d.blockCh)
+// Hold is a handle to a single connection attempt. It can be used to block,
+// fail and succeed connection attempts.
+type Hold struct {
+	// dialer is the dialer that created this hold.
+	dialer *BlockingDialer
+	// waitCh is closed when a connection attempt is received.
+	waitCh chan struct{}
+	// blockCh receives the value to return from DialContext for this connection
+	// attempt (nil on resume, an error on fail). It receives at most 1 value.
+	blockCh chan error
+	// addr is the address that this hold is for.
+	addr string
+}
+
+// Hold blocks the dialer when a connection attempt is made to the given addr.
+// A hold is valid for exactly one connection attempt. Multiple holds for an
+// addr can be added, and they will apply in the order that the connections are
+// attempted.
+func (d *BlockingDialer) Hold(addr string) *Hold {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	h := Hold{dialer: d, blockCh: make(chan error, 1), waitCh: make(chan struct{}), addr: addr}
+	d.holds[addr] = append(d.holds[addr], &h)
+	return &h
+}
+
+// Wait blocks until there is a connection attempt on this Hold, or the context
+// expires. Return false if the context has expired, true otherwise.
+func (h *Hold) Wait(ctx context.Context) bool {
+	logger.Infof("Hold %p: Waiting for a connection attempt to addr %q", h, h.addr)
+	select {
+	case <-ctx.Done():
+		return false
+	case <-h.waitCh:
+		return true
+	}
+}
+
+// Resume unblocks the dialer for the given addr. Either Resume or Fail must be
+// called at most once on a hold. Otherwise, Resume panics.
+func (h *Hold) Resume() {
+	logger.Infof("Hold %p: Resuming connection attempt to addr %q", h, h.addr)
+	h.blockCh <- nil
+	close(h.blockCh)
+}
+
+// Fail fails the connection attempt. Either Resume or Fail must be
+// called at most once on a hold. Otherwise, Resume panics.
+func (h *Hold) Fail(err error) {
+	logger.Infof("Hold %p: Failing connection attempt to addr %q", h, h.addr)
+	h.blockCh <- err
+	close(h.blockCh)
 }
diff --git a/internal/testutils/blocking_context_dialer_test.go b/internal/testutils/blocking_context_dialer_test.go
new file mode 100644
index 00000000..d2b59541
--- /dev/null
+++ b/internal/testutils/blocking_context_dialer_test.go
@@ -0,0 +1,201 @@
+/*
+ *
+ * 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 testutils
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+const (
+	testTimeout      = 5 * time.Second
+	testShortTimeout = 10 * time.Millisecond
+)
+
+func (s) TestBlockingDialer_NoHold(t *testing.T) {
+	lis, err := LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen: %v", err)
+	}
+	defer lis.Close()
+
+	d := NewBlockingDialer()
+
+	// This should not block.
+	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+	conn, err := d.DialContext(ctx, lis.Addr().String())
+	if err != nil {
+		t.Fatalf("Failed to dial: %v", err)
+	}
+	conn.Close()
+}
+
+func (s) TestBlockingDialer_HoldWaitResume(t *testing.T) {
+	lis, err := LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen: %v", err)
+	}
+	defer lis.Close()
+
+	d := NewBlockingDialer()
+	h := d.Hold(lis.Addr().String())
+
+	done := make(chan struct{})
+	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+	go func() {
+		defer close(done)
+		conn, err := d.DialContext(ctx, lis.Addr().String())
+		if err != nil {
+			t.Errorf("BlockingDialer.DialContext() got error: %v, want success", err)
+			return
+		}
+		conn.Close()
+	}()
+
+	// This should block until the goroutine above is scheduled.
+	if !h.Wait(ctx) {
+		t.Fatalf("Timeout while waiting for a connection attempt to %q", h.addr)
+	}
+	select {
+	case <-done:
+		t.Fatalf("Expected dialer to be blocked.")
+	case <-time.After(testShortTimeout):
+	}
+
+	h.Resume() // Unblock the above goroutine.
+
+	select {
+	case <-done:
+	case <-ctx.Done():
+		t.Errorf("Timeout waiting for connection attempt to resume.")
+	}
+}
+
+func (s) TestBlockingDialer_HoldWaitFail(t *testing.T) {
+	lis, err := LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen: %v", err)
+	}
+	defer lis.Close()
+
+	d := NewBlockingDialer()
+	h := d.Hold(lis.Addr().String())
+
+	wantErr := errors.New("test error")
+
+	dialError := make(chan error)
+	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+	go func() {
+		_, err := d.DialContext(ctx, lis.Addr().String())
+		dialError <- err
+	}()
+
+	if !h.Wait(ctx) {
+		t.Fatalf("Timeout while waiting for a connection attempt to " + h.addr)
+	}
+	select {
+	case err = <-dialError:
+		t.Errorf("DialContext got unblocked with err %v. Want DialContext to still be blocked after Wait()", err)
+	case <-time.After(testShortTimeout):
+	}
+
+	h.Fail(wantErr)
+
+	select {
+	case err = <-dialError:
+		if !errors.Is(err, wantErr) {
+			t.Errorf("BlockingDialer.DialContext() after Fail(): got error %v, want %v", err, wantErr)
+		}
+	case <-ctx.Done():
+		t.Errorf("Timeout waiting for connection attempt to fail.")
+	}
+}
+
+func (s) TestBlockingDialer_ContextCanceled(t *testing.T) {
+	lis, err := LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen: %v", err)
+	}
+	defer lis.Close()
+
+	d := NewBlockingDialer()
+	h := d.Hold(lis.Addr().String())
+
+	dialErr := make(chan error)
+	testCtx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+
+	ctx, cancel := context.WithCancel(testCtx)
+	defer cancel()
+	go func() {
+		_, err := d.DialContext(ctx, lis.Addr().String())
+		dialErr <- err
+	}()
+	if !h.Wait(testCtx) {
+		t.Errorf("Timeout while waiting for a connection attempt to %q", h.addr)
+	}
+
+	cancel()
+
+	select {
+	case err = <-dialErr:
+		if !errors.Is(err, context.Canceled) {
+			t.Errorf("BlockingDialer.DialContext() after context cancel: got error %v, want %v", err, context.Canceled)
+		}
+	case <-testCtx.Done():
+		t.Errorf("Timeout while waiting for Wait to return.")
+	}
+
+	h.Resume() // noop, just make sure nothing bad happen.
+}
+
+func (s) TestBlockingDialer_CancelWait(t *testing.T) {
+	lis, err := LocalTCPListener()
+	if err != nil {
+		t.Fatalf("Failed to listen: %v", err)
+	}
+	defer lis.Close()
+
+	d := NewBlockingDialer()
+	h := d.Hold(lis.Addr().String())
+
+	testCtx, cancel := context.WithTimeout(context.Background(), testTimeout)
+	defer cancel()
+
+	ctx, cancel := context.WithCancel(testCtx)
+	cancel()
+	done := make(chan struct{})
+	go func() {
+		if h.Wait(ctx) {
+			t.Errorf("Expected cancel to return false when context expires")
+		}
+		done <- struct{}{}
+	}()
+
+	select {
+	case <-done:
+	case <-testCtx.Done():
+		t.Errorf("Timeout while waiting for Wait to return.")
+	}
+}
diff --git a/internal/transport/grpchttp2/errors_test.go b/internal/transport/grpchttp2/errors_test.go
new file mode 100644
index 00000000..133efdd2
--- /dev/null
+++ b/internal/transport/grpchttp2/errors_test.go
@@ -0,0 +1,67 @@
+/*
+ *
+ * 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 grpchttp2
+
+import (
+	"testing"
+
+	"google.golang.org/grpc/internal/grpctest"
+)
+
+type s struct {
+	grpctest.Tester
+}
+
+func Test(t *testing.T) {
+	grpctest.RunSubTests(t, s{})
+}
+
+func (s) TestErrorCodeString(t *testing.T) {
+	tests := []struct {
+		err  ErrCode
+		want string
+	}{
+		// Known error cases
+		{err: ErrCodeNoError, want: "NO_ERROR"},
+		{err: ErrCodeProtocol, want: "PROTOCOL_ERROR"},
+		{err: ErrCodeInternal, want: "INTERNAL_ERROR"},
+		{err: ErrCodeFlowControl, want: "FLOW_CONTROL_ERROR"},
+		{err: ErrCodeSettingsTimeout, want: "SETTINGS_TIMEOUT"},
+		{err: ErrCodeStreamClosed, want: "STREAM_CLOSED"},
+		{err: ErrCodeFrameSize, want: "FRAME_SIZE_ERROR"},
+		{err: ErrCodeRefusedStream, want: "REFUSED_STREAM"},
+		{err: ErrCodeCancel, want: "CANCEL"},
+		{err: ErrCodeCompression, want: "COMPRESSION_ERROR"},
+		{err: ErrCodeConnect, want: "CONNECT_ERROR"},
+		{err: ErrCodeEnhanceYourCalm, want: "ENHANCE_YOUR_CALM"},
+		{err: ErrCodeIndaequateSecurity, want: "INADEQUATE_SECURITY"},
+		{err: ErrCodeHTTP11Required, want: "HTTP_1_1_REQUIRED"},
+		// Type casting known error case
+		{err: ErrCode(0x1), want: "PROTOCOL_ERROR"},
+		// Unknown error case
+		{err: ErrCode(0xf), want: "unknown error code 0xf"},
+	}
+
+	for _, test := range tests {
+		got := test.err.String()
+		if got != test.want {
+			t.Errorf("ErrCode.String(%#x) = %q, want %q", int(test.err), got, test.want)
+		}
+	}
+}
diff --git a/stats/opentelemetry/csm/observability_test.go b/stats/opentelemetry/csm/observability_test.go
index edb7a858..d2b2884c 100644
--- a/stats/opentelemetry/csm/observability_test.go
+++ b/stats/opentelemetry/csm/observability_test.go
@@ -206,13 +206,13 @@ func (s) TestCSMPluginOptionUnary(t *testing.T) {
 				serverOptionWithCSMPluginOption(opentelemetry.Options{
 					MetricsOptions: opentelemetry.MetricsOptions{
 						MeterProvider: provider,
-						Metrics:       opentelemetry.DefaultMetrics,
+						Metrics:       opentelemetry.DefaultMetrics(),
 					}}, po),
 			}
 			dopts := []grpc.DialOption{dialOptionWithCSMPluginOption(opentelemetry.Options{
 				MetricsOptions: opentelemetry.MetricsOptions{
 					MeterProvider:  provider,
-					Metrics:        opentelemetry.DefaultMetrics,
+					Metrics:        opentelemetry.DefaultMetrics(),
 					OptionalLabels: []string{"csm.service_name", "csm.service_namespace_name"},
 				},
 			}, po)}
@@ -368,13 +368,13 @@ func (s) TestCSMPluginOptionStreaming(t *testing.T) {
 				serverOptionWithCSMPluginOption(opentelemetry.Options{
 					MetricsOptions: opentelemetry.MetricsOptions{
 						MeterProvider: provider,
-						Metrics:       opentelemetry.DefaultMetrics,
+						Metrics:       opentelemetry.DefaultMetrics(),
 					}}, po),
 			}
 			dopts := []grpc.DialOption{dialOptionWithCSMPluginOption(opentelemetry.Options{
 				MetricsOptions: opentelemetry.MetricsOptions{
 					MeterProvider:  provider,
-					Metrics:        opentelemetry.DefaultMetrics,
+					Metrics:        opentelemetry.DefaultMetrics(),
 					OptionalLabels: []string{"csm.service_name", "csm.service_namespace_name"},
 				},
 			}, po)}
@@ -460,7 +460,7 @@ func (s) TestXDSLabels(t *testing.T) {
 	dopts := []grpc.DialOption{dialOptionWithCSMPluginOption(opentelemetry.Options{
 		MetricsOptions: opentelemetry.MetricsOptions{
 			MeterProvider:  provider,
-			Metrics:        opentelemetry.DefaultMetrics,
+			Metrics:        opentelemetry.DefaultMetrics(),
 			OptionalLabels: []string{"csm.service_name", "csm.service_namespace_name"},
 		},
 	}, po), grpc.WithUnaryInterceptor(unaryInterceptorAttachXDSLabels)}
diff --git a/stats/opentelemetry/e2e_test.go b/stats/opentelemetry/e2e_test.go
index 2aaa4706..0e8558aa 100644
--- a/stats/opentelemetry/e2e_test.go
+++ b/stats/opentelemetry/e2e_test.go
@@ -52,9 +52,7 @@ func Test(t *testing.T) {
 // component and the server.
 func setup(t *testing.T, methodAttributeFilter func(string) bool) (*metric.ManualReader, *stubserver.StubServer) {
 	reader := metric.NewManualReader()
-	provider := metric.NewMeterProvider(
-		metric.WithReader(reader),
-	)
+	provider := metric.NewMeterProvider(metric.WithReader(reader))
 	ss := &stubserver.StubServer{
 		UnaryCallF: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
 			return &testpb.SimpleResponse{Payload: &testpb.Payload{
@@ -74,12 +72,12 @@ func setup(t *testing.T, methodAttributeFilter func(string) bool) (*metric.Manua
 	if err := ss.Start([]grpc.ServerOption{opentelemetry.ServerOption(opentelemetry.Options{
 		MetricsOptions: opentelemetry.MetricsOptions{
 			MeterProvider:         provider,
-			Metrics:               opentelemetry.DefaultMetrics,
+			Metrics:               opentelemetry.DefaultMetrics(),
 			MethodAttributeFilter: methodAttributeFilter,
 		}})}, opentelemetry.DialOption(opentelemetry.Options{
 		MetricsOptions: opentelemetry.MetricsOptions{
 			MeterProvider: provider,
-			Metrics:       opentelemetry.DefaultMetrics,
+			Metrics:       opentelemetry.DefaultMetrics(),
 		},
 	})); err != nil {
 		t.Fatalf("Error starting endpoint server: %v", err)
diff --git a/stats/opentelemetry/example_test.go b/stats/opentelemetry/example_test.go
index 156918e2..66676f3c 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"
-	stats2 "google.golang.org/grpc/experimental/stats"
+	estats "google.golang.org/grpc/experimental/stats"
 	"google.golang.org/grpc/stats/opentelemetry"
 
 	"go.opentelemetry.io/otel/sdk/metric"
@@ -53,7 +53,7 @@ func Example_dialOption() {
 	opts := opentelemetry.Options{
 		MetricsOptions: opentelemetry.MetricsOptions{
 			MeterProvider: provider,
-			Metrics:       opentelemetry.DefaultMetrics, // equivalent to unset - distinct from empty
+			Metrics:       opentelemetry.DefaultMetrics(), // equivalent to unset - distinct from empty
 		},
 	}
 	do := opentelemetry.DialOption(opts)
@@ -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.ClientAttemptDuration, opentelemetry.ClientAttemptRcvdCompressedTotalMessageSize),
 		},
 	}
 	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: stats2.NewMetrics(), // Distinct to nil, which creates default metrics. This empty set creates no metrics.
+			Metrics: estats.NewMetrics(), // 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: stats2.NewMetrics(opentelemetry.ClientAttemptDuration, opentelemetry.ClientAttemptRcvdCompressedTotalMessageSize), // only create these metrics
+			Metrics: estats.NewMetrics(opentelemetry.ClientAttemptDuration, opentelemetry.ClientAttemptRcvdCompressedTotalMessageSize), // only create these metrics
 		},
 	}
 	do := opentelemetry.DialOption(opts)
diff --git a/stats/opentelemetry/metricsregistry_test.go b/stats/opentelemetry/metricsregistry_test.go
new file mode 100644
index 00000000..0d2bb956
--- /dev/null
+++ b/stats/opentelemetry/metricsregistry_test.go
@@ -0,0 +1,309 @@
+/*
+ * 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 opentelemetry
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	estats "google.golang.org/grpc/experimental/stats"
+	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/grpctest"
+
+	"go.opentelemetry.io/otel/attribute"
+	otelmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+)
+
+var defaultTestTimeout = 5 * time.Second
+
+type s struct {
+	grpctest.Tester
+}
+
+func Test(t *testing.T) {
+	grpctest.RunSubTests(t, s{})
+}
+
+type metricsRecorderForTest interface {
+	estats.MetricsRecorder
+	initializeMetrics()
+}
+
+func newClientStatsHandler(options MetricsOptions) metricsRecorderForTest {
+	return &clientStatsHandler{options: Options{MetricsOptions: options}}
+}
+
+func newServerStatsHandler(options MetricsOptions) metricsRecorderForTest {
+	return &serverStatsHandler{options: Options{MetricsOptions: options}}
+}
+
+// TestMetricsRegistryMetrics tests the OpenTelemetry behavior with respect to
+// registered metrics. It registers metrics in the metrics registry. It then
+// creates an OpenTelemetry client and server stats handler This test then makes
+// measurements on those instruments using one of the stats handlers, then tests
+// the expected metrics emissions, which includes default metrics and optional
+// label assertions.
+func (s) TestMetricsRegistryMetrics(t *testing.T) {
+	internal.SnapshotMetricRegistryForTesting.(func(t *testing.T))(t)
+	intCountHandle1 := estats.RegisterInt64Count(estats.MetricDescriptor{
+		Name:           "int-counter-1",
+		Description:    "Sum of calls from test",
+		Unit:           "int",
+		Labels:         []string{"int counter 1 label key"},
+		OptionalLabels: []string{"int counter 1 optional label key"},
+		Default:        true,
+	})
+	// A non default metric. If not specified in OpenTelemetry constructor, this
+	// will become a no-op, so measurements recorded on it won't show up in
+	// emitted metrics.
+	intCountHandle2 := estats.RegisterInt64Count(estats.MetricDescriptor{
+		Name:           "int-counter-2",
+		Description:    "Sum of calls from test",
+		Unit:           "int",
+		Labels:         []string{"int counter 2 label key"},
+		OptionalLabels: []string{"int counter 2 optional label key"},
+		Default:        false,
+	})
+	// Register another non default metric. This will get added to the default
+	// metrics set in the OpenTelemetry constructor options, so metrics recorded
+	// on this should show up in metrics emissions.
+	intCountHandle3 := estats.RegisterInt64Count(estats.MetricDescriptor{
+		Name:           "int-counter-3",
+		Description:    "sum of calls from test",
+		Unit:           "int",
+		Labels:         []string{"int counter 3 label key"},
+		OptionalLabels: []string{"int counter 3 optional label key"},
+		Default:        false,
+	})
+	floatCountHandle := estats.RegisterFloat64Count(estats.MetricDescriptor{
+		Name:           "float-counter",
+		Description:    "sum of calls from test",
+		Unit:           "float",
+		Labels:         []string{"float counter label key"},
+		OptionalLabels: []string{"float counter optional label key"},
+		Default:        true,
+	})
+	bounds := []float64{0, 5, 10}
+	intHistoHandle := estats.RegisterInt64Histo(estats.MetricDescriptor{
+		Name:           "int-histo",
+		Description:    "histogram of call values from tests",
+		Unit:           "int",
+		Labels:         []string{"int histo label key"},
+		OptionalLabels: []string{"int histo optional label key"},
+		Default:        true,
+		Bounds:         bounds,
+	})
+	floatHistoHandle := estats.RegisterFloat64Histo(estats.MetricDescriptor{
+		Name:           "float-histo",
+		Description:    "histogram of call values from tests",
+		Unit:           "float",
+		Labels:         []string{"float histo label key"},
+		OptionalLabels: []string{"float histo optional label key"},
+		Default:        true,
+		Bounds:         bounds,
+	})
+	intGaugeHandle := estats.RegisterInt64Gauge(estats.MetricDescriptor{
+		Name:           "simple-gauge",
+		Description:    "the most recent int emitted by test",
+		Unit:           "int",
+		Labels:         []string{"int gauge label key"},
+		OptionalLabels: []string{"int gauge optional label key"},
+		Default:        true,
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Only float optional labels are configured, so only float optional labels should show up.
+	// All required labels should show up.
+	wantMetrics := []metricdata.Metrics{
+		{
+			Name:        "int-counter-1",
+			Description: "Sum of calls from test",
+			Unit:        "int",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(attribute.String("int counter 1 label key", "int counter 1 label value")), // No optional label, not float.
+						Value:      1,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "int-counter-3",
+			Description: "sum of calls from test",
+			Unit:        "int",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(attribute.String("int counter 3 label key", "int counter 3 label value")), // No optional label, not float.
+						Value:      4,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "float-counter",
+			Description: "sum of calls from test",
+			Unit:        "float",
+			Data: metricdata.Sum[float64]{
+				DataPoints: []metricdata.DataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(attribute.String("float counter label key", "float counter label value"), attribute.String("float counter optional label key", "float counter optional label value")),
+						Value:      1.2,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "int-histo",
+			Description: "histogram of call values from tests",
+			Unit:        "int",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(attribute.String("int histo label key", "int histo label value")), // No optional label, not float.
+						Count:        1,
+						Bounds:       bounds,
+						BucketCounts: []uint64{0, 1, 0, 0},
+						Min:          metricdata.NewExtrema(int64(3)),
+						Max:          metricdata.NewExtrema(int64(3)),
+						Sum:          3,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "float-histo",
+			Description: "histogram of call values from tests",
+			Unit:        "float",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes:   attribute.NewSet(attribute.String("float histo label key", "float histo label value"), attribute.String("float histo optional label key", "float histo optional label value")),
+						Count:        1,
+						Bounds:       bounds,
+						BucketCounts: []uint64{0, 1, 0, 0},
+						Min:          metricdata.NewExtrema(float64(4.3)),
+						Max:          metricdata.NewExtrema(float64(4.3)),
+						Sum:          4.3,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "simple-gauge",
+			Description: "the most recent int emitted by test",
+			Unit:        "int",
+			Data: metricdata.Gauge[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(attribute.String("int gauge label key", "int gauge label value")), // No optional label, not float.
+						Value:      8,
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range []struct {
+		name        string
+		constructor func(options MetricsOptions) metricsRecorderForTest
+	}{
+		{
+			name:        "client stats handler",
+			constructor: newClientStatsHandler,
+		},
+		{
+			name:        "server stats handler",
+			constructor: newServerStatsHandler,
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			reader := otelmetric.NewManualReader()
+			provider := otelmetric.NewMeterProvider(otelmetric.WithReader(reader))
+
+			// This configures the defaults alongside int counter 3. All the instruments
+			// registered except int counter 2 and 3 are default, so all measurements
+			// recorded should show up in reader collected metrics except those for int
+			// counter 2.
+			// This also only toggles the float count and float histo optional labels,
+			// so only those should show up in metrics emissions. All the required
+			// labels should show up in metrics emissions.
+			mo := MetricsOptions{
+				Metrics:        DefaultMetrics().Add("int-counter-3"),
+				OptionalLabels: []string{"float counter optional label key", "float histo optional label key"},
+				MeterProvider:  provider,
+			}
+			mr := test.constructor(mo)
+			mr.initializeMetrics()
+			// These Record calls are guaranteed at a layer underneath OpenTelemetry for
+			// labels emitted to match the length of labels + optional labels.
+			intCountHandle1.Record(mr, 1, []string{"int counter 1 label value", "int counter 1 optional label value"}...)
+			// int-counter-2 is not part of metrics specified (not default), so this
+			// record call shouldn't show up in the reader.
+			intCountHandle2.Record(mr, 2, []string{"int counter 2 label value", "int counter 2 optional label value"}...)
+			// int-counter-3 is part of metrics specified, so this call should show up
+			// in the reader.
+			intCountHandle3.Record(mr, 4, []string{"int counter 3 label value", "int counter 3 optional label value"}...)
+
+			// All future recording points should show up in emissions as all of these are defaults.
+			floatCountHandle.Record(mr, 1.2, []string{"float counter label value", "float counter optional label value"}...)
+			intHistoHandle.Record(mr, 3, []string{"int histo label value", "int histo optional label value"}...)
+			floatHistoHandle.Record(mr, 4.3, []string{"float histo label value", "float histo optional label value"}...)
+			intGaugeHandle.Record(mr, 7, []string{"int gauge label value", "int gauge optional label value"}...)
+			// This second gauge call should take the place of the previous gauge call.
+			intGaugeHandle.Record(mr, 8, []string{"int gauge label value", "int gauge optional label value"}...)
+			rm := &metricdata.ResourceMetrics{}
+			reader.Collect(ctx, rm)
+			gotMetrics := map[string]metricdata.Metrics{}
+			for _, sm := range rm.ScopeMetrics {
+				for _, m := range sm.Metrics {
+					gotMetrics[m.Name] = m
+				}
+			}
+
+			for _, metric := range wantMetrics {
+				val, ok := gotMetrics[metric.Name]
+				if !ok {
+					t.Fatalf("Metric %v not present in recorded metrics", metric.Name)
+				}
+				if !metricdatatest.AssertEqual(t, metric, val, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars()) {
+					t.Fatalf("Metrics data type not equal for metric: %v", metric.Name)
+				}
+			}
+
+			// int-counter-2 is not a default metric and wasn't specified in
+			// constructor, so emissions should not show up.
+			if _, ok := gotMetrics["int-counter-2"]; ok {
+				t.Fatalf("Metric int-counter-2 present in recorded metrics, was not configured")
+			}
+		})
+	}
+}
diff --git a/test/xds/xds_server_certificate_providers_test.go b/test/xds/xds_server_certificate_providers_test.go
index 17748e03..3359bc58 100644
--- a/test/xds/xds_server_certificate_providers_test.go
+++ b/test/xds/xds_server_certificate_providers_test.go
@@ -120,7 +120,7 @@ func (s) TestServerSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *test
 			}
 			select {
 			case nackCh <- struct{}{}:
-			case <-ctx.Done():
+			default:
 			}
 			return nil
 		},
@@ -247,7 +247,7 @@ func (s) TestServerSideXDS_WithValidAndInvalidSecurityConfiguration(t *testing.T
 			}
 			select {
 			case nackCh <- struct{}{}:
-			case <-ctx.Done():
+			default:
 			}
 			return nil
 		},
diff --git a/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go b/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
index 078313fa..df6b84fb 100644
--- a/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
+++ b/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
@@ -22,7 +22,9 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"math/rand"
 	"net"
+	"slices"
 	"testing"
 	"time"
 
@@ -31,6 +33,7 @@ import (
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
@@ -43,6 +46,7 @@ import (
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
+	"google.golang.org/grpc/status"
 
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@@ -74,6 +78,15 @@ const (
 	virtualHostName = "test.server"
 )
 
+// fastConnectParams disables connection attempts backoffs and lowers delays.
+// This speeds up tests that rely on subchannel to move to transient failure.
+var fastConnectParams = grpc.ConnectParams{
+	Backoff: backoff.Config{
+		BaseDelay: 10 * time.Millisecond,
+	},
+	MinConnectTimeout: 100 * time.Millisecond,
+}
+
 // Tests the case where the ring contains a single subConn, and verifies that
 // when the server goes down, the LB policy on the client automatically
 // reconnects until the subChannel moves out of TRANSIENT_FAILURE.
@@ -99,6 +112,7 @@ func (s) TestRingHash_ReconnectToMoveOutOfTransientFailure(t *testing.T) {
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithResolvers(r),
 		grpc.WithDefaultServiceConfig(ringHashServiceConfig),
+		grpc.WithConnectParams(fastConnectParams),
 	}
 	cc, err := grpc.NewClient(r.Scheme()+":///test.server", dopts...)
 	if err != nil {
@@ -146,33 +160,28 @@ func (s) TestRingHash_ReconnectToMoveOutOfTransientFailure(t *testing.T) {
 	}
 }
 
-// startTestServiceBackends starts num stub servers.
-func startTestServiceBackends(t *testing.T, num int) ([]*stubserver.StubServer, func()) {
+// startTestServiceBackends starts num stub servers. It returns their addresses.
+// Servers are closed when the test is stopped.
+func startTestServiceBackends(t *testing.T, num int) []string {
 	t.Helper()
 
-	var servers []*stubserver.StubServer
+	addrs := make([]string, 0, num)
 	for i := 0; i < num; i++ {
-		servers = append(servers, stubserver.StartTestService(t, nil))
-	}
-
-	return servers, func() {
-		for _, server := range servers {
-			server.Stop()
-		}
+		server := stubserver.StartTestService(t, nil)
+		t.Cleanup(server.Stop)
+		addrs = append(addrs, server.Address)
 	}
+	return addrs
 }
 
-// backendOptions returns a slice of e2e.BackendOptions for the given stub
-// servers.
-func backendOptions(t *testing.T, servers []*stubserver.StubServer) []e2e.BackendOptions {
+// backendOptions returns a slice of e2e.BackendOptions for the given server
+// addresses.
+func backendOptions(t *testing.T, serverAddrs []string) []e2e.BackendOptions {
 	t.Helper()
 
 	var backendOpts []e2e.BackendOptions
-	for _, server := range servers {
-		backendOpts = append(backendOpts, e2e.BackendOptions{
-			Port:   testutils.ParsePort(t, server.Address),
-			Weight: 1,
-		})
+	for _, addr := range serverAddrs {
+		backendOpts = append(backendOpts, e2e.BackendOptions{Port: testutils.ParsePort(t, addr)})
 	}
 	return backendOpts
 }
@@ -210,10 +219,12 @@ func checkRPCSendOK(ctx context.Context, t *testing.T, client testgrpc.TestServi
 	return backendCount
 }
 
-// makeNonExistentBackends returns a slice of e2e.BackendOptions with num
-// listeners, each of which is closed immediately. Useful to simulate servers
-// that are unreachable.
-func makeNonExistentBackends(t *testing.T, num int) []e2e.BackendOptions {
+// makeNonExistentBackends returns a slice of strings with num listeners, each
+// of which is closed immediately. Useful to simulate servers that are
+// unreachable.
+func makeNonExistentBackends(t *testing.T, num int) []string {
+	t.Helper()
+
 	closedListeners := make([]net.Listener, 0, num)
 	for i := 0; i < num; i++ {
 		lis, err := testutils.LocalTCPListener()
@@ -224,16 +235,14 @@ func makeNonExistentBackends(t *testing.T, num int) []e2e.BackendOptions {
 	}
 
 	// Stop the servers that we want to be unreachable and collect their
-	// addresses.
-	backendOptions := make([]e2e.BackendOptions, 0, num)
+	// addresses. We don't close them in the loop above to make sure ports are
+	// not reused across them.
+	addrs := make([]string, 0, num)
 	for _, lis := range closedListeners {
-		backendOptions = append(backendOptions, e2e.BackendOptions{
-			Port:   testutils.ParsePort(t, lis.Addr().String()),
-			Weight: 1,
-		})
+		addrs = append(addrs, lis.Addr().String())
 		lis.Close()
 	}
-	return backendOptions
+	return addrs
 }
 
 // setupManagementServerAndResolver sets up an xDS management server, creates
@@ -244,6 +253,8 @@ func makeNonExistentBackends(t *testing.T, num int) []e2e.BackendOptions {
 //
 // Returns the management server, node ID and the xDS resolver builder.
 func setupManagementServerAndResolver(t *testing.T) (*e2e.ManagementServer, string, resolver.Builder) {
+	t.Helper()
+
 	// Start an xDS management server.
 	xdsServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
 
@@ -264,14 +275,22 @@ func setupManagementServerAndResolver(t *testing.T) (*e2e.ManagementServer, stri
 	return xdsServer, nodeID, r
 }
 
+// xdsUpdateOpts returns an e2e.UpdateOptions for the given node ID with the given xDS resources.
+func xdsUpdateOpts(nodeID string, endpoints *v3endpointpb.ClusterLoadAssignment, cluster *v3clusterpb.Cluster, route *v3routepb.RouteConfiguration, listener *v3listenerpb.Listener) e2e.UpdateOptions {
+	return e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
+		Clusters:  []*v3clusterpb.Cluster{cluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	}
+}
+
 // Tests that when an aggregate cluster is configured with ring hash policy, and
 // the first cluster is in transient failure, all RPCs are sent to the second
 // cluster using the ring hash policy.
 func (s) TestRingHash_AggregateClusterFallBackFromRingHashAtStartup(t *testing.T) {
-	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
-
-	servers, stop := startTestServiceBackends(t, 2)
-	defer stop()
+	addrs := startTestServiceBackends(t, 2)
 
 	const primaryClusterName = "new_cluster_1"
 	const primaryServiceName = "new_eds_service_1"
@@ -284,7 +303,7 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashAtStartup(t *testing.T
 		Localities: []e2e.LocalityOptions{{
 			Name:     "locality0",
 			Weight:   1,
-			Backends: makeNonExistentBackends(t, 2),
+			Backends: backendOptions(t, makeNonExistentBackends(t, 2)),
 		}},
 	})
 	ep2 := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
@@ -292,14 +311,14 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashAtStartup(t *testing.T
 		Localities: []e2e.LocalityOptions{{
 			Name:     "locality0",
 			Weight:   1,
-			Backends: backendOptions(t, servers),
+			Backends: backendOptions(t, addrs),
 		}},
 	})
 	primaryCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
 		ClusterName: primaryClusterName,
 		ServiceName: primaryServiceName,
 	})
-	secundaryCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+	secondaryCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
 		ClusterName: secondaryClusterName,
 		ServiceName: secondaryServiceName,
 	})
@@ -317,13 +336,15 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashAtStartup(t *testing.T
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	if err := xdsServer.Update(ctx, e2e.UpdateOptions{
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	updateOpts := e2e.UpdateOptions{
 		NodeID:    nodeID,
 		Endpoints: []*v3endpointpb.ClusterLoadAssignment{ep1, ep2},
-		Clusters:  []*v3clusterpb.Cluster{cluster, primaryCluster, secundaryCluster},
+		Clusters:  []*v3clusterpb.Cluster{cluster, primaryCluster, secondaryCluster},
 		Routes:    []*v3routepb.RouteConfiguration{route},
 		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
+	}
+	if err := xdsServer.Update(ctx, updateOpts); err != nil {
 		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
 
@@ -334,7 +355,8 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashAtStartup(t *testing.T
 	defer conn.Close()
 	client := testgrpc.NewTestServiceClient(conn)
 
-	gotPerBackend := checkRPCSendOK(ctx, t, client, 100)
+	const numRPCs = 100
+	gotPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
 
 	// Since this is using ring hash with the channel ID as the key, all RPCs
 	// are routed to the same backend of the secondary locality.
@@ -346,17 +368,10 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashAtStartup(t *testing.T
 	var got int
 	for backend, got = range gotPerBackend {
 	}
-	found := false
-	for _, server := range servers {
-		if backend == server.Address {
-			found = true
-			break
-		}
-	}
-	if !found {
-		t.Errorf("Got RPCs routed to an unexpected backend: %v, want one of %v", backend, servers)
+	if !slices.Contains(addrs, backend) {
+		t.Errorf("Got RPCs routed to an unexpected backend: %v, want one of %v", backend, addrs)
 	}
-	if got != 100 {
+	if got != numRPCs {
 		t.Errorf("Got %v RPCs routed to a backend, want %v", got, 100)
 	}
 }
@@ -376,21 +391,18 @@ func replaceDNSResolver(t *testing.T) *manual.Resolver {
 // logical DNS cluster, all RPCs are sent to the second cluster using the ring
 // hash policy.
 func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartup(t *testing.T) {
-	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
-
 	const edsClusterName = "eds_cluster"
 	const logicalDNSClusterName = "logical_dns_cluster"
 	const clusterName = "aggregate_cluster"
 
-	backends, stop := startTestServiceBackends(t, 1)
-	defer stop()
+	backends := startTestServiceBackends(t, 1)
 
 	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
 		ClusterName: edsClusterName,
 		Localities: []e2e.LocalityOptions{{
 			Name:     "locality0",
 			Weight:   1,
-			Backends: makeNonExistentBackends(t, 1),
+			Backends: backendOptions(t, makeNonExistentBackends(t, 1)),
 			Priority: 0,
 		}},
 	})
@@ -421,18 +433,20 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartup(
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	if err := xdsServer.Update(ctx, e2e.UpdateOptions{
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	updateOpts := e2e.UpdateOptions{
 		NodeID:    nodeID,
 		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
 		Clusters:  []*v3clusterpb.Cluster{cluster, edsCluster, logicalDNSCluster},
 		Routes:    []*v3routepb.RouteConfiguration{route},
 		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
+	}
+	if err := xdsServer.Update(ctx, updateOpts); err != nil {
 		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
 
 	dnsR := replaceDNSResolver(t)
-	dnsR.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: backends[0].Address}}})
+	dnsR.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: backends[0]}}})
 
 	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
@@ -445,7 +459,7 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartup(
 	var got string
 	for got = range gotPerBackend {
 	}
-	if want := backends[0].Address; got != want {
+	if want := backends[0]; got != want {
 		t.Errorf("Got RPCs routed to an unexpected got: %v, want %v", got, want)
 	}
 }
@@ -454,22 +468,18 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartup(
 // it's first child is in transient failure, and the fallback is a logical DNS,
 // the later recovers from transient failure when its backend becomes available.
 func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartupNoFailedRPCs(t *testing.T) {
-	// https://github.com/grpc/grpc/blob/083bbee4805c14ce62e6c9535fe936f68b854c4f/test/cpp/end2end/xds/xds_ring_hash_end2end_test.cc#L225
-	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
-
 	const edsClusterName = "eds_cluster"
 	const logicalDNSClusterName = "logical_dns_cluster"
 	const clusterName = "aggregate_cluster"
 
-	backends, stop := startTestServiceBackends(t, 1)
-	defer stop()
+	backends := startTestServiceBackends(t, 1)
 
 	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
 		ClusterName: edsClusterName,
 		Localities: []e2e.LocalityOptions{{
 			Name:     "locality0",
 			Weight:   1,
-			Backends: makeNonExistentBackends(t, 1),
+			Backends: backendOptions(t, makeNonExistentBackends(t, 1)),
 			Priority: 0,
 		}},
 	})
@@ -500,18 +510,20 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartupN
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	if err := xdsServer.Update(ctx, e2e.UpdateOptions{
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	updateOpts := e2e.UpdateOptions{
 		NodeID:    nodeID,
 		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
 		Clusters:  []*v3clusterpb.Cluster{cluster, edsCluster, logicalDNSCluster},
 		Routes:    []*v3routepb.RouteConfiguration{route},
 		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
+	}
+	if err := xdsServer.Update(ctx, updateOpts); err != nil {
 		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
 
 	dnsR := replaceDNSResolver(t)
-	dnsR.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: backends[0].Address}}})
+	dnsR.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: backends[0]}}})
 
 	dialer := testutils.NewBlockingDialer()
 	cp := grpc.ConnectParams{
@@ -522,18 +534,20 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartupN
 		},
 		MinConnectTimeout: 0,
 	}
-	opts := []grpc.DialOption{
+	dopts := []grpc.DialOption{
 		grpc.WithResolvers(xdsResolver),
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithContextDialer(dialer.DialContext),
 		grpc.WithConnectParams(cp)}
-	conn, err := grpc.NewClient("xds:///test.server", opts...)
+	conn, err := grpc.NewClient("xds:///test.server", dopts...)
 	if err != nil {
 		t.Fatalf("Failed to create client: %s", err)
 	}
 	defer conn.Close()
 	client := testgrpc.NewTestServiceClient(conn)
 
+	hold := dialer.Hold(backends[0])
+
 	errCh := make(chan error, 2)
 	go func() {
 		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
@@ -562,15 +576,19 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartupN
 		errCh <- nil
 	}()
 
+	// Wait for a connection attempt to backends[0].
+	if !hold.Wait(ctx) {
+		t.Fatalf("Timeout while waiting for a connection attempt to %s", backends[0])
+	}
 	// Allow the connection attempts to complete.
-	dialer.Resume()
+	hold.Resume()
 
 	// RPCs should complete successfully.
 	for range []int{0, 1} {
 		select {
 		case err := <-errCh:
 			if err != nil {
-				t.Errorf("Expected 2 rpc to succeed, but failed: %v", err)
+				t.Errorf("Expected 2 rpc to succeed, but at least one failed: %v", err)
 			}
 		case <-ctx.Done():
 			t.Fatalf("Timed out waiting for RPCs to complete")
@@ -578,22 +596,37 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartupN
 	}
 }
 
-// Tests that ring hash policy that hashes using channel id ensures all RPCs to
-// go 1 particular backend.
-func (s) TestRingHash_ChannelIdHashing(t *testing.T) {
-	backends, stop := startTestServiceBackends(t, 4)
-	defer stop()
+// endpointResource creates a ClusterLoadAssignment containing a single locality
+// with the given addresses.
+func endpointResource(t *testing.T, clusterName string, addrs []string) *v3endpointpb.ClusterLoadAssignment {
+	t.Helper()
 
-	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	// We must set the host name socket address in EDS, as the ring hash policy
+	// uses it to construct the ring.
+	host, _, err := net.SplitHostPort(addrs[0])
+	if err != nil {
+		t.Fatalf("Failed to split host and port from stubserver: %v", err)
+	}
 
-	const clusterName = "cluster"
-	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+	return e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
 		ClusterName: clusterName,
+		Host:        host,
 		Localities: []e2e.LocalityOptions{{
-			Backends: backendOptions(t, backends),
+			Backends: backendOptions(t, addrs),
 			Weight:   1,
 		}},
 	})
+}
+
+// Tests that ring hash policy that hashes using channel id ensures all RPCs to
+// go 1 particular backend.
+func (s) TestRingHash_ChannelIdHashing(t *testing.T) {
+	backends := startTestServiceBackends(t, 4)
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+
+	const clusterName = "cluster"
+	endpoints := endpointResource(t, clusterName, backends)
 	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
 		ClusterName: clusterName,
 		ServiceName: clusterName,
@@ -605,13 +638,7 @@ func (s) TestRingHash_ChannelIdHashing(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	if err := xdsServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
-		Clusters:  []*v3clusterpb.Cluster{cluster},
-		Routes:    []*v3routepb.RouteConfiguration{route},
-		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
 		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
 
@@ -622,15 +649,16 @@ func (s) TestRingHash_ChannelIdHashing(t *testing.T) {
 	defer conn.Close()
 	client := testgrpc.NewTestServiceClient(conn)
 
-	received := checkRPCSendOK(ctx, t, client, 100)
+	const numRPCs = 100
+	received := checkRPCSendOK(ctx, t, client, numRPCs)
 	if len(received) != 1 {
 		t.Errorf("Got RPCs routed to %v backends, want %v", len(received), 1)
 	}
-	var count int
-	for _, count = range received {
+	var got int
+	for _, got = range received {
 	}
-	if count != 100 {
-		t.Errorf("Got %v RPCs routed to a backend, want %v", count, 100)
+	if got != numRPCs {
+		t.Errorf("Got %v RPCs routed to a backend, want %v", got, numRPCs)
 	}
 }
 
@@ -650,30 +678,15 @@ func headerHashRoute(routeName, virtualHostName, clusterName, header string) *v3
 	return route
 }
 
-// Tests that ring hash policy that hashes using a header value can spread RPCs
-// across all the backends.
+// Tests that ring hash policy that hashes using a header value can send RPCs
+// to specific backends based on their hash.
 func (s) TestRingHash_HeaderHashing(t *testing.T) {
-	backends, stop := startTestServiceBackends(t, 4)
-	defer stop()
-
-	// We must set the host name socket address in EDS, as the ring hash policy
-	// uses it to construct the ring.
-	host, _, err := net.SplitHostPort(backends[0].Address)
-	if err != nil {
-		t.Fatalf("Failed to split host and port from stubserver: %v", err)
-	}
+	backends := startTestServiceBackends(t, 4)
 
 	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
 
 	const clusterName = "cluster"
-	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
-		ClusterName: clusterName,
-		Host:        host,
-		Localities: []e2e.LocalityOptions{{
-			Backends: backendOptions(t, backends),
-			Weight:   1,
-		}},
-	})
+	endpoints := endpointResource(t, clusterName, backends)
 	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
 		ClusterName: clusterName,
 		ServiceName: clusterName,
@@ -685,13 +698,7 @@ func (s) TestRingHash_HeaderHashing(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	if err := xdsServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
-		Clusters:  []*v3clusterpb.Cluster{cluster},
-		Routes:    []*v3routepb.RouteConfiguration{route},
-		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
 		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
 
@@ -706,10 +713,11 @@ func (s) TestRingHash_HeaderHashing(t *testing.T) {
 	// to a specific backend as the header value matches the value used to
 	// create the entry in the ring.
 	for _, backend := range backends {
-		ctx := metadata.NewOutgoingContext(ctx, metadata.Pairs("address_hash", backend.Address+"_0"))
-		reqPerBackend := checkRPCSendOK(ctx, t, client, 1)
-		if reqPerBackend[backend.Address] != 1 {
-			t.Errorf("Got RPC routed to backend %v, want %v", reqPerBackend, backend.Address)
+		ctx := metadata.NewOutgoingContext(ctx, metadata.Pairs("address_hash", backend+"_0"))
+		numRPCs := 10
+		reqPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
+		if reqPerBackend[backend] != numRPCs {
+			t.Errorf("Got RPC routed to addresses %v, want all RPCs routed to %v", reqPerBackend, backend)
 		}
 	}
 }
@@ -717,27 +725,10 @@ func (s) TestRingHash_HeaderHashing(t *testing.T) {
 // Tests that ring hash policy that hashes using a header value and regex
 // rewrite to aggregate RPCs to 1 backend.
 func (s) TestRingHash_HeaderHashingWithRegexRewrite(t *testing.T) {
-	backends, stop := startTestServiceBackends(t, 4)
-	defer stop()
-
-	// We must set the host name socket address in EDS, as the ring hash policy
-	// uses it to construct the ring.
-	host, _, err := net.SplitHostPort(backends[0].Address)
-	if err != nil {
-		t.Fatalf("Failed to split host and port from stubserver: %v", err)
-	}
-
-	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	backends := startTestServiceBackends(t, 4)
 
 	clusterName := "cluster"
-	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
-		ClusterName: clusterName,
-		Host:        host,
-		Localities: []e2e.LocalityOptions{{
-			Backends: backendOptions(t, backends),
-			Weight:   1,
-		}},
-	})
+	endpoints := endpointResource(t, clusterName, backends)
 	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
 		ClusterName: clusterName,
 		ServiceName: clusterName,
@@ -757,13 +748,8 @@ func (s) TestRingHash_HeaderHashingWithRegexRewrite(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	if err := xdsServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
-		Clusters:  []*v3clusterpb.Cluster{cluster},
-		Routes:    []*v3routepb.RouteConfiguration{route},
-		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
 		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
 
@@ -781,7 +767,7 @@ func (s) TestRingHash_HeaderHashingWithRegexRewrite(t *testing.T) {
 	// hashing to the same value.
 	gotPerBackend := make(map[string]int)
 	for _, backend := range backends {
-		ctx := metadata.NewOutgoingContext(ctx, metadata.Pairs("address_hash", backend.Address+"_0"))
+		ctx := metadata.NewOutgoingContext(ctx, metadata.Pairs("address_hash", backend+"_0"))
 		res := checkRPCSendOK(ctx, t, client, 100)
 		for addr, count := range res {
 			gotPerBackend[addr] += count
@@ -816,7 +802,7 @@ func computeIdealNumberOfRPCs(t *testing.T, p, errorTolerance float64) int {
 // minimum ring size to ensure that the ring is large enough to distribute
 // requests more uniformly across endpoints when a random hash is used.
 func setRingHashLBPolicyWithHighMinRingSize(t *testing.T, cluster *v3clusterpb.Cluster) {
-	minRingSize := uint64(100000)
+	const minRingSize = 100000
 	oldVal := envconfig.RingHashCap
 	envconfig.RingHashCap = minRingSize
 	t.Cleanup(func() {
@@ -839,20 +825,11 @@ func setRingHashLBPolicyWithHighMinRingSize(t *testing.T, cluster *v3clusterpb.C
 
 // Tests that ring hash policy that hashes using a random value.
 func (s) TestRingHash_NoHashPolicy(t *testing.T) {
-	backends, stop := startTestServiceBackends(t, 2)
-	defer stop()
+	backends := startTestServiceBackends(t, 2)
 	numRPCs := computeIdealNumberOfRPCs(t, .5, errorTolerance)
 
-	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
-
 	const clusterName = "cluster"
-	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
-		ClusterName: clusterName,
-		Localities: []e2e.LocalityOptions{{
-			Backends: backendOptions(t, backends),
-			Weight:   1,
-		}},
-	})
+	endpoints := endpointResource(t, clusterName, backends)
 	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
 		ClusterName: clusterName,
 		ServiceName: clusterName,
@@ -864,13 +841,8 @@ func (s) TestRingHash_NoHashPolicy(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	if err := xdsServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
-		Clusters:  []*v3clusterpb.Cluster{cluster},
-		Routes:    []*v3routepb.RouteConfiguration{route},
-		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
 		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
 
@@ -884,26 +856,23 @@ func (s) TestRingHash_NoHashPolicy(t *testing.T) {
 	// Send a large number of RPCs and check that they are distributed randomly.
 	gotPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
 	for _, backend := range backends {
-		got := float64(gotPerBackend[backend.Address]) / float64(numRPCs)
+		got := float64(gotPerBackend[backend]) / float64(numRPCs)
 		want := .5
 		if !cmp.Equal(got, want, cmpopts.EquateApprox(0, errorTolerance)) {
-			t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[2].Address, got, want, errorTolerance)
+			t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backend, got, want, errorTolerance)
 		}
 	}
 }
 
 // Tests that we observe endpoint weights.
 func (s) TestRingHash_EndpointWeights(t *testing.T) {
-	backends, stop := startTestServiceBackends(t, 3)
-	defer stop()
-
-	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	backends := startTestServiceBackends(t, 3)
 
 	const clusterName = "cluster"
 	backendOpts := []e2e.BackendOptions{
-		{Port: testutils.ParsePort(t, backends[0].Address)},
-		{Port: testutils.ParsePort(t, backends[1].Address)},
-		{Port: testutils.ParsePort(t, backends[2].Address), Weight: 2},
+		{Port: testutils.ParsePort(t, backends[0])},
+		{Port: testutils.ParsePort(t, backends[1])},
+		{Port: testutils.ParsePort(t, backends[2]), Weight: 2},
 	}
 
 	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
@@ -928,13 +897,8 @@ func (s) TestRingHash_EndpointWeights(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	if err := xdsServer.Update(ctx, e2e.UpdateOptions{
-		NodeID:    nodeID,
-		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
-		Clusters:  []*v3clusterpb.Cluster{cluster},
-		Routes:    []*v3routepb.RouteConfiguration{route},
-		Listeners: []*v3listenerpb.Listener{listener},
-	}); err != nil {
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
 		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
 
@@ -949,18 +913,656 @@ func (s) TestRingHash_EndpointWeights(t *testing.T) {
 	numRPCs := computeIdealNumberOfRPCs(t, .25, errorTolerance)
 	gotPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
 
-	got := float64(gotPerBackend[backends[0].Address]) / float64(numRPCs)
+	got := float64(gotPerBackend[backends[0]]) / float64(numRPCs)
 	want := .25
 	if !cmp.Equal(got, want, cmpopts.EquateApprox(0, errorTolerance)) {
-		t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[0].Address, got, want, errorTolerance)
+		t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[0], got, want, errorTolerance)
 	}
-	got = float64(gotPerBackend[backends[1].Address]) / float64(numRPCs)
+	got = float64(gotPerBackend[backends[1]]) / float64(numRPCs)
 	if !cmp.Equal(got, want, cmpopts.EquateApprox(0, errorTolerance)) {
-		t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[1].Address, got, want, errorTolerance)
+		t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[1], got, want, errorTolerance)
 	}
-	got = float64(gotPerBackend[backends[2].Address]) / float64(numRPCs)
+	got = float64(gotPerBackend[backends[2]]) / float64(numRPCs)
 	want = .50
 	if !cmp.Equal(got, want, cmpopts.EquateApprox(0, errorTolerance)) {
-		t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[2].Address, got, want, errorTolerance)
+		t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[2], got, want, errorTolerance)
+	}
+}
+
+// Tests that ring hash policy evaluation will continue past the terminal hash
+// policy if no results are produced yet.
+func (s) TestRingHash_ContinuesPastTerminalPolicyThatDoesNotProduceResult(t *testing.T) {
+	backends := startTestServiceBackends(t, 2)
+
+	const clusterName = "cluster"
+	endpoints := endpointResource(t, clusterName, backends)
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+
+	route := e2e.DefaultRouteConfig("new_route", "test.server", clusterName)
+
+	// Even though this hash policy is terminal, since it produces no result, we
+	// continue past it to find a policy that produces results.
+	hashPolicy := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_Header_{
+			Header: &v3routepb.RouteAction_HashPolicy_Header{
+				HeaderName: "header_not_present",
+			},
+		},
+		Terminal: true,
+	}
+	hashPolicy2 := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_Header_{
+			Header: &v3routepb.RouteAction_HashPolicy_Header{
+				HeaderName: "address_hash",
+			},
+		},
+	}
+	action := route.VirtualHosts[0].Routes[0].Action.(*v3routepb.Route_Route)
+	action.Route.HashPolicy = []*v3routepb.RouteAction_HashPolicy{&hashPolicy, &hashPolicy2}
+
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	// - The first hash policy does not match because the header is not present.
+	//   If this hash policy was applied, it would spread the load across
+	//   backend 0 and 1, since a random hash would be used.
+	// - In the second hash policy, each type of RPC contains a header
+	//   value that always hashes to backend 0, as the header value
+	//   matches the value used to create the entry in the ring.
+	// We verify that the second hash policy is used by checking that all RPCs
+	// are being routed to backend 0.
+	wantBackend := backends[0]
+	ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs("address_hash", wantBackend+"_0"))
+	const numRPCs = 100
+	gotPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
+	if got := gotPerBackend[wantBackend]; got != numRPCs {
+		t.Errorf("Got %v RPCs routed to backend %v, want %v", got, wantBackend, numRPCs)
+	}
+}
+
+// Tests that a random hash is used when header hashing policy specified a
+// header field that the RPC did not have.
+func (s) TestRingHash_HashOnHeaderThatIsNotPresent(t *testing.T) {
+	backends := startTestServiceBackends(t, 2)
+	wantFractionPerBackend := .5
+	numRPCs := computeIdealNumberOfRPCs(t, wantFractionPerBackend, errorTolerance)
+
+	const clusterName = "cluster"
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Localities: []e2e.LocalityOptions{{
+			Backends: backendOptions(t, backends),
+			Weight:   1,
+		}},
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+	})
+	setRingHashLBPolicyWithHighMinRingSize(t, cluster)
+	route := headerHashRoute("new_route", virtualHostName, clusterName, "header_not_present")
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	// The first hash policy does not apply because the header is not present in
+	// the RPCs that we are about to send. As a result, a random hash should be
+	// used instead, resulting in a random request distribution.
+	// We verify this by checking that the RPCs are distributed randomly.
+	gotPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
+	for _, backend := range backends {
+		got := float64(gotPerBackend[backend]) / float64(numRPCs)
+		if !cmp.Equal(got, wantFractionPerBackend, cmpopts.EquateApprox(0, errorTolerance)) {
+			t.Errorf("fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backend, got, wantFractionPerBackend, errorTolerance)
+		}
+	}
+}
+
+// Tests that a random hash is used when only unsupported hash policies are
+// configured.
+func (s) TestRingHash_UnsupportedHashPolicyDefaultToRandomHashing(t *testing.T) {
+	backends := startTestServiceBackends(t, 2)
+	wantFractionPerBackend := .5
+	numRPCs := computeIdealNumberOfRPCs(t, wantFractionPerBackend, errorTolerance)
+
+	const clusterName = "cluster"
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Localities: []e2e.LocalityOptions{{
+			Backends: backendOptions(t, backends),
+			Weight:   1,
+		}},
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+	})
+	setRingHashLBPolicyWithHighMinRingSize(t, cluster)
+	route := e2e.DefaultRouteConfig("new_route", "test.server", clusterName)
+	unsupportedHashPolicy1 := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_Cookie_{
+			Cookie: &v3routepb.RouteAction_HashPolicy_Cookie{Name: "cookie"},
+		},
+	}
+	unsupportedHashPolicy2 := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_ConnectionProperties_{
+			ConnectionProperties: &v3routepb.RouteAction_HashPolicy_ConnectionProperties{SourceIp: true},
+		},
+	}
+	unsupportedHashPolicy3 := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_QueryParameter_{
+			QueryParameter: &v3routepb.RouteAction_HashPolicy_QueryParameter{Name: "query_parameter"},
+		},
+	}
+	action := route.VirtualHosts[0].Routes[0].Action.(*v3routepb.Route_Route)
+	action.Route.HashPolicy = []*v3routepb.RouteAction_HashPolicy{&unsupportedHashPolicy1, &unsupportedHashPolicy2, &unsupportedHashPolicy3}
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	// Since none of the hash policy are supported, a random hash should be
+	// generated for every request.
+	// We verify this by checking that the RPCs are distributed randomly.
+	gotPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
+	for _, backend := range backends {
+		got := float64(gotPerBackend[backend]) / float64(numRPCs)
+		if !cmp.Equal(got, wantFractionPerBackend, cmpopts.EquateApprox(0, errorTolerance)) {
+			t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backend, got, wantFractionPerBackend, errorTolerance)
+		}
+	}
+}
+
+// Tests that unsupported hash policy types are all ignored before a supported
+// hash policy.
+func (s) TestRingHash_UnsupportedHashPolicyUntilChannelIdHashing(t *testing.T) {
+	backends := startTestServiceBackends(t, 2)
+
+	const clusterName = "cluster"
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Localities: []e2e.LocalityOptions{{
+			Backends: backendOptions(t, backends),
+			Weight:   1,
+		}},
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+	})
+	setRingHashLBPolicyWithHighMinRingSize(t, cluster)
+	route := e2e.DefaultRouteConfig("new_route", "test.server", clusterName)
+	unsupportedHashPolicy1 := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_Cookie_{
+			Cookie: &v3routepb.RouteAction_HashPolicy_Cookie{Name: "cookie"},
+		},
+	}
+	unsupportedHashPolicy2 := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_ConnectionProperties_{
+			ConnectionProperties: &v3routepb.RouteAction_HashPolicy_ConnectionProperties{SourceIp: true},
+		},
+	}
+	unsupportedHashPolicy3 := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_QueryParameter_{
+			QueryParameter: &v3routepb.RouteAction_HashPolicy_QueryParameter{Name: "query_parameter"},
+		},
+	}
+	channelIDhashPolicy := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_FilterState_{
+			FilterState: &v3routepb.RouteAction_HashPolicy_FilterState{
+				Key: "io.grpc.channel_id",
+			},
+		},
+	}
+	action := route.VirtualHosts[0].Routes[0].Action.(*v3routepb.Route_Route)
+	action.Route.HashPolicy = []*v3routepb.RouteAction_HashPolicy{&unsupportedHashPolicy1, &unsupportedHashPolicy2, &unsupportedHashPolicy3, &channelIDhashPolicy}
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	// Since only unsupported policies are present except for the last one
+	// which is using the channel ID hashing policy, all requests should be
+	// routed to the same backend.
+	const numRPCs = 100
+	gotPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
+	if len(gotPerBackend) != 1 {
+		t.Errorf("Got RPCs routed to %v backends, want 1", len(gotPerBackend))
+	}
+	var got int
+	for _, got = range gotPerBackend {
+	}
+	if got != numRPCs {
+		t.Errorf("Got %v RPCs routed to a backend, want %v", got, numRPCs)
+	}
+}
+
+// Tests that ring hash policy that hashes using a random value can spread RPCs
+// across all the backends according to locality weight.
+func (s) TestRingHash_RandomHashingDistributionAccordingToLocalityAndEndpointWeight(t *testing.T) {
+	backends := startTestServiceBackends(t, 2)
+
+	const clusterName = "cluster"
+	const locality1Weight = uint32(1)
+	const endpoint1Weight = uint32(1)
+	const locality2Weight = uint32(2)
+	const endpoint2Weight = uint32(2)
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Localities: []e2e.LocalityOptions{
+			{
+				Backends: []e2e.BackendOptions{{
+					Port:   testutils.ParsePort(t, backends[0]),
+					Weight: endpoint1Weight,
+				}},
+				Weight: locality1Weight,
+			},
+			{
+				Backends: []e2e.BackendOptions{{
+					Port:   testutils.ParsePort(t, backends[1]),
+					Weight: endpoint2Weight,
+				}},
+				Weight: locality2Weight,
+			},
+		},
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+	})
+	setRingHashLBPolicyWithHighMinRingSize(t, cluster)
+	route := e2e.DefaultRouteConfig("new_route", "test.server", clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	const weight1 = endpoint1Weight * locality1Weight
+	const weight2 = endpoint2Weight * locality2Weight
+	const wantRPCs1 = float64(weight1) / float64(weight1+weight2)
+	const wantRPCs2 = float64(weight2) / float64(weight1+weight2)
+	numRPCs := computeIdealNumberOfRPCs(t, math.Min(wantRPCs1, wantRPCs2), errorTolerance)
+
+	// Send a large number of RPCs and check that they are distributed randomly.
+	gotPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)
+	got := float64(gotPerBackend[backends[0]]) / float64(numRPCs)
+	if !cmp.Equal(got, wantRPCs1, cmpopts.EquateApprox(0, errorTolerance)) {
+		t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[2], got, wantRPCs1, errorTolerance)
+	}
+	got = float64(gotPerBackend[backends[1]]) / float64(numRPCs)
+	if !cmp.Equal(got, wantRPCs2, cmpopts.EquateApprox(0, errorTolerance)) {
+		t.Errorf("Fraction of RPCs to backend %s: got %v, want %v (margin: +-%v)", backends[2], got, wantRPCs2, errorTolerance)
+	}
+}
+
+// Tests that ring hash policy that hashes using a fixed string ensures all RPCs
+// to go 1 particular backend; and that subsequent hashing policies are ignored
+// due to the setting of terminal.
+func (s) TestRingHash_FixedHashingTerminalPolicy(t *testing.T) {
+	backends := startTestServiceBackends(t, 2)
+	const clusterName = "cluster"
+	endpoints := endpointResource(t, clusterName, backends)
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+
+	route := e2e.DefaultRouteConfig("new_route", "test.server", clusterName)
+
+	hashPolicy := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_Header_{
+			Header: &v3routepb.RouteAction_HashPolicy_Header{
+				HeaderName: "fixed_string",
+			},
+		},
+		Terminal: true,
+	}
+	hashPolicy2 := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_Header_{
+			Header: &v3routepb.RouteAction_HashPolicy_Header{
+				HeaderName: "random_string",
+			},
+		},
+	}
+	action := route.VirtualHosts[0].Routes[0].Action.(*v3routepb.Route_Route)
+	action.Route.HashPolicy = []*v3routepb.RouteAction_HashPolicy{&hashPolicy, &hashPolicy2}
+
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	// Check that despite the matching random string header, since the fixed
+	// string hash policy is terminal, only the fixed string hash policy applies
+	// and requests all get routed to the same host.
+	gotPerBackend := make(map[string]int)
+	const numRPCs = 100
+	for i := 0; i < numRPCs; i++ {
+		ctx := metadata.NewOutgoingContext(ctx, metadata.Pairs(
+			"fixed_string", backends[0]+"_0",
+			"random_string", fmt.Sprintf("%d", rand.Int())),
+		)
+		var remote peer.Peer
+		_, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.Peer(&remote))
+		if err != nil {
+			t.Fatalf("rpc EmptyCall() failed: %v", err)
+		}
+		gotPerBackend[remote.Addr.String()]++
+	}
+
+	if len(gotPerBackend) != 1 {
+		t.Error("Got RPCs routed to multiple backends, want a single backend")
+	}
+	if got := gotPerBackend[backends[0]]; got != numRPCs {
+		t.Errorf("Got %v RPCs routed to %v, want %v", got, backends[0], numRPCs)
+	}
+}
+
+// TestRingHash_IdleToReady tests that the channel will go from idle to ready
+// via connecting; (though it is not possible to catch the connecting state
+// before moving to ready via the public API).
+// TODO: we should be able to catch all state transitions by using the internal.SubscribeToConnectivityStateChanges API.
+func (s) TestRingHash_IdleToReady(t *testing.T) {
+	backends := startTestServiceBackends(t, 1)
+
+	const clusterName = "cluster"
+	endpoints := endpointResource(t, clusterName, backends)
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := channelIDHashRoute("new_route", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	if got, want := conn.GetState(), connectivity.Idle; got != want {
+		t.Errorf("conn.GetState(): got %v, want %v", got, want)
+	}
+
+	checkRPCSendOK(ctx, t, client, 1)
+
+	if got, want := conn.GetState(), connectivity.Ready; got != want {
+		t.Errorf("conn.GetState(): got %v, want %v", got, want)
+	}
+}
+
+// Test that the channel will transition to READY once it starts
+// connecting even if there are no RPCs being sent to the picker.
+func (s) TestRingHash_ContinuesConnectingWithoutPicks(t *testing.T) {
+	backend := stubserver.StartTestService(t, &stubserver.StubServer{
+		// We expect the server EmptyCall to not be call here because the
+		// aggregated channel state is never READY when the call is pending.
+		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+			t.Errorf("EmptyCall() should not have been called")
+			return &testpb.Empty{}, nil
+		},
+	})
+	defer backend.Stop()
+
+	nonExistantServerAddr := makeNonExistentBackends(t, 1)[0]
+
+	const clusterName = "cluster"
+	endpoints := endpointResource(t, clusterName, []string{backend.Address, nonExistantServerAddr})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := headerHashRoute("new_route", virtualHostName, clusterName, "address_hash")
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	dialer := testutils.NewBlockingDialer()
+	dopts := []grpc.DialOption{
+		grpc.WithResolvers(xdsResolver),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithContextDialer(dialer.DialContext),
+	}
+	conn, err := grpc.NewClient("xds:///test.server", dopts...)
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	hold := dialer.Hold(backend.Address)
+
+	rpcCtx, rpcCancel := context.WithCancel(ctx)
+	go func() {
+		rpcCtx = metadata.NewOutgoingContext(rpcCtx, metadata.Pairs("address_hash", nonExistantServerAddr+"_0"))
+		_, err := client.EmptyCall(rpcCtx, &testpb.Empty{})
+		if status.Code(err) != codes.Canceled {
+			t.Errorf("Expected RPC to be canceled, got error: %v", err)
+		}
+	}()
+
+	// Wait for the connection attempt to the real backend.
+	if !hold.Wait(ctx) {
+		t.Fatalf("Timeout waiting for connection attempt to backend %v.", backend.Address)
+	}
+	// Now cancel the RPC while we are still connecting.
+	rpcCancel()
+
+	// This allows the connection attempts to continue. The RPC was cancelled
+	// before the backend was connected, but the backend is up. The conn
+	// becomes Ready due to the connection attempt to the existing backend
+	// succeeding, despite no new RPC being sent.
+	hold.Resume()
+
+	testutils.AwaitState(ctx, t, conn, connectivity.Ready)
+}
+
+// Tests that when the first pick is down leading to a transient failure, we
+// will move on to the next ring hash entry.
+func (s) TestRingHash_TransientFailureCheckNextOne(t *testing.T) {
+	backends := startTestServiceBackends(t, 1)
+	nonExistentBackends := makeNonExistentBackends(t, 1)
+
+	const clusterName = "cluster"
+	endpoints := endpointResource(t, clusterName, append(nonExistentBackends, backends...))
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := headerHashRoute("new_route", virtualHostName, clusterName, "address_hash")
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	// Note each type of RPC contains a header value that will always be hashed
+	// the value that was used to place the non-existent endpoint on the ring,
+	// but it still gets routed to the backend that is up.
+	ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs("address_hash", nonExistentBackends[0]+"_0"))
+	reqPerBackend := checkRPCSendOK(ctx, t, client, 1)
+	var got string
+	for got = range reqPerBackend {
+	}
+	if want := backends[0]; got != want {
+		t.Errorf("Got RPC routed to addr %v, want %v", got, want)
+	}
+}
+
+// Tests for a bug seen in the wild in c-core, where ring_hash started with no
+// endpoints and reported TRANSIENT_FAILURE, then got an update with endpoints
+// and reported IDLE, but the picker update was squelched, so it failed to ever
+// get reconnected.
+func (s) TestRingHash_ReattemptWhenGoingFromTransientFailureToIdle(t *testing.T) {
+	const clusterName = "cluster"
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Localities:  []e2e.LocalityOptions{{}}, // note the empty locality (no endpoint).
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := e2e.DefaultRouteConfig("new_route", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	if got, want := conn.GetState(), connectivity.Idle; got != want {
+		t.Errorf("conn.GetState(): got %v, want %v", got, want)
+	}
+
+	// There are no endpoints in EDS. RPCs should fail and the channel should
+	// transition to transient failure.
+	if _, err = client.EmptyCall(ctx, &testpb.Empty{}); err == nil {
+		t.Errorf("rpc EmptyCall() succeeded, want error")
+	}
+	if got, want := conn.GetState(), connectivity.TransientFailure; got != want {
+		t.Errorf("conn.GetState(): got %v, want %v", got, want)
+	}
+
+	backends := startTestServiceBackends(t, 1)
+
+	t.Log("Updating EDS with a new backend endpoint.")
+	endpoints = e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Localities: []e2e.LocalityOptions{{
+			Backends: backendOptions(t, backends),
+			Weight:   1,
+		}},
+	})
+	if err = xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	// A WaitForReady RPC should succeed, and the channel should report READY.
+	if _, err = client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
+		t.Errorf("rpc EmptyCall() failed: %v", err)
+	}
+	if got, want := conn.GetState(), connectivity.Ready; got != want {
+		t.Errorf("conn.GetState(): got %v, want %v", got, want)
 	}
 }