From 55c756ed0fb22fa73e9006a47eae3594dd6b41ef Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Thu, 10 Sep 2026 15:57:05 +0800 Subject: [PATCH] data path support fallbackFull Signed-off-by: Lyndon-Li --- pkg/controller/data_download_controller.go | 1 + pkg/controller/data_upload_controller.go | 1 + .../pod_volume_backup_controller.go | 1 + .../pod_volume_restore_controller.go | 1 + pkg/datapath/data_path.go | 11 +++-- pkg/datapath/data_path_test.go | 12 +++-- pkg/datapath/types.go | 2 + pkg/uploader/provider/mocks/Provider.go | 49 ++++++++++++------- pkg/uploader/provider/provider.go | 4 +- 9 files changed, 55 insertions(+), 27 deletions(-) diff --git a/pkg/controller/data_download_controller.go b/pkg/controller/data_download_controller.go index 053a083e7..3507edfe5 100644 --- a/pkg/controller/data_download_controller.go +++ b/pkg/controller/data_download_controller.go @@ -502,6 +502,7 @@ func (r *DataDownloadReconciler) OnDataDownloadCompleted(ctx context.Context, na dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseCompleted dd.Status.IncrementalBytes = ptr.To(result.Restore.IncrementalBytes) + dd.Status.FallbackFull = result.Restore.FallbackFull dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()} delete(dd.Labels, exposer.ExposeOnGoingLabel) diff --git a/pkg/controller/data_upload_controller.go b/pkg/controller/data_upload_controller.go index 534458acc..84504153f 100644 --- a/pkg/controller/data_upload_controller.go +++ b/pkg/controller/data_upload_controller.go @@ -514,6 +514,7 @@ func (r *DataUploadReconciler) OnDataUploadCompleted(ctx context.Context, namesp du.Status.SnapshotID = result.Backup.SnapshotID du.Status.IncrementalBytes = result.Backup.IncrementalBytes du.Status.SourceSize = result.Backup.SourceSize + du.Status.FallbackFull = result.Backup.FallbackFull du.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()} if result.Backup.EmptySnapshot { diff --git a/pkg/controller/pod_volume_backup_controller.go b/pkg/controller/pod_volume_backup_controller.go index fc1570325..399f4a3b8 100644 --- a/pkg/controller/pod_volume_backup_controller.go +++ b/pkg/controller/pod_volume_backup_controller.go @@ -551,6 +551,7 @@ func (r *PodVolumeBackupReconciler) OnDataPathCompleted(ctx context.Context, nam pvb.Status.CompletionTimestamp = &completionTime pvb.Status.IncrementalBytes = result.Backup.IncrementalBytes pvb.Status.SourceSize = result.Backup.SourceSize + pvb.Status.FallbackFull = result.Backup.FallbackFull if result.Backup.EmptySnapshot { pvb.Status.Message = "volume was empty so no snapshot was taken" } diff --git a/pkg/controller/pod_volume_restore_controller.go b/pkg/controller/pod_volume_restore_controller.go index d8ffb0e1a..81540aa15 100644 --- a/pkg/controller/pod_volume_restore_controller.go +++ b/pkg/controller/pod_volume_restore_controller.go @@ -838,6 +838,7 @@ func (r *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, na pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseCompleted pvr.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()} pvr.Status.IncrementalBytes = ptr.To(result.Restore.IncrementalBytes) + pvr.Status.FallbackFull = result.Restore.FallbackFull delete(pvr.Labels, exposer.ExposeOnGoingLabel) diff --git a/pkg/datapath/data_path.go b/pkg/datapath/data_path.go index bbc577230..2e4ee238e 100644 --- a/pkg/datapath/data_path.go +++ b/pkg/datapath/data_path.go @@ -197,7 +197,7 @@ func (dp *generalDataPath) StartBackup(source AccessPoint, uploaderConfig map[st dp.wgDataPath.Done() }() - snapshotID, emptySnapshot, totalBytes, incrementalBytes, sourceSize, err := dp.uploaderProv.RunBackup( + snapshotID, emptySnapshot, totalBytes, incrementalBytes, sourceSize, fallback, err := dp.uploaderProv.RunBackup( dp.ctx, source.ByPath, backupParam.RealSource, @@ -233,6 +233,7 @@ func (dp *generalDataPath) StartBackup(source AccessPoint, uploaderConfig map[st TotalBytes: totalBytes, IncrementalBytes: ptr.To(incrementalBytes), SourceSize: sourceSize, + FallbackFull: fallback, }}) } }() @@ -257,7 +258,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u dp.wgDataPath.Done() }() - incrementalBytes, totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental, + incrementalBytes, totalBytes, fallback, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental, provider.CBTParam{ Source: cbtservice.SourceInfo{ Snapshot: restoreParam.VolumeSnapshotName, @@ -275,7 +276,11 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u } dp.callbacks.OnFailed(context.Background(), dp.namespace, dp.jobName, dataPathErr) } else { - dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target, TotalBytes: totalBytes, IncrementalBytes: incrementalBytes}}) + dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target, + TotalBytes: totalBytes, + IncrementalBytes: incrementalBytes, + FallbackFull: fallback, + }}) } }() diff --git a/pkg/datapath/data_path_test.go b/pkg/datapath/data_path_test.go index 8f8cf285f..7bf70f209 100644 --- a/pkg/datapath/data_path_test.go +++ b/pkg/datapath/data_path_test.go @@ -89,6 +89,7 @@ func TestAsyncBackup(t *testing.T) { TotalBytes: 3000, IncrementalBytes: ptr.To(int64(200)), SourceSize: 2000, + FallbackFull: true, }, }, path: "fake-path", @@ -112,6 +113,7 @@ func TestAsyncBackup(t *testing.T) { TotalBytes: 2000, IncrementalBytes: ptr.To(int64(200)), SourceSize: 2000, + FallbackFull: false, }, }, path: "fake-path", @@ -126,7 +128,7 @@ func TestAsyncBackup(t *testing.T) { if test.result.Backup.IncrementalBytes != nil { incrementalBytes = *test.result.Backup.IncrementalBytes } - mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, incrementalBytes, test.result.Backup.SourceSize, test.err) + mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, incrementalBytes, test.result.Backup.SourceSize, test.result.Backup.FallbackFull, test.err) mockProvider.On("Close", mock.Anything).Return(nil) dp.uploaderProv = mockProvider dp.initialized = true @@ -201,8 +203,10 @@ func TestAsyncRestore(t *testing.T) { }, result: Result{ Restore: RestoreResult{ - Target: AccessPoint{ByPath: "fake-path"}, - TotalBytes: 1000, + Target: AccessPoint{ByPath: "fake-path"}, + TotalBytes: 1000, + IncrementalBytes: 500, + FallbackFull: true, }, }, path: "fake-path", @@ -214,7 +218,7 @@ func TestAsyncRestore(t *testing.T) { t.Run(test.name, func(t *testing.T) { dp := newGeneralDataPath("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*generalDataPath) mockProvider := providerMock.NewProvider(t) - mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.IncrementalBytes, test.result.Restore.TotalBytes, test.err) + mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.IncrementalBytes, test.result.Restore.TotalBytes, test.result.Restore.FallbackFull, test.err) mockProvider.On("Close", mock.Anything).Return(nil) dp.uploaderProv = mockProvider dp.initialized = true diff --git a/pkg/datapath/types.go b/pkg/datapath/types.go index 7fbf2a957..dfb545a10 100644 --- a/pkg/datapath/types.go +++ b/pkg/datapath/types.go @@ -40,6 +40,7 @@ type BackupResult struct { // unmarshals to a non-nil zero, distinguishing "measured zero" from "not measured". IncrementalBytes *int64 `json:"incrementalBytes,omitempty"` SourceSize int64 `json:"sourceSize,omitempty"` + FallbackFull bool `json:"fallbackFull,omitempty"` } // RestoreResult represents the result of a restore @@ -47,6 +48,7 @@ type RestoreResult struct { Target AccessPoint `json:"target,omitempty"` TotalBytes int64 `json:"totalBytes,omitempty"` IncrementalBytes int64 `json:"incrementalBytes,omitempty"` + FallbackFull bool `json:"fallbackFull,omitempty"` } // Callbacks defines the collection of callbacks during backup/restore diff --git a/pkg/uploader/provider/mocks/Provider.go b/pkg/uploader/provider/mocks/Provider.go index f2b8d764a..a50dc17af 100644 --- a/pkg/uploader/provider/mocks/Provider.go +++ b/pkg/uploader/provider/mocks/Provider.go @@ -91,7 +91,7 @@ func (_c *Provider_Close_Call) RunAndReturn(run func(ctx context.Context) error) } // RunBackup provides a mock function for the type Provider -func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, error) { +func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error) { ret := _mock.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) if len(ret) == 0 { @@ -103,8 +103,9 @@ func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource st var r2 int64 var r3 int64 var r4 int64 - var r5 error - if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, int64, error)); ok { + var r5 bool + var r6 error + if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error)); ok { return returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) } if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok { @@ -132,12 +133,17 @@ func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource st } else { r4 = ret.Get(4).(int64) } - if returnFunc, ok := ret.Get(5).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { + if returnFunc, ok := ret.Get(5).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok { r5 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) } else { - r5 = ret.Error(5) + r5 = ret.Get(5).(bool) } - return r0, r1, r2, r3, r4, r5 + if returnFunc, ok := ret.Get(6).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { + r6 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) + } else { + r6 = ret.Error(6) + } + return r0, r1, r2, r3, r4, r5, r6 } // Provider_RunBackup_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunBackup' @@ -218,18 +224,18 @@ func (_c *Provider_RunBackup_Call) Run(run func(ctx context.Context, path string return _c } -func (_c *Provider_RunBackup_Call) Return(_a0 string, _a1 bool, _a2 int64, _a3 int64, _a4 int64, _a5 error) *Provider_RunBackup_Call { - _c.Call.Return(_a0, _a1, _a2, _a3, _a4, _a5) +func (_c *Provider_RunBackup_Call) Return(_a0 string, _a1 bool, _a2 int64, _a3 int64, _a4 int64, _a5 bool, _a6 error) *Provider_RunBackup_Call { + _c.Call.Return(_a0, _a1, _a2, _a3, _a4, _a5, _a6) return _c } -func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, error)) *Provider_RunBackup_Call { +func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error)) *Provider_RunBackup_Call { _c.Call.Return(run) return _c } // RunRestore provides a mock function for the type Provider -func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, error) { +func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, bool, error) { ret := _mock.Called(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater) if len(ret) == 0 { @@ -238,8 +244,9 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume var r0 int64 var r1 int64 - var r2 error - if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, int64, error)); ok { + var r2 bool + var r3 error + if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, int64, bool, error)); ok { return returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater) } if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { @@ -254,13 +261,19 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume r1 = ret.Get(1).(int64) } - if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { + if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok { r2 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater) } else { - r2 = ret.Error(2) + r2 = ret.Get(2).(bool) } - return r0, r1, r2 + if returnFunc, ok := ret.Get(3).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { + r3 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater) + } else { + r3 = ret.Error(3) + } + + return r0, r1, r2, r3 } // Provider_RunRestore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunRestore' @@ -329,12 +342,12 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID return _c } -func (_c *Provider_RunRestore_Call) Return(_a0 int64, _a1 int64, _a2 error) *Provider_RunRestore_Call { - _c.Call.Return(_a0, _a1, _a2) +func (_c *Provider_RunRestore_Call) Return(_a0 int64, _a1 int64, _a2 bool, _a3 error) *Provider_RunRestore_Call { + _c.Call.Return(_a0, _a1, _a2, _a3) return _c } -func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, error)) *Provider_RunRestore_Call { +func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, bool, error)) *Provider_RunRestore_Call { _c.Call.Return(run) return _c } diff --git a/pkg/uploader/provider/provider.go b/pkg/uploader/provider/provider.go index 53a44a4fb..79ea92f53 100644 --- a/pkg/uploader/provider/provider.go +++ b/pkg/uploader/provider/provider.go @@ -57,7 +57,7 @@ type Provider interface { cbtParam CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, - updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, error) + updater uploader.ProgressUpdater) (string, bool, int64, int64, int64, bool, error) // RunRestore which will do restore for one specific volume with given snapshot id and return error // updater is used for updating backup progress which implement by third-party RunRestore( @@ -68,7 +68,7 @@ type Provider interface { cbtParam CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, - updater uploader.ProgressUpdater) (int64, int64, error) + updater uploader.ProgressUpdater) (int64, int64, bool, error) // Close which will close related repository Close(ctx context.Context) error }