Files
nomad/command/agent/fs_endpoint_test.go
2016-07-25 11:16:01 -07:00

952 lines
21 KiB
Go

package agent
import (
"bytes"
"io"
"io/ioutil"
"math"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"reflect"
"strconv"
"strings"
"testing"
"time"
"github.com/hashicorp/nomad/client/allocdir"
"github.com/hashicorp/nomad/testutil"
"github.com/pmezard/go-difflib/difflib"
"github.com/ugorji/go/codec"
)
func TestAllocDirFS_List_MissingParams(t *testing.T) {
httpTest(t, nil, func(s *TestServer) {
req, err := http.NewRequest("GET", "/v1/client/fs/ls/", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW := httptest.NewRecorder()
_, err = s.Server.DirectoryListRequest(respW, req)
if err != allocIDNotPresentErr {
t.Fatalf("expected err: %v, actual: %v", allocIDNotPresentErr, err)
}
})
}
func TestAllocDirFS_Stat_MissingParams(t *testing.T) {
httpTest(t, nil, func(s *TestServer) {
req, err := http.NewRequest("GET", "/v1/client/fs/stat/", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW := httptest.NewRecorder()
_, err = s.Server.FileStatRequest(respW, req)
if err != allocIDNotPresentErr {
t.Fatalf("expected err: %v, actual: %v", allocIDNotPresentErr, err)
}
req, err = http.NewRequest("GET", "/v1/client/fs/stat/foo", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW = httptest.NewRecorder()
_, err = s.Server.FileStatRequest(respW, req)
if err != fileNameNotPresentErr {
t.Fatalf("expected err: %v, actual: %v", allocIDNotPresentErr, err)
}
})
}
func TestAllocDirFS_ReadAt_MissingParams(t *testing.T) {
httpTest(t, nil, func(s *TestServer) {
req, err := http.NewRequest("GET", "/v1/client/fs/readat/", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW := httptest.NewRecorder()
_, err = s.Server.FileReadAtRequest(respW, req)
if err == nil {
t.Fatal("expected error")
}
req, err = http.NewRequest("GET", "/v1/client/fs/readat/foo", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW = httptest.NewRecorder()
_, err = s.Server.FileReadAtRequest(respW, req)
if err == nil {
t.Fatal("expected error")
}
req, err = http.NewRequest("GET", "/v1/client/fs/readat/foo?path=/path/to/file", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW = httptest.NewRecorder()
_, err = s.Server.FileReadAtRequest(respW, req)
if err == nil {
t.Fatal("expected error")
}
})
}
type WriteCloseChecker struct {
io.WriteCloser
Closed bool
}
func (w *WriteCloseChecker) Close() error {
w.Closed = true
return w.WriteCloser.Close()
}
// This test checks, that even if the frame size has not been hit, a flush will
// periodically occur.
func TestStreamFramer_Flush(t *testing.T) {
// Create the stream framer
r, w := io.Pipe()
wrappedW := &WriteCloseChecker{WriteCloser: w}
hRate, bWindow := 100*time.Millisecond, 100*time.Millisecond
sf := NewStreamFramer(wrappedW, hRate, bWindow, 100)
sf.Run()
// Create a decoder
dec := codec.NewDecoder(r, jsonHandle)
f := "foo"
fe := "bar"
d := []byte{0xa}
o := int64(10)
// Start the reader
resultCh := make(chan struct{})
go func() {
for {
var frame StreamFrame
if err := dec.Decode(&frame); err != nil {
t.Fatalf("failed to decode")
}
if frame.IsHeartbeat() {
continue
}
if reflect.DeepEqual(frame.Data, d) && frame.Offset == o && frame.File == f && frame.FileEvent == fe {
resultCh <- struct{}{}
return
}
}
}()
// Write only 1 byte so we do not hit the frame size
if err := sf.Send(f, fe, d, o); err != nil {
t.Fatalf("Send() failed %v", err)
}
select {
case <-resultCh:
case <-time.After(2 * bWindow):
t.Fatalf("failed to flush")
}
// Close the reader and wait. This should cause the runner to exit
if err := r.Close(); err != nil {
t.Fatalf("failed to close reader")
}
select {
case <-sf.ExitCh():
case <-time.After(2 * hRate):
t.Fatalf("exit channel should close")
}
sf.Destroy()
if !wrappedW.Closed {
t.Fatalf("writer not closed")
}
}
// This test checks that frames will be batched till the frame size is hit (in
// the case that is before the flush).
func TestStreamFramer_Batch(t *testing.T) {
// Create the stream framer
r, w := io.Pipe()
wrappedW := &WriteCloseChecker{WriteCloser: w}
// Ensure the batch window doesn't get hit
hRate, bWindow := 100*time.Millisecond, 500*time.Millisecond
sf := NewStreamFramer(wrappedW, hRate, bWindow, 3)
sf.Run()
// Create a decoder
dec := codec.NewDecoder(r, jsonHandle)
f := "foo"
fe := "bar"
d := []byte{0xa, 0xb, 0xc}
o := int64(10)
// Start the reader
resultCh := make(chan struct{})
go func() {
for {
var frame StreamFrame
if err := dec.Decode(&frame); err != nil {
t.Fatalf("failed to decode")
}
if frame.IsHeartbeat() {
continue
}
if reflect.DeepEqual(frame.Data, d) && frame.Offset == o && frame.File == f && frame.FileEvent == fe {
resultCh <- struct{}{}
return
}
}
}()
// Write only 1 byte so we do not hit the frame size
if err := sf.Send(f, fe, d[:1], o); err != nil {
t.Fatalf("Send() failed %v", err)
}
// Ensure we didn't get any data
select {
case <-resultCh:
t.Fatalf("Got data before frame size reached")
case <-time.After(bWindow / 2):
}
// Write the rest so we hit the frame size
if err := sf.Send(f, fe, d[1:], o); err != nil {
t.Fatalf("Send() failed %v", err)
}
// Ensure we get data
select {
case <-resultCh:
case <-time.After(2 * bWindow):
t.Fatalf("Did not receive data after batch size reached")
}
// Close the reader and wait. This should cause the runner to exit
if err := r.Close(); err != nil {
t.Fatalf("failed to close reader")
}
select {
case <-sf.ExitCh():
case <-time.After(2 * hRate):
t.Fatalf("exit channel should close")
}
sf.Destroy()
if !wrappedW.Closed {
t.Fatalf("writer not closed")
}
}
func TestStreamFramer_Heartbeat(t *testing.T) {
// Create the stream framer
r, w := io.Pipe()
wrappedW := &WriteCloseChecker{WriteCloser: w}
hRate, bWindow := 100*time.Millisecond, 100*time.Millisecond
sf := NewStreamFramer(wrappedW, hRate, bWindow, 100)
sf.Run()
// Create a decoder
dec := codec.NewDecoder(r, jsonHandle)
// Start the reader
resultCh := make(chan struct{})
go func() {
for {
var frame StreamFrame
if err := dec.Decode(&frame); err != nil {
t.Fatalf("failed to decode")
}
if frame.IsHeartbeat() {
resultCh <- struct{}{}
return
}
}
}()
select {
case <-resultCh:
case <-time.After(2 * hRate):
t.Fatalf("failed to heartbeat")
}
// Close the reader and wait. This should cause the runner to exit
if err := r.Close(); err != nil {
t.Fatalf("failed to close reader")
}
select {
case <-sf.ExitCh():
case <-time.After(2 * hRate):
t.Fatalf("exit channel should close")
}
sf.Destroy()
if !wrappedW.Closed {
t.Fatalf("writer not closed")
}
}
// This test checks that frames are received in order
func TestStreamFramer_Order(t *testing.T) {
// Create the stream framer
r, w := io.Pipe()
wrappedW := &WriteCloseChecker{WriteCloser: w}
// Ensure the batch window doesn't get hit
hRate, bWindow := 100*time.Millisecond, 10*time.Millisecond
sf := NewStreamFramer(wrappedW, hRate, bWindow, 10)
sf.Run()
// Create a decoder
dec := codec.NewDecoder(r, jsonHandle)
files := []string{"1", "2", "3", "4", "5"}
input := bytes.NewBuffer(make([]byte, 0, 100000))
for i := 0; i <= 1000; i++ {
str := strconv.Itoa(i) + ","
input.WriteString(str)
}
expected := bytes.NewBuffer(make([]byte, 0, 100000))
for _, _ = range files {
expected.Write(input.Bytes())
}
receivedBuf := bytes.NewBuffer(make([]byte, 0, 100000))
// Start the reader
resultCh := make(chan struct{})
go func() {
for {
var frame StreamFrame
if err := dec.Decode(&frame); err != nil {
t.Fatalf("failed to decode")
}
if frame.IsHeartbeat() {
continue
}
receivedBuf.Write(frame.Data)
if reflect.DeepEqual(expected, receivedBuf) {
resultCh <- struct{}{}
return
}
}
}()
// Send the data
b := input.Bytes()
shards := 10
each := len(b) / shards
for _, f := range files {
for i := 0; i < shards; i++ {
l, r := each*i, each*(i+1)
if i == shards-1 {
r = len(b)
}
if err := sf.Send(f, "", b[l:r], 0); err != nil {
t.Fatalf("Send() failed %v", err)
}
}
}
// Ensure we get data
select {
case <-resultCh:
case <-time.After(10 * bWindow):
got := receivedBuf.String()
want := expected.String()
diff := difflib.ContextDiff{
A: difflib.SplitLines(strings.Replace(got, ",", "\n", -1)),
B: difflib.SplitLines(strings.Replace(want, ",", "\n", -1)),
FromFile: "Got",
ToFile: "Want",
Context: 3,
Eol: "\n",
}
result, _ := difflib.GetContextDiffString(diff)
t.Fatalf(strings.Replace(result, "\t", " ", -1))
}
// Close the reader and wait. This should cause the runner to exit
if err := r.Close(); err != nil {
t.Fatalf("failed to close reader")
}
select {
case <-sf.ExitCh():
case <-time.After(2 * hRate):
t.Fatalf("exit channel should close")
}
sf.Destroy()
if !wrappedW.Closed {
t.Fatalf("writer not closed")
}
}
func TestHTTP_Stream_MissingParams(t *testing.T) {
httpTest(t, nil, func(s *TestServer) {
req, err := http.NewRequest("GET", "/v1/client/fs/stream/", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW := httptest.NewRecorder()
_, err = s.Server.Stream(respW, req)
if err == nil {
t.Fatal("expected error")
}
req, err = http.NewRequest("GET", "/v1/client/fs/stream/foo", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW = httptest.NewRecorder()
_, err = s.Server.Stream(respW, req)
if err == nil {
t.Fatal("expected error")
}
req, err = http.NewRequest("GET", "/v1/client/fs/stream/foo?path=/path/to/file", nil)
if err != nil {
t.Fatalf("err: %v", err)
}
respW = httptest.NewRecorder()
_, err = s.Server.Stream(respW, req)
if err == nil {
t.Fatal("expected error")
}
})
}
// tempAllocDir returns a new alloc dir that is rooted in a temp dir. The caller
// should destroy the temp dir.
func tempAllocDir(t *testing.T) *allocdir.AllocDir {
dir, err := ioutil.TempDir("", "")
if err != nil {
t.Fatalf("TempDir() failed: %v", err)
}
return allocdir.NewAllocDir(dir)
}
type nopWriteCloser struct {
io.Writer
}
func (n nopWriteCloser) Close() error {
return nil
}
func TestHTTP_Stream_NoFile(t *testing.T) {
httpTest(t, nil, func(s *TestServer) {
// Get a temp alloc dir
ad := tempAllocDir(t)
defer os.RemoveAll(ad.AllocDir)
framer := NewStreamFramer(nopWriteCloser{ioutil.Discard}, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
defer framer.Destroy()
if err := s.Server.stream(0, "foo", ad, framer, nil); err == nil {
t.Fatalf("expected an error when streaming unknown file")
}
})
}
func TestHTTP_Stream_Modify(t *testing.T) {
httpTest(t, nil, func(s *TestServer) {
// Get a temp alloc dir
ad := tempAllocDir(t)
defer os.RemoveAll(ad.AllocDir)
// Create a file in the temp dir
streamFile := "stream_file"
f, err := os.Create(filepath.Join(ad.AllocDir, streamFile))
if err != nil {
t.Fatalf("Failed to create file: %v", err)
}
defer f.Close()
// Create a decoder
r, w := io.Pipe()
defer r.Close()
defer w.Close()
dec := codec.NewDecoder(r, jsonHandle)
data := []byte("helloworld")
// Start the reader
resultCh := make(chan struct{})
go func() {
var collected []byte
for {
var frame StreamFrame
if err := dec.Decode(&frame); err != nil {
t.Fatalf("failed to decode: %v", err)
}
if frame.IsHeartbeat() {
continue
}
collected = append(collected, frame.Data...)
if reflect.DeepEqual(data, collected) {
resultCh <- struct{}{}
return
}
}
}()
// Write a few bytes
if _, err := f.Write(data[:3]); err != nil {
t.Fatalf("write failed: %v", err)
}
framer := NewStreamFramer(w, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
defer framer.Destroy()
// Start streaming
go func() {
if err := s.Server.stream(0, streamFile, ad, framer, nil); err != nil {
t.Fatalf("stream() failed: %v", err)
}
}()
// Sleep a little before writing more. This lets us check if the watch
// is working.
time.Sleep(1 * time.Second)
if _, err := f.Write(data[3:]); err != nil {
t.Fatalf("write failed: %v", err)
}
select {
case <-resultCh:
case <-time.After(2 * streamBatchWindow):
t.Fatalf("failed to send new data")
}
})
}
func TestHTTP_Stream_Truncate(t *testing.T) {
httpTest(t, nil, func(s *TestServer) {
// Get a temp alloc dir
ad := tempAllocDir(t)
defer os.RemoveAll(ad.AllocDir)
// Create a file in the temp dir
streamFile := "stream_file"
streamFilePath := filepath.Join(ad.AllocDir, streamFile)
f, err := os.Create(streamFilePath)
if err != nil {
t.Fatalf("Failed to create file: %v", err)
}
defer f.Close()
// Create a decoder
r, w := io.Pipe()
defer r.Close()
defer w.Close()
dec := codec.NewDecoder(r, jsonHandle)
data := []byte("helloworld")
// Start the reader
truncateCh := make(chan struct{})
dataPostTruncCh := make(chan struct{})
go func() {
var collected []byte
for {
var frame StreamFrame
if err := dec.Decode(&frame); err != nil {
t.Fatalf("failed to decode: %v", err)
}
if frame.IsHeartbeat() {
continue
}
if frame.FileEvent == truncateEvent {
close(truncateCh)
}
collected = append(collected, frame.Data...)
if reflect.DeepEqual(data, collected) {
close(dataPostTruncCh)
return
}
}
}()
// Write a few bytes
if _, err := f.Write(data[:3]); err != nil {
t.Fatalf("write failed: %v", err)
}
framer := NewStreamFramer(w, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
defer framer.Destroy()
// Start streaming
go func() {
if err := s.Server.stream(0, streamFile, ad, framer, nil); err != nil {
t.Fatalf("stream() failed: %v", err)
}
}()
// Sleep a little before truncating. This lets us check if the watch
// is working.
time.Sleep(1 * time.Second)
if err := f.Truncate(0); err != nil {
t.Fatalf("truncate failed: %v", err)
}
if err := f.Sync(); err != nil {
t.Fatalf("sync failed: %v", err)
}
if err := f.Close(); err != nil {
t.Fatalf("failed to close file: %v", err)
}
f2, err := os.OpenFile(streamFilePath, os.O_RDWR, 0)
if err != nil {
t.Fatalf("failed to reopen file: %v", err)
}
defer f2.Close()
if _, err := f2.Write(data[3:5]); err != nil {
t.Fatalf("write failed: %v", err)
}
select {
case <-truncateCh:
case <-time.After(2 * streamBatchWindow):
t.Fatalf("did not receive truncate")
}
// Sleep a little before writing more. This lets us check if the watch
// is working.
time.Sleep(1 * time.Second)
if _, err := f2.Write(data[5:]); err != nil {
t.Fatalf("write failed: %v", err)
}
select {
case <-dataPostTruncCh:
case <-time.After(2 * streamBatchWindow):
t.Fatalf("did not receive post truncate data")
}
})
}
func TestHTTP_Stream_Delete(t *testing.T) {
httpTest(t, nil, func(s *TestServer) {
// Get a temp alloc dir
ad := tempAllocDir(t)
defer os.RemoveAll(ad.AllocDir)
// Create a file in the temp dir
streamFile := "stream_file"
streamFilePath := filepath.Join(ad.AllocDir, streamFile)
f, err := os.Create(streamFilePath)
if err != nil {
t.Fatalf("Failed to create file: %v", err)
}
defer f.Close()
// Create a decoder
r, w := io.Pipe()
wrappedW := &WriteCloseChecker{WriteCloser: w}
defer r.Close()
defer w.Close()
dec := codec.NewDecoder(r, jsonHandle)
data := []byte("helloworld")
// Start the reader
deleteCh := make(chan struct{})
go func() {
for {
var frame StreamFrame
if err := dec.Decode(&frame); err != nil {
t.Fatalf("failed to decode: %v", err)
}
if frame.IsHeartbeat() {
continue
}
if frame.FileEvent == deleteEvent {
close(deleteCh)
return
}
}
}()
// Write a few bytes
if _, err := f.Write(data[:3]); err != nil {
t.Fatalf("write failed: %v", err)
}
framer := NewStreamFramer(wrappedW, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
// Start streaming
go func() {
if err := s.Server.stream(0, streamFile, ad, framer, nil); err != nil {
t.Fatalf("stream() failed: %v", err)
}
}()
// Sleep a little before deleting. This lets us check if the watch
// is working.
time.Sleep(1 * time.Second)
if err := os.Remove(streamFilePath); err != nil {
t.Fatalf("delete failed: %v", err)
}
select {
case <-deleteCh:
case <-time.After(4 * streamBatchWindow):
t.Fatalf("did not receive delete")
}
framer.Destroy()
testutil.WaitForResult(func() (bool, error) {
return wrappedW.Closed, nil
}, func(err error) {
t.Fatalf("connection not closed")
})
})
}
func TestLogs_findClosest(t *testing.T) {
task := "foo"
entries := []*allocdir.AllocFileInfo{
{
Name: "foo.stdout.0",
Size: 100,
},
{
Name: "foo.stdout.1",
Size: 100,
},
{
Name: "foo.stdout.2",
Size: 100,
},
{
Name: "foo.stdout.3",
Size: 100,
},
{
Name: "foo.stderr.0",
Size: 100,
},
{
Name: "foo.stderr.1",
Size: 100,
},
{
Name: "foo.stderr.2",
Size: 100,
},
}
cases := []struct {
Entries []*allocdir.AllocFileInfo
DesiredIdx int64
DesiredOffset int64
Task string
LogType string
ExpectedFile string
ExpectedIdx int64
ExpectedOffset int64
Error bool
}{
// Test error cases
{
Entries: nil,
DesiredIdx: 0,
Task: task,
LogType: "stdout",
Error: true,
},
{
Entries: entries[0:3],
DesiredIdx: 0,
Task: task,
LogType: "stderr",
Error: true,
},
// Test begining cases
{
Entries: entries,
DesiredIdx: 0,
Task: task,
LogType: "stdout",
ExpectedFile: entries[0].Name,
ExpectedIdx: 0,
},
{
// Desired offset should be ignored at edges
Entries: entries,
DesiredIdx: 0,
DesiredOffset: -100,
Task: task,
LogType: "stdout",
ExpectedFile: entries[0].Name,
ExpectedIdx: 0,
ExpectedOffset: 0,
},
{
// Desired offset should be ignored at edges
Entries: entries,
DesiredIdx: 1,
DesiredOffset: -1000,
Task: task,
LogType: "stdout",
ExpectedFile: entries[0].Name,
ExpectedIdx: 0,
ExpectedOffset: 0,
},
{
Entries: entries,
DesiredIdx: 0,
Task: task,
LogType: "stderr",
ExpectedFile: entries[4].Name,
ExpectedIdx: 0,
},
{
Entries: entries,
DesiredIdx: 0,
Task: task,
LogType: "stdout",
ExpectedFile: entries[0].Name,
ExpectedIdx: 0,
},
// Test middle cases
{
Entries: entries,
DesiredIdx: 1,
Task: task,
LogType: "stdout",
ExpectedFile: entries[1].Name,
ExpectedIdx: 1,
},
{
Entries: entries,
DesiredIdx: 1,
DesiredOffset: 10,
Task: task,
LogType: "stdout",
ExpectedFile: entries[1].Name,
ExpectedIdx: 1,
ExpectedOffset: 10,
},
{
Entries: entries,
DesiredIdx: 1,
DesiredOffset: 110,
Task: task,
LogType: "stdout",
ExpectedFile: entries[2].Name,
ExpectedIdx: 2,
ExpectedOffset: 10,
},
{
Entries: entries,
DesiredIdx: 1,
Task: task,
LogType: "stderr",
ExpectedFile: entries[5].Name,
ExpectedIdx: 1,
},
// Test end cases
{
Entries: entries,
DesiredIdx: math.MaxInt64,
Task: task,
LogType: "stdout",
ExpectedFile: entries[3].Name,
ExpectedIdx: 3,
},
{
Entries: entries,
DesiredIdx: math.MaxInt64,
DesiredOffset: math.MaxInt64,
Task: task,
LogType: "stdout",
ExpectedFile: entries[3].Name,
ExpectedIdx: 3,
ExpectedOffset: 100,
},
{
Entries: entries,
DesiredIdx: math.MaxInt64,
DesiredOffset: -10,
Task: task,
LogType: "stdout",
ExpectedFile: entries[3].Name,
ExpectedIdx: 3,
ExpectedOffset: 90,
},
{
Entries: entries,
DesiredIdx: math.MaxInt64,
Task: task,
LogType: "stderr",
ExpectedFile: entries[6].Name,
ExpectedIdx: 2,
},
}
for i, c := range cases {
entry, idx, offset, err := findClosest(c.Entries, c.DesiredIdx, c.DesiredOffset, c.Task, c.LogType)
if err != nil {
if !c.Error {
t.Fatalf("case %d: Unexpected error: %v", i, err)
}
continue
}
if entry.Name != c.ExpectedFile {
t.Fatalf("case %d: Got file %q; want %q", i, entry.Name, c.ExpectedFile)
}
if idx != c.ExpectedIdx {
t.Fatalf("case %d: Got index %d; want %d", i, idx, c.ExpectedIdx)
}
if offset != c.ExpectedOffset {
t.Fatalf("case %d: Got offset %d; want %d", i, offset, c.ExpectedOffset)
}
}
}