Fix kmsg watcher restart and shutdown issues

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,16 +140,26 @@ 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
 			}
 		}
 	}
 }
 
 // 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
 		}

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.