Add VolumeInfo metadata structures and object get method.

Modify design according to comments.
Add PVInfo structure.
Add backup VolumeInfo's object storage's put and get methods.

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

changelogs/unreleased/7070-blackpiglet

@@ -0,0 +1 @@
+Add VolumeInfo metadata structures.

design/pv_backup_info.md

@@ -42,37 +42,39 @@ The 1 version of `VolumeInfo` definition is:
 type VolumeInfoV1 struct {
     PVCName        string    // The PVC's name. The format should be <namespace-name>/<PVC-name>
     PVName         string    // The PV name.
-    BackupMethod   string    // The way the volume data is backed up. The valid value includes `VeleroNativeSnapshot`, `PodVolumeBackup`, `CSISnapshot` and `Skipped`.
-    SnapshotDataMovement bool // Whether the volume's snapshot data is moved to specified storage.
+    BackupMethod   string    // The way the volume data is backed up. The valid value includes `VeleroNativeSnapshot`, `PodVolumeBackup` and `CSISnapshot`.
+    SnapshotDataMoved bool   // Whether the volume's snapshot data is moved to specified storage.
 
+    Skipped         boolean   // Whether the Volume is skipped in this backup.
     SkippedReason   string    // The reason for the volume is skipped in the backup.
     StartTimestamp  *metav1.Time // Snapshot starts timestamp.
 
     CSISnapshotInfo          CSISnapshotInfo
-    SnapshotDataMoveInfo     SnapshotDataMoveInfo
+    SnapshotDataMovementInfo SnapshotDataMovementInfo
     NativeSnapshotInfo       VeleroNativeSnapshotInfo
     PVBInfo                  PodVolumeBackupInfo
+    PVInfo                   PVInfo
 }
 
 // CSISnapshotInfo is used for displaying the CSI snapshot status
 type CSISnapshotInfo struct {
-    SnapshotHandle  string       // The actual snapshot ID. It can be the cloud provider's snapshot ID or the file-system uploader's snapshot.
+    SnapshotHandle  string       // It's the storage provider's snapshot ID for CSI.
     Size            int64        // The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.
 
     Driver          string  // The name of the CSI driver.
     VSCName         string // The name of the VolumeSnapshotContent. 
 }
 
