Merge pull request #1052 from percona/PBM-1226-logpath-opt-no-cfg

PBM-1226: `logpath` option for pbm-agent

Author: boris-ilijic

cmd/pbm-agent/main.go

@@ -49,6 +49,20 @@ func main() {
 		versionFormat = versionCmd.Flag("format", "Output format <json or \"\">").
 				Default("").
 				String()
+
+		logPath = pbmCmd.Flag("log-path", "Path to file").
+			Envar("LOG_PATH").
+			Default("/dev/stderr").
+			String()
+		logJSON = pbmCmd.Flag("log-json", "Enable JSON output").
+			Envar("LOG_JSON").
+			Bool()
+		logLevel = pbmCmd.Flag(
+			"log-level",
+			"Minimal log level based on severity level: D, I, W, E or F, low to high. Choosing one includes higher levels too.").
+			Envar("LOG_LEVEL").
+			Default(log.D).
+			Enum(log.D, log.I, log.W, log.E, log.F)
 	)
 
 	cmd, err := pbmCmd.DefaultEnvars().Parse(os.Args[1:])
@@ -74,19 +88,24 @@ func main() {
 
 	hidecreds()
 
-	fmt.Print(perconaSquadNotice)
+	logOpts := &log.Opts{
+		LogPath:  *logPath,
+		LogLevel: *logLevel,
+		LogJSON:  *logJSON,
+	}
 
-	err = runAgent(url, *dumpConns)
+	err = runAgent(url, *dumpConns, logOpts)
 	stdlog.Println("Exit:", err)
 	if err != nil {
 		os.Exit(1)
 	}
 }
 
