VOL-1497 : Add more control to kv/memory access
- Added kv locking mechanism (etcd only)
- (watch) control path access whenever possible
- (watch) use a transaction for updates and merge with memory
- cleaned up vendoring
- misc changes to fix exceptions found along the way
Amendments:
- Copyright header got removed in auto-generated file
- Changed default locking to false for KV list operation
- Updated backend api to allow the passing of locking parameter
Change-Id: Ie1a55d3ca8b9d92ae71a85ce42bb22fcf1419e2c
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/balancer.go b/vendor/go.etcd.io/etcd/clientv3/balancer/balancer.go
new file mode 100644
index 0000000..25dc2b7
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/balancer.go
@@ -0,0 +1,275 @@
+// Copyright 2018 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 balancer
+
+import (
+ "fmt"
+ "strconv"
+ "sync"
+ "time"
+
+ "go.etcd.io/etcd/clientv3/balancer/picker"
+
+ "go.uber.org/zap"
+ "google.golang.org/grpc/balancer"
+ "google.golang.org/grpc/connectivity"
+ "google.golang.org/grpc/resolver"
+ _ "google.golang.org/grpc/resolver/dns" // register DNS resolver
+ _ "google.golang.org/grpc/resolver/passthrough" // register passthrough resolver
+)
+
+// RegisterBuilder creates and registers a builder. Since this function calls balancer.Register, it
+// must be invoked at initialization time.
+func RegisterBuilder(cfg Config) {
+ bb := &builder{cfg}
+ balancer.Register(bb)
+
+ bb.cfg.Logger.Info(
+ "registered balancer",
+ zap.String("policy", bb.cfg.Policy.String()),
+ zap.String("name", bb.cfg.Name),
+ )
+}
+
+type builder struct {
+ cfg Config
+}
+
+// Build is called initially when creating "ccBalancerWrapper".
+// "grpc.Dial" is called to this client connection.
+// Then, resolved addresses will be handled via "HandleResolvedAddrs".
+func (b *builder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer {
+ bb := &baseBalancer{
+ id: strconv.FormatInt(time.Now().UnixNano(), 36),
+ policy: b.cfg.Policy,
+ name: b.cfg.Policy.String(),
+ lg: b.cfg.Logger,
+
+ addrToSc: make(map[resolver.Address]balancer.SubConn),
+ scToAddr: make(map[balancer.SubConn]resolver.Address),
+ scToSt: make(map[balancer.SubConn]connectivity.State),
+
+ currentConn: nil,
+ csEvltr: &connectivityStateEvaluator{},
+
+ // initialize picker always returns "ErrNoSubConnAvailable"
+ Picker: picker.NewErr(balancer.ErrNoSubConnAvailable),
+ }
+ if b.cfg.Name != "" {
+ bb.name = b.cfg.Name
+ }
+ if bb.lg == nil {
+ bb.lg = zap.NewNop()
+ }
+
+ // TODO: support multiple connections
+ bb.mu.Lock()
+ bb.currentConn = cc
+ bb.mu.Unlock()
+
+ bb.lg.Info(
+ "built balancer",
+ zap.String("balancer-id", bb.id),
+ zap.String("policy", bb.policy.String()),
+ zap.String("resolver-target", cc.Target()),
+ )
+ return bb
+}
+
+// Name implements "grpc/balancer.Builder" interface.
+func (b *builder) Name() string { return b.cfg.Name }
+
+// Balancer defines client balancer interface.
+type Balancer interface {
+ // Balancer is called on specified client connection. Client initiates gRPC
+ // connection with "grpc.Dial(addr, grpc.WithBalancerName)", and then those resolved
+ // addresses are passed to "grpc/balancer.Balancer.HandleResolvedAddrs".
+ // For each resolved address, balancer calls "balancer.ClientConn.NewSubConn".
+ // "grpc/balancer.Balancer.HandleSubConnStateChange" is called when connectivity state
+ // changes, thus requires failover logic in this method.
+ balancer.Balancer
+
+ // Picker calls "Pick" for every client request.
+ picker.Picker
+}
+
+type baseBalancer struct {
+ id string
+ policy picker.Policy
+ name string
+ lg *zap.Logger
+
+ mu sync.RWMutex
+
+ addrToSc map[resolver.Address]balancer.SubConn
+ scToAddr map[balancer.SubConn]resolver.Address
+ scToSt map[balancer.SubConn]connectivity.State
+
+ currentConn balancer.ClientConn
+ currentState connectivity.State
+ csEvltr *connectivityStateEvaluator
+
+ picker.Picker
+}
+
+// HandleResolvedAddrs implements "grpc/balancer.Balancer" interface.
+// gRPC sends initial or updated resolved addresses from "Build".
+func (bb *baseBalancer) HandleResolvedAddrs(addrs []resolver.Address, err error) {
+ if err != nil {
+ bb.lg.Warn("HandleResolvedAddrs called with error", zap.String("balancer-id", bb.id), zap.Error(err))
+ return
+ }
+ bb.lg.Info("resolved", zap.String("balancer-id", bb.id), zap.Strings("addresses", addrsToStrings(addrs)))
+
+ bb.mu.Lock()
+ defer bb.mu.Unlock()
+
+ resolved := make(map[resolver.Address]struct{})
+ for _, addr := range addrs {
+ resolved[addr] = struct{}{}
+ if _, ok := bb.addrToSc[addr]; !ok {
+ sc, err := bb.currentConn.NewSubConn([]resolver.Address{addr}, balancer.NewSubConnOptions{})
+ if err != nil {
+ bb.lg.Warn("NewSubConn failed", zap.String("balancer-id", bb.id), zap.Error(err), zap.String("address", addr.Addr))
+ continue
+ }
+ bb.addrToSc[addr] = sc
+ bb.scToAddr[sc] = addr
+ bb.scToSt[sc] = connectivity.Idle
+ sc.Connect()
+ }
+ }
+
+ for addr, sc := range bb.addrToSc {
+ if _, ok := resolved[addr]; !ok {
+ // was removed by resolver or failed to create subconn
+ bb.currentConn.RemoveSubConn(sc)
+ delete(bb.addrToSc, addr)
+
+ bb.lg.Info(
+ "removed subconn",
+ zap.String("balancer-id", bb.id),
+ zap.String("address", addr.Addr),
+ zap.String("subconn", scToString(sc)),
+ )
+
+ // Keep the state of this sc in bb.scToSt until sc's state becomes Shutdown.
+ // The entry will be deleted in HandleSubConnStateChange.
+ // (DO NOT) delete(bb.scToAddr, sc)
+ // (DO NOT) delete(bb.scToSt, sc)
+ }
+ }
+}
+
+// HandleSubConnStateChange implements "grpc/balancer.Balancer" interface.
+func (bb *baseBalancer) HandleSubConnStateChange(sc balancer.SubConn, s connectivity.State) {
+ bb.mu.Lock()
+ defer bb.mu.Unlock()
+
+ old, ok := bb.scToSt[sc]
+ if !ok {
+ bb.lg.Warn(
+ "state change for an unknown subconn",
+ zap.String("balancer-id", bb.id),
+ zap.String("subconn", scToString(sc)),
+ zap.String("state", s.String()),
+ )
+ return
+ }
+
+ bb.lg.Info(
+ "state changed",
+ zap.String("balancer-id", bb.id),
+ zap.Bool("connected", s == connectivity.Ready),
+ zap.String("subconn", scToString(sc)),
+ zap.String("address", bb.scToAddr[sc].Addr),
+ zap.String("old-state", old.String()),
+ zap.String("new-state", s.String()),
+ )
+
+ bb.scToSt[sc] = s
+ switch s {
+ case connectivity.Idle:
+ sc.Connect()
+ case connectivity.Shutdown:
+ // When an address was removed by resolver, b called RemoveSubConn but
+ // kept the sc's state in scToSt. Remove state for this sc here.
+ delete(bb.scToAddr, sc)
+ delete(bb.scToSt, sc)
+ }
+
+ oldAggrState := bb.currentState
+ bb.currentState = bb.csEvltr.recordTransition(old, s)
+
+ // Regenerate picker when one of the following happens:
+ // - this sc became ready from not-ready
+ // - this sc became not-ready from ready
+ // - the aggregated state of balancer became TransientFailure from non-TransientFailure
+ // - the aggregated state of balancer became non-TransientFailure from TransientFailure
+ if (s == connectivity.Ready) != (old == connectivity.Ready) ||
+ (bb.currentState == connectivity.TransientFailure) != (oldAggrState == connectivity.TransientFailure) {
+ bb.regeneratePicker()
+ }
+
+ bb.currentConn.UpdateBalancerState(bb.currentState, bb.Picker)
+ return
+}
+
+func (bb *baseBalancer) regeneratePicker() {
+ if bb.currentState == connectivity.TransientFailure {
+ bb.lg.Info(
+ "generated transient error picker",
+ zap.String("balancer-id", bb.id),
+ zap.String("policy", bb.policy.String()),
+ )
+ bb.Picker = picker.NewErr(balancer.ErrTransientFailure)
+ return
+ }
+
+ // only pass ready subconns to picker
+ scs := make([]balancer.SubConn, 0)
+ addrToSc := make(map[resolver.Address]balancer.SubConn)
+ scToAddr := make(map[balancer.SubConn]resolver.Address)
+ for addr, sc := range bb.addrToSc {
+ if st, ok := bb.scToSt[sc]; ok && st == connectivity.Ready {
+ scs = append(scs, sc)
+ addrToSc[addr] = sc
+ scToAddr[sc] = addr
+ }
+ }
+
+ switch bb.policy {
+ case picker.RoundrobinBalanced:
+ bb.Picker = picker.NewRoundrobinBalanced(bb.lg, scs, addrToSc, scToAddr)
+
+ default:
+ panic(fmt.Errorf("invalid balancer picker policy (%d)", bb.policy))
+ }
+
+ bb.lg.Info(
+ "generated picker",
+ zap.String("balancer-id", bb.id),
+ zap.String("policy", bb.policy.String()),
+ zap.Strings("subconn-ready", scsToStrings(addrToSc)),
+ zap.Int("subconn-size", len(addrToSc)),
+ )
+}
+
+// Close implements "grpc/balancer.Balancer" interface.
+// Close is a nop because base balancer doesn't have internal state to clean up,
+// and it doesn't need to call RemoveSubConn for the SubConns.
+func (bb *baseBalancer) Close() {
+ // TODO
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/config.go b/vendor/go.etcd.io/etcd/clientv3/balancer/config.go
new file mode 100644
index 0000000..0339a84
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/config.go
@@ -0,0 +1,36 @@
+// Copyright 2018 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 balancer
+
+import (
+ "go.etcd.io/etcd/clientv3/balancer/picker"
+
+ "go.uber.org/zap"
+)
+
+// Config defines balancer configurations.
+type Config struct {
+ // Policy configures balancer policy.
+ Policy picker.Policy
+
+ // Name defines an additional name for balancer.
+ // Useful for balancer testing to avoid register conflicts.
+ // If empty, defaults to policy name.
+ Name string
+
+ // Logger configures balancer logging.
+ // If nil, logs are discarded.
+ Logger *zap.Logger
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/connectivity.go b/vendor/go.etcd.io/etcd/clientv3/balancer/connectivity.go
new file mode 100644
index 0000000..6cdeb3f
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/connectivity.go
@@ -0,0 +1,58 @@
+// Copyright 2018 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 balancer
+
+import "google.golang.org/grpc/connectivity"
+
+// connectivityStateEvaluator gets updated by addrConns when their
+// states transition, based on which it evaluates the state of
+// ClientConn.
+type connectivityStateEvaluator struct {
+ numReady uint64 // Number of addrConns in ready state.
+ numConnecting uint64 // Number of addrConns in connecting state.
+ numTransientFailure uint64 // Number of addrConns in transientFailure.
+}
+
+// recordTransition records state change happening in every subConn and based on
+// that it evaluates what aggregated state should be.
+// It can only transition between Ready, Connecting and TransientFailure. Other states,
+// Idle and Shutdown are transitioned into by ClientConn; in the beginning of the connection
+// before any subConn is created ClientConn is in idle state. In the end when ClientConn
+// closes it is in Shutdown state.
+//
+// recordTransition should only be called synchronously from the same goroutine.
+func (cse *connectivityStateEvaluator) recordTransition(oldState, newState connectivity.State) connectivity.State {
+ // Update counters.
+ for idx, state := range []connectivity.State{oldState, newState} {
+ updateVal := 2*uint64(idx) - 1 // -1 for oldState and +1 for new.
+ switch state {
+ case connectivity.Ready:
+ cse.numReady += updateVal
+ case connectivity.Connecting:
+ cse.numConnecting += updateVal
+ case connectivity.TransientFailure:
+ cse.numTransientFailure += updateVal
+ }
+ }
+
+ // Evaluate.
+ if cse.numReady > 0 {
+ return connectivity.Ready
+ }
+ if cse.numConnecting > 0 {
+ return connectivity.Connecting
+ }
+ return connectivity.TransientFailure
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/doc.go b/vendor/go.etcd.io/etcd/clientv3/balancer/doc.go
new file mode 100644
index 0000000..45af5e9
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/doc.go
@@ -0,0 +1,16 @@
+// Copyright 2018 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 balancer implements client balancer.
+package balancer
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/grpc1.7-health.go b/vendor/go.etcd.io/etcd/clientv3/balancer/grpc1.7-health.go
new file mode 100644
index 0000000..7d24b93
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/grpc1.7-health.go
@@ -0,0 +1,657 @@
+// Copyright 2018 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 balancer
+
+import (
+ "context"
+ "errors"
+ "io/ioutil"
+ "net/url"
+ "strings"
+ "sync"
+ "time"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/grpclog"
+ healthpb "google.golang.org/grpc/health/grpc_health_v1"
+ "google.golang.org/grpc/status"
+)
+
+// TODO: replace with something better
+var lg = grpclog.NewLoggerV2(ioutil.Discard, ioutil.Discard, ioutil.Discard)
+
+const (
+ minHealthRetryDuration = 3 * time.Second
+ unknownService = "unknown service grpc.health.v1.Health"
+)
+
+// ErrNoAddrAvilable is returned by Get() when the balancer does not have
+// any active connection to endpoints at the time.
+// This error is returned only when opts.BlockingWait is true.
+var ErrNoAddrAvilable = status.Error(codes.Unavailable, "there is no address available")
+
+type NotifyMsg int
+
+const (
+ NotifyReset NotifyMsg = iota
+ NotifyNext
+)
+
+// GRPC17Health does the bare minimum to expose multiple eps
+// to the grpc reconnection code path
+type GRPC17Health struct {
+ // addrs are the client's endpoint addresses for grpc
+ addrs []grpc.Address
+
+ // eps holds the raw endpoints from the client
+ eps []string
+
+ // notifyCh notifies grpc of the set of addresses for connecting
+ notifyCh chan []grpc.Address
+
+ // readyc closes once the first connection is up
+ readyc chan struct{}
+ readyOnce sync.Once
+
+ // healthCheck checks an endpoint's health.
+ healthCheck func(ep string) (bool, error)
+ healthCheckTimeout time.Duration
+
+ unhealthyMu sync.RWMutex
+ unhealthyHostPorts map[string]time.Time
+
+ // mu protects all fields below.
+ mu sync.RWMutex
+
+ // upc closes when pinAddr transitions from empty to non-empty or the balancer closes.
+ upc chan struct{}
+
+ // downc closes when grpc calls down() on pinAddr
+ downc chan struct{}
+
+ // stopc is closed to signal updateNotifyLoop should stop.
+ stopc chan struct{}
+ stopOnce sync.Once
+ wg sync.WaitGroup
+
+ // donec closes when all goroutines are exited
+ donec chan struct{}
+
+ // updateAddrsC notifies updateNotifyLoop to update addrs.
+ updateAddrsC chan NotifyMsg
+
+ // grpc issues TLS cert checks using the string passed into dial so
+ // that string must be the host. To recover the full scheme://host URL,
+ // have a map from hosts to the original endpoint.
+ hostPort2ep map[string]string
+
+ // pinAddr is the currently pinned address; set to the empty string on
+ // initialization and shutdown.
+ pinAddr string
+
+ closed bool
+}
+
+// DialFunc defines gRPC dial function.
+type DialFunc func(ep string, dopts ...grpc.DialOption) (*grpc.ClientConn, error)
+
+// NewGRPC17Health returns a new health balancer with gRPC v1.7.
+func NewGRPC17Health(
+ eps []string,
+ timeout time.Duration,
+ dialFunc DialFunc,
+) *GRPC17Health {
+ notifyCh := make(chan []grpc.Address)
+ addrs := eps2addrs(eps)
+ hb := &GRPC17Health{
+ addrs: addrs,
+ eps: eps,
+ notifyCh: notifyCh,
+ readyc: make(chan struct{}),
+ healthCheck: func(ep string) (bool, error) { return grpcHealthCheck(ep, dialFunc) },
+ unhealthyHostPorts: make(map[string]time.Time),
+ upc: make(chan struct{}),
+ stopc: make(chan struct{}),
+ downc: make(chan struct{}),
+ donec: make(chan struct{}),
+ updateAddrsC: make(chan NotifyMsg),
+ hostPort2ep: getHostPort2ep(eps),
+ }
+ if timeout < minHealthRetryDuration {
+ timeout = minHealthRetryDuration
+ }
+ hb.healthCheckTimeout = timeout
+
+ close(hb.downc)
+ go hb.updateNotifyLoop()
+ hb.wg.Add(1)
+ go func() {
+ defer hb.wg.Done()
+ hb.updateUnhealthy()
+ }()
+ return hb
+}
+
+func (b *GRPC17Health) Start(target string, config grpc.BalancerConfig) error { return nil }
+
+func (b *GRPC17Health) ConnectNotify() <-chan struct{} {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.upc
+}
+
+func (b *GRPC17Health) UpdateAddrsC() chan NotifyMsg { return b.updateAddrsC }
+func (b *GRPC17Health) StopC() chan struct{} { return b.stopc }
+
+func (b *GRPC17Health) Ready() <-chan struct{} { return b.readyc }
+
+func (b *GRPC17Health) Endpoint(hostPort string) string {
+ b.mu.RLock()
+ defer b.mu.RUnlock()
+ return b.hostPort2ep[hostPort]
+}
+
+func (b *GRPC17Health) Pinned() string {
+ b.mu.RLock()
+ defer b.mu.RUnlock()
+ return b.pinAddr
+}
+
+func (b *GRPC17Health) HostPortError(hostPort string, err error) {
+ if b.Endpoint(hostPort) == "" {
+ lg.Infof("clientv3/balancer: %q is stale (skip marking as unhealthy on %q)", hostPort, err.Error())
+ return
+ }
+
+ b.unhealthyMu.Lock()
+ b.unhealthyHostPorts[hostPort] = time.Now()
+ b.unhealthyMu.Unlock()
+ lg.Infof("clientv3/balancer: %q is marked unhealthy (%q)", hostPort, err.Error())
+}
+
+func (b *GRPC17Health) removeUnhealthy(hostPort, msg string) {
+ if b.Endpoint(hostPort) == "" {
+ lg.Infof("clientv3/balancer: %q was not in unhealthy (%q)", hostPort, msg)
+ return
+ }
+
+ b.unhealthyMu.Lock()
+ delete(b.unhealthyHostPorts, hostPort)
+ b.unhealthyMu.Unlock()
+ lg.Infof("clientv3/balancer: %q is removed from unhealthy (%q)", hostPort, msg)
+}
+
+func (b *GRPC17Health) countUnhealthy() (count int) {
+ b.unhealthyMu.RLock()
+ count = len(b.unhealthyHostPorts)
+ b.unhealthyMu.RUnlock()
+ return count
+}
+
+func (b *GRPC17Health) isUnhealthy(hostPort string) (unhealthy bool) {
+ b.unhealthyMu.RLock()
+ _, unhealthy = b.unhealthyHostPorts[hostPort]
+ b.unhealthyMu.RUnlock()
+ return unhealthy
+}
+
+func (b *GRPC17Health) cleanupUnhealthy() {
+ b.unhealthyMu.Lock()
+ for k, v := range b.unhealthyHostPorts {
+ if time.Since(v) > b.healthCheckTimeout {
+ delete(b.unhealthyHostPorts, k)
+ lg.Infof("clientv3/balancer: removed %q from unhealthy after %v", k, b.healthCheckTimeout)
+ }
+ }
+ b.unhealthyMu.Unlock()
+}
+
+func (b *GRPC17Health) liveAddrs() ([]grpc.Address, map[string]struct{}) {
+ unhealthyCnt := b.countUnhealthy()
+
+ b.mu.RLock()
+ defer b.mu.RUnlock()
+
+ hbAddrs := b.addrs
+ if len(b.addrs) == 1 || unhealthyCnt == 0 || unhealthyCnt == len(b.addrs) {
+ liveHostPorts := make(map[string]struct{}, len(b.hostPort2ep))
+ for k := range b.hostPort2ep {
+ liveHostPorts[k] = struct{}{}
+ }
+ return hbAddrs, liveHostPorts
+ }
+
+ addrs := make([]grpc.Address, 0, len(b.addrs)-unhealthyCnt)
+ liveHostPorts := make(map[string]struct{}, len(addrs))
+ for _, addr := range b.addrs {
+ if !b.isUnhealthy(addr.Addr) {
+ addrs = append(addrs, addr)
+ liveHostPorts[addr.Addr] = struct{}{}
+ }
+ }
+ return addrs, liveHostPorts
+}
+
+func (b *GRPC17Health) updateUnhealthy() {
+ for {
+ select {
+ case <-time.After(b.healthCheckTimeout):
+ b.cleanupUnhealthy()
+ pinned := b.Pinned()
+ if pinned == "" || b.isUnhealthy(pinned) {
+ select {
+ case b.updateAddrsC <- NotifyNext:
+ case <-b.stopc:
+ return
+ }
+ }
+ case <-b.stopc:
+ return
+ }
+ }
+}
+
+// NeedUpdate returns true if all connections are down or
+// addresses do not include current pinned address.
+func (b *GRPC17Health) NeedUpdate() bool {
+ // updating notifyCh can trigger new connections,
+ // need update addrs if all connections are down
+ // or addrs does not include pinAddr.
+ b.mu.RLock()
+ update := !hasAddr(b.addrs, b.pinAddr)
+ b.mu.RUnlock()
+ return update
+}
+
+func (b *GRPC17Health) UpdateAddrs(eps ...string) {
+ np := getHostPort2ep(eps)
+
+ b.mu.Lock()
+ defer b.mu.Unlock()
+
+ match := len(np) == len(b.hostPort2ep)
+ if match {
+ for k, v := range np {
+ if b.hostPort2ep[k] != v {
+ match = false
+ break
+ }
+ }
+ }
+ if match {
+ // same endpoints, so no need to update address
+ return
+ }
+
+ b.hostPort2ep = np
+ b.addrs, b.eps = eps2addrs(eps), eps
+
+ b.unhealthyMu.Lock()
+ b.unhealthyHostPorts = make(map[string]time.Time)
+ b.unhealthyMu.Unlock()
+}
+
+func (b *GRPC17Health) Next() {
+ b.mu.RLock()
+ downc := b.downc
+ b.mu.RUnlock()
+ select {
+ case b.updateAddrsC <- NotifyNext:
+ case <-b.stopc:
+ }
+ // wait until disconnect so new RPCs are not issued on old connection
+ select {
+ case <-downc:
+ case <-b.stopc:
+ }
+}
+
+func (b *GRPC17Health) updateNotifyLoop() {
+ defer close(b.donec)
+
+ for {
+ b.mu.RLock()
+ upc, downc, addr := b.upc, b.downc, b.pinAddr
+ b.mu.RUnlock()
+ // downc or upc should be closed
+ select {
+ case <-downc:
+ downc = nil
+ default:
+ }
+ select {
+ case <-upc:
+ upc = nil
+ default:
+ }
+ switch {
+ case downc == nil && upc == nil:
+ // stale
+ select {
+ case <-b.stopc:
+ return
+ default:
+ }
+ case downc == nil:
+ b.notifyAddrs(NotifyReset)
+ select {
+ case <-upc:
+ case msg := <-b.updateAddrsC:
+ b.notifyAddrs(msg)
+ case <-b.stopc:
+ return
+ }
+ case upc == nil:
+ select {
+ // close connections that are not the pinned address
+ case b.notifyCh <- []grpc.Address{{Addr: addr}}:
+ case <-downc:
+ case <-b.stopc:
+ return
+ }
+ select {
+ case <-downc:
+ b.notifyAddrs(NotifyReset)
+ case msg := <-b.updateAddrsC:
+ b.notifyAddrs(msg)
+ case <-b.stopc:
+ return
+ }
+ }
+ }
+}
+
+func (b *GRPC17Health) notifyAddrs(msg NotifyMsg) {
+ if msg == NotifyNext {
+ select {
+ case b.notifyCh <- []grpc.Address{}:
+ case <-b.stopc:
+ return
+ }
+ }
+ b.mu.RLock()
+ pinAddr := b.pinAddr
+ downc := b.downc
+ b.mu.RUnlock()
+ addrs, hostPorts := b.liveAddrs()
+
+ var waitDown bool
+ if pinAddr != "" {
+ _, ok := hostPorts[pinAddr]
+ waitDown = !ok
+ }
+
+ select {
+ case b.notifyCh <- addrs:
+ if waitDown {
+ select {
+ case <-downc:
+ case <-b.stopc:
+ }
+ }
+ case <-b.stopc:
+ }
+}
+
+func (b *GRPC17Health) Up(addr grpc.Address) func(error) {
+ if !b.mayPin(addr) {
+ return func(err error) {}
+ }
+
+ b.mu.Lock()
+ defer b.mu.Unlock()
+
+ // gRPC might call Up after it called Close. We add this check
+ // to "fix" it up at application layer. Otherwise, will panic
+ // if b.upc is already closed.
+ if b.closed {
+ return func(err error) {}
+ }
+
+ // gRPC might call Up on a stale address.
+ // Prevent updating pinAddr with a stale address.
+ if !hasAddr(b.addrs, addr.Addr) {
+ return func(err error) {}
+ }
+
+ if b.pinAddr != "" {
+ lg.Infof("clientv3/balancer: %q is up but not pinned (already pinned %q)", addr.Addr, b.pinAddr)
+ return func(err error) {}
+ }
+
+ // notify waiting Get()s and pin first connected address
+ close(b.upc)
+ b.downc = make(chan struct{})
+ b.pinAddr = addr.Addr
+ lg.Infof("clientv3/balancer: pin %q", addr.Addr)
+
+ // notify client that a connection is up
+ b.readyOnce.Do(func() { close(b.readyc) })
+
+ return func(err error) {
+ // If connected to a black hole endpoint or a killed server, the gRPC ping
+ // timeout will induce a network I/O error, and retrying until success;
+ // finding healthy endpoint on retry could take several timeouts and redials.
+ // To avoid wasting retries, gray-list unhealthy endpoints.
+ b.HostPortError(addr.Addr, err)
+
+ b.mu.Lock()
+ b.upc = make(chan struct{})
+ close(b.downc)
+ b.pinAddr = ""
+ b.mu.Unlock()
+ lg.Infof("clientv3/balancer: unpin %q (%q)", addr.Addr, err.Error())
+ }
+}
+
+func (b *GRPC17Health) mayPin(addr grpc.Address) bool {
+ if b.Endpoint(addr.Addr) == "" { // stale host:port
+ return false
+ }
+
+ b.unhealthyMu.RLock()
+ unhealthyCnt := len(b.unhealthyHostPorts)
+ failedTime, bad := b.unhealthyHostPorts[addr.Addr]
+ b.unhealthyMu.RUnlock()
+
+ b.mu.RLock()
+ skip := len(b.addrs) == 1 || unhealthyCnt == 0 || len(b.addrs) == unhealthyCnt
+ b.mu.RUnlock()
+ if skip || !bad {
+ return true
+ }
+
+ // prevent isolated member's endpoint from being infinitely retried, as follows:
+ // 1. keepalive pings detects GoAway with http2.ErrCodeEnhanceYourCalm
+ // 2. balancer 'Up' unpins with grpc: failed with network I/O error
+ // 3. grpc-healthcheck still SERVING, thus retry to pin
+ // instead, return before grpc-healthcheck if failed within healthcheck timeout
+ if elapsed := time.Since(failedTime); elapsed < b.healthCheckTimeout {
+ lg.Infof("clientv3/balancer: %q is up but not pinned (failed %v ago, require minimum %v after failure)", addr.Addr, elapsed, b.healthCheckTimeout)
+ return false
+ }
+
+ if ok, _ := b.healthCheck(addr.Addr); ok {
+ b.removeUnhealthy(addr.Addr, "health check success")
+ return true
+ }
+
+ b.HostPortError(addr.Addr, errors.New("health check failed"))
+ return false
+}
+
+func (b *GRPC17Health) Get(ctx context.Context, opts grpc.BalancerGetOptions) (grpc.Address, func(), error) {
+ var (
+ addr string
+ closed bool
+ )
+
+ // If opts.BlockingWait is false (for fail-fast RPCs), it should return
+ // an address it has notified via Notify immediately instead of blocking.
+ if !opts.BlockingWait {
+ b.mu.RLock()
+ closed = b.closed
+ addr = b.pinAddr
+ b.mu.RUnlock()
+ if closed {
+ return grpc.Address{Addr: ""}, nil, grpc.ErrClientConnClosing
+ }
+ if addr == "" {
+ return grpc.Address{Addr: ""}, nil, ErrNoAddrAvilable
+ }
+ return grpc.Address{Addr: addr}, func() {}, nil
+ }
+
+ for {
+ b.mu.RLock()
+ ch := b.upc
+ b.mu.RUnlock()
+ select {
+ case <-ch:
+ case <-b.donec:
+ return grpc.Address{Addr: ""}, nil, grpc.ErrClientConnClosing
+ case <-ctx.Done():
+ return grpc.Address{Addr: ""}, nil, ctx.Err()
+ }
+ b.mu.RLock()
+ closed = b.closed
+ addr = b.pinAddr
+ b.mu.RUnlock()
+ // Close() which sets b.closed = true can be called before Get(), Get() must exit if balancer is closed.
+ if closed {
+ return grpc.Address{Addr: ""}, nil, grpc.ErrClientConnClosing
+ }
+ if addr != "" {
+ break
+ }
+ }
+ return grpc.Address{Addr: addr}, func() {}, nil
+}
+
+func (b *GRPC17Health) Notify() <-chan []grpc.Address { return b.notifyCh }
+
+func (b *GRPC17Health) Close() error {
+ b.mu.Lock()
+ // In case gRPC calls close twice. TODO: remove the checking
+ // when we are sure that gRPC wont call close twice.
+ if b.closed {
+ b.mu.Unlock()
+ <-b.donec
+ return nil
+ }
+ b.closed = true
+ b.stopOnce.Do(func() { close(b.stopc) })
+ b.pinAddr = ""
+
+ // In the case of following scenario:
+ // 1. upc is not closed; no pinned address
+ // 2. client issues an RPC, calling invoke(), which calls Get(), enters for loop, blocks
+ // 3. client.conn.Close() calls balancer.Close(); closed = true
+ // 4. for loop in Get() never exits since ctx is the context passed in by the client and may not be canceled
+ // we must close upc so Get() exits from blocking on upc
+ select {
+ case <-b.upc:
+ default:
+ // terminate all waiting Get()s
+ close(b.upc)
+ }
+
+ b.mu.Unlock()
+ b.wg.Wait()
+
+ // wait for updateNotifyLoop to finish
+ <-b.donec
+ close(b.notifyCh)
+
+ return nil
+}
+
+func grpcHealthCheck(ep string, dialFunc func(ep string, dopts ...grpc.DialOption) (*grpc.ClientConn, error)) (bool, error) {
+ conn, err := dialFunc(ep)
+ if err != nil {
+ return false, err
+ }
+ defer conn.Close()
+ cli := healthpb.NewHealthClient(conn)
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ resp, err := cli.Check(ctx, &healthpb.HealthCheckRequest{})
+ cancel()
+ if err != nil {
+ if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable {
+ if s.Message() == unknownService { // etcd < v3.3.0
+ return true, nil
+ }
+ }
+ return false, err
+ }
+ return resp.Status == healthpb.HealthCheckResponse_SERVING, nil
+}
+
+func hasAddr(addrs []grpc.Address, targetAddr string) bool {
+ for _, addr := range addrs {
+ if targetAddr == addr.Addr {
+ return true
+ }
+ }
+ return false
+}
+
+func getHost(ep string) string {
+ url, uerr := url.Parse(ep)
+ if uerr != nil || !strings.Contains(ep, "://") {
+ return ep
+ }
+ return url.Host
+}
+
+func eps2addrs(eps []string) []grpc.Address {
+ addrs := make([]grpc.Address, len(eps))
+ for i := range eps {
+ addrs[i].Addr = getHost(eps[i])
+ }
+ return addrs
+}
+
+func getHostPort2ep(eps []string) map[string]string {
+ hm := make(map[string]string, len(eps))
+ for i := range eps {
+ _, host, _ := parseEndpoint(eps[i])
+ hm[host] = eps[i]
+ }
+ return hm
+}
+
+func parseEndpoint(endpoint string) (proto string, host string, scheme string) {
+ proto = "tcp"
+ host = endpoint
+ url, uerr := url.Parse(endpoint)
+ if uerr != nil || !strings.Contains(endpoint, "://") {
+ return proto, host, scheme
+ }
+ scheme = url.Scheme
+
+ // strip scheme:// prefix since grpc dials by host
+ host = url.Host
+ switch url.Scheme {
+ case "http", "https":
+ case "unix", "unixs":
+ proto = "unix"
+ host = url.Host + url.Path
+ default:
+ proto, host = "", ""
+ }
+ return proto, host, scheme
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/picker/doc.go b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/doc.go
new file mode 100644
index 0000000..35dabf5
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/doc.go
@@ -0,0 +1,16 @@
+// Copyright 2018 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 picker defines/implements client balancer picker policy.
+package picker
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/picker/err.go b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/err.go
new file mode 100644
index 0000000..c70ce15
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/err.go
@@ -0,0 +1,34 @@
+// Copyright 2018 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 picker
+
+import (
+ "context"
+
+ "google.golang.org/grpc/balancer"
+)
+
+// NewErr returns a picker that always returns err on "Pick".
+func NewErr(err error) Picker {
+ return &errPicker{err: err}
+}
+
+type errPicker struct {
+ err error
+}
+
+func (p *errPicker) Pick(context.Context, balancer.PickOptions) (balancer.SubConn, func(balancer.DoneInfo), error) {
+ return nil, nil, p.err
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/picker/picker.go b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/picker.go
new file mode 100644
index 0000000..7ea761b
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/picker.go
@@ -0,0 +1,24 @@
+// Copyright 2018 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 picker
+
+import (
+ "google.golang.org/grpc/balancer"
+)
+
+// Picker defines balancer Picker methods.
+type Picker interface {
+ balancer.Picker
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/picker/picker_policy.go b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/picker_policy.go
new file mode 100644
index 0000000..463ddc2
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/picker_policy.go
@@ -0,0 +1,49 @@
+// Copyright 2018 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 picker
+
+import "fmt"
+
+// Policy defines balancer picker policy.
+type Policy uint8
+
+const (
+ // TODO: custom picker is not supported yet.
+ // custom defines custom balancer picker.
+ custom Policy = iota
+
+ // RoundrobinBalanced balance loads over multiple endpoints
+ // and implements failover in roundrobin fashion.
+ RoundrobinBalanced Policy = iota
+
+ // TODO: only send loads to pinned address "RoundrobinFailover"
+ // just like how 3.3 client works
+ //
+ // TODO: priotize leader
+ // TODO: health-check
+ // TODO: weighted roundrobin
+ // TODO: power of two random choice
+)
+
+func (p Policy) String() string {
+ switch p {
+ case custom:
+ panic("'custom' picker policy is not supported yet")
+ case RoundrobinBalanced:
+ return "etcd-client-roundrobin-balanced"
+ default:
+ panic(fmt.Errorf("invalid balancer picker policy (%d)", p))
+ }
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/picker/roundrobin_balanced.go b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/roundrobin_balanced.go
new file mode 100644
index 0000000..b043d57
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/picker/roundrobin_balanced.go
@@ -0,0 +1,92 @@
+// Copyright 2018 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 picker
+
+import (
+ "context"
+ "sync"
+
+ "go.uber.org/zap"
+ "go.uber.org/zap/zapcore"
+ "google.golang.org/grpc/balancer"
+ "google.golang.org/grpc/resolver"
+)
+
+// NewRoundrobinBalanced returns a new roundrobin balanced picker.
+func NewRoundrobinBalanced(
+ lg *zap.Logger,
+ scs []balancer.SubConn,
+ addrToSc map[resolver.Address]balancer.SubConn,
+ scToAddr map[balancer.SubConn]resolver.Address,
+) Picker {
+ return &rrBalanced{
+ lg: lg,
+ scs: scs,
+ addrToSc: addrToSc,
+ scToAddr: scToAddr,
+ }
+}
+
+type rrBalanced struct {
+ lg *zap.Logger
+
+ mu sync.RWMutex
+ next int
+ scs []balancer.SubConn
+
+ addrToSc map[resolver.Address]balancer.SubConn
+ scToAddr map[balancer.SubConn]resolver.Address
+}
+
+// Pick is called for every client request.
+func (rb *rrBalanced) Pick(ctx context.Context, opts balancer.PickOptions) (balancer.SubConn, func(balancer.DoneInfo), error) {
+ rb.mu.RLock()
+ n := len(rb.scs)
+ rb.mu.RUnlock()
+ if n == 0 {
+ return nil, nil, balancer.ErrNoSubConnAvailable
+ }
+
+ rb.mu.Lock()
+ cur := rb.next
+ sc := rb.scs[cur]
+ picked := rb.scToAddr[sc].Addr
+ rb.next = (rb.next + 1) % len(rb.scs)
+ rb.mu.Unlock()
+
+ rb.lg.Debug(
+ "picked",
+ zap.String("address", picked),
+ zap.Int("subconn-index", cur),
+ zap.Int("subconn-size", n),
+ )
+
+ doneFunc := func(info balancer.DoneInfo) {
+ // TODO: error handling?
+ fss := []zapcore.Field{
+ zap.Error(info.Err),
+ zap.String("address", picked),
+ zap.Bool("success", info.Err == nil),
+ zap.Bool("bytes-sent", info.BytesSent),
+ zap.Bool("bytes-received", info.BytesReceived),
+ }
+ if info.Err == nil {
+ rb.lg.Debug("balancer done", fss...)
+ } else {
+ rb.lg.Warn("balancer failed", fss...)
+ }
+ }
+ return sc, doneFunc, nil
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/resolver/endpoint/endpoint.go b/vendor/go.etcd.io/etcd/clientv3/balancer/resolver/endpoint/endpoint.go
new file mode 100644
index 0000000..1f32039
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/resolver/endpoint/endpoint.go
@@ -0,0 +1,240 @@
+// Copyright 2018 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 endpoint resolves etcd entpoints using grpc targets of the form 'endpoint://<id>/<endpoint>'.
+package endpoint
+
+import (
+ "fmt"
+ "net/url"
+ "strings"
+ "sync"
+
+ "google.golang.org/grpc/resolver"
+)
+
+const scheme = "endpoint"
+
+var (
+ targetPrefix = fmt.Sprintf("%s://", scheme)
+
+ bldr *builder
+)
+
+func init() {
+ bldr = &builder{
+ resolverGroups: make(map[string]*ResolverGroup),
+ }
+ resolver.Register(bldr)
+}
+
+type builder struct {
+ mu sync.RWMutex
+ resolverGroups map[string]*ResolverGroup
+}
+
+// NewResolverGroup creates a new ResolverGroup with the given id.
+func NewResolverGroup(id string) (*ResolverGroup, error) {
+ return bldr.newResolverGroup(id)
+}
+
+// ResolverGroup keeps all endpoints of resolvers using a common endpoint://<id>/ target
+// up-to-date.
+type ResolverGroup struct {
+ mu sync.RWMutex
+ id string
+ endpoints []string
+ resolvers []*Resolver
+}
+
+func (e *ResolverGroup) addResolver(r *Resolver) {
+ e.mu.Lock()
+ addrs := epsToAddrs(e.endpoints...)
+ e.resolvers = append(e.resolvers, r)
+ e.mu.Unlock()
+ r.cc.NewAddress(addrs)
+}
+
+func (e *ResolverGroup) removeResolver(r *Resolver) {
+ e.mu.Lock()
+ for i, er := range e.resolvers {
+ if er == r {
+ e.resolvers = append(e.resolvers[:i], e.resolvers[i+1:]...)
+ break
+ }
+ }
+ e.mu.Unlock()
+}
+
+// SetEndpoints updates the endpoints for ResolverGroup. All registered resolver are updated
+// immediately with the new endpoints.
+func (e *ResolverGroup) SetEndpoints(endpoints []string) {
+ addrs := epsToAddrs(endpoints...)
+ e.mu.Lock()
+ e.endpoints = endpoints
+ for _, r := range e.resolvers {
+ r.cc.NewAddress(addrs)
+ }
+ e.mu.Unlock()
+}
+
+// Target constructs a endpoint target using the endpoint id of the ResolverGroup.
+func (e *ResolverGroup) Target(endpoint string) string {
+ return Target(e.id, endpoint)
+}
+
+// Target constructs a endpoint resolver target.
+func Target(id, endpoint string) string {
+ return fmt.Sprintf("%s://%s/%s", scheme, id, endpoint)
+}
+
+// IsTarget checks if a given target string in an endpoint resolver target.
+func IsTarget(target string) bool {
+ return strings.HasPrefix(target, "endpoint://")
+}
+
+func (e *ResolverGroup) Close() {
+ bldr.close(e.id)
+}
+
+// Build creates or reuses an etcd resolver for the etcd cluster name identified by the authority part of the target.
+func (b *builder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOption) (resolver.Resolver, error) {
+ if len(target.Authority) < 1 {
+ return nil, fmt.Errorf("'etcd' target scheme requires non-empty authority identifying etcd cluster being routed to")
+ }
+ id := target.Authority
+ es, err := b.getResolverGroup(id)
+ if err != nil {
+ return nil, fmt.Errorf("failed to build resolver: %v", err)
+ }
+ r := &Resolver{
+ endpointID: id,
+ cc: cc,
+ }
+ es.addResolver(r)
+ return r, nil
+}
+
+func (b *builder) newResolverGroup(id string) (*ResolverGroup, error) {
+ b.mu.RLock()
+ _, ok := b.resolverGroups[id]
+ b.mu.RUnlock()
+ if ok {
+ return nil, fmt.Errorf("Endpoint already exists for id: %s", id)
+ }
+
+ es := &ResolverGroup{id: id}
+ b.mu.Lock()
+ b.resolverGroups[id] = es
+ b.mu.Unlock()
+ return es, nil
+}
+
+func (b *builder) getResolverGroup(id string) (*ResolverGroup, error) {
+ b.mu.RLock()
+ es, ok := b.resolverGroups[id]
+ b.mu.RUnlock()
+ if !ok {
+ return nil, fmt.Errorf("ResolverGroup not found for id: %s", id)
+ }
+ return es, nil
+}
+
+func (b *builder) close(id string) {
+ b.mu.Lock()
+ delete(b.resolverGroups, id)
+ b.mu.Unlock()
+}
+
+func (b *builder) Scheme() string {
+ return scheme
+}
+
+// Resolver provides a resolver for a single etcd cluster, identified by name.
+type Resolver struct {
+ endpointID string
+ cc resolver.ClientConn
+ sync.RWMutex
+}
+
+// TODO: use balancer.epsToAddrs
+func epsToAddrs(eps ...string) (addrs []resolver.Address) {
+ addrs = make([]resolver.Address, 0, len(eps))
+ for _, ep := range eps {
+ addrs = append(addrs, resolver.Address{Addr: ep})
+ }
+ return addrs
+}
+
+func (*Resolver) ResolveNow(o resolver.ResolveNowOption) {}
+
+func (r *Resolver) Close() {
+ es, err := bldr.getResolverGroup(r.endpointID)
+ if err != nil {
+ return
+ }
+ es.removeResolver(r)
+}
+
+// ParseEndpoint endpoint parses an endpoint of the form
+// (http|https)://<host>*|(unix|unixs)://<path>)
+// and returns a protocol ('tcp' or 'unix'),
+// host (or filepath if a unix socket),
+// scheme (http, https, unix, unixs).
+func ParseEndpoint(endpoint string) (proto string, host string, scheme string) {
+ proto = "tcp"
+ host = endpoint
+ url, uerr := url.Parse(endpoint)
+ if uerr != nil || !strings.Contains(endpoint, "://") {
+ return proto, host, scheme
+ }
+ scheme = url.Scheme
+
+ // strip scheme:// prefix since grpc dials by host
+ host = url.Host
+ switch url.Scheme {
+ case "http", "https":
+ case "unix", "unixs":
+ proto = "unix"
+ host = url.Host + url.Path
+ default:
+ proto, host = "", ""
+ }
+ return proto, host, scheme
+}
+
+// ParseTarget parses a endpoint://<id>/<endpoint> string and returns the parsed id and endpoint.
+// If the target is malformed, an error is returned.
+func ParseTarget(target string) (string, string, error) {
+ noPrefix := strings.TrimPrefix(target, targetPrefix)
+ if noPrefix == target {
+ return "", "", fmt.Errorf("malformed target, %s prefix is required: %s", targetPrefix, target)
+ }
+ parts := strings.SplitN(noPrefix, "/", 2)
+ if len(parts) != 2 {
+ return "", "", fmt.Errorf("malformed target, expected %s://<id>/<endpoint>, but got %s", scheme, target)
+ }
+ return parts[0], parts[1], nil
+}
+
+// ParseHostPort splits a "<host>:<port>" string into the host and port parts.
+// The port part is optional.
+func ParseHostPort(hostPort string) (host string, port string) {
+ parts := strings.SplitN(hostPort, ":", 2)
+ host = parts[0]
+ if len(parts) > 1 {
+ port = parts[1]
+ }
+ return host, port
+}
diff --git a/vendor/go.etcd.io/etcd/clientv3/balancer/utils.go b/vendor/go.etcd.io/etcd/clientv3/balancer/utils.go
new file mode 100644
index 0000000..a11faeb
--- /dev/null
+++ b/vendor/go.etcd.io/etcd/clientv3/balancer/utils.go
@@ -0,0 +1,68 @@
+// Copyright 2018 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 balancer
+
+import (
+ "fmt"
+ "net/url"
+ "sort"
+ "sync/atomic"
+ "time"
+
+ "google.golang.org/grpc/balancer"
+ "google.golang.org/grpc/resolver"
+)
+
+func scToString(sc balancer.SubConn) string {
+ return fmt.Sprintf("%p", sc)
+}
+
+func scsToStrings(scs map[resolver.Address]balancer.SubConn) (ss []string) {
+ ss = make([]string, 0, len(scs))
+ for a, sc := range scs {
+ ss = append(ss, fmt.Sprintf("%s (%s)", a.Addr, scToString(sc)))
+ }
+ sort.Strings(ss)
+ return ss
+}
+
+func addrsToStrings(addrs []resolver.Address) (ss []string) {
+ ss = make([]string, len(addrs))
+ for i := range addrs {
+ ss[i] = addrs[i].Addr
+ }
+ sort.Strings(ss)
+ return ss
+}
+
+func epsToAddrs(eps ...string) (addrs []resolver.Address) {
+ addrs = make([]resolver.Address, 0, len(eps))
+ for _, ep := range eps {
+ u, err := url.Parse(ep)
+ if err != nil {
+ addrs = append(addrs, resolver.Address{Addr: ep, Type: resolver.Backend})
+ continue
+ }
+ addrs = append(addrs, resolver.Address{Addr: u.Host, Type: resolver.Backend})
+ }
+ return addrs
+}
+
+var genN = new(uint32)
+
+func genName() string {
+ now := time.Now().UnixNano()
+ return fmt.Sprintf("%X%X", now, atomic.AddUint32(genN, 1))
+}