diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go index 9b1729f63..c82da3811 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 @@ -92,8 +94,11 @@ func (k *kernelLogWatcher) Stop() { func (k *kernelLogWatcher) watchLoop() { kmsgs := k.kmsgParser.Parse() defer func() { - if err := k.kmsgParser.Close(); err != nil { - klog.Errorf("Failed to close kmsg parser: %v", err) + // kmsgParser is nil when stopping interrupted a restart. + if k.kmsgParser != nil { + if err := k.kmsgParser.Close(); err != nil { + klog.Errorf("Failed to close kmsg parser: %v", err) + } } close(k.logCh) k.tomb.Done() @@ -108,13 +113,14 @@ func (k *kernelLogWatcher) watchLoop() { if !ok { klog.Error("Kmsg channel closed, attempting to restart kmsg parser") - // Close the old parser + // Close the old parser and clear the reference so the + // deferred cleanup doesn't close it a second time. if err := k.kmsgParser.Close(); err != nil { klog.Errorf("Failed to close kmsg parser: %v", err) } + k.kmsgParser = nil - // 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 { @@ -134,9 +140,16 @@ func (k *kernelLogWatcher) watchLoop() { continue } - k.logCh <- &logtypes.Log{ + // The consumer stops draining logCh before calling Stop(), so a + // plain send on a full channel could block forever and deadlock Stop(). + select { + case k.logCh <- &logtypes.Log{ Message: strings.TrimSpace(msg.Message), Timestamp: msg.Timestamp, + }: + case <-k.tomb.Stopping(): + klog.Infof("Stop watching kernel log") + return } } } @@ -144,6 +157,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 +168,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 +188,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 50d24e39a..25c181c15 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,147 @@ 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()") + } +} + +// TestStopDuringRestartClosesOldParserOnce verifies that when Stop() arrives +// while the watcher is in the restart path, the already-closed old parser is +// not closed a second time by watchLoop's deferred cleanup. +func TestStopDuringRestartClosesOldParserOnce(t *testing.T) { + now := time.Now() + + // Closing the channel after sending drives watchLoop into the restart path. + mock := &mockKmsgParser{ + kmsgs: []kmsgparser.Message{{Message: "msg", Timestamp: now}}, + closeAfterSend: true, + } + + factoryCalled := make(chan struct{}, 1) + w := &kernelLogWatcher{ + cfg: types.WatcherConfig{}, + startTime: now.Add(-time.Minute), + tomb: tomb.NewTomb(), + logCh: make(chan *logtypes.Log, 100), + kmsgParser: mock, + newParser: func() (kmsgparser.Parser, error) { + select { + case factoryCalled <- struct{}{}: + default: + } + // Keep the watcher in the retry loop until Stop() is called. + return nil, fmt.Errorf("kmsg unavailable") + }, + } + + logCh, err := w.Watch() + assert.NoError(t, err) + <-logCh + + // Wait until watchLoop has entered the restart path. + select { + case <-factoryCalled: + case <-time.After(time.Second): + t.Fatal("timeout waiting for restart attempt") + } + + w.Stop() + + 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()") + } + + assert.Equal(t, 1, mock.CloseCallCount(), + "old parser must be closed exactly once, not again by watchLoop's defer") +} + +// TestStopDoesNotDeadlockWhenLogChannelFull verifies that Stop() returns even +// when logCh is full and nobody is draining it. +func TestStopDoesNotDeadlockWhenLogChannelFull(t *testing.T) { + now := time.Now() + + // More messages than logCh capacity so watchLoop ends up blocked sending. + kmsgs := make([]kmsgparser.Message, 150) + for i := range kmsgs { + kmsgs[i] = kmsgparser.Message{Message: fmt.Sprintf("msg-%d", i), Timestamp: now} + } + + w := &kernelLogWatcher{ + cfg: types.WatcherConfig{}, + startTime: now.Add(-time.Minute), + tomb: tomb.NewTomb(), + logCh: make(chan *logtypes.Log, 100), + kmsgParser: &mockKmsgParser{kmsgs: kmsgs}, + } + + // Watch but never read logCh, mimicking the log monitor after it has + // decided to stop. + _, err := w.Watch() + assert.NoError(t, err) + + // Let watchLoop fill the channel and block on the send. + time.Sleep(300 * time.Millisecond) + + stopped := make(chan struct{}) + go func() { + w.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(2 * time.Second): + t.Fatal("Stop() deadlocked while logCh was full") + } +} + // TestWatcherProcessesMessageContent verifies watchLoop's per-message // handling: empty messages are dropped, and surrounding whitespace is // trimmed before forwarding.