add structured logging (#5219)

Co-authored-by: aler9 <46489434+aler9@users.noreply.github.com>

api/openapi.yaml

@@ -73,6 +73,8 @@ components:
           type: array
           items:
             type: string
+        logStructured:
+          type: boolean
         logFile:
           type: string
         sysLogPrefix:

docs/2-usage/17-logging.md

@@ -43,6 +43,21 @@ If _MediaMTX_ is also running as a [system service](start-on-boot), log entries
 journalctl -u mediamtx
 ```
 
+## Structured logging
+
+Log collectors (like Loki, Logstash, CloudWatch and fluentd) parse logs in a more reliable way if they are fed with entries in structured format (JSONL). This can be enabled with the `logStructured` parameter:
+
+```yml
+# When destination is "stdout" or "file", emit logs in structured format (JSON).
+logStructured: true
+```
+
+Obtaining:
+
+```
+{"timestamp":"2003-05-01T20:34:14+01:00","level":"INF","message":"[RTSP] listener opened on :8554 (TCP), :8000 (UDP/RTP), :8001 (UDP/RTCP)"}
+```
+
 ## Log file rotation
 
 The log file can be periodically rotated or truncated by using an external utility.

internal/conf/conf.go

@@ -152,6 +152,7 @@ type Conf struct {
 	// General
 	LogLevel            LogLevel        `json:"logLevel"`
 	LogDestinations     LogDestinations `json:"logDestinations"`
+	LogStructured       bool            `json:"logStructured"`
 	LogFile             string          `json:"logFile"`
 	SysLogPrefix        string          `json:"sysLogPrefix"`
 	ReadTimeout         Duration        `json:"readTimeout"`
@@ -318,6 +319,7 @@ func (conf *Conf) setDefaults() {
 	// General
 	conf.LogLevel = LogLevel(logger.Info)
 	conf.LogDestinations = LogDestinations{logger.DestinationStdout}
+	conf.LogStructured = false
 	conf.LogFile = "mediamtx.log"
 	conf.SysLogPrefix = "mediamtx"
 	conf.ReadTimeout = 10 * Duration(time.Second)

internal/core/core.go

@@ -157,7 +157,14 @@ func New(args []string) (*Core, bool) {
 		done:           make(chan struct{}),
 	}
 
-	tempLogger, _ := logger.New(logger.Warn, []logger.Destination{logger.DestinationStdout}, "", "")
+	tempLogger := &logger.Logger{
+		Level:        logger.Warn,
+		Destinations: []logger.Destination{logger.DestinationStdout},
+		Structured:   false,
+		File:         "",
+		SysLogPrefix: "",
+	}
+	tempLogger.Initialize() //nolint:errcheck
 
 	confPaths := append([]string(nil), defaultConfPaths...)
 	if runtime.GOOS != "windows" {
@@ -263,15 +270,18 @@ func (p *Core) createResources(initial bool) error {
 	var err error
 
 	if p.logger == nil {
-		p.logger, err = logger.New(
-			logger.Level(p.conf.LogLevel),
-			p.conf.LogDestinations,
-			p.conf.LogFile,
-			p.conf.SysLogPrefix,
-		)
+		i := &logger.Logger{
+			Level:        logger.Level(p.conf.LogLevel),
+			Destinations: p.conf.LogDestinations,
+			Structured:   p.conf.LogStructured,
+			File:         p.conf.LogFile,
+			SysLogPrefix: p.conf.SysLogPrefix,
+		}
+		err = i.Initialize()
 		if err != nil {
 			return err
 		}
+		p.logger = i
 	}
 
 	if initial {
@@ -691,7 +701,8 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
 		newConf.LogLevel != p.conf.LogLevel ||
 		!reflect.DeepEqual(newConf.LogDestinations, p.conf.LogDestinations) ||
 		newConf.LogFile != p.conf.LogFile ||
-		newConf.SysLogPrefix != p.conf.SysLogPrefix
+		newConf.SysLogPrefix != p.conf.SysLogPrefix ||
+		newConf.LogStructured != p.conf.LogStructured
 
 	closeAuthManager := newConf == nil ||
 		newConf.AuthMethod != p.conf.AuthMethod ||

internal/logger/destination_file.go

@@ -4,31 +4,48 @@ import (
 	"bytes"
 	"fmt"
 	"os"
+	"strconv"
 	"time"
 )
 
 type destinationFile struct {
-	file *os.File
-	buf  bytes.Buffer
+	structured bool
+	file       *os.File
+	buf        bytes.Buffer
 }
 
-func newDestinationFile(filePath string) (destination, error) {
+func newDestinationFile(structured bool, filePath string) (destination, error) {
 	f, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
 	if err != nil {
 		return nil, err
 	}
 
 	return &destinationFile{
-		file: f,
+		structured: structured,
+		file:       f,
 	}, nil
 }
 
 func (d *destinationFile) log(t time.Time, level Level, format string, args ...any) {
 	d.buf.Reset()
-	writeTime(&d.buf, t, false)
-	writeLevel(&d.buf, level, false)
-	fmt.Fprintf(&d.buf, format, args...)
-	d.buf.WriteByte('\n')
+
+	if d.structured {
+		d.buf.WriteString(`{"timestamp":"`)
+		d.buf.WriteString(t.Format(time.RFC3339))
+		d.buf.WriteString(`","level":"`)
+		writeLevel(&d.buf, level, false)
+		d.buf.WriteString(`","message":`)
+		d.buf.WriteString(strconv.Quote(fmt.Sprintf(format, args...)))
+		d.buf.WriteString(`}`)
+		d.buf.WriteByte('\n')
+	} else {
+		writePlainTime(&d.buf, t, false)
+		writeLevel(&d.buf, level, false)
+		d.buf.WriteByte(' ')
+		fmt.Fprintf(&d.buf, format, args...)
+		d.buf.WriteByte('\n')
+	}
+
 	d.file.Write(d.buf.Bytes()) //nolint:errcheck
 }
 

