[VOL-5486] Fix deprecated versions
Change-Id: I3e03ea246020547ae75fa92ce8cf5cbba7e8f3bb
Signed-off-by: Abhay Kumar <abhay.kumar@radisys.com>
diff --git a/vendor/go.etcd.io/etcd/client/v3/client.go b/vendor/go.etcd.io/etcd/client/v3/client.go
new file mode 100644
index 0000000..24f5988
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/client/v3/client.go
@@ -0,0 +1,655 @@
+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/coreos/go-semver/semver"
+ "go.uber.org/zap"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ grpccredentials "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/grpc/keepalive"
+ "google.golang.org/grpc/status"
+
+ "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
+ "go.etcd.io/etcd/api/v3/version"
+ "go.etcd.io/etcd/client/pkg/v3/logutil"
+ "go.etcd.io/etcd/client/pkg/v3/verify"
+ "go.etcd.io/etcd/client/v3/credentials"
+ "go.etcd.io/etcd/client/v3/internal/endpoint"
+ "go.etcd.io/etcd/client/v3/internal/resolver"
+)
+
+var (
+ ErrNoAvailableEndpoints = errors.New("etcdclient: no available endpoints")
+ ErrOldCluster = errors.New("etcdclient: old cluster version")
+)
+
+// Client provides and manages an etcd v3 client session.
+type Client struct {
+ Cluster
+ KV
+ Lease
+ Watcher
+ Auth
+ Maintenance
+
+ conn *grpc.ClientConn
+
+ cfg Config
+ creds grpccredentials.TransportCredentials
+ resolver *resolver.EtcdManualResolver
+
+ epMu *sync.RWMutex
+ endpoints []string
+
+ ctx context.Context
+ cancel context.CancelFunc
+
+ // Username is a user name for authentication.
+ Username string
+ // Password is a password for authentication.
+ Password string
+ authTokenBundle credentials.PerRPCCredentialsBundle
+
+ callOpts []grpc.CallOption
+
+ lgMu *sync.RWMutex
+ lg *zap.Logger
+}
+
+// New creates a new etcdv3 client from a given configuration.
+func New(cfg Config) (*Client, error) {
+ if len(cfg.Endpoints) == 0 {
+ return nil, ErrNoAvailableEndpoints
+ }
+
+ return newClient(&cfg)
+}
+
+// NewCtxClient creates a client with a context but no underlying grpc
+// connection. This is useful for embedded cases that override the
+// service interface implementations and do not need connection management.
+func NewCtxClient(ctx context.Context, opts ...Option) *Client {
+ cctx, cancel := context.WithCancel(ctx)
+ c := &Client{ctx: cctx, cancel: cancel, lgMu: new(sync.RWMutex), epMu: new(sync.RWMutex)}
+ for _, opt := range opts {
+ opt(c)
+ }
+ if c.lg == nil {
+ c.lg = zap.NewNop()
+ }
+ return c
+}
+
+// Option is a function type that can be passed as argument to NewCtxClient to configure client
+type Option func(*Client)
+
+// NewFromURL creates a new etcdv3 client from a URL.
+func NewFromURL(url string) (*Client, error) {
+ return New(Config{Endpoints: []string{url}})
+}
+
+// NewFromURLs creates a new etcdv3 client from URLs.
+func NewFromURLs(urls []string) (*Client, error) {
+ return New(Config{Endpoints: urls})
+}
+
+// WithZapLogger is a NewCtxClient option that overrides the logger
+func WithZapLogger(lg *zap.Logger) Option {
+ return func(c *Client) {
+ c.lg = lg
+ }
+}
+
+// WithLogger overrides the logger.
+//
+// Deprecated: Please use WithZapLogger or Logger field in clientv3.Config
+//
+// Does not changes grpcLogger, that can be explicitly configured
+// using grpc_zap.ReplaceGrpcLoggerV2(..) method.
+func (c *Client) WithLogger(lg *zap.Logger) *Client {
+ c.lgMu.Lock()
+ c.lg = lg
+ c.lgMu.Unlock()
+ return c
+}
+
+// GetLogger gets the logger.
+// NOTE: This method is for internal use of etcd-client library and should not be used as general-purpose logger.
+func (c *Client) GetLogger() *zap.Logger {
+ c.lgMu.RLock()
+ l := c.lg
+ c.lgMu.RUnlock()
+ return l
+}
+
+// Close shuts down the client's etcd connections.
+func (c *Client) Close() error {
+ c.cancel()
+ if c.Watcher != nil {
+ c.Watcher.Close()
+ }
+ if c.Lease != nil {
+ c.Lease.Close()
+ }
+ if c.conn != nil {
+ return ContextError(c.ctx, c.conn.Close())
+ }
+ return c.ctx.Err()
+}
+
+// Ctx is a context for "out of band" messages (e.g., for sending
+// "clean up" message when another context is canceled). It is
+// canceled on client Close().
+func (c *Client) Ctx() context.Context { return c.ctx }
+
+// Endpoints lists the registered endpoints for the client.
+func (c *Client) Endpoints() []string {
+ // copy the slice; protect original endpoints from being changed
+ c.epMu.RLock()
+ defer c.epMu.RUnlock()
+ eps := make([]string, len(c.endpoints))
+ copy(eps, c.endpoints)
+ return eps
+}
+
+// SetEndpoints updates client's endpoints.
+func (c *Client) SetEndpoints(eps ...string) {
+ c.epMu.Lock()
+ defer c.epMu.Unlock()
+ c.endpoints = eps
+
+ c.resolver.SetEndpoints(eps)
+}
+
+// Sync synchronizes client's endpoints with the known endpoints from the etcd membership.
+func (c *Client) Sync(ctx context.Context) error {
+ mresp, err := c.MemberList(ctx)
+ if err != nil {
+ return err
+ }
+ var eps []string
+ for _, m := range mresp.Members {
+ if len(m.Name) != 0 && !m.IsLearner {
+ eps = append(eps, m.ClientURLs...)
+ }
+ }
+ // The linearizable `MemberList` returned successfully, so the
+ // endpoints shouldn't be empty.
+ verify.Verify(func() {
+ if len(eps) == 0 {
+ panic("empty endpoints returned from etcd cluster")
+ }
+ })
+ c.SetEndpoints(eps...)
+ c.lg.Debug("set etcd endpoints by autoSync", zap.Strings("endpoints", eps))
+ return nil
+}
+
+func (c *Client) autoSync() {
+ if c.cfg.AutoSyncInterval == time.Duration(0) {
+ return
+ }
+
+ for {
+ select {
+ case <-c.ctx.Done():
+ return
+ case <-time.After(c.cfg.AutoSyncInterval):
+ ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
+ err := c.Sync(ctx)
+ cancel()
+ if err != nil && !errors.Is(err, c.ctx.Err()) {
+ c.lg.Info("Auto sync endpoints failed.", zap.Error(err))
+ }
+ }
+ }
+}
+
+// dialSetupOpts gives the dial opts prior to any authentication.
+func (c *Client) dialSetupOpts(creds grpccredentials.TransportCredentials, dopts ...grpc.DialOption) []grpc.DialOption {
+ var opts []grpc.DialOption
+
+ if c.cfg.DialKeepAliveTime > 0 {
+ params := keepalive.ClientParameters{
+ Time: c.cfg.DialKeepAliveTime,
+ Timeout: c.cfg.DialKeepAliveTimeout,
+ PermitWithoutStream: c.cfg.PermitWithoutStream,
+ }
+ opts = append(opts, grpc.WithKeepaliveParams(params))
+ }
+ opts = append(opts, dopts...)
+
+ if creds != nil {
+ opts = append(opts, grpc.WithTransportCredentials(creds))
+ } else {
+ opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+ }
+
+ unaryMaxRetries := defaultUnaryMaxRetries
+ if c.cfg.MaxUnaryRetries > 0 {
+ unaryMaxRetries = c.cfg.MaxUnaryRetries
+ }
+
+ backoffWaitBetween := defaultBackoffWaitBetween
+ if c.cfg.BackoffWaitBetween > 0 {
+ backoffWaitBetween = c.cfg.BackoffWaitBetween
+ }
+
+ backoffJitterFraction := defaultBackoffJitterFraction
+ if c.cfg.BackoffJitterFraction > 0 {
+ backoffJitterFraction = c.cfg.BackoffJitterFraction
+ }
+
+ // Interceptor retry and backoff.
+ // TODO: Replace all of clientv3/retry.go with RetryPolicy:
+ // https://github.com/grpc/grpc-proto/blob/cdd9ed5c3d3f87aef62f373b93361cf7bddc620d/grpc/service_config/service_config.proto#L130
+ rrBackoff := withBackoff(c.roundRobinQuorumBackoff(backoffWaitBetween, backoffJitterFraction))
+ opts = append(opts,
+ // Disable stream retry by default since go-grpc-middleware/retry does not support client streams.
+ // Streams that are safe to retry are enabled individually.
+ grpc.WithStreamInterceptor(c.streamClientInterceptor(withMax(0), rrBackoff)),
+ grpc.WithUnaryInterceptor(c.unaryClientInterceptor(withMax(unaryMaxRetries), rrBackoff)),
+ )
+
+ return opts
+}
+
+// Dial connects to a single endpoint using the client's config.
+func (c *Client) Dial(ep string) (*grpc.ClientConn, error) {
+ creds := c.credentialsForEndpoint(ep)
+
+ // Using ad-hoc created resolver, to guarantee only explicitly given
+ // endpoint is used.
+ return c.dial(creds, grpc.WithResolvers(resolver.New(ep)))
+}
+
+func (c *Client) getToken(ctx context.Context) error {
+ var err error // return last error in a case of fail
+
+ if c.Username == "" || c.Password == "" {
+ return nil
+ }
+
+ resp, err := c.Auth.Authenticate(ctx, c.Username, c.Password)
+ if err != nil {
+ if errors.Is(err, rpctypes.ErrAuthNotEnabled) {
+ c.authTokenBundle.UpdateAuthToken("")
+ return nil
+ }
+ return err
+ }
+ c.authTokenBundle.UpdateAuthToken(resp.Token)
+ return nil
+}
+
+// dialWithBalancer dials the client's current load balanced resolver group. The scheme of the host
+// of the provided endpoint determines the scheme used for all endpoints of the client connection.
+func (c *Client) dialWithBalancer(dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
+ creds := c.credentialsForEndpoint(c.Endpoints()[0])
+ opts := append(dopts, grpc.WithResolvers(c.resolver))
+ return c.dial(creds, opts...)
+}
+
+// dial configures and dials any grpc balancer target.
+func (c *Client) dial(creds grpccredentials.TransportCredentials, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
+ opts := c.dialSetupOpts(creds, dopts...)
+
+ if c.authTokenBundle != nil {
+ opts = append(opts, grpc.WithPerRPCCredentials(c.authTokenBundle.PerRPCCredentials()))
+ }
+
+ opts = append(opts, c.cfg.DialOptions...)
+
+ dctx := c.ctx
+ if c.cfg.DialTimeout > 0 {
+ var cancel context.CancelFunc
+ dctx, cancel = context.WithTimeout(c.ctx, c.cfg.DialTimeout)
+ defer cancel() // TODO: Is this right for cases where grpc.WithBlock() is not set on the dial options?
+ }
+ target := fmt.Sprintf("%s://%p/%s", resolver.Schema, c, authority(c.endpoints[0]))
+ conn, err := grpc.DialContext(dctx, target, opts...)
+ if err != nil {
+ return nil, err
+ }
+ return conn, nil
+}
+
+func authority(endpoint string) string {
+ spl := strings.SplitN(endpoint, "://", 2)
+ if len(spl) < 2 {
+ if strings.HasPrefix(endpoint, "unix:") {
+ return endpoint[len("unix:"):]
+ }
+ if strings.HasPrefix(endpoint, "unixs:") {
+ return endpoint[len("unixs:"):]
+ }
+ return endpoint
+ }
+ return spl[1]
+}
+
+func (c *Client) credentialsForEndpoint(ep string) grpccredentials.TransportCredentials {
+ r := endpoint.RequiresCredentials(ep)
+ switch r {
+ case endpoint.CredsDrop:
+ return nil
+ case endpoint.CredsOptional:
+ return c.creds
+ case endpoint.CredsRequire:
+ if c.creds != nil {
+ return c.creds
+ }
+ return credentials.NewTransportCredential(nil)
+ default:
+ panic(fmt.Errorf("unsupported CredsRequirement: %v", r))
+ }
+}
+
+func newClient(cfg *Config) (*Client, error) {
+ if cfg == nil {
+ cfg = &Config{}
+ }
+ var creds grpccredentials.TransportCredentials
+ if cfg.TLS != nil {
+ creds = credentials.NewTransportCredential(cfg.TLS)
+ }
+
+ // use a temporary skeleton client to bootstrap first connection
+ baseCtx := context.TODO()
+ if cfg.Context != nil {
+ baseCtx = cfg.Context
+ }
+
+ ctx, cancel := context.WithCancel(baseCtx)
+ client := &Client{
+ conn: nil,
+ cfg: *cfg,
+ creds: creds,
+ ctx: ctx,
+ cancel: cancel,
+ epMu: new(sync.RWMutex),
+ callOpts: defaultCallOpts,
+ lgMu: new(sync.RWMutex),
+ }
+
+ var err error
+ if cfg.Logger != nil {
+ client.lg = cfg.Logger
+ } else if cfg.LogConfig != nil {
+ client.lg, err = cfg.LogConfig.Build()
+ } else {
+ client.lg, err = logutil.CreateDefaultZapLogger(etcdClientDebugLevel())
+ if client.lg != nil {
+ client.lg = client.lg.Named("etcd-client")
+ }
+ }
+ if err != nil {
+ return nil, err
+ }
+
+ if cfg.Username != "" && cfg.Password != "" {
+ client.Username = cfg.Username
+ client.Password = cfg.Password
+ client.authTokenBundle = credentials.NewPerRPCCredentialBundle()
+ }
+ if cfg.MaxCallSendMsgSize > 0 || cfg.MaxCallRecvMsgSize > 0 {
+ if cfg.MaxCallRecvMsgSize > 0 && cfg.MaxCallSendMsgSize > cfg.MaxCallRecvMsgSize {
+ return nil, fmt.Errorf("gRPC message recv limit (%d bytes) must be greater than send limit (%d bytes)", cfg.MaxCallRecvMsgSize, cfg.MaxCallSendMsgSize)
+ }
+ callOpts := []grpc.CallOption{
+ defaultWaitForReady,
+ defaultMaxCallSendMsgSize,
+ defaultMaxCallRecvMsgSize,
+ }
+ if cfg.MaxCallSendMsgSize > 0 {
+ callOpts[1] = grpc.MaxCallSendMsgSize(cfg.MaxCallSendMsgSize)
+ }
+ if cfg.MaxCallRecvMsgSize > 0 {
+ callOpts[2] = grpc.MaxCallRecvMsgSize(cfg.MaxCallRecvMsgSize)
+ }
+ client.callOpts = callOpts
+ }
+
+ client.resolver = resolver.New(cfg.Endpoints...)
+
+ if len(cfg.Endpoints) < 1 {
+ client.cancel()
+ return nil, errors.New("at least one Endpoint is required in client config")
+ }
+ client.SetEndpoints(cfg.Endpoints...)
+
+ // Use a provided endpoint target so that for https:// without any tls config given, then
+ // grpc will assume the certificate server name is the endpoint host.
+ conn, err := client.dialWithBalancer()
+ if err != nil {
+ client.cancel()
+ client.resolver.Close()
+ // TODO: Error like `fmt.Errorf(dialing [%s] failed: %v, strings.Join(cfg.Endpoints, ";"), err)` would help with debugging a lot.
+ return nil, err
+ }
+ client.conn = conn
+
+ client.Cluster = NewCluster(client)
+ client.KV = NewKV(client)
+ client.Lease = NewLease(client)
+ client.Watcher = NewWatcher(client)
+ client.Auth = NewAuth(client)
+ client.Maintenance = NewMaintenance(client)
+
+ // get token with established connection
+ ctx, cancel = client.ctx, func() {}
+ if client.cfg.DialTimeout > 0 {
+ ctx, cancel = context.WithTimeout(ctx, client.cfg.DialTimeout)
+ }
+ err = client.getToken(ctx)
+ if err != nil {
+ client.Close()
+ cancel()
+ // TODO: Consider fmt.Errorf("communicating with [%s] failed: %v", strings.Join(cfg.Endpoints, ";"), err)
+ return nil, err
+ }
+ cancel()
+
+ if cfg.RejectOldCluster {
+ if err := client.checkVersion(); err != nil {
+ client.Close()
+ return nil, err
+ }
+ }
+
+ go client.autoSync()
+ return client, nil
+}
+
+// roundRobinQuorumBackoff retries against quorum between each backoff.
+// This is intended for use with a round robin load balancer.
+func (c *Client) roundRobinQuorumBackoff(waitBetween time.Duration, jitterFraction float64) backoffFunc {
+ return func(attempt uint) time.Duration {
+ // after each round robin across quorum, backoff for our wait between duration
+ n := uint(len(c.Endpoints()))
+ quorum := (n/2 + 1)
+ if attempt%quorum == 0 {
+ c.lg.Debug("backoff", zap.Uint("attempt", attempt), zap.Uint("quorum", quorum), zap.Duration("waitBetween", waitBetween), zap.Float64("jitterFraction", jitterFraction))
+ return jitterUp(waitBetween, jitterFraction)
+ }
+ c.lg.Debug("backoff skipped", zap.Uint("attempt", attempt), zap.Uint("quorum", quorum))
+ return 0
+ }
+}
+
+// minSupportedVersion returns the minimum version supported, which is the previous minor release.
+func minSupportedVersion() *semver.Version {
+ ver := semver.Must(semver.NewVersion(version.Version))
+ // consider only major and minor version
+ ver = &semver.Version{Major: ver.Major, Minor: ver.Minor}
+ for i := range version.AllVersions {
+ if version.AllVersions[i].Equal(*ver) {
+ if i == 0 {
+ return ver
+ }
+ return &version.AllVersions[i-1]
+ }
+ }
+ panic("current version is not in the version list")
+}
+
+func (c *Client) checkVersion() (err error) {
+ var wg sync.WaitGroup
+
+ eps := c.Endpoints()
+ errc := make(chan error, len(eps))
+ ctx, cancel := context.WithCancel(c.ctx)
+ if c.cfg.DialTimeout > 0 {
+ cancel()
+ ctx, cancel = context.WithTimeout(c.ctx, c.cfg.DialTimeout)
+ }
+
+ wg.Add(len(eps))
+ for _, ep := range eps {
+ // if cluster is current, any endpoint gives a recent version
+ go func(e string) {
+ defer wg.Done()
+ resp, rerr := c.Status(ctx, e)
+ if rerr != nil {
+ errc <- rerr
+ return
+ }
+ vs, serr := semver.NewVersion(resp.Version)
+ if serr != nil {
+ errc <- serr
+ return
+ }
+
+ if vs.LessThan(*minSupportedVersion()) {
+ rerr = ErrOldCluster
+ }
+ errc <- rerr
+ }(ep)
+ }
+ // wait for success
+ for range eps {
+ if err = <-errc; err != nil {
+ break
+ }
+ }
+ cancel()
+ wg.Wait()
+ return err
+}
+
+// ActiveConnection returns the current in-use connection
+func (c *Client) ActiveConnection() *grpc.ClientConn { return c.conn }
+
+// isHaltErr returns true if the given error and context indicate no forward
+// progress can be made, even after reconnecting.
+func isHaltErr(ctx context.Context, err error) bool {
+ if ctx != nil && ctx.Err() != nil {
+ return true
+ }
+ if err == nil {
+ return false
+ }
+ ev, _ := status.FromError(err)
+ // Unavailable codes mean the system will be right back.
+ // (e.g., can't connect, lost leader)
+ // Treat Internal codes as if something failed, leaving the
+ // system in an inconsistent state, but retrying could make progress.
+ // (e.g., failed in middle of send, corrupted frame)
+ // TODO: are permanent Internal errors possible from grpc?
+ return ev.Code() != codes.Unavailable && ev.Code() != codes.Internal
+}
+
+// isUnavailableErr returns true if the given error is an unavailable error
+func isUnavailableErr(ctx context.Context, err error) bool {
+ if ctx != nil && ctx.Err() != nil {
+ return false
+ }
+ if err == nil {
+ return false
+ }
+ ev, ok := status.FromError(err)
+ if ok {
+ // Unavailable codes mean the system will be right back.
+ // (e.g., can't connect, lost leader)
+ return ev.Code() == codes.Unavailable
+ }
+ return false
+}
+
+// ContextError converts the error into an EtcdError if the error message matches one of
+// the defined messages; otherwise, it tries to retrieve the context error.
+func ContextError(ctx context.Context, err error) error {
+ if err == nil {
+ return nil
+ }
+ err = rpctypes.Error(err)
+ var serverErr rpctypes.EtcdError
+ if errors.As(err, &serverErr) {
+ return err
+ }
+ if ev, ok := status.FromError(err); ok {
+ code := ev.Code()
+ switch code {
+ case codes.DeadlineExceeded:
+ fallthrough
+ case codes.Canceled:
+ if ctx.Err() != nil {
+ err = ctx.Err()
+ }
+ }
+ }
+ return err
+}
+
+func canceledByCaller(stopCtx context.Context, err error) bool {
+ if stopCtx.Err() == nil || err == nil {
+ return false
+ }
+
+ return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
+}
+
+// IsConnCanceled returns true, if error is from a closed gRPC connection.
+// ref. https://github.com/grpc/grpc-go/pull/1854
+func IsConnCanceled(err error) bool {
+ if err == nil {
+ return false
+ }
+
+ // >= gRPC v1.23.x
+ s, ok := status.FromError(err)
+ if ok {
+ // connection is canceled or server has already closed the connection
+ return s.Code() == codes.Canceled || s.Message() == "transport is closing"
+ }
+
+ // >= gRPC v1.10.x
+ if errors.Is(err, context.Canceled) {
+ return true
+ }
+
+ // <= gRPC v1.7.x returns 'errors.New("grpc: the client connection is closing")'
+ return strings.Contains(err.Error(), "grpc: the client connection is closing")
+}