Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9abe05be61 | ||
|
|
0288735aaf |
@@ -245,6 +245,17 @@ func (p *Packer) Flush(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DiscardCurrentBlob closes and removes the temp file of the blob being
|
||||
// packed, if any, and drops that blob. A caller that stops before Flush
|
||||
// must call it, or the temp file is left behind. Thread-safe.
|
||||
func (p *Packer) DiscardCurrentBlob() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
p.cleanupTempFile()
|
||||
p.currentBlob = nil
|
||||
}
|
||||
|
||||
// FinalizeBlob finalizes the current blob being assembled.
|
||||
// This compresses the accumulated chunks, encrypts the result, and computes
|
||||
// the content-addressed hash. The finalized blob is either passed to the
|
||||
|
||||
@@ -251,6 +251,12 @@ func NewConfigCommand() *cobra.Command {
|
||||
Use: "config",
|
||||
Short: "Manage the configuration file",
|
||||
Long: "Commands for creating, editing, and querying the vaultik config file.",
|
||||
// cobra checks Args only on a command that has a Run: without
|
||||
// one, a mistyped subcommand printed help and exited 0.
|
||||
Args: cobra.NoArgs,
|
||||
Run: func(cmd *cobra.Command, _ []string) {
|
||||
_ = cmd.Help()
|
||||
},
|
||||
}
|
||||
|
||||
cmd.AddCommand(newConfigInitCommand())
|
||||
|
||||
@@ -15,6 +15,12 @@ func NewDatabaseCommand() *cobra.Command {
|
||||
Use: "database",
|
||||
Short: "Manage the local state database",
|
||||
Long: `Commands for managing the local SQLite state database.`,
|
||||
// cobra checks Args only on a command that has a Run: without
|
||||
// one, a mistyped subcommand printed help and exited 0.
|
||||
Args: cobra.NoArgs,
|
||||
Run: func(cmd *cobra.Command, _ []string) {
|
||||
_ = cmd.Help()
|
||||
},
|
||||
}
|
||||
|
||||
cmd.AddCommand(
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
package cli //nolint:testpackage // shares runEntry and the command-word constants
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// TestEntryGroupRejectsUnknownSubcommand runs each group command with a
|
||||
// mistyped subcommand, as a crontab line might carry one. It must fail
|
||||
// with exit 1 and name the word, as `vaultik bogus` does at the top
|
||||
// level, instead of printing the group's help and exiting 0 having done
|
||||
// nothing. The group name alone still prints its help.
|
||||
//
|
||||
//nolint:paralleltest // replaces os.Args, os.Stdout and os.Stderr
|
||||
func TestEntryGroupRejectsUnknownSubcommand(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
group string
|
||||
mistyped string
|
||||
}{
|
||||
{group: cmdSnapshot, mistyped: "craete"},
|
||||
{group: "config", mistyped: "inti"},
|
||||
{group: cmdRemote, mistyped: "ifno"},
|
||||
{group: "database", mistyped: "delte"},
|
||||
{group: "completion", mistyped: "nosuchshell"},
|
||||
} {
|
||||
t.Run(testCase.group, func(t *testing.T) {
|
||||
code, stdout, stderr := runEntry(t,
|
||||
testCase.group, testCase.mistyped)
|
||||
|
||||
assert.Equal(t, 1, code)
|
||||
assert.Empty(t, stdout)
|
||||
assert.Contains(t, stderr,
|
||||
`unknown command "`+testCase.mistyped+`"`)
|
||||
|
||||
code, stdout, _ = runEntry(t, testCase.group)
|
||||
|
||||
assert.Equal(t, 0, code)
|
||||
assert.Contains(t, stdout,
|
||||
programName+" "+testCase.group+" [command]")
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -18,6 +18,12 @@ func NewRemoteCommand() *cobra.Command {
|
||||
Use: "remote",
|
||||
Short: "Remote storage management commands",
|
||||
Long: "Commands for inspecting and managing remote storage",
|
||||
// cobra checks Args only on a command that has a Run: without
|
||||
// one, a mistyped subcommand printed help and exited 0.
|
||||
Args: cobra.NoArgs,
|
||||
Run: func(cmd *cobra.Command, _ []string) {
|
||||
_ = cmd.Help()
|
||||
},
|
||||
}
|
||||
|
||||
// Add subcommands
|
||||
|
||||
@@ -74,6 +74,21 @@ on the source system.`,
|
||||
NewDatabaseCommand(),
|
||||
)
|
||||
|
||||
// cobra adds the completion command only when Execute runs, and with
|
||||
// no Run, so it skipped its own cobra.NoArgs and a mistyped shell name
|
||||
// printed help and exited 0. Adding it here lets it have a Run. Its
|
||||
// scripts go to os.Stdout as it is now: a later SetOut does not
|
||||
// redirect them.
|
||||
cmd.InitDefaultCompletionCmd()
|
||||
|
||||
for _, sub := range cmd.Commands() {
|
||||
if sub.Name() == "completion" {
|
||||
sub.Run = func(completion *cobra.Command, _ []string) {
|
||||
_ = completion.Help()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return cmd
|
||||
}
|
||||
|
||||
|
||||
@@ -41,6 +41,12 @@ func NewSnapshotCommand() *cobra.Command {
|
||||
Use: "snapshot",
|
||||
Short: "Snapshot management commands",
|
||||
Long: "Commands for creating, listing, and managing snapshots",
|
||||
// cobra checks Args only on a command that has a Run: without
|
||||
// one, a mistyped subcommand printed help and exited 0.
|
||||
Args: cobra.NoArgs,
|
||||
Run: func(cmd *cobra.Command, _ []string) {
|
||||
_ = cmd.Help()
|
||||
},
|
||||
}
|
||||
|
||||
// Add subcommands
|
||||
|
||||
@@ -1295,6 +1295,11 @@ func (s *Scanner) batchAddFilesToSnapshot(
|
||||
func (s *Scanner) processPhase(
|
||||
ctx context.Context, filesToProcess []*FileToProcess, result *ScanResult,
|
||||
) error {
|
||||
// On an error or cancellation the blob being packed is never
|
||||
// finalized, so its temp file must be removed here. After a
|
||||
// successful flush there is no such blob and this does nothing.
|
||||
defer s.packer.DiscardCurrentBlob()
|
||||
|
||||
// Calculate total bytes to process
|
||||
var totalBytes int64
|
||||
for _, f := range filesToProcess {
|
||||
|
||||
@@ -381,3 +381,60 @@ func TestScannerCancelStopsSkipErrorsRun(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestScannerRemovesBlobTempFileWhenRunStops checks that a run stopped
|
||||
// part-way, by a file it cannot read or by a cancellation, removes the
|
||||
// temp file of the blob it was packing.
|
||||
func TestScannerRemovesBlobTempFileWhenRunStops(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// a.txt is backed up first and starts a blob; the run stops at b.txt.
|
||||
const target = "/source/b.txt"
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
newFs func(cancel context.CancelFunc) afero.Fs
|
||||
}{
|
||||
{
|
||||
name: "file cannot be read",
|
||||
newFs: func(context.CancelFunc) afero.Fs {
|
||||
return &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "cancelled",
|
||||
newFs: func(cancel context.CancelFunc) afero.Fs {
|
||||
return &cancelOnOpenFs{
|
||||
Fs: afero.NewMemMapFs(), target: target, cancel: cancel,
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
fs := tt.newFs(cancel)
|
||||
writeSkipErrorTestFile(t, fs, "/source/a.txt", "first file content")
|
||||
writeSkipErrorTestFile(t, fs, target, "second file content")
|
||||
|
||||
_, err := runSkipErrorScan(ctx, t, fs, "/source", false, nil)
|
||||
if err == nil {
|
||||
t.Fatal("expected the scan to stop with an error, got nil")
|
||||
}
|
||||
|
||||
left, err := afero.Glob(fs, filepath.Join(os.TempDir(), "vaultik-blob-*"))
|
||||
if err != nil {
|
||||
t.Fatalf("listing blob temp files: %v", err)
|
||||
}
|
||||
|
||||
if len(left) != 0 {
|
||||
t.Fatalf("expected no blob temp file left, got %v", left)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,20 +1,15 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
"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"
|
||||
)
|
||||
@@ -82,116 +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()
|
||||
tempDir := t.TempDir()
|
||||
|
||||
dataDir := filepath.Join(tempDir, "source")
|
||||
storeDir := filepath.Join(tempDir, "remote")
|
||||
dbPath := filepath.Join(tempDir, "index.sqlite")
|
||||
|
||||
chunkSize := int64(32 * 1024)
|
||||
maxBlobSize := int64(128 * 1024)
|
||||
|
||||
// Random data does not compress or deduplicate, so it fills several
|
||||
// blobs.
|
||||
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))
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
cfg, storer, snapshotID := runFileStorageBackup(
|
||||
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
|
||||
|
||||
healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "ok", healthy.Status)
|
||||
require.GreaterOrEqual(t, healthy.BlobCount, 3)
|
||||
require.Equal(t, healthy.BlobCount, healthy.Verified)
|
||||
require.Zero(t, healthy.Missing)
|
||||
|
||||
// Deep verify reads blobs in hash order, which is the sorted order of
|
||||
// their storage keys.
|
||||
keys, err := storer.List(ctx, "blobs/")
|
||||
require.NoError(t, err)
|
||||
slices.Sort(keys)
|
||||
require.Len(t, keys, healthy.BlobCount)
|
||||
|
||||
// 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 := os.ReadFile(lastPath)
|
||||
require.NoError(t, err)
|
||||
|
||||
blob[len(blob)/2] ^= 0xff
|
||||
require.NoError(t, os.WriteFile(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)
|
||||
}
|
||||
|
||||
// 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) {
|
||||
|
||||
+17
-35
@@ -32,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.
|
||||
@@ -137,6 +135,7 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
|
||||
}
|
||||
|
||||
result.Status = "ok"
|
||||
result.Verified = len(dbBlobs)
|
||||
|
||||
if opts.JSON {
|
||||
return v.outputVerifyJSON(result)
|
||||
@@ -291,7 +290,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)
|
||||
}
|
||||
@@ -302,8 +301,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)
|
||||
}
|
||||
@@ -655,13 +653,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.
|
||||
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 {
|
||||
@@ -674,21 +667,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 err != nil:
|
||||
log.Warn("Blob missing from storage",
|
||||
"hash", blob.Hash, "error", err)
|
||||
if err != nil {
|
||||
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
|
||||
}
|
||||
|
||||
result.Missing++
|
||||
result.MissingSize += blob.CompressedSize
|
||||
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
|
||||
@@ -702,22 +691,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
|
||||
@@ -740,7 +723,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