-
Notifications
You must be signed in to change notification settings - Fork 711
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Reading pod logs returns all container logs
This is achieved by issuing an http request for each container to kubernetes' API, which yields one Reader for the corresponding container. `logReadCloser' then reads from the above readers in parallel as data is available, buffering when necessary, forwarding it to clients by implementing the io.ReadCloser interface.
- Loading branch information
Roberto Bruggemann
committed
Jan 3, 2018
1 parent
90cbd8d
commit 899b15e
Showing
5 changed files
with
261 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| package kubernetes | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "io" | ||
|
|
||
| log "github.com/Sirupsen/logrus" | ||
| ) | ||
|
|
||
| const ( | ||
| internalBufferSize = 1024 | ||
| ) | ||
|
|
||
| type logReadCloser struct { | ||
| readClosers []io.ReadCloser | ||
| eof []bool | ||
| buffer bytes.Buffer | ||
| dataChannel chan []byte | ||
| stopChannels []chan struct{} | ||
| eofChannel chan int | ||
| } | ||
|
|
||
| // NewLogReadCloser takes multiple io.ReadCloser and reads where data is available. | ||
| func NewLogReadCloser(readClosers ...io.ReadCloser) io.ReadCloser { | ||
| stopChannels := make([]chan struct{}, len(readClosers)) | ||
| for i := range readClosers { | ||
| stopChannels[i] = make(chan struct{}) | ||
| } | ||
|
|
||
| l := logReadCloser{ | ||
| readClosers: readClosers, | ||
| dataChannel: make(chan []byte), | ||
| stopChannels: stopChannels, | ||
| eofChannel: make(chan int), | ||
| eof: make([]bool, len(readClosers)), | ||
| } | ||
|
|
||
| for idx := range l.readClosers { | ||
| go l.readInput(idx) | ||
| } | ||
|
|
||
| return &l | ||
| } | ||
|
|
||
| func (l *logReadCloser) Read(p []byte) (int, error) { | ||
| if len(p) <= l.buffer.Len() { | ||
| return l.readInternalBuffer(p) | ||
| } | ||
|
|
||
| // if there's data available to read, read it, | ||
| // otherwise block | ||
| byteCount := 0 | ||
| if l.buffer.Len() > 0 { | ||
| n, err := l.readInternalBuffer(p) | ||
| if err != nil { | ||
| return n, err | ||
| } | ||
| byteCount += n | ||
| } else { | ||
| // block on read or EOF | ||
| received := false | ||
| for !received && !l.isEOF() { | ||
| select { | ||
| case data := <-l.dataChannel: | ||
| l.buffer.Write(data) | ||
| received = true | ||
| case idx := <-l.eofChannel: | ||
| l.eof[idx] = true | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // check if there's more data to read, without blocking | ||
| empty := false | ||
| for !empty && l.buffer.Len() < len(p) { | ||
| select { | ||
| case data := <-l.dataChannel: | ||
| l.buffer.Write(data) | ||
| case idx := <-l.eofChannel: | ||
| l.eof[idx] = true | ||
| default: | ||
| empty = true | ||
| } | ||
| } | ||
|
|
||
| return l.readInternalBuffer(p[byteCount:]) | ||
| } | ||
|
|
||
| func (l *logReadCloser) Close() error { | ||
| for i, rc := range l.readClosers { | ||
| err := rc.Close() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // synchronous stop: | ||
| // the routines write to dataChannel which will be closed by this thread | ||
| select { | ||
| case <-l.stopChannels[i]: | ||
| break | ||
| } | ||
| close(l.stopChannels[i]) | ||
| } | ||
|
|
||
| close(l.dataChannel) | ||
| close(l.eofChannel) | ||
| return nil | ||
| } | ||
|
|
||
| func (l *logReadCloser) readInternalBuffer(p []byte) (int, error) { | ||
| n, err := l.buffer.Read(p) | ||
| if err == io.EOF && !l.isEOF() { | ||
| return n, nil | ||
| } | ||
|
|
||
| return n, err | ||
| } | ||
|
|
||
| func (l *logReadCloser) readInput(idx int) { | ||
| tmpBuffer := make([]byte, internalBufferSize) | ||
| for { | ||
| n, err := l.readClosers[idx].Read(tmpBuffer) | ||
| if err == io.EOF { | ||
| if n > 0 { | ||
| l.dataChannel <- tmpBuffer[:n] | ||
| } | ||
| l.eofChannel <- idx | ||
| break | ||
| } | ||
| if err != nil { | ||
| log.Errorf("Failed to read: %v", err) | ||
| break | ||
| } | ||
| l.dataChannel <- tmpBuffer[:n] | ||
| } | ||
|
|
||
| // signal the routine won't write to dataChannel | ||
| l.stopChannels[idx] <- struct{}{} | ||
| } | ||
|
|
||
| func (l *logReadCloser) isEOF() bool { | ||
| for _, e := range l.eof { | ||
| if !e { | ||
| return false | ||
| } | ||
| } | ||
| return true | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| package kubernetes_test | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "io" | ||
| "io/ioutil" | ||
| "testing" | ||
|
|
||
| "github.com/weaveworks/scope/probe/kubernetes" | ||
| ) | ||
|
|
||
| func TestLogReadCloser(t *testing.T) { | ||
| s0 := []byte("abcdefghijklmnopqrstuvwxyz") | ||
| s1 := []byte("ABCDEFGHIJKLMNOPQRSTUVWXYZ") | ||
| s2 := []byte("0123456789012345") | ||
|
|
||
| r0 := ioutil.NopCloser(bytes.NewReader(s0)) | ||
| r1 := ioutil.NopCloser(bytes.NewReader(s1)) | ||
| r2 := ioutil.NopCloser(bytes.NewReader(s2)) | ||
|
|
||
| l := kubernetes.NewLogReadCloser(r0, r1, r2) | ||
|
|
||
| buf := make([]byte, 3000) | ||
| count := 0 | ||
| for { | ||
| n, err := l.Read(buf[count:]) | ||
| if err == io.EOF { | ||
| break | ||
| } | ||
| if err != nil { | ||
| t.Error(err) | ||
| } | ||
| count += n | ||
| } | ||
|
|
||
| total := len(s0) + len(s1) + len(s2) | ||
| if count != total { | ||
| t.Errorf("Must read %v characters, but got %v", total, count) | ||
| } | ||
|
|
||
| // check every byte | ||
| byteCounter := map[byte]int{} | ||
| byteCount(byteCounter, s0) | ||
| byteCount(byteCounter, s1) | ||
| byteCount(byteCounter, s2) | ||
|
|
||
| for i := 0; i < count; i++ { | ||
| b := buf[i] | ||
| v, ok := byteCounter[b] | ||
| if ok { | ||
| v-- | ||
| byteCounter[b] = v | ||
| } | ||
| } | ||
|
|
||
| for b, c := range byteCounter { | ||
| if c != 0 { | ||
| t.Errorf("%v should be 0 instead of %v", b, c) | ||
| } | ||
| } | ||
|
|
||
| err := l.Close() | ||
| if err != nil { | ||
| t.Errorf("Close must not return an error: %v", err) | ||
| } | ||
| } | ||
|
|
||
| func byteCount(accumulator map[byte]int, s []byte) { | ||
| for _, b := range s { | ||
| v, ok := accumulator[b] | ||
| if !ok { | ||
| v = 0 | ||
| } | ||
| v++ | ||
| accumulator[b] = v | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters