From 9cada8fc11dacc31f050c914b4232fc7a31ea3b7 Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Thu, 26 Feb 2026 13:43:58 +0800 Subject: [PATCH] issue 9460: flush buffer when uploader completes Signed-off-by: Lyndon-Li --- changelogs/unreleased/9561-Lyndon-Li‎‎ | 1 + go.mod | 2 +- pkg/uploader/kopia/block_restore.go | 21 +++++++++++++-- pkg/uploader/kopia/flush_volume_linux.go | 11 +++++--- .../kopia/{flush.go => restore_output.go} | 9 +++++-- pkg/uploader/kopia/snapshot.go | 27 ++++++++++--------- 6 files changed, 50 insertions(+), 21 deletions(-) create mode 100644 changelogs/unreleased/9561-Lyndon-Li‎‎ rename pkg/uploader/kopia/{flush.go => restore_output.go} (82%) diff --git a/changelogs/unreleased/9561-Lyndon-Li‎‎ b/changelogs/unreleased/9561-Lyndon-Li‎‎ new file mode 100644 index 000000000..e6661e44d --- /dev/null +++ b/changelogs/unreleased/9561-Lyndon-Li‎‎ @@ -0,0 +1 @@ +Fix issue #9460, flush buffer before data mover completes \ No newline at end of file diff --git a/go.mod b/go.mod index 65177f63d..8fbfbef90 100644 --- a/go.mod +++ b/go.mod @@ -43,6 +43,7 @@ require ( go.uber.org/zap v1.27.1 golang.org/x/mod v0.30.0 golang.org/x/oauth2 v0.33.0 + golang.org/x/sys v0.38.0 golang.org/x/text v0.31.0 google.golang.org/api v0.256.0 google.golang.org/grpc v1.77.0 @@ -183,7 +184,6 @@ require ( golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect golang.org/x/net v0.47.0 // indirect golang.org/x/sync v0.18.0 // indirect - golang.org/x/sys v0.38.0 // indirect golang.org/x/term v0.37.0 // indirect golang.org/x/time v0.14.0 // indirect golang.org/x/tools v0.38.0 // indirect diff --git a/pkg/uploader/kopia/block_restore.go b/pkg/uploader/kopia/block_restore.go index 1065f3293..4f28a59de 100644 --- a/pkg/uploader/kopia/block_restore.go +++ b/pkg/uploader/kopia/block_restore.go @@ -35,6 +35,7 @@ type BlockOutput struct { *restore.FilesystemOutput targetFileName string + targetFile *os.File } var _ restore.Output = &BlockOutput{} @@ -52,7 +53,7 @@ func (o *BlockOutput) WriteFile(ctx context.Context, relativePath string, remote if err != nil { return errors.Wrapf(err, "failed to open file %s", o.targetFileName) } - defer targetFile.Close() + o.targetFile = targetFile buffer := make([]byte, bufferSize) @@ -103,5 +104,21 @@ func (o *BlockOutput) BeginDirectory(ctx context.Context, relativePath string, e } func (o *BlockOutput) Flush() error { - return flushVolume(o.targetFileName) + if o.targetFile != nil { + if err := o.targetFile.Sync(); err != nil { + return errors.Wrapf(err, "error syncing block dev %v", o.targetFileName) + } + } + + return nil +} + +func (o *BlockOutput) Terminate() error { + if o.targetFile != nil { + if err := o.targetFile.Close(); err != nil { + return errors.Wrapf(err, "error closing block dev %v", o.targetFileName) + } + } + + return nil } diff --git a/pkg/uploader/kopia/flush_volume_linux.go b/pkg/uploader/kopia/flush_volume_linux.go index 3f4901614..98234e1b9 100644 --- a/pkg/uploader/kopia/flush_volume_linux.go +++ b/pkg/uploader/kopia/flush_volume_linux.go @@ -37,11 +37,14 @@ func flushVolume(dirPath string) error { return errors.Wrapf(err, "error getting handle of dir %v", dirPath) } - raw.Control(func(fd uintptr) { + var syncErr error + if err := raw.Control(func(fd uintptr) { if e := unix.Syncfs(int(fd)); e != nil { - err = e + syncErr = e } - }) + }); err != nil { + return errors.Wrapf(err, "error calling fs sync from %v", dirPath) + } - return errors.Wrapf(err, "error syncing fs from %v", dirPath) + return errors.Wrapf(syncErr, "error syncing fs from %v", dirPath) } diff --git a/pkg/uploader/kopia/flush.go b/pkg/uploader/kopia/restore_output.go similarity index 82% rename from pkg/uploader/kopia/flush.go rename to pkg/uploader/kopia/restore_output.go index 04c10c7f6..74311d38a 100644 --- a/pkg/uploader/kopia/flush.go +++ b/pkg/uploader/kopia/restore_output.go @@ -16,10 +16,15 @@ limitations under the License. package kopia -import "github.com/pkg/errors" +import ( + "github.com/kopia/kopia/snapshot/restore" + "github.com/pkg/errors" +) var errFlushUnsupported = errors.New("flush is not supported") -type Flusher interface { +type RestoreOutput interface { + restore.Output Flush() error + Terminate() error } diff --git a/pkg/uploader/kopia/snapshot.go b/pkg/uploader/kopia/snapshot.go index ca05f7568..1924ed35b 100644 --- a/pkg/uploader/kopia/snapshot.go +++ b/pkg/uploader/kopia/snapshot.go @@ -384,6 +384,10 @@ func (o *fileSystemRestoreOutput) Flush() error { return flushVolumeFunc(o.TargetPath) } +func (o *fileSystemRestoreOutput) Terminate() error { + return nil +} + // Restore restore specific sourcePath with given snapshotID and update progress func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) { @@ -443,24 +447,23 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, return 0, 0, errors.Wrap(err, "error to init output") } - var output restore.Output - var flusher Flusher + var output RestoreOutput if volMode == uploader.PersistentVolumeBlock { - o := &BlockOutput{ + output = &BlockOutput{ FilesystemOutput: fsOutput, } - - output = o - flusher = o } else { - o := &fileSystemRestoreOutput{ + output = &fileSystemRestoreOutput{ FilesystemOutput: fsOutput, } - - output = o - flusher = o } + defer func() { + if err := output.Terminate(); err != nil { + log.Warnf("error terminating restore output for %v", path) + } + }() + stat, err := restoreEntryFunc(kopiaCtx, rep, output, rootEntry, restore.Options{ Parallel: restoreConcurrency, RestoreDirEntryAtDepth: math.MaxInt32, @@ -474,14 +477,14 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, return 0, 0, errors.Wrapf(err, "Failed to copy snapshot data to the target") } - if err := flusher.Flush(); err != nil { + if err := output.Flush(); err != nil { if err == errFlushUnsupported { log.Warnf("Skip flushing data for %v under the current OS %v", path, runtime.GOOS) } else { return 0, 0, errors.Wrapf(err, "Failed to flush data to target") } } else { - log.Warnf("Flush done for volume dir %v", path) + log.Infof("Flush done for volume dir %v", path) } return stat.RestoredTotalFileSize, stat.RestoredFileCount, nil