-// SnapshotDataMoveInfo is used for displaying the snapshot data mover status.
-type SnapshotDataMoveInfo struct {
+// SnapshotDataMovementInfo is used for displaying the snapshot data mover status.
+type SnapshotDataMovementInfo struct {
     DataMover        string    // The data mover used by the backup. The valid values are `velero` and ``(equals to `velero`).
     UploaderType     string    // The type of the uploader that uploads the snapshot data. The valid values are `kopia` and `restic`. It's useful for file-system backup and snapshot data mover.
-    RetainedSnapshot string    // The name or ID of the snapshot associated object(SAO).
+    RetainedSnapshot string    // The name or ID of the snapshot associated object(SAO). SAO is used to support local snapshots for the snapshot data mover, e.g. it could be a VolumeSnapshot for CSI snapshot data movement.
 }
 
 // VeleroNativeSnapshotInfo is used for displaying the Velero native snapshot status.
 type VeleroNativeSnapshotInfo struct {
-    SnapshotHandle      string       // The actual snapshot ID. It can be the cloud provider's snapshot ID or the file-system uploader's snapshot.
+    SnapshotHandle      string       // It's the storage provider's snapshot ID for the Velero-native snapshot.
     Size                int64        // The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.
 
     VolumeType string    // The cloud provider snapshot volume type.
@@ -82,12 +84,20 @@ type VeleroNativeSnapshotInfo struct {
 
 // PodVolumeBackupInfo is used for displaying the PodVolumeBackup snapshot status.
 type PodVolumeBackupInfo struct {
-    SnapshotHandle      string       // The actual snapshot ID. It can be the cloud provider's snapshot ID or the file-system uploader's snapshot.
+    SnapshotHandle      string       // It's the file-system uploader's snapshot ID for PodVolumeBackup.
     Size                int64        // The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.
 
     UploaderType  string    // The type of the uploader that uploads the data. The valid values are `kopia` and `restic`. It's useful for file-system backup and snapshot data mover.
-    VolumeName    string // The PVC's corresponding volume name used by Pod
-    PodName       string // The Pod name mounting this PVC. The format should be <namespace-name>/<pod-name>.
+    VolumeName    string   // The PVC's corresponding volume name used by Pod: https://github.com/kubernetes/kubernetes/blob/e4b74dd12fa8cb63c174091d5536a10b8ec19d34/pkg/apis/core/types.go#L48
+    PodName       string   // The Pod name mounting this PVC. The format should be <namespace-name>/<pod-name>.
+    NodeName      string   // The PVB-taken k8s node's name.
+}
+
+// PVInfo is used to store some PV information modified after creation.
+// Those information are lost after PV recreation.
+type PVInfo struct {
+    ReclaimPolicy string            // ReclaimPolicy of PV. It could be different from the referenced StorageClass.
+    labels        map[string]string // The PV's labels should be kept after recreation.
 }
 ```
 
@@ -164,11 +174,12 @@ After introducing the VolumeInfo array, the following logic will be added.
             ...
         case CSISnapshot:
             ...
-        case Skipped:
-            // Check whether the Velero server should restore the PV depending on the DeletionPolicy setting.
         default:
             // Need to check whether the volume is backed up by the SnapshotDataMover.
             if volumeInfo.SnapshotDataMovement:
+
+            // Check whether the Velero server should restore the PV depending on the DeletionPolicy setting.
+            if volumeInfo.Skipped:
 ```
 
 ### How the VolumeInfo metadata file is deleted

pkg/controller/backup_controller.go

@@ -849,6 +849,7 @@ func persistBackup(backup *pkgbackup.Request,
 ) []error {
 	persistErrs := []error{}
 	backupJSON := new(bytes.Buffer)
+	volumeInfos := make([]volume.VolumeInfoV1, 0)
 
 	if err := encode.To(backup.Backup, "json", backupJSON); err != nil {
 		persistErrs = append(persistErrs, errors.Wrap(err, "error encoding backup"))
@@ -895,6 +896,11 @@ func persistBackup(backup *pkgbackup.Request,
 		persistErrs = append(persistErrs, errs...)
 	}
 
+	volumeInfoJSON, errs := encode.ToJSONGzip(volumeInfos, "backup volumes information")
+	if errs != nil {
+		persistErrs = append(persistErrs, errs...)
+	}
+
 	if len(persistErrs) > 0 {
 		// Don't upload the JSON files or backup tarball if encoding to json fails.
 		backupJSON = nil
@@ -906,6 +912,7 @@ func persistBackup(backup *pkgbackup.Request,
 		csiSnapshotContentsJSON = nil
 		csiSnapshotClassesJSON = nil
 		backupResult = nil
+		volumeInfoJSON = nil
 	}
 
 	backupInfo := persistence.BackupInfo{
@@ -921,6 +928,7 @@ func persistBackup(backup *pkgbackup.Request,
 		CSIVolumeSnapshots:        csiSnapshotJSON,
 		CSIVolumeSnapshotContents: csiSnapshotContentsJSON,
 		CSIVolumeSnapshotClasses:  csiSnapshotClassesJSON,
+		BackupVolumeInfo:          volumeInfoJSON,
 	}
 	if err := backupStore.PutBackup(backupInfo); err != nil {
 		persistErrs = append(persistErrs, err)

pkg/persistence/object_store.go

@@ -17,8 +17,10 @@ limitations under the License.
 package persistence
 
 import (
+	"bytes"
 	"compress/gzip"
 	"encoding/json"
+	"fmt"
 	"io"
 	"strings"
 	"time"
@@ -50,7 +52,8 @@ type BackupInfo struct {
 	BackupResourceList,
 	CSIVolumeSnapshots,
 	CSIVolumeSnapshotContents,
-	CSIVolumeSnapshotClasses io.Reader
+	CSIVolumeSnapshotClasses,
+	BackupVolumeInfo io.Reader
 }
 
 // BackupStore defines operations for creating, retrieving, and deleting
@@ -270,6 +273,7 @@ func (s *objectBackupStore) PutBackup(info BackupInfo) error {
 		s.layout.getCSIVolumeSnapshotContentsKey(info.Name): info.CSIVolumeSnapshotContents,
 		s.layout.getCSIVolumeSnapshotClassesKey(info.Name):  info.CSIVolumeSnapshotClasses,
 		s.layout.getBackupResultsKey(info.Name):             info.BackupResults,
+		s.layout.getBackupVolumeInfoKey(info.Name):          info.BackupVolumeInfo,
 	}
 
 	for key, reader := range backupObjs {
@@ -491,6 +495,46 @@ func (s *objectBackupStore) GetPodVolumeBackups(name string) ([]*velerov1api.Pod
 	return podVolumeBackups, nil
 }
 
+func (s *objectBackupStore) GetBackupVolumeInfos(name string) (*volume.VolumeInfos, error) {
+	var volumeInfos *volume.VolumeInfos
+	var volumeInfoVersion volume.VolumeInfoVersion
+
+	res, err := tryGet(s.objectStore, s.bucket, s.layout.getBackupVolumeInfoKey(name))
+	if err != nil {
+		return volumeInfos, err
+	}
+	if res == nil {
+		return volumeInfos, nil
+	}
+	defer res.Close()
+
+	// The res's reader is a stream reader. If cannot be read twice.
+	// So copy it before using.
+	var copyReader bytes.Buffer
+	teeReader := io.TeeReader(res, &copyReader)
+
+	if err := decode(teeReader, &volumeInfoVersion); err != nil {
+		return volumeInfos, err
+	}
+
+	switch volumeInfoVersion.Version {
+	case volume.VolumeInfoVersionV1:
+		volumeInfos = new(volume.VolumeInfos)
+		volumeInfos.Version = volume.VolumeInfoVersionV1
+		volumeInfosV1 := new(volume.VolumeInfosV1)
+		if err := decode(&copyReader, &volumeInfosV1); err != nil {
+			return volumeInfos, err
+		}
+
+		volumeInfos.VolumeInfosV1 = volumeInfosV1.Infos
+
+	default:
+		return volumeInfos, fmt.Errorf("unknown backup VolumeInfo version: %s", volumeInfoVersion.Version)
+	}
+
+	return volumeInfos, nil
+}
+
 func (s *objectBackupStore) GetBackupContents(name string) (io.ReadCloser, error) {
 	return s.objectStore.GetObject(s.bucket, s.layout.getBackupContentsKey(name))
 }

pkg/persistence/object_store_layout.go

@@ -128,3 +128,7 @@ func (l *ObjectStoreLayout) getCSIVolumeSnapshotClassesKey(backup string) string
 func (l *ObjectStoreLayout) getBackupResultsKey(backup string) string {
 	return path.Join(l.subdirs["backups"], backup, fmt.Sprintf("%s-results.gz", backup))
 }
+
+func (l *ObjectStoreLayout) getBackupVolumeInfoKey(backup string) string {
+	return path.Join(l.subdirs["backups"], backup, fmt.Sprintf("%s-volumeinfos.json.gz", backup))
+}

pkg/persistence/object_store_test.go

@@ -227,6 +227,7 @@ func TestPutBackup(t *testing.T) {
 		snapshots            io.Reader
 		backupItemOperations io.Reader
 		resourceList         io.Reader
+		backupVolumeInfo     io.Reader
 		expectedErr          string
 		expectedKeys         []string
 	}{
@@ -239,6 +240,7 @@ func TestPutBackup(t *testing.T) {
 			snapshots:            newStringReadSeeker("snapshots"),
 			backupItemOperations: newStringReadSeeker("backupItemOperations"),
 			resourceList:         newStringReadSeeker("resourceList"),
+			backupVolumeInfo:     newStringReadSeeker("backupVolumeInfo"),
 			expectedErr:          "",
 			expectedKeys: []string{
 				"backups/backup-1/velero-backup.json",
@@ -248,6 +250,7 @@ func TestPutBackup(t *testing.T) {
 				"backups/backup-1/backup-1-volumesnapshots.json.gz",
 				"backups/backup-1/backup-1-itemoperations.json.gz",
 				"backups/backup-1/backup-1-resource-list.json.gz",
+				"backups/backup-1/backup-1-volumeinfos.json.gz",
 			},
 		},
 		{
@@ -260,6 +263,7 @@ func TestPutBackup(t *testing.T) {
 			snapshots:            newStringReadSeeker("snapshots"),
 			backupItemOperations: newStringReadSeeker("backupItemOperations"),
 			resourceList:         newStringReadSeeker("resourceList"),
+			backupVolumeInfo:     newStringReadSeeker("backupVolumeInfo"),
 			expectedErr:          "",
 			expectedKeys: []string{
 				"prefix-1/backups/backup-1/velero-backup.json",
@@ -269,6 +273,7 @@ func TestPutBackup(t *testing.T) {
 				"prefix-1/backups/backup-1/backup-1-volumesnapshots.json.gz",
 				"prefix-1/backups/backup-1/backup-1-itemoperations.json.gz",
 				"prefix-1/backups/backup-1/backup-1-resource-list.json.gz",
+				"prefix-1/backups/backup-1/backup-1-volumeinfos.json.gz",
 			},
 		},
 		{
@@ -280,6 +285,7 @@ func TestPutBackup(t *testing.T) {
 			snapshots:            newStringReadSeeker("snapshots"),
 			backupItemOperations: newStringReadSeeker("backupItemOperations"),
 			resourceList:         newStringReadSeeker("resourceList"),
+			backupVolumeInfo:     newStringReadSeeker("backupVolumeInfo"),
 			expectedErr:          "error readers return errors",
 			expectedKeys:         []string{"backups/backup-1/backup-1-logs.gz"},
 		},
@@ -291,6 +297,7 @@ func TestPutBackup(t *testing.T) {
 			snapshots:            newStringReadSeeker("snapshots"),
 			backupItemOperations: newStringReadSeeker("backupItemOperations"),
 			resourceList:         newStringReadSeeker("resourceList"),
+			backupVolumeInfo:     newStringReadSeeker("backupVolumeInfo"),
 			expectedErr:          "error readers return errors",
 			expectedKeys:         []string{"backups/backup-1/backup-1-logs.gz"},
 		},
@@ -303,6 +310,7 @@ func TestPutBackup(t *testing.T) {
 			snapshots:            newStringReadSeeker("snapshots"),
 			backupItemOperations: newStringReadSeeker("backupItemOperations"),
 			resourceList:         newStringReadSeeker("resourceList"),
+			backupVolumeInfo:     newStringReadSeeker("backupVolumeInfo"),
 			expectedErr:          "",
 			expectedKeys: []string{
 				"backups/backup-1/velero-backup.json",
@@ -311,23 +319,26 @@ func TestPutBackup(t *testing.T) {
 				"backups/backup-1/backup-1-volumesnapshots.json.gz",
 				"backups/backup-1/backup-1-itemoperations.json.gz",
 				"backups/backup-1/backup-1-resource-list.json.gz",
+				"backups/backup-1/backup-1-volumeinfos.json.gz",
 			},
 		},
 		{
-			name:            "data should be uploaded even when metadata is nil",
-			metadata:        nil,
-			contents:        newStringReadSeeker("contents"),
-			log:             newStringReadSeeker("log"),
-			podVolumeBackup: newStringReadSeeker("podVolumeBackup"),
-			snapshots:       newStringReadSeeker("snapshots"),
-			resourceList:    newStringReadSeeker("resourceList"),
-			expectedErr:     "",
+			name:             "data should be uploaded even when metadata is nil",
+			metadata:         nil,
+			contents:         newStringReadSeeker("contents"),
+			log:              newStringReadSeeker("log"),
+			podVolumeBackup:  newStringReadSeeker("podVolumeBackup"),
+			snapshots:        newStringReadSeeker("snapshots"),
+			resourceList:     newStringReadSeeker("resourceList"),
+			backupVolumeInfo: newStringReadSeeker("backupVolumeInfo"),
+			expectedErr:      "",
 			expectedKeys: []string{
 				"backups/backup-1/backup-1.tar.gz",
 				"backups/backup-1/backup-1-logs.gz",
 				"backups/backup-1/backup-1-podvolumebackups.json.gz",
 				"backups/backup-1/backup-1-volumesnapshots.json.gz",
 				"backups/backup-1/backup-1-resource-list.json.gz",
+				"backups/backup-1/backup-1-volumeinfos.json.gz",
 			},
 		},
 	}
@@ -345,6 +356,7 @@ func TestPutBackup(t *testing.T) {
 				VolumeSnapshots:      tc.snapshots,
 				BackupItemOperations: tc.backupItemOperations,
 				BackupResourceList:   tc.resourceList,
+				BackupVolumeInfo:     tc.backupVolumeInfo,
 			}
 			err := harness.PutBackup(backupInfo)
 
@@ -1045,6 +1057,78 @@ func TestNewObjectBackupStoreGetterConfig(t *testing.T) {
 	}
 }
 
+func TestGetBackupVolumeInfos(t *testing.T) {
+	tests := []struct {
+		name           string
+		volumeInfo     *volume.VolumeInfosV1
+		expectedErr    string
+		expectedResult []volume.VolumeInfoV1
+	}{
+		{
+			name: "No VolumeInfos, expect no error.",
+		},
+		{
+			name: "Invalid VolumeInfo version, should fail.",
+			volumeInfo: &volume.VolumeInfosV1{
+				Version: "invalid",
+			},
+			expectedErr: "unknown backup VolumeInfo version: invalid",
+		},
+		{
+			name: "Valid VolumeInfo version, should pass.",
+			volumeInfo: &volume.VolumeInfosV1{
+				Version: "1",
+				Infos: []volume.VolumeInfoV1{
+					{
+						PVCName:           "pvcName",
+						PVName:            "pvName",
+						Skipped:           true,
+						SnapshotDataMoved: false,
+					},
+				},
+			},
+			expectedResult: []volume.VolumeInfoV1{
+				{
+					PVCName:           "pvcName",
+					PVName:            "pvName",
+					Skipped:           true,
+					SnapshotDataMoved: false,
+				},
+			},
+		},
+	}
+
+	harness := newObjectBackupStoreTestHarness("test-bucket", "")
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.volumeInfo != nil {
+				obj := new(bytes.Buffer)
+				gzw := gzip.NewWriter(obj)
+
+				require.NoError(t, json.NewEncoder(gzw).Encode(tc.volumeInfo))
+				require.NoError(t, gzw.Close())
+				harness.objectStore.PutObject(harness.bucket, "backups/test-backup/test-backup-volumeinfos.json.gz", obj)
+			}
+
+			result, err := harness.GetBackupVolumeInfos("test-backup")
+			if tc.expectedErr != "" {
+				require.Equal(t, tc.expectedErr, err.Error())
+			} else {
+				if err != nil {
+					fmt.Println(err.Error())
+				}
+				require.NoError(t, err)
+			}
+
+			if len(tc.expectedResult) > 0 {
+				require.Equal(t, tc.expectedResult, result.VolumeInfosV1)
+			}
+
+		})
+	}
+}
+
 func encodeToBytes(obj runtime.Object) []byte {
 	res, err := encode.Encode(obj, "json")
 	if err != nil {