functional,tests: git mv functional tests/

Modularization: Mechanical move of "functional" directory into tests module using:

% git mv functional tests/

tests/functional/runner/election_command.go

@@ -0,0 +1,144 @@
+// Copyright 2016 The etcd 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 runner
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"go.etcd.io/etcd/v3/clientv3/concurrency"
+
+	"github.com/spf13/cobra"
+)
+
+// NewElectionCommand returns the cobra command for "election runner".
+func NewElectionCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "election [election name (defaults to 'elector')]",
+		Short: "Performs election operation",
+		Run:   runElectionFunc,
+	}
+	cmd.Flags().IntVar(&totalClientConnections, "total-client-connections", 10, "total number of client connections")
+	return cmd
+}
+
+func runElectionFunc(cmd *cobra.Command, args []string) {
+	election := "elector"
+	if len(args) == 1 {
+		election = args[0]
+	}
+	if len(args) > 1 {
+		ExitWithError(ExitBadArgs, errors.New("election takes at most one argument"))
+	}
+
+	rcs := make([]roundClient, totalClientConnections)
+	validatec := make(chan struct{}, len(rcs))
+	// nextc closes when election is ready for next round.
+	nextc := make(chan struct{})
+	eps := endpointsFromFlag(cmd)
+
+	for i := range rcs {
+		v := fmt.Sprintf("%d", i)
+		observedLeader := ""
+		validateWaiters := 0
+		var rcNextc chan struct{}
+		setRcNextc := func() {
+			rcNextc = nextc
+		}
+
+		rcs[i].c = newClient(eps, dialTimeout)
+		var (
+			s   *concurrency.Session
+			err error
+		)
+		for {
+			s, err = concurrency.NewSession(rcs[i].c)
+			if err == nil {
+				break
+			}
+		}
+
+		e := concurrency.NewElection(s, election)
+		rcs[i].acquire = func() (err error) {
+			ctx, cancel := context.WithCancel(context.Background())
+			donec := make(chan struct{})
+			go func() {
+				defer close(donec)
+				for ctx.Err() == nil {
+					if ol, ok := <-e.Observe(ctx); ok {
+						observedLeader = string(ol.Kvs[0].Value)
+						break
+					}
+				}
+				if observedLeader != v {
+					cancel()
+				}
+			}()
+			err = e.Campaign(ctx, v)
+			cancel()
+			<-donec
+			if err == nil {
+				observedLeader = v
+			}
+			if observedLeader == v {
+				validateWaiters = len(rcs)
+			}
+			select {
+			case <-ctx.Done():
+				return nil
+			default:
+				return err
+			}
+		}
+		rcs[i].validate = func() error {
+			l, err := e.Leader(context.TODO())
+			if err == nil && string(l.Kvs[0].Value) != observedLeader {
+				return fmt.Errorf("expected leader %q, got %q", observedLeader, l.Kvs[0].Value)
+			}
+			if err != nil {
+				return err
+			}
+			setRcNextc()
+			validatec <- struct{}{}
+			return nil
+		}
+		rcs[i].release = func() error {
+			for validateWaiters > 0 {
+				select {
+				case <-validatec:
+					validateWaiters--
+				default:
+					return fmt.Errorf("waiting on followers")
+				}
+			}
+			if err := e.Resign(context.TODO()); err != nil {
+				return err
+			}
+			if observedLeader == v {
+				oldNextc := nextc
+				nextc = make(chan struct{})
+				close(oldNextc)
+
+			}
+			<-rcNextc
+			observedLeader = ""
+			return nil
+		}
+	}
+	// each client creates 1 key from Campaign() and delete it from Resign()
+	// a round involves in 2*len(rcs) requests.
+	doRounds(rcs, rounds, 2*len(rcs))
+}

tests/functional/runner/error.go

@@ -0,0 +1,42 @@
+// Copyright 2015 The etcd 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 runner
+
+import (
+	"fmt"
+	"os"
+
+	"go.etcd.io/etcd/v3/client"
+)
+
+const (
+	// http://tldp.org/LDP/abs/html/exitcodes.html
+	ExitSuccess = iota
+	ExitError
+	ExitBadConnection
+	ExitInvalidInput // for txn, watch command
+	ExitBadFeature   // provided a valid flag with an unsupported value
+	ExitInterrupted
+	ExitIO
+	ExitBadArgs = 128
+)
+
+func ExitWithError(code int, err error) {
+	fmt.Fprintln(os.Stderr, "Error: ", err)
+	if cerr, ok := err.(*client.ClusterError); ok {
+		fmt.Fprintln(os.Stderr, cerr.Detail())
+	}
+	os.Exit(code)
+}

