pkg/util/log: Implement handling for oversized log messages in bufferedSink

This update introduces functionality to gracefully drop oversized log messages that exceed the max buffer size without crashing the CRDB process when exit-on-error is enabled. A warning is logged to indicate that the message was dropped, ensuring that the logging system remains stable.

Part of: CRDB-53951
Epic: CRDB-56325
Release note: None

Author: Abhinav1299

pkg/util/log/buffered_sink.go

@@ -221,6 +221,12 @@ func (bs *bufferedSink) output(b []byte, opts sinkOutputOptions) error {
 
 	var errC chan error
 
+	// Variables to log an oversized message warning after releasing the lock.
+	// We can't call Ops.Warningf() while holding bs.mu because that would
+	// cause a deadlock (the warning message would try to log through this same sink).
+	var logOversizedWarning bool
+	var maxSize uint64
+
 	err := func() error {
 		bs.mu.Lock()
 		defer bs.mu.Unlock()
@@ -229,6 +235,16 @@ func (bs *bufferedSink) output(b []byte, opts sinkOutputOptions) error {
 		if err != nil {
 			// Release the msg buffer, since our append failed.
 			putBuffer(msg)
+			// If the message is too large to fit in the buffer, we log a warning
+			// but don't return an error. This prevents the server from crashing
+			// due to exit-on-error when a single log message exceeds max-buffer-size.
+			// Instead, we gracefully drop the oversized message.
+			if errors.Is(err, errMsgTooLarge) {
+				// Set flag to log warning after releasing the lock (to avoid recursion/deadlock)
+				logOversizedWarning = true
+				maxSize = bs.mu.buf.maxSizeBytes
+				return nil // Don't propagate the error - handle gracefully
+			}
 			return err
 		}
 
@@ -281,6 +297,14 @@ func (bs *bufferedSink) output(b []byte, opts sinkOutputOptions) error {
 		}
 		return nil
 	}()
+
+	// Log the oversized message warning after releasing the lock to avoid deadlock.
+	// This warning will be logged through the normal logging system and will appear
+	// in the log files.
+	if logOversizedWarning {
+		Ops.Warningf(context.Background(), "dropped log message because it exceeds max-buffer-size (%d bytes) for sink %T", maxSize, bs.child)
+	}
+
 	if err != nil {
 		return err
 	}

pkg/util/log/buffered_sink_test.go

@@ -630,3 +630,46 @@ func (t *testWaitGroupSink) exitCode() exit.Code {
 }
 
 var _ logSink = (*testWaitGroupSink)(nil)
+
+// TestBufferedSinkOversizedMessage tests that oversized messages (larger than
+// maxBufferSize) are dropped gracefully with a warning, rather than causing
+// the process to crash when exit-on-error is enabled.
+func TestBufferedSinkOversizedMessage(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer Scope(t).Close(t)
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mock := NewMockLogSink(ctrl)
+
+	// Set up a buffered sink with a small max buffer size.
+	maxBufferSize := uint64(10)
+	sink := newBufferedSink(mock, noMaxStaleness, noSizeTrigger, maxBufferSize, false /* crashOnAsyncFlushErr */, nil)
+	closer := newBufferedSinkCloser()
+	sink.Start(closer)
+	defer func() { require.NoError(t, closer.Close(defaultCloserTimeout)) }()
+
+	// Create a message that exceeds the max buffer size.
+	oversizedMessage := bytes.Repeat([]byte("x"), int(maxBufferSize)+10)
+
+	// We expect that the oversized message will NOT be sent to the child sink.
+	// Instead, a warning message will be logged (via Ops.Warningf).
+	// The warning message itself is small and should succeed.
+	mock.EXPECT().
+		output(gomock.Any(), gomock.Any()).
+		Do(func(b []byte, opts sinkOutputOptions) {
+			// Verify this is the warning message, not the oversized message.
+			msg := string(b)
+			require.Contains(t, msg, "dropped log message because it exceeds max-buffer-size")
+			require.NotContains(t, msg, string(oversizedMessage))
+		}).
+		MaxTimes(1) // The warning may or may not be logged through this same sink
+
+	// Attempt to output the oversized message. This should succeed (return nil)
+	// and not cause the process to crash.
+	err := sink.output(oversizedMessage, sinkOutputOptions{})
+	require.NoError(t, err, "oversized message should be handled gracefully without returning an error")
+
+	// Give a moment for any async warning to be logged.
+	time.Sleep(100 * time.Millisecond)
+}