diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go index 9b1729f6..83818e8e 100644 --- a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go +++ b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go @@ -44,6 +44,8 @@ type kernelLogWatcher struct { kmsgParser kmsgparser.Parser // newParser creates a kmsgparser. Overridable in tests; defaults to kmsgparser.NewParser. newParser func() (kmsgparser.Parser, error) + // lastRestart is when the parser was last restarted; used to rate-limit restarts. + lastRestart time.Time } // NewKmsgWatcher creates a watcher which will read messages from /dev/kmsg @@ -113,8 +115,7 @@ func (k *kernelLogWatcher) watchLoop() { klog.Errorf("Failed to close kmsg parser: %v", err) } - // Try to restart immediately. retryCreateParser() applies backoff only - // after a failed NewParser() or SeekEnd() attempt. + // Try to restart. retryCreateParser() waits between attempts. var restarted bool kmsgs, restarted = k.retryCreateParser() if !restarted { @@ -144,6 +145,9 @@ func (k *kernelLogWatcher) watchLoop() { // retryCreateParser attempts to create a new kmsg parser. // It tries immediately first, then waits retryDelay between subsequent failures. +// The first attempt is also delayed if the previous restart was less than +// retryDelay ago, so a parser that keeps failing right after a successful +// restart cannot drive a hot restart loop. // On success, it seeks the new parser to the end of the kmsg ring buffer to // avoid replaying messages that were already processed before the restart. // Any messages written to kmsg between the old parser closing and the new @@ -152,6 +156,15 @@ func (k *kernelLogWatcher) watchLoop() { // restart was triggered by a kmsg flood. // It returns the new message channel and true on success, or nil and false if stopping was signaled. func (k *kernelLogWatcher) retryCreateParser() (<-chan kmsgparser.Message, bool) { + if since := time.Since(k.lastRestart); since < retryDelay { + select { + case <-k.tomb.Stopping(): + klog.Infof("Stop watching kernel log during restart attempt") + return nil, false + case <-time.After(retryDelay - since): + } + } + for { parser, err := k.newParser() if err != nil { @@ -163,6 +176,7 @@ func (k *kernelLogWatcher) retryCreateParser() (<-chan kmsgparser.Message, bool) } } else { k.kmsgParser = parser + k.lastRestart = time.Now() klog.Infof("Successfully restarted kmsg parser") return parser.Parse(), true } diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go index 50d24e39..bc4db02b 100644 --- a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go +++ b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go @@ -19,6 +19,7 @@ package kmsg import ( "fmt" "sync" + "sync/atomic" "testing" "time" @@ -418,6 +419,55 @@ func TestWatcherRestartsOnUnexpectedChannelClose(t *testing.T) { assert.Equal(t, 1, second.SeekEndCallCount(), "SeekEnd should be called once on the restart parser") } +// TestWatcherRateLimitsRestarts verifies that a parser whose channel closes +// right after every restart does not drive a hot restart loop: only the first +// restart is immediate, the next attempt waits retryDelay. +func TestWatcherRateLimitsRestarts(t *testing.T) { + now := time.Now() + + var factoryCalls int64 + w := &kernelLogWatcher{ + cfg: types.WatcherConfig{}, + startTime: now.Add(-time.Minute), + tomb: tomb.NewTomb(), + logCh: make(chan *logtypes.Log, 100), + // Every parser closes its channel immediately, triggering restarts. + kmsgParser: &mockKmsgParser{closeAfterSend: true}, + newParser: func() (kmsgparser.Parser, error) { + atomic.AddInt64(&factoryCalls, 1) + return &mockKmsgParser{closeAfterSend: true}, nil + }, + } + + logCh, err := w.Watch() + assert.NoError(t, err) + + // The first restart is immediate; the second must wait retryDelay (5s), + // so within this window the factory must be called exactly once. + time.Sleep(500 * time.Millisecond) + assert.EqualValues(t, 1, atomic.LoadInt64(&factoryCalls), + "only one restart should happen within retryDelay") + + // Stop() must return promptly while the watcher waits out the rate limit. + stopped := make(chan struct{}) + go func() { + w.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("timeout waiting for Stop() during restart rate-limit wait") + } + + select { + case _, ok := <-logCh: + assert.False(t, ok, "log channel should be closed after Stop()") + case <-time.After(time.Second): + t.Fatal("timeout waiting for log channel to close after Stop()") + } +} + // TestWatcherProcessesMessageContent verifies watchLoop's per-message // handling: empty messages are dropped, and surrounding whitespace is // trimmed before forwarding.