From 9c23921c11b96f254bb91000cd23b6324b8797a1 Mon Sep 17 00:00:00 2001 From: Euan Kemp Date: Thu, 9 Mar 2017 20:40:49 -0800 Subject: [PATCH] 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). --- .../logwatchers/kmsg/log_watcher.go | 115 +++++++++++++ .../logwatchers/kmsg/log_watcher_test.go | 160 ++++++++++++++++++ .../logwatchers/register_kmsg.go | 23 +++ 3 files changed, 298 insertions(+) create mode 100644 pkg/systemlogmonitor/logwatchers/kmsg/log_watcher.go create mode 100644 pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_test.go create mode 100644 pkg/systemlogmonitor/logwatchers/register_kmsg.go diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher.go new file mode 100644 index 00000000..f5d6f4e7 --- /dev/null +++ b/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, + } + } + } +} diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_test.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_test.go new file mode 100644 index 00000000..f34037c5 --- /dev/null +++ b/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 +} diff --git a/pkg/systemlogmonitor/logwatchers/register_kmsg.go b/pkg/systemlogmonitor/logwatchers/register_kmsg.go new file mode 100644 index 00000000..37c4c67b --- /dev/null +++ b/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) +}