output_file_test.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)
}