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.
This commit is contained in:
Ciprian Hacman
2026-07-11 10:28:27 +03:00
parent b77c1eb03f
commit a20306e6a1
2 changed files with 61 additions and 3 deletions
@@ -94,8 +94,11 @@ func (k *kernelLogWatcher) Stop() {
func (k *kernelLogWatcher) watchLoop() { func (k *kernelLogWatcher) watchLoop() {
kmsgs := k.kmsgParser.Parse() kmsgs := k.kmsgParser.Parse()
defer func() { defer func() {
if err := k.kmsgParser.Close(); err != nil { // kmsgParser is nil when stopping interrupted a restart.
klog.Errorf("Failed to close kmsg parser: %v", err) if k.kmsgParser != nil {
if err := k.kmsgParser.Close(); err != nil {
klog.Errorf("Failed to close kmsg parser: %v", err)
}
} }
close(k.logCh) close(k.logCh)
k.tomb.Done() k.tomb.Done()
@@ -110,10 +113,12 @@ func (k *kernelLogWatcher) watchLoop() {
if !ok { if !ok {
klog.Error("Kmsg channel closed, attempting to restart kmsg parser") 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 { if err := k.kmsgParser.Close(); err != nil {
klog.Errorf("Failed to close kmsg parser: %v", err) klog.Errorf("Failed to close kmsg parser: %v", err)
} }
k.kmsgParser = nil
// Try to restart. retryCreateParser() waits between attempts. // Try to restart. retryCreateParser() waits between attempts.
var restarted bool var restarted bool
@@ -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 // TestStopDoesNotDeadlockWhenLogChannelFull verifies that Stop() returns even
// when logCh is full and nobody is draining it. // when logCh is full and nobody is draining it.
func TestStopDoesNotDeadlockWhenLogChannelFull(t *testing.T) { func TestStopDoesNotDeadlockWhenLogChannelFull(t *testing.T) {