From 625ed54f1ebcc006e7d04a27544807fbaff08508 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Thu, 8 Jul 2021 10:30:22 +0300 Subject: [PATCH 1/4] Allow input file read ahead Input file now pre-reads N requests, sort them by timestamp and emit on demand. You can control read depth using --input-file-read-depth which is 100 by default. It makes implementaiton faster, and it fix various issues when due to concurrenccy, or another issues requests gets addeed out of order. --- emitter.go | 2 +- input_file.go | 143 ++++++++++++++++++++++++++++++++++++-------- input_file_test.go | 12 ++-- output_dummy.go | 1 + output_file_test.go | 2 +- plugins.go | 3 +- s3_test.go | 2 +- settings.go | 10 ++-- 8 files changed, 134 insertions(+), 41 deletions(-) diff --git a/emitter.go b/emitter.go index 3db337c..1c985f1 100644 --- a/emitter.go +++ b/emitter.go @@ -152,7 +152,7 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error { } } else { for _, dst := range writers { - if _, err := dst.PluginWrite(msg); err != nil { + if _, err := dst.PluginWrite(msg); err != nil && err != io.ErrClosedPipe { return err } } diff --git a/input_file.go b/input_file.go index 5447492..3fa6d11 100644 --- a/input_file.go +++ b/input_file.go @@ -4,6 +4,7 @@ import ( "bufio" "bytes" "compress/gzip" + "container/heap" "errors" "fmt" "io" @@ -20,27 +21,71 @@ import ( "github.com/aws/aws-sdk-go/service/s3" ) -type fileInputReader struct { - reader *bufio.Reader +type filePayload struct { data []byte - file io.ReadCloser timestamp int64 - closed int32 // Value of 0 indicates that the file is still open. - s3 bool } -func (f *fileInputReader) parseNext() error { +// An IntHeap is a min-heap of ints. +type payloadQueue struct { + sync.RWMutex + s []*filePayload +} + +func (h payloadQueue) Len() int { return len(h.s) } +func (h payloadQueue) Less(i, j int) bool { return h.s[i].timestamp < h.s[j].timestamp } +func (h payloadQueue) Swap(i, j int) { h.s[i], h.s[j] = h.s[j], h.s[i] } + +func (h *payloadQueue) Push(x interface{}) { + // Push and Pop use pointer receivers because they modify the slice's length, + // not just its contents. + h.s = append(h.s, x.(*filePayload)) +} + +func (h *payloadQueue) Pop() interface{} { + old := h.s + n := len(old) + x := old[n-1] + h.s = old[0 : n-1] + return x +} + +func (h payloadQueue) Idx(i int) *filePayload { + h.RLock() + defer h.RUnlock() + + return h.s[i] +} + +type fileInputReader struct { + reader *bufio.Reader + file io.ReadCloser + closed int32 // Value of 0 indicates that the file is still open. + s3 bool + queue payloadQueue + readDepth int +} + +func (f *fileInputReader) parse(init chan struct{}) error { payloadSeparatorAsBytes := []byte(payloadSeparator) var buffer bytes.Buffer + var initialized bool + for { line, err := f.reader.ReadBytes('\n') if err != nil { if err != io.EOF { Debug(1, err) - } else { - f.Close() } + + f.Close() + + if !initialized { + close(init) + initialized = true + } + return err } @@ -48,21 +93,51 @@ func (f *fileInputReader) parseNext() error { asBytes := buffer.Bytes() meta := payloadMeta(asBytes) - f.timestamp, _ = strconv.ParseInt(string(meta[2]), 10, 64) - f.data = asBytes[:len(asBytes)-1] + timestamp, _ := strconv.ParseInt(string(meta[2]), 10, 64) + data := asBytes[:len(asBytes)-1] - return nil + f.queue.Lock() + heap.Push(&f.queue, &filePayload{ + timestamp: timestamp, + data: data, + }) + f.queue.Unlock() + + for { + if f.queue.Len() < f.readDepth { + break + } + + if !initialized { + close(init) + initialized = true + } + + time.Sleep(100 * time.Millisecond) + } + + buffer = bytes.Buffer{} + continue } buffer.Write(line) } - } -func (f *fileInputReader) ReadPayload() []byte { - defer f.parseNext() +func (f *fileInputReader) wait() { + for { + if atomic.LoadInt32(&f.closed) == 1 { + return + } - return f.data + if f.queue.Len() > 0 { + return + } + + time.Sleep(100 * time.Millisecond) + } + + return } // Close closes this plugin @@ -75,7 +150,7 @@ func (f *fileInputReader) Close() error { return nil } -func newFileInputReader(path string) *fileInputReader { +func newFileInputReader(path string, readDepth int) *fileInputReader { var file io.ReadCloser var err error @@ -90,7 +165,7 @@ func newFileInputReader(path string) *fileInputReader { return nil } - r := &fileInputReader{file: file, closed: 0} + r := &fileInputReader{file: file, closed: 0, readDepth: readDepth} if strings.HasSuffix(path, ".gz") { gzReader, err := gzip.NewReader(file) if err != nil { @@ -102,7 +177,11 @@ func newFileInputReader(path string) *fileInputReader { r.reader = bufio.NewReader(file) } - r.parseNext() + heap.Init(&r.queue) + + init := make(chan struct{}) + go r.parse(init) + <-init return r } @@ -116,16 +195,18 @@ type FileInput struct { readers []*fileInputReader speedFactor float64 loop bool + readDepth int } // NewFileInput constructor for FileInput. Accepts file path as argument. -func NewFileInput(path string, loop bool) (i *FileInput) { +func NewFileInput(path string, loop bool, readDepth int) (i *FileInput) { i = new(FileInput) i.data = make(chan []byte, 1000) i.exit = make(chan bool) i.path = path i.speedFactor = 1 i.loop = loop + i.readDepth = readDepth if err := i.init(); err != nil { return @@ -176,7 +257,7 @@ func (i *FileInput) init() (err error) { i.readers = make([]*fileInputReader, len(matches)) for idx, p := range matches { - i.readers[idx] = newFileInputReader(p) + i.readers[idx] = newFileInputReader(p, i.readDepth) } return nil @@ -201,11 +282,17 @@ func (i *FileInput) String() string { // Find reader with smallest timestamp e.g next payload in row func (i *FileInput) nextReader() (next *fileInputReader) { for _, r := range i.readers { - if r == nil || atomic.LoadInt32(&r.closed) != 0 { + if r == nil { continue } - if next == nil || r.timestamp < next.timestamp { + r.wait() + + if r.queue.Len() == 0 { + continue + } + + if next == nil || r.queue.Idx(0).timestamp > next.queue.Idx(0).timestamp { next = r continue } @@ -236,19 +323,23 @@ func (i *FileInput) emit() { } } + reader.queue.RLock() + payload := heap.Pop(&reader.queue).(*filePayload) + reader.queue.RUnlock() + if lastTime != -1 { - diff := reader.timestamp - lastTime + diff := payload.timestamp - lastTime if i.speedFactor != 1 { diff = int64(float64(diff) / i.speedFactor) } if diff >= 0 { - lastTime = reader.timestamp + lastTime = payload.timestamp time.Sleep(time.Duration(diff)) } } else { - lastTime = reader.timestamp + lastTime = payload.timestamp } // Recheck if we have exited since last check. @@ -256,7 +347,7 @@ func (i *FileInput) emit() { case <-i.exit: return default: - i.data <- reader.ReadPayload() + i.data <- payload.data } } diff --git a/input_file_test.go b/input_file_test.go index 91b9bcc..c6dcd8d 100644 --- a/input_file_test.go +++ b/input_file_test.go @@ -104,7 +104,7 @@ func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) { file2.Write([]byte(payloadSeparator)) file2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) for i := '1'; i <= '4'; i++ { msg, _ := input.PluginRead() @@ -130,7 +130,7 @@ func TestInputFileRequestsWithLatency(t *testing.T) { file.Write([]byte("1 3 250000000\nrequest3")) file.Write([]byte(payloadSeparator)) - input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false) + input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100) start := time.Now().UnixNano() for i := 0; i < 3; i++ { @@ -170,7 +170,7 @@ func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) { file2.Write([]byte(payloadSeparator)) file2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) for i := '1'; i <= '4'; i++ { msg, _ := input.PluginRead() @@ -198,7 +198,7 @@ func TestInputFileLoop(t *testing.T) { file.Write([]byte(payloadSeparator)) file.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true) + input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100) // Even if we have just 2 requests in file, it should indifinitly loop for i := 0; i < 1000; i++ { @@ -226,7 +226,7 @@ func TestInputFileCompressed(t *testing.T) { name2 := output2.file.Name() output2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) for i := 0; i < 2000; i++ { input.PluginRead() } @@ -326,7 +326,7 @@ func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile { func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback) (err error) { wg := new(sync.WaitGroup) - input := NewFileInput(captureFile.Name(), false) + input := NewFileInput(captureFile.Name(), false, 100) output := NewTestOutput(func(msg *Message) { callback(msg) wg.Done() diff --git a/output_dummy.go b/output_dummy.go index e28653f..3d67a59 100644 --- a/output_dummy.go +++ b/output_dummy.go @@ -24,6 +24,7 @@ func (i *DummyOutput) PluginWrite(msg *Message) (int, error) { n += nn nn, err = os.Stdout.Write(payloadSeparatorAsBytes) n += nn + return n, err } diff --git a/output_file_test.go b/output_file_test.go index d68c183..53794fb 100644 --- a/output_file_test.go +++ b/output_file_test.go @@ -39,7 +39,7 @@ func TestFileOutput(t *testing.T) { emitter.Close() var counter int64 - input2 := NewFileInput("/tmp/test_requests.gor", false) + input2 := NewFileInput("/tmp/test_requests.gor", false, 100) output2 := NewTestOutput(func(*Message) { atomic.AddInt64(&counter, 1) wg.Done() diff --git a/plugins.go b/plugins.go index b1cef48..1f1363c 100644 --- a/plugins.go +++ b/plugins.go @@ -83,7 +83,6 @@ func (plugins *InOutPlugins) registerPlugin(constructor interface{}, options ... plugins.Outputs = append(plugins.Outputs, w) } plugins.All = append(plugins.All, plugin) - } // NewPlugins specify and initialize all available plugins @@ -119,7 +118,7 @@ func NewPlugins() *InOutPlugins { } for _, options := range Settings.InputFile { - plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop) + plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth) } for _, path := range Settings.OutputFile { diff --git a/s3_test.go b/s3_test.go index fc48be5..f1f85ff 100644 --- a/s3_test.go +++ b/s3_test.go @@ -127,7 +127,7 @@ func TestInputFileFromS3(t *testing.T) { <-output.closeCh } - input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false) + input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd, 100), false) buf := make([]byte, 1000) for i := 0; i <= 19999; i++ { diff --git a/settings.go b/settings.go index 954aa0a..2ecdfc7 100644 --- a/settings.go +++ b/settings.go @@ -45,10 +45,11 @@ type AppSettings struct { OutputTCPConfig TCPOutputConfig OutputTCPStats bool `json:"output-tcp-stats"` - InputFile MultiOption `json:"input-file"` - InputFileLoop bool `json:"input-file-loop"` - OutputFile MultiOption `json:"output-file"` - OutputFileConfig FileOutputConfig + InputFile MultiOption `json:"input-file"` + InputFileLoop bool `json:"input-file-loop"` + InputFileReadDepth int `json:"input-file-read-depth"` + OutputFile MultiOption `json:"output-file"` + OutputFileConfig FileOutputConfig InputRAW MultiOption `json:"input_raw"` RAWInputConfig @@ -113,6 +114,7 @@ func init() { flag.Var(&Settings.InputFile, "input-file", "Read requests from file: \n\tgor --input-file ./requests.gor --output-http staging.com") flag.BoolVar(&Settings.InputFileLoop, "input-file-loop", false, "Loop input files, useful for performance testing.") + flag.IntVar(&Settings.InputFileReadDepth, "input-file-read-depth", 100, "GoReplay tries to read and cache multiple records, in advance. In parallel it also perform sorting of requests, if they came out of order. Since it needs hold this buffer in memory, bigger values can cause worse performance") flag.Var(&Settings.OutputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor") flag.DurationVar(&Settings.OutputFileConfig.FlushInterval, "output-file-flush-interval", time.Second, "Interval for forcing buffer flush to the file, default: 1s.") From 61b377d123be6f5a107bf3d8582d672b739a9a44 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Thu, 8 Jul 2021 23:00:12 +0300 Subject: [PATCH 2/4] Do not use buffer pool for messages This buffers are get re-used and data gets corrupted --- tcp/tcp_message.go | 122 ++------------------------------------------- 1 file changed, 3 insertions(+), 119 deletions(-) diff --git a/tcp/tcp_message.go b/tcp/tcp_message.go index e89d5f8..66f1000 100644 --- a/tcp/tcp_message.go +++ b/tcp/tcp_message.go @@ -9,108 +9,9 @@ import ( "time" "unsafe" - "github.com/buger/goreplay/simpletime" "github.com/buger/goreplay/size" ) -var bufferPool = NewBufferPool(1000, 1) - -type buf struct { - b []byte - created time.Time - gc bool -} - -type bufPool struct { - buffers chan *buf - ttl int -} - -func NewBufferPool(max int, ttl int) *bufPool { - pool := &bufPool{ - buffers: make(chan *buf, max), - ttl: ttl, - } - - // Ensure that memory released over time - go func() { - var released int - // GC - for { - for i := 0; i < 100; i++ { - select { - case c := <-pool.buffers: - if simpletime.Now.Sub(c.created) < time.Duration(ttl)*time.Second { - select { - case pool.buffers <- c: - default: - stats.Add("active_buffer_count", -1) - c.b = nil - c.gc = true - released++ - } - } else { - stats.Add("active_buffer_count", -1) - // Else GC - c.b = nil - c.gc = true - released++ - } - default: - break - } - } - - bufPoolCount.Set(int64(len(pool.buffers))) - releasedCount.Set(int64(released)) - - time.Sleep(1000 * time.Millisecond) - } - }() - - return pool -} - -// Borrow a Client from the pool. -func (p *bufPool) Get() *buf { - var c *buf - select { - case c = <-p.buffers: - default: - stats.Add("total_alloc_buffer_count", 1) - stats.Add("active_buffer_count", 1) - - c = new(buf) - c.b = make([]byte, 1024) - c.created = simpletime.Now - - // Use this technique to find if pool leaks, and objects get GCd - // - // runtime.SetFinalizer(c, func(p *buf) { - // if !p.gc { - // panic("Pool leak") - // } - // }) - } - return c -} - -// Return returns a Client to the pool. -func (p *bufPool) Put(c *buf) { - select { - case p.buffers <- c: - default: - stats.Add("active_buffers", -1) - c.gc = true - c.b = nil - // if pool overloaded, let it go - } -} - -func (p *bufPool) Len() int { - return len(p.buffers) -} - // Stats every message carry its own stats object type Stats struct { LostData int @@ -130,7 +31,6 @@ type Message struct { packets []*Packet parser *MessageParser feedback interface{} - dataBuf *buf Stats } @@ -164,8 +64,6 @@ func (m *Message) UUID() []byte { } func (m *Message) add(packet *Packet) bool { - // fmt.Println("SEQ:", packet.Seq, " - ", len(packet.Payload)) - // Skip duplicates for _, p := range m.packets { if p.Seq == packet.Seq { @@ -228,20 +126,10 @@ func (m *Message) PacketData() [][]byte { // Data returns data in this message func (m *Message) Data() []byte { - m.dataBuf = bufferPool.Get() + var tmp []byte + tmp, _ = copySlice(tmp, m.PacketData()...) - // var totalLen int - // for _, p := range m.packets { - // totalLen += len(p.Payload) - // } - // tmp := make([]byte, totalLen) - var n int - if m.dataBuf == nil { - panic("asdsd") - } - m.dataBuf.b, n = copySlice(m.dataBuf.b, m.PacketData()...) - - return m.dataBuf.b[:n] + return tmp } // SetProtocolState set feedback/data that can be used later, e.g with End or Start hint @@ -264,10 +152,6 @@ func (m *Message) Finalize() { for _, p := range m.packets { packetPool.Put(p) } - - if m.dataBuf != nil { - bufferPool.Put(m.dataBuf) - } } // Emitter message handler From 8e76559b492402ef1d2aee99e0106d085db7f935 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Thu, 8 Jul 2021 23:09:50 +0300 Subject: [PATCH 3/4] Add --input-file-dry-run option Now you can get information about file content without performing the actual replay. For example, it can tell you how many requests in your files, and how long it will take to replay them. --- input_file.go | 66 +++++++++++++++++++++++++++++++++++++++------ input_file_test.go | 12 ++++----- input_raw_test.go | 9 ++++--- output_file_test.go | 2 +- output_http.go | 1 + plugins.go | 2 +- s3_test.go | 2 +- settings.go | 2 ++ 8 files changed, 75 insertions(+), 21 deletions(-) diff --git a/input_file.go b/input_file.go index 3fa6d11..8f45c99 100644 --- a/input_file.go +++ b/input_file.go @@ -6,8 +6,10 @@ import ( "compress/gzip" "container/heap" "errors" + "expvar" "fmt" "io" + "math" "os" "path/filepath" "strconv" @@ -51,9 +53,6 @@ func (h *payloadQueue) Pop() interface{} { } func (h payloadQueue) Idx(i int) *filePayload { - h.RLock() - defer h.RUnlock() - return h.s[i] } @@ -196,10 +195,13 @@ type FileInput struct { speedFactor float64 loop bool readDepth int + dryRun bool + + stats *expvar.Map } // NewFileInput constructor for FileInput. Accepts file path as argument. -func NewFileInput(path string, loop bool, readDepth int) (i *FileInput) { +func NewFileInput(path string, loop bool, readDepth int, dryRun bool) (i *FileInput) { i = new(FileInput) i.data = make(chan []byte, 1000) i.exit = make(chan bool) @@ -207,6 +209,8 @@ func NewFileInput(path string, loop bool, readDepth int) (i *FileInput) { i.speedFactor = 1 i.loop = loop i.readDepth = readDepth + i.stats = expvar.NewMap("file-" + path) + i.dryRun = dryRun if err := i.init(); err != nil { return @@ -246,7 +250,6 @@ func (i *FileInput) init() (err error) { } else if matches, err = filepath.Glob(i.path); err != nil { Debug(0, "[INPUT-FILE] Wrong file pattern", i.path, err) return - } if len(matches) == 0 { @@ -260,6 +263,8 @@ func (i *FileInput) init() (err error) { i.readers[idx] = newFileInputReader(p, i.readDepth) } + i.stats.Add("reader_count", int64(len(matches))) + return nil } @@ -270,6 +275,7 @@ func (i *FileInput) PluginRead() (*Message, error) { case <-i.exit: return nil, ErrorStopped case buf := <-i.data: + i.stats.Add("read_from", 1) msg.Meta, msg.Data = payloadMetaWithBody(buf) return &msg, nil } @@ -292,7 +298,7 @@ func (i *FileInput) nextReader() (next *fileInputReader) { continue } - if next == nil || r.queue.Idx(0).timestamp > next.queue.Idx(0).timestamp { + if next == nil || r.queue.Idx(0).timestamp < next.queue.Idx(0).timestamp { next = r continue } @@ -304,6 +310,11 @@ func (i *FileInput) nextReader() (next *fileInputReader) { func (i *FileInput) emit() { var lastTime int64 = -1 + var maxWait, firstWait, minWait int64 + minWait = math.MaxInt64 + + i.stats.Add("negative_wait", 0) + for { select { case <-i.exit: @@ -325,18 +336,39 @@ func (i *FileInput) emit() { reader.queue.RLock() payload := heap.Pop(&reader.queue).(*filePayload) + i.stats.Add("total_counter", 1) + i.stats.Add("total_bytes", int64(len(payload.data))) reader.queue.RUnlock() if lastTime != -1 { diff := payload.timestamp - lastTime + if firstWait == 0 { + firstWait = diff + } + if i.speedFactor != 1 { diff = int64(float64(diff) / i.speedFactor) } if diff >= 0 { lastTime = payload.timestamp - time.Sleep(time.Duration(diff)) + + if !i.dryRun { + time.Sleep(time.Duration(diff)) + } + + i.stats.Add("total_wait", diff) + + if diff > maxWait { + maxWait = diff + } + + if diff < minWait { + minWait = diff + } + } else { + i.stats.Add("negative_wait", 1) } } else { lastTime = payload.timestamp @@ -347,12 +379,30 @@ func (i *FileInput) emit() { case <-i.exit: return default: - i.data <- payload.data + if !i.dryRun { + i.data <- payload.data + } } } + i.stats.Set("first_wait", time.Duration(firstWait)) + i.stats.Set("max_wait", time.Duration(maxWait)) + i.stats.Set("min_wait", time.Duration(minWait)) + Debug(0, fmt.Sprintf("[INPUT-FILE] FileInput: end of file '%s'\n", i.path)) + if i.dryRun { + fmt.Printf("Records found: %v\nFiles processed: %v\nBytes processed: %v\nMax wait: %v\nMin wait: %v\nFirst wait: %v\nIt will take `%v` to replay at current speed.\nFound %v records with out of order timestamp\n", + i.stats.Get("total_counter"), + i.stats.Get("reader_count"), + i.stats.Get("total_bytes"), + i.stats.Get("max_wait"), + i.stats.Get("min_wait"), + i.stats.Get("first_wait"), + time.Duration(i.stats.Get("total_wait").(*expvar.Int).Value()), + i.stats.Get("negative_wait"), + ) + } } // Close closes this plugin diff --git a/input_file_test.go b/input_file_test.go index c6dcd8d..7fef0f8 100644 --- a/input_file_test.go +++ b/input_file_test.go @@ -104,7 +104,7 @@ func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) { file2.Write([]byte(payloadSeparator)) file2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, false) for i := '1'; i <= '4'; i++ { msg, _ := input.PluginRead() @@ -130,7 +130,7 @@ func TestInputFileRequestsWithLatency(t *testing.T) { file.Write([]byte("1 3 250000000\nrequest3")) file.Write([]byte(payloadSeparator)) - input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100) + input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100, false) start := time.Now().UnixNano() for i := 0; i < 3; i++ { @@ -170,7 +170,7 @@ func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) { file2.Write([]byte(payloadSeparator)) file2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, false) for i := '1'; i <= '4'; i++ { msg, _ := input.PluginRead() @@ -198,7 +198,7 @@ func TestInputFileLoop(t *testing.T) { file.Write([]byte(payloadSeparator)) file.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100) + input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100, false) // Even if we have just 2 requests in file, it should indifinitly loop for i := 0; i < 1000; i++ { @@ -226,7 +226,7 @@ func TestInputFileCompressed(t *testing.T) { name2 := output2.file.Name() output2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, false) for i := 0; i < 2000; i++ { input.PluginRead() } @@ -326,7 +326,7 @@ func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile { func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback) (err error) { wg := new(sync.WaitGroup) - input := NewFileInput(captureFile.Name(), false, 100) + input := NewFileInput(captureFile.Name(), false, 100, false) output := NewTestOutput(func(msg *Message) { callback(msg) wg.Done() diff --git a/input_raw_test.go b/input_raw_test.go index d844f6a..969e5a7 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -233,10 +233,11 @@ func TestInputRAWChunkedEncoding(t *testing.T) { originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) conf := RAWInputConfig{ - Engine: capture.EnginePcap, - Expire: time.Second, - Protocol: ProtocolHTTP, - TrackResponse: true, + Engine: capture.EnginePcap, + Expire: time.Second, + Protocol: ProtocolHTTP, + TrackResponse: true, + AllowIncomplete: true, } input := NewRAWInput(originAddr, conf) diff --git a/output_file_test.go b/output_file_test.go index 53794fb..4190df8 100644 --- a/output_file_test.go +++ b/output_file_test.go @@ -39,7 +39,7 @@ func TestFileOutput(t *testing.T) { emitter.Close() var counter int64 - input2 := NewFileInput("/tmp/test_requests.gor", false, 100) + input2 := NewFileInput("/tmp/test_requests.gor", false, 100, false) output2 := NewTestOutput(func(*Message) { atomic.AddInt64(&counter, 1) wg.Done() diff --git a/output_http.go b/output_http.go index 43c8240..cb0c0de 100644 --- a/output_http.go +++ b/output_http.go @@ -215,6 +215,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, msg *Message) { if !isRequestPayload(msg.Meta) { return } + uuid := payloadID(msg.Meta) start := time.Now() resp, err := client.Send(msg.Data) diff --git a/plugins.go b/plugins.go index 1f1363c..54d4e7c 100644 --- a/plugins.go +++ b/plugins.go @@ -118,7 +118,7 @@ func NewPlugins() *InOutPlugins { } for _, options := range Settings.InputFile { - plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth) + plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth, Settings.InputFileDryRun) } for _, path := range Settings.OutputFile { diff --git a/s3_test.go b/s3_test.go index f1f85ff..65453d3 100644 --- a/s3_test.go +++ b/s3_test.go @@ -127,7 +127,7 @@ func TestInputFileFromS3(t *testing.T) { <-output.closeCh } - input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd, 100), false) + input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false, 100, false) buf := make([]byte, 1000) for i := 0; i <= 19999; i++ { diff --git a/settings.go b/settings.go index 2ecdfc7..c64c7cb 100644 --- a/settings.go +++ b/settings.go @@ -48,6 +48,7 @@ type AppSettings struct { InputFile MultiOption `json:"input-file"` InputFileLoop bool `json:"input-file-loop"` InputFileReadDepth int `json:"input-file-read-depth"` + InputFileDryRun bool `json:"input-file-dry-run"` OutputFile MultiOption `json:"output-file"` OutputFileConfig FileOutputConfig @@ -115,6 +116,7 @@ func init() { flag.Var(&Settings.InputFile, "input-file", "Read requests from file: \n\tgor --input-file ./requests.gor --output-http staging.com") flag.BoolVar(&Settings.InputFileLoop, "input-file-loop", false, "Loop input files, useful for performance testing.") flag.IntVar(&Settings.InputFileReadDepth, "input-file-read-depth", 100, "GoReplay tries to read and cache multiple records, in advance. In parallel it also perform sorting of requests, if they came out of order. Since it needs hold this buffer in memory, bigger values can cause worse performance") + flag.BoolVar(&Settings.InputFileDryRun, "input-file-dry-run", false, "Simulate reading from the data source without replaying it. You will get information about expected replay time, number of found records etc.") flag.Var(&Settings.OutputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor") flag.DurationVar(&Settings.OutputFileConfig.FlushInterval, "output-file-flush-interval", time.Second, "Interval for forcing buffer flush to the file, default: 1s.") From 19ad90aa4bb0320c8ba623d5509a0690e8c3ec25 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Thu, 8 Jul 2021 23:23:02 +0300 Subject: [PATCH 4/4] Added time --input-raw-max-wait In some low traffic cases you can have cases when time between request minutes. Additionally increasing speed can be not an option. Now you can "skip" this pauses, by seetting max wait time --- input_file.go | 8 +++++++- input_file_test.go | 12 ++++++------ output_file_test.go | 2 +- plugins.go | 2 +- s3_test.go | 2 +- settings.go | 12 +++++++----- 6 files changed, 23 insertions(+), 15 deletions(-) diff --git a/input_file.go b/input_file.go index 8f45c99..d1e4230 100644 --- a/input_file.go +++ b/input_file.go @@ -196,12 +196,13 @@ type FileInput struct { loop bool readDepth int dryRun bool + maxWait time.Duration stats *expvar.Map } // NewFileInput constructor for FileInput. Accepts file path as argument. -func NewFileInput(path string, loop bool, readDepth int, dryRun bool) (i *FileInput) { +func NewFileInput(path string, loop bool, readDepth int, maxWait time.Duration, dryRun bool) (i *FileInput) { i = new(FileInput) i.data = make(chan []byte, 1000) i.exit = make(chan bool) @@ -211,6 +212,7 @@ func NewFileInput(path string, loop bool, readDepth int, dryRun bool) (i *FileIn i.readDepth = readDepth i.stats = expvar.NewMap("file-" + path) i.dryRun = dryRun + i.maxWait = maxWait if err := i.init(); err != nil { return @@ -351,6 +353,10 @@ func (i *FileInput) emit() { diff = int64(float64(diff) / i.speedFactor) } + if i.maxWait > 0 && diff > int64(i.maxWait) { + diff = int64(i.maxWait) + } + if diff >= 0 { lastTime = payload.timestamp diff --git a/input_file_test.go b/input_file_test.go index 7fef0f8..0538e62 100644 --- a/input_file_test.go +++ b/input_file_test.go @@ -104,7 +104,7 @@ func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) { file2.Write([]byte(payloadSeparator)) file2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false) for i := '1'; i <= '4'; i++ { msg, _ := input.PluginRead() @@ -130,7 +130,7 @@ func TestInputFileRequestsWithLatency(t *testing.T) { file.Write([]byte("1 3 250000000\nrequest3")) file.Write([]byte(payloadSeparator)) - input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100, false) + input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100, 0, false) start := time.Now().UnixNano() for i := 0; i < 3; i++ { @@ -170,7 +170,7 @@ func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) { file2.Write([]byte(payloadSeparator)) file2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false) for i := '1'; i <= '4'; i++ { msg, _ := input.PluginRead() @@ -198,7 +198,7 @@ func TestInputFileLoop(t *testing.T) { file.Write([]byte(payloadSeparator)) file.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100, false) + input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100, 0, false) // Even if we have just 2 requests in file, it should indifinitly loop for i := 0; i < 1000; i++ { @@ -226,7 +226,7 @@ func TestInputFileCompressed(t *testing.T) { name2 := output2.file.Name() output2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false) for i := 0; i < 2000; i++ { input.PluginRead() } @@ -326,7 +326,7 @@ func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile { func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback) (err error) { wg := new(sync.WaitGroup) - input := NewFileInput(captureFile.Name(), false, 100, false) + input := NewFileInput(captureFile.Name(), false, 100, 0, false) output := NewTestOutput(func(msg *Message) { callback(msg) wg.Done() diff --git a/output_file_test.go b/output_file_test.go index 4190df8..3fd00d6 100644 --- a/output_file_test.go +++ b/output_file_test.go @@ -39,7 +39,7 @@ func TestFileOutput(t *testing.T) { emitter.Close() var counter int64 - input2 := NewFileInput("/tmp/test_requests.gor", false, 100, false) + input2 := NewFileInput("/tmp/test_requests.gor", false, 100, 0, false) output2 := NewTestOutput(func(*Message) { atomic.AddInt64(&counter, 1) wg.Done() diff --git a/plugins.go b/plugins.go index 54d4e7c..e6b1a1e 100644 --- a/plugins.go +++ b/plugins.go @@ -118,7 +118,7 @@ func NewPlugins() *InOutPlugins { } for _, options := range Settings.InputFile { - plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth, Settings.InputFileDryRun) + plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth, Settings.InputFileMaxWait, Settings.InputFileDryRun) } for _, path := range Settings.OutputFile { diff --git a/s3_test.go b/s3_test.go index 65453d3..a6b12f9 100644 --- a/s3_test.go +++ b/s3_test.go @@ -127,7 +127,7 @@ func TestInputFileFromS3(t *testing.T) { <-output.closeCh } - input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false, 100, false) + input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false, 100, 0, false) buf := make([]byte, 1000) for i := 0; i <= 19999; i++ { diff --git a/settings.go b/settings.go index c64c7cb..b992731 100644 --- a/settings.go +++ b/settings.go @@ -45,11 +45,12 @@ type AppSettings struct { OutputTCPConfig TCPOutputConfig OutputTCPStats bool `json:"output-tcp-stats"` - InputFile MultiOption `json:"input-file"` - InputFileLoop bool `json:"input-file-loop"` - InputFileReadDepth int `json:"input-file-read-depth"` - InputFileDryRun bool `json:"input-file-dry-run"` - OutputFile MultiOption `json:"output-file"` + InputFile MultiOption `json:"input-file"` + InputFileLoop bool `json:"input-file-loop"` + InputFileReadDepth int `json:"input-file-read-depth"` + InputFileDryRun bool `json:"input-file-dry-run"` + InputFileMaxWait time.Duration `json:"input-file-max-wait"` + OutputFile MultiOption `json:"output-file"` OutputFileConfig FileOutputConfig InputRAW MultiOption `json:"input_raw"` @@ -117,6 +118,7 @@ func init() { flag.BoolVar(&Settings.InputFileLoop, "input-file-loop", false, "Loop input files, useful for performance testing.") flag.IntVar(&Settings.InputFileReadDepth, "input-file-read-depth", 100, "GoReplay tries to read and cache multiple records, in advance. In parallel it also perform sorting of requests, if they came out of order. Since it needs hold this buffer in memory, bigger values can cause worse performance") flag.BoolVar(&Settings.InputFileDryRun, "input-file-dry-run", false, "Simulate reading from the data source without replaying it. You will get information about expected replay time, number of found records etc.") + flag.DurationVar(&Settings.InputFileMaxWait, "input-file-max-wait", 0, "Set the maximum time between requests. Can help in situations when you have too long periods between request, and you want to skip them. Example: --input-raw-max-wait 1s") flag.Var(&Settings.OutputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor") flag.DurationVar(&Settings.OutputFileConfig.FlushInterval, "output-file-flush-interval", time.Second, "Interval for forcing buffer flush to the file, default: 1s.")