Merge branch 'dev' into PBM-1514-Improve-resync-performance-updates

Author: veceraj

cmd/pbm-agent/restore.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"runtime"
 	"time"
 
@@ -238,7 +239,9 @@ func addRestoreMetaWithError(
 	errStr string,
 	args ...any,
 ) error {
-	l.Error(errStr, args...)
+	errMsg := fmt.Sprintf(errStr, args...)
+
+	l.Error(errMsg)
 
 	meta := &restore.RestoreMeta{
 		Type:     defs.LogicalBackup,
@@ -248,7 +251,7 @@ func addRestoreMetaWithError(
 		PITR:     int64(cmd.OplogTS.T),
 		StartTS:  time.Now().UTC().Unix(),
 		Status:   defs.StatusError,
-		Error:    errStr,
+		Error:    errMsg,
 		Replsets: []restore.RestoreReplset{},
 	}
 	err := restore.SetRestoreMetaIfNotExists(ctx, conn, meta)
@@ -260,7 +263,7 @@ func addRestoreMetaWithError(
 		Name:       setName,
 		StartTS:    time.Now().UTC().Unix(),
 		Status:     defs.StatusError,
-		Error:      errStr,
+		Error:      errMsg,
 		Conditions: restore.Conditions{},
 	}
 	err = restore.AddRestoreRSMeta(ctx, conn, cmd.Name, rs)

pbm/restore/physical.go

@@ -220,16 +220,12 @@ func (r *PhysRestore) close(noerr, cleanup bool) {
 			r.log.Error("remove tmp config %s: %v", r.tmpConf.Name(), err)
 		}
 	}
-	// clean-up internal mongod log only if there is no error
+
+	// if there is no error clean-up internal restore files, internal log file(s) should stay
 	if noerr {
-		r.log.Debug("rm tmp logs")
-		err := os.Remove(path.Join(r.dbpath, internalMongodLog))
-		if err != nil {
-			r.log.Warning("remove tmp mongod logs %s: %v", path.Join(r.dbpath, internalMongodLog), err)
-		}
 		extMeta := filepath.Join(r.dbpath,
 			fmt.Sprintf(defs.ExternalRsMetaFile, util.MakeReverseRSMapFunc(r.rsMap)(r.nodeInfo.SetName)))
-		err = os.Remove(extMeta)
+		err := os.Remove(extMeta)
 		if err != nil && !errors.Is(err, os.ErrNotExist) {
 			r.log.Warning("remove external rs meta <%s>: %v", extMeta, err)
 		}
@@ -252,7 +248,7 @@ func (r *PhysRestore) close(noerr, cleanup bool) {
 		}
 	} else if cleanup { // clean-up dbpath on err if needed (cluster is done or partlyDone)
 		r.log.Debug("clean-up dbpath")
-		err := removeAll(r.dbpath, nil, r.log)
+		err := removeAll(r.dbpath, r.log, getInternalLogFileSkipRule())
 		if err != nil {
 			r.log.Error("flush dbpath %s: %v", r.dbpath, err)
 		}
@@ -378,6 +374,10 @@ func (r *PhysRestore) migrateDBDirToFallbackDir() error {
 	if err != nil {
 		return errors.Wrap(err, "remove fallback db path")
 	}
+	err = removeInternalMongoLogs(dbpath, r.log)
+	if err != nil {
+		return errors.Wrap(err, "remove internal mongod log(s)")
+	}
 
 	r.log.Debug("create %s", fallbackPath)
 	info, err := os.Stat(dbpath)
@@ -402,7 +402,7 @@ func (r *PhysRestore) migrateDBDirToFallbackDir() error {
 // moves all content from fallback path.
 func (r *PhysRestore) migrateFromFallbackDirToDBDir() error {
 	r.log.Debug("clean-up dbpath")
-	err := removeAll(r.dbpath, []string{fallbackDir}, r.log)
+	err := removeAll(r.dbpath, r.log, getFallbackSyncFileSkipRule(), getInternalLogFileSkipRule())
 	if err != nil {
 		r.log.Error("flush dbpath %s: %v", r.dbpath, err)
 		return errors.Wrap(err, "remove all from dbpath")
@@ -2592,7 +2592,9 @@ func moveAll(fromDir, toDir string, toIgnore []string, l log.LogEvent) error {
 	return nil
 }
 
-func removeAll(dir string, toIgnore []string, l log.LogEvent) error {
+// removeAll removes all files and directories from specified dir.
+// It ignores files selected with filesSkipRules parameter.
+func removeAll(dir string, l log.LogEvent, fileSkipRules ...fileSkipRule) error {
 	d, err := os.Open(dir)
 	if err != nil {
 		return errors.Wrap(err, "open dir")
@@ -2604,7 +2606,7 @@ func removeAll(dir string, toIgnore []string, l log.LogEvent) error {
 		return errors.Wrap(err, "read file names")
 	}
 	for _, n := range names {
-		if n == internalMongodLog || slices.Contains(toIgnore, n) {
+		if isFileToSkip(n, fileSkipRules...) {
 			continue
 		}
 		err = os.RemoveAll(filepath.Join(dir, n))
@@ -2616,6 +2618,65 @@ func removeAll(dir string, toIgnore []string, l log.LogEvent) error {
 	return nil
 }
 
+// removeInternalMongoLogs removes internal mongod logs from directory.
+// It'll remove everything that starts with 'pbm.restore.log'
+func removeInternalMongoLogs(dir string, l log.LogEvent) error {
+	d, err := os.Open(dir)
+	if err != nil {
+		return errors.Wrap(err, "open dir")
+	}
+	defer d.Close()
+
+	names, err := d.Readdirnames(-1)
+	if err != nil {
+		return errors.Wrap(err, "read file names")
+	}
+	for _, n := range names {
+		if isInternalMongoLog(n) {
+			err = os.RemoveAll(filepath.Join(dir, n))
+			if err != nil {
+				return errors.Wrapf(err, "remove '%s", n)
+			}
+
+			l.Debug("remove %s", filepath.Join(dir, n))
+		}
+	}
+
+	return nil
+}
+
+// isInternalMongoLog checks whether the file with the name f
+// is internal mongo log file
+func isInternalMongoLog(f string) bool {
+	return strings.HasPrefix(f, internalMongodLog)
+}
+
+type fileSkipRule func(string) bool
+
+// isFileToSkip for the given file name f and the given set of skip rules: skipRules,
+// function returns true if at least one rule is satisfied. In case when all rules
+// are false it returns false, which means that file shouldn't be skipped.
+func isFileToSkip(f string, skipRules ...fileSkipRule) bool {
+	for _, rule := range skipRules {
+		if rule(f) {
+			return true
+		}
+	}
+	return false
+}
+
+func getInternalLogFileSkipRule() fileSkipRule {
+	return func(f string) bool {
+		return isInternalMongoLog(f)
+	}
+}
+
+func getFallbackSyncFileSkipRule() fileSkipRule {
+	return func(f string) bool {
+		return f == fallbackDir
+	}
+}
+
 func majmin(v string) string {
 	if len(v) == 0 {
 		return v

pbm/restore/physical_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"path"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -181,3 +182,162 @@ func TestWaitMgoFreePort(t *testing.T) {
 		}
 	})
 }
+
+func TestRemoveAll(t *testing.T) {
+	t.Run("removes all files in dir", func(t *testing.T) {
+		tmpDir := setupTestFiles(t)
+
+		err := removeAll(tmpDir, log.DiscardEvent)
+		if err != nil {
+			t.Fatalf("got error when removing all files, err=%v", err)
+		}
+
+		files := readDir(t, tmpDir)
+		if len(files) != 0 {
+			t.Fatalf("dir should be empty, got=%d files", len(files))
+		}
+	})
+
+	t.Run("skipping internal mongod log files", func(t *testing.T) {
+		tmpDir := setupTestFiles(t)
+
+		err := removeAll(tmpDir, log.DiscardEvent, getInternalLogFileSkipRule())
+		if err != nil {
+			t.Fatalf("got error when removing all files, err=%v", err)
+		}
+
+		files := readDir(t, tmpDir)
+		if len(files) != 3 {
+			t.Fatalf("expected to have 3 log files, got=%d files, names=%s", len(files), files)
+		}
+	})
+
+	t.Run("skipping fallback dir", func(t *testing.T) {
+		tmpDir := setupTestFiles(t)
+
+		err := removeAll(tmpDir, log.DiscardEvent, getFallbackSyncFileSkipRule())
+		if err != nil {
+			t.Fatalf("got error when removing all files, err=%v", err)
+		}
+
+		files := readDir(t, tmpDir)
+		if len(files) != 1 || files[0] != fallbackDir {
+			t.Fatalf("expected to have fallback dir, got=%d files/dirs, names=%s", len(files), files)
+		}
+
+		files = readDir(t, path.Join(tmpDir, fallbackDir))
+		if len(files) != 1 || files[0] != "file.fallback" {
+			t.Fatalf("expected to have 1 file within fallback dir, got=%d files/dirs, names=%s", len(files), files)
+		}
+	})
+
+	t.Run("skipping all pbm related", func(t *testing.T) {
+		tmpDir := setupTestFiles(t)
+
+		err := removeAll(tmpDir, log.DiscardEvent, getFallbackSyncFileSkipRule(), getInternalLogFileSkipRule())
+		if err != nil {
+			t.Fatalf("got error when removing all files, err=%v", err)
+		}
+
+		files := readDir(t, tmpDir)
+		if len(files) != 4 {
+			t.Fatalf("expected to have fallback dir and mongod log file, got=%d files/dirs, names=%s", len(files), files)
+		}
+	})
+}
+
+func TestRemoveInternalMongoLogs(t *testing.T) {
+	tmpDir := setupTestFiles(t)
+
+	err := removeInternalMongoLogs(tmpDir, log.DiscardEvent)
+	if err != nil {
+		t.Fatalf("got error when removing internal mongod logs, err=%v", err)
+	}
+
+	files := readDir(t, tmpDir)
+	if len(files) != 6 {
+		t.Fatalf("only mongod log files should be removed, expected 6 files, got=%d files", len(files))
+	}
+	for _, f := range files {
+		if strings.HasPrefix(f, internalMongodLog) {
+			t.Fatalf("internal mongod log file is not deleted: %s", f)
+		}
+	}
+}
+
+func readDir(t *testing.T, dir string) []string {
+	t.Helper()
+
+	d, err := os.Open(dir)
+	if err != nil {
+		t.Fatalf("failing opening dir, err=%v", err)
+	}
+	defer d.Close()
+
+	files, err := d.Readdirnames(-1)
+	if err != nil {
+		t.Fatalf("failing reading dir, err=%v", err)
+	}
+
+	return files
+}
+
+func setupTestFiles(t *testing.T) string {
+	tmpDir, err := os.MkdirTemp("", "phys-test-*")
+	if err != nil {
+		t.Fatalf("error while creating setup files: %v", err)
+	}
+
+	t.Cleanup(func() {
+		os.RemoveAll(tmpDir)
+	})
+
+	// tmpDir/
+	//   - file1.txt
+	//   - file2.log
+	//   - file3.txt.tmp
+	//   - subdir/
+	//     - file4.txt
+	//     - file5.log
+	//     - file6.txt.tmp
+	//   - empty/
+	//   - pbm.restore.log
+	//   - pbm.restore.log.2025-05-15T14-33-34
+	//   - pbm.restore.log.2025-05-15T14-37-34
+	//   - .fallbacksync/
+	//     - file.fs.txt
+	createTestFile(t, filepath.Join(tmpDir, "file1.txt"), "content1")
+	createTestFile(t, filepath.Join(tmpDir, "file2.log"), "content2")
+	createTestFile(t, filepath.Join(tmpDir, "file3.txt.tmp"), "content3")
+
+	createTestDir(t, filepath.Join(tmpDir, "subdir"))
+	createTestFile(t, filepath.Join(tmpDir, "subdir", "file4.txt"), "content4")
+	createTestFile(t, filepath.Join(tmpDir, "subdir", "file5.log"), "content5")
+	createTestFile(t, filepath.Join(tmpDir, "subdir", "file6.txt.tmp"), "content6")
+
+	createTestDir(t, filepath.Join(tmpDir, "empty"))
+
+	// files & dirs with additional semantic:
+	createTestFile(t, filepath.Join(tmpDir, "pbm.restore.log"), "log-content-1")
+	createTestFile(t, filepath.Join(tmpDir, "pbm.restore.log.2025-05-15T14-33-34"), "log-content-2")
+	createTestFile(t, filepath.Join(tmpDir, "pbm.restore.log.2025-05-15T14-37-34"), "log-content-3")
+
+	createTestDir(t, filepath.Join(tmpDir, fallbackDir))
+	createTestFile(t, filepath.Join(tmpDir, fallbackDir, "file.fallback"), "content")
+
+	return tmpDir
+}
+
+func createTestFile(t *testing.T, path, content string) {
+	t.Helper()
+	if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
+		t.Fatalf("error while creating file %s: %v", path, err)
+	}
+}
+
+func createTestDir(t *testing.T, path string) {
+	t.Helper()
+	if err := os.Mkdir(path, 0o755); err != nil {
+		t.Fatalf("error while creating dir %s: %v", path, err)
+	}
+}