From a41cb1190250a531f1948d37755a37013055b83f Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Tue, 11 Aug 2026 12:56:23 +0800 Subject: [PATCH 1/7] add prefetch options for repo interface Signed-off-by: Lyndon-Li --- pkg/repository/udmrepo/kopialib/lib_repo.go | 4 ++-- pkg/repository/udmrepo/repo.go | 7 ++++++- pkg/uploader/block/uploader.go | 2 +- pkg/uploader/kopia/shim.go | 2 +- 4 files changed, 10 insertions(+), 5 deletions(-) diff --git a/pkg/repository/udmrepo/kopialib/lib_repo.go b/pkg/repository/udmrepo/kopialib/lib_repo.go index c7bb65a43..b26edb76f 100644 --- a/pkg/repository/udmrepo/kopialib/lib_repo.go +++ b/pkg/repository/udmrepo/kopialib/lib_repo.go @@ -338,7 +338,7 @@ func (km *kopiaMaintenance) maintainProgress(uploaded int64) { } } -func (kr *kopiaRepository) OpenObject(ctx context.Context, id udmrepo.ID) (udmrepo.ObjectReader, error) { +func (kr *kopiaRepository) OpenObject(ctx context.Context, id udmrepo.ID, opt udmrepo.ObjectReadOptions) (udmrepo.ObjectReader, error) { if kr.rawRepo == nil { return nil, errors.New("repo is closed or not open") } @@ -550,7 +550,7 @@ func (kr *kopiaRepository) WriteMetadata(ctx context.Context, meta *udmrepo.Meta } func (kr *kopiaRepository) ReadMetadata(ctx context.Context, id udmrepo.ID) (*udmrepo.Metadata, error) { - reader, err := kr.OpenObject(ctx, id) + reader, err := kr.OpenObject(ctx, id, udmrepo.ObjectReadOptions{}) if err != nil { return nil, errors.Wrapf(err, "error to open metadata object %v", id) } diff --git a/pkg/repository/udmrepo/repo.go b/pkg/repository/udmrepo/repo.go index 76cdc5f1d..5873db743 100644 --- a/pkg/repository/udmrepo/repo.go +++ b/pkg/repository/udmrepo/repo.go @@ -72,6 +72,11 @@ type ObjectWriteOptions struct { ParentObject ID // The object in the previous snapshot, for incremental backup } +type ObjectReadOptions struct { + Prefetch bool + PrefetchBudgetMB int +} + type AdvancedFeatureInfo struct { MultiPartBackup bool // if set to true, it means the repo supports multiple-part backup } @@ -136,7 +141,7 @@ type BackupRepoService interface { type BackupRepo interface { // OpenObject opens an existing object for read. // id: the object's unified identifier. - OpenObject(ctx context.Context, id ID) (ObjectReader, error) + OpenObject(ctx context.Context, id ID, opt ObjectReadOptions) (ObjectReader, error) // GetManifest gets a manifest data from the backup repository. GetManifest(ctx context.Context, id ID, mani *RepoManifest) error diff --git a/pkg/uploader/block/uploader.go b/pkg/uploader/block/uploader.go index 824c0ae9f..4d8d86e84 100644 --- a/pkg/uploader/block/uploader.go +++ b/pkg/uploader/block/uploader.go @@ -176,7 +176,7 @@ func (blkup *blockUploader) Restore(snapshot udmrepo.Snapshot, dest destInfo, bi return 0, errors.Errorf("dest dev(%s) size is too small (%v vs. %v)", dest.path, dest.size, sourceSize) } - reader, err := blkup.repoWriter.OpenObject(blkup.ctx, meta.SubObjects[0].ID) + reader, err := blkup.repoWriter.OpenObject(blkup.ctx, meta.SubObjects[0].ID, udmrepo.ObjectReadOptions{}) if err != nil { return 0, errors.Wrapf(err, "error opening bdev object %v", meta.SubObjects[0].Name) } diff --git a/pkg/uploader/kopia/shim.go b/pkg/uploader/kopia/shim.go index 4a3908185..465b76c04 100644 --- a/pkg/uploader/kopia/shim.go +++ b/pkg/uploader/kopia/shim.go @@ -56,7 +56,7 @@ func NewShimRepo(repo udmrepo.BackupRepo) repo.RepositoryWriter { // OpenObject open specific object func (sr *shimRepository) OpenObject(ctx context.Context, id object.ID) (object.Reader, error) { - reader, err := sr.udmRepo.OpenObject(ctx, udmrepo.ID(id.String())) + reader, err := sr.udmRepo.OpenObject(ctx, udmrepo.ID(id.String()), udmrepo.ObjectReadOptions{}) if err != nil { return nil, errors.Wrapf(err, "failed to open object with id %v", id) } From b74824f9c0b7ad39260dbb5dda382490bd276746 Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Tue, 11 Aug 2026 13:00:44 +0800 Subject: [PATCH 2/7] extend object reader for prefetch Signed-off-by: Lyndon-Li --- pkg/repository/udmrepo/kopialib/lib_repo.go | 138 +++++++++++++++++++- 1 file changed, 134 insertions(+), 4 deletions(-) diff --git a/pkg/repository/udmrepo/kopialib/lib_repo.go b/pkg/repository/udmrepo/kopialib/lib_repo.go index b26edb76f..0efde34a7 100644 --- a/pkg/repository/udmrepo/kopialib/lib_repo.go +++ b/pkg/repository/udmrepo/kopialib/lib_repo.go @@ -73,8 +73,22 @@ type logThrottle struct { interval time.Duration } +type objectPrefetch struct { + ctx context.Context + cancel context.CancelFunc + entries []object.IndirectObjectEntry + curOffset int64 + cond *sync.Cond + mu sync.Mutex + nextEntry int + budget int64 +} + type kopiaObjectReader struct { rawReader object.Reader + rawRepo repo.Repository + prefetch *objectPrefetch + logger logrus.FieldLogger } type kopiaObjectWriter struct { @@ -353,9 +367,43 @@ func (kr *kopiaRepository) OpenObject(ctx context.Context, id udmrepo.ID, opt ud return nil, errors.Wrap(err, "error to open object") } - return &kopiaObjectReader{ + var prefetch *objectPrefetch + if opt.Prefetch { + if e, err := kr.getFlattenedEntries(ctx, objID); err != nil { + kr.logger.WithError(err).Warnf("Failed to load entries for object %v, skip prefetch", id) + } else { + pCtx, pCancel := context.WithCancel(ctx) + prefetch = &objectPrefetch{ + ctx: pCtx, + cancel: pCancel, + budget: int64(opt.PrefetchBudgetMB) << 20, + entries: e, + } + + prefetch.cond = sync.NewCond(&prefetch.mu) + } + + } + + rd := &kopiaObjectReader{ rawReader: reader, - }, nil + rawRepo: kr.rawRepo, + prefetch: prefetch, + logger: kr.logger, + } + + if rd.prefetch != nil { + go rd.prefetchProc() + + go func() { + <-rd.prefetch.ctx.Done() + prefetch.mu.Lock() + prefetch.cond.Broadcast() + prefetch.mu.Unlock() + }() + } + + return rd, nil } func (kr *kopiaRepository) GetManifest(ctx context.Context, id udmrepo.ID, mani *udmrepo.RepoManifest) error { @@ -792,7 +840,16 @@ func (kor *kopiaObjectReader) Read(p []byte) (int, error) { return 0, errors.New("object reader is closed or not open") } - return kor.rawReader.Read(p) + n, err := kor.rawReader.Read(p) + if n > 0 { + if kor.prefetch != nil { + kor.prefetch.mu.Lock() + kor.prefetch.curOffset += int64(n) + kor.prefetch.cond.Signal() + kor.prefetch.mu.Unlock() + } + } + return n, err } func (kor *kopiaObjectReader) Seek(offset int64, whence int) (int64, error) { @@ -800,10 +857,83 @@ func (kor *kopiaObjectReader) Seek(offset int64, whence int) (int64, error) { return -1, errors.New("object reader is closed or not open") } - return kor.rawReader.Seek(offset, whence) + off, err := kor.rawReader.Seek(offset, whence) + if err == nil { + if kor.prefetch != nil { + kor.prefetch.mu.Lock() + kor.prefetch.curOffset = off + kor.prefetch.cond.Signal() + kor.prefetch.mu.Unlock() + } + } + + return off, err +} + +func (kor *kopiaObjectReader) prefetchProc() { + prefetch := kor.prefetch + if prefetch == nil { + return + } + + for { + prefetch.mu.Lock() + + select { + case <-prefetch.ctx.Done(): + prefetch.mu.Unlock() + return + default: + } + + curOffset := prefetch.curOffset + + for prefetch.nextEntry < len(prefetch.entries) { + entry := prefetch.entries[prefetch.nextEntry] + if entry.Start+entry.Length <= curOffset { + prefetch.nextEntry++ + } else { + break + } + } + + if prefetch.nextEntry >= len(prefetch.entries) { + prefetch.mu.Unlock() + return + } + + var toFetch []object.ID + for prefetch.nextEntry < len(prefetch.entries) { + entry := prefetch.entries[prefetch.nextEntry] + + if entry.Start > curOffset+prefetch.budget { + break + } + + toFetch = append(toFetch, entry.Object) + prefetch.nextEntry++ + } + + if len(toFetch) == 0 { + prefetch.cond.Wait() + prefetch.mu.Unlock() + continue + } + + prefetch.mu.Unlock() + + _, err := kor.rawRepo.PrefetchObjects(prefetch.ctx, toFetch, "") + if err != nil && err != context.Canceled { + kor.logger.WithError(err).Warnf("Failed to prefetch contents for offset %v", curOffset) + } + } } func (kor *kopiaObjectReader) Close() error { + if kor.prefetch != nil && kor.prefetch.cancel != nil { + kor.prefetch.cancel() + } + if kor.rawReader == nil { return nil } From de87def9d902df85d48c6add91c278bb754698ea Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Tue, 11 Aug 2026 13:02:11 +0800 Subject: [PATCH 3/7] enable object reader prefetch for block uploader Signed-off-by: Lyndon-Li --- pkg/uploader/block/uploader.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pkg/uploader/block/uploader.go b/pkg/uploader/block/uploader.go index 4d8d86e84..275fd18ed 100644 --- a/pkg/uploader/block/uploader.go +++ b/pkg/uploader/block/uploader.go @@ -176,7 +176,10 @@ func (blkup *blockUploader) Restore(snapshot udmrepo.Snapshot, dest destInfo, bi return 0, errors.Errorf("dest dev(%s) size is too small (%v vs. %v)", dest.path, dest.size, sourceSize) } - reader, err := blkup.repoWriter.OpenObject(blkup.ctx, meta.SubObjects[0].ID, udmrepo.ObjectReadOptions{}) + reader, err := blkup.repoWriter.OpenObject(blkup.ctx, meta.SubObjects[0].ID, udmrepo.ObjectReadOptions{ + Prefetch: true, + PrefetchBudgetMB: 256, + }) if err != nil { return 0, errors.Wrapf(err, "error opening bdev object %v", meta.SubObjects[0].Name) } From c27343fc4e256647310d754e3c52e01afb2d45a3 Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Tue, 11 Aug 2026 13:07:27 +0800 Subject: [PATCH 4/7] fix UT errors Signed-off-by: Lyndon-Li --- .../udmrepo/kopialib/lib_repo_test.go | 2 +- pkg/repository/udmrepo/mocks/BackupRepo.go | 30 +++++++++++-------- 2 files changed, 19 insertions(+), 13 deletions(-) diff --git a/pkg/repository/udmrepo/kopialib/lib_repo_test.go b/pkg/repository/udmrepo/kopialib/lib_repo_test.go index 370b82b9e..0cc52261a 100644 --- a/pkg/repository/udmrepo/kopialib/lib_repo_test.go +++ b/pkg/repository/udmrepo/kopialib/lib_repo_test.go @@ -318,7 +318,7 @@ func TestOpenObject(t *testing.T) { kr.rawRepo = tc.rawRepo } - _, err := kr.OpenObject(t.Context(), udmrepo.ID(tc.objectID)) + _, err := kr.OpenObject(t.Context(), udmrepo.ID(tc.objectID), udmrepo.ObjectReadOptions{}) if tc.expectedErr == "" { assert.NoError(t, err) diff --git a/pkg/repository/udmrepo/mocks/BackupRepo.go b/pkg/repository/udmrepo/mocks/BackupRepo.go index 623c4d70d..3206422b4 100644 --- a/pkg/repository/udmrepo/mocks/BackupRepo.go +++ b/pkg/repository/udmrepo/mocks/BackupRepo.go @@ -699,8 +699,8 @@ func (_c *BackupRepo_NewObjectWriter_Call) RunAndReturn(run func(ctx context.Con } // OpenObject provides a mock function for the type BackupRepo -func (_mock *BackupRepo) OpenObject(ctx context.Context, id udmrepo.ID) (udmrepo.ObjectReader, error) { - ret := _mock.Called(ctx, id) +func (_mock *BackupRepo) OpenObject(ctx context.Context, id udmrepo.ID, opt udmrepo.ObjectReadOptions) (udmrepo.ObjectReader, error) { + ret := _mock.Called(ctx, id, opt) if len(ret) == 0 { panic("no return value specified for OpenObject") @@ -708,18 +708,18 @@ func (_mock *BackupRepo) OpenObject(ctx context.Context, id udmrepo.ID) (udmrepo var r0 udmrepo.ObjectReader var r1 error - if returnFunc, ok := ret.Get(0).(func(context.Context, udmrepo.ID) (udmrepo.ObjectReader, error)); ok { - return returnFunc(ctx, id) + if returnFunc, ok := ret.Get(0).(func(context.Context, udmrepo.ID, udmrepo.ObjectReadOptions) (udmrepo.ObjectReader, error)); ok { + return returnFunc(ctx, id, opt) } - if returnFunc, ok := ret.Get(0).(func(context.Context, udmrepo.ID) udmrepo.ObjectReader); ok { - r0 = returnFunc(ctx, id) + if returnFunc, ok := ret.Get(0).(func(context.Context, udmrepo.ID, udmrepo.ObjectReadOptions) udmrepo.ObjectReader); ok { + r0 = returnFunc(ctx, id, opt) } else { if ret.Get(0) != nil { r0 = ret.Get(0).(udmrepo.ObjectReader) } } - if returnFunc, ok := ret.Get(1).(func(context.Context, udmrepo.ID) error); ok { - r1 = returnFunc(ctx, id) + if returnFunc, ok := ret.Get(1).(func(context.Context, udmrepo.ID, udmrepo.ObjectReadOptions) error); ok { + r1 = returnFunc(ctx, id, opt) } else { r1 = ret.Error(1) } @@ -734,11 +734,12 @@ type BackupRepo_OpenObject_Call struct { // OpenObject is a helper method to define mock.On call // - ctx context.Context // - id udmrepo.ID -func (_e *BackupRepo_Expecter) OpenObject(ctx interface{}, id interface{}) *BackupRepo_OpenObject_Call { - return &BackupRepo_OpenObject_Call{Call: _e.mock.On("OpenObject", ctx, id)} +// - opt udmrepo.ObjectReadOptions +func (_e *BackupRepo_Expecter) OpenObject(ctx interface{}, id interface{}, opt interface{}) *BackupRepo_OpenObject_Call { + return &BackupRepo_OpenObject_Call{Call: _e.mock.On("OpenObject", ctx, id, opt)} } -func (_c *BackupRepo_OpenObject_Call) Run(run func(ctx context.Context, id udmrepo.ID)) *BackupRepo_OpenObject_Call { +func (_c *BackupRepo_OpenObject_Call) Run(run func(ctx context.Context, id udmrepo.ID, opt udmrepo.ObjectReadOptions)) *BackupRepo_OpenObject_Call { _c.Call.Run(func(args mock.Arguments) { var arg0 context.Context if args[0] != nil { @@ -748,9 +749,14 @@ func (_c *BackupRepo_OpenObject_Call) Run(run func(ctx context.Context, id udmre if args[1] != nil { arg1 = args[1].(udmrepo.ID) } + var arg2 udmrepo.ObjectReadOptions + if args[2] != nil { + arg2 = args[2].(udmrepo.ObjectReadOptions) + } run( arg0, arg1, + arg2, ) }) return _c @@ -761,7 +767,7 @@ func (_c *BackupRepo_OpenObject_Call) Return(objectReader udmrepo.ObjectReader, return _c } -func (_c *BackupRepo_OpenObject_Call) RunAndReturn(run func(ctx context.Context, id udmrepo.ID) (udmrepo.ObjectReader, error)) *BackupRepo_OpenObject_Call { +func (_c *BackupRepo_OpenObject_Call) RunAndReturn(run func(ctx context.Context, id udmrepo.ID, opt udmrepo.ObjectReadOptions) (udmrepo.ObjectReader, error)) *BackupRepo_OpenObject_Call { _c.Call.Return(run) return _c } From 92bcf5a3b33dc27649e12e81eb755c4520a7e33d Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Tue, 11 Aug 2026 13:36:26 +0800 Subject: [PATCH 5/7] add UT for prefetch Signed-off-by: Lyndon-Li --- .../udmrepo/kopialib/lib_repo_test.go | 216 +++++++++++++++++- 1 file changed, 212 insertions(+), 4 deletions(-) diff --git a/pkg/repository/udmrepo/kopialib/lib_repo_test.go b/pkg/repository/udmrepo/kopialib/lib_repo_test.go index 0cc52261a..6bd64009c 100644 --- a/pkg/repository/udmrepo/kopialib/lib_repo_test.go +++ b/pkg/repository/udmrepo/kopialib/lib_repo_test.go @@ -22,12 +22,14 @@ import ( "encoding/json" "math" "os" + "sync" "testing" "time" "github.com/cockroachdb/errors" "github.com/kopia/kopia/fs" "github.com/kopia/kopia/repo" + "github.com/kopia/kopia/repo/content" "github.com/kopia/kopia/repo/manifest" "github.com/kopia/kopia/repo/object" "github.com/kopia/kopia/snapshot" @@ -285,6 +287,7 @@ func TestOpenObject(t *testing.T) { name string rawRepo *repomocks.MockRepository objectID string + opt udmrepo.ObjectReadOptions retErr error expectedErr string }{ @@ -304,21 +307,38 @@ func TestOpenObject(t *testing.T) { retErr: errors.New("fake-open-error"), expectedErr: "error to open object: fake-open-error", }, + { + name: "raw open success, without prefetch", + rawRepo: repomocks.NewMockRepository(t), + objectID: "D0123456789abcdef0123456789abcdef", + }, + { + name: "raw open success, with prefetch", + rawRepo: repomocks.NewMockRepository(t), + objectID: "D0123456789abcdef0123456789abcdef", + opt: udmrepo.ObjectReadOptions{Prefetch: true, PrefetchBudgetMB: 10}, + }, } for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - kr := &kopiaRepository{} + kr := &kopiaRepository{ + logger: velerotest.NewLogger(), + } if tc.rawRepo != nil { - if tc.retErr != nil { - tc.rawRepo.On("OpenObject", mock.Anything, mock.Anything).Return(nil, tc.retErr) + if tc.name != "objectID is invalid" { + if tc.retErr != nil { + tc.rawRepo.On("OpenObject", mock.Anything, mock.Anything).Return(nil, tc.retErr) + } else { + tc.rawRepo.On("OpenObject", mock.Anything, mock.Anything).Return(nil, nil) + } } kr.rawRepo = tc.rawRepo } - _, err := kr.OpenObject(t.Context(), udmrepo.ID(tc.objectID), udmrepo.ObjectReadOptions{}) + _, err := kr.OpenObject(t.Context(), udmrepo.ID(tc.objectID), tc.opt) if tc.expectedErr == "" { assert.NoError(t, err) @@ -845,6 +865,7 @@ func TestReaderClose(t *testing.T) { name string rawObjReader *repomocks.Reader rawReaderRetErr error + withPrefetch bool expectedErr string }{ { @@ -860,6 +881,11 @@ func TestReaderClose(t *testing.T) { name: "succeed", rawObjReader: repomocks.NewReader(t), }, + { + name: "succeed with prefetch", + rawObjReader: repomocks.NewReader(t), + withPrefetch: true, + }, } for _, tc := range testCases { @@ -871,8 +897,20 @@ func TestReaderClose(t *testing.T) { kr.rawReader = tc.rawObjReader } + if tc.withPrefetch { + ctx, cancel := context.WithCancel(t.Context()) + kr.prefetch = &objectPrefetch{ + ctx: ctx, + cancel: cancel, + } + } + err := kr.Close() + if tc.withPrefetch { + assert.ErrorIs(t, kr.prefetch.ctx.Err(), context.Canceled) + } + if tc.expectedErr == "" { assert.NoError(t, err) } else { @@ -1832,3 +1870,173 @@ func TestListSnapshot(t *testing.T) { }) } } + +func mustParseID(s string) object.ID { + id, _ := object.ParseID(s) + return id +} + +func TestPrefetchProc(t *testing.T) { + testCases := []struct { + name string + setupPrefetch func(ctx context.Context, cancel context.CancelFunc) *objectPrefetch + mockRepo func(mockRepo *repomocks.MockRepository) + runConcurrently bool + trigger func(prefetch *objectPrefetch) + }{ + { + name: "nil prefetch", + setupPrefetch: func(ctx context.Context, cancel context.CancelFunc) *objectPrefetch { + return nil + }, + }, + { + name: "context canceled", + setupPrefetch: func(ctx context.Context, cancel context.CancelFunc) *objectPrefetch { + cancel() + p := &objectPrefetch{ + ctx: ctx, + } + p.cond = sync.NewCond(&p.mu) + return p + }, + }, + { + name: "fetch all entries and exit", + setupPrefetch: func(ctx context.Context, cancel context.CancelFunc) *objectPrefetch { + p := &objectPrefetch{ + ctx: ctx, + entries: []object.IndirectObjectEntry{ + {Start: 0, Length: 100, Object: mustParseID("D0123456789abcdef0123456789abcdef")}, + {Start: 100, Length: 100, Object: mustParseID("D0123456789abcdef0123456789abcdeg")}, + }, + budget: 200, + curOffset: 0, + } + p.cond = sync.NewCond(&p.mu) + return p + }, + mockRepo: func(mockRepo *repomocks.MockRepository) { + mockRepo.On("PrefetchObjects", mock.Anything, []object.ID{mustParseID("D0123456789abcdef0123456789abcdef"), mustParseID("D0123456789abcdef0123456789abcdeg")}, "").Return(([]content.ID)(nil), nil).Once() + }, + }, + { + name: "fetch partial, wait, and fetch rest", + runConcurrently: true, + setupPrefetch: func(ctx context.Context, cancel context.CancelFunc) *objectPrefetch { + p := &objectPrefetch{ + ctx: ctx, + entries: []object.IndirectObjectEntry{ + {Start: 0, Length: 100, Object: mustParseID("D0123456789abcdef0123456789abcdef")}, + {Start: 100, Length: 100, Object: mustParseID("D0123456789abcdef0123456789abcdeg")}, + }, + budget: 50, + curOffset: 0, + } + p.cond = sync.NewCond(&p.mu) + return p + }, + mockRepo: func(mockRepo *repomocks.MockRepository) { + mockRepo.On("PrefetchObjects", mock.Anything, []object.ID{mustParseID("D0123456789abcdef0123456789abcdef")}, "").Return(([]content.ID)(nil), nil).Once() + mockRepo.On("PrefetchObjects", mock.Anything, []object.ID{mustParseID("D0123456789abcdef0123456789abcdeg")}, "").Return(([]content.ID)(nil), nil).Once() + }, + trigger: func(prefetch *objectPrefetch) { + // Wait a bit for the first fetch and wait to happen + time.Sleep(50 * time.Millisecond) + prefetch.mu.Lock() + prefetch.curOffset = 100 + prefetch.cond.Signal() + prefetch.mu.Unlock() + }, + }, + { + name: "cancel while waiting on cond", + runConcurrently: true, + setupPrefetch: func(ctx context.Context, cancel context.CancelFunc) *objectPrefetch { + p := &objectPrefetch{ + ctx: ctx, + cancel: cancel, + entries: []object.IndirectObjectEntry{ + {Start: 0, Length: 100, Object: mustParseID("D0123456789abcdef0123456789abcdef")}, + {Start: 100, Length: 100, Object: mustParseID("D0123456789abcdef0123456789abcdeg")}, + }, + budget: 50, + curOffset: 0, + } + p.cond = sync.NewCond(&p.mu) + // Simulate the watcher goroutine spawned in OpenObject + go func() { + <-ctx.Done() + p.mu.Lock() + p.cond.Broadcast() + p.mu.Unlock() + }() + return p + }, + mockRepo: func(mockRepo *repomocks.MockRepository) { + mockRepo.On("PrefetchObjects", mock.Anything, []object.ID{mustParseID("D0123456789abcdef0123456789abcdef")}, "").Return(([]content.ID)(nil), nil).Once() + }, + trigger: func(prefetch *objectPrefetch) { + // Wait a bit for the first fetch and wait to happen + time.Sleep(50 * time.Millisecond) + prefetch.cancel() // This triggers the watcher, broadcasts, and exits prefetchProc + }, + }, + { + name: "prefetch error should not panic and continue", + setupPrefetch: func(ctx context.Context, cancel context.CancelFunc) *objectPrefetch { + p := &objectPrefetch{ + ctx: ctx, + entries: []object.IndirectObjectEntry{ + {Start: 0, Length: 100, Object: mustParseID("D0123456789abcdef0123456789abcdef")}, + }, + budget: 200, + curOffset: 0, + } + p.cond = sync.NewCond(&p.mu) + return p + }, + mockRepo: func(mockRepo *repomocks.MockRepository) { + mockRepo.On("PrefetchObjects", mock.Anything, []object.ID{mustParseID("D0123456789abcdef0123456789abcdef")}, "").Return(([]content.ID)(nil), errors.New("fake-error")).Once() + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + mockRepo := repomocks.NewMockRepository(t) + if tc.mockRepo != nil { + tc.mockRepo(mockRepo) + } + + kor := &kopiaObjectReader{ + rawRepo: mockRepo, + logger: velerotest.NewLogger(), + prefetch: tc.setupPrefetch(ctx, cancel), + } + + if tc.runConcurrently { + done := make(chan struct{}) + go func() { + kor.prefetchProc() + close(done) + }() + if tc.trigger != nil { + tc.trigger(kor.prefetch) + } + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("prefetchProc did not finish in time") + } + } else { + kor.prefetchProc() + } + + mockRepo.AssertExpectations(t) + }) + } +} From c8127e243b78b8dc67b8fda6858088195cafd04e Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Tue, 11 Aug 2026 13:38:41 +0800 Subject: [PATCH 6/7] object reader throughput improvement Signed-off-by: Lyndon-Li --- changelogs/unreleased/10225-Lyndon-Li | 1 + pkg/repository/udmrepo/kopialib/lib_repo.go | 1 - pkg/repository/udmrepo/kopialib/lib_repo_test.go | 2 +- pkg/uploader/block/uploader_test.go | 2 +- pkg/uploader/kopia/shim_test.go | 6 +++--- 5 files changed, 6 insertions(+), 6 deletions(-) create mode 100644 changelogs/unreleased/10225-Lyndon-Li diff --git a/changelogs/unreleased/10225-Lyndon-Li b/changelogs/unreleased/10225-Lyndon-Li new file mode 100644 index 000000000..435da13d1 --- /dev/null +++ b/changelogs/unreleased/10225-Lyndon-Li @@ -0,0 +1 @@ +Add prefetch mechanism to object reader so as to improve the restore throughput of block data mover \ No newline at end of file diff --git a/pkg/repository/udmrepo/kopialib/lib_repo.go b/pkg/repository/udmrepo/kopialib/lib_repo.go index 0efde34a7..e60128358 100644 --- a/pkg/repository/udmrepo/kopialib/lib_repo.go +++ b/pkg/repository/udmrepo/kopialib/lib_repo.go @@ -382,7 +382,6 @@ func (kr *kopiaRepository) OpenObject(ctx context.Context, id udmrepo.ID, opt ud prefetch.cond = sync.NewCond(&prefetch.mu) } - } rd := &kopiaObjectReader{ diff --git a/pkg/repository/udmrepo/kopialib/lib_repo_test.go b/pkg/repository/udmrepo/kopialib/lib_repo_test.go index 6bd64009c..b4d487c43 100644 --- a/pkg/repository/udmrepo/kopialib/lib_repo_test.go +++ b/pkg/repository/udmrepo/kopialib/lib_repo_test.go @@ -908,7 +908,7 @@ func TestReaderClose(t *testing.T) { err := kr.Close() if tc.withPrefetch { - assert.ErrorIs(t, kr.prefetch.ctx.Err(), context.Canceled) + require.ErrorIs(t, kr.prefetch.ctx.Err(), context.Canceled) } if tc.expectedErr == "" { diff --git a/pkg/uploader/block/uploader_test.go b/pkg/uploader/block/uploader_test.go index 3b8930476..1b1eddbce 100644 --- a/pkg/uploader/block/uploader_test.go +++ b/pkg/uploader/block/uploader_test.go @@ -663,7 +663,7 @@ func TestBlockUploaderRestore(t *testing.T) { objReader.On("Read", mock.Anything).Return(0, io.EOF) objReader.On("Close").Return(nil) - repoWriter.On("OpenObject", mock.Anything, udmrepo.ID("data-id")).Return(objReader, nil) + repoWriter.On("OpenObject", mock.Anything, udmrepo.ID("data-id"), mock.Anything).Return(objReader, nil) snap := udmrepo.Snapshot{ Description: "test snapshot", diff --git a/pkg/uploader/kopia/shim_test.go b/pkg/uploader/kopia/shim_test.go index 7933ec6b8..3c7941405 100644 --- a/pkg/uploader/kopia/shim_test.go +++ b/pkg/uploader/kopia/shim_test.go @@ -81,7 +81,7 @@ func TestOpenObject(t *testing.T) { name: "Success", backupRepo: func() *mocks.BackupRepo { backupRepo := &mocks.BackupRepo{} - backupRepo.On("OpenObject", mock.Anything, mock.Anything).Return(&shimObjectReader{}, nil) + backupRepo.On("OpenObject", mock.Anything, mock.Anything, mock.Anything).Return(&shimObjectReader{}, nil) return backupRepo }(), }, @@ -89,7 +89,7 @@ func TestOpenObject(t *testing.T) { name: "Open object error", backupRepo: func() *mocks.BackupRepo { backupRepo := &mocks.BackupRepo{} - backupRepo.On("OpenObject", mock.Anything, mock.Anything).Return(&shimObjectReader{}, errors.New("Error open object")) + backupRepo.On("OpenObject", mock.Anything, mock.Anything, mock.Anything).Return(&shimObjectReader{}, errors.New("Error open object")) return backupRepo }(), isOpenObjectError: true, @@ -98,7 +98,7 @@ func TestOpenObject(t *testing.T) { name: "Get nil reader", backupRepo: func() *mocks.BackupRepo { backupRepo := &mocks.BackupRepo{} - backupRepo.On("OpenObject", mock.Anything, mock.Anything).Return(nil, nil) + backupRepo.On("OpenObject", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) return backupRepo }(), isReaderNil: true, From 8b7951426b5d282f041b3708f70db0f661c0e80d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Wenkai=20Yin=28=E5=B0=B9=E6=96=87=E5=BC=80=29?= Date: Tue, 11 Aug 2026 15:48:31 +0800 Subject: [PATCH 7/7] Add "SnapshotClass" to DataUploadResult (#10227) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add "SnapshotClass" to DataUploadResult Signed-off-by: Wenkai Yin(尹文开) --- changelogs/unreleased/10227-ywk253100 | 1 + pkg/apis/velero/v2alpha1/data_upload_types.go | 4 ++++ .../actions/dataupload_retrieve_action.go | 3 +++ .../dataupload_retrieve_action_test.go | 23 +++++++++++++++++++ 4 files changed, 31 insertions(+) create mode 100644 changelogs/unreleased/10227-ywk253100 diff --git a/changelogs/unreleased/10227-ywk253100 b/changelogs/unreleased/10227-ywk253100 new file mode 100644 index 000000000..dcbbd6ac5 --- /dev/null +++ b/changelogs/unreleased/10227-ywk253100 @@ -0,0 +1 @@ +Add "SnapshotClass" to DataUploadResult \ No newline at end of file diff --git a/pkg/apis/velero/v2alpha1/data_upload_types.go b/pkg/apis/velero/v2alpha1/data_upload_types.go index 606502254..37e273b2b 100644 --- a/pkg/apis/velero/v2alpha1/data_upload_types.go +++ b/pkg/apis/velero/v2alpha1/data_upload_types.go @@ -268,4 +268,8 @@ type DataUploadResult struct { // FSType is the file system type of the volume. // +optional FSType string `json:"fsType,omitempty"` + + // SnapshotClass is the name of the snapshot class that the volume snapshot is created with + // +optional + SnapshotClass string `json:"snapshotClass,omitempty"` } diff --git a/pkg/restore/actions/dataupload_retrieve_action.go b/pkg/restore/actions/dataupload_retrieve_action.go index 77e4766f5..27db07471 100644 --- a/pkg/restore/actions/dataupload_retrieve_action.go +++ b/pkg/restore/actions/dataupload_retrieve_action.go @@ -82,6 +82,9 @@ func (d *DataUploadRetrieveAction) Execute(input *velero.RestoreItemActionExecut NodeOS: dataUpload.Status.NodeOS, FSType: dataUpload.Spec.SourceFSType, } + if dataUpload.Spec.CSISnapshot != nil { + dataUploadResult.SnapshotClass = dataUpload.Spec.CSISnapshot.SnapshotClass + } jsonBytes, err := json.Marshal(dataUploadResult) if err != nil { diff --git a/pkg/restore/actions/dataupload_retrieve_action_test.go b/pkg/restore/actions/dataupload_retrieve_action_test.go index 64be241bf..33a46a0a3 100644 --- a/pkg/restore/actions/dataupload_retrieve_action_test.go +++ b/pkg/restore/actions/dataupload_retrieve_action_test.go @@ -66,6 +66,29 @@ func TestDataUploadRetrieveActionExectue(t *testing.T) { }, expectedDataUploadResult: builder.ForConfigMap("velero", "").ObjectMeta(builder.WithGenerateName("testDU-"), builder.WithLabels(velerov1.PVCNamespaceNameLabel, "testNamespace.testPVC", velerov1.RestoreUIDLabel, "testingUID", velerov1.ResourceUsageLabel, string(velerov1.VeleroResourceUsageDataUploadResult))).Data("testingUID", `{"backupStorageLocation":"testLocation","snapshotID":"fake-id","sourceNamespace":"testNamespace","snapshotSize":1000}`).Result(), }, + { + name: "DataUploadRetrieve Action test with optional fields", + dataUpload: func() *velerov2alpha1.DataUpload { + du := builder.ForDataUpload("velero", "testDU"). + SourceNamespace("testNamespace"). + SourcePVC("testPVC"). + SnapshotID("fake-id"). + TotalBytes(1000). + DataMover("velero"). + NodeOS("linux"). + CSISnapshot(&velerov2alpha1.CSISnapshotSpec{SnapshotClass: "testClass"}). + Result() + du.Status.DataMoverResult = &map[string]string{"key": "value"} + du.Spec.SourceFSType = "ext4" + return du + }(), + restore: builder.ForRestore("velero", "testRestore").ObjectMeta(builder.WithUID("testingUID")).Backup("testBackup").Result(), + runtimeScheme: scheme, + veleroObjs: []runtime.Object{ + builder.ForBackup("velero", "testBackup").StorageLocation("testLocation").Result(), + }, + expectedDataUploadResult: builder.ForConfigMap("velero", "").ObjectMeta(builder.WithGenerateName("testDU-"), builder.WithLabels(velerov1.PVCNamespaceNameLabel, "testNamespace.testPVC", velerov1.RestoreUIDLabel, "testingUID", velerov1.ResourceUsageLabel, string(velerov1.VeleroResourceUsageDataUploadResult))).Data("testingUID", `{"backupStorageLocation":"testLocation","datamover":"velero","snapshotID":"fake-id","sourceNamespace":"testNamespace","dataMoverResult":{"key":"value"},"nodeOS":"linux","snapshotSize":1000,"fsType":"ext4","snapshotClass":"testClass"}`).Result(), + }, { name: "Long source namespace and PVC name should also work", dataUpload: builder.ForDataUpload("velero", "testDU").SourceNamespace("migre209d0da-49c7-45ba-8d5a-3e59fd591ec1").SourcePVC("kibishii-data-kibishii-deployment-0").Result(),