1
2
3
4
5
6
7
8
9
10
11
12
13
14
15 package etcdserver
16
17 import (
18 "fmt"
19 "reflect"
20 "strings"
21 "time"
22
23 "github.com/golang/protobuf/proto"
24 pb "go.etcd.io/etcd/api/v3/etcdserverpb"
25 "go.etcd.io/etcd/client/pkg/v3/types"
26 "go.etcd.io/etcd/server/v3/etcdserver/api/membership"
27 "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
28
29 "go.uber.org/zap"
30 )
31
32
33
34 func isConnectedToQuorumSince(transport rafthttp.Transporter, since time.Time, self types.ID, members []*membership.Member) bool {
35 return numConnectedSince(transport, since, self, members) >= (len(members)/2)+1
36 }
37
38
39
40 func isConnectedSince(transport rafthttp.Transporter, since time.Time, remote types.ID) bool {
41 t := transport.ActiveSince(remote)
42 return !t.IsZero() && t.Before(since)
43 }
44
45
46
47 func isConnectedFullySince(transport rafthttp.Transporter, since time.Time, self types.ID, members []*membership.Member) bool {
48 return numConnectedSince(transport, since, self, members) == len(members)
49 }
50
51
52
53 func numConnectedSince(transport rafthttp.Transporter, since time.Time, self types.ID, members []*membership.Member) int {
54 connectedNum := 0
55 for _, m := range members {
56 if m.ID == self || isConnectedSince(transport, since, m.ID) {
57 connectedNum++
58 }
59 }
60 return connectedNum
61 }
62
63
64
65 func longestConnected(tp rafthttp.Transporter, membs []types.ID) (types.ID, bool) {
66 var longest types.ID
67 var oldest time.Time
68 for _, id := range membs {
69 tm := tp.ActiveSince(id)
70 if tm.IsZero() {
71 continue
72 }
73
74 if oldest.IsZero() {
75 oldest = tm
76 longest = id
77 }
78
79 if tm.Before(oldest) {
80 oldest = tm
81 longest = id
82 }
83 }
84 if uint64(longest) == 0 {
85 return longest, false
86 }
87 return longest, true
88 }
89
90 type notifier struct {
91 c chan struct{}
92 err error
93 }
94
95 func newNotifier() *notifier {
96 return ¬ifier{
97 c: make(chan struct{}),
98 }
99 }
100
101 func (nc *notifier) notify(err error) {
102 nc.err = err
103 close(nc.c)
104 }
105
106 func warnOfExpensiveRequest(lg *zap.Logger, warningApplyDuration time.Duration, now time.Time, reqStringer fmt.Stringer, respMsg proto.Message, err error) {
107 if time.Since(now) <= warningApplyDuration {
108 return
109 }
110 var resp string
111 if !isNil(respMsg) {
112 resp = fmt.Sprintf("size:%d", proto.Size(respMsg))
113 }
114 warnOfExpensiveGenericRequest(lg, warningApplyDuration, now, reqStringer, "", resp, err)
115 }
116
117 func warnOfFailedRequest(lg *zap.Logger, now time.Time, reqStringer fmt.Stringer, respMsg proto.Message, err error) {
118 var resp string
119 if !isNil(respMsg) {
120 resp = fmt.Sprintf("size:%d", proto.Size(respMsg))
121 }
122 d := time.Since(now)
123 lg.Warn(
124 "failed to apply request",
125 zap.Duration("took", d),
126 zap.String("request", reqStringer.String()),
127 zap.String("response", resp),
128 zap.Error(err),
129 )
130 }
131
132 func warnOfExpensiveReadOnlyTxnRequest(lg *zap.Logger, warningApplyDuration time.Duration, now time.Time, r *pb.TxnRequest, txnResponse *pb.TxnResponse, err error) {
133 if time.Since(now) <= warningApplyDuration {
134 return
135 }
136 reqStringer := pb.NewLoggableTxnRequest(r)
137 var resp string
138 if !isNil(txnResponse) {
139 var resps []string
140 for _, r := range txnResponse.Responses {
141 switch op := r.Response.(type) {
142 case *pb.ResponseOp_ResponseRange:
143 if op.ResponseRange != nil {
144 resps = append(resps, fmt.Sprintf("range_response_count:%d", len(op.ResponseRange.Kvs)))
145 } else {
146 resps = append(resps, "range_response:nil")
147 }
148 default:
149
150 }
151 }
152 resp = fmt.Sprintf("responses:<%s> size:%d", strings.Join(resps, " "), txnResponse.Size())
153 }
154 warnOfExpensiveGenericRequest(lg, warningApplyDuration, now, reqStringer, "read-only txn ", resp, err)
155 }
156
157 func warnOfExpensiveReadOnlyRangeRequest(lg *zap.Logger, warningApplyDuration time.Duration, now time.Time, reqStringer fmt.Stringer, rangeResponse *pb.RangeResponse, err error) {
158 if time.Since(now) <= warningApplyDuration {
159 return
160 }
161 var resp string
162 if !isNil(rangeResponse) {
163 resp = fmt.Sprintf("range_response_count:%d size:%d", len(rangeResponse.Kvs), rangeResponse.Size())
164 }
165 warnOfExpensiveGenericRequest(lg, warningApplyDuration, now, reqStringer, "read-only range ", resp, err)
166 }
167
168
169 func warnOfExpensiveGenericRequest(lg *zap.Logger, warningApplyDuration time.Duration, now time.Time, reqStringer fmt.Stringer, prefix string, resp string, err error) {
170 lg.Warn(
171 "apply request took too long",
172 zap.Duration("took", time.Since(now)),
173 zap.Duration("expected-duration", warningApplyDuration),
174 zap.String("prefix", prefix),
175 zap.String("request", reqStringer.String()),
176 zap.String("response", resp),
177 zap.Error(err),
178 )
179 slowApplies.Inc()
180 }
181
182 func isNil(msg proto.Message) bool {
183 return msg == nil || reflect.ValueOf(msg).IsNil()
184 }
185
186
187
188
189 type panicAlternativeStringer struct {
190 stringer fmt.Stringer
191 alternative func() string
192 }
193
194 func (n panicAlternativeStringer) String() (s string) {
195 defer func() {
196 if err := recover(); err != nil {
197 s = n.alternative()
198 }
199 }()
200 s = n.stringer.String()
201 return s
202 }
203
View as plain text