From 84bee825758ef06c288fd3744c528023f615833e Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Thu, 9 Jul 2026 11:49:18 +0800 Subject: [PATCH] block uploader backup implementation Signed-off-by: Lyndon-Li --- changelogs/unreleased/9979-Lyndon-Li | 1 + pkg/uploader/block/uploader.go | 32 ++++++++++++++-------------- pkg/uploader/block/uploader_test.go | 26 +++++++++++----------- 3 files changed, 30 insertions(+), 29 deletions(-) create mode 100644 changelogs/unreleased/9979-Lyndon-Li diff --git a/changelogs/unreleased/9979-Lyndon-Li b/changelogs/unreleased/9979-Lyndon-Li new file mode 100644 index 000000000..78134da35 --- /dev/null +++ b/changelogs/unreleased/9979-Lyndon-Li @@ -0,0 +1 @@ +Add the backup implementation for block data mover \ No newline at end of file diff --git a/pkg/uploader/block/uploader.go b/pkg/uploader/block/uploader.go index 9d4dde9cb..75e913cb7 100644 --- a/pkg/uploader/block/uploader.go +++ b/pkg/uploader/block/uploader.go @@ -71,8 +71,8 @@ func NewUploader(ctx context.Context, repoWriter udmrepo.BackupRepo, progress up } } -func (bu *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, bitmap cbt.Iterator, configs map[string]string) (udmrepo.Snapshot, int64, error) { - snapStart := bu.repoWriter.Time() +func (blkup *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, bitmap cbt.Iterator, configs map[string]string) (udmrepo.Snapshot, int64, error) { + snapStart := blkup.repoWriter.Time() if bitmap == nil { return udmrepo.Snapshot{}, 0, errors.New("bitmap is not available") @@ -83,7 +83,7 @@ func (bu *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, bitm backupMode = udmrepo.ObjectDataBackupModeFull } - destObj, err := bu.repoWriter.NewObjectWriter(bu.ctx, udmrepo.ObjectWriteOptions{ + destObj, err := blkup.repoWriter.NewObjectWriter(blkup.ctx, udmrepo.ObjectWriteOptions{ Description: "BDEV:" + getObjectName(source.realSource), DataType: udmrepo.ObjectDataTypeData, AccessMode: udmrepo.ObjectDataAccessModeBlock, @@ -97,12 +97,12 @@ func (bu *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, bitm defer destObj.Close() - id, backupSize, objectSize, err := bu.backupObject(source.dev, destObj, bitmap, source.size) + id, backupSize, objectSize, err := blkup.backupObject(source.dev, destObj, bitmap, source.size) if err != nil { return udmrepo.Snapshot{}, 0, errors.Wrapf(err, "error backing up bdev %s", source.realSource) } - entryId, err := bu.repoWriter.WriteMetadata(bu.ctx, &udmrepo.Metadata{ + entryID, err := blkup.repoWriter.WriteMetadata(blkup.ctx, &udmrepo.Metadata{ SubObjects: []udmrepo.ObjectMetadata{ { ID: id, @@ -120,7 +120,7 @@ func (bu *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, bitm return udmrepo.Snapshot{}, 0, errors.Wrap(err, "error writing metadata") } - snapEnd := bu.repoWriter.Time() + snapEnd := blkup.repoWriter.Time() return udmrepo.Snapshot{ Source: source.realSource, @@ -129,7 +129,7 @@ func (bu *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, bitm Description: source.realSource, TotalSize: objectSize, RootObject: udmrepo.ObjectMetadata{ - ID: entryId, + ID: entryID, Name: "bdev-root", Type: udmrepo.ObjectDataTypeMetadata, Permissions: 0o777, @@ -138,12 +138,12 @@ func (bu *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, bitm } // TODO implement in following PRs -func (bu *blockUploader) Restore(snapshot udmrepo.Snapshot, dest destInfo, bitmap cbt.Iterator, configs map[string]string) (int64, error) { +func (blkup *blockUploader) Restore(snapshot udmrepo.Snapshot, dest destInfo, bitmap cbt.Iterator, configs map[string]string) (int64, error) { return 0, errors.New("not implemented") } -func (bu *blockUploader) backupObject(dev *os.File, dest udmrepo.ObjectWriter, bitmap cbt.Iterator, totalLength int64) (udmrepo.ID, int64, int64, error) { - backupSize, objectSize, err := bu.backupData(dev, dest, bitmap, totalLength) +func (blkup *blockUploader) backupObject(dev *os.File, dest udmrepo.ObjectWriter, bitmap cbt.Iterator, totalLength int64) (udmrepo.ID, int64, int64, error) { + backupSize, objectSize, err := blkup.backupData(dev, dest, bitmap, totalLength) if err != nil { return "", backupSize, objectSize, err } @@ -165,7 +165,7 @@ func (r *readResult) resetBuffer(list *freelist.FreeList) { } } -func (bu *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.ObjectWriter, bitmap cbt.Iterator, totalLength int64) (int64, int64, error) { +func (blkup *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.ObjectWriter, bitmap cbt.Iterator, totalLength int64) (int64, int64, error) { blockSize := bitmap.BlockSize() list := freelist.New(bufferSize, int(blockSize)) resultChan := make(chan readResult, list.Capacity()) @@ -182,7 +182,7 @@ func (bu *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.ObjectWri var buffer []byte for valid { select { - case <-bu.ctx.Done(): + case <-blkup.ctx.Done(): return case <-quit: return @@ -229,11 +229,11 @@ func (bu *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.ObjectWri for curCount < int64(totalCount) { select { - case <-bu.ctx.Done(): + case <-blkup.ctx.Done(): writeErr = ErrCanceled case result, readerRunning = <-resultChan: if !readerRunning { - if bu.ctx.Err() != nil { + if blkup.ctx.Err() != nil { writeErr = ErrCanceled } else { writeErr = io.ErrUnexpectedEOF @@ -266,7 +266,7 @@ func (bu *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.ObjectWri result.resetBuffer(list) curCount++ - bu.progress.UpdateProgress(&uploader.Progress{BytesDone: lastPos, TotalBytes: aligned}) + blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: lastPos, TotalBytes: aligned}) } result.resetBuffer(list) @@ -283,7 +283,7 @@ func (bu *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.ObjectWri written += s - bu.progress.UpdateProgress(&uploader.Progress{BytesDone: aligned, TotalBytes: aligned}) + blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: aligned, TotalBytes: aligned}) } return written, aligned, nil diff --git a/pkg/uploader/block/uploader_test.go b/pkg/uploader/block/uploader_test.go index 2032e31ad..88fd4771e 100644 --- a/pkg/uploader/block/uploader_test.go +++ b/pkg/uploader/block/uploader_test.go @@ -24,7 +24,7 @@ import ( "testing" "time" - "github.com/pkg/errors" + "github.com/cockroachdb/errors" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -53,12 +53,12 @@ func TestNewUploader(t *testing.T) { uploader := NewUploader(ctx, repoWriter, progress, log) - bu, ok := uploader.(*blockUploader) + blkup, ok := uploader.(*blockUploader) assert.True(t, ok) - assert.Equal(t, ctx, bu.ctx) - assert.Equal(t, repoWriter, bu.repoWriter) - assert.Equal(t, progress, bu.progress) - assert.Equal(t, log, bu.log) + assert.Equal(t, ctx, blkup.ctx) + assert.Equal(t, repoWriter, blkup.repoWriter) + assert.Equal(t, progress, blkup.progress) + assert.Equal(t, log, blkup.log) } func TestGetObjectName(t *testing.T) { @@ -158,7 +158,7 @@ func TestCopyTailData(t *testing.T) { if tc.expectErr { assert.Error(t, err) } else { - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, tc.expected, n) } }) @@ -261,9 +261,9 @@ func TestBlockUploaderBackup(t *testing.T) { log := logrus.New() log.Out = io.Discard - bu := NewUploader(ctx, repoWriter, progress, log) + blkup := NewUploader(ctx, repoWriter, progress, log) - f, err := os.CreateTemp("", "blktest-*") + f, err := os.CreateTemp(t.TempDir(), "blktest-*") require.NoError(t, err) defer os.Remove(f.Name()) defer f.Close() @@ -317,7 +317,7 @@ func TestBlockUploaderBackup(t *testing.T) { } else if tc.cancelCtx { iterMock.On("BlockSize").Return(uint(1048576)) iterMock.On("Count").Return(uint64(1)) - iterMock.On("Next").Return(uint64(0), true) + iterMock.On("Next").Return(uint64(0), true).Maybe() objWriter.On("Result").Return(udmrepo.ID(""), errors.New("write failed")).Maybe() } else if tc.shortWrite { @@ -361,15 +361,15 @@ func TestBlockUploaderBackup(t *testing.T) { })).Return(objWriter, tc.createObjErr) } - snap, size, err := bu.Backup(srcInfo, tc.parentObj, iterator, nil) + snap, size, err := blkup.Backup(srcInfo, tc.parentObj, iterator, nil) if tc.expectErr { - assert.Error(t, err) + require.Error(t, err) if tc.expectErrStr != "" { assert.Contains(t, err.Error(), tc.expectErrStr) } } else { - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, "/data/volume1", snap.Source) assert.Equal(t, udmrepo.ID("meta-01"), snap.RootObject.ID) assert.Equal(t, int64(0), size)