blob: c49e4ba1237aa7bc9833f986144909da593a28cd [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"
khenaidooac637102019-01-14 15:44:34 -050019 "errors"
20 "fmt"
21 "net"
Stephane Barbarie260a5632019-02-26 16:12:49 -050022 "os"
khenaidooac637102019-01-14 15:44:34 -050023 "strconv"
24 "strings"
25 "sync"
26 "time"
27
khenaidood948f772021-08-11 17:49:24 -040028 "github.com/coreos/etcd/clientv3/balancer"
29 "github.com/coreos/etcd/clientv3/balancer/picker"
30 "github.com/coreos/etcd/clientv3/balancer/resolver/endpoint"
31 "github.com/coreos/etcd/clientv3/credentials"
32 "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
33 "github.com/coreos/etcd/pkg/logutil"
34 "github.com/coreos/pkg/capnslog"
Stephane Barbarie260a5632019-02-26 16:12:49 -050035 "github.com/google/uuid"
Stephane Barbarie260a5632019-02-26 16:12:49 -050036 "go.uber.org/zap"
khenaidooac637102019-01-14 15:44:34 -050037 "google.golang.org/grpc"
38 "google.golang.org/grpc/codes"
Scott Baker8461e152019-10-01 14:44:30 -070039 grpccredentials "google.golang.org/grpc/credentials"
khenaidooac637102019-01-14 15:44:34 -050040 "google.golang.org/grpc/keepalive"
khenaidooac637102019-01-14 15:44:34 -050041 "google.golang.org/grpc/status"
42)
43
44var (
45 ErrNoAvailableEndpoints = errors.New("etcdclient: no available endpoints")
46 ErrOldCluster = errors.New("etcdclient: old cluster version")
Stephane Barbarie260a5632019-02-26 16:12:49 -050047
48 roundRobinBalancerName = fmt.Sprintf("etcd-%s", picker.RoundrobinBalanced.String())
khenaidooac637102019-01-14 15:44:34 -050049)
50
khenaidood948f772021-08-11 17:49:24 -040051var (
52 plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "clientv3")
53)
54
Stephane Barbarie260a5632019-02-26 16:12:49 -050055func init() {
56 lg := zap.NewNop()
57 if os.Getenv("ETCD_CLIENT_DEBUG") != "" {
Scott Baker8461e152019-10-01 14:44:30 -070058 lcfg := logutil.DefaultZapLoggerConfig
59 lcfg.Level = zap.NewAtomicLevelAt(zap.DebugLevel)
60
Stephane Barbarie260a5632019-02-26 16:12:49 -050061 var err error
Scott Baker8461e152019-10-01 14:44:30 -070062 lg, err = lcfg.Build() // info level logging
Stephane Barbarie260a5632019-02-26 16:12:49 -050063 if err != nil {
64 panic(err)
65 }
66 }
Scott Baker8461e152019-10-01 14:44:30 -070067
68 // TODO: support custom balancer
Stephane Barbarie260a5632019-02-26 16:12:49 -050069 balancer.RegisterBuilder(balancer.Config{
70 Policy: picker.RoundrobinBalanced,
71 Name: roundRobinBalancerName,
72 Logger: lg,
73 })
74}
75
khenaidooac637102019-01-14 15:44:34 -050076// Client provides and manages an etcd v3 client session.
77type Client struct {
78 Cluster
79 KV
80 Lease
81 Watcher
82 Auth
83 Maintenance
84
Stephane Barbarie260a5632019-02-26 16:12:49 -050085 conn *grpc.ClientConn
khenaidooac637102019-01-14 15:44:34 -050086
Stephane Barbarie260a5632019-02-26 16:12:49 -050087 cfg Config
Scott Baker8461e152019-10-01 14:44:30 -070088 creds grpccredentials.TransportCredentials
Stephane Barbarie260a5632019-02-26 16:12:49 -050089 resolverGroup *endpoint.ResolverGroup
Scott Baker8461e152019-10-01 14:44:30 -070090 mu *sync.RWMutex
khenaidooac637102019-01-14 15:44:34 -050091
92 ctx context.Context
93 cancel context.CancelFunc
94
95 // Username is a user name for authentication.
96 Username string
97 // Password is a password for authentication.
Scott Baker8461e152019-10-01 14:44:30 -070098 Password string
99 authTokenBundle credentials.Bundle
khenaidooac637102019-01-14 15:44:34 -0500100
101 callOpts []grpc.CallOption
Stephane Barbarie260a5632019-02-26 16:12:49 -0500102
103 lg *zap.Logger
khenaidooac637102019-01-14 15:44:34 -0500104}
105
106// New creates a new etcdv3 client from a given configuration.
107func New(cfg Config) (*Client, error) {
108 if len(cfg.Endpoints) == 0 {
109 return nil, ErrNoAvailableEndpoints
110 }
111
112 return newClient(&cfg)
113}
114
115// NewCtxClient creates a client with a context but no underlying grpc
116// connection. This is useful for embedded cases that override the
117// service interface implementations and do not need connection management.
118func NewCtxClient(ctx context.Context) *Client {
119 cctx, cancel := context.WithCancel(ctx)
120 return &Client{ctx: cctx, cancel: cancel}
121}
122
123// NewFromURL creates a new etcdv3 client from a URL.
124func NewFromURL(url string) (*Client, error) {
125 return New(Config{Endpoints: []string{url}})
126}
127
Stephane Barbarie260a5632019-02-26 16:12:49 -0500128// NewFromURLs creates a new etcdv3 client from URLs.
129func NewFromURLs(urls []string) (*Client, error) {
130 return New(Config{Endpoints: urls})
131}
132
khenaidooac637102019-01-14 15:44:34 -0500133// Close shuts down the client's etcd connections.
134func (c *Client) Close() error {
135 c.cancel()
khenaidood948f772021-08-11 17:49:24 -0400136 c.Watcher.Close()
137 c.Lease.Close()
Stephane Barbarie260a5632019-02-26 16:12:49 -0500138 if c.resolverGroup != nil {
139 c.resolverGroup.Close()
140 }
khenaidooac637102019-01-14 15:44:34 -0500141 if c.conn != nil {
142 return toErr(c.ctx, c.conn.Close())
143 }
144 return c.ctx.Err()
145}
146
147// Ctx is a context for "out of band" messages (e.g., for sending
148// "clean up" message when another context is canceled). It is
149// canceled on client Close().
150func (c *Client) Ctx() context.Context { return c.ctx }
151
152// Endpoints lists the registered endpoints for the client.
Scott Baker8461e152019-10-01 14:44:30 -0700153func (c *Client) Endpoints() []string {
khenaidooac637102019-01-14 15:44:34 -0500154 // copy the slice; protect original endpoints from being changed
Scott Baker8461e152019-10-01 14:44:30 -0700155 c.mu.RLock()
156 defer c.mu.RUnlock()
157 eps := make([]string, len(c.cfg.Endpoints))
khenaidooac637102019-01-14 15:44:34 -0500158 copy(eps, c.cfg.Endpoints)
Scott Baker8461e152019-10-01 14:44:30 -0700159 return eps
khenaidooac637102019-01-14 15:44:34 -0500160}
161
162// SetEndpoints updates client's endpoints.
163func (c *Client) SetEndpoints(eps ...string) {
164 c.mu.Lock()
Stephane Barbarie260a5632019-02-26 16:12:49 -0500165 defer c.mu.Unlock()
khenaidooac637102019-01-14 15:44:34 -0500166 c.cfg.Endpoints = eps
Stephane Barbarie260a5632019-02-26 16:12:49 -0500167 c.resolverGroup.SetEndpoints(eps)
khenaidooac637102019-01-14 15:44:34 -0500168}
169
170// Sync synchronizes client's endpoints with the known endpoints from the etcd membership.
171func (c *Client) Sync(ctx context.Context) error {
172 mresp, err := c.MemberList(ctx)
173 if err != nil {
174 return err
175 }
176 var eps []string
177 for _, m := range mresp.Members {
178 eps = append(eps, m.ClientURLs...)
179 }
180 c.SetEndpoints(eps...)
181 return nil
182}
183
184func (c *Client) autoSync() {
185 if c.cfg.AutoSyncInterval == time.Duration(0) {
186 return
187 }
188
189 for {
190 select {
191 case <-c.ctx.Done():
192 return
193 case <-time.After(c.cfg.AutoSyncInterval):
194 ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
195 err := c.Sync(ctx)
196 cancel()
197 if err != nil && err != c.ctx.Err() {
Stephane Barbarie260a5632019-02-26 16:12:49 -0500198 lg.Lvl(4).Infof("Auto sync endpoints failed: %v", err)
khenaidooac637102019-01-14 15:44:34 -0500199 }
200 }
201 }
202}
203
Scott Baker8461e152019-10-01 14:44:30 -0700204func (c *Client) processCreds(scheme string) (creds grpccredentials.TransportCredentials) {
khenaidooac637102019-01-14 15:44:34 -0500205 creds = c.creds
206 switch scheme {
207 case "unix":
208 case "http":
209 creds = nil
210 case "https", "unixs":
211 if creds != nil {
212 break
213 }
Scott Baker8461e152019-10-01 14:44:30 -0700214 creds = credentials.NewBundle(credentials.Config{}).TransportCredentials()
khenaidooac637102019-01-14 15:44:34 -0500215 default:
216 creds = nil
217 }
218 return creds
219}
220
Stephane Barbarie260a5632019-02-26 16:12:49 -0500221// dialSetupOpts gives the dial opts prior to any authentication.
Scott Baker8461e152019-10-01 14:44:30 -0700222func (c *Client) dialSetupOpts(creds grpccredentials.TransportCredentials, dopts ...grpc.DialOption) (opts []grpc.DialOption, err error) {
khenaidooac637102019-01-14 15:44:34 -0500223 if c.cfg.DialKeepAliveTime > 0 {
224 params := keepalive.ClientParameters{
Stephane Barbarie260a5632019-02-26 16:12:49 -0500225 Time: c.cfg.DialKeepAliveTime,
226 Timeout: c.cfg.DialKeepAliveTimeout,
227 PermitWithoutStream: c.cfg.PermitWithoutStream,
khenaidooac637102019-01-14 15:44:34 -0500228 }
229 opts = append(opts, grpc.WithKeepaliveParams(params))
230 }
231 opts = append(opts, dopts...)
232
khenaidood948f772021-08-11 17:49:24 -0400233 dialer := endpoint.Dialer
khenaidooac637102019-01-14 15:44:34 -0500234 if creds != nil {
Scott Baker8461e152019-10-01 14:44:30 -0700235 opts = append(opts, grpc.WithTransportCredentials(creds))
khenaidood948f772021-08-11 17:49:24 -0400236 // gRPC load balancer workaround. See credentials.transportCredential for details.
237 if credsDialer, ok := creds.(TransportCredentialsWithDialer); ok {
238 dialer = credsDialer.Dialer
239 }
khenaidooac637102019-01-14 15:44:34 -0500240 } else {
241 opts = append(opts, grpc.WithInsecure())
242 }
khenaidood948f772021-08-11 17:49:24 -0400243 opts = append(opts, grpc.WithContextDialer(dialer))
khenaidooac637102019-01-14 15:44:34 -0500244
Stephane Barbarie260a5632019-02-26 16:12:49 -0500245 // Interceptor retry and backoff.
246 // TODO: Replace all of clientv3/retry.go with interceptor based retry, or with
247 // https://github.com/grpc/proposal/blob/master/A6-client-retries.md#retry-policy
248 // once it is available.
249 rrBackoff := withBackoff(c.roundRobinQuorumBackoff(defaultBackoffWaitBetween, defaultBackoffJitterFraction))
250 opts = append(opts,
251 // Disable stream retry by default since go-grpc-middleware/retry does not support client streams.
252 // Streams that are safe to retry are enabled individually.
253 grpc.WithStreamInterceptor(c.streamClientInterceptor(c.lg, withMax(0), rrBackoff)),
254 grpc.WithUnaryInterceptor(c.unaryClientInterceptor(c.lg, withMax(defaultUnaryMaxRetries), rrBackoff)),
255 )
256
257 return opts, nil
khenaidooac637102019-01-14 15:44:34 -0500258}
259
260// Dial connects to a single endpoint using the client's config.
Stephane Barbarie260a5632019-02-26 16:12:49 -0500261func (c *Client) Dial(ep string) (*grpc.ClientConn, error) {
khenaidood948f772021-08-11 17:49:24 -0400262 creds, err := c.directDialCreds(ep)
263 if err != nil {
264 return nil, err
265 }
Stephane Barbarie260a5632019-02-26 16:12:49 -0500266 // Use the grpc passthrough resolver to directly dial a single endpoint.
267 // This resolver passes through the 'unix' and 'unixs' endpoints schemes used
268 // by etcd without modification, allowing us to directly dial endpoints and
269 // using the same dial functions that we use for load balancer dialing.
270 return c.dial(fmt.Sprintf("passthrough:///%s", ep), creds)
khenaidooac637102019-01-14 15:44:34 -0500271}
272
273func (c *Client) getToken(ctx context.Context) error {
274 var err error // return last error in a case of fail
275 var auth *authenticator
276
Scott Baker8461e152019-10-01 14:44:30 -0700277 eps := c.Endpoints()
278 for _, ep := range eps {
khenaidooac637102019-01-14 15:44:34 -0500279 // use dial options without dopts to avoid reusing the client balancer
Stephane Barbarie260a5632019-02-26 16:12:49 -0500280 var dOpts []grpc.DialOption
281 _, host, _ := endpoint.ParseEndpoint(ep)
282 target := c.resolverGroup.Target(host)
283 creds := c.dialWithBalancerCreds(ep)
284 dOpts, err = c.dialSetupOpts(creds, c.cfg.DialOptions...)
285 if err != nil {
286 err = fmt.Errorf("failed to configure auth dialer: %v", err)
287 continue
288 }
289 dOpts = append(dOpts, grpc.WithBalancerName(roundRobinBalancerName))
290 auth, err = newAuthenticator(ctx, target, dOpts, c)
khenaidooac637102019-01-14 15:44:34 -0500291 if err != nil {
292 continue
293 }
294 defer auth.close()
295
296 var resp *AuthenticateResponse
297 resp, err = auth.authenticate(ctx, c.Username, c.Password)
298 if err != nil {
Stephane Barbarie260a5632019-02-26 16:12:49 -0500299 // return err without retrying other endpoints
300 if err == rpctypes.ErrAuthNotEnabled {
301 return err
302 }
khenaidooac637102019-01-14 15:44:34 -0500303 continue
304 }
305
Scott Baker8461e152019-10-01 14:44:30 -0700306 c.authTokenBundle.UpdateAuthToken(resp.Token)
khenaidooac637102019-01-14 15:44:34 -0500307 return nil
308 }
309
310 return err
311}
312
Stephane Barbarie260a5632019-02-26 16:12:49 -0500313// dialWithBalancer dials the client's current load balanced resolver group. The scheme of the host
314// of the provided endpoint determines the scheme used for all endpoints of the client connection.
315func (c *Client) dialWithBalancer(ep string, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
316 _, host, _ := endpoint.ParseEndpoint(ep)
317 target := c.resolverGroup.Target(host)
318 creds := c.dialWithBalancerCreds(ep)
319 return c.dial(target, creds, dopts...)
320}
321
322// dial configures and dials any grpc balancer target.
Scott Baker8461e152019-10-01 14:44:30 -0700323func (c *Client) dial(target string, creds grpccredentials.TransportCredentials, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
Stephane Barbarie260a5632019-02-26 16:12:49 -0500324 opts, err := c.dialSetupOpts(creds, dopts...)
325 if err != nil {
326 return nil, fmt.Errorf("failed to configure dialer: %v", err)
327 }
328
khenaidooac637102019-01-14 15:44:34 -0500329 if c.Username != "" && c.Password != "" {
Scott Baker8461e152019-10-01 14:44:30 -0700330 c.authTokenBundle = credentials.NewBundle(credentials.Config{})
khenaidooac637102019-01-14 15:44:34 -0500331
Stephane Barbarie260a5632019-02-26 16:12:49 -0500332 ctx, cancel := c.ctx, func() {}
khenaidooac637102019-01-14 15:44:34 -0500333 if c.cfg.DialTimeout > 0 {
Stephane Barbarie260a5632019-02-26 16:12:49 -0500334 ctx, cancel = context.WithTimeout(ctx, c.cfg.DialTimeout)
khenaidooac637102019-01-14 15:44:34 -0500335 }
336
Stephane Barbarie260a5632019-02-26 16:12:49 -0500337 err = c.getToken(ctx)
khenaidooac637102019-01-14 15:44:34 -0500338 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 }
Stephane Barbarie260a5632019-02-26 16:12:49 -0500343 cancel()
khenaidooac637102019-01-14 15:44:34 -0500344 return nil, err
345 }
346 } else {
Scott Baker8461e152019-10-01 14:44:30 -0700347 opts = append(opts, grpc.WithPerRPCCredentials(c.authTokenBundle.PerRPCCredentials()))
khenaidooac637102019-01-14 15:44:34 -0500348 }
Stephane Barbarie260a5632019-02-26 16:12:49 -0500349 cancel()
khenaidooac637102019-01-14 15:44:34 -0500350 }
351
352 opts = append(opts, c.cfg.DialOptions...)
353
Stephane Barbarie260a5632019-02-26 16:12:49 -0500354 dctx := c.ctx
355 if c.cfg.DialTimeout > 0 {
356 var cancel context.CancelFunc
357 dctx, cancel = context.WithTimeout(c.ctx, c.cfg.DialTimeout)
358 defer cancel() // TODO: Is this right for cases where grpc.WithBlock() is not set on the dial options?
359 }
360
361 conn, err := grpc.DialContext(dctx, target, opts...)
khenaidooac637102019-01-14 15:44:34 -0500362 if err != nil {
363 return nil, err
364 }
365 return conn, nil
366}
367
khenaidood948f772021-08-11 17:49:24 -0400368func (c *Client) directDialCreds(ep string) (grpccredentials.TransportCredentials, error) {
369 _, host, scheme := endpoint.ParseEndpoint(ep)
Stephane Barbarie260a5632019-02-26 16:12:49 -0500370 creds := c.creds
371 if len(scheme) != 0 {
372 creds = c.processCreds(scheme)
373 if creds != nil {
Scott Baker8461e152019-10-01 14:44:30 -0700374 clone := creds.Clone()
Stephane Barbarie260a5632019-02-26 16:12:49 -0500375 // Set the server name must to the endpoint hostname without port since grpc
376 // otherwise attempts to check if x509 cert is valid for the full endpoint
377 // including the scheme and port, which fails.
khenaidood948f772021-08-11 17:49:24 -0400378 overrideServerName, _, err := net.SplitHostPort(host)
379 if err != nil {
380 // Either the host didn't have a port or the host could not be parsed. Either way, continue with the
381 // original host string.
382 overrideServerName = host
383 }
384 clone.OverrideServerName(overrideServerName)
Scott Baker8461e152019-10-01 14:44:30 -0700385 creds = clone
Stephane Barbarie260a5632019-02-26 16:12:49 -0500386 }
387 }
khenaidood948f772021-08-11 17:49:24 -0400388 return creds, nil
Stephane Barbarie260a5632019-02-26 16:12:49 -0500389}
390
Scott Baker8461e152019-10-01 14:44:30 -0700391func (c *Client) dialWithBalancerCreds(ep string) grpccredentials.TransportCredentials {
Stephane Barbarie260a5632019-02-26 16:12:49 -0500392 _, _, scheme := endpoint.ParseEndpoint(ep)
393 creds := c.creds
394 if len(scheme) != 0 {
395 creds = c.processCreds(scheme)
396 }
397 return creds
398}
399
khenaidooac637102019-01-14 15:44:34 -0500400func newClient(cfg *Config) (*Client, error) {
401 if cfg == nil {
402 cfg = &Config{}
403 }
Scott Baker8461e152019-10-01 14:44:30 -0700404 var creds grpccredentials.TransportCredentials
khenaidooac637102019-01-14 15:44:34 -0500405 if cfg.TLS != nil {
Scott Baker8461e152019-10-01 14:44:30 -0700406 creds = credentials.NewBundle(credentials.Config{TLSConfig: cfg.TLS}).TransportCredentials()
khenaidooac637102019-01-14 15:44:34 -0500407 }
408
409 // use a temporary skeleton client to bootstrap first connection
410 baseCtx := context.TODO()
411 if cfg.Context != nil {
412 baseCtx = cfg.Context
413 }
414
415 ctx, cancel := context.WithCancel(baseCtx)
416 client := &Client{
417 conn: nil,
khenaidooac637102019-01-14 15:44:34 -0500418 cfg: *cfg,
419 creds: creds,
420 ctx: ctx,
421 cancel: cancel,
Scott Baker8461e152019-10-01 14:44:30 -0700422 mu: new(sync.RWMutex),
khenaidooac637102019-01-14 15:44:34 -0500423 callOpts: defaultCallOpts,
424 }
Stephane Barbarie260a5632019-02-26 16:12:49 -0500425
William Kurkiandaa6bb22019-03-07 12:26:28 -0500426 lcfg := logutil.DefaultZapLoggerConfig
Stephane Barbarie260a5632019-02-26 16:12:49 -0500427 if cfg.LogConfig != nil {
428 lcfg = *cfg.LogConfig
429 }
430 var err error
431 client.lg, err = lcfg.Build()
432 if err != nil {
433 return nil, err
434 }
435
khenaidooac637102019-01-14 15:44:34 -0500436 if cfg.Username != "" && cfg.Password != "" {
437 client.Username = cfg.Username
438 client.Password = cfg.Password
439 }
440 if cfg.MaxCallSendMsgSize > 0 || cfg.MaxCallRecvMsgSize > 0 {
441 if cfg.MaxCallRecvMsgSize > 0 && cfg.MaxCallSendMsgSize > cfg.MaxCallRecvMsgSize {
442 return nil, fmt.Errorf("gRPC message recv limit (%d bytes) must be greater than send limit (%d bytes)", cfg.MaxCallRecvMsgSize, cfg.MaxCallSendMsgSize)
443 }
444 callOpts := []grpc.CallOption{
445 defaultFailFast,
446 defaultMaxCallSendMsgSize,
447 defaultMaxCallRecvMsgSize,
448 }
449 if cfg.MaxCallSendMsgSize > 0 {
450 callOpts[1] = grpc.MaxCallSendMsgSize(cfg.MaxCallSendMsgSize)
451 }
452 if cfg.MaxCallRecvMsgSize > 0 {
453 callOpts[2] = grpc.MaxCallRecvMsgSize(cfg.MaxCallRecvMsgSize)
454 }
455 client.callOpts = callOpts
456 }
457
Stephane Barbarie260a5632019-02-26 16:12:49 -0500458 // Prepare a 'endpoint://<unique-client-id>/' resolver for the client and create a endpoint target to pass
459 // to dial so the client knows to use this resolver.
460 client.resolverGroup, err = endpoint.NewResolverGroup(fmt.Sprintf("client-%s", uuid.New().String()))
khenaidooac637102019-01-14 15:44:34 -0500461 if err != nil {
462 client.cancel()
khenaidooac637102019-01-14 15:44:34 -0500463 return nil, err
464 }
Stephane Barbarie260a5632019-02-26 16:12:49 -0500465 client.resolverGroup.SetEndpoints(cfg.Endpoints)
khenaidooac637102019-01-14 15:44:34 -0500466
Stephane Barbarie260a5632019-02-26 16:12:49 -0500467 if len(cfg.Endpoints) < 1 {
468 return nil, fmt.Errorf("at least one Endpoint must is required in client config")
khenaidooac637102019-01-14 15:44:34 -0500469 }
Stephane Barbarie260a5632019-02-26 16:12:49 -0500470 dialEndpoint := cfg.Endpoints[0]
471
472 // Use a provided endpoint target so that for https:// without any tls config given, then
473 // grpc will assume the certificate server name is the endpoint host.
474 conn, err := client.dialWithBalancer(dialEndpoint, grpc.WithBalancerName(roundRobinBalancerName))
475 if err != nil {
476 client.cancel()
477 client.resolverGroup.Close()
478 return nil, err
479 }
480 // TODO: With the old grpc balancer interface, we waited until the dial timeout
481 // for the balancer to be ready. Is there an equivalent wait we should do with the new grpc balancer interface?
482 client.conn = conn
khenaidooac637102019-01-14 15:44:34 -0500483
484 client.Cluster = NewCluster(client)
485 client.KV = NewKV(client)
486 client.Lease = NewLease(client)
487 client.Watcher = NewWatcher(client)
488 client.Auth = NewAuth(client)
489 client.Maintenance = NewMaintenance(client)
490
491 if cfg.RejectOldCluster {
492 if err := client.checkVersion(); err != nil {
493 client.Close()
494 return nil, err
495 }
496 }
497
498 go client.autoSync()
499 return client, nil
500}
501
Stephane Barbarie260a5632019-02-26 16:12:49 -0500502// roundRobinQuorumBackoff retries against quorum between each backoff.
503// This is intended for use with a round robin load balancer.
504func (c *Client) roundRobinQuorumBackoff(waitBetween time.Duration, jitterFraction float64) backoffFunc {
505 return func(attempt uint) time.Duration {
506 // after each round robin across quorum, backoff for our wait between duration
507 n := uint(len(c.Endpoints()))
508 quorum := (n/2 + 1)
509 if attempt%quorum == 0 {
William Kurkiandaa6bb22019-03-07 12:26:28 -0500510 c.lg.Debug("backoff", zap.Uint("attempt", attempt), zap.Uint("quorum", quorum), zap.Duration("waitBetween", waitBetween), zap.Float64("jitterFraction", jitterFraction))
Stephane Barbarie260a5632019-02-26 16:12:49 -0500511 return jitterUp(waitBetween, jitterFraction)
512 }
William Kurkiandaa6bb22019-03-07 12:26:28 -0500513 c.lg.Debug("backoff skipped", zap.Uint("attempt", attempt), zap.Uint("quorum", quorum))
Stephane Barbarie260a5632019-02-26 16:12:49 -0500514 return 0
515 }
516}
517
khenaidooac637102019-01-14 15:44:34 -0500518func (c *Client) checkVersion() (err error) {
519 var wg sync.WaitGroup
Scott Baker8461e152019-10-01 14:44:30 -0700520
521 eps := c.Endpoints()
522 errc := make(chan error, len(eps))
khenaidooac637102019-01-14 15:44:34 -0500523 ctx, cancel := context.WithCancel(c.ctx)
524 if c.cfg.DialTimeout > 0 {
Scott Baker8461e152019-10-01 14:44:30 -0700525 cancel()
526 ctx, cancel = context.WithTimeout(c.ctx, c.cfg.DialTimeout)
khenaidooac637102019-01-14 15:44:34 -0500527 }
Scott Baker8461e152019-10-01 14:44:30 -0700528
529 wg.Add(len(eps))
530 for _, ep := range eps {
khenaidooac637102019-01-14 15:44:34 -0500531 // if cluster is current, any endpoint gives a recent version
532 go func(e string) {
533 defer wg.Done()
534 resp, rerr := c.Status(ctx, e)
535 if rerr != nil {
536 errc <- rerr
537 return
538 }
539 vs := strings.Split(resp.Version, ".")
540 maj, min := 0, 0
541 if len(vs) >= 2 {
Scott Baker8461e152019-10-01 14:44:30 -0700542 var serr error
543 if maj, serr = strconv.Atoi(vs[0]); serr != nil {
544 errc <- serr
545 return
546 }
547 if min, serr = strconv.Atoi(vs[1]); serr != nil {
548 errc <- serr
549 return
550 }
khenaidooac637102019-01-14 15:44:34 -0500551 }
552 if maj < 3 || (maj == 3 && min < 2) {
553 rerr = ErrOldCluster
554 }
555 errc <- rerr
556 }(ep)
557 }
558 // wait for success
Scott Baker8461e152019-10-01 14:44:30 -0700559 for range eps {
khenaidooac637102019-01-14 15:44:34 -0500560 if err = <-errc; err == nil {
561 break
562 }
563 }
564 cancel()
565 wg.Wait()
566 return err
567}
568
569// ActiveConnection returns the current in-use connection
570func (c *Client) ActiveConnection() *grpc.ClientConn { return c.conn }
571
572// isHaltErr returns true if the given error and context indicate no forward
573// progress can be made, even after reconnecting.
574func isHaltErr(ctx context.Context, err error) bool {
575 if ctx != nil && ctx.Err() != nil {
576 return true
577 }
578 if err == nil {
579 return false
580 }
581 ev, _ := status.FromError(err)
582 // Unavailable codes mean the system will be right back.
583 // (e.g., can't connect, lost leader)
584 // Treat Internal codes as if something failed, leaving the
585 // system in an inconsistent state, but retrying could make progress.
586 // (e.g., failed in middle of send, corrupted frame)
587 // TODO: are permanent Internal errors possible from grpc?
588 return ev.Code() != codes.Unavailable && ev.Code() != codes.Internal
589}
590
591// isUnavailableErr returns true if the given error is an unavailable error
592func isUnavailableErr(ctx context.Context, err error) bool {
593 if ctx != nil && ctx.Err() != nil {
594 return false
595 }
596 if err == nil {
597 return false
598 }
Scott Baker8461e152019-10-01 14:44:30 -0700599 ev, ok := status.FromError(err)
600 if ok {
601 // Unavailable codes mean the system will be right back.
602 // (e.g., can't connect, lost leader)
603 return ev.Code() == codes.Unavailable
604 }
605 return false
khenaidooac637102019-01-14 15:44:34 -0500606}
607
608func toErr(ctx context.Context, err error) error {
609 if err == nil {
610 return nil
611 }
612 err = rpctypes.Error(err)
613 if _, ok := err.(rpctypes.EtcdError); ok {
614 return err
615 }
Stephane Barbarie260a5632019-02-26 16:12:49 -0500616 if ev, ok := status.FromError(err); ok {
617 code := ev.Code()
618 switch code {
619 case codes.DeadlineExceeded:
620 fallthrough
621 case codes.Canceled:
622 if ctx.Err() != nil {
623 err = ctx.Err()
624 }
khenaidooac637102019-01-14 15:44:34 -0500625 }
khenaidooac637102019-01-14 15:44:34 -0500626 }
627 return err
628}
629
630func canceledByCaller(stopCtx context.Context, err error) bool {
631 if stopCtx.Err() == nil || err == nil {
632 return false
633 }
634
635 return err == context.Canceled || err == context.DeadlineExceeded
636}
Stephane Barbarie260a5632019-02-26 16:12:49 -0500637
638// IsConnCanceled returns true, if error is from a closed gRPC connection.
639// ref. https://github.com/grpc/grpc-go/pull/1854
640func IsConnCanceled(err error) bool {
641 if err == nil {
642 return false
643 }
Scott Baker8461e152019-10-01 14:44:30 -0700644
645 // >= gRPC v1.23.x
Stephane Barbarie260a5632019-02-26 16:12:49 -0500646 s, ok := status.FromError(err)
647 if ok {
648 // connection is canceled or server has already closed the connection
649 return s.Code() == codes.Canceled || s.Message() == "transport is closing"
650 }
Scott Baker8461e152019-10-01 14:44:30 -0700651
Stephane Barbarie260a5632019-02-26 16:12:49 -0500652 // >= gRPC v1.10.x
653 if err == context.Canceled {
654 return true
655 }
Scott Baker8461e152019-10-01 14:44:30 -0700656
Stephane Barbarie260a5632019-02-26 16:12:49 -0500657 // <= gRPC v1.7.x returns 'errors.New("grpc: the client connection is closing")'
658 return strings.Contains(err.Error(), "grpc: the client connection is closing")
659}
khenaidood948f772021-08-11 17:49:24 -0400660
661// TransportCredentialsWithDialer is for a gRPC load balancer workaround. See credentials.transportCredential for details.
662type TransportCredentialsWithDialer interface {
663 grpccredentials.TransportCredentials
664 Dialer(ctx context.Context, dialEp string) (net.Conn, error)
665}