ebs br: serialize volume backup deletion by GC

Signed-off-by: BornChanger <dawn_catcher@126.com>
pingcap · Dec 11, 2023 · 4b3be4c · 4b3be4c
1 parent 3863d3e
commit 4b3be4c
Showing 1 changed file with 35 additions and 25 deletions.
diff --git a/pkg/fedvolumebackup/backupschedule/backup_schedule_manager.go b/pkg/fedvolumebackup/backupschedule/backup_schedule_manager.go
@@ -258,18 +258,25 @@ func (bm *backupScheduleManager) backupGCByMaxReservedTime(vbs *v1alpha1.VolumeB
 		return
 	}
 
-	for _, backup := range expiredBackups {
-		// delete the expired backup
-		if err = bm.deps.FedVolumeBackupControl.DeleteVolumeBackup(backup); err != nil {
-			klog.Errorf("backup schedule %s/%s gc backup %s failed, err %v", ns, bsName, backup.GetName(), err)
+	// In order to avoid throttling, we choose to do delete volumebackup one by one.
+	// Delete the oldest expired backup
+	if len(expiredBackups) > 0 {
+		backup := expiredBackups[0]
+		if backup.DeletionTimestamp != nil {
+			klog.Infof("Deletion is ongoing for backup schedule %s/%s, backup %s", ns, bsName, backup.GetName())
 			return
-		}
-		klog.Infof("backup schedule %s/%s gc backup %s success", ns, bsName, backup.GetName())
-	}
+		} else {
+			if err = bm.deps.FedVolumeBackupControl.DeleteVolumeBackup(backup); err != nil {
+				klog.Errorf("backup schedule %s/%s gc backup %s failed, err %v", ns, bsName, backup.GetName(), err)
+				return
+			}
+			klog.Infof("backup schedule %s/%s gc backup %s success", ns, bsName, backup.GetName())
 
-	if len(expiredBackups) == len(backupsList) && len(expiredBackups) > 0 {
-		// All backups have been deleted, so the last backup information in the backupSchedule should be reset
-		bm.resetLastBackup(vbs)
+			if len(expiredBackups) == 1 {
+				// All backups have been deleted, so the last backup information in the backupSchedule should be reset
+				bm.resetLastBackup(vbs)
+			}
+		}
 	}
 }
 
@@ -346,23 +353,26 @@ func (bm *backupScheduleManager) backupGCByMaxBackups(vbs *v1alpha1.VolumeBackup
 	}
 
 	sort.Sort(byCreateTimeDesc(backupsList))
-	var deleteCount int
-	for i, backup := range backupsList {
-		if i < int(*vbs.Spec.MaxBackups) {
-			continue
-		}
-		// delete the backup
-		if err := bm.deps.FedVolumeBackupControl.DeleteVolumeBackup(backup); err != nil {
-			klog.Errorf("backup schedule %s/%s gc backup %s failed, err %v", ns, bsName, backup.GetName(), err)
+
+	// In order to avoid throttling, we choose to do delete volumebackup one by one.
+	// Delete the oldest expired backup
+	if len(backupsList) > int(*vbs.Spec.MaxBackups) {
+		backup := backupsList[int(*vbs.Spec.MaxBackups)]
+		if backup.DeletionTimestamp != nil {
+			klog.Infof("Deletion is ongoing for backup schedule %s/%s, backup %s", ns, bsName, backup.GetName())
 			return
-		}
-		deleteCount += 1
-		klog.Infof("backup schedule %s/%s gc backup %s success", ns, bsName, backup.GetName())
-	}
+		} else {
+			if err = bm.deps.FedVolumeBackupControl.DeleteVolumeBackup(backup); err != nil {
+				klog.Errorf("backup schedule %s/%s gc backup %s failed, err %v", ns, bsName, backup.GetName(), err)
+				return
+			}
+			klog.Infof("backup schedule %s/%s gc backup %s success", ns, bsName, backup.GetName())
 
-	if deleteCount == len(backupsList) && deleteCount > 0 {
-		// All backups have been deleted, so the last backup information in the backupSchedule should be reset
-		bm.resetLastBackup(vbs)
+			if len(backupsList) == 1 {
+				// All backups have been deleted, so the last backup information in the backupSchedule should be reset
+				bm.resetLastBackup(vbs)
+			}
+		}
 	}
 }