blob: 7132807767ebb5e602cb75f0a3be673b069c56d0 [file] [log] [blame]
khenaidooac637102019-01-14 15:44:34 -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
15package clientv3
16
17import (
18 "context"
19 "crypto/tls"
20 "errors"
21 "fmt"
22 "net"
23 "net/url"
24 "strconv"
25 "strings"
26 "sync"
27 "time"
28
29 "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
30
31 "google.golang.org/grpc"
32 "google.golang.org/grpc/codes"
33 "google.golang.org/grpc/credentials"
34 "google.golang.org/grpc/keepalive"
35 "google.golang.org/grpc/metadata"
36 "google.golang.org/grpc/status"
37)
38
39var (
40 ErrNoAvailableEndpoints = errors.New("etcdclient: no available endpoints")
41 ErrOldCluster = errors.New("etcdclient: old cluster version")
42)
43
44// Client provides and manages an etcd v3 client session.
45type Client struct {
46 Cluster
47 KV
48 Lease
49 Watcher
50 Auth
51 Maintenance
52
53 conn *grpc.ClientConn
54 dialerrc chan error
55
56 cfg Config
57 creds *credentials.TransportCredentials
58 balancer *healthBalancer
59 mu *sync.Mutex
60
61 ctx context.Context
62 cancel context.CancelFunc
63
64 // Username is a user name for authentication.
65 Username string
66 // Password is a password for authentication.
67 Password string
68 // tokenCred is an instance of WithPerRPCCredentials()'s argument
69 tokenCred *authTokenCredential
70
71 callOpts []grpc.CallOption
72}
73
74// New creates a new etcdv3 client from a given configuration.
75func New(cfg Config) (*Client, error) {
76 if len(cfg.Endpoints) == 0 {
77 return nil, ErrNoAvailableEndpoints
78 }
79
80 return newClient(&cfg)
81}
82
83// NewCtxClient creates a client with a context but no underlying grpc
84// connection. This is useful for embedded cases that override the
85// service interface implementations and do not need connection management.
86func NewCtxClient(ctx context.Context) *Client {
87 cctx, cancel := context.WithCancel(ctx)
88 return &Client{ctx: cctx, cancel: cancel}
89}
90
91// NewFromURL creates a new etcdv3 client from a URL.
92func NewFromURL(url string) (*Client, error) {
93 return New(Config{Endpoints: []string{url}})
94}
95
96// Close shuts down the client's etcd connections.
97func (c *Client) Close() error {
98 c.cancel()
99 c.Watcher.Close()
100 c.Lease.Close()
101 if c.conn != nil {
102 return toErr(c.ctx, c.conn.Close())
103 }
104 return c.ctx.Err()
105}
106
107// Ctx is a context for "out of band" messages (e.g., for sending
108// "clean up" message when another context is canceled). It is
109// canceled on client Close().
110func (c *Client) Ctx() context.Context { return c.ctx }
111
112// Endpoints lists the registered endpoints for the client.
113func (c *Client) Endpoints() (eps []string) {
114 // copy the slice; protect original endpoints from being changed
115 eps = make([]string, len(c.cfg.Endpoints))
116 copy(eps, c.cfg.Endpoints)
117 return
118}
119
120// SetEndpoints updates client's endpoints.
121func (c *Client) SetEndpoints(eps ...string) {
122 c.mu.Lock()
123 c.cfg.Endpoints = eps
124 c.mu.Unlock()
125 c.balancer.updateAddrs(eps...)
126
127 // updating notifyCh can trigger new connections,
128 // need update addrs if all connections are down
129 // or addrs does not include pinAddr.
130 c.balancer.mu.RLock()
131 update := !hasAddr(c.balancer.addrs, c.balancer.pinAddr)
132 c.balancer.mu.RUnlock()
133 if update {
134 select {
135 case c.balancer.updateAddrsC <- notifyNext:
136 case <-c.balancer.stopc:
137 }
138 }
139}
140
141// Sync synchronizes client's endpoints with the known endpoints from the etcd membership.
142func (c *Client) Sync(ctx context.Context) error {
143 mresp, err := c.MemberList(ctx)
144 if err != nil {
145 return err
146 }
147 var eps []string
148 for _, m := range mresp.Members {
149 eps = append(eps, m.ClientURLs...)
150 }
151 c.SetEndpoints(eps...)
152 return nil
153}
154
155func (c *Client) autoSync() {
156 if c.cfg.AutoSyncInterval == time.Duration(0) {
157 return
158 }
159
160 for {
161 select {
162 case <-c.ctx.Done():
163 return
164 case <-time.After(c.cfg.AutoSyncInterval):
165 ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
166 err := c.Sync(ctx)
167 cancel()
168 if err != nil && err != c.ctx.Err() {
169 logger.Println("Auto sync endpoints failed:", err)
170 }
171 }
172 }
173}
174
175type authTokenCredential struct {
176 token string
177 tokenMu *sync.RWMutex
178}
179
180func (cred authTokenCredential) RequireTransportSecurity() bool {
181 return false
182}
183
184func (cred authTokenCredential) GetRequestMetadata(ctx context.Context, s ...string) (map[string]string, error) {
185 cred.tokenMu.RLock()
186 defer cred.tokenMu.RUnlock()
187 return map[string]string{
188 "token": cred.token,
189 }, nil
190}
191
192func parseEndpoint(endpoint string) (proto string, host string, scheme string) {
193 proto = "tcp"
194 host = endpoint
195 url, uerr := url.Parse(endpoint)
196 if uerr != nil || !strings.Contains(endpoint, "://") {
197 return proto, host, scheme
198 }
199 scheme = url.Scheme
200
201 // strip scheme:// prefix since grpc dials by host
202 host = url.Host
203 switch url.Scheme {
204 case "http", "https":
205 case "unix", "unixs":
206 proto = "unix"
207 host = url.Host + url.Path
208 default:
209 proto, host = "", ""
210 }
211 return proto, host, scheme
212}
213
214func (c *Client) processCreds(scheme string) (creds *credentials.TransportCredentials) {
215 creds = c.creds
216 switch scheme {
217 case "unix":
218 case "http":
219 creds = nil
220 case "https", "unixs":
221 if creds != nil {
222 break
223 }
224 tlsconfig := &tls.Config{}
225 emptyCreds := credentials.NewTLS(tlsconfig)
226 creds = &emptyCreds
227 default:
228 creds = nil
229 }
230 return creds
231}
232
233// dialSetupOpts gives the dial opts prior to any authentication
234func (c *Client) dialSetupOpts(endpoint string, dopts ...grpc.DialOption) (opts []grpc.DialOption) {
235 if c.cfg.DialTimeout > 0 {
236 opts = []grpc.DialOption{grpc.WithTimeout(c.cfg.DialTimeout)}
237 }
238 if c.cfg.DialKeepAliveTime > 0 {
239 params := keepalive.ClientParameters{
240 Time: c.cfg.DialKeepAliveTime,
241 Timeout: c.cfg.DialKeepAliveTimeout,
242 }
243 opts = append(opts, grpc.WithKeepaliveParams(params))
244 }
245 opts = append(opts, dopts...)
246
247 f := func(host string, t time.Duration) (net.Conn, error) {
248 proto, host, _ := parseEndpoint(c.balancer.endpoint(host))
249 if host == "" && endpoint != "" {
250 // dialing an endpoint not in the balancer; use
251 // endpoint passed into dial
252 proto, host, _ = parseEndpoint(endpoint)
253 }
254 if proto == "" {
255 return nil, fmt.Errorf("unknown scheme for %q", host)
256 }
257 select {
258 case <-c.ctx.Done():
259 return nil, c.ctx.Err()
260 default:
261 }
262 dialer := &net.Dialer{Timeout: t}
263 conn, err := dialer.DialContext(c.ctx, proto, host)
264 if err != nil {
265 select {
266 case c.dialerrc <- err:
267 default:
268 }
269 }
270 return conn, err
271 }
272 opts = append(opts, grpc.WithDialer(f))
273
274 creds := c.creds
275 if _, _, scheme := parseEndpoint(endpoint); len(scheme) != 0 {
276 creds = c.processCreds(scheme)
277 }
278 if creds != nil {
279 opts = append(opts, grpc.WithTransportCredentials(*creds))
280 } else {
281 opts = append(opts, grpc.WithInsecure())
282 }
283
284 return opts
285}
286
287// Dial connects to a single endpoint using the client's config.
288func (c *Client) Dial(endpoint string) (*grpc.ClientConn, error) {
289 return c.dial(endpoint)
290}
291
292func (c *Client) getToken(ctx context.Context) error {
293 var err error // return last error in a case of fail
294 var auth *authenticator
295
296 for i := 0; i < len(c.cfg.Endpoints); i++ {
297 endpoint := c.cfg.Endpoints[i]
298 host := getHost(endpoint)
299 // use dial options without dopts to avoid reusing the client balancer
300 auth, err = newAuthenticator(host, c.dialSetupOpts(endpoint), c)
301 if err != nil {
302 continue
303 }
304 defer auth.close()
305
306 var resp *AuthenticateResponse
307 resp, err = auth.authenticate(ctx, c.Username, c.Password)
308 if err != nil {
309 continue
310 }
311
312 c.tokenCred.tokenMu.Lock()
313 c.tokenCred.token = resp.Token
314 c.tokenCred.tokenMu.Unlock()
315
316 return nil
317 }
318
319 return err
320}
321
322func (c *Client) dial(endpoint string, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
323 opts := c.dialSetupOpts(endpoint, dopts...)
324 host := getHost(endpoint)
325 if c.Username != "" && c.Password != "" {
326 c.tokenCred = &authTokenCredential{
327 tokenMu: &sync.RWMutex{},
328 }
329
330 ctx := c.ctx
331 if c.cfg.DialTimeout > 0 {
332 cctx, cancel := context.WithTimeout(ctx, c.cfg.DialTimeout)
333 defer cancel()
334 ctx = cctx
335 }
336
337 err := c.getToken(ctx)
338 if err != nil {
339 if toErr(ctx, err) != rpctypes.ErrAuthNotEnabled {
340 if err == ctx.Err() && ctx.Err() != c.ctx.Err() {
341 err = context.DeadlineExceeded
342 }
343 return nil, err
344 }
345 } else {
346 opts = append(opts, grpc.WithPerRPCCredentials(c.tokenCred))
347 }
348 }
349
350 opts = append(opts, c.cfg.DialOptions...)
351
352 conn, err := grpc.DialContext(c.ctx, host, opts...)
353 if err != nil {
354 return nil, err
355 }
356 return conn, nil
357}
358
359// WithRequireLeader requires client requests to only succeed
360// when the cluster has a leader.
361func WithRequireLeader(ctx context.Context) context.Context {
362 md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
363 return metadata.NewOutgoingContext(ctx, md)
364}
365
366func newClient(cfg *Config) (*Client, error) {
367 if cfg == nil {
368 cfg = &Config{}
369 }
370 var creds *credentials.TransportCredentials
371 if cfg.TLS != nil {
372 c := credentials.NewTLS(cfg.TLS)
373 creds = &c
374 }
375
376 // use a temporary skeleton client to bootstrap first connection
377 baseCtx := context.TODO()
378 if cfg.Context != nil {
379 baseCtx = cfg.Context
380 }
381
382 ctx, cancel := context.WithCancel(baseCtx)
383 client := &Client{
384 conn: nil,
385 dialerrc: make(chan error, 1),
386 cfg: *cfg,
387 creds: creds,
388 ctx: ctx,
389 cancel: cancel,
390 mu: new(sync.Mutex),
391 callOpts: defaultCallOpts,
392 }
393 if cfg.Username != "" && cfg.Password != "" {
394 client.Username = cfg.Username
395 client.Password = cfg.Password
396 }
397 if cfg.MaxCallSendMsgSize > 0 || cfg.MaxCallRecvMsgSize > 0 {
398 if cfg.MaxCallRecvMsgSize > 0 && cfg.MaxCallSendMsgSize > cfg.MaxCallRecvMsgSize {
399 return nil, fmt.Errorf("gRPC message recv limit (%d bytes) must be greater than send limit (%d bytes)", cfg.MaxCallRecvMsgSize, cfg.MaxCallSendMsgSize)
400 }
401 callOpts := []grpc.CallOption{
402 defaultFailFast,
403 defaultMaxCallSendMsgSize,
404 defaultMaxCallRecvMsgSize,
405 }
406 if cfg.MaxCallSendMsgSize > 0 {
407 callOpts[1] = grpc.MaxCallSendMsgSize(cfg.MaxCallSendMsgSize)
408 }
409 if cfg.MaxCallRecvMsgSize > 0 {
410 callOpts[2] = grpc.MaxCallRecvMsgSize(cfg.MaxCallRecvMsgSize)
411 }
412 client.callOpts = callOpts
413 }
414
415 client.balancer = newHealthBalancer(cfg.Endpoints, cfg.DialTimeout, func(ep string) (bool, error) {
416 return grpcHealthCheck(client, ep)
417 })
418
419 // use Endpoints[0] so that for https:// without any tls config given, then
420 // grpc will assume the certificate server name is the endpoint host.
421 conn, err := client.dial(cfg.Endpoints[0], grpc.WithBalancer(client.balancer))
422 if err != nil {
423 client.cancel()
424 client.balancer.Close()
425 return nil, err
426 }
427 client.conn = conn
428
429 // wait for a connection
430 if cfg.DialTimeout > 0 {
431 hasConn := false
432 waitc := time.After(cfg.DialTimeout)
433 select {
434 case <-client.balancer.ready():
435 hasConn = true
436 case <-ctx.Done():
437 case <-waitc:
438 }
439 if !hasConn {
440 err := context.DeadlineExceeded
441 select {
442 case err = <-client.dialerrc:
443 default:
444 }
445 client.cancel()
446 client.balancer.Close()
447 conn.Close()
448 return nil, err
449 }
450 }
451
452 client.Cluster = NewCluster(client)
453 client.KV = NewKV(client)
454 client.Lease = NewLease(client)
455 client.Watcher = NewWatcher(client)
456 client.Auth = NewAuth(client)
457 client.Maintenance = NewMaintenance(client)
458
459 if cfg.RejectOldCluster {
460 if err := client.checkVersion(); err != nil {
461 client.Close()
462 return nil, err
463 }
464 }
465
466 go client.autoSync()
467 return client, nil
468}
469
470func (c *Client) checkVersion() (err error) {
471 var wg sync.WaitGroup
472 errc := make(chan error, len(c.cfg.Endpoints))
473 ctx, cancel := context.WithCancel(c.ctx)
474 if c.cfg.DialTimeout > 0 {
475 ctx, cancel = context.WithTimeout(ctx, c.cfg.DialTimeout)
476 }
477 wg.Add(len(c.cfg.Endpoints))
478 for _, ep := range c.cfg.Endpoints {
479 // if cluster is current, any endpoint gives a recent version
480 go func(e string) {
481 defer wg.Done()
482 resp, rerr := c.Status(ctx, e)
483 if rerr != nil {
484 errc <- rerr
485 return
486 }
487 vs := strings.Split(resp.Version, ".")
488 maj, min := 0, 0
489 if len(vs) >= 2 {
490 maj, _ = strconv.Atoi(vs[0])
491 min, rerr = strconv.Atoi(vs[1])
492 }
493 if maj < 3 || (maj == 3 && min < 2) {
494 rerr = ErrOldCluster
495 }
496 errc <- rerr
497 }(ep)
498 }
499 // wait for success
500 for i := 0; i < len(c.cfg.Endpoints); i++ {
501 if err = <-errc; err == nil {
502 break
503 }
504 }
505 cancel()
506 wg.Wait()
507 return err
508}
509
510// ActiveConnection returns the current in-use connection
511func (c *Client) ActiveConnection() *grpc.ClientConn { return c.conn }
512
513// isHaltErr returns true if the given error and context indicate no forward
514// progress can be made, even after reconnecting.
515func isHaltErr(ctx context.Context, err error) bool {
516 if ctx != nil && ctx.Err() != nil {
517 return true
518 }
519 if err == nil {
520 return false
521 }
522 ev, _ := status.FromError(err)
523 // Unavailable codes mean the system will be right back.
524 // (e.g., can't connect, lost leader)
525 // Treat Internal codes as if something failed, leaving the
526 // system in an inconsistent state, but retrying could make progress.
527 // (e.g., failed in middle of send, corrupted frame)
528 // TODO: are permanent Internal errors possible from grpc?
529 return ev.Code() != codes.Unavailable && ev.Code() != codes.Internal
530}
531
532// isUnavailableErr returns true if the given error is an unavailable error
533func isUnavailableErr(ctx context.Context, err error) bool {
534 if ctx != nil && ctx.Err() != nil {
535 return false
536 }
537 if err == nil {
538 return false
539 }
540 ev, _ := status.FromError(err)
541 // Unavailable codes mean the system will be right back.
542 // (e.g., can't connect, lost leader)
543 return ev.Code() == codes.Unavailable
544}
545
546func toErr(ctx context.Context, err error) error {
547 if err == nil {
548 return nil
549 }
550 err = rpctypes.Error(err)
551 if _, ok := err.(rpctypes.EtcdError); ok {
552 return err
553 }
554 ev, _ := status.FromError(err)
555 code := ev.Code()
556 switch code {
557 case codes.DeadlineExceeded:
558 fallthrough
559 case codes.Canceled:
560 if ctx.Err() != nil {
561 err = ctx.Err()
562 }
563 case codes.Unavailable:
564 case codes.FailedPrecondition:
565 err = grpc.ErrClientConnClosing
566 }
567 return err
568}
569
570func canceledByCaller(stopCtx context.Context, err error) bool {
571 if stopCtx.Err() == nil || err == nil {
572 return false
573 }
574
575 return err == context.Canceled || err == context.DeadlineExceeded
576}