add log file tailing and logrotate support

utils/oomparser/oomparser.go

@@ -18,7 +18,6 @@ import (
 	"bufio"
 	"fmt"
 	"io"
-	"os"
 	"os/exec"
 	"path"
 	"regexp"
@@ -26,6 +25,7 @@ import (
 	"time"
 
 	"github.com/google/cadvisor/utils"
+	"github.com/google/cadvisor/utils/tail"
 
 	"github.com/golang/glog"
 )
@@ -105,27 +105,29 @@ func checkIfStartOfOomMessages(line string) bool {
 // Should prevent EOF errors that occur when lines are read before being fully
 // written to the log. It reads line by line splitting on
 // the "\n" character.
-func readLinesFromFile(lineChannel chan string, ioreader *bufio.Reader) {
+func readLinesFromFile(lineChannel chan string, ioreader *bufio.Reader) error {
 	linefragment := ""
 	var line string
 	var err error
 	for true {
 		line, err = ioreader.ReadString('\n')
-		if err == io.EOF {
-			if line != "" {
+		if err != nil && err != io.EOF {
+			glog.Errorf("exiting analyzeLinesHelper with error %v", err)
+			close(lineChannel)
+			break
+		}
+		if line == "" {
+			time.Sleep(100 * time.Millisecond)
+			continue
+		}
+		if err == nil {
+			lineChannel <- linefragment + line
+			linefragment = ""
+		} else { // err == io.EOF
 			linefragment += line
 		}
-			time.Sleep(100 * time.Millisecond)
-		} else if err == nil {
-			if linefragment != "" {
-				line = linefragment + line
-				linefragment = ""
-			}
-			lineChannel <- line
-		} else if err != nil && err != io.EOF {
-			glog.Errorf("exiting analyzeLinesHelper with error %v", err)
-		}
 	}
+	return err
 }
 
 // Calls goroutine for readLinesFromFile, which feeds it complete lines.
@@ -160,7 +162,7 @@ func (self *OomParser) StreamOoms(outStream chan *OomInstance) {
 			outStream <- oomCurrentInstance
 		}
 	}
-	glog.Infof("exiting analyzeLines")
+	glog.Infof("exiting analyzeLines. OOM events will not be reported.")
 }
 
 func callJournalctl() (io.ReadCloser, error) {
@@ -184,7 +186,6 @@ func trySystemd() (*OomParser, error) {
 	return &OomParser{
 		ioreader: bufio.NewReader(readcloser),
 	}, nil
-
 }
 
 // List of possible kernel log files. These are prioritized in order so that
@@ -193,7 +194,7 @@ var kernelLogFiles = []string{"/var/log/kern.log", "/var/log/messages", "/var/lo
 
 // looks for system files that contain kernel messages and if one is found, sets
 // the systemFile attribute of the OomParser object
-func getSystemFile() (string, error) {
+func getLogFile() (string, error) {
 	for _, logFile := range kernelLogFiles {
 		if utils.FileExists(logFile) {
 			glog.Infof("OOM parser using kernel log file: %q", logFile)
@@ -203,22 +204,29 @@ func getSystemFile() (string, error) {
 	return "", fmt.Errorf("unable to find any kernel log file available from our set: %v", kernelLogFiles)
 }
 
-// initializes an OomParser object and calls getSystemFile to set the systemFile
-// attribute.  Returns and OomParser object and an error
+func tryLogFile() (*OomParser, error) {
+	logFile, err := getLogFile()
+	if err != nil {
+		return nil, err
+	}
+	tail, err := tail.NewTail(logFile)
+	if err != nil {
+		return nil, err
+	}
+	return &OomParser{
+		ioreader: bufio.NewReader(tail),
+	}, nil
+}
+
+// initializes an OomParser object. Returns an OomParser object and an error.
 func New() (*OomParser, error) {
 	parser, err := trySystemd()
 	if err == nil {
 		return parser, nil
 	}
-	systemFile, err := getSystemFile()
-	if err != nil {
+	parser, err = tryLogFile()
+	if err == nil {
+		return parser, nil
+	}
 	return nil, err
 }
-	file, err := os.Open(systemFile)
-	if err != nil {
-		return nil, err
-	}
-	return &OomParser{
-		ioreader: bufio.NewReader(file),
-	}, nil
-}

utils/tail/tail.go

@@ -0,0 +1,157 @@
+// Copyright 2016 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package tail implements "tail -F" functionality following rotated logs
+package tail
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/golang/glog"
+	"golang.org/x/exp/inotify"
+)
+
+const (
+	readerStateOpening = 1 << iota
+	readerStateOpened
+	readerStateError
+)
+
+type Tail struct {
+	reader      *bufio.Reader
+	readerState int
+	readerLock  sync.Mutex
+	filename    string
+	file        *os.File
+	stop        chan bool
+	watcher     *inotify.Watcher
+}
+
+const retryOpenInterval = time.Second
+const maxOpenAttempts = 3
+
+// NewTail starts opens the given file and watches it for deletion/rotation
+func NewTail(filename string) (*Tail, error) {
+	t := &Tail{
+		filename: filename,
+	}
+	var err error
+	t.stop = make(chan bool)
+	t.watcher, err = inotify.NewWatcher()
+	if err != nil {
+		return nil, fmt.Errorf("inotify init failed on %s: %v", t.filename, err)
+	}
+	t.readerState = readerStateOpening
+	go t.watchLoop()
+	return t, nil
+}
+
+// Read implements the io.Reader interface for Tail
+func (t *Tail) Read(p []byte) (int, error) {
+	t.readerLock.Lock()
+	defer t.readerLock.Unlock()
+	if t.reader == nil {
+		if t.readerState == readerStateOpening {
+			return 0, nil
+		}
+		return 0, fmt.Errorf("can't open log file %s", t.filename)
+	}
+	return t.reader.Read(p)
+}
+
+var _ io.Reader = &Tail{}
+
+// Close stops watching and closes the file
+func (t *Tail) Close() {
+	close(t.stop)
+}
+
+func isEvent(event *inotify.Event, flag uint32) bool {
+	return event.Mask&flag == flag
+}
+
+func (t *Tail) fileChanged() error {
+	for {
+		select {
+		case event := <-t.watcher.Event:
+			// We don't get IN_DELETE because we are holding the file open
+			if isEvent(event, inotify.IN_ATTRIB) || isEvent(event, inotify.IN_MOVE_SELF) {
+				return nil
+			}
+		case <-t.stop:
+			return fmt.Errorf("watch was cancelled")
+		}
+	}
+}
+
+func (t *Tail) attemptOpen() (err error) {
+	t.readerLock.Lock()
+	defer t.readerLock.Unlock()
+	for attempt := 1; attempt <= maxOpenAttempts; attempt++ {
+		glog.V(4).Infof("Opening %s (attempt %d of %d)", t.filename, attempt, maxOpenAttempts)
+		t.file, err = os.Open(t.filename)
+		if err == nil {
+			// TODO: not interested in old events?
+			//t.file.Seek(0, os.SEEK_END)
+			t.reader = bufio.NewReader(t.file)
+			t.readerState = readerStateOpened
+			return nil
+		}
+		select {
+		case <-time.After(retryOpenInterval):
+		case <-t.stop:
+			t.readerState = readerStateError
+			return fmt.Errorf("watch was cancelled")
+		}
+	}
+	t.readerState = readerStateError
+	return err
+}
+
+func (t *Tail) watchLoop() {
+	for {
+		err := t.watchFile()
+		if err != nil {
+			glog.Errorf("Tail failed on %s: %v", t.filename, err)
+			break
+		}
+	}
+}
+
+func (t *Tail) watchFile() error {
+	err := t.attemptOpen()
+	if err != nil {
+		return err
+	}
+	defer t.file.Close()
+	err = t.watcher.Watch(t.filename)
+	if err != nil {
+		return err
+	}
+	defer t.watcher.RemoveWatch(t.filename)
+	err = t.fileChanged()
+	if err != nil {
+		return err
+	}
+	glog.V(4).Infof("Log file %s moved/deleted", t.filename)
+	t.readerLock.Lock()
+	defer t.readerLock.Unlock()
+	t.readerState = readerStateOpening
+	return nil
+}