luxolus
/
open-nomad
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

Merge pull request #2235 from hashicorp/f-http-logs 

Logs via a browser
This commit is contained in:
Alex Dadgar 2017-01-23 16:59:37 -08:00 committed by GitHub
parent 4e5de453c7 c61f8d2526
commit a9aa992f0c
3 changed files with 141 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
command/agent/fs_endpoint.go
View File

@ -255,6 +255,9 @@ func (s *StreamFrame) IsCleared() bool {
// StreamFramer is used to buffer and send frames as well as heartbeat.
type StreamFramer struct {
// plainTxt determines whether we frame or just send plain text data.
plainTxt bool
out io.WriteCloser
enc *codec.Encoder
encLock sync.Mutex
@ -281,8 +284,11 @@ type StreamFramer struct {
}
// NewStreamFramer creates a new stream framer that will output StreamFrames to
// the passed output.
func NewStreamFramer(out io.WriteCloser, heartbeatRate, batchWindow time.Duration, frameSize int) *StreamFramer {
// the passed output. If plainTxt is set we do not frame and just batch plain
// text data.
func NewStreamFramer(out io.WriteCloser, plainTxt bool,
heartbeatRate, batchWindow time.Duration, frameSize int) *StreamFramer {
// Create a JSON encoder
enc := codec.NewEncoder(out, jsonHandle)
@ -291,6 +297,7 @@ func NewStreamFramer(out io.WriteCloser, heartbeatRate, batchWindow time.Duratio
flusher := time.NewTicker(batchWindow)
return &StreamFramer{
plainTxt: plainTxt,
out: out,
enc: enc,
frameSize: frameSize,
@ -390,6 +397,10 @@ OUTER:
func (s *StreamFramer) send(f *StreamFrame) error {
s.encLock.Lock()
defer s.encLock.Unlock()
if s.plainTxt {
_, err := io.Copy(s.out, bytes.NewReader(f.Data))
return err
}
return s.enc.Encode(f)
}
@ -549,7 +560,7 @@ func (s *HTTPServer) Stream(resp http.ResponseWriter, req *http.Request) (interf
output := ioutils.NewWriteFlusher(resp)
// Create the framer
framer := NewStreamFramer(output, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer := NewStreamFramer(output, false, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
defer framer.Destroy()
@ -697,7 +708,7 @@ OUTER:
// applied. Defaults to "start".
func (s *HTTPServer) Logs(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
var allocID, task, logType string
var follow bool
var plain, follow bool
var err error
q := req.URL.Query()
@ -710,8 +721,16 @@ func (s *HTTPServer) Logs(resp http.ResponseWriter, req *http.Request) (interfac
return nil, taskNotPresentErr
}
if follow, err = strconv.ParseBool(q.Get("follow")); err != nil {
return nil, fmt.Errorf("Failed to parse follow field to boolean: %v", err)
if followStr := q.Get("follow"); followStr != "" {
if follow, err = strconv.ParseBool(followStr); err != nil {
return nil, fmt.Errorf("Failed to parse follow field to boolean: %v", err)
}
}
if plainStr := q.Get("plain"); plainStr != "" {
if plain, err = strconv.ParseBool(plainStr); err != nil {
return nil, fmt.Errorf("Failed to parse plain field to boolean: %v", err)
}
}
logType = q.Get("type")
@ -747,15 +766,15 @@ func (s *HTTPServer) Logs(resp http.ResponseWriter, req *http.Request) (interfac
// Create an output that gets flushed on every write
output := ioutils.NewWriteFlusher(resp)
return nil, s.logs(follow, offset, origin, task, logType, fs, output)
return nil, s.logs(follow, plain, offset, origin, task, logType, fs, output)
}
func (s *HTTPServer) logs(follow bool, offset int64,
func (s *HTTPServer) logs(follow, plain bool, offset int64,
origin, task, logType string,
fs allocdir.AllocDirFS, output io.WriteCloser) error {
// Create the framer
framer := NewStreamFramer(output, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer := NewStreamFramer(output, plain, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
defer framer.Destroy()

119
command/agent/fs_endpoint_test.go
View File

@ -14,6 +14,7 @@ import (
"reflect"
"runtime"
"strconv"
"strings"
"testing"
"time"
@ -118,7 +119,7 @@ func TestStreamFramer_Flush(t *testing.T) {
r, w := io.Pipe()
wrappedW := &WriteCloseChecker{WriteCloser: w}
hRate, bWindow := 100*time.Millisecond, 100*time.Millisecond
sf := NewStreamFramer(wrappedW, hRate, bWindow, 100)
sf := NewStreamFramer(wrappedW, false, hRate, bWindow, 100)
sf.Run()
// Create a decoder
@ -186,7 +187,7 @@ func TestStreamFramer_Batch(t *testing.T) {
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 := NewStreamFramer(wrappedW, false, hRate, bWindow, 3)
sf.Run()
// Create a decoder
@ -263,7 +264,7 @@ func TestStreamFramer_Heartbeat(t *testing.T) {
r, w := io.Pipe()
wrappedW := &WriteCloseChecker{WriteCloser: w}
hRate, bWindow := 100*time.Millisecond, 100*time.Millisecond
sf := NewStreamFramer(wrappedW, hRate, bWindow, 100)
sf := NewStreamFramer(wrappedW, false, hRate, bWindow, 100)
sf.Run()
// Create a decoder
@ -315,7 +316,7 @@ func TestStreamFramer_Order(t *testing.T) {
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 := NewStreamFramer(wrappedW, false, hRate, bWindow, 10)
sf.Run()
// Create a decoder
@ -401,6 +402,102 @@ func TestStreamFramer_Order(t *testing.T) {
}
}
// This test checks that frames are received in order
func TestStreamFramer_Order_PlainText(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, true, hRate, bWindow, 10)
sf.Run()
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() {
OUTER:
for {
if _, err := receivedBuf.ReadFrom(r); err != nil {
if strings.Contains(err.Error(), "closed pipe") {
resultCh <- struct{}{}
return
}
t.Fatalf("bad read: %v", err)
}
if expected.Len() != receivedBuf.Len() {
continue
}
expectedBytes := expected.Bytes()
actualBytes := receivedBuf.Bytes()
for i, e := range expectedBytes {
if a := actualBytes[i]; a != e {
continue OUTER
}
}
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 * time.Duration(testutil.TestMultiplier()) * bWindow):
if expected.Len() != receivedBuf.Len() {
t.Fatalf("Got %v; want %v", expected.Len(), receivedBuf.Len())
}
expectedBytes := expected.Bytes()
actualBytes := receivedBuf.Bytes()
for i, e := range expectedBytes {
if a := actualBytes[i]; a != e {
t.Fatalf("Index %d; Got %q; want %q", i, a, e)
}
}
}
// Close the reader and wait. This should cause the runner to exit
if err := r.Close(); err != nil {
t.Fatalf("failed to close reader")
}
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)
@ -467,7 +564,7 @@ func TestHTTP_Stream_NoFile(t *testing.T) {
ad := tempAllocDir(t)
defer os.RemoveAll(ad.AllocDir)
framer := NewStreamFramer(nopWriteCloser{ioutil.Discard}, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer := NewStreamFramer(nopWriteCloser{ioutil.Discard}, false, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
defer framer.Destroy()
@ -526,7 +623,7 @@ func TestHTTP_Stream_Modify(t *testing.T) {
t.Fatalf("write failed: %v", err)
}
framer := NewStreamFramer(w, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer := NewStreamFramer(w, false, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
defer framer.Destroy()
@ -607,7 +704,7 @@ func TestHTTP_Stream_Truncate(t *testing.T) {
t.Fatalf("write failed: %v", err)
}
framer := NewStreamFramer(w, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer := NewStreamFramer(w, false, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
defer framer.Destroy()
@ -710,7 +807,7 @@ func TestHTTP_Stream_Delete(t *testing.T) {
t.Fatalf("write failed: %v", err)
}
framer := NewStreamFramer(wrappedW, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer := NewStreamFramer(wrappedW, false, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
framer.Run()
// Start streaming
@ -804,7 +901,7 @@ func TestHTTP_Logs_NoFollow(t *testing.T) {
// Start streaming logs
go func() {
if err := s.Server.logs(false, 0, OriginStart, task, logType, ad, wrappedW); err != nil {
if err := s.Server.logs(false, false, 0, OriginStart, task, logType, ad, wrappedW); err != nil {
t.Fatalf("logs() failed: %v", err)
}
}()
@ -893,7 +990,7 @@ func TestHTTP_Logs_Follow(t *testing.T) {
// Start streaming logs
go func() {
if err := s.Server.logs(true, 0, OriginStart, task, logType, ad, wrappedW); err != nil {
if err := s.Server.logs(true, false, 0, OriginStart, task, logType, ad, wrappedW); err != nil {
t.Fatalf("logs() failed: %v", err)
}
}()
@ -1006,7 +1103,7 @@ func BenchmarkHTTP_Logs_Follow(t *testing.B) {
// Start streaming logs
go func() {
if err := s.Server.logs(true, 0, OriginStart, task, logType, ad, wrappedW); err != nil {
if err := s.Server.logs(true, false, 0, OriginStart, task, logType, ad, wrappedW); err != nil {
t.Fatalf("logs() failed: %v", err)
}
}()

5
website/source/docs/http/client-fs.html.md
View File

@ -227,6 +227,11 @@ allocation was placed.
Origin can be either "start" or "end" and applies the offset relative to
either the start or end of the logs respectively. Defaults to "start".
</li>
<li>
<span class="param">plain</span>
A boolean of whether to return just the plain text without framing.
This can be usef when viewing logs in a browser.
</li>
</ul>
</dd>