Fall back to other decompression method if the current one fails

The mitigation will be removed after identifying longhorn/longhorn#7687

Longhorn 7687

Signed-off-by: Derek Su <[email protected]>

backupbackingimage/backupbackingimage.go

@@ -1,10 +1,12 @@
 package backupbackingimage
 
 import (
+	"compress/gzip"
 	"context"
 	"fmt"
 	"io"
 	"os"
+	"strings"
 	"sync"
 
 	"github.com/pkg/errors"
@@ -484,7 +486,26 @@ func restoreBlockToFile(bsDriver backupstore.BackupStoreDriver, backingImageFile
 	defer rc.Close()
 	r, err := util.DecompressAndVerify(decompression, rc, blk.BlockChecksum)
 	if err != nil {
-		return err
+		// Fall back to other decompression method if the current one fails
+		// The mitigation will be removed after identifying https://github.com/longhorn/longhorn/issues/7687
+		// Seek rc to offset 0
+		seeker, ok := rc.(io.Seeker)
+		if !ok {
+			return errors.New("reader does not support seeking")
+		}
+
+		if _, err := seeker.Seek(0, io.SeekStart); err != nil {
+			return errors.Wrapf(err, "failed to seek to offset 0 for block %v", blk.BlockChecksum)
+		}
+
+		if strings.Contains(err.Error(), gzip.ErrHeader.Error()) {
+			r, err = util.DecompressAndVerify("lz4", rc, blk.BlockChecksum)
+		} else if strings.Contains(err.Error(), "lz4: bad magic number") {
+			r, err = util.DecompressAndVerify("gzip", rc, blk.BlockChecksum)
+		}
+	}
+	if err != nil {
+		return errors.Wrapf(err, "failed to decompress and verify block %v", blk.BlockChecksum)
 	}
 
 	if _, err := backingImageFile.Seek(blk.Offset, 0); err != nil {

deltablock.go

@@ -1,11 +1,13 @@
 package backupstore
 
 import (
+	"compress/gzip"
 	"context"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -252,6 +254,10 @@ func CreateDeltaBlockBackup(backupName string, config *DeltaBackupConfig) (isInc
 		},
 	}
 
+	log = logrus.WithFields(logrus.Fields{
+		"compressionMethod": volume.CompressionMethod,
+	})
+
 	// keep lock alive for async go routine.
 	if err := lock.Lock(); err != nil {
 		deltaOps.CloseSnapshot(snapshot.Name, volume.Name)
@@ -264,6 +270,7 @@ func CreateDeltaBlockBackup(backupName string, config *DeltaBackupConfig) (isInc
 		deltaOps.UpdateBackupStatus(snapshot.Name, volume.Name, string(types.ProgressStateInProgress), 0, "", "")
 
 		log.Info("Performing delta block backup")
+
 		if progress, backup, err := performBackup(bsDriver, config, delta, deltaBackup, backupRequest.lastBackup); err != nil {
 			logrus.WithError(err).Errorf("Failed to perform backup for volume %v snapshot %v", volume.Name, snapshot.Name)
 			deltaOps.UpdateBackupStatus(snapshot.Name, volume.Name, string(types.ProgressStateInProgress), progress, "", err.Error())
@@ -738,18 +745,39 @@ func restoreBlockToFile(bsDriver BackupStoreDriver, volumeName string, volDev *o
 	blkFile := getBlockFilePath(volumeName, blk.BlockChecksum)
 	rc, err := bsDriver.Read(blkFile)
 	if err != nil {
-		return err
+		return errors.Wrapf(err, "failed to read block %v from backupstore %v", blk.BlockChecksum, blkFile)
 	}
 	defer rc.Close()
+
 	r, err := util.DecompressAndVerify(decompression, rc, blk.BlockChecksum)
 	if err != nil {
-		return err
+		// Fall back to other decompression method if the current one fails
+		// The mitigation will be removed after identifying https://github.com/longhorn/longhorn/issues/7687
+		// Seek rc to offset 0
+		seeker, ok := rc.(io.Seeker)
+		if !ok {
+			return errors.New("reader does not support seeking")
+		}
+
+		if _, err := seeker.Seek(0, io.SeekStart); err != nil {
+			return errors.Wrapf(err, "failed to seek to offset 0 for block %v", blk.BlockChecksum)
+		}
+
+		if strings.Contains(err.Error(), gzip.ErrHeader.Error()) {
+			r, err = util.DecompressAndVerify("lz4", rc, blk.BlockChecksum)
+		} else if strings.Contains(err.Error(), "lz4: bad magic number") {
+			r, err = util.DecompressAndVerify("gzip", rc, blk.BlockChecksum)
+		}
+	}
+	if err != nil {
+		return errors.Wrapf(err, "failed to decompress and verify block %v", blk.BlockChecksum)
 	}
+
 	if _, err := volDev.Seek(blk.Offset, 0); err != nil {
-		return err
+		return errors.Wrapf(err, "failed to seek to offset %v for decompressed block %v", blk.Offset, blk.BlockChecksum)
 	}
 	_, err = io.CopyN(volDev, r, DEFAULT_BLOCK_SIZE)
-	return err
+	return errors.Wrapf(err, "failed to write decompressed block %v to volume %v", blk.BlockChecksum, volumeName)
 }
 
 func RestoreDeltaBlockBackupIncrementally(ctx context.Context, config *DeltaRestoreConfig) error {

util/util.go

@@ -120,12 +120,12 @@ func CompressData(method string, data []byte) (io.ReadSeeker, error) {
 func DecompressAndVerify(method string, src io.Reader, checksum string) (io.Reader, error) {
 	r, err := newDecompressionReader(method, src)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "failed to create decompression reader")
 	}
 	defer r.Close()
 	block, err := io.ReadAll(r)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "failed to read decompressed data")
 	}
 	if GetChecksum(block) != checksum {
 		return nil, fmt.Errorf("checksum verification failed for block")