feat(log): add async logger in util

. #1000453008

Signed-off-by: slasher <shenjie1@oppo.com>

blobstore/util/log/logext.go

@@ -20,7 +20,6 @@ import (
 	"io"
 	"os"
 	"runtime"
-	"sync"
 	"sync/atomic"
 	"time"
 )
@@ -29,11 +28,24 @@ type logger struct {
 	level     int32
 	calldepth int
 	writer    atomic.Value
-	pool      sync.Pool
+}
+
+type syncWriter struct{ io.Writer }
+
+func (syncWriter) AsyncWrite(_ AsyncBuffer) (int, error) {
+	panic("log: not implement AsyncWriter")
 }
 
 type logWriter struct {
-	io.Writer
+	AsyncWriter
+	async bool
+}
+
+func newLogWriter(w io.Writer) *logWriter {
+	if aw, ok := w.(AsyncWriter); ok {
+		return &logWriter{AsyncWriter: aw, async: true}
+	}
+	return &logWriter{AsyncWriter: syncWriter{Writer: w}, async: false}
 }
 
 // New return a logger with default level Linfo.
@@ -42,13 +54,8 @@ func New(out io.Writer, calldepth int) Logger {
 	l := &logger{
 		level:     int32(Linfo),
 		calldepth: calldepth,
-		pool: sync.Pool{
-			New: func() interface{} {
-				return new(bytes.Buffer)
-			},
-		},
 	}
-	l.writer.Store(&logWriter{out})
+	l.writer.Store(newLogWriter(out))
 	return l
 }
 
