blob: 445ecfeb1eb23b6376cd6f03f159e5e6c2038494 [file] [log] [blame]
William Kurkianea869482019-04-09 15:16:11 -04001// 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
15package clientv3
16
17import (
18 "context"
19 "crypto/tls"
20 "errors"
21 "fmt"
22 "net"
23 "os"
24 "strconv"
25 "strings"
26 "sync"
27 "time"
28
29 "github.com/google/uuid"
30 "go.etcd.io/etcd/clientv3/balancer"
31 "go.etcd.io/etcd/clientv3/balancer/picker"
32 "go.etcd.io/etcd/clientv3/balancer/resolver/endpoint"
33 "go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
34 "go.etcd.io/etcd/pkg/logutil"
35 "go.uber.org/zap"
36 "google.golang.org/grpc"
37 "google.golang.org/grpc/codes"
38 "google.golang.org/grpc/credentials"
39 "google.golang.org/grpc/keepalive"
40 "google.golang.org/grpc/metadata"
41 "google.golang.org/grpc/status"
42)
43
44var (
45 ErrNoAvailableEndpoints = errors.New("etcdclient: no available endpoints")
46 ErrOldCluster = errors.New("etcdclient: old cluster version")
47
48 roundRobinBalancerName = fmt.Sprintf("etcd-%s", picker.RoundrobinBalanced.String())
49)
50
51func init() {
52 lg := zap.NewNop()
53 if os.Getenv("ETCD_CLIENT_DEBUG") != "" {
54 var err error
55 lg, err = zap.NewProductionConfig().Build() // info level logging
56 if err != nil {
57 panic(err)
58 }
59 }
60 balancer.RegisterBuilder(balancer.Config{
61 Policy: picker.RoundrobinBalanced,
62 Name: roundRobinBalancerName,
63 Logger: lg,
64 })
65}
66
67// Client provides and manages an etcd v3 client session.
68type Client struct {
69 Cluster
70 KV
71 Lease
72 Watcher
73 Auth
74 Maintenance
75
76 conn *grpc.ClientConn
77
78 cfg Config
79 creds *credentials.TransportCredentials
80 balancer balancer.Balancer
81 resolverGroup *endpoint.ResolverGroup
82 mu *sync.Mutex
83
84 ctx context.Context
85 cancel context.CancelFunc
86
87 // Username is a user name for authentication.
88 Username string
89 // Password is a password for authentication.
90 Password string
91 // tokenCred is an instance of WithPerRPCCredentials()'s argument
92 tokenCred *authTokenCredential
93
94 callOpts []grpc.CallOption
95
96 lg *zap.Logger
97}
98
99// New creates a new etcdv3 client from a given configuration.
100func New(cfg Config) (*Client, error) {
101 if len(cfg.Endpoints) == 0 {
102 return nil, ErrNoAvailableEndpoints
103 }
104
105 return newClient(&cfg)
106}
107
108// NewCtxClient creates a client with a context but no underlying grpc
109// connection. This is useful for embedded cases that override the
110// service interface implementations and do not need connection management.
111func NewCtxClient(ctx context.Context) *Client {
112 cctx, cancel := context.WithCancel(ctx)
113 return &Client{ctx: cctx, cancel: cancel}
114}
115
116// NewFromURL creates a new etcdv3 client from a URL.
117func NewFromURL(url string) (*Client, error) {
118 return New(Config{Endpoints: []string{url}})
119}
120
121// NewFromURLs creates a new etcdv3 client from URLs.
122func NewFromURLs(urls []string) (*Client, error) {
123 return New(Config{Endpoints: urls})
124}
125
126// Close shuts down the client's etcd connections.
127func (c *Client) Close() error {
128 c.cancel()
129 c.Watcher.Close()
130 c.Lease.Close()
131 if c.resolverGroup != nil {
132 c.resolverGroup.Close()
133 }
134 if c.conn != nil {
135 return toErr(c.ctx, c.conn.Close())
136 }
137 return c.ctx.Err()
138}
139
140// Ctx is a context for "out of band" messages (e.g., for sending
141// "clean up" message when another context is canceled). It is
142// canceled on client Close().
143func (c *Client) Ctx() context.Context { return c.ctx }
144
145// Endpoints lists the registered endpoints for the client.
146func (c *Client) Endpoints() (eps []string) {
147 // copy the slice; protect original endpoints from being changed
148 eps = make([]string, len(c.cfg.Endpoints))
149 copy(eps, c.cfg.Endpoints)
150 return
151}
152
153// SetEndpoints updates client's endpoints.
154func (c *Client) SetEndpoints(eps ...string) {
155 c.mu.Lock()
156 defer c.mu.Unlock()
157 c.cfg.Endpoints = eps
158 c.resolverGroup.SetEndpoints(eps)
159}
160
161// Sync synchronizes client's endpoints with the known endpoints from the etcd membership.
162func (c *Client) Sync(ctx context.Context) error {
163 mresp, err := c.MemberList(ctx)
164 if err != nil {
165 return err
166 }
167 var eps []string
168 for _, m := range mresp.Members {
169 eps = append(eps, m.ClientURLs...)
170 }
171 c.SetEndpoints(eps...)
172 return nil
173}
174
175func (c *Client) autoSync() {
176 if c.cfg.AutoSyncInterval == time.Duration(0) {
177 return
178 }
179
180 for {
181 select {
182 case <-c.ctx.Done():
183 return
184 case <-time.After(c.cfg.AutoSyncInterval):
185 ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
186 err := c.Sync(ctx)
187 cancel()
188 if err != nil && err != c.ctx.Err() {
189 lg.Lvl(4).Infof("Auto sync endpoints failed: %v", err)
190 }
191 }
192 }
193}
194
195type authTokenCredential struct {
196 token string
197 tokenMu *sync.RWMutex
198}
199
200func (cred authTokenCredential) RequireTransportSecurity() bool {
201 return false
202}
203
204func (cred authTokenCredential) GetRequestMetadata(ctx context.Context, s ...string) (map[string]string, error) {
205 cred.tokenMu.RLock()
206 defer cred.tokenMu.RUnlock()
207 return map[string]string{
208 rpctypes.TokenFieldNameGRPC: cred.token,
209 }, nil
210}
211
212func (c *Client) processCreds(scheme string) (creds *credentials.TransportCredentials) {
213 creds = c.creds
214 switch scheme {
215 case "unix":
216 case "http":
217 creds = nil
218 case "https", "unixs":
219 if creds != nil {
220 break
221 }
222 tlsconfig := &tls.Config{}
223 emptyCreds := credentials.NewTLS(tlsconfig)
224 creds = &emptyCreds
225 default:
226 creds = nil
227 }
228 return creds
229}
230
231// dialSetupOpts gives the dial opts prior to any authentication.
232func (c *Client) dialSetupOpts(creds *credentials.TransportCredentials, dopts ...grpc.DialOption) (opts []grpc.DialOption, err error) {
233 if c.cfg.DialKeepAliveTime > 0 {
234 params := keepalive.ClientParameters{
235 Time: c.cfg.DialKeepAliveTime,
236 Timeout: c.cfg.DialKeepAliveTimeout,
237 PermitWithoutStream: c.cfg.PermitWithoutStream,
238 }
239 opts = append(opts, grpc.WithKeepaliveParams(params))
240 }
241 opts = append(opts, dopts...)
242
243 // Provide a net dialer that supports cancelation and timeout.
244 f := func(dialEp string, t time.Duration) (net.Conn, error) {
245 proto, host, _ := endpoint.ParseEndpoint(dialEp)
246 select {
247 case <-c.ctx.Done():
248 return nil, c.ctx.Err()
249 default:
250 }
251 dialer := &net.Dialer{Timeout: t}
252 return dialer.DialContext(c.ctx, proto, host)
253 }
254 opts = append(opts, grpc.WithDialer(f))
255
256 if creds != nil {
257 opts = append(opts, grpc.WithTransportCredentials(*creds))
258 } else {
259 opts = append(opts, grpc.WithInsecure())
260 }
261
262 // Interceptor retry and backoff.
263 // TODO: Replace all of clientv3/retry.go with interceptor based retry, or with
264 // https://github.com/grpc/proposal/blob/master/A6-client-retries.md#retry-policy
265 // once it is available.
266 rrBackoff := withBackoff(c.roundRobinQuorumBackoff(defaultBackoffWaitBetween, defaultBackoffJitterFraction))
267 opts = append(opts,
268 // Disable stream retry by default since go-grpc-middleware/retry does not support client streams.
269 // Streams that are safe to retry are enabled individually.
270 grpc.WithStreamInterceptor(c.streamClientInterceptor(c.lg, withMax(0), rrBackoff)),
271 grpc.WithUnaryInterceptor(c.unaryClientInterceptor(c.lg, withMax(defaultUnaryMaxRetries), rrBackoff)),
272 )
273
274 return opts, nil
275}
276
277// Dial connects to a single endpoint using the client's config.
278func (c *Client) Dial(ep string) (*grpc.ClientConn, error) {
279 creds := c.directDialCreds(ep)
280 // Use the grpc passthrough resolver to directly dial a single endpoint.
281 // This resolver passes through the 'unix' and 'unixs' endpoints schemes used
282 // by etcd without modification, allowing us to directly dial endpoints and
283 // using the same dial functions that we use for load balancer dialing.
284 return c.dial(fmt.Sprintf("passthrough:///%s", ep), creds)
285}
286
287func (c *Client) getToken(ctx context.Context) error {
288 var err error // return last error in a case of fail
289 var auth *authenticator
290
291 for i := 0; i < len(c.cfg.Endpoints); i++ {
292 ep := c.cfg.Endpoints[i]
293 // use dial options without dopts to avoid reusing the client balancer
294 var dOpts []grpc.DialOption
295 _, host, _ := endpoint.ParseEndpoint(ep)
296 target := c.resolverGroup.Target(host)
297 creds := c.dialWithBalancerCreds(ep)
298 dOpts, err = c.dialSetupOpts(creds, c.cfg.DialOptions...)
299 if err != nil {
300 err = fmt.Errorf("failed to configure auth dialer: %v", err)
301 continue
302 }
303 dOpts = append(dOpts, grpc.WithBalancerName(roundRobinBalancerName))
304 auth, err = newAuthenticator(ctx, target, dOpts, c)
305 if err != nil {
306 continue
307 }
308 defer auth.close()
309
310 var resp *AuthenticateResponse
311 resp, err = auth.authenticate(ctx, c.Username, c.Password)
312 if err != nil {
313 // return err without retrying other endpoints
314 if err == rpctypes.ErrAuthNotEnabled {
315 return err
316 }
317 continue
318 }
319
320 c.tokenCred.tokenMu.Lock()
321 c.tokenCred.token = resp.Token
322 c.tokenCred.tokenMu.Unlock()
323
324 return nil
325 }
326
327 return err
328}
329
330// dialWithBalancer dials the client's current load balanced resolver group. The scheme of the host
331// of the provided endpoint determines the scheme used for all endpoints of the client connection.
332func (c *Client) dialWithBalancer(ep string, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
333 _, host, _ := endpoint.ParseEndpoint(ep)
334 target := c.resolverGroup.Target(host)
335 creds := c.dialWithBalancerCreds(ep)
336 return c.dial(target, creds, dopts...)
337}
338
339// dial configures and dials any grpc balancer target.
340func (c *Client) dial(target string, creds *credentials.TransportCredentials, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
341 opts, err := c.dialSetupOpts(creds, dopts...)
342 if err != nil {
343 return nil, fmt.Errorf("failed to configure dialer: %v", err)
344 }
345
346 if c.Username != "" && c.Password != "" {
347 c.tokenCred = &authTokenCredential{
348 tokenMu: &sync.RWMutex{},
349 }
350
351 ctx, cancel := c.ctx, func() {}
352 if c.cfg.DialTimeout > 0 {
353 ctx, cancel = context.WithTimeout(ctx, c.cfg.DialTimeout)
354 }
355
356 err = c.getToken(ctx)
357 if err != nil {
358 if toErr(ctx, err) != rpctypes.ErrAuthNotEnabled {
359 if err == ctx.Err() && ctx.Err() != c.ctx.Err() {
360 err = context.DeadlineExceeded
361 }
362 cancel()
363 return nil, err
364 }
365 } else {
366 opts = append(opts, grpc.WithPerRPCCredentials(c.tokenCred))
367 }
368 cancel()
369 }
370
371 opts = append(opts, c.cfg.DialOptions...)
372
373 dctx := c.ctx
374 if c.cfg.DialTimeout > 0 {
375 var cancel context.CancelFunc
376 dctx, cancel = context.WithTimeout(c.ctx, c.cfg.DialTimeout)
377 defer cancel() // TODO: Is this right for cases where grpc.WithBlock() is not set on the dial options?
378 }
379
380 conn, err := grpc.DialContext(dctx, target, opts...)
381 if err != nil {
382 return nil, err
383 }
384 return conn, nil
385}
386
387func (c *Client) directDialCreds(ep string) *credentials.TransportCredentials {
388 _, hostPort, scheme := endpoint.ParseEndpoint(ep)
389 creds := c.creds
390 if len(scheme) != 0 {
391 creds = c.processCreds(scheme)
392 if creds != nil {
393 c := *creds
394 clone := c.Clone()
395 // Set the server name must to the endpoint hostname without port since grpc
396 // otherwise attempts to check if x509 cert is valid for the full endpoint
397 // including the scheme and port, which fails.
398 host, _ := endpoint.ParseHostPort(hostPort)
399 clone.OverrideServerName(host)
400 creds = &clone
401 }
402 }
403 return creds
404}
405
406func (c *Client) dialWithBalancerCreds(ep string) *credentials.TransportCredentials {
407 _, _, scheme := endpoint.ParseEndpoint(ep)
408 creds := c.creds
409 if len(scheme) != 0 {
410 creds = c.processCreds(scheme)
411 }
412 return creds
413}
414
415// WithRequireLeader requires client requests to only succeed
416// when the cluster has a leader.
417func WithRequireLeader(ctx context.Context) context.Context {
418 md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
419 return metadata.NewOutgoingContext(ctx, md)
420}
421
422func newClient(cfg *Config) (*Client, error) {
423 if cfg == nil {
424 cfg = &Config{}
425 }
426 var creds *credentials.TransportCredentials
427 if cfg.TLS != nil {
428 c := credentials.NewTLS(cfg.TLS)
429 creds = &c
430 }
431
432 // use a temporary skeleton client to bootstrap first connection
433 baseCtx := context.TODO()
434 if cfg.Context != nil {
435 baseCtx = cfg.Context
436 }
437
438 ctx, cancel := context.WithCancel(baseCtx)
439 client := &Client{
440 conn: nil,
441 cfg: *cfg,
442 creds: creds,
443 ctx: ctx,
444 cancel: cancel,
445 mu: new(sync.Mutex),
446 callOpts: defaultCallOpts,
447 }
448
449 lcfg := logutil.DefaultZapLoggerConfig
450 if cfg.LogConfig != nil {
451 lcfg = *cfg.LogConfig
452 }
453 var err error
454 client.lg, err = lcfg.Build()
455 if err != nil {
456 return nil, err
457 }
458
459 if cfg.Username != "" && cfg.Password != "" {
460 client.Username = cfg.Username
461 client.Password = cfg.Password
462 }
463 if cfg.MaxCallSendMsgSize > 0 || cfg.MaxCallRecvMsgSize > 0 {
464 if cfg.MaxCallRecvMsgSize > 0 && cfg.MaxCallSendMsgSize > cfg.MaxCallRecvMsgSize {
465 return nil, fmt.Errorf("gRPC message recv limit (%d bytes) must be greater than send limit (%d bytes)", cfg.MaxCallRecvMsgSize, cfg.MaxCallSendMsgSize)
466 }
467 callOpts := []grpc.CallOption{
468 defaultFailFast,
469 defaultMaxCallSendMsgSize,
470 defaultMaxCallRecvMsgSize,
471 }
472 if cfg.MaxCallSendMsgSize > 0 {
473 callOpts[1] = grpc.MaxCallSendMsgSize(cfg.MaxCallSendMsgSize)
474 }
475 if cfg.MaxCallRecvMsgSize > 0 {
476 callOpts[2] = grpc.MaxCallRecvMsgSize(cfg.MaxCallRecvMsgSize)
477 }
478 client.callOpts = callOpts
479 }
480
481 // Prepare a 'endpoint://<unique-client-id>/' resolver for the client and create a endpoint target to pass
482 // to dial so the client knows to use this resolver.
483 client.resolverGroup, err = endpoint.NewResolverGroup(fmt.Sprintf("client-%s", uuid.New().String()))
484 if err != nil {
485 client.cancel()
486 return nil, err
487 }
488 client.resolverGroup.SetEndpoints(cfg.Endpoints)
489
490 if len(cfg.Endpoints) < 1 {
491 return nil, fmt.Errorf("at least one Endpoint must is required in client config")
492 }
493 dialEndpoint := cfg.Endpoints[0]
494
495 // Use a provided endpoint target so that for https:// without any tls config given, then
496 // grpc will assume the certificate server name is the endpoint host.
497 conn, err := client.dialWithBalancer(dialEndpoint, grpc.WithBalancerName(roundRobinBalancerName))
498 if err != nil {
499 client.cancel()
500 client.resolverGroup.Close()
501 return nil, err
502 }
503 // TODO: With the old grpc balancer interface, we waited until the dial timeout
504 // for the balancer to be ready. Is there an equivalent wait we should do with the new grpc balancer interface?
505 client.conn = conn
506
507 client.Cluster = NewCluster(client)
508 client.KV = NewKV(client)
509 client.Lease = NewLease(client)
510 client.Watcher = NewWatcher(client)
511 client.Auth = NewAuth(client)
512 client.Maintenance = NewMaintenance(client)
513
514 if cfg.RejectOldCluster {
515 if err := client.checkVersion(); err != nil {
516 client.Close()
517 return nil, err
518 }
519 }
520
521 go client.autoSync()
522 return client, nil
523}
524
525// roundRobinQuorumBackoff retries against quorum between each backoff.
526// This is intended for use with a round robin load balancer.
527func (c *Client) roundRobinQuorumBackoff(waitBetween time.Duration, jitterFraction float64) backoffFunc {
528 return func(attempt uint) time.Duration {
529 // after each round robin across quorum, backoff for our wait between duration
530 n := uint(len(c.Endpoints()))
531 quorum := (n/2 + 1)
532 if attempt%quorum == 0 {
533 c.lg.Debug("backoff", zap.Uint("attempt", attempt), zap.Uint("quorum", quorum), zap.Duration("waitBetween", waitBetween), zap.Float64("jitterFraction", jitterFraction))
534 return jitterUp(waitBetween, jitterFraction)
535 }
536 c.lg.Debug("backoff skipped", zap.Uint("attempt", attempt), zap.Uint("quorum", quorum))
537 return 0
538 }
539}
540
541func (c *Client) checkVersion() (err error) {
542 var wg sync.WaitGroup
543 errc := make(chan error, len(c.cfg.Endpoints))
544 ctx, cancel := context.WithCancel(c.ctx)
545 if c.cfg.DialTimeout > 0 {
546 ctx, cancel = context.WithTimeout(ctx, c.cfg.DialTimeout)
547 }
548 wg.Add(len(c.cfg.Endpoints))
549 for _, ep := range c.cfg.Endpoints {
550 // if cluster is current, any endpoint gives a recent version
551 go func(e string) {
552 defer wg.Done()
553 resp, rerr := c.Status(ctx, e)
554 if rerr != nil {
555 errc <- rerr
556 return
557 }
558 vs := strings.Split(resp.Version, ".")
559 maj, min := 0, 0
560 if len(vs) >= 2 {
561 maj, _ = strconv.Atoi(vs[0])
562 min, rerr = strconv.Atoi(vs[1])
563 }
564 if maj < 3 || (maj == 3 && min < 2) {
565 rerr = ErrOldCluster
566 }
567 errc <- rerr
568 }(ep)
569 }
570 // wait for success
571 for i := 0; i < len(c.cfg.Endpoints); i++ {
572 if err = <-errc; err == nil {
573 break
574 }
575 }
576 cancel()
577 wg.Wait()
578 return err
579}
580
581// ActiveConnection returns the current in-use connection
582func (c *Client) ActiveConnection() *grpc.ClientConn { return c.conn }
583
584// isHaltErr returns true if the given error and context indicate no forward
585// progress can be made, even after reconnecting.
586func isHaltErr(ctx context.Context, err error) bool {
587 if ctx != nil && ctx.Err() != nil {
588 return true
589 }
590 if err == nil {
591 return false
592 }
593 ev, _ := status.FromError(err)
594 // Unavailable codes mean the system will be right back.
595 // (e.g., can't connect, lost leader)
596 // Treat Internal codes as if something failed, leaving the
597 // system in an inconsistent state, but retrying could make progress.
598 // (e.g., failed in middle of send, corrupted frame)
599 // TODO: are permanent Internal errors possible from grpc?
600 return ev.Code() != codes.Unavailable && ev.Code() != codes.Internal
601}
602
603// isUnavailableErr returns true if the given error is an unavailable error
604func isUnavailableErr(ctx context.Context, err error) bool {
605 if ctx != nil && ctx.Err() != nil {
606 return false
607 }
608 if err == nil {
609 return false
610 }
611 ev, _ := status.FromError(err)
612 // Unavailable codes mean the system will be right back.
613 // (e.g., can't connect, lost leader)
614 return ev.Code() == codes.Unavailable
615}
616
617func toErr(ctx context.Context, err error) error {
618 if err == nil {
619 return nil
620 }
621 err = rpctypes.Error(err)
622 if _, ok := err.(rpctypes.EtcdError); ok {
623 return err
624 }
625 if ev, ok := status.FromError(err); ok {
626 code := ev.Code()
627 switch code {
628 case codes.DeadlineExceeded:
629 fallthrough
630 case codes.Canceled:
631 if ctx.Err() != nil {
632 err = ctx.Err()
633 }
634 case codes.Unavailable:
635 case codes.FailedPrecondition:
636 err = grpc.ErrClientConnClosing
637 }
638 }
639 return err
640}
641
642func canceledByCaller(stopCtx context.Context, err error) bool {
643 if stopCtx.Err() == nil || err == nil {
644 return false
645 }
646
647 return err == context.Canceled || err == context.DeadlineExceeded
648}
649
650// IsConnCanceled returns true, if error is from a closed gRPC connection.
651// ref. https://github.com/grpc/grpc-go/pull/1854
652func IsConnCanceled(err error) bool {
653 if err == nil {
654 return false
655 }
656 // >= gRPC v1.10.x
657 s, ok := status.FromError(err)
658 if ok {
659 // connection is canceled or server has already closed the connection
660 return s.Code() == codes.Canceled || s.Message() == "transport is closing"
661 }
662 // >= gRPC v1.10.x
663 if err == context.Canceled {
664 return true
665 }
666 // <= gRPC v1.7.x returns 'errors.New("grpc: the client connection is closing")'
667 return strings.Contains(err.Error(), "grpc: the client connection is closing")
668}