Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ac8dd776f5 |
@@ -293,8 +293,8 @@ needed.
|
||||
**`snapshot list`**: Show every snapshot known to this host — the union
|
||||
of the local index and the backup destination store — with timestamps
|
||||
and three sizes per snapshot (compressed remote size; total
|
||||
uncompressed chunk size; size of the chunks that snapshot stored that
|
||||
were not stored before, recorded when its backup ran).
|
||||
uncompressed chunk size; size of chunks newly referenced by that
|
||||
snapshot).
|
||||
|
||||
Listing the destination store does **not** require the age secret key,
|
||||
so it works in vaultik's intended configuration, where the backed-up
|
||||
@@ -307,8 +307,8 @@ only the hash, so hostname and snapshot name exist solely in the local
|
||||
index and in the encrypted per-snapshot database. Snapshots found only
|
||||
on the destination store are therefore listed as
|
||||
`<remote only:<abbreviated remote key>>` and show `<remote only>` in
|
||||
the uncompressed and "new chunk" columns, which come only from the
|
||||
local index. Their timestamp and compressed size are real,
|
||||
the uncompressed and "new chunk" columns, which can only be computed
|
||||
from the local index. Their timestamp and compressed size are real,
|
||||
read from the manifest.
|
||||
|
||||
Snapshots in the local index with no counterpart on the destination
|
||||
|
||||
@@ -119,7 +119,6 @@ Tracks backup snapshots.
|
||||
- `completed_at` (INTEGER) - Completion timestamp (NULL if in progress)
|
||||
- `file_count` (INTEGER) - Number of files in snapshot
|
||||
- `chunk_count` (INTEGER) - Number of chunks this snapshot stored that were not stored before
|
||||
- `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
|
||||
- `total_size` (INTEGER) - Total size of all files
|
||||
- `blob_size` (INTEGER) - Total compressed size of all referenced blobs
|
||||
|
||||
@@ -86,7 +86,6 @@ CREATE TABLE IF NOT EXISTS snapshots (
|
||||
completed_at INTEGER,
|
||||
file_count INTEGER NOT NULL DEFAULT 0,
|
||||
chunk_count INTEGER NOT NULL DEFAULT 0,
|
||||
new_chunk_size INTEGER NOT NULL DEFAULT 0,
|
||||
blob_count INTEGER NOT NULL DEFAULT 0,
|
||||
total_size INTEGER NOT NULL DEFAULT 0,
|
||||
blob_size INTEGER NOT NULL DEFAULT 0,
|
||||
|
||||
@@ -129,7 +129,6 @@ func (r *SnapshotRepository) UpdateExtendedStats(
|
||||
compressionLevel int,
|
||||
uploadBytes int64,
|
||||
uploadDurationMs int64,
|
||||
newChunkSize int64,
|
||||
) error {
|
||||
compressionRatio, err := r.extendedCompressionRatio(
|
||||
ctx, tx, snapshotID, blobUncompressedSize,
|
||||
@@ -144,19 +143,18 @@ func (r *SnapshotRepository) UpdateExtendedStats(
|
||||
compression_ratio = ?,
|
||||
compression_level = ?,
|
||||
upload_bytes = ?,
|
||||
upload_duration_ms = ?,
|
||||
new_chunk_size = ?
|
||||
upload_duration_ms = ?
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
if tx != nil {
|
||||
_, err = tx.ExecContext(ctx, query,
|
||||
blobUncompressedSize, compressionRatio, compressionLevel,
|
||||
uploadBytes, uploadDurationMs, newChunkSize, snapshotID)
|
||||
uploadBytes, uploadDurationMs, snapshotID)
|
||||
} else {
|
||||
_, err = r.db.ExecWithLog(ctx, query,
|
||||
blobUncompressedSize, compressionRatio, compressionLevel,
|
||||
uploadBytes, uploadDurationMs, newChunkSize, snapshotID)
|
||||
uploadBytes, uploadDurationMs, snapshotID)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
@@ -596,24 +594,45 @@ func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
|
||||
return totalSize, nil
|
||||
}
|
||||
|
||||
// GetSnapshotNewChunkSize returns the total plaintext size of the chunks the
|
||||
// snapshot's backup stored that were not stored before, as the backup
|
||||
// recorded it. It is not derived from the snapshot's files: a file's row is
|
||||
// shared by every snapshot that lists it and is rewritten when the file
|
||||
// changes, so it shows only the file's latest chunks.
|
||||
// GetSnapshotNewChunkSize returns the sum of plaintext sizes of chunks that are
|
||||
// referenced by this snapshot but not by any earlier completed snapshot known to
|
||||
// the local database. The result is the marginal uncompressed data this snapshot
|
||||
// added to the dedup pool — i.e., the delta from prior snapshots.
|
||||
func (r *SnapshotRepository) GetSnapshotNewChunkSize(
|
||||
ctx context.Context, snapshotID string,
|
||||
) (int64, error) {
|
||||
query := `SELECT new_chunk_size FROM snapshots WHERE id = ?`
|
||||
query := `
|
||||
WITH this_snap_chunks AS (
|
||||
SELECT DISTINCT fc.chunk_hash
|
||||
FROM snapshot_files sf
|
||||
JOIN file_chunks fc ON sf.file_id = fc.file_id
|
||||
WHERE sf.snapshot_id = ?
|
||||
),
|
||||
prior_chunks AS (
|
||||
SELECT DISTINCT fc.chunk_hash
|
||||
FROM snapshots s
|
||||
JOIN snapshot_files sf ON sf.snapshot_id = s.id
|
||||
JOIN file_chunks fc ON fc.file_id = sf.file_id
|
||||
WHERE s.completed_at IS NOT NULL
|
||||
AND s.id != ?
|
||||
AND s.started_at < (SELECT started_at FROM snapshots WHERE id = ?)
|
||||
)
|
||||
SELECT COALESCE(SUM(c.size), 0)
|
||||
FROM chunks c
|
||||
JOIN this_snap_chunks t ON c.chunk_hash = t.chunk_hash
|
||||
WHERE c.chunk_hash NOT IN (SELECT chunk_hash FROM prior_chunks)
|
||||
`
|
||||
|
||||
var newChunkSize int64
|
||||
var totalSize int64
|
||||
|
||||
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&newChunkSize)
|
||||
err := r.db.conn.QueryRowContext(
|
||||
ctx, query, snapshotID, snapshotID, snapshotID,
|
||||
).Scan(&totalSize)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("querying new chunk size: %w", err)
|
||||
}
|
||||
|
||||
return newChunkSize, nil
|
||||
return totalSize, nil
|
||||
}
|
||||
|
||||
// GetIncompleteSnapshots returns all snapshots that haven't been completed
|
||||
|
||||
@@ -135,9 +135,7 @@ type ScannerConfig struct {
|
||||
// are counted per file: BytesScanned is the size of the new and changed
|
||||
// files, BytesSkipped that of the unchanged ones. FilesFailed counts the
|
||||
// new and changed files that phase 2 could not store; FilesScanned
|
||||
// includes them and BytesScanned does not. ChunksCreated counts the
|
||||
// chunks that were not already stored, and NewChunkSize is their total
|
||||
// size.
|
||||
// includes them and BytesScanned does not.
|
||||
type ScanResult struct {
|
||||
FilesScanned int
|
||||
FilesSkipped int
|
||||
@@ -147,7 +145,6 @@ type ScanResult struct {
|
||||
BytesSkipped int64
|
||||
BytesDeleted int64
|
||||
ChunksCreated int
|
||||
NewChunkSize int64
|
||||
BlobsCreated int
|
||||
BlobsUploaded int
|
||||
BytesUploaded int64
|
||||
@@ -1869,12 +1866,11 @@ func (s *Scanner) processFileStreaming(
|
||||
return nil
|
||||
}
|
||||
|
||||
// updateChunkStats counts a chunk that was not already stored, and its
|
||||
// size. The scan result's file counts, BytesScanned and BytesSkipped are
|
||||
// not touched here: the scan phase counts each file once.
|
||||
// updateChunkStats counts a chunk that was not already stored. The scan
|
||||
// result's file counts, BytesScanned and BytesSkipped are not touched
|
||||
// here: the scan phase counts each file once.
|
||||
func (s *Scanner) updateChunkStats(chunkSize int64, result *ScanResult) {
|
||||
result.ChunksCreated++
|
||||
result.NewChunkSize += chunkSize
|
||||
|
||||
if s.progress != nil {
|
||||
s.progress.GetStats().ChunksCreated.Add(1)
|
||||
|
||||
@@ -199,7 +199,6 @@ func (sm *SnapshotManager) UpdateSnapshotStatsExtended(
|
||||
stats.CompressionLevel,
|
||||
stats.BytesUploaded,
|
||||
stats.UploadDurationMs,
|
||||
stats.NewChunkSize,
|
||||
)
|
||||
})
|
||||
}
|
||||
@@ -907,7 +906,6 @@ type ExtendedBackupStats struct {
|
||||
BlobUncompressedSize int64 // Total uncompressed size of all referenced blobs
|
||||
CompressionLevel int // Compression level used for this snapshot
|
||||
UploadDurationMs int64 // Total milliseconds spent uploading to S3
|
||||
NewChunkSize int64 // Total size of the chunks counted in ChunksCreated
|
||||
}
|
||||
|
||||
// deleteOtherSnapshots deletes all snapshots except the current one
|
||||
|
||||
+24
-5
@@ -29,7 +29,10 @@ var (
|
||||
ErrURLCredentials = errors.New(
|
||||
"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")
|
||||
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.
|
||||
@@ -90,8 +93,9 @@ func parseFileURL(rawPath string) (*URL, error) {
|
||||
}
|
||||
|
||||
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
||||
// the userinfo and any query parameter other than endpoint, region and
|
||||
// ssl, so a credential-bearing URL is never stored or echoed.
|
||||
// the userinfo or in the endpoint parameter's 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,6 +118,14 @@ func parseS3URL(rawURL string) (*URL, error) {
|
||||
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,
|
||||
@@ -125,8 +137,9 @@ func parseS3URL(rawURL string) (*URL, error) {
|
||||
}
|
||||
|
||||
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
||||
// query parameters, so credentials in the userinfo and any parameter at
|
||||
// all are rejected rather than silently ignored.
|
||||
// query parameters, so credentials in the userinfo, parameters in the
|
||||
// remote name and any query parameter at all are rejected rather than
|
||||
// silently ignored.
|
||||
func parseRcloneURL(rawURL string) (*URL, error) {
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
@@ -142,6 +155,12 @@ func parseRcloneURL(rawURL string) (*URL, error) {
|
||||
return nil, ErrMissingRemote
|
||||
}
|
||||
|
||||
// An rclone remote name never contains a comma; one starts rclone's
|
||||
// remote,param=value form, whose values can be credentials.
|
||||
if strings.Contains(remote, ",") {
|
||||
return nil, ErrRcloneRemoteParams
|
||||
}
|
||||
|
||||
err = rejectUnknownParams(u.Query())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -150,6 +150,19 @@ 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,
|
||||
@@ -203,6 +216,27 @@ 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.
|
||||
func TestParseStorageURLRejectsRcloneRemoteParams(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const secret = "topsecret"
|
||||
|
||||
raw := "rclone://gdrive,client_secret=" + secret + "/backups"
|
||||
|
||||
_, err := storage.ParseStorageURL(raw)
|
||||
if !errors.Is(err, storage.ErrRcloneRemoteParams) {
|
||||
t.Fatalf("ParseStorageURL(%q) error = %v, want %v",
|
||||
raw, err, storage.ErrRcloneRemoteParams)
|
||||
}
|
||||
|
||||
if strings.Contains(err.Error(), secret) {
|
||||
t.Errorf("error message leaked %q: %v", secret, 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,133 +0,0 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"maps"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/chunker"
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
|
||||
// TestNewChunkSizeOfChangedFileStaysFixed covers
|
||||
// https://git.eeqj.de/sneak/vaultik/issues/311. A backup whose only change
|
||||
// is a modified file lists the size of the chunks that change added as its
|
||||
// NEW CHUNK SIZE, and a later backup that changes the file again leaves
|
||||
// every earlier snapshot's figure as it was. The file's row in the local
|
||||
// index is shared by all three snapshots and rewritten by each backup.
|
||||
func TestNewChunkSizeOfChangedFileStaysFixed(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"))
|
||||
|
||||
firstFiles := maps.Clone(files)
|
||||
firstID := localSnapshotID(ctx, t, repos, "first")
|
||||
|
||||
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
|
||||
require.NoError(t, backUp(v, "second"))
|
||||
|
||||
secondID := localSnapshotID(ctx, t, repos, "second")
|
||||
secondAdded := sizeOfNewChunks(t, files, firstFiles)
|
||||
require.Positive(t, secondAdded)
|
||||
assert.Equal(t, secondAdded, listedNewChunkSizes(t, v)[secondID],
|
||||
"the second snapshot must count the chunks the change added")
|
||||
|
||||
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
|
||||
require.NoError(t, backUp(v, "third"))
|
||||
|
||||
sizes := listedNewChunkSizes(t, v)
|
||||
assert.Equal(t, sizeOfNewChunks(t, firstFiles, nil), sizes[firstID],
|
||||
"a later backup must not change the first snapshot's figure")
|
||||
assert.Equal(t, secondAdded, sizes[secondID],
|
||||
"a later backup must not change the second snapshot's figure")
|
||||
}
|
||||
|
||||
// sizeOfNewChunks returns the total size of the distinct chunks of the
|
||||
// files in after that no file in before has, cut the way the backup cuts
|
||||
// them.
|
||||
func sizeOfNewChunks(t *testing.T, after, before map[string][]byte) int64 {
|
||||
t.Helper()
|
||||
|
||||
known := make(map[string]bool)
|
||||
for _, chunk := range chunksOf(t, before) {
|
||||
known[chunk.Hash] = true
|
||||
}
|
||||
|
||||
var total int64
|
||||
|
||||
for _, chunk := range chunksOf(t, after) {
|
||||
if !known[chunk.Hash] {
|
||||
known[chunk.Hash] = true
|
||||
total += chunk.Size
|
||||
}
|
||||
}
|
||||
|
||||
return total
|
||||
}
|
||||
|
||||
// chunksOf returns the chunks of every file in files.
|
||||
func chunksOf(t *testing.T, files map[string][]byte) []chunker.Chunk {
|
||||
t.Helper()
|
||||
|
||||
c := chunker.NewChunker(faultChunkSize)
|
||||
|
||||
var all []chunker.Chunk
|
||||
|
||||
for _, content := range files {
|
||||
chunks, err := c.ChunkReader(bytes.NewReader(content))
|
||||
require.NoError(t, err)
|
||||
|
||||
all = append(all, chunks...)
|
||||
}
|
||||
|
||||
return all
|
||||
}
|
||||
|
||||
// listedNewChunkSizes runs `snapshot list --json` and returns each listed
|
||||
// snapshot's new_chunk_size by snapshot ID.
|
||||
func listedNewChunkSizes(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.NewChunkSize
|
||||
}
|
||||
|
||||
return sizes
|
||||
}
|
||||
@@ -209,7 +209,6 @@ type snapshotStats struct {
|
||||
totalFiles int
|
||||
totalBytes int64
|
||||
totalChunks int
|
||||
newChunkSize int64
|
||||
totalBlobs int
|
||||
totalBytesSkipped int64
|
||||
totalFilesSkipped int
|
||||
@@ -342,7 +341,6 @@ func (v *Vaultik) scanAllDirectories(
|
||||
stats.totalFiles += result.FilesScanned
|
||||
stats.totalBytes += result.BytesScanned
|
||||
stats.totalChunks += result.ChunksCreated
|
||||
stats.newChunkSize += result.NewChunkSize
|
||||
stats.totalBlobs += result.BlobsCreated
|
||||
stats.totalFilesSkipped += result.FilesSkipped
|
||||
stats.totalFilesFailed += result.FilesFailed
|
||||
@@ -405,7 +403,6 @@ func (v *Vaultik) finalizeSnapshotMetadata(
|
||||
BlobUncompressedSize: stats.blobUncompressedSize,
|
||||
CompressionLevel: v.Config.CompressionLevel,
|
||||
UploadDurationMs: stats.uploadDuration.Milliseconds(),
|
||||
NewChunkSize: stats.newChunkSize,
|
||||
}
|
||||
|
||||
err = v.SnapshotManager.UpdateSnapshotStatsExtended(v.ctx, snapshotID, extStats)
|
||||
|
||||
@@ -445,7 +445,6 @@ type listJSONRow struct {
|
||||
RemoteKey string `json:"remote_key"`
|
||||
Timestamp string `json:"timestamp"`
|
||||
CompressedSize int64 `json:"compressed_size"`
|
||||
NewChunkSize int64 `json:"new_chunk_size"`
|
||||
LocallyTracked bool `json:"locally_tracked"`
|
||||
RemotePresent *bool `json:"remote_present"`
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user