From ba2b95697602447be080bc11bbdf5b9d9972b774 Mon Sep 17 00:00:00 2001
From: Chin-Ya Huang <chin-ya.huang@suse.com>
Date: Mon, 29 Jul 2024 15:42:02 +0800
Subject: [PATCH] feat(restore): implememt incremental restore

longhorn/longhorn-6613

Signed-off-by: Chin-Ya Huang <chin-ya.huang@suse.com>
---
 pkg/spdk/engine.go  |  5 ++++-
 pkg/spdk/replica.go | 29 ++++++++++++++++++++++++++++-
 2 files changed, 32 insertions(+), 2 deletions(-)

diff --git a/pkg/spdk/engine.go b/pkg/spdk/engine.go
index 0c04565a..f5e3ef74 100644
--- a/pkg/spdk/engine.go
+++ b/pkg/spdk/engine.go
@@ -1754,7 +1754,10 @@ func (e *Engine) BackupRestore(spdkClient *spdkclient.Client, backupUrl, engineN
 			e.log.Infof("Generating a snapshot name %s for the full restore", snapshotName)
 		}
 	} else {
-		return nil, errors.Errorf("incremental restore is not supported yet")
+		if snapshotName == "" {
+			snapshotName = util.UUID()
+			e.log.Infof("Generating a snapshot name %s for the incremental restore", snapshotName)
+		}
 	}
 
 	resp := &spdkrpc.EngineBackupRestoreResponse{
diff --git a/pkg/spdk/replica.go b/pkg/spdk/replica.go
index fbcae452..11deef7a 100644
--- a/pkg/spdk/replica.go
+++ b/pkg/spdk/replica.go
@@ -1930,14 +1930,18 @@ func (r *Replica) BackupRestore(spdkClient *spdkclient.Client, backupUrl, snapsh
 			return grpcstatus.Errorf(grpccodes.Internal, err.Error())
 		}
 	} else {
+		r.log.Info("Resetting the restore for backup %v", backupUrl)
+
 		var lvolName string
 		var snapshotNameToBeRestored string
 
 		validLastRestoredBackup := r.canDoIncrementalRestore(restore, backupUrl, backupName)
 		if validLastRestoredBackup {
+			r.log.Infof("Starting an incremental restore for backup %v", backupUrl)
 			lvolName = GetReplicaSnapshotLvolName(r.Name, restore.LastRestored)
 			snapshotNameToBeRestored = restore.LastRestored
 		} else {
+			r.log.Infof("Starting a full restore for backup %v", backupUrl)
 			lvolName = GetReplicaSnapshotLvolName(r.Name, snapshotName)
 			snapshotNameToBeRestored = snapshotName
 		}
@@ -1960,7 +1964,11 @@ func (r *Replica) BackupRestore(spdkClient *spdkclient.Client, backupUrl, snapsh
 		}
 		r.log.Infof("Successfully initiated full restore for %v to %v", backupUrl, newRestore.LvolName)
 	} else {
-		return fmt.Errorf("incremental restore is not supported yet")
+		r.log.Infof("Starting an incremental restore for backup %v", backupUrl)
+		if err := r.backupRestoreIncrementally(backupUrl, newRestore.LastRestored, newRestore.LvolName, concurrentLimit); err != nil {
+			return errors.Wrapf(err, "failed to start incremental backup restore")
+		}
+		r.log.Infof("Successfully initiated incremental restore for %v to %v", backupUrl, newRestore.LvolName)
 	}
 
 	go func() {
@@ -1973,6 +1981,25 @@ func (r *Replica) BackupRestore(spdkClient *spdkclient.Client, backupUrl, snapsh
 
 }
 
+func (r *Replica) backupRestoreIncrementally(backupURL, lastRestored, snapshotLvolName string, concurrentLimit int32) error {
+	backupURL = butil.UnescapeURL(backupURL)
+
+	logrus.WithFields(logrus.Fields{
+		"backupURL":        backupURL,
+		"lastRestored":     lastRestored,
+		"snapshotLvolName": snapshotLvolName,
+		"concurrentLimit":  concurrentLimit,
+	}).Info("Start restoring backup incrementally")
+
+	return backupstore.RestoreDeltaBlockBackupIncrementally(r.ctx, &backupstore.DeltaRestoreConfig{
+		BackupURL:       backupURL,
+		DeltaOps:        r.restore,
+		LastBackupName:  lastRestored,
+		Filename:        snapshotLvolName,
+		ConcurrentLimit: int32(concurrentLimit),
+	})
+}
+
 func (r *Replica) backupRestore(backupURL, snapshotLvolName string, concurrentLimit int32) error {
 	backupURL = butil.UnescapeURL(backupURL)