From d73e0d60cea059df3adef2c15d53fd4cfde943d7 Mon Sep 17 00:00:00 2001 From: Steve Joachim Date: Wed, 30 Jun 2021 16:46:51 -0400 Subject: [PATCH] robustness: reduce memory consumption (#1155) Detailed changes: * Implement Kopia API client * Implement kopia client * Implement new persister * Separate storeLoad and Delete concurrency tests * Update Store interface to return error for Delete * Return errors from os.RemoveAll * Robustness test updates * Push robustness metadata without writing to fs * Fix testenv.AssertNoError references * Update tests to use new kopia persister * Minor updates to multiclient test cases * Use require.NoError instead of assertNoError * Add context to store interface * Update logging to be less verbose * Use io instead of ioutil package * Simplify restore by using object ID * Accommodate repository.NewWriter signature change * Improve tests to increase code coverage * Spelling and error string fixes * Address lint errors Co-authored-by: Nick --- tests/robustness/checker/checker.go | 40 +-- tests/robustness/engine/engine.go | 12 +- tests/robustness/engine/engine_test.go | 20 +- tests/robustness/engine/metadata.go | 25 +- .../multiclient_test/framework/harness.go | 13 +- .../multiclient_test/multiclient_test.go | 14 +- tests/robustness/persister.go | 8 +- tests/robustness/robustness_test/main_test.go | 13 +- .../robustness_test/robustness_test.go | 9 +- .../snapmeta/kopia_persister_light.go | 114 ++++++++ .../snapmeta/kopia_persister_light_test.go | 168 ++++++++++++ tests/robustness/snapmeta/simple.go | 10 +- tests/robustness/snapmeta/simple_test.go | 13 +- tests/tools/kopiaclient/kopiaclient.go | 248 ++++++++++++++++++ 14 files changed, 618 insertions(+), 89 deletions(-) create mode 100644 tests/robustness/snapmeta/kopia_persister_light.go create mode 100644 tests/robustness/snapmeta/kopia_persister_light_test.go create mode 100644 tests/tools/kopiaclient/kopiaclient.go diff --git a/tests/robustness/checker/checker.go b/tests/robustness/checker/checker.go index 7bad3c33e..96fed9408 100644 --- a/tests/robustness/checker/checker.go +++ b/tests/robustness/checker/checker.go @@ -98,11 +98,11 @@ func (ssMeta *SnapshotMetadata) IsDeleted() bool { } // GetSnapshotMetadata gets the metadata associated with the given snapshot ID. -func (chk *Checker) GetSnapshotMetadata(snapID string) (*SnapshotMetadata, error) { +func (chk *Checker) GetSnapshotMetadata(ctx context.Context, snapID string) (*SnapshotMetadata, error) { chk.mu.RLock() defer chk.mu.RUnlock() - return chk.loadSnapshotMetadata(snapID) + return chk.loadSnapshotMetadata(ctx, snapID) } // GetLiveSnapIDs gets the list of snapshot IDs being tracked by the checker's snapshot store @@ -116,8 +116,8 @@ func (chk *Checker) GetLiveSnapIDs() []string { // IsSnapshotIDDeleted reports whether the metadata associated with the provided snapshot ID // has it marked as deleted. -func (chk *Checker) IsSnapshotIDDeleted(snapID string) (bool, error) { - md, err := chk.loadSnapshotMetadata(snapID) +func (chk *Checker) IsSnapshotIDDeleted(ctx context.Context, snapID string) (bool, error) { + md, err := chk.loadSnapshotMetadata(ctx, snapID) if err != nil { return false, err } @@ -157,7 +157,7 @@ func (chk *Checker) VerifySnapshotMetadata(ctx context.Context) error { if chk.RecoveryMode { chk.mu.Lock() - chk.snapshotMetadataStore.Delete(metaSnapID) + chk.snapshotMetadataStore.Delete(ctx, metaSnapID) chk.SnapIDIndex.RemoveFromIndex(metaSnapID, LiveSnapshotsIdxName) chk.mu.Unlock() } else { @@ -222,7 +222,7 @@ func (chk *Checker) TakeSnapshot(ctx context.Context, sourceDir string, opts map chk.mu.Lock() defer chk.mu.Unlock() - err = chk.saveSnapshotMetadata(ssMeta) + err = chk.saveSnapshotMetadata(ctx, ssMeta) if err != nil { return snapID, err } @@ -252,7 +252,7 @@ func (chk *Checker) RestoreSnapshot(ctx context.Context, snapID string, reportOu // using the Checker's Snapshotter, and performs a data consistency check on the // resulting tree using the saved snapshot data. func (chk *Checker) RestoreSnapshotToPath(ctx context.Context, snapID, destPath string, reportOut io.Writer, opts map[string]string) error { - ssMeta, err := chk.safeRestorePrepare(snapID) + ssMeta, err := chk.safeRestorePrepare(ctx, snapID) if err != nil { return err } @@ -262,7 +262,7 @@ func (chk *Checker) RestoreSnapshotToPath(ctx context.Context, snapID, destPath // safeRestorePrepare wraps key indexing operations that need to be // executed together under one mutex lock. -func (chk *Checker) safeRestorePrepare(snapID string) (*SnapshotMetadata, error) { +func (chk *Checker) safeRestorePrepare(ctx context.Context, snapID string) (*SnapshotMetadata, error) { chk.mu.RLock() defer chk.mu.RUnlock() @@ -273,7 +273,7 @@ func (chk *Checker) safeRestorePrepare(snapID string) (*SnapshotMetadata, error) return nil, robustness.ErrNoOp } - ssMeta, err := chk.loadSnapshotMetadata(snapID) + ssMeta, err := chk.loadSnapshotMetadata(ctx, snapID) if err != nil { return nil, err } @@ -309,7 +309,7 @@ func (chk *Checker) RestoreVerifySnapshot(ctx context.Context, snapID, destPath chk.mu.Lock() defer chk.mu.Unlock() - err = chk.saveSnapshotMetadata(ssMeta) + err = chk.saveSnapshotMetadata(ctx, ssMeta) if err != nil { return err } @@ -333,7 +333,7 @@ func (chk *Checker) DeleteSnapshot(ctx context.Context, snapID string, opts map[ // Load the metadata, ensure the snapshot hasn't already been deleted. // If not, remove it from the list of live snapshots. These two tasks must // be done atomically. - ssMeta, err := chk.safeDeletePrepare(snapID) + ssMeta, err := chk.safeDeletePrepare(ctx, snapID) if err != nil { return err } @@ -346,7 +346,7 @@ func (chk *Checker) DeleteSnapshot(ctx context.Context, snapID string, opts map[ ssMeta.DeletionTime = clock.Now() ssMeta.ValidationData = nil - return chk.safeDeleteFinish(ssMeta) + return chk.safeDeleteFinish(ctx, ssMeta) } // safeDeletePrepare is a routine that will remove the snapshot ID from @@ -354,7 +354,7 @@ func (chk *Checker) DeleteSnapshot(ctx context.Context, snapID string, opts map[ // The check for deletion must be done atomically with the removal from the // index to ensure no races, which could lead to false failures if the Delete // operation runs against a snapshot that has already been deleted. -func (chk *Checker) safeDeletePrepare(snapID string) (*SnapshotMetadata, error) { +func (chk *Checker) safeDeletePrepare(ctx context.Context, snapID string) (*SnapshotMetadata, error) { chk.mu.Lock() defer chk.mu.Unlock() @@ -364,7 +364,7 @@ func (chk *Checker) safeDeletePrepare(snapID string) (*SnapshotMetadata, error) return nil, robustness.ErrNoOp } - ssMeta, err := chk.loadSnapshotMetadata(snapID) + ssMeta, err := chk.loadSnapshotMetadata(ctx, snapID) if err != nil { return nil, err } @@ -381,13 +381,13 @@ func (chk *Checker) safeDeletePrepare(snapID string) (*SnapshotMetadata, error) // safeDelete finish will atomically update the metadata repository with the // deleted snapshot metadata and add the ID to the deleted snapshots index. -func (chk *Checker) safeDeleteFinish(ssMeta *SnapshotMetadata) error { +func (chk *Checker) safeDeleteFinish(ctx context.Context, ssMeta *SnapshotMetadata) error { chk.mu.Lock() defer chk.mu.Unlock() chk.SnapIDIndex.AddToIndex(ssMeta.SnapID, DeletedSnapshotsIdxName) - err := chk.saveSnapshotMetadata(ssMeta) + err := chk.saveSnapshotMetadata(ctx, ssMeta) if err != nil { return err } @@ -395,13 +395,13 @@ func (chk *Checker) safeDeleteFinish(ssMeta *SnapshotMetadata) error { return nil } -func (chk *Checker) saveSnapshotMetadata(ssMeta *SnapshotMetadata) error { +func (chk *Checker) saveSnapshotMetadata(ctx context.Context, ssMeta *SnapshotMetadata) error { ssMetaRaw, err := json.Marshal(ssMeta) if err != nil { return err } - err = chk.snapshotMetadataStore.Store(ssMeta.SnapID, ssMetaRaw) + err = chk.snapshotMetadataStore.Store(ctx, ssMeta.SnapID, ssMetaRaw) if err != nil { return err } @@ -409,9 +409,9 @@ func (chk *Checker) saveSnapshotMetadata(ssMeta *SnapshotMetadata) error { return nil } -func (chk *Checker) loadSnapshotMetadata(snapID string) (*SnapshotMetadata, error) { +func (chk *Checker) loadSnapshotMetadata(ctx context.Context, snapID string) (*SnapshotMetadata, error) { // Lookup metadata by snapshot ID - b, err := chk.snapshotMetadataStore.Load(snapID) + b, err := chk.snapshotMetadataStore.Load(ctx, snapID) if err != nil { return nil, err } diff --git a/tests/robustness/engine/engine.go b/tests/robustness/engine/engine.go index 287c042a7..ba54c6b4d 100644 --- a/tests/robustness/engine/engine.go +++ b/tests/robustness/engine/engine.go @@ -134,17 +134,17 @@ func (e *Engine) Shutdown(ctx context.Context) error { defer e.cleanComponents() if e.MetaStore != nil { - err := e.saveLog() + err := e.saveLog(ctx) if err != nil { return err } - err = e.saveStats() + err = e.saveStats(ctx) if err != nil { return err } - err = e.saveSnapIDIndex() + err = e.saveSnapIDIndex(ctx) if err != nil { return err } @@ -193,19 +193,19 @@ func (e *Engine) Init(ctx context.Context) error { return err } - err = e.loadStats() + err = e.loadStats(ctx) if err != nil { return err } e.CumulativeStats.RunCounter++ - err = e.loadLog() + err = e.loadLog(ctx) if err != nil { return err } - err = e.loadSnapIDIndex() + err = e.loadSnapIDIndex(ctx) if err != nil { return err } diff --git a/tests/robustness/engine/engine_test.go b/tests/robustness/engine/engine_test.go index a79c665c9..406a86fe7 100644 --- a/tests/robustness/engine/engine_test.go +++ b/tests/robustness/engine/engine_test.go @@ -298,7 +298,7 @@ func TestSnapshotVerificationFail(t *testing.T) { require.NoError(t, err) // Get the metadata collected on that snapshot - ssMeta1, err := eng.Checker.GetSnapshotMetadata(snapID1) + ssMeta1, err := eng.Checker.GetSnapshotMetadata(ctx, snapID1) require.NoError(t, err) // Do additional writes, writing 1 extra byte than before @@ -310,7 +310,7 @@ func TestSnapshotVerificationFail(t *testing.T) { require.NoError(t, err) // Get the second snapshot's metadata - ssMeta2, err := eng.Checker.GetSnapshotMetadata(snapID2) + ssMeta2, err := eng.Checker.GetSnapshotMetadata(ctx, snapID2) require.NoError(t, err) // Swap second snapshot's validation data into the first's metadata @@ -369,11 +369,11 @@ func TestDataPersistency(t *testing.T) { require.NoError(t, err) // Get the walk data associated with the snapshot that was taken - dataDirWalk, err := eng.Checker.GetSnapshotMetadata(snapID) + dataDirWalk, err := eng.Checker.GetSnapshotMetadata(ctx, snapID) require.NoError(t, err) // Save the snapshot ID index - err = eng.saveSnapIDIndex() + err = eng.saveSnapIDIndex(ctx) require.NoError(t, err) // Flush the snapshot metadata to persistent storage @@ -693,6 +693,8 @@ func TestIOLimitPerWriteAction(t *testing.T) { } func TestStatsPersist(t *testing.T) { + ctx := context.Background() + tmpDir, err := ioutil.TempDir("", "stats-persist-test") require.NoError(t, err) @@ -729,7 +731,7 @@ func TestStatsPersist(t *testing.T) { }, } - err = eng.saveStats() + err = eng.saveStats(ctx) require.NoError(t, err) err = eng.MetaStore.FlushMetadata() @@ -749,7 +751,7 @@ func TestStatsPersist(t *testing.T) { MetaStore: snapStoreNew, } - err = engNew.loadStats() + err = engNew.loadStats(ctx) require.NoError(t, err) if got, want := engNew.Stats(), eng.Stats(); got != want { @@ -761,6 +763,8 @@ func TestStatsPersist(t *testing.T) { } func TestLogsPersist(t *testing.T) { + ctx := context.Background() + tmpDir, err := ioutil.TempDir("", "logs-persist-test") require.NoError(t, err) @@ -799,7 +803,7 @@ func TestLogsPersist(t *testing.T) { EngineLog: logData, } - err = eng.saveLog() + err = eng.saveLog(ctx) require.NoError(t, err) err = eng.MetaStore.FlushMetadata() @@ -819,7 +823,7 @@ func TestLogsPersist(t *testing.T) { MetaStore: snapStoreNew, } - err = engNew.loadLog() + err = engNew.loadLog(ctx) require.NoError(t, err) if got, want := engNew.EngineLog.String(), eng.EngineLog.String(); got != want { diff --git a/tests/robustness/engine/metadata.go b/tests/robustness/engine/metadata.go index 3749c1e9b..24ef231da 100644 --- a/tests/robustness/engine/metadata.go +++ b/tests/robustness/engine/metadata.go @@ -3,6 +3,7 @@ package engine import ( + "context" "encoding/json" "errors" @@ -17,18 +18,18 @@ const ( ) // saveLog saves the engine Log in the metadata store. -func (e *Engine) saveLog() error { +func (e *Engine) saveLog(ctx context.Context) error { b, err := json.Marshal(e.EngineLog) if err != nil { return err } - return e.MetaStore.Store(engineLogsStoreKey, b) + return e.MetaStore.Store(ctx, engineLogsStoreKey, b) } // loadLog loads the engine log from the metadata store. -func (e *Engine) loadLog() error { - b, err := e.MetaStore.Load(engineLogsStoreKey) +func (e *Engine) loadLog(ctx context.Context) error { + b, err := e.MetaStore.Load(ctx, engineLogsStoreKey) if err != nil { if errors.Is(err, robustness.ErrKeyNotFound) { // Swallow key-not-found error. May not have historical logs @@ -49,18 +50,18 @@ func (e *Engine) loadLog() error { } // saveStats saves the engine Stats in the metadata store. -func (e *Engine) saveStats() error { +func (e *Engine) saveStats(ctx context.Context) error { cumulStatRaw, err := json.Marshal(e.CumulativeStats) if err != nil { return err } - return e.MetaStore.Store(engineStatsStoreKey, cumulStatRaw) + return e.MetaStore.Store(ctx, engineStatsStoreKey, cumulStatRaw) } // loadStats loads the engine Stats from the metadata store. -func (e *Engine) loadStats() error { - b, err := e.MetaStore.Load(engineStatsStoreKey) +func (e *Engine) loadStats(ctx context.Context) error { + b, err := e.MetaStore.Load(ctx, engineStatsStoreKey) if err != nil { if errors.Is(err, robustness.ErrKeyNotFound) { // Swallow key-not-found error. We may not have historical @@ -78,18 +79,18 @@ func (e *Engine) loadStats() error { } // saveSnapIDIndex saves the Checker's snapshot ID index in the metadata store. -func (e *Engine) saveSnapIDIndex() error { +func (e *Engine) saveSnapIDIndex(ctx context.Context) error { snapIDIdxRaw, err := json.Marshal(e.Checker.SnapIDIndex) if err != nil { return err } - return e.MetaStore.Store(snapIDIndexStoreKey, snapIDIdxRaw) + return e.MetaStore.Store(ctx, snapIDIndexStoreKey, snapIDIdxRaw) } // loadSnapIDIndex loads the Checker's snapshot ID index from the metadata store. -func (e *Engine) loadSnapIDIndex() error { - b, err := e.MetaStore.Load(snapIDIndexStoreKey) +func (e *Engine) loadSnapIDIndex(ctx context.Context) error { + b, err := e.MetaStore.Load(ctx, snapIDIndexStoreKey) if err != nil { if errors.Is(err, robustness.ErrKeyNotFound) { // Swallow key-not-found error. We may not have historical diff --git a/tests/robustness/multiclient_test/framework/harness.go b/tests/robustness/multiclient_test/framework/harness.go index f62cd3a91..0ca3aa597 100644 --- a/tests/robustness/multiclient_test/framework/harness.go +++ b/tests/robustness/multiclient_test/framework/harness.go @@ -44,7 +44,7 @@ type TestHarness struct { baseDirPath string fileWriter *MultiClientFileWriter snapshotter *MultiClientSnapshotter - persister *snapmeta.KopiaPersister + persister *snapmeta.KopiaPersisterLight engine *engine.Engine skipTest bool @@ -149,16 +149,9 @@ func (th *TestHarness) getSnapshotter() bool { } func (th *TestHarness) getPersister() bool { - kp, err := snapmeta.NewPersister(th.baseDirPath) + kp, err := snapmeta.NewPersisterLight(th.baseDirPath) if err != nil { - if errors.Is(err, kopiarunner.ErrExeVariableNotSet) { - log.Println("Skipping robustness tests because KOPIA_EXE is not set") - - th.skipTest = true - } else { - log.Println("Error creating kopia Persister:", err) - } - + log.Println("Error creating kopia Persister:", err) return false } diff --git a/tests/robustness/multiclient_test/multiclient_test.go b/tests/robustness/multiclient_test/multiclient_test.go index 273051e83..507487cfb 100644 --- a/tests/robustness/multiclient_test/multiclient_test.go +++ b/tests/robustness/multiclient_test/multiclient_test.go @@ -44,14 +44,14 @@ func TestManySmallFiles(t *testing.T) { _, err = eng.ExecAction(ctx, engine.WriteRandomFilesActionKey, fileWriteOpts) require.NoError(t, err) - _, err = eng.ExecAction(ctx, engine.SnapshotDirActionKey, nil) + snapOut, err := eng.ExecAction(ctx, engine.SnapshotDirActionKey, nil) require.NoError(t, err) - _, err = eng.ExecAction(ctx, engine.RestoreSnapshotActionKey, nil) + _, err = eng.ExecAction(ctx, engine.RestoreSnapshotActionKey, snapOut) require.NoError(t, err) } - ctx := testlogging.Context(t) + ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo) th.RunN(ctx, t, numClients, f) } @@ -82,7 +82,7 @@ func TestOneLargeFile(t *testing.T) { require.NoError(t, err) } - ctx := testlogging.Context(t) + ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo) th.RunN(ctx, t, numClients, f) } @@ -117,12 +117,12 @@ func TestManySmallFilesAcrossDirecoryTree(t *testing.T) { require.NoError(t, err) } - ctx := testlogging.Context(t) + ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo) th.RunN(ctx, t, numClients, f) } func TestRandomizedSmall(t *testing.T) { - numClients := 2 + numClients := 4 st := clock.Now() opts := engine.ActionOpts{ @@ -151,7 +151,7 @@ func TestRandomizedSmall(t *testing.T) { } } - ctx := testlogging.Context(t) + ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo) th.RunN(ctx, t, numClients, f) } diff --git a/tests/robustness/persister.go b/tests/robustness/persister.go index 5fdecff24..16d3c0679 100644 --- a/tests/robustness/persister.go +++ b/tests/robustness/persister.go @@ -2,12 +2,14 @@ package robustness +import "context" + // Store describes the ability to store and retrieve // a buffer of metadata, indexed by a string key. type Store interface { - Store(key string, val []byte) error - Load(key string) ([]byte, error) - Delete(key string) + Store(ctx context.Context, key string, val []byte) error + Load(ctx context.Context, key string) ([]byte, error) + Delete(ctx context.Context, key string) error } // Persister describes the ability to flush metadata diff --git a/tests/robustness/robustness_test/main_test.go b/tests/robustness/robustness_test/main_test.go index 35e615411..b4a052ed8 100644 --- a/tests/robustness/robustness_test/main_test.go +++ b/tests/robustness/robustness_test/main_test.go @@ -73,7 +73,7 @@ type kopiaRobustnessTestHarness struct { baseDirPath string fileWriter *fiofilewriter.FileWriter snapshotter *snapmeta.KopiaSnapshotter - persister *snapmeta.KopiaPersister + persister *snapmeta.KopiaPersisterLight engine *engine.Engine skipTest bool @@ -154,16 +154,9 @@ func (th *kopiaRobustnessTestHarness) getSnapshotter() bool { } func (th *kopiaRobustnessTestHarness) getPersister() bool { - kp, err := snapmeta.NewPersister(th.baseDirPath) + kp, err := snapmeta.NewPersisterLight(th.baseDirPath) if err != nil { - if errors.Is(err, kopiarunner.ErrExeVariableNotSet) { - log.Println("Skipping robustness tests because KOPIA_EXE is not set") - - th.skipTest = true - } else { - log.Println("Error creating kopia Persister:", err) - } - + log.Println("Error creating kopia Persister:", err) return false } diff --git a/tests/robustness/robustness_test/robustness_test.go b/tests/robustness/robustness_test/robustness_test.go index 749d2a087..a180fe436 100644 --- a/tests/robustness/robustness_test/robustness_test.go +++ b/tests/robustness/robustness_test/robustness_test.go @@ -29,7 +29,7 @@ func TestManySmallFiles(t *testing.T) { fiofilewriter.MinNumFilesPerWriteField: strconv.Itoa(numFiles), } - ctx := testlogging.Context(t) + ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo) _, err := eng.ExecAction(ctx, engine.WriteRandomFilesActionKey, fileWriteOpts) require.NoError(t, err) @@ -53,7 +53,7 @@ func TestOneLargeFile(t *testing.T) { fiofilewriter.MinNumFilesPerWriteField: strconv.Itoa(numFiles), } - ctx := testlogging.Context(t) + ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo) _, err := eng.ExecAction(ctx, engine.WriteRandomFilesActionKey, fileWriteOpts) require.NoError(t, err) @@ -81,7 +81,7 @@ func TestManySmallFilesAcrossDirecoryTree(t *testing.T) { engine.ActionRepeaterField: strconv.Itoa(actionRepeats), } - ctx := testlogging.Context(t) + ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo) _, err := eng.ExecAction(ctx, engine.WriteRandomFilesActionKey, fileWriteOpts) require.NoError(t, err) @@ -112,9 +112,8 @@ func TestRandomizedSmall(t *testing.T) { }, } + ctx := testlogging.ContextWithLevel(t, testlogging.LevelInfo) for clock.Since(st) <= *randomizedTestDur { - ctx := testlogging.Context(t) - err := eng.RandomAction(ctx, opts) if errors.Is(err, robustness.ErrNoOp) { t.Log("Random action resulted in no-op") diff --git a/tests/robustness/snapmeta/kopia_persister_light.go b/tests/robustness/snapmeta/kopia_persister_light.go new file mode 100644 index 000000000..668267d54 --- /dev/null +++ b/tests/robustness/snapmeta/kopia_persister_light.go @@ -0,0 +1,114 @@ +// +build darwin,amd64 linux,amd64 + +package snapmeta + +import ( + "context" + "log" + "os" + "sync" + + "github.com/kopia/kopia/tests/robustness" + "github.com/kopia/kopia/tests/tools/kopiaclient" +) + +// KopiaPersisterLight is a wrapper for KopiaClient that satisfies the Persister +// interface. +type KopiaPersisterLight struct { + kc *kopiaclient.KopiaClient + keysInProcess map[string]bool + c *sync.Cond + baseDir string +} + +var _ robustness.Persister = (*KopiaPersisterLight)(nil) + +// NewPersisterLight returns a new KopiaPersisterLight. +func NewPersisterLight(baseDir string) (*KopiaPersisterLight, error) { + persistenceDir, err := os.MkdirTemp(baseDir, "kopia-persistence-root-") + if err != nil { + return nil, err + } + + return &KopiaPersisterLight{ + kc: kopiaclient.NewKopiaClient(persistenceDir), + keysInProcess: map[string]bool{}, + c: sync.NewCond(&sync.Mutex{}), + baseDir: persistenceDir, + }, nil +} + +// ConnectOrCreateRepo creates a new Kopia repo or connects to an existing one if possible. +func (kpl *KopiaPersisterLight) ConnectOrCreateRepo(repoPath string) error { + bucketName := os.Getenv(S3BucketNameEnvKey) + return kpl.kc.CreateOrConnectRepo(context.Background(), repoPath, bucketName) +} + +// Store pushes the key value pair to the Kopia repository. +func (kpl *KopiaPersisterLight) Store(ctx context.Context, key string, val []byte) error { + kpl.waitFor(key) + defer kpl.doneWith(key) + + log.Println("pushing metadata for", key) + + return kpl.kc.SnapshotCreate(ctx, key, val) +} + +// Load pulls the key value pair from the Kopia repo and returns the value. +func (kpl *KopiaPersisterLight) Load(ctx context.Context, key string) ([]byte, error) { + kpl.waitFor(key) + defer kpl.doneWith(key) + + log.Println("pulling metadata for", key) + + return kpl.kc.SnapshotRestore(ctx, key) +} + +// Delete deletes all snapshots associated with the given key. +func (kpl *KopiaPersisterLight) Delete(ctx context.Context, key string) error { + kpl.waitFor(key) + defer kpl.doneWith(key) + + log.Println("deleting metadata for", key) + + return kpl.kc.SnapshotDelete(ctx, key) +} + +// LoadMetadata is a no-op. It is included to satisfy the Persister interface. +func (kpl *KopiaPersisterLight) LoadMetadata() error { + return nil +} + +// FlushMetadata is a no-op. It is included to satisfy the Persister interface. +func (kpl *KopiaPersisterLight) FlushMetadata() error { + return nil +} + +// GetPersistDir returns the persistence directory. +func (kpl *KopiaPersisterLight) GetPersistDir() string { + return kpl.baseDir +} + +// Cleanup removes the persistence directory and closes the Kopia repo. +func (kpl *KopiaPersisterLight) Cleanup() { + if err := os.RemoveAll(kpl.baseDir); err != nil { + log.Println("cannot remove persistence dir") + } +} + +func (kpl *KopiaPersisterLight) waitFor(key string) { + kpl.c.L.Lock() + for kpl.keysInProcess[key] { + kpl.c.Wait() + } + + kpl.keysInProcess[key] = true + kpl.c.L.Unlock() +} + +func (kpl *KopiaPersisterLight) doneWith(key string) { + kpl.c.L.Lock() + delete(kpl.keysInProcess, key) + kpl.c.L.Unlock() + kpl.c.Signal() +} diff --git a/tests/robustness/snapmeta/kopia_persister_light_test.go b/tests/robustness/snapmeta/kopia_persister_light_test.go new file mode 100644 index 000000000..1487c5b06 --- /dev/null +++ b/tests/robustness/snapmeta/kopia_persister_light_test.go @@ -0,0 +1,168 @@ +// +build darwin,amd64 linux,amd64 + +package snapmeta + +import ( + "bytes" + "context" + "fmt" + "log" + "os" + "testing" +) + +var ( + key = "mykey" + val = []byte("myval") +) + +func TestStoreLoadDelete(t *testing.T) { + ctx := context.Background() + + repoPath, err := os.MkdirTemp("", "kopia-test-repo-") + assertNoError(t, err) + + defer os.RemoveAll(repoPath) + + kpl := initKPL(t, repoPath) + defer kpl.Cleanup() + + kpl.testStoreLoad(ctx, t, key, val) + kpl.testDelete(ctx, t, key) +} + +func TestConcurrency(t *testing.T) { + ctx := context.Background() + + repoPath, err := os.MkdirTemp("", "kopia-test-repo-") + assertNoError(t, err) + + defer os.RemoveAll(repoPath) + + kpl := initKPL(t, repoPath) + defer kpl.Cleanup() + + keys := []string{"key1", "key2", "key3"} + vals := [][]byte{[]byte("val1"), []byte("val2"), []byte("val3")} + + t.Run("storeLoad", func(t *testing.T) { + for i := 0; i < 9; i++ { + j := i + t.Run(fmt.Sprint(i), func(t *testing.T) { + t.Parallel() + kpl.testStoreLoad(ctx, t, keys[j%3], vals[j%3]) + }) + } + }) + + t.Run("delete", func(t *testing.T) { + for i := 0; i < 9; i++ { + j := i + t.Run(fmt.Sprint(i), func(t *testing.T) { + t.Parallel() + kpl.testDelete(ctx, t, keys[j%3]) + }) + } + }) +} + +// Store and test that subsequent Load succeeds. +func (kpl *KopiaPersisterLight) testStoreLoad(ctx context.Context, t *testing.T, key string, val []byte) { //nolint:thelper + err := kpl.Store(ctx, key, val) + assertNoError(t, err) + + valOut, err := kpl.Load(ctx, key) + assertNoError(t, err) + + if !bytes.Equal(valOut, val) { + t.Fatal("loaded value does not equal stored value", valOut, val) + } +} + +// Delete and test that subsequent Load fails. +func (kpl *KopiaPersisterLight) testDelete(ctx context.Context, t *testing.T, key string) { //nolint:thelper + kpl.Delete(ctx, key) + + _, err := kpl.Load(ctx, key) + log.Println("err:", err) + + if err == nil { + t.Fatal("snapshot was not deleted properly") + } +} + +func TestPersistence(t *testing.T) { + ctx := context.Background() + + repoPath, err := os.MkdirTemp("", "kopia-test-repo-") + assertNoError(t, err) + + kpl := initKPL(t, repoPath) + + // Persistence directory should be set. + if persistDir := kpl.GetPersistDir(); persistDir == "" { + t.Error("could not get persistence directory") + } + + // These are no-ops and should always succeed. + err = kpl.LoadMetadata() + assertNoError(t, err) + err = kpl.FlushMetadata() + assertNoError(t, err) + + // Store and cleanup kpl + err = kpl.Store(ctx, key, val) + assertNoError(t, err) + + kpl.Cleanup() + + // Re-initialize and Load + kpl = initKPL(t, repoPath) + valOut, err := kpl.Load(ctx, key) + assertNoError(t, err) + + if !bytes.Equal(valOut, val) { + t.Fatal("loaded value does not equal stored value") + } + + kpl.Cleanup() + os.RemoveAll(repoPath) +} + +func TestS3Connect(t *testing.T) { + repoPath, err := os.MkdirTemp("", "kopia-test-repo-") + assertNoError(t, err) + + // Test the S3 code path by attempting to connect to a nonexistent bucket. + os.Setenv(S3BucketNameEnvKey, "does-not-exist") + + kpl, err := NewPersisterLight("") + assertNoError(t, err) + + if err := kpl.ConnectOrCreateRepo(repoPath); err == nil { + t.Error("should not be able to connect to nonexistent S3 bucket") + } + + kpl.Cleanup() + os.RemoveAll(repoPath) +} + +func initKPL(t *testing.T, repoPath string) *KopiaPersisterLight { //nolint:thelper + os.Unsetenv(S3BucketNameEnvKey) + + kpl, err := NewPersisterLight("") + assertNoError(t, err) + + err = kpl.ConnectOrCreateRepo(repoPath) + assertNoError(t, err) + + return kpl +} + +func assertNoError(t *testing.T, err error) { + t.Helper() + + if err != nil { + t.Errorf("err: %v", err) + } +} diff --git a/tests/robustness/snapmeta/simple.go b/tests/robustness/snapmeta/simple.go index 5a2480cc5..c15097fef 100644 --- a/tests/robustness/snapmeta/simple.go +++ b/tests/robustness/snapmeta/simple.go @@ -3,6 +3,8 @@ package snapmeta import ( + "context" + "github.com/kopia/kopia/tests/robustness" ) @@ -24,7 +26,7 @@ func NewSimple() *Simple { } // Store implements the Storer interface Store method. -func (s *Simple) Store(key string, val []byte) error { +func (s *Simple) Store(ctx context.Context, key string, val []byte) error { buf := make([]byte, len(val)) _ = copy(buf, val) @@ -34,7 +36,7 @@ func (s *Simple) Store(key string, val []byte) error { } // Load implements the Storer interface Load method. -func (s *Simple) Load(key string) ([]byte, error) { +func (s *Simple) Load(ctx context.Context, key string) ([]byte, error) { if buf, found := s.Data[key]; found { retBuf := make([]byte, len(buf)) _ = copy(retBuf, buf) @@ -46,6 +48,8 @@ func (s *Simple) Load(key string) ([]byte, error) { } // Delete implements the Storer interface Delete method. -func (s *Simple) Delete(key string) { +func (s *Simple) Delete(ctx context.Context, key string) error { delete(s.Data, key) + + return nil } diff --git a/tests/robustness/snapmeta/simple_test.go b/tests/robustness/snapmeta/simple_test.go index ee27b7e54..97e1e81d7 100644 --- a/tests/robustness/snapmeta/simple_test.go +++ b/tests/robustness/snapmeta/simple_test.go @@ -4,6 +4,7 @@ package snapmeta import ( "bytes" + "context" "errors" "testing" @@ -11,9 +12,11 @@ import ( ) func TestSimpleBasic(t *testing.T) { + ctx := context.Background() + simple := NewSimple() - gotData, err := simple.Load("non-existent-key") + gotData, err := simple.Load(ctx, "non-existent-key") if !errors.Is(err, robustness.ErrKeyNotFound) { t.Fatalf("Did not get expected error: %q", err) } @@ -24,9 +27,9 @@ func TestSimpleBasic(t *testing.T) { storeKey := "key-to-store" data := []byte("some stored data") - simple.Store(storeKey, data) + simple.Store(ctx, storeKey, data) - gotData, err = simple.Load(storeKey) + gotData, err = simple.Load(ctx, storeKey) if err != nil { t.Fatalf("Error getting key: %v", err) } @@ -35,9 +38,9 @@ func TestSimpleBasic(t *testing.T) { t.Fatalf("Did not get the correct data") } - simple.Delete(storeKey) + simple.Delete(ctx, storeKey) - gotData, err = simple.Load(storeKey) + gotData, err = simple.Load(ctx, storeKey) if !errors.Is(err, robustness.ErrKeyNotFound) { t.Fatalf("Did not get expected error: %q", err) } diff --git a/tests/tools/kopiaclient/kopiaclient.go b/tests/tools/kopiaclient/kopiaclient.go new file mode 100644 index 000000000..b33075c48 --- /dev/null +++ b/tests/tools/kopiaclient/kopiaclient.go @@ -0,0 +1,248 @@ +// +build darwin,amd64 linux,amd64 + +// Package kopiaclient provides a client to interact with a Kopia repo. +package kopiaclient + +import ( + "bytes" + "context" + "io" + "log" + "os" + "path/filepath" + + "github.com/pkg/errors" + + "github.com/kopia/kopia/fs" + "github.com/kopia/kopia/fs/virtualfs" + "github.com/kopia/kopia/internal/units" + "github.com/kopia/kopia/repo" + "github.com/kopia/kopia/repo/blob" + "github.com/kopia/kopia/repo/blob/filesystem" + "github.com/kopia/kopia/repo/blob/s3" + "github.com/kopia/kopia/snapshot" + "github.com/kopia/kopia/snapshot/policy" + "github.com/kopia/kopia/snapshot/snapshotfs" + "github.com/kopia/kopia/tests/robustness" +) + +// KopiaClient uses a Kopia repo to create, restore, and delete snapshots. +type KopiaClient struct { + configPath string + pw string +} + +const ( + configFileName = "config" + password = "kj13498po&_EXAMPLE" //nolint:gosec + s3Endpoint = "s3.amazonaws.com" + awsAccessKeyIDEnvKey = "AWS_ACCESS_KEY_ID" + awsSecretAccessKeyEnvKey = "AWS_SECRET_ACCESS_KEY" //nolint:gosec + dataFileName = "data" +) + +// NewKopiaClient returns a new KopiaClient. +func NewKopiaClient(basePath string) *KopiaClient { + return &KopiaClient{ + configPath: filepath.Join(basePath, configFileName), + pw: password, + } +} + +// CreateOrConnectRepo creates a new Kopia repo or connects to an existing one if possible. +func (kc *KopiaClient) CreateOrConnectRepo(ctx context.Context, repoDir, bucketName string) error { + st, err := kc.getStorage(ctx, repoDir, bucketName) + if err != nil { + return err + } + + if iErr := repo.Initialize(ctx, st, &repo.NewRepositoryOptions{}, kc.pw); iErr != nil { + if !errors.Is(iErr, repo.ErrAlreadyInitialized) { + return errors.Wrap(iErr, "repo is already initialized") + } + + log.Println("connecting to existing repository") + } + + if iErr := repo.Connect(ctx, kc.configPath, st, kc.pw, &repo.ConnectOptions{}); iErr != nil { + return errors.Wrap(iErr, "error connecting to repository") + } + + return errors.Wrap(err, "unable to open repository") +} + +// SnapshotCreate creates a snapshot for the given path. +func (kc *KopiaClient) SnapshotCreate(ctx context.Context, key string, val []byte) error { + r, err := repo.Open(ctx, kc.configPath, kc.pw, &repo.Options{}) + if err != nil { + return errors.Wrap(err, "cannot open repository") + } + + ctx, rw, err := r.NewWriter(ctx, repo.WriteSessionOptions{}) + if err != nil { + return errors.Wrap(err, "cannot get new repository writer") + } + + si := kc.getSourceInfoFromKey(r, key) + + policyTree, err := policy.TreeForSource(ctx, r, si) + if err != nil { + return errors.Wrap(err, "cannot get policy tree for source") + } + + source := kc.getSourceForKeyVal(key, val) + u := snapshotfs.NewUploader(rw) + + man, err := u.Upload(ctx, source, policyTree, si) + if err != nil { + return errors.Wrap(err, "cannot get manifest") + } + + log.Printf("snapshotting %v", units.BytesStringBase10(man.Stats.TotalFileSize)) + + if _, err := snapshot.SaveSnapshot(ctx, rw, man); err != nil { + return errors.Wrap(err, "cannot save snapshot") + } + + if err := rw.Flush(ctx); err != nil { + return err + } + + return r.Close(ctx) +} + +// SnapshotRestore restores the latest snapshot for the given path. +func (kc *KopiaClient) SnapshotRestore(ctx context.Context, key string) ([]byte, error) { + r, err := repo.Open(ctx, kc.configPath, kc.pw, &repo.Options{}) + if err != nil { + return nil, errors.Wrap(err, "cannot open repository") + } + + mans, err := kc.getSnapshotsFromKey(ctx, r, key) + if err != nil { + return nil, errors.Wrap(err, "cannot get snapshots from key") + } + + man := kc.latestManifest(mans) + rootOIDWithPath := string(man.RootObjectID()) + "/" + dataFileName + + oid, err := snapshotfs.ParseObjectIDWithPath(ctx, r, rootOIDWithPath) + if err != nil { + return nil, errors.Wrapf(err, "cannot parse object ID %s", rootOIDWithPath) + } + + or, err := r.OpenObject(ctx, oid) + if err != nil { + return nil, errors.Wrapf(err, "cannot open object %s", string(oid)) + } + + val, err := io.ReadAll(or) + if err != nil { + return nil, err + } + + log.Printf("restored %v", units.BytesStringBase10(int64(len(val)))) + + if err := r.Close(ctx); err != nil { + return nil, err + } + + return val, nil +} + +// SnapshotDelete deletes all snapshots for a given path. +func (kc *KopiaClient) SnapshotDelete(ctx context.Context, key string) error { + r, err := repo.Open(ctx, kc.configPath, kc.pw, &repo.Options{}) + if err != nil { + return errors.Wrap(err, "cannot open repository") + } + + ctx, rw, err := r.NewWriter(ctx, repo.WriteSessionOptions{}) + if err != nil { + return errors.Wrap(err, "cannot get new repository writer") + } + + mans, err := kc.getSnapshotsFromKey(ctx, r, key) + if err != nil { + return errors.Wrap(err, "cannot get snapshots from key") + } + + for _, man := range mans { + if err := rw.DeleteManifest(ctx, man.ID); err != nil { + return errors.Wrap(err, "cannot delete manifest") + } + } + + if err := rw.Flush(ctx); err != nil { + return err + } + + return r.Close(ctx) +} + +func (kc *KopiaClient) getStorage(ctx context.Context, repoDir, bucketName string) (st blob.Storage, err error) { + if bucketName != "" { + s3Opts := &s3.Options{ + BucketName: bucketName, + Prefix: repoDir, + Endpoint: s3Endpoint, + AccessKeyID: os.Getenv(awsAccessKeyIDEnvKey), + SecretAccessKey: os.Getenv(awsSecretAccessKeyEnvKey), + } + st, err = s3.New(ctx, s3Opts) + } else { + if iErr := os.MkdirAll(repoDir, 0o700); iErr != nil { + return nil, errors.Wrap(iErr, "cannot create directory") + } + + fsOpts := &filesystem.Options{ + Path: repoDir, + } + st, err = filesystem.New(ctx, fsOpts) + } + + return st, errors.Wrap(err, "unable to get storage") +} + +// getSourceForKeyVal creates a virtual directory for `key` that contains a single virtual file that +// reads its contents from `val`. +func (kc *KopiaClient) getSourceForKeyVal(key string, val []byte) fs.Entry { + return virtualfs.NewStaticDirectory(key, fs.Entries{ + virtualfs.StreamingFileFromReader(dataFileName, bytes.NewReader(val)), + }) +} + +func (kc *KopiaClient) getSnapshotsFromKey(ctx context.Context, r repo.Repository, key string) ([]*snapshot.Manifest, error) { + si := kc.getSourceInfoFromKey(r, key) + + manifests, err := snapshot.ListSnapshots(ctx, r, si) + if err != nil { + return nil, errors.Wrap(err, "cannot list snapshots") + } + + if len(manifests) == 0 { + return nil, robustness.ErrKeyNotFound + } + + return manifests, nil +} + +func (kc *KopiaClient) getSourceInfoFromKey(r repo.Repository, key string) snapshot.SourceInfo { + return snapshot.SourceInfo{ + Host: r.ClientOptions().Hostname, + UserName: r.ClientOptions().Username, + Path: key, + } +} + +func (kc *KopiaClient) latestManifest(mans []*snapshot.Manifest) *snapshot.Manifest { + latest := mans[0] + + for _, m := range mans { + if m.StartTime.After(latest.StartTime) { + latest = m + } + } + + return latest +}