diff --git a/go.mod b/go.mod index 8654104a..94554b4a 100644 --- a/go.mod +++ b/go.mod @@ -20,7 +20,6 @@ require ( require ( github.com/daaku/go.zipexe v1.0.2 // indirect github.com/golang/protobuf v1.5.3 // indirect - github.com/google/renameio v1.0.1 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect go.uber.org/multierr v1.11.0 // indirect diff --git a/go.sum b/go.sum index 2a9193ea..d0b912de 100644 --- a/go.sum +++ b/go.sum @@ -14,8 +14,6 @@ github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiu github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= -github.com/google/renameio v1.0.1 h1:Lh/jXZmvZxb0BBeSY5VKEfidcbcbenKjZFzM/q0fSeU= -github.com/google/renameio v1.0.1/go.mod h1:t/HQoYBZSsWSNK35C6CO/TpPLDVWvxOHboWUAweKUpk= github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1 h1:HcUWd006luQPljE73d5sk+/VgYPGUReEVz2y1/qylwY= github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1/go.mod h1:w9Y7gY31krpLmrVU5ZPG9H7l9fZuRu5/3R3S3FMtVQ4= github.com/grpc-ecosystem/grpc-gateway/v2 v2.18.1 h1:6UKoz5ujsI55KNpsJH3UwCq3T8kKbZwNZBNPuTTje8U= diff --git a/internal/orchestrator/repo.go b/internal/orchestrator/repo.go index 2c3267f7..45ce8984 100644 --- a/internal/orchestrator/repo.go +++ b/internal/orchestrator/repo.go @@ -89,9 +89,6 @@ func (r *RepoOrchestrator) Backup(ctx context.Context, plan *v1.Plan, progressCa } func (r *RepoOrchestrator) ListSnapshotFiles(ctx context.Context, snapshotId string, path string) ([]*v1.LsEntry, error) { - r.mu.Lock() - defer r.mu.Unlock() - _, entries, err := r.repo.ListDirectory(ctx, snapshotId, path) if err != nil { return nil, fmt.Errorf("failed to list snapshot files: %w", err) diff --git a/pkg/restic/error.go b/pkg/restic/error.go index 9f458c86..30ad12d8 100644 --- a/pkg/restic/error.go +++ b/pkg/restic/error.go @@ -5,10 +5,12 @@ import ( "os/exec" ) +const outputBufferLimit = 1000 + type CmdError struct { Command string - Err error - Output string + Err error + Output string } func (e *CmdError) Error() string { @@ -28,18 +30,23 @@ func (e *CmdError) Is(target error) bool { return ok } -// NewCmdError creates a new error indicating that running a command failed. -func NewCmdError(cmd *exec.Cmd, output []byte, err error) *CmdError { +// newCmdError creates a new error indicating that running a command failed. +func newCmdError(cmd *exec.Cmd, output string, err error) *CmdError { cerr := &CmdError{ Command: cmd.String(), - Err: err, + Err: err, } - if len(output) > 0 { - if len(output) > 1000 { - output = output[:1000] - } - cerr.Output = string(output) + if len(output) >= outputBufferLimit { + cerr.Output = output[:outputBufferLimit] + "\n...[truncated]" } + return cerr -} \ No newline at end of file +} + +func newCmdErrorPreformatted(cmd *exec.Cmd, output string, err error) *CmdError { + return &CmdError{ + Command: cmd.String(), + Err: err, + } +} diff --git a/pkg/restic/io.go b/pkg/restic/io.go new file mode 100644 index 00000000..2e86cf3e --- /dev/null +++ b/pkg/restic/io.go @@ -0,0 +1,86 @@ +package restic + +import ( + "fmt" + "io" +) + +// headWriter keeps the first 'limit' bytes in memory. +type headWriter struct { + buf []byte + limit int +} + +var _ io.Writer = &headWriter{} + +func (w *headWriter) Write(p []byte) (n int, err error) { + if len(w.buf) >= w.limit { + return len(p), nil + } + w.buf = append(w.buf, p...) + if len(w.buf) > w.limit { + w.buf = w.buf[:w.limit] + } + return len(p), nil +} + +func (w *headWriter) Bytes() []byte { + return w.buf +} + +// tailWriter keeps the last 'limit' bytes in memory. +type tailWriter struct { + buf []byte + limit int +} + +var _ io.Writer = &tailWriter{} + +func (w *tailWriter) Write(p []byte) (n int, err error) { + w.buf = append(w.buf, p...) + if len(w.buf) > w.limit { + w.buf = w.buf[len(w.buf)-w.limit:] + } + return len(p), nil +} + +func (w *tailWriter) Bytes() []byte { + return w.buf +} + +type outputCapturer struct { + headWriter + tailWriter + limit int + totalBytes int +} + +var _ io.Writer = &outputCapturer{} + +func newOutputCapturer(limit int) *outputCapturer { + return &outputCapturer{ + headWriter: headWriter{limit: limit}, + tailWriter: tailWriter{limit: limit}, + limit: limit, + } +} + +func (w *outputCapturer) Write(p []byte) (n int, err error) { + w.headWriter.Write(p) + w.tailWriter.Write(p) + w.totalBytes += len(p) + return len(p), nil +} + +func (w *outputCapturer) String() string { + head := w.headWriter.Bytes() + tail := w.tailWriter.Bytes() + if w.totalBytes <= w.limit { + return string(head) + } + + head = head[:w.limit/2] + tail = tail[len(tail)-w.limit/2:] + + return fmt.Sprintf("%s...[%v bytes dropped]...%s", string(head), w.totalBytes-len(head)-len(tail), string(tail)) +} diff --git a/pkg/restic/io_test.go b/pkg/restic/io_test.go new file mode 100644 index 00000000..af995bd8 --- /dev/null +++ b/pkg/restic/io_test.go @@ -0,0 +1,24 @@ +package restic + +import "testing" + +func TestOutputCapture(t *testing.T) { + c := newOutputCapturer(100) + + c.Write([]byte("hello")) + + if c.String() != "hello" { + t.Errorf("expected 'hello', got '%s'", c.String()) + } +} + +func TestOutputCaptureDrops(t *testing.T) { + c := newOutputCapturer(2) + + c.Write([]byte("hello")) + + want := "h...[3 bytes dropped]...o" + if c.String() != want { + t.Errorf("expected '%s', got '%s'", want, c.String()) + } +} diff --git a/pkg/restic/limitwriter.go b/pkg/restic/limitwriter.go deleted file mode 100644 index 8ab36883..00000000 --- a/pkg/restic/limitwriter.go +++ /dev/null @@ -1,34 +0,0 @@ -package restic - -import "io" - -// limitWriter silently stops writing after 'limit' bytes. -type limitWriter struct { - written int64 - limit int64 - w io.Writer -} - -var _ io.Writer = &limitWriter{} - -func (w *limitWriter) Write(p []byte) (n int, err error) { - r := len(p) - if w.written >= w.limit { - return r, nil - } - - if w.written+int64(len(p)) > w.limit { - p = p[:w.limit-w.written] - } - - n, err = w.w.Write(p) - w.written += int64(n) - return r, err -} - -func newLimitWriter(w io.Writer, limit int64) io.Writer { - return &limitWriter{ - w: w, - limit: limit, - } -} diff --git a/pkg/restic/outputs.go b/pkg/restic/outputs.go index a09a78d9..c8e71110 100644 --- a/pkg/restic/outputs.go +++ b/pkg/restic/outputs.go @@ -105,7 +105,7 @@ func readBackupProgressEntries(cmd *exec.Cmd, output io.Reader, callback func(ev bytes = append(bytes, scanner.Bytes()...) } - return nil, NewCmdError(cmd, bytes, fmt.Errorf("command output was not JSON: %w", err)) + return nil, newCmdError(cmd, string(bytes), fmt.Errorf("command output was not JSON: %w", err)) } if err := event.Validate(); err != nil { return nil, err @@ -254,7 +254,7 @@ func readRestoreProgressEntries(cmd *exec.Cmd, output io.Reader, callback func(e bytes = append(bytes, scanner.Bytes()...) } - return nil, NewCmdError(cmd, bytes, fmt.Errorf("command output was not JSON: %w", err)) + return nil, newCmdError(cmd, string(bytes), fmt.Errorf("command output was not JSON: %w", err)) } if err := event.Validate(); err != nil { return nil, err @@ -271,10 +271,12 @@ func readRestoreProgressEntries(cmd *exec.Cmd, output io.Reader, callback func(e for scanner.Scan() { var event RestoreProgressEntry if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { - return nil, fmt.Errorf("failed to parse JSON: %w", err) + // skip it. Best effort parsing, restic will return with a non-zero exit code if it fails. + continue } if err := event.Validate(); err != nil { - return nil, err + // skip it. Best effort parsing, restic will return with a non-zero exit code if it fails. + continue } if callback != nil { diff --git a/pkg/restic/restic.go b/pkg/restic/restic.go index 49141ee0..fb62044c 100644 --- a/pkg/restic/restic.go +++ b/pkg/restic/restic.go @@ -17,8 +17,6 @@ import ( var errAlreadyInitialized = errors.New("repo already initialized") -const outputBufferLimit = 1000 - type Repo struct { mu sync.Mutex cmd string @@ -71,7 +69,7 @@ func (r *Repo) init(ctx context.Context) error { if strings.Contains(string(output), "config file already exists") || strings.Contains(string(output), "already initialized") { return errAlreadyInitialized } - return NewCmdError(cmd, output, err) + return newCmdError(cmd, string(output), err) } r.initialized = true @@ -79,8 +77,6 @@ func (r *Repo) init(ctx context.Context) error { } func (r *Repo) Init(ctx context.Context) error { - r.mu.Lock() - defer r.mu.Unlock() if err := r.init(ctx); err != nil && !errors.Is(err, errAlreadyInitialized) { return fmt.Errorf("init failed: %w", err) } @@ -88,9 +84,6 @@ func (r *Repo) Init(ctx context.Context) error { } func (r *Repo) Backup(ctx context.Context, progressCallback func(*BackupProgressEntry), opts ...BackupOption) (*BackupProgressEntry, error) { - r.mu.Lock() - defer r.mu.Unlock() - opt := &BackupOpts{} for _, o := range opts { o(opt) @@ -107,9 +100,9 @@ func (r *Repo) Backup(ctx context.Context, progressCallback func(*BackupProgress args = append(args, opt.paths...) args = append(args, opt.extraArgs...) - output := bytes.NewBuffer(nil) + output := newOutputCapturer(outputBufferLimit) reader, writer := io.Pipe() - capture := io.MultiWriter(newLimitWriter(output, outputBufferLimit), writer) + capture := io.MultiWriter(output, writer) cmd := exec.CommandContext(ctx, r.cmd, args...) cmd.Env = append(cmd.Env, r.buildEnv()...) @@ -117,7 +110,7 @@ func (r *Repo) Backup(ctx context.Context, progressCallback func(*BackupProgress cmd.Stdout = capture if err := cmd.Start(); err != nil { - return nil, NewCmdError(cmd, nil, err) + return nil, newCmdError(cmd, "", err) } var wg sync.WaitGroup @@ -147,16 +140,13 @@ func (r *Repo) Backup(ctx context.Context, progressCallback func(*BackupProgress wg.Wait() if cmdErr != nil || readErr != nil { - return nil, NewCmdError(cmd, output.Bytes(), errors.Join(cmdErr, readErr)) + return nil, newCmdErrorPreformatted(cmd, output.String(), errors.Join(cmdErr, readErr)) } return summary, nil } func (r *Repo) Snapshots(ctx context.Context, opts ...GenericOption) ([]*Snapshot, error) { - r.mu.Lock() - defer r.mu.Unlock() - opt := resolveOpts(opts) args := []string{"snapshots", "--json"} @@ -169,12 +159,12 @@ func (r *Repo) Snapshots(ctx context.Context, opts ...GenericOption) ([]*Snapsho output, err := cmd.CombinedOutput() if err != nil { - return nil, NewCmdError(cmd, output, err) + return nil, newCmdError(cmd, "", err) } var snapshots []*Snapshot if err := json.Unmarshal(output, &snapshots); err != nil { - return nil, NewCmdError(cmd, output, fmt.Errorf("command output is not valid JSON: %w", err)) + return nil, newCmdError(cmd, "", fmt.Errorf("command output is not valid JSON: %w", err)) } for _, snapshot := range snapshots { if err := snapshot.Validate(); err != nil { @@ -185,9 +175,6 @@ func (r *Repo) Snapshots(ctx context.Context, opts ...GenericOption) ([]*Snapsho } func (r *Repo) Forget(ctx context.Context, policy *RetentionPolicy, opts ...GenericOption) (*ForgetResult, error) { - r.mu.Lock() - defer r.mu.Unlock() - // first run the forget command opt := resolveOpts(opts) @@ -202,27 +189,24 @@ func (r *Repo) Forget(ctx context.Context, policy *RetentionPolicy, opts ...Gene output, err := cmd.CombinedOutput() if err != nil { - return nil, NewCmdError(cmd, output, err) + return nil, newCmdError(cmd, string(output), err) } var result []ForgetResult if err := json.Unmarshal(output, &result); err != nil { - return nil, NewCmdError(cmd, output, fmt.Errorf("command output is not valid JSON: %w", err)) + return nil, newCmdError(cmd, string(output), 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, output, fmt.Errorf("invalid forget result: %w", err)) + return nil, newCmdError(cmd, string(output), fmt.Errorf("invalid forget result: %w", err)) } return &result[0], nil } func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...GenericOption) error { - r.mu.Lock() - defer r.mu.Unlock() - opt := resolveOpts(opts) args := []string{"prune"} @@ -233,10 +217,10 @@ func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...Generic cmd.Env = append(cmd.Env, r.buildEnv()...) cmd.Env = append(cmd.Env, opt.extraEnv...) - buf := bytes.NewBuffer(nil) - var writer io.Writer = newLimitWriter(buf, outputBufferLimit) + var output = newOutputCapturer(outputBufferLimit) + var writer io.Writer = output if pruneOutput != nil { - writer = io.MultiWriter(pruneOutput, buf) + writer = io.MultiWriter(pruneOutput, output) } cmd.Stdout = writer cmd.Stderr = writer @@ -244,25 +228,22 @@ func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...Generic writer.Write([]byte("command: " + strings.Join(cmd.Args, " ") + "\n")) if err := cmd.Run(); err != nil { - return NewCmdError(cmd, buf.Bytes(), err) + return newCmdErrorPreformatted(cmd, output.String(), err) } return nil } func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*RestoreProgressEntry), opts ...GenericOption) (*RestoreProgressEntry, error) { - r.mu.Lock() - defer r.mu.Unlock() - opt := resolveOpts(opts) args := []string{"restore", snapshot, "--json"} args = append(args, r.extraArgs...) args = append(args, opt.extraArgs...) - output := bytes.NewBuffer(nil) + output := newOutputCapturer(outputBufferLimit) reader, writer := io.Pipe() - capture := io.MultiWriter(newLimitWriter(output, 1000), writer) + capture := io.MultiWriter(output, writer) cmd := exec.CommandContext(ctx, r.cmd, args...) cmd.Env = append(cmd.Env, r.buildEnv()...) @@ -271,7 +252,7 @@ func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*Rest cmd.Stdout = capture if err := cmd.Start(); err != nil { - return nil, NewCmdError(cmd, nil, err) + return nil, newCmdError(cmd, "", err) } var wg sync.WaitGroup @@ -301,16 +282,13 @@ func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*Rest wg.Wait() if cmdErr != nil || readErr != nil { - return nil, NewCmdError(cmd, output.Bytes(), errors.Join(cmdErr, readErr)) + return nil, newCmdError(cmd, output.String(), errors.Join(cmdErr, readErr)) } return summary, nil } func (r *Repo) ListDirectory(ctx context.Context, snapshot string, path string, opts ...GenericOption) (*Snapshot, []*LsEntry, error) { - r.mu.Lock() - defer r.mu.Unlock() - if path == "" { // an empty path can trigger very expensive operations (e.g. iterates all files in the snapshot) return nil, nil, errors.New("path must not be empty") @@ -328,21 +306,18 @@ func (r *Repo) ListDirectory(ctx context.Context, snapshot string, path string, output, err := cmd.CombinedOutput() if err != nil { - return nil, nil, NewCmdError(cmd, output, err) + return nil, nil, newCmdError(cmd, string(output), err) } snapshots, entries, err := readLs(bytes.NewBuffer(output)) if err != nil { - return nil, nil, NewCmdError(cmd, output, err) + return nil, nil, newCmdError(cmd, string(output), err) } return snapshots, entries, nil } func (r *Repo) Unlock(ctx context.Context, opts ...GenericOption) error { - r.mu.Lock() - defer r.mu.Unlock() - opt := resolveOpts(opts) args := []string{"unlock"} @@ -355,7 +330,7 @@ func (r *Repo) Unlock(ctx context.Context, opts ...GenericOption) error { output, err := cmd.CombinedOutput() if err != nil { - return NewCmdError(cmd, output, err) + return newCmdError(cmd, string(output), err) } return nil diff --git a/pkg/restic/restic_test.go b/pkg/restic/restic_test.go index d854583e..15a54e0b 100644 --- a/pkg/restic/restic_test.go +++ b/pkg/restic/restic_test.go @@ -79,8 +79,10 @@ 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) { t.Logf("backup event: %v", event) + gotEvent = true }, tc.opts...) if (err != nil) != tc.wantErr { t.Fatalf("wanted error: %v, got: %v", tc.wantErr, err) @@ -97,6 +99,10 @@ func TestResticBackup(t *testing.T) { if summary.TotalFilesProcessed != tc.files { t.Errorf("wanted %d files, got: %d", tc.files, summary.TotalFilesProcessed) } + + if !gotEvent { + t.Errorf("wanted backup event, got: false") + } }) } } diff --git a/webui/src/views/AddRepoModal.tsx b/webui/src/views/AddRepoModal.tsx index 8cf1a7ed..a66d9a59 100644 --- a/webui/src/views/AddRepoModal.tsx +++ b/webui/src/views/AddRepoModal.tsx @@ -349,7 +349,7 @@ export const AddRepoModal = ({ {/* Repo.flags */} - + {(fields, { add, remove }, { errors }) => ( <> {fields.map((field, index) => (