Retrieve DataUpload into backup result ConfigMap during volume snapsh…

…ot restore.

Fix issue vmware-tanzu#6117.
Add CSI plugin needs builder functions.

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

changelogs/unreleased/6410-blackpiglet

@@ -0,0 +1 @@
+Retrieve DataUpload into backup result ConfigMap during volume snapshot restore.

pkg/apis/velero/v1/labels_annotations.go

@@ -75,4 +75,18 @@ const (
 	// ResourceTimeoutAnnotation is the annotation key used to carry the global resource
 	// timeout value for backup to plugins.
 	ResourceTimeoutAnnotation = "velero.io/resource-timeout"
+
+	// PVCNameLabel is the label key used to identify the the PVC's namespace and name.
+	// The format is <namespace>/<name>.
+	PVCNamespaceNameLabel = "velero.io/pvc-namespace-name"
+
+	// DynamicPVRestoreLabel is the label key for dynamic PV restore
+	DynamicPVRestoreLabel = "velero.io/dynamic-pv-restore"
+)
+
+type AsyncOperationIDPrefix string
+
+const (
+	AsyncOperationIDPrefixDataDownload AsyncOperationIDPrefix = "dd-"
+	AsyncOperationIDPrefixDataUpload   AsyncOperationIDPrefix = "du-"
 )

pkg/builder/data_download_builder.go

@@ -0,0 +1,64 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package builder
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+)
+
+// DataDownloadBuilder builds DataDownload objects.
+type DataDownloadBuilder struct {
+	object *velerov2alpha1api.DataDownload
+}
+
+// ForDataDownload is the constructor of DataDownloadBuilder
+func ForDataDownload(namespace, name string) *DataDownloadBuilder {
+	return &DataDownloadBuilder{
+		object: &velerov2alpha1api.DataDownload{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "DataDownload",
+				APIVersion: velerov2alpha1api.SchemeGroupVersion.String(),
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      name,
+				Namespace: namespace,
+			},
+		},
+	}
+}
+
+// Result returns the built DataDownload
+func (b *DataDownloadBuilder) Result() *velerov2alpha1api.DataDownload {
+	return b.object
+}
+
+// TargetVolume sets DataDownload's spec.targetVolume
+func (b *DataDownloadBuilder) TargetVolume(targetVolume velerov2alpha1api.TargetVolumeSpec) *DataDownloadBuilder {
+	b.object.Spec.TargetVolume = targetVolume
+	return b
+}
+
+// ObjectMeta applies functional options to the DataDownload's ObjectMeta.
+func (b *DataDownloadBuilder) ObjectMeta(opts ...ObjectMetaOpt) *DataDownloadBuilder {
+	for _, opt := range opts {
+		opt(b.object)
+	}
+
+	return b
+}

pkg/builder/object_meta.go

@@ -153,3 +153,10 @@ func WithManagedFields(val []metav1.ManagedFieldsEntry) func(obj metav1.Object)
 		obj.SetManagedFields(val)
 	}
 }
+
+// WithOwnerReference is a functional option that applies the specified OwnerReference to an object.
+func WithOwnerReference(val []metav1.OwnerReference) func(obj metav1.Object) {
+	return func(obj metav1.Object) {
+		obj.SetOwnerReferences(val)
+	}
+}

pkg/builder/persistent_volume_claim_builder.go

