github/node-problem-detector
1
0
Fork 0
mirror of https://github.com/kubernetes/node-problem-detector.git synced 2026-08-19 20:26:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #41 from euank/kmsg-parser

Browse Source 
logwatchers: add new kmsg-based kernel log watcher
This commit is contained in:
Lantao Liu
2017-05-30 15:53:24 -07:00
committed by GitHub
parent 9f20903fe0 73cba49db0
commit be6c516cfd
9 changed files with 777 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/systemlogmonitor/README.md
+11 -9
View File
@@ -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
+115
View File
@@ -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
+160
View File
@@ -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
+23
View File
@@ -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)
}