Merge pull request #9288 from xiang90/wl
etcdserver: improve request took too long warningrelease-3.4
commit
11214aa33b
|
@ -107,6 +107,8 @@ func (s *EtcdServer) newApplierV3() applierV3 {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *applierV3backend) Apply(r *pb.InternalRaftRequest) *applyResult {
|
func (a *applierV3backend) Apply(r *pb.InternalRaftRequest) *applyResult {
|
||||||
|
defer warnOfExpensiveRequest(time.Now(), r)
|
||||||
|
|
||||||
ar := &applyResult{}
|
ar := &applyResult{}
|
||||||
|
|
||||||
// call into a.s.applyV3.F instead of a.F so upper appliers can check individual calls
|
// call into a.s.applyV3.F instead of a.F so upper appliers can check individual calls
|
||||||
|
|
|
@ -107,6 +107,8 @@ func (a *applierV2store) Sync(r *RequestV2) Response {
|
||||||
// applyV2Request interprets r as a call to store.X and returns a Response interpreted
|
// applyV2Request interprets r as a call to store.X and returns a Response interpreted
|
||||||
// from store.Event
|
// from store.Event
|
||||||
func (s *EtcdServer) applyV2Request(r *RequestV2) Response {
|
func (s *EtcdServer) applyV2Request(r *RequestV2) Response {
|
||||||
|
defer warnOfExpensiveRequest(time.Now(), r)
|
||||||
|
|
||||||
switch r.Method {
|
switch r.Method {
|
||||||
case "POST":
|
case "POST":
|
||||||
return s.applyV2.Post(r)
|
return s.applyV2.Post(r)
|
||||||
|
|
|
@ -794,14 +794,8 @@ func (s *EtcdServer) run() {
|
||||||
|
|
||||||
func (s *EtcdServer) applyAll(ep *etcdProgress, apply *apply) {
|
func (s *EtcdServer) applyAll(ep *etcdProgress, apply *apply) {
|
||||||
s.applySnapshot(ep, apply)
|
s.applySnapshot(ep, apply)
|
||||||
st := time.Now()
|
|
||||||
s.applyEntries(ep, apply)
|
s.applyEntries(ep, apply)
|
||||||
d := time.Since(st)
|
|
||||||
entriesNum := len(apply.entries)
|
|
||||||
if entriesNum != 0 && d > time.Duration(entriesNum)*warnApplyDuration {
|
|
||||||
plog.Warningf("apply entries took too long [%v for %d entries]", d, len(apply.entries))
|
|
||||||
plog.Warningf("avoid queries with large range/delete range!")
|
|
||||||
}
|
|
||||||
proposalsApplied.Set(float64(ep.appliedi))
|
proposalsApplied.Set(float64(ep.appliedi))
|
||||||
s.applyWait.Trigger(ep.appliedi)
|
s.applyWait.Trigger(ep.appliedi)
|
||||||
// wait for the raft routine to finish the disk writes before triggering a
|
// wait for the raft routine to finish the disk writes before triggering a
|
||||||
|
|
|
@ -15,6 +15,7 @@
|
||||||
package etcdserver
|
package etcdserver
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"fmt"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/coreos/etcd/etcdserver/membership"
|
"github.com/coreos/etcd/etcdserver/membership"
|
||||||
|
@ -95,3 +96,19 @@ func (nc *notifier) notify(err error) {
|
||||||
nc.err = err
|
nc.err = err
|
||||||
close(nc.c)
|
close(nc.c)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func warnOfExpensiveRequest(now time.Time, stringer fmt.Stringer) {
|
||||||
|
warnOfExpensiveGenericRequest(now, stringer, "")
|
||||||
|
}
|
||||||
|
|
||||||
|
func warnOfExpensiveReadOnlyRangeRequest(now time.Time, stringer fmt.Stringer) {
|
||||||
|
warnOfExpensiveGenericRequest(now, stringer, "read-only range ")
|
||||||
|
}
|
||||||
|
|
||||||
|
func warnOfExpensiveGenericRequest(now time.Time, stringer fmt.Stringer, prefix string) {
|
||||||
|
// TODO: add metrics
|
||||||
|
d := time.Since(now)
|
||||||
|
if d > warnApplyDuration {
|
||||||
|
plog.Warningf("%srequest %q took too long (%v) to execute", prefix, stringer.String(), d)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
@ -158,3 +158,8 @@ func (r *RequestV2) Handle(ctx context.Context, v2api RequestV2Handler) (Respons
|
||||||
}
|
}
|
||||||
return Response{}, ErrUnknownMethod
|
return Response{}, ErrUnknownMethod
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (r *RequestV2) String() string {
|
||||||
|
rpb := pb.Request(*r)
|
||||||
|
return rpb.String()
|
||||||
|
}
|
||||||
|
|
|
@ -84,6 +84,8 @@ type Authenticator interface {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *EtcdServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
|
func (s *EtcdServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
|
||||||
|
defer warnOfExpensiveReadOnlyRangeRequest(time.Now(), r)
|
||||||
|
|
||||||
if !r.Serializable {
|
if !r.Serializable {
|
||||||
err := s.linearizableReadNotify(ctx)
|
err := s.linearizableReadNotify(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -95,6 +97,7 @@ func (s *EtcdServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeRe
|
||||||
chk := func(ai *auth.AuthInfo) error {
|
chk := func(ai *auth.AuthInfo) error {
|
||||||
return s.authStore.IsRangePermitted(ai, r.Key, r.RangeEnd)
|
return s.authStore.IsRangePermitted(ai, r.Key, r.RangeEnd)
|
||||||
}
|
}
|
||||||
|
|
||||||
get := func() { resp, err = s.applyV3Base.Range(nil, r) }
|
get := func() { resp, err = s.applyV3Base.Range(nil, r) }
|
||||||
if serr := s.doSerialize(ctx, chk, get); serr != nil {
|
if serr := s.doSerialize(ctx, chk, get); serr != nil {
|
||||||
return nil, serr
|
return nil, serr
|
||||||
|
@ -131,12 +134,16 @@ func (s *EtcdServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse
|
||||||
chk := func(ai *auth.AuthInfo) error {
|
chk := func(ai *auth.AuthInfo) error {
|
||||||
return checkTxnAuth(s.authStore, ai, r)
|
return checkTxnAuth(s.authStore, ai, r)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
defer warnOfExpensiveReadOnlyRangeRequest(time.Now(), r)
|
||||||
|
|
||||||
get := func() { resp, err = s.applyV3Base.Txn(r) }
|
get := func() { resp, err = s.applyV3Base.Txn(r) }
|
||||||
if serr := s.doSerialize(ctx, chk, get); serr != nil {
|
if serr := s.doSerialize(ctx, chk, get); serr != nil {
|
||||||
return nil, serr
|
return nil, serr
|
||||||
}
|
}
|
||||||
return resp, err
|
return resp, err
|
||||||
}
|
}
|
||||||
|
|
||||||
resp, err := s.raftRequest(ctx, pb.InternalRaftRequest{Txn: r})
|
resp, err := s.raftRequest(ctx, pb.InternalRaftRequest{Txn: r})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|
Loading…
Reference in New Issue