Cherry-pick #8874 to v1.79.x (#8904)

Cherry picks [PR](#8874) into 1.79.x

RELEASE NOTES:
* stats: only process RPC stats/tracing in health and ORCA producers if
a handler is configured, preventing unnecessary error logging

Author: mbissa

internal/transport/client_stream.go

@@ -24,6 +24,7 @@ import (
 	"golang.org/x/net/http2"
 	"google.golang.org/grpc/mem"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
 )
 
@@ -46,10 +47,11 @@ type ClientStream struct {
 	// meaningful after headerChan is closed (always call waitOnHeader() before
 	// reading its value).
 	headerValid      bool
-	noHeaders        bool        // set if the client never received headers (set only after the stream is done).
-	headerChanClosed uint32      // set when headerChan is closed. Used to avoid closing headerChan multiple times.
-	bytesReceived    atomic.Bool // indicates whether any bytes have been received on this stream
-	unprocessed      atomic.Bool // set if the server sends a refused stream or GOAWAY including this stream
+	noHeaders        bool          // set if the client never received headers (set only after the stream is done).
+	headerChanClosed uint32        // set when headerChan is closed. Used to avoid closing headerChan multiple times.
+	bytesReceived    atomic.Bool   // indicates whether any bytes have been received on this stream
+	unprocessed      atomic.Bool   // set if the server sends a refused stream or GOAWAY including this stream
+	statsHandler     stats.Handler // nil for internal streams (e.g., health check, ORCA) where telemetry is not supported.
 }
 
 // Read reads an n byte message from the input stream.

internal/transport/http2_client.go

@@ -478,18 +478,19 @@ func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 	return t, nil
 }
 