tests/functional/runner/global.go

@@ -0,0 +1,114 @@
+// Copyright 2016 The etcd 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 runner
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"go.etcd.io/etcd/v3/clientv3"
+
+	"github.com/spf13/cobra"
+	"golang.org/x/time/rate"
+)
+
+// shared flags
+var (
+	totalClientConnections int // total number of client connections to be made with server
+	endpoints              []string
+	dialTimeout            time.Duration
+	rounds                 int // total number of rounds to run; set to <= 0 to run forever.
+	reqRate                int // maximum number of requests per second.
+)
+
+type roundClient struct {
+	c        *clientv3.Client
+	progress int
+	acquire  func() error
+	validate func() error
+	release  func() error
+}
+
+func newClient(eps []string, timeout time.Duration) *clientv3.Client {
+	c, err := clientv3.New(clientv3.Config{
+		Endpoints:   eps,
+		DialTimeout: timeout * time.Second,
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+	return c
+}
+
+func doRounds(rcs []roundClient, rounds int, requests int) {
+	var wg sync.WaitGroup
+
+	wg.Add(len(rcs))
+	finished := make(chan struct{})
+	limiter := rate.NewLimiter(rate.Limit(reqRate), reqRate)
+	for i := range rcs {
+		go func(rc *roundClient) {
+			defer wg.Done()
+			for rc.progress < rounds || rounds <= 0 {
+				if err := limiter.WaitN(context.Background(), requests/len(rcs)); err != nil {
+					log.Panicf("rate limiter error %v", err)
+				}
+
+				for rc.acquire() != nil { /* spin */
+				}
+
+				if err := rc.validate(); err != nil {
+					log.Fatal(err)
+				}
+
+				time.Sleep(10 * time.Millisecond)
+				rc.progress++
+				finished <- struct{}{}
+
+				for rc.release() != nil { /* spin */
+				}
+			}
+		}(&rcs[i])
+	}
+
+	start := time.Now()
+	for i := 1; i < len(rcs)*rounds+1 || rounds <= 0; i++ {
+		select {
+		case <-finished:
+			if i%100 == 0 {
+				fmt.Printf("finished %d, took %v\n", i, time.Since(start))
+				start = time.Now()
+			}
+		case <-time.After(time.Minute):
+			log.Panic("no progress after 1 minute!")
+		}
+	}
+	wg.Wait()
+
+	for _, rc := range rcs {
+		rc.c.Close()
+	}
+}
+
+func endpointsFromFlag(cmd *cobra.Command) []string {
+	eps, err := cmd.Flags().GetStringSlice("endpoints")
+	if err != nil {
+		ExitWithError(ExitError, err)
+	}
+	return eps
+}

tests/functional/runner/help.go

@@ -0,0 +1,175 @@
+// Copyright 2015 The etcd 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.
+
+// copied from https://github.com/rkt/rkt/blob/master/rkt/help.go
+
+package runner
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"text/tabwriter"
+	"text/template"
+
+	"go.etcd.io/etcd/api/v3/version"
+
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+)
+
+var (
+	commandUsageTemplate *template.Template
+	templFuncs           = template.FuncMap{
+		"descToLines": func(s string) []string {
+			// trim leading/trailing whitespace and split into slice of lines
+			return strings.Split(strings.Trim(s, "\n\t "), "\n")
+		},
+		"cmdName": func(cmd *cobra.Command, startCmd *cobra.Command) string {
+			parts := []string{cmd.Name()}
+			for cmd.HasParent() && cmd.Parent().Name() != startCmd.Name() {
+				cmd = cmd.Parent()
+				parts = append([]string{cmd.Name()}, parts...)
+			}
+			return strings.Join(parts, " ")
+		},
+	}
+)
+
+func init() {
+	commandUsage := `
+{{ $cmd := .Cmd }}\
+{{ $cmdname := cmdName .Cmd .Cmd.Root }}\
+NAME:
+{{ if not .Cmd.HasParent }}\
+{{printf "\t%s - %s" .Cmd.Name .Cmd.Short}}
+{{else}}\
+{{printf "\t%s - %s" $cmdname .Cmd.Short}}
+{{end}}\
+
+USAGE:
+{{printf "\t%s" .Cmd.UseLine}}
+{{ if not .Cmd.HasParent }}\
+
+VERSION:
+{{printf "\t%s" .Version}}
+{{end}}\
+{{if .Cmd.HasSubCommands}}\
+
+API VERSION:
+{{printf "\t%s" .APIVersion}}
+{{end}}\
+{{if .Cmd.HasSubCommands}}\
+
+
+COMMANDS:
+{{range .SubCommands}}\
+{{ $cmdname := cmdName . $cmd }}\
+{{ if .Runnable }}\
+{{printf "\t%s\t%s" $cmdname .Short}}
+{{end}}\
+{{end}}\
+{{end}}\
+{{ if .Cmd.Long }}\
+
+DESCRIPTION:
+{{range $line := descToLines .Cmd.Long}}{{printf "\t%s" $line}}
+{{end}}\
+{{end}}\
+{{if .Cmd.HasLocalFlags}}\
+
+OPTIONS:
+{{.LocalFlags}}\
+{{end}}\
+{{if .Cmd.HasInheritedFlags}}\
+
+GLOBAL OPTIONS:
+{{.GlobalFlags}}\
+{{end}}
+`[1:]
+
+	commandUsageTemplate = template.Must(template.New("command_usage").Funcs(templFuncs).Parse(strings.Replace(commandUsage, "\\\n", "", -1)))
+}
+
+func etcdFlagUsages(flagSet *pflag.FlagSet) string {
+	x := new(bytes.Buffer)
+
+	flagSet.VisitAll(func(flag *pflag.Flag) {
+		if len(flag.Deprecated) > 0 {
+			return
+		}
+		var format string
+		if len(flag.Shorthand) > 0 {
+			format = "  -%s, --%s"
+		} else {
+			format = "   %s   --%s"
+		}
+		if len(flag.NoOptDefVal) > 0 {
+			format = format + "["
+		}
+		if flag.Value.Type() == "string" {
+			// put quotes on the value
+			format = format + "=%q"
+		} else {
+			format = format + "=%s"
+		}
+		if len(flag.NoOptDefVal) > 0 {
+			format = format + "]"
+		}
+		format = format + "\t%s\n"
+		shorthand := flag.Shorthand
+		fmt.Fprintf(x, format, shorthand, flag.Name, flag.DefValue, flag.Usage)
+	})
+
+	return x.String()
+}
+
+func getSubCommands(cmd *cobra.Command) []*cobra.Command {
+	var subCommands []*cobra.Command
+	for _, subCmd := range cmd.Commands() {
+		subCommands = append(subCommands, subCmd)
+		subCommands = append(subCommands, getSubCommands(subCmd)...)
+	}
+	return subCommands
+}
+
+func usageFunc(cmd *cobra.Command) error {
+	subCommands := getSubCommands(cmd)
+	tabOut := getTabOutWithWriter(os.Stdout)
+	commandUsageTemplate.Execute(tabOut, struct {
+		Cmd         *cobra.Command
+		LocalFlags  string
+		GlobalFlags string
+		SubCommands []*cobra.Command
+		Version     string
+		APIVersion  string
+	}{
+		cmd,
+		etcdFlagUsages(cmd.LocalFlags()),
+		etcdFlagUsages(cmd.InheritedFlags()),
+		subCommands,
+		version.Version,
+		version.APIVersion,
+	})
+	tabOut.Flush()
+	return nil
+}
+
+func getTabOutWithWriter(writer io.Writer) *tabwriter.Writer {
+	aTabOut := new(tabwriter.Writer)
+	aTabOut.Init(writer, 0, 8, 1, '\t', 0)
+	return aTabOut
+}

tests/functional/runner/lease_renewer_command.go

@@ -0,0 +1,91 @@
+// Copyright 2016 The etcd 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 runner
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	"go.etcd.io/etcd/v3/clientv3"
+
+	"github.com/spf13/cobra"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var (
+	leaseTTL int64
+)
+
+// NewLeaseRenewerCommand returns the cobra command for "lease-renewer runner".
+func NewLeaseRenewerCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "lease-renewer",
+		Short: "Performs lease renew operation",
+		Run:   runLeaseRenewerFunc,
+	}
+	cmd.Flags().Int64Var(&leaseTTL, "ttl", 5, "lease's ttl")
+	return cmd
+}
+
+func runLeaseRenewerFunc(cmd *cobra.Command, args []string) {
+	if len(args) > 0 {
+		ExitWithError(ExitBadArgs, errors.New("lease-renewer does not take any argument"))
+	}
+
+	eps := endpointsFromFlag(cmd)
+	c := newClient(eps, dialTimeout)
+	ctx := context.Background()
+
+	for {
+		var (
+			l   *clientv3.LeaseGrantResponse
+			lk  *clientv3.LeaseKeepAliveResponse
+			err error
+		)
+		for {
+			l, err = c.Lease.Grant(ctx, leaseTTL)
+			if err == nil {
+				break
+			}
+		}
+		expire := time.Now().Add(time.Duration(l.TTL-1) * time.Second)
+
+		for {
+			lk, err = c.Lease.KeepAliveOnce(ctx, l.ID)
+			if ev, ok := status.FromError(err); ok && ev.Code() == codes.NotFound {
+				if time.Since(expire) < 0 {
+					log.Fatalf("bad renew! exceeded: %v", time.Since(expire))
+					for {
+						lk, err = c.Lease.KeepAliveOnce(ctx, l.ID)
+						fmt.Println(lk, err)
+						time.Sleep(time.Second)
+					}
+				}
+				log.Fatalf("lost lease %d, expire: %v\n", l.ID, expire)
+				break
+			}
+			if err != nil {
+				continue
+			}
+			expire = time.Now().Add(time.Duration(lk.TTL-1) * time.Second)
+			log.Printf("renewed lease %d, expire: %v\n", lk.ID, expire)
+			time.Sleep(time.Duration(lk.TTL-2) * time.Second)
+		}
+	}
+}

