update grpc-go version to v1.32.0 which has some breaking api changes

clientv3/balancer/balancer.go

@@ -136,15 +136,35 @@ type baseBalancer struct {
 	connectivityRecorder connectivity.Recorder
 
 	picker picker.Picker
+
+	resolverErr error // the last error reported by the resolver; cleared on successful resolution
 }
 
-// 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))
+// ResolverError implements "grpc/balancer.Balancer" interface.
+func (bb *baseBalancer) ResolverError(err error) {
+	bb.resolverErr = err
+	if len(bb.addrToSc) == 0 {
+		bb.connectivityRecorder.RecordTransition(bb.connectivityRecorder.GetCurrentState(), grpcconnectivity.TransientFailure)
+	}
+
+	if bb.connectivityRecorder.GetCurrentState() != grpcconnectivity.TransientFailure {
+		// The picker will not change since the balancer does not currently
+		// report an error.
 		return
 	}
+	bb.updatePicker()
+	bb.currentConn.UpdateState(balancer.State{
+		ConnectivityState: bb.connectivityRecorder.GetCurrentState(),
+		Picker:            bb.picker,
+	})
+}
+
+// UpdateClientConnState implements "grpc/balancer.Balancer" interface.
+func (bb *baseBalancer) UpdateClientConnState(state balancer.ClientConnState) error {
+	addrs := state.ResolverState.Addresses
+	// Successful resolution; clear resolver error and ensure we return nil.
+	bb.resolverErr = nil
+
 	bb.lg.Info("resolved",
 		zap.String("picker", bb.picker.String()),
 		zap.String("balancer-id", bb.id),
@@ -191,10 +211,14 @@ func (bb *baseBalancer) HandleResolvedAddrs(addrs []resolver.Address, err error)
 			// (DO NOT) delete(bb.scToSt, sc)
 		}
 	}
+
+	return nil
 }
 
