Use VolumeInfo to help restore the PV.

Add VolumeInfo for left PVs during backup.

Signed-off-by: Xun Jiang <jxun@vmware.com>

changelogs/unreleased/7138-blackpiglet

@@ -0,0 +1 @@
+Use VolumeInfo to help restore the PV.

design/pv_backup_info.md

@@ -108,6 +108,7 @@ type PVInfo struct {
 ### How the VolumeInfo array is generated.
 The function `persistBackup` has `backup *pkgbackup.Request` in parameters.
 From it, the `VolumeSnapshots`, `PodVolumeBackups`, `CSISnapshots`, `itemOperationsList`, and `SkippedPVTracker` can be read. All of them will be iterated and merged into the `VolumeInfo` array, and then persisted into backup repository in function `persistBackup`.
+After going through all the available sources, Velero will check whether there are still some PVs left. If there is any, Velero will generate VolumeInfo for them. The VolumeInfo will contain the PVC namespace, PVC name, PV name and the PVInfo structure.
 
 Please notice that the change happened in async operations are not reflected in the new metadata file. The file only covers the volume changes happen in the Velero server process scope.
 
@@ -125,7 +126,7 @@ type BackupStore interface {
 ### How the VolumeInfo array is used.
 
 #### Generate the PVC backed-up information summary
-The downstream tools can use this VolumeInfo array to format and display their volume information. This is in the scope of this feature.
+The downstream tools can use this VolumeInfo array to format and display their volume information. This is not in the scope of this feature.
 
 #### Retrieve volume backed-up information for `velero backup describe` command
 The `velero backup describe` can also use this VolumeInfo array structure to display the volume information. The snapshot data mover volume should use this structure at first, then the Velero native snapshot, CSI snapshot, and PodVolumeBackup can also use this structure. The detailed implementation is also not in this feature's scope.

pkg/controller/backup_controller.go

@@ -933,6 +933,9 @@ func generateVolumeInfo(backup *pkgbackup.Request, csiVolumeSnapshots []snapshot
 	dataUploadVolumeInfos := generateVolumeInfoFromDataUpload(backup, crClient, logger)
 	volumeInfos = append(volumeInfos, dataUploadVolumeInfos...)
 
+	leftPVVolumeInfos := generateVolumeInfoForLeftPV(backup, volumeInfos, logger)
+	volumeInfos = append(volumeInfos, leftPVVolumeInfos...)
+
 	return volumeInfos
 }
 
@@ -1237,3 +1240,36 @@ func generateVolumeInfoFromDataUpload(backup *pkgbackup.Request, crClient kbclie
 
 	return tmpVolumeInfos
 }
+
+// generateVolumeInfoForLeftPV generates VolumeInfo for all the left
+// PVs (not skipped, no native snapshot, not CSI snapshot, no DataUpload, no PVB)
+func generateVolumeInfoForLeftPV(backup *pkgbackup.Request, volumeInfos []volume.VolumeInfo, logger logrus.FieldLogger) []volume.VolumeInfo {
+	tmpVolumeInfos := make([]volume.VolumeInfo, 0)
+
+	volumeInfoMap := make(map[string]volume.VolumeInfo)
+	for _, volumeInfo := range volumeInfos {
+		volumeInfoMap[volumeInfo.PVName] = volumeInfo
+	}
+
+	for key, pvcPVInfo := range backup.PVMap {
+		// PVMap's key is either PVC name or PV name.
+		// PV name's format is pvcNamespace/pvName.
+		// This check only wants PV name key from PVMap.
+		if !strings.Contains(key, "/") {
+			if _, ok := volumeInfoMap[key]; !ok {
+				volumeInfo := volume.VolumeInfo{
+					PVCName:      pvcPVInfo.PVCName,
+					PVCNamespace: pvcPVInfo.PVCNamespace,
+					PVName:       pvcPVInfo.PV.Name,
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
+						Labels:        pvcPVInfo.PV.Labels,
+					},
+				}
+				tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
+			}
+		}
+	}
+
+	return tmpVolumeInfos
+}

pkg/controller/backup_controller_test.go

@@ -2480,6 +2480,59 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) {
 	}
 }
 
