diff --git a/changelogs/unreleased/9887-Lyndon-Li b/changelogs/unreleased/9887-Lyndon-Li new file mode 100644 index 000000000..a44418c9c --- /dev/null +++ b/changelogs/unreleased/9887-Lyndon-Li @@ -0,0 +1 @@ +Add WriteAt implementation for Incremental aware object writer for 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 bb70e1b18..6e9355765 100644 --- a/pkg/repository/udmrepo/kopialib/lib_repo.go +++ b/pkg/repository/udmrepo/kopialib/lib_repo.go @@ -92,7 +92,6 @@ type kopiaObjectWriterEx struct { description string compressor compression.Name splitter string - zeroBuffer []byte zeroObject object.ID writeLock sync.Mutex asyncWritesSem chan struct{} @@ -1045,6 +1044,11 @@ func (kow *kopiaObjectWriterEx) WriteAt(p []byte, offset int64) (int, error) { for i := startEntry; i < endEntry; i++ { e := kow.parentEntries[i] + + if e.Start != int64(i)*kow.blockSize { + return 0, errors.Errorf("parent entry %v start %v does not match expected start %v", i, e.Start, int64(i)*kow.blockSize) + } + if e.Length != kow.blockSize { return 0, errors.Errorf("parent entry %v length %v does not match child block size %v", i, e.Length, kow.blockSize) } @@ -1069,7 +1073,7 @@ func (kow *kopiaObjectWriterEx) WriteAt(p []byte, offset int64) (int, error) { objName := fmt.Sprintf("%s-b%v", kow.description, entryID) if err := kow.writeZeroObject(objName, entryID); err != nil { - return 0, errors.Wrapf(err, "error writting zero object for %s", objName) + return 0, errors.Wrapf(err, "error writing zero object for %s", objName) } curPos += kow.blockSize diff --git a/pkg/repository/udmrepo/kopialib/lib_repo_ex_test.go b/pkg/repository/udmrepo/kopialib/lib_repo_ex_test.go index 428ed0f11..fdaeb9f69 100644 --- a/pkg/repository/udmrepo/kopialib/lib_repo_ex_test.go +++ b/pkg/repository/udmrepo/kopialib/lib_repo_ex_test.go @@ -258,6 +258,7 @@ func TestKopiaObjectWriterEx_Write(t *testing.T) { { name: "write object returns nil writer", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockRepoWriter.On("NewObjectWriter", mock.Anything, mock.Anything).Return(nil) @@ -271,14 +272,16 @@ func TestKopiaObjectWriterEx_Write(t *testing.T) { inputData: make([]byte, 1024), expectedLen: 1024, verify: func(t *testing.T, kow *kopiaObjectWriterEx) { + t.Helper() err := kow.getWriteError() - assert.Error(t, err) - assert.Contains(t, err.Error(), "error openning writer for -b0") + require.Error(t, err) + assert.Contains(t, err.Error(), "error opening writer for -b0") }, }, { name: "write object result error", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -299,8 +302,9 @@ func TestKopiaObjectWriterEx_Write(t *testing.T) { inputData: make([]byte, 1024), expectedLen: 1024, verify: func(t *testing.T, kow *kopiaObjectWriterEx) { + t.Helper() err := kow.getWriteError() - assert.Error(t, err) + require.Error(t, err) assert.Contains(t, err.Error(), "simulated result error") }, }, @@ -462,6 +466,7 @@ func TestKopiaObjectWriterEx_Result(t *testing.T) { { name: "write indirect object encoding failure", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -481,6 +486,7 @@ func TestKopiaObjectWriterEx_Result(t *testing.T) { { name: "write indirect object result failure", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -652,6 +658,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "writer is closed", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() return &kopiaObjectWriterEx{ rawRepoWriter: nil, } @@ -663,6 +670,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "invalid offset", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() return &kopiaObjectWriterEx{ rawRepoWriter: repomocks.NewMockRepositoryWriter(t), blockSize: 1024, @@ -675,6 +683,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "invalid length", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() return &kopiaObjectWriterEx{ rawRepoWriter: repomocks.NewMockRepositoryWriter(t), blockSize: 1024, @@ -687,6 +696,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "cannot write back", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() return &kopiaObjectWriterEx{ rawRepoWriter: repomocks.NewMockRepositoryWriter(t), blockSize: 1024, @@ -702,6 +712,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "success write at cur pos", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -724,13 +735,15 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { offset: 0, expectedLen: 1024, verify: func(t *testing.T, kow *kopiaObjectWriterEx) { - assert.Equal(t, 1, len(kow.entries)) + t.Helper() + assert.Len(t, kow.entries, 1) assert.Equal(t, int64(0), kow.entries[0].Start) }, }, { name: "success write with gap filling zeros", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -754,7 +767,8 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { offset: 1024, expectedLen: 1024, verify: func(t *testing.T, kow *kopiaObjectWriterEx) { - assert.Equal(t, 2, len(kow.entries)) + t.Helper() + assert.Len(t, kow.entries, 2) assert.Equal(t, int64(0), kow.entries[0].Start) id, _ := object.ParseID("I12345") assert.Equal(t, id, kow.entries[0].Object) @@ -765,6 +779,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "success write with gap filling from parent", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -791,7 +806,8 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { offset: 1024, expectedLen: 1024, verify: func(t *testing.T, kow *kopiaObjectWriterEx) { - assert.Equal(t, 2, len(kow.entries)) + t.Helper() + assert.Len(t, kow.entries, 2) assert.Equal(t, int64(0), kow.entries[0].Start) parentID, _ := object.ParseID("Iparent") assert.Equal(t, parentID, kow.entries[0].Object) @@ -801,6 +817,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "success write zero length", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) return &kopiaObjectWriterEx{ ctx: context.Background(), @@ -813,12 +830,14 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { offset: 0, expectedLen: 0, verify: func(t *testing.T, kow *kopiaObjectWriterEx) { - assert.Equal(t, 0, len(kow.entries)) + t.Helper() + assert.Empty(t, kow.entries) }, }, { name: "gap filling with invalid parent entry length", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) return &kopiaObjectWriterEx{ ctx: context.Background(), @@ -837,6 +856,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "gap filling partially with parent and rest with zeros", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -864,7 +884,8 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { offset: 2048, expectedLen: 1024, verify: func(t *testing.T, kow *kopiaObjectWriterEx) { - assert.Equal(t, 3, len(kow.entries)) + t.Helper() + assert.Len(t, kow.entries, 3) assert.Equal(t, int64(0), kow.entries[0].Start) parentID, _ := object.ParseID("Iparent") @@ -880,6 +901,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { { name: "writeZeroObject failure", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -898,11 +920,12 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { }, inputData: make([]byte, 1024), offset: 1024, - expectedErr: "error writting zero object for -b0: error writting for -b0: simulated zero object write error", + expectedErr: "error writing zero object for -b0: error writing for -b0: simulated zero object write error", }, { name: "writeObject short write", setupWriter: func(t *testing.T) *kopiaObjectWriterEx { + t.Helper() mockRepoWriter := repomocks.NewMockRepositoryWriter(t) mockWriter := repomocks.NewWriter(t) @@ -922,8 +945,9 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { offset: 0, expectedLen: 1024, verify: func(t *testing.T, kow *kopiaObjectWriterEx) { + t.Helper() err := kow.getWriteError() - assert.Error(t, err) + require.Error(t, err) assert.Contains(t, err.Error(), "short write for -b0") }, }, @@ -941,7 +965,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) { if tc.expectedErr != "" { assert.EqualError(t, err, tc.expectedErr) } else { - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, tc.expectedLen, l) if tc.verify != nil { tc.verify(t, kow) @@ -972,14 +996,14 @@ func TestKopiaObjectWriterEx_MultipleWriteAt(t *testing.T) { } l, err := kow.WriteAt(make([]byte, 1024), 0) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, 1024, l) l, err = kow.WriteAt(make([]byte, 1024), 2048) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, 1024, l) - assert.Equal(t, 3, len(kow.entries)) + assert.Len(t, kow.entries, 3) assert.Equal(t, int64(0), kow.entries[0].Start) assert.Equal(t, int64(1024), kow.entries[1].Start) assert.Equal(t, id, kow.entries[1].Object) @@ -1028,7 +1052,7 @@ func TestKopiaObjectWriterEx_ConcurrentWriteAt(t *testing.T) { close(start) wg.Wait() - assert.Greater(t, len(kow.entries), 0) + assert.NotEmpty(t, kow.entries) } type dummyObjectWriter struct { @@ -1078,11 +1102,11 @@ func TestKopiaObjectWriterEx_LargeSequentialWrite(t *testing.T) { for i := 0; i < blocks; i++ { l, err := kow.Write(data) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, int(blockSize), l) } - assert.Equal(t, blocks, len(kow.entries)) + assert.Len(t, kow.entries, blocks) assert.Equal(t, int64(blocks-1)*blockSize, kow.entries[blocks-1].Start) } @@ -1103,11 +1127,11 @@ func TestKopiaObjectWriterEx_LargeSparseWriteAt(t *testing.T) { data := make([]byte, blockSize) l, err := kow.WriteAt(data, offset) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, int(blockSize), l) expectedEntries := 5121 - assert.Equal(t, expectedEntries, len(kow.entries)) + assert.Len(t, kow.entries, expectedEntries) assert.Equal(t, int64(0), kow.entries[0].Start) assert.Equal(t, offset, kow.entries[expectedEntries-1].Start) } @@ -1137,21 +1161,21 @@ func TestKopiaObjectWriterEx_MixedWriteAndWriteAt(t *testing.T) { // 1. Write 1 block sequentially data1 := make([]byte, blockSize) l, err := kow.Write(data1) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, int(blockSize), l) // Entries: [0:1024] - assert.Equal(t, 1, len(kow.entries)) + assert.Len(t, kow.entries, 1) assert.Equal(t, int64(0), kow.entries[0].Start) // 2. WriteAt with gap (offset = 2048). This creates a gap block at 1024 data2 := make([]byte, blockSize) l, err = kow.WriteAt(data2, 2048) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, int(blockSize), l) // Entries should now be 3: [0:1024, 1024:2048(zero object), 2048:3072] - assert.Equal(t, 3, len(kow.entries)) + assert.Len(t, kow.entries, 3) assert.Equal(t, int64(0), kow.entries[0].Start) assert.Equal(t, int64(1024), kow.entries[1].Start) assert.Equal(t, id, kow.entries[1].Object) // filled with zero block @@ -1160,11 +1184,11 @@ func TestKopiaObjectWriterEx_MixedWriteAndWriteAt(t *testing.T) { // 3. Write another block sequentially. It should append at 3072. data3 := make([]byte, blockSize) l, err = kow.Write(data3) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, int(blockSize), l) // Entries should now be 4: [0:1024, 1024:2048(zero object), 2048:3072, 3072:4096] - assert.Equal(t, 4, len(kow.entries)) + assert.Len(t, kow.entries, 4) assert.Equal(t, int64(3072), kow.entries[3].Start) } @@ -1194,12 +1218,14 @@ func TestKopiaObjectWriterEx_ConcurrentAsyncErrors(t *testing.T) { // Issue multiple writes so they all spawn async goroutines // First few writes shouldn't fail immediately until getWriteError catches the asynchronous fault for i := 0; i < 10; i++ { - kow.Write(data) + l, err := kow.Write(data) + require.NoError(t, err) + assert.Equal(t, 1024, l) } id, err := kow.Result() - assert.Error(t, err) + require.Error(t, err) assert.Contains(t, err.Error(), "simulated async error") assert.Equal(t, udmrepo.ID(""), id) } @@ -1232,7 +1258,9 @@ func TestKopiaObjectWriterEx_ConcurrentWriteAndWriteAt(t *testing.T) { go func() { defer wg.Done() <-start - kow.Write(make([]byte, 1024)) + l, err := kow.Write(make([]byte, 1024)) + require.NoError(t, err) + assert.Equal(t, 1024, l) }() } @@ -1255,13 +1283,13 @@ func TestKopiaObjectWriterEx_ConcurrentWriteAndWriteAt(t *testing.T) { wg.Wait() // We only care that the locking effectively mitigated a panic or slice data corruption - assert.Greater(t, len(kow.entries), 0) + assert.NotEmpty(t, kow.entries) } func TestKopiaObjectWriterEx_Checkpoint(t *testing.T) { kow := &kopiaObjectWriterEx{} id, err := kow.Checkpoint() - assert.Error(t, err) + require.Error(t, err) assert.Equal(t, udmrepo.ID(""), id) assert.Equal(t, "not supported", err.Error()) }