@@ -18,6 +18,7 @@ package builder
 
 import (
 	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -73,3 +74,39 @@ func (b *PersistentVolumeClaimBuilder) Phase(phase corev1api.PersistentVolumeCla
 	b.object.Status.Phase = phase
 	return b
 }
+
+// RequestResource sets the PersistentVolumeClaim's spec.Resources.Requests.
+func (b *PersistentVolumeClaimBuilder) RequestResource(requests corev1api.ResourceList) *PersistentVolumeClaimBuilder {
+	if b.object.Spec.Resources.Requests == nil {
+		b.object.Spec.Resources.Requests = make(map[corev1api.ResourceName]resource.Quantity)
+	}
+	b.object.Spec.Resources.Requests = requests
+	return b
+}
+
+// LimitResource sets the PersistentVolumeClaim's spec.Resources.Limits.
+func (b *PersistentVolumeClaimBuilder) LimitResource(limits corev1api.ResourceList) *PersistentVolumeClaimBuilder {
+	if b.object.Spec.Resources.Limits == nil {
+		b.object.Spec.Resources.Limits = make(map[corev1api.ResourceName]resource.Quantity)
+	}
+	b.object.Spec.Resources.Limits = limits
+	return b
+}
+
+// DataSource sets the PersistentVolumeClaim's spec.DataSource.
+func (b *PersistentVolumeClaimBuilder) DataSource(dataSource *corev1api.TypedLocalObjectReference) *PersistentVolumeClaimBuilder {
+	b.object.Spec.DataSource = dataSource
+	return b
+}
+
+// DataSourceRef sets the PersistentVolumeClaim's spec.DataSourceRef.
+func (b *PersistentVolumeClaimBuilder) DataSourceRef(dataSourceRef *corev1api.TypedLocalObjectReference) *PersistentVolumeClaimBuilder {
+	b.object.Spec.DataSourceRef = dataSourceRef
+	return b
+}
+
+// Selector sets the PersistentVolumeClaim's spec.Selector.
+func (b *PersistentVolumeClaimBuilder) Selector(labelSelector *metav1.LabelSelector) *PersistentVolumeClaimBuilder {
+	b.object.Spec.Selector = labelSelector
+	return b
+}

pkg/cmd/server/server.go

@@ -521,6 +521,7 @@ High priorities:
   - VolumeSnapshotContents are needed as they contain the handle to the volume snapshot in the
     storage provider
   - VolumeSnapshots are needed to create PVCs using the VolumeSnapshot as their data source.
+  - DataUploads need to restore before PVC for Snapshot DataMover to work, because PVC needs the DataUploadResults to create DataDownloads.
   - PVs go before PVCs because PVCs depend on them.
   - PVCs go before pods or controllers so they can be mounted as volumes.
   - Service accounts go before secrets so service account token secrets can be filled automatically.
@@ -551,6 +552,7 @@ var defaultRestorePriorities = restore.Priorities{
 		"volumesnapshotclass.snapshot.storage.k8s.io",
 		"volumesnapshotcontents.snapshot.storage.k8s.io",
 		"volumesnapshots.snapshot.storage.k8s.io",
+		"datauploads.velero.io",
 		"persistentvolumes",
 		"persistentvolumeclaims",
 		"serviceaccounts",

pkg/restore/dataupload_retrieve_action.go

@@ -0,0 +1,104 @@
+/*
+Copyright 2020 the Velero contributors.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package restore
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/label"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+)
+
+type DataUploadRetrieveAction struct {
+	logger          logrus.FieldLogger
+	configMapClient corev1client.ConfigMapInterface
+}
+
+func NewDataUploadRetrieveAction(logger logrus.FieldLogger, configMapClient corev1client.ConfigMapInterface) *DataUploadRetrieveAction {
+	return &DataUploadRetrieveAction{
+		logger:          logger,
+		configMapClient: configMapClient,
+	}
+}
+
+func (d *DataUploadRetrieveAction) AppliesTo() (velero.ResourceSelector, error) {
+	return velero.ResourceSelector{
+		IncludedResources: []string{"datauploads.velero.io"},
+	}, nil
+}
+
+func (d *DataUploadRetrieveAction) Execute(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+	d.logger.Info("Executing DataUploadRetrieveAction")
+
+	dataUpload := velerov2alpha1.DataUpload{}
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(input.ItemFromBackup.UnstructuredContent(), &dataUpload); err != nil {
+		d.logger.Errorf("unable to convert unstructured item to DataUpload: %s", err.Error())
+		return nil, errors.Wrap(err, "unable to convert unstructured item to DataUpload.")
+	}
+
+	dataUploadResult := velerov2alpha1.DataUploadResult{
+		BackupStorageLocation: dataUpload.Spec.BackupStorageLocation,
+		DataMover:             dataUpload.Spec.DataMover,
+		SnapshotID:            dataUpload.Status.SnapshotID,
+		SourceNamespace:       dataUpload.Spec.SourceNamespace,
+		DataMoverResult:       dataUpload.Status.DataMoverResult,
+	}
+
+	jsonBytes, err := json.Marshal(dataUploadResult)
+	if err != nil {
+		d.logger.Errorf("fail to convert DataUploadResult to JSON: %s", err.Error())
+		return nil, errors.Wrap(err, "fail to convert DataUploadResult to JSON")
+	}
+
+	cm := corev1api.ConfigMap{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "ConfigMap",
+			APIVersion: corev1api.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: dataUpload.Name + "-",
+			Namespace:    dataUpload.Namespace,
+			Labels: map[string]string{
+				velerov1api.RestoreUIDLabel:       label.GetValidName(string(input.Restore.UID)),
+				velerov1api.PVCNamespaceNameLabel: dataUpload.Spec.SourceNamespace + "." + dataUpload.Spec.SourcePVC,
+			},
+		},
+		Data: map[string]string{
+			string(input.Restore.UID): string(jsonBytes),
+		},
+	}
+
+	_, err = d.configMapClient.Create(context.Background(), &cm, metav1.CreateOptions{})
+	if err != nil {
+		d.logger.Errorf("fail to create DataUploadResult ConfigMap %s/%s: %s", cm.Namespace, cm.Name, err.Error())
+		return nil, errors.Wrap(err, "fail to create DataUploadResult ConfigMap")
+	}
+
+	return &velero.RestoreItemActionExecuteOutput{
+		SkipRestore: true,
+	}, nil
+}

pkg/restore/dataupload_retrieve_action_test.go

@@ -0,0 +1,86 @@
+/*
+Copyright 2020 the Velero contributors.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package restore
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+
+	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func TestDataUploadRetrieveActionExectue(t *testing.T) {
+	tests := []struct {
+		name                     string
+		dataUpload               *velerov2alpha1.DataUpload
+		restore                  *velerov1.Restore
+		expectedDataUploadResult *corev1.ConfigMap
+		expectedErr              string
+	}{
+		{
+			name:                     "DataUploadRetrieve Action test",
+			dataUpload:               builder.ForDataUpload("velero", "testDU").SourceNamespace("testNamespace").SourcePVC("testPVC").Result(),
+			restore:                  builder.ForRestore("velero", "testRestore").ObjectMeta(builder.WithUID("testingUID")).Result(),
+			expectedDataUploadResult: builder.ForConfigMap("velero", "").ObjectMeta(builder.WithGenerateName("testDU-"), builder.WithLabels(velerov1.PVCNamespaceNameLabel, "testNamespace.testPVC", velerov1.RestoreUIDLabel, "testingUID")).Data("testingUID", `{"backupStorageLocation":"","sourceNamespace":"testNamespace"}`).Result(),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			logger := velerotest.NewLogger()
+			cmClient := fake.NewSimpleClientset()
+
+			var unstructuredDataUpload map[string]interface{}
+			if tc.dataUpload != nil {
+				var err error
+				unstructuredDataUpload, err = runtime.DefaultUnstructuredConverter.ToUnstructured(tc.dataUpload)
+				require.NoError(t, err)
+			}
+			input := velero.RestoreItemActionExecuteInput{
+				Restore:        tc.restore,
+				ItemFromBackup: &unstructured.Unstructured{Object: unstructuredDataUpload},
+			}
+
+			action := NewDataUploadRetrieveAction(logger, cmClient.CoreV1().ConfigMaps("velero"))
+			_, err := action.Execute(&input)
+			if tc.expectedErr != "" {
+				require.Equal(t, tc.expectedErr, err.Error())
+			}
+			require.NoError(t, err)
+
+			if tc.expectedDataUploadResult != nil {
+				cmList, err := cmClient.CoreV1().ConfigMaps("velero").List(context.Background(), metav1.ListOptions{
+					LabelSelector: fmt.Sprintf("%s=%s,%s=%s", velerov1.RestoreUIDLabel, "testingUID", velerov1.PVCNamespaceNameLabel, tc.dataUpload.Spec.SourceNamespace+"."+tc.dataUpload.Spec.SourcePVC),
+				})
+				require.NoError(t, err)
+				require.Equal(t, *tc.expectedDataUploadResult, cmList.Items[0])
+			}
+		})
+	}
+}

pkg/restore/restore.go

@@ -535,6 +535,20 @@ func (ctx *restoreContext) execute() (results.Result, results.Result) {
 	// Close the progress update channel.
 	quit <- struct{}{}
 
+	// Clean the DataUploadResult ConfigMaps
+	defer func() {
+		opts := []crclient.DeleteAllOfOption{
+			crclient.InNamespace(""),
+			crclient.MatchingLabels{
+				velerov1api.RestoreUIDLabel: string(ctx.restore.UID),
+			},
+		}
+		err := ctx.kbClient.DeleteAllOf(go_context.Background(), &v1.ConfigMap{}, opts...)
+		if err != nil {
+			ctx.log.Errorf("Fail to batch delete DataUploadResult ConfigMaps for restore %s: %s", ctx.restore.Name, err.Error())
+		}
+	}()
+
 	// Do a final progress update as stopping the ticker might have left last few
 	// updates from taking place.
 	updated := ctx.restore.DeepCopy()