Repository: grpc/grpc-go

Issue:
Linter rule for using context.Background() without a timeout in tests
### Use case(s) - what problem will this feature solve?

We want timeouts in every test, so anytime an operation needs a context (even during test setup), we want to avoid code using `context.Background()` or `context.TODO()`  is needed there should be something like `context.WithTimeout(context.Background(), defaultTestTimeout)`.

### Proposed Solution

Add a linter rule that makes sure whenever we use `context.Background()` in test, it is using `WithTimeout`. Something like

```
git grep 'context.Background()' -- "*_test.go" | not grep -v 'context.WithTimeout(context.Background()'
```

### Alternatives Considered

Do not lint this and continue relying on reviews to catch this.

### Additional Context

Prompted by https://github.com/grpc/grpc-go/pull/7271#discussion_r1626538233

Discussion:
@atollena: There are many cases where we use `context.Background` because we don't use the context for cancellation:

```
+ git grep 'context.Background()' -- '*_test.go'
+ not grep -v 'context.WithTimeout(context.Background()'
+ grep -v 'context.WithTimeout(context.Background()'
benchmark/primitives/context_test.go:	ctx, cancel := context.WithCancel(context.Background())
benchmark/primitives/context_test.go:	ctx, cancel := context.WithCancel(context.Background())
benchmark/primitives/context_test.go:	ctx, cancel := context.WithCancel(context.Background())
benchmark/primitives/context_test.go:	ctx, cancel := context.WithCancel(context.Background())
benchmark/primitives/context_test.go:	pCtx := context.Background()
clientconn_test.go:	ctx, cancel := context.WithCancel(context.Background())
credentials/alts/internal/handshaker/handshaker_test.go:	hs, err := NewClientHandshaker(context.Background(), clientConn, conn, opts)
credentials/alts/internal/handshaker/handshaker_test.go:	hs, err := NewServerHandshaker(context.Background(), clientConn, conn, opts)
credentials/google/google_test.go:				ctx:     context.Background(),
credentials/google/google_test.go:				ctx: icredentials.NewClientHandshakeInfoContext(context.Background(), credentials.ClientHandshakeInfo{
credentials/google/google_test.go:				ctx: icredentials.NewClientHandshakeInfoContext(context.Background(), credentials.ClientHandshakeInfo{
credentials/google/google_test.go:				ctx: icredentials.NewClientHandshakeInfoContext(context.Background(), credentials.ClientHandshakeInfo{
credentials/google/google_test.go:				ctx: icredentials.NewClientHandshakeInfoContext(context.Background(), credentials.ClientHandshakeInfo{
credentials/google/xds_test.go:		return icredentials.NewClientHandshakeInfoContext(context.Background(), credentials.ClientHandshakeInfo{
credentials/google/xds_test.go:		{"not an xDS cluster", context.Background(), false},
gcp/observability/observability_test.go:	if err := Start(context.Background()); err == nil {
gcp/observability/observability_test.go:	if err := Start(context.Background()); err == nil {
gcp/observability/observability_test.go:	if err := Start(context.Background()); err == nil {
gcp/observability/observability_test.go:	if err := Start(context.Background()); err == nil {
gcp/observability/observability_test.go:	if err := Start(context.Background()); err == nil {
gcp/observability/observability_test.go:	if err := Start(context.Background()); err == nil {
internal/binarylog/method_logger_test.go:		ml.Log(context.Background(), tc.config)
internal/transport/handler_server_test.go:		context.Background(), func(s *Stream) { go handleStream(s) },
internal/transport/handler_server_test.go:		context.Background(), func(s *Stream) { go handleStream(s) },
internal/transport/handler_server_test.go:		context.Background(), func(s *Stream) { go runStream(s) },
internal/transport/handler_server_test.go:		context.Background(), func(s *Stream) { go handleStream(st, s) },
internal/transport/handler_server_test.go:		context.Background(), func(s *Stream) { go handleStream(s) },
internal/transport/transport_test.go:			go transport.HandleStreams(context.Background(), h.handleStreamAndNotify)
internal/transport/transport_test.go:			go transport.HandleStreams(context.Background(), func(*Stream) {})
internal/transport/transport_test.go:			go transport.HandleStreams(context.Background(), func(s *Stream) {
internal/transport/transport_test.go:			go transport.HandleStreams(context.Background(), func(s *Stream) {
internal/transport/transport_test.go:			go transport.HandleStreams(context.Background(), func(s *Stream) {
internal/transport/transport_test.go:			go transport.HandleStreams(context.Background(), func(s *Stream) {
internal/transport/transport_test.go:			go transport.HandleStreams(context.Background(), func(s *Stream) {
internal/transport/transport_test.go:			go transport.HandleStreams(context.Background(), func(s *Stream) {
internal/transport/transport_test.go:	connectCtx, cancel := context.WithDeadline(context.Background(), time.Now().Add(2*time.Second))
internal/transport/transport_test.go:	ct, connErr := NewClientTransport(connectCtx, context.Background(), addr, copts, func(GoAwayReason) {})
internal/transport/transport_test.go:	connectCtx, cancel := context.WithDeadline(context.Background(), time.Now().Add(2*time.Second))
internal/transport/transport_test.go:	tr, err := NewClientTransport(connectCtx, context.Background(), resolver.Address{Addr: lis.Addr().String()}, copts, func(GoAwayReason) {})
internal/transport/transport_test.go:	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10))
internal/transport/transport_test.go:	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10))
internal/transport/transport_test.go:	pctx, cancel := context.WithCancel(context.Background())
internal/transport/transport_test.go:		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10))
internal/transport/transport_test.go:	_, err = NewClientTransport(connectCtx, context.Background(), resolver.Address{Addr: lis.Addr().String()}, copts, func(GoAwayReason) {})
internal/transport/transport_test.go:	_, err = NewClientTransport(connectCtx, context.Background(), resolver.Address{Addr: lis.Addr().String()}, copts, func(GoAwayReason) {})
internal/transport/transport_test.go:	connectCtx, cancel := context.WithDeadline(context.Background(), time.Now().Add(2*time.Second))
internal/transport/transport_test.go:	ct, err := NewClientTransport(connectCtx, context.Background(), resolver.Address{Addr: lis.Addr().String()}, copts, func(GoAwayReason) {})
internal/transport/transport_test.go:		ctx:         context.Background(),
internal/transport/transport_test.go:	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(2*time.Second))
internal/transport/transport_test.go:	tr, err := NewClientTransport(ctx, context.Background(), addr, copts, func(GoAwayReason) {})
internal/transport/transport_test.go:	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(2*time.Second))
internal/transport/transport_test.go:	tr, err := NewClientTransport(ctx, context.Background(), addr, copts, func(GoAwayReason) {})
internal/transport/transport_test.go:	ct, err := NewClientTransport(ctx, context.Background(), resolver.Address{Addr: lis.Addr().String()}, ConnectOptions{}, func(GoAwayReason) {})
internal/xds/bootstrap/tlscreds/bundle_ext_test.go:	if _, err = client.EmptyCall(context.Background(), &testpb.Empty{}); err != nil {
internal/xds/bootstrap/tlscreds/bundle_test.go:	_, err = client.EmptyCall(context.Background(), &testpb.Empty{})
internal/xds/rbac/rbac_engine_test.go:					ctx := metadata.NewIncomingContext(context.Background(), data.rpcData.md)
metadata/metadata_test.go:	ctx := NewIncomingContext(context.Background(), md)
metadata/metadata_test.go:	ctx := NewIncomingContext(context.Background(), md)
metadata/metadata_test.go:	ctx := NewIncomingContext(context.Background(), md)
metadata/metadata_test.go:	ctx := NewIncomingContext(context.Background(), md)
peer/peer_test.go:			ctx := NewContext(context.Background(), tc.peer)
picker_wrapper_test.go:			if tr, _, err := bp.pick(context.Background(), true, balancer.PickInfo{}); err != nil || tr != testT {
picker_wrapper_test.go:			if tr, _, err := bp.pick(context.Background(), true, balancer.PickInfo{}); err != nil || tr != testT {
picker_wrapper_test.go:			if tr, _, err := bp.pick(context.Background(), false, balancer.PickInfo{}); err != nil || tr != testT {
picker_wrapper_test.go:			if tr, _, err := bp.pick(context.Background(), true, balancer.PickInfo{}); err != nil || tr != testT {
resolver_balancer_ext_test.go:		ctx, cancel = context.WithCancel(context.Background())
security/advancedtls/advancedtls_test.go:			_, clientAuthInfo, handshakeErr := clientTLS.ClientHandshake(context.Background(),
server_test.go:	ii(context.Background(), nil, nil, handler)
server_test.go:	ctx := NewContextWithServerTransportStream(context.Background(), expectedStream)
server_test.go:				if _, err := s.opts.unaryInt(context.Background(), nil, nil,
test/gracefulstop_test.go:		_, err := clt.UnaryCall(context.Background(), &testpb.SimpleRequest{})
test/gracefulstop_test.go:		_, err := clt.UnaryCall(context.Background(), &testpb.SimpleRequest{})
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_1"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_2"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "notacluster"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_1"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_2"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:notacluster"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_1"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_2"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_3"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:notacluster"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_1"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_2"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:notacluster"),
xds/internal/balancer/clustermanager/clustermanager_test.go:		gotSCSt, err := p2.Pick(balancer.PickInfo{Ctx: SetPickedCluster(context.Background(), "cds:notacluster")})
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_1"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:cluster_2"),
xds/internal/balancer/clustermanager/clustermanager_test.go:				Ctx: SetPickedCluster(context.Background(), "cds:notacluster"),
xds/internal/balancer/clustermanager/clustermanager_test.go:		Ctx: SetPickedCluster(context.Background(), "csp:cluster"),
xds/internal/balancer/ringhash/picker_test.go:				Ctx: SetRequestHash(context.Background(), tt.hash),
xds/internal/balancer/ringhash/picker_test.go:	_, err := p.Pick(balancer.PickInfo{Ctx: SetRequestHash(context.Background(), 5)})
xds/internal/balancer/ringhash/picker_test.go:	pr, err := p.Pick(balancer.PickInfo{Ctx: SetRequestHash(context.Background(), 5)})
xds/internal/balancer/ringhash/picker_test.go:	_, err := p.Pick(balancer.PickInfo{Ctx: SetRequestHash(context.Background(), 5)})
xds/internal/balancer/ringhash/ringhash_test.go:	return SetRequestHash(context.Background(), h)
xds/internal/balancer/ringhash/ringhash_test.go:	p0.Pick(balancer.PickInfo{Ctx: context.Background()})
xds/internal/resolver/serviceconfig_test.go:				Context: metadata.NewOutgoingContext(context.Background(), metadata.Pairs(":path", "/products")),
xds/internal/resolver/serviceconfig_test.go:				Context: metadata.NewOutgoingContext(context.Background(), metadata.Pairs(":path", "abc")),
xds/internal/resolver/serviceconfig_test.go:				Context: metadata.NewOutgoingContext(context.Background(), metadata.Pairs("something-bin", "xyz")),
xds/internal/resolver/serviceconfig_test.go:					metadata.NewOutgoingContext(context.Background(), metadata.Pairs("content-type", "user value")),
xds/internal/resolver/xds_resolver_test.go:					_, err = res.Interceptor.NewStream(context.Background(), iresolver.RPCInfo{}, func() {}, func(ctx context.Context, done func()) (iresolver.ClientStream, error) {
xds/internal/xdsclient/xdsresource/filter_chain_test.go:						errs = append(errs, int.AllowRPC(context.Background()).Error())
xds/internal/xdsclient/xdsresource/matcher_test.go:				Context: metadata.NewOutgoingContext(context.Background(), metadata.Pairs("th", "tv")),
xds/internal/xdsclient/xdsresource/matcher_test.go:				Context: metadata.NewOutgoingContext(context.Background(), metadata.Pairs("th", "tv")),
xds/internal/xdsclient/xdsresource/matcher_test.go:				Context: metadata.NewOutgoingContext(context.Background(), metadata.Pairs("th", "tv")),
xds/internal/xdsclient/xdsresource/matcher_test.go:				Context: metadata.NewOutgoingContext(context.Background(), metadata.Pairs("th", "tv")),
xds/internal/xdsclient/xdsresource/matcher_test.go:				Context: grpcutil.WithExtraMetadata(context.Background(), metadata.Pairs(
xds/internal/xdsclient/xdsresource/matcher_test.go:					metadata.NewOutgoingContext(context.Background(), metadata.Pairs("t-bin", "123")), metadata.Pairs(
```@easwars: Thanks for filing this issue. That's an embarrassing number of usages of the background context.
@hasson82: @atollena I am willing to work on it if it is still vacant.

I will add here a question I have, to avoid ping pong later. 
What do we think about the benchmark tests for contexts. 

```
benchmark/primitives/context_test.go:	ctx, cancel := context.WithCancel(context.Background())
benchmark/primitives/context_test.go:	ctx, cancel := context.WithCancel(context.Background())
benchmark/primitives/context_test.go:	ctx, cancel := context.WithCancel(context.Background())
benchmark/primitives/context_test.go:	ctx, cancel := context.WithCancel(context.Background())
benchmark/primitives/context_test.go:	pCtx := context.Background()
```
Do we want to add timeouts there? Can it affect the test?@atollena: No-one is working on this at the moment.

I think you should exclude benchmark/primitives/context_test from the linter rule.

Solution patch:
diff --git a/doc.go b/doc.go
index 0022859a..e7b532b6 100644
--- a/doc.go
+++ b/doc.go
@@ -16,7 +16,7 @@
  *
  */
 
