fix: serialize logger initialization and publish the console target safely

Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-09-08 12:58:21 +08:00
parent 8e2392e48f
commit 079ebb1926
3 changed files with 69 additions and 18 deletions
+4 -4
View File
@@ -395,9 +395,9 @@ func consoleLogIf(ctx context.Context, subsystem string, err error, errKind ...a
if err == nil { if err == nil {
return return
} }
if consoleTgt != nil { if console := consoleTgt.Load(); console != nil {
entry := errToEntry(ctx, subsystem, err, errKind...) entry := errToEntry(ctx, subsystem, err, errKind...)
consoleTgt.Send(ctx, entry) (*console).Send(ctx, entry)
} }
} }
@@ -423,8 +423,8 @@ func sendLog(ctx context.Context, entry log.Entry) {
// Iterate over all logger targets to send the log entry // Iterate over all logger targets to send the log entry
for _, t := range systemTgts { for _, t := range systemTgts {
if err := t.Send(ctx, entry); err != nil { if err := t.Send(ctx, entry); err != nil {
if consoleTgt != nil { // Sending to the console never fails if console := consoleTgt.Load(); console != nil { // Sending to the console never fails
consoleTgt.Send(ctx, errToEntry(ctx, "logging", fmt.Errorf("unable to send log event to Logger target (%s): %v", t.String(), err), entry.Level)) (*console).Send(ctx, errToEntry(ctx, "logging", fmt.Errorf("unable to send log event to Logger target (%s): %v", t.String(), err), entry.Level))
} }
} }
} }
+10 -5
View File
@@ -22,6 +22,7 @@ import (
"fmt" "fmt"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"github.com/minio/minio/internal/logger/target/http" "github.com/minio/minio/internal/logger/target/http"
"github.com/minio/minio/internal/logger/target/kafka" "github.com/minio/minio/internal/logger/target/kafka"
@@ -106,7 +107,10 @@ var (
auditTargets = newTargetsList() auditTargets = newTargetsList()
// This is always set represent /dev/console target // This is always set represent /dev/console target
consoleTgt Target consoleTgt atomic.Pointer[Target]
// Init may emit logs, so serialize registration without holding the list lock.
systemTargetInitMu sync.Mutex
) )
// SystemTargets returns active targets. // SystemTargets returns active targets.
@@ -153,6 +157,9 @@ func CurrentStats() map[string]types.TargetStats {
// registered is a no-op, so callers may safely re-register // registered is a no-op, so callers may safely re-register
// long-lived targets such as the console logger. // long-lived targets such as the console logger.
func AddSystemTarget(ctx context.Context, t Target) error { func AddSystemTarget(ctx context.Context, t Target) error {
systemTargetInitMu.Lock()
defer systemTargetInitMu.Unlock()
if systemTargets.contains(t) { if systemTargets.contains(t) {
return nil return nil
} }
@@ -161,10 +168,8 @@ func AddSystemTarget(ctx context.Context, t Target) error {
return err return err
} }
if consoleTgt == nil { if t.Type() == types.TargetConsole {
if t.Type() == types.TargetConsole { consoleTgt.CompareAndSwap(nil, &t)
consoleTgt = t
}
} }
systemTargets.addIfAbsent(t) systemTargets.addIfAbsent(t)
+55 -9
View File
@@ -19,6 +19,7 @@ package logger
import ( import (
"context" "context"
"errors"
"sync" "sync"
"sync/atomic" "sync/atomic"
"testing" "testing"
@@ -31,12 +32,19 @@ type fakeTarget struct {
name string name string
kind types.TargetType kind types.TargetType
inits atomic.Int32 inits atomic.Int32
init func() error
} }
func (f *fakeTarget) String() string { return f.name } func (f *fakeTarget) String() string { return f.name }
func (f *fakeTarget) Endpoint() string { return "" } func (f *fakeTarget) Endpoint() string { return "" }
func (f *fakeTarget) Stats() types.TargetStats { return types.TargetStats{} } func (f *fakeTarget) Stats() types.TargetStats { return types.TargetStats{} }
func (f *fakeTarget) Init(context.Context) error { f.inits.Add(1); return nil } func (f *fakeTarget) Init(context.Context) error {
f.inits.Add(1)
if f.init != nil {
return f.init()
}
return nil
}
func (f *fakeTarget) IsOnline(context.Context) bool { return true } func (f *fakeTarget) IsOnline(context.Context) bool { return true }
func (f *fakeTarget) Cancel() {} func (f *fakeTarget) Cancel() {}
func (f *fakeTarget) Send(context.Context, any) error { return nil } func (f *fakeTarget) Send(context.Context, any) error { return nil }
@@ -45,10 +53,12 @@ func (f *fakeTarget) Type() types.TargetType { return f.kind }
// swapSystemTargets isolates the package-level registry for a test. // swapSystemTargets isolates the package-level registry for a test.
func swapSystemTargets(t *testing.T) { func swapSystemTargets(t *testing.T) {
t.Helper() t.Helper()
prevTargets, prevConsole := systemTargets, consoleTgt prevTargets, prevConsole := systemTargets, consoleTgt.Load()
systemTargets, consoleTgt = newTargetsList(), nil systemTargets = newTargetsList()
consoleTgt.Store(nil)
t.Cleanup(func() { t.Cleanup(func() {
systemTargets, consoleTgt = prevTargets, prevConsole systemTargets = prevTargets
consoleTgt.Store(prevConsole)
}) })
} }
@@ -71,7 +81,7 @@ func TestAddSystemTargetIdempotent(t *testing.T) {
if got := console.inits.Load(); got != 1 { if got := console.inits.Load(); got != 1 {
t.Fatalf("expected Init to run once, ran %d times", got) t.Fatalf("expected Init to run once, ran %d times", got)
} }
if consoleTgt != console { if got := consoleTgt.Load(); got == nil || *got != console {
t.Fatalf("consoleTgt not set to the console target") t.Fatalf("consoleTgt not set to the console target")
} }
@@ -89,17 +99,53 @@ func TestAddSystemTargetConcurrent(t *testing.T) {
ctx := context.Background() ctx := context.Background()
console := &fakeTarget{name: "console+http", kind: types.TargetConsole} console := &fakeTarget{name: "console+http", kind: types.TargetConsole}
console.init = func() error {
// Initialization is allowed to inspect or write to the existing logger.
_ = SystemTargets()
consoleLogIf(ctx, "test", errors.New("initializing logger"))
return nil
}
var wg sync.WaitGroup var wg sync.WaitGroup
start := make(chan struct{})
for range 32 { for range 32 {
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
_ = AddSystemTarget(ctx, console) <-start
if err := AddSystemTarget(ctx, console); err != nil {
t.Errorf("AddSystemTarget: %v", err)
}
consoleLogIf(ctx, "test", errors.New("concurrent logger"))
}() }()
} }
close(start)
wg.Wait() wg.Wait()
if got := len(SystemTargets()); got != 1 { if got := len(SystemTargets()); got != 1 {
t.Fatalf("expected 1 system target after concurrent adds, got %d", got) t.Fatalf("expected 1 system target after concurrent adds, got %d", got)
} }
if got := console.inits.Load(); got != 1 {
t.Fatalf("expected one concurrent initialization, got %d", got)
}
}
func TestAddSystemTargetRetriesFailedInit(t *testing.T) {
swapSystemTargets(t)
ctx := context.Background()
initErr := errors.New("target unavailable")
target := &fakeTarget{name: "console", kind: types.TargetConsole}
target.init = func() error { return initErr }
if err := AddSystemTarget(ctx, target); !errors.Is(err, initErr) {
t.Fatalf("expected initialization error, got %v", err)
}
if len(SystemTargets()) != 0 || consoleTgt.Load() != nil {
t.Fatal("failed initialization published a target")
}
target.init = nil
if err := AddSystemTarget(ctx, target); err != nil {
t.Fatal(err)
}
if got := len(SystemTargets()); got != 1 || target.inits.Load() != 2 {
t.Fatalf("retry: targets=%d, initializations=%d", got, target.inits.Load())
}
} }