From 185120bc23ab2da319611286d3e52d84ec05506d Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Wed, 23 Sep 2026 19:19:36 +0300 Subject: [PATCH] fix(logger): hand psrpc the service logger so its bus failures are logged psrpc keeps a package logger of its own for the failures only it can see on the Redis bus: a receive that errors on a dropped pub/sub connection, a subscription that cannot be reconciled, a publish pipeline that fails. It defaults to logr.Discard, and neither livekit-server nor livekit-sip ever calls psrpc.SetLogger, so those errors vanish. A dial the bus loses then surfaces only as a "no response from servers" from the caller, with nothing on either side saying why. SetLogger is the one path both binaries take, so it now hands psrpc the same logger as component "psrpc". The level follows component_levels.psrpc like every other component, including through the live config reload, since the logr is backed by the component's own zap core. ToSlogHandler now passes the zap logger's name through to zapslog, which otherwise drops it: the psrpc lines would carry no "livekit.psrpc" name to find them by. The root logger's name is empty, so slog.Default output does not change. Signed-off-by: Cosmin Staicu --- .changeset/psrpc-service-logger.md | 6 ++++ logger/logger.go | 4 +++ logger/logger_test.go | 55 ++++++++++++++++++++++++++++++ logger/slog.go | 2 +- 4 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 .changeset/psrpc-service-logger.md 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()) }