Delete moved snapshots when the backup is deleted

This commit introduces a deleteItemAction which writes a temporary configmap to
record the snapshot info so that the controller can trigger repo manager
to remove the snapshot

This process is a bit chatty and we should consider to refactor the code
so it's easier to connect to the repo directly in the DIA

Signed-off-by: Daniel Jiang <jiangd@vmware.com>

changelogs/unreleased/6547-reasonerjt

@@ -0,0 +1 @@
+Delete moved snapshots when the backup is deleted

pkg/apis/velero/v1/labels_annotations.go

@@ -57,6 +57,10 @@ const (
 	// DataUploadLabel is the label key used to identify the dataupload for snapshot backup pod
 	DataUploadLabel = "velero.io/data-upload"
 
+	// DataUploadSnapshotInfoLabel is used to identify the configmap that contains the snapshot info of a data upload
+	// normally the value of the label should the "true" or "false"
+	DataUploadSnapshotInfoLabel = "velero.io/data-upload-snapshot-info"
+
 	// DataDownloadLabel is the label key used to identify the datadownload for snapshot restore pod
 	DataDownloadLabel = "velero.io/data-download"
 

pkg/cmd/server/plugin/plugin.go

@@ -21,6 +21,8 @@ import (
 	"github.com/spf13/cobra"
 	apiextensions "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 
+	"github.com/vmware-tanzu/velero/pkg/datamover"
+
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/backup"
 	"github.com/vmware-tanzu/velero/pkg/client"
@@ -59,7 +61,9 @@ func NewCommand(f client.Factory) *cobra.Command {
 				RegisterRestoreItemAction("velero.io/apiservice", newAPIServiceRestoreItemAction).
 				RegisterRestoreItemAction("velero.io/admission-webhook-configuration", newAdmissionWebhookConfigurationAction).
 				RegisterRestoreItemAction("velero.io/secret", newSecretRestoreItemAction(f)).
-				RegisterRestoreItemAction("velero.io/dataupload", newDataUploadRetrieveAction(f))
+				RegisterRestoreItemAction("velero.io/dataupload", newDataUploadRetrieveAction(f)).
+				RegisterDeleteItemAction("velero.io/dataupload-delete", newDateUploadDeleteItemAction(f))
+
 			if !features.IsEnabled(velerov1api.APIGroupVersionsFeatureFlag) {
 				// Do not register crd-remap-version BIA if the API Group feature flag is enabled, so that the v1 CRD can be backed up
 				pluginServer = pluginServer.RegisterBackupItemAction("velero.io/crd-remap-version", newRemapCRDVersionAction(f))
@@ -256,3 +260,13 @@ func newDataUploadRetrieveAction(f client.Factory) plugincommon.HandlerInitializ
 		return restore.NewDataUploadRetrieveAction(logger, client), nil
 	}
 }
+
+func newDateUploadDeleteItemAction(f client.Factory) plugincommon.HandlerInitializer {
+	return func(logger logrus.FieldLogger) (interface{}, error) {
+		client, err := f.KubebuilderClient()
+		if err != nil {
+			return nil, err
+		}
+		return datamover.NewDataUploadDeleteAction(logger, client), nil
+	}
+}

pkg/controller/backup_deletion_controller.go

@@ -22,6 +22,11 @@ import (
 	"fmt"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
+
+	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+
 	jsonpatch "github.com/evanphx/json-patch"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -314,6 +319,33 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		}
 	}
 
+	if boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData) {
+		log.Info("Removing snapshot data by data mover")
+		if deleteErrs := r.deleteMovedSnapshots(ctx, backup); len(deleteErrs) > 0 {
+			for _, err := range deleteErrs {
+				errs = append(errs, err.Error())
+			}
+		}
+		duList := &velerov2alpha1.DataUploadList{}
+		log.Info("Removing local datauploads")
+		if err := r.Client.List(ctx, duList, &client.ListOptions{
+			Namespace: backup.Namespace,
+			LabelSelector: labels.SelectorFromSet(map[string]string{
+				velerov1api.BackupNameLabel: label.GetValidName(backup.Name),
+			}),
+		}); err != nil {
+			log.WithError(err).Error("Error listing datauploads")
+			errs = append(errs, err.Error())
+		} else {
+			for i := range duList.Items {
+				du := duList.Items[i]
+				if err := r.Delete(ctx, &du); err != nil {
+					errs = append(errs, err.Error())
+				}
+			}
+		}
+	}
+
 	if backupStore != nil {
 		log.Info("Removing backup from backup storage")
 		if err := backupStore.DeleteBackup(backup.Name); err != nil {
@@ -469,6 +501,41 @@ func (r *backupDeletionReconciler) deletePodVolumeSnapshots(ctx context.Context,
 	return errs
 }
 
+func (r *backupDeletionReconciler) deleteMovedSnapshots(ctx context.Context, backup *velerov1api.Backup) []error {
+	if r.repoMgr == nil {
+		return nil
+	}
+	list := &corev1.ConfigMapList{}
+	if err := r.Client.List(ctx, list, &client.ListOptions{
+		Namespace: backup.Namespace,
+		LabelSelector: labels.SelectorFromSet(
+			map[string]string{
+				velerov1api.BackupNameLabel:             label.GetValidName(backup.Name),
+				velerov1api.DataUploadSnapshotInfoLabel: "true",
+			}),
+	}); err != nil {
+		return []error{errors.Wrapf(err, "failed to retrieve config for snapshot info")}
+	}
+	var errs []error
+	for i := range list.Items {
+		cm := list.Items[i]
+		snapshot := repository.SnapshotIdentifier{}
+		b, _ := json.Marshal(cm.Data)
+		if err := json.Unmarshal(b, &snapshot); err != nil {
+			errs = append(errs, errors.Wrapf(err, "failed to unmarshal snapshot info"))
+			continue
+		}
+		if err := r.repoMgr.Forget(ctx, snapshot); err != nil {
+			errs = append(errs, errors.Wrapf(err, "failed to delete snapshot %s, namespace: %s", snapshot.SnapshotID, snapshot.VolumeNamespace))
+		}
+		r.logger.Infof("Deleted snapshot %s, namespace: %s, repo type: %s", snapshot.SnapshotID, snapshot.VolumeNamespace, snapshot.RepositoryType)
+		if err := r.Client.Delete(ctx, &cm); err != nil {
+			r.logger.Warnf("Failed to delete snapshot info configmap %s/%s: %v", cm.Namespace, cm.Name, err)
+		}
+	}
+	return errs
+}
+
 func (r *backupDeletionReconciler) patchDeleteBackupRequest(ctx context.Context, req *velerov1api.DeleteBackupRequest, mutate func(*velerov1api.DeleteBackupRequest)) (*velerov1api.DeleteBackupRequest, error) {
 	original := req.DeepCopy()
 	mutate(req)

pkg/controller/data_download_controller.go

@@ -109,7 +109,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 		return ctrl.Result{}, err
 	}
 
-	if dd.Spec.DataMover != "" && dd.Spec.DataMover != dataMoverType {
+	if !datamover.IsBuiltInUploader(dd.Spec.DataMover) {
 		log.WithField("data mover", dd.Spec.DataMover).Info("it is not one built-in data mover which is not supported by Velero")
 		return ctrl.Result{}, nil
 	}

pkg/controller/data_upload_controller.go

@@ -53,10 +53,10 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
-const dataMoverType string = "velero"
-const dataUploadDownloadRequestor string = "snapshot-data-upload-download"
-
-const preparingMonitorFrequency time.Duration = time.Minute
+const (
+	dataUploadDownloadRequestor string        = "snapshot-data-upload-download"
+	preparingMonitorFrequency   time.Duration = time.Minute
+)
 
 // DataUploadReconciler reconciles a DataUpload object
 type DataUploadReconciler struct {
@@ -116,7 +116,7 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, errors.Wrap(err, "getting DataUpload")
 	}
 
-	if du.Spec.DataMover != "" && du.Spec.DataMover != dataMoverType {
+	if !datamover.IsBuiltInUploader(du.Spec.DataMover) {
 		log.WithField("Data mover", du.Spec.DataMover).Debug("it is not one built-in data mover which is not supported by Velero")
 		return ctrl.Result{}, nil
 	}

pkg/datamover/dataupload_delete_action.go

@@ -0,0 +1,86 @@
+package datamover
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"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"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	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/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/repository"
+)
+
+type DataUploadDeleteAction struct {
+	logger logrus.FieldLogger
+	client client.Client
+}
+
+func (d *DataUploadDeleteAction) AppliesTo() (velero.ResourceSelector, error) {
+	return velero.ResourceSelector{
+		IncludedResources: []string{"datauploads.velero.io"},
+	}, nil
+}
+
+func (d *DataUploadDeleteAction) Execute(input *velero.DeleteItemActionExecuteInput) error {
+	d.logger.Infof("Executing DataUploadDeleteAction")
+	du := &velerov2alpha1.DataUpload{}
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(input.Item.UnstructuredContent(), &du); err != nil {
+		return errors.WithStack(errors.Wrapf(err, "failed to convert input.Item from unstructured"))
+	}
+	cm := genConfigmap(input.Backup, *du)
+	if cm == nil {
+		// will not fail the backup deletion
+		return nil
+	}
+	err := d.client.Create(context.Background(), cm)
+	if err != nil {
+		return errors.WithStack(errors.Wrapf(err, "failed to create the configmap for DataUpload %s/%s", du.Namespace, du.Name))
+	}
+	return nil
+}
+
+// generate the configmap which is to be created and used as a way to communicate the snapshot info to the backup deletion controller
+func genConfigmap(bak *velerov1.Backup, du velerov2alpha1.DataUpload) *corev1api.ConfigMap {
+	if !IsBuiltInUploader(du.Spec.DataMover) || du.Status.SnapshotID == "" {
+		return nil
+	}
+	snapshot := repository.SnapshotIdentifier{
+		VolumeNamespace:       du.Spec.SourceNamespace,
+		BackupStorageLocation: bak.Spec.StorageLocation,
+		SnapshotID:            du.Status.SnapshotID,
+		RepositoryType:        GetUploaderType(du.Spec.DataMover),
+	}
+	b, _ := json.Marshal(snapshot)
+	data := make(map[string]string)
+	_ = json.Unmarshal(b, &data)
+	return &corev1api.ConfigMap{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: corev1api.SchemeGroupVersion.String(),
+			Kind:       "ConfigMap",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: bak.Namespace,
+			Name:      fmt.Sprintf("%s-info", du.Name),
+			Labels: map[string]string{
+				velerov1.BackupNameLabel:             bak.Name,
+				velerov1.DataUploadSnapshotInfoLabel: "true",
+			},
+		},
+		Data: data,
+	}
+}
+
+func NewDataUploadDeleteAction(logger logrus.FieldLogger, client client.Client) *DataUploadDeleteAction {
+	return &DataUploadDeleteAction{
+		logger: logger,
+		client: client,
+	}
+}

pkg/datamover/util.go

@@ -23,3 +23,7 @@ func GetUploaderType(dataMover string) string {
 		return dataMover
 	}
 }
+
+func IsBuiltInUploader(dataMover string) bool {
+	return dataMover == "" || dataMover == "velero"
+}

pkg/datamover/util_test.go

@@ -0,0 +1,70 @@
+package datamover
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsBuiltInUploader(t *testing.T) {
+	testcases := []struct {
+		name      string
+		dataMover string
+		want      bool
+	}{
+		{
+			name:      "empty dataMover is builtin",
+			dataMover: "",
+			want:      true,
+		},
+		{
+			name:      "velero dataMover is builtin",
+			dataMover: "velero",
+			want:      true,
+		},
+		{
+			name:      "kopia dataMover is not builtin",
+			dataMover: "kopia",
+			want:      false,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			assert.Equal(tt, tc.want, IsBuiltInUploader(tc.dataMover))
+		})
+	}
+}
+
+func TestGetUploaderType(t *testing.T) {
+	testcases := []struct {
+		name  string
+		input string
+		want  string
+	}{
+		{
+			name:  "empty dataMover is kopia",
+			input: "",
+			want:  "kopia",
+		},
+		{
+			name:  "velero dataMover is kopia",
+			input: "velero",
+			want:  "kopia",
+		},
+		{
+			name:  "kopia dataMover is kopia",
+			input: "kopia",
+			want:  "kopia",
+		},
+		{
+			name:  "restic dataMover is restic",
+			input: "restic",
+			want:  "restic",
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			assert.Equal(tt, tc.want, GetUploaderType(tc.input))
+		})
+	}
+}

pkg/repository/manager.go

@@ -36,18 +36,18 @@ import (
 type SnapshotIdentifier struct {
 	// VolumeNamespace is the namespace of the pod/volume that
 	// the snapshot is for.
-	VolumeNamespace string
+	VolumeNamespace string `json:"volumeNamespace"`
 
 	// BackupStorageLocation is the backup's storage location
 	// name.
-	BackupStorageLocation string
+	BackupStorageLocation string `json:"backupStorageLocation"`
 
 	// SnapshotID is the short ID of the snapshot.
-	SnapshotID string
+	SnapshotID string `json:"snapshotID"`
 
 	// RepositoryType is the type of the repository where the
 	// snapshot is stored
-	RepositoryType string
+	RepositoryType string `json:"repositoryType"`
 }
 
 // Manager manages backup repositories.