diff --git a/internal/orchestrator/tasks/taskrestore.go b/internal/orchestrator/tasks/taskrestore.go index 5f1ff1f7..c74f13c8 100644 --- a/internal/orchestrator/tasks/taskrestore.go +++ b/internal/orchestrator/tasks/taskrestore.go @@ -86,7 +86,7 @@ func restoreHelper(ctx context.Context, st ScheduledTask, taskRunner TaskRunner, }) if err != nil { - return fmt.Errorf("restore failed: %w", err) + return err } restoreOp.LastStatus = summary diff --git a/pkg/restic/outputs.go b/pkg/restic/outputs.go index 52b57247..d3bc55f8 100644 --- a/pkg/restic/outputs.go +++ b/pkg/restic/outputs.go @@ -126,7 +126,7 @@ func (b *BackupProgressEntry) Validate() error { } // readBackupProgressEntries returns the summary event or an error if the command failed. -func readBackupProgressEntries(output io.Reader, callback func(event *BackupProgressEntry)) (*BackupProgressEntry, error) { +func readBackupProgressEntries(output io.Reader, logger io.Writer, callback func(event *BackupProgressEntry)) (*BackupProgressEntry, error) { scanner := bufio.NewScanner(output) scanner.Split(bufio.ScanLines) @@ -139,16 +139,28 @@ func readBackupProgressEntries(output io.Reader, callback func(event *BackupProg var event BackupProgressEntry if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { nonJSONOutput.Write(scanner.Bytes()) + if logger != nil { + logger.Write(scanner.Bytes()) + } continue } if err := event.Validate(); err != nil { nonJSONOutput.Write(scanner.Bytes()) + if logger != nil { + logger.Write(scanner.Bytes()) + } continue } + if event.MessageType == "error" && logger != nil { + logger.Write(scanner.Bytes()) + } if callback != nil { callback(&event) } if event.MessageType == "summary" { + if logger != nil { + logger.Write(scanner.Bytes()) + } summary = &event } } @@ -250,7 +262,7 @@ func (e *RestoreProgressEntry) Validate() error { } // readRestoreProgressEntries returns the summary event or an error if the command failed. -func readRestoreProgressEntries(output io.Reader, callback func(event *RestoreProgressEntry)) (*RestoreProgressEntry, error) { +func readRestoreProgressEntries(output io.Reader, logger io.Writer, callback func(event *RestoreProgressEntry)) (*RestoreProgressEntry, error) { scanner := bufio.NewScanner(output) scanner.Split(bufio.ScanLines) @@ -263,18 +275,30 @@ func readRestoreProgressEntries(output io.Reader, callback func(event *RestorePr var event RestoreProgressEntry if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { nonJSONOutput.Write(scanner.Bytes()) + if logger != nil { + logger.Write(scanner.Bytes()) + } continue } if err := event.Validate(); err != nil { // skip it. Best effort parsing, restic will return with a non-zero exit code if it fails. nonJSONOutput.Write(scanner.Bytes()) + if logger != nil { + logger.Write(scanner.Bytes()) + } continue } + if event.MessageType == "error" && logger != nil { + logger.Write(scanner.Bytes()) + } if callback != nil { callback(&event) } if event.MessageType == "summary" { + if logger != nil { + logger.Write(scanner.Bytes()) + } summary = &event } } diff --git a/pkg/restic/outputs_test.go b/pkg/restic/outputs_test.go index 1d8bab30..af52d07a 100644 --- a/pkg/restic/outputs_test.go +++ b/pkg/restic/outputs_test.go @@ -12,7 +12,7 @@ func TestReadBackupProgressEntries(t *testing.T) { b := bytes.NewBuffer([]byte(testInput)) - summary, err := readBackupProgressEntries(b, func(event *BackupProgressEntry) { + summary, err := readBackupProgressEntries(b, nil, func(event *BackupProgressEntry) { t.Logf("event: %v", event) }) if err != nil { diff --git a/pkg/restic/restic.go b/pkg/restic/restic.go index 3a8f9103..17f40746 100644 --- a/pkg/restic/restic.go +++ b/pkg/restic/restic.go @@ -162,8 +162,16 @@ func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func args = append(args, paths...) opts = append(slices.Clone(opts), WithEnv("RESTIC_PROGRESS_FPS=2")) + logger := LoggerFromContext(ctx) cmdCtx, cancel := context.WithCancel(ctx) + cmdCtx = ContextWithLogger(cmdCtx, nil) // ensure no logger is used cmd := r.commandWithContext(cmdCtx, args, opts...) + + // Ensure the command is logged since we're overriding the logger + if logger != nil { + fmt.Fprintf(logger, "command: %v %v\n", cmd.Path, strings.Join(cmd.Args, " ")) + } + buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing reader, writer := nio.Pipe(buf) r.pipeCmdOutputToWriter(cmd, writer) @@ -176,7 +184,7 @@ func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func defer wg.Done() defer cancel() var err error - summary, err = readBackupProgressEntries(reader, progressCallback) + summary, err = readBackupProgressEntries(reader, logger, progressCallback) if err != nil { readErr = fmt.Errorf("processing command output: %w", err) } @@ -204,8 +212,14 @@ func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func 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")) + + logger := LoggerFromContext(ctx) cmdCtx, cancel := context.WithCancel(ctx) + cmdCtx = ContextWithLogger(cmdCtx, nil) // ensure no logger is used cmd := r.commandWithContext(cmdCtx, []string{"restore", "--json", snapshot}, opts...) + if logger != nil { + fmt.Fprintf(logger, "command: %v %v\n", cmd.Path, strings.Join(cmd.Args, " ")) + } buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing reader, writer := nio.Pipe(buf) r.pipeCmdOutputToWriter(cmd, writer) @@ -218,7 +232,7 @@ func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*Rest defer wg.Done() defer cancel() var err error - summary, err = readRestoreProgressEntries(reader, callback) + summary, err = readRestoreProgressEntries(reader, logger, callback) if err != nil { readErr = fmt.Errorf("processing command output: %w", err) }