-func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *ClientStream {
+func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr, handler stats.Handler) *ClientStream {
 	// TODO(zhaoq): Handle uint32 overflow of Stream.id.
 	s := &ClientStream{
 		Stream: Stream{
 			method:         callHdr.Method,
 			sendCompress:   callHdr.SendCompress,
 			contentSubtype: callHdr.ContentSubtype,
 		},
-		ct:         t,
-		done:       make(chan struct{}),
-		headerChan: make(chan struct{}),
-		doneFunc:   callHdr.DoneFunc,
+		ct:           t,
+		done:         make(chan struct{}),
+		headerChan:   make(chan struct{}),
+		doneFunc:     callHdr.DoneFunc,
+		statsHandler: handler,
 	}
 	s.Stream.buf.init()
 	s.Stream.wq.init(defaultWriteQuota, s.done)
@@ -744,7 +745,7 @@ func (e NewStreamError) Error() string {
 
 // NewStream creates a stream and registers it into the transport as "active"
 // streams.  All non-nil errors returned will be *NewStreamError.
-func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientStream, error) {
+func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr, handler stats.Handler) (*ClientStream, error) {
 	ctx = peer.NewContext(ctx, t.Peer())
 
 	// ServerName field of the resolver returned address takes precedence over
@@ -781,7 +782,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientS
 	if err != nil {
 		return nil, &NewStreamError{Err: err, AllowTransparentRetry: false}
 	}
-	s := t.newStream(ctx, callHdr)
+	s := t.newStream(ctx, callHdr, handler)
 	cleanup := func(err error) {
 		if s.swapState(streamDone) == streamDone {
 			// If it was already done, return.
@@ -902,7 +903,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientS
 			return nil, &NewStreamError{Err: ErrConnClosing, AllowTransparentRetry: true}
 		}
 	}
-	if t.statsHandler != nil {
+	if s.statsHandler != nil {
 		header, ok := metadata.FromOutgoingContext(ctx)
 		if ok {
 			header.Set("user-agent", t.userAgent)
@@ -911,7 +912,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientS
 		}
 		// Note: The header fields are compressed with hpack after this call returns.
 		// No WireLength field is set here.
-		t.statsHandler.HandleRPC(s.ctx, &stats.OutHeader{
+		s.statsHandler.HandleRPC(s.ctx, &stats.OutHeader{
 			Client:      true,
 			FullMethod:  callHdr.Method,
 			RemoteAddr:  t.remoteAddr,
@@ -1587,16 +1588,16 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		}
 	}
 
-	if t.statsHandler != nil {
+	if s.statsHandler != nil {
 		if !endStream {
-			t.statsHandler.HandleRPC(s.ctx, &stats.InHeader{
+			s.statsHandler.HandleRPC(s.ctx, &stats.InHeader{
 				Client:      true,
 				WireLength:  int(frame.Header().Length),
 				Header:      metadata.MD(mdata).Copy(),
 				Compression: s.recvCompress,
 			})
 		} else {
-			t.statsHandler.HandleRPC(s.ctx, &stats.InTrailer{
+			s.statsHandler.HandleRPC(s.ctx, &stats.InTrailer{
 				Client:     true,
 				WireLength: int(frame.Header().Length),
 				Trailer:    metadata.MD(mdata).Copy(),

internal/transport/keepalive_test.go

@@ -69,7 +69,7 @@ func (s) TestMaxConnectionIdle(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	stream, err := client.NewStream(ctx, &CallHdr{})
+	stream, err := client.NewStream(ctx, &CallHdr{}, nil)
 	if err != nil {
 		t.Fatalf("client.NewStream() failed: %v", err)
 	}
@@ -111,7 +111,7 @@ func (s) TestMaxConnectionIdleBusyClient(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	_, err := client.NewStream(ctx, &CallHdr{})
+	_, err := client.NewStream(ctx, &CallHdr{}, nil)
 	if err != nil {
 		t.Fatalf("client.NewStream() failed: %v", err)
 	}
@@ -150,7 +150,7 @@ func (s) TestMaxConnectionAge(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	if _, err := client.NewStream(ctx, &CallHdr{}); err != nil {
+	if _, err := client.NewStream(ctx, &CallHdr{}, nil); err != nil {
 		t.Fatalf("client.NewStream() failed: %v", err)
 	}
 
@@ -372,7 +372,7 @@ func (s) TestKeepaliveClientClosesWithActiveStreams(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	// Create a stream, but send no data on it.
-	if _, err := client.NewStream(ctx, &CallHdr{}); err != nil {
+	if _, err := client.NewStream(ctx, &CallHdr{}, nil); err != nil {
 		t.Fatalf("Stream creation failed: %v", err)
 	}
 
@@ -514,7 +514,7 @@ func (s) TestKeepaliveServerEnforcementWithAbusiveClientWithRPC(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	if _, err := client.NewStream(ctx, &CallHdr{}); err != nil {
+	if _, err := client.NewStream(ctx, &CallHdr{}, nil); err != nil {
 		t.Fatalf("Stream creation failed: %v", err)
 	}
 
@@ -743,7 +743,7 @@ func (s) TestTCPUserTimeout(t *testing.T) {
 
 		ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 		defer cancel()
-		stream, err := client.NewStream(ctx, &CallHdr{})
+		stream, err := client.NewStream(ctx, &CallHdr{}, nil)
 		if err != nil {
 			t.Fatalf("client.NewStream() failed: %v", err)
 		}
@@ -810,7 +810,7 @@ func makeTLSCreds(t *testing.T, certPath, keyPath, rootsPath string) credentials
 func checkForHealthyStream(client *http2Client) error {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	stream, err := client.NewStream(ctx, &CallHdr{})
+	stream, err := client.NewStream(ctx, &CallHdr{}, nil)
 	stream.Close(err)
 	return err
 }
@@ -819,7 +819,7 @@ func pollForStreamCreationError(client *http2Client) error {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	for {
-		if _, err := client.NewStream(ctx, &CallHdr{}); err != nil {
+		if _, err := client.NewStream(ctx, &CallHdr{}, nil); err != nil {
 			break
 		}
 		time.Sleep(50 * time.Millisecond)
@@ -845,7 +845,7 @@ func waitForGoAwayTooManyPings(client *http2Client) error {
 		return fmt.Errorf("test timed out before getting GoAway with reason:GoAwayTooManyPings from server")
 	}
 
-	if _, err := client.NewStream(ctx, &CallHdr{}); err == nil {
+	if _, err := client.NewStream(ctx, &CallHdr{}, nil); err == nil {
 		return fmt.Errorf("stream creation succeeded after receiving a GoAway from the server")
 	}
 	return nil

internal/transport/transport.go

@@ -617,7 +617,7 @@ type ClientTransport interface {
 	GracefulClose()
 
 	// NewStream creates a Stream for an RPC.
-	NewStream(ctx context.Context, callHdr *CallHdr) (*ClientStream, error)
+	NewStream(ctx context.Context, callHdr *CallHdr, handler stats.Handler) (*ClientStream, error)
 
 	// Error returns a channel that is closed when some I/O error
 	// happens. Typically the caller should have a goroutine to monitor