@@ -78,7 +85,8 @@ func (l *logger) Outputf(id string, lvl Level, calldepth int, format string, a .
 
 func (l *logger) write(id string, lvl Level, file string, line int, s string) error {
 	now := time.Now()
-	buf := l.pool.Get().(*bytes.Buffer)
+	buffer := NewAsyncBuffer()
+	buf := buffer.Buffer()
 
 	buf.Reset()
 	l.formatOutput(buf, now, file, line, lvl)
@@ -92,9 +100,14 @@ func (l *logger) write(id string, lvl Level, file string, line int, s string) er
 	if len(s) > 0 && s[len(s)-1] != '\n' {
 		buf.WriteByte('\n')
 	}
-	out := l.writer.Load().(io.Writer)
-	_, err := out.Write(buf.Bytes())
-	l.pool.Put(buf)
+	out := l.writer.Load().(*logWriter)
+	var err error
+	if out.async {
+		_, err = out.AsyncWriter.AsyncWrite(buffer)
+	} else {
+		_, err = out.AsyncWriter.Write(buf.Bytes())
+		buffer.Release()
+	}
 	return err
 }
 
@@ -150,7 +163,7 @@ func (l *logger) GetOutputLevel() Level {
 }
 
 func (l *logger) SetOutput(w io.Writer) {
-	l.writer.Store(&logWriter{w})
+	l.writer.Store(newLogWriter(w))
 }
 
 func (l *logger) SetOutputLevel(lvl Level) {

blobstore/util/log/logger.go

@@ -0,0 +1,218 @@
+// Copyright 2025 The CubeFS Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package log
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/cubefs/cubefs/blobstore/util"
+	"gopkg.in/natefinch/lumberjack.v2"
+)
+
+var (
+	ErrWriteTimeout = errors.New("log: async write timeout")
+
+	bufferPool = sync.Pool{New: func() any { return new(bytes.Buffer) }}
+
+	_ AsyncWriter = (*AsyncLogger)(nil)
+)
+
+// AsyncWriter no-copy buffer async writer.
+type AsyncWriter interface {
+	Write(p []byte) (n int, err error)
+	// AsyncWrite should release async buffer.
+	AsyncWrite(buf AsyncBuffer) (n int, err error)
+}
+
+// AsyncBuffer reuseable buffer for asynchronous write.
+type AsyncBuffer interface {
+	Buffer() *bytes.Buffer
+	Release()
+}
+
+func NewAsyncBuffer() AsyncBuffer {
+	return &asyncBuffer{buf: bufferPool.New().(*bytes.Buffer)}
+}
+
+type asyncBuffer struct{ buf *bytes.Buffer }
+
+func (ab *asyncBuffer) Buffer() *bytes.Buffer { return ab.buf }
+func (ab *asyncBuffer) Release() {
+	if ab.buf != nil {
+		bufferPool.Put(ab.buf)
+		ab.buf = nil
+	}
+}
+
+// AsyncLogger wraps an io.Writer and performs writes asynchronously.
+// It uses a buffered channel to queue write operations and a goroutine
+// to process them in the background.
+// But QueueSize == 0, it is a synchronous logger.
+//
+// AsyncLogger must not be copied after first use. It contains sync.Once,
+// sync.WaitGroup, and channels that cannot be safely copied.
+type AsyncLogger struct {
+	// configures of lumberjack.Logger
+	Filename   string `json:"filename" yaml:"filename"`
+	MaxSize    int    `json:"maxsize" yaml:"maxsize"`
+	MaxAge     int    `json:"maxage" yaml:"maxage"`
+	MaxBackups int    `json:"maxbackups" yaml:"maxbackups"`
+	LocalTime  bool   `json:"localtime" yaml:"localtime"`
+	Compress   bool   `json:"compress" yaml:"compress"`
+
+	// QueueSize is the size of the async write queue. Default is 0.
+	// If it is 0, synchronous writing is used (no async wrapper).
+	QueueSize int `json:"queuesize" yaml:"queuesize"`
+
+	// WriteTimeout is the timeout for async writes. Default is 0, means no timeout.
+	WriteTimeout util.Duration `json:"writetimeout" yaml:"writetimeout"`
+
+	logger *lumberjack.Logger
+
+	initOnce  sync.Once
+	closeOnce sync.Once
+	closeCh   chan struct{}
+	wg        sync.WaitGroup
+	flushCh   chan chan struct{}
+	queue     chan AsyncBuffer
+}
+
+func (al *AsyncLogger) init() {
+	al.initOnce.Do(func() {
+		if al.QueueSize > 1<<20 {
+			panic(fmt.Sprintf("log: large queue size %d for logger", al.QueueSize))
+		}
+		if al.QueueSize > 0 {
+			al.queue = make(chan AsyncBuffer, al.QueueSize)
+		}
+
+		al.logger = &lumberjack.Logger{
+			Filename:   al.Filename,
+			MaxSize:    al.MaxSize,
+			MaxAge:     al.MaxAge,
+			MaxBackups: al.MaxBackups,
+			LocalTime:  al.LocalTime,
+			Compress:   al.Compress,
+		}
+
+		al.closeCh = make(chan struct{})
+		al.flushCh = make(chan chan struct{})
+
+		al.wg.Add(1)
+		go al.run()
+	})
+}
+
+func (al *AsyncLogger) run() {
+	defer al.wg.Done()
+	for {
+		select {
+		case buffer := <-al.queue:
+			al.logger.Write(buffer.Buffer().Bytes())
+			buffer.Release()
+		case <-al.closeCh: // close
+			for {
+				select {
+				case buffer := <-al.queue:
+					al.logger.Write(buffer.Buffer().Bytes())
+					buffer.Release()
+				default:
+					return
+				}
+			}
+		case done := <-al.flushCh: // flush
+			flushed := false
+			for !flushed {
+				select {
+				case buffer := <-al.queue:
+					al.logger.Write(buffer.Buffer().Bytes())
+					buffer.Release()
+				default:
+					close(done)
+					flushed = true
+				}
+			}
+		}
+	}
+}
+
+func (al *AsyncLogger) Flush() error {
+	al.init()
+	done := make(chan struct{})
+	select {
+	case al.flushCh <- done:
+		<-done
+		return nil
+	case <-al.closeCh:
+		return io.ErrClosedPipe
+	}
+}
+
+func (al *AsyncLogger) Rotate() error {
+	al.init()
+	return al.logger.Rotate()
+}
+
+func (al *AsyncLogger) Close() error {
+	al.init()
+	al.closeOnce.Do(func() { close(al.closeCh) })
+	al.wg.Wait()
+	return al.logger.Close()
+}
+
+// Write queues the data for asynchronous writing.
+// It never blocks; if the queue is full, it will drop the write.
+func (al *AsyncLogger) Write(p []byte) (n int, err error) {
+	al.init()
+	if al.queue == nil {
+		return al.logger.Write(p)
+	}
+
+	buffer := NewAsyncBuffer()
+	buffer.Buffer().Write(p) // copy
+	return al.AsyncWrite(buffer)
+}
+
+func (al *AsyncLogger) AsyncWrite(buffer AsyncBuffer) (n int, err error) {
+	al.init()
+	if al.queue == nil {
+		n, err = al.logger.Write(buffer.Buffer().Bytes())
+		buffer.Release()
+		return
+	}
+
+	var timerCh <-chan time.Time
+	if al.WriteTimeout.Duration > 0 {
+		timer := util.TimerAcquire(al.WriteTimeout.Duration)
+		timerCh = timer.C
+		defer func() { util.TimerRelease(timer) }()
+	}
+	l := buffer.Buffer().Len()
+	select {
+	case al.queue <- buffer:
+		return l, nil
+	case <-timerCh:
+		buffer.Release()
+		return 0, ErrWriteTimeout
+	case <-al.closeCh:
+		buffer.Release()
+		return 0, io.ErrClosedPipe
+	}
+}

blobstore/util/log/logger_test.go

@@ -0,0 +1,142 @@
+// Copyright 2025 The CubeFS Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package log
+
+import (
+	"io"
+	"os"
+	"path"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cubefs/cubefs/blobstore/util"
+)
+
+var tempBuffer = make([]byte, 64<<10)
+
+func tempFilename() (string, string) {
+	dir := path.Join(os.TempDir(), "log-async-"+util.Any2String(time.Now().Nanosecond()))
+	path := path.Join(dir, "log.log")
+	return dir, path
+}
+
+func TestAyncLoggerSync(t *testing.T) {
+	dir, filename := tempFilename()
+	defer os.RemoveAll(dir)
+
+	require.Panics(t, func() {
+		l := &AsyncLogger{Filename: filename, QueueSize: (1 << 20) + 1}
+		l.Write(make([]byte, 1))
+	})
+
+	logger := &AsyncLogger{Filename: filename}
+	var wg sync.WaitGroup
+	wg.Add(1000)
+	for range [1000]struct{}{} {
+		go func() {
+			logger.Write(make([]byte, 1<<10))
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	buffer := NewAsyncBuffer()
+	logger.AsyncWrite(buffer)
+	logger.Rotate()
+	logger.Flush()
+	logger.Close()
+}
+
+func TestAyncLoggerAsync(t *testing.T) {
+	dir, filename := tempFilename()
+	defer os.RemoveAll(dir)
+
+	{
+		logger := &AsyncLogger{
+			Filename:   filename,
+			MaxSize:    10,
+			MaxBackups: 3,
+			QueueSize:  1,
+		}
+		var wg sync.WaitGroup
+		wg.Add(1000)
+		for range [1000]struct{}{} {
+			go func() {
+				logger.Write(tempBuffer)
+				wg.Done()
+			}()
+		}
+		wg.Wait()
+		logger.Rotate()
+		logger.Flush()
+		logger.Close()
+	}
+	{
+		logger := &AsyncLogger{
+			Filename:   filename,
+			MaxSize:    10,
+			MaxBackups: 3,
+			QueueSize:  1024,
+		}
+		for range [10000]struct{}{} {
+			go func() {
+				logger.Write(tempBuffer)
+			}()
+		}
+		logger.Flush()
+		for range [1000]struct{}{} {
+			go func() {
+				logger.Write(tempBuffer)
+			}()
+		}
+		logger.Close()
+		require.ErrorIs(t, logger.Flush(), io.ErrClosedPipe)
+	}
+}
+
+func TestAyncLoggerTimeout(t *testing.T) {
+	dir, filename := tempFilename()
+	defer os.RemoveAll(dir)
+
+	{
+		logger := &AsyncLogger{
+			Filename:     filename,
+			MaxSize:      10,
+			MaxBackups:   3,
+			QueueSize:    4,
+			WriteTimeout: util.Duration{Duration: 100 * time.Nanosecond},
+		}
+		for range [10000]struct{}{} {
+			go func() {
+				logger.Write(tempBuffer)
+			}()
+		}
+		logger.Flush()
+		for range [1000]struct{}{} {
+			go func() {
+				logger.Write(tempBuffer)
+			}()
+		}
+		logger.Close()
+		for range [1000]struct{}{} {
+			go func() {
+				logger.Write(tempBuffer)
+			}()
+		}
+		require.ErrorIs(t, logger.Flush(), io.ErrClosedPipe)
+	}
+}

blobstore/util/util.go

@@ -22,6 +22,7 @@ import (
 	"net"
 	"os"
 	"strconv"
+	"sync"
 	"time"
 	"unsafe"
 
@@ -405,3 +406,25 @@ func String2Any(str string, pvalue interface{}) error {
 	}
 	return err
 }
+
+var poolTimer = sync.Pool{New: func() any { return time.NewTimer(time.Hour) }}
+
+func TimerAcquire(d time.Duration) *time.Timer {
+	timer := poolTimer.Get().(*time.Timer)
+	timer.Stop()
+	select {
+	case <-timer.C:
+	default:
+	}
+	timer.Reset(d)
+	return timer
+}
+
+func TimerRelease(timer *time.Timer) {
+	timer.Stop()
+	select {
+	case <-timer.C:
+	default:
+	}
+	poolTimer.Put(timer) // nolint:staticcheck
+}

blobstore/util/util_test.go

@@ -287,6 +287,19 @@ func TestStringConvert(t *testing.T) {
 	require.Error(t, String2Any("not-bool", &valBool))
 }
 
+func TestTimerPool(t *testing.T) {
+	{
+		timer := TimerAcquire(10 * time.Millisecond)
+		time.Sleep(20 * time.Millisecond)
+		TimerRelease(timer)
+	}
+	{
+		timer := TimerAcquire(10 * time.Millisecond)
+		<-timer.C
+		TimerRelease(timer)
+	}
+}
+
 func BenchmarkStrFmt2String(b *testing.B) {
 	var val int64 = 77
 	for ii := 0; ii < b.N; ii++ {