Skip to content

Commit

Permalink
etcdserver: Add response byte size and range response count to took t…
Browse files Browse the repository at this point in the history
…oo long warning
  • Loading branch information
jpbetz committed Jun 11, 2018
1 parent 25f4d80 commit b47e148
Show file tree
Hide file tree
Showing 4 changed files with 63 additions and 16 deletions.
5 changes: 3 additions & 2 deletions etcdserver/apply.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
2 changes: 1 addition & 1 deletion etcdserver/apply_v2.go
Original file line number Diff line number Diff line change
Expand Up @@ -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":
Expand Down
55 changes: 48 additions & 7 deletions etcdserver/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -99,15 +103,40 @@ func (nc *notifier) notify(err error) {
close(nc.c)
}

func warnOfExpensiveRequest(lg *zap.Logger, now time.Time, stringer fmt.Stringer) {
warnOfExpensiveGenericRequest(lg, now, stringer, "")
func warnOfExpensiveRequest(lg *zap.Logger, now time.Time, reqStringer fmt.Stringer, respMsg proto.Message, err error) {
var resp string
if !isNil(respMsg) {
resp = fmt.Sprintf("size:%d", proto.Size(respMsg))
}
warnOfExpensiveGenericRequest(lg, now, reqStringer, "", resp, err)
}

func warnOfExpensiveReadOnlyTxnRequest(lg *zap.Logger, now time.Time, reqStringer fmt.Stringer, txnResponse *pb.TxnResponse, err error) {
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 warnOfExpensiveReadOnlyRangeRequest(lg *zap.Logger, now time.Time, stringer fmt.Stringer) {
warnOfExpensiveGenericRequest(lg, now, stringer, "read-only range ")
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, stringer fmt.Stringer, prefix string) {
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 {
Expand All @@ -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()
}
17 changes: 11 additions & 6 deletions etcdserver/v3_server.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down Expand Up @@ -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 {
Expand Down

0 comments on commit b47e148

Please sign in to comment.