etcd-runner: make command compliant

tools/functional-tester/etcd-runner/command/election_command.go

@@ -0,0 +1,126 @@
+// 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 command
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/coreos/etcd/clientv3/concurrency"
+	"github.com/spf13/cobra"
+)
+
+// NewElectionCommand returns the cobra command for "election runner".
+func NewElectionCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "election",
+		Short: "Performs election operation",
+		Run:   runElectionFunc,
+	}
+	cmd.Flags().IntVar(&rounds, "rounds", 100, "number of rounds to run")
+	cmd.Flags().IntVar(&totalClientConnections, "total-client-connections", 10, "total number of client connections")
+	return cmd
+}
+
+func runElectionFunc(cmd *cobra.Command, args []string) {
+	if len(args) > 0 {
+		ExitWithError(ExitBadArgs, errors.New("election does not take any argument"))
+	}
+
+	rcs := make([]roundClient, totalClientConnections)
+	validatec, releasec := make(chan struct{}, len(rcs)), make(chan struct{}, len(rcs))
+	for range rcs {
+		releasec <- struct{}{}
+	}
+
+	eps := endpointsFromFlag(cmd)
+	dialTimeout := dialTimeoutFromCmd(cmd)
+
+	for i := range rcs {
+		v := fmt.Sprintf("%d", i)
+		observedLeader := ""
+		validateWaiters := 0
+
+		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, "electors")
+
+		rcs[i].acquire = func() error {
+			<-releasec
+			ctx, cancel := context.WithCancel(context.Background())
+			go func() {
+				if ol, ok := <-e.Observe(ctx); ok {
+					observedLeader = string(ol.Kvs[0].Value)
+					if observedLeader != v {
+						cancel()
+					}
+				}
+			}()
+			err = e.Campaign(ctx, v)
+			if err == nil {
+				observedLeader = v
+			}
+			if observedLeader == v {
+				validateWaiters = len(rcs)
+			}
+			select {
+			case <-ctx.Done():
+				return nil
+			default:
+				cancel()
+				return err
+			}
+		}
+		rcs[i].validate = func() error {
+			if l, err := e.Leader(context.TODO()); err == nil && l != observedLeader {
+				return fmt.Errorf("expected leader %q, got %q", observedLeader, l)
+			}
+			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 {
+				for range rcs {
+					releasec <- struct{}{}
+				}
+			}
+			observedLeader = ""
+			return nil
+		}
+	}
+
+	doRounds(rcs, rounds)
+}

tools/functional-tester/etcd-runner/command/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 command
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/coreos/etcd/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)
+}

tools/functional-tester/etcd-runner/command/global.go

@@ -0,0 +1,125 @@
+// 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 command
+
+import (
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/coreos/etcd/clientv3"
+
+	"github.com/spf13/cobra"
+)
+
+var (
+	rounds                 int
+	totalClientConnections int
+)
+
+// GlobalFlags are flags that defined globally
+// and are inherited to all sub-commands.
+type GlobalFlags struct {
+	Endpoints   []string
+	DialTimeout time.Duration
+}
+
+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: time.Duration(timeout) * time.Second,
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+	return c
+}
+
+func doRounds(rcs []roundClient, rounds int) {
+	var mu sync.Mutex
+	var wg sync.WaitGroup
+
+	wg.Add(len(rcs))
+	finished := make(chan struct{}, 0)
+	for i := range rcs {
+		go func(rc *roundClient) {
+			defer wg.Done()
+			for rc.progress < rounds {
+				for rc.acquire() != nil { /* spin */
+				}
+
+				mu.Lock()
+				if err := rc.validate(); err != nil {
+					log.Fatal(err)
+				}
+				mu.Unlock()
+
+				time.Sleep(10 * time.Millisecond)
+				rc.progress++
+				finished <- struct{}{}
+
+				mu.Lock()
+				for rc.release() != nil {
+					mu.Unlock()
+					mu.Lock()
+				}
+				mu.Unlock()
+			}
+		}(&rcs[i])
+	}
+
+	start := time.Now()
+	for i := 1; i < len(rcs)*rounds+1; 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 {
+	endpoints, err := cmd.Flags().GetStringSlice("endpoints")
+	if err != nil {
+		ExitWithError(ExitError, err)
+	}
+	return endpoints
+}
+
+func dialTimeoutFromCmd(cmd *cobra.Command) time.Duration {
+	dialTimeout, err := cmd.Flags().GetDuration("dial-timeout")
+	if err != nil {
+		ExitWithError(ExitError, err)
+	}
+	return dialTimeout
+}

tools/functional-tester/etcd-runner/command/lease_renewer_command.go

@@ -0,0 +1,86 @@
+// 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 command
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/coreos/etcd/clientv3"
+	"github.com/spf13/cobra"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+)
+
+// 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,
+	}
+	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)
+	dialTimeout := dialTimeoutFromCmd(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, 5)
+			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 grpc.Code(err) == codes.NotFound {
+				if time.Since(expire) < 0 {
+					log.Printf("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.Printf("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)
+		}
+	}
+}

