add tracing for streaming rpc

This commit is contained in:
yangzhouhan
2015-06-17 19:30:57 -07:00
parent 4782e693c2
commit 58dcee7526
3 changed files with 43 additions and 13 deletions

View File

@ -102,9 +102,6 @@ type callInfo struct {
traceInfo traceInfo // in trace.go traceInfo traceInfo // in trace.go
} }
// EnableTracing controls whether to trace RPCs using the golang.org/x/net/trace package.
// This should only be set before any RPCs are sent or received by this program.
var EnableTracing = true
// Invoke is called by the generated code. It sends the RPC request on the // Invoke is called by the generated code. It sends the RPC request on the
// wire and returns after response is received. // wire and returns after response is received.

View File

@ -36,8 +36,10 @@ package grpc
import ( import (
"errors" "errors"
"io" "io"
"time"
"golang.org/x/net/context" "golang.org/x/net/context"
"golang.org/x/net/trace"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata" "google.golang.org/grpc/metadata"
"google.golang.org/grpc/transport" "google.golang.org/grpc/transport"
@ -94,6 +96,7 @@ type ClientStream interface {
// by generated code. // by generated code.
func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) { func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
// TODO(zhaoq): CallOption is omitted. Add support when it is needed. // TODO(zhaoq): CallOption is omitted. Add support when it is needed.
var trInfo traceInfo
callHdr := &transport.CallHdr{ callHdr := &transport.CallHdr{
Host: cc.authority, Host: cc.authority,
Method: method, Method: method,
@ -102,6 +105,14 @@ func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
if err != nil { if err != nil {
return nil, toRPCErr(err) return nil, toRPCErr(err)
} }
if EnableTracing {
trInfo.tr = trace.New("Sent."+methodFamily(desc.StreamName), desc.StreamName)
trInfo.firstLine.client = true
if deadline, ok := ctx.Deadline(); ok {
trInfo.firstLine.deadline = deadline.Sub(time.Now())
}
trInfo.tr.LazyLog(&trInfo.firstLine, false)
}
s, err := t.NewStream(ctx, callHdr) s, err := t.NewStream(ctx, callHdr)
if err != nil { if err != nil {
return nil, toRPCErr(err) return nil, toRPCErr(err)
@ -112,6 +123,7 @@ func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
p: &parser{s: s}, p: &parser{s: s},
desc: desc, desc: desc,
codec: cc.dopts.codec, codec: cc.dopts.codec,
traceInfo: trInfo,
}, nil }, nil
} }
@ -122,6 +134,7 @@ type clientStream struct {
p *parser p *parser
desc *StreamDesc desc *StreamDesc
codec Codec codec Codec
traceInfo traceInfo
} }
func (cs *clientStream) Context() context.Context { func (cs *clientStream) Context() context.Context {
@ -152,6 +165,7 @@ func (cs *clientStream) SendMsg(m interface{}) (err error) {
} }
err = toRPCErr(err) err = toRPCErr(err)
}() }()
out, err := encode(cs.codec, m, compressionNone) out, err := encode(cs.codec, m, compressionNone)
if err != nil { if err != nil {
return transport.StreamErrorf(codes.Internal, "grpc: %v", err) return transport.StreamErrorf(codes.Internal, "grpc: %v", err)
@ -172,23 +186,38 @@ func (cs *clientStream) RecvMsg(m interface{}) (err error) {
return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>")) return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
} }
if err == io.EOF { if err == io.EOF {
if EnableTracing {
defer cs.traceInfo.tr.Finish()
}
if cs.s.StatusCode() == codes.OK { if cs.s.StatusCode() == codes.OK {
return nil return nil
} }
return Errorf(cs.s.StatusCode(), cs.s.StatusDesc()) return Errorf(cs.s.StatusCode(), cs.s.StatusDesc())
} }
if EnableTracing {
cs.traceInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
cs.traceInfo.tr.SetError()
}
return toRPCErr(err) return toRPCErr(err)
} }
if _, ok := err.(transport.ConnectionError); !ok { if _, ok := err.(transport.ConnectionError); !ok {
cs.t.CloseStream(cs.s, err) cs.t.CloseStream(cs.s, err)
} }
if err == io.EOF { if err == io.EOF {
if EnableTracing {
cs.traceInfo.tr.Finish()
}
if cs.s.StatusCode() == codes.OK { if cs.s.StatusCode() == codes.OK {
// Returns io.EOF to indicate the end of the stream. // Returns io.EOF to indicate the end of the stream.
return return
} }
return Errorf(cs.s.StatusCode(), cs.s.StatusDesc()) return Errorf(cs.s.StatusCode(), cs.s.StatusDesc())
} }
if EnableTracing {
cs.traceInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
cs.traceInfo.tr.SetError()
}
return toRPCErr(err) return toRPCErr(err)
} }

View File

@ -44,6 +44,10 @@ import (
"golang.org/x/net/trace" "golang.org/x/net/trace"
) )
// EnableTracing controls whether to trace RPCs using the golang.org/x/net/trace package.
// This should only be set before any RPCs are sent or received by this program.
var EnableTracing = true
// methodFamily returns the trace family for the given method. // methodFamily returns the trace family for the given method.
// It turns "/pkg.Service/GetFoo" into "pkg.Service". // It turns "/pkg.Service/GetFoo" into "pkg.Service".
func methodFamily(m string) string { func methodFamily(m string) string {