etcd-tester: initial stresser

tools/functional-tester/etcd-tester/stresser.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"sync"
+	"time"
+
+	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
+	"github.com/coreos/etcd/client"
+)
+
+type Stresser interface {
+	// Stress starts to stress the etcd cluster
+	Stress() error
+	// Cancel cancels the stress test on the etcd cluster
+	Cancel()
+	// Report reports the success and failure of the stress test
+	Report() (success int, failure int)
+}
+
+type stresser struct {
+	Endpoint    string
+	SuffexRange int
+
+	N        int
+	Interval time.Duration
+
+	mu      sync.Mutex
+	failure int
+	success int
+
+	cancel func()
+}
+
+func (s *stresser) Stress() error {
+	cfg := client.Config{Endpoints: []string{s.Endpoint}}
+	c, err := client.New(cfg)
+	if err != nil {
+		return err
+	}
+
+	kv := client.NewKeysAPI(c)
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancel = cancel
+
+	for i := 0; i < s.N; i++ {
+		go func() {
+			for {
+				_, err := kv.Set(ctx, "foo", "bar", nil)
+				if err == context.Canceled {
+					return
+				}
+				s.mu.Lock()
+				if err != nil {
+					s.failure++
+				}
+				s.success++
+				s.mu.Unlock()
+			}
+		}()
+	}
+
+	<-ctx.Done()
+	return nil
+}
+
+func (s *stresser) Cancel() {
+	s.cancel()
+}
+
+func (s *stresser) Report() (success int, failure int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.success, s.failure
+}