[query] Query handlers refactoring

.golangci.yml

@@ -108,6 +108,8 @@ linters-settings:
     locale: US
     ignore-words:
       - someword
+  exhaustive:
+    default-signifies-exhaustive: true
   lll:
     # max line length, lines longer will be reported. Default is 120.
     # '\t' is counted as 1 character by default, and can be changed with the tab-width option
@@ -172,7 +174,6 @@ linters:
     - goconst
     - gocritic
     - gocyclo
-    - godox
     - goimports
     - golint
     - gosimple
@@ -211,10 +212,17 @@ linters:
     - exhaustivestruct
     # We allow cuddling assignment following conditions because there are valid 
     # logical groupings for this use-case (e.g. when evaluating config values).
-    - wsl
+    - wsl 
+    # Wrapcheck can cause errors until all callsites checking explicit error
+    # types like io.EOF are converted to use errors.Is instead. Re-enable this
+    # linter once all error checks are upgraded.
+    - wrapcheck
+    # godox prevents using TODOs or FIXMEs which can be useful for demarkation
+    # of future work.
+    - godox 
     # New line required before return would require a large fraction of the
     # code base to need updating, it's not worth the perceived benefit.
-    - nlreturn
+    - nlreturn 
   disable-all: false
   presets:
     # bodyclose, errcheck, gosec, govet, scopelint, staticcheck, typecheck

Makefile

@@ -165,7 +165,7 @@ tools-linux-amd64:
 	$(LINUX_AMD64_ENV) make tools
 
 .PHONY: all
-all: lint test-ci-unit test-ci-integration services tools
+all: test-ci-unit test-ci-integration services tools
 	@echo Made all successfully
 
 .PHONY: install-tools
@@ -256,7 +256,7 @@ SUBDIR_TARGETS := \
 	asset-gen       \
 	genny-gen       \
 	license-gen     \
-	all-gen			\
+	all-gen         \
 	lint
 
 .PHONY: test-ci-unit
@@ -384,6 +384,7 @@ endef
 
 # generate targets across SUBDIRS for each SUBDIR_TARGET. i.e. generate rules
 # which allow `make all-gen` to invoke `make all-gen-dbnode all-gen-coordinator ...`
+# NB: we skip lint explicity as it runs as a separate CI step.
 $(foreach SUBDIR_TARGET, $(SUBDIR_TARGETS), $(eval $(SUBDIR_TARGET_RULE)))
 
 # Builds the single kube bundle from individual manifest files.
@@ -401,7 +402,7 @@ go-mod-tidy:
 .PHONY: all-gen
 all-gen: \
 	install-tools \
-	$(foreach SUBDIR_TARGET, $(SUBDIR_TARGETS), $(SUBDIR_TARGET)) \
+	$(foreach SUBDIR_TARGET, $(filter-out lint all-gen,$(SUBDIR_TARGETS)), $(SUBDIR_TARGET)) \
 	kube-gen-all \
 	go-mod-tidy
 

go.mod

