blob: e48a00367c35e723d6eda33516a28f3a3650aadc [file] [log] [blame]
Stephane Barbarie260a5632019-02-26 16:12:49 -05001// Copyright 2016 The etcd Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15// Based on github.com/grpc-ecosystem/go-grpc-middleware/retry, but modified to support the more
16// fine grained error checking required by write-at-most-once retry semantics of etcd.
17
18package clientv3
19
20import (
21 "context"
22 "io"
23 "sync"
24 "time"
25
26 "go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
27 "go.uber.org/zap"
28 "google.golang.org/grpc"
29 "google.golang.org/grpc/codes"
30 "google.golang.org/grpc/metadata"
31)
32
33// unaryClientInterceptor returns a new retrying unary client interceptor.
34//
35// The default configuration of the interceptor is to not retry *at all*. This behaviour can be
36// changed through options (e.g. WithMax) on creation of the interceptor or on call (through grpc.CallOptions).
37func (c *Client) unaryClientInterceptor(logger *zap.Logger, optFuncs ...retryOption) grpc.UnaryClientInterceptor {
38 intOpts := reuseOrNewWithCallOptions(defaultOptions, optFuncs)
39 return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
40 grpcOpts, retryOpts := filterCallOptions(opts)
41 callOpts := reuseOrNewWithCallOptions(intOpts, retryOpts)
42 // short circuit for simplicity, and avoiding allocations.
43 if callOpts.max == 0 {
44 return invoker(ctx, method, req, reply, cc, grpcOpts...)
45 }
46 var lastErr error
47 for attempt := uint(0); attempt < callOpts.max; attempt++ {
48 if err := waitRetryBackoff(ctx, attempt, callOpts); err != nil {
49 return err
50 }
William Kurkiandaa6bb22019-03-07 12:26:28 -050051 logger.Debug(
Stephane Barbarie260a5632019-02-26 16:12:49 -050052 "retrying of unary invoker",
53 zap.String("target", cc.Target()),
54 zap.Uint("attempt", attempt),
55 )
56 lastErr = invoker(ctx, method, req, reply, cc, grpcOpts...)
57 if lastErr == nil {
58 return nil
59 }
60 logger.Warn(
61 "retrying of unary invoker failed",
62 zap.String("target", cc.Target()),
63 zap.Uint("attempt", attempt),
64 zap.Error(lastErr),
65 )
66 if isContextError(lastErr) {
67 if ctx.Err() != nil {
68 // its the context deadline or cancellation.
69 return lastErr
70 }
71 // its the callCtx deadline or cancellation, in which case try again.
72 continue
73 }
74 if callOpts.retryAuth && rpctypes.Error(lastErr) == rpctypes.ErrInvalidAuthToken {
75 gterr := c.getToken(ctx)
76 if gterr != nil {
77 logger.Warn(
78 "retrying of unary invoker failed to fetch new auth token",
79 zap.String("target", cc.Target()),
80 zap.Error(gterr),
81 )
82 return lastErr // return the original error for simplicity
83 }
84 continue
85 }
86 if !isSafeRetry(c.lg, lastErr, callOpts) {
87 return lastErr
88 }
89 }
90 return lastErr
91 }
92}
93
94// streamClientInterceptor returns a new retrying stream client interceptor for server side streaming calls.
95//
96// The default configuration of the interceptor is to not retry *at all*. This behaviour can be
97// changed through options (e.g. WithMax) on creation of the interceptor or on call (through grpc.CallOptions).
98//
99// Retry logic is available *only for ServerStreams*, i.e. 1:n streams, as the internal logic needs
100// to buffer the messages sent by the client. If retry is enabled on any other streams (ClientStreams,
101// BidiStreams), the retry interceptor will fail the call.
102func (c *Client) streamClientInterceptor(logger *zap.Logger, optFuncs ...retryOption) grpc.StreamClientInterceptor {
103 intOpts := reuseOrNewWithCallOptions(defaultOptions, optFuncs)
104 return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
105 grpcOpts, retryOpts := filterCallOptions(opts)
106 callOpts := reuseOrNewWithCallOptions(intOpts, retryOpts)
107 // short circuit for simplicity, and avoiding allocations.
108 if callOpts.max == 0 {
109 return streamer(ctx, desc, cc, method, grpcOpts...)
110 }
111 if desc.ClientStreams {
112 return nil, grpc.Errorf(codes.Unimplemented, "clientv3/retry_interceptor: cannot retry on ClientStreams, set Disable()")
113 }
114 newStreamer, err := streamer(ctx, desc, cc, method, grpcOpts...)
William Kurkiandaa6bb22019-03-07 12:26:28 -0500115 logger.Warn("retry stream intercept", zap.Error(err))
Stephane Barbarie260a5632019-02-26 16:12:49 -0500116 if err != nil {
117 // TODO(mwitkow): Maybe dial and transport errors should be retriable?
118 return nil, err
119 }
120 retryingStreamer := &serverStreamingRetryingStream{
121 client: c,
122 ClientStream: newStreamer,
123 callOpts: callOpts,
124 ctx: ctx,
125 streamerCall: func(ctx context.Context) (grpc.ClientStream, error) {
126 return streamer(ctx, desc, cc, method, grpcOpts...)
127 },
128 }
129 return retryingStreamer, nil
130 }
131}
132
133// type serverStreamingRetryingStream is the implementation of grpc.ClientStream that acts as a
134// proxy to the underlying call. If any of the RecvMsg() calls fail, it will try to reestablish
135// a new ClientStream according to the retry policy.
136type serverStreamingRetryingStream struct {
137 grpc.ClientStream
138 client *Client
139 bufferedSends []interface{} // single message that the client can sen
140 receivedGood bool // indicates whether any prior receives were successful
141 wasClosedSend bool // indicates that CloseSend was closed
142 ctx context.Context
143 callOpts *options
144 streamerCall func(ctx context.Context) (grpc.ClientStream, error)
145 mu sync.RWMutex
146}
147
148func (s *serverStreamingRetryingStream) setStream(clientStream grpc.ClientStream) {
149 s.mu.Lock()
150 s.ClientStream = clientStream
151 s.mu.Unlock()
152}
153
154func (s *serverStreamingRetryingStream) getStream() grpc.ClientStream {
155 s.mu.RLock()
156 defer s.mu.RUnlock()
157 return s.ClientStream
158}
159
160func (s *serverStreamingRetryingStream) SendMsg(m interface{}) error {
161 s.mu.Lock()
162 s.bufferedSends = append(s.bufferedSends, m)
163 s.mu.Unlock()
164 return s.getStream().SendMsg(m)
165}
166
167func (s *serverStreamingRetryingStream) CloseSend() error {
168 s.mu.Lock()
169 s.wasClosedSend = true
170 s.mu.Unlock()
171 return s.getStream().CloseSend()
172}
173
174func (s *serverStreamingRetryingStream) Header() (metadata.MD, error) {
175 return s.getStream().Header()
176}
177
178func (s *serverStreamingRetryingStream) Trailer() metadata.MD {
179 return s.getStream().Trailer()
180}
181
182func (s *serverStreamingRetryingStream) RecvMsg(m interface{}) error {
183 attemptRetry, lastErr := s.receiveMsgAndIndicateRetry(m)
184 if !attemptRetry {
185 return lastErr // success or hard failure
186 }
187 // We start off from attempt 1, because zeroth was already made on normal SendMsg().
188 for attempt := uint(1); attempt < s.callOpts.max; attempt++ {
189 if err := waitRetryBackoff(s.ctx, attempt, s.callOpts); err != nil {
190 return err
191 }
192 newStream, err := s.reestablishStreamAndResendBuffer(s.ctx)
193 if err != nil {
194 // TODO(mwitkow): Maybe dial and transport errors should be retriable?
195 return err
196 }
197 s.setStream(newStream)
198 attemptRetry, lastErr = s.receiveMsgAndIndicateRetry(m)
199 //fmt.Printf("Received message and indicate: %v %v\n", attemptRetry, lastErr)
200 if !attemptRetry {
201 return lastErr
202 }
203 }
204 return lastErr
205}
206
207func (s *serverStreamingRetryingStream) receiveMsgAndIndicateRetry(m interface{}) (bool, error) {
208 s.mu.RLock()
209 wasGood := s.receivedGood
210 s.mu.RUnlock()
211 err := s.getStream().RecvMsg(m)
212 if err == nil || err == io.EOF {
213 s.mu.Lock()
214 s.receivedGood = true
215 s.mu.Unlock()
216 return false, err
217 } else if wasGood {
218 // previous RecvMsg in the stream succeeded, no retry logic should interfere
219 return false, err
220 }
221 if isContextError(err) {
222 if s.ctx.Err() != nil {
223 return false, err
224 }
225 // its the callCtx deadline or cancellation, in which case try again.
226 return true, err
227 }
228 if s.callOpts.retryAuth && rpctypes.Error(err) == rpctypes.ErrInvalidAuthToken {
229 gterr := s.client.getToken(s.ctx)
230 if gterr != nil {
William Kurkiandaa6bb22019-03-07 12:26:28 -0500231 s.client.lg.Warn("retry failed to fetch new auth token", zap.Error(gterr))
Stephane Barbarie260a5632019-02-26 16:12:49 -0500232 return false, err // return the original error for simplicity
233 }
234 return true, err
235
236 }
237 return isSafeRetry(s.client.lg, err, s.callOpts), err
238}
239
240func (s *serverStreamingRetryingStream) reestablishStreamAndResendBuffer(callCtx context.Context) (grpc.ClientStream, error) {
241 s.mu.RLock()
242 bufferedSends := s.bufferedSends
243 s.mu.RUnlock()
244 newStream, err := s.streamerCall(callCtx)
245 if err != nil {
246 return nil, err
247 }
248 for _, msg := range bufferedSends {
249 if err := newStream.SendMsg(msg); err != nil {
250 return nil, err
251 }
252 }
253 if err := newStream.CloseSend(); err != nil {
254 return nil, err
255 }
256 return newStream, nil
257}
258
259func waitRetryBackoff(ctx context.Context, attempt uint, callOpts *options) error {
260 waitTime := time.Duration(0)
261 if attempt > 0 {
262 waitTime = callOpts.backoffFunc(attempt)
263 }
264 if waitTime > 0 {
265 timer := time.NewTimer(waitTime)
266 select {
267 case <-ctx.Done():
268 timer.Stop()
269 return contextErrToGrpcErr(ctx.Err())
270 case <-timer.C:
271 }
272 }
273 return nil
274}
275
276// isSafeRetry returns "true", if request is safe for retry with the given error.
277func isSafeRetry(lg *zap.Logger, err error, callOpts *options) bool {
278 if isContextError(err) {
279 return false
280 }
281 switch callOpts.retryPolicy {
282 case repeatable:
283 return isSafeRetryImmutableRPC(err)
284 case nonRepeatable:
285 return isSafeRetryMutableRPC(err)
286 default:
287 lg.Warn("unrecognized retry policy", zap.String("retryPolicy", callOpts.retryPolicy.String()))
288 return false
289 }
290}
291
292func isContextError(err error) bool {
293 return grpc.Code(err) == codes.DeadlineExceeded || grpc.Code(err) == codes.Canceled
294}
295
296func contextErrToGrpcErr(err error) error {
297 switch err {
298 case context.DeadlineExceeded:
299 return grpc.Errorf(codes.DeadlineExceeded, err.Error())
300 case context.Canceled:
301 return grpc.Errorf(codes.Canceled, err.Error())
302 default:
303 return grpc.Errorf(codes.Unknown, err.Error())
304 }
305}
306
307var (
308 defaultOptions = &options{
309 retryPolicy: nonRepeatable,
310 max: 0, // disable
311 backoffFunc: backoffLinearWithJitter(50*time.Millisecond /*jitter*/, 0.10),
312 retryAuth: true,
313 }
314)
315
316// backoffFunc denotes a family of functions that control the backoff duration between call retries.
317//
318// They are called with an identifier of the attempt, and should return a time the system client should
319// hold off for. If the time returned is longer than the `context.Context.Deadline` of the request
320// the deadline of the request takes precedence and the wait will be interrupted before proceeding
321// with the next iteration.
322type backoffFunc func(attempt uint) time.Duration
323
324// withRetryPolicy sets the retry policy of this call.
325func withRetryPolicy(rp retryPolicy) retryOption {
326 return retryOption{applyFunc: func(o *options) {
327 o.retryPolicy = rp
328 }}
329}
330
331// withAuthRetry sets enables authentication retries.
332func withAuthRetry(retryAuth bool) retryOption {
333 return retryOption{applyFunc: func(o *options) {
334 o.retryAuth = retryAuth
335 }}
336}
337
338// withMax sets the maximum number of retries on this call, or this interceptor.
339func withMax(maxRetries uint) retryOption {
340 return retryOption{applyFunc: func(o *options) {
341 o.max = maxRetries
342 }}
343}
344
345// WithBackoff sets the `BackoffFunc `used to control time between retries.
346func withBackoff(bf backoffFunc) retryOption {
347 return retryOption{applyFunc: func(o *options) {
348 o.backoffFunc = bf
349 }}
350}
351
352type options struct {
353 retryPolicy retryPolicy
354 max uint
355 backoffFunc backoffFunc
356 retryAuth bool
357}
358
359// retryOption is a grpc.CallOption that is local to clientv3's retry interceptor.
360type retryOption struct {
361 grpc.EmptyCallOption // make sure we implement private after() and before() fields so we don't panic.
362 applyFunc func(opt *options)
363}
364
365func reuseOrNewWithCallOptions(opt *options, retryOptions []retryOption) *options {
366 if len(retryOptions) == 0 {
367 return opt
368 }
369 optCopy := &options{}
370 *optCopy = *opt
371 for _, f := range retryOptions {
372 f.applyFunc(optCopy)
373 }
374 return optCopy
375}
376
377func filterCallOptions(callOptions []grpc.CallOption) (grpcOptions []grpc.CallOption, retryOptions []retryOption) {
378 for _, opt := range callOptions {
379 if co, ok := opt.(retryOption); ok {
380 retryOptions = append(retryOptions, co)
381 } else {
382 grpcOptions = append(grpcOptions, opt)
383 }
384 }
385 return grpcOptions, retryOptions
386}
387
388// BackoffLinearWithJitter waits a set period of time, allowing for jitter (fractional adjustment).
389//
390// For example waitBetween=1s and jitter=0.10 can generate waits between 900ms and 1100ms.
391func backoffLinearWithJitter(waitBetween time.Duration, jitterFraction float64) backoffFunc {
392 return func(attempt uint) time.Duration {
393 return jitterUp(waitBetween, jitterFraction)
394 }
395}