rls, grpc: inherit parent channel stats handlers and interceptors on the RLS control channel

RouteLookup RPCs issued on the RLS balancer's control channel were
invisible to any stats handler configured on the parent ClientConn
(e.g. OpenTelemetry's grpc.client.attempt.duration histogram), because
the control channel was created via a bare grpc.NewClient call that had
no way to see the parent channel's stats handlers or client
interceptors.

Introduce an internal hook, NewChannelForBalancer, that creates a new
ClientConn while inheriting the parent's stats handlers and unary/stream
interceptors (single and chained). The parent is reached by walking the
balancer.ClientConn wrapping chain using the Unwrap()
balancer.ClientConn contract that every wrapper in the tree now
implements (added in a prior change). A seen-set guards against cycles.

RLS now dials via this hook instead of grpc.NewClient. Verified
end-to-end against a real Bigtable DirectPath workload: RouteLookup
attempts to dns:///bigtablerls.googleapis.com show up in
grpc.client.attempt.duration with method, target, and status labels
populated correctly. Regression test added in
balancer/rls/metrics_test.go.

RELEASE NOTES:
* rls: RouteLookup RPCs issued on the RLS balancer's control channel now inherit stats handlers and client interceptors configured on the parent ClientConn, so per-attempt telemetry (e.g. grpc.client.attempt.duration) covers RouteLookup RPCs.

Author: sushanb

balancer/rls/balancer.go

@@ -367,7 +367,7 @@ func (b *rlsBalancer) handleControlChannelUpdate(newCfg *lbConfig) {
 	backToReadyFn := func() {
 		b.updateCh.Put(controlChannelReady{})
 	}
-	ctrlCh, err := newControlChannel(newCfg.lookupService, newCfg.controlChannelServiceConfig, newCfg.lookupServiceTimeout, b.bopts, backToReadyFn)
+	ctrlCh, err := newControlChannel(b.cc, newCfg.lookupService, newCfg.controlChannelServiceConfig, newCfg.lookupServiceTimeout, b.bopts, backToReadyFn)
 	if err != nil {
 		// This is very uncommon and usually represents a non-transient error.
 		// There is not much we can do here other than wait for another update

balancer/rls/control_channel.go

@@ -72,7 +72,14 @@ type controlChannel struct {
 // newControlChannel creates a controlChannel to rlsServerName and uses
 // serviceConfig, if non-empty, as the default service config for the underlying
 // gRPC channel.
-func newControlChannel(rlsServerName, serviceConfig string, rpcTimeout time.Duration, bOpts balancer.BuildOptions, backToReadyFunc func()) (*controlChannel, error) {
+//
+// parentCC is the balancer.ClientConn passed to Balancer.Build. It is used to
+// inherit the parent channel's stats handlers and client interceptors so that
+// per-attempt telemetry configured on the parent channel also covers
+// RouteLookup RPCs. A nil parentCC is tolerated (control channel is created
+// without inheriting parent telemetry) for tests that construct the control
+// channel directly.
+func newControlChannel(parentCC balancer.ClientConn, rlsServerName, serviceConfig string, rpcTimeout time.Duration, bOpts balancer.BuildOptions, backToReadyFunc func()) (*controlChannel, error) {
 	ctrlCh := &controlChannel{
 		rpcTimeout:      rpcTimeout,
 		backToReadyFunc: backToReadyFunc,
@@ -84,7 +91,8 @@ func newControlChannel(rlsServerName, serviceConfig string, rpcTimeout time.Dura
 	if err != nil {
 		return nil, err
 	}
-	ctrlCh.cc, err = grpc.NewClient(rlsServerName, dopts...)
+	newChan := internal.NewChannelForBalancer.(func(balancer.ClientConn, string, ...grpc.DialOption) (*grpc.ClientConn, error))
+	ctrlCh.cc, err = newChan(parentCC, rlsServerName, dopts...)
 	if err != nil {
 		return nil, err
 	}

balancer/rls/control_channel_test.go

@@ -51,7 +51,7 @@ func (s) TestControlChannelThrottled(t *testing.T) {
 	overrideAdaptiveThrottler(t, alwaysThrottlingThrottler())
 
 	// Create a control channel to the fake RLS server.
-	ctrlCh, err := newControlChannel(rlsServer.Address, "", defaultTestTimeout, balancer.BuildOptions{}, nil)
+	ctrlCh, err := newControlChannel(nil, rlsServer.Address, "", defaultTestTimeout, balancer.BuildOptions{}, nil)
 	if err != nil {
 		t.Fatalf("Failed to create control channel to RLS server: %v", err)
 	}
@@ -79,7 +79,7 @@ func (s) TestLookupFailure(t *testing.T) {
 	})
 
 	// Create a control channel to the fake RLS server.
-	ctrlCh, err := newControlChannel(rlsServer.Address, "", defaultTestTimeout, balancer.BuildOptions{}, nil)
+	ctrlCh, err := newControlChannel(nil, rlsServer.Address, "", defaultTestTimeout, balancer.BuildOptions{}, nil)
 	if err != nil {
 		t.Fatalf("Failed to create control channel to RLS server: %v", err)
 	}
@@ -118,7 +118,7 @@ func (s) TestLookupDeadlineExceeded(t *testing.T) {
 	overrideAdaptiveThrottler(t, neverThrottlingThrottler())
 
 	// Create a control channel with a small deadline.
-	ctrlCh, err := newControlChannel(rlsServer.Address, "", defaultTestShortTimeout, balancer.BuildOptions{}, nil)
+	ctrlCh, err := newControlChannel(nil, rlsServer.Address, "", defaultTestShortTimeout, balancer.BuildOptions{}, nil)
 	if err != nil {
 		t.Fatalf("Failed to create control channel to RLS server: %v", err)
 	}
@@ -272,7 +272,7 @@ func testControlChannelCredsSuccess(t *testing.T, sopts []grpc.ServerOption, bop
 	})
 
 	// Create a control channel to the fake server.
-	ctrlCh, err := newControlChannel(rlsServer.Address, "", defaultTestTimeout, bopts, nil)
+	ctrlCh, err := newControlChannel(nil, rlsServer.Address, "", defaultTestTimeout, bopts, nil)
 	if err != nil {
 		t.Fatalf("Failed to create control channel to RLS server: %v", err)
 	}
@@ -360,7 +360,7 @@ func testControlChannelCredsFailure(t *testing.T, sopts []grpc.ServerOption, bop
 	overrideAdaptiveThrottler(t, neverThrottlingThrottler())
 
 	// Create the control channel to the fake server.
-	ctrlCh, err := newControlChannel(rlsServer.Address, "", defaultTestTimeout, bopts, nil)
+	ctrlCh, err := newControlChannel(nil, rlsServer.Address, "", defaultTestTimeout, bopts, nil)
 	if err != nil {
 		t.Fatalf("Failed to create control channel to RLS server: %v", err)
 	}
@@ -457,7 +457,7 @@ func (s) TestNewControlChannelUnsupportedCredsBundle(t *testing.T) {
 	rlsServer, _ := rlstest.SetupFakeRLSServer(t, nil)
 
 	// Create the control channel to the fake server.
-	ctrlCh, err := newControlChannel(rlsServer.Address, "", defaultTestTimeout, balancer.BuildOptions{CredsBundle: &unsupportedCredsBundle{}}, nil)
+	ctrlCh, err := newControlChannel(nil, rlsServer.Address, "", defaultTestTimeout, balancer.BuildOptions{CredsBundle: &unsupportedCredsBundle{}}, nil)
 	if err == nil {
 		ctrlCh.close()
 		t.Fatal("newControlChannel succeeded when expected to fail")

balancer/rls/metrics_test.go

@@ -373,3 +373,63 @@ func (s) TestRLSFailedRPCMetric(t *testing.T) {
 		}
 	}
 }
+
+// TestRLSControlChannelAttemptDurationMetric verifies that per-attempt
+// telemetry configured on the parent channel (grpc.client.attempt.duration)
+// also covers the RouteLookup RPC issued by the RLS balancer's control
+// channel. Regression test for the plumbing that inherits parent stats
+// handlers and interceptors onto balancer-owned control channels.
+func (s) TestRLSControlChannelAttemptDurationMetric(t *testing.T) {
+	rlsServer, _ := rlstest.SetupFakeRLSServer(t, nil)
+	rlsConfig := buildBasicRLSConfigWithChildPolicy(t, t.Name(), rlsServer.Address)
+	backend := &stubserver.StubServer{
+		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
+			return &testpb.Empty{}, nil
+		},
+	}
+	if err := backend.StartServer(); err != nil {
+		t.Fatalf("Failed to start backend: %v", err)
+	}
+	defer backend.Stop()
+	rlsConfig.RouteLookupConfig.DefaultTarget = backend.Address
+
+	r := startManualResolverWithConfig(t, rlsConfig)
+	reader := metric.NewManualReader()
+	provider := metric.NewMeterProvider(metric.WithReader(reader))
+	mo := opentelemetry.MetricsOptions{MeterProvider: provider}
+	cc, err := grpc.NewClient(r.Scheme()+":///", grpc.WithResolvers(r), grpc.WithTransportCredentials(insecure.NewCredentials()), opentelemetry.DialOption(opentelemetry.Options{MetricsOptions: mo}))
+	if err != nil {
+		t.Fatalf("Failed to dial local test server: %v", err)
+	}
+	defer cc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if _, err := testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}); err != nil {
+		t.Fatalf("client.EmptyCall failed: %v", err)
+	}
+
+	const rlsMethod = "grpc.lookup.v1.RouteLookupService/RouteLookup"
+	md, ok := metricsDataFromReader(ctx, reader)["grpc.client.attempt.duration"]
+	if !ok {
+		t.Fatal("grpc.client.attempt.duration metric not recorded")
+	}
+	hist, ok := md.Data.(metricdata.Histogram[float64])
+	if !ok {
+		t.Fatalf("grpc.client.attempt.duration has unexpected data type %T, want Histogram[float64]", md.Data)
+	}
+	for _, dp := range hist.DataPoints {
+		if v, ok := dp.Attributes.Value("grpc.method"); ok && v.AsString() == rlsMethod {
+			return
+		}
+	}
+	t.Fatalf("grpc.client.attempt.duration did not record a data point for method %q; got attribute sets %v", rlsMethod, attributeSets(hist.DataPoints))
+}
+
+func attributeSets(dps []metricdata.HistogramDataPoint[float64]) []attribute.Set {
+	out := make([]attribute.Set, 0, len(dps))
+	for _, dp := range dps {
+		out = append(out, dp.Attributes)
+	}
+	return out
+}

clientconn.go

@@ -790,6 +790,67 @@ func init() {
 	internal.ExitIdleModeForTesting = func(cc *ClientConn) {
 		cc.idlenessMgr.ExitIdleMode()
 	}
+	internal.NewChannelForBalancer = newChannelForBalancer
+}
+
+// newChannelForBalancer creates a new ClientConn that inherits parent's stats
+// handlers and client interceptors. Used by LB policies that open their own
+// control-plane channels (e.g. RLS) so per-attempt telemetry configured on the
+// parent channel also covers those control-plane RPCs. Inherited options are
+// prepended to opts, so callers may still override them by passing their own
+// stats handlers or interceptors in opts.
+//
+// The balancer.ClientConn handed to Balancer.Build is often wrapped several
+// layers deep in xDS/DirectPath configs (gracefulswitch, balancer group,
+// cluster_impl, outlier_detection, ...). Every wrapping layer exposes its
+// delegate via Unwrap() balancer.ClientConn (see the same-named methods on
+// those types); walk that chain until we find the underlying
+// *ccBalancerWrapper that holds the parent *ClientConn. If none is found,
+// fall back to a plain NewClient so the caller still gets a working channel.
+func newChannelForBalancer(parent balancer.ClientConn, target string, opts ...DialOption) (*ClientConn, error) {
+	type unwrapper interface{ Unwrap() balancer.ClientConn }
+	var ccb *ccBalancerWrapper
+	seen := make(map[balancer.ClientConn]struct{})
+	for cur := parent; cur != nil; {
+		if c, ok := cur.(*ccBalancerWrapper); ok {
+			ccb = c
+			break
+		}
+		if _, dup := seen[cur]; dup {
+			break
+		}
+		seen[cur] = struct{}{}
+		u, ok := cur.(unwrapper)
+		if !ok {
+			break
+		}
+		next := u.Unwrap()
+		if next == nil || next == cur {
+			break
+		}
+		cur = next
+	}
+	if ccb == nil || ccb.cc == nil {
+		return NewClient(target, opts...)
+	}
+	pd := ccb.cc.dopts
+	inherited := make([]DialOption, 0, len(pd.copts.StatsHandlers)+4)
+	for _, sh := range pd.copts.StatsHandlers {
+		inherited = append(inherited, WithStatsHandler(sh))
+	}
+	if pd.unaryInt != nil {
+		inherited = append(inherited, WithUnaryInterceptor(pd.unaryInt))
+	}
+	if len(pd.chainUnaryInts) > 0 {
+		inherited = append(inherited, WithChainUnaryInterceptor(pd.chainUnaryInts...))
+	}
+	if pd.streamInt != nil {
+		inherited = append(inherited, WithStreamInterceptor(pd.streamInt))
+	}
+	if len(pd.chainStreamInts) > 0 {
+		inherited = append(inherited, WithChainStreamInterceptor(pd.chainStreamInts...))
+	}
+	return NewClient(target, append(inherited, opts...)...)
 }
 
 func (cc *ClientConn) maybeApplyDefaultServiceConfig() {

internal/internal.go

@@ -143,6 +143,14 @@ var (
 	// provided grpc.ClientConn.
 	SubscribeToConnectivityStateChanges any // func(*grpc.ClientConn, grpcsync.Subscriber)
 
+	// NewChannelForBalancer creates a new grpc.ClientConn that inherits the
+	// parent ClientConn's stats handlers and client interceptors. It is meant
+	// to be used by LB policies that open their own control-plane channels
+	// (for example, RLS) so that per-attempt telemetry configured on the
+	// parent channel also covers those control-plane RPCs. The parent argument
+	// must be the balancer.ClientConn that gRPC passed to Balancer.Build.
+	NewChannelForBalancer any // func(parent balancer.ClientConn, target string, opts ...grpc.DialOption) (*grpc.ClientConn, error)
+
 	// NewXDSResolverWithConfigForTesting creates a new xds resolver builder using
 	// the provided xds bootstrap config instead of the global configuration from
 	// the supported environment variables.  The resolver.Builder is meant to be