Merge pull request #41 from euank/kmsg-parser

logwatchers: add new kmsg-based kernel log watcher

node-problem-detector.yaml

@@ -27,6 +27,9 @@ spec:
         - name: log
           mountPath: /var/log
           readOnly: true
+        - name: kmsg
+          mountPath: /dev/kmsg
+          readOnly: true
         # Make sure node problem detector is in the same timezone
         # with the host.
         - name: localtime
@@ -40,6 +43,9 @@ spec:
         # Config `log` to your system log directory
         hostPath:
           path: /var/log/
+      - name: kmsg
+        hostPath:
+          path: /dev/kmsg
       - name: localtime
         hostPath:
           path: /etc/localtime

pkg/systemlogmonitor/README.md

@@ -8,10 +8,11 @@ the configuration files. (
 [`config/kernel-monitor.json`](https://github.com/kubernetes/node-problem-detector/blob/master/config/kernel-monitor.json) as an example).
 The rule list is extensible.
 
-## Limitations
+## Supported sources
 
-* System Log Monitor only supports file based log and journald now, but it is easy
-  to extend it with [new log watcher](#new-log-watcher)
+* System Log Monitor currently supports file-based logs, journald, and kmsg.
+  Additional sources can be added by implementing a [new log
+  watcher](#new-log-watcher).
 
 ## Add New NodeConditions
 
@@ -44,10 +45,10 @@ with new rule definition:
 
 System log monitor supports different log management tools with different log
 watchers:
-* [filelog](https://github.com/kubernetes/node-problem-detector/blob/master/pkg/systemlogmonitor/logwatchers/filelog): Log watcher for
+* [filelog](./logwatchers/filelog): Log watcher for
 arbitrary file based log.
-* [journald](https://github.com/kubernetes/node-problem-detector/blob/master/pkg/systemlogmonitor/logwatchers/journald): Log watcher for
-journald.
+* [journald](.//logwatchers/journald): Log watcher for
+* [kmsg](./logwatchers/kmsg): Log watcher for the kernel ring buffer device, /dev/kmsg.
 Set `plugin` in the configuration file to specify log watcher.
 
 ### Plugin Configuration
@@ -66,6 +67,7 @@ Log watcher specific configurations are configured in `pluginConfig`.
   * timestampFormat: The format of the timestamp. The format string is the time
     `2006-01-02T15:04:05Z07:00` in the expected format. (See
     [golang timestamp format](https://golang.org/pkg/time/#pkg-constants))
+* **kmsg**
 
 ### Change Log Path
 
@@ -78,6 +80,6 @@ field in the configurtion file is the log path. You can always configure
 
 ### New Log Watcher
 
-System log monitor uses [Log
-Watcher](https://github.com/kubernetes/node-problem-detector/blob/master/pkg/systemlogmonitor/logwatchers/types/log_watcher.go) to support different log management tools.
-It is easy to implement a new log watcher.
+System log monitor uses [Log Watcher](./logwatchers/types/log_watcher.go) to
+support different log management tools.  It is easy to implement a new log
+watcher.

pkg/systemlogmonitor/logwatchers/kmsg/log_watcher.go

@@ -0,0 +1,115 @@
+/*
+Copyright 2017 The Kubernetes Authors 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 kmsg
+
+import (
+	"bufio"
+	"fmt"
+	"strings"
+	"time"
+
+	utilclock "code.cloudfoundry.org/clock"
+	"github.com/euank/go-kmsg-parser/kmsgparser"
+	"github.com/golang/glog"
+
+	"k8s.io/node-problem-detector/pkg/systemlogmonitor/logwatchers/types"
+	logtypes "k8s.io/node-problem-detector/pkg/systemlogmonitor/types"
+	"k8s.io/node-problem-detector/pkg/systemlogmonitor/util"
+)
+
+type kernelLogWatcher struct {
+	cfg    types.WatcherConfig
+	logCh  chan *logtypes.Log
+	tomb   *util.Tomb
+	reader *bufio.Reader
+
+	kmsgParser kmsgparser.Parser
+	clock      utilclock.Clock
+}
+
+// NewKmsgWatcher creates a watcher which will read messages from /dev/kmsg
+func NewKmsgWatcher(cfg types.WatcherConfig) types.LogWatcher {
+	kmsgparser.NewParser()
+	return &kernelLogWatcher{
+		cfg:  cfg,
+		tomb: util.NewTomb(),
+		// Arbitrary capacity
+		logCh: make(chan *logtypes.Log, 100),
+		clock: utilclock.NewClock(),
+	}
+}
+
+var _ types.WatcherCreateFunc = NewKmsgWatcher
+
+func (k *kernelLogWatcher) Watch() (<-chan *logtypes.Log, error) {
+	if k.kmsgParser == nil {
+		// nil-check to make mocking easier
+		parser, err := kmsgparser.NewParser()
+		if err != nil {
+			return nil, fmt.Errorf("failed to create kmsg parser: %v", err)
+		}
+		k.kmsgParser = parser
+	}
+
+	lookback, err := time.ParseDuration(k.cfg.Lookback)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse lookback duration %q: %v", k.cfg.Lookback, err)
+	}
+
+	go k.watchLoop(lookback)
+	return k.logCh, nil
+}
+
+// Stop closes the kmsgparser
+func (k *kernelLogWatcher) Stop() {
+	k.kmsgParser.Close()
+	k.tomb.Stop()
+}
+
+// watchLoop is the main watch loop of kernel log watcher.
+func (k *kernelLogWatcher) watchLoop(lookback time.Duration) {
+	defer func() {
+		close(k.logCh)
+		k.tomb.Done()
+	}()
+	kmsgs := k.kmsgParser.Parse()
+
+	for {
+		select {
+		case <-k.tomb.Stopping():
+			glog.Infof("Stop watching kernel log")
+			k.kmsgParser.Close()
+			return
+		case msg := <-kmsgs:
+			glog.V(5).Infof("got kernel message: %+v", msg)
+			if msg.Message == "" {
+				continue
+			}
+
+			// Discard too old messages
+			if k.clock.Since(msg.Timestamp) > lookback {
+				glog.V(5).Infof("throwing away msg %v for being too old: %v > %v", msg.Message, msg.Timestamp.String(), lookback.String())
+				continue
+			}
+
+			k.logCh <- &logtypes.Log{
+				Message:   strings.TrimSpace(msg.Message),
+				Timestamp: msg.Timestamp,
+			}
+		}
+	}
+}

pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_test.go

@@ -0,0 +1,160 @@
+/*
+Copyright 2017 The Kubernetes Authors 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 kmsg
+
+import (
+	"io"
+	"testing"
+
+	"code.cloudfoundry.org/clock/fakeclock"
+	"github.com/euank/go-kmsg-parser/kmsgparser"
+	"github.com/stretchr/testify/assert"
+
+	"time"
+
+	"k8s.io/node-problem-detector/pkg/systemlogmonitor/logwatchers/types"
+	logtypes "k8s.io/node-problem-detector/pkg/systemlogmonitor/types"
+)
+
+type mockKmsgParser struct {
+	kmsgs []kmsgparser.Message
+}
+
+func (m *mockKmsgParser) SetLogger(kmsgparser.Logger) {}
+func (m *mockKmsgParser) Close() error                { return nil }
+func (m *mockKmsgParser) Parse() <-chan kmsgparser.Message {
+	c := make(chan kmsgparser.Message)
+	go func() {
+		for _, msg := range m.kmsgs {
+			c <- msg
+		}
+	}()
+	return c
+}
+func (m *mockKmsgParser) SeekEnd() error { return nil }
+
+func TestWatch(t *testing.T) {
+	now := time.Date(time.Now().Year(), time.January, 2, 3, 4, 5, 0, time.Local)
+	fakeClock := fakeclock.NewFakeClock(now)
+	testCases := []struct {
+		log      *mockKmsgParser
+		logs     []logtypes.Log
+		lookback string
+	}{
+		{
+			// The start point is at the head of the log file.
+			log: &mockKmsgParser{kmsgs: []kmsgparser.Message{
+				{Message: "1", Timestamp: now.Add(0 * time.Second)},
+				{Message: "2", Timestamp: now.Add(1 * time.Second)},
+				{Message: "3", Timestamp: now.Add(2 * time.Second)},
+			}},
+			logs: []logtypes.Log{
+				{
+					Timestamp: now,
+					Message:   "1",
+				},
+				{
+					Timestamp: now.Add(time.Second),
+					Message:   "2",
+				},
+				{
+					Timestamp: now.Add(2 * time.Second),
+					Message:   "3",
+				},
+			},
+			lookback: "0",
+		},
+		{
+			// The start point is in the middle of the log file.
+			log: &mockKmsgParser{kmsgs: []kmsgparser.Message{
+				{Message: "1", Timestamp: now.Add(-1 * time.Second)},
+				{Message: "2", Timestamp: now.Add(0 * time.Second)},
+				{Message: "3", Timestamp: now.Add(1 * time.Second)},
+			}},
+			logs: []logtypes.Log{
+				{
+					Timestamp: now,
+					Message:   "2",
+				},
+				{
+					Timestamp: now.Add(time.Second),
+					Message:   "3",
+				},
+			},
+			lookback: "0",
+		},
+		{
+			// The start point is at the end of the log file, but we look back.
+			log: &mockKmsgParser{kmsgs: []kmsgparser.Message{
+				{Message: "1", Timestamp: now.Add(-2 * time.Second)},
+				{Message: "2", Timestamp: now.Add(-1 * time.Second)},
+				{Message: "3", Timestamp: now.Add(0 * time.Second)},
+			}},
+			lookback: "1s",
+			logs: []logtypes.Log{
+				{
+					Timestamp: now.Add(-time.Second),
+					Message:   "2",
+				},
+				{
+					Timestamp: now,
+					Message:   "3",
+				},
+			},
+		},
+	}
+	for _, test := range testCases {
+		w := NewKmsgWatcher(types.WatcherConfig{Lookback: test.lookback})
+		w.(*kernelLogWatcher).clock = fakeClock
+		w.(*kernelLogWatcher).kmsgParser = test.log
+		logCh, err := w.Watch()
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer w.Stop()
+		for _, expected := range test.logs {
+			got := <-logCh
+			assert.Equal(t, &expected, got)
+		}
+		// The log channel should have already been drained
+		// There could stil be future messages sent into the channel, but the chance is really slim.
+		timeout := time.After(100 * time.Millisecond)
+		select {
+		case log := <-logCh:
+			t.Errorf("unexpected extra log: %+v", *log)
+		case <-timeout:
+		}
+	}
+}
+
+type fakeKmsgReader struct {
+	logLines []string
+}
+
+func (r *fakeKmsgReader) Read(data []byte) (int, error) {
+	if len(r.logLines) == 0 {
+		return 0, io.EOF
+	}
+	l := r.logLines[0]
+	r.logLines = r.logLines[1:]
+	copy(data, []byte(l))
+	return len(l), nil
+}
+
+func (r *fakeKmsgReader) Close() error {
+	return nil
+}

pkg/systemlogmonitor/logwatchers/register_kmsg.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2017 The Kubernetes Authors 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 logwatchers
+
+import "k8s.io/node-problem-detector/pkg/systemlogmonitor/logwatchers/kmsg"
+
+func init() {
+	registerLogWatcher("kmsg", kmsg.NewKmsgWatcher)
+}