stats: add BeginTime to stats.End (grpc#1907)

server.go

@@ -823,13 +823,15 @@ func (s *Server) sendResponse(t transport.ServerTransport, stream *transport.Str
 func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, md *MethodDesc, trInfo *traceInfo) (err error) {
 	sh := s.opts.statsHandler
 	if sh != nil {
+		beginTime := time.Now()
 		begin := &stats.Begin{
-			BeginTime: time.Now(),
+			BeginTime: beginTime,
 		}
 		sh.HandleRPC(stream.Context(), begin)
 		defer func() {
 			end := &stats.End{
-				EndTime: time.Now(),
+				BeginTime: beginTime,
+				EndTime:   time.Now(),
 			}
 			if err != nil && err != io.EOF {
 				end.Error = toRPCErr(err)
@@ -1023,13 +1025,15 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
 func (s *Server) processStreamingRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, sd *StreamDesc, trInfo *traceInfo) (err error) {
 	sh := s.opts.statsHandler
 	if sh != nil {
+		beginTime := time.Now()
 		begin := &stats.Begin{
-			BeginTime: time.Now(),
+			BeginTime: beginTime,
 		}
 		sh.HandleRPC(stream.Context(), begin)
 		defer func() {
 			end := &stats.End{
-				EndTime: time.Now(),
+				BeginTime: beginTime,
+				EndTime:   time.Now(),
 			}
 			if err != nil && err != io.EOF {
 				end.Error = toRPCErr(err)

stats/stats.go

@@ -169,6 +169,8 @@ func (s *OutTrailer) isRPCStats() {}
 type End struct {
 	// Client is true if this End is from client side.
 	Client bool
+	// BeginTime is the time when the RPC began.
+	BeginTime time.Time
 	// EndTime is the time when the RPC ends.
 	EndTime time.Time
 	// Error is the error the RPC ended with. It is an error generated from

stats/stats_test.go

@@ -638,6 +638,9 @@ func checkEnd(t *testing.T, d *gotData, e *expectedData) {
 	if d.ctx == nil {
 		t.Fatalf("d.ctx = nil, want <non-nil>")
 	}
+	if st.BeginTime.IsZero() {
+		t.Fatalf("st.BeginTime = %v, want <non-zero>", st.BeginTime)
+	}
 	if st.EndTime.IsZero() {
 		t.Fatalf("st.EndTime = %v, want <non-zero>", st.EndTime)
 	}

stream.go

@@ -205,20 +205,24 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 	}
 	ctx = newContextWithRPCInfo(ctx, c.failFast)
 	sh := cc.dopts.copts.StatsHandler
+	var beginTime time.Time
 	if sh != nil {
 		ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method, FailFast: c.failFast})
+		beginTime = time.Now()
 		begin := &stats.Begin{
 			Client:    true,
-			BeginTime: time.Now(),
+			BeginTime: beginTime,
 			FailFast:  c.failFast,
 		}
 		sh.HandleRPC(ctx, begin)
 		defer func() {
 			if err != nil {
 				// Only handle end stats if err != nil.
 				end := &stats.End{
-					Client: true,
-					Error:  err,
+					Client:    true,
+					Error:     err,
+					BeginTime: beginTime,
+					EndTime:   time.Now(),
 				}
 				sh.HandleRPC(ctx, end)
 			}
@@ -283,6 +287,7 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 
 		statsCtx:     ctx,
 		statsHandler: cc.dopts.copts.StatsHandler,
+		beginTime:    beginTime,
 	}
 	if desc != unaryStreamDesc {
 		// Listen on cc and stream contexts to cleanup when the user closes the
@@ -337,6 +342,7 @@ type clientStream struct {
 	// so that all the generated stats for a particular RPC can be associated in the processing phase.
 	statsCtx     context.Context
 	statsHandler stats.Handler
+	beginTime    time.Time
 }
 
 func (cs *clientStream) Context() context.Context {
@@ -512,9 +518,10 @@ func (cs *clientStream) finish(err error) {
 	}
 	if cs.statsHandler != nil {
 		end := &stats.End{
-			Client:  true,
-			EndTime: time.Now(),
-			Error:   err,
+			Client:    true,
+			BeginTime: cs.beginTime,
+			EndTime:   time.Now(),
+			Error:     err,
 		}
 		cs.statsHandler.HandleRPC(cs.statsCtx, end)
 	}