data path support fallbackFull

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

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)

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 {

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"
 		}

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)
 

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,
+			}})
 		}
 	}()
 

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

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

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
 }

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
 }