Merge pull request #69 from skipor/feature/56

Ammo num, autotagging and minor fixes.

.travis.yml

@@ -9,9 +9,13 @@ before_install:
   - sudo add-apt-repository ppa:masterminds/glide -y
   - sudo apt-get update -q
   - sudo apt-get install glide -y
+  # Build go tools in separate $GOPATH.
+  # That allows to keep main $GOPATH clean and be sure that only glide.lock deps used.
+  - mkdir $HOME/tools
+  - GOPATH=$HOME/tools make tools
+  - export PATH=$HOME/tools/bin:$PATH
 
 install:
-  - make tools
-  - go get -t `glide novendor`
+  - glide install
 
 script: make travis

cli/cli.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
 	"net/http"
 	"os"
 	"os/signal"
@@ -154,7 +153,7 @@ func startMonitoring(conf monitoringConfig) (stop func()) {
 	if conf.CPUProfile != "" {
 		f, err := os.Create(conf.MemProfile)
 		if err != nil {
-			log.Fatal(err)
+			zap.L().Fatal("Memory profile file create fail", zap.Error(err))
 		}
 		pprof.StartCPUProfile(f)
 		stops = append(stops, func() {
@@ -163,9 +162,9 @@ func startMonitoring(conf monitoringConfig) (stop func()) {
 		})
 	}
 	if conf.MemProfile != "" {
-		f, err := os.Create(conf.MemProfile)
+		f, err := os.Create(conf.CPUProfile)
 		if err != nil {
-			log.Fatal(err)
+			zap.L().Fatal("CPU profile file create fail", zap.Error(err))
 		}
 		stops = append(stops, func() {
 			pprof.WriteHeapProfile(f)

components/phttp/ammo/simple/ammo.go

@@ -17,19 +17,25 @@ type Ammo struct {
 	// Need to research is it possible. http.Transport can hold reference to http.Request.
 	req *http.Request
 	tag string
-}
-
-func NewAmmo(req *http.Request, tag string) *Ammo {
-	return &Ammo{req, tag}
+	id  int
 }
 
 func (a *Ammo) Request() (*http.Request, *netsample.Sample) {
 	sample := netsample.Acquire(a.tag)
+	sample.SetId(a.id)
 	return a.req, sample
 }
 
 func (a *Ammo) Reset(req *http.Request, tag string) {
-	*a = Ammo{req, tag}
+	*a = Ammo{req, tag, -1}
+}
+
+func (a *Ammo) SetId(id int) {
+	a.id = id
+}
+
+func (a *Ammo) Id() int {
+	return a.id
 }
 
 var _ phttp.Ammo = (*Ammo)(nil)

components/phttp/ammo/simple/jsonline/provider.go

@@ -10,7 +10,7 @@ import (
 	"context"
 	"net/http"
 
-	"github.com/facebookgo/stackerr"
+	"github.com/pkg/errors"
 	"github.com/spf13/afero"
 	"go.uber.org/zap"
 
@@ -49,7 +49,7 @@ func (p *Provider) start(ctx context.Context, ammoFile afero.File) error {
 			data := scanner.Bytes()
 			a, err := decodeAmmo(data, p.Pool.Get().(*simple.Ammo))
 			if err != nil {
-				return stackerr.Newf("failed to decode ammo at line: %v; data: %q; error: %s", line, data, err)
+				return errors.Wrapf(err, "failed to decode ammo at line: %v; data: %q", line, data)
 			}
 			ammoNum++
 			select {
@@ -71,7 +71,7 @@ func decodeAmmo(jsonDoc []byte, am *simple.Ammo) (*simple.Ammo, error) {
 	var data data
 	err := data.UnmarshalJSON(jsonDoc)
 	if err != nil {
-		return nil, stackerr.Wrap(err)
+		return nil, errors.WithStack(err)
 	}
 	req, err := data.ToRequest()
 	if err != nil {
@@ -85,7 +85,7 @@ func (d *data) ToRequest() (req *http.Request, err error) {
 	uri := "http://" + d.Host + d.Uri
 	req, err = http.NewRequest(d.Method, uri, nil)
 	if err != nil {
-		return nil, stackerr.Wrap(err)
+		return nil, errors.WithStack(err)
 	}
 	for k, v := range d.Headers {
 		req.Header.Set(k, v)

components/phttp/ammo/simple/provider.go

@@ -9,8 +9,10 @@ import (
 	"context"
 	"sync"
 
-	"github.com/facebookgo/stackerr"
+	"github.com/pkg/errors"
 	"github.com/spf13/afero"
+	"go.uber.org/atomic"
+
 	"github.com/yandex/pandora/core"
 )
 
@@ -25,16 +27,20 @@ func NewProvider(fs afero.Fs, fileName string, start func(ctx context.Context, f
 }
 
 type Provider struct {
-	fs       afero.Fs
-	fileName string
-	start    func(ctx context.Context, file afero.File) error
-	Sink     chan *Ammo
-	Pool     sync.Pool
+	fs        afero.Fs
+	fileName  string
+	start     func(ctx context.Context, file afero.File) error
+	Sink      chan *Ammo
+	Pool      sync.Pool
+	idCounter atomic.Int64
 }
 
-func (p *Provider) Acquire() (ammo core.Ammo, ok bool) {
-	ammo, ok = <-p.Sink
-	return
+func (p *Provider) Acquire() (core.Ammo, bool) {
+	ammo, ok := <-p.Sink
+	if ok {
+		ammo.SetId(int(p.idCounter.Inc() - 1))
+	}
+	return ammo, ok
 }
 
 func (p *Provider) Release(a core.Ammo) {
@@ -45,8 +51,7 @@ func (p *Provider) Run(ctx context.Context) error {
 	defer close(p.Sink)
 	file, err := p.fs.Open(p.fileName)
 	if err != nil {
-		// TODO(skipor): instead of passing stacktrace log error here.
-		return stackerr.Newf("failed to open ammo Sink: %v", err)
+		return errors.Wrap(err, "failed to open ammo file")
 	}
 	defer file.Close()
 	return p.start(ctx, file)

components/phttp/ammo/simple/raw/decoder.go

@@ -13,8 +13,6 @@ func decodeHeader(headerString []byte) (reqSize int, tag string, err error) {
 	reqSize, err = strconv.Atoi(parts[0])
 	if len(parts) > 1 {
 		tag = parts[1]
-	} else {
-		tag = "__EMPTY__"
 	}
 	return
 }

components/phttp/ammo/simple/raw/decoder_test.go

@@ -23,7 +23,7 @@ var _ = Describe("Decoder", func() {
 		reqSize, tag, err := decodeHeader([]byte(raw))
 		Expect(err).To(BeNil())
 		Expect(reqSize).To(Equal(123))
-		Expect(tag).To(Equal("__EMPTY__"))
+		Expect(tag).To(Equal(""))
 	})
 	It("should parse GET request", func() {
 		raw := "GET /some/path HTTP/1.0\r\n" +

components/phttp/ammo/simple/raw/provider.go

@@ -4,13 +4,12 @@ import (
 	"bufio"
 	"context"
 	"io"
-	"log"
-	"os"
 
-	"github.com/facebookgo/stackerr"
+	"github.com/pkg/errors"
 	"github.com/spf13/afero"
 
 	"github.com/yandex/pandora/components/phttp/ammo/simple"
+	"go.uber.org/zap"
 )
 
 /*
@@ -44,7 +43,7 @@ User-Agent: xxx (shell 1)
 */
 
 func filePosition(file afero.File) (position int64) {
-	position, _ = file.Seek(0, os.SEEK_CUR)
+	position, _ = file.Seek(0, io.SeekCurrent)
 	return
 }
 
@@ -80,13 +79,13 @@ func (p *Provider) start(ctx context.Context, ammoFile afero.File) error {
 		for p.Limit == 0 || ammoNum < p.Limit {
 			data, isPrefix, err := reader.ReadLine()
 			if isPrefix {
-				return stackerr.Newf("Too long header in ammo at position %v", filePosition(ammoFile))
+				return errors.Errorf("too long header in ammo at position %v", filePosition(ammoFile))
 			}
 			if err == io.EOF {
 				break // start over from the beginning
 			}
 			if err != nil {
-				return stackerr.Newf("error reading ammo at position: %v; error: %s", filePosition(ammoFile), err)
+				return errors.Wrapf(err, "reading ammo failed at position: %v", filePosition(ammoFile))
 			}
 			if len(data) == 0 {
 				continue // skip empty lines
@@ -97,11 +96,11 @@ func (p *Provider) start(ctx context.Context, ammoFile afero.File) error {
 			}
 			buff := make([]byte, reqSize)
 			if n, err := io.ReadFull(reader, buff); err != nil {
-				return stackerr.Newf("failed to read ammo at position: %v; tried to read: %v; have read: %v; error: %s", filePosition(ammoFile), reqSize, n, err)
+				return errors.Wrapf(err, "failed to read ammo at position: %v; tried to read: %v; have read: %v", filePosition(ammoFile), reqSize, n)
 			}
 			req, err := decodeRequest(buff)
 			if err != nil {
-				return stackerr.Newf("failed to decode ammo at position: %v; data: %q; error: %s", filePosition(ammoFile), buff, err)
+				return errors.Wrapf(err, "failed to decode ammo at position: %v; data: %q", filePosition(ammoFile), buff)
 			}
 			sh := p.Pool.Get().(*simple.Ammo)
 			sh.Reset(req, tag)
@@ -114,13 +113,13 @@ func (p *Provider) start(ctx context.Context, ammoFile afero.File) error {
 			}
 		}
 		if ammoNum == 0 {
-			return stackerr.Newf("no ammo in file")
+			return errors.New("no ammo in file")
 		}
 		if p.Passes != 0 && passNum >= p.Passes {
 			break
 		}
 		ammoFile.Seek(0, 0)
 	}
-	log.Println("Ran out of ammo")
+	zap.L().Debug("Ran out of ammo")
 	return nil
 }

components/phttp/ammo/simple/raw/provider_test.go

@@ -21,8 +21,8 @@ const testFile = "./ammo.stpd"
 const testFileData = "../../../testdata/ammo.stpd"
 
 var testData = []ammoData{
-	{"GET", "www.ya.ru", "/", http.Header{"Connection": []string{"close"}}, "__EMPTY__", ""},
-	{"GET", "www.ya.ru", "/test", http.Header{"Connection": []string{"close"}}, "__EMPTY__", ""},
+	{"GET", "www.ya.ru", "/", http.Header{"Connection": []string{"close"}}, "", ""},
+	{"GET", "www.ya.ru", "/test", http.Header{"Connection": []string{"close"}}, "", ""},
 	{"GET", "www.ya.ru", "/test2", http.Header{"Connection": []string{"close"}}, "tag", ""},
 	{"POST", "www.ya.ru", "/test3", http.Header{"Connection": []string{"close"}, "Content-Length": []string{"5"}}, "tag", "hello"},
 }

components/phttp/ammo/simple/uri/decoder.go

@@ -12,7 +12,7 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/facebookgo/stackerr"
+	"github.com/pkg/errors"
 
 	"github.com/yandex/pandora/components/phttp/ammo/simple"
 )
@@ -37,15 +37,15 @@ func newDecoder(ctx context.Context, sink chan<- *simple.Ammo, pool *sync.Pool)
 
 func (d *decoder) Decode(line []byte) error {
 	if len(line) == 0 {
-		return stackerr.Newf("empty line")
+		return errors.New("empty line")
 	}
 	switch line[0] {
 	case '/':
 		return d.decodeURI(line)
 	case '[':
 		return d.decodeHeader(line)
 	}
-	return stackerr.Newf("every line should begin with '[' or '/'")
+	return errors.New("every line should begin with '[' or '/'")
 }
 
 func (d *decoder) decodeURI(line []byte) error {
@@ -55,12 +55,10 @@ func (d *decoder) decodeURI(line []byte) error {
 	var tag string
 	if len(parts) > 1 {
 		tag = parts[1]
-	} else {
-		tag = "__EMPTY__"
 	}
 	req, err := http.NewRequest("GET", string(url), nil)
 	if err != nil {
-		return stackerr.Newf("uri decode error: ", err)
+		return errors.Wrap(err, "uri decode")
 	}
 	for k, v := range d.header {
 		// http.Request.Write sends Host header based on Host or URL.Host.
@@ -84,17 +82,17 @@ func (d *decoder) decodeURI(line []byte) error {
 
 func (d *decoder) decodeHeader(line []byte) error {
 	if len(line) < 3 || line[0] != '[' || line[len(line)-1] != ']' {
-		return stackerr.Newf("header line should be like '[key: value]")
+		return errors.New("header line should be like '[key: value]")
 	}
 	line = line[1 : len(line)-1]
 	colonIdx := bytes.IndexByte(line, ':')
 	if colonIdx < 0 {
-		return stackerr.Newf("missing colon")
+		return errors.New("missing colon")
 	}
 	key := string(bytes.TrimSpace(line[:colonIdx]))
 	val := string(bytes.TrimSpace(line[colonIdx+1:]))
 	if key == "" {
-		return stackerr.Newf("missing header key")
+		return errors.New("missing header key")
 	}
 	d.header.Set(key, val)
 	return nil

components/phttp/ammo/simple/uri/decoder_test.go

@@ -79,7 +79,7 @@ var _ = Describe("Decoder", func() {
 		header.Set("Host", host)
 		Expect(decoder.header).To(Equal(header))
 		Expect(decoder.ammoNum).To(Equal(1))
-		Expect(sample.Tags()).To(Equal("__EMPTY__"))
+		Expect(sample.Tags()).To(Equal(""))
 	})
 	It("uri and tag", func() {
 		header := http.Header{"a": []string{"b"}, "c": []string{"d"}}

components/phttp/ammo/simple/uri/provider_test.go

@@ -30,16 +30,16 @@ const testFileData = `/0
 `
 
 var testData = []ammoData{
-	{"", "/0", http.Header{}, "__EMPTY__"},
-	{"", "/1", http.Header{"A": []string{"b"}}, "__EMPTY__"},
+	{"", "/0", http.Header{}, ""},
+	{"", "/1", http.Header{"A": []string{"b"}}, ""},
 	{"example.com", "/2", http.Header{
 		"A": []string{"b"},
 		"C": []string{"d"},
-	}, "__EMPTY__"},
+	}, ""},
 	{"other.net", "/3", http.Header{
 		"A": []string{""},
 		"C": []string{"d"},
-	}, "__EMPTY__"},
+	}, ""},
 	{"other.net", "/4", http.Header{
 		"A": []string{""},
 		"C": []string{"d"},