feat: calc the backup size from snapshot storage usage (#4819)

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

@@ -27,7 +27,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/dustin/go-humanize"
 	backupUtil "github.com/pingcap/tidb-operator/cmd/backup-manager/app/util"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/backup/constants"
@@ -92,23 +91,13 @@ func (bo *Options) backupData(
 		logCallback = func(line string) {
 			if strings.Contains(line, successTag) {
 				extract := strings.Split(line, successTag)[1]
-				sizeStr := regexp.MustCompile(`size=(\d+)`).FindString(extract)
-				size := strings.ReplaceAll(sizeStr, "size=", "")
 				tsStr := regexp.MustCompile(`resolved_ts=(\d+)`).FindString(extract)
 				ts := strings.ReplaceAll(tsStr, "resolved_ts=", "")
-				klog.Infof("%s size: %s, resolved_ts: %s", successTag, size, ts)
+				klog.Infof("%s resolved_ts: %s", successTag, ts)
 
-				backupSize, err := strconv.ParseInt(size, 10, 64)
-				if err != nil {
-					klog.Warningf("Failed to parse BackupSize %s, %v", size, err)
-				}
-				backupSize = backupSize << 30 // Convert GiB to bytes.
-				backupSizeReadable := humanize.Bytes(uint64(backupSize))
 				progress := 100.0
 				if err := statusUpdater.Update(backup, nil, &controller.BackupUpdateStatus{
 					CommitTs:           &ts,
-					BackupSize:         &backupSize,
-					BackupSizeReadable: &backupSizeReadable,
 					ProgressStep:       &progressStep,
 					Progress:           &progress,
 					ProgressUpdateTime: &metav1.Time{Time: time.Now()},

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

@@ -321,11 +321,21 @@ func (bm *Manager) performBackup(ctx context.Context, backup *v1alpha1.Backup, d
 	var updateStatus *controller.BackupUpdateStatus
 	switch bm.Mode {
 	case string(v1alpha1.BackupModeVolumeSnapshot):
-		// In volume snapshot mode, commitTS and size have been updated according to the
-		// br command output, so we don't need to update them here.
+		// In volume snapshot mode, commitTS have been updated according to the
+		// br command output, so we don't need to update it here.
+		backupSize, err := util.CalcVolSnapBackupSize(ctx, backup.Spec.StorageProvider)
+
+		if err != nil {
+			klog.Warningf("Failed to calc volume snapshot backup size %d bytes, %v", backupSize, err)
+		}
+
+		backupSizeReadable := humanize.Bytes(uint64(backupSize))
+
 		updateStatus = &controller.BackupUpdateStatus{
-			TimeStarted:   &metav1.Time{Time: started},
-			TimeCompleted: &metav1.Time{Time: time.Now()},
+			TimeStarted:        &metav1.Time{Time: started},
+			TimeCompleted:      &metav1.Time{Time: time.Now()},
+			BackupSize:         &backupSize,
+			BackupSizeReadable: &backupSizeReadable,
 		}
 	default:
 		backupMeta, err := util.GetBRMetaData(ctx, backup.Spec.StorageProvider)

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

@@ -16,12 +16,15 @@ package clean
 import (
 	"context"
 	"fmt"
+	"sort"
 
+	"github.com/dustin/go-humanize"
 	"github.com/pingcap/tidb-operator/cmd/backup-manager/app/util"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	listers "github.com/pingcap/tidb-operator/pkg/client/listers/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/controller"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	errorutils "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/klog/v2"
 )
@@ -72,8 +75,25 @@ func (bm *Manager) performCleanBackup(ctx context.Context, backup *v1alpha1.Back
 	var errs []error
 	var err error
 	// volume-snapshot backup requires to delete the snapshot firstly, then delete the backup meta file
+	// volume-snapshot is incremental snapshot per volume. Any backup deletion will take effects on next volume-snapshot backup
+	// we need update backup size of the impacted the volume-snapshot backup.
 	if backup.Spec.Mode == v1alpha1.BackupModeVolumeSnapshot {
+		nextNackup := bm.getNextBackup(ctx, backup)
+		if nextNackup == nil {
+			klog.Errorf("get next backup for cluster %s backup is nil", bm)
+		}
+
+		// clean backup will delete all vol snapshots
 		err = bm.cleanBackupMetaWithVolSnapshots(ctx, backup)
+		if err != nil {
+			klog.Errorf("delete backup %s for cluster %s backup failure", backup.Name, bm)
+		}
+
+		// update the next backup size
+		if nextNackup != nil {
+			bm.updateVolumeSnapshotBackupSize(ctx, nextNackup)
+		}
+
 	} else {
 		if backup.Spec.BR != nil {
 			err = bm.cleanBRRemoteBackupData(ctx, backup)
@@ -102,3 +122,60 @@ func (bm *Manager) performCleanBackup(ctx context.Context, backup *v1alpha1.Back
 		Status: corev1.ConditionTrue,
 	}, nil)
 }
+
+// getNextBackup to get next backup sorted by start time
+func (bm *Manager) getNextBackup(ctx context.Context, backup *v1alpha1.Backup) *v1alpha1.Backup {
+	var err error
+	bks, err := bm.backupLister.Backups(backup.Namespace).List(labels.Everything())
+	if err != nil {
+		return nil
+	}
+
+	// sort the backup list by TimeStarted, since volume snapshot is point-in-time (start time) backup
+	sort.Slice(bks, func(i, j int) bool {
+		return bks[i].Status.TimeStarted.Before(&bks[j].Status.TimeStarted)
+	})
+
+	for i, bk := range bks {
+		if backup.Name == bk.Name {
+			return bm.getVolumeSnapshotBackup(bks[i+1:])
+		}
+	}
+
+	return nil
+}
+
+// getVolumeSnapshotBackup get the first volume-snapshot backup from backup list, which may contain non-volume snapshot
+func (bm *Manager) getVolumeSnapshotBackup(backups []*v1alpha1.Backup) *v1alpha1.Backup {
+	for _, bk := range backups {
+		if bk.Spec.Mode == v1alpha1.BackupModeVolumeSnapshot {
+			return bk
+		}
+	}
+
+	// reach end of backup list, there is no volume snapshot backups
+	return nil
+}
+
+// updateVolumeSnapshotBackupSize update a volume-snapshot backup size
+func (bm *Manager) updateVolumeSnapshotBackupSize(ctx context.Context, backup *v1alpha1.Backup) error {
+	var updateStatus *controller.BackupUpdateStatus
+
+	backupSize, err := util.CalcVolSnapBackupSize(ctx, backup.Spec.StorageProvider)
+
+	if err != nil {
+		klog.Warningf("Failed to parse BackupSize %d KB, %v", backupSize, err)
+	}
+
+	backupSizeReadable := humanize.Bytes(uint64(backupSize))
+
+	updateStatus = &controller.BackupUpdateStatus{
+		BackupSize:         &backupSize,
+		BackupSizeReadable: &backupSizeReadable,
+	}
+
+	return bm.StatusUpdater.Update(backup, &v1alpha1.BackupCondition{
+		Type:   v1alpha1.BackupComplete,
+		Status: corev1.ConditionTrue,
+	}, updateStatus)
+}