1// 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} 669