mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-08-15 19:56:06 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into block-uploader-restore-progress

Browse Source
This commit is contained in:
Lyndon-Li
2026-08-11 15:53:08 +08:00
parent bbb0f11f33 23c215d054
commit 882ad944f3
13 changed files with 412 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelogs/unreleased/10225-Lyndon-Li
+1
View File
@@ -0,0 +1 @@
Add prefetch mechanism to object reader so as to improve the restore throughput of block data mover
changelogs/unreleased/10227-ywk253100
+1
View File
@@ -0,0 +1 @@
Add "SnapshotClass" to DataUploadResult
pkg/apis/velero/v2alpha1/data_upload_types.go
+4
View File
@@ -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"`
}
pkg/repository/udmrepo/kopialib/lib_repo.go
+135 -6
View File
@@ -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 {
@@ -338,7 +352,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")
}
@@ -353,9 +367,42 @@ func (kr *kopiaRepository) OpenObject(ctx context.Context, id udmrepo.ID) (udmre
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 {
@@ -550,7 +597,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)
}
@@ -792,7 +839,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 +856,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
}
pkg/repository/udmrepo/kopialib/lib_repo_test.go
+212 -4
View File
@@ -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))
_, 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 {
require.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)
})
}
}
pkg/repository/udmrepo/mocks/BackupRepo.go
+18 -12
View File
@@ -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
}
pkg/repository/udmrepo/repo.go
+6 -1
View File
@@ -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
pkg/restore/actions/dataupload_retrieve_action.go
+3
View File
@@ -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 {
pkg/restore/actions/dataupload_retrieve_action_test.go
+23
View File
@@ -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(),
pkg/uploader/block/uploader.go
+4 -1
View File
@@ -176,7 +176,10 @@ func (blkup *blockUploader) Restore(snapshot udmrepo.Snapshot, dest destInfo, bi
return 0, 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{
Prefetch: true,
PrefetchBudgetMB: 256,
})
if err != nil {
return 0, 0, errors.Wrapf(err, "error opening bdev object %v", meta.SubObjects[0].Name)
}
pkg/uploader/block/uploader_test.go
+1 -1
View File
@@ -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",
pkg/uploader/kopia/shim.go
+1 -1
View File
@@ -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)
}
pkg/uploader/kopia/shim_test.go
+3 -3
View File
@@ -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,