From 7bfd2ac5e185dfec7b8ccb903fc9bcd17db16dc4 Mon Sep 17 00:00:00 2001 From: garethgeorge Date: Sat, 30 Mar 2024 11:18:28 -0700 Subject: [PATCH] feat: improve consistency of restic command execution and output capture --- internal/hook/hook.go | 13 +-- internal/orchestrator/repo.go | 37 ++++-- pkg/restic/logging.go | 31 +++++ pkg/restic/restic.go | 213 +++++++++++----------------------- pkg/restic/restic_test.go | 45 ++++--- 5 files changed, 154 insertions(+), 185 deletions(-) create mode 100644 pkg/restic/logging.go diff --git a/internal/hook/hook.go b/internal/hook/hook.go index 5d28a56f..2a2e6e8f 100644 --- a/internal/hook/hook.go +++ b/internal/hook/hook.go @@ -1,7 +1,6 @@ package hook import ( - "bufio" "bytes" "fmt" "io" @@ -103,20 +102,12 @@ func (e *HookExecutor) executeHook(op *v1.Operation, hook *Hook, event v1.Hook_C } output := &bytes.Buffer{} - pr, pw := io.Pipe() - go func() { - defer pr.Close() - scanner := bufio.NewScanner(pr) - for scanner.Scan() { - zap.S().Debugf("hook output: %v", scanner.Text()) - } - }() - defer pw.Close() - if err := hook.Do(event, vars, io.MultiWriter(output, pw)); err != nil { + if err := hook.Do(event, vars, io.MultiWriter(output)); err != nil { output.Write([]byte(fmt.Sprintf("Error: %v", err))) op.DisplayMessage = err.Error() op.Status = v1.OperationStatus_STATUS_ERROR + zap.S().Errorf("execute hook: %v", err) } else { op.Status = v1.OperationStatus_STATUS_SUCCESS } diff --git a/internal/orchestrator/repo.go b/internal/orchestrator/repo.go index b795fc07..58cf86fc 100644 --- a/internal/orchestrator/repo.go +++ b/internal/orchestrator/repo.go @@ -4,7 +4,9 @@ import ( "context" "fmt" "io" + "slices" "sort" + "strings" "sync" "time" @@ -47,6 +49,13 @@ func NewRepoOrchestrator(repoConfig *v1.Repo, resticPath string) (*RepoOrchestra opts = append(opts, restic.WithFlags(args...)) } + // Add BatchMode=yes to sftp.args if it's not already set. + if slices.IndexFunc(repoConfig.GetFlags(), func(a string) bool { + return strings.Contains(a, "sftp.args") + }) == -1 { + opts = append(opts, restic.WithFlags("-o", "sftp.args=-oBatchMode=yes")) + } + if env := repoConfig.GetEnv(); len(env) != 0 { opts = append(opts, restic.WithEnv(repoConfig.GetEnv()...)) } @@ -103,25 +112,29 @@ func (r *RepoOrchestrator) Backup(ctx context.Context, plan *v1.Plan, progressCa startTime := time.Now() - var opts []restic.BackupOption - opts = append(opts, restic.WithBackupPaths(plan.Paths...)) - opts = append(opts, restic.WithBackupExcludes(plan.Excludes...)) - opts = append(opts, restic.WithBackupIExcludes(plan.Iexcludes...)) - opts = append(opts, restic.WithBackupTags(tagForPlan(plan))) + var opts []restic.GenericOption + opts = append(opts, restic.WithFlags("--exclude-caches")) + opts = append(opts, restic.WithFlags("--tag", tagForPlan(plan))) + for _, exclude := range plan.Excludes { + opts = append(opts, restic.WithFlags("--exclude", exclude)) + } + for _, iexclude := range plan.Iexcludes { + opts = append(opts, restic.WithFlags("--iexclude", iexclude)) + } + + if len(snapshots) > 0 { + opts = append(opts, restic.WithFlags("--parent", snapshots[len(snapshots)-1].Id)) + } + for _, f := range plan.GetBackupFlags() { args, err := shlex.Split(f) if err != nil { return nil, fmt.Errorf("failed to parse backup flag %q for plan %q: %w", f, plan.Id, err) } - opts = append(opts, restic.WithBackupFlags(args...)) + opts = append(opts, restic.WithFlags(args...)) } - if len(snapshots) > 0 { - // TODO: design a test strategy to verify that the backup parent is used correctly. - opts = append(opts, restic.WithBackupParent(snapshots[len(snapshots)-1].Id)) - } - - summary, err := r.repo.Backup(ctx, progressCallback, opts...) + summary, err := r.repo.Backup(ctx, plan.Paths, progressCallback, opts...) if err != nil { return summary, fmt.Errorf("failed to backup: %w", err) } diff --git a/pkg/restic/logging.go b/pkg/restic/logging.go new file mode 100644 index 00000000..c3e367b4 --- /dev/null +++ b/pkg/restic/logging.go @@ -0,0 +1,31 @@ +package restic + +import ( + "context" + "io" + "os/exec" +) + +var loggerKey = struct{}{} + +func ContextWithLogger(ctx context.Context, logger io.Writer) context.Context { + return context.WithValue(ctx, loggerKey, logger) +} + +func LoggerFromContext(ctx context.Context) io.Writer { + writer, _ := ctx.Value(loggerKey).(io.Writer) + return writer +} + +func addLoggingToCommand(ctx context.Context, cmd *exec.Cmd) { + logger := LoggerFromContext(ctx) + if logger == nil { + return + } + if cmd.Stdout != nil { + cmd.Stdout = io.MultiWriter(cmd.Stdout, logger) + } + if cmd.Stderr != nil { + cmd.Stderr = io.MultiWriter(cmd.Stderr, logger) + } +} diff --git a/pkg/restic/restic.go b/pkg/restic/restic.go index 3c47800c..340b16b4 100644 --- a/pkg/restic/restic.go +++ b/pkg/restic/restic.go @@ -34,12 +34,6 @@ func NewRepo(resticBin string, uri string, opts ...GenericOption) *Repo { o(opt) } - if slices.IndexFunc(opt.extraArgs, func(a string) bool { - return strings.Contains(a, "sftp.args") - }) == -1 { - // opt.extraArgs = append(opt.extraArgs, "-o", "sftp.args=-oBatchMode=yes") - } - opt.extraEnv = append(opt.extraEnv, "RESTIC_REPOSITORY="+uri) return &Repo{ @@ -51,15 +45,9 @@ func NewRepo(resticBin string, uri string, opts ...GenericOption) *Repo { } } -// init initializes the repo, the command will be cancelled with the context. -func (r *Repo) init(ctx context.Context, opts ...GenericOption) error { - if r.initialized { - return nil - } - +func (r *Repo) commandWithContext(ctx context.Context, args []string, opts ...GenericOption) *exec.Cmd { opt := resolveOpts(opts) - var args = []string{"init", "--json"} args = append(args, r.extraArgs...) args = append(args, opt.extraArgs...) @@ -67,11 +55,45 @@ func (r *Repo) init(ctx context.Context, opts ...GenericOption) error { cmd.Env = append(cmd.Env, r.extraEnv...) cmd.Env = append(cmd.Env, opt.extraEnv...) - if output, err := cmd.CombinedOutput(); err != nil { - if strings.Contains(string(output), "config file already exists") || strings.Contains(string(output), "already initialized") { + addLoggingToCommand(ctx, cmd) + + if logger := LoggerFromContext(ctx); logger != nil { + fmt.Fprintf(logger, "command: %v %v\n", r.cmd, strings.Join(args, " ")) + } + + return cmd +} + +func (r *Repo) pipeCmdOutputToWriter(cmd *exec.Cmd, handlers ...io.Writer) { + stdoutHandlers := slices.Clone(handlers) + stderrHandlers := slices.Clone(handlers) + + if cmd.Stdout != nil { + handlers = append(stdoutHandlers, cmd.Stdout) + } + if cmd.Stderr != nil { + handlers = append(stderrHandlers, cmd.Stderr) + } + + cmd.Stdout = io.MultiWriter(handlers...) + cmd.Stderr = io.MultiWriter(handlers...) +} + +// init initializes the repo, the command will be cancelled with the context. +func (r *Repo) init(ctx context.Context, opts ...GenericOption) error { + if r.initialized { + return nil + } + + cmd := r.commandWithContext(ctx, []string{"init", "--json"}, opts...) + output := bytes.NewBuffer(nil) + r.pipeCmdOutputToWriter(cmd, output) + + if err := cmd.Run(); err != nil { + if strings.Contains(output.String(), "config file already exists") || strings.Contains(output.String(), "already initialized") { return errAlreadyInitialized } - return newCmdError(cmd, string(output), err) + return newCmdError(cmd, output.String(), err) } r.initialized = true @@ -85,32 +107,20 @@ func (r *Repo) Init(ctx context.Context, opts ...GenericOption) error { return nil } -func (r *Repo) Backup(ctx context.Context, progressCallback func(*BackupProgressEntry), opts ...BackupOption) (*BackupProgressEntry, error) { - opt := &BackupOpts{} - for _, o := range opts { - o(opt) - } - - for _, p := range opt.paths { +func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func(*BackupProgressEntry), opts ...GenericOption) (*BackupProgressEntry, error) { + for _, p := range paths { if _, err := os.Stat(p); err != nil { return nil, fmt.Errorf("path %s does not exist: %w", p, err) } } args := []string{"backup", "--json", "--exclude-caches"} - args = append(args, r.extraArgs...) - args = append(args, opt.paths...) - args = append(args, opt.extraArgs...) + args = append(args, paths...) - output := newOutputCapturer(outputBufferLimit) + cmd := r.commandWithContext(ctx, args, opts...) + capture := newOutputCapturer(outputBufferLimit) reader, writer := io.Pipe() - capture := io.MultiWriter(output, writer) - - cmd := exec.CommandContext(ctx, r.cmd, args...) - cmd.Env = append(cmd.Env, r.extraEnv...) - cmd.StdoutPipe() - cmd.Stderr = capture - cmd.Stdout = capture + r.pipeCmdOutputToWriter(cmd, writer, capture) if err := cmd.Start(); err != nil { return nil, newCmdError(cmd, "", err) @@ -152,32 +162,26 @@ func (r *Repo) Backup(ctx context.Context, progressCallback func(*BackupProgress wg.Wait() if cmdErr != nil || readErr != nil { - return summary, newCmdErrorPreformatted(cmd, output.String(), errors.Join(cmdErr, readErr)) + return summary, newCmdErrorPreformatted(cmd, capture.String(), errors.Join(cmdErr, readErr)) } return summary, nil } func (r *Repo) Snapshots(ctx context.Context, opts ...GenericOption) ([]*Snapshot, error) { - opt := resolveOpts(opts) + cmd := r.commandWithContext(ctx, []string{"snapshots", "--json"}, opts...) + output := bytes.NewBuffer(nil) + r.pipeCmdOutputToWriter(cmd, output) - args := []string{"snapshots", "--json"} - args = append(args, r.extraArgs...) - args = append(args, opt.extraArgs...) - - cmd := exec.CommandContext(ctx, r.cmd, args...) - cmd.Env = append(cmd.Env, r.extraEnv...) - cmd.Env = append(cmd.Env, opt.extraEnv...) - - output, err := cmd.CombinedOutput() - if err != nil { - return nil, newCmdError(cmd, "", err) + if err := cmd.Run(); err != nil { + return nil, newCmdError(cmd, output.String(), err) } var snapshots []*Snapshot - if err := json.Unmarshal(output, &snapshots); err != nil { - return nil, newCmdError(cmd, string(output), fmt.Errorf("command output is not valid JSON: %w", err)) + if err := json.Unmarshal(output.Bytes(), &snapshots); err != nil { + return nil, newCmdError(cmd, output.String(), fmt.Errorf("command output is not valid JSON: %w", err)) } + for _, snapshot := range snapshots { if err := snapshot.Validate(); err != nil { return nil, fmt.Errorf("invalid snapshot: %w", err) @@ -187,82 +191,54 @@ func (r *Repo) Snapshots(ctx context.Context, opts ...GenericOption) ([]*Snapsho } func (r *Repo) Forget(ctx context.Context, policy *RetentionPolicy, opts ...GenericOption) (*ForgetResult, error) { - // first run the forget command - opt := resolveOpts(opts) - args := []string{"forget", "--json"} - args = append(args, r.extraArgs...) - args = append(args, opt.extraArgs...) args = append(args, policy.toForgetFlags()...) - cmd := exec.CommandContext(ctx, r.cmd, args...) - cmd.Env = append(cmd.Env, r.extraEnv...) - cmd.Env = append(cmd.Env, opt.extraEnv...) - - output, err := cmd.CombinedOutput() - if err != nil { - return nil, newCmdError(cmd, string(output), err) + cmd := r.commandWithContext(ctx, args, opts...) + output := bytes.NewBuffer(nil) + r.pipeCmdOutputToWriter(cmd, output) + if err := cmd.Run(); err != nil { + return nil, newCmdError(cmd, output.String(), err) } var result []ForgetResult - if err := json.Unmarshal(output, &result); err != nil { - return nil, newCmdError(cmd, string(output), fmt.Errorf("command output is not valid JSON: %w", err)) + if err := json.Unmarshal(output.Bytes(), &result); err != nil { + return nil, newCmdError(cmd, output.String(), fmt.Errorf("command output is not valid JSON: %w", err)) } if len(result) != 1 { return nil, fmt.Errorf("expected 1 output from forget, got %v", len(result)) } if err := result[0].Validate(); err != nil { - return nil, newCmdError(cmd, string(output), fmt.Errorf("invalid forget result: %w", err)) + return nil, newCmdError(cmd, output.String(), fmt.Errorf("invalid forget result: %w", err)) } return &result[0], nil } func (r *Repo) ForgetSnapshot(ctx context.Context, snapshotId string, opts ...GenericOption) error { - opt := resolveOpts(opts) - args := []string{"forget", "--json", snapshotId} - args = append(args, r.extraArgs...) - args = append(args, opt.extraArgs...) - args = append(args, snapshotId) - cmd := exec.CommandContext(ctx, r.cmd, args...) - cmd.Env = append(cmd.Env, r.extraEnv...) - cmd.Env = append(cmd.Env, opt.extraEnv...) - - output, err := cmd.CombinedOutput() - if err != nil { - return newCmdError(cmd, string(output), err) + cmd := r.commandWithContext(ctx, args, opts...) + output := bytes.NewBuffer(nil) + r.pipeCmdOutputToWriter(cmd, output) + if err := cmd.Run(); err != nil { + return newCmdError(cmd, output.String(), err) } return nil } func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...GenericOption) error { - opt := resolveOpts(opts) - args := []string{"prune"} - args = append(args, r.extraArgs...) - args = append(args, opt.extraArgs...) - - cmd := exec.CommandContext(ctx, r.cmd, args...) - cmd.Env = append(cmd.Env, r.extraEnv...) - cmd.Env = append(cmd.Env, opt.extraEnv...) - - var output = newOutputCapturer(outputBufferLimit) - var writer io.Writer = output + cmd := r.commandWithContext(ctx, args, opts...) + output := bytes.NewBuffer(nil) + r.pipeCmdOutputToWriter(cmd, output) if pruneOutput != nil { - writer = io.MultiWriter(pruneOutput, output) + r.pipeCmdOutputToWriter(cmd, pruneOutput) } - cmd.Stdout = writer - cmd.Stderr = writer - - writer.Write([]byte("command: " + strings.Join(cmd.Args, " ") + "\n")) - if err := cmd.Run(); err != nil { return newCmdErrorPreformatted(cmd, output.String(), err) } - return nil } @@ -428,55 +404,6 @@ func (r *RetentionPolicy) toForgetFlags() []string { return flags } -type BackupOpts struct { - paths []string - extraArgs []string -} - -type BackupOption func(opts *BackupOpts) - -func WithBackupPaths(paths ...string) BackupOption { - return func(opts *BackupOpts) { - opts.paths = append(opts.paths, paths...) - } -} - -func WithBackupExcludes(excludes ...string) BackupOption { - return func(opts *BackupOpts) { - for _, exclude := range excludes { - opts.extraArgs = append(opts.extraArgs, "--exclude", exclude) - } - } -} - -func WithBackupIExcludes(iexcludes ...string) BackupOption { - return func(opts *BackupOpts) { - for _, iexclude := range iexcludes { - opts.extraArgs = append(opts.extraArgs, "--iexclude", iexclude) - } - } -} - -func WithBackupTags(tags ...string) BackupOption { - return func(opts *BackupOpts) { - for _, tag := range tags { - opts.extraArgs = append(opts.extraArgs, "--tag", tag) - } - } -} - -func WithBackupParent(parent string) BackupOption { - return func(opts *BackupOpts) { - opts.extraArgs = append(opts.extraArgs, "--parent", parent) - } -} - -func WithBackupFlags(flags ...string) BackupOption { - return func(opts *BackupOpts) { - opts.extraArgs = append(opts.extraArgs, flags...) - } -} - type GenericOpts struct { extraArgs []string extraEnv []string diff --git a/pkg/restic/restic_test.go b/pkg/restic/restic_test.go index c01f7d2d..67f764ec 100644 --- a/pkg/restic/restic_test.go +++ b/pkg/restic/restic_test.go @@ -40,38 +40,45 @@ func TestResticBackup(t *testing.T) { var tests = []struct { name string - opts []BackupOption + opts []GenericOption + paths []string files int // expected files at the end of the backup wantErr bool }{ { name: "no options", - opts: []BackupOption{WithBackupPaths(testData)}, + paths: []string{testData}, + opts: []GenericOption{}, files: 100, }, { name: "with two paths", - opts: []BackupOption{WithBackupPaths(testData), WithBackupPaths(testData2)}, + paths: []string{testData, testData2}, + opts: []GenericOption{}, files: 200, }, { name: "with exclude", - opts: []BackupOption{WithBackupPaths(testData), WithBackupExcludes("file1*")}, + paths: []string{testData}, + opts: []GenericOption{WithFlags("--exclude", "file1*")}, files: 90, }, { name: "with exclude pattern", - opts: []BackupOption{WithBackupPaths(testData), WithBackupExcludes("file*")}, + paths: []string{testData}, + opts: []GenericOption{WithFlags("--iexclude=file*")}, files: 0, }, { name: "with nothing to backup", - opts: []BackupOption{}, + paths: []string{}, + opts: []GenericOption{}, wantErr: true, }, { name: "with unreadable file", - opts: []BackupOption{WithBackupPaths(testData), WithBackupPaths(testDataUnreadable)}, + paths: []string{testData, testDataUnreadable}, + opts: []GenericOption{}, wantErr: true, }, } @@ -79,7 +86,7 @@ func TestResticBackup(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { gotEvent := false - summary, err := r.Backup(context.Background(), func(event *BackupProgressEntry) { + summary, err := r.Backup(context.Background(), tc.paths, func(event *BackupProgressEntry) { t.Logf("backup event: %v", event) gotEvent = true }, tc.opts...) @@ -121,9 +128,9 @@ func TestResticPartialBackup(t *testing.T) { var entries []*BackupProgressEntry - summary, err := r.Backup(context.Background(), func(entry *BackupProgressEntry) { + summary, err := r.Backup(context.Background(), []string{testDataUnreadable}, func(entry *BackupProgressEntry) { entries = append(entries, entry) - }, WithBackupPaths(testDataUnreadable)) + }) if !errors.Is(err, ErrPartialBackup) { t.Fatalf("wanted error to be partial backup, got: %v", err) } @@ -158,9 +165,9 @@ func TestResticBackupLots(t *testing.T) { // backup 25 times for i := 0; i < 25; i++ { - _, err := r.Backup(context.Background(), func(e *BackupProgressEntry) { + _, err := r.Backup(context.Background(), []string{testData}, func(e *BackupProgressEntry) { t.Logf("backup event: %+v", e) - }, WithBackupPaths(testData)) + }) if err != nil { t.Fatalf("failed to backup and create new snapshot: %v", err) } @@ -180,7 +187,7 @@ func TestSnapshot(t *testing.T) { testData := helpers.CreateTestData(t) for i := 0; i < 10; i++ { - _, err := r.Backup(context.Background(), nil, WithBackupPaths(testData), WithBackupTags(fmt.Sprintf("tag%d", i))) + _, err := r.Backup(context.Background(), []string{testData}, nil, WithFlags("--tag", fmt.Sprintf("tag%d", i))) if err != nil { t.Fatalf("failed to backup and create new snapshot: %v", err) } @@ -235,7 +242,7 @@ func TestLs(t *testing.T) { testData := helpers.CreateTestData(t) - snapshot, err := r.Backup(context.Background(), nil, WithBackupPaths(testData)) + snapshot, err := r.Backup(context.Background(), []string{testData}, nil) if err != nil { t.Fatalf("failed to backup and create new snapshot: %v", err) } @@ -264,7 +271,7 @@ func TestResticForget(t *testing.T) { ids := make([]string, 0) for i := 0; i < 10; i++ { - output, err := r.Backup(context.Background(), nil, WithBackupPaths(testData)) + output, err := r.Backup(context.Background(), []string{testData}, nil) if err != nil { t.Fatalf("failed to backup and create new snapshot: %v", err) } @@ -318,7 +325,7 @@ func TestForgetSnapshotId(t *testing.T) { ids := make([]string, 0) for i := 0; i < 5; i++ { - output, err := r.Backup(context.Background(), nil, WithBackupPaths(testData)) + output, err := r.Backup(context.Background(), []string{testData}, nil) if err != nil { t.Fatalf("failed to backup and create new snapshot: %v", err) } @@ -352,7 +359,7 @@ func TestResticPrune(t *testing.T) { testData := helpers.CreateTestData(t) for i := 0; i < 3; i++ { - _, err := r.Backup(context.Background(), nil, WithBackupPaths(testData)) + _, err := r.Backup(context.Background(), []string{testData}, nil) if err != nil { t.Fatalf("failed to backup: %v", err) } @@ -390,7 +397,7 @@ func TestResticRestore(t *testing.T) { testData := helpers.CreateTestData(t) - snapshot, err := r.Backup(context.Background(), nil, WithBackupPaths(testData)) + snapshot, err := r.Backup(context.Background(), []string{testData}, nil) if err != nil { t.Fatalf("failed to backup and create new snapshot: %v", err) } @@ -420,7 +427,7 @@ func TestResticStats(t *testing.T) { testData := helpers.CreateTestData(t) - _, err := r.Backup(context.Background(), nil, WithBackupPaths(testData)) + _, err := r.Backup(context.Background(), []string{testData}, nil) if err != nil { t.Fatalf("failed to backup and create new snapshot: %v", err) }