CBG-5405: fix race when InitLogging replaces memory loggers

base/logger_audit.go

@@ -187,7 +187,7 @@ func (l *AuditLogger) getAuditLoggerConfig() *AuditLoggerConfig {
 }
 
 // NewAuditLogger returns a new AuditLogger from a config.
-func NewAuditLogger(ctx context.Context, config *AuditLoggerConfig, logFilePath string, minAge int, buffer *strings.Builder, globalFields map[string]any) (*AuditLogger, error) {
+func NewAuditLogger(ctx context.Context, config *AuditLoggerConfig, logFilePath string, minAge int, globalFields map[string]any) (*AuditLogger, error) {
 	if config == nil {
 		config = &AuditLoggerConfig{}
 	}
@@ -203,7 +203,7 @@ func NewAuditLogger(ctx context.Context, config *AuditLoggerConfig, logFilePath
 		config.CollationBufferSize = new(defaultFileLoggerCollateBufferSize)
 	}
 
-	fl, err := NewFileLogger(ctx, &config.FileLoggerConfig, LevelNone, auditLogName, logFilePath, minAge, nil, buffer)
+	fl, err := NewFileLogger(ctx, &config.FileLoggerConfig, LevelNone, auditLogName, logFilePath, minAge, nil)
 	if err != nil {
 		return nil, err
 	}

base/logger_audit_test.go

@@ -9,10 +9,14 @@
 package base
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"io"
+	"log"
 	"maps"
+	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/couchbase/sync_gateway/testing/assert"
@@ -137,7 +141,7 @@ func TestAuditLoggerGlobalFields(t *testing.T) {
 			if testCase.contextFields != nil {
 				ctx = AuditLogCtx(ctx, testCase.contextFields)
 			}
-			logger, err := NewAuditLogger(ctx, &AuditLoggerConfig{FileLoggerConfig: FileLoggerConfig{Enabled: new(true)}}, tmpdir, 0, nil, testCase.globalFields)
+			logger, err := NewAuditLogger(ctx, &AuditLoggerConfig{FileLoggerConfig: FileLoggerConfig{Enabled: new(true)}}, tmpdir, 0, testCase.globalFields)
 			require.NoError(t, err)
 			auditLogger.Store(logger)
 
@@ -298,7 +302,7 @@ func BenchmarkAuditFieldwork(b *testing.B) {
 			Output:              buf,
 			CollationBufferSize: new(0),
 		},
-	}, b.TempDir(), auditMinAge, nil, map[string]any{"foo": "bar", "buzz": 1234})
+	}, b.TempDir(), auditMinAge, map[string]any{"foo": "bar", "buzz": 1234})
 	require.NoError(b, err)
 	auditLogger.Store(al)
 
@@ -355,3 +359,32 @@ func Test_expandFieldsAdditionalDataReadOnly(t *testing.T) {
 	// additionalData should not be modified
 	assert.Len(t, additionalData, 1)
 }
+
+func TestSwapAuditLoggerReplacesMemoryLogger(t *testing.T) {
+	prev := &AuditLogger{FileLogger: FileLogger{Enabled: AtomicBool{1}, closed: make(chan struct{})}}
+	prev.output = &prev.buffer
+	prev.logger = log.New(&prev.buffer, "", 0)
+	prev.log("buffered")
+	var ptr atomic.Pointer[AuditLogger]
+	ptr.Store(prev)
+
+	var output bytes.Buffer
+	logger, err := NewAuditLogger(TestCtx(t), &AuditLoggerConfig{
+		FileLoggerConfig: FileLoggerConfig{
+			Enabled:             new(true),
+			CollationBufferSize: new(0),
+			Output:              &output,
+		},
+	}, t.TempDir(), auditMinAge, nil)
+	require.NoError(t, err)
+	defer func() { assert.NoError(t, logger.Close()) }()
+	swapLogger(&ptr, logger)
+	require.Same(t, logger, ptr.Load())
+
+	// a goroutine that loaded the global logger before it was replaced still writes to prev
+	prev.log("forwarded")
+	logger.FlushBufferToLog()
+
+	// forwarded lines go straight to output, so they can precede the buffered lines
+	assert.ElementsMatch(t, []string{"buffered", "forwarded"}, strings.Fields(output.String()))
+}