internal/logger/destination_stdout.go

@@ -3,30 +3,50 @@ package logger
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"os"
+	"strconv"
 	"time"
 
 	"golang.org/x/term"
 )
 
 type destinationStdout struct {
-	useColor bool
-	buf      bytes.Buffer
+	structured bool
+	stdout     io.Writer
+	useColor   bool
+	buf        bytes.Buffer
 }
 
-func newDestionationStdout() destination {
+func newDestionationStdout(structured bool, stdout io.Writer) destination {
 	return &destinationStdout{
-		useColor: term.IsTerminal(int(os.Stdout.Fd())),
+		structured: structured,
+		stdout:     stdout,
+		useColor:   term.IsTerminal(int(os.Stdout.Fd())),
 	}
 }
 
 func (d *destinationStdout) log(t time.Time, level Level, format string, args ...any) {
 	d.buf.Reset()
-	writeTime(&d.buf, t, d.useColor)
-	writeLevel(&d.buf, level, d.useColor)
-	fmt.Fprintf(&d.buf, format, args...)
-	d.buf.WriteByte('\n')
-	os.Stdout.Write(d.buf.Bytes()) //nolint:errcheck
+
+	if d.structured {
+		d.buf.WriteString(`{"timestamp":"`)
+		d.buf.WriteString(t.Format(time.RFC3339))
+		d.buf.WriteString(`","level":"`)
+		writeLevel(&d.buf, level, false)
+		d.buf.WriteString(`","message":`)
+		d.buf.WriteString(strconv.Quote(fmt.Sprintf(format, args...)))
+		d.buf.WriteString(`}`)
+		d.buf.WriteByte('\n')
+	} else {
+		writePlainTime(&d.buf, t, d.useColor)
+		writeLevel(&d.buf, level, d.useColor)
+		d.buf.WriteByte(' ')
+		fmt.Fprintf(&d.buf, format, args...)
+		d.buf.WriteByte('\n')
+	}
+
+	d.stdout.Write(d.buf.Bytes()) //nolint:errcheck
 }
 
 func (d *destinationStdout) close() {

internal/logger/logger.go

@@ -3,6 +3,8 @@ package logger
 
 import (
 	"bytes"
+	"io"
+	"os"
 	"sync"
 	"time"
 
@@ -11,47 +13,56 @@ import (
 
 // Logger is a log handler.
 type Logger struct {
-	level Level
+	Level        Level
+	Destinations []Destination
+	Structured   bool
+	File         string
+	SysLogPrefix string
 
+	timeNow      func() time.Time
+	stdout       io.Writer
 	destinations []destination
 	mutex        sync.Mutex
 }
 
-// New allocates a log handler.
-func New(level Level, destinations []Destination, filePath string, sysLogPrefix string) (*Logger, error) {
-	lh := &Logger{
-		level: level,
+// Initialize initializes Logger.
+func (l *Logger) Initialize() error {
+	if l.timeNow == nil {
+		l.timeNow = time.Now
+	}
+	if l.stdout == nil {
+		l.stdout = os.Stdout
 	}
 
-	for _, destType := range destinations {
+	for _, destType := range l.Destinations {
 		switch destType {
 		case DestinationStdout:
-			lh.destinations = append(lh.destinations, newDestionationStdout())
+			l.destinations = append(l.destinations, newDestionationStdout(l.Structured, l.stdout))
 
 		case DestinationFile:
-			dest, err := newDestinationFile(filePath)
+			dest, err := newDestinationFile(l.Structured, l.File)
 			if err != nil {
-				lh.Close()
-				return nil, err
+				l.Close()
+				return err
 			}
-			lh.destinations = append(lh.destinations, dest)
+			l.destinations = append(l.destinations, dest)
 
 		case DestinationSyslog:
-			dest, err := newDestinationSyslog(sysLogPrefix)
+			dest, err := newDestinationSyslog(l.SysLogPrefix)
 			if err != nil {
-				lh.Close()
-				return nil, err
+				l.Close()
+				return err
 			}
-			lh.destinations = append(lh.destinations, dest)
+			l.destinations = append(l.destinations, dest)
 		}
 	}
 
-	return lh, nil
+	return nil
 }
 
 // Close closes a log handler.
-func (lh *Logger) Close() {
-	for _, dest := range lh.destinations {
+func (l *Logger) Close() {
+	for _, dest := range l.destinations {
 		dest.close()
 	}
 }
@@ -73,7 +84,7 @@ func itoa(i int, wid int) []byte {
 	return b[bp:]
 }
 
-func writeTime(buf *bytes.Buffer, t time.Time, useColor bool) {
+func writePlainTime(buf *bytes.Buffer, t time.Time, useColor bool) {
 	var intbuf bytes.Buffer
 
 	// date
@@ -131,21 +142,20 @@ func writeLevel(buf *bytes.Buffer, level Level, useColor bool) {
 			buf.WriteString("ERR")
 		}
 	}
-	buf.WriteByte(' ')
 }
 
 // Log writes a log entry.
-func (lh *Logger) Log(level Level, format string, args ...any) {
-	if level < lh.level {
+func (l *Logger) Log(level Level, format string, args ...any) {
+	if level < l.Level {
 		return
 	}
 
-	lh.mutex.Lock()
-	defer lh.mutex.Unlock()
+	l.mutex.Lock()
+	defer l.mutex.Unlock()
 
-	t := time.Now()
+	t := l.timeNow()
 
-	for _, dest := range lh.destinations {
+	for _, dest := range l.destinations {
 		dest.log(t, level, format, args...)
 	}
 }

internal/logger/logger_test.go

@@ -0,0 +1,77 @@
+package logger
+
+import (
+	"bytes"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLoggerToStdout(t *testing.T) {
+	for _, ca := range []string{
+		"plain",
+		"structured",
+	} {
+		t.Run(ca, func(t *testing.T) {
+			var buf bytes.Buffer
+
+			l := &Logger{
+				Destinations: []Destination{DestinationStdout},
+				Structured:   (ca == "structured"),
+				timeNow:      func() time.Time { return time.Date(2003, 11, 4, 23, 15, 8, 0, time.UTC) },
+				stdout:       &buf,
+			}
+			err := l.Initialize()
+			require.NoError(t, err)
+			defer l.Close()
+
+			l.Log(Info, "test format %d", 123)
+
+			if ca == "plain" {
+				require.Equal(t, "2003/11/04 23:15:08 INF test format 123\n", buf.String())
+			} else {
+				require.Equal(t, `{"timestamp":"2003-11-04T23:15:08Z",`+
+					`"level":"INF","message":"test format 123"}`+"\n", buf.String())
+			}
+		})
+	}
+}
+
+func TestLoggerToFile(t *testing.T) {
+	for _, ca := range []string{
+		"plain",
+		"structured",
+	} {
+		t.Run(ca, func(t *testing.T) {
+			tempFile, err := os.CreateTemp(os.TempDir(), "mtx-logger-")
+			require.NoError(t, err)
+			defer os.Remove(tempFile.Name())
+			defer tempFile.Close()
+
+			l := &Logger{
+				Level:        Debug,
+				Destinations: []Destination{DestinationFile},
+				Structured:   ca == "structured",
+				File:         tempFile.Name(),
+				timeNow:      func() time.Time { return time.Date(2003, 11, 4, 23, 15, 8, 0, time.UTC) },
+			}
+			err = l.Initialize()
+			require.NoError(t, err)
+			defer l.Close()
+
+			l.Log(Info, "test format %d", 123)
+
+			buf, err := os.ReadFile(tempFile.Name())
+			require.NoError(t, err)
+
+			if ca == "plain" {
+				require.Equal(t, "2003/11/04 23:15:08 INF test format 123\n", string(buf))
+			} else {
+				require.Equal(t, `{"timestamp":"2003-11-04T23:15:08Z",`+
+					`"level":"INF","message":"test format 123"}`+"\n", string(buf))
+			}
+		})
+	}
+}

mediamtx.yml

@@ -10,9 +10,11 @@
 logLevel: info
 # Destinations of log messages; available values are "stdout", "file" and "syslog".
 logDestinations: [stdout]
-# If "file" is in logDestinations, this is the file which will receive logs.
+# When destination is "stdout" or "file", emit logs in structured format (JSON).
+logStructured: no
+# When "file" is in logDestinations, this is the file which will receive logs.
 logFile: mediamtx.log
-# If "syslog" is in logDestinations, use prefix for logs.
+# When "syslog" is in logDestinations, use prefix for logs.
 sysLogPrefix: mediamtx
 
 # Timeout of read operations.