diff --git a/internal/orchestrator/logging/logging.go b/internal/orchestrator/logging/logging.go index 91276117..5e8178c3 100644 --- a/internal/orchestrator/logging/logging.go +++ b/internal/orchestrator/logging/logging.go @@ -35,7 +35,7 @@ func Logger(ctx context.Context) *zap.Logger { } p := zap.NewProductionEncoderConfig() p.EncodeTime = zapcore.ISO8601TimeEncoder - fe := zapcore.NewJSONEncoder(p) + fe := zapcore.NewConsoleEncoder(p) l := zap.New(zapcore.NewTee( zap.L().Core(), zapcore.NewCore(fe, zapcore.AddSync(writer), zapcore.DebugLevel), diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 78114435..d62af28d 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -325,6 +325,11 @@ func (o *Orchestrator) Run(ctx context.Context) { } err := t.Task.Run(taskCtx, t.ScheduledTask, runner) + if err != nil { + fmt.Fprintf(logs, "\ntask %q returned error: %v\n", t.Task.Name(), err) + } else { + fmt.Fprintf(logs, "\ntask %q completed successfully\n", t.Task.Name()) + } if op != nil { // write logs to log storage for this task. diff --git a/internal/orchestrator/repo/logging.go b/internal/orchestrator/repo/logging.go index d72793ff..369229d3 100644 --- a/internal/orchestrator/repo/logging.go +++ b/internal/orchestrator/repo/logging.go @@ -1,9 +1,11 @@ package repo import ( + "bytes" "context" + "fmt" + "io" - "github.com/garethgeorge/backrest/internal/ioutil" "github.com/garethgeorge/backrest/internal/orchestrator/logging" "github.com/garethgeorge/backrest/pkg/restic" ) @@ -15,8 +17,78 @@ func forwardResticLogs(ctx context.Context) (context.Context, func()) { if writer == nil { return ctx, func() {} } - capture := ioutil.NewOutputCapturer(64 * 1024) + limit := &limitWriter{w: writer, n: 64 * 1024} + capture := &linePrefixer{w: limit, prefix: []byte("[restic] ")} return restic.ContextWithLogger(ctx, capture), func() { - _, _ = writer.Write(capture.Bytes()) + if limit.d > 0 { + fmt.Fprintf(writer, "Output truncated, %d bytes dropped\n", limit.d) + } + capture.Close() } } + +// limitWriter is a writer that limits the number of bytes written to it. +type limitWriter struct { + w io.Writer + n int + d int +} + +func (l *limitWriter) Write(p []byte) (rnw int, err error) { + rnw = len(p) + if l.n <= 0 { + l.d += len(p) + return 0, nil + } + if len(p) > l.n { + l.d += len(p) - l.n + p = p[:l.n] + } + _, err = l.w.Write(p) + l.n -= len(p) + return +} + +type linePrefixer struct { + w io.Writer + buf []byte + prefix []byte +} + +func (l *linePrefixer) Write(p []byte) (n int, err error) { + n = len(p) + l.buf = append(l.buf, p...) + + bufOrig := l.buf + wroteLines := false + for { + i := bytes.IndexByte(l.buf, '\n') + if i < 0 { + break + } + wroteLines = true + if _, err := l.w.Write(l.prefix); err != nil { + return 0, err + } + if _, err := l.w.Write(l.buf[:i+1]); err != nil { + return 0, err + } + l.buf = l.buf[i+1:] + } + if wroteLines { + l.buf = append(bufOrig[:0], l.buf...) + } + return +} + +func (l *linePrefixer) Close() error { + if len(l.buf) > 0 { + if _, err := l.w.Write(l.prefix); err != nil { + return err + } + if _, err := l.w.Write(l.buf); err != nil { + return err + } + } + return nil +} diff --git a/pkg/restic/error.go b/pkg/restic/error.go index bb1da609..a4c9af9a 100644 --- a/pkg/restic/error.go +++ b/pkg/restic/error.go @@ -29,13 +29,14 @@ func (e *CmdError) Is(target error) bool { // newCmdError creates a new error indicating that running a command failed. func newCmdError(ctx context.Context, cmd *exec.Cmd, err error) *CmdError { - cerr := &CmdError{ - Command: cmd.String(), - Err: err, + shortCmd := cmd.String() + if len(shortCmd) > 100 { + shortCmd = shortCmd[:100] + "..." } - if logger := LoggerFromContext(ctx); logger != nil { - logger.Write([]byte(cerr.Error())) + cerr := &CmdError{ + Command: shortCmd, + Err: err, } return cerr } diff --git a/pkg/restic/restic.go b/pkg/restic/restic.go index 88287669..c1ed7194 100644 --- a/pkg/restic/restic.go +++ b/pkg/restic/restic.go @@ -66,9 +66,6 @@ func (r *Repo) commandWithContext(ctx context.Context, args []string, opts ...Ge sw := &ioutil.SynchronizedWriter{W: logger} cmd.Stderr = sw cmd.Stdout = sw - } - - if logger := LoggerFromContext(ctx); logger != nil { fmt.Fprintf(logger, "\ncommand: %v %v\n", fullCmd[0], strings.Join(fullCmd[1:], " ")) } @@ -185,7 +182,49 @@ func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func } } } - return summary, newCmdError(ctx, cmd, newErrorWithOutput(errors.Join(cmdErr, readErr), outputForErr.String())) + return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr)) + } + return summary, nil +} + +func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*RestoreProgressEntry), opts ...GenericOption) (*RestoreProgressEntry, error) { + opts = append(slices.Clone(opts), WithEnv("RESTIC_PROGRESS_FPS=2")) + cmdCtx, cancel := context.WithCancel(ctx) + cmd := r.commandWithContext(cmdCtx, []string{"restore", "--json", snapshot}, opts...) + buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing + reader, writer := nio.Pipe(buf) + r.pipeCmdOutputToWriter(cmd, writer) + + var readErr error + var summary *RestoreProgressEntry + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + defer cancel() + var err error + summary, err = readRestoreProgressEntries(reader, callback) + if err != nil { + readErr = fmt.Errorf("processing command output: %w", err) + } + }() + + cmdErr := cmd.Run() + writer.Close() + wg.Wait() + if cmdErr != nil || readErr != nil { + if cmdErr != nil { + var exitErr *exec.ExitError + if errors.As(cmdErr, &exitErr) { + if exitErr.ExitCode() == 3 { + cmdErr = ErrPartialBackup + } else { + cmdErr = fmt.Errorf("exit code %d: %w", exitErr.ExitCode(), ErrBackupFailed) + } + } + } + + return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr)) } return summary, nil } @@ -275,48 +314,6 @@ func (r *Repo) Check(ctx context.Context, checkOutput io.Writer, opts ...Generic return nil } -func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*RestoreProgressEntry), opts ...GenericOption) (*RestoreProgressEntry, error) { - opts = append(slices.Clone(opts), WithEnv("RESTIC_PROGRESS_FPS=2")) - cmdCtx, cancel := context.WithCancel(ctx) - cmd := r.commandWithContext(cmdCtx, []string{"restore", "--json", snapshot}, opts...) - buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing - reader, writer := nio.Pipe(buf) - r.pipeCmdOutputToWriter(cmd, writer) - - var readErr error - var summary *RestoreProgressEntry - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - defer cancel() - var err error - summary, err = readRestoreProgressEntries(reader, callback) - if err != nil { - readErr = fmt.Errorf("processing command output: %w", err) - } - }() - - cmdErr := cmd.Run() - writer.Close() - wg.Wait() - if cmdErr != nil || readErr != nil { - if cmdErr != nil { - var exitErr *exec.ExitError - if errors.As(cmdErr, &exitErr) { - if exitErr.ExitCode() == 3 { - cmdErr = ErrPartialBackup - } else { - cmdErr = fmt.Errorf("exit code %d: %w", exitErr.ExitCode(), ErrBackupFailed) - } - } - } - - return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr)) - } - return summary, nil -} - func (r *Repo) ListDirectory(ctx context.Context, snapshot string, path string, opts ...GenericOption) (*Snapshot, []*LsEntry, error) { if path == "" { // an empty path can trigger very expensive operations (e.g. iterates all files in the snapshot)