tests/functional/runner/lock_racer_command.go

@@ -0,0 +1,94 @@
+// Copyright 2016 The etcd 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 runner
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"go.etcd.io/etcd/v3/clientv3/concurrency"
+
+	"github.com/spf13/cobra"
+)
+
+// NewLockRacerCommand returns the cobra command for "lock-racer runner".
+func NewLockRacerCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "lock-racer [name of lock (defaults to 'racers')]",
+		Short: "Performs lock race operation",
+		Run:   runRacerFunc,
+	}
+	cmd.Flags().IntVar(&totalClientConnections, "total-client-connections", 10, "total number of client connections")
+	return cmd
+}
+
+func runRacerFunc(cmd *cobra.Command, args []string) {
+	racers := "racers"
+	if len(args) == 1 {
+		racers = args[0]
+	}
+
+	if len(args) > 1 {
+		ExitWithError(ExitBadArgs, errors.New("lock-racer takes at most one argument"))
+	}
+
+	rcs := make([]roundClient, totalClientConnections)
+	ctx := context.Background()
+	// mu ensures validate and release funcs are atomic.
+	var mu sync.Mutex
+	cnt := 0
+
+	eps := endpointsFromFlag(cmd)
+
+	for i := range rcs {
+		var (
+			s   *concurrency.Session
+			err error
+		)
+
+		rcs[i].c = newClient(eps, dialTimeout)
+
+		for {
+			s, err = concurrency.NewSession(rcs[i].c)
+			if err == nil {
+				break
+			}
+		}
+		m := concurrency.NewMutex(s, racers)
+		rcs[i].acquire = func() error { return m.Lock(ctx) }
+		rcs[i].validate = func() error {
+			mu.Lock()
+			defer mu.Unlock()
+			if cnt++; cnt != 1 {
+				return fmt.Errorf("bad lock; count: %d", cnt)
+			}
+			return nil
+		}
+		rcs[i].release = func() error {
+			mu.Lock()
+			defer mu.Unlock()
+			if err := m.Unlock(ctx); err != nil {
+				return err
+			}
+			cnt = 0
+			return nil
+		}
+	}
+	// each client creates 1 key from NewMutex() and delete it from Unlock()
+	// a round involves in 2*len(rcs) requests.
+	doRounds(rcs, rounds, 2*len(rcs))
+}

