Repository: go-kit/kit
Pull request #1009 (2020-08-30 11:02:07)

Fix etcdv3 client won't return error when no endpoint is available
Kit 's etcdv3 client would not return any error if no endpoint is avilable now.

1. Adding an option to make etcd client will connect happen in background returning the client object before initial connect.
2. Add two test about creating new etcd client.

**Releated Commit:**
- <https://github.com/etcd-io/etcd/commit/4b0af5b4ac160e66cd7af5ef93e94e9d4498b41a>

**Releated Issues:**
-  <https://github.com/etcd-io/etcd/issues/9877>

wayjam: @peterbourgon updated

Patch:
diff --git a/sd/etcdv3/client.go b/sd/etcdv3/client.go
index ffc86fe..0ccb9e6 100644
--- a/sd/etcdv3/client.go
+++ b/sd/etcdv3/client.go
@@ -8,6 +8,7 @@ import (
 
 	"go.etcd.io/etcd/clientv3"
 	"go.etcd.io/etcd/pkg/transport"
+	"google.golang.org/grpc"
 )
 
 var (
@@ -73,8 +74,14 @@ type ClientOptions struct {
 	CACert        string
 	DialTimeout   time.Duration
 	DialKeepAlive time.Duration
-	Username      string
-	Password      string
+
+	// DialOptions is a list of dial options for the gRPC client (e.g., for interceptors).
+	// For example, pass grpc.WithBlock() to block until the underlying connection is up.
+	// Without this, Dial returns immediately and connecting the server happens in background.
+	DialOptions []grpc.DialOption
+
+	Username string
+	Password string
 }
 
 // NewClient returns Client with a connection to the named machines. It will
@@ -107,6 +114,7 @@ func NewClient(ctx context.Context, machines []string, options ClientOptions) (C
 		Endpoints:         machines,
 		DialTimeout:       options.DialTimeout,
 		DialKeepAliveTime: options.DialKeepAlive,
+		DialOptions:       options.DialOptions,
 		TLS:               tlscfg,
 		Username:          options.Username,
 		Password:          options.Password,
diff --git a/transport/http/jsonrpc/client.go b/transport/http/jsonrpc/client.go
index 07b673d..86e4123 100644
--- a/transport/http/jsonrpc/client.go
+++ b/transport/http/jsonrpc/client.go
@@ -196,6 +196,10 @@ func (c Client) Endpoint() endpoint.Endpoint {
 			defer resp.Body.Close()
 		}
 
+		for _, f := range c.after {
+			ctx = f(ctx, resp)
+		}
+
 		// Decode the body into an object
 		var rpcRes Response
 		err = json.NewDecoder(resp.Body).Decode(&rpcRes)
@@ -203,10 +207,6 @@ func (c Client) Endpoint() endpoint.Endpoint {
 			return nil, err
 		}
 
-		for _, f := range c.after {
-			ctx = f(ctx, resp)
-		}
-
 		return c.dec(ctx, rpcRes)
 	}
 }


