Use structured logging with slog

Configuration file can set the logging handler between two: text,
which is the default, and json. Default level is info which matches
previous behavior.

A bit more info level logging is enabled by default to default
restore to being verbose compared to quiet and snapshotting.

Trace logging has been converted to be on DEBUG level and always
enabled if the configured level is high enough.

Bumps Go requirement to 1.21

.github/workflows/commit.yml

@@ -8,7 +8,7 @@ jobs:
       - uses: actions/checkout@v2
       - uses: actions/setup-go@v2
         with:
-          go-version: '1.20'
+          go-version: '1.21'
       - uses: actions/cache@v2
         with:
           path: ~/go/pkg/mod

.github/workflows/release.linux.yml

@@ -31,7 +31,7 @@ jobs:
       - uses: actions/checkout@v2
       - uses: actions/setup-go@v2
         with:
-          go-version: '1.20'
+          go-version: '1.21'
 
       - id: release
         uses: bruceadams/[email protected]

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.20.1 as builder
+FROM golang:1.21.3 as builder
 
 WORKDIR /src/litestream
 COPY . .

cmd/litestream/generations.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
 	"os"
 	"text/tabwriter"
 	"time"
@@ -87,15 +86,15 @@ func (c *GenerationsCommand) Run(ctx context.Context, args []string) (err error)
 	for _, r := range replicas {
 		generations, err := r.Client.Generations(ctx)
 		if err != nil {
-			log.Printf("%s: cannot list generations: %s", r.Name(), err)
+			r.Logger().Error("cannot list generations", "error", err)
 			continue
 		}
 
 		// Iterate over each generation for the replica.
 		for _, generation := range generations {
 			createdAt, updatedAt, err := r.GenerationTimeBounds(ctx, generation)
 			if err != nil {
-				log.Printf("%s: cannot determine generation time bounds: %s", r.Name(), err)
+				r.Logger().Error("cannot determine generation time bounds", "error", err)
 				continue
 			}
 

cmd/litestream/main.go

@@ -6,7 +6,7 @@ import (
 	"flag"
 	"fmt"
 	"io/ioutil"
-	"log"
+	"log/slog"
 	"net/url"
 	"os"
 	"os/user"
@@ -37,13 +37,11 @@ var (
 var errStop = errors.New("stop")
 
 func main() {
-	log.SetFlags(0)
-
 	m := NewMain()
 	if err := m.Run(context.Background(), os.Args[1:]); err == flag.ErrHelp || err == errStop {
 		os.Exit(1)
 	} else if err != nil {
-		log.Println(err)
+		slog.Error("failed to run", "error", err)
 		os.Exit(1)
 	}
 }
@@ -172,6 +170,16 @@ type Config struct {
 	// Global S3 settings
 	AccessKeyID     string `yaml:"access-key-id"`
 	SecretAccessKey string `yaml:"secret-access-key"`
+
+	// Logging
+	Logging LoggingConfig `yaml:"logging"`
+}
+
+// LoggingConfig configures logging.
+type LoggingConfig struct {
+	Level  string `yaml:"level"`
+	Type   string `yaml:"type"`
+	Stderr bool   `yaml:"stderr"`
 }
 
 // propagateGlobalSettings copies global S3 settings to replica configs.
@@ -241,6 +249,36 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	// Propage settings from global config to replica configs.
 	config.propagateGlobalSettings()
 
+	// Configure logging.
+	logOutput := os.Stdout
+	if config.Logging.Stderr {
+		logOutput = os.Stderr
+	}
+
+	logOptions := slog.HandlerOptions{
+		Level: slog.LevelInfo,
+	}
+
+	switch strings.ToUpper(config.Logging.Level) {
+	case "DEBUG":
+		logOptions.Level = slog.LevelDebug
+	case "WARN", "WARNING":
+		logOptions.Level = slog.LevelWarn
+	case "ERROR":
+		logOptions.Level = slog.LevelError
+	}
+
+	var logHandler slog.Handler
+	switch config.Logging.Type {
+	case "json":
+		logHandler = slog.NewJSONHandler(logOutput, &logOptions)
+	case "text", "":
+		logHandler = slog.NewTextHandler(logOutput, &logOptions)
+	}
+
+	// Set global default logger.
+	slog.SetDefault(slog.New(logHandler))
+
 	return config, nil
 }
 

cmd/litestream/main_notwindows.go

@@ -1,4 +1,4 @@
-// +build !windows
+//go:build !windows
 
 package main
 

cmd/litestream/main_windows.go

@@ -1,11 +1,10 @@
-// +build windows
+//go:build windows
 
 package main
 
 import (
 	"context"
 	"io"
-	"log"
 	"os"
 	"os/signal"
 
@@ -63,13 +62,13 @@ func (s *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, stat
 	// Instantiate replication command and load configuration.
 	c := NewReplicateCommand()
 	if c.Config, err = ReadConfigFile(DefaultConfigPath(), true); err != nil {
-		log.Printf("cannot load configuration: %s", err)
+		slog.Error("cannot load configuration", "error", err)
 		return true, 1
 	}
 
 	// Execute replication command.
 	if err := c.Run(s.ctx); err != nil {
-		log.Printf("cannot replicate: %s", err)
+		slog.Error("cannot replicate", "error", err)
 		statusCh <- svc.Status{State: svc.StopPending}
 		return true, 2
 	}
@@ -88,7 +87,7 @@ func (s *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, stat
 			case svc.Interrogate:
 				statusCh <- req.CurrentStatus
 			default:
-				log.Printf("Litestream service received unexpected change request cmd: %d", req.Cmd)
+				slog.Error("Litestream service received unexpected change request", "cmd", req.Cmd)
 			}
 		}
 	}

cmd/litestream/replicate.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
+	"log/slog"
 	"net"
 	"net/http"
 	_ "net/http/pprof"
@@ -42,7 +42,6 @@ func NewReplicateCommand() *ReplicateCommand {
 func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err error) {
 	fs := flag.NewFlagSet("litestream-replicate", flag.ContinueOnError)
 	execFlag := fs.String("exec", "", "execute subcommand")
-	tracePath := fs.String("trace", "", "trace path")
 	configPath, noExpandEnv := registerConfigFlag(fs)
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
@@ -80,27 +79,17 @@ func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err e
 		c.Config.Exec = *execFlag
 	}
 
-	// Enable trace logging.
-	if *tracePath != "" {
-		f, err := os.Create(*tracePath)
-		if err != nil {
-			return err
-		}
-		defer f.Close()
-		litestream.Tracef = log.New(f, "", log.LstdFlags|log.Lmicroseconds|log.LUTC|log.Lshortfile).Printf
-	}
-
 	return nil
 }
 
 // Run loads all databases specified in the configuration.
 func (c *ReplicateCommand) Run() (err error) {
 	// Display version information.
-	log.Printf("litestream %s", Version)
+	slog.Info("litestream", "version", Version)
 
 	// Setup databases.
 	if len(c.Config.DBs) == 0 {
-		log.Println("no databases specified in configuration")
+		slog.Error("no databases specified in configuration")
 	}
 
 	for _, dbConfig := range c.Config.DBs {
@@ -118,21 +107,22 @@ func (c *ReplicateCommand) Run() (err error) {
 
 	// Notify user that initialization is done.
 	for _, db := range c.DBs {
-		log.Printf("initialized db: %s", db.Path())
+		slog.Info("initialized db", "path", db.Path())
 		for _, r := range db.Replicas {
+			slog := slog.With("name", r.Name(), "type", r.Client.Type(), "sync-interval", r.SyncInterval)
 			switch client := r.Client.(type) {
 			case *file.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q path=%q", r.Name(), client.Type(), client.Path())
+				slog.Info("replicating to", "path", client.Path())
 			case *s3.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q region=%q endpoint=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, client.Region, client.Endpoint, r.SyncInterval)
+				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "region", client.Region, "endpoint", client.Endpoint)
 			case *gcs.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, r.SyncInterval)
+				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path)
 			case *abs.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q endpoint=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, client.Endpoint, r.SyncInterval)
+				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "endpoint", client.Endpoint)
 			case *sftp.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q host=%q user=%q path=%q sync-interval=%s", r.Name(), client.Type(), client.Host, client.User, client.Path, r.SyncInterval)
+				slog.Info("replicating to", "host", client.Host, "user", client.User, "path", client.Path)
 			default:
-				log.Printf("replicating to: name=%q type=%q", r.Name(), client.Type())
+				slog.Info("replicating to")
 			}
 		}
 	}
