Fixing volume tagging

cmd/backup-manager/app/clean/clean.go

@@ -105,7 +105,11 @@ func (bo *Options) deleteSnapshotsAndBackupMeta(ctx context.Context, backup *v1a
 	}
 
 	//2. delete the snapshot
-	if err = bo.deleteVolumeSnapshots(metaInfo, backup.Spec.CleanOption.SnapshotsDeleteRatio); err != nil {
+	var deleteRatio = 1.0
+	if backup.Spec.CleanOption != nil {
+		deleteRatio = backup.Spec.CleanOption.SnapshotsDeleteRatio
+	}
+	if err = bo.deleteVolumeSnapshots(metaInfo, deleteRatio); err != nil {
 		klog.Errorf("delete volume snapshot failure, a mannual check or delete aciton require.")
 		return err
 	}

pkg/backup/restore/restore_manager.go

@@ -154,50 +154,6 @@ func (rm *restoreManager) syncRestoreJob(restore *v1alpha1.Restore) error {
 			if !tc.AllTiKVsAreAvailable() {
 				return controller.RequeueErrorf("restore %s/%s: waiting for all TiKVs are available in tidbcluster %s/%s", ns, name, tc.Namespace, tc.Name)
 			} else {
-				sel, err := label.New().Instance(tc.Name).TiKV().Selector()
-				if err != nil {
-					rm.statusUpdater.Update(restore, &v1alpha1.RestoreCondition{
-						Type:    v1alpha1.RestoreRetryFailed,
-						Status:  corev1.ConditionTrue,
-						Reason:  "BuildTiKVSelectorFailed",
-						Message: err.Error(),
-					}, nil)
-					return err
-				}
-
-				pvs, err := rm.deps.PVLister.List(sel)
-				if err != nil {
-					rm.statusUpdater.Update(restore, &v1alpha1.RestoreCondition{
-						Type:    v1alpha1.RestoreRetryFailed,
-						Status:  corev1.ConditionTrue,
-						Reason:  "ListPVsFailed",
-						Message: err.Error(),
-					}, nil)
-					return err
-				}
-
-				s, reason, err := snapshotter.NewSnapshotterForRestore(restore.Spec.Mode, rm.deps)
-				if err != nil {
-					rm.statusUpdater.Update(restore, &v1alpha1.RestoreCondition{
-						Type:    v1alpha1.RestoreRetryFailed,
-						Status:  corev1.ConditionTrue,
-						Reason:  reason,
-						Message: err.Error(),
-					}, nil)
-					return err
-				}
-
-				err = s.AddVolumeTags(pvs)
-				if err != nil {
-					rm.statusUpdater.Update(restore, &v1alpha1.RestoreCondition{
-						Type:    v1alpha1.RestoreRetryFailed,
-						Status:  corev1.ConditionTrue,
-						Reason:  "AddVolumeTagFailed",
-						Message: err.Error(),
-					}, nil)
-					return err
-				}
-
 				return rm.statusUpdater.Update(restore, &v1alpha1.RestoreCondition{
 					Type:   v1alpha1.RestoreTiKVComplete,
 					Status: corev1.ConditionTrue,

pkg/backup/snapshotter/snapshotter.go

@@ -175,6 +175,11 @@ func (s *BaseSnapshotter) prepareRestoreMetadata(r *v1alpha1.Restore, csb *Cloud
 		return reason, err
 	}
 
+	// add volume tags here
+	if err := m.snapshotter.AddVolumeTags(csb.Kubernetes.PVs); err != nil {
+		return "volume_tag_adding_failed", err
+	}
+
 	return "", nil
 }
 

pkg/backup/snapshotter/snapshotter_aws.go

@@ -30,6 +30,7 @@ const (
 	CloudAPIConcurrency = 8
 	PVCTagKey           = "CSIVolumeName"
 	PodTagKey           = "kubernetes.io/created-for/pvc/name"
+	PodNSTagKey         = "kubernetes.io/created-for/pvc/namespace"
 )
 
 // AWSSnapshotter is the snapshotter for creating snapshots from volumes (during a backup)
@@ -102,15 +103,12 @@ func (s *AWSSnapshotter) AddVolumeTags(pvs []*corev1.PersistentVolume) error {
 	resourcesTags := make(map[string]util.TagMap)
 
 	for _, pv := range pvs {
-		podName := pv.GetAnnotations()[label.AnnPodNameKey]
-		pvcName := pv.GetName()
-		volId := pv.Spec.CSI.VolumeHandle
-
 		tags := make(map[string]string)
-		tags[PVCTagKey] = pvcName
-		tags[PodTagKey] = podName
+		tags[PVCTagKey] = pv.GetName()
+		tags[PodTagKey] = pv.GetAnnotations()[label.AnnPodNameKey]
+		tags[PodNSTagKey] = pv.GetLabels()[label.NamespaceLabelKey]
 
-		resourcesTags[volId] = tags
+		resourcesTags[pv.GetAnnotations()[constants.AnnRestoredVolumeID]] = tags
 	}
 	ec2Session, err := util.NewEC2Session(CloudAPIConcurrency)
 	if err != nil {

pkg/backup/snapshotter/snapshotter_test.go

@@ -906,8 +906,8 @@ func TestPrepareRestoreMetadata(t *testing.T) {
 	_ = json.Unmarshal([]byte(meta), csb)
 	// happy path
 	reason, err = s.PrepareRestoreMetadata(restore, csb)
-	require.Empty(t, reason)
-	require.NoError(t, err)
+	require.Equal(t, reason, "volume_tag_adding_failed")
+	require.Error(t, err)
 }
 
 func TestProcessCSBPVCsAndPVs(t *testing.T) {