+func TestGenerateVolumeInfoForLeftPV(t *testing.T) {
+	tests := []struct {
+		name                string
+		pvMap               map[string]backup.PvcPvInfo
+		volumeInfos         []volume.VolumeInfo
+		expectedVolumeInfos []volume.VolumeInfo
+	}{
+		{
+			name: "generate VolumeInfo for the PV not included in the input volumeInfos",
+			pvMap: map[string]backup.PvcPvInfo{
+				"velero/testPVC": {},
+				"testPV1":        {},
+				"testPV2": {
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV2",
+							Labels: nil,
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+			},
+			volumeInfos: []volume.VolumeInfo{
+				{PVName: "testPV1"},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{
+				{
+					PVName: "testPV2",
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+						Labels:        nil,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			request := new(backup.Request)
+			if tc.pvMap != nil {
+				request.PVMap = tc.pvMap
+			}
+			logger := logging.DefaultLogger(logrus.DebugLevel, logging.FormatJSON)
+
+			volumeInfos := generateVolumeInfoForLeftPV(request, tc.volumeInfos, logger)
+			require.Equal(t, tc.expectedVolumeInfos, volumeInfos)
+		})
+	}
+}
+
 func int64Ptr(val int) *int64 {
 	i := int64(val)
 	return &i

pkg/controller/restore_controller.go

@@ -56,6 +56,7 @@ import (
 	kubeutil "github.com/vmware-tanzu/velero/pkg/util/kube"
 	"github.com/vmware-tanzu/velero/pkg/util/logging"
 	"github.com/vmware-tanzu/velero/pkg/util/results"
+	"github.com/vmware-tanzu/velero/pkg/volume"
 )
 
 // nonRestorableResources is an exclusion list  for the restoration process. Any resources
@@ -520,6 +521,16 @@ func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backu
 		return errors.Wrap(err, "fail to fetch CSI VolumeSnapshots metadata")
 	}
 
+	backupVolumeInfoMap := make(map[string]volume.VolumeInfo)
+	volumeInfos, err := backupStore.GetBackupVolumeInfos(restore.Spec.BackupName)
+	if err != nil || volumeInfos == nil {
+		restoreLog.WithError(err).Infof("Backup %s doesn't have volumeinfos metadata file.", restore.Spec.BackupName)
+	} else {
+		for _, volumeInfo := range volumeInfos.VolumeInfos {
+			backupVolumeInfoMap[volumeInfo.PVName] = volumeInfo
+		}
+	}
+
 	restoreLog.Info("starting restore")
 
 	var podVolumeBackups []*api.PodVolumeBackup
@@ -537,6 +548,7 @@ func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backu
 		ResourceModifiers:    resourceModifiers,
 		DisableInformerCache: r.disableInformerCache,
 		CSIVolumeSnapshots:   csiVolumeSnapshots,
+		VolumeInfoMap:        backupVolumeInfoMap,
 	}
 	restoreWarnings, restoreErrors := r.restorer.RestoreWithResolvers(restoreReq, actionsResolver, pluginManager)
 

pkg/controller/restore_controller_test.go

@@ -235,6 +235,7 @@ func TestRestoreReconcile(t *testing.T) {
 		putRestoreLogErr                error
 		expectedFinalPhase              string
 		addValidFinalizer               bool
+		emptyVolumeInfo                 bool
 	}{
 		{
 			name:                     "restore with both namespace in both includedNamespaces and excludedNamespaces fails validation",
@@ -415,6 +416,18 @@ func TestRestoreReconcile(t *testing.T) {
 			backup:      defaultBackup().StorageLocation("default").Result(),
 			expectedErr: false,
 		},
+		{
+			name:                  "valid restore with empty VolumeInfos",
+			location:              defaultStorageLocation,
+			restore:               NewRestore("foo", "bar", "backup-1", "ns-1", "", velerov1api.RestorePhaseNew).Result(),
+			backup:                defaultBackup().StorageLocation("default").Result(),
+			emptyVolumeInfo:       true,
+			expectedErr:           false,
+			expectedPhase:         string(velerov1api.RestorePhaseInProgress),
+			expectedStartTime:     &timestamp,
+			expectedCompletedTime: &timestamp,
+			expectedRestorerCall:  NewRestore("foo", "bar", "backup-1", "ns-1", "", velerov1api.RestorePhaseInProgress).Result(),
+		},
 	}
 
 	formatFlag := logging.FormatText
@@ -482,6 +495,11 @@ func TestRestoreReconcile(t *testing.T) {
 				backupStore.On("PutRestoreResults", test.backup.Name, test.restore.Name, mock.Anything).Return(nil)
 				backupStore.On("PutRestoredResourceList", test.restore.Name, mock.Anything).Return(nil)
 				backupStore.On("PutRestoreItemOperations", mock.Anything, mock.Anything).Return(nil)
+				if test.emptyVolumeInfo == true {
+					backupStore.On("GetBackupVolumeInfos", test.backup.Name).Return(nil, nil)
+				} else {
+					backupStore.On("GetBackupVolumeInfos", test.backup.Name).Return(&volume.VolumeInfos{}, nil)
+				}
 
 				volumeSnapshots := []*volume.Snapshot{
 					{

pkg/persistence/object_store.go

@@ -73,6 +73,7 @@ type BackupStore interface {
 	GetCSIVolumeSnapshots(name string) ([]*snapshotv1api.VolumeSnapshot, error)
 	GetCSIVolumeSnapshotContents(name string) ([]*snapshotv1api.VolumeSnapshotContent, error)
 	GetCSIVolumeSnapshotClasses(name string) ([]*snapshotv1api.VolumeSnapshotClass, error)
+	GetBackupVolumeInfos(name string) (*volume.VolumeInfos, error)
 
 	// BackupExists checks if the backup metadata file exists in object storage.
 	BackupExists(bucket, backupName string) (bool, error)

pkg/restore/request.go

@@ -62,6 +62,7 @@ type Request struct {
 	ResourceModifiers    *resourcemodifiers.ResourceModifiers
 	DisableInformerCache bool
 	CSIVolumeSnapshots   []*snapshotv1api.VolumeSnapshot
+	VolumeInfoMap        map[string]volume.VolumeInfo
 }
 
 type restoredItemStatus struct {