-//go:generate ./regenerate.sh
+//go:generate ./scripts/regenerate.sh
 
 /*
 Package grpc implements an RPC system called gRPC.
diff --git a/examples/features/csm_observability/README.md b/examples/features/csm_observability/README.md
new file mode 100644
index 00000000..c7b8c5ad
--- /dev/null
+++ b/examples/features/csm_observability/README.md
@@ -0,0 +1,38 @@
+# CSM Observability
+
+This examples shows how to configure CSM Observability for gRPC client and
+server applications (configured once per binary), and shows what type of
+telemetry data it can produce for certain RPCs with additional CSM Labels. The
+gRPC Client accepts configuration from an xDS Control plane as the default
+address that it connects to is "xds:///helloworld:50051", but this can be
+overridden with the command line flag --server_addr. This can be plugged into
+the steps outlined in the CSM Observability User Guide.
+
+## Try it (locally if overwritten xDS Address)
+
+```
+go run server/main.go
+```
+
+```
+go run client/main.go
+```
+
+Curl to the port where Prometheus exporter is outputting metrics data:
+```
+curl localhost:9464/metrics
+```
+
+# Building
+From the grpc-go directory:
+
+Client:
+docker build -t <TAG> -f examples/features/csm_observability/client/Dockerfile .
+
+Server:
+docker build -t <TAG> -f examples/features/csm_observability/server/Dockerfile .
+
+Note that this example will not work by default, as the client uses an xDS
+Scheme and thus needs xDS Resources to connect to the server. Deploy the built
+client and server containers within Cloud Service Mesh in order for this example
+to work, or overwrite target to point to :<server serving port>.
diff --git a/examples/features/csm_observability/client/Dockerfile b/examples/features/csm_observability/client/Dockerfile
new file mode 100644
index 00000000..ad6a37b7
--- /dev/null
+++ b/examples/features/csm_observability/client/Dockerfile
@@ -0,0 +1,35 @@
+# 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.
+
+# Dockerfile for building the example client. To build the image, run the
+# following command from grpc-go directory:
+# docker build -t <TAG> -f examples/features/csm_observability/client/Dockerfile .
+FROM golang:1.21-alpine as build
+
+RUN apk --no-cache add curl
+
+# Make a grpc-go directory and copy the repo into it.
+WORKDIR /go/src/grpc-go
+COPY . .
+
+# Build a static binary without cgo so that we can copy just the binary in the
+# final image, and can get rid of the Go compiler and gRPC-Go dependencies.
+RUN cd examples/features/csm_observability/client && go build -tags osusergo,netgo .
+
+FROM alpine
+RUN apk --no-cache add curl
+COPY --from=build /go/src/grpc-go/examples/features/csm_observability/client/client .
+ENV GRPC_GO_LOG_VERBOSITY_LEVEL=99
+ENV GRPC_GO_LOG_SEVERITY_LEVEL="info"
+ENTRYPOINT ["./client"]
diff --git a/examples/features/csm_observability/client/main.go b/examples/features/csm_observability/client/main.go
new file mode 100644
index 00000000..c89676c1
--- /dev/null
+++ b/examples/features/csm_observability/client/main.go
@@ -0,0 +1,77 @@
+/*
+ *
+ * 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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/examples/features/proto/echo"
+	"google.golang.org/grpc/stats/opentelemetry"
+	"google.golang.org/grpc/stats/opentelemetry/csm"
+	_ "google.golang.org/grpc/xds" // To install the xds resolvers and balancers.
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"go.opentelemetry.io/otel/exporters/prometheus"
+	"go.opentelemetry.io/otel/sdk/metric"
+)
+
+var (
+	target             = flag.String("target", "xds:///helloworld:50051", "the server address to connect to")
+	prometheusEndpoint = flag.String("prometheus_endpoint", ":9464", "the Prometheus exporter endpoint")
+)
+
+func main() {
+	flag.Parse()
+	exporter, err := prometheus.New()
+	if err != nil {
+		log.Fatalf("Failed to start prometheus exporter: %v", err)
+	}
+	provider := metric.NewMeterProvider(metric.WithReader(exporter))
+	go http.ListenAndServe(*prometheusEndpoint, promhttp.Handler())
+
+	cleanup := csm.EnableObservability(context.Background(), opentelemetry.Options{MetricsOptions: opentelemetry.MetricsOptions{MeterProvider: provider}})
+	defer cleanup()
+
+	cc, err := grpc.NewClient(*target, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		log.Fatalf("Failed to start NewClient: %v", err)
+	}
+	defer cc.Close()
+	c := echo.NewEchoClient(cc)
+
+	// Make a RPC every second. This should trigger telemetry to be emitted from
+	// the client and the server.
+	for {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+		r, err := c.UnaryEcho(ctx, &echo.EchoRequest{Message: "this is examples/opentelemetry"})
+		if err != nil {
+			log.Printf("UnaryEcho failed: %v", err)
+		}
+		fmt.Println(r)
+		time.Sleep(time.Second)
+		cancel()
+	}
+}
diff --git a/examples/features/csm_observability/server/Dockerfile b/examples/features/csm_observability/server/Dockerfile
new file mode 100644
index 00000000..de6836a3
--- /dev/null
+++ b/examples/features/csm_observability/server/Dockerfile
@@ -0,0 +1,34 @@
+# 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.
+
+# Dockerfile for building the example server. To build the image, run the
+# following command from grpc-go directory:
+# docker build -t <TAG> -f examples/features/csm_observability/server/Dockerfile .
+
+FROM golang:1.21-alpine as build
+RUN apk --no-cache add curl
+# Make a grpc-go directory and copy the repo into it.
+WORKDIR /go/src/grpc-go
+COPY . .
+
+# Build a static binary without cgo so that we can copy just the binary in the
+# final image, and can get rid of the Go compiler and gRPC-Go dependencies.
+RUN cd examples/features/csm_observability/server && go build -tags osusergo,netgo .
+
+FROM alpine
+RUN apk --no-cache add curl
+COPY --from=build /go/src/grpc-go/examples/features/csm_observability/server/server .
+ENV GRPC_GO_LOG_VERBOSITY_LEVEL=99
+ENV GRPC_GO_LOG_SEVERITY_LEVEL="info"
+ENTRYPOINT ["./server"]
diff --git a/examples/features/csm_observability/server/main.go b/examples/features/csm_observability/server/main.go
new file mode 100644
index 00000000..151d3dba
--- /dev/null
+++ b/examples/features/csm_observability/server/main.go
@@ -0,0 +1,77 @@
+/*
+ *
+ * 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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+
+	"google.golang.org/grpc"
+	pb "google.golang.org/grpc/examples/features/proto/echo"
+	"google.golang.org/grpc/stats/opentelemetry"
+	"google.golang.org/grpc/stats/opentelemetry/csm"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"go.opentelemetry.io/otel/exporters/prometheus"
+	"go.opentelemetry.io/otel/sdk/metric"
+)
+
+var (
+	port               = flag.String("port", "50051", "the server address to connect to")
+	prometheusEndpoint = flag.String("prometheus_endpoint", ":9464", "the Prometheus exporter endpoint")
+)
+
+type echoServer struct {
+	pb.UnimplementedEchoServer
+	addr string
+}
+
+func (s *echoServer) UnaryEcho(ctx context.Context, req *pb.EchoRequest) (*pb.EchoResponse, error) {
+	return &pb.EchoResponse{Message: fmt.Sprintf("%s (from %s)", req.Message, s.addr)}, nil
+}
+
+func main() {
+	flag.Parse()
+	exporter, err := prometheus.New()
+	if err != nil {
+		log.Fatalf("Failed to start prometheus exporter: %v", err)
+	}
+	provider := metric.NewMeterProvider(metric.WithReader(exporter))
+	go http.ListenAndServe(*prometheusEndpoint, promhttp.Handler())
+
+	cleanup := csm.EnableObservability(context.Background(), opentelemetry.Options{MetricsOptions: opentelemetry.MetricsOptions{MeterProvider: provider}})
+	defer cleanup()
+
+	lis, err := net.Listen("tcp", ":"+*port)
+	if err != nil {
+		log.Fatalf("Failed to listen: %v", err)
+	}
+	s := grpc.NewServer()
+	pb.RegisterEchoServer(s, &echoServer{addr: ":" + *port})
+
+	log.Printf("Serving on %s\n", *port)
+
+	if err := s.Serve(lis); err != nil {
+		log.Fatalf("Failed to serve: %v", err)
+	}
+}
diff --git a/examples/go.mod b/examples/go.mod
index 5f1f474d..6399807d 100644
--- a/examples/go.mod
+++ b/examples/go.mod
@@ -4,10 +4,14 @@ go 1.21
 
 require (
 	github.com/cncf/xds/go v0.0.0-20240423153145-555b57ec207b
+	github.com/prometheus/client_golang v1.19.1
+	go.opentelemetry.io/otel/exporters/prometheus v0.49.0
+	go.opentelemetry.io/otel/sdk/metric v1.27.0
 	golang.org/x/oauth2 v0.21.0
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240604185151-ef581f913117
 	google.golang.org/grpc v1.64.0
 	google.golang.org/grpc/gcp/observability v1.0.1
+	google.golang.org/grpc/stats/opentelemetry v0.0.0-20240604165302-6d236200ea68
 	google.golang.org/protobuf v1.34.1
 )
 
@@ -22,10 +26,11 @@ require (
 	cloud.google.com/go/monitoring v1.19.0 // indirect
 	cloud.google.com/go/trace v1.10.7 // indirect
 	contrib.go.opencensus.io/exporter/stackdriver v0.13.15-0.20230702191903-2de6d2748484 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.23.0 // indirect
 	github.com/aws/aws-sdk-go-v2 v1.27.1 // indirect
-	github.com/aws/aws-sdk-go-v2/config v1.27.17 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.17 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.4 // indirect
+	github.com/aws/aws-sdk-go-v2/config v1.27.4 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.4 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.15.2 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.8 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.8 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 // indirect
@@ -35,6 +40,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.24.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.28.11 // indirect
 	github.com/aws/smithy-go v1.20.2 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/envoyproxy/go-control-plane v0.12.0 // indirect
@@ -48,11 +54,16 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
 	github.com/googleapis/gax-go/v2 v2.12.4 // indirect
+	github.com/prometheus/client_model v0.6.1 // indirect
+	github.com/prometheus/common v0.53.0 // indirect
+	github.com/prometheus/procfs v0.15.0 // indirect
 	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/contrib/detectors/gcp v1.27.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.52.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.52.0 // indirect
 	go.opentelemetry.io/otel v1.27.0 // indirect
 	go.opentelemetry.io/otel/metric v1.27.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.27.0 // indirect
 	go.opentelemetry.io/otel/trace v1.27.0 // indirect
 	golang.org/x/crypto v0.24.0 // indirect
 	golang.org/x/net v0.26.0 // indirect
@@ -67,3 +78,5 @@ require (
 )
 
 replace google.golang.org/grpc => ../
+
+replace google.golang.org/grpc/stats/opentelemetry => ../stats/opentelemetry
diff --git a/examples/go.sum b/examples/go.sum
index e6129207..9a7ae4f7 100644
--- a/examples/go.sum
+++ b/examples/go.sum
@@ -774,6 +774,8 @@ gioui.org v0.0.0-20210308172011-57750fc8a0a6/go.mod h1:RSH6KIUZ0p2xy5zHDxgAM4zum
 git.sr.ht/~sbinet/gg v0.3.1/go.mod h1:KGYtlADtqsqANL9ueOFkWymvzUvLMQllU5Ixo+8v3pc=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.23.0 h1:yRhWveg9NbJcJYoJL4FoSauT2dxnt4N9MIAJ7tvU/mQ=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.23.0/go.mod h1:p2puVVSKjQ84Qb1gzw2XHLs34WQyHTYFZLaVxypAFYs=
 github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c/go.mod h1:X0CRv0ky0k6m906ixxpzmDRLvX58TFUKS2eePweuyxk=
 github.com/ajstarks/deck v0.0.0-20200831202436-30c9fc6549a9/go.mod h1:JynElWSGnm/4RlzPXRlREEwqTHAN3T56Bv2ITsFT3gY=
 github.com/ajstarks/deck/generate v0.0.0-20210309230005-c3f852c02e19/go.mod h1:T13YZdzov6OU0A1+RfKZiZN9ca6VeKdBdyDV+BY97Tk=
@@ -787,12 +789,12 @@ github.com/apache/arrow/go/v12 v12.0.0/go.mod h1:d+tV/eHZZ7Dz7RPrFKtPK02tpr+c9/P
 github.com/apache/thrift v0.16.0/go.mod h1:PHK3hniurgQaNMZYaCLEqXKsYK8upmhPbmdP2FXSqgU=
 github.com/aws/aws-sdk-go-v2 v1.27.1 h1:xypCL2owhog46iFxBKKpBcw+bPTX/RJzwNj8uSilENw=
 github.com/aws/aws-sdk-go-v2 v1.27.1/go.mod h1:ffIFB97e2yNsv4aTSGkqtHnppsIJzw7G7BReUZ3jCXM=
-github.com/aws/aws-sdk-go-v2/config v1.27.17 h1:L0JZN7Gh7pT6u5CJReKsLhGKparqNKui+mcpxMXjDZc=
-github.com/aws/aws-sdk-go-v2/config v1.27.17/go.mod h1:MzM3balLZeaafYcPz8IihAmam/aCz6niPQI0FdprxW0=
-github.com/aws/aws-sdk-go-v2/credentials v1.17.17 h1:b3Dk9uxQByS9sc6r0sc2jmxsJKO75eOcb9nNEiaUBLM=
-github.com/aws/aws-sdk-go-v2/credentials v1.17.17/go.mod h1:e4khg9iY08LnFK/HXQDWMf9GDaiMari7jWPnXvKAuBU=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.4 h1:0cSfTYYL9qiRcdi4Dvz+8s3JUgNR2qvbgZkXcwPEEEk=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.4/go.mod h1:Wjn5O9eS7uSi7vlPKt/v0MLTncANn9EMmoDvnzJli6o=
+github.com/aws/aws-sdk-go-v2/config v1.27.4 h1:AhfWb5ZwimdsYTgP7Od8E9L1u4sKmDW2ZVeLcf2O42M=
+github.com/aws/aws-sdk-go-v2/config v1.27.4/go.mod h1:zq2FFXK3A416kiukwpsd+rD4ny6JC7QSkp4QdN1Mp2g=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.4 h1:h5Vztbd8qLppiPwX+y0Q6WiwMZgpd9keKe2EAENgAuI=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.4/go.mod h1:+30tpwrkOgvkJL1rUZuRLoxcJwtI/OkeBLYnHxJtVe0=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.15.2 h1:AK0J8iYBFeUk2Ax7O8YpLtFsfhdOByh2QIkHmigpRYk=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.15.2/go.mod h1:iRlGzMix0SExQEviAyptRWRGdYNo3+ufW/lCzvKVTUc=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.8 h1:RnLB7p6aaFMRfyQkD6ckxR7myCC9SABIqSz4czYUUbU=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.8/go.mod h1:XH7dQJd+56wEbP1I4e4Duo+QhSMxNArE8VP7NuUOTeM=
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.8 h1:jzApk2f58L9yW9q1GEab3BMMFWUkkiZhyrRUtbwUbKU=
@@ -811,6 +813,8 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.28.11 h1:HYS0csS7UJxdYRoG+bGgUYrSwVn
 github.com/aws/aws-sdk-go-v2/service/sts v1.28.11/go.mod h1:QXnthRM35zI92048MMwfFChjFmoufTdhtHmouwNfhhU=
 github.com/aws/smithy-go v1.20.2 h1:tbp628ireGtzcHDDmLT/6ADHidqnwgF57XOXZe6tp4Q=
 github.com/aws/smithy-go v1.20.2/go.mod h1:krry+ya/rV9RDcV/Q16kpu6ypI4K2czasz0NC3qS14E=
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/boombuler/barcode v1.0.0/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/boombuler/barcode v1.0.1/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -1041,10 +1045,18 @@ github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZ
 github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v1.19.1 h1:wZWJDwK+NameRJuPGDhlnFgx8e8HN3XHQeLaYJFJBOE=
+github.com/prometheus/client_golang v1.19.1/go.mod h1:mP78NwGzrVks5S2H6ab8+ZZGJLZUq1hoULYBAYBw1Ho=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.3.0/go.mod h1:LDGWKZIo7rky3hgvBe+caln+Dr3dPggB5dvjtD7w9+w=
 github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI=
+github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
+github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
+github.com/prometheus/common v0.53.0 h1:U2pL9w9nmJwJDa4qqLQ3ZaePJ6ZTwt7cMD3AG3+aLCE=
+github.com/prometheus/common v0.53.0/go.mod h1:BrxBKv3FWBIGXw89Mg1AeBq7FSyRzXWI3l3e7W3RN5U=
+github.com/prometheus/procfs v0.15.0 h1:A82kmvXJq2jTu5YUhSGNlYoxh85zLnKgPz4bMZgI5Ek=
+github.com/prometheus/procfs v0.15.0/go.mod h1:Y0RJ/Y5g5wJpkTisOtqwDSo4HwhGmLB4VQSw2sQJLHk=
 github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
@@ -1089,16 +1101,22 @@ go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
 go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
 go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
+go.opentelemetry.io/contrib/detectors/gcp v1.27.0 h1:eVfDeFAPnMFZUhNdDZ/BbpEmC7/xxDKTSba5NhJH88s=
+go.opentelemetry.io/contrib/detectors/gcp v1.27.0/go.mod h1:amd+4uZxqJAUx7zI1JvygUtAc2EVWtQeyz8D+3161SQ=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.52.0 h1:vS1Ao/R55RNV4O7TA2Qopok8yN+X0LIP6RVWLFkprck=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.52.0/go.mod h1:BMsdeOxN04K0L5FNUBfjFdvwWGNe/rkmSwH4Aelu/X0=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.52.0 h1:9l89oX4ba9kHbBol3Xin3leYJ+252h0zszDtBwyKe2A=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.52.0/go.mod h1:XLZfZboOJWHNKUv7eH0inh0E9VV6eWDFB/9yJyTLPp0=
 go.opentelemetry.io/otel v1.27.0 h1:9BZoF3yMK/O1AafMiQTVu0YDj5Ea4hPhxCs7sGva+cg=
 go.opentelemetry.io/otel v1.27.0/go.mod h1:DMpAK8fzYRzs+bi3rS5REupisuqTheUlSZJ1WnZaPAQ=
+go.opentelemetry.io/otel/exporters/prometheus v0.49.0 h1:Er5I1g/YhfYv9Affk9nJLfH/+qCCVVg1f2R9AbJfqDQ=
+go.opentelemetry.io/otel/exporters/prometheus v0.49.0/go.mod h1:KfQ1wpjf3zsHjzP149P4LyAwWRupc6c7t1ZJ9eXpKQM=
 go.opentelemetry.io/otel/metric v1.27.0 h1:hvj3vdEKyeCi4YaYfNjv2NUje8FqKqUY8IlF0FxV/ik=
 go.opentelemetry.io/otel/metric v1.27.0/go.mod h1:mVFgmRlhljgBiuk/MP/oKylr4hs85GZAylncepAX/ak=
-go.opentelemetry.io/otel/sdk v1.24.0 h1:YMPPDNymmQN3ZgczicBY3B6sf9n62Dlj9pWD3ucgoDw=
-go.opentelemetry.io/otel/sdk v1.24.0/go.mod h1:KVrIYw6tEubO9E96HQpcmpTKDVn9gdv35HoYiQWGDFg=
+go.opentelemetry.io/otel/sdk v1.27.0 h1:mlk+/Y1gLPLn84U4tI8d3GNJmGT/eXe3ZuOXN9kTWmI=
+go.opentelemetry.io/otel/sdk v1.27.0/go.mod h1:Ha9vbLwJE6W86YstIywK2xFfPjbWlCuwPtMkKdz/Y4A=
+go.opentelemetry.io/otel/sdk/metric v1.27.0 h1:5uGNOlpXi+Hbo/DRoI31BSb1v+OGcpv2NemcCrOL8gI=
+go.opentelemetry.io/otel/sdk/metric v1.27.0/go.mod h1:we7jJVrYN2kh3mVBlswtPU22K0SA+769l93J6bsyvqw=
 go.opentelemetry.io/otel/trace v1.27.0 h1:IqYb813p7cmbHk0a5y6pD5JPakbVfftRXABGt5/Rscw=
 go.opentelemetry.io/otel/trace v1.27.0/go.mod h1:6RiD1hkAprV4/q+yd2ln1HG9GoPx39SuvvstaLBl+l4=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
diff --git a/scripts/install_protoc.sh b/scripts/install_protoc.sh
new file mode 100755
index 00000000..f0b8aef5
--- /dev/null
+++ b/scripts/install_protoc.sh
@@ -0,0 +1,69 @@
+#!/bin/bash
+# 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.
+#
+# This script ensures the installation of protobuf on client machine.
+# In case of manual run of this script, make sure you pass the args
+# expected at
+# https://github.com/grpc/grpc-go/blob/master/scripts/install_protoc.sh#L60
+
+set -eu -o pipefail
+
+source "$(dirname $0)/vet-common.sh"
+
+# The version of protoc that will be installed.
+PROTOC_VERSION="25.2"
+
+# Function to download pre-built binaries for Linux with
+# ARCH as $1, OS as $2, and INSTALL_PATH as $3 arguments.
+download_binary() {
+  # 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
+      else
+        echo "Unable to determine installed protoc version. Starting the installation."
+      fi
+  fi
+  DOWNLOAD_URL="https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOC_VERSION}/protoc-${PROTOC_VERSION}-$2-$1.zip"
+  # Download and unzip
+  curl -LO "$DOWNLOAD_URL"
+  INSTALL_DIR="${3:-${GOBIN:-${GOPATH:-$HOME/go}}}"
+  unzip "protoc-${PROTOC_VERSION}-$2-$1.zip" -d $INSTALL_DIR
+  rm "protoc-${PROTOC_VERSION}-$2-$1.zip"
+  rm "${INSTALL_DIR}/readme.txt"
+}
+
+# Detect the architecture
+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."
+esac
+
+# Detect the Operating System
+INSTALL_PATH=${1:+"$1"}
+case "$(uname -s)" in
+  "Darwin") download_binary $ARCH "osx" "$INSTALL_PATH";;
+  "Linux") download_binary $ARCH "linux" "$INSTALL_PATH";;
+*) die "Unsupported OS. Please consider manual installation from \
+   https://github.com/protocolbuffers/protobuf/releases/ and add to PATH" ;;
+esac
diff --git a/regenerate.sh b/scripts/regenerate.sh
similarity index 63%
rename from regenerate.sh
rename to scripts/regenerate.sh
index 3edca296..c5d7b1a6 100755
--- a/regenerate.sh
+++ b/scripts/regenerate.sh
@@ -19,12 +19,15 @@ WORKDIR=$(mktemp -d)
 
 function finish {
   rm -rf "$WORKDIR"
+  # Revert back the PATH to client's original value
+  export PATH=$ORIGINAL_PATH
 }
 trap finish EXIT
 
-export GOBIN=${WORKDIR}/bin
-export PATH=${GOBIN}:${PATH}
-mkdir -p ${GOBIN}
+GOBIN="${WORKDIR}"/bin
+ORIGINAL_PATH=$PATH
+export PATH="${GOBIN}:${PATH}"
+mkdir -p "${GOBIN}"
 
 echo "remove existing generated files"
 # grpc_testing_not_regenerate/*.pb.go is not re-generated,
@@ -38,39 +41,41 @@ echo "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
+git clone --quiet https://github.com/grpc/grpc-proto "${WORKDIR}/grpc-proto"
 
 echo "git clone https://github.com/protocolbuffers/protobuf"
-git clone --quiet https://github.com/protocolbuffers/protobuf ${WORKDIR}/protobuf
+git clone --quiet https://github.com/protocolbuffers/protobuf "${WORKDIR}/protobuf"
 
 # Pull in code.proto as a proto dependency
-mkdir -p ${WORKDIR}/googleapis/google/rpc
+mkdir -p "${WORKDIR}/googleapis/google/rpc"
 echo "curl 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
+curl --silent https://raw.githubusercontent.com/googleapis/googleapis/master/google/rpc/code.proto > "${WORKDIR}/googleapis/google/rpc/code.proto"
 
-mkdir -p ${WORKDIR}/out
+source ./scripts/install_protoc.sh $WORKDIR
+
+mkdir -p "${WORKDIR}/out"
 
 # Generates sources without the embed requirement
 LEGACY_SOURCES=(
-  ${WORKDIR}/grpc-proto/grpc/binlog/v1/binarylog.proto
-  ${WORKDIR}/grpc-proto/grpc/channelz/v1/channelz.proto
-  ${WORKDIR}/grpc-proto/grpc/health/v1/health.proto
-  ${WORKDIR}/grpc-proto/grpc/lb/v1/load_balancer.proto
+  "${WORKDIR}/grpc-proto/grpc/binlog/v1/binarylog.proto"
+  "${WORKDIR}/grpc-proto/grpc/channelz/v1/channelz.proto"
+  "${WORKDIR}/grpc-proto/grpc/health/v1/health.proto"
+  "${WORKDIR}/grpc-proto/grpc/lb/v1/load_balancer.proto"
   profiling/proto/service.proto
-  ${WORKDIR}/grpc-proto/grpc/reflection/v1alpha/reflection.proto
-  ${WORKDIR}/grpc-proto/grpc/reflection/v1/reflection.proto
+  "${WORKDIR}/grpc-proto/grpc/reflection/v1alpha/reflection.proto"
+  "${WORKDIR}/grpc-proto/grpc/reflection/v1/reflection.proto"
 )
 
 # Generates only the new gRPC Service symbols
 SOURCES=(
   $(git ls-files --exclude-standard --cached --others "*.proto" | grep -v '^profiling/proto/service.proto$')
-  ${WORKDIR}/grpc-proto/grpc/gcp/altscontext.proto
-  ${WORKDIR}/grpc-proto/grpc/gcp/handshaker.proto
-  ${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/gcp/altscontext.proto"
+  "${WORKDIR}/grpc-proto/grpc/gcp/handshaker.proto"
+  "${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"
 )
 
 # These options of the form 'Mfoo.proto=bar' instruct the codegen to use an
@@ -95,29 +100,29 @@ for src in ${SOURCES[@]}; do
   echo "protoc ${src}"
   protoc --go_out=${OPTS}:${WORKDIR}/out --go-grpc_out=${OPTS},use_generic_streams_experimental=true:${WORKDIR}/out \
     -I"." \
-    -I${WORKDIR}/grpc-proto \
-    -I${WORKDIR}/googleapis \
-    -I${WORKDIR}/protobuf/src \
-    ${src}
+    -I"${WORKDIR}/grpc-proto" \
+    -I"${WORKDIR}/googleapis" \
+    -I"${WORKDIR}/protobuf/src" \
+    "${src}"
 done
 
 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 \
-    -I${WORKDIR}/protobuf/src \
-    ${src}
+    -I"${WORKDIR}/grpc-proto" \
+    -I"${WORKDIR}/googleapis" \
+    -I"${WORKDIR}/protobuf/src" \
+    "${src}"
 done
 
 # The go_package option in grpc/lookup/v1/rls.proto doesn't match the
 # current location. Move it into the right place.
-mkdir -p ${WORKDIR}/out/google.golang.org/grpc/internal/proto/grpc_lookup_v1
-mv ${WORKDIR}/out/google.golang.org/grpc/lookup/grpc_lookup_v1/* ${WORKDIR}/out/google.golang.org/grpc/internal/proto/grpc_lookup_v1
+mkdir -p "${WORKDIR}/out/google.golang.org/grpc/internal/proto/grpc_lookup_v1"
+mv "${WORKDIR}"/out/google.golang.org/grpc/lookup/grpc_lookup_v1/* "${WORKDIR}/out/google.golang.org/grpc/internal/proto/grpc_lookup_v1"
 
 # grpc_testing_not_regenerate/*.pb.go are not re-generated,
 # see grpc_testing_not_regenerate/README.md for details.
-rm ${WORKDIR}/out/google.golang.org/grpc/reflection/test/grpc_testing_not_regenerate/*.pb.go
+rm "${WORKDIR}"/out/google.golang.org/grpc/reflection/test/grpc_testing_not_regenerate/*.pb.go
 
-cp -R ${WORKDIR}/out/google.golang.org/grpc/* .
+cp -R "${WORKDIR}"/out/google.golang.org/grpc/* .
diff --git a/scripts/vet-proto.sh b/scripts/vet-proto.sh
index 26c1ac65..56b42359 100755
--- a/scripts/vet-proto.sh
+++ b/scripts/vet-proto.sh
@@ -20,18 +20,11 @@ trap cleanup EXIT
 # consistent with the place where all binaries installed by scripts in this repo
 # go.)
 if [[ "$1" = "-install" ]]; then
-  if [[ "${GITHUB_ACTIONS}" = "true" ]]; then
-    PROTOBUF_VERSION=25.2 # Shows up in pb.go files as v4.22.0
-    PROTOC_FILENAME=protoc-${PROTOBUF_VERSION}-linux-x86_64.zip
-    pushd /home/runner/go
-      wget https://github.com/google/protobuf/releases/download/v${PROTOBUF_VERSION}/${PROTOC_FILENAME}
-      unzip ${PROTOC_FILENAME}
-      protoc --version # Check that the binary works.
-    popd
-  else 
-    # TODO: replace with install protoc when https://github.com/grpc/grpc-go/pull/7064 is merged.
-    die "-install currently intended for use in CI only."
-  fi
+    if [[ "${GITHUB_ACTIONS}" = "true" ]]; then
+      source ./scripts/install_protoc.sh "/home/runner/go"
+    else
+      die "run protoc installer https://github.com/grpc/grpc-go/blob/master/scripts/install_protoc.sh"
+    fi
   echo SUCCESS
   exit 0
 elif [[ "$#" -ne 0 ]]; then
diff --git a/stats/opentelemetry/client_metrics.go b/stats/opentelemetry/client_metrics.go
index cbc4e36f..86541321 100644
--- a/stats/opentelemetry/client_metrics.go
+++ b/stats/opentelemetry/client_metrics.go
@@ -63,7 +63,7 @@ func (h *clientStatsHandler) initializeMetrics() {
 
 func (h *clientStatsHandler) unaryInterceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
 	ci := &callInfo{
-		target: h.determineTarget(cc),
+		target: cc.CanonicalTarget(),
 		method: h.determineMethod(method, opts...),
 	}
 	ctx = setCallInfo(ctx, ci)
@@ -83,17 +83,6 @@ func (h *clientStatsHandler) unaryInterceptor(ctx context.Context, method string
 	return err
 }
 
-// determineTarget determines the target to record attributes with. This will be
-// "other" if target filter is set and specifies, the target name as is
-// otherwise.
-func (h *clientStatsHandler) determineTarget(cc *grpc.ClientConn) string {
-	target := cc.CanonicalTarget()
-	if f := h.options.MetricsOptions.TargetAttributeFilter; f != nil && !f(target) {
-		target = "other"
-	}
-	return target
-}
-
 // determineMethod determines the method to record attributes with. This will be
 // "other" if StaticMethod isn't specified or if method filter is set and
 // specifies, the method name as is otherwise.
@@ -108,7 +97,7 @@ func (h *clientStatsHandler) determineMethod(method string, opts ...grpc.CallOpt
 
 func (h *clientStatsHandler) streamInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
 	ci := &callInfo{
-		target: h.determineTarget(cc),
+		target: cc.CanonicalTarget(),
 		method: h.determineMethod(method, opts...),
 	}
 	ctx = setCallInfo(ctx, ci)
diff --git a/stats/opentelemetry/opentelemetry.go b/stats/opentelemetry/opentelemetry.go
index 6a6bcd16..4bc195b4 100644
--- a/stats/opentelemetry/opentelemetry.go
+++ b/stats/opentelemetry/opentelemetry.go
@@ -120,12 +120,6 @@ type MetricsOptions struct {
 	// will be recorded.
 	Metrics *Metrics
 
-	// TargetAttributeFilter is a callback that takes the target string of the
-	// channel and returns a bool representing whether to use target as a label
-	// value or use the string "other". If unset, will use the target string as
-	// is. This only applies for client side metrics.
-	TargetAttributeFilter func(string) bool
-
 	// MethodAttributeFilter is to record the method name of RPCs handled by
 	// grpc.UnknownServiceHandler, but take care to limit the values allowed, as
 	// allowing too many will increase cardinality and could cause severe memory
diff --git a/xds/internal/internal.go b/xds/internal/internal.go
index fda4c7f5..70919905 100644
--- a/xds/internal/internal.go
+++ b/xds/internal/internal.go
@@ -83,3 +83,10 @@ func SetLocalityID(addr resolver.Address, l LocalityID) resolver.Address {
 
 // ResourceTypeMapForTesting maps TypeUrl to corresponding ResourceType.
 var ResourceTypeMapForTesting map[string]any
+
+// UnknownCSMLabels are TelemetryLabels emitted from CDS if CSM Telemetry Label
+// data is not present in the CDS Resource.
+var UnknownCSMLabels = map[string]string{
+	"csm.service_name":           "unknown",
+	"csm.service_namespace_name": "unknown",
+}
diff --git a/xds/internal/xdsclient/xdsresource/unmarshal_cds.go b/xds/internal/xdsclient/xdsresource/unmarshal_cds.go
index 9eaef4d9..8ede639a 100644
--- a/xds/internal/xdsclient/xdsresource/unmarshal_cds.go
+++ b/xds/internal/xdsclient/xdsresource/unmarshal_cds.go
@@ -100,6 +100,15 @@ func validateClusterAndConstructClusterUpdate(cluster *v3clusterpb.Cluster, serv
 			}
 		}
 	}
+	// "The values for the service labels csm.service_name and
+	// csm.service_namespace_name come from xDS, “unknown” if not present." -
+	// CSM Design.
+	if _, ok := telemetryLabels["csm.service_name"]; !ok {
+		telemetryLabels["csm.service_name"] = "unknown"
+	}
+	if _, ok := telemetryLabels["csm.service_namespace_name"]; !ok {
+		telemetryLabels["csm.service_namespace_name"] = "unknown"
+	}
 
 	var lbPolicy json.RawMessage
 	var err error



Test patch:
diff --git a/internal/testutils/blocking_context_dialer.go b/internal/testutils/blocking_context_dialer.go
new file mode 100644
index 00000000..ea7a8519
--- /dev/null
+++ b/internal/testutils/blocking_context_dialer.go
@@ -0,0 +1,56 @@
+/*
+ *
+ * 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"
+	"net"
+)
+
+// BlockingDialer is a dialer that waits for Resume() to be called before
+// dialing.
+type BlockingDialer struct {
+	dialer  *net.Dialer
+	blockCh chan struct{}
+}
+
+// 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{}),
+	}
+}
+
+// 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) {
+	select {
+	case <-d.blockCh:
+	case <-ctx.Done():
+		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)
+}
diff --git a/internal/testutils/xds/e2e/clientresources.go b/internal/testutils/xds/e2e/clientresources.go
index 175490e9..d511b473 100644
--- a/internal/testutils/xds/e2e/clientresources.go
+++ b/internal/testutils/xds/e2e/clientresources.go
@@ -675,6 +675,8 @@ type LocalityOptions struct {
 	Weight uint32
 	// Backends is a set of backends belonging to this locality.
 	Backends []BackendOptions
+	// Priority is the priority of the locality. Defaults to 0.
+	Priority uint32
 }
 
 // BackendOptions contains options to configure individual backends in a
@@ -686,6 +688,8 @@ type BackendOptions struct {
 	// Health status of the backend. Default is UNKNOWN which is treated the
 	// same as HEALTHY.
 	HealthStatus v3corepb.HealthStatus
+	// Weight sets the backend weight. Defaults to 1.
+	Weight uint32
 }
 
 // EndpointOptions contains options to configure an Endpoint (or
@@ -708,7 +712,7 @@ type EndpointOptions struct {
 func DefaultEndpoint(clusterName string, host string, ports []uint32) *v3endpointpb.ClusterLoadAssignment {
 	var bOpts []BackendOptions
 	for _, p := range ports {
-		bOpts = append(bOpts, BackendOptions{Port: p})
+		bOpts = append(bOpts, BackendOptions{Port: p, Weight: 1})
 	}
 	return EndpointResourceWithOptions(EndpointOptions{
 		ClusterName: clusterName,
@@ -729,6 +733,10 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 	for i, locality := range opts.Localities {
 		var lbEndpoints []*v3endpointpb.LbEndpoint
 		for _, b := range locality.Backends {
+			// Weight defaults to 1.
+			if b.Weight == 0 {
+				b.Weight = 1
+			}
 			lbEndpoints = append(lbEndpoints, &v3endpointpb.LbEndpoint{
 				HostIdentifier: &v3endpointpb.LbEndpoint_Endpoint{Endpoint: &v3endpointpb.Endpoint{
 					Address: &v3corepb.Address{Address: &v3corepb.Address_SocketAddress{
@@ -740,7 +748,7 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 					}},
 				}},
 				HealthStatus:        b.HealthStatus,
-				LoadBalancingWeight: &wrapperspb.UInt32Value{Value: 1},
+				LoadBalancingWeight: &wrapperspb.UInt32Value{Value: b.Weight},
 			})
 		}
 
@@ -752,7 +760,7 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 			},
 			LbEndpoints:         lbEndpoints,
 			LoadBalancingWeight: &wrapperspb.UInt32Value{Value: locality.Weight},
-			Priority:            0,
+			Priority:            locality.Priority,
 		})
 	}
 
diff --git a/stats/opentelemetry/csm/observability_test.go b/stats/opentelemetry/csm/observability_test.go
new file mode 100644
index 00000000..efa97199
--- /dev/null
+++ b/stats/opentelemetry/csm/observability_test.go
@@ -0,0 +1,611 @@
+/*
+ *
+ * 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 csm
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"testing"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding/gzip"
+	istats "google.golang.org/grpc/internal/stats"
+	"google.golang.org/grpc/internal/stubserver"
+	"google.golang.org/grpc/internal/testutils/xds/bootstrap"
+	testgrpc "google.golang.org/grpc/interop/grpc_testing"
+	testpb "google.golang.org/grpc/interop/grpc_testing"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/stats/opentelemetry"
+	"google.golang.org/grpc/stats/opentelemetry/internal/testutils"
+)
+
+// setupEnv configures the environment for CSM Observability Testing. It sets
+// the bootstrap env var to a bootstrap file with a nodeID provided. It sets CSM
+// Env Vars as well, and mocks the resource detector's returned attribute set to
+// simulate the environment. It registers a cleanup function on the provided t
+// to restore the environment to it's original state.
+func setupEnv(t *testing.T, resourceDetectorEmissions map[string]string, nodeID, csmCanonicalServiceName, csmWorkloadName string) {
+	cleanup, err := bootstrap.CreateFile(bootstrap.Options{
+		NodeID:    nodeID,
+		ServerURI: "xds_server_uri",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create bootstrap: %v", err)
+	}
+	oldCSMCanonicalServiceName, csmCanonicalServiceNamePresent := os.LookupEnv("CSM_CANONICAL_SERVICE_NAME")
+	oldCSMWorkloadName, csmWorkloadNamePresent := os.LookupEnv("CSM_WORKLOAD_NAME")
+	os.Setenv("CSM_CANONICAL_SERVICE_NAME", csmCanonicalServiceName)
+	os.Setenv("CSM_WORKLOAD_NAME", csmWorkloadName)
+
+	var attributes []attribute.KeyValue
+	for k, v := range resourceDetectorEmissions {
+		attributes = append(attributes, attribute.String(k, v))
+	}
+	// Return the attributes configured as part of the test in place
+	// of reading from resource.
+	attrSet := attribute.NewSet(attributes...)
+	origGetAttrSet := getAttrSetFromResourceDetector
+	getAttrSetFromResourceDetector = func(context.Context) *attribute.Set {
+		return &attrSet
+	}
+	t.Cleanup(func() {
+		cleanup()
+		if csmCanonicalServiceNamePresent {
+			os.Setenv("CSM_CANONICAL_SERVICE_NAME", oldCSMCanonicalServiceName)
+		} else {
+			os.Unsetenv("CSM_CANONICAL_SERVICE_NAME")
+		}
+		if csmWorkloadNamePresent {
+			os.Setenv("CSM_WORKLOAD_NAME", oldCSMWorkloadName)
+		} else {
+			os.Unsetenv("CSM_WORKLOAD_NAME")
+		}
+
+		getAttrSetFromResourceDetector = origGetAttrSet
+	})
+}
+
+// TestCSMPluginOptionUnary tests the CSM Plugin Option and labels. It
+// configures the environment for the CSM Plugin Option to read from. It then
+// configures a system with a gRPC Client and gRPC server with the OpenTelemetry
+// Dial and Server Option configured with a CSM Plugin Option with a certain
+// unary handler set to induce different ways of setting metadata exchange
+// labels, and makes a Unary RPC. This RPC should cause certain recording for
+// each registered metric observed through a Manual Metrics Reader on the
+// provided OpenTelemetry SDK's Meter Provider. The CSM Labels emitted from the
+// plugin option should be attached to the relevant metrics.
+func (s) TestCSMPluginOptionUnary(t *testing.T) {
+	resourceDetectorEmissions := map[string]string{
+		"cloud.platform":     "gcp_kubernetes_engine",
+		"cloud.region":       "cloud_region_val", // availability_zone isn't present, so this should become location
+		"cloud.account.id":   "cloud_account_id_val",
+		"k8s.namespace.name": "k8s_namespace_name_val",
+		"k8s.cluster.name":   "k8s_cluster_name_val",
+	}
+	const nodeID = "projects/12345/networks/mesh:mesh_id/nodes/aaaa-aaaa-aaaa-aaaa"
+	const csmCanonicalServiceName = "csm_canonical_service_name"
+	const csmWorkloadName = "csm_workload_name"
+	setupEnv(t, resourceDetectorEmissions, nodeID, csmCanonicalServiceName, csmWorkloadName)
+
+	attributesWant := map[string]string{
+		"csm.workload_canonical_service": csmCanonicalServiceName, // from env
+		"csm.mesh_id":                    "mesh_id",               // from bootstrap env var
+
+		// No xDS Labels - this happens in a test below.
+
+		"csm.remote_workload_type":              "gcp_kubernetes_engine",
+		"csm.remote_workload_canonical_service": csmCanonicalServiceName,
+		"csm.remote_workload_project_id":        "cloud_account_id_val",
+		"csm.remote_workload_cluster_name":      "k8s_cluster_name_val",
+		"csm.remote_workload_namespace_name":    "k8s_namespace_name_val",
+		"csm.remote_workload_location":          "cloud_region_val",
+		"csm.remote_workload_name":              csmWorkloadName,
+	}
+
+	var csmLabels []attribute.KeyValue
+	for k, v := range attributesWant {
+		csmLabels = append(csmLabels, attribute.String(k, v))
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	tests := []struct {
+		name string
+		// To test the different operations for Unary RPC's from the interceptor
+		// level that can plumb metadata exchange header in.
+		unaryCallFunc func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error)
+		opts          testutils.MetricDataOptions
+	}{
+		{
+			name: "normal-flow",
+			unaryCallFunc: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+				return &testpb.SimpleResponse{Payload: &testpb.Payload{
+					Body: make([]byte, len(in.GetPayload().GetBody())),
+				}}, nil
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels:                  csmLabels,
+				UnaryCompressedMessageSize: float64(57),
+			},
+		},
+		{
+			name: "trailers-only",
+			unaryCallFunc: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+				return nil, errors.New("some error") // return an error and no message - this triggers trailers only - no messages or headers sent
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels:       csmLabels,
+				UnaryCallFailed: true,
+			},
+		},
+		{
+			name: "set-header",
+			unaryCallFunc: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+				grpc.SetHeader(ctx, metadata.New(map[string]string{"some-metadata": "some-metadata-val"}))
+
+				return &testpb.SimpleResponse{Payload: &testpb.Payload{
+					Body: make([]byte, len(in.GetPayload().GetBody())),
+				}}, nil
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels:                  csmLabels,
+				UnaryCompressedMessageSize: float64(57),
+			},
+		},
+		{
+			name: "send-header",
+			unaryCallFunc: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+				grpc.SendHeader(ctx, metadata.New(map[string]string{"some-metadata": "some-metadata-val"}))
+
+				return &testpb.SimpleResponse{Payload: &testpb.Payload{
+					Body: make([]byte, len(in.GetPayload().GetBody())),
+				}}, nil
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels:                  csmLabels,
+				UnaryCompressedMessageSize: float64(57),
+			},
+		},
+		{
+			name: "send-msg",
+			unaryCallFunc: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+				return &testpb.SimpleResponse{Payload: &testpb.Payload{
+					Body: make([]byte, len(in.GetPayload().GetBody())),
+				}}, nil
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels:                  csmLabels,
+				UnaryCompressedMessageSize: float64(57),
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			reader := metric.NewManualReader()
+			provider := metric.NewMeterProvider(metric.WithReader(reader))
+			ss := &stubserver.StubServer{UnaryCallF: test.unaryCallFunc}
+			po := newPluginOption(ctx)
+			sopts := []grpc.ServerOption{
+				serverOptionWithCSMPluginOption(opentelemetry.Options{
+					MetricsOptions: opentelemetry.MetricsOptions{
+						MeterProvider: provider,
+						Metrics:       opentelemetry.DefaultMetrics,
+					}}, po),
+			}
+			dopts := []grpc.DialOption{dialOptionWithCSMPluginOption(opentelemetry.Options{
+				MetricsOptions: opentelemetry.MetricsOptions{
+					MeterProvider:  provider,
+					Metrics:        opentelemetry.DefaultMetrics,
+					OptionalLabels: []string{"csm.service_name", "csm.service_namespace_name"},
+				},
+			}, po)}
+			if err := ss.Start(sopts, dopts...); err != nil {
+				t.Fatalf("Error starting endpoint server: %v", err)
+			}
+			defer ss.Stop()
+
+			var request *testpb.SimpleRequest
+			if test.opts.UnaryCompressedMessageSize != 0 {
+				request = &testpb.SimpleRequest{Payload: &testpb.Payload{
+					Body: make([]byte, 10000),
+				}}
+			}
+			// Make a Unary RPC. These should cause certain metrics to be
+			// emitted, which should be able to be observed through the Metric
+			// Reader.
+			ss.Client.UnaryCall(ctx, request, grpc.UseCompressor(gzip.Name))
+			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
+				}
+			}
+
+			opts := test.opts
+			opts.Target = ss.Target
+			wantMetrics := testutils.MetricDataUnary(opts)
+			testutils.CompareMetrics(ctx, t, reader, gotMetrics, wantMetrics)
+		})
+	}
+}
+
+// TestCSMPluginOptionStreaming tests the CSM Plugin Option and labels. It
+// configures the environment for the CSM Plugin Option to read from. It then
+// configures a system with a gRPC Client and gRPC server with the OpenTelemetry
+// Dial and Server Option configured with a CSM Plugin Option with a certain
+// streaming handler set to induce different ways of setting metadata exchange
+// labels, and makes a Streaming RPC. This RPC should cause certain recording
+// for each registered metric observed through a Manual Metrics Reader on the
+// provided OpenTelemetry SDK's Meter Provider. The CSM Labels emitted from the
+// plugin option should be attached to the relevant metrics.
+func (s) TestCSMPluginOptionStreaming(t *testing.T) {
+	resourceDetectorEmissions := map[string]string{
+		"cloud.platform":     "gcp_kubernetes_engine",
+		"cloud.region":       "cloud_region_val", // availability_zone isn't present, so this should become location
+		"cloud.account.id":   "cloud_account_id_val",
+		"k8s.namespace.name": "k8s_namespace_name_val",
+		"k8s.cluster.name":   "k8s_cluster_name_val",
+	}
+	const nodeID = "projects/12345/networks/mesh:mesh_id/nodes/aaaa-aaaa-aaaa-aaaa"
+	const csmCanonicalServiceName = "csm_canonical_service_name"
+	const csmWorkloadName = "csm_workload_name"
+	setupEnv(t, resourceDetectorEmissions, nodeID, csmCanonicalServiceName, csmWorkloadName)
+
+	attributesWant := map[string]string{
+		"csm.workload_canonical_service": csmCanonicalServiceName, // from env
+		"csm.mesh_id":                    "mesh_id",               // from bootstrap env var
+
+		// No xDS Labels - this happens in a test below.
+
+		"csm.remote_workload_type":              "gcp_kubernetes_engine",
+		"csm.remote_workload_canonical_service": csmCanonicalServiceName,
+		"csm.remote_workload_project_id":        "cloud_account_id_val",
+		"csm.remote_workload_cluster_name":      "k8s_cluster_name_val",
+		"csm.remote_workload_namespace_name":    "k8s_namespace_name_val",
+		"csm.remote_workload_location":          "cloud_region_val",
+		"csm.remote_workload_name":              csmWorkloadName,
+	}
+
+	var csmLabels []attribute.KeyValue
+	for k, v := range attributesWant {
+		csmLabels = append(csmLabels, attribute.String(k, v))
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	tests := []struct {
+		name string
+		// To test the different operations for Streaming RPC's from the
+		// interceptor level that can plumb metadata exchange header in.
+		streamingCallFunc func(stream testgrpc.TestService_FullDuplexCallServer) error
+		opts              testutils.MetricDataOptions
+	}{
+		{
+			name: "trailers-only",
+			streamingCallFunc: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+				for {
+					if _, err := stream.Recv(); err == io.EOF {
+						return nil
+					}
+				}
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels: csmLabels,
+			},
+		},
+		{
+			name: "set-header",
+			streamingCallFunc: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+				stream.SetHeader(metadata.New(map[string]string{"some-metadata": "some-metadata-val"}))
+				for {
+					if _, err := stream.Recv(); err == io.EOF {
+						return nil
+					}
+				}
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels: csmLabels,
+			},
+		},
+		{
+			name: "send-header",
+			streamingCallFunc: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+				stream.SendHeader(metadata.New(map[string]string{"some-metadata": "some-metadata-val"}))
+				for {
+					if _, err := stream.Recv(); err == io.EOF {
+						return nil
+					}
+				}
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels: csmLabels,
+			},
+		},
+		{
+			name: "send-msg",
+			streamingCallFunc: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+				stream.Send(&testpb.StreamingOutputCallResponse{Payload: &testpb.Payload{
+					Body: make([]byte, 10000),
+				}})
+				for {
+					if _, err := stream.Recv(); err == io.EOF {
+						return nil
+					}
+				}
+			},
+			opts: testutils.MetricDataOptions{
+				CSMLabels:                      csmLabels,
+				StreamingCompressedMessageSize: float64(57),
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			reader := metric.NewManualReader()
+			provider := metric.NewMeterProvider(metric.WithReader(reader))
+			ss := &stubserver.StubServer{FullDuplexCallF: test.streamingCallFunc}
+			po := newPluginOption(ctx)
+			sopts := []grpc.ServerOption{
+				serverOptionWithCSMPluginOption(opentelemetry.Options{
+					MetricsOptions: opentelemetry.MetricsOptions{
+						MeterProvider: provider,
+						Metrics:       opentelemetry.DefaultMetrics,
+					}}, po),
+			}
+			dopts := []grpc.DialOption{dialOptionWithCSMPluginOption(opentelemetry.Options{
+				MetricsOptions: opentelemetry.MetricsOptions{
+					MeterProvider:  provider,
+					Metrics:        opentelemetry.DefaultMetrics,
+					OptionalLabels: []string{"csm.service_name", "csm.service_namespace_name"},
+				},
+			}, po)}
+			if err := ss.Start(sopts, dopts...); err != nil {
+				t.Fatalf("Error starting endpoint server: %v", err)
+			}
+			defer ss.Stop()
+
+			stream, err := ss.Client.FullDuplexCall(ctx, grpc.UseCompressor(gzip.Name))
+			if err != nil {
+				t.Fatalf("ss.Client.FullDuplexCall failed: %f", err)
+			}
+
+			if test.opts.StreamingCompressedMessageSize != 0 {
+				if err := stream.Send(&testpb.StreamingOutputCallRequest{Payload: &testpb.Payload{
+					Body: make([]byte, 10000),
+				}}); err != nil {
+					t.Fatalf("stream.Send failed")
+				}
+				if _, err := stream.Recv(); err != nil {
+					t.Fatalf("stream.Recv failed with error: %v", err)
+				}
+			}
+
+			stream.CloseSend()
+			if _, err = stream.Recv(); err != io.EOF {
+				t.Fatalf("stream.Recv received an unexpected error: %v, expected an EOF error", err)
+			}
+
+			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
+				}
+			}
+
+			opts := test.opts
+			opts.Target = ss.Target
+			wantMetrics := testutils.MetricDataStreaming(opts)
+			testutils.CompareMetrics(ctx, t, reader, gotMetrics, wantMetrics)
+		})
+	}
+}
+
+func unaryInterceptorAttachXDSLabels(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+	ctx = istats.SetLabels(ctx, &istats.Labels{
+		TelemetryLabels: map[string]string{
+			// mock what the cluster impl would write here ("csm." xDS Labels)
+			"csm.service_name":           "service_name_val",
+			"csm.service_namespace_name": "service_namespace_val",
+		},
+	})
+
+	// TagRPC will just see this in the context and set it's xDS Labels to point
+	// to this map on the heap.
+	return invoker(ctx, method, req, reply, cc, opts...)
+}
+
+// TestXDSLabels tests that xDS Labels get emitted from OpenTelemetry metrics.
+// This test configures OpenTelemetry with the CSM Plugin Option, and xDS
+// Optional Labels turned on. It then configures an interceptor to attach
+// labels, representing the cluster_impl picker. It then makes a unary RPC, and
+// expects xDS Labels labels to be attached to emitted relevant metrics. Full
+// xDS System alongside OpenTelemetry will be tested with interop. (there is
+// a test for xDS -> Stats handler and this tests -> OTel -> emission).
+func (s) TestXDSLabels(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	reader := metric.NewManualReader()
+	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{
+				Body: make([]byte, len(in.GetPayload().GetBody())),
+			}}, nil
+		},
+	}
+
+	po := newPluginOption(ctx)
+	dopts := []grpc.DialOption{dialOptionWithCSMPluginOption(opentelemetry.Options{
+		MetricsOptions: opentelemetry.MetricsOptions{
+			MeterProvider:  provider,
+			Metrics:        opentelemetry.DefaultMetrics,
+			OptionalLabels: []string{"csm.service_name", "csm.service_namespace_name"},
+		},
+	}, po), grpc.WithUnaryInterceptor(unaryInterceptorAttachXDSLabels)}
+	if err := ss.Start(nil, dopts...); err != nil {
+		t.Fatalf("Error starting endpoint server: %v", err)
+	}
+
+	defer ss.Stop()
+	ss.Client.UnaryCall(ctx, &testpb.SimpleRequest{Payload: &testpb.Payload{
+		Body: make([]byte, 10000),
+	}}, grpc.UseCompressor(gzip.Name))
+
+	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
+		}
+	}
+
+	unaryMethodAttr := attribute.String("grpc.method", "grpc.testing.TestService/UnaryCall")
+	targetAttr := attribute.String("grpc.target", ss.Target)
+	unaryStatusAttr := attribute.String("grpc.status", "OK")
+
+	serviceNameAttr := attribute.String("csm.service_name", "service_name_val")
+	serviceNamespaceAttr := attribute.String("csm.service_namespace_name", "service_namespace_val")
+	meshIDAttr := attribute.String("csm.mesh_id", "unknown")
+	workloadCanonicalServiceAttr := attribute.String("csm.workload_canonical_service", "unknown")
+	remoteWorkloadTypeAttr := attribute.String("csm.remote_workload_type", "unknown")
+	remoteWorkloadCanonicalServiceAttr := attribute.String("csm.remote_workload_canonical_service", "unknown")
+
+	unaryMethodClientSideEnd := []attribute.KeyValue{
+		unaryMethodAttr,
+		targetAttr,
+		unaryStatusAttr,
+		serviceNameAttr,
+		serviceNamespaceAttr,
+		meshIDAttr,
+		workloadCanonicalServiceAttr,
+		remoteWorkloadTypeAttr,
+		remoteWorkloadCanonicalServiceAttr,
+	}
+
+	unaryCompressedBytesSentRecv := int64(57) // Fixed 10000 bytes with gzip assumption.
+	unaryBucketCounts := []uint64{0x0, 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0}
+	unaryExtrema := metricdata.NewExtrema(int64(57))
+	wantMetrics := []metricdata.Metrics{
+		{
+			Name:        "grpc.client.attempt.started",
+			Description: "Number of client call attempts started.",
+			Unit:        "attempt",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(unaryMethodAttr, targetAttr),
+						Value:      1,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		}, // Doesn't have xDS Labels, CSM Labels start from header or trailer from server, whichever comes first, so this doesn't need it
+		{
+			Name:        "grpc.client.attempt.duration",
+			Description: "End-to-end time taken to complete a client call attempt.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(unaryMethodClientSideEnd...),
+						Count:      1,
+						Bounds:     testutils.DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.sent_total_compressed_message_size",
+			Description: "Compressed message bytes sent per client call attempt.",
+			Unit:        "By",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(unaryMethodClientSideEnd...),
+						Count:        1,
+						Bounds:       testutils.DefaultSizeBounds,
+						BucketCounts: unaryBucketCounts,
+						Min:          unaryExtrema,
+						Max:          unaryExtrema,
+						Sum:          unaryCompressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.rcvd_total_compressed_message_size",
+			Description: "Compressed message bytes received per call attempt.",
+			Unit:        "By",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(unaryMethodClientSideEnd...),
+						Count:        1,
+						Bounds:       testutils.DefaultSizeBounds,
+						BucketCounts: unaryBucketCounts,
+						Min:          unaryExtrema,
+						Max:          unaryExtrema,
+						Sum:          unaryCompressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.call.duration",
+			Description: "Time taken by gRPC to complete an RPC from application's perspective.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(unaryMethodAttr, targetAttr, unaryStatusAttr),
+						Count:      1,
+						Bounds:     testutils.DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+	}
+
+	testutils.CompareMetrics(ctx, t, reader, gotMetrics, wantMetrics)
+}
+
+// TestObservability tests that Observability global function compiles and runs
+// without error. The actual functionality of this function will be verified in
+// interop tests.
+func (s) TestObservability(t *testing.T) {
+	cleanup := EnableObservability(context.Background(), opentelemetry.Options{})
+	cleanup()
+}
diff --git a/stats/opentelemetry/e2e_test.go b/stats/opentelemetry/e2e_test.go
index c0850d62..aa473474 100644
--- a/stats/opentelemetry/e2e_test.go
+++ b/stats/opentelemetry/e2e_test.go
@@ -14,11 +14,10 @@
  * limitations under the License.
  */
 
