Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2c4af6d0df | ||
|
|
42eb50b151 | ||
|
|
25ee9211d3 | ||
|
|
3361e9ebd9 |
+24
-3
@@ -5,7 +5,16 @@
|
|||||||
FROM golangci/golangci-lint:v2.14.0@sha256:ad862ba6b3798cbe0fd9fd7408d498fd74fbd2623a92406b2fd3898faf0bf98f AS lint
|
FROM golangci/golangci-lint:v2.14.0@sha256:ad862ba6b3798cbe0fd9fd7408d498fd74fbd2623a92406b2fd3898faf0bf98f AS lint
|
||||||
WORKDIR /src
|
WORKDIR /src
|
||||||
COPY go.mod go.sum ./
|
COPY go.mod go.sum ./
|
||||||
RUN go mod download
|
# Without the time limit a stalled connection to the module proxy holds
|
||||||
|
# this step, and every build on the host sharing it, until the kernel
|
||||||
|
# drops the connection, which can take an hour. The limit is well above
|
||||||
|
# the three minutes a working download can take on a busy build host.
|
||||||
|
RUN for attempt in 1 2 3; do \
|
||||||
|
timeout 300 go mod download && exit 0; \
|
||||||
|
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
|
||||||
|
done; \
|
||||||
|
echo "module download failed: giving up after 3 attempts" >&2; \
|
||||||
|
exit 1
|
||||||
COPY . .
|
COPY . .
|
||||||
# `golangci-lint run` silently ignores an unknown top-level key in
|
# `golangci-lint run` silently ignores an unknown top-level key in
|
||||||
# .golangci.yml, such as a misspelt `linters:`; `config verify` fails on it.
|
# .golangci.yml, such as a misspelt `linters:`; `config verify` fails on it.
|
||||||
@@ -18,7 +27,13 @@ RUN golangci-lint run --config .golangci.yml ./...
|
|||||||
FROM golang:1.26.1@sha256:cd78d88e00afadbedd272f977d375a6247455f3a4b1178f8ae8bbcb201743a8a AS test
|
FROM golang:1.26.1@sha256:cd78d88e00afadbedd272f977d375a6247455f3a4b1178f8ae8bbcb201743a8a AS test
|
||||||
WORKDIR /src
|
WORKDIR /src
|
||||||
COPY go.mod go.sum ./
|
COPY go.mod go.sum ./
|
||||||
RUN go mod download
|
# Limited and retried as in the lint phase above.
|
||||||
|
RUN for attempt in 1 2 3; do \
|
||||||
|
timeout 300 go mod download && exit 0; \
|
||||||
|
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
|
||||||
|
done; \
|
||||||
|
echo "module download failed: giving up after 3 attempts" >&2; \
|
||||||
|
exit 1
|
||||||
COPY . .
|
COPY . .
|
||||||
RUN go test -timeout 90s -race -cover ./... || \
|
RUN go test -timeout 90s -race -cover ./... || \
|
||||||
{ echo "--- Rerunning with -v for details ---"; \
|
{ echo "--- Rerunning with -v for details ---"; \
|
||||||
@@ -36,7 +51,13 @@ RUN apk add --no-cache git
|
|||||||
RUN git config --system --add safe.directory /src
|
RUN git config --system --add safe.directory /src
|
||||||
WORKDIR /src
|
WORKDIR /src
|
||||||
COPY go.mod go.sum ./
|
COPY go.mod go.sum ./
|
||||||
RUN go mod download
|
# Limited and retried as in the lint phase above.
|
||||||
|
RUN for attempt in 1 2 3; do \
|
||||||
|
timeout 300 go mod download && exit 0; \
|
||||||
|
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
|
||||||
|
done; \
|
||||||
|
echo "module download failed: giving up after 3 attempts" >&2; \
|
||||||
|
exit 1
|
||||||
COPY . .
|
COPY . .
|
||||||
|
|
||||||
# The VERSION build arg when one is given, otherwise
|
# The VERSION build arg when one is given, otherwise
|
||||||
|
|||||||
@@ -294,7 +294,8 @@ needed.
|
|||||||
of the local index and the backup destination store — with timestamps
|
of the local index and the backup destination store — with timestamps
|
||||||
and three sizes per snapshot (compressed remote size; total
|
and three sizes per snapshot (compressed remote size; total
|
||||||
uncompressed chunk size; size of the chunks that snapshot stored that
|
uncompressed chunk size; size of the chunks that snapshot stored that
|
||||||
were not stored before, recorded when its backup ran).
|
were not stored before). Both chunk sizes are recorded when the
|
||||||
|
snapshot's backup runs.
|
||||||
|
|
||||||
Listing the destination store does **not** require the age secret key,
|
Listing the destination store does **not** require the age secret key,
|
||||||
so it works in vaultik's intended configuration, where the backed-up
|
so it works in vaultik's intended configuration, where the backed-up
|
||||||
@@ -338,7 +339,9 @@ local index alone, and still exits zero.
|
|||||||
|
|
||||||
**`snapshot purge`**: Remove old snapshots based on criteria. Retention is
|
**`snapshot purge`**: Remove old snapshots based on criteria. Retention is
|
||||||
per-snapshot-name (`--keep-latest` keeps the latest of each name, not the
|
per-snapshot-name (`--keep-latest` keeps the latest of each name, not the
|
||||||
latest globally).
|
latest globally). If a snapshot cannot be removed from the local index, the
|
||||||
|
command stops there, leaves that snapshot's metadata on the destination store
|
||||||
|
and exits non-zero; `snapshot create --prune` fails the same way.
|
||||||
* `--keep-latest`: Keep only the most recent snapshot of each name
|
* `--keep-latest`: Keep only the most recent snapshot of each name
|
||||||
* `--older-than <duration>`: Remove snapshots older than duration (e.g. `30d`,
|
* `--older-than <duration>`: Remove snapshots older than duration (e.g. `30d`,
|
||||||
`4w`, `6mo`, `1y`; `m` is minutes, `mo` is months)
|
`4w`, `6mo`, `1y`; `m` is minutes, `mo` is months)
|
||||||
|
|||||||
@@ -0,0 +1,45 @@
|
|||||||
|
package main_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestDockerfileBoundsEveryModuleDownload fails unless every
|
||||||
|
// `go mod download` in the Dockerfile runs under a time limit, is tried
|
||||||
|
// a fixed number of times, and then fails the build with a message that
|
||||||
|
// names the module download (issue #351). Like the version tests, this
|
||||||
|
// parses the Dockerfile; that a stalled download really is cut off and
|
||||||
|
// retried was checked by hand.
|
||||||
|
func TestDockerfileBoundsEveryModuleDownload(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
var downloads []string
|
||||||
|
|
||||||
|
for _, instruction := range instructions(t, productDockerfile) {
|
||||||
|
if strings.Contains(instruction, "go mod download") {
|
||||||
|
downloads = append(downloads, instruction)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NotEmpty(t, downloads, "%s must download modules",
|
||||||
|
productDockerfile)
|
||||||
|
|
||||||
|
for _, download := range downloads {
|
||||||
|
assert.Contains(t, download, "timeout 300 go mod download",
|
||||||
|
"%s: a module download must have a time limit: %s",
|
||||||
|
productDockerfile, download)
|
||||||
|
assert.Contains(t, download, "for attempt in 1 2 3",
|
||||||
|
"%s: a module download must be retried: %s",
|
||||||
|
productDockerfile, download)
|
||||||
|
assert.Contains(t, download, "module download failed",
|
||||||
|
"%s: a module download that never succeeds must say so: %s",
|
||||||
|
productDockerfile, download)
|
||||||
|
assert.True(t, strings.HasSuffix(download, "exit 1"),
|
||||||
|
"%s: a module download that never succeeds must fail the"+
|
||||||
|
" build: %s", productDockerfile, download)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -122,6 +122,7 @@ Tracks backup snapshots.
|
|||||||
- `new_chunk_size` (INTEGER) - Total size of those chunks before compression; `snapshot list` shows it as NEW CHUNK SIZE
|
- `new_chunk_size` (INTEGER) - Total size of those chunks before compression; `snapshot list` shows it as NEW CHUNK SIZE
|
||||||
- `blob_count` (INTEGER) - Number of blobs this snapshot created
|
- `blob_count` (INTEGER) - Number of blobs this snapshot created
|
||||||
- `total_size` (INTEGER) - Total size of all files
|
- `total_size` (INTEGER) - Total size of all files
|
||||||
|
- `uncompressed_chunk_size` (INTEGER) - Total size before compression of the distinct chunks the snapshot's files reference, recorded when its backup ran; `snapshot list` shows it as UNCOMPRESSED SIZE
|
||||||
- `blob_size` (INTEGER) - Total compressed size of all referenced blobs
|
- `blob_size` (INTEGER) - Total compressed size of all referenced blobs
|
||||||
- `blob_uncompressed_size` (INTEGER) - Total uncompressed size of all referenced blobs
|
- `blob_uncompressed_size` (INTEGER) - Total uncompressed size of all referenced blobs
|
||||||
- `compression_ratio` (REAL) - Compression ratio achieved
|
- `compression_ratio` (REAL) - Compression ratio achieved
|
||||||
|
|||||||
@@ -0,0 +1,90 @@
|
|||||||
|
package cli //nolint:testpackage // shares runEntry and the hermetic config helpers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||||
|
"sneak.berlin/go/vaultik/internal/types"
|
||||||
|
)
|
||||||
|
|
||||||
|
// The `snapshot purge` command word and flags used below.
|
||||||
|
const (
|
||||||
|
cmdPurge = "purge"
|
||||||
|
flagKeepLatest = "--keep-latest"
|
||||||
|
flagForce = "--force"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestEntrySnapshotPurgeKeepsMetadataWhenLocalIndexDeleteFails makes
|
||||||
|
// removing a snapshot from the local index fail and checks that
|
||||||
|
// `snapshot purge` exits non-zero and leaves that snapshot's metadata on
|
||||||
|
// the destination store. It used to log the error, delete the metadata
|
||||||
|
// anyway and exit 0.
|
||||||
|
//
|
||||||
|
//nolint:paralleltest // replaces os.Args, os.Stdout, os.Stderr and the xdg globals
|
||||||
|
func TestEntrySnapshotPurgeKeepsMetadataWhenLocalIndexDeleteFails(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
configPath := writeHermeticPruneConfig(t, false)
|
||||||
|
|
||||||
|
// writeHermeticPruneConfig puts the index and the store directory
|
||||||
|
// beside the config file.
|
||||||
|
dir := filepath.Dir(configPath)
|
||||||
|
|
||||||
|
// Two snapshots of the same name: --keep-latest purges the older.
|
||||||
|
olderStart := time.Date(2026, 5, 1, 9, 0, 0, 0, time.UTC)
|
||||||
|
newerStart := olderStart.Add(time.Hour)
|
||||||
|
olderID := "test-host_test_" + olderStart.Format(time.RFC3339)
|
||||||
|
newerID := "test-host_test_" + newerStart.Format(time.RFC3339)
|
||||||
|
|
||||||
|
db, err := database.New(ctx, filepath.Join(dir, "index.sqlite"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
manifestPaths := make(map[string]string)
|
||||||
|
|
||||||
|
for id, startedAt := range map[string]time.Time{
|
||||||
|
olderID: olderStart,
|
||||||
|
newerID: newerStart,
|
||||||
|
} {
|
||||||
|
completedAt := startedAt.Add(time.Minute)
|
||||||
|
snap := &database.Snapshot{
|
||||||
|
ID: types.SnapshotID(id),
|
||||||
|
Hostname: "test-host",
|
||||||
|
VaultikVersion: "test",
|
||||||
|
StartedAt: startedAt,
|
||||||
|
CompletedAt: &completedAt,
|
||||||
|
}
|
||||||
|
|
||||||
|
err = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
return repos.Snapshots.Create(ctx, tx, snap)
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
manifestPaths[id] = filepath.Join(dir, "store", "metadata",
|
||||||
|
snapshot.RemoteSnapshotKey(id), "manifest.json.zst")
|
||||||
|
require.NoError(t, os.MkdirAll(filepath.Dir(manifestPaths[id]), 0o750))
|
||||||
|
require.NoError(t, os.WriteFile(manifestPaths[id], []byte("stub"), 0o600))
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = db.Conn().ExecContext(ctx, `
|
||||||
|
CREATE TRIGGER refuse_snapshot_delete BEFORE DELETE ON snapshots
|
||||||
|
BEGIN SELECT RAISE(ABORT, 'simulated index error'); END`)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, db.Close())
|
||||||
|
|
||||||
|
code, _, stderr := runEntry(t,
|
||||||
|
flagConfig, configPath, cmdSnapshot, cmdPurge, flagKeepLatest, flagForce)
|
||||||
|
|
||||||
|
assert.Equal(t, 1, code)
|
||||||
|
assert.Contains(t, stderr, "simulated index error")
|
||||||
|
assert.FileExists(t, manifestPaths[olderID],
|
||||||
|
"a snapshot still in the local index keeps its metadata")
|
||||||
|
}
|
||||||
@@ -89,6 +89,7 @@ CREATE TABLE IF NOT EXISTS snapshots (
|
|||||||
new_chunk_size INTEGER NOT NULL DEFAULT 0,
|
new_chunk_size INTEGER NOT NULL DEFAULT 0,
|
||||||
blob_count INTEGER NOT NULL DEFAULT 0,
|
blob_count INTEGER NOT NULL DEFAULT 0,
|
||||||
total_size INTEGER NOT NULL DEFAULT 0,
|
total_size INTEGER NOT NULL DEFAULT 0,
|
||||||
|
uncompressed_chunk_size INTEGER NOT NULL DEFAULT 0,
|
||||||
blob_size INTEGER NOT NULL DEFAULT 0,
|
blob_size INTEGER NOT NULL DEFAULT 0,
|
||||||
blob_uncompressed_size INTEGER NOT NULL DEFAULT 0,
|
blob_uncompressed_size INTEGER NOT NULL DEFAULT 0,
|
||||||
compression_ratio REAL NOT NULL DEFAULT 1.0,
|
compression_ratio REAL NOT NULL DEFAULT 1.0,
|
||||||
|
|||||||
@@ -120,7 +120,11 @@ func (r *SnapshotRepository) UpdateCounts(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpdateExtendedStats updates extended statistics for a snapshot
|
// UpdateExtendedStats updates extended statistics for a snapshot. It also
|
||||||
|
// records the total plaintext size of the distinct chunks the snapshot's
|
||||||
|
// files reference. Only the snapshot's own backup can take that total: a
|
||||||
|
// file's row is shared by every snapshot that lists it, and a later
|
||||||
|
// backup of a changed file rewrites the row with the new chunks.
|
||||||
func (r *SnapshotRepository) UpdateExtendedStats(
|
func (r *SnapshotRepository) UpdateExtendedStats(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
tx *sql.Tx,
|
tx *sql.Tx,
|
||||||
@@ -145,7 +149,17 @@ func (r *SnapshotRepository) UpdateExtendedStats(
|
|||||||
compression_level = ?,
|
compression_level = ?,
|
||||||
upload_bytes = ?,
|
upload_bytes = ?,
|
||||||
upload_duration_ms = ?,
|
upload_duration_ms = ?,
|
||||||
new_chunk_size = ?
|
new_chunk_size = ?,
|
||||||
|
uncompressed_chunk_size = (
|
||||||
|
SELECT COALESCE(SUM(c.size), 0)
|
||||||
|
FROM (
|
||||||
|
SELECT DISTINCT fc.chunk_hash
|
||||||
|
FROM snapshot_files sf
|
||||||
|
JOIN file_chunks fc ON sf.file_id = fc.file_id
|
||||||
|
WHERE sf.snapshot_id = snapshots.id
|
||||||
|
) sc
|
||||||
|
JOIN chunks c ON sc.chunk_hash = c.chunk_hash
|
||||||
|
)
|
||||||
WHERE id = ?
|
WHERE id = ?
|
||||||
`
|
`
|
||||||
|
|
||||||
@@ -555,30 +569,22 @@ func (r *SnapshotRepository) GetSnapshotBlobSizes(
|
|||||||
return compressed, uncompressed, nil
|
return compressed, uncompressed, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetSnapshotUncompressedChunkSize returns the sum of plaintext sizes of all unique
|
// GetSnapshotUncompressedChunkSize returns the total plaintext size of the
|
||||||
// chunks referenced by a snapshot (via snapshot_files → file_chunks → chunks).
|
// distinct chunks the snapshot's files reference, as the snapshot's backup
|
||||||
|
// recorded it (see UpdateExtendedStats).
|
||||||
func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
|
func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
|
||||||
ctx context.Context, snapshotID string,
|
ctx context.Context, snapshotID string,
|
||||||
) (int64, error) {
|
) (int64, error) {
|
||||||
query := `
|
query := `SELECT uncompressed_chunk_size FROM snapshots WHERE id = ?`
|
||||||
SELECT COALESCE(SUM(c.size), 0)
|
|
||||||
FROM (
|
|
||||||
SELECT DISTINCT fc.chunk_hash
|
|
||||||
FROM snapshot_files sf
|
|
||||||
JOIN file_chunks fc ON sf.file_id = fc.file_id
|
|
||||||
WHERE sf.snapshot_id = ?
|
|
||||||
) sc
|
|
||||||
JOIN chunks c ON sc.chunk_hash = c.chunk_hash
|
|
||||||
`
|
|
||||||
|
|
||||||
var totalSize int64
|
var uncompressedChunkSize int64
|
||||||
|
|
||||||
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&totalSize)
|
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&uncompressedChunkSize)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, fmt.Errorf("querying uncompressed chunk size: %w", err)
|
return 0, fmt.Errorf("querying uncompressed chunk size: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return totalSize, nil
|
return uncompressedChunkSize, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetSnapshotNewChunkSize returns the total plaintext size of the chunks the
|
// GetSnapshotNewChunkSize returns the total plaintext size of the chunks the
|
||||||
|
|||||||
+6
-32
@@ -29,10 +29,7 @@ var (
|
|||||||
ErrURLCredentials = errors.New(
|
ErrURLCredentials = errors.New(
|
||||||
"storage URL must not carry credentials; " +
|
"storage URL must not carry credentials; " +
|
||||||
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
||||||
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
||||||
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.
|
// URL represents a parsed storage URL.
|
||||||
@@ -93,9 +90,8 @@ func parseFileURL(rawPath string) (*URL, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
||||||
// the userinfo or in the endpoint parameter's userinfo, and any query
|
// the userinfo and any query parameter other than endpoint, region and
|
||||||
// parameter other than endpoint, region and ssl, so a credential-bearing
|
// ssl, so a credential-bearing URL is never stored or echoed.
|
||||||
// URL is never stored or echoed.
|
|
||||||
func parseS3URL(rawURL string) (*URL, error) {
|
func parseS3URL(rawURL string) (*URL, error) {
|
||||||
u, err := url.Parse(rawURL)
|
u, err := url.Parse(rawURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -118,14 +114,6 @@ func parseS3URL(rawURL string) (*URL, error) {
|
|||||||
return nil, err
|
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{
|
return &URL{
|
||||||
Scheme: schemeS3,
|
Scheme: schemeS3,
|
||||||
Bucket: bucket,
|
Bucket: bucket,
|
||||||
@@ -137,23 +125,9 @@ func parseS3URL(rawURL string) (*URL, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
||||||
// query parameters, so credentials in the userinfo, parameters in the
|
// query parameters, so credentials in the userinfo and any parameter at
|
||||||
// remote name and any query parameter at all are rejected rather than
|
// all are rejected rather than silently ignored.
|
||||||
// silently ignored.
|
|
||||||
func parseRcloneURL(rawURL string) (*URL, error) {
|
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)
|
u, err := url.Parse(rawURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, wrapParseError(err)
|
return nil, wrapParseError(err)
|
||||||
@@ -163,7 +137,7 @@ func parseRcloneURL(rawURL string) (*URL, error) {
|
|||||||
return nil, ErrURLCredentials
|
return nil, ErrURLCredentials
|
||||||
}
|
}
|
||||||
|
|
||||||
remote = u.Host
|
remote := u.Host
|
||||||
if remote == "" {
|
if remote == "" {
|
||||||
return nil, ErrMissingRemote
|
return nil, ErrMissingRemote
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -150,19 +150,6 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
|||||||
wantErr: storage.ErrURLCredentials,
|
wantErr: storage.ErrURLCredentials,
|
||||||
secrets: []string{key, secret, "mybucket"},
|
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",
|
name: "s3 unknown query param",
|
||||||
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
||||||
@@ -216,54 +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())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse
|
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse
|
||||||
// itself fails, the wrapped error carries only the inner cause, not the
|
// itself fails, the wrapped error carries only the inner cause, not the
|
||||||
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
package vaultik_test
|
package vaultik_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
|
"crypto/rand"
|
||||||
"io"
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -9,7 +11,9 @@ import (
|
|||||||
|
|
||||||
"github.com/spf13/afero"
|
"github.com/spf13/afero"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
"sneak.berlin/go/vaultik/internal/ui"
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
)
|
)
|
||||||
@@ -77,6 +81,174 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) {
|
|||||||
"deep verify should fail on a corrupted blob")
|
"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
|
// corruptOneBlob flips a middle byte of the first blob file found under
|
||||||
// blobsDir, leaving the file length unchanged.
|
// blobsDir, leaving the file length unchanged.
|
||||||
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
||||||
|
|||||||
@@ -627,7 +627,8 @@ func selectSnapshotsToPurge(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// confirmAndExecutePurge shows deletion candidates, confirms with the
|
// confirmAndExecutePurge shows deletion candidates, confirms with the
|
||||||
// user, and deletes snapshots.
|
// user, and deletes snapshots. It stops with an error at the first
|
||||||
|
// snapshot it cannot delete.
|
||||||
func (v *Vaultik) confirmAndExecutePurge(
|
func (v *Vaultik) confirmAndExecutePurge(
|
||||||
toDelete []SnapshotInfo, force, quiet bool,
|
toDelete []SnapshotInfo, force, quiet bool,
|
||||||
) error {
|
) error {
|
||||||
@@ -669,10 +670,11 @@ func (v *Vaultik) confirmAndExecutePurge(
|
|||||||
snapshotID := snap.ID.String()
|
snapshotID := snap.ID.String()
|
||||||
log.Info("Deleting snapshot", "id", snapshotID)
|
log.Info("Deleting snapshot", "id", snapshotID)
|
||||||
|
|
||||||
|
// Stop before the remote delete: a snapshot still in the local
|
||||||
|
// index must keep its metadata on the destination store.
|
||||||
err := v.deleteSnapshotFromLocalDB(snapshotID)
|
err := v.deleteSnapshotFromLocalDB(snapshotID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("Failed to delete from local database",
|
return fmt.Errorf("deleting snapshot %s from local index: %w", snapshotID, err)
|
||||||
"snapshot_id", snapshotID, "error", err)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
_, err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
|
_, err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
|
||||||
|
|||||||
@@ -483,13 +483,14 @@ func TestListSnapshots_UnreadableManifestDoesNotHideOthers(t *testing.T) {
|
|||||||
//
|
//
|
||||||
//nolint:tagliatelle // snake_case is the established output format
|
//nolint:tagliatelle // snake_case is the established output format
|
||||||
type listJSONRow struct {
|
type listJSONRow struct {
|
||||||
ID string `json:"id"`
|
ID string `json:"id"`
|
||||||
RemoteKey string `json:"remote_key"`
|
RemoteKey string `json:"remote_key"`
|
||||||
Timestamp string `json:"timestamp"`
|
Timestamp string `json:"timestamp"`
|
||||||
CompressedSize int64 `json:"compressed_size"`
|
CompressedSize int64 `json:"compressed_size"`
|
||||||
NewChunkSize int64 `json:"new_chunk_size"`
|
UncompressedSize int64 `json:"uncompressed_size"`
|
||||||
LocallyTracked bool `json:"locally_tracked"`
|
NewChunkSize int64 `json:"new_chunk_size"`
|
||||||
RemotePresent *bool `json:"remote_present"`
|
LocallyTracked bool `json:"locally_tracked"`
|
||||||
|
RemotePresent *bool `json:"remote_present"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// decodeListJSON parses the command's stdout, which must contain
|
// decodeListJSON parses the command's stdout, which must contain
|
||||||
|
|||||||
@@ -0,0 +1,85 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestUncompressedSizeOfChangedFileStaysFixed covers
|
||||||
|
// https://git.eeqj.de/sneak/vaultik/issues/360. A backup that changes a
|
||||||
|
// file leaves the earlier snapshot's UNCOMPRESSED SIZE as it was when that
|
||||||
|
// snapshot's backup completed. The file's row in the local index is shared
|
||||||
|
// by both snapshots and rewritten by the second backup.
|
||||||
|
func TestUncompressedSizeOfChangedFileStaysFixed(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
tempDir := t.TempDir()
|
||||||
|
dataDir := filepath.Join(tempDir, "src")
|
||||||
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
||||||
|
changedPath := filepath.Join(dataDir, "changed.bin")
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
files := writeFaultSourceTree(t, fs, dataDir)
|
||||||
|
cfg := changedFileConfig(dataDir, dbPath)
|
||||||
|
|
||||||
|
appendRandomBytes(t, fs, files, changedPath, appendedFileSize)
|
||||||
|
|
||||||
|
store, err := storage.NewFileStorer(filepath.Join(tempDir, "remote"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
db, err := database.New(ctx, dbPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
||||||
|
|
||||||
|
require.NoError(t, backUp(v, "first"))
|
||||||
|
|
||||||
|
// With nothing to compare against, sizeOfNewChunks totals every
|
||||||
|
// distinct chunk of the files.
|
||||||
|
firstID := localSnapshotID(ctx, t, repos, "first")
|
||||||
|
firstSize := sizeOfNewChunks(t, files, nil)
|
||||||
|
require.Equal(t, firstSize, listedUncompressedSizes(t, v)[firstID],
|
||||||
|
"the first snapshot must count the chunks of every file it holds")
|
||||||
|
|
||||||
|
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
|
||||||
|
require.NoError(t, backUp(v, "second"))
|
||||||
|
|
||||||
|
secondID := localSnapshotID(ctx, t, repos, "second")
|
||||||
|
sizes := listedUncompressedSizes(t, v)
|
||||||
|
assert.Equal(t, firstSize, sizes[firstID],
|
||||||
|
"a later backup must not change the first snapshot's figure")
|
||||||
|
assert.Equal(t, sizeOfNewChunks(t, files, nil), sizes[secondID],
|
||||||
|
"the second snapshot must count the changed file's chunks")
|
||||||
|
}
|
||||||
|
|
||||||
|
// listedUncompressedSizes runs `snapshot list --json` and returns each
|
||||||
|
// listed snapshot's uncompressed_size by snapshot ID.
|
||||||
|
func listedUncompressedSizes(t *testing.T, v *vaultik.Vaultik) map[string]int64 {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
var out bytes.Buffer
|
||||||
|
|
||||||
|
v.Stdout = &out
|
||||||
|
require.NoError(t, v.ListSnapshots(true))
|
||||||
|
|
||||||
|
sizes := make(map[string]int64)
|
||||||
|
for _, row := range decodeListJSON(t, out.String()) {
|
||||||
|
sizes[row.ID] = row.UncompressedSize
|
||||||
|
}
|
||||||
|
|
||||||
|
return sizes
|
||||||
|
}
|
||||||
+39
-17
@@ -17,6 +17,7 @@ import (
|
|||||||
"sneak.berlin/go/vaultik/internal/database"
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Sentinel errors for snapshot verification failures.
|
// Sentinel errors for snapshot verification failures.
|
||||||
@@ -32,7 +33,9 @@ var (
|
|||||||
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
||||||
errManifestMissingBlob = errors.New(
|
errManifestMissingBlob = errors.New(
|
||||||
"manifest omits blob present in database")
|
"manifest omits blob present in database")
|
||||||
errBlobSizeMismatch = errors.New("blob size mismatch")
|
errBlobSizeMismatch = errors.New("blob size mismatch")
|
||||||
|
errBlobsNotInStorage = errors.New(
|
||||||
|
"blobs missing from storage or stored with the wrong size")
|
||||||
)
|
)
|
||||||
|
|
||||||
// verifyStatusFailed is the JSON status value for a failed verification.
|
// verifyStatusFailed is the JSON status value for a failed verification.
|
||||||
@@ -149,7 +152,6 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
result.Status = "ok"
|
result.Status = "ok"
|
||||||
result.Verified = len(dbBlobs)
|
|
||||||
|
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
return v.outputVerifyJSON(result)
|
return v.outputVerifyJSON(result)
|
||||||
@@ -306,7 +308,7 @@ func (v *Vaultik) runVerificationSteps(
|
|||||||
v.stdoutf("Checking blob existence in remote storage...\n")
|
v.stdoutf("Checking blob existence in remote storage...\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.verifyBlobExistenceFromDB(dbBlobs)
|
err = v.verifyBlobExistenceFromDB(dbBlobs, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||||
}
|
}
|
||||||
@@ -317,7 +319,8 @@ func (v *Vaultik) runVerificationSteps(
|
|||||||
len(dbBlobs), ubytes(totalSize))
|
len(dbBlobs), ubytes(totalSize))
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
|
err = v.performDeepVerificationFromDB(
|
||||||
|
dbBlobs, tdb.db.Conn(), opts, result, identities)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||||
}
|
}
|
||||||
@@ -669,8 +672,14 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
|
// verifyBlobExistenceFromDB checks that every blob from the database is
|
||||||
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
// 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 {
|
||||||
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
||||||
|
|
||||||
for i, blob := range blobs {
|
for i, blob := range blobs {
|
||||||
@@ -683,17 +692,23 @@ func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
|||||||
// Construct blob path
|
// Construct blob path
|
||||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
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)
|
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
||||||
if err != nil {
|
switch {
|
||||||
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
|
case errors.Is(err, storage.ErrNotFound):
|
||||||
}
|
log.Warn("Blob missing from storage", "hash", blob.Hash)
|
||||||
|
|
||||||
// Verify size matches
|
result.Missing++
|
||||||
if stat.Size != blob.CompressedSize {
|
result.MissingSize += blob.CompressedSize
|
||||||
return fmt.Errorf(
|
case err != nil:
|
||||||
"%w: blob %s: S3 has %d bytes, database has %d bytes",
|
return fmt.Errorf("checking blob %s in storage: %w",
|
||||||
errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
|
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++
|
||||||
}
|
}
|
||||||
|
|
||||||
// Progress update every 100 blobs
|
// Progress update every 100 blobs
|
||||||
@@ -707,16 +722,22 @@ func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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")
|
log.Info("✓ All blobs exist in storage")
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// performDeepVerificationFromDB downloads and verifies the content of
|
// performDeepVerificationFromDB downloads and verifies the content of
|
||||||
// each blob using the database as source.
|
// each blob using the database as source, stopping at the first blob that
|
||||||
|
// fails. Each blob that passes is counted in result.Verified.
|
||||||
func (v *Vaultik) performDeepVerificationFromDB(
|
func (v *Vaultik) performDeepVerificationFromDB(
|
||||||
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
||||||
identities []age.Identity,
|
result *VerifyResult, identities []age.Identity,
|
||||||
) error {
|
) error {
|
||||||
// Calculate total bytes for ETA
|
// Calculate total bytes for ETA
|
||||||
var totalBytesExpected int64
|
var totalBytesExpected int64
|
||||||
@@ -739,6 +760,7 @@ func (v *Vaultik) performDeepVerificationFromDB(
|
|||||||
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
result.Verified++
|
||||||
bytesProcessed += blobInfo.CompressedSize
|
bytesProcessed += blobInfo.CompressedSize
|
||||||
elapsed := time.Since(startTime)
|
elapsed := time.Since(startTime)
|
||||||
remaining := len(blobs) - (i + 1)
|
remaining := len(blobs) - (i + 1)
|
||||||
|
|||||||
Reference in New Issue
Block a user