-func runAgent(mongoURI string, dumpConns int) error {
-	mtLog.SetDateFormat(log.LogTimeFormat)
-	mtLog.SetVerbosity(&options.Verbosity{VLevel: mtLog.DebugLow})
-
+func runAgent(
+	mongoURI string,
+	dumpConns int,
+	logOpts *log.Opts,
+) error {
 	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
 	defer cancel()
 
@@ -95,16 +114,34 @@ func runAgent(mongoURI string, dumpConns int) error {
 		return errors.Wrap(err, "connect to PBM")
 	}
 
+	err = setupNewDB(ctx, leadConn)
+	if err != nil {
+		return errors.Wrap(err, "setup pbm collections")
+	}
+
 	agent, err := newAgent(ctx, leadConn, mongoURI, dumpConns)
 	if err != nil {
 		return errors.Wrap(err, "connect to the node")
 	}
 
-	logger := log.New(agent.leadConn.LogCollection(), agent.brief.SetName, agent.brief.Me)
+	logger := log.NewWithOpts(
+		ctx,
+		agent.leadConn,
+		agent.brief.SetName,
+		agent.brief.Me,
+		logOpts)
 	defer logger.Close()
 
 	ctx = log.SetLoggerToContext(ctx, logger)
 
+	mtLog.SetDateFormat(log.LogTimeFormat)
+	mtLog.SetVerbosity(&options.Verbosity{VLevel: mtLog.DebugLow})
+	mtLog.SetWriter(logger)
+
+	logger.Printf(perconaSquadNotice)
+	logger.Printf("log options: log-path=%s, log-level:%s, log-json:%t",
+		logOpts.LogPath, logOpts.LogLevel, logOpts.LogJSON)
+
 	canRunSlicer := true
 	if err := agent.CanStart(ctx); err != nil {
 		if errors.Is(err, ErrArbiterNode) || errors.Is(err, ErrDelayedNode) {
@@ -114,11 +151,6 @@ func runAgent(mongoURI string, dumpConns int) error {
 		}
 	}
 
-	err = setupNewDB(ctx, agent.leadConn)
-	if err != nil {
-		return errors.Wrap(err, "setup pbm collections")
-	}
-
 	agent.showIncompatibilityWarning(ctx)
 
 	if canRunSlicer {

cmd/pbm/main.go

@@ -495,7 +495,7 @@ func main() {
 		if err != nil {
 			exitErr(errors.Wrap(err, "connect to mongodb"), pbmOutF)
 		}
-		ctx = log.SetLoggerToContext(ctx, log.New(conn.LogCollection(), "", ""))
+		ctx = log.SetLoggerToContext(ctx, log.New(conn, "", ""))
 
 		ver, err := version.GetMongoVersion(ctx, conn.MongoClient())
 		if err != nil {

pbm/log/discard.go

@@ -28,6 +28,10 @@ func (discardLoggerImpl) PauseMgo() {
 func (discardLoggerImpl) ResumeMgo() {
 }
 
+func (discardLoggerImpl) Write([]byte) (int, error) {
+	return 0, nil
+}
+
 func (discardLoggerImpl) Printf(msg string, args ...any) {
 }
 

pbm/log/log.go

@@ -26,6 +26,8 @@ type Logger interface {
 	PauseMgo()
 	ResumeMgo()
 
+	Write(p []byte) (n int, err error)
+
 	Printf(msg string, args ...any)
 	Debug(event, obj, opid string, epoch primitive.Timestamp, msg string, args ...any)
 	Info(event, obj, opid string, epoch primitive.Timestamp, msg string, args ...any)
@@ -60,19 +62,44 @@ const (
 	Debug
 )
 
+const (
+	F = "F"
+	E = "E"
+	W = "W"
+	I = "I"
+	D = "D"
+)
+
 func (s Severity) String() string {
 	switch s {
 	case Fatal:
-		return "F"
+		return F
 	case Error:
-		return "E"
+		return E
 	case Warning:
-		return "W"
+		return W
 	case Info:
-		return "I"
+		return I
 	case Debug:
-		return "D"
+		return D
 	default:
 		return ""
 	}
 }
+
+func strToSeverity(s string) Severity {
+	switch s {
+	case F:
+		return Fatal
+	case E:
+		return Error
+	case W:
+		return Warning
+	case I:
+		return Info
+	case D:
+		return Debug
+	default:
+		return Debug
+	}
+}

pbm/log/logger.go

@@ -5,35 +5,149 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"io/fs"
 	"log"
 	"os"
+	"path/filepath"
+	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
 	"go.mongodb.org/mongo-driver/bson/primitive"
-	"go.mongodb.org/mongo-driver/mongo"
 
+	"github.com/percona/percona-backup-mongodb/pbm/connect"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 )
 
+const (
+	logPathStdErr = "/dev/stderr"
+)
+
 type loggerImpl struct {
-	cn   *mongo.Collection
-	out  io.Writer
+	conn connect.Client
+
+	mu     sync.Mutex
+	logger *logger
+
 	rs   string
 	node string
 
 	buf    Buffer
 	bufSet atomic.Uint32
 
 	pauseMgo int32
+
+	logLevel Severity
+	logJSON  bool
 }
 
-func New(cn *mongo.Collection, rs, node string) Logger {
+// New creates default logger which outputs to stderr.
+func New(conn connect.Client, rs, node string) Logger {
 	return &loggerImpl{
-		cn:   cn,
-		out:  os.Stderr,
-		rs:   rs,
-		node: node,
+		conn:     conn,
+		logger:   newStdLogger(),
+		rs:       rs,
+		node:     node,
+		logLevel: Debug,
+	}
+}
+
+// NewWithOpts creates logger based on provided options.
+func NewWithOpts(ctx context.Context, conn connect.Client, rs, node string, opts *Opts) Logger {
+	l := &loggerImpl{
+		conn:     conn,
+		rs:       rs,
+		node:     node,
+		logLevel: strToSeverity(opts.LogLevel),
+		logJSON:  opts.LogJSON,
+	}
+
+	l.mu.Lock()
+	l.createLogger(opts.LogPath)
+	l.mu.Unlock()
+
+	return l
+}
+
+// Opts represents options that are specified by the user.
+type Opts struct {
+	LogPath  string
+	LogJSON  bool
+	LogLevel string
+}
+
+type logger struct {
+	out     io.Writer
+	logPath string
+}
+
+func newStdLogger() *logger {
+	return &logger{
+		out:     os.Stderr,
+		logPath: logPathStdErr,
+	}
+}
+
+func newFileLogger(logPath string) (*logger, error) {
+	fullpath, err := filepath.Abs(logPath)
+	if err != nil {
+		return nil, errors.Wrap(err, "abs")
+	}
+
+	fileInfo, err := os.Stat(fullpath)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			return nil, errors.Wrap(err, "stat")
+		}
+
+		dirpath := filepath.Dir(fullpath)
+		err = os.MkdirAll(dirpath, fs.ModeDir|0o777)
+		if err != nil {
+			return nil, errors.Wrap(err, "mkdir -p")
+		}
+	} else if fileInfo.IsDir() {
+		return nil, errors.New("path is dir")
+	}
+
+	out, err := os.OpenFile(fullpath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
+	if err != nil {
+		return nil, errors.Wrap(err, "open file")
+	}
+
+	return &logger{
+		out:     out,
+		logPath: logPath,
+	}, nil
+}
+
+func (l *logger) close() {
+	if l == nil || l.out == nil ||
+		l.logPath == "" || l.logPath == logPathStdErr {
+		return
+	}
+	if o, ok := l.out.(io.Closer); ok {
+		o.Close()
+	}
+}
+
+// createLogger creates file/stderr type of logger based on logPath.
+// In case of an error during file logger creation, it falls back to stderr logger.
+func (l *loggerImpl) createLogger(logPath string) {
+	// close old one first
+	l.logger.close()
+
+	// and create new one
+	if strings.TrimSpace(logPath) == "" || logPath == logPathStdErr {
+		l.logger = newStdLogger()
+	} else {
+		fl, err := newFileLogger(logPath)
+		if err != nil {
+			l.logger = newStdLogger()
+			log.Printf("[ERROR] error while creating file logger: %v", err)
+		} else {
+			l.logger = fl
+		}
 	}
 }
 
@@ -77,6 +191,13 @@ func (l *loggerImpl) ResumeMgo() {
 	atomic.StoreInt32(&l.pauseMgo, 0)
 }
 
+func (l *loggerImpl) Write(p []byte) (int, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	return l.logger.out.Write(p)
+}
+
 func (l *loggerImpl) output(
 	s Severity,
 	event,
@@ -111,7 +232,7 @@ func (l *loggerImpl) output(
 
 	err := l.Output(context.TODO(), e)
 	if err != nil {
-		log.Printf("[ERROR] wrting log: %v, entry: %s", err, e)
+		log.Printf("[ERROR] writing log: %v, entry: %s", err, e)
 	}
 }
 
@@ -142,8 +263,9 @@ func (l *loggerImpl) Fatal(event, obj, opid string, epoch primitive.Timestamp, m
 func (l *loggerImpl) Output(ctx context.Context, e *Entry) error {
 	var rerr error
 
-	if l.cn != nil && atomic.LoadInt32(&l.pauseMgo) == 0 {
-		_, err := l.cn.InsertOne(ctx, e)
+	// conn is nil during physical restore
+	if l.conn != nil && l.conn.LogCollection() != nil && atomic.LoadInt32(&l.pauseMgo) == 0 {
+		_, err := l.conn.LogCollection().InsertOne(ctx, e)
 		if err != nil {
 			rerr = errors.Wrap(err, "db")
 		}
@@ -162,10 +284,18 @@ func (l *loggerImpl) Output(ctx context.Context, e *Entry) error {
 		}
 	}
 
-	if l.out != nil {
-		_, err := l.out.Write(append([]byte(e.String()), '\n'))
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.logger != nil && l.logLevel >= e.Severity {
+		var err error
+		if l.logJSON {
+			err = json.NewEncoder(l.logger.out).Encode(e)
+			err = errors.Wrap(err, "io json")
+		} else {
+			_, err = l.logger.out.Write(append([]byte(e.String()), '\n'))
+			err = errors.Wrap(err, "io text")
+		}
 
-		err = errors.Wrap(err, "io")
 		if rerr != nil {
 			rerr = errors.Errorf("%v, %v", rerr, err)
 		} else {