From 5c88bcf1a6e77de010e90f9ee36c84bbbf959024 Mon Sep 17 00:00:00 2001 From: ashmrtn <3891298+ashmrtn@users.noreply.github.com> Date: Mon, 22 Aug 2022 12:42:27 -0700 Subject: [PATCH] feat(snapshots): Callback for when uploader finishes processing a file (#2331) * Make callback for upload file completion Callback does not indicate that a file will be reachable immediately in the resulting snapshot, but does indicate that the uploader is done processing the file in some way (either via uploading data or finding a previous version in the repo) and whether there was an error processing the file. * Tests for new FinishedFile callback Ensure hadErr is properly populated and FinishedFile is called even if the file was considered cached. * Refine comment on interface function slightly * Give callback error instead of bool about error * Add locks around concurrent accesses in test --- internal/mockfs/mockfs.go | 19 ++++++ internal/server/source_manager.go | 6 ++ snapshot/snapshotfs/upload.go | 17 ++++- snapshot/snapshotfs/upload_progress.go | 12 ++++ snapshot/snapshotfs/upload_test.go | 86 ++++++++++++++++++++++++++ 5 files changed, 137 insertions(+), 3 deletions(-) diff --git a/internal/mockfs/mockfs.go b/internal/mockfs/mockfs.go index a181d920a..1d660bda9 100644 --- a/internal/mockfs/mockfs.go +++ b/internal/mockfs/mockfs.go @@ -115,6 +115,25 @@ func (imd *Directory) AddFile(name string, content []byte, permissions os.FileMo return file } +// AddFileWithSource adds a mock file with the specified name, permissions, and +// given source function for getting a Reader instance. +func (imd *Directory) AddFileWithSource(name string, permissions os.FileMode, source func() (ReaderSeekerCloser, error)) *File { + imd, name = imd.resolveSubdir(name) + file := &File{ + entry: entry{ + name: name, + mode: permissions, + size: 0, + modTime: DefaultModTime, + }, + source: source, + } + + imd.addChild(file) + + return file +} + // AddSymlink adds a mock symlink with the specified name, target and permissions. func (imd *Directory) AddSymlink(name, target string, permissions os.FileMode) *Symlink { imd, name = imd.resolveSubdir(name) diff --git a/internal/server/source_manager.go b/internal/server/source_manager.go index 98dbb4703..8420ecc44 100644 --- a/internal/server/source_manager.go +++ b/internal/server/source_manager.go @@ -543,6 +543,12 @@ func (t *uitaskProgress) FinishedHashingFile(fname string, numBytes int64) { t.maybeReport() } +// FinishedFile is emitted when the system is done examining a file. +func (t *uitaskProgress) FinishedFile(fname string, err error) { + t.p.FinishedFile(fname, err) + t.maybeReport() +} + // HashedBytes is emitted while hashing any blocks of bytes. func (t *uitaskProgress) HashedBytes(numBytes int64) { t.p.HashedBytes(numBytes) diff --git a/snapshot/snapshotfs/upload.go b/snapshot/snapshotfs/upload.go index d64b0ebff..d281c14c1 100644 --- a/snapshot/snapshotfs/upload.go +++ b/snapshot/snapshotfs/upload.go @@ -136,8 +136,12 @@ func (u *Uploader) incompleteReason() string { return "" } -func (u *Uploader) uploadFileInternal(ctx context.Context, parentCheckpointRegistry *checkpointRegistry, relativePath string, f fs.File, pol *policy.Policy) (*snapshot.DirEntry, error) { +func (u *Uploader) uploadFileInternal(ctx context.Context, parentCheckpointRegistry *checkpointRegistry, relativePath string, f fs.File, pol *policy.Policy) (dirEntry *snapshot.DirEntry, ret error) { u.Progress.HashingFile(relativePath) + + defer func() { + u.Progress.FinishedFile(relativePath, ret) + }() defer u.Progress.FinishedHashingFile(relativePath, f.Size()) if pf, ok := f.(snapshot.HasDirEntryOrNil); ok { @@ -294,8 +298,12 @@ func (u *Uploader) uploadFileData(ctx context.Context, parentCheckpointRegistry return de, nil } -func (u *Uploader) uploadSymlinkInternal(ctx context.Context, relativePath string, f fs.Symlink) (*snapshot.DirEntry, error) { +func (u *Uploader) uploadSymlinkInternal(ctx context.Context, relativePath string, f fs.Symlink) (dirEntry *snapshot.DirEntry, ret error) { u.Progress.HashingFile(relativePath) + + defer func() { + u.Progress.FinishedFile(relativePath, ret) + }() defer u.Progress.FinishedHashingFile(relativePath, f.Size()) target, err := f.Readlink(ctx) @@ -328,7 +336,7 @@ func (u *Uploader) uploadSymlinkInternal(ctx context.Context, relativePath strin return de, nil } -func (u *Uploader) uploadStreamingFileInternal(ctx context.Context, relativePath string, f fs.StreamingFile) (*snapshot.DirEntry, error) { +func (u *Uploader) uploadStreamingFileInternal(ctx context.Context, relativePath string, f fs.StreamingFile) (dirEntry *snapshot.DirEntry, ret error) { reader, err := f.GetReader(ctx) if err != nil { return nil, errors.Wrap(err, "unable to get streaming file reader") @@ -340,6 +348,7 @@ func (u *Uploader) uploadStreamingFileInternal(ctx context.Context, relativePath defer func() { u.Progress.FinishedHashingFile(relativePath, streamSize) + u.Progress.FinishedFile(relativePath, ret) }() writer := u.repo.NewObjectWriter(ctx, object.WriterOptions{ @@ -805,6 +814,8 @@ func (u *Uploader) processSingle( // compute entryResult now, cachedEntry is short-lived cachedDirEntry, err := newDirEntry(entry, entry.Name(), cachedEntry.(object.HasObjectID).ObjectID()) + u.Progress.FinishedFile(entryRelativePath, err) + if err != nil { return errors.Wrap(err, "unable to create dir entry") } diff --git a/snapshot/snapshotfs/upload_progress.go b/snapshot/snapshotfs/upload_progress.go index c07d4f70c..4f078460e 100644 --- a/snapshot/snapshotfs/upload_progress.go +++ b/snapshot/snapshotfs/upload_progress.go @@ -30,6 +30,12 @@ type UploadProgress interface { // FinishedHashingFile is emitted at the end of hashing of a given file. FinishedHashingFile(fname string, numBytes int64) + // FinishedFile is emitted when the uploader is done with a file, regardless of if it was hashed + // or cached. If an error was encountered it reports that too. A call to FinishedFile gives no + // information about the reachability of the file in checkpoints that may occur close to the + // time this function is called. + FinishedFile(fname string, err error) + // HashedBytes is emitted while hashing any blocks of bytes. HashedBytes(numBytes int64) @@ -82,6 +88,9 @@ func (p *NullUploadProgress) HashingFile(fname string) {} // FinishedHashingFile implements UploadProgress. func (p *NullUploadProgress) FinishedHashingFile(fname string, numBytes int64) {} +// FinishedFile implements UploadProgress. +func (p *NullUploadProgress) FinishedFile(fname string, err error) {} + // StartedDirectory implements UploadProgress. func (p *NullUploadProgress) StartedDirectory(dirname string) {} @@ -170,6 +179,9 @@ func (p *CountingUploadProgress) FinishedHashingFile(fname string, numBytes int6 atomic.AddInt32(&p.counters.TotalHashedFiles, 1) } +// FinishedFile implements UploadProgress. +func (p *CountingUploadProgress) FinishedFile(fname string, err error) {} + // ExcludedDir implements UploadProgress. func (p *CountingUploadProgress) ExcludedDir(dirname string) { atomic.AddInt32(&p.counters.TotalExcludedDirs, 1) diff --git a/snapshot/snapshotfs/upload_test.go b/snapshot/snapshotfs/upload_test.go index 78f698cb9..28bc4eb38 100644 --- a/snapshot/snapshotfs/upload_test.go +++ b/snapshot/snapshotfs/upload_test.go @@ -13,6 +13,7 @@ "runtime/debug" "sort" "strings" + "sync" "sync/atomic" "testing" "time" @@ -523,6 +524,91 @@ func TestUpload_SubDirectoryReadFailureSomeIgnoredNoFailFast(t *testing.T) { ) } +type mockProgress struct { + UploadProgress + finishedFileCheck func(string, error) +} + +func (mp *mockProgress) FinishedFile(relativePath string, err error) { + defer mp.UploadProgress.FinishedFile(relativePath, err) + + mp.finishedFileCheck(relativePath, err) +} + +func TestUpload_FinishedFileProgress(t *testing.T) { + ctx := testlogging.Context(t) + th := newUploadTestHarness(ctx, t) + mu := sync.Mutex{} + filesFinished := 0 + + defer th.cleanup() + + t.Logf("checking FinishedFile callbacks") + + root := mockfs.NewDirectory() + root.AddFile("f1", []byte{'1', '2', '3'}, 0o777) + root.AddFileWithSource("f2", 0o777, func() (mockfs.ReaderSeekerCloser, error) { + return nil, assert.AnError + }) + + u := NewUploader(th.repo) + u.ForceHashPercentage = 0 + u.Progress = &mockProgress{ + UploadProgress: u.Progress, + finishedFileCheck: func(relativePath string, err error) { + defer func() { + mu.Lock() + defer mu.Unlock() + + filesFinished++ + }() + + assert.Contains(t, []string{"f1", "f2"}, filepath.Base(relativePath)) + + if strings.Contains(relativePath, "f2") { + assert.Error(t, err) + return + } + + assert.NoError(t, err) + }, + } + + trueValue := policy.OptionalBool(true) + policyTree := policy.BuildTree(map[string]*policy.Policy{ + ".": { + ErrorHandlingPolicy: policy.ErrorHandlingPolicy{ + IgnoreFileErrors: &trueValue, + IgnoreDirectoryErrors: &trueValue, + }, + }, + }, policy.DefaultPolicy) + + man, err := u.Upload(ctx, root, policyTree, snapshot.SourceInfo{}) + require.NoError(t, err) + + assert.Equal(t, int32(0), atomic.LoadInt32(&man.Stats.ErrorCount), "ErrorCount") + assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.IgnoredErrorCount), "IgnoredErrorCount") + assert.Equal(t, int32(0), atomic.LoadInt32(&man.Stats.CachedFiles), "CachedFiles") + assert.Equal(t, int32(2), atomic.LoadInt32(&man.Stats.NonCachedFiles), "NonCachedFiles") + assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.TotalDirectoryCount), "TotalDirectoryCount") + assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.TotalFileCount), "TotalFileCount") + assert.Equal(t, 2, filesFinished, "FinishedFile calls") + + // Upload a second time to check for cached files. + filesFinished = 0 + man, err = u.Upload(ctx, root, policyTree, snapshot.SourceInfo{}, man) + require.NoError(t, err) + + assert.Equal(t, int32(0), atomic.LoadInt32(&man.Stats.ErrorCount), "ErrorCount") + assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.IgnoredErrorCount), "IgnoredErrorCount") + assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.CachedFiles), "CachedFiles") + assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.NonCachedFiles), "NonCachedFiles") + assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.TotalDirectoryCount), "TotalDirectoryCount") + assert.Equal(t, int32(0), atomic.LoadInt32(&man.Stats.TotalFileCount), "TotalFileCount") + assert.Equal(t, 2, filesFinished, "FinishedFile calls") +} + func TestUploadWithCheckpointing(t *testing.T) { ctx := testlogging.Context(t) th := newUploadTestHarness(ctx, t)