mirror of
https://github.com/buger/goreplay.git
synced 2024-04-21 12:32:02 +00:00
Dima GolomozyandGitHub
40946831b9
change package from `main -> goreplay` this will allow importing `goreplay` as a package
332 lines
8.9 KiB
Go
332 lines
8.9 KiB
Go
package goreplay
|
|
|
|
import (
|
|
"fmt"
|
|
"github.com/buger/goreplay/internal/size"
|
|
"math/rand"
|
|
"os"
|
|
"reflect"
|
|
"sort"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
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.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)
|
|
}
|