logging improvements

go.mod

@@ -5,7 +5,7 @@ go 1.23.0
 require (
 	github.com/FZambia/eagle v0.1.0
 	github.com/FZambia/statik v0.1.2-0.20180217151304-b9f012bb2a1b
-	github.com/centrifugal/centrifuge v0.33.5-0.20241222164738-7c718f40f348
+	github.com/centrifugal/centrifuge v0.33.5-0.20241224182106-98fa466b7243
 	github.com/centrifugal/protocol v0.13.5-0.20241111155425-6c360178091e
 	github.com/cristalhq/jwt/v5 v5.4.0
 	github.com/go-viper/mapstructure/v2 v2.2.1

go.sum

@@ -6,8 +6,8 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
 github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
-github.com/centrifugal/centrifuge v0.33.5-0.20241222164738-7c718f40f348 h1:F2UZujbtzJwNxcXt6BoXfXFe5NIUkVjYjlvPnzLf8pE=
-github.com/centrifugal/centrifuge v0.33.5-0.20241222164738-7c718f40f348/go.mod h1:hKy1+IjduZJge3EDS3NSSZdTpWd4qhz+AlANNfyv/jE=
+github.com/centrifugal/centrifuge v0.33.5-0.20241224182106-98fa466b7243 h1:mb42mNCu+SOmEDdC1QyAGyoc/ItG/0F6YO2A847LV7c=
+github.com/centrifugal/centrifuge v0.33.5-0.20241224182106-98fa466b7243/go.mod h1:hKy1+IjduZJge3EDS3NSSZdTpWd4qhz+AlANNfyv/jE=
 github.com/centrifugal/protocol v0.13.5-0.20241111155425-6c360178091e h1:+GbuEwJybDuHz6e8S17t/f0I4aTDnZjk37c0aGNFbwc=
 github.com/centrifugal/protocol v0.13.5-0.20241111155425-6c360178091e/go.mod h1:7V5vI30VcoxJe4UD87xi7bOsvI0bmEhvbQuMjrFM2L4=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=

internal/consuming/kafka.go

@@ -528,7 +528,7 @@ func (pc *partitionConsumer) processRecords(records []*kgo.Record) {
 			retries++
 			backoffDuration = getNextBackoffDuration(backoffDuration, retries)
 			pc.metrics.errorsTotal.WithLabelValues(pc.name).Inc()
-			log.Error().Err(err).Str("consumer_name", pc.name).Str("topic", record.Topic).Int32("partition", record.Partition).Dur("next_attempt_in", backoffDuration).Msg("error processing consumed record")
+			log.Error().Err(err).Str("consumer_name", pc.name).Str("topic", record.Topic).Int32("partition", record.Partition).Str("next_attempt_in", backoffDuration.String()).Msg("error processing consumed record")
 			select {
 			case <-time.After(backoffDuration):
 			case <-pc.partitionCtx.Done():

internal/logging/logging.go

@@ -1,15 +1,16 @@
 package logging
 
 import (
+	"io"
 	"os"
 	"runtime"
 	"strings"
 
 	"github.com/centrifugal/centrifugo/v5/internal/config"
 	"github.com/centrifugal/centrifugo/v5/internal/logutils"
+	"github.com/mattn/go-isatty"
 
 	"github.com/centrifugal/centrifuge"
-	"github.com/mattn/go-isatty"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )
@@ -80,8 +81,16 @@ func (h *centrifugeLogHandler) readEntries() {
 			continue
 		}
 		if entry.Fields != nil {
+			if entry.Error != nil {
+				l = l.Err(entry.Error)
+				delete(entry.Fields, "error")
+			}
 			l.Fields(entry.Fields).Msg(entry.Message)
 		} else {
+			if entry.Error != nil {
+				l = l.Err(entry.Error)
+				delete(entry.Fields, "error")
+			}
 			l.Msg(entry.Message)
 		}
 	}
@@ -95,40 +104,59 @@ func (h *centrifugeLogHandler) Handle(entry centrifuge.LogEntry) {
 	}
 }
 
-func configureConsoleWriter() {
+func configureConsoleWriter() *zerolog.ConsoleWriter {
 	if isTerminalAttached() {
-		log.Logger = log.Output(zerolog.ConsoleWriter{
+		return &zerolog.ConsoleWriter{
 			Out:                 os.Stdout,
 			TimeFormat:          "2006-01-02 15:04:05",
 			FormatLevel:         logutils.ConsoleFormatLevel(),
 			FormatErrFieldName:  logutils.ConsoleFormatErrFieldName(),
 			FormatErrFieldValue: logutils.ConsoleFormatErrFieldValue(),
-		})
+		}
 	}
+	return nil
 }
 
 func isTerminalAttached() bool {
 	return isatty.IsTerminal(os.Stdout.Fd()) && runtime.GOOS != "windows"
 }
 
 func Setup(cfg config.Config) (centrifuge.LogHandler, func()) {
-	configureConsoleWriter()
+	var writers []io.Writer
+
+	var file *os.File
+	if cfg.LogFile != "" {
+		var err error
+		file, err = os.OpenFile(cfg.LogFile, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
+		if err != nil {
+			log.Fatal().Err(err).Msg("error opening log file")
+		}
+		writers = append(writers, file)
+	} else {
+		consoleWriter := configureConsoleWriter()
+		if consoleWriter != nil {
+			writers = append(writers, consoleWriter)
+		} else {
+			writers = append(writers, os.Stderr)
+		}
+	}
+
 	logLevel, ok := logLevelMatches[strings.ToUpper(cfg.LogLevel)]
 	if !ok {
 		logLevel = zerolog.InfoLevel
 	}
 	zerolog.SetGlobalLevel(logLevel)
-	if cfg.LogFile != "" {
-		f, err := os.OpenFile(cfg.LogFile, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
-		if err != nil {
-			log.Fatal().Err(err).Msg("error opening log file")
-		}
-		log.Logger = log.Output(f)
-		return newCentrifugeLogHandler().Handle, func() {
-			_ = f.Close()
+
+	if len(writers) > 0 {
+		mw := io.MultiWriter(writers...)
+		log.Logger = log.Output(mw)
+	}
+
+	return newCentrifugeLogHandler().Handle, func() {
+		if file != nil {
+			_ = file.Close()
 		}
 	}
-	return newCentrifugeLogHandler().Handle, nil
 }
 
 // Enabled checks if a specific logging level is enabled

internal/unigrpc/grpc.go

@@ -64,7 +64,7 @@ func (s *Service) Consume(req *unistream.ConnectRequest, stream unistream.Centri
 	if logging.Enabled(logging.DebugLevel) {
 		log.Debug().Str("transport", transport.Name()).Str("client", c.ID()).Msg("client connection established")
 		defer func(started time.Time) {
-			log.Debug().Str("transport", transport.Name()).Str("client", c.ID()).Dur("duration", time.Since(started)).Msg("client connection completed")
+			log.Debug().Str("transport", transport.Name()).Str("client", c.ID()).Str("duration", time.Since(started).String()).Msg("client connection completed")
 		}(time.Now())
 	}
 

internal/unihttpstream/handler.go

@@ -73,7 +73,7 @@ func (h Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if logging.Enabled(logging.DebugLevel) {
 		log.Debug().Str("transport", transportName).Str("client", c.ID()).Msg("client connection established")
 		defer func(started time.Time) {
-			log.Debug().Str("transport", transportName).Str("client", c.ID()).Dur("duration", time.Since(started)).Msg("client connection completed")
+			log.Debug().Str("transport", transportName).Str("client", c.ID()).Str("duration", time.Since(started).String()).Msg("client connection completed")
 		}(time.Now())
 	}
 

internal/unisse/handler.go

@@ -83,7 +83,7 @@ func (h Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if logging.Enabled(logging.DebugLevel) {
 		log.Debug().Str("transport", transportName).Str("client", c.ID()).Msg("client connection established")
 		defer func(started time.Time) {
-			log.Debug().Str("transport", transportName).Str("client", c.ID()).Dur("duration", time.Since(started)).Msg("client connection completed")
+			log.Debug().Str("transport", transportName).Str("client", c.ID()).Str("duration", time.Since(started).String()).Msg("client connection completed")
 		}(time.Now())
 	}
 

internal/uniws/handler.go

@@ -142,7 +142,7 @@ func (s *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 		if logging.Enabled(logging.DebugLevel) {
 			log.Debug().Str("transport", transportName).Str("client", c.ID()).Msg("client connection established")
 			defer func(started time.Time) {
-				log.Debug().Str("transport", transportName).Str("client", c.ID()).Dur("duration", time.Since(started)).Msg("client connection completed")
+				log.Debug().Str("transport", transportName).Str("client", c.ID()).Str("duration", time.Since(started).String()).Msg("client connection completed")
 			}(time.Now())
 		}
 

internal/usage/usage.go

@@ -141,7 +141,7 @@ func (s *Sender) isDev() bool {
 func (s *Sender) Run(ctx context.Context) error {
 	firstTimeSend := time.Now().Add(initialDelay)
 	if s.isDev() {
-		log.Debug().Dur("delay", initialDelay).Msg("usage stats: schedule next send")
+		log.Debug().Str("delay", initialDelay.String()).Msg("usage stats: schedule next send")
 	}
 
 	// Wait 1/4 of a delay to randomize hourly ticks on different nodes.

internal/wt/handler.go

@@ -64,7 +64,7 @@ func (s *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 	if logging.Enabled(logging.DebugLevel) {
 		log.Debug().Str("transport", transportName).Str("client", c.ID()).Msg("client connection established")
 		defer func(started time.Time) {
-			log.Debug().Str("transport", transportName).Str("client", c.ID()).Dur("duration", time.Since(started)).Msg("client connection completed")
+			log.Debug().Str("transport", transportName).Str("client", c.ID()).Str("duration", time.Since(started).String()).Msg("client connection completed")
 		}(time.Now())
 	}