Merge pull request #5786 from qiuming-best/unsupport-type-error

Set Kopia IgnoreUnknownTypes in ErrorHandlingPolicy to True

changelogs/unreleased/5786-qiuming-best

@@ -0,0 +1 @@
+Set Kopia IgnoreUnknownTypes in ErrorHandlingPolicy to True for ignoring backup unknown file type

pkg/uploader/kopia/progress.go

@@ -20,6 +20,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/sirupsen/logrus"
+
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 )
 
@@ -63,6 +65,7 @@ type KopiaProgress struct {
 	processedBytes int64                    // which statistic all bytes has been processed currently
 	outputThrottle Throttle                 // which control the frequency of update progress
 	Updater        uploader.ProgressUpdater //which kopia progress will call the UpdateProgress interface, the third party will implement the interface to do the progress update
+	Log            logrus.FieldLogger       // output info into log when backup
 }
 
 //UploadedBytes the total bytes has uploaded currently
@@ -77,8 +80,10 @@ func (p *KopiaProgress) UploadedBytes(numBytes int64) {
 func (p *KopiaProgress) Error(path string, err error, isIgnored bool) {
 	if isIgnored {
 		atomic.AddInt32(&p.ignoredErrorCount, 1)
+		p.Log.Warnf("Ignored error when processing %v: %v", path, err)
 	} else {
 		atomic.AddInt32(&p.fatalErrorCount, 1)
+		p.Log.Errorf("Error when processing %v: %v", path, err)
 	}
 }
 

pkg/uploader/kopia/snapshot.go

@@ -18,6 +18,7 @@ package kopia
 
 import (
 	"context"
+	"fmt"
 	"io/ioutil"
 	"math"
 	"os"
@@ -61,8 +62,14 @@ type SnapshotUploader interface {
 	) (*snapshot.Manifest, error)
 }
 
-func newOptionalInt(b policy.OptionalInt) *policy.OptionalInt {
-	return &b
+func newOptionalInt(b int) *policy.OptionalInt {
+	ob := policy.OptionalInt(b)
+	return &ob
+}
+
+func newOptionalBool(b bool) *policy.OptionalBool {
+	ob := policy.OptionalBool(b)
+	return &ob
 }
 
 //setupDefaultPolicy set default policy for kopia
@@ -75,11 +82,14 @@ func setupDefaultPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceIn
 			CompressorName: "none",
 		},
 		UploadPolicy: policy.UploadPolicy{
-			MaxParallelFileReads: newOptionalInt(policy.OptionalInt(runtime.NumCPU())),
+			MaxParallelFileReads: newOptionalInt(runtime.NumCPU()),
 		},
 		SchedulingPolicy: policy.SchedulingPolicy{
 			Manual: true,
 		},
+		ErrorHandlingPolicy: policy.ErrorHandlingPolicy{
+			IgnoreUnknownTypes: newOptionalBool(true),
+		},
 	})
 }
 
@@ -212,19 +222,23 @@ func SnapshotSource(
 		return "", 0, errors.Wrapf(err, "Failed to flush kopia repository")
 	}
 	log.Infof("Created snapshot with root %v and ID %v in %v", manifest.RootObjectID(), manifest.ID, time.Since(snapshotStartTime).Truncate(time.Second))
-	return reportSnapshotStatus(manifest)
+	return reportSnapshotStatus(manifest, policyTree)
 }
 
-func reportSnapshotStatus(manifest *snapshot.Manifest) (string, int64, error) {
+func reportSnapshotStatus(manifest *snapshot.Manifest, policyTree *policy.Tree) (string, int64, error) {
 	manifestID := manifest.ID
 	snapSize := manifest.Stats.TotalFileSize
 
 	var errs []string
 	if ds := manifest.RootEntry.DirSummary; ds != nil {
 		for _, ent := range ds.FailedEntries {
-			errs = append(errs, ent.Error)
+			policy := policyTree.DefinedPolicy()
+			if !(policy != nil && *policy.ErrorHandlingPolicy.IgnoreUnknownTypes == true && strings.Contains(ent.Error, fs.ErrUnknown.Error())) {
+				errs = append(errs, fmt.Sprintf("Error when processing %v: %v", ent.EntryPath, ent.Error))
+			}
 		}
 	}
+
 	if len(errs) != 0 {
 		return "", 0, errors.New(strings.Join(errs, "\n"))
 	}

pkg/uploader/provider/kopia.go

@@ -119,10 +119,11 @@ func (kp *kopiaProvider) RunBackup(
 	})
 	repoWriter := kopia.NewShimRepo(kp.bkRepo)
 	kpUploader := snapshotfs.NewUploader(repoWriter)
-	prorgess := new(kopia.KopiaProgress)
-	prorgess.InitThrottle(backupProgressCheckInterval)
-	prorgess.Updater = updater
-	kpUploader.Progress = prorgess
+	progress := new(kopia.KopiaProgress)
+	progress.InitThrottle(backupProgressCheckInterval)
+	progress.Updater = updater
+	progress.Log = log
+	kpUploader.Progress = progress
 	quit := make(chan struct{})
 	log.Info("Starting backup")
 	go kp.CheckContext(ctx, quit, nil, kpUploader)