@@ -146,11 +136,11 @@ func (c *ReplicateCommand) Run() (err error) {
 			hostport = net.JoinHostPort("localhost", port)
 		}
 
-		log.Printf("serving metrics on http://%s/metrics", hostport)
+		slog.Info("serving metrics on", "url", fmt.Sprintf("http://%s/metrics", hostport))
 		go func() {
 			http.Handle("/metrics", promhttp.Handler())
 			if err := http.ListenAndServe(c.Config.Addr, nil); err != nil {
-				log.Printf("cannot start metrics server: %s", err)
+				slog.Error("cannot start metrics server", "error", err)
 			}
 		}()
 	}
@@ -179,7 +169,7 @@ func (c *ReplicateCommand) Run() (err error) {
 func (c *ReplicateCommand) Close() (err error) {
 	for _, db := range c.DBs {
 		if e := db.Close(); e != nil {
-			log.Printf("error closing db: path=%s err=%s", db.Path(), e)
+			db.Logger.Error("error closing db", "error", e)
 			if err == nil {
 				err = e
 			}
@@ -215,8 +205,5 @@ Arguments:
 	-no-expand-env
 	    Disables environment variable expansion in configuration file.
 
-	-trace PATH
-	    Write verbose trace logging to PATH.
-
 `[1:], DefaultConfigPath())
 }

cmd/litestream/restore.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"log"
 	"os"
 	"strconv"
 	"time"
@@ -19,7 +18,6 @@ type RestoreCommand struct{}
 // Run executes the command.
 func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 	opt := litestream.NewRestoreOptions()
-	opt.Verbose = true
 
 	fs := flag.NewFlagSet("litestream-restore", flag.ContinueOnError)
 	configPath, noExpandEnv := registerConfigFlag(fs)
@@ -31,7 +29,6 @@ func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 	ifDBNotExists := fs.Bool("if-db-not-exists", false, "")
 	ifReplicaExists := fs.Bool("if-replica-exists", false, "")
 	timestampStr := fs.String("timestamp", "", "timestamp")
-	verbose := fs.Bool("v", false, "verbose output")
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -48,11 +45,6 @@ func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 		}
 	}
 
-	// Instantiate logger if verbose output is enabled.
-	if *verbose {
-		opt.Logger = log.New(os.Stderr, "", log.LstdFlags|log.Lmicroseconds)
-	}
-
 	// Determine replica & generation to restore from.
 	var r *litestream.Replica
 	if isURL(fs.Arg(0)) {

cmd/litestream/snapshots.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
+	"log/slog"
 	"os"
 	"text/tabwriter"
 	"time"
@@ -82,7 +82,7 @@ func (c *SnapshotsCommand) Run(ctx context.Context, args []string) (err error) {
 	for _, r := range replicas {
 		infos, err := r.Snapshots(ctx)
 		if err != nil {
-			log.Printf("cannot determine snapshots: %s", err)
+			slog.Error("cannot determine snapshots", "error", err)
 			continue
 		}
 		for _, info := range infos {

cmd/litestream/wal.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
 	"os"
 	"text/tabwriter"
 	"time"
@@ -86,7 +85,7 @@ func (c *WALCommand) Run(ctx context.Context, args []string) (err error) {
 			generations = []string{*generation}
 		} else {
 			if generations, err = r.Client.Generations(ctx); err != nil {
-				log.Printf("%s: cannot determine generations: %s", r.Name(), err)
+				r.Logger().Error("cannot determine generations", "error", err)
 				continue
 			}
 		}
@@ -113,7 +112,7 @@ func (c *WALCommand) Run(ctx context.Context, args []string) (err error) {
 				}
 				return itr.Close()
 			}(); err != nil {
-				log.Printf("%s: cannot fetch wal segments: %s", r.Name(), err)
+				r.Logger().Error("cannot fetch wal segments", "error", err)
 				continue
 			}
 		}