From 187d4d30d6d7ece67ab597bf9ec213d23c024aa5 Mon Sep 17 00:00:00 2001 From: Ciprian Hacman Date: Sat, 11 Jul 2026 10:20:34 +0300 Subject: [PATCH 1/3] fix(logwatchers/kmsg): rate-limit parser restarts to prevent hot loop If the restarted parser's channel closes again right away (reads keep failing on the reopened /dev/kmsg), the watcher restarts in a tight loop with no delay, spinning a CPU core and flooding the logs (measured 37k restarts in 200ms). Delay the first attempt when the previous restart was less than retryDelay ago. --- .../logwatchers/kmsg/log_watcher_linux.go | 18 ++++++- .../kmsg/log_watcher_linux_test.go | 50 +++++++++++++++++++ 2 files changed, 66 insertions(+), 2 deletions(-) diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go index 9b1729f63..83818e8eb 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 50d24e39a..bc4db02bf 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. From b77c1eb03fa75e4fed4c8d25902e1127f9884067 Mon Sep 17 00:00:00 2001 From: Ciprian Hacman Date: Sat, 11 Jul 2026 10:21:43 +0300 Subject: [PATCH 2/3] fix(logwatchers/kmsg): don't block Stop() when the log channel is full The log monitor stops draining logCh before calling watcher.Stop(), so with a full channel (e.g. a kmsg burst at shutdown) watchLoop blocked on the send forever, never called tomb.Done(), and Stop() hung. Select on tomb.Stopping() alongside the send. --- .../logwatchers/kmsg/log_watcher_linux.go | 9 ++++- .../kmsg/log_watcher_linux_test.go | 39 +++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go index 83818e8eb..fa6d8cf25 100644 --- a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go +++ b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go @@ -135,9 +135,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 } } } diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go index bc4db02bf..3767b9860 100644 --- a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go +++ b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go @@ -468,6 +468,45 @@ func TestWatcherRateLimitsRestarts(t *testing.T) { } } +// 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. From a20306e6a14a5cf199d03548d749c08092002e7a Mon Sep 17 00:00:00 2001 From: Ciprian Hacman Date: Sat, 11 Jul 2026 10:24:03 +0300 Subject: [PATCH 3/3] fix(logwatchers/kmsg): don't close the old parser twice when Stop() interrupts a restart The restart path closes the failed parser before retrying. If stopping is signaled during the retry wait, watchLoop's deferred cleanup closed the same parser again, logging a spurious 'file already closed' error at shutdown. Clear the reference after closing and nil-check the defer. --- .../logwatchers/kmsg/log_watcher_linux.go | 11 ++-- .../kmsg/log_watcher_linux_test.go | 53 +++++++++++++++++++ 2 files changed, 61 insertions(+), 3 deletions(-) diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go index fa6d8cf25..c82da3811 100644 --- a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go +++ b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go @@ -94,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() @@ -110,10 +113,12 @@ 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. retryCreateParser() waits between attempts. var restarted bool diff --git a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go index 3767b9860..25c181c15 100644 --- a/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go +++ b/pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux_test.go @@ -468,6 +468,59 @@ func TestWatcherRateLimitsRestarts(t *testing.T) { } } +// 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) {