@@ -119,6 +119,7 @@ require (
 	go.etcd.io/etcd v0.5.0-alpha.5.0.20200824191128-ae9734ed278b
 	go.uber.org/atomic v1.6.0
 	go.uber.org/config v1.4.0
+	go.uber.org/goleak v1.1.10
 	go.uber.org/zap v1.13.0
 	golang.org/x/net v0.0.0-20200822124328-c89045814202
 	golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208

go.sum

@@ -922,6 +922,8 @@ go.uber.org/atomic v1.6.0 h1:Ezj3JGmsOnG1MoRWQkPBsKLe9DwWD9QeXzTRzzldNVk=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/config v1.4.0 h1:upnMPpMm6WlbZtXoasNkK4f0FhxwS+W4Iqz5oNznehQ=
 go.uber.org/config v1.4.0/go.mod h1:aCyrMHmUAc/s2h9sv1koP84M9ZF/4K+g2oleyESO/Ig=
+go.uber.org/goleak v1.1.10 h1:z+mqJhf6ss6BSfSM671tgKyZBFPTTJM+HLxnhPC3wu0=
+go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
 go.uber.org/multierr v1.4.0 h1:f3WCSC2KzAcBXGATIxAB1E2XuCpNU255wNKZ505qi3E=
@@ -1069,6 +1071,7 @@ golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191104232314-dc038396d1f0/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191203134012-c197fd4bf371/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=

src/cluster/kv/etcd/store_test.go

@@ -35,12 +35,13 @@ import (
 	"github.com/m3db/m3/src/cluster/kv"
 	"github.com/m3db/m3/src/cluster/mocks"
 	xclock "github.com/m3db/m3/src/x/clock"
+	"github.com/m3db/m3/src/x/retry"
 
-	"go.etcd.io/etcd/clientv3"
-	"go.etcd.io/etcd/integration"
 	"github.com/coreos/pkg/capnslog"
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/require"
+	"go.etcd.io/etcd/clientv3"
+	"go.etcd.io/etcd/integration"
 	"golang.org/x/net/context"
 )
 
@@ -89,6 +90,8 @@ func TestGetAndSet(t *testing.T) {
 }
 
 func TestNoCache(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 
 	store, err := NewStore(ec, ec, opts)
@@ -155,6 +158,8 @@ func TestCacheDirCreation(t *testing.T) {
 }
 
 func TestCache(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 
 	f, err := ioutil.TempFile("", "")
@@ -207,6 +212,8 @@ func TestCache(t *testing.T) {
 }
 
 func TestSetIfNotExist(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -226,6 +233,8 @@ func TestSetIfNotExist(t *testing.T) {
 }
 
 func TestCheckAndSet(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -252,6 +261,8 @@ func TestCheckAndSet(t *testing.T) {
 }
 
 func TestWatchClose(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -280,6 +291,7 @@ func TestWatchClose(t *testing.T) {
 		if !ok {
 			break
 		}
+		time.Sleep(1 * time.Millisecond)
 	}
 
 	// getting a new watch will create a new watchale and thread to watch for updates
@@ -300,6 +312,8 @@ func TestWatchClose(t *testing.T) {
 }
 
 func TestWatchLastVersion(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -311,7 +325,7 @@ func TestWatchLastVersion(t *testing.T) {
 	require.Nil(t, w.Get())
 
 	var errs int32
-	lastVersion := 100
+	lastVersion := 50
 	go func() {
 		for i := 1; i <= lastVersion; i++ {
 			_, err := store.Set("foo", genProto(fmt.Sprintf("bar%d", i)))
@@ -334,6 +348,8 @@ func TestWatchLastVersion(t *testing.T) {
 }
 
 func TestWatchFromExist(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -371,6 +387,8 @@ func TestWatchFromExist(t *testing.T) {
 }
 
 func TestWatchFromNotExist(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -414,6 +432,8 @@ func TestGetFromKvNotFound(t *testing.T) {
 }
 
 func TestMultipleWatchesFromExist(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -464,6 +484,8 @@ func TestMultipleWatchesFromExist(t *testing.T) {
 }
 
 func TestMultipleWatchesFromNotExist(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -506,6 +528,8 @@ func TestMultipleWatchesFromNotExist(t *testing.T) {
 }
 
 func TestWatchNonBlocking(t *testing.T) {
+	t.Parallel()
+
 	ec, opts, closeFn := testStore(t)
 	defer closeFn()
 
@@ -1093,7 +1117,7 @@ func TestWatchWithStartRevision(t *testing.T) {
 			store, err := NewStore(ec, ec, opts)
 			require.NoError(t, err)
 
-			for i := 1; i <= 100; i++ {
+			for i := 1; i <= 50; i++ {
 				_, err = store.Set("foo", genProto(fmt.Sprintf("bar-%d", i)))
 				require.NoError(t, err)
 			}
@@ -1110,7 +1134,7 @@ func TestWatchWithStartRevision(t *testing.T) {
 			w1, err := store.Watch("foo")
 			require.NoError(t, err)
 			<-w1.C()
-			verifyValue(t, w1.Get(), "bar-100", 100)
+			verifyValue(t, w1.Get(), "bar-50", 50)
 		})
 	}
 
@@ -1137,7 +1161,11 @@ func testStore(t *testing.T) (*clientv3.Client, Options, func()) {
 	}
 
 	opts := NewOptions().
-		SetWatchChanCheckInterval(10 * time.Millisecond).
+		SetWatchChanCheckInterval(50 * time.Millisecond).
+		SetWatchChanResetInterval(150 * time.Millisecond).
+		SetWatchChanInitTimeout(150 * time.Millisecond).
+		SetRequestTimeout(100 * time.Millisecond).
+		SetRetryOptions(retry.NewOptions().SetMaxRetries(1).SetMaxBackoff(0)).
 		SetPrefix("test")
 
 	return ec, opts, closer