Test patch:
diff --git a/sd/etcdv3/client_test.go b/sd/etcdv3/client_test.go
new file mode 100644
index 0000000..ff29a73
--- /dev/null
+++ b/sd/etcdv3/client_test.go
@@ -0,0 +1,80 @@
+package etcdv3
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+)
+
+const (
+	// irrelevantEndpoint is an address which does not exists.
+	irrelevantEndpoint = "http://irrelevant:12345"
+)
+
+func TestNewClient(t *testing.T) {
+	client, err := NewClient(
+		context.Background(),
+		[]string{irrelevantEndpoint},
+		ClientOptions{
+			DialTimeout:   3 * time.Second,
+			DialKeepAlive: 3 * time.Second,
+		},
+	)
+	if err != nil {
+		t.Fatalf("unexpected error creating client: %v", err)
+	}
+	if client == nil {
+		t.Fatal("expected new Client, got nil")
+	}
+}
+
+func TestClientOptions(t *testing.T) {
+	client, err := NewClient(
+		context.Background(),
+		[]string{},
+		ClientOptions{
+			Cert:          "",
+			Key:           "",
+			CACert:        "",
+			DialTimeout:   3 * time.Second,
+			DialKeepAlive: 3 * time.Second,
+		},
+	)
+	if err == nil {
+		t.Errorf("expected error: %v", err)
+	}
+	if client != nil {
+		t.Fatalf("expected client to be nil on failure")
+	}
+
+	_, err = NewClient(
+		context.Background(),
+		[]string{irrelevantEndpoint},
+		ClientOptions{
+			Cert:          "does-not-exist.crt",
+			Key:           "does-not-exist.key",
+			CACert:        "does-not-exist.CACert",
+			DialTimeout:   3 * time.Second,
+			DialKeepAlive: 3 * time.Second,
+		},
+	)
+	if err == nil {
+		t.Errorf("expected error: %v", err)
+	}
+
+	client, err = NewClient(
+		context.Background(),
+		[]string{irrelevantEndpoint},
+		ClientOptions{
+			DialOptions: []grpc.DialOption{grpc.WithBlock()},
+		},
+	)
+	if err == nil {
+		t.Errorf("expected connection should fail")
+	}
+	if client != nil {
+		t.Errorf("expected client to be nil on failure")
+	}
+}
diff --git a/sd/etcdv3/example_test.go b/sd/etcdv3/example_test.go
index 1422e31..8acf1d3 100644
--- a/sd/etcdv3/example_test.go
+++ b/sd/etcdv3/example_test.go
@@ -9,6 +9,7 @@ import (
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/sd"
 	"github.com/go-kit/kit/sd/lb"
+	"google.golang.org/grpc"
 )
 
 func Example() {
@@ -44,6 +45,9 @@ func Example() {
 
 		// If DialKeepAlive is 0, it defaults to 3s
 		DialKeepAlive: time.Second * 3,
+
+		// If passing `grpc.WithBlock`, dial connection will block until success.
+		DialOptions: []grpc.DialOption{grpc.WithBlock()},
 	}
 
 	// Build the client.
diff --git a/transport/http/jsonrpc/client_test.go b/transport/http/jsonrpc/client_test.go
index bbc3748..462e0fc 100644
--- a/transport/http/jsonrpc/client_test.go
+++ b/transport/http/jsonrpc/client_test.go
@@ -18,23 +18,107 @@ type TestResponse struct {
 	String string
 }
 
-func TestCanCallBeforeFunc(t *testing.T) {
-	called := false
-	u, _ := url.Parse("http://senseye.io/jsonrpc")
-	sut := jsonrpc.NewClient(
-		u,
-		"add",
-		jsonrpc.ClientBefore(func(ctx context.Context, req *http.Request) context.Context {
-			called = true
-			return ctx
-		}),
-	)
+type testServerResponseOptions struct {
+	Body   string
+	Status int
+}
 
-	sut.Endpoint()(context.TODO(), "foo")
+func httptestServer(t *testing.T) *httptest.Server {
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
 
-	if !called {
-		t.Fatal("Expected client before func to be called. Wasn't.")
+		var testReq jsonrpc.Request
+		if err := json.NewDecoder(r.Body).Decode(&testReq); err != nil {
+			t.Fatal(err)
+		}
+
+		var options testServerResponseOptions
+		if err := json.Unmarshal(testReq.Params, &options); err != nil {
+			t.Fatal(err)
+		}
+
+		if options.Status == 0 {
+			options.Status = http.StatusOK
+		}
+
+		w.WriteHeader(options.Status)
+		w.Write([]byte(options.Body))
+	}))
+}
+
+func TestBeforeAfterFuncs(t *testing.T) {
+	t.Parallel()
+
+	var tests = []struct {
+		name   string
+		status int
+		body   string
+	}{
+		{
+			name: "empty body",
+			body: "",
+		},
+		{
+			name:   "empty body 500",
+			body:   "",
+			status: 500,
+		},
+
+		{
+			name: "empty json body",
+			body: "{}",
+		},
+		{
+			name: "error",
+			body: `{"jsonrpc":"2.0","error":{"code":32603,"message":"Bad thing happened."}}`,
+		},
 	}
+
+	server := httptestServer(t)
+	defer server.Close()
+
+	testUrl, err := url.Parse(server.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			beforeCalled := false
+			afterCalled := false
+			finalizerCalled := false
+
+			sut := jsonrpc.NewClient(
+				testUrl,
+				"dummy",
+				jsonrpc.ClientBefore(func(ctx context.Context, req *http.Request) context.Context {
+					beforeCalled = true
+					return ctx
+				}),
+				jsonrpc.ClientAfter(func(ctx context.Context, resp *http.Response) context.Context {
+					afterCalled = true
+					return ctx
+				}),
+				jsonrpc.ClientFinalizer(func(ctx context.Context, err error) {
+					finalizerCalled = true
+				}),
+			)
+
+			sut.Endpoint()(context.TODO(), testServerResponseOptions{Body: tt.body, Status: tt.status})
+			if !beforeCalled {
+				t.Fatal("Expected client before func to be called. Wasn't.")
+			}
+			if !afterCalled {
+				t.Fatal("Expected client after func to be called. Wasn't.")
+			}
+			if !finalizerCalled {
+				t.Fatal("Expected client finalizer func to be called. Wasn't.")
+			}
+
+		})
+
+	}
+
 }
 
 type staticIDGenerator int