-// HandleSubConnStateChange implements "grpc/balancer.Balancer" interface.
-func (bb *baseBalancer) HandleSubConnStateChange(sc balancer.SubConn, s grpcconnectivity.State) {
+// UpdateSubConnState implements "grpc/balancer.Balancer" interface.
+func (bb *baseBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
+	s := state.ConnectivityState
+
 	bb.mu.Lock()
 	defer bb.mu.Unlock()
 
@@ -247,7 +271,7 @@ func (bb *baseBalancer) HandleSubConnStateChange(sc balancer.SubConn, s grpcconn
 		bb.updatePicker()
 	}
 
-	bb.currentConn.UpdateBalancerState(bb.connectivityRecorder.GetCurrentState(), bb.picker)
+	bb.currentConn.UpdateState(balancer.State{ConnectivityState: bb.connectivityRecorder.GetCurrentState(), Picker: bb.picker})
 }
 
 func (bb *baseBalancer) updatePicker() {

clientv3/balancer/picker/err.go

@@ -15,8 +15,6 @@
 package picker
 
 import (
-	"context"
-
 	"google.golang.org/grpc/balancer"
 )
 
@@ -34,6 +32,6 @@ func (ep *errPicker) String() string {
 	return ep.p.String()
 }
 
-func (ep *errPicker) Pick(context.Context, balancer.PickInfo) (balancer.SubConn, func(balancer.DoneInfo), error) {
-	return nil, nil, ep.err
+func (ep *errPicker) Pick(balancer.PickInfo) (balancer.PickResult, error) {
+	return balancer.PickResult{}, ep.err
 }

clientv3/balancer/picker/roundrobin_balanced.go

@@ -15,7 +15,6 @@
 package picker
 
 import (
-	"context"
 	"sync"
 
 	"go.uber.org/zap"
@@ -52,12 +51,12 @@ type rrBalanced struct {
 func (rb *rrBalanced) String() string { return rb.p.String() }
 
 // Pick is called for every client request.
-func (rb *rrBalanced) Pick(ctx context.Context, opts balancer.PickInfo) (balancer.SubConn, func(balancer.DoneInfo), error) {
+func (rb *rrBalanced) Pick(opts balancer.PickInfo) (balancer.PickResult, error) {
 	rb.mu.RLock()
 	n := len(rb.scs)
 	rb.mu.RUnlock()
 	if n == 0 {
-		return nil, nil, balancer.ErrNoSubConnAvailable
+		return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
 	}
 
 	rb.mu.Lock()
@@ -91,5 +90,5 @@ func (rb *rrBalanced) Pick(ctx context.Context, opts balancer.PickInfo) (balance
 			rb.lg.Warn("balancer failed", fss...)
 		}
 	}
-	return sc, doneFunc, nil
+	return balancer.PickResult{SubConn: sc, Done: doneFunc}, nil
 }

clientv3/naming/grpc.go

@@ -18,11 +18,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-
 	etcd "go.etcd.io/etcd/v3/clientv3"
+	gnaming "go.etcd.io/etcd/v3/clientv3/naming/grpcnaming"
 
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/naming"
 	"google.golang.org/grpc/status"
 )
 
@@ -34,23 +33,23 @@ type GRPCResolver struct {
 	Client *etcd.Client
 }
 
-func (gr *GRPCResolver) Update(ctx context.Context, target string, nm naming.Update, opts ...etcd.OpOption) (err error) {
+func (gr *GRPCResolver) Update(ctx context.Context, target string, nm gnaming.Update, opts ...etcd.OpOption) (err error) {
 	switch nm.Op {
-	case naming.Add:
+	case gnaming.Add:
 		var v []byte
 		if v, err = json.Marshal(nm); err != nil {
 			return status.Error(codes.InvalidArgument, err.Error())
 		}
 		_, err = gr.Client.KV.Put(ctx, target+"/"+nm.Addr, string(v), opts...)
-	case naming.Delete:
+	case gnaming.Delete:
 		_, err = gr.Client.Delete(ctx, target+"/"+nm.Addr, opts...)
 	default:
 		return status.Error(codes.InvalidArgument, "naming: bad naming op")
 	}
 	return err
 }
 
-func (gr *GRPCResolver) Resolve(target string) (naming.Watcher, error) {
+func (gr *GRPCResolver) Resolve(target string) (gnaming.Watcher, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	w := &gRPCWatcher{c: gr.Client, target: target + "/", ctx: ctx, cancel: cancel}
 	return w, nil
@@ -68,7 +67,7 @@ type gRPCWatcher struct {
 // Next gets the next set of updates from the etcd resolver.
 // Calls to Next should be serialized; concurrent calls are not safe since
 // there is no way to reconcile the update ordering.
-func (gw *gRPCWatcher) Next() ([]*naming.Update, error) {
+func (gw *gRPCWatcher) Next() ([]*gnaming.Update, error) {
 	if gw.wch == nil {
 		// first Next() returns all addresses
 		return gw.firstNext()
@@ -87,17 +86,17 @@ func (gw *gRPCWatcher) Next() ([]*naming.Update, error) {
 		return nil, gw.err
 	}
 
-	updates := make([]*naming.Update, 0, len(wr.Events))
+	updates := make([]*gnaming.Update, 0, len(wr.Events))
 	for _, e := range wr.Events {
-		var jupdate naming.Update
+		var jupdate gnaming.Update
 		var err error
 		switch e.Type {
 		case etcd.EventTypePut:
 			err = json.Unmarshal(e.Kv.Value, &jupdate)
-			jupdate.Op = naming.Add
+			jupdate.Op = gnaming.Add
 		case etcd.EventTypeDelete:
 			err = json.Unmarshal(e.PrevKv.Value, &jupdate)
-			jupdate.Op = naming.Delete
+			jupdate.Op = gnaming.Delete
 		default:
 			continue
 		}
@@ -108,17 +107,17 @@ func (gw *gRPCWatcher) Next() ([]*naming.Update, error) {
 	return updates, nil
 }
 
-func (gw *gRPCWatcher) firstNext() ([]*naming.Update, error) {
+func (gw *gRPCWatcher) firstNext() ([]*gnaming.Update, error) {
 	// Use serialized request so resolution still works if the target etcd
 	// server is partitioned away from the quorum.
 	resp, err := gw.c.Get(gw.ctx, gw.target, etcd.WithPrefix(), etcd.WithSerializable())
 	if gw.err = err; err != nil {
 		return nil, err
 	}
 
-	updates := make([]*naming.Update, 0, len(resp.Kvs))
+	updates := make([]*gnaming.Update, 0, len(resp.Kvs))
 	for _, kv := range resp.Kvs {
-		var jupdate naming.Update
+		var jupdate gnaming.Update
 		if err := json.Unmarshal(kv.Value, &jupdate); err != nil {
 			continue
 		}

clientv3/naming/grpcnaming/naming.go

@@ -0,0 +1,68 @@
+/*
+ *
+ * Copyright 2014 gRPC 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 naming defines the naming API and related data structures for gRPC.
+//
+// This package is deprecated: please use package resolver instead.
+package grpcnaming
+
+// Operation defines the corresponding operations for a name resolution change.
+//
+// Deprecated: please use package resolver.
+type Operation uint8
+
+const (
+	// Add indicates a new address is added.
+	Add Operation = iota
+	// Delete indicates an existing address is deleted.
+	Delete
+)
+
+// Update defines a name resolution update. Notice that it is not valid having both
+// empty string Addr and nil Metadata in an Update.
+//
+// Deprecated: please use package resolver.
+type Update struct {
+	// Op indicates the operation of the update.
+	Op Operation
+	// Addr is the updated address. It is empty string if there is no address update.
+	Addr string
+	// Metadata is the updated metadata. It is nil if there is no metadata update.
+	// Metadata is not required for a custom naming implementation.
+	Metadata interface{}
+}
+
+// Resolver creates a Watcher for a target to track its resolution changes.
+//
+// Deprecated: please use package resolver.
+type Resolver interface {
+	// Resolve creates a Watcher for target.
+	Resolve(target string) (Watcher, error)
+}
+
+// Watcher watches for the updates on the specified target.
+//
+// Deprecated: please use package resolver.
+type Watcher interface {
+	// Next blocks until an update or error happens. It may return one or more
+	// updates. The first call should get the full set of the results. It should
+	// return an error if and only if Watcher cannot recover.
+	Next() ([]*Update, error)
+	// Close closes the Watcher.
+	Close()
+}

go.mod

@@ -3,6 +3,7 @@ module go.etcd.io/etcd/v3
 go 1.15
 
 require (
+	cloud.google.com/go/bigquery v1.6.0 // indirect
 	github.com/bgentry/speakeasy v0.1.0
 	github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa
 	github.com/coreos/go-semver v0.2.0
@@ -11,9 +12,13 @@ require (
 	github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4
 	github.com/fatih/color v1.7.0 // indirect
 	github.com/gogo/protobuf v1.3.1
-	github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903
+	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e
+	github.com/golang/mock v1.4.4 // indirect
 	github.com/golang/protobuf v1.3.5
 	github.com/google/btree v1.0.0
+	github.com/google/go-cmp v0.5.1 // indirect
+	github.com/google/martian/v3 v3.0.0 // indirect
+	github.com/google/pprof v0.0.0-20200708004538-1a94d8640e99 // indirect
 	github.com/google/uuid v1.0.0
 	github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4
@@ -39,14 +44,20 @@ require (
 	go.etcd.io/bbolt v1.3.5
 	go.etcd.io/etcd/api/v3 v3.0.0-00010101000000-000000000000
 	go.etcd.io/etcd/pkg/v3 v3.0.0-00010101000000-000000000000
+	go.opencensus.io v0.22.4 // indirect
 	go.uber.org/zap v1.15.0
-	golang.org/x/crypto v0.0.0-20191002192127-34f69633bfdc
-	golang.org/x/net v0.0.0-20191002035440-2ec189313ef0
+	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
+	golang.org/x/net v0.0.0-20200707034311-ab3426394381
 	golang.org/x/text v0.3.3 // indirect
-	golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2
+	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
+	golang.org/x/tools v0.0.0-20200806022845-90696ccdc692 // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	google.golang.org/api v0.29.0 // indirect
+	google.golang.org/appengine v1.6.6 // indirect
 	google.golang.org/genproto v0.0.0-20200513103714-09dca8ec2884
-	google.golang.org/grpc v1.29.1
+	google.golang.org/grpc v1.32.0
 	gopkg.in/cheggaaa/pb.v1 v1.0.25
+	honnef.co/go/tools v0.0.1-2020.1.4 // indirect
 	sigs.k8s.io/yaml v1.1.0
 )