mirror of
https://github.com/buger/goreplay.git
synced 2024-04-21 12:32:02 +00:00
Leonid Bugaev
19ad90aa4b
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
334 lines
8.9 KiB
Go
334 lines
8.9 KiB
Go
package main
|
|
|
|
import (
|
|
"fmt"
|
|
"math/rand"
|
|
"os"
|
|
"reflect"
|
|
"sort"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/buger/goreplay/size"
|
|
)
|
|
|
|
func TestFileOutput(t *testing.T) {
|
|
wg := new(sync.WaitGroup)
|
|
|
|
input := NewTestInput()
|
|
output := NewFileOutput("/tmp/test_requests.gor", &FileOutputConfig{FlushInterval: time.Minute, Append: true})
|
|
|
|
plugins := &InOutPlugins{
|
|
Inputs: []PluginReader{input},
|
|
Outputs: []PluginWriter{output},
|
|
}
|
|
plugins.All = append(plugins.All, input, output)
|
|
|
|
emitter := NewEmitter()
|
|
go emitter.Start(plugins, Settings.Middleware)
|
|
|
|
for i := 0; i < 100; i++ {
|
|
wg.Add(2)
|
|
input.EmitGET()
|
|
input.EmitPOST()
|
|
}
|
|
time.Sleep(100 * time.Millisecond)
|
|
output.flush()
|
|
emitter.Close()
|
|
|
|
var counter int64
|
|
input2 := NewFileInput("/tmp/test_requests.gor", false, 100, 0, false)
|
|
output2 := NewTestOutput(func(*Message) {
|
|
atomic.AddInt64(&counter, 1)
|
|
wg.Done()
|
|
})
|
|
|
|
plugins2 := &InOutPlugins{
|
|
Inputs: []PluginReader{input2},
|
|
Outputs: []PluginWriter{output2},
|
|
}
|
|
plugins2.All = append(plugins2.All, input2, output2)
|
|
|
|
emitter2 := NewEmitter()
|
|
go emitter2.Start(plugins2, Settings.Middleware)
|
|
|
|
wg.Wait()
|
|
emitter2.Close()
|
|
}
|
|
|
|
func TestFileOutputWithNameCleaning(t *testing.T) {
|
|
output := &FileOutput{pathTemplate: "./test_requests.gor", config: &FileOutputConfig{FlushInterval: time.Minute, Append: false}}
|
|
expectedFileName := "test_requests_0.gor"
|
|
output.updateName()
|
|
|
|
if expectedFileName != output.currentName {
|
|
t.Errorf("Expected path %s but got %s", expectedFileName, output.currentName)
|
|
}
|
|
|
|
}
|
|
|
|
func TestFileOutputPathTemplate(t *testing.T) {
|
|
output := &FileOutput{pathTemplate: "/tmp/log-%Y-%m-%d-%S-%t", config: &FileOutputConfig{FlushInterval: time.Minute, Append: true}}
|
|
now := time.Now()
|
|
output.payloadType = []byte("3")
|
|
expectedPath := fmt.Sprintf("/tmp/log-%s-%s-%s-%s-3", now.Format("2006"), now.Format("01"), now.Format("02"), now.Format("05"))
|
|
path := output.filename()
|
|
|
|
if expectedPath != path {
|
|
t.Errorf("Expected path %s but got %s", expectedPath, path)
|
|
}
|
|
}
|
|
|
|
func TestFileOutputMultipleFiles(t *testing.T) {
|
|
output := NewFileOutput("/tmp/log-%Y-%m-%d-%S", &FileOutputConfig{Append: true, FlushInterval: time.Minute})
|
|
|
|
if output.file != nil {
|
|
t.Error("Should not initialize file if no writes")
|
|
}
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name1 := output.file.Name()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name2 := output.file.Name()
|
|
|
|
time.Sleep(time.Second)
|
|
output.updateName()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name3 := output.file.Name()
|
|
|
|
if name2 != name1 {
|
|
t.Error("Fast changes should happen in same file:", name1, name2, name3)
|
|
}
|
|
|
|
if name3 == name1 {
|
|
t.Error("File name should change:", name1, name2, name3)
|
|
}
|
|
|
|
os.Remove(name1)
|
|
os.Remove(name3)
|
|
}
|
|
|
|
func TestFileOutputFilePerRequest(t *testing.T) {
|
|
output := NewFileOutput("/tmp/log-%Y-%m-%d-%S-%r", &FileOutputConfig{Append: true})
|
|
|
|
if output.file != nil {
|
|
t.Error("Should not initialize file if no writes")
|
|
}
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name1 := output.file.Name()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 2 1\r\n"), Data: []byte("test")})
|
|
name2 := output.file.Name()
|
|
|
|
time.Sleep(time.Second)
|
|
output.updateName()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 3 1\r\n"), Data: []byte("test")})
|
|
name3 := output.file.Name()
|
|
|
|
if name3 == name2 || name2 == name1 || name3 == name1 {
|
|
t.Error("File name should change:", name1, name2, name3)
|
|
}
|
|
|
|
os.Remove(name1)
|
|
os.Remove(name2)
|
|
os.Remove(name3)
|
|
}
|
|
|
|
func TestFileOutputCompression(t *testing.T) {
|
|
output := NewFileOutput("/tmp/log-%Y-%m-%d-%S.gz", &FileOutputConfig{Append: true, FlushInterval: time.Minute})
|
|
|
|
if output.file != nil {
|
|
t.Error("Should not initialize file if no writes")
|
|
}
|
|
|
|
for i := 0; i < 1000; i++ {
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
}
|
|
|
|
name := output.file.Name()
|
|
output.Close()
|
|
|
|
s, _ := os.Stat(name)
|
|
if s.Size() == 12*1000 {
|
|
t.Error("Should be compressed file:", s.Size())
|
|
}
|
|
|
|
os.Remove(name)
|
|
}
|
|
|
|
func TestGetFileIndex(t *testing.T) {
|
|
var tests = []struct {
|
|
path string
|
|
index int
|
|
}{
|
|
{"/tmp/logs", -1},
|
|
{"/tmp/logs_1", 1},
|
|
{"/tmp/logs_2.gz", 2},
|
|
{"/tmp/logs_0.gz", 0},
|
|
}
|
|
|
|
for _, c := range tests {
|
|
if getFileIndex(c.path) != c.index {
|
|
t.Error(c.path, "should be", c.index, "instead", getFileIndex(c.path))
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestSetFileIndex(t *testing.T) {
|
|
var tests = []struct {
|
|
path string
|
|
index int
|
|
newPath string
|
|
}{
|
|
{"/tmp/logs", 0, "/tmp/logs_0"},
|
|
{"/tmp/logs.gz", 1, "/tmp/logs_1.gz"},
|
|
{"/tmp/logs_1", 0, "/tmp/logs_0"},
|
|
{"/tmp/logs_0", 10, "/tmp/logs_10"},
|
|
{"/tmp/logs_0.gz", 10, "/tmp/logs_10.gz"},
|
|
{"/tmp/logs_underscores.gz", 10, "/tmp/logs_underscores_10.gz"},
|
|
}
|
|
|
|
for _, c := range tests {
|
|
if setFileIndex(c.path, c.index) != c.newPath {
|
|
t.Error(c.path, "should be", c.newPath, "instead", setFileIndex(c.path, c.index))
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestFileOutputAppendQueueLimitOverflow(t *testing.T) {
|
|
rnd := rand.Int63()
|
|
name := fmt.Sprintf("/tmp/%d", rnd)
|
|
|
|
output := NewFileOutput(name, &FileOutputConfig{Append: false, FlushInterval: time.Minute, QueueLimit: 2})
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name1 := output.file.Name()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name2 := output.file.Name()
|
|
|
|
output.updateName()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name3 := output.file.Name()
|
|
|
|
if name2 != name1 || name1 != fmt.Sprintf("/tmp/%d_0", rnd) {
|
|
t.Error("Fast changes should happen in same file:", name1, name2, name3)
|
|
}
|
|
|
|
if name3 == name1 || name3 != fmt.Sprintf("/tmp/%d_1", rnd) {
|
|
t.Error("File name should change:", name1, name2, name3)
|
|
}
|
|
|
|
os.Remove(name1)
|
|
os.Remove(name3)
|
|
}
|
|
|
|
func TestFileOutputAppendQueueLimitNoOverflow(t *testing.T) {
|
|
rnd := rand.Int63()
|
|
name := fmt.Sprintf("/tmp/%d", rnd)
|
|
|
|
output := NewFileOutput(name, &FileOutputConfig{Append: false, FlushInterval: time.Minute, QueueLimit: 3})
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name1 := output.file.Name()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name2 := output.file.Name()
|
|
|
|
output.updateName()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name3 := output.file.Name()
|
|
|
|
if name2 != name1 || name1 != fmt.Sprintf("/tmp/%d_0", rnd) {
|
|
t.Error("Fast changes should happen in same file:", name1, name2, name3)
|
|
}
|
|
|
|
if name3 != name1 || name3 != fmt.Sprintf("/tmp/%d_0", rnd) {
|
|
t.Error("File name should not change:", name1, name2, name3)
|
|
}
|
|
|
|
os.Remove(name1)
|
|
os.Remove(name3)
|
|
}
|
|
|
|
func TestFileOutputAppendQueueLimitGzips(t *testing.T) {
|
|
rnd := rand.Int63()
|
|
name := fmt.Sprintf("/tmp/%d.gz", rnd)
|
|
|
|
output := NewFileOutput(name, &FileOutputConfig{Append: false, FlushInterval: time.Minute, QueueLimit: 2})
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name1 := output.file.Name()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name2 := output.file.Name()
|
|
|
|
output.updateName()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name3 := output.file.Name()
|
|
|
|
if name2 != name1 || name1 != fmt.Sprintf("/tmp/%d_0.gz", rnd) {
|
|
t.Error("Fast changes should happen in same file:", name1, name2, name3)
|
|
}
|
|
|
|
if name3 == name1 || name3 != fmt.Sprintf("/tmp/%d_1.gz", rnd) {
|
|
t.Error("File name should change:", name1, name2, name3)
|
|
}
|
|
|
|
os.Remove(name1)
|
|
os.Remove(name3)
|
|
}
|
|
|
|
func TestFileOutputSort(t *testing.T) {
|
|
var files = []string{"2016_0", "2014_10", "2015_0", "2015_10", "2015_2"}
|
|
var expected = []string{"2014_10", "2015_0", "2015_2", "2015_10", "2016_0"}
|
|
sort.Sort(sortByFileIndex(files))
|
|
|
|
if !reflect.DeepEqual(files, expected) {
|
|
t.Error("Should properly sort file names using indexes", files, expected)
|
|
}
|
|
}
|
|
|
|
func TestFileOutputAppendSizeLimitOverflow(t *testing.T) {
|
|
rnd := rand.Int63()
|
|
name := fmt.Sprintf("/tmp/%d", rnd)
|
|
|
|
message := []byte("1 1 1\r\ntest")
|
|
|
|
messageSize := len(message) + len(payloadSeparator)
|
|
|
|
output := NewFileOutput(name, &FileOutputConfig{Append: false, FlushInterval: time.Minute, SizeLimit: size.Size(2 * messageSize)})
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name1 := output.file.Name()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name2 := output.file.Name()
|
|
|
|
output.flush()
|
|
output.updateName()
|
|
|
|
output.PluginWrite(&Message{Meta: []byte("1 1 1\r\n"), Data: []byte("test")})
|
|
name3 := output.file.Name()
|
|
|
|
if name2 != name1 || name1 != fmt.Sprintf("/tmp/%d_0", rnd) {
|
|
t.Error("Fast changes should happen in same file:", name1, name2, name3)
|
|
}
|
|
|
|
if name3 == name1 || name3 != fmt.Sprintf("/tmp/%d_1", rnd) {
|
|
t.Error("File name should change:", name1, name2, name3)
|
|
}
|
|
|
|
os.Remove(name1)
|
|
os.Remove(name3)
|
|
}
|