From 2859e9f94608b44c9c351c7ccfcff21665573627 Mon Sep 17 00:00:00 2001

From: Sky Ao <aoxiaojian@gmail.com>

Date: Thu, 15 Oct 2020 15:34:16 +0800

Subject: [PATCH 1/4] update grpc-go version to v1.32.0 which has some breaking

 api changes in naming and loadbalancer package

---

 clientv3/balancer/balancer.go                 |  39 ++-

 clientv3/balancer/picker/err.go               |   6 +-

 .../balancer/picker/roundrobin_balanced.go    |   7 +-

 clientv3/naming/grpc.go                       |  27 +-

 clientv3/naming/grpcnaming/naming.go          |  73 +++++

 proxy/grpcproxy/cluster.go                    |   2 +-

 proxy/grpcproxy/register.go                   |   2 +-

 7 files changed, 424 insertions(+), 46 deletions(-)

 create mode 100644 clientv3/naming/grpcnaming/naming.go

diff --git a/clientv3/balancer/balancer.go b/clientv3/balancer/balancer.go

index 3eecb9d1d2..d3e672c632 100644

--- a/clientv3/balancer/balancer.go

+++ b/clientv3/balancer/balancer.go

@@ -136,15 +136,34 @@ 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))

+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 +210,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 +270,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() {

diff --git a/clientv3/balancer/picker/err.go b/clientv3/balancer/picker/err.go

index f4b941d652..a37baa7bd2 100644

--- a/clientv3/balancer/picker/err.go

+++ b/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

 }

diff --git a/clientv3/balancer/picker/roundrobin_balanced.go b/clientv3/balancer/picker/roundrobin_balanced.go

index e3971ecc42..993c05ec13 100644

--- a/clientv3/balancer/picker/roundrobin_balanced.go

+++ b/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

 }

diff --git a/clientv3/naming/grpc.go b/clientv3/naming/grpc.go

index b680274bb3..f7cd2bd8a6 100644

--- a/clientv3/naming/grpc.go

+++ b/clientv3/naming/grpc.go

@@ -20,9 +20,9 @@ import (

 	"fmt"

 	etcd "go.etcd.io/etcd/clientv3"

+	gnaming "go.etcd.io/etcd/clientv3/naming/grpcnaming"

 	"google.golang.org/grpc/codes"

-	"google.golang.org/grpc/naming"

 	"google.golang.org/grpc/status"

 )

@@ -34,15 +33,15 @@ 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")

@@ -50,7 +49,7 @@ func (gr *GRPCResolver) Update(ctx context.Context, target string, nm naming.Upd

 	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,7 +107,7 @@ 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())

@@ -116,9 +115,9 @@ func (gw *gRPCWatcher) firstNext() ([]*naming.Update, error) {

 		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

 		}

diff --git a/clientv3/naming/grpcnaming/naming.go b/clientv3/naming/grpcnaming/naming.go

new file mode 100644

index 0000000000..9b1f20cc61

--- /dev/null

+++ b/clientv3/naming/grpcnaming/naming.go

@@ -0,0 +1,73 @@

+/*

+ *

+ * 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.

+

+

+// Notice: this file is a copy of naming/naming.go from grpc-go v1.29.1.

+// The package of grpc naming is removed since grpc-go v1.30.0.

+// This is a work around to make etcd work with grpc new version (>=v1.30.0) without too many code change.

+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()

+}

\ No newline at end of file

diff --git a/proxy/grpcproxy/cluster.go b/proxy/grpcproxy/cluster.go

index 5f3ab76584..a786457a6a 100644

--- a/proxy/grpcproxy/cluster.go

+++ b/proxy/grpcproxy/cluster.go

@@ -18,6 +18,7 @@ import (

 	"context"

 	"errors"

 	"fmt"

+	gnaming "go.etcd.io/etcd/clientv3/naming/grpcnaming"

 	"os"

 	"sync"

@@ -27,7 +28,6 @@ import (

 	pb "go.etcd.io/etcd/etcdserver/etcdserverpb"

 	"golang.org/x/time/rate"

-	gnaming "google.golang.org/grpc/naming"

 )

 // allow maximum 1 retry per second

diff --git a/proxy/grpcproxy/register.go b/proxy/grpcproxy/register.go

index e74cd5ca90..912e147d77 100644

--- a/proxy/grpcproxy/register.go

+++ b/proxy/grpcproxy/register.go

@@ -16,6 +16,7 @@ package grpcproxy

 import (

 	"encoding/json"

+	gnaming "go.etcd.io/etcd/clientv3/naming/grpcnaming"

 	"os"

 	"go.etcd.io/etcd/clientv3"

@@ -23,7 +24,6 @@ import (

 	"go.etcd.io/etcd/clientv3/naming"

 	"golang.org/x/time/rate"

-	gnaming "google.golang.org/grpc/naming"

 )

 // allow maximum 1 retry per second

From b1d600ee05737adf6fc2ba3a5c3d92c1037bb6a6 Mon Sep 17 00:00:00 2001

From: Sky Ao <aoxiaojian@gmail.com>

Date: Thu, 22 Oct 2020 11:20:17 +0800

Subject: [PATCH 3/4] add doc for method ResolveError()

---

 clientv3/balancer/balancer.go | 1 +

 1 file changed, 1 insertion(+)

diff --git a/clientv3/balancer/balancer.go b/clientv3/balancer/balancer.go

index d3e672c632..1488580331 100644

--- a/clientv3/balancer/balancer.go

+++ b/clientv3/balancer/balancer.go

@@ -140,6 +140,7 @@ type baseBalancer struct {

 	resolverErr error // the last error reported by the resolver; cleared on successful resolution

 }

+// ResolverError implements "grpc/balancer.Balancer" interface.

 func (bb *baseBalancer) ResolverError(err error) {

 	bb.resolverErr = err

 	if len(bb.addrToSc) == 0 {

From 5ba58f2283eda7247f51b4e9dd07c367c4d77235 Mon Sep 17 00:00:00 2001

From: Sky Ao <aoxiaojian@gmail.com>

Date: Thu, 22 Oct 2020 11:36:03 +0800

Subject: [PATCH 4/4] update to pass fmt checking

---

 clientv3/naming/grpcnaming/naming.go | 7 +------

 1 file changed, 1 insertion(+), 6 deletions(-)

diff --git a/clientv3/naming/grpcnaming/naming.go b/clientv3/naming/grpcnaming/naming.go

index 9b1f20cc61..a4415a6541 100644

--- a/clientv3/naming/grpcnaming/naming.go

+++ b/clientv3/naming/grpcnaming/naming.go

@@ -19,11 +19,6 @@

 // Package naming defines the naming API and related data structures for gRPC.

 //

 // This package is deprecated: please use package resolver instead.

-

-

-// Notice: this file is a copy of naming/naming.go from grpc-go v1.29.1.

-// The package of grpc naming is removed since grpc-go v1.30.0.

-// This is a work around to make etcd work with grpc new version (>=v1.30.0) without too many code change.

 package grpcnaming

 // Operation defines the corresponding operations for a name resolution change.

@@ -70,4 +65,4 @@ type Watcher interface {

 	Next() ([]*Update, error)

 	// Close closes the Watcher.

 	Close()

-}

\ No newline at end of file

+}