tools/functional-tester/etcd-runner/command/lock_racer_command.go

@@ -0,0 +1,81 @@
+// 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 command
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/coreos/etcd/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",
+		Short: "Performs lock race operation",
+		Run:   runRacerFunc,
+	}
+	cmd.Flags().IntVar(&rounds, "rounds", 100, "number of rounds to run")
+	cmd.Flags().IntVar(&totalClientConnections, "total-client-connections", 10, "total number of client connections")
+	return cmd
+}
+
+func runRacerFunc(cmd *cobra.Command, args []string) {
+	if len(args) > 0 {
+		ExitWithError(ExitBadArgs, errors.New("lock-racer does not take any argument"))
+	}
+
+	rcs := make([]roundClient, totalClientConnections)
+	ctx := context.Background()
+	cnt := 0
+
+	eps := endpointsFromFlag(cmd)
+	dialTimeout := dialTimeoutFromCmd(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 {
+			if cnt++; cnt != 1 {
+				return fmt.Errorf("bad lock; count: %d", cnt)
+			}
+			return nil
+		}
+		rcs[i].release = func() error {
+			if err := m.Unlock(ctx); err != nil {
+				return err
+			}
+			cnt = 0
+			return nil
+		}
+	}
+	doRounds(rcs, rounds)
+}

tools/functional-tester/etcd-runner/command/watch_command.go

@@ -0,0 +1,202 @@
+// 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 command
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/etcd/pkg/stringutil"
+	"github.com/spf13/cobra"
+	"golang.org/x/time/rate"
+)
+
+// 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().IntVar(&rounds, "rounds", 100, "number of rounds to run")
+	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; round++ {
+		fmt.Println("round", round)
+		performWatchOnPrefixes(ctx, cmd, round)
+	}
+}
+
+func performWatchOnPrefixes(ctx context.Context, cmd *cobra.Command, round int) {
+	runningTime := 60 * time.Second // time for which operation should be performed
+	noOfPrefixes := 36              // total number of prefixes which will be watched upon
+	watchPerPrefix := 10            // number of watchers per prefix
+	reqRate := 30                   // put request per second
+	keyPrePrefix := 30              // max number of keyPrePrefixs for put operation
+
+	prefixes := stringutil.UniqueStrings(5, noOfPrefixes)
+	keys := stringutil.RandomStrings(10, keyPrePrefix)
+
+	roundPrefix := fmt.Sprintf("%16x", round)
+
+	eps := endpointsFromFlag(cmd)
+	dialTimeout := dialTimeoutFromCmd(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))
+	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, 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)
+
+			watchPrefix := roundPrefix + "-" + prefix
+
+			wc := rc.Watch(ctxc, watchPrefix, clientv3.WithPrefix(), clientv3.WithRev(revision))
+			wcs = append(wcs, wc)
+
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				checkWatchResponse(wc, watchPrefix, 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, roundPrefix); 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", len(keys), n, 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()
+}