tests/functional/runner/root.go

@@ -0,0 +1,70 @@
+// Copyright 2017 The etcd 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 runner implements individual etcd-runner commands for the etcd-runner utility.
+package runner
+
+import (
+	"log"
+	"math/rand"
+	"time"
+
+	"github.com/spf13/cobra"
+)
+
+const (
+	cliName        = "etcd-runner"
+	cliDescription = "Stress tests using clientv3 functionality.."
+
+	defaultDialTimeout = 2 * time.Second
+)
+
+var (
+	rootCmd = &cobra.Command{
+		Use:        cliName,
+		Short:      cliDescription,
+		SuggestFor: []string{"etcd-runner"},
+	}
+)
+
+func init() {
+	cobra.EnablePrefixMatching = true
+
+	rand.Seed(time.Now().UnixNano())
+
+	log.SetFlags(log.Lmicroseconds)
+
+	rootCmd.PersistentFlags().StringSliceVar(&endpoints, "endpoints", []string{"127.0.0.1:2379"}, "gRPC endpoints")
+	rootCmd.PersistentFlags().DurationVar(&dialTimeout, "dial-timeout", defaultDialTimeout, "dial timeout for client connections")
+	rootCmd.PersistentFlags().IntVar(&reqRate, "req-rate", 30, "maximum number of requests per second")
+	rootCmd.PersistentFlags().IntVar(&rounds, "rounds", 100, "number of rounds to run; 0 to run forever")
+
+	rootCmd.AddCommand(
+		NewElectionCommand(),
+		NewLeaseRenewerCommand(),
+		NewLockRacerCommand(),
+		NewWatchCommand(),
+	)
+}
+
+func Start() {
+	rootCmd.SetUsageFunc(usageFunc)
+
+	// Make help just show the usage
+	rootCmd.SetHelpTemplate(`{{.UsageString}}`)
+
+	if err := rootCmd.Execute(); err != nil {
+		ExitWithError(ExitError, err)
+	}
+}

