Merge pull request #9826 from jpbetz/response_sizes

etcdserver: Add response byte size and range response count to took too long warning

etcdserver/apply.go

@@ -109,9 +109,10 @@ func (s *EtcdServer) newApplierV3() applierV3 {
 }
 
 func (a *applierV3backend) Apply(r *pb.InternalRaftRequest) *applyResult {
-	defer warnOfExpensiveRequest(a.s.getLogger(), time.Now(), &pb.InternalRaftStringer{Request: r})
-
 	ar := &applyResult{}
+	defer func(start time.Time) {
+		warnOfExpensiveRequest(a.s.getLogger(), start, &pb.InternalRaftStringer{Request: r}, ar.resp, ar.err)
+	}(time.Now())
 
 	// call into a.s.applyV3.F instead of a.F so upper appliers can check individual calls
 	switch {

etcdserver/apply_v2.go

@@ -114,7 +114,7 @@ func (a *applierV2store) Sync(r *RequestV2) Response {
 // applyV2Request interprets r as a call to v2store.X
 // and returns a Response interpreted from v2store.Event
 func (s *EtcdServer) applyV2Request(r *RequestV2) Response {
-	defer warnOfExpensiveRequest(s.getLogger(), time.Now(), r)
+	defer warnOfExpensiveRequest(s.getLogger(), time.Now(), r, nil, nil)
 
 	switch r.Method {
 	case "POST":

etcdserver/util.go

@@ -16,11 +16,15 @@ package etcdserver
 
 import (
 	"fmt"
+	"reflect"
+	"strings"
 	"time"
 
 	"github.com/coreos/etcd/etcdserver/api/membership"
 	"github.com/coreos/etcd/etcdserver/api/rafthttp"
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 	"github.com/coreos/etcd/pkg/types"
+	"github.com/golang/protobuf/proto"
 
 	"go.uber.org/zap"
 )
@@ -99,15 +103,40 @@ func (nc *notifier) notify(err error) {
 	close(nc.c)
 }
 
-func warnOfExpensiveRequest(lg *zap.Logger, now time.Time, stringer fmt.Stringer) {
+func warnOfExpensiveRequest(lg *zap.Logger, now time.Time, reqStringer fmt.Stringer, respMsg proto.Message, err error) {
-	warnOfExpensiveGenericRequest(lg, now, stringer, "")
+	var resp string
+	if !isNil(respMsg) {
+		resp = fmt.Sprintf("size:%d", proto.Size(respMsg))
+	}
+	warnOfExpensiveGenericRequest(lg, now, reqStringer, "", resp, err)
 }
 
-func warnOfExpensiveReadOnlyRangeRequest(lg *zap.Logger, now time.Time, stringer fmt.Stringer) {
+func warnOfExpensiveReadOnlyTxnRequest(lg *zap.Logger, now time.Time, reqStringer fmt.Stringer, txnResponse *pb.TxnResponse, err error) {
-	warnOfExpensiveGenericRequest(lg, now, stringer, "read-only range ")
+	var resp string
+	if !isNil(txnResponse) {
+		var resps []string
+		for _, r := range txnResponse.Responses {
+			switch op := r.Response.(type) {
+			case *pb.ResponseOp_ResponseRange:
+				resps = append(resps, fmt.Sprintf("range_response_count:%d", len(op.ResponseRange.Kvs)))
+			default:
+				// only range responses should be in a read only txn request
+			}
+		}
+		resp = fmt.Sprintf("responses:<%s> size:%d", strings.Join(resps, " "), proto.Size(txnResponse))
+	}
+	warnOfExpensiveGenericRequest(lg, now, reqStringer, "read-only range ", resp, err)
 }
 
-func warnOfExpensiveGenericRequest(lg *zap.Logger, now time.Time, stringer fmt.Stringer, prefix string) {
+func warnOfExpensiveReadOnlyRangeRequest(lg *zap.Logger, now time.Time, reqStringer fmt.Stringer, rangeResponse *pb.RangeResponse, err error) {
+	var resp string
+	if !isNil(rangeResponse) {
+		resp = fmt.Sprintf("range_response_count:%d size:%d", len(rangeResponse.Kvs), proto.Size(rangeResponse))
+	}
+	warnOfExpensiveGenericRequest(lg, now, reqStringer, "read-only range ", resp, err)
+}
+
+func warnOfExpensiveGenericRequest(lg *zap.Logger, now time.Time, reqStringer fmt.Stringer, prefix string, resp string, err error) {
 	// TODO: add metrics
 	d := time.Since(now)
 	if d > warnApplyDuration {
@@ -117,11 +146,23 @@ func warnOfExpensiveGenericRequest(lg *zap.Logger, now time.Time, stringer fmt.S
 				zap.Duration("took", d),
 				zap.Duration("expected-duration", warnApplyDuration),
 				zap.String("prefix", prefix),
-				zap.String("request", stringer.String()),
+				zap.String("request", reqStringer.String()),
+				zap.String("response", resp),
+				zap.Error(err),
 			)
 		} else {
-			plog.Warningf("%srequest %q took too long (%v) to execute", prefix, stringer.String(), d)
+			var result string
+			if err != nil {
+				result = fmt.Sprintf("error:%v", err)
+			} else {
+				result = resp
+			}
+			plog.Warningf("%srequest %q with result %q took too long (%v) to execute", prefix, reqStringer.String(), result, d)
 		}
 		slowApplies.Inc()
 	}
 }
+
+func isNil(msg proto.Message) bool {
+	return msg == nil || reflect.ValueOf(msg).IsNil()
+}

etcdserver/v3_server.go

@@ -86,23 +86,26 @@ type Authenticator interface {
 }
 
 func (s *EtcdServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
-	defer warnOfExpensiveReadOnlyRangeRequest(s.getLogger(), time.Now(), r)
+	var resp *pb.RangeResponse
+	var err error
+	defer func(start time.Time) {
+		warnOfExpensiveReadOnlyRangeRequest(s.getLogger(), start, r, resp, err)
+	}(time.Now())
 
 	if !r.Serializable {
-		err := s.linearizableReadNotify(ctx)
+		err = s.linearizableReadNotify(ctx)
 		if err != nil {
 			return nil, err
 		}
 	}
-	var resp *pb.RangeResponse
-	var err error
 	chk := func(ai *auth.AuthInfo) error {
 		return s.authStore.IsRangePermitted(ai, r.Key, r.RangeEnd)
 	}
 
 	get := func() { resp, err = s.applyV3Base.Range(nil, r) }
 	if serr := s.doSerialize(ctx, chk, get); serr != nil {
-		return nil, serr
+		err = serr
+		return nil, err
 	}
 	return resp, err
 }
@@ -137,7 +140,9 @@ func (s *EtcdServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse
 			return checkTxnAuth(s.authStore, ai, r)
 		}
 
-		defer warnOfExpensiveReadOnlyRangeRequest(s.getLogger(), time.Now(), r)
+		defer func(start time.Time) {
+			warnOfExpensiveReadOnlyTxnRequest(s.getLogger(), start, r, resp, err)
+		}(time.Now())
 
 		get := func() { resp, err = s.applyV3Base.Txn(r) }
 		if serr := s.doSerialize(ctx, chk, get); serr != nil {