-package opentelemetry
+package opentelemetry_test
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"testing"
 	"time"
@@ -29,6 +28,8 @@ import (
 	"google.golang.org/grpc/internal/stubserver"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
+	"google.golang.org/grpc/stats/opentelemetry"
+	"google.golang.org/grpc/stats/opentelemetry/internal/testutils"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/metric"
@@ -46,35 +47,10 @@ func Test(t *testing.T) {
 	grpctest.RunSubTests(t, s{})
 }
 
-// waitForServerCompletedRPCs waits until the unary and streaming stats.End
-// calls are finished processing. It does this by waiting for the expected
-// metric triggered by stats.End to appear through the passed in metrics reader.
-func waitForServerCompletedRPCs(ctx context.Context, reader metric.Reader, wantMetric metricdata.Metrics, t *testing.T) (map[string]metricdata.Metrics, error) {
-	for ; ctx.Err() == nil; <-time.After(time.Millisecond) {
-		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
-			}
-		}
-		val, ok := gotMetrics[wantMetric.Name]
-		if !ok {
-			continue
-		}
-		if !metricdatatest.AssertEqual(t, wantMetric, val, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars()) {
-			continue
-		}
-		return gotMetrics, nil
-	}
-	return nil, fmt.Errorf("error waiting for metric %v: %v", wantMetric, ctx.Err())
-}
-
 // setup creates a stub server with OpenTelemetry component configured on client
 // and server side. It returns a reader for metrics emitted from OpenTelemetry
 // component and the server.