@@ -42,6 +126,8 @@ type staticIDGenerator int
 func (g staticIDGenerator) Generate() interface{} { return g }
 
 func TestClientHappyPath(t *testing.T) {
+	t.Parallel()
+
 	var (
 		afterCalledKey    = "AC"
 		beforeHeaderKey   = "BF"
@@ -92,6 +178,7 @@ func TestClientHappyPath(t *testing.T) {
 		w.WriteHeader(http.StatusOK)
 		w.Write([]byte(testbody))
 	}))
+	defer server.Close()
 
 	sut := jsonrpc.NewClient(
 		mustParse(server.URL),
@@ -153,6 +240,8 @@ func TestClientHappyPath(t *testing.T) {
 }
 
 func TestCanUseDefaults(t *testing.T) {
+	t.Parallel()
+
 	var (
 		testbody    = `{"jsonrpc":"2.0", "result":"boogaloo"}`
 		requestBody []byte
@@ -168,6 +257,7 @@ func TestCanUseDefaults(t *testing.T) {
 		w.WriteHeader(http.StatusOK)
 		w.Write([]byte(testbody))
 	}))
+	defer server.Close()
 
 	sut := jsonrpc.NewClient(
 		mustParse(server.URL),
@@ -210,6 +300,8 @@ func TestCanUseDefaults(t *testing.T) {
 }
 
 func TestClientCanHandleJSONRPCError(t *testing.T) {
+	t.Parallel()
+
 	var testbody = `{
 		"jsonrpc": "2.0",
 		"error": {
@@ -221,6 +313,7 @@ func TestClientCanHandleJSONRPCError(t *testing.T) {
 		w.WriteHeader(http.StatusOK)
 		w.Write([]byte(testbody))
 	}))
+	defer server.Close()
 
 	sut := jsonrpc.NewClient(mustParse(server.URL), "add")
 
@@ -255,6 +348,8 @@ func TestClientCanHandleJSONRPCError(t *testing.T) {
 }
 
 func TestDefaultAutoIncrementer(t *testing.T) {
+	t.Parallel()
+
 	sut := jsonrpc.NewAutoIncrementID(0)
 	var want uint64
 	for ; want < 100; want++ {