Add support for named attacks

This commit introduces attack names and includes that information in
the Result struct as well as a sequence number for each result.

This allows for the plot report to work with multiple overlaid result
sets.

README.md

@@ -61,6 +61,8 @@ attack command:
       Local IP address (default 0.0.0.0)
   -lazy
       Read targets lazily
+  -name string
+      Attack name
   -output string
       Output file (default "stdout")
   -rate uint

attack.go

@@ -23,6 +23,7 @@ func attackCmd() command {
 		laddr:   localAddr{&vegeta.DefaultLocalAddr},
 	}
 
+	fs.StringVar(&opts.name, "name", "", "Attack name")
 	fs.StringVar(&opts.targetsf, "targets", "stdin", "Targets file")
 	fs.StringVar(&opts.outputf, "output", "stdout", "Output file")
 	fs.StringVar(&opts.bodyf, "body", "", "Requests body file")
@@ -56,6 +57,7 @@ var (
 
 // attackOpts aggregates the attack function command options
 type attackOpts struct {
+	name        string
 	targetsf    string
 	outputf     string
 	bodyf       string
@@ -138,7 +140,7 @@ func attack(opts *attackOpts) (err error) {
 		vegeta.H2C(opts.h2c),
 	)
 
-	res := atk.Attack(tr, opts.rate, opts.duration)
+	res := atk.Attack(tr, opts.rate, opts.duration, opts.name)
 	enc := vegeta.NewEncoder(out)
 	sig := make(chan os.Signal, 1)
 	signal.Notify(sig, os.Interrupt)

lib/attack.go

@@ -186,16 +186,16 @@ func H2C(enabled bool) func(*Attacker) {
 }
 
 // Attack reads its Targets from the passed Targeter and attacks them at
-// the rate specified for duration time. When the duration is zero the attack
-// runs until Stop is called. Results are put into the returned channel as soon
-// as they arrive.
-func (a *Attacker) Attack(tr Targeter, rate uint64, du time.Duration) <-chan *Result {
+// the rate specified for the given duration. When the duration is zero the attack
+// runs until Stop is called. Results are sent to the returned channel as soon
+// as they arrive and will have their Attack field set to the given name.
+func (a *Attacker) Attack(tr Targeter, rate uint64, du time.Duration, name string) <-chan *Result {
 	var workers sync.WaitGroup
 	results := make(chan *Result)
-	ticks := make(chan time.Time)
+	ticks := make(chan uint64)
 	for i := uint64(0); i < a.workers; i++ {
 		workers.Add(1)
-		go a.attack(tr, &workers, ticks, results)
+		go a.attack(tr, name, &workers, ticks, results)
 	}
 
 	go func() {
@@ -204,20 +204,20 @@ func (a *Attacker) Attack(tr Targeter, rate uint64, du time.Duration) <-chan *Re
 		defer close(ticks)
 		interval := 1e9 / rate
 		hits := rate * uint64(du.Seconds())
-		began, done := time.Now(), uint64(0)
+		began, seq := time.Now(), uint64(0)
 		for {
-			now, next := time.Now(), began.Add(time.Duration(done*interval))
+			now, next := time.Now(), began.Add(time.Duration(seq*interval))
 			time.Sleep(next.Sub(now))
 			select {
-			case ticks <- max(next, now):
-				if done++; done == hits {
+			case ticks <- seq:
+				if seq++; seq == hits {
 					return
 				}
 			case <-a.stopch:
 				return
 			default: // all workers are blocked. start one more and try again
 				workers.Add(1)
-				go a.attack(tr, &workers, ticks, results)
+				go a.attack(tr, name, &workers, ticks, results)
 			}
 		}
 	}()
@@ -235,16 +235,16 @@ func (a *Attacker) Stop() {
 	}
 }
 
-func (a *Attacker) attack(tr Targeter, workers *sync.WaitGroup, ticks <-chan time.Time, results chan<- *Result) {
+func (a *Attacker) attack(tr Targeter, name string, workers *sync.WaitGroup, ticks <-chan uint64, results chan<- *Result) {
 	defer workers.Done()
-	for tm := range ticks {
-		results <- a.hit(tr, tm)
+	for seq := range ticks {
+		results <- a.hit(tr, name, seq)
 	}
 }
 
-func (a *Attacker) hit(tr Targeter, tm time.Time) *Result {
+func (a *Attacker) hit(tr Targeter, name string, seq uint64) *Result {
 	var (
-		res Result
+		res = Result{Attack: name, Seq: seq}
 		tgt Target
 		err error
 	)
@@ -288,10 +288,3 @@ func (a *Attacker) hit(tr Targeter, tm time.Time) *Result {
 
 	return &res
 }
-
-func max(a, b time.Time) time.Time {
-	if a.After(b) {
-		return a
-	}
-	return b
-}

lib/attack_test.go

@@ -25,7 +25,7 @@ func TestAttackRate(t *testing.T) {
 	rate := uint64(100)
 	atk := NewAttacker()
 	var hits uint64
-	for range atk.Attack(tr, rate, 1*time.Second) {
+	for range atk.Attack(tr, rate, 1*time.Second, "") {
 		hits++
 	}
 	if got, want := hits, rate; got != want {
@@ -44,7 +44,7 @@ func TestAttackDuration(t *testing.T) {
 	time.AfterFunc(2*time.Second, func() { t.Fatal("Timed out") })
 
 	rate, hits := uint64(100), uint64(0)
-	for range atk.Attack(tr, rate, 0) {
+	for range atk.Attack(tr, rate, 0, "") {
 		if hits++; hits == 100 {
 			atk.Stop()
 			break
@@ -76,7 +76,7 @@ func TestRedirects(t *testing.T) {
 	redirects := 2
 	atk := NewAttacker(Redirects(redirects))
 	tr := NewStaticTargeter(Target{Method: "GET", URL: server.URL})
-	res := atk.hit(tr, time.Now())
+	res := atk.hit(tr, "", 0)
 	want := fmt.Sprintf("stopped after %d redirects", redirects)
 	if got := res.Error; !strings.HasSuffix(got, want) {
 		t.Fatalf("want: '%v' in '%v'", want, got)
@@ -93,7 +93,7 @@ func TestNoFollow(t *testing.T) {
 	defer server.Close()
 	atk := NewAttacker(Redirects(NoFollow))
 	tr := NewStaticTargeter(Target{Method: "GET", URL: server.URL})
-	res := atk.hit(tr, time.Now())
+	res := atk.hit(tr, "", 0)
 	if res.Error != "" {
 		t.Fatalf("got err: %v", res.Error)
 	}
@@ -112,7 +112,7 @@ func TestTimeout(t *testing.T) {
 	defer server.Close()
 	atk := NewAttacker(Timeout(10 * time.Millisecond))
 	tr := NewStaticTargeter(Target{Method: "GET", URL: server.URL})
-	res := atk.hit(tr, time.Now())
+	res := atk.hit(tr, "", 0)
 	want := "net/http: timeout awaiting response headers"
 	if got := res.Error; !strings.HasSuffix(got, want) {
 		t.Fatalf("want: '%v' in '%v'", want, got)
@@ -137,7 +137,7 @@ func TestLocalAddr(t *testing.T) {
 	defer server.Close()
 	atk := NewAttacker(LocalAddr(*addr))
 	tr := NewStaticTargeter(Target{Method: "GET", URL: server.URL})
-	atk.hit(tr, time.Now())
+	atk.hit(tr, "", 0)
 }
 
 func TestKeepAlive(t *testing.T) {
@@ -171,7 +171,7 @@ func TestStatusCodeErrors(t *testing.T) {
 	defer server.Close()
 	atk := NewAttacker()
 	tr := NewStaticTargeter(Target{Method: "GET", URL: server.URL})
-	res := atk.hit(tr, time.Now())
+	res := atk.hit(tr, "", 0)
 	if got, want := res.Error, "400 Bad Request"; got != want {
 		t.Fatalf("got: %v, want: %v", got, want)
 	}
@@ -181,7 +181,7 @@ func TestBadTargeterError(t *testing.T) {
 	t.Parallel()
 	atk := NewAttacker()
 	tr := func(*Target) error { return io.EOF }
-	res := atk.hit(tr, time.Now())
+	res := atk.hit(tr, "", 0)
 	if got, want := res.Error, io.EOF.Error(); got != want {
 		t.Fatalf("got: %v, want: %v", got, want)
 	}
@@ -199,7 +199,7 @@ func TestResponseBodyCapture(t *testing.T) {
 	defer server.Close()
 	atk := NewAttacker()
 	tr := NewStaticTargeter(Target{Method: "GET", URL: server.URL})
-	res := atk.hit(tr, time.Now())
+	res := atk.hit(tr, "", 0)
 	if got := res.Body; !bytes.Equal(got, want) {
 		t.Fatalf("got: %v, want: %v", got, want)
 	}
@@ -226,7 +226,7 @@ func TestProxyOption(t *testing.T) {
 	}))
 
 	tr := NewStaticTargeter(Target{Method: "GET", URL: "http://127.0.0.2"})
-	res := atk.hit(tr, time.Now())
+	res := atk.hit(tr, "", 0)
 	if got, want := res.Error, ""; got != want {
 		t.Errorf("got error: %q, want %q", got, want)
 	}

lib/reporters.go

@@ -7,6 +7,8 @@ import (
 	"strconv"
 	"strings"
 	"text/tabwriter"
+
+	"github.com/lucasb-eyer/go-colorful"
 )
 
 // A Report represents the state a Reporter uses to write out its reports.
@@ -112,35 +114,69 @@ func NewPlotReporter(title string, rs *Results) Reporter {
 			return err
 		}
 
-		buf := make([]byte, 0, 128)
-		for i, result := range *rs {
-			buf = append(buf, '[')
-			buf = append(buf, strconv.FormatFloat(
-				result.Timestamp.Sub((*rs)[0].Timestamp).Seconds(), 'f', -1, 32)...)
-			buf = append(buf, ',')
-
-			latency := strconv.FormatFloat(result.Latency.Seconds()*1000, 'f', -1, 32)
-			if result.Error == "" {
-				buf = append(buf, "NaN,"...)
-				buf = append(buf, latency...)
-				buf = append(buf, ']', ',')
-			} else {
-				buf = append(buf, latency...)
-				buf = append(buf, ",NaN],"...)
-			}
+		attacks := make(map[string]Results, len(*rs))
+		for _, r := range *rs {
+			attacks[r.Attack] = append(attacks[r.Attack], r)
+		}
 
-			if i == len(*rs)-1 {
-				buf = buf[:len(buf)-1]
-			}
+		const series = 2 // OK and Errors
+		i, offsets := 0, make(map[string]int, len(attacks))
+		for attack := range attacks {
+			offsets[attack] = 1 + i*series
+			i++
+		}
 
-			if _, err = w.Write(buf); err != nil {
-				return err
+		const nan = "NaN"
+
+		data := make([]string, 1+len(attacks)*series)
+		for attack, results := range attacks {
+			for i, r := range results {
+				for j := range data {
+					data[j] = nan
+				}
+
+				offset := offsets[attack]
+				if r.Error == "" {
+					offset++
+				}
+
+				ts := r.Timestamp.Sub(results[0].Timestamp).Seconds()
+				data[0] = strconv.FormatFloat(ts, 'f', -1, 32)
+
+				latency := r.Latency.Seconds() * 1000
+				data[offset] = strconv.FormatFloat(latency, 'f', -1, 32)
+
+				s := "[" + strings.Join(data, ",") + "]"
+
+				if i < len(*rs)-1 {
+					s += ","
+				}
+
+				if _, err = io.WriteString(w, s); err != nil {
+					return err
+				}
 			}
+		}
+
+		labels := make([]string, len(data))
+		labels[0] = strconv.Quote("Seconds")
+
+		for attack, offset := range offsets {
+			labels[offset] = strconv.Quote(attack + " - ERR")
+			labels[offset+1] = strconv.Quote(attack + " - OK")
+		}
+
+		colors := make([]string, 0, len(labels)-1)
+		palette, err := colorful.HappyPalette(cap(colors))
+		if err != nil {
+			return err
+		}
 
-			buf = buf[:0]
+		for _, color := range palette {
+			colors = append(colors, strconv.Quote(color.Hex()))
 		}
 
-		_, err = fmt.Fprintf(w, plotsTemplateTail, title)
+		_, err = fmt.Fprintf(w, plotsTemplateTail, title, strings.Join(labels, ","), strings.Join(colors, ","))
 		return err
 	}
 }
@@ -163,11 +199,11 @@ const (
 	plotsTemplateTail = `],
     {
       title: '%s',
-      labels: ['Seconds', 'ERR', 'OK'],
+      labels: [%s],
       ylabel: 'Latency (ms)',
       xlabel: 'Seconds elapsed',
+      colors: [%s],
       showRoller: true,
-      colors: ['#FA7878', '#8AE234'],
       legend: 'always',
       logscale: true,
       strokeWidth: 1.3

lib/results.go

@@ -18,6 +18,8 @@ func init() {
 
 // Result contains the results of a single Target hit.
 type Result struct {
+	Attack    string        `json:"attack"`
+	Seq       uint64        `json:"seq"`
 	Code      uint16        `json:"code"`
 	Timestamp time.Time     `json:"timestamp"`
 	Latency   time.Duration `json:"latency"`
@@ -32,7 +34,9 @@ func (r *Result) End() time.Time { return r.Timestamp.Add(r.Latency) }
 
 // Equal returns true if the given Result is equal to the receiver.
 func (r Result) Equal(other Result) bool {
-	return r.Code == other.Code &&
+	return r.Attack == other.Attack &&
+		r.Seq == other.Seq &&
+		r.Code == other.Code &&
 		r.Timestamp.Equal(other.Timestamp) &&
 		r.Latency == other.Latency &&
 		r.BytesIn == other.BytesIn &&
@@ -115,6 +119,8 @@ func NewCSVEncoder(w io.Writer) Encoder {
 			strconv.FormatUint(r.BytesIn, 10),
 			r.Error,
 			base64.StdEncoding.EncodeToString(r.Body),
+			r.Attack,
+			strconv.FormatUint(r.Seq, 10),
 		})
 
 		if err != nil {
@@ -165,6 +171,11 @@ func NewCSVDecoder(rd io.Reader) Decoder {
 		r.Error = rec[5]
 		r.Body, err = base64.StdEncoding.DecodeString(rec[6])
 
+		r.Attack = rec[7]
+		if r.Seq, err = strconv.ParseUint(rec[8], 10, 64); err != nil {
+			return err
+		}
+
 		return err
 	}
 }

lib/results_test.go

@@ -63,8 +63,10 @@ func TestEncoding(t *testing.T) {
 			var buf bytes.Buffer
 			enc := tc.enc(&buf)
 			dec := tc.dec(&buf)
-			err := quick.Check(func(code uint16, ts uint32, latency time.Duration, bsIn, bsOut uint64, body []byte, e string) bool {
+			err := quick.Check(func(code uint16, ts uint32, latency time.Duration, seq, bsIn, bsOut uint64, body []byte, attack, e string) bool {
 				want := Result{
+					Attack:    attack,
+					Seq:       seq,
 					Code:      code,
 					Timestamp: time.Unix(int64(ts), 0),
 					Latency:   latency,