clientv3: use grpc codes to translate raw grpc errors

clientv3/client.go

@@ -29,6 +29,7 @@ import (
 
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/metadata"
 )
@@ -294,17 +295,7 @@ func isHaltErr(ctx context.Context, err error) bool {
 	if err == nil {
 		return false
 	}
-	eErr := rpctypes.Error(err)
-	if _, ok := eErr.(rpctypes.EtcdError); ok {
-		return eErr != rpctypes.ErrStopped && eErr != rpctypes.ErrNoLeader
-	}
-	// treat etcdserver errors not recognized by the client as halting
-	return isConnClosing(err) || strings.Contains(err.Error(), "etcdserver:")
-}
-
-// isConnClosing returns true if the error matches a grpc client closing error
-func isConnClosing(err error) bool {
-	return strings.Contains(err.Error(), grpc.ErrClientConnClosing.Error())
+	return grpc.Code(err) != codes.Unavailable
 }
 
 func toErr(ctx context.Context, err error) error {
@@ -312,12 +303,20 @@ func toErr(ctx context.Context, err error) error {
 		return nil
 	}
 	err = rpctypes.Error(err)
-	switch {
-	case ctx.Err() != nil && strings.Contains(err.Error(), "context"):
-		err = ctx.Err()
-	case strings.Contains(err.Error(), ErrNoAvailableEndpoints.Error()):
+	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:
 		err = ErrNoAvailableEndpoints
-	case strings.Contains(err.Error(), grpc.ErrClientConnClosing.Error()):
+	case codes.FailedPrecondition:
 		err = grpc.ErrClientConnClosing
 	}
 	return err

clientv3/client_test.go

@@ -19,7 +19,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/coreos/etcd/etcdserver"
+	"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
 	"github.com/coreos/etcd/pkg/testutil"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
@@ -72,11 +72,11 @@ func TestIsHaltErr(t *testing.T) {
 	if !isHaltErr(nil, fmt.Errorf("etcdserver: some etcdserver error")) {
 		t.Errorf(`error prefixed with "etcdserver: " should be Halted by default`)
 	}
-	if isHaltErr(nil, etcdserver.ErrStopped) {
-		t.Errorf("error %v should not halt", etcdserver.ErrStopped)
+	if isHaltErr(nil, rpctypes.ErrGRPCStopped) {
+		t.Errorf("error %v should not halt", rpctypes.ErrGRPCStopped)
 	}
-	if isHaltErr(nil, etcdserver.ErrNoLeader) {
-		t.Errorf("error %v should not halt", etcdserver.ErrNoLeader)
+	if isHaltErr(nil, rpctypes.ErrGRPCNoLeader) {
+		t.Errorf("error %v should not halt", rpctypes.ErrGRPCNoLeader)
 	}
 	ctx, cancel := context.WithCancel(context.TODO())
 	if isHaltErr(ctx, nil) {

clientv3/retry.go

@@ -15,9 +15,11 @@
 package clientv3
 
 import (
+	"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
 	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 )
 
 type rpcFunc func(ctx context.Context) error
@@ -27,8 +29,16 @@ func (c *Client) newRetryWrapper() retryRpcFunc {
 	return func(rpcCtx context.Context, f rpcFunc) {
 		for {
 			err := f(rpcCtx)
-			// ignore grpc conn closing on fail-fast calls; they are transient errors
-			if err == nil || !isConnClosing(err) {
+			if err == nil {
+				return
+			}
+			// only retry if unavailable
+			if grpc.Code(err) != codes.Unavailable {
+				return
+			}
+			// always stop retry on etcd errors
+			eErr := rpctypes.Error(err)
+			if _, ok := eErr.(rpctypes.EtcdError); ok {
 				return
 			}
 			select {