mirror of
https://github.com/kubernetes/node-problem-detector.git
synced 2026-08-20 04:36:41 +00:00
logwatchers/kmsg: add initial kmsg watcher impl
This adds a logwatcher which is able to parse kernel messages directly from the /dev/kmsg interface. This supports any modern linux distro, while also avoiding any dependency on libraries (e.g. as journald needs).
This commit is contained in:
Euan Kemp
@@ -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,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
Reference in New Issue
Block a user