Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
178f9c0fcf |
@@ -413,11 +413,12 @@ under `blobs/` whose name is not a blob hash is not counted as a blob,
|
||||
since `prune` never deletes it.
|
||||
* `--json`: Output as JSON
|
||||
|
||||
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
||||
backup destination store, leaving the bucket prefix empty. Destructive and
|
||||
irreversible. This is the single supported way to wipe the entire
|
||||
destination store. If anything cannot be deleted, the command exits
|
||||
non-zero; running it again deletes what is left.
|
||||
**`remote nuke`**: Delete every object under `metadata/` and `blobs/` on
|
||||
the backup destination store, whatever its name, leaving the bucket prefix
|
||||
empty. Destructive and irreversible. This is the single supported way to
|
||||
wipe the entire destination store. If a listing afterwards still finds
|
||||
anything there, the command names each object left and exits non-zero;
|
||||
running it again deletes what is left.
|
||||
* `--force`: Required to confirm destruction.
|
||||
|
||||
**`database delete`**: Delete the local SQLite state database file
|
||||
|
||||
+13
-63
@@ -120,10 +120,9 @@ func (f *FileStorer) Delete(_ context.Context, key string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// List returns all keys with the given prefix, reading through symlinked
|
||||
// directories. It fails when the destination directory is missing, or
|
||||
// when the prefix or a directory above it is a symlink whose target is
|
||||
// missing; a prefix that is truly missing is an empty listing.
|
||||
// List returns all keys with the given prefix. It fails when the
|
||||
// destination directory is missing; a missing prefix under it is an
|
||||
// empty listing.
|
||||
func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error) {
|
||||
var keys []string
|
||||
|
||||
@@ -134,7 +133,8 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
||||
|
||||
basePath := f.fullPath(prefix)
|
||||
|
||||
exists, err := f.prefixExists(prefix)
|
||||
// Check if the prefix exists
|
||||
exists, err := afero.Exists(f.fs, basePath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("checking path: %w", err)
|
||||
}
|
||||
@@ -143,7 +143,7 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
||||
return keys, nil // Empty list for non-existent prefix
|
||||
}
|
||||
|
||||
err = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
||||
err = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -176,9 +176,7 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
||||
}
|
||||
|
||||
// ListStream returns a channel of ObjectInfo for large result sets. Like
|
||||
// List, it reads through symlinked directories and sends an error when
|
||||
// the destination directory is missing, or when the prefix or a directory
|
||||
// above it is a symlink whose target is missing.
|
||||
// List, it sends an error when the destination directory is missing.
|
||||
func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan ObjectInfo {
|
||||
ch := make(chan ObjectInfo)
|
||||
|
||||
@@ -194,7 +192,8 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
||||
|
||||
basePath := f.fullPath(prefix)
|
||||
|
||||
exists, err := f.prefixExists(prefix)
|
||||
// Check if the prefix exists
|
||||
exists, err := afero.Exists(f.fs, basePath)
|
||||
if err != nil {
|
||||
ch <- ObjectInfo{Err: fmt.Errorf("checking path: %w", err)}
|
||||
|
||||
@@ -205,7 +204,7 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
||||
return // Empty channel for non-existent prefix
|
||||
}
|
||||
|
||||
_ = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
||||
_ = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||
// Check context cancellation
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@@ -244,12 +243,11 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
||||
}
|
||||
|
||||
// DeletePartialUploads removes every file under prefix whose name ends in
|
||||
// tempSuffix. Like List, it reads through symlinked directories and fails
|
||||
// on one whose target is missing. A missing prefix has none to remove.
|
||||
// tempSuffix. A missing prefix has none to remove.
|
||||
func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
||||
basePath := f.fullPath(prefix)
|
||||
|
||||
exists, err := f.prefixExists(prefix)
|
||||
exists, err := afero.Exists(f.fs, basePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("checking path: %w", err)
|
||||
}
|
||||
@@ -258,7 +256,7 @@ func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) er
|
||||
return nil
|
||||
}
|
||||
|
||||
err = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
||||
err = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -354,54 +352,6 @@ func (f *FileStorer) writeAtomic(
|
||||
return nil
|
||||
}
|
||||
|
||||
// walk is afero.Walk, except that it reads through a symlinked directory
|
||||
// the way Get and Stat do, where afero.Walk reports the link as one file.
|
||||
// afero.Walk uses Lstat, which does not follow a link, whenever the
|
||||
// filesystem has one; the struct below carries only the afero.Fs methods,
|
||||
// so the walk uses Stat instead. A link back up the tree is followed until
|
||||
// the operating system refuses the path for too many levels of links, and
|
||||
// the walk then fails.
|
||||
func (f *FileStorer) walk(root string, fn filepath.WalkFunc) error {
|
||||
return afero.Walk(struct{ afero.Fs }{f.fs}, root, fn)
|
||||
}
|
||||
|
||||
// prefixExists reports whether prefix exists under the destination
|
||||
// directory. It checks each directory on the way down to prefix in turn,
|
||||
// and the first one Stat cannot find decides: if it is not there at all,
|
||||
// prefix is missing; if it is a symlink whose target is missing, such as
|
||||
// a link to a volume that is not mounted, prefixExists fails, since
|
||||
// listing it as empty would let prune delete every blob.
|
||||
func (f *FileStorer) prefixExists(prefix string) (bool, error) {
|
||||
lstater, canLstat := f.fs.(afero.Lstater)
|
||||
path := f.basePath
|
||||
|
||||
for name := range strings.SplitSeq(prefix, "/") {
|
||||
path = filepath.Join(path, name)
|
||||
|
||||
_, err := f.fs.Stat(path)
|
||||
if err == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if !os.IsNotExist(err) {
|
||||
return false, err
|
||||
}
|
||||
|
||||
// Lstat does not follow a symlink at path, so it finds the link
|
||||
// that Stat could not follow.
|
||||
if canLstat {
|
||||
_, _, lstatErr := lstater.LstatIfPossible(path)
|
||||
if lstatErr == nil {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
|
||||
return false, nil
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// fullPath returns the full filesystem path for a key.
|
||||
func (f *FileStorer) fullPath(key string) string {
|
||||
return filepath.Join(f.basePath, key)
|
||||
|
||||
@@ -4,10 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
@@ -63,232 +60,3 @@ func TestFileStorerListMissingDestination(t *testing.T) {
|
||||
t.Errorf("ListStream error = %v, want a not-exist error", streamErr)
|
||||
}
|
||||
}
|
||||
|
||||
// The two directories at the top of a destination, either of which the
|
||||
// tests below replace with a symlink.
|
||||
const (
|
||||
metadataDir = "metadata"
|
||||
blobsDir = "blobs"
|
||||
)
|
||||
|
||||
// moveBehindSymlink moves base/dir to a directory elsewhere and puts a
|
||||
// symlink to it in its place.
|
||||
func moveBehindSymlink(t *testing.T, base, dir string) {
|
||||
t.Helper()
|
||||
|
||||
linkPath := filepath.Join(base, dir)
|
||||
movedPath := filepath.Join(t.TempDir(), dir)
|
||||
|
||||
err := os.Rename(linkPath, movedPath)
|
||||
if err != nil {
|
||||
t.Fatalf("moving %s away: %v", dir, err)
|
||||
}
|
||||
|
||||
err = os.Symlink(movedPath, linkPath)
|
||||
if err != nil {
|
||||
t.Fatalf("linking %s: %v", dir, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFileStorerListsThroughSymlinkedDirectory checks that List and
|
||||
// ListStream read through a symlinked metadata/ or blobs/ directory, as Get
|
||||
// and Stat do, instead of listing the link as one key. Listing the whole
|
||||
// destination meets the link part-way through the walk; listing its prefix
|
||||
// starts the walk at the link. A prefix missing behind the link lists as
|
||||
// empty.
|
||||
func TestFileStorerListsThroughSymlinkedDirectory(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
for _, dir := range []string{metadataDir, blobsDir} {
|
||||
t.Run(dir, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
base := t.TempDir()
|
||||
key := dir + "/aa/bb/object"
|
||||
|
||||
s, err := storage.NewFileStorer(base)
|
||||
if err != nil {
|
||||
t.Fatalf("NewFileStorer: %v", err)
|
||||
}
|
||||
|
||||
err = s.Put(context.Background(), key, strings.NewReader("data"))
|
||||
if err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
|
||||
moveBehindSymlink(t, base, dir)
|
||||
|
||||
want := []string{key}
|
||||
|
||||
for _, prefix := range []string{"", dir + "/"} {
|
||||
got := listSorted(t, s, prefix)
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("List(%q) = %v, want %v", prefix, got, want)
|
||||
}
|
||||
|
||||
got = listStreamKeys(t, s, prefix)
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("ListStream(%q) = %v, want %v", prefix, got, want)
|
||||
}
|
||||
}
|
||||
|
||||
absent := dir + "/absent/"
|
||||
|
||||
if got := listSorted(t, s, absent); len(got) != 0 {
|
||||
t.Errorf("List(%q) = %v, want empty", absent, got)
|
||||
}
|
||||
|
||||
if got := listStreamKeys(t, s, absent); len(got) != 0 {
|
||||
t.Errorf("ListStream(%q) = %v, want empty", absent, got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestFileStorerListFailsOnSymlinkWithMissingTarget checks that List and
|
||||
// ListStream fail on a symlinked metadata/ or blobs/ directory whose target
|
||||
// is missing, such as a link to a volume that is not mounted, as they do on
|
||||
// a missing destination directory, instead of listing it as empty. That
|
||||
// holds for a prefix below the link too.
|
||||
func TestFileStorerListFailsOnSymlinkWithMissingTarget(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
for _, dir := range []string{metadataDir, blobsDir} {
|
||||
t.Run(dir, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
base := t.TempDir()
|
||||
|
||||
s, err := storage.NewFileStorer(base)
|
||||
if err != nil {
|
||||
t.Fatalf("NewFileStorer: %v", err)
|
||||
}
|
||||
|
||||
unmounted := filepath.Join(t.TempDir(), "unmounted")
|
||||
|
||||
err = os.Symlink(unmounted, filepath.Join(base, dir))
|
||||
if err != nil {
|
||||
t.Fatalf("Symlink: %v", err)
|
||||
}
|
||||
|
||||
for _, prefix := range []string{"", dir + "/", dir + "/aa/"} {
|
||||
keys, err := s.List(ctx, prefix)
|
||||
if !errors.Is(err, fs.ErrNotExist) {
|
||||
t.Errorf("List(%q) = %v, %v; want a not-exist error",
|
||||
prefix, keys, err)
|
||||
}
|
||||
|
||||
var streamErr error
|
||||
|
||||
for object := range s.ListStream(ctx, prefix) {
|
||||
if object.Err != nil {
|
||||
streamErr = object.Err
|
||||
}
|
||||
}
|
||||
|
||||
if !errors.Is(streamErr, fs.ErrNotExist) {
|
||||
t.Errorf("ListStream(%q) error = %v, want a not-exist error",
|
||||
prefix, streamErr)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestFileStorerListFailsOnSymlinkLoop checks that a symlink pointing back
|
||||
// at the destination makes List and ListStream fail instead of walking it
|
||||
// forever.
|
||||
func TestFileStorerListFailsOnSymlinkLoop(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
base := t.TempDir()
|
||||
|
||||
s, err := storage.NewFileStorer(base)
|
||||
if err != nil {
|
||||
t.Fatalf("NewFileStorer: %v", err)
|
||||
}
|
||||
|
||||
err = s.Put(ctx, testBlobKey, strings.NewReader("data"))
|
||||
if err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
|
||||
err = os.Symlink(base, filepath.Join(base, "blobs", "loop"))
|
||||
if err != nil {
|
||||
t.Fatalf("Symlink: %v", err)
|
||||
}
|
||||
|
||||
keys, err := s.List(ctx, "blobs/")
|
||||
if err == nil {
|
||||
t.Errorf("List = %d keys and no error, want an error", len(keys))
|
||||
}
|
||||
|
||||
var streamErr error
|
||||
|
||||
for object := range s.ListStream(ctx, "blobs/") {
|
||||
if object.Err != nil {
|
||||
streamErr = object.Err
|
||||
}
|
||||
}
|
||||
|
||||
if streamErr == nil {
|
||||
t.Error("ListStream sent no error, want one")
|
||||
}
|
||||
}
|
||||
|
||||
// TestFileStorerDeletePartialUploadsThroughSymlinkedDirectory checks that
|
||||
// DeletePartialUploads, which remote nuke uses, removes a leftover temp
|
||||
// file behind a symlinked metadata/ or blobs/ directory and keeps the
|
||||
// object at the real key.
|
||||
func TestFileStorerDeletePartialUploadsThroughSymlinkedDirectory(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
for _, dir := range []string{metadataDir, blobsDir} {
|
||||
t.Run(dir, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
base := t.TempDir()
|
||||
key := dir + "/aa/bb/object"
|
||||
|
||||
s, err := storage.NewFileStorer(base)
|
||||
if err != nil {
|
||||
t.Fatalf("NewFileStorer: %v", err)
|
||||
}
|
||||
|
||||
err = s.Put(ctx, key, strings.NewReader("data"))
|
||||
if err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
|
||||
leftover := filepath.Join(base, key+"-123456.partial")
|
||||
|
||||
err = os.WriteFile(leftover, []byte("half"), 0o600)
|
||||
if err != nil {
|
||||
t.Fatalf("writing leftover temp file: %v", err)
|
||||
}
|
||||
|
||||
moveBehindSymlink(t, base, dir)
|
||||
|
||||
err = s.DeletePartialUploads(ctx, dir+"/")
|
||||
if err != nil {
|
||||
t.Fatalf("DeletePartialUploads: %v", err)
|
||||
}
|
||||
|
||||
// leftover runs through the link.
|
||||
_, err = os.Stat(leftover)
|
||||
if !os.IsNotExist(err) {
|
||||
t.Errorf("leftover temp file was not removed: %v", err)
|
||||
}
|
||||
|
||||
_, err = s.Stat(ctx, key)
|
||||
if err != nil {
|
||||
t.Errorf("Stat of the real key: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+6
-40
@@ -30,12 +30,6 @@ var (
|
||||
"storage URL must not carry credentials; " +
|
||||
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
||||
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
||||
ErrURLBadQuery = errors.New(
|
||||
"storage URL query cannot be decoded; " +
|
||||
"check it for a ; or a % not followed by two hex digits")
|
||||
ErrRcloneRemoteParams = errors.New(
|
||||
"rclone URL must not carry parameters in the remote name; " +
|
||||
"set them on the remote with rclone config instead")
|
||||
)
|
||||
|
||||
// URL represents a parsed storage URL.
|
||||
@@ -96,9 +90,8 @@ func parseFileURL(rawPath string) (*URL, error) {
|
||||
}
|
||||
|
||||
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
||||
// the userinfo or in the endpoint parameter's userinfo, a query that
|
||||
// cannot be decoded, and any query parameter other than endpoint, region
|
||||
// and ssl, so a credential-bearing URL is never stored or echoed.
|
||||
// the userinfo and any query parameter other than endpoint, region and
|
||||
// ssl, so a credential-bearing URL is never stored or echoed.
|
||||
func parseS3URL(rawURL string) (*URL, error) {
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
@@ -114,26 +107,13 @@ func parseS3URL(rawURL string) (*URL, error) {
|
||||
return nil, ErrMissingBucket
|
||||
}
|
||||
|
||||
// u.Query() would silently drop a part it cannot decode, endpoint
|
||||
// included. The decode error is not wrapped: it can quote the value.
|
||||
query, err := url.ParseQuery(u.RawQuery)
|
||||
if err != nil {
|
||||
return nil, ErrURLBadQuery
|
||||
}
|
||||
query := u.Query()
|
||||
|
||||
err = rejectUnknownParams(query, "endpoint", "region", "ssl")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// A host name never contains "@", so one in an endpoint, with or
|
||||
// without a scheme, can only introduce userinfo.
|
||||
for _, endpoint := range query["endpoint"] {
|
||||
if strings.Contains(endpoint, "@") {
|
||||
return nil, ErrURLCredentials
|
||||
}
|
||||
}
|
||||
|
||||
return &URL{
|
||||
Scheme: schemeS3,
|
||||
Bucket: bucket,
|
||||
@@ -145,23 +125,9 @@ func parseS3URL(rawURL string) (*URL, error) {
|
||||
}
|
||||
|
||||
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
||||
// query parameters, so credentials in the userinfo, parameters in the
|
||||
// remote name and any query parameter at all are rejected rather than
|
||||
// silently ignored.
|
||||
// query parameters, so credentials in the userinfo and any parameter at
|
||||
// all are rejected rather than silently ignored.
|
||||
func parseRcloneURL(rawURL string) (*URL, error) {
|
||||
// The remote name runs up to the first "/", "?" or "#". It never
|
||||
// contains a comma; one starts rclone's remote,param=value form,
|
||||
// whose values can be credentials. This is checked before url.Parse,
|
||||
// whose error for such a name can print part of it.
|
||||
remote := strings.TrimPrefix(rawURL, "rclone://")
|
||||
if end := strings.IndexAny(remote, "/?#"); end >= 0 {
|
||||
remote = remote[:end]
|
||||
}
|
||||
|
||||
if strings.Contains(remote, ",") {
|
||||
return nil, ErrRcloneRemoteParams
|
||||
}
|
||||
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return nil, wrapParseError(err)
|
||||
@@ -171,7 +137,7 @@ func parseRcloneURL(rawURL string) (*URL, error) {
|
||||
return nil, ErrURLCredentials
|
||||
}
|
||||
|
||||
remote = u.Host
|
||||
remote := u.Host
|
||||
if remote == "" {
|
||||
return nil, ErrMissingRemote
|
||||
}
|
||||
|
||||
@@ -150,19 +150,6 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
||||
wantErr: storage.ErrURLCredentials,
|
||||
secrets: []string{key, secret, "mybucket"},
|
||||
},
|
||||
{
|
||||
name: "s3 userinfo in endpoint",
|
||||
raw: "s3://mybucket?endpoint=https://" +
|
||||
key + ":" + secret + "@minio.example.com",
|
||||
wantErr: storage.ErrURLCredentials,
|
||||
secrets: []string{key, secret},
|
||||
},
|
||||
{
|
||||
name: "s3 userinfo in endpoint without scheme",
|
||||
raw: "s3://mybucket?endpoint=" + key + ":" + secret + "@minio.example.com",
|
||||
wantErr: storage.ErrURLCredentials,
|
||||
secrets: []string{key, secret},
|
||||
},
|
||||
{
|
||||
name: "s3 unknown query param",
|
||||
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
||||
@@ -216,104 +203,6 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestParseStorageURLRejectsRcloneRemoteParams checks that an rclone
|
||||
// remote name carrying parameters (rclone's "remote,param=value" form,
|
||||
// which can hold a credential) is rejected without echoing the value,
|
||||
// including the URLs that url.Parse itself refuses.
|
||||
func TestParseStorageURLRejectsRcloneRemoteParams(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
raw string
|
||||
secrets []string // must not appear in the error message
|
||||
}{
|
||||
{
|
||||
name: "configured remote",
|
||||
raw: "rclone://gdrive,client_secret=topsecret/backups",
|
||||
secrets: []string{"topsecret"},
|
||||
},
|
||||
{
|
||||
name: "remote defined inline",
|
||||
raw: "rclone://:s3,access_key_id=KEY,secret_access_key=SECRET/bucket",
|
||||
secrets: []string{"KEY", "SECRET"},
|
||||
},
|
||||
{
|
||||
name: "colon in a parameter value",
|
||||
raw: "rclone://gdrive,client_secret=ab:cd/backups",
|
||||
secrets: []string{"cd"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
_, err := storage.ParseStorageURL(tc.raw)
|
||||
if !errors.Is(err, storage.ErrRcloneRemoteParams) {
|
||||
t.Fatalf("ParseStorageURL(%q) error = %v, want %v",
|
||||
tc.raw, err, storage.ErrRcloneRemoteParams)
|
||||
}
|
||||
|
||||
for _, secret := range tc.secrets {
|
||||
if strings.Contains(err.Error(), secret) {
|
||||
t.Errorf("error message leaked %q: %v", secret, err.Error())
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestParseStorageURLRejectsUndecodableQuery checks that an s3 URL whose
|
||||
// query cannot be decoded is rejected, not read with that part of the
|
||||
// query left out, and that the error does not echo it.
|
||||
func TestParseStorageURLRejectsUndecodableQuery(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
key = "AKIAKEY"
|
||||
secret = "topsecret"
|
||||
)
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
raw string
|
||||
}{
|
||||
{
|
||||
name: "semicolon",
|
||||
raw: "s3://mybucket/prefix?endpoint=https://" + key + ":" + secret +
|
||||
"@minio.example.com:9000;region=us-east-1",
|
||||
},
|
||||
{
|
||||
name: "bad percent escape",
|
||||
raw: "s3://mybucket/prefix?endpoint=https://" + key + ":" + secret +
|
||||
"%zz@minio.example.com:9000",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
_, err := storage.ParseStorageURL(tc.raw)
|
||||
if err == nil {
|
||||
t.Fatalf("ParseStorageURL(%q) returned no error", tc.raw)
|
||||
}
|
||||
|
||||
if !errors.Is(err, storage.ErrURLBadQuery) {
|
||||
t.Errorf("ParseStorageURL(%q) error = %v, want %v",
|
||||
tc.raw, err, storage.ErrURLBadQuery)
|
||||
}
|
||||
|
||||
for _, s := range []string{key, secret, "minio.example.com"} {
|
||||
if strings.Contains(err.Error(), s) {
|
||||
t.Errorf("error message leaked %q: %v", s, err.Error())
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse
|
||||
// itself fails, the wrapped error carries only the inner cause, not the
|
||||
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
||||
|
||||
@@ -1,9 +1,7 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
@@ -11,9 +9,7 @@ import (
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/config"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
"sneak.berlin/go/vaultik/internal/ui"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
@@ -81,174 +77,6 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) {
|
||||
"deep verify should fail on a corrupted blob")
|
||||
}
|
||||
|
||||
// TestDeepVerifyJSONCountsBlobs checks the verified and missing counts
|
||||
// that deep verify reports under --json: on a healthy snapshot, with a
|
||||
// damaged blob, and with missing blobs. Deep verify stops at the first
|
||||
// damaged blob, so the blobs read before it are the ones verified; when
|
||||
// blobs are missing it reads no blob contents, but counts every missing
|
||||
// blob.
|
||||
func TestDeepVerifyJSONCountsBlobs(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||
|
||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||
|
||||
healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "ok", healthy.Status)
|
||||
require.Len(t, keys, healthy.BlobCount)
|
||||
require.Equal(t, healthy.BlobCount, healthy.Verified)
|
||||
require.Zero(t, healthy.Missing)
|
||||
|
||||
// Damage the last blob read, keeping its length, so every blob before
|
||||
// it is read and passes first.
|
||||
lastPath := filepath.Join(storeDir, keys[len(keys)-1])
|
||||
blob, err := afero.ReadFile(fs, lastPath)
|
||||
require.NoError(t, err)
|
||||
|
||||
blob[len(blob)/2] ^= 0xff
|
||||
require.NoError(t, afero.WriteFile(fs, lastPath, blob, 0o600))
|
||||
|
||||
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.Error(t, err)
|
||||
require.Equal(t, "failed", got.Status)
|
||||
require.Equal(t, healthy.BlobCount-1, got.Verified)
|
||||
require.Zero(t, got.Missing)
|
||||
|
||||
// Delete the first blob and the damaged last one: both must be
|
||||
// counted as missing, not only the first one found.
|
||||
var deletedSize int64
|
||||
|
||||
for _, key := range []string{keys[0], keys[len(keys)-1]} {
|
||||
info, statErr := storer.Stat(ctx, key)
|
||||
require.NoError(t, statErr)
|
||||
|
||||
deletedSize += info.Size
|
||||
|
||||
require.NoError(t, storer.Delete(ctx, key))
|
||||
}
|
||||
|
||||
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.Error(t, err)
|
||||
require.Equal(t, "failed", got.Status)
|
||||
require.Equal(t, 2, got.Missing)
|
||||
require.Equal(t, deletedSize, got.MissingSize)
|
||||
require.Zero(t, got.Verified)
|
||||
}
|
||||
|
||||
// TestDeepVerifyJSONCountsWrongSizeBlob checks that under --deep --json a
|
||||
// blob of the wrong size is counted in mismatched and fails the run, and
|
||||
// that the check goes on past it to count a missing blob after it.
|
||||
func TestDeepVerifyJSONCountsWrongSizeBlob(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||
|
||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||
|
||||
// Make the first blob checked longer, and delete the last one.
|
||||
firstPath := filepath.Join(storeDir, keys[0])
|
||||
blob, err := afero.ReadFile(fs, firstPath)
|
||||
require.NoError(t, err)
|
||||
|
||||
blob = append(blob, []byte("extra")...)
|
||||
require.NoError(t, afero.WriteFile(fs, firstPath, blob, 0o600))
|
||||
require.NoError(t, storer.Delete(ctx, keys[len(keys)-1]))
|
||||
|
||||
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.Error(t, err)
|
||||
require.Equal(t, "failed", got.Status)
|
||||
require.Equal(t, 1, got.Mismatched)
|
||||
require.Equal(t, 1, got.Missing)
|
||||
require.Zero(t, got.Verified)
|
||||
}
|
||||
|
||||
// TestDeepVerifyStopsOnBlobThatCannotBeChecked checks that a blob that is
|
||||
// present in storage but cannot be checked is not counted as missing:
|
||||
// deep verify stops and fails with the storage error.
|
||||
func TestDeepVerifyStopsOnBlobThatCannotBeChecked(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||
|
||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||
failing := statFailingStorer{Storer: storer, failKeys: []string{keys[0]}}
|
||||
|
||||
got, err := deepVerifyJSON(ctx, t, cfg, failing, snapshotID)
|
||||
require.Error(t, err)
|
||||
require.Equal(t, "failed", got.Status)
|
||||
require.Zero(t, got.Missing)
|
||||
require.Contains(t, got.ErrorMessage, errStatDenied.Error())
|
||||
}
|
||||
|
||||
// backUpSeveralBlobs backs up random data, which does not compress or
|
||||
// deduplicate, so it fills at least three blobs in a store on disk at
|
||||
// storeDir. It returns the blobs' storage keys sorted, which is hash
|
||||
// order, the order deep verify reads them in.
|
||||
func backUpSeveralBlobs(
|
||||
ctx context.Context, t *testing.T, storeDir string,
|
||||
) (*config.Config, *storage.FileStorer, string, []string) {
|
||||
t.Helper()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
tempDir := t.TempDir()
|
||||
dataDir := filepath.Join(tempDir, "source")
|
||||
dbPath := filepath.Join(tempDir, "index.sqlite")
|
||||
|
||||
chunkSize := int64(32 * 1024)
|
||||
maxBlobSize := int64(128 * 1024)
|
||||
|
||||
data := make([]byte, maxBlobSize*4)
|
||||
_, err := rand.Read(data)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
||||
require.NoError(t, afero.WriteFile(fs,
|
||||
filepath.Join(dataDir, "data.bin"), data, 0o644))
|
||||
|
||||
cfg, storer, snapshotID := runFileStorageBackup(
|
||||
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
|
||||
|
||||
keys := storedBlobKeys(t, storeDir)
|
||||
require.GreaterOrEqual(t, len(keys), 3)
|
||||
|
||||
return cfg, storer, snapshotID, keys
|
||||
}
|
||||
|
||||
// deepVerifyJSON runs deep verify with --json on snapshotID and returns
|
||||
// the decoded result and the error RunDeepVerify returned.
|
||||
func deepVerifyJSON(
|
||||
ctx context.Context, t *testing.T,
|
||||
cfg *config.Config, storer storage.Storer, snapshotID string,
|
||||
) (vaultik.VerifyResult, error) {
|
||||
t.Helper()
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
v := &vaultik.Vaultik{
|
||||
Config: cfg,
|
||||
Storage: storer,
|
||||
Fs: afero.NewOsFs(),
|
||||
Stdout: &out,
|
||||
Stderr: io.Discard,
|
||||
UI: ui.NewWithColor(io.Discard, false),
|
||||
}
|
||||
v.SetContext(ctx)
|
||||
|
||||
err := v.RunDeepVerify(snapshotID,
|
||||
&vaultik.VerifyOptions{Deep: true, JSON: true})
|
||||
|
||||
return decodeVerifyResult(t, out.Bytes()), err
|
||||
}
|
||||
|
||||
// corruptOneBlob flips a middle byte of the first blob file found under
|
||||
// blobsDir, leaving the file length unchanged.
|
||||
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
||||
|
||||
@@ -18,7 +18,8 @@ import (
|
||||
// TestNukeRemoteLeavesNoFiles checks that remote nuke leaves no file
|
||||
// under a file:// destination, including the `.partial` files uploads
|
||||
// killed part-way leave next to a snapshot's metadata and next to where a
|
||||
// blob would have been.
|
||||
// blob would have been, a snapshot's database whose manifest was never
|
||||
// uploaded, and a file under blobs/ whose name is not a blob hash.
|
||||
func TestNukeRemoteLeavesNoFiles(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
@@ -38,6 +39,9 @@ func TestNukeRemoteLeavesNoFiles(t *testing.T) {
|
||||
"db.zst.age-123456.partial"),
|
||||
filepath.Join(storeDir, "blobs", hash[:2], hash[2:4],
|
||||
hash+"-123456.partial"),
|
||||
filepath.Join(storeDir, "metadata",
|
||||
snapshot.RemoteSnapshotKey("no-manifest"), "db.zst.age"),
|
||||
filepath.Join(storeDir, "blobs", "0b", "01", ".DS_Store"),
|
||||
}
|
||||
|
||||
for _, leftover := range leftovers {
|
||||
@@ -100,6 +104,34 @@ func TestNukeRemoteFailsWhenADeleteFails(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestNukeRemoteNamesWhatIsLeft checks that remote nuke, when an object
|
||||
// it does not recognise cannot be deleted, does not report the store
|
||||
// empty but names that object and fails.
|
||||
func TestNukeRemoteNamesWhatIsLeft(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||
|
||||
const leftKey = "blobs/0b/01/.DS_Store"
|
||||
|
||||
leftPath := filepath.Join(storeDir, filepath.FromSlash(leftKey))
|
||||
require.NoError(t, os.MkdirAll(filepath.Dir(leftPath), 0o750))
|
||||
require.NoError(t, os.WriteFile(leftPath, []byte("finder"), 0o600))
|
||||
|
||||
store := faultstore.New(v.Storage)
|
||||
store.OnDelete = func(key string) bool { return key == leftKey }
|
||||
v.Storage = store
|
||||
|
||||
require.Error(t, v.NukeRemote(true))
|
||||
|
||||
assert.Contains(t, out.String(), leftKey)
|
||||
assert.NotContains(t, out.String(), "now empty")
|
||||
assert.Contains(t, out.String(), "vaultik remote nuke --force")
|
||||
}
|
||||
|
||||
// TestNukeRemoteRefusesAnotherDestinationWithoutRetryAdvice checks that
|
||||
// remote nuke, when the local index belongs to a different destination
|
||||
// than the configured one, fails without advising to run nuke again:
|
||||
|
||||
+60
-16
@@ -22,6 +22,11 @@ var errNukeRequiresForce = errors.New(
|
||||
// errBlobDeleteFailed reports unreferenced blobs prune could not delete.
|
||||
var errBlobDeleteFailed = errors.New("blob(s) could not be deleted")
|
||||
|
||||
// errObjectsLeft reports objects remote nuke still finds on the backup
|
||||
// destination store after deleting.
|
||||
var errObjectsLeft = errors.New(
|
||||
"object(s) still on the backup destination store")
|
||||
|
||||
// nukeCommandHint is the command suggested when a remote nuke could not
|
||||
// delete everything: running it again deletes what is left, which
|
||||
// `vaultik prune` does not do for snapshot metadata.
|
||||
@@ -35,10 +40,11 @@ const metadataDirName = "metadata"
|
||||
// (blobs/<aa>/<bb>/<hash>).
|
||||
const blobKeyParts = 4
|
||||
|
||||
// NukeRemote deletes every snapshot's metadata and every blob from remote
|
||||
// storage, along with any object an upload cut off part-way left under a
|
||||
// temporary `.partial` name. After this returns successfully the bucket
|
||||
// prefix is empty and the next backup starts from scratch.
|
||||
// NukeRemote deletes every object under metadata/ and blobs/ in remote
|
||||
// storage, whatever its name, along with any object an upload cut off
|
||||
// part-way left under a temporary `.partial` name. It returns
|
||||
// successfully only when a listing afterwards finds nothing left there,
|
||||
// and the next backup then starts from scratch.
|
||||
//
|
||||
// Refuses to run unless force is true. The caller is responsible for
|
||||
// confirming with the user.
|
||||
@@ -68,7 +74,8 @@ func (v *Vaultik) NukeRemote(force bool) error {
|
||||
}
|
||||
|
||||
// emptyDestinationStore does the deleting for NukeRemote and stops at
|
||||
// the first step that fails.
|
||||
// the first step that fails. It also fails when objects are still
|
||||
// listed at the end, naming each one.
|
||||
func (v *Vaultik) emptyDestinationStore() error {
|
||||
v.UI.Beginf("Removing all snapshot metadata from backup destination store.")
|
||||
|
||||
@@ -77,25 +84,62 @@ func (v *Vaultik) emptyDestinationStore() error {
|
||||
return fmt.Errorf("removing all snapshots: %w", err)
|
||||
}
|
||||
|
||||
v.UI.Beginf("Removing any blobs still present in backup destination store.")
|
||||
v.UI.Beginf("Removing everything still present in backup destination store.")
|
||||
|
||||
err = v.PruneBlobs(&PruneOptions{Force: true})
|
||||
if err != nil {
|
||||
return fmt.Errorf("pruning blobs: %w", err)
|
||||
var left []string
|
||||
|
||||
for _, prefix := range []string{"metadata/", "blobs/"} {
|
||||
keys, err := v.deleteEverythingUnder(prefix)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
left = append(left, keys...)
|
||||
}
|
||||
|
||||
// The file and rclone listings skip `.partial` objects, so the two
|
||||
// steps above never delete them.
|
||||
for _, prefix := range []string{"metadata/", "blobs/"} {
|
||||
err = v.Storage.DeletePartialUploads(v.ctx, prefix)
|
||||
if err != nil {
|
||||
return fmt.Errorf("deleting partial uploads: %w", err)
|
||||
}
|
||||
for _, key := range left {
|
||||
v.UI.Warningf("Still on the backup destination store: %s", key)
|
||||
}
|
||||
|
||||
if len(left) > 0 {
|
||||
return fmt.Errorf("%d %w", len(left), errObjectsLeft)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// deleteEverythingUnder deletes every object under prefix, whatever its
|
||||
// name, and returns the keys a listing afterwards still finds. A delete
|
||||
// that fails is logged and the others still run; that object then
|
||||
// shows up in the listing afterwards.
|
||||
func (v *Vaultik) deleteEverythingUnder(prefix string) ([]string, error) {
|
||||
keys, err := v.Storage.List(v.ctx, prefix)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("listing %s: %w", prefix, err)
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
err = v.Storage.Delete(v.ctx, key)
|
||||
if err != nil {
|
||||
log.Error("Failed to delete object", "key", key, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// The file and rclone listings skip `.partial` objects, so the
|
||||
// deletes above never reach them.
|
||||
err = v.Storage.DeletePartialUploads(v.ctx, prefix)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("deleting partial uploads: %w", err)
|
||||
}
|
||||
|
||||
left, err := v.Storage.List(v.ctx, prefix)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("listing %s: %w", prefix, err)
|
||||
}
|
||||
|
||||
return left, nil
|
||||
}
|
||||
|
||||
// PruneBlobsResult contains the result of a blob prune operation
|
||||
//
|
||||
//nolint:tagliatelle // snake_case is the established JSON output format
|
||||
|
||||
@@ -1,147 +0,0 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
|
||||
// TestPruneDeletesNothingThroughSymlinkedDirectory checks that prune keeps
|
||||
// every blob and the local snapshot record when the metadata/ or blobs/
|
||||
// directory of a file:// destination is a symlink to a directory elsewhere
|
||||
// (https://git.eeqj.de/sneak/vaultik/issues/296). A backup writes through
|
||||
// such a link, so prune has to list through it too.
|
||||
func TestPruneDeletesNothingThroughSymlinkedDirectory(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
for _, dir := range []string{"metadata", "blobs"} {
|
||||
t.Run(dir, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||
blobFiles := blobFilesIn(t, storeDir)
|
||||
moveBehindSymlink(t, storeDir, dir)
|
||||
|
||||
require.NoError(t, v.Prune(&vaultik.PruneOptions{Force: true}))
|
||||
|
||||
// When blobs/ is the link, these paths run through it.
|
||||
for _, blobFile := range blobFiles {
|
||||
assert.FileExists(t, blobFile, "prune must delete no blob")
|
||||
}
|
||||
|
||||
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, snapshots, 1, "prune must delete no local snapshot record")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestPruneFailsOnSymlinkedDirectoryWithMissingTarget checks that prune
|
||||
// fails, deleting no blob and keeping the local snapshot record, when the
|
||||
// metadata/ or blobs/ directory of a file:// destination is a symlink
|
||||
// whose target is missing, as a link to a volume that is not mounted is.
|
||||
func TestPruneFailsOnSymlinkedDirectoryWithMissingTarget(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
for _, dir := range []string{"metadata", "blobs"} {
|
||||
t.Run(dir, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||
blobFiles := blobFilesIn(t, storeDir)
|
||||
targetPath := moveBehindSymlink(t, storeDir, dir)
|
||||
|
||||
// Unmount the volume the link points to, and mount it again
|
||||
// once prune has run.
|
||||
unmountedPath := targetPath + "-unmounted"
|
||||
require.NoError(t, os.Rename(targetPath, unmountedPath))
|
||||
|
||||
err := v.Prune(&vaultik.PruneOptions{Force: true})
|
||||
require.ErrorIs(t, err, fs.ErrNotExist)
|
||||
|
||||
require.NoError(t, os.Rename(unmountedPath, targetPath))
|
||||
|
||||
// When blobs/ is the link, these paths run through it.
|
||||
for _, blobFile := range blobFiles {
|
||||
assert.FileExists(t, blobFile, "prune must delete no blob")
|
||||
}
|
||||
|
||||
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, snapshots, 1, "prune must delete no local snapshot record")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestRemoveSnapshotWarnsOnSymlinkedMetadataWithMissingTarget checks that
|
||||
// snapshot remove warns that the metadata could not be removed from the
|
||||
// destination when metadata/ is a symlink whose target is missing, as it
|
||||
// does when the destination directory is missing, instead of finding no
|
||||
// metadata to remove.
|
||||
func TestRemoveSnapshotWarnsOnSymlinkedMetadataWithMissingTarget(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, repos, out := backUpToFileDestination(ctx, t, storeDir)
|
||||
targetPath := moveBehindSymlink(t, storeDir, "metadata")
|
||||
require.NoError(t, os.Rename(targetPath, targetPath+"-unmounted"))
|
||||
|
||||
result, err := v.RemoveSnapshot(localSnapshotID(ctx, t, repos, "first"),
|
||||
&vaultik.RemoveOptions{Force: true})
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.False(t, result.RemoteRemoved)
|
||||
assert.Contains(t, out.String(),
|
||||
"Could not remove snapshot metadata from remote")
|
||||
assert.NotContains(t, out.String(),
|
||||
"Removed snapshot metadata from remote storage")
|
||||
}
|
||||
|
||||
// blobFilesIn returns the paths of the blob files under storeDir/blobs, and
|
||||
// fails the test when there are none.
|
||||
func blobFilesIn(t *testing.T, storeDir string) []string {
|
||||
t.Helper()
|
||||
|
||||
var blobFiles []string
|
||||
|
||||
err := filepath.WalkDir(filepath.Join(storeDir, "blobs"),
|
||||
func(path string, entry fs.DirEntry, err error) error {
|
||||
if err == nil && !entry.IsDir() {
|
||||
blobFiles = append(blobFiles, path)
|
||||
}
|
||||
|
||||
return err
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotEmpty(t, blobFiles)
|
||||
|
||||
return blobFiles
|
||||
}
|
||||
|
||||
// moveBehindSymlink moves storeDir/dir to a directory elsewhere, puts a
|
||||
// symlink to it in its place, and returns where it went.
|
||||
func moveBehindSymlink(t *testing.T, storeDir, dir string) string {
|
||||
t.Helper()
|
||||
|
||||
linkPath := filepath.Join(storeDir, dir)
|
||||
targetPath := filepath.Join(t.TempDir(), dir)
|
||||
require.NoError(t, os.Rename(linkPath, targetPath))
|
||||
require.NoError(t, os.Symlink(targetPath, linkPath))
|
||||
|
||||
return targetPath
|
||||
}
|
||||
+17
-39
@@ -17,7 +17,6 @@ import (
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
)
|
||||
|
||||
// Sentinel errors for snapshot verification failures.
|
||||
@@ -33,9 +32,7 @@ var (
|
||||
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
||||
errManifestMissingBlob = errors.New(
|
||||
"manifest omits blob present in database")
|
||||
errBlobSizeMismatch = errors.New("blob size mismatch")
|
||||
errBlobsNotInStorage = errors.New(
|
||||
"blobs missing from storage or stored with the wrong size")
|
||||
errBlobSizeMismatch = errors.New("blob size mismatch")
|
||||
)
|
||||
|
||||
// verifyStatusFailed is the JSON status value for a failed verification.
|
||||
@@ -152,6 +149,7 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
|
||||
}
|
||||
|
||||
result.Status = "ok"
|
||||
result.Verified = len(dbBlobs)
|
||||
|
||||
if opts.JSON {
|
||||
return v.outputVerifyJSON(result)
|
||||
@@ -308,7 +306,7 @@ func (v *Vaultik) runVerificationSteps(
|
||||
v.stdoutf("Checking blob existence in remote storage...\n")
|
||||
}
|
||||
|
||||
err = v.verifyBlobExistenceFromDB(dbBlobs, result)
|
||||
err = v.verifyBlobExistenceFromDB(dbBlobs)
|
||||
if err != nil {
|
||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||
}
|
||||
@@ -319,8 +317,7 @@ func (v *Vaultik) runVerificationSteps(
|
||||
len(dbBlobs), ubytes(totalSize))
|
||||
}
|
||||
|
||||
err = v.performDeepVerificationFromDB(
|
||||
dbBlobs, tdb.db.Conn(), opts, result, identities)
|
||||
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
|
||||
if err != nil {
|
||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||
}
|
||||
@@ -672,14 +669,8 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
|
||||
return nil
|
||||
}
|
||||
|
||||
// verifyBlobExistenceFromDB checks that every blob from the database is
|
||||
// in storage with the size the database records. It checks them all,
|
||||
// counting the missing and wrong-size blobs in result, and fails if it
|
||||
// found any. Any storage error other than not found stops the check and
|
||||
// fails with that error.
|
||||
func (v *Vaultik) verifyBlobExistenceFromDB(
|
||||
blobs []snapshot.BlobInfo, result *VerifyResult,
|
||||
) error {
|
||||
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
|
||||
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
||||
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
||||
|
||||
for i, blob := range blobs {
|
||||
@@ -692,23 +683,17 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
|
||||
// Construct blob path
|
||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
||||
|
||||
// Check blob exists
|
||||
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
||||
switch {
|
||||
case errors.Is(err, storage.ErrNotFound):
|
||||
log.Warn("Blob missing from storage", "hash", blob.Hash)
|
||||
if err != nil {
|
||||
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
|
||||
}
|
||||
|
||||
result.Missing++
|
||||
result.MissingSize += blob.CompressedSize
|
||||
case err != nil:
|
||||
return fmt.Errorf("checking blob %s in storage: %w",
|
||||
blob.Hash, err)
|
||||
case stat.Size != blob.CompressedSize:
|
||||
log.Warn("Blob has the wrong size in storage",
|
||||
"hash", blob.Hash,
|
||||
"storage_size", stat.Size,
|
||||
"database_size", blob.CompressedSize)
|
||||
|
||||
result.Mismatched++
|
||||
// Verify size matches
|
||||
if stat.Size != blob.CompressedSize {
|
||||
return fmt.Errorf(
|
||||
"%w: blob %s: S3 has %d bytes, database has %d bytes",
|
||||
errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
|
||||
}
|
||||
|
||||
// Progress update every 100 blobs
|
||||
@@ -722,22 +707,16 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
|
||||
}
|
||||
}
|
||||
|
||||
if result.Missing > 0 || result.Mismatched > 0 {
|
||||
return fmt.Errorf("%w: %d missing, %d with the wrong size",
|
||||
errBlobsNotInStorage, result.Missing, result.Mismatched)
|
||||
}
|
||||
|
||||
log.Info("✓ All blobs exist in storage")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// performDeepVerificationFromDB downloads and verifies the content of
|
||||
// each blob using the database as source, stopping at the first blob that
|
||||
// fails. Each blob that passes is counted in result.Verified.
|
||||
// each blob using the database as source.
|
||||
func (v *Vaultik) performDeepVerificationFromDB(
|
||||
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
||||
result *VerifyResult, identities []age.Identity,
|
||||
identities []age.Identity,
|
||||
) error {
|
||||
// Calculate total bytes for ETA
|
||||
var totalBytesExpected int64
|
||||
@@ -760,7 +739,6 @@ func (v *Vaultik) performDeepVerificationFromDB(
|
||||
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
||||
}
|
||||
|
||||
result.Verified++
|
||||
bytesProcessed += blobInfo.CompressedSize
|
||||
elapsed := time.Since(startTime)
|
||||
remaining := len(blobs) - (i + 1)
|
||||
|
||||
Reference in New Issue
Block a user