fix: prevent async logging from blocking shutdown (#53360)

issue: #53358 - async logging backpressure: use bounded channel enqueue waits for all levels; higher-level logs get a longer best-effort wait and may still be dropped when the sink is stuck - logger shutdown: bound Stop waiting by the configured timeout even when the underlying sink is blocked - logging configuration: clarify that nonDroppableLevel does not guarantee lossless logging --------- Signed-off-by: chyezh <[email protected]>
2026-10-12 02:19:10 +00:00 · 2026-09-19 02:47:36 +08:00
parent c51f08efa0
commit 08bef73f78
4 changed files with 195 additions and 22 deletions
@@ -78,13 +78,17 @@ type Config struct {
 	// AsyncWriteFlushInterval is the interval to flush the logs
 	AsyncWriteFlushInterval time.Duration `toml:"async-write-flush-interval" json:"async-write-flush-interval"`

-	// AsyncWriteDroppedTimeout is the timeout to drop the write request if the buffer is full
+	// AsyncWriteDroppedTimeout is the maximum time a log below AsyncWriteNonDroppableLevel waits for queue space.
+	// The log is dropped if the queue remains full when the timeout expires.
 	AsyncWriteDroppedTimeout time.Duration `toml:"async-write-dropped-timeout" json:"async-write-dropped-timeout"`

-	// AsyncWriteNonDroppableLevel is the level that will not be dropped when the buffer is full
+	// AsyncWriteNonDroppableLevel is the level at which a log uses a best-effort extended enqueue wait.
+	// At or above this level, a log waits at most max(AsyncWriteDroppedTimeout, AsyncWriteStopTimeout)
+	// for queue space and can still be dropped. The name is retained for compatibility.
 	AsyncWriteNonDroppableLevel string `toml:"async-write-non-droppable-level" json:"async-write-non-droppable-level"`

-	// AsyncWriteStopTimeout is the timeout to stop the async write
+	// AsyncWriteStopTimeout is the maximum time to wait for async logging to stop. It also contributes
+	// to the maximum enqueue wait for logs at or above AsyncWriteNonDroppableLevel.
 	AsyncWriteStopTimeout time.Duration `toml:"async-write-stop-timeout" json:"async-write-stop-timeout"`

 	// AsyncWritePendingLength is the maximum number of pending write requests, the exceeded log operation will be dropped
@@ -159,22 +159,32 @@ func (s *asyncTextIOCore) WriteWithCEntry(ent CEntry) {
 func (s *asyncTextIOCore) write(ent entryItem) {
 	length := ent.buf.Len()
 	if length == 0 {
+		ent.buf.Free()
 		return
 	}
-	var writeDroppedTimeout <-chan time.Time
-	if ent.level < s.nonDroppableLevel {
-		writeDroppedTimeout = time.After(s.writeDroppedTimeout)
+
+	timeout := s.writeDroppedTimeout
+	if ent.level >= s.nonDroppableLevel {
+		// Give important entries at least as much time to wait for space, but never block
+		// indefinitely when the underlying writer is stuck.
+		timeout = max(timeout, s.stopTimeout)
 	}
+
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
 	select {
 	case s.pending <- ent:
 		metrics.LoggingPendingWriteTotal.Inc()
-	case <-writeDroppedTimeout:
-		metrics.LoggingDroppedWriteTotal.Inc()
-		// drop the entry if the write is dropped due to timeout
-		ent.buf.Free()
+	case <-timer.C:
+		s.dropEntry(ent)
 	}
 }

+func (s *asyncTextIOCore) dropEntry(ent entryItem) {
+	metrics.LoggingDroppedWriteTotal.Inc()
+	ent.buf.Free()
+}
+
 type CEntryTextIOCore interface {
 	WriteWithCEntry(ent CEntry)
 }
@@ -284,7 +294,14 @@ func (s *asyncTextIOCore) flushAllPendingWrites(done chan struct{}) {
 	}
 }

+// Stop requests shutdown and waits for at most the configured timeout. If the
+// underlying writer is blocked, its background goroutine may outlive Stop.
 func (s *asyncTextIOCore) Stop() {
 	s.notifier.Cancel()
-	s.notifier.BlockUntilFinish()
+	timer := time.NewTimer(s.stopTimeout)
+	defer timer.Stop()
+	select {
+	case <-s.notifier.FinishChan():
+	case <-timer.C:
+	}
 }
@@ -25,8 +25,13 @@ import (
 	"time"

 	"github.com/cockroachdb/errors"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
+
+	"github.com/milvus-io/milvus/pkg/v3/metrics"
 )

 func TestAsyncBufferedWriteSyncer(t *testing.T) {
@@ -84,3 +89,149 @@ func (s *blockWriter) Sync() error {
 	}
 	return nil
 }
+
+func TestAsyncBufferedWriteSyncerHighLevelWaitsLongerForSpace(t *testing.T) {
+	writer := newBlockingWriteSyncer()
+	defer writer.unblock()
+	syncer := newAsyncTextIOCoreForBlockedWriter(writer, 20*time.Millisecond, 500*time.Millisecond)
+
+	require.NoError(t, syncer.Write(zapcore.Entry{Level: zap.InfoLevel, Message: "writing"}, nil))
+	writer.waitUntilBlocked(t)
+	require.NoError(t, syncer.Write(zapcore.Entry{Level: zap.InfoLevel, Message: "pending"}, nil))
+
+	droppedBefore := testutil.ToFloat64(metrics.LoggingDroppedWriteTotal)
+	done := make(chan error, 1)
+	go func() {
+		done <- syncer.Write(zapcore.Entry{Level: zap.ErrorLevel, Message: "important"}, nil)
+	}()
+
+	// An ordinary entry gives up after droppedTimeout, while the Error entry
+	// continues waiting for space rather than evicting the pending entry.
+	require.NoError(t, syncer.Write(zapcore.Entry{Level: zap.InfoLevel, Message: "droppable"}, nil))
+	assert.Equal(t, droppedBefore+1, testutil.ToFloat64(metrics.LoggingDroppedWriteTotal))
+	select {
+	case <-done:
+		require.FailNow(t, "error log returned before queue space became available")
+	default:
+	}
+
+	writer.unblock()
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		require.FailNow(t, "error log did not resume after queue space became available")
+	}
+	assert.Equal(t, droppedBefore+1, testutil.ToFloat64(metrics.LoggingDroppedWriteTotal))
+	syncer.Stop()
+}
+
+func TestAsyncBufferedWriteSyncerHighLevelWriteIsBounded(t *testing.T) {
+	writer := newBlockingWriteSyncer()
+	defer writer.unblock()
+	syncer := newAsyncTextIOCoreForBlockedWriter(writer, 10*time.Millisecond, 50*time.Millisecond)
+
+	require.NoError(t, syncer.Write(zapcore.Entry{Level: zap.InfoLevel, Message: "writing"}, nil))
+	writer.waitUntilBlocked(t)
+	require.NoError(t, syncer.Write(zapcore.Entry{Level: zap.InfoLevel, Message: "pending"}, nil))
+
+	droppedBefore := testutil.ToFloat64(metrics.LoggingDroppedWriteTotal)
+	done := make(chan error, 1)
+	go func() {
+		done <- syncer.Write(zapcore.Entry{Level: zap.ErrorLevel, Message: "important"}, nil)
+	}()
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		require.FailNow(t, "error log blocked indefinitely on a full pending queue")
+	}
+	assert.Equal(t, droppedBefore+1, testutil.ToFloat64(metrics.LoggingDroppedWriteTotal))
+	pending := <-syncer.pending
+	assert.Contains(t, pending.buf.String(), "pending")
+	syncer.pending <- pending
+
+	writer.unblock()
+	syncer.Stop()
+}
+
+func TestAsyncBufferedWriteSyncerStopTimesOutOnBlockedWriter(t *testing.T) {
+	writer := newBlockingWriteSyncer()
+	defer writer.unblock()
+	syncer := newAsyncTextIOCoreForBlockedWriter(writer, 10*time.Millisecond, 50*time.Millisecond)
+
+	require.NoError(t, syncer.Write(zapcore.Entry{Level: zap.InfoLevel, Message: "writing"}, nil))
+	writer.waitUntilBlocked(t)
+
+	done := make(chan struct{})
+	go func() {
+		syncer.Stop()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		require.FailNow(t, "async logger stop blocked on the underlying writer")
+	}
+
+	writer.unblock()
+	select {
+	case <-syncer.notifier.FinishChan():
+	case <-time.After(time.Second):
+		require.FailNow(t, "async logger did not finish after the writer was unblocked")
+	}
+}
+
+func newAsyncTextIOCoreForBlockedWriter(writer zapcore.WriteSyncer, droppedTimeout, stopTimeout time.Duration) *asyncTextIOCore {
+	return NewAsyncTextIOCore(
+		&Config{
+			Format:                      "text",
+			AsyncWriteFlushInterval:     time.Hour,
+			AsyncWriteDroppedTimeout:    droppedTimeout,
+			AsyncWriteNonDroppableLevel: zap.ErrorLevel.String(),
+			AsyncWriteStopTimeout:       stopTimeout,
+			AsyncWritePendingLength:     1,
+			AsyncWriteBufferSize:        1,
+			AsyncWriteMaxBytesPerLog:    1024,
+		},
+		writer,
+		zap.DebugLevel,
+	)
+}
+
+type blockingWriteSyncer struct {
+	started     chan struct{}
+	release     chan struct{}
+	startedOnce sync.Once
+	releaseOnce sync.Once
+}
+
+func newBlockingWriteSyncer() *blockingWriteSyncer {
+	return &blockingWriteSyncer{
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+	}
+}
+
+func (s *blockingWriteSyncer) Write(p []byte) (int, error) {
+	s.startedOnce.Do(func() { close(s.started) })
+	<-s.release
+	return len(p), nil
+}
+
+func (s *blockingWriteSyncer) Sync() error {
+	return nil
+}
+
+func (s *blockingWriteSyncer) waitUntilBlocked(t *testing.T) {
+	t.Helper()
+	select {
+	case <-s.started:
+	case <-time.After(time.Second):
+		require.FailNow(t, "writer was not called")
+	}
+}
+
+func (s *blockingWriteSyncer) unblock() {
+	s.releaseOnce.Do(func() { close(s.release) })
+}
@@ -2251,10 +2251,9 @@ Faster the logging writes can be seen by the underlying file system.`,
 		Key:          "log.asyncWrite.droppedTimeout",
 		DefaultValue: "100ms",
 		Version:      "2.6.7",
-		Doc: `The timeout to drop the write operation if the buffer is full.
-Once the underlying buffered writer is blocked or too slow and
-the pending length is larger than the pending length threshold,
-the new incoming write operation will be dropped if it exceeds the timeout.`,
+		Doc: `The maximum time a log below nonDroppableLevel waits for queue space.
+If the underlying buffered writer is blocked or too slow and the pending queue
+remains full when this timeout expires, the incoming log is dropped.`,
 		Export: false,
 	}
 	l.AsyncWriteDroppedTimeout.Init(base.mgr)
@@ -2263,9 +2262,10 @@ the new incoming write operation will be dropped if it exceeds the timeout.`,
 		Key:          "log.asyncWrite.nonDroppableLevel",
 		DefaultValue: "error",
 		Version:      "2.6.7",
-		Doc: `The level that will not be dropped when the buffer is full.
-Once the level is greater or equal to the non-droppable level, 
-the write operation will not be dropped because the buffer is full`,
+		Doc: `The level at which a log uses a best-effort extended wait for queue space.
+At or above this level, a log waits at most max(droppedTimeout, stopTimeout).
+The name is retained for compatibility; it does not guarantee delivery, and these
+logs are dropped if the queue remains full when that timeout expires.`,
 		Export: false,
 	}
 	l.AsyncWriteNonDroppableLevel.Init(base.mgr)
@@ -2274,10 +2274,11 @@ the write operation will not be dropped because the buffer is full`,
 		Key:          "log.asyncWrite.stopTimeout",
 		DefaultValue: "1s",
 		Version:      "2.6.7",
-		Doc: `The timeout to stop the async write.
-When the milvus is on shutdown, 
-the async writer of logger will try to flush all the pending write operations 
-to the underlying file system until reaching the timeout.`,
+		Doc: `The maximum time to wait for async logging to stop. It also contributes to
+the maximum enqueue wait for logs at or above nonDroppableLevel: their limit is
+max(droppedTimeout, stopTimeout).
+When Milvus shuts down, the async logger tries to flush pending writes to the
+underlying file system until this timeout expires.`,
 		Export: false,
 	}
 	l.AsyncWriteStopTimeout.Init(base.mgr)