tests/functional/runner/watch_command.go

@@ -0,0 +1,210 @@
+// Copyright 2016 The etcd 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 runner
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"go.etcd.io/etcd/v3/clientv3"
+	"go.etcd.io/etcd/v3/pkg/stringutil"
+
+	"github.com/spf13/cobra"
+	"golang.org/x/time/rate"
+)
+
+var (
+	runningTime    time.Duration // time for which operation should be performed
+	noOfPrefixes   int           // total number of prefixes which will be watched upon
+	watchPerPrefix int           // number of watchers per prefix
+	watchPrefix    string        // prefix append to keys in watcher
+	totalKeys      int           // total number of keys for operation
+)
+
+// NewWatchCommand returns the cobra command for "watcher runner".
+func NewWatchCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "watcher",
+		Short: "Performs watch operation",
+		Run:   runWatcherFunc,
+	}
+	cmd.Flags().DurationVar(&runningTime, "running-time", 60, "number of seconds to run")
+	cmd.Flags().StringVar(&watchPrefix, "prefix", "", "the prefix to append on all keys")
+	cmd.Flags().IntVar(&noOfPrefixes, "total-prefixes", 10, "total no of prefixes to use")
+	cmd.Flags().IntVar(&watchPerPrefix, "watch-per-prefix", 10, "number of watchers per prefix")
+	cmd.Flags().IntVar(&totalKeys, "total-keys", 1000, "total number of keys to watch")
+
+	return cmd
+}
+
+func runWatcherFunc(cmd *cobra.Command, args []string) {
+	if len(args) > 0 {
+		ExitWithError(ExitBadArgs, errors.New("watcher does not take any argument"))
+	}
+
+	ctx := context.Background()
+	for round := 0; round < rounds || rounds <= 0; round++ {
+		fmt.Println("round", round)
+		performWatchOnPrefixes(ctx, cmd, round)
+	}
+}
+
+func performWatchOnPrefixes(ctx context.Context, cmd *cobra.Command, round int) {
+	keyPerPrefix := totalKeys / noOfPrefixes
+	prefixes := stringutil.UniqueStrings(5, noOfPrefixes)
+	keys := stringutil.RandomStrings(10, keyPerPrefix)
+
+	roundPrefix := fmt.Sprintf("%16x", round)
+
+	eps := endpointsFromFlag(cmd)
+
+	var (
+		revision int64
+		wg       sync.WaitGroup
+		gr       *clientv3.GetResponse
+		err      error
+	)
+
+	client := newClient(eps, dialTimeout)
+	defer client.Close()
+
+	gr, err = getKey(ctx, client, "non-existent")
+	if err != nil {
+		log.Fatalf("failed to get the initial revision: %v", err)
+	}
+	revision = gr.Header.Revision
+
+	ctxt, cancel := context.WithDeadline(ctx, time.Now().Add(runningTime*time.Second))
+	defer cancel()
+
+	// generate and put keys in cluster
+	limiter := rate.NewLimiter(rate.Limit(reqRate), reqRate)
+
+	go func() {
+		for _, key := range keys {
+			for _, prefix := range prefixes {
+				if err = limiter.Wait(ctxt); err != nil {
+					return
+				}
+				if err = putKeyAtMostOnce(ctxt, client, watchPrefix+"-"+roundPrefix+"-"+prefix+"-"+key); err != nil {
+					log.Fatalf("failed to put key: %v", err)
+					return
+				}
+			}
+		}
+	}()
+
+	ctxc, cancelc := context.WithCancel(ctx)
+
+	wcs := make([]clientv3.WatchChan, 0)
+	rcs := make([]*clientv3.Client, 0)
+
+	for _, prefix := range prefixes {
+		for j := 0; j < watchPerPrefix; j++ {
+			rc := newClient(eps, dialTimeout)
+			rcs = append(rcs, rc)
+
+			wprefix := watchPrefix + "-" + roundPrefix + "-" + prefix
+
+			wc := rc.Watch(ctxc, wprefix, clientv3.WithPrefix(), clientv3.WithRev(revision))
+			wcs = append(wcs, wc)
+
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				checkWatchResponse(wc, wprefix, keys)
+			}()
+		}
+	}
+	wg.Wait()
+
+	cancelc()
+
+	// verify all watch channels are closed
+	for e, wc := range wcs {
+		if _, ok := <-wc; ok {
+			log.Fatalf("expected wc to be closed, but received %v", e)
+		}
+	}
+
+	for _, rc := range rcs {
+		rc.Close()
+	}
+
+	if err = deletePrefix(ctx, client, watchPrefix); err != nil {
+		log.Fatalf("failed to clean up keys after test: %v", err)
+	}
+}
+
+func checkWatchResponse(wc clientv3.WatchChan, prefix string, keys []string) {
+	for n := 0; n < len(keys); {
+		wr, more := <-wc
+		if !more {
+			log.Fatalf("expect more keys (received %d/%d) for %s", n, len(keys), prefix)
+		}
+		for _, event := range wr.Events {
+			expectedKey := prefix + "-" + keys[n]
+			receivedKey := string(event.Kv.Key)
+			if expectedKey != receivedKey {
+				log.Fatalf("expected key %q, got %q for prefix : %q\n", expectedKey, receivedKey, prefix)
+			}
+			n++
+		}
+	}
+}
+
+func putKeyAtMostOnce(ctx context.Context, client *clientv3.Client, key string) error {
+	gr, err := getKey(ctx, client, key)
+	if err != nil {
+		return err
+	}
+
+	var modrev int64
+	if len(gr.Kvs) > 0 {
+		modrev = gr.Kvs[0].ModRevision
+	}
+
+	for ctx.Err() == nil {
+		_, err := client.Txn(ctx).If(clientv3.Compare(clientv3.ModRevision(key), "=", modrev)).Then(clientv3.OpPut(key, key)).Commit()
+
+		if err == nil {
+			return nil
+		}
+	}
+
+	return ctx.Err()
+}
+
+func deletePrefix(ctx context.Context, client *clientv3.Client, key string) error {
+	for ctx.Err() == nil {
+		if _, err := client.Delete(ctx, key, clientv3.WithPrefix()); err == nil {
+			return nil
+		}
+	}
+	return ctx.Err()
+}
+
+func getKey(ctx context.Context, client *clientv3.Client, key string) (*clientv3.GetResponse, error) {
+	for ctx.Err() == nil {
+		if gr, err := client.Get(ctx, key); err == nil {
+			return gr, nil
+		}
+	}
+	return nil, ctx.Err()
+}