diff options
Diffstat (limited to 'vendor/google.golang.org/grpc/transport')
3 files changed, 35 insertions, 25 deletions
diff --git a/vendor/google.golang.org/grpc/transport/http2_client.go b/vendor/google.golang.org/grpc/transport/http2_client.go index 605b1e5..892f8ba 100644 --- a/vendor/google.golang.org/grpc/transport/http2_client.go +++ b/vendor/google.golang.org/grpc/transport/http2_client.go @@ -99,6 +99,8 @@ type http2Client struct { creds []credentials.PerRPCCredentials + statsHandler stats.Handler + mu sync.Mutex // guard the following variables state transportState // the state of underlying connection activeStreams map[uint32]*Stream @@ -208,16 +210,17 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) ( creds: opts.PerRPCCredentials, maxStreams: math.MaxInt32, streamSendQuota: defaultWindowSize, + statsHandler: opts.StatsHandler, } - if stats.On() { - t.ctx = stats.TagConn(t.ctx, &stats.ConnTagInfo{ + if t.statsHandler != nil { + t.ctx = t.statsHandler.TagConn(t.ctx, &stats.ConnTagInfo{ RemoteAddr: t.remoteAddr, LocalAddr: t.localAddr, }) connBegin := &stats.ConnBegin{ Client: true, } - stats.HandleConn(t.ctx, connBegin) + t.statsHandler.HandleConn(t.ctx, connBegin) } // Start the reader goroutine for incoming message. Each transport has // a dedicated goroutine which reads HTTP2 frame from network. Then it @@ -470,7 +473,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea return nil, connectionErrorf(true, err, "transport: %v", err) } } - if stats.On() { + if t.statsHandler != nil { outHeader := &stats.OutHeader{ Client: true, WireLength: bufLen, @@ -479,7 +482,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea LocalAddr: t.localAddr, Compression: callHdr.SendCompress, } - stats.HandleRPC(s.clientStatsCtx, outHeader) + t.statsHandler.HandleRPC(s.clientStatsCtx, outHeader) } t.writableChan <- 0 return s, nil @@ -559,11 +562,11 @@ func (t *http2Client) Close() (err error) { s.mu.Unlock() s.write(recvMsg{err: ErrConnClosing}) } - if stats.On() { + if t.statsHandler != nil { connEnd := &stats.ConnEnd{ Client: true, } - stats.HandleConn(t.ctx, connEnd) + t.statsHandler.HandleConn(t.ctx, connEnd) } return } @@ -911,19 +914,19 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) { endStream := frame.StreamEnded() var isHeader bool defer func() { - if stats.On() { + if t.statsHandler != nil { if isHeader { inHeader := &stats.InHeader{ Client: true, WireLength: int(frame.Header().Length), } - stats.HandleRPC(s.clientStatsCtx, inHeader) + t.statsHandler.HandleRPC(s.clientStatsCtx, inHeader) } else { inTrailer := &stats.InTrailer{ Client: true, WireLength: int(frame.Header().Length), } - stats.HandleRPC(s.clientStatsCtx, inTrailer) + t.statsHandler.HandleRPC(s.clientStatsCtx, inTrailer) } } }() diff --git a/vendor/google.golang.org/grpc/transport/http2_server.go b/vendor/google.golang.org/grpc/transport/http2_server.go index 316188e..a095dd0 100644 --- a/vendor/google.golang.org/grpc/transport/http2_server.go +++ b/vendor/google.golang.org/grpc/transport/http2_server.go @@ -88,6 +88,8 @@ type http2Server struct { // sendQuotaPool provides flow control to outbound message. sendQuotaPool *quotaPool + stats stats.Handler + mu sync.Mutex // guard the following state transportState activeStreams map[uint32]*Stream @@ -146,14 +148,15 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err shutdownChan: make(chan struct{}), activeStreams: make(map[uint32]*Stream), streamSendQuota: defaultWindowSize, + stats: config.StatsHandler, } - if stats.On() { - t.ctx = stats.TagConn(t.ctx, &stats.ConnTagInfo{ + if t.stats != nil { + t.ctx = t.stats.TagConn(t.ctx, &stats.ConnTagInfo{ RemoteAddr: t.remoteAddr, LocalAddr: t.localAddr, }) connBegin := &stats.ConnBegin{} - stats.HandleConn(t.ctx, connBegin) + t.stats.HandleConn(t.ctx, connBegin) } go t.controller() t.writableChan <- 0 @@ -250,8 +253,8 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func( t.updateWindow(s, uint32(n)) } s.ctx = traceCtx(s.ctx, s.method) - if stats.On() { - s.ctx = stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method}) + if t.stats != nil { + s.ctx = t.stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method}) inHeader := &stats.InHeader{ FullMethod: s.method, RemoteAddr: t.remoteAddr, @@ -259,7 +262,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func( Compression: s.recvCompress, WireLength: int(frame.Header().Length), } - stats.HandleRPC(s.ctx, inHeader) + t.stats.HandleRPC(s.ctx, inHeader) } handle(s) return @@ -540,11 +543,11 @@ func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error { if err := t.writeHeaders(s, t.hBuf, false); err != nil { return err } - if stats.On() { + if t.stats != nil { outHeader := &stats.OutHeader{ WireLength: bufLen, } - stats.HandleRPC(s.Context(), outHeader) + t.stats.HandleRPC(s.Context(), outHeader) } t.writableChan <- 0 return nil @@ -603,11 +606,11 @@ func (t *http2Server) WriteStatus(s *Stream, statusCode codes.Code, statusDesc s t.Close() return err } - if stats.On() { + if t.stats != nil { outTrailer := &stats.OutTrailer{ WireLength: bufLen, } - stats.HandleRPC(s.Context(), outTrailer) + t.stats.HandleRPC(s.Context(), outTrailer) } t.closeStream(s) t.writableChan <- 0 @@ -789,9 +792,9 @@ func (t *http2Server) Close() (err error) { for _, s := range streams { s.cancel() } - if stats.On() { + if t.stats != nil { connEnd := &stats.ConnEnd{} - stats.HandleConn(t.ctx, connEnd) + t.stats.HandleConn(t.ctx, connEnd) } return } diff --git a/vendor/google.golang.org/grpc/transport/transport.go b/vendor/google.golang.org/grpc/transport/transport.go index 4726bb2..d465991 100644 --- a/vendor/google.golang.org/grpc/transport/transport.go +++ b/vendor/google.golang.org/grpc/transport/transport.go @@ -48,6 +48,7 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/stats" "google.golang.org/grpc/tap" ) @@ -357,9 +358,10 @@ const ( // ServerConfig consists of all the configurations to establish a server transport. type ServerConfig struct { - MaxStreams uint32 - AuthInfo credentials.AuthInfo - InTapHandle tap.ServerInHandle + MaxStreams uint32 + AuthInfo credentials.AuthInfo + InTapHandle tap.ServerInHandle + StatsHandler stats.Handler } // NewServerTransport creates a ServerTransport with conn or non-nil error @@ -380,6 +382,8 @@ type ConnectOptions struct { PerRPCCredentials []credentials.PerRPCCredentials // TransportCredentials stores the Authenticator required to setup a client connection. TransportCredentials credentials.TransportCredentials + // StatsHandler stores the handler for stats. + StatsHandler stats.Handler } // TargetInfo contains the information of the target such as network address and metadata. |