ebs br: serial execution of volume backup gc

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)
+			}
+		}
 	}
 }
 

pkg/fedvolumebackup/backupschedule/backup_schedule_manager_test.go

@@ -123,14 +123,14 @@ func TestManager(t *testing.T) {
 	bs.Spec.MaxBackups = pointer.Int32Ptr(5)
 	err = m.Sync(bs)
 	g.Expect(err).Should(BeNil())
-	helper.checkBacklist(bs.Namespace, 5)
+	helper.checkBacklist(bs.Namespace, 9)
 
 	t.Log("test setting MaxReservedTime")
 	bs.Spec.MaxBackups = nil
 	bs.Spec.MaxReservedTime = pointer.StringPtr("71h")
 	err = m.Sync(bs)
 	g.Expect(err).Should(BeNil())
-	helper.checkBacklist(bs.Namespace, 3)
+	helper.checkBacklist(bs.Namespace, 8)
 }
 
 func TestGetLastScheduledTime(t *testing.T) {