diff --git a/.changeset/psrpc-service-logger.md b/.changeset/psrpc-service-logger.md new file mode 100644 index 000000000..955e987ed --- /dev/null +++ b/.changeset/psrpc-service-logger.md @@ -0,0 +1,6 @@ +--- +"github.com/livekit/protocol": patch +"@livekit/protocol": patch +--- + +Hand psrpc the service logger in SetLogger so its bus errors are logged diff --git a/logger/logger.go b/logger/logger.go index 0d1af7981..1ebee455a 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -23,6 +23,7 @@ import ( "github.com/go-logr/logr" "github.com/go-logr/logr/funcr" + "github.com/livekit/psrpc" "go.uber.org/zap" "go.uber.org/zap/zapcore" @@ -59,6 +60,9 @@ func SetLogger(l Logger, name string) { defaultLogger = l.WithName(name) // pkg wrapper needs to drop two levels of depth pkgLogger = l.WithCallDepth(1).WithName(name) + // psrpc reports its own bus failures (a dropped Redis subscription, a failed publish) only + // through its own package logger, which discards until one is set. + psrpc.SetLogger(logr.FromSlogHandler(ToSlogHandler(defaultLogger.WithComponent("psrpc")))) } func Debugw(msg string, keysAndValues ...any) { diff --git a/logger/logger_test.go b/logger/logger_test.go index 60f30da4c..dfb3b2efb 100644 --- a/logger/logger_test.go +++ b/logger/logger_test.go @@ -1,13 +1,21 @@ package logger import ( + "bytes" + "context" "encoding/json" + "errors" "fmt" + "net" "os" "runtime" "strings" + "sync/atomic" "testing" + "time" + "github.com/livekit/psrpc/pkg/bus/redisbus" + "github.com/redis/go-redis/v9" "github.com/stretchr/testify/require" "go.uber.org/zap/zapcore" "go.uber.org/zap/zaptest/observer" @@ -410,3 +418,50 @@ func TestLoggerCallDepth(t *testing.T) { }) } } + +// chanWriteSyncer hands each write to the test, so a line logged from another goroutine is read +// without racing on a shared buffer. +type chanWriteSyncer chan []byte + +func (c chanWriteSyncer) Write(p []byte) (int, error) { + c <- bytes.Clone(p) + return len(p), nil +} + +func (chanWriteSyncer) Sync() error { return nil } + +func TestSetLoggerRoutesPsrpcBusErrors(t *testing.T) { + t.Cleanup(func() { + defaultLogger = LogRLogger(discardLogger) + pkgLogger = LogRLogger(discardLogger) + }) + + lines := make(chanWriteSyncer, 1) + SetLogger(must.Get(NewZapLogger(&Config{}, jsonTee(lines))), "TEST") + + dialErr := errors.New("redis is unreachable") + var dials atomic.Int32 + rc := redis.NewClient(&redis.Options{ + Dialer: func(context.Context, string, string) (net.Conn, error) { + if dials.Add(1) == 1 { + return nil, dialErr + } + // Park every reconnect after the first failure, so the bus logs once and then stays + // quiet instead of retrying into later tests. + select {} + }, + }) + redisbus.New(rc) + + var line map[string]any + select { + case b := <-lines: + require.NoError(t, json.Unmarshal(b, &line)) + case <-time.After(5 * time.Second): + t.Fatal("the psrpc bus error never reached the configured logger") + } + require.Equal(t, "error", line["level"]) + require.Equal(t, "redis receive message failed", line["msg"]) + require.Equal(t, "TEST.psrpc", line["logger"]) + require.Contains(t, fmt.Sprint(line), dialErr.Error()) +} diff --git a/logger/slog.go b/logger/slog.go index 694ece1fd..fe4cbd1e6 100644 --- a/logger/slog.go +++ b/logger/slog.go @@ -32,7 +32,7 @@ func ToSlogHandler(log Logger) slog.Handler { switch log := log.(type) { case ZapLogger: zlog := log.ToZap().Desugar() - return zapslog.NewHandler(zlog.Core()) + return zapslog.NewHandler(zlog.Core(), zapslog.WithName(zlog.Name())) case LogRLogger: return logr.ToSlogHandler(log.toLogr()) }