2016-05-13 06:50:58 +03:00
|
|
|
// Copyright 2016 The etcd Authors
|
2016-01-27 20:59:47 +03:00
|
|
|
//
|
|
|
|
// 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 (
|
2016-05-26 01:31:39 +03:00
|
|
|
"crypto/tls"
|
2016-02-02 22:01:58 +03:00
|
|
|
"errors"
|
2016-05-26 01:31:39 +03:00
|
|
|
"fmt"
|
2016-03-27 08:05:22 +03:00
|
|
|
"io/ioutil"
|
|
|
|
"log"
|
2016-01-31 06:14:36 +03:00
|
|
|
"net"
|
|
|
|
"net/url"
|
2016-02-06 02:13:24 +03:00
|
|
|
"strings"
|
2016-01-28 23:54:03 +03:00
|
|
|
"time"
|
2016-01-28 06:23:26 +03:00
|
|
|
|
2016-05-13 05:25:42 +03:00
|
|
|
"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
|
|
|
|
|
2016-03-23 03:10:28 +03:00
|
|
|
"golang.org/x/net/context"
|
|
|
|
"google.golang.org/grpc"
|
2016-08-24 02:07:03 +03:00
|
|
|
"google.golang.org/grpc/codes"
|
2016-03-23 03:10:28 +03:00
|
|
|
"google.golang.org/grpc/credentials"
|
2016-05-13 05:25:42 +03:00
|
|
|
"google.golang.org/grpc/metadata"
|
2016-01-27 20:59:47 +03:00
|
|
|
)
|
|
|
|
|
2016-02-02 22:01:58 +03:00
|
|
|
var (
|
|
|
|
ErrNoAvailableEndpoints = errors.New("etcdclient: no available endpoints")
|
|
|
|
)
|
|
|
|
|
2016-01-28 06:23:26 +03:00
|
|
|
// Client provides and manages an etcd v3 client session.
|
2016-01-27 20:59:47 +03:00
|
|
|
type Client struct {
|
2016-02-25 09:33:28 +03:00
|
|
|
Cluster
|
|
|
|
KV
|
|
|
|
Lease
|
|
|
|
Watcher
|
2016-03-02 08:57:40 +03:00
|
|
|
Auth
|
2016-03-10 20:29:30 +03:00
|
|
|
Maintenance
|
2016-01-27 20:59:47 +03:00
|
|
|
|
2016-08-13 07:16:09 +03:00
|
|
|
conn *grpc.ClientConn
|
|
|
|
cfg Config
|
|
|
|
creds *credentials.TransportCredentials
|
|
|
|
balancer *simpleBalancer
|
|
|
|
retryWrapper retryRpcFunc
|
2016-03-03 23:43:03 +03:00
|
|
|
|
|
|
|
ctx context.Context
|
|
|
|
cancel context.CancelFunc
|
2016-05-05 02:57:45 +03:00
|
|
|
|
2016-05-20 20:19:09 +03:00
|
|
|
// Username is a username for authentication
|
2016-05-13 10:43:42 +03:00
|
|
|
Username string
|
2016-05-20 20:19:09 +03:00
|
|
|
// Password is a password for authentication
|
2016-05-13 10:43:42 +03:00
|
|
|
Password string
|
2016-01-27 20:59:47 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// New creates a new etcdv3 client from a given configuration.
|
|
|
|
func New(cfg Config) (*Client, error) {
|
2016-02-02 22:01:58 +03:00
|
|
|
if len(cfg.Endpoints) == 0 {
|
|
|
|
return nil, ErrNoAvailableEndpoints
|
|
|
|
}
|
|
|
|
|
2016-01-31 06:14:36 +03:00
|
|
|
return newClient(&cfg)
|
2016-01-27 20:59:47 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// NewFromURL creates a new etcdv3 client from a URL.
|
|
|
|
func NewFromURL(url string) (*Client, error) {
|
|
|
|
return New(Config{Endpoints: []string{url}})
|
|
|
|
}
|
|
|
|
|
2016-04-01 00:01:16 +03:00
|
|
|
// NewFromConfigFile creates a new etcdv3 client from a configuration file.
|
|
|
|
func NewFromConfigFile(path string) (*Client, error) {
|
|
|
|
cfg, err := configFromFile(path)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
return New(*cfg)
|
|
|
|
}
|
|
|
|
|
2016-01-27 20:59:47 +03:00
|
|
|
// Close shuts down the client's etcd connections.
|
2016-06-08 03:03:48 +03:00
|
|
|
func (c *Client) Close() error {
|
|
|
|
c.cancel()
|
|
|
|
return toErr(c.ctx, c.conn.Close())
|
2016-01-27 20:59:47 +03:00
|
|
|
}
|
|
|
|
|
2016-03-03 23:43:03 +03:00
|
|
|
// 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 }
|
|
|
|
|
2016-01-28 06:23:26 +03:00
|
|
|
// Endpoints lists the registered endpoints for the client.
|
|
|
|
func (c *Client) Endpoints() []string { return c.cfg.Endpoints }
|
|
|
|
|
2016-09-19 22:36:01 +03:00
|
|
|
// SetEndpoints updates client's endpoints.
|
|
|
|
func (c *Client) SetEndpoints(eps ...string) {
|
|
|
|
c.cfg.Endpoints = eps
|
|
|
|
c.balancer.updateAddrs(eps)
|
|
|
|
}
|
|
|
|
|
2016-09-21 19:10:25 +03:00
|
|
|
// 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 {
|
|
|
|
eps = append(eps, m.ClientURLs...)
|
|
|
|
}
|
|
|
|
c.SetEndpoints(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, _ := context.WithTimeout(c.ctx, 5*time.Second)
|
|
|
|
if err := c.Sync(ctx); err != nil && err != c.ctx.Err() {
|
|
|
|
logger.Println("Auto sync endpoints failed:", err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-05-13 10:43:42 +03:00
|
|
|
type authTokenCredential struct {
|
|
|
|
token string
|
|
|
|
}
|
|
|
|
|
|
|
|
func (cred authTokenCredential) RequireTransportSecurity() bool {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
|
|
|
func (cred authTokenCredential) GetRequestMetadata(ctx context.Context, s ...string) (map[string]string, error) {
|
|
|
|
return map[string]string{
|
|
|
|
"token": cred.token,
|
|
|
|
}, nil
|
|
|
|
}
|
|
|
|
|
2016-09-19 22:36:01 +03:00
|
|
|
func parseEndpoint(endpoint string) (proto string, host string, scheme bool) {
|
2016-06-08 02:28:08 +03:00
|
|
|
proto = "tcp"
|
|
|
|
host = endpoint
|
|
|
|
url, uerr := url.Parse(endpoint)
|
|
|
|
if uerr != nil || !strings.Contains(endpoint, "://") {
|
|
|
|
return
|
|
|
|
}
|
2016-09-19 22:36:01 +03:00
|
|
|
scheme = true
|
|
|
|
|
2016-06-08 02:28:08 +03:00
|
|
|
// strip scheme:// prefix since grpc dials by host
|
|
|
|
host = url.Host
|
|
|
|
switch url.Scheme {
|
2016-09-19 22:36:01 +03:00
|
|
|
case "http", "https":
|
2016-06-08 02:28:08 +03:00
|
|
|
case "unix":
|
|
|
|
proto = "unix"
|
2016-09-19 22:36:01 +03:00
|
|
|
default:
|
|
|
|
proto, host = "", ""
|
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *Client) processCreds(protocol string) (creds *credentials.TransportCredentials) {
|
|
|
|
creds = c.creds
|
|
|
|
switch protocol {
|
|
|
|
case "unix":
|
2016-06-08 02:28:08 +03:00
|
|
|
case "http":
|
|
|
|
creds = nil
|
|
|
|
case "https":
|
|
|
|
if creds != nil {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
tlsconfig := &tls.Config{}
|
|
|
|
emptyCreds := credentials.NewTLS(tlsconfig)
|
|
|
|
creds = &emptyCreds
|
|
|
|
default:
|
2016-09-19 22:36:01 +03:00
|
|
|
creds = nil
|
2016-06-08 02:28:08 +03:00
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-08-12 19:42:18 +03:00
|
|
|
// dialSetupOpts gives the dial opts prior to any authentication
|
|
|
|
func (c *Client) dialSetupOpts(endpoint string, dopts ...grpc.DialOption) (opts []grpc.DialOption) {
|
|
|
|
if c.cfg.DialTimeout > 0 {
|
2016-07-01 21:29:37 +03:00
|
|
|
opts = []grpc.DialOption{grpc.WithTimeout(c.cfg.DialTimeout)}
|
2016-01-29 03:09:14 +03:00
|
|
|
}
|
2016-06-08 02:28:08 +03:00
|
|
|
opts = append(opts, dopts...)
|
|
|
|
|
|
|
|
f := func(host string, t time.Duration) (net.Conn, error) {
|
2016-09-19 22:36:01 +03:00
|
|
|
proto, host, _ := parseEndpoint(c.balancer.getEndpoint(host))
|
2016-06-08 02:28:08 +03:00
|
|
|
if proto == "" {
|
2016-06-09 20:10:55 +03:00
|
|
|
return nil, fmt.Errorf("unknown scheme for %q", host)
|
2016-05-26 01:31:39 +03:00
|
|
|
}
|
2016-03-08 00:42:52 +03:00
|
|
|
select {
|
|
|
|
case <-c.ctx.Done():
|
|
|
|
return nil, c.ctx.Err()
|
|
|
|
default:
|
|
|
|
}
|
2016-06-08 02:28:08 +03:00
|
|
|
return net.DialTimeout(proto, host, t)
|
2016-03-08 00:42:52 +03:00
|
|
|
}
|
|
|
|
opts = append(opts, grpc.WithDialer(f))
|
2016-01-31 06:14:36 +03:00
|
|
|
|
2016-09-19 22:36:01 +03:00
|
|
|
creds := c.creds
|
|
|
|
if proto, _, scheme := parseEndpoint(endpoint); scheme {
|
|
|
|
creds = c.processCreds(proto)
|
|
|
|
}
|
2016-05-26 01:31:39 +03:00
|
|
|
if creds != nil {
|
|
|
|
opts = append(opts, grpc.WithTransportCredentials(*creds))
|
|
|
|
} else {
|
|
|
|
opts = append(opts, grpc.WithInsecure())
|
|
|
|
}
|
|
|
|
|
2016-06-09 20:10:55 +03:00
|
|
|
return opts
|
|
|
|
}
|
|
|
|
|
|
|
|
// Dial connects to a single endpoint using the client's config.
|
|
|
|
func (c *Client) Dial(endpoint string) (*grpc.ClientConn, error) {
|
|
|
|
return c.dial(endpoint)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *Client) dial(endpoint string, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
|
|
|
|
opts := c.dialSetupOpts(endpoint, dopts...)
|
|
|
|
host := getHost(endpoint)
|
2016-05-13 10:43:42 +03:00
|
|
|
if c.Username != "" && c.Password != "" {
|
2016-06-09 20:10:55 +03:00
|
|
|
// use dial options without dopts to avoid reusing the client balancer
|
|
|
|
auth, err := newAuthenticator(host, c.dialSetupOpts(endpoint))
|
2016-05-13 10:43:42 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
defer auth.close()
|
|
|
|
|
|
|
|
resp, err := auth.authenticate(c.ctx, c.Username, c.Password)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
opts = append(opts, grpc.WithPerRPCCredentials(authTokenCredential{token: resp.Token}))
|
|
|
|
}
|
|
|
|
|
2016-06-08 02:28:08 +03:00
|
|
|
conn, err := grpc.Dial(host, opts...)
|
2016-01-28 23:54:03 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
return conn, nil
|
2016-01-28 06:23:26 +03:00
|
|
|
}
|
|
|
|
|
2016-05-13 05:25:42 +03:00
|
|
|
// WithRequireLeader requires client requests to only succeed
|
|
|
|
// when the cluster has a leader.
|
|
|
|
func WithRequireLeader(ctx context.Context) context.Context {
|
|
|
|
md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
|
|
|
|
return metadata.NewContext(ctx, md)
|
|
|
|
}
|
|
|
|
|
2016-01-31 06:14:36 +03:00
|
|
|
func newClient(cfg *Config) (*Client, error) {
|
2016-01-28 06:23:26 +03:00
|
|
|
if cfg == nil {
|
2016-06-08 03:03:48 +03:00
|
|
|
cfg = &Config{}
|
2016-01-28 06:23:26 +03:00
|
|
|
}
|
2016-06-11 06:40:46 +03:00
|
|
|
var creds *credentials.TransportCredentials
|
2016-01-29 03:09:14 +03:00
|
|
|
if cfg.TLS != nil {
|
2016-03-08 02:00:14 +03:00
|
|
|
c := credentials.NewTLS(cfg.TLS)
|
2016-01-29 03:09:14 +03:00
|
|
|
creds = &c
|
|
|
|
}
|
2016-05-13 10:43:42 +03:00
|
|
|
|
2016-01-31 06:14:36 +03:00
|
|
|
// use a temporary skeleton client to bootstrap first connection
|
2016-03-08 00:42:52 +03:00
|
|
|
ctx, cancel := context.WithCancel(context.TODO())
|
2016-02-25 09:33:28 +03:00
|
|
|
client := &Client{
|
2016-06-08 03:03:48 +03:00
|
|
|
conn: nil,
|
|
|
|
cfg: *cfg,
|
|
|
|
creds: creds,
|
|
|
|
ctx: ctx,
|
|
|
|
cancel: cancel,
|
2016-02-25 09:33:28 +03:00
|
|
|
}
|
2016-05-13 10:43:42 +03:00
|
|
|
if cfg.Username != "" && cfg.Password != "" {
|
|
|
|
client.Username = cfg.Username
|
|
|
|
client.Password = cfg.Password
|
|
|
|
}
|
2016-06-08 02:28:08 +03:00
|
|
|
|
2016-08-13 07:16:09 +03:00
|
|
|
client.balancer = newSimpleBalancer(cfg.Endpoints)
|
|
|
|
conn, err := client.dial(cfg.Endpoints[0], grpc.WithBalancer(client.balancer))
|
2016-06-08 03:03:48 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
client.conn = conn
|
2016-08-13 07:16:09 +03:00
|
|
|
client.retryWrapper = client.newRetryWrapper()
|
2016-05-05 02:57:45 +03:00
|
|
|
|
2016-07-01 21:29:37 +03:00
|
|
|
// wait for a connection
|
|
|
|
if cfg.DialTimeout > 0 {
|
|
|
|
hasConn := false
|
|
|
|
waitc := time.After(cfg.DialTimeout)
|
|
|
|
select {
|
|
|
|
case <-client.balancer.readyc:
|
|
|
|
hasConn = true
|
|
|
|
case <-ctx.Done():
|
|
|
|
case <-waitc:
|
|
|
|
}
|
|
|
|
if !hasConn {
|
|
|
|
client.cancel()
|
|
|
|
conn.Close()
|
|
|
|
return nil, grpc.ErrClientConnTimeout
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-02-25 09:33:28 +03:00
|
|
|
client.Cluster = NewCluster(client)
|
|
|
|
client.KV = NewKV(client)
|
|
|
|
client.Lease = NewLease(client)
|
|
|
|
client.Watcher = NewWatcher(client)
|
2016-03-02 08:57:40 +03:00
|
|
|
client.Auth = NewAuth(client)
|
2016-03-30 05:24:44 +03:00
|
|
|
client.Maintenance = NewMaintenance(client)
|
2016-03-29 01:12:31 +03:00
|
|
|
if cfg.Logger != nil {
|
|
|
|
logger.Set(cfg.Logger)
|
2016-03-27 08:05:22 +03:00
|
|
|
} else {
|
2016-03-29 01:12:31 +03:00
|
|
|
// disable client side grpc by default
|
|
|
|
logger.Set(log.New(ioutil.Discard, "", 0))
|
2016-03-27 08:05:22 +03:00
|
|
|
}
|
2016-02-25 09:33:28 +03:00
|
|
|
|
2016-09-21 19:10:25 +03:00
|
|
|
go client.autoSync()
|
2016-02-25 09:33:28 +03:00
|
|
|
return client, nil
|
2016-01-28 06:23:26 +03:00
|
|
|
}
|
|
|
|
|
2016-02-21 16:05:03 +03:00
|
|
|
// ActiveConnection returns the current in-use connection
|
2016-06-08 03:03:48 +03:00
|
|
|
func (c *Client) ActiveConnection() *grpc.ClientConn { return c.conn }
|
2016-06-02 20:54:01 +03:00
|
|
|
|
2016-06-08 03:03:48 +03:00
|
|
|
// 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
|
2016-06-02 20:54:01 +03:00
|
|
|
}
|
2016-06-08 03:03:48 +03:00
|
|
|
if err == nil {
|
|
|
|
return false
|
2016-01-27 20:59:47 +03:00
|
|
|
}
|
2016-08-29 05:41:27 +03:00
|
|
|
code := grpc.Code(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 code != codes.Unavailable && code != codes.Internal
|
2016-05-05 02:57:45 +03:00
|
|
|
}
|
|
|
|
|
2016-06-08 03:03:48 +03:00
|
|
|
func toErr(ctx context.Context, err error) error {
|
|
|
|
if err == nil {
|
|
|
|
return nil
|
2016-05-05 02:57:45 +03:00
|
|
|
}
|
2016-06-08 03:03:48 +03:00
|
|
|
err = rpctypes.Error(err)
|
2016-08-24 02:07:03 +03:00
|
|
|
if _, ok := err.(rpctypes.EtcdError); ok {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
code := grpc.Code(err)
|
|
|
|
switch code {
|
|
|
|
case codes.DeadlineExceeded:
|
|
|
|
fallthrough
|
|
|
|
case codes.Canceled:
|
|
|
|
if ctx.Err() != nil {
|
|
|
|
err = ctx.Err()
|
|
|
|
}
|
|
|
|
case codes.Unavailable:
|
2016-07-01 21:29:37 +03:00
|
|
|
err = ErrNoAvailableEndpoints
|
2016-08-24 02:07:03 +03:00
|
|
|
case codes.FailedPrecondition:
|
2016-06-08 03:03:48 +03:00
|
|
|
err = grpc.ErrClientConnClosing
|
2016-05-05 02:57:45 +03:00
|
|
|
}
|
2016-06-08 03:03:48 +03:00
|
|
|
return err
|
2016-01-28 08:13:23 +03:00
|
|
|
}
|