merge: access log rotation and enhancements

internal/net/gphttp/accesslog/access_logger.go

@@ -2,59 +2,99 @@ package accesslog
 
 import (
 	"bufio"
-	"bytes"
 	"io"
 	"net/http"
 	"sync"
 	"time"
 
+	"github.com/rs/zerolog"
 	"github.com/yusing/go-proxy/internal/gperr"
 	"github.com/yusing/go-proxy/internal/logging"
 	"github.com/yusing/go-proxy/internal/task"
+	"github.com/yusing/go-proxy/internal/utils/synk"
+	"golang.org/x/time/rate"
 )
 
 type (
 	AccessLogger struct {
-		task     *task.Task
-		cfg      *Config
-		io       AccessLogIO
-		buffered *bufio.Writer
+		task          *task.Task
+		cfg           *Config
+		io            AccessLogIO
+		buffered      *bufio.Writer
+		supportRotate bool
+
+		lineBufPool *synk.BytesPool // buffer pool for formatting a single log line
+
+		errRateLimiter *rate.Limiter
+
+		logger zerolog.Logger
 
-		lineBufPool sync.Pool // buffer pool for formatting a single log line
 		Formatter
 	}
 
 	AccessLogIO interface {
-		io.ReadWriteCloser
-		io.ReadWriteSeeker
-		io.ReaderAt
+		io.Writer
 		sync.Locker
 		Name() string // file name or path
-		Truncate(size int64) error
 	}
 
 	Formatter interface {
-		// Format writes a log line to line without a trailing newline
-		Format(line *bytes.Buffer, req *http.Request, res *http.Response)
-		SetGetTimeNow(getTimeNow func() time.Time)
+		// AppendLog appends a log line to line with or without a trailing newline
+		AppendLog(line []byte, req *http.Request, res *http.Response) []byte
 	}
 )
 