-func setup(t *testing.T, tafOn bool, maf func(string) bool) (*metric.ManualReader, *stubserver.StubServer) {
+func setup(t *testing.T, methodAttributeFilter func(string) bool) (*metric.ManualReader, *stubserver.StubServer) {
 	reader := metric.NewManualReader()
 	provider := metric.NewMeterProvider(
 		metric.WithReader(reader),
@@ -82,7 +58,7 @@ func setup(t *testing.T, tafOn bool, maf func(string) bool) (*metric.ManualReade
 	ss := &stubserver.StubServer{
 		UnaryCallF: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
 			return &testpb.SimpleResponse{Payload: &testpb.Payload{
-				Body: make([]byte, 10000),
+				Body: make([]byte, len(in.GetPayload().GetBody())),
 			}}, nil
 		},
 		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
@@ -94,24 +70,16 @@ func setup(t *testing.T, tafOn bool, maf func(string) bool) (*metric.ManualReade
 			}
 		},
 	}
-	var taf func(string) bool
-	if tafOn {
-		taf = func(str string) bool {
-			return str != ss.Target
-		}
-	}
-	if err := ss.Start([]grpc.ServerOption{ServerOption(Options{
-		MetricsOptions: MetricsOptions{
-			MeterProvider:         provider,
-			Metrics:               DefaultMetrics,
-			TargetAttributeFilter: taf,
-			MethodAttributeFilter: maf,
-		}})}, DialOption(Options{
-		MetricsOptions: MetricsOptions{
+
+	if err := ss.Start([]grpc.ServerOption{opentelemetry.ServerOption(opentelemetry.Options{
+		MetricsOptions: opentelemetry.MetricsOptions{
 			MeterProvider:         provider,
-			Metrics:               DefaultMetrics,
-			TargetAttributeFilter: taf,
-			MethodAttributeFilter: maf,
+			Metrics:               opentelemetry.DefaultMetrics,
+			MethodAttributeFilter: methodAttributeFilter,
+		}})}, opentelemetry.DialOption(opentelemetry.Options{
+		MetricsOptions: opentelemetry.MetricsOptions{
+			MeterProvider: provider,
+			Metrics:       opentelemetry.DefaultMetrics,
 		},
 	})); err != nil {
 		t.Fatalf("Error starting endpoint server: %v", err)
@@ -119,20 +87,19 @@ func setup(t *testing.T, tafOn bool, maf func(string) bool) (*metric.ManualReade
 	return reader, ss
 }
 
-// TestMethodTargetAttributeFilter tests the method and target attribute filter.
-// The method and target filter set should bucket the grpc.method/grpc.target
-// attribute into "other" if filter specifies.
-func (s) TestMethodTargetAttributeFilter(t *testing.T) {
+// TestMethodAttributeFilter tests the method attribute filter. The method
+// filter set should bucket the grpc.method attribute into "other" if the method
+// attribute filter specifies.
+func (s) TestMethodAttributeFilter(t *testing.T) {
 	maf := func(str string) bool {
 		// Will allow duplex/any other type of RPC.
-		return str != "/grpc.testing.TestService/UnaryCall"
+		return str != testpb.TestService_UnaryCall_FullMethodName
 	}
-	// pull out setup into a helper
-	reader, ss := setup(t, true, maf)
+	reader, ss := setup(t, maf)
 	defer ss.Stop()
 
-	// make a single RPC (unary rpc), and filter out the target and method
-	// that would correspond.
+	// Make a Unary and Streaming RPC. The Unary RPC should be filtered by the
+	// method attribute filter, and the Full Duplex (Streaming) RPC should not.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	if _, err := ss.Client.UnaryCall(ctx, &testpb.SimpleRequest{Payload: &testpb.Payload{
@@ -147,10 +114,16 @@ func (s) TestMethodTargetAttributeFilter(t *testing.T) {
 
 	stream.CloseSend()
 	if _, err = stream.Recv(); err != io.EOF {
-		t.Fatalf("unexpected error: %v, expected an EOF error", err)
+		t.Fatalf("stream.Recv received an unexpected error: %v, expected an EOF error", err)
 	}
 	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
+		}
+	}
 
 	wantMetrics := []metricdata.Metrics{
 		{
@@ -160,11 +133,11 @@ func (s) TestMethodTargetAttributeFilter(t *testing.T) {
 			Data: metricdata.Sum[int64]{
 				DataPoints: []metricdata.DataPoint[int64]{
 					{
-						Attributes: attribute.NewSet(attribute.String("grpc.method", "grpc.testing.TestService/UnaryCall"), attribute.String("grpc.target", "other")),
+						Attributes: attribute.NewSet(attribute.String("grpc.method", "grpc.testing.TestService/UnaryCall"), attribute.String("grpc.target", ss.Target)),
 						Value:      1,
 					},
 					{
-						Attributes: attribute.NewSet(attribute.String("grpc.method", "grpc.testing.TestService/FullDuplexCall"), attribute.String("grpc.target", "other")),
+						Attributes: attribute.NewSet(attribute.String("grpc.method", "grpc.testing.TestService/FullDuplexCall"), attribute.String("grpc.target", ss.Target)),
 						Value:      1,
 					},
 				},
@@ -172,54 +145,29 @@ func (s) TestMethodTargetAttributeFilter(t *testing.T) {
 				IsMonotonic: true,
 			},
 		},
+		{
+			Name:        "grpc.server.call.duration",
+			Description: "End-to-end time taken to complete a call from server transport's perspective.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{ // Method should go to "other" due to the method attribute filter.
+						Attributes: attribute.NewSet(attribute.String("grpc.method", "other"), attribute.String("grpc.status", "OK")),
+						Count:      1,
+						Bounds:     testutils.DefaultLatencyBounds,
+					},
+					{
+						Attributes: attribute.NewSet(attribute.String("grpc.method", "grpc.testing.TestService/FullDuplexCall"), attribute.String("grpc.status", "OK")),
+						Count:      1,
+						Bounds:     testutils.DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
 	}
-	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)
-		}
-	}
-}
 
-// assertDataPointWithinFiveSeconds asserts the metric passed in contains
-// a histogram with dataPoints that fall within buckets that are <=5.
-func assertDataPointWithinFiveSeconds(metric metricdata.Metrics) error {
-	histo, ok := metric.Data.(metricdata.Histogram[float64])
-	if !ok {
-		return fmt.Errorf("metric data is not histogram")
-	}
-	for _, dataPoint := range histo.DataPoints {
-		var boundWithFive int
-		for i, bucket := range dataPoint.Bounds {
-			if bucket >= 5 {
-				boundWithFive = i
-			}
-		}
-		foundPoint := false
-		for i, bucket := range dataPoint.BucketCounts {
-			if i >= boundWithFive {
-				return fmt.Errorf("data point not found in bucket <=5 seconds")
-			}
-			if bucket == 1 {
-				foundPoint = true
-				break
-			}
-		}
-		if !foundPoint {
-			return fmt.Errorf("no data point found for metric")
-		}
-	}
-	return nil
+	testutils.CompareMetrics(ctx, t, reader, gotMetrics, wantMetrics)
 }
 
 // TestAllMetricsOneFunction tests emitted metrics from OpenTelemetry
@@ -232,7 +180,7 @@ func assertDataPointWithinFiveSeconds(metric metricdata.Metrics) error {
 // on the Client (no StaticMethodCallOption set) and Server. The method
 // attribute on subsequent metrics should be bucketed in "other".
 func (s) TestAllMetricsOneFunction(t *testing.T) {
-	reader, ss := setup(t, false, nil)
+	reader, ss := setup(t, nil)
 	defer ss.Stop()
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -251,7 +199,7 @@ func (s) TestAllMetricsOneFunction(t *testing.T) {
 
 	stream.CloseSend()
 	if _, err = stream.Recv(); err != io.EOF {
-		t.Fatalf("unexpected error: %v, expected an EOF error", err)
+		t.Fatalf("stream.Recv received an unexpected error: %v, expected an EOF error", err)
 	}
 
 	rm := &metricdata.ResourceMetrics{}
@@ -264,257 +212,11 @@ func (s) TestAllMetricsOneFunction(t *testing.T) {
 		}
 	}
 
-	unaryMethodAttr := attribute.String("grpc.method", "grpc.testing.TestService/UnaryCall")
-	duplexMethodAttr := attribute.String("grpc.method", "grpc.testing.TestService/FullDuplexCall")
-
-	targetAttr := attribute.String("grpc.target", ss.Target)
-	statusAttr := attribute.String("grpc.status", "OK")
-
-	wantMetrics := []metricdata.Metrics{
-		{
-			Name:        "grpc.client.attempt.started",
-			Description: "Number of client call attempts started.",
-			Unit:        "attempt",
-			Data: metricdata.Sum[int64]{
-				DataPoints: []metricdata.DataPoint[int64]{
-					{
-						Attributes: attribute.NewSet(unaryMethodAttr, targetAttr),
-						Value:      1,
-					},
-					{
-						Attributes: attribute.NewSet(duplexMethodAttr, targetAttr),
-						Value:      1,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-				IsMonotonic: true,
-			},
-		},
-		{
-			Name:        "grpc.client.attempt.duration",
-			Description: "End-to-end time taken to complete a client call attempt.",
-			Unit:        "s",
-			Data: metricdata.Histogram[float64]{
-				DataPoints: []metricdata.HistogramDataPoint[float64]{
-					{
-						Attributes: attribute.NewSet(unaryMethodAttr, targetAttr, statusAttr),
-						Count:      1,
-						Bounds:     DefaultLatencyBounds,
-					},
-					{
-						Attributes: attribute.NewSet(duplexMethodAttr, targetAttr, statusAttr),
-						Count:      1,
-						Bounds:     DefaultLatencyBounds,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-			},
-		},
-		{
-			Name:        "grpc.client.attempt.sent_total_compressed_message_size",
-			Description: "Compressed message bytes sent per client call attempt.",
-			Unit:        "By",
-			Data: metricdata.Histogram[int64]{
-				DataPoints: []metricdata.HistogramDataPoint[int64]{
-					{
-						Attributes:   attribute.NewSet(unaryMethodAttr, targetAttr, statusAttr),
-						Count:        1,
-						Bounds:       DefaultSizeBounds,
-						BucketCounts: []uint64{0x0, 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
-						Min:          metricdata.NewExtrema(int64(57)),
-						Max:          metricdata.NewExtrema(int64(57)),
-						Sum:          57,
-					},
-					{
-						Attributes:   attribute.NewSet(duplexMethodAttr, targetAttr, statusAttr),
-						Count:        1,
-						Bounds:       DefaultSizeBounds,
-						BucketCounts: []uint64{0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
-						Min:          metricdata.NewExtrema(int64(0)),
-						Max:          metricdata.NewExtrema(int64(0)),
-						Sum:          0,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-			},
-		},
-		{
-			Name:        "grpc.client.attempt.rcvd_total_compressed_message_size",
-			Description: "Compressed message bytes received per call attempt.",
-			Unit:        "By",
-			Data: metricdata.Histogram[int64]{
-				DataPoints: []metricdata.HistogramDataPoint[int64]{
-					{
-						Attributes:   attribute.NewSet(unaryMethodAttr, targetAttr, statusAttr),
-						Count:        1,
-						Bounds:       DefaultSizeBounds,
-						BucketCounts: []uint64{0x0, 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
-						Min:          metricdata.NewExtrema(int64(57)),
-						Max:          metricdata.NewExtrema(int64(57)),
-						Sum:          57,
-					},
-					{
-						Attributes:   attribute.NewSet(duplexMethodAttr, targetAttr, statusAttr),
-						Count:        1,
-						Bounds:       DefaultSizeBounds,
-						BucketCounts: []uint64{0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
-						Min:          metricdata.NewExtrema(int64(0)),
-						Max:          metricdata.NewExtrema(int64(0)),
-						Sum:          0,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-			},
-		},
-		{
-			Name:        "grpc.client.call.duration",
-			Description: "Time taken by gRPC to complete an RPC from application's perspective.",
-			Unit:        "s",
-			Data: metricdata.Histogram[float64]{
-				DataPoints: []metricdata.HistogramDataPoint[float64]{
-					{
-						Attributes: attribute.NewSet(unaryMethodAttr, targetAttr, statusAttr),
-						Count:      1,
-						Bounds:     DefaultLatencyBounds,
-					},
-					{
-						Attributes: attribute.NewSet(duplexMethodAttr, targetAttr, statusAttr),
-						Count:      1,
-						Bounds:     DefaultLatencyBounds,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-			},
-		},
-		{
-			Name:        "grpc.server.call.started",
-			Description: "Number of server calls started.",
-			Unit:        "call",
-			Data: metricdata.Sum[int64]{
-				DataPoints: []metricdata.DataPoint[int64]{
-					{
-						Attributes: attribute.NewSet(unaryMethodAttr),
-						Value:      1,
-					},
-					{
-						Attributes: attribute.NewSet(duplexMethodAttr),
-						Value:      1,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-				IsMonotonic: true,
-			},
-		},
-		{
-			Name:        "grpc.server.call.sent_total_compressed_message_size",
-			Unit:        "By",
-			Description: "Compressed message bytes sent per server call.",
-			Data: metricdata.Histogram[int64]{
-				DataPoints: []metricdata.HistogramDataPoint[int64]{
-					{
-						Attributes:   attribute.NewSet(unaryMethodAttr, statusAttr),
-						Count:        1,
-						Bounds:       DefaultSizeBounds,
-						BucketCounts: []uint64{0x0, 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
-						Min:          metricdata.NewExtrema(int64(57)),
-						Max:          metricdata.NewExtrema(int64(57)),
-						Sum:          57,
-					},
-					{
-						Attributes:   attribute.NewSet(duplexMethodAttr, statusAttr),
-						Count:        1,
-						Bounds:       DefaultSizeBounds,
-						BucketCounts: []uint64{0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
-						Min:          metricdata.NewExtrema(int64(0)),
-						Max:          metricdata.NewExtrema(int64(0)),
-						Sum:          0,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-			},
-		},
-		{
-			Name:        "grpc.server.call.rcvd_total_compressed_message_size",
-			Unit:        "By",
-			Description: "Compressed message bytes received per server call.",
-			Data: metricdata.Histogram[int64]{
-				DataPoints: []metricdata.HistogramDataPoint[int64]{
-					{
-						Attributes:   attribute.NewSet(unaryMethodAttr, statusAttr),
-						Count:        1,
-						Bounds:       DefaultSizeBounds,
-						BucketCounts: []uint64{0x0, 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
-						Min:          metricdata.NewExtrema(int64(57)),
-						Max:          metricdata.NewExtrema(int64(57)),
-						Sum:          57,
-					},
-					{
-						Attributes:   attribute.NewSet(duplexMethodAttr, statusAttr),
-						Count:        1,
-						Bounds:       DefaultSizeBounds,
-						BucketCounts: []uint64{0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0},
-						Min:          metricdata.NewExtrema(int64(0)),
-						Max:          metricdata.NewExtrema(int64(0)),
-						Sum:          0,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-			},
-		},
-		{
-			Name:        "grpc.server.call.duration",
-			Description: "End-to-end time taken to complete a call from server transport's perspective.",
-			Unit:        "s",
-			Data: metricdata.Histogram[float64]{
-				DataPoints: []metricdata.HistogramDataPoint[float64]{
-					{
-						Attributes: attribute.NewSet(unaryMethodAttr, statusAttr),
-						Count:      1,
-						Bounds:     DefaultLatencyBounds,
-					},
-					{
-						Attributes: attribute.NewSet(duplexMethodAttr, statusAttr),
-						Count:      1,
-						Bounds:     DefaultLatencyBounds,
-					},
-				},
-				Temporality: metricdata.CumulativeTemporality,
-			},
-		},
-	}
-
-	for _, metric := range wantMetrics {
-		if metric.Name == "grpc.server.call.sent_total_compressed_message_size" || metric.Name == "grpc.server.call.rcvd_total_compressed_message_size" {
-			// Sync the metric reader to see the event because stats.End is
-			// handled async server side. Thus, poll until metrics created from
-			// stats.End show up.
-			if gotMetrics, err = waitForServerCompletedRPCs(ctx, reader, metric, t); err != nil {
-				t.Fatalf("error waiting for sent total compressed message size for metric: %v", metric.Name)
-			}
-			continue
-		}
-
-		// If one of the duration metrics, ignore the bucket counts, and make
-		// sure it count falls within a bucket <= 5 seconds (maximum duration of
-		// test due to context).
-		val, ok := gotMetrics[metric.Name]
-		if !ok {
-			t.Fatalf("metric %v not present in recorded metrics", metric.Name)
-		}
-		if metric.Name == "grpc.client.attempt.duration" || metric.Name == "grpc.client.call.duration" || metric.Name == "grpc.server.call.duration" {
-			if !metricdatatest.AssertEqual(t, metric, val, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars(), metricdatatest.IgnoreValue()) {
-				t.Fatalf("metrics data type not equal for metric: %v", metric.Name)
-			}
-			if err := assertDataPointWithinFiveSeconds(val); err != nil {
-				t.Fatalf("Data point not within five seconds for metric %v: %v", metric.Name, err)
-			}
-			continue
-		}
-
-		if !metricdatatest.AssertEqual(t, metric, val, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars()) {
-			t.Fatalf("metrics data type not equal for metric: %v", metric.Name)
-		}
-	}
+	wantMetrics := testutils.MetricData(testutils.MetricDataOptions{
+		Target:                     ss.Target,
+		UnaryCompressedMessageSize: float64(57),
+	})
+	testutils.CompareMetrics(ctx, t, reader, gotMetrics, wantMetrics)
 
 	stream, err = ss.Client.FullDuplexCall(ctx)
 	if err != nil {
@@ -523,7 +225,7 @@ func (s) TestAllMetricsOneFunction(t *testing.T) {
 
 	stream.CloseSend()
 	if _, err = stream.Recv(); err != io.EOF {
-		t.Fatalf("unexpected error: %v, expected an EOF error", err)
+		t.Fatalf("stream.Recv received an unexpected error: %v, expected an EOF error", err)
 	}
 	// This Invoke doesn't pass the StaticMethodCallOption. Thus, the method
 	// attribute should become "other" on client side metrics. Since it is also
@@ -541,6 +243,10 @@ func (s) TestAllMetricsOneFunction(t *testing.T) {
 			gotMetrics[m.Name] = m
 		}
 	}
+	unaryMethodAttr := attribute.String("grpc.method", "grpc.testing.TestService/UnaryCall")
+	duplexMethodAttr := attribute.String("grpc.method", "grpc.testing.TestService/FullDuplexCall")
+
+	targetAttr := attribute.String("grpc.target", ss.Target)
 	otherMethodAttr := attribute.String("grpc.method", "other")
 	wantMetrics = []metricdata.Metrics{
 		{
@@ -593,10 +299,10 @@ func (s) TestAllMetricsOneFunction(t *testing.T) {
 	for _, metric := range wantMetrics {
 		val, ok := gotMetrics[metric.Name]
 		if !ok {
-			t.Fatalf("metric %v not present in recorded metrics", metric.Name)
+			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)
+			t.Fatalf("Metrics data type not equal for metric: %v", metric.Name)
 		}
 	}
 }
diff --git a/stats/opentelemetry/example_test.go b/stats/opentelemetry/example_test.go
index 607bcf55..b4d6755b 100644
--- a/stats/opentelemetry/example_test.go
+++ b/stats/opentelemetry/example_test.go
@@ -17,8 +17,6 @@
 package opentelemetry_test
 
 import (
-	"strings"
-
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/stats/opentelemetry"
@@ -55,9 +53,6 @@ func Example_dialOption() {
 		MetricsOptions: opentelemetry.MetricsOptions{
 			MeterProvider: provider,
 			Metrics:       opentelemetry.DefaultMetrics, // equivalent to unset - distinct from empty
-			TargetAttributeFilter: func(str string) bool {
-				return !strings.HasPrefix(str, "dns") // Filter out DNS targets.
-			},
 		},
 	}
 	do := opentelemetry.DialOption(opts)
diff --git a/stats/opentelemetry/internal/testutils/testutils.go b/stats/opentelemetry/internal/testutils/testutils.go
new file mode 100644
index 00000000..10856409
--- /dev/null
+++ b/stats/opentelemetry/internal/testutils/testutils.go
@@ -0,0 +1,796 @@
+/*
+ * 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 contains helpers for OpenTelemetry tests.
+package testutils
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+)
+
+// Redefine default bounds here to avoid a cyclic dependency with top level
+// opentelemetry package. Could define once through internal, but would make
+// external opentelemetry godoc less readable.
+var (
+	// DefaultLatencyBounds are the default bounds for latency metrics.
+	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}
+	// 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}
+)
+
+// waitForServerCompletedRPCs waits until the unary and streaming stats.End
+// calls are finished processing. It does this by waiting for the expected
+// metric triggered by stats.End to appear through the passed in metrics reader.
+//
+// Returns a new gotMetrics map containing the metric data being polled for, or
+// an error if failed to wait for metric.
+func waitForServerCompletedRPCs(ctx context.Context, t *testing.T, reader metric.Reader, wantMetric metricdata.Metrics) (map[string]metricdata.Metrics, error) {
+	for ; ctx.Err() == nil; <-time.After(time.Millisecond) {
+		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
+			}
+		}
+		val, ok := gotMetrics[wantMetric.Name]
+		if !ok {
+			continue
+		}
+		metricdatatest.AssertEqual(t, wantMetric, val, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+		return gotMetrics, nil
+	}
+	return nil, fmt.Errorf("error waiting for metric %v: %v", wantMetric, ctx.Err())
+}
+
+// checkDataPointWithinFiveSeconds checks if the metric passed in contains a
+// histogram with dataPoints that fall within buckets that are <=5. Returns an
+// error if check fails.
+func checkDataPointWithinFiveSeconds(metric metricdata.Metrics) error {
+	histo, ok := metric.Data.(metricdata.Histogram[float64])
+	if !ok {
+		return fmt.Errorf("metric data is not histogram")
+	}
+	for _, dataPoint := range histo.DataPoints {
+		var boundWithFive int
+		for i, bound := range dataPoint.Bounds {
+			if bound >= 5 {
+				boundWithFive = i
+			}
+		}
+		foundPoint := false
+		for i, count := range dataPoint.BucketCounts {
+			if i >= boundWithFive {
+				return fmt.Errorf("data point not found in bucket <=5 seconds")
+			}
+			if count == 1 {
+				foundPoint = true
+				break
+			}
+		}
+		if !foundPoint {
+			return fmt.Errorf("no data point found for metric")
+		}
+	}
+	return nil
+}
+
+// MetricDataOptions are the options used to configure the metricData emissions
+// of expected metrics data from NewMetricData.
+type MetricDataOptions struct {
+	// CSMLabels are the csm labels to attach to metrics which receive csm
+	// labels (all A66 expect client call and started RPC's client and server
+	// side).
+	CSMLabels []attribute.KeyValue
+	// Target is the target of the client and server.
+	Target string
+	// UnaryCallFailed is whether the Unary Call failed, which would trigger
+	// trailers only.
+	UnaryCallFailed bool
+	// UnaryCompressedMessageSize is the compressed message size of the Unary
+	// RPC. This assumes both client and server sent the same message size.
+	UnaryCompressedMessageSize float64
+	// StreamingCompressedMessageSize is the compressed message size of the
+	// Streaming RPC. This assumes both client and server sent the same message
+	// size.
+	StreamingCompressedMessageSize float64
+}
+
+// createBucketCounts creates a list of bucket counts based off the
+// recordingPoints and bounds. Both recordingPoints and bounds are assumed to be
+// in order.
+func createBucketCounts(recordingPoints []float64, bounds []float64) []uint64 {
+	var bucketCounts []uint64
+	var recordingPointIndex int
+	for _, bound := range bounds {
+		var bucketCount uint64
+		if recordingPointIndex >= len(recordingPoints) {
+			bucketCounts = append(bucketCounts, bucketCount)
+			continue
+		}
+		for recordingPoints[recordingPointIndex] <= bound {
+			bucketCount += 1
+			recordingPointIndex += 1
+			if recordingPointIndex >= len(recordingPoints) {
+				break
+			}
+		}
+		bucketCounts = append(bucketCounts, bucketCount)
+	}
+	// The rest of the recording points are last bound -> infinity.
+	bucketCounts = append(bucketCounts, uint64(len(recordingPoints)-recordingPointIndex))
+	return bucketCounts
+}
+
+// MetricDataUnary returns a list of expected metrics defined in A66 for a
+// client and server for one unary RPC.
+func MetricDataUnary(options MetricDataOptions) []metricdata.Metrics {
+	methodAttr := attribute.String("grpc.method", "grpc.testing.TestService/UnaryCall")
+	targetAttr := attribute.String("grpc.target", options.Target)
+	statusAttr := attribute.String("grpc.status", "OK")
+	if options.UnaryCallFailed {
+		statusAttr = attribute.String("grpc.status", "UNKNOWN")
+	}
+	clientSideEnd := []attribute.KeyValue{
+		methodAttr,
+		targetAttr,
+		statusAttr,
+	}
+	serverSideEnd := []attribute.KeyValue{
+		methodAttr,
+		statusAttr,
+	}
+	clientSideEnd = append(clientSideEnd, options.CSMLabels...)
+	serverSideEnd = append(serverSideEnd, options.CSMLabels...)
+	compressedBytesSentRecv := int64(options.UnaryCompressedMessageSize)
+	bucketCounts := createBucketCounts([]float64{options.UnaryCompressedMessageSize}, DefaultSizeBounds)
+	extrema := metricdata.NewExtrema(int64(options.UnaryCompressedMessageSize))
+	return []metricdata.Metrics{
+		{
+			Name:        "grpc.client.attempt.started",
+			Description: "Number of client call attempts started.",
+			Unit:        "attempt",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(methodAttr, targetAttr),
+						Value:      1,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.duration",
+			Description: "End-to-end time taken to complete a client call attempt.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(clientSideEnd...),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.sent_total_compressed_message_size",
+			Description: "Compressed message bytes sent per client call attempt.",
+			Unit:        "By",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(clientSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: bucketCounts,
+						Min:          extrema,
+						Max:          extrema,
+						Sum:          compressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.rcvd_total_compressed_message_size",
+			Description: "Compressed message bytes received per call attempt.",
+			Unit:        "By",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(clientSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: bucketCounts,
+						Min:          extrema,
+						Max:          extrema,
+						Sum:          compressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.call.duration",
+			Description: "Time taken by gRPC to complete an RPC from application's perspective.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(methodAttr, targetAttr, statusAttr),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.started",
+			Description: "Number of server calls started.",
+			Unit:        "call",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(methodAttr),
+						Value:      1,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "grpc.server.call.sent_total_compressed_message_size",
+			Unit:        "By",
+			Description: "Compressed message bytes sent per server call.",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(serverSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: bucketCounts,
+						Min:          extrema,
+						Max:          extrema,
+						Sum:          compressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.rcvd_total_compressed_message_size",
+			Unit:        "By",
+			Description: "Compressed message bytes received per server call.",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(serverSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: bucketCounts,
+						Min:          extrema,
+						Max:          extrema,
+						Sum:          compressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.duration",
+			Description: "End-to-end time taken to complete a call from server transport's perspective.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(serverSideEnd...),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+	}
+}
+
+// MetricDataStreaming returns a list of expected metrics defined in A66 for a
+// client and server for one streaming RPC.
+func MetricDataStreaming(options MetricDataOptions) []metricdata.Metrics {
+	methodAttr := attribute.String("grpc.method", "grpc.testing.TestService/FullDuplexCall")
+	targetAttr := attribute.String("grpc.target", options.Target)
+	statusAttr := attribute.String("grpc.status", "OK")
+	clientSideEnd := []attribute.KeyValue{
+		methodAttr,
+		targetAttr,
+		statusAttr,
+	}
+	serverSideEnd := []attribute.KeyValue{
+		methodAttr,
+		statusAttr,
+	}
+	clientSideEnd = append(clientSideEnd, options.CSMLabels...)
+	serverSideEnd = append(serverSideEnd, options.CSMLabels...)
+	compressedBytesSentRecv := int64(options.StreamingCompressedMessageSize)
+	bucketCounts := createBucketCounts([]float64{options.StreamingCompressedMessageSize}, DefaultSizeBounds)
+	extrema := metricdata.NewExtrema(int64(options.StreamingCompressedMessageSize))
+	return []metricdata.Metrics{
+		{
+			Name:        "grpc.client.attempt.started",
+			Description: "Number of client call attempts started.",
+			Unit:        "attempt",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(methodAttr, targetAttr),
+						Value:      1,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.duration",
+			Description: "End-to-end time taken to complete a client call attempt.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(clientSideEnd...),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.sent_total_compressed_message_size",
+			Description: "Compressed message bytes sent per client call attempt.",
+			Unit:        "By",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(clientSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: bucketCounts,
+						Min:          extrema,
+						Max:          extrema,
+						Sum:          compressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.rcvd_total_compressed_message_size",
+			Description: "Compressed message bytes received per call attempt.",
+			Unit:        "By",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(clientSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: bucketCounts,
+						Min:          extrema,
+						Max:          extrema,
+						Sum:          compressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.call.duration",
+			Description: "Time taken by gRPC to complete an RPC from application's perspective.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(methodAttr, targetAttr, statusAttr),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.started",
+			Description: "Number of server calls started.",
+			Unit:        "call",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(methodAttr),
+						Value:      1,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "grpc.server.call.sent_total_compressed_message_size",
+			Unit:        "By",
+			Description: "Compressed message bytes sent per server call.",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(serverSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: bucketCounts,
+						Min:          extrema,
+						Max:          extrema,
+						Sum:          compressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.rcvd_total_compressed_message_size",
+			Unit:        "By",
+			Description: "Compressed message bytes received per server call.",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(serverSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: bucketCounts,
+						Min:          extrema,
+						Max:          extrema,
+						Sum:          compressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.duration",
+			Description: "End-to-end time taken to complete a call from server transport's perspective.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(serverSideEnd...),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+	}
+}
+
+// MetricData returns a metricsDataSlice for A66 metrics for client and server
+// with a unary RPC and streaming RPC with certain compression and message flow
+// sent. If csmAttributes is set to true, the corresponding CSM Metrics (not
+// client side call metrics, or started on client and server side).
+func MetricData(options MetricDataOptions) []metricdata.Metrics {
+	unaryMethodAttr := attribute.String("grpc.method", "grpc.testing.TestService/UnaryCall")
+	duplexMethodAttr := attribute.String("grpc.method", "grpc.testing.TestService/FullDuplexCall")
+	targetAttr := attribute.String("grpc.target", options.Target)
+	unaryStatusAttr := attribute.String("grpc.status", "OK")
+	streamingStatusAttr := attribute.String("grpc.status", "OK")
+	if options.UnaryCallFailed {
+		unaryStatusAttr = attribute.String("grpc.status", "UNKNOWN")
+	}
+	unaryMethodClientSideEnd := []attribute.KeyValue{
+		unaryMethodAttr,
+		targetAttr,
+		unaryStatusAttr,
+	}
+	streamingMethodClientSideEnd := []attribute.KeyValue{
+		duplexMethodAttr,
+		targetAttr,
+		streamingStatusAttr,
+	}
+	unaryMethodServerSideEnd := []attribute.KeyValue{
+		unaryMethodAttr,
+		unaryStatusAttr,
+	}
+	streamingMethodServerSideEnd := []attribute.KeyValue{
+		duplexMethodAttr,
+		streamingStatusAttr,
+	}
+
+	unaryMethodClientSideEnd = append(unaryMethodClientSideEnd, options.CSMLabels...)
+	streamingMethodClientSideEnd = append(streamingMethodClientSideEnd, options.CSMLabels...)
+	unaryMethodServerSideEnd = append(unaryMethodServerSideEnd, options.CSMLabels...)
+	streamingMethodServerSideEnd = append(streamingMethodServerSideEnd, options.CSMLabels...)
+	unaryCompressedBytesSentRecv := int64(options.UnaryCompressedMessageSize)
+	unaryBucketCounts := createBucketCounts([]float64{options.UnaryCompressedMessageSize}, DefaultSizeBounds)
+	unaryExtrema := metricdata.NewExtrema(int64(options.UnaryCompressedMessageSize))
+
+	streamingCompressedBytesSentRecv := int64(options.StreamingCompressedMessageSize)
+	streamingBucketCounts := createBucketCounts([]float64{options.StreamingCompressedMessageSize}, DefaultSizeBounds)
+	streamingExtrema := metricdata.NewExtrema(int64(options.StreamingCompressedMessageSize))
+
+	return []metricdata.Metrics{
+		{
+			Name:        "grpc.client.attempt.started",
+			Description: "Number of client call attempts started.",
+			Unit:        "attempt",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(unaryMethodAttr, targetAttr),
+						Value:      1,
+					},
+					{
+						Attributes: attribute.NewSet(duplexMethodAttr, targetAttr),
+						Value:      1,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.duration",
+			Description: "End-to-end time taken to complete a client call attempt.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(unaryMethodClientSideEnd...),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+					{
+						Attributes: attribute.NewSet(streamingMethodClientSideEnd...),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.sent_total_compressed_message_size",
+			Description: "Compressed message bytes sent per client call attempt.",
+			Unit:        "By",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(unaryMethodClientSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: unaryBucketCounts,
+						Min:          unaryExtrema,
+						Max:          unaryExtrema,
+						Sum:          unaryCompressedBytesSentRecv,
+					},
+					{
+						Attributes:   attribute.NewSet(streamingMethodClientSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: streamingBucketCounts,
+						Min:          streamingExtrema,
+						Max:          streamingExtrema,
+						Sum:          streamingCompressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.attempt.rcvd_total_compressed_message_size",
+			Description: "Compressed message bytes received per call attempt.",
+			Unit:        "By",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(unaryMethodClientSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: unaryBucketCounts,
+						Min:          unaryExtrema,
+						Max:          unaryExtrema,
+						Sum:          unaryCompressedBytesSentRecv,
+					},
+					{
+						Attributes:   attribute.NewSet(streamingMethodClientSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: streamingBucketCounts,
+						Min:          streamingExtrema,
+						Max:          streamingExtrema,
+						Sum:          streamingCompressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.client.call.duration",
+			Description: "Time taken by gRPC to complete an RPC from application's perspective.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(unaryMethodAttr, targetAttr, unaryStatusAttr),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+					{
+						Attributes: attribute.NewSet(duplexMethodAttr, targetAttr, streamingStatusAttr),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.started",
+			Description: "Number of server calls started.",
+			Unit:        "call",
+			Data: metricdata.Sum[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(unaryMethodAttr),
+						Value:      1,
+					},
+					{
+						Attributes: attribute.NewSet(duplexMethodAttr),
+						Value:      1,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+			},
+		},
+		{
+			Name:        "grpc.server.call.sent_total_compressed_message_size",
+			Unit:        "By",
+			Description: "Compressed message bytes sent per server call.",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(unaryMethodServerSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: unaryBucketCounts,
+						Min:          unaryExtrema,
+						Max:          unaryExtrema,
+						Sum:          unaryCompressedBytesSentRecv,
+					},
+					{
+						Attributes:   attribute.NewSet(streamingMethodServerSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: streamingBucketCounts,
+						Min:          streamingExtrema,
+						Max:          streamingExtrema,
+						Sum:          streamingCompressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.rcvd_total_compressed_message_size",
+			Unit:        "By",
+			Description: "Compressed message bytes received per server call.",
+			Data: metricdata.Histogram[int64]{
+				DataPoints: []metricdata.HistogramDataPoint[int64]{
+					{
+						Attributes:   attribute.NewSet(unaryMethodServerSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: unaryBucketCounts,
+						Min:          unaryExtrema,
+						Max:          unaryExtrema,
+						Sum:          unaryCompressedBytesSentRecv,
+					},
+					{
+						Attributes:   attribute.NewSet(streamingMethodServerSideEnd...),
+						Count:        1,
+						Bounds:       DefaultSizeBounds,
+						BucketCounts: streamingBucketCounts,
+						Min:          streamingExtrema,
+						Max:          streamingExtrema,
+						Sum:          streamingCompressedBytesSentRecv,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+		{
+			Name:        "grpc.server.call.duration",
+			Description: "End-to-end time taken to complete a call from server transport's perspective.",
+			Unit:        "s",
+			Data: metricdata.Histogram[float64]{
+				DataPoints: []metricdata.HistogramDataPoint[float64]{
+					{
+						Attributes: attribute.NewSet(unaryMethodServerSideEnd...),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+					{
+						Attributes: attribute.NewSet(streamingMethodServerSideEnd...),
+						Count:      1,
+						Bounds:     DefaultLatencyBounds,
+					},
+				},
+				Temporality: metricdata.CumulativeTemporality,
+			},
+		},
+	}
+}
+
+// CompareMetrics asserts wantMetrics are what we expect. It polls for eventual
+// server metrics (not emitted synchronously with client side rpc returning),
+// and for duration metrics makes sure the data point is within possible testing
+// time (five seconds from context timeout).
+func CompareMetrics(ctx context.Context, t *testing.T, mr *metric.ManualReader, gotMetrics map[string]metricdata.Metrics, wantMetrics []metricdata.Metrics) {
+	for _, metric := range wantMetrics {
+		if metric.Name == "grpc.server.call.sent_total_compressed_message_size" || metric.Name == "grpc.server.call.rcvd_total_compressed_message_size" {
+			// Sync the metric reader to see the event because stats.End is
+			// handled async server side. Thus, poll until metrics created from
+			// stats.End show up.
+			var err error
+			if gotMetrics, err = waitForServerCompletedRPCs(ctx, t, mr, metric); err != nil { // move to shared helper
+				t.Fatal(err)
+			}
+			continue
+		}
+
+		// If one of the duration metrics, ignore the bucket counts, and make
+		// sure it count falls within a bucket <= 5 seconds (maximum duration of
+		// test due to context).
+		val, ok := gotMetrics[metric.Name]
+		if !ok {
+			t.Fatalf("Metric %v not present in recorded metrics", metric.Name)
+		}
+		if metric.Name == "grpc.client.attempt.duration" || metric.Name == "grpc.client.call.duration" || metric.Name == "grpc.server.call.duration" {
+			if !metricdatatest.AssertEqual(t, metric, val, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars(), metricdatatest.IgnoreValue()) {
+				t.Fatalf("Metrics data type not equal for metric: %v", metric.Name)
+			}
+			if err := checkDataPointWithinFiveSeconds(val); err != nil {
+				t.Fatalf("Data point not within five seconds for metric %v: %v", metric.Name, err)
+			}
+			continue
+		}
+
+		if !metricdatatest.AssertEqual(t, metric, val, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars()) {
+			t.Fatalf("Metrics data type not equal for metric: %v", metric.Name)
+		}
+	}
+}
diff --git a/xds/internal/balancer/cdsbalancer/aggregate_cluster_test.go b/xds/internal/balancer/cdsbalancer/aggregate_cluster_test.go
index c95f0ff6..a37b1609 100644
--- a/xds/internal/balancer/cdsbalancer/aggregate_cluster_test.go
+++ b/xds/internal/balancer/cdsbalancer/aggregate_cluster_test.go
@@ -33,6 +33,7 @@ import (
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/serviceconfig"
 	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/xds/internal"
 	"google.golang.org/grpc/xds/internal/balancer/clusterresolver"
 
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
@@ -85,6 +86,7 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 					Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 					EDSServiceName:   serviceName,
 					OutlierDetection: json.RawMessage(`{}`),
+					TelemetryLabels:  internal.UnknownCSMLabels,
 				}},
 				XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 			},
@@ -94,6 +96,7 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 					Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 					EDSServiceName:   serviceName + "-new",
 					OutlierDetection: json.RawMessage(`{}`),
+					TelemetryLabels:  internal.UnknownCSMLabels,
 				}},
 				XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 			},
@@ -108,6 +111,7 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 					Type:             clusterresolver.DiscoveryMechanismTypeLogicalDNS,
 					DNSHostname:      "dns_host:8080",
 					OutlierDetection: json.RawMessage(`{}`),
+					TelemetryLabels:  internal.UnknownCSMLabels,
 				}},
 				XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 			},
@@ -117,6 +121,7 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
 					Type:             clusterresolver.DiscoveryMechanismTypeLogicalDNS,
 					DNSHostname:      "dns_host_new:8080",
 					OutlierDetection: json.RawMessage(`{}`),
+					TelemetryLabels:  internal.UnknownCSMLabels,
 				}},
 				XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 			},
@@ -211,12 +216,14 @@ func (s) TestAggregateClusterSuccess_ThenUpdateChildClusters(t *testing.T) {
 				Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 				EDSServiceName:   serviceName,
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 			{
 				Cluster:          dnsClusterName,
 				Type:             clusterresolver.DiscoveryMechanismTypeLogicalDNS,
 				DNSHostname:      fmt.Sprintf("%s:%d", dnsHostName, dnsPort),
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 		},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
@@ -247,12 +254,14 @@ func (s) TestAggregateClusterSuccess_ThenUpdateChildClusters(t *testing.T) {
 				Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 				EDSServiceName:   serviceName,
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 			{
 				Cluster:          dnsClusterNameNew,
 				Type:             clusterresolver.DiscoveryMechanismTypeLogicalDNS,
 				DNSHostname:      fmt.Sprintf("%s:%d", dnsHostNameNew, dnsPort),
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 		},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
@@ -298,12 +307,14 @@ func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) {
 				Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 				EDSServiceName:   serviceName,
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 			{
 				Cluster:          dnsClusterName,
 				Type:             clusterresolver.DiscoveryMechanismTypeLogicalDNS,
 				DNSHostname:      fmt.Sprintf("%s:%d", dnsHostName, dnsPort),
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 		},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
@@ -329,6 +340,7 @@ func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) {
 			Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 			EDSServiceName:   serviceName,
 			OutlierDetection: json.RawMessage(`{}`),
+			TelemetryLabels:  internal.UnknownCSMLabels,
 		}},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 	}
@@ -363,6 +375,7 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T
 			Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 			EDSServiceName:   serviceName,
 			OutlierDetection: json.RawMessage(`{}`),
+			TelemetryLabels:  internal.UnknownCSMLabels,
 		}},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 	}
@@ -391,12 +404,14 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T
 				Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 				EDSServiceName:   serviceName,
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 			{
 				Cluster:          dnsClusterName,
 				Type:             clusterresolver.DiscoveryMechanismTypeLogicalDNS,
 				DNSHostname:      fmt.Sprintf("%s:%d", dnsHostName, dnsPort),
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 		},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
@@ -420,6 +435,7 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T
 			Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 			EDSServiceName:   serviceName,
 			OutlierDetection: json.RawMessage(`{}`),
+			TelemetryLabels:  internal.UnknownCSMLabels,
 		}},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 	}
@@ -572,6 +588,7 @@ func (s) TestAggregatedClusterSuccess_DiamondDependency(t *testing.T) {
 			Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 			EDSServiceName:   serviceName,
 			OutlierDetection: json.RawMessage(`{}`),
+			TelemetryLabels:  internal.UnknownCSMLabels,
 		}},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 	}
@@ -639,12 +656,14 @@ func (s) TestAggregatedClusterSuccess_IgnoreDups(t *testing.T) {
 				Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 				EDSServiceName:   serviceName,
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 			{
 				Cluster:          clusterNameD,
 				Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 				EDSServiceName:   serviceName,
 				OutlierDetection: json.RawMessage(`{}`),
+				TelemetryLabels:  internal.UnknownCSMLabels,
 			},
 		},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
@@ -727,6 +746,7 @@ func (s) TestAggregatedCluster_NodeChildOfItself(t *testing.T) {
 			Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 			EDSServiceName:   serviceName,
 			OutlierDetection: json.RawMessage(`{}`),
+			TelemetryLabels:  internal.UnknownCSMLabels,
 		}},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 	}
@@ -832,6 +852,7 @@ func (s) TestAggregatedCluster_CycleWithLeafNode(t *testing.T) {
 			Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 			EDSServiceName:   serviceName,
 			OutlierDetection: json.RawMessage(`{}`),
+			TelemetryLabels:  internal.UnknownCSMLabels,
 		}},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 	}
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
index 5a654638..82f4fb5f 100644
--- a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
+++ b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
@@ -42,6 +42,7 @@ import (
 	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/serviceconfig"
 	"google.golang.org/grpc/status"
+	xdsinternal "google.golang.org/grpc/xds/internal"
 	"google.golang.org/grpc/xds/internal/balancer/clusterresolver"
 	"google.golang.org/grpc/xds/internal/xdsclient"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
@@ -195,7 +196,7 @@ func registerWrappedCDSPolicy(t *testing.T) chan balancer.Balancer {
 //   - the nodeID expected by the management server
 //   - the grpc channel to the test backend service
 //   - the manual resolver configured on the channel
-//   - the xDS cient used the grpc channel
+//   - the xDS client used the grpc channel
 //   - a channel on which requested cluster resource names are sent
 //   - a channel used to signal that previously requested cluster resources are
 //     no longer requested
@@ -456,6 +457,7 @@ func (s) TestClusterUpdate_Success(t *testing.T) {
 					EDSServiceName:        serviceName,
 					MaxConcurrentRequests: newUint32(512),
 					OutlierDetection:      json.RawMessage(`{}`),
+					TelemetryLabels:       xdsinternal.UnknownCSMLabels,
 				}},
 				XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 			},
@@ -483,6 +485,7 @@ func (s) TestClusterUpdate_Success(t *testing.T) {
 					Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 					EDSServiceName:   serviceName,
 					OutlierDetection: json.RawMessage(`{}`),
+					TelemetryLabels:  xdsinternal.UnknownCSMLabels,
 				}},
 				XDSLBPolicy: json.RawMessage(`[{"ring_hash_experimental": {"minRingSize":100, "maxRingSize":1000}}]`),
 			},
@@ -505,6 +508,7 @@ func (s) TestClusterUpdate_Success(t *testing.T) {
 					Type:             clusterresolver.DiscoveryMechanismTypeEDS,
 					EDSServiceName:   serviceName,
 					OutlierDetection: json.RawMessage(`{"successRateEjection":{}}`),
+					TelemetryLabels:  xdsinternal.UnknownCSMLabels,
 				}},
 				XDSLBPolicy: json.RawMessage(`[{"ring_hash_experimental": {"minRingSize":1024, "maxRingSize":8388608}}]`),
 			},
@@ -557,6 +561,7 @@ func (s) TestClusterUpdate_Success(t *testing.T) {
 							"requestVolume": 50
 						}
 					}`),
+					TelemetryLabels: xdsinternal.UnknownCSMLabels,
 				}},
 				XDSLBPolicy: json.RawMessage(`[{"ring_hash_experimental": {"minRingSize":1024, "maxRingSize":8388608}}]`),
 			},
@@ -609,6 +614,7 @@ func (s) TestClusterUpdate_SuccessWithLRS(t *testing.T) {
 			EDSServiceName:      serviceName,
 			LoadReportingServer: lrsServerCfg,
 			OutlierDetection:    json.RawMessage(`{}`),
+			TelemetryLabels:     xdsinternal.UnknownCSMLabels,
 		}},
 		XDSLBPolicy: json.RawMessage(`[{"xds_wrr_locality_experimental": {"childPolicy": [{"round_robin": {}}]}}]`),
 	}
diff --git a/xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go b/xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go
index 1fb8ea73..88659318 100644
--- a/xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go
+++ b/xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go
@@ -79,26 +79,18 @@ func makeLogicalDNSClusterResource(name, dnsHost string, dnsPort uint32) *v3clus
 // Returns the following:
 //   - a channel onto which the DNS target being resolved is written to by the
 //     mock DNS resolver
-//   - a channel to notify close of the DNS resolver
-//   - a channel to notify re-resolution requests to the DNS resolver
 //   - a manual resolver which is used to mock the actual DNS resolution
-//   - a cleanup function which re-registers the original DNS resolver
-func setupDNS() (chan resolver.Target, chan struct{}, chan resolver.ResolveNowOptions, *manual.Resolver, func()) {
+func setupDNS(t *testing.T) (chan resolver.Target, *manual.Resolver) {
 	targetCh := make(chan resolver.Target, 1)
-	closeCh := make(chan struct{}, 1)
-	resolveNowCh := make(chan resolver.ResolveNowOptions, 1)
 
 	mr := manual.NewBuilderWithScheme("dns")
-	mr.BuildCallback = func(target resolver.Target, _ resolver.ClientConn, _ resolver.BuildOptions) {
-		targetCh <- target
-	}
-	mr.CloseCallback = func() { closeCh <- struct{}{} }
-	mr.ResolveNowCallback = func(opts resolver.ResolveNowOptions) { resolveNowCh <- opts }
+	mr.BuildCallback = func(target resolver.Target, _ resolver.ClientConn, _ resolver.BuildOptions) { targetCh <- target }
 
 	dnsResolverBuilder := resolver.Get("dns")
 	resolver.Register(mr)
 
-	return targetCh, closeCh, resolveNowCh, mr, func() { resolver.Register(dnsResolverBuilder) }
+	t.Cleanup(func() { resolver.Register(dnsResolverBuilder) })
+	return targetCh, mr
 }
 
 // TestAggregateCluster_WithTwoEDSClusters tests the case where the top-level
@@ -471,8 +463,7 @@ func (s) TestAggregateCluster_WithOneDNSCluster_HostnameChange(t *testing.T) {
 // cluster. The test verifies that RPCs fail until both clusters are resolved to
 // endpoints, and RPCs are routed to the higher priority EDS cluster.
 func (s) TestAggregateCluster_WithEDSAndDNS(t *testing.T) {
-	dnsTargetCh, _, _, dnsR, cleanup1 := setupDNS()
-	defer cleanup1()
+	dnsTargetCh, dnsR := setupDNS(t)
 
 	// Start an xDS management server that pushes the name of the requested EDS
 	// resource onto a channel.
@@ -661,8 +652,7 @@ func (s) TestAggregateCluster_SwitchEDSAndDNS(t *testing.T) {
 // still successful. This is the expected behavior because the cluster resolver
 // policy eats errors from DNS Resolver after it has returned an error.
 func (s) TestAggregateCluster_BadEDS_GoodToBadDNS(t *testing.T) {
-	dnsTargetCh, _, _, dnsR, cleanup1 := setupDNS()
-	defer cleanup1()
+	dnsTargetCh, dnsR := setupDNS(t)
 
 	// Start an xDS management server.
 	managementServer, nodeID, bootstrapContents, _, cleanup2 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
diff --git a/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go b/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
index e04dbe1c..7ac6306c 100644
--- a/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
+++ b/xds/internal/balancer/clusterresolver/e2e_test/balancer_test.go
@@ -41,6 +41,7 @@ import (
 	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/serviceconfig"
 	"google.golang.org/grpc/status"
+	xdsinternal "google.golang.org/grpc/xds/internal"
 	"google.golang.org/grpc/xds/internal/balancer/clusterimpl"
 	"google.golang.org/grpc/xds/internal/balancer/outlierdetection"
 	"google.golang.org/grpc/xds/internal/balancer/priority"
@@ -400,8 +401,9 @@ func (s) TestOutlierDetectionConfigPropagationToChildPolicy(t *testing.T) {
 						ChildPolicy: &iserviceconfig.BalancerConfig{
 							Name: clusterimpl.Name,
 							Config: &clusterimpl.LBConfig{
-								Cluster:        clusterName,
-								EDSServiceName: edsServiceName,
+								Cluster:         clusterName,
+								EDSServiceName:  edsServiceName,
+								TelemetryLabels: xdsinternal.UnknownCSMLabels,
 								ChildPolicy: &iserviceconfig.BalancerConfig{
 									Name: wrrlocality.Name,
 									Config: &wrrlocality.LBConfig{
diff --git a/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go b/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
index 5eb8ffd1..2ca3b5ad 100644
--- a/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
+++ b/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
@@ -20,21 +20,40 @@ package ringhash_test
 
 import (
 	"context"
+	"fmt"
+	"math"
+	"net"
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/grpctest"
+	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/grpc/internal/testutils/xds/e2e"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 
+	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	v3ringhashpb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/ring_hash/v3"
+	v3matcherpb "github.com/envoyproxy/go-control-plane/envoy/type/matcher/v3"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 
-	_ "google.golang.org/grpc/xds/internal/balancer/ringhash" // Register the ring_hash_experimental LB policy.
+	_ "google.golang.org/grpc/xds"
 )
 
 type s struct {
@@ -46,22 +65,16 @@ func Test(t *testing.T) {
 }
 
 const (
-	defaultTestTimeout      = 10 * time.Second
-	defaultTestShortTimeout = 10 * time.Millisecond // For events expected to *not* happen.
-)
+	defaultTestTimeout = 10 * time.Second
 
-type testService struct {
-	testgrpc.TestServiceServer
-}
+	errorTolerance = .05 // For tests that rely on statistical significance.
 
-func (*testService) EmptyCall(context.Context, *testpb.Empty) (*testpb.Empty, error) {
-	return &testpb.Empty{}, nil
-}
+	virtualHostName = "test.server"
+)
 
-// TestRingHash_ReconnectToMoveOutOfTransientFailure 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.
+// 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.
 func (s) TestRingHash_ReconnectToMoveOutOfTransientFailure(t *testing.T) {
 	// Create a restartable listener to simulate server being down.
 	l, err := testutils.LocalTCPListener()
@@ -69,16 +82,11 @@ func (s) TestRingHash_ReconnectToMoveOutOfTransientFailure(t *testing.T) {
 		t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
 	}
 	lis := testutils.NewRestartableListener(l)
-
-	// Start a server backend exposing the test service.
-	server := grpc.NewServer()
-	defer server.Stop()
-	testgrpc.RegisterTestServiceServer(server, &testService{})
-	go func() {
-		if err := server.Serve(lis); err != nil {
-			t.Errorf("Serve() failed: %v", err)
-		}
-	}()
+	srv := stubserver.StartTestService(t, &stubserver.StubServer{
+		Listener:   lis,
+		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) { return &testpb.Empty{}, nil },
+	})
+	defer srv.Stop()
 
 	// Create a clientConn with a manual resolver (which is used to push the
 	// address of the test backend), and a default service config pointing to
@@ -90,14 +98,14 @@ func (s) TestRingHash_ReconnectToMoveOutOfTransientFailure(t *testing.T) {
 		grpc.WithResolvers(r),
 		grpc.WithDefaultServiceConfig(ringHashServiceConfig),
 	}
-	cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
+	cc, err := grpc.NewClient(r.Scheme()+":///test.server", dopts...)
 	if err != nil {
-		t.Fatalf("failed to dial local test server: %v", err)
+		t.Fatalf("Failed to dial local test server: %v", err)
 	}
 	defer cc.Close()
 
 	// Push the address of the test backend through the manual resolver.
-	r.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: lis.Addr().String()}}})
+	r.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: lis.Addr().String()}}})
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -128,17 +136,811 @@ func (s) TestRingHash_ReconnectToMoveOutOfTransientFailure(t *testing.T) {
 	// attempt to reconnect on its own and come out of TRANSIENT_FAILURE, even
 	// without an RPC attempt.
 	lis.Restart()
-	for ; ctx.Err() == nil; <-time.After(defaultTestShortTimeout) {
-		if cc.GetState() == connectivity.Ready {
+	testutils.AwaitState(ctx, t, cc, connectivity.Ready)
+
+	// An RPC at this point is expected to succeed.
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+		t.Fatalf("rpc EmptyCall() failed: %v", err)
+	}
+}
+
+// startTestServiceBackends starts num stub servers.
+func startTestServiceBackends(t *testing.T, num int) ([]*stubserver.StubServer, func()) {
+	t.Helper()
+
+	var servers []*stubserver.StubServer
+	for i := 0; i < num; i++ {
+		servers = append(servers, stubserver.StartTestService(t, nil))
+	}
+
+	return servers, func() {
+		for _, server := range servers {
+			server.Stop()
+		}
+	}
+}
+
+// backendOptions returns a slice of e2e.BackendOptions for the given stub
+// servers.
+func backendOptions(t *testing.T, servers []*stubserver.StubServer) []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,
+		})
+	}
+	return backendOpts
+}
+
+// channelIDHashRoute returns a RouteConfiguration with a hash policy that
+// hashes based on the channel ID.
+func channelIDHashRoute(routeName, virtualHostDomain, clusterName string) *v3routepb.RouteConfiguration {
+	route := e2e.DefaultRouteConfig(routeName, virtualHostDomain, clusterName)
+	hashPolicy := 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{&hashPolicy}
+	return route
+}
+
+// checkRPCSendOK sends num RPCs to the client. It returns a map of backend
+// addresses as keys and number of RPCs sent to this address as value. Abort the
+// test if any RPC fails.
+func checkRPCSendOK(t *testing.T, ctx context.Context, client testpb.TestServiceClient, num int) map[string]int {
+	t.Helper()
+
+	backendCount := make(map[string]int)
+	for i := 0; i < num; i++ {
+		var remote peer.Peer
+		if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.Peer(&remote)); err != nil {
+			t.Fatalf("rpc EmptyCall() failed: %v", err)
+		}
+		backendCount[remote.Addr.String()]++
+	}
+	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 {
+	closedListeners := make([]net.Listener, 0, num)
+	for i := 0; i < num; i++ {
+		lis, err := testutils.LocalTCPListener()
+		if err != nil {
+			t.Fatalf("testutils.LocalTCPListener() failed: %v", err)
+		}
+		closedListeners = append(closedListeners, lis)
+	}
+
+	// Stop the servers that we want to be unreachable and collect their
+	// addresses.
+	backendOptions := make([]e2e.BackendOptions, 0, num)
+	for _, lis := range closedListeners {
+		backendOptions = append(backendOptions, e2e.BackendOptions{
+			Port:   testutils.ParsePort(t, lis.Addr().String()),
+			Weight: 1,
+		})
+		lis.Close()
+	}
+	return backendOptions
+}
+
+// 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, stop := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer stop()
+
+	servers, stop := startTestServiceBackends(t, 2)
+	defer stop()
+
+	const primaryClusterName = "new_cluster_1"
+	const primaryServiceName = "new_eds_service_1"
+	const secondaryClusterName = "new_cluster_2"
+	const secondaryServiceName = "new_eds_service_2"
+	const clusterName = "aggregate_cluster"
+
+	ep1 := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: primaryServiceName,
+		Localities: []e2e.LocalityOptions{{
+			Name:     "locality0",
+			Weight:   1,
+			Backends: makeNonExistentBackends(t, 2),
+		}},
+	})
+	ep2 := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: secondaryServiceName,
+		Localities: []e2e.LocalityOptions{{
+			Name:     "locality0",
+			Weight:   1,
+			Backends: backendOptions(t, servers),
+		}},
+	})
+	primaryCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: primaryClusterName,
+		ServiceName: primaryServiceName,
+	})
+	secundaryCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: secondaryClusterName,
+		ServiceName: secondaryServiceName,
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		Type:        e2e.ClusterTypeAggregate,
+		// TODO: when "A75: xDS Aggregate Cluster Behavior Fixes" is implemented, the
+		// policy will have to be set on the child clusters.
+		Policy:     e2e.LoadBalancingPolicyRingHash,
+		ChildNames: []string{primaryClusterName, secondaryClusterName},
+	})
+	route := channelIDHashRoute("new_route", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	err := xdsServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{ep1, ep2},
+		Clusters:  []*v3clusterpb.Cluster{cluster, primaryCluster, secundaryCluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	})
+	if 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)
+
+	gotPerBackend := checkRPCSendOK(t, ctx, client, 100)
+
+	// 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.
+	if len(gotPerBackend) != 1 {
+		t.Errorf("Got RPCs routed to %v backends, want %v", len(gotPerBackend), 1)
+	}
+
+	var backend string
+	var got int
+	for backend, got = range gotPerBackend {
+	}
+	found := false
+	for _, server := range servers {
+		if backend == server.Address {
+			found = true
 			break
 		}
 	}
-	if err := ctx.Err(); err != nil {
-		t.Fatalf("Timeout waiting for channel to reach READT after server restart: %v", err)
+	if !found {
+		t.Errorf("Got RPCs routed to an unexpected backend: %v, want one of %v", backend, servers)
+	}
+	if got != 100 {
+		t.Errorf("Got %v RPCs routed to a backend, want %v", got, 100)
 	}
+}
 
-	// An RPC at this point is expected to fail.
-	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
-		t.Fatalf("rpc EmptyCall() failed: %v", err)
+func replaceDNSResolver(t *testing.T) *manual.Resolver {
+	mr := manual.NewBuilderWithScheme("dns")
+
+	dnsResolverBuilder := resolver.Get("dns")
+	resolver.Register(mr)
+
+	t.Cleanup(func() { resolver.Register(dnsResolverBuilder) })
+	return mr
+}
+
+// Tests that when an aggregate cluster is configured with ring hash policy, and
+// the first is an EDS cluster in transient failure, and the fallback is a
+// 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, stop := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer stop()
+
+	const edsClusterName = "eds_cluster"
+	const logicalDNSClusterName = "logical_dns_cluster"
+	const clusterName = "aggregate_cluster"
+
+	backends, stop := startTestServiceBackends(t, 1)
+	defer stop()
+
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: edsClusterName,
+		Localities: []e2e.LocalityOptions{{
+			Name:     "locality0",
+			Weight:   1,
+			Backends: makeNonExistentBackends(t, 1),
+			Priority: 0,
+		}},
+	})
+	edsCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: edsClusterName,
+		ServiceName: edsClusterName,
+	})
+
+	logicalDNSCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		Type:        e2e.ClusterTypeLogicalDNS,
+		ClusterName: logicalDNSClusterName,
+		// The DNS values are not used because we fake DNS later on, but they
+		// are required to be present for the resource to be valid.
+		DNSHostName: "server.example.com",
+		DNSPort:     443,
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		Type:        e2e.ClusterTypeAggregate,
+		// TODO: when "A75: xDS Aggregate Cluster Behavior Fixes" is merged, the
+		// policy will have to be set on the child clusters.
+		Policy:     e2e.LoadBalancingPolicyRingHash,
+		ChildNames: []string{edsClusterName, logicalDNSClusterName},
+	})
+	route := channelIDHashRoute("new_route", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	err := xdsServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
+		Clusters:  []*v3clusterpb.Cluster{cluster, edsCluster, logicalDNSCluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	})
+	if 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}}})
+
+	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)
+
+	gotPerBackend := checkRPCSendOK(t, ctx, client, 1)
+	var got string
+	for got = range gotPerBackend {
+	}
+	if want := backends[0].Address; got != want {
+		t.Errorf("Got RPCs routed to an unexpected got: %v, want %v", got, want)
+	}
+}
+
+// Tests that when an aggregate cluster is configured with ring hash policy, and
+// 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, stop := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer stop()
+
+	const edsClusterName = "eds_cluster"
+	const logicalDNSClusterName = "logical_dns_cluster"
+	const clusterName = "aggregate_cluster"
+
+	backends, stop := startTestServiceBackends(t, 1)
+	defer stop()
+
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: edsClusterName,
+		Localities: []e2e.LocalityOptions{{
+			Name:     "locality0",
+			Weight:   1,
+			Backends: makeNonExistentBackends(t, 1),
+			Priority: 0,
+		}},
+	})
+	edsCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: edsClusterName,
+		ServiceName: edsClusterName,
+	})
+
+	logicalDNSCluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		Type:        e2e.ClusterTypeLogicalDNS,
+		ClusterName: logicalDNSClusterName,
+		// The DNS values are not used because we fake DNS later on, but they
+		// are required to be present for the resource to be valid.
+		DNSHostName: "server.example.com",
+		DNSPort:     443,
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		Type:        e2e.ClusterTypeAggregate,
+		// TODO: when "A75: xDS Aggregate Cluster Behavior Fixes" is merged, the
+		// policy will have to be set on the child clusters.
+		Policy:     e2e.LoadBalancingPolicyRingHash,
+		ChildNames: []string{edsClusterName, logicalDNSClusterName},
+	})
+	route := channelIDHashRoute("new_route", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	err := xdsServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
+		Clusters:  []*v3clusterpb.Cluster{cluster, edsCluster, logicalDNSCluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	})
+	if 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}}})
+
+	dialer := testutils.NewBlockingDialer()
+	cp := grpc.ConnectParams{
+		// Increase backoff time, so that subconns stay in TRANSIENT_FAILURE
+		// for long enough to trigger potential problems.
+		Backoff: backoff.Config{
+			BaseDelay: defaultTestTimeout,
+		},
+		MinConnectTimeout: 0,
+	}
+	opts := []grpc.DialOption{
+		grpc.WithResolvers(xdsResolver),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithContextDialer(dialer.DialContext),
+		grpc.WithConnectParams(cp)}
+	conn, err := grpc.NewClient("xds:///test.server", opts...)
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	errCh := make(chan error, 2)
+	go func() {
+		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+			errCh <- fmt.Errorf("first rpc UnaryCall() failed: %v", err)
+			return
+		}
+		errCh <- nil
+	}()
+
+	testutils.AwaitState(ctx, t, conn, connectivity.Connecting)
+
+	go func() {
+		// Start a second RPC at this point, which should be queued as well.
+		// This will fail if the priority policy fails to update the picker to
+		// point to the LOGICAL_DNS child; if it leaves it pointing to the EDS
+		// priority 1, then the RPC will fail, because all subchannels are in
+		// transient failure.
+		//
+		// Note that sending only the first RPC does not catch this case,
+		// because if the priority policy fails to update the picker, then the
+		// pick for the first RPC will not be retried.
+		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+			errCh <- fmt.Errorf("second UnaryCall() failed: %v", err)
+			return
+		}
+		errCh <- nil
+	}()
+
+	// Allow the connection attempts to complete.
+	dialer.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)
+			}
+		case <-ctx.Done():
+			t.Fatalf("Timed out waiting for RPCs to complete")
+		}
+	}
+}
+
+// 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()
+
+	xdsServer, nodeID, _, xdsResolver, stop := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer stop()
+	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,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := channelIDHashRoute("new_route", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	err := xdsServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
+		Clusters:  []*v3clusterpb.Cluster{cluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	})
+	if 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)
+
+	received := checkRPCSendOK(t, ctx, client, 100)
+	if len(received) != 1 {
+		t.Errorf("Got RPCs routed to %v backends, want %v", len(received), 1)
+	}
+	var count int
+	for _, count = range received {
+	}
+	if count != 100 {
+		t.Errorf("Got %v RPCs routed to a backend, want %v", count, 100)
+	}
+}
+
+// headerHashRoute creates a RouteConfiguration with a hash policy that uses the
+// provided header.
+func headerHashRoute(routeName, virtualHostName, clusterName, header string) *v3routepb.RouteConfiguration {
+	route := e2e.DefaultRouteConfig(routeName, virtualHostName, clusterName)
+	hashPolicy := v3routepb.RouteAction_HashPolicy{
+		PolicySpecifier: &v3routepb.RouteAction_HashPolicy_Header_{
+			Header: &v3routepb.RouteAction_HashPolicy_Header{
+				HeaderName: header,
+			},
+		},
+	}
+	action := route.VirtualHosts[0].Routes[0].Action.(*v3routepb.Route_Route)
+	action.Route.HashPolicy = []*v3routepb.RouteAction_HashPolicy{&hashPolicy}
+	return route
+}
+
+// Tests that ring hash policy that hashes using a header value can spread RPCs
+// across all the backends.
+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)
+	}
+
+	xdsServer, nodeID, _, xdsResolver, stop := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer stop()
+	const clusterName = "cluster"
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Host:        host,
+		Localities: []e2e.LocalityOptions{{
+			Backends: backendOptions(t, backends),
+			Weight:   1,
+		}},
+	})
+	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()
+
+	err = xdsServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
+		Clusters:  []*v3clusterpb.Cluster{cluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	})
+	if 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
+	// 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(t, ctx, client, 1)
+		if reqPerBackend[backend.Address] != 1 {
+			t.Errorf("Got RPC routed to backend %v, want %v", reqPerBackend, backend.Address)
+		}
+	}
+}
+
+// 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, stop := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer stop()
+	clusterName := "cluster"
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Host:        host,
+		Localities: []e2e.LocalityOptions{{
+			Backends: backendOptions(t, backends),
+			Weight:   1,
+		}},
+	})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := headerHashRoute("new_route", virtualHostName, clusterName, "address_hash")
+	action := route.VirtualHosts[0].Routes[0].Action.(*v3routepb.Route_Route)
+	action.Route.HashPolicy[0].GetHeader().RegexRewrite = &v3matcherpb.RegexMatchAndSubstitute{
+		Pattern: &v3matcherpb.RegexMatcher{
+			EngineType: &v3matcherpb.RegexMatcher_GoogleRe2{},
+			Regex:      "[0-9]+",
+		},
+		Substitution: "foo",
+	}
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	err = xdsServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
+		Clusters:  []*v3clusterpb.Cluster{cluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	})
+	if 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 would always be hashed
+	// to a specific backend as the header value matches the value used to
+	// create the entry in the ring. However, the regex rewrites all numbers to
+	// "foo", and header values only differ by numbers, so they all end up
+	// 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"))
+		res := checkRPCSendOK(t, ctx, client, 100)
+		for addr, count := range res {
+			gotPerBackend[addr] += count
+		}
+	}
+	if want := 1; len(gotPerBackend) != want {
+		t.Errorf("Got RPCs routed to %v backends, want %v", len(gotPerBackend), want)
+	}
+	var got int
+	for _, got = range gotPerBackend {
+	}
+	if want := 400; got != want {
+		t.Errorf("Got %v RPCs routed to a backend, want %v", got, want)
+	}
+}
+
+// computeIdealNumberOfRPCs computes the ideal number of RPCs to send so that
+// we can observe an event happening with probability p, and the result will
+// have value p with the given error tolerance.
+//
+// See https://github.com/grpc/grpc/blob/4f6e13bdda9e8c26d6027af97db4b368ca2b3069/test/cpp/end2end/xds/xds_end2end_test_lib.h#L941
+// for an explanation of the formula.
+func computeIdealNumberOfRPCs(t *testing.T, p, errorTolerance float64) int {
+	if p < 0 || p > 1 {
+		t.Fatal("p must be in (0, 1)")
+	}
+	numRPCs := math.Ceil(p * (1 - p) * 5. * 5. / errorTolerance / errorTolerance)
+	return int(numRPCs + 1000.) // add 1k as a buffer to avoid flakyness.
+}
+
+// setRingHashLBPolicyWithHighMinRingSize sets the ring hash policy with a high
+// 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)
+	oldVal := envconfig.RingHashCap
+	envconfig.RingHashCap = minRingSize
+	t.Cleanup(func() {
+		envconfig.RingHashCap = oldVal
+	})
+	// Increasing min ring size for random distribution.
+	config := testutils.MarshalAny(t, &v3ringhashpb.RingHash{
+		HashFunction:    v3ringhashpb.RingHash_XX_HASH,
+		MinimumRingSize: &wrapperspb.UInt64Value{Value: minRingSize},
+	})
+	cluster.LoadBalancingPolicy = &v3clusterpb.LoadBalancingPolicy{
+		Policies: []*v3clusterpb.LoadBalancingPolicy_Policy{{
+			TypedExtensionConfig: &v3corepb.TypedExtensionConfig{
+				Name:        "envoy.load_balancing_policies.ring_hash",
+				TypedConfig: config,
+			},
+		}},
+	}
+}
+
+// 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()
+	numRPCs := computeIdealNumberOfRPCs(t, .5, errorTolerance)
+
+	xdsServer, nodeID, _, xdsResolver, stop := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer stop()
+	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", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	err := xdsServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
+		Clusters:  []*v3clusterpb.Cluster{cluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	})
+	if 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)
+
+	// Send a large number of RPCs and check that they are distributed randomly.
+	gotPerBackend := checkRPCSendOK(t, ctx, client, numRPCs)
+	for _, backend := range backends {
+		got := float64(gotPerBackend[backend.Address]) / 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)
+		}
+	}
+}
+
+// Tests that we observe endpoint weights.
+func (s) TestRingHash_EndpointWeights(t *testing.T) {
+	backends, stop := startTestServiceBackends(t, 3)
+	defer stop()
+	xdsServer, nodeID, _, xdsResolver, stop := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer stop()
+	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},
+	}
+
+	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
+		ClusterName: clusterName,
+		Localities: []e2e.LocalityOptions{{
+			Backends: backendOpts,
+			Weight:   1,
+		}},
+	})
+	endpoints.Endpoints[0].LbEndpoints[0].LoadBalancingWeight = wrapperspb.UInt32(uint32(1))
+	endpoints.Endpoints[0].LbEndpoints[1].LoadBalancingWeight = wrapperspb.UInt32(uint32(1))
+	endpoints.Endpoints[0].LbEndpoints[2].LoadBalancingWeight = wrapperspb.UInt32(uint32(2))
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+	})
+	// Increasing min ring size for random distribution.
+	setRingHashLBPolicyWithHighMinRingSize(t, cluster)
+	route := e2e.DefaultRouteConfig("new_route", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	err := xdsServer.Update(ctx, e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{endpoints},
+		Clusters:  []*v3clusterpb.Cluster{cluster},
+		Routes:    []*v3routepb.RouteConfiguration{route},
+		Listeners: []*v3listenerpb.Listener{listener},
+	})
+	if 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)
+
+	// Send a large number of RPCs and check that they are distributed randomly.
+	numRPCs := computeIdealNumberOfRPCs(t, .25, errorTolerance)
+	gotPerBackend := checkRPCSendOK(t, ctx, client, numRPCs)
+
+	got := float64(gotPerBackend[backends[0].Address]) / 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)
+	}
+	got = float64(gotPerBackend[backends[1].Address]) / 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)
+	}
+	got = float64(gotPerBackend[backends[2].Address]) / 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)
 	}
 }
diff --git a/xds/internal/xdsclient/tests/cds_watchers_test.go b/xds/internal/xdsclient/tests/cds_watchers_test.go
index ace15d8d..c7780889 100644
--- a/xds/internal/xdsclient/tests/cds_watchers_test.go
+++ b/xds/internal/xdsclient/tests/cds_watchers_test.go
@@ -103,7 +103,7 @@ func verifyClusterUpdate(ctx context.Context, updateCh *testutils.Channel, wantU
 			return fmt.Errorf("received update with error type %v, want %v", gotType, wantType)
 		}
 	}
-	cmpOpts := []cmp.Option{cmpopts.EquateEmpty(), cmpopts.IgnoreFields(xdsresource.ClusterUpdate{}, "Raw", "LBPolicy")}
+	cmpOpts := []cmp.Option{cmpopts.EquateEmpty(), cmpopts.IgnoreFields(xdsresource.ClusterUpdate{}, "Raw", "LBPolicy", "TelemetryLabels")}
 	if diff := cmp.Diff(wantUpdate.update, got.update, cmpOpts...); diff != "" {
 		return fmt.Errorf("received unexpected diff in the cluster resource update: (-want, got):\n%s", diff)
 	}
diff --git a/xds/internal/xdsclient/tests/resource_update_test.go b/xds/internal/xdsclient/tests/resource_update_test.go
index 7451e67a..5ac0ab42 100644
--- a/xds/internal/xdsclient/tests/resource_update_test.go
+++ b/xds/internal/xdsclient/tests/resource_update_test.go
@@ -875,7 +875,7 @@ func (s) TestHandleClusterResponseFromManagementServer(t *testing.T) {
 			}
 			cmpOpts := []cmp.Option{
 				cmpopts.EquateEmpty(),
-				cmpopts.IgnoreFields(xdsresource.ClusterUpdate{}, "Raw", "LBPolicy"),
+				cmpopts.IgnoreFields(xdsresource.ClusterUpdate{}, "Raw", "LBPolicy", "TelemetryLabels"),
 			}
 			if diff := cmp.Diff(test.wantUpdate, gotUpdate, cmpOpts...); diff != "" {
 				t.Fatalf("Unexpected diff in metadata, diff (-want +got):\n%s", diff)
diff --git a/xds/internal/xdsclient/xdsresource/tests/unmarshal_cds_test.go b/xds/internal/xdsclient/xdsresource/tests/unmarshal_cds_test.go
index dde89b27..75ca714c 100644
--- a/xds/internal/xdsclient/xdsresource/tests/unmarshal_cds_test.go
+++ b/xds/internal/xdsclient/xdsresource/tests/unmarshal_cds_test.go
@@ -34,6 +34,7 @@ import (
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/internal/xds/bootstrap"
 	"google.golang.org/grpc/serviceconfig"
+	"google.golang.org/grpc/xds/internal"
 	"google.golang.org/grpc/xds/internal/balancer/ringhash"
 	"google.golang.org/grpc/xds/internal/balancer/wrrlocality"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
@@ -141,9 +142,10 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				},
 			},
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName: clusterName,
-				ClusterType: xdsresource.ClusterTypeLogicalDNS,
-				DNSHostName: "dns_host:8080",
+				ClusterName:     clusterName,
+				ClusterType:     xdsresource.ClusterTypeLogicalDNS,
+				DNSHostName:     "dns_host:8080",
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: wrrlocality.Name,
@@ -172,6 +174,7 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				ClusterName:             clusterName,
 				ClusterType:             xdsresource.ClusterTypeAggregate,
 				PrioritizedClusterNames: []string{"a", "b", "c"},
+				TelemetryLabels:         internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: wrrlocality.Name,
@@ -183,9 +186,12 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 			},
 		},
 		{
-			name:       "happy-case-no-service-name-no-lrs",
-			cluster:    e2e.DefaultCluster(clusterName, "", e2e.SecurityLevelNone),
-			wantUpdate: xdsresource.ClusterUpdate{ClusterName: clusterName},
+			name:    "happy-case-no-service-name-no-lrs",
+			cluster: e2e.DefaultCluster(clusterName, "", e2e.SecurityLevelNone),
+			wantUpdate: xdsresource.ClusterUpdate{
+				ClusterName:     clusterName,
+				TelemetryLabels: internal.UnknownCSMLabels,
+			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: wrrlocality.Name,
 				Config: &wrrlocality.LBConfig{
@@ -199,8 +205,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 			name:    "happy-case-no-lrs",
 			cluster: e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone),
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: wrrlocality.Name,
@@ -223,6 +230,7 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				ClusterName:     clusterName,
 				EDSServiceName:  serviceName,
 				LRSServerConfig: serverCfg,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: wrrlocality.Name,
@@ -261,6 +269,7 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				EDSServiceName:  serviceName,
 				LRSServerConfig: serverCfg,
 				MaxRequests:     func() *uint32 { i := uint32(512); return &i }(),
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: wrrlocality.Name,
@@ -279,8 +288,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				return c
 			}(),
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: "ring_hash_experimental",
@@ -306,8 +316,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				LbPolicy: v3clusterpb.Cluster_LEAST_REQUEST,
 			},
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: "least_request_experimental",
@@ -330,8 +341,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				return c
 			}(),
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: "ring_hash_experimental",
@@ -362,8 +374,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				},
 			},
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: "least_request_experimental",
@@ -400,8 +413,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				},
 			},
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: "ring_hash_experimental",
@@ -435,8 +449,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				},
 			},
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: wrrlocality.Name,
@@ -474,8 +489,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				},
 			},
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: wrrlocality.Name,
@@ -522,8 +538,9 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 				},
 			},
 			wantUpdate: xdsresource.ClusterUpdate{
-				ClusterName:    clusterName,
-				EDSServiceName: serviceName,
+				ClusterName:     clusterName,
+				EDSServiceName:  serviceName,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 			wantLBConfig: &iserviceconfig.BalancerConfig{
 				Name: "ring_hash_experimental",
diff --git a/xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go b/xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go
index eeea3d58..5f285867 100644
--- a/xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go
+++ b/xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go
@@ -29,6 +29,7 @@ import (
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/xds/bootstrap"
 	"google.golang.org/grpc/internal/xds/matcher"
+	"google.golang.org/grpc/xds/internal"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
 
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
@@ -873,6 +874,7 @@ func (s) TestValidateClusterWithSecurityConfig(t *testing.T) {
 					RootInstanceName: rootPluginInstance,
 					RootCertName:     rootCertName,
 				},
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 		{
@@ -914,6 +916,7 @@ func (s) TestValidateClusterWithSecurityConfig(t *testing.T) {
 					RootInstanceName: rootPluginInstance,
 					RootCertName:     rootCertName,
 				},
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 		{
@@ -959,6 +962,7 @@ func (s) TestValidateClusterWithSecurityConfig(t *testing.T) {
 					IdentityInstanceName: identityPluginInstance,
 					IdentityCertName:     identityCertName,
 				},
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 		{
@@ -1006,6 +1010,7 @@ func (s) TestValidateClusterWithSecurityConfig(t *testing.T) {
 					IdentityInstanceName: identityPluginInstance,
 					IdentityCertName:     identityCertName,
 				},
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 		{
@@ -1072,6 +1077,7 @@ func (s) TestValidateClusterWithSecurityConfig(t *testing.T) {
 						matcher.StringMatcherForTesting(nil, nil, nil, newStringP(sanContains), nil, false),
 					},
 				},
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 		{
@@ -1138,6 +1144,7 @@ func (s) TestValidateClusterWithSecurityConfig(t *testing.T) {
 						matcher.StringMatcherForTesting(nil, nil, nil, newStringP(sanContains), nil, false),
 					},
 				},
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 	}
@@ -1331,6 +1338,7 @@ func (s) TestUnmarshalCluster(t *testing.T) {
 				EDSServiceName:  v3Service,
 				LRSServerConfig: serverCfg,
 				Raw:             v3ClusterAny,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 		{
@@ -1343,6 +1351,7 @@ func (s) TestUnmarshalCluster(t *testing.T) {
 				EDSServiceName:  v3Service,
 				LRSServerConfig: serverCfg,
 				Raw:             v3ClusterAny,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 		{
@@ -1355,6 +1364,7 @@ func (s) TestUnmarshalCluster(t *testing.T) {
 				EDSServiceName:  v3Service,
 				LRSServerConfig: serverCfg,
 				Raw:             v3ClusterAnyWithEDSConfigSourceSelf,
+				TelemetryLabels: internal.UnknownCSMLabels,
 			},
 		},
 		{
@@ -1384,7 +1394,8 @@ func (s) TestUnmarshalCluster(t *testing.T) {
 				LRSServerConfig: serverCfg,
 				Raw:             v3ClusterAnyWithTelemetryLabelsIgnoreSome,
 				TelemetryLabels: map[string]string{
-					"csm.service_name": "grpc-service",
+					"csm.service_name":           "grpc-service",
+					"csm.service_namespace_name": "unknown",
 				},
 			},
 		},