base/logger_file.go

@@ -21,6 +21,7 @@ import (
 	"runtime/debug"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/pkg/errors"
@@ -89,7 +90,7 @@ type logRotationConfig struct {
 }
 
 // NewFileLogger returns a new FileLogger from a config.
-func NewFileLogger(ctx context.Context, config *FileLoggerConfig, level LogLevel, name string, logFilePath string, minAge int, defaultMaxAgeOverride *int, buffer *strings.Builder) (*FileLogger, error) {
+func NewFileLogger(ctx context.Context, config *FileLoggerConfig, level LogLevel, name string, logFilePath string, minAge int, defaultMaxAgeOverride *int) (*FileLogger, error) {
 	if config == nil {
 		config = &FileLoggerConfig{}
 	}
@@ -116,10 +117,6 @@ func NewFileLogger(ctx context.Context, config *FileLoggerConfig, level LogLevel
 	}
 	logger.Enabled.Set(*config.Enabled)
 
-	if buffer != nil {
-		logger.buffer = *buffer
-	}
-
 	// Only create the collateBuffer channel and worker if required.
 	if *config.CollationBufferSize > 1 {
 		logger.collateBuffer = make(chan string, *config.CollationBufferSize)
@@ -133,6 +130,37 @@ func NewFileLogger(ctx context.Context, config *FileLoggerConfig, level LogLevel
 	return logger, nil
 }
 
+// swapLogger stores next in ptr, and moves buffered and later writes from the old memory logger to next.
+func swapLogger[T any, P interface {
+	*T
+	fileLogger() *FileLogger
+}](ptr *atomic.Pointer[T], next P) {
+	if prev := P(ptr.Load()); prev != nil {
+		prev, next := prev.fileLogger(), next.fileLogger()
+		if prev.output == &prev.buffer {
+			// SetOutput waits for in-flight writes to prev, so the buffer is stable once it returns.
+			prev.logger.SetOutput(loggerForwarder{next})
+		}
+		next.buffer.WriteString(prev.buffer.String())
+	}
+	ptr.Store(next)
+}
+
+// fileLogger returns l. Loggers that embed FileLogger use it to give swapLogger their FileLogger.
+func (l *FileLogger) fileLogger() *FileLogger {
+	return l
+}
+
+// loggerForwarder is an io.Writer that writes each log line to a FileLogger.
+type loggerForwarder struct {
+	logger *FileLogger
+}
+
+func (f loggerForwarder) Write(p []byte) (int, error) {
+	f.logger.conditionalPrint(strings.TrimSuffix(string(p), "\n"))
+	return len(p), nil
+}
+
 func (l *FileLogger) FlushBufferToLog() {
 	// Need to clear hanging new line to avoid empty line
 	logString := strings.TrimSuffix(l.buffer.String(), "\n")

base/logger_file_test.go

@@ -11,11 +11,14 @@ licenses/APL2.txt.
 package base
 
 import (
+	"bytes"
 	"fmt"
 	"io"
 	"log"
 	"os"
 	"path/filepath"
+	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/couchbase/sync_gateway/testing/assert"
@@ -276,3 +279,51 @@ func getDirFiles(t *testing.T, dir string) []string {
 	}
 	return fileNames
 }
+
+func TestSwapFileLoggerReplacesMemoryLogger(t *testing.T) {
+	testCases := []struct {
+		name          string
+		enabled       bool
+		collationSize int
+		expected      []string
+	}{
+		{name: "uncollated", enabled: true, collationSize: 0, expected: []string{"buffered", "forwarded"}},
+		{name: "collated", enabled: true, collationSize: 10, expected: []string{"buffered", "forwarded"}},
+		{name: "disabled", enabled: false, collationSize: 0, expected: nil},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			prev := NewMemoryLogger(LevelInfo)
+			prev.log("buffered")
+			var ptr atomic.Pointer[FileLogger]
+			ptr.Store(prev)
+
+			var output bytes.Buffer
+			config := &FileLoggerConfig{
+				Enabled:             new(tc.enabled),
+				CollationBufferSize: new(tc.collationSize),
+				Output:              &output,
+			}
+			logger, err := NewFileLogger(TestCtx(t), config, LevelInfo, "test", "", 0, nil)
+			require.NoError(t, err)
+			defer func() { assert.NoError(t, logger.Close()) }()
+			swapLogger(&ptr, logger)
+			require.Same(t, logger, ptr.Load())
+
+			// a goroutine that loaded the global logger before it was replaced still writes to prev
+			prev.log("forwarded")
+			logger.FlushBufferToLog()
+
+			if logger.collateBuffer != nil {
+				flushLogMutex.Lock()
+				defer flushLogMutex.Unlock()
+				logger.collateBufferWg.Wait()
+				flushLogBuffersWaitGroup.Add(1)
+				logger.flushChan <- struct{}{}
+				flushLogBuffersWaitGroup.Wait()
+			}
+			// forwarded lines go straight to output, so they can precede the buffered lines
+			assert.ElementsMatch(t, tc.expected, strings.Fields(output.String()))
+		})
+	}
+}

base/logging_config.go

@@ -16,7 +16,6 @@ import (
 	"log"
 	"os"
 	"path/filepath"
-	"strings"
 	"time"
 
 	"github.com/pkg/errors"
@@ -92,53 +91,48 @@ func InitLogging(ctx context.Context, logFilePath string,
 		ConsolefCtx(ctx, LevelInfo, KeyNone, "Logging: Audit to %v", auditLogFilePath)
 	}
 
-	rawErrorlogger, err := NewFileLogger(ctx, error, LevelError, LevelError.String(), logFilePath, errorMinAge, nil, &errorLogger.Load().buffer)
+	rawErrorlogger, err := NewFileLogger(ctx, error, LevelError, LevelError.String(), logFilePath, errorMinAge, nil)
 	if err != nil {
 		return err
 	}
-	errorLogger.Store(rawErrorlogger)
+	swapLogger(&errorLogger, rawErrorlogger)
 
-	rawWarnLogger, err := NewFileLogger(ctx, warn, LevelWarn, LevelWarn.String(), logFilePath, warnMinAge, nil, &warnLogger.Load().buffer)
+	rawWarnLogger, err := NewFileLogger(ctx, warn, LevelWarn, LevelWarn.String(), logFilePath, warnMinAge, nil)
 	if err != nil {
 		return err
 	}
-	warnLogger.Store(rawWarnLogger)
+	swapLogger(&warnLogger, rawWarnLogger)
 
-	rawInfoLogger, err := NewFileLogger(ctx, info, LevelInfo, LevelInfo.String(), logFilePath, infoMinAge, nil, &infoLogger.Load().buffer)
+	rawInfoLogger, err := NewFileLogger(ctx, info, LevelInfo, LevelInfo.String(), logFilePath, infoMinAge, nil)
 	if err != nil {
 		return err
 	}
-	infoLogger.Store(rawInfoLogger)
+	swapLogger(&infoLogger, rawInfoLogger)
 
-	rawDebugLogger, err := NewFileLogger(ctx, debug, LevelDebug, LevelDebug.String(), logFilePath, debugMinAge, nil, &debugLogger.Load().buffer)
+	rawDebugLogger, err := NewFileLogger(ctx, debug, LevelDebug, LevelDebug.String(), logFilePath, debugMinAge, nil)
 	if err != nil {
 		return err
 	}
-	debugLogger.Store(rawDebugLogger)
+	swapLogger(&debugLogger, rawDebugLogger)
 
-	rawTraceLogger, err := NewFileLogger(ctx, trace, LevelTrace, LevelTrace.String(), logFilePath, traceMinAge, nil, &traceLogger.Load().buffer)
+	rawTraceLogger, err := NewFileLogger(ctx, trace, LevelTrace, LevelTrace.String(), logFilePath, traceMinAge, nil)
 	if err != nil {
 		return err
 	}
-	traceLogger.Store(rawTraceLogger)
+	swapLogger(&traceLogger, rawTraceLogger)
 
 	// Since there is no level checking in the stats logging, use LevelNone for the level.
-	rawStatsLogger, err := NewFileLogger(ctx, stats, LevelNone, "stats", logFilePath, statsMinAge, new(statsDefaultMaxAgeOverride), &statsLogger.Load().buffer)
+	rawStatsLogger, err := NewFileLogger(ctx, stats, LevelNone, "stats", logFilePath, statsMinAge, new(statsDefaultMaxAgeOverride))
 	if err != nil {
 		return err
 	}
-	statsLogger.Store(rawStatsLogger)
+	swapLogger(&statsLogger, rawStatsLogger)
 
-	var auditLoggerBuffer *strings.Builder
-	prevAuditLogger := auditLogger.Load()
-	if prevAuditLogger != nil {
-		auditLoggerBuffer = &prevAuditLogger.buffer
-	}
-	rawAuditLogger, err := NewAuditLogger(ctx, audit, auditLogFilePath, auditMinAge, auditLoggerBuffer, auditLogGlobalFields)
+	rawAuditLogger, err := NewAuditLogger(ctx, audit, auditLogFilePath, auditMinAge, auditLogGlobalFields)
 	if err != nil {
 		return err
 	}
-	auditLogger.Store(rawAuditLogger)
+	swapLogger(&auditLogger, rawAuditLogger)
 
 	// Pick up any console log level/key changes for the external loggers too
 	updateExternalLoggers()

base/logging_test.go

@@ -78,7 +78,7 @@ func TestLogRotationInterval(t *testing.T) {
 	t.Logf("countBefore: %d", countBefore)
 
 	ctx := TestCtx(t)
-	fl, err := NewFileLogger(ctx, config, LevelTrace, "test", logPath, 0, nil, nil)
+	fl, err := NewFileLogger(ctx, config, LevelTrace, "test", logPath, 0, nil)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, fl.Close())

rest/audit_test.go

@@ -6,8 +6,6 @@
 // software will be governed by the Apache License, Version 2.0, included in
 // the file licenses/APL2.txt.
 
-//go:build !race
-
 package rest
 
 import (
@@ -1627,9 +1625,16 @@ func createAuditLoggingRestTester(t *testing.T) *RestTester {
 				},
 				Console: &base.ConsoleLoggerConfig{
 					FileLoggerConfig: base.FileLoggerConfig{
-						Enabled: new(true),
+						Enabled:             new(true),
+						CollationBufferSize: new(0), // avoid data race in collation with FlushLogBuffers test code
 					},
 				},
+				Error: &base.FileLoggerConfig{
+					CollationBufferSize: new(0), // avoid data race in collation with FlushLogBuffers test code
+				},
+				Warn: &base.FileLoggerConfig{
+					CollationBufferSize: new(0), // avoid data race in collation with FlushLogBuffers test code
+				},
 				Info: &base.FileLoggerConfig{
 					Enabled:             new(false),
 					CollationBufferSize: new(0), // avoid data race in collation with FlushLogBuffers test code
@@ -1642,6 +1647,9 @@ func createAuditLoggingRestTester(t *testing.T) *RestTester {
 					Enabled:             new(false),
 					CollationBufferSize: new(0), // avoid data race in collation with FlushLogBuffers test code
 				},
+				Stats: &base.FileLoggerConfig{
+					CollationBufferSize: new(0), // avoid data race in collation with FlushLogBuffers test code
+				},
 			}
 			require.NoError(t, config.SetupAndValidateLogging(base.TestCtx(t)))
 		},

rest/cluster_compat_audit_test.go

@@ -6,8 +6,6 @@
 // software will be governed by the Apache License, Version 2.0, included in
 // the file licenses/APL2.txt.
 
-//go:build !race
-
 package rest
 
 import (