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.

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
 		}

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.