Add backup VolumeInfo's object storage's put and get methods.

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

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 {