Merge pull request #22 from Random-Liu/add-look-back

Kernel Monitor: Add look back support and kernel panic handling

pkg/kernelmonitor/kernel_log_watcher.go

@@ -18,7 +18,9 @@ package kernelmonitor
 
 import (
 	"bufio"
+	"fmt"
 	"os"
+	"time"
 
 	"k8s.io/node-problem-detector/pkg/kernelmonitor/translator"
 	"k8s.io/node-problem-detector/pkg/kernelmonitor/types"
@@ -26,6 +28,7 @@ import (
 
 	"github.com/golang/glog"
 	"github.com/hpcloud/tail"
+	utilclock "github.com/pivotal-golang/clock"
 )
 
 const (
@@ -34,7 +37,12 @@ const (
 
 // WatcherConfig is the configuration of kernel log watcher.
 type WatcherConfig struct {
+	// KernelLogPath is the path to the kernel log
 	KernelLogPath string `json:"logPath, omitempty"`
+	// StartPattern is the pattern of the start line
+	StartPattern string `json:"startPattern, omitempty"`
+	// Lookback is the time kernel watcher looks up
+	Lookback string `json:"lookback, omitempty"`
 }
 
 // KernelLogWatcher watches and translates the kernel log. Once there is new log line,
@@ -53,6 +61,7 @@ type kernelLogWatcher struct {
 	tl    *tail.Tail
 	logCh chan *types.KernelLog
 	tomb  *util.Tomb
+	clock utilclock.Clock
 }
 
 // NewKernelLogWatcher creates a new kernel log watcher.
@@ -63,6 +72,7 @@ func NewKernelLogWatcher(cfg WatcherConfig) KernelLogWatcher {
 		tomb:  util.NewTomb(),
 		// A capacity 1000 buffer should be enough
 		logCh: make(chan *types.KernelLog, 1000),
+		clock: utilclock.NewClock(),
 	}
 }
 
@@ -144,39 +154,46 @@ func (k *kernelLogWatcher) watchLoop() {
 	}
 }
 
-// getStartPoint parses the newest kernel log file and try to find the latest reboot point.
-// Currently we rely on the kernel log timestamp to find the reboot point. The basic idea
-// is straight forward: In the whole lifecycle of a node, the kernel log timestamp should
-// always increase, only when it is reboot, the timestamp will decrease. We just parse the
-// log and find the latest timestamp decreasing, then it should be the latest reboot point.
-// TODO(random-liu): A drawback is that if the node is started long time ago, we'll only get
-// logs in the newest kernel log file. We may want to improve this in the future.
+// getStartPoint finds the start point to parse the log. The start point is either
+// the line at (now - lookback) or the first line of kernel log.
+// Notice that, kernel log watcher doesn't look back to the rolled out logs.
 func (k *kernelLogWatcher) getStartPoint(path string) (int64, error) {
 	f, err := os.Open(path)
 	if err != nil {
-		return -1, err
+		return 0, fmt.Errorf("failed to open file %q: %v", path, err)
 	}
 	defer f.Close()
+	lookback, err := parseDuration(k.cfg.Lookback)
+	if err != nil {
+		return 0, fmt.Errorf("failed to parse duration %q: %v", k.cfg.Lookback, err)
+	}
 	start := int64(0)
-	total := 0
-	lastTimestamp := int64(0)
 	reader := bufio.NewReader(f)
 	done := false
 	for !done {
 		line, err := reader.ReadString('\n')
 		if err != nil {
+			if len(line) == 0 {
+				// No need to continue parsing if nothing is read
+				break
+			}
 			done = true
 		}
-		total += len(line)
 		log, err := k.trans.Translate(line)
 		if err != nil {
 			glog.Infof("unable to parse line: %q, %v", line, err)
-			continue
+		} else if k.clock.Since(log.Timestamp) <= lookback {
+			break
 		}
-		if log.Timestamp < lastTimestamp {
-			start = int64(total - len(line))
-		}
-		lastTimestamp = log.Timestamp
+		start += int64(len(line))
 	}
 	return start, nil
 }
+
+func parseDuration(s string) (time.Duration, error) {
+	// If the duration is not configured, just return 0 by default
+	if s == "" {
+		return 0, nil
+	}
+	return time.ParseDuration(s)
+}

pkg/kernelmonitor/kernel_log_watcher_test.go

@@ -24,50 +24,73 @@ import (
 	"time"
 
 	"k8s.io/node-problem-detector/pkg/kernelmonitor/types"
+
+	"github.com/pivotal-golang/clock/fakeclock"
 )
 
 func TestGetStartPoint(t *testing.T) {
+	// now is a fake time
+	now := time.Date(time.Now().Year(), time.January, 2, 3, 4, 5, 0, time.Local)
+	fakeClock := fakeclock.NewFakeClock(now)
 	testCases := []struct {
-		log  string
-		logs []types.KernelLog
-		err  bool
+		log      string
+		logs     []types.KernelLog
+		lookback string
 	}{
 		{
 			// The start point is at the head of the log file.
-			log: `kernel: [1.000000] 1
-			kernel: [2.000000] 2
-			kernel: [3.000000] 3
+			log: `Jan  2 03:04:05 kernel: [0.000000] 1
+			Jan  2 03:04:06 kernel: [1.000000] 2
+			Jan  2 03:04:07 kernel: [2.000000] 3
 			`,
 			logs: []types.KernelLog{
 				{
-					Timestamp: 1000000,
+					Timestamp: now,
 					Message:   "1",
 				},
 				{
-					Timestamp: 2000000,
+					Timestamp: now.Add(time.Second),
 					Message:   "2",
 				},
 				{
-					Timestamp: 3000000,
+					Timestamp: now.Add(2 * time.Second),
 					Message:   "3",
 				},
 			},
 		},
 		{
 			// The start point is in the middle of the log file.
-			log: `kernel: [3.000000] 3
-			kernel: [1.000000] 1
-			kernel: [2.000000] 2
+			log: `Jan  2 03:04:04 kernel: [0.000000] 1
+			Jan  2 03:04:05 kernel: [1.000000] 2
+			Jan  2 03:04:06 kernel: [2.000000] 3
 			`,
 			logs: []types.KernelLog{
 				{
-					Timestamp: 1000000,
-					Message:   "1",
+					Timestamp: now,
+					Message:   "2",
 				},
 				{
-					Timestamp: 2000000,
+					Timestamp: now.Add(time.Second),
+					Message:   "3",
+				},
+			},
+		},
+		{
+			// The start point is at the end of the log file, but we look back.
+			log: `Jan  2 03:04:03 kernel: [0.000000] 1
+			Jan  2 03:04:04 kernel: [1.000000] 2
+			Jan  2 03:04:05 kernel: [2.000000] 3
+			`,
+			lookback: "1s",
+			logs: []types.KernelLog{
+				{
+					Timestamp: now.Add(-time.Second),
 					Message:   "2",
 				},
+				{
+					Timestamp: now,
+					Message:   "3",
+				},
 			},
 		},
 	}
@@ -84,7 +107,9 @@ func TestGetStartPoint(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		w := NewKernelLogWatcher(WatcherConfig{KernelLogPath: f.Name()})
+		w := NewKernelLogWatcher(WatcherConfig{KernelLogPath: f.Name(), Lookback: test.lookback})
+		// Set the fake clock.
+		w.(*kernelLogWatcher).clock = fakeClock
 		logCh, err := w.Watch()
 		if err != nil {
 			t.Fatal(err)

pkg/kernelmonitor/kernel_monitor.go

@@ -58,7 +58,6 @@ type kernelMonitor struct {
 	buffer     LogBuffer
 	config     MonitorConfig
 	conditions []types.Condition
-	uptime     time.Time
 	logCh      <-chan *kerntypes.KernelLog
 	output     chan *types.Status
 	tomb       *util.Tomb
@@ -77,8 +76,6 @@ func NewKernelMonitorOrDie(configPath string) KernelMonitor {
 	if err != nil {
 		panic(err)
 	}
-	// Initialize the default node conditions
-	k.conditions = initialConditions(k.config.DefaultConditions)
 	err = validateRules(k.config.Rules)
 	if err != nil {
 		panic(err)
@@ -89,8 +86,6 @@ func NewKernelMonitorOrDie(configPath string) KernelMonitor {
 	if err != nil {
 		panic(err)
 	}
-	k.uptime = time.Now().Add(time.Duration(-info.Uptime * int64(time.Second)))
-	glog.Infof("Got system boot time: %v", k.uptime)
 	k.watcher = NewKernelLogWatcher(k.config.WatcherConfig)
 	k.buffer = NewLogBuffer(k.config.BufferSize)
 	// A 1000 size channel should be big enough.
@@ -117,22 +112,11 @@ func (k *kernelMonitor) Stop() {
 // monitorLoop is the main loop of kernel monitor.
 func (k *kernelMonitor) monitorLoop() {
 	defer k.tomb.Done()
-	k.output <- k.initialStatus() // Update the initial status
+	k.initializeStatus()
 	for {
 		select {
 		case log := <-k.logCh:
-			// Once there is new log, kernel monitor will push it into the log buffer and try
-			// to match each rule. If any rule is matched, kernel monitor will report a status.
-			k.buffer.Push(log)
-			for _, rule := range k.config.Rules {
-				matched := k.buffer.Match(rule.Pattern)
-				if len(matched) == 0 {
-					continue
-				}
-				status := k.generateStatus(matched, rule)
-				glog.Infof("New status generated: %+v", status)
-				k.output <- status
-			}
+			k.parseLog(log)
 		case <-k.tomb.Stopping():
 			k.watcher.Stop()
 			glog.Infof("Kernel monitor stopped")
@@ -141,15 +125,33 @@ func (k *kernelMonitor) monitorLoop() {
 	}
 }
 
+// parseLog parses one log line.
+func (k *kernelMonitor) parseLog(log *kerntypes.KernelLog) {
+	// Once there is new log, kernel monitor will push it into the log buffer and try
+	// to match each rule. If any rule is matched, kernel monitor will report a status.
+	k.buffer.Push(log)
+	if matched := k.buffer.Match(k.config.StartPattern); len(matched) != 0 {
+		// Reset the condition if a start log shows up.
+		glog.Infof("Found start log %q, re-initialize the status", generateMessage(matched))
+		k.initializeStatus()
+		return
+	}
+	for _, rule := range k.config.Rules {
+		matched := k.buffer.Match(rule.Pattern)
+		if len(matched) == 0 {
+			continue
+		}
+		status := k.generateStatus(matched, rule)
+		glog.Infof("New status generated: %+v", status)
+		k.output <- status
+	}
+}
+
 // generateStatus generates status from the logs.
 func (k *kernelMonitor) generateStatus(logs []*kerntypes.KernelLog, rule kerntypes.Rule) *types.Status {
 	// We use the timestamp of the first log line as the timestamp of the status.
-	timestamp := k.generateTimestamp(logs[0].Timestamp)
-	messages := []string{}
-	for _, log := range logs {
-		messages = append(messages, log.Message)
-	}
-	message := concatLogs(messages)
+	timestamp := logs[0].Timestamp
+	message := generateMessage(logs)
 	var events []types.Event
 	if rule.Type == kerntypes.Temp {
 		// For temporary error only generate event
@@ -181,14 +183,13 @@ func (k *kernelMonitor) generateStatus(logs []*kerntypes.KernelLog, rule kerntyp
 	}
 }
 
-// generateTimestamp converts the kernel log time to real time.
-func (k *kernelMonitor) generateTimestamp(timestamp int64) time.Time {
-	return k.uptime.Add(time.Duration(timestamp * int64(time.Microsecond)))
-}
-
-// initialStatus returns the initial status with initial condition.
-func (k *kernelMonitor) initialStatus() *types.Status {
-	return &types.Status{
+// initializeStatus initializes the internal condition and also reports it to the node problem detector.
+func (k *kernelMonitor) initializeStatus() {
+	// Initialize the default node conditions
+	k.conditions = initialConditions(k.config.DefaultConditions)
+	glog.Infof("Initalize condition generated: %+v", k.conditions)
+	// Update the initial status
+	k.output <- &types.Status{
 		Source:     k.config.Source,
 		Conditions: k.conditions,
 	}
@@ -215,3 +216,11 @@ func validateRules(rules []kerntypes.Rule) error {
 	}
 	return nil
 }
+
+func generateMessage(logs []*kerntypes.KernelLog) string {
+	messages := []string{}
+	for _, log := range logs {
+		messages = append(messages, log.Message)
+	}
+	return concatLogs(messages)
+}

pkg/kernelmonitor/kernel_monitor_test.go

@@ -32,7 +32,6 @@ const (
 )
 
 func TestGenerateStatus(t *testing.T) {
-	uptime := time.Unix(1000, 0)
 	initConditions := []types.Condition{
 		{
 			Type:       testConditionA,
@@ -47,11 +46,11 @@ func TestGenerateStatus(t *testing.T) {
 	}
 	logs := []*kerntypes.KernelLog{
 		{
-			Timestamp: 100000,
+			Timestamp: time.Unix(1000, 1000),
 			Message:   "test message 1",
 		},
 		{
-			Timestamp: 200000,
+			Timestamp: time.Unix(2000, 2000),
 			Message:   "test message 2",
 		},
 	}
@@ -72,7 +71,7 @@ func TestGenerateStatus(t *testing.T) {
 					{
 						Type:       testConditionA,
 						Status:     true,
-						Transition: time.Unix(1000, 100000*1000),
+						Transition: time.Unix(1000, 1000),
 						Reason:     "test reason",
 						Message:    "test message 1\ntest message 2",
 					},
@@ -89,7 +88,7 @@ func TestGenerateStatus(t *testing.T) {
 				Source: testSource,
 				Events: []types.Event{{
 					Severity:  types.Warn,
-					Timestamp: time.Unix(1000, 100000*1000),
+					Timestamp: time.Unix(1000, 1000),
 					Reason:    "test reason",
 					Message:   "test message 1\ntest message 2",
 				}},
@@ -102,7 +101,6 @@ func TestGenerateStatus(t *testing.T) {
 				Source: testSource,
 			},
 			conditions: initConditions,
-			uptime:     uptime,
 		}
 		got := k.generateStatus(logs, test.rule)
 		if !reflect.DeepEqual(&test.expected, got) {

pkg/kernelmonitor/translator/translator.go

@@ -18,8 +18,8 @@ package translator
 
 import (
 	"fmt"
-	"strconv"
 	"strings"
+	"time"
 
 	"k8s.io/node-problem-detector/pkg/kernelmonitor/types"
 )
@@ -41,11 +41,7 @@ func NewDefaultTranslator() Translator {
 }
 
 func (t *defaultTranslator) Translate(line string) (*types.KernelLog, error) {
-	timestr, message, err := parseLine(line)
-	if err != nil {
-		return nil, err
-	}
-	timestamp, err := parseTimestamp(timestr)
+	timestamp, message, err := t.parseLine(line)
 	if err != nil {
 		return nil, err
 	}
@@ -55,32 +51,35 @@ func (t *defaultTranslator) Translate(line string) (*types.KernelLog, error) {
 	}, nil
 }
 
-func parseLine(line string) (string, string, error) {
+var (
+	timestampLen  = 15
+	messagePrefix = "]"
+)
+
+func (t *defaultTranslator) parseLine(line string) (time.Time, string, error) {
+	// Trim the spaces to make sure timestamp could be found
+	line = strings.TrimSpace(line)
+	if len(line) < timestampLen {
+		return time.Time{}, "", fmt.Errorf("the line is too short: %q", line)
+	}
 	// Example line: Jan  1 00:00:00 hostname kernel: [0.000000] component: log message
-	timestampPrefix := "kernel: ["
-	timestampSuffix := "]"
-	idx := strings.Index(line, timestampPrefix)
-	if idx == -1 {
-		return "", "", fmt.Errorf("can't find timestamp prefix %q in line %q", timestampPrefix, line)
+	now := time.Now()
+	// There is no time zone information in kernel log timestamp, apply the current time
+	// zone.
+	timestamp, err := time.ParseInLocation(time.Stamp, line[:timestampLen], time.Local)
+	if err != nil {
+		return time.Time{}, "", fmt.Errorf("error parsing timestamp in line %q: %v", line, err)
 	}
-	line = line[idx+len(timestampPrefix):]
+	// There is no year information in kernel log timestamp, apply the current year.
+	// This could go wrong during looking back phase after kernel monitor is started,
+	// and the old logs are generated in old year.
+	timestamp = timestamp.AddDate(now.Year(), 0, 0)
 
-	idx = strings.Index(line, timestampSuffix)
-	if idx == -1 {
-		return "", "", fmt.Errorf("can't find timestamp suffix %q in line %q", timestampSuffix, line)
+	loc := strings.Index(line, messagePrefix)
+	if loc == -1 {
+		return timestamp, "", fmt.Errorf("can't find message prefix %q in line %q", messagePrefix, line)
 	}
-
-	timestamp := strings.Trim(line[:idx], " ")
-	message := strings.Trim(line[idx+1:], " ")
+	message := strings.Trim(line[loc+1:], " ")
 
 	return timestamp, message, nil
 }
-
-func parseTimestamp(timestamp string) (int64, error) {
-	f, err := strconv.ParseFloat(timestamp, 64)
-	if err != nil {
-		return 0, err
-	}
-	// seconds to microseconds
-	return int64(f * 1000000), nil
-}

pkg/kernelmonitor/translator/translator_test.go

@@ -18,33 +18,32 @@ package translator
 
 import (
 	"testing"
+	"time"
 )
 
 func TestDefaultTranslator(t *testing.T) {
 	tr := NewDefaultTranslator()
-
+	year := time.Now().Year()
 	testCases := []struct {
 		input     string
 		err       bool
-		timestamp int64
+		timestamp time.Time
 		message   string
 	}{
 		{
-			input:     "Jan  1 00:00:00 hostname kernel: [9.999999] component: log message",
-			timestamp: 9999999,
+			input:     "May  1 12:23:45 hostname kernel: [0.000000] component: log message",
+			timestamp: time.Date(year, time.May, 1, 12, 23, 45, 0, time.Local),
 			message:   "component: log message",
 		},
 		{
-			input:     "Jan  1 00:00:00 hostname kernel: [9.999999]",
-			timestamp: 9999999,
+			// no log message
+			input:     "May 21 12:23:45 hostname kernel: [9.999999]",
+			timestamp: time.Date(year, time.May, 21, 12, 23, 45, 0, time.Local),
 			message:   "",
 		},
 		{
-			input: "Jan  1 00:00:00 hostname kernel: [9.999999 component: log message",
-			err:   true,
-		},
-		{
-			input: "Jan  1 00:00:00 hostname user: [9.999999] component: log message",
+			// the right square bracket is missing
+			input: "May 21 12:23:45 hostname kernel: [9.999999 component: log message",
 			err:   true,
 		},
 	}
@@ -58,7 +57,7 @@ func TestDefaultTranslator(t *testing.T) {
 			continue
 		}
 		if test.timestamp != log.Timestamp || test.message != log.Message {
-			t.Errorf("case %d: expect timestamp: %d, message: %q; got %+v", c+1, test.timestamp, test.message, log)
+			t.Errorf("case %d: expect %v, %q; got %v, %q", c+1, test.timestamp, test.message, log.Timestamp, log.Message)
 		}
 	}
 }

pkg/kernelmonitor/types/types.go

@@ -16,10 +16,14 @@ limitations under the License.
 
 package types
 
+import (
+	"time"
+)
+
 // KernelLog is the log item returned by translator. It's very easy to extend this
 // to support other log monitoring, such as docker log monitoring.
 type KernelLog struct {
-	Timestamp int64 // microseconds since kernel boot
+	Timestamp time.Time
 	Message   string
 }