-func NewAccessLogger(parent task.Parent, io AccessLogIO, cfg *Config) *AccessLogger {
+const MinBufferSize = 4 * kilobyte
+
+const (
+	flushInterval  = 30 * time.Second
+	rotateInterval = time.Hour
+)
+
+func NewAccessLogger(parent task.Parent, cfg *Config) (*AccessLogger, error) {
+	var ios []AccessLogIO
+
+	if cfg.Stdout {
+		ios = append(ios, stdoutIO)
+	}
+
+	if cfg.Path != "" {
+		io, err := newFileIO(cfg.Path)
+		if err != nil {
+			return nil, err
+		}
+		ios = append(ios, io)
+	}
+
+	if len(ios) == 0 {
+		return nil, nil
+	}
+
+	return NewAccessLoggerWithIO(parent, NewMultiWriter(ios...), cfg), nil
+}
+
+func NewMockAccessLogger(parent task.Parent, cfg *Config) *AccessLogger {
+	return NewAccessLoggerWithIO(parent, NewMockFile(), cfg)
+}
+
+func NewAccessLoggerWithIO(parent task.Parent, io AccessLogIO, cfg *Config) *AccessLogger {
 	if cfg.BufferSize == 0 {
 		cfg.BufferSize = DefaultBufferSize
 	}
-	if cfg.BufferSize < 4096 {
-		cfg.BufferSize = 4096
+	if cfg.BufferSize < MinBufferSize {
+		cfg.BufferSize = MinBufferSize
 	}
 	l := &AccessLogger{
-		task:     parent.Subtask("accesslog"),
-		cfg:      cfg,
-		io:       io,
-		buffered: bufio.NewWriterSize(io, cfg.BufferSize),
+		task:           parent.Subtask("accesslog."+io.Name(), true),
+		cfg:            cfg,
+		io:             io,
+		buffered:       bufio.NewWriterSize(io, cfg.BufferSize),
+		lineBufPool:    synk.NewBytesPool(1024, synk.DefaultMaxBytes),
+		errRateLimiter: rate.NewLimiter(rate.Every(time.Second), 1),
+		logger:         logging.With().Str("file", io.Name()).Logger(),
 	}
 
-	fmt := CommonFormatter{cfg: &l.cfg.Fields, GetTimeNow: time.Now}
+	fmt := CommonFormatter{cfg: &l.cfg.Fields}
 	switch l.cfg.Format {
 	case FormatCommon:
 		l.Formatter = &fmt
@@ -66,14 +106,19 @@ func NewAccessLogger(parent task.Parent, io AccessLogIO, cfg *Config) *AccessLog
 		panic("invalid access log format")
 	}
 
-	l.lineBufPool.New = func() any {
-		return bytes.NewBuffer(make([]byte, 0, 1024))
+	if _, ok := l.io.(supportRotate); ok {
+		l.supportRotate = true
 	}
+
 	go l.start()
 	return l
 }
 
-func (l *AccessLogger) checkKeep(req *http.Request, res *http.Response) bool {
+func (l *AccessLogger) Config() *Config {
+	return l.cfg
+}
+
+func (l *AccessLogger) shouldLog(req *http.Request, res *http.Response) bool {
 	if !l.cfg.Filters.StatusCodes.CheckKeep(req, res) ||
 		!l.cfg.Filters.Method.CheckKeep(req, res) ||
 		!l.cfg.Filters.Headers.CheckKeep(req, res) ||
@@ -84,53 +129,63 @@ func (l *AccessLogger) checkKeep(req *http.Request, res *http.Response) bool {
 }
 
 func (l *AccessLogger) Log(req *http.Request, res *http.Response) {
-	if !l.checkKeep(req, res) {
+	if !l.shouldLog(req, res) {
 		return
 	}
 
-	line := l.lineBufPool.Get().(*bytes.Buffer)
-	line.Reset()
+	line := l.lineBufPool.Get()
 	defer l.lineBufPool.Put(line)
-	l.Formatter.Format(line, req, res)
-	line.WriteRune('\n')
-	l.write(line.Bytes())
+	line = l.Formatter.AppendLog(line, req, res)
+	if line[len(line)-1] != '\n' {
+		line = append(line, '\n')
+	}
+	l.lockWrite(line)
 }
 
 func (l *AccessLogger) LogError(req *http.Request, err error) {
 	l.Log(req, &http.Response{StatusCode: http.StatusInternalServerError, Status: err.Error()})
 }
 
-func (l *AccessLogger) Config() *Config {
-	return l.cfg
+func (l *AccessLogger) ShouldRotate() bool {
+	return l.cfg.Retention.IsValid() && l.supportRotate
 }
 
-func (l *AccessLogger) Rotate() error {
-	if l.cfg.Retention == nil {
-		return nil
+func (l *AccessLogger) Rotate() (result *RotateResult, err error) {
+	if !l.ShouldRotate() {
+		return nil, nil
 	}
+
 	l.io.Lock()
 	defer l.io.Unlock()
 
-	return l.rotate()
+	return rotateLogFile(l.io.(supportRotate), l.cfg.Retention)
 }
 
 func (l *AccessLogger) handleErr(err error) {
-	gperr.LogError("failed to write access log", err)
+	if l.errRateLimiter.Allow() {
+		gperr.LogError("failed to write access log", err)
+	} else {
+		gperr.LogError("too many errors, stopping access log", err)
+		l.task.Finish(err)
+	}
 }
 
 func (l *AccessLogger) start() {
 	defer func() {
+		defer l.task.Finish(nil)
+		defer l.close()
 		if err := l.Flush(); err != nil {
 			l.handleErr(err)
 		}
-		l.close()
-		l.task.Finish(nil)
 	}()
 
 	// flushes the buffer every 30 seconds
 	flushTicker := time.NewTicker(30 * time.Second)
 	defer flushTicker.Stop()
 
+	rotateTicker := time.NewTicker(rotateInterval)
+	defer rotateTicker.Stop()
+
 	for {
 		select {
 		case <-l.task.Context().Done():
@@ -139,6 +194,18 @@ func (l *AccessLogger) start() {
 			if err := l.Flush(); err != nil {
 				l.handleErr(err)
 			}
+		case <-rotateTicker.C:
+			if !l.ShouldRotate() {
+				continue
+			}
+			l.logger.Info().Msg("rotating access log file")
+			if res, err := l.Rotate(); err != nil {
+				l.handleErr(err)
+			} else if res != nil {
+				res.Print(&l.logger)
+			} else {
+				l.logger.Info().Msg("no rotation needed")
+			}
 		}
 	}
 }
@@ -150,18 +217,20 @@ func (l *AccessLogger) Flush() error {
 }
 
 func (l *AccessLogger) close() {
-	l.io.Lock()
-	defer l.io.Unlock()
-	l.io.Close()
+	if r, ok := l.io.(io.Closer); ok {
+		l.io.Lock()
+		defer l.io.Unlock()
+		r.Close()
+	}
 }
 
-func (l *AccessLogger) write(data []byte) {
+func (l *AccessLogger) lockWrite(data []byte) {
 	l.io.Lock() // prevent concurrent write, i.e. log rotation, other access loggers
 	_, err := l.buffered.Write(data)
 	l.io.Unlock()
 	if err != nil {
 		l.handleErr(err)
 	} else {
-		logging.Debug().Msg("access log flushed to " + l.io.Name())
+		logging.Trace().Msg("access log flushed to " + l.io.Name())
 	}
 }