Synchronize gpbackup history with the standby coordinator. (#111)
`gpbackup_history.db` is stored on the active primary coordinator and is not automatically available on the standby. After a failover, the promoted coordinator may have missing or outdated backup history, which affects backup discovery and management.
This change keeps the cluster history database synchronized with an available standby coordinator:
- `gpbackup` automatically synchronizes history after a successful backup.
- gpBackMan automatically synchronizes history after commands that delete backups or clean history.
- The new `gpbackman history-sync` command allows synchronization to be started manually.
- Automatic synchronization is best-effort: failures are reported as warnings and do not change the result of a successful primary command.
- Automatic synchronization can be disabled with `--no-history-sync-standby`.
Only `gpbackup_history.db` is synchronized. Backup data, reports, and other backup artifacts are not included.
Unit and end-to-end coverage were added, and the related user documentation was updated.
Commit history:
* Add gpbackman coordinator discovery helpers.
* Add gpbackup standby history sync.
* Add gpbackman standby history sync engine.
* Add gpbackman history sync command and mutation hooks.
* Add standby history sync end-to-end coverage.
* Document standby history database synchronization.
* Tighten standby history sync validation.
* Log standby history sync cleanup errors.
Keep the primary operation result while making database and row close failures visible.
* Simplify the handling of history synchronization errors in standby mode.
Return connection, SQLite, and local cleanup failures to the policy boundary without changing best-effort exit status.
* Isolate standby history sync end-to-end coverage.
Disable automatic sync in shared helpers and restore the original standby history after the dedicated scenario.
* Simplify standby history synchronization documentation.
* Fix standby history sync tests on macOS.
* Fix gpbackman report-info examples.
* Fix link in gpBackMan docs.
* Fix standby history sync rsync destination.
Pass remote paths directly to rsync while retaining shell quoting for SSH install and cleanup commands.
* Add configurable standby history sync timeout.
Expose --history-sync-standby-timeout as integer seconds in gpbackup and sync-capable gpBackMan commands. Use int to match existing CLI conventions, default to 300 seconds, and cap values at one day to catch accidental settings without excluding slow transfers.
Create one context deadline after SQLite snapshot validation and share its remaining budget across rsync and remote install. This keeps discovery and local snapshot work outside the limit and prevents each transport stage from restarting the timeout.
Run rsync and ssh with CommandContext and BatchMode so stalled processes are terminated without waiting for interactive authentication.
Use an independent 120-second context for failure cleanup so remote temporary files can still be removed after the transport deadline expires while cleanup remains bounded. Preserve both primary and cleanup errors and keep automatic synchronization best-effort while history-sync stays strict.
* Protect standby history sync rsync paths.
Pass remote paths with rsync's `-s` option so shell metacharacters are not interpreted by the remote shell, and document the resulting rsync 3.0.0 requirement.
* Guard standby history sync context error checks with command result.
When rsync or ssh succeeds and the context deadline fires immediately after, the successful transfer was incorrectly treated as a timeout. Only prefer ctx.Err() when the command itself returned an error.
* Extend standby history sync context error guards to gpbackup.
diff --git a/README.md b/README.md
index eb2fafa..db40bc2 100644
--- a/README.md
+++ b/README.md
@@ -89,6 +89,60 @@
Run `--help` with either command for a complete list of options.
+### Standby history database synchronization
+
+After a successful backup, `gpbackup` automatically copies a consistent
+snapshot of the coordinator's `gpbackup_history.db` to an up standby
+coordinator. Synchronization starts only after the final `Success` history row
+has been written and the local SQLite connection has been closed.
+
+This synchronization is best effort. If no up standby exists, synchronization
+is skipped. If discovery, snapshot creation, transfer, or installation fails,
+`gpbackup` logs a warning but keeps the successful backup exit status. A
+failed or terminated backup is not synchronized. Synchronization also does not
+run when `--no-history` is used or when the final history update fails. Use
+`--no-history-sync-standby` to keep writing local history while disabling
+standby synchronization for one backup:
+
+```bash
+gpbackup --dbname <your_db_name> --no-history-sync-standby
+```
+
+Configure the sync timeout with `--history-sync-standby-timeout SECONDS`. The
+default is 300 seconds; the supported range is 1 to 86400 seconds. The timeout
+is one shared budget for `rsync` and remote install. It starts after snapshot
+validation. Standby discovery and SQLite snapshot creation and validation
+(`VACUUM INTO` and `PRAGMA quick_check`) are outside this budget. If a
+transport step fails, remote cleanup of the temporary file uses its own fixed
+120-second timeout, independent of `--history-sync-standby-timeout`.
+
+The synchronization process:
+
+1. Takes a non-waiting lock next to the canonical source database.
+2. Creates a consistent SQLite snapshot with `VACUUM INTO` and accepts it only
+ when `PRAGMA quick_check` returns `ok`.
+3. Transfers the snapshot with `rsync -p -s` to a unique temporary file in the
+ standby coordinator data directory.
+4. Preserves the existing standby file's owner, group, and mode when it
+ exists, then atomically renames the temporary file to
+ `gpbackup_history.db`.
+
+`rsync` 3.0.0 or later must be installed on both the host running `gpbackup`
+and the standby coordinator. The `gpbackup` host must also have `ssh`, and the
+current OS user must have non-interactive SSH access to the standby host. That
+user must be able to create files in the standby coordinator data directory
+and preserve the destination file's ownership and permissions. The cluster
+must expose an up standby in `gp_segment_configuration`.
+
+The atomic rename prevents readers from observing a partially copied database,
+but it is not a failover coordination mechanism. A coordinator role change
+during synchronization can race with discovery and installation. Processes
+that already have the old standby database open continue reading that old
+inode until they close and reopen it.
+
+For automatic synchronization after history maintenance and for the strict
+manual command, see [gpBackMan history synchronization](./gpbackman/README.md#standby-history-db-sync).
+
## Additional tools
This repository also includes the following tools:
@@ -196,4 +250,4 @@
## Acknowledgment
Thanks to all the Greenplum Backup contributors, more details in its [GitHub
-page](https://github.com/greenplum-db/gpbackup-archive).
\ No newline at end of file
+page](https://github.com/greenplum-db/gpbackup-archive).
diff --git a/backup/backup.go b/backup/backup.go
index 9e12918..c003274 100644
--- a/backup/backup.go
+++ b/backup/backup.go
@@ -509,6 +509,7 @@
// failure; in either case, update the end time to the actual value. Between our signal handler and recovering
// panics, there should be no way for gpbackup to exit that leaves the entry in the initial status.
+ historyUpdated := false
if !MustGetFlagBool(options.NO_HISTORY) {
var statusString string
if backupFailed {
@@ -525,9 +526,12 @@
historyDB.Close()
if err != nil {
gplog.Error("Unable to update history database. Error: %v", err)
+ } else {
+ historyUpdated = true
}
}
}
+ syncBackupHistoryToStandbyAfterCleanup(backupFailed, historyUpdated)
err := backupLockFile.Unlock()
if err != nil && backupLockFile != "" {
diff --git a/backup/history_standby_sync.go b/backup/history_standby_sync.go
new file mode 100644
index 0000000..53a7374
--- /dev/null
+++ b/backup/history_standby_sync.go
@@ -0,0 +1,459 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package backup
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "net/url"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/apache/cloudberry-backup/options"
+ "github.com/apache/cloudberry-go-libs/gplog"
+ "github.com/apache/cloudberry-go-libs/operating"
+ _ "github.com/mattn/go-sqlite3"
+ "github.com/nightlyone/lockfile"
+)
+
+const (
+ backupHistoryDBName = "gpbackup_history.db"
+ // Leave enough time for the 30-second SSH connection timeout and remote removal
+ // while keeping failure cleanup bounded.
+ backupHistoryStandbySyncCleanupTimeout = 120 * time.Second
+ backupHistoryStandbySyncSSHOptions = "ssh -o BatchMode=yes -o StrictHostKeyChecking=no -o ConnectTimeout=30"
+ backupHistoryStandbySyncTempDirPattern = "gpbackup-history-standby-sync-*"
+ backupHistoryStandbySyncStandbySQL = "SELECT hostname, datadir FROM gp_segment_configuration WHERE content = -1 AND role = 'm' AND status = 'u';"
+ // Cap the timeout at one day to catch accidentally oversized CLI values.
+ // A longer transport deadline is not meaningful for standby history synchronization.
+ maxHistorySyncStandbyTimeoutSeconds = int(24 * time.Hour / time.Second)
+)
+
+type backupHistoryStandbySyncTarget struct {
+ sourceDBPath string
+ standbyHost string
+ standbyDataDir string
+ standbyHistoryDBPath string
+}
+
+type backupHistoryStandbySyncStandby struct {
+ Hostname string `db:"hostname"`
+ DataDir string `db:"datadir"`
+}
+
+type backupHistoryStandbySyncCommand interface {
+ CombinedOutput() ([]byte, error)
+}
+
+var (
+ backupHistoryStandbySync = syncBackupHistoryToStandby
+
+ backupHistoryStandbySyncCommandExec = func(ctx context.Context, name string, args ...string) backupHistoryStandbySyncCommand {
+ return exec.CommandContext(ctx, name, args...)
+ }
+ backupHistoryStandbySyncContextWithTimeout = context.WithTimeout
+ backupHistoryStandbySyncOpenSQLite = sql.Open
+ backupHistoryStandbySyncMkdirTemp = os.MkdirTemp
+ backupHistoryStandbySyncRemoveAll = os.RemoveAll
+ backupHistoryStandbySyncCurrentUser = func() (string, error) {
+ currentUser, err := operating.System.CurrentUser()
+ if err != nil {
+ return "", err
+ }
+ return currentUser.Username, nil
+ }
+)
+
+func syncBackupHistoryToStandbyBestEffort(disabled bool) (string, error) {
+ if disabled {
+ skipReason := "disabled by --" + options.NO_HISTORY_SYNC_STANDBY
+ gplog.Info("Skipping history db sync to standby coordinator: %s", skipReason)
+ return skipReason, nil
+ }
+
+ skipReason, err := backupHistoryStandbySync()
+ if err != nil {
+ gplog.Warn("History db sync to standby coordinator failed; standby history may be stale: %v", err)
+ return "", err
+ }
+ if skipReason != "" {
+ gplog.Debug("Skipping history db sync to standby coordinator: %s", skipReason)
+ }
+ return skipReason, nil
+}
+
+func syncBackupHistoryToStandbyAfterCleanup(backupFailed bool, historyUpdated bool) {
+ if backupFailed || !historyUpdated || MustGetFlagBool(options.NO_HISTORY) {
+ return
+ }
+ _, _ = syncBackupHistoryToStandbyBestEffort(MustGetFlagBool(options.NO_HISTORY_SYNC_STANDBY))
+}
+
+func syncBackupHistoryToStandby() (string, error) {
+ sourceDBPath, sourceInfo, err := canonicalBackupHistoryStandbySyncSource(globalFPInfo.GetBackupHistoryDatabasePath())
+ if err != nil {
+ return "", err
+ }
+
+ target, skipReason, err := discoverBackupHistoryStandbySyncTarget(sourceDBPath)
+ if err != nil {
+ return "", err
+ }
+ if skipReason != "" {
+ return skipReason, nil
+ }
+
+ userName, err := backupHistoryStandbySyncCurrentUser()
+ if err != nil {
+ return "", fmt.Errorf("resolve current OS user for standby history sync: %w", err)
+ }
+
+ err = withBackupHistoryStandbySyncLock(sourceDBPath, func() error {
+ return withBackupHistoryStandbySyncSnapshot(sourceDBPath, sourceInfo.Mode().Perm(), func(snapshotPath string) error {
+ timeoutSeconds := MustGetFlagInt(options.HISTORY_SYNC_STANDBY_TIMEOUT)
+ ctx, cancel := backupHistoryStandbySyncContextWithTimeout(context.Background(), time.Duration(timeoutSeconds)*time.Second)
+ defer cancel()
+
+ transportErr := syncBackupHistoryStandbySnapshot(ctx, target, userName, snapshotPath)
+ if transportErr != nil && errors.Is(ctx.Err(), context.DeadlineExceeded) {
+ return fmt.Errorf("standby history sync transport timed out after %d seconds: %w", timeoutSeconds, transportErr)
+ }
+ return transportErr
+ })
+ })
+ if err != nil {
+ return "", err
+ }
+ return "", nil
+}
+
+func canonicalBackupHistoryStandbySyncSource(sourceDBPath string) (string, os.FileInfo, error) {
+ absoluteSourceDBPath, err := filepath.Abs(filepath.Clean(sourceDBPath))
+ if err != nil {
+ return "", nil, fmt.Errorf("resolve absolute source history db path for standby sync: %w", err)
+ }
+ canonicalSourceDBPath, err := filepath.EvalSymlinks(absoluteSourceDBPath)
+ if err != nil {
+ return "", nil, fmt.Errorf("resolve canonical source history db path for standby sync: %w", err)
+ }
+ sourceInfo, err := os.Stat(canonicalSourceDBPath)
+ if err != nil {
+ return "", nil, fmt.Errorf("stat source history db for standby sync: %w", err)
+ }
+ if !sourceInfo.Mode().IsRegular() {
+ return "", nil, fmt.Errorf("source history db for standby sync is not a regular file: %s", canonicalSourceDBPath)
+ }
+ return canonicalSourceDBPath, sourceInfo, nil
+}
+
+func discoverBackupHistoryStandbySyncTarget(sourceDBPath string) (*backupHistoryStandbySyncTarget, string, error) {
+ standby, err := queryBackupHistoryStandbySyncStandby()
+ if err != nil {
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, "no up standby coordinator found", nil
+ }
+ return nil, "", fmt.Errorf("query up standby coordinator for standby history sync discovery: %w", err)
+ }
+ target := &backupHistoryStandbySyncTarget{
+ sourceDBPath: sourceDBPath,
+ standbyHost: standby.Hostname,
+ standbyDataDir: standby.DataDir,
+ standbyHistoryDBPath: filepath.Join(standby.DataDir, backupHistoryDBName),
+ }
+ gplog.Debug("Discovered standby history sync target: source=%s standby=%s:%s", target.sourceDBPath, target.standbyHost, target.standbyHistoryDBPath)
+ return target, "", nil
+}
+
+func queryBackupHistoryStandbySyncStandby() (backupHistoryStandbySyncStandby, error) {
+ var standby backupHistoryStandbySyncStandby
+ if connectionPool == nil {
+ return standby, errors.New("connection pool is not initialized")
+ }
+ err := connectionPool.Get(&standby, backupHistoryStandbySyncStandbySQL)
+ return standby, err
+}
+
+func withBackupHistoryStandbySyncLock(sourceDBPath string, syncFn func() error) error {
+ lockPath := backupHistoryStandbySyncLockPath(sourceDBPath)
+ sourceLock, err := lockfile.New(lockPath)
+ if err != nil {
+ return fmt.Errorf("create standby history sync lock %s: %w", lockPath, err)
+ }
+ if err := sourceLock.TryLock(); err != nil {
+ return fmt.Errorf("lock standby history sync source %s: %w", sourceDBPath, err)
+ }
+
+ syncErr := syncFn()
+ unlockErr := sourceLock.Unlock()
+ if syncErr != nil {
+ if unlockErr != nil {
+ return fmt.Errorf("%w; additionally failed to release standby history sync lock %s: %v", syncErr, lockPath, unlockErr)
+ }
+ return syncErr
+ }
+ if unlockErr != nil {
+ return fmt.Errorf("release standby history sync lock %s: %w", lockPath, unlockErr)
+ }
+ return nil
+}
+
+func backupHistoryStandbySyncLockPath(sourceDBPath string) string {
+ return sourceDBPath + ".sync.lock"
+}
+
+func withBackupHistoryStandbySyncSnapshot(sourceDBPath string, sourceMode os.FileMode, syncFn func(string) error) (retErr error) {
+ snapshotPath, tempDir, err := createBackupHistoryStandbySyncSnapshot(sourceDBPath, sourceMode)
+ if tempDir != "" {
+ defer func() {
+ retErr = errors.Join(retErr, cleanupBackupHistoryStandbySyncTempDir(tempDir))
+ }()
+ }
+ if err != nil {
+ return err
+ }
+ return syncFn(snapshotPath)
+}
+
+func createBackupHistoryStandbySyncSnapshot(sourceDBPath string, sourceMode os.FileMode) (string, string, error) {
+ tempDir, err := backupHistoryStandbySyncMkdirTemp("", backupHistoryStandbySyncTempDirPattern)
+ if err != nil {
+ return "", "", fmt.Errorf("create local standby history sync temp directory: %w", err)
+ }
+ snapshotPath := filepath.Join(tempDir, backupHistoryDBName)
+ if err := vacuumBackupHistoryStandbySyncSnapshot(sourceDBPath, snapshotPath); err != nil {
+ return "", "", errors.Join(err, cleanupBackupHistoryStandbySyncTempDir(tempDir))
+ }
+ if err := os.Chmod(snapshotPath, sourceMode); err != nil {
+ return "", "", errors.Join(
+ fmt.Errorf("set standby history sync snapshot permissions from source history db: %w", err),
+ cleanupBackupHistoryStandbySyncTempDir(tempDir),
+ )
+ }
+ if err := validateBackupHistoryStandbySyncSnapshot(snapshotPath); err != nil {
+ return "", "", errors.Join(err, cleanupBackupHistoryStandbySyncTempDir(tempDir))
+ }
+ return snapshotPath, tempDir, nil
+}
+
+func vacuumBackupHistoryStandbySyncSnapshot(sourceDBPath, snapshotPath string) (retErr error) {
+ sourceDB, err := backupHistoryStandbySyncOpenSQLite("sqlite3", backupHistoryStandbySyncSQLiteURI(sourceDBPath, "ro"))
+ if err != nil {
+ return fmt.Errorf("open source history db for standby sync snapshot: %w", err)
+ }
+ defer func() {
+ if closeErr := sourceDB.Close(); closeErr != nil {
+ retErr = errors.Join(retErr, fmt.Errorf("close source history db for standby sync snapshot: %w", closeErr))
+ }
+ }()
+
+ if _, err := sourceDB.Exec("VACUUM main INTO ?", snapshotPath); err != nil {
+ return fmt.Errorf("create standby history sync snapshot with VACUUM INTO: %w", err)
+ }
+ return nil
+}
+
+func validateBackupHistoryStandbySyncSnapshot(snapshotPath string) error {
+ results, err := runBackupHistoryStandbySyncQuickCheck(snapshotPath)
+ if err != nil {
+ return err
+ }
+ if len(results) != 1 || results[0] != "ok" {
+ return fmt.Errorf("validate standby history sync snapshot quick_check: expected single ok result, got %v", results)
+ }
+ return nil
+}
+
+func runBackupHistoryStandbySyncQuickCheck(snapshotPath string) (results []string, retErr error) {
+ snapshotDB, err := backupHistoryStandbySyncOpenSQLite("sqlite3", backupHistoryStandbySyncSQLiteURI(snapshotPath, "ro"))
+ if err != nil {
+ return nil, fmt.Errorf("open standby history sync snapshot read-only: %w", err)
+ }
+ defer func() {
+ if closeErr := snapshotDB.Close(); closeErr != nil {
+ retErr = errors.Join(retErr, fmt.Errorf("close standby history sync snapshot: %w", closeErr))
+ }
+ }()
+
+ rows, err := snapshotDB.Query("PRAGMA quick_check")
+ if err != nil {
+ return nil, fmt.Errorf("run PRAGMA quick_check on standby history sync snapshot: %w", err)
+ }
+ defer func() {
+ if closeErr := rows.Close(); closeErr != nil {
+ retErr = errors.Join(retErr, fmt.Errorf("close standby history sync quick_check rows: %w", closeErr))
+ }
+ }()
+
+ results = make([]string, 0)
+ for rows.Next() {
+ var result string
+ if err := rows.Scan(&result); err != nil {
+ return nil, fmt.Errorf("scan PRAGMA quick_check result for standby history sync snapshot: %w", err)
+ }
+ results = append(results, result)
+ }
+ if err := rows.Err(); err != nil {
+ return nil, fmt.Errorf("read PRAGMA quick_check results for standby history sync snapshot: %w", err)
+ }
+ return results, nil
+}
+
+func cleanupBackupHistoryStandbySyncTempDir(tempDir string) error {
+ if err := backupHistoryStandbySyncRemoveAll(tempDir); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return fmt.Errorf("remove local standby history sync temp directory %s: %w", tempDir, err)
+ }
+ return nil
+}
+
+func syncBackupHistoryStandbySnapshot(ctx context.Context, target *backupHistoryStandbySyncTarget, userName, snapshotPath string) error {
+ remoteTempPath := newBackupHistoryStandbySyncRemoteTempPath(target.standbyDataDir, snapshotPath)
+ if err := rsyncBackupHistoryStandbySyncSnapshot(ctx, snapshotPath, target.standbyHost, userName, remoteTempPath); err != nil {
+ return cleanupBackupHistoryStandbySyncRemoteTempAfterError(err, target.standbyHost, userName, remoteTempPath)
+ }
+ if err := installBackupHistoryStandbySyncSnapshot(ctx, target, userName, remoteTempPath); err != nil {
+ return cleanupBackupHistoryStandbySyncRemoteTempAfterError(err, target.standbyHost, userName, remoteTempPath)
+ }
+ return nil
+}
+
+func newBackupHistoryStandbySyncRemoteTempPath(standbyDataDir, snapshotPath string) string {
+ return filepath.Join(standbyDataDir, fmt.Sprintf(".%s.%s.tmp", backupHistoryDBName, filepath.Base(filepath.Dir(snapshotPath))))
+}
+
+func rsyncBackupHistoryStandbySyncSnapshot(ctx context.Context, snapshotPath, standbyHost, userName, remoteTempPath string) error {
+ args := buildBackupHistoryStandbySyncRsyncArgs(snapshotPath, standbyHost, userName, remoteTempPath)
+ gplog.Debug("Transfer history db snapshot to standby coordinator: %s -> %s:%s", snapshotPath, standbyHost, remoteTempPath)
+ output, err := backupHistoryStandbySyncCommandExec(ctx, "rsync", args...).CombinedOutput()
+ if ctxErr := ctx.Err(); ctxErr != nil && err != nil {
+ err = ctxErr
+ }
+ if err != nil {
+ return fmt.Errorf("rsync standby history snapshot to %s:%s failed: %w%s", standbyHost, remoteTempPath, err, formatBackupHistoryStandbySyncCommandOutput(output))
+ }
+ return nil
+}
+
+func buildBackupHistoryStandbySyncRsyncArgs(snapshotPath, standbyHost, userName, remoteTempPath string) []string {
+ return []string{
+ "-p",
+ "-s",
+ "-e",
+ backupHistoryStandbySyncSSHOptions,
+ "--",
+ snapshotPath,
+ fmt.Sprintf("%s@%s:%s", userName, standbyHost, remoteTempPath),
+ }
+}
+
+func installBackupHistoryStandbySyncSnapshot(ctx context.Context, target *backupHistoryStandbySyncTarget, userName, remoteTempPath string) error {
+ command := buildBackupHistoryStandbySyncRemoteInstallCommand(remoteTempPath, target.standbyHistoryDBPath)
+ gplog.Debug("Install history db snapshot on standby coordinator: %s:%s", target.standbyHost, target.standbyHistoryDBPath)
+ output, err := runBackupHistoryStandbySyncSSHCommand(ctx, command, target.standbyHost, userName)
+ if err != nil {
+ return fmt.Errorf("install standby history snapshot on %s:%s failed: %w%s", target.standbyHost, target.standbyHistoryDBPath, err, formatBackupHistoryStandbySyncCommandOutput(output))
+ }
+ return nil
+}
+
+func buildBackupHistoryStandbySyncRemoteInstallCommand(remoteTempPath, standbyHistoryDBPath string) string {
+ quotedTempPath := shellQuoteBackupHistoryStandbySyncPath(remoteTempPath)
+ quotedHistoryDBPath := shellQuoteBackupHistoryStandbySyncPath(standbyHistoryDBPath)
+ return fmt.Sprintf(
+ "test -f %s && if test -e %s; then chown --reference=%s -- %s && chmod --reference=%s -- %s; fi && mv -f -- %s %s",
+ quotedTempPath,
+ quotedHistoryDBPath,
+ quotedHistoryDBPath,
+ quotedTempPath,
+ quotedHistoryDBPath,
+ quotedTempPath,
+ quotedTempPath,
+ quotedHistoryDBPath,
+ )
+}
+
+func cleanupBackupHistoryStandbySyncRemoteTempAfterError(primaryErr error, standbyHost, userName, remoteTempPath string) error {
+ cleanupCtx, cancel := context.WithTimeout(context.Background(), backupHistoryStandbySyncCleanupTimeout)
+ defer cancel()
+
+ if cleanupErr := cleanupBackupHistoryStandbySyncRemoteTemp(cleanupCtx, standbyHost, userName, remoteTempPath); cleanupErr != nil {
+ return fmt.Errorf("%w; additionally failed to clean up remote temp file: %w", primaryErr, cleanupErr)
+ }
+ return primaryErr
+}
+
+func cleanupBackupHistoryStandbySyncRemoteTemp(ctx context.Context, standbyHost, userName, remoteTempPath string) error {
+ command := buildBackupHistoryStandbySyncRemoteCleanupCommand(remoteTempPath)
+ gplog.Debug("Clean up remote standby history sync temp file: %s:%s", standbyHost, remoteTempPath)
+ output, err := runBackupHistoryStandbySyncSSHCommand(ctx, command, standbyHost, userName)
+ if err != nil {
+ return fmt.Errorf("clean up remote standby history sync temp file %s:%s failed: %w%s", standbyHost, remoteTempPath, err, formatBackupHistoryStandbySyncCommandOutput(output))
+ }
+ return nil
+}
+
+func buildBackupHistoryStandbySyncRemoteCleanupCommand(remoteTempPath string) string {
+ return fmt.Sprintf("rm -f -- %s", shellQuoteBackupHistoryStandbySyncPath(remoteTempPath))
+}
+
+func runBackupHistoryStandbySyncSSHCommand(ctx context.Context, remoteCommand, standbyHost, userName string) ([]byte, error) {
+ output, err := backupHistoryStandbySyncCommandExec(
+ ctx,
+ "ssh",
+ "-o",
+ "BatchMode=yes",
+ "-o",
+ "StrictHostKeyChecking=no",
+ "-o",
+ "ConnectTimeout=30",
+ fmt.Sprintf("%s@%s", userName, standbyHost),
+ remoteCommand,
+ ).CombinedOutput()
+ if ctxErr := ctx.Err(); ctxErr != nil && err != nil {
+ return output, ctxErr
+ }
+ return output, err
+}
+
+func backupHistoryStandbySyncSQLiteURI(dbPath, mode string) string {
+ query := url.Values{}
+ query.Set("mode", mode)
+ dbURI := url.URL{Scheme: "file", Path: dbPath, RawQuery: query.Encode()}
+ return dbURI.String()
+}
+
+func shellQuoteBackupHistoryStandbySyncPath(value string) string {
+ if value == "" {
+ return "''"
+ }
+ return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
+}
+
+func formatBackupHistoryStandbySyncCommandOutput(output []byte) string {
+ trimmedOutput := strings.TrimSpace(string(output))
+ if trimmedOutput == "" {
+ return ""
+ }
+ return ": " + trimmedOutput
+}
diff --git a/backup/history_standby_sync_test.go b/backup/history_standby_sync_test.go
new file mode 100644
index 0000000..dc22f90
--- /dev/null
+++ b/backup/history_standby_sync_test.go
@@ -0,0 +1,630 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package backup
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "regexp"
+ "strconv"
+ "time"
+
+ "github.com/DATA-DOG/go-sqlmock"
+ backupfilepath "github.com/apache/cloudberry-backup/filepath"
+ "github.com/apache/cloudberry-backup/options"
+ "github.com/apache/cloudberry-go-libs/dbconn"
+ "github.com/apache/cloudberry-go-libs/gplog"
+ "github.com/apache/cloudberry-go-libs/testhelper"
+ "github.com/jmoiron/sqlx"
+ "github.com/nightlyone/lockfile"
+ "github.com/spf13/pflag"
+
+ . "github.com/onsi/ginkgo/v2"
+ . "github.com/onsi/gomega"
+)
+
+type backupHistoryStandbySyncCommandCall struct {
+ ctx context.Context
+ ctxErr error
+ deadline time.Time
+ hasDeadline bool
+ name string
+ args []string
+}
+
+type backupHistoryStandbySyncCommandResponse struct {
+ output []byte
+ err error
+}
+
+type backupHistoryStandbySyncFakeCommand struct {
+ output []byte
+ err error
+}
+
+func (c backupHistoryStandbySyncFakeCommand) CombinedOutput() ([]byte, error) {
+ return c.output, c.err
+}
+
+var _ = Describe("backup history standby sync", func() {
+ var (
+ originalSync func() (string, error)
+ originalOpenSQLite func(string, string) (*sql.DB, error)
+ originalContextWithTimeout func(context.Context, time.Duration) (context.Context, context.CancelFunc)
+ )
+
+ BeforeEach(func() {
+ testhelper.SetupTestLogger()
+ cmdFlags = pflag.NewFlagSet("gpbackup", pflag.ContinueOnError)
+ options.SetBackupFlagDefaults(cmdFlags)
+ globalFPInfo = backupfilepath.FilePathInfo{}
+ connectionPool = nil
+ originalSync = backupHistoryStandbySync
+ originalOpenSQLite = backupHistoryStandbySyncOpenSQLite
+ originalContextWithTimeout = backupHistoryStandbySyncContextWithTimeout
+ backupHistoryStandbySync = syncBackupHistoryToStandby
+ backupHistoryStandbySyncOpenSQLite = sql.Open
+ backupHistoryStandbySyncContextWithTimeout = context.WithTimeout
+ backupHistoryStandbySyncCommandExec = func(ctx context.Context, name string, args ...string) backupHistoryStandbySyncCommand {
+ return backupHistoryStandbySyncFakeCommand{}
+ }
+ backupHistoryStandbySyncMkdirTemp = os.MkdirTemp
+ backupHistoryStandbySyncRemoveAll = os.RemoveAll
+ backupHistoryStandbySyncCurrentUser = func() (string, error) {
+ return "gpadmin", nil
+ }
+ })
+
+ AfterEach(func() {
+ backupHistoryStandbySync = originalSync
+ backupHistoryStandbySyncOpenSQLite = originalOpenSQLite
+ backupHistoryStandbySyncContextWithTimeout = originalContextWithTimeout
+ backupHistoryStandbySyncCommandExec = func(ctx context.Context, name string, args ...string) backupHistoryStandbySyncCommand {
+ return backupHistoryStandbySyncFakeCommand{}
+ }
+ backupHistoryStandbySyncMkdirTemp = os.MkdirTemp
+ backupHistoryStandbySyncRemoveAll = os.RemoveAll
+ backupHistoryStandbySyncCurrentUser = func() (string, error) {
+ return "gpadmin", nil
+ }
+ if connectionPool != nil {
+ connectionPool.Close()
+ }
+ })
+
+ It("creates a verified snapshot with source permissions", func() {
+ tmpDir := GinkgoT().TempDir()
+ sourcePath := filepath.Join(tmpDir, backupHistoryDBName)
+ createBackupHistoryStandbySyncSQLiteDB(sourcePath)
+ Expect(os.Chmod(sourcePath, 0o640)).To(Succeed())
+
+ snapshotPath, tempDir, err := createBackupHistoryStandbySyncSnapshot(sourcePath, 0o640)
+ Expect(err).ToNot(HaveOccurred())
+ defer cleanupBackupHistoryStandbySyncTempDir(tempDir)
+
+ Expect(snapshotPath).To(Equal(filepath.Join(tempDir, backupHistoryDBName)))
+ snapshotInfo, err := os.Stat(snapshotPath)
+ Expect(err).ToNot(HaveOccurred())
+ Expect(snapshotInfo.Mode().Perm()).To(Equal(os.FileMode(0o640)))
+
+ snapshotDB, err := sql.Open("sqlite3", backupHistoryStandbySyncSQLiteURI(snapshotPath, "ro"))
+ Expect(err).ToNot(HaveOccurred())
+ defer snapshotDB.Close()
+ var value string
+ Expect(snapshotDB.QueryRow("SELECT value FROM sync_test WHERE id = 1").Scan(&value)).To(Succeed())
+ Expect(value).To(Equal("present"))
+ Expect(validateBackupHistoryStandbySyncSnapshot(snapshotPath)).To(Succeed())
+ })
+
+ It("rejects corrupted SQLite sources before transport", func() {
+ tmpDir := GinkgoT().TempDir()
+ sourcePath := filepath.Join(tmpDir, backupHistoryDBName)
+ Expect(os.WriteFile(sourcePath, []byte("not sqlite"), 0o600)).To(Succeed())
+
+ _, tempDir, err := createBackupHistoryStandbySyncSnapshot(sourcePath, 0o600)
+ Expect(err).To(HaveOccurred())
+ Expect(err.Error()).To(ContainSubstring("VACUUM INTO"))
+ Expect(tempDir).To(BeEmpty())
+ })
+
+ It("returns SQLite close errors without changing the gpbackup error code", func() {
+ sqlDB, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ Expect(err).ToNot(HaveOccurred())
+ closeErr := errors.New("close failed")
+ mock.ExpectExec("VACUUM main INTO ?").
+ WithArgs("/tmp/snapshot.db").
+ WillReturnResult(sqlmock.NewResult(0, 0))
+ mock.ExpectClose().WillReturnError(closeErr)
+ backupHistoryStandbySyncOpenSQLite = func(driverName, dataSourceName string) (*sql.DB, error) {
+ Expect(driverName).To(Equal("sqlite3"))
+ return sqlDB, nil
+ }
+ originalErrorCode := gplog.GetErrorCode()
+ DeferCleanup(gplog.SetErrorCode, originalErrorCode)
+ gplog.SetErrorCode(0)
+
+ err = vacuumBackupHistoryStandbySyncSnapshot("/tmp/source.db", "/tmp/snapshot.db")
+
+ Expect(errors.Is(err, closeErr)).To(BeTrue())
+ Expect(err.Error()).To(ContainSubstring("close source history db for standby sync snapshot"))
+ Expect(gplog.GetErrorCode()).To(Equal(0))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("returns local cleanup errors after success and joins them with sync errors", func() {
+ sourcePath := filepath.Join(GinkgoT().TempDir(), backupHistoryDBName)
+ createBackupHistoryStandbySyncSQLiteDB(sourcePath)
+ cleanupErr := errors.New("cleanup failed")
+ backupHistoryStandbySyncMkdirTemp = func(dir, pattern string) (string, error) {
+ return GinkgoT().TempDir(), nil
+ }
+ backupHistoryStandbySyncRemoveAll = func(path string) error {
+ return cleanupErr
+ }
+
+ err := withBackupHistoryStandbySyncSnapshot(sourcePath, 0o600, func(string) error {
+ return nil
+ })
+ Expect(errors.Is(err, cleanupErr)).To(BeTrue())
+
+ syncErr := errors.New("sync failed")
+ err = withBackupHistoryStandbySyncSnapshot(sourcePath, 0o600, func(string) error {
+ return syncErr
+ })
+ Expect(errors.Is(err, syncErr)).To(BeTrue())
+ Expect(errors.Is(err, cleanupErr)).To(BeTrue())
+ })
+
+ It("joins snapshot creation and local cleanup errors", func() {
+ sourcePath := filepath.Join(GinkgoT().TempDir(), backupHistoryDBName)
+ Expect(os.WriteFile(sourcePath, []byte("not sqlite"), 0o600)).To(Succeed())
+ cleanupErr := errors.New("cleanup failed")
+ backupHistoryStandbySyncRemoveAll = func(path string) error {
+ return cleanupErr
+ }
+
+ _, tempDir, err := createBackupHistoryStandbySyncSnapshot(sourcePath, 0o600)
+
+ Expect(tempDir).To(BeEmpty())
+ Expect(err.Error()).To(ContainSubstring("VACUUM INTO"))
+ Expect(errors.Is(err, cleanupErr)).To(BeTrue())
+ })
+
+ It("canonicalizes symlink sources and builds the shared lock path from the canonical source", func() {
+ tmpDir := GinkgoT().TempDir()
+ realDir := filepath.Join(tmpDir, "real")
+ linkDir := filepath.Join(tmpDir, "link")
+ Expect(os.Mkdir(realDir, 0o700)).To(Succeed())
+ Expect(os.Mkdir(linkDir, 0o700)).To(Succeed())
+ // Resolve any symlinks in the OS temp dir itself (e.g. macOS /var -> /private/var)
+ // so the expected path matches the canonicalization performed by the code under test.
+ canonicalRealDir, err := filepath.EvalSymlinks(realDir)
+ Expect(err).ToNot(HaveOccurred())
+ realDir = canonicalRealDir
+ realSourcePath := filepath.Join(realDir, backupHistoryDBName)
+ createBackupHistoryStandbySyncSQLiteDB(realSourcePath)
+ linkSourcePath := filepath.Join(linkDir, backupHistoryDBName)
+ Expect(os.Symlink(realSourcePath, linkSourcePath)).To(Succeed())
+
+ canonicalSourcePath, _, err := canonicalBackupHistoryStandbySyncSource(linkSourcePath)
+ Expect(err).ToNot(HaveOccurred())
+ Expect(canonicalSourcePath).To(Equal(realSourcePath))
+ Expect(backupHistoryStandbySyncLockPath(canonicalSourcePath)).To(Equal(realSourcePath + ".sync.lock"))
+ })
+
+ It("rejects a non-regular source", func() {
+ sourcePath := GinkgoT().TempDir()
+
+ _, _, err := canonicalBackupHistoryStandbySyncSource(sourcePath)
+
+ Expect(err).To(MatchError(ContainSubstring("is not a regular file")))
+ })
+
+ It("skips when no up standby coordinator exists", func() {
+ tmpDir := GinkgoT().TempDir()
+ sourcePath := filepath.Join(tmpDir, backupHistoryDBName)
+ createBackupHistoryStandbySyncSQLiteDB(sourcePath)
+ globalFPInfo = backupfilepath.FilePathInfo{SegDirMap: map[int]string{-1: tmpDir}}
+ mock := setupBackupHistoryStandbySyncConnection()
+ mock.ExpectQuery(regexp.QuoteMeta(backupHistoryStandbySyncStandbySQL)).WillReturnError(sql.ErrNoRows)
+
+ commandCalls := setBackupHistoryStandbySyncCommands(nil)
+ skipReason, err := syncBackupHistoryToStandby()
+
+ Expect(err).ToNot(HaveOccurred())
+ Expect(skipReason).To(Equal("no up standby coordinator found"))
+ Expect(*commandCalls).To(BeEmpty())
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("orchestrates discovery, snapshot, rsync transport, atomic install, and local cleanup", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby data")
+ snapshotDir := filepath.Join(tmpDir, "snapshot dir")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ Expect(os.Mkdir(standbyDataDir, 0o700)).To(Succeed())
+ sourcePath := filepath.Join(primaryDataDir, backupHistoryDBName)
+ createBackupHistoryStandbySyncSQLiteDB(sourcePath)
+ globalFPInfo = backupfilepath.FilePathInfo{SegDirMap: map[int]string{-1: primaryDataDir}}
+ mock := setupBackupHistoryStandbySyncConnection()
+ expectBackupHistoryStandbySyncStandby(mock, "sdw-standby", standbyDataDir)
+ backupHistoryStandbySyncMkdirTemp = func(dir, pattern string) (string, error) {
+ Expect(dir).To(Equal(""))
+ Expect(pattern).To(Equal(backupHistoryStandbySyncTempDirPattern))
+ Expect(os.Mkdir(snapshotDir, 0o700)).To(Succeed())
+ return snapshotDir, nil
+ }
+
+ commandCalls := setBackupHistoryStandbySyncCommands([]backupHistoryStandbySyncCommandResponse{{}, {}})
+ start := time.Now()
+ Expect(cmdFlags.Set(options.HISTORY_SYNC_STANDBY_TIMEOUT, "600")).To(Succeed())
+ skipReason, err := syncBackupHistoryToStandby()
+ finished := time.Now()
+
+ Expect(err).ToNot(HaveOccurred())
+ Expect(skipReason).To(BeEmpty())
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ Expect(*commandCalls).To(HaveLen(2))
+ snapshotPath := filepath.Join(snapshotDir, backupHistoryDBName)
+ remoteTempPath := newBackupHistoryStandbySyncRemoteTempPath(standbyDataDir, snapshotPath)
+ Expect((*commandCalls)[0].name).To(Equal("rsync"))
+ Expect((*commandCalls)[0].args).To(Equal(buildBackupHistoryStandbySyncRsyncArgs(snapshotPath, "sdw-standby", "gpadmin", remoteTempPath)))
+ Expect((*commandCalls)[1].name).To(Equal("ssh"))
+ Expect((*commandCalls)[1].ctx).To(BeIdenticalTo((*commandCalls)[0].ctx))
+ deadline, ok := (*commandCalls)[0].ctx.Deadline()
+ Expect(ok).To(BeTrue())
+ Expect(deadline).To(BeTemporally(">=", start.Add(600*time.Second)))
+ Expect(deadline).To(BeTemporally("<=", finished.Add(600*time.Second)))
+ Expect((*commandCalls)[1].args).To(Equal([]string{
+ "-o",
+ "BatchMode=yes",
+ "-o",
+ "StrictHostKeyChecking=no",
+ "-o",
+ "ConnectTimeout=30",
+ "gpadmin@sdw-standby",
+ buildBackupHistoryStandbySyncRemoteInstallCommand(remoteTempPath, filepath.Join(standbyDataDir, backupHistoryDBName)),
+ }))
+ _, err = os.Stat(snapshotDir)
+ Expect(errors.Is(err, os.ErrNotExist)).To(BeTrue())
+ })
+
+ It("returns the rsync stage, configured seconds, and DeadlineExceeded without waiting", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourcePath := filepath.Join(primaryDataDir, backupHistoryDBName)
+ createBackupHistoryStandbySyncSQLiteDB(sourcePath)
+ globalFPInfo = backupfilepath.FilePathInfo{SegDirMap: map[int]string{-1: primaryDataDir}}
+ mock := setupBackupHistoryStandbySyncConnection()
+ expectBackupHistoryStandbySyncStandby(mock, "sdw-standby", standbyDataDir)
+ Expect(cmdFlags.Set(options.HISTORY_SYNC_STANDBY_TIMEOUT, "600")).To(Succeed())
+ backupHistoryStandbySyncContextWithTimeout = func(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) {
+ Expect(timeout).To(Equal(600 * time.Second))
+ return context.WithDeadline(parent, time.Now().Add(-time.Second))
+ }
+ commandCalls := setBackupHistoryStandbySyncCommands([]backupHistoryStandbySyncCommandResponse{
+ {err: context.DeadlineExceeded},
+ {},
+ })
+ start := time.Now()
+
+ _, err := syncBackupHistoryToStandby()
+ finished := time.Now()
+
+ Expect(err).To(HaveOccurred())
+ Expect(errors.Is(err, context.DeadlineExceeded)).To(BeTrue())
+ Expect(err.Error()).To(ContainSubstring("rsync standby history snapshot"))
+ Expect(err.Error()).To(ContainSubstring("timed out after 600 seconds"))
+ Expect(*commandCalls).To(HaveLen(2))
+ Expect((*commandCalls)[0].ctxErr).To(Equal(context.DeadlineExceeded))
+ Expect((*commandCalls)[1].ctx).ToNot(BeIdenticalTo((*commandCalls)[0].ctx))
+ Expect((*commandCalls)[1].ctxErr).ToNot(HaveOccurred())
+ Expect((*commandCalls)[1].hasDeadline).To(BeTrue())
+ Expect((*commandCalls)[1].deadline).To(BeTemporally(">=", start.Add(backupHistoryStandbySyncCleanupTimeout)))
+ Expect((*commandCalls)[1].deadline).To(BeTemporally("<=", finished.Add(backupHistoryStandbySyncCleanupTimeout)))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("releases the source lock after transport errors", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourcePath := filepath.Join(primaryDataDir, backupHistoryDBName)
+ createBackupHistoryStandbySyncSQLiteDB(sourcePath)
+ globalFPInfo = backupfilepath.FilePathInfo{SegDirMap: map[int]string{-1: primaryDataDir}}
+ mock := setupBackupHistoryStandbySyncConnection()
+ expectBackupHistoryStandbySyncStandby(mock, "sdw-standby", standbyDataDir)
+ setBackupHistoryStandbySyncCommands([]backupHistoryStandbySyncCommandResponse{
+ {output: []byte("rsync failed"), err: errors.New("exit status 1")},
+ {},
+ })
+
+ _, err := syncBackupHistoryToStandby()
+ Expect(err).To(HaveOccurred())
+ Expect(err.Error()).To(ContainSubstring("rsync standby history snapshot"))
+ Expect(err.Error()).To(ContainSubstring("rsync failed"))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+
+ sourceLock, err := lockfile.New(backupHistoryStandbySyncLockPath(sourcePath))
+ Expect(err).ToNot(HaveOccurred())
+ Expect(sourceLock.TryLock()).To(Succeed())
+ Expect(sourceLock.Unlock()).To(Succeed())
+ })
+
+ It("returns lock contention as an error without creating a snapshot", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourcePath := filepath.Join(primaryDataDir, backupHistoryDBName)
+ createBackupHistoryStandbySyncSQLiteDB(sourcePath)
+ globalFPInfo = backupfilepath.FilePathInfo{SegDirMap: map[int]string{-1: primaryDataDir}}
+ mock := setupBackupHistoryStandbySyncConnection()
+ expectBackupHistoryStandbySyncStandby(mock, "sdw-standby", standbyDataDir)
+ lockPath := backupHistoryStandbySyncLockPath(sourcePath)
+ Expect(os.WriteFile(lockPath, []byte(fmt.Sprintf("%d\n", os.Getppid())), 0o600)).To(Succeed())
+ defer os.Remove(lockPath)
+ mkdirTempCalls := 0
+ backupHistoryStandbySyncMkdirTemp = func(dir, pattern string) (string, error) {
+ mkdirTempCalls++
+ return "", errors.New("snapshot should not be created")
+ }
+
+ _, err := syncBackupHistoryToStandby()
+
+ Expect(err).To(HaveOccurred())
+ Expect(err.Error()).To(ContainSubstring("lock standby history sync source"))
+ Expect(mkdirTempCalls).To(Equal(0))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("protects rsync paths and quotes remote shell paths", func() {
+ remoteTempPath := "/data dir/standby's/.gpbackup_history.db.tmp"
+ destPath := "/data dir/standby's/gpbackup_history.db"
+ Expect(backupHistoryStandbySyncSSHOptions).To(ContainSubstring("BatchMode=yes"))
+
+ Expect(buildBackupHistoryStandbySyncRsyncArgs("/tmp/snapshot", "sdw-standby", "gpadmin", remoteTempPath)).To(Equal([]string{
+ "-p",
+ "-s",
+ "-e",
+ backupHistoryStandbySyncSSHOptions,
+ "--",
+ "/tmp/snapshot",
+ "gpadmin@sdw-standby:" + remoteTempPath,
+ }))
+ installCommand := buildBackupHistoryStandbySyncRemoteInstallCommand(remoteTempPath, destPath)
+ Expect(installCommand).To(ContainSubstring("test -f " + shellQuoteBackupHistoryStandbySyncPath(remoteTempPath)))
+ Expect(installCommand).To(ContainSubstring("chown --reference=" + shellQuoteBackupHistoryStandbySyncPath(destPath) + " -- " + shellQuoteBackupHistoryStandbySyncPath(remoteTempPath)))
+ Expect(installCommand).To(ContainSubstring("chmod --reference=" + shellQuoteBackupHistoryStandbySyncPath(destPath) + " -- " + shellQuoteBackupHistoryStandbySyncPath(remoteTempPath)))
+ Expect(installCommand).To(ContainSubstring("mv -f -- " + shellQuoteBackupHistoryStandbySyncPath(remoteTempPath) + " " + shellQuoteBackupHistoryStandbySyncPath(destPath)))
+ Expect(buildBackupHistoryStandbySyncRemoteCleanupCommand(remoteTempPath)).To(Equal("rm -f -- " + shellQuoteBackupHistoryStandbySyncPath(remoteTempPath)))
+ })
+
+ It("keeps protected rsync paths out of the remote shell command", func() {
+ rsyncPath := requireBackupHistoryStandbySyncRsync3()
+ tmpDir := GinkgoT().TempDir()
+ remoteArgsPath := filepath.Join(tmpDir, "remote-args")
+ fakeShellPath := filepath.Join(tmpDir, "fake-ssh")
+ fakeShell := "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$RSYNC_REMOTE_ARGS_FILE\"\nexit 1\n"
+ Expect(os.WriteFile(fakeShellPath, []byte(fakeShell), 0o700)).To(Succeed())
+
+ snapshotPath := filepath.Join(tmpDir, "snapshot")
+ Expect(os.WriteFile(snapshotPath, []byte("snapshot"), 0o600)).To(Succeed())
+ remoteTempPath := "/data dir/standby's/$HOME/[history]*;RSYNC_REMOTE_PATH_SENTINEL"
+ args := buildBackupHistoryStandbySyncRsyncArgs(snapshotPath, "sdw-standby", "gpadmin", remoteTempPath)
+ remoteShellReplaced := false
+ for i := range args {
+ if args[i] == "-e" && i+1 < len(args) {
+ args[i+1] = fakeShellPath
+ remoteShellReplaced = true
+ break
+ }
+ }
+ Expect(remoteShellReplaced).To(BeTrue())
+
+ command := exec.Command(rsyncPath, args...)
+ command.Env = append(os.Environ(), "RSYNC_REMOTE_ARGS_FILE="+remoteArgsPath)
+ _, err := command.CombinedOutput()
+ Expect(err).To(HaveOccurred())
+
+ remoteArgs, err := os.ReadFile(remoteArgsPath)
+ Expect(err).ToNot(HaveOccurred())
+ Expect(string(remoteArgs)).To(ContainSubstring("--server"))
+ Expect(string(remoteArgs)).ToNot(ContainSubstring("RSYNC_REMOTE_PATH_SENTINEL"))
+ })
+
+ It("chains remote cleanup errors onto the primary transport error", func() {
+ cleanupCommandErr := fmt.Errorf("cleanup timeout: %w", context.DeadlineExceeded)
+ commandCalls := setBackupHistoryStandbySyncCommands([]backupHistoryStandbySyncCommandResponse{
+ {output: []byte("cleanup failed"), err: cleanupCommandErr},
+ })
+ primaryErr := errors.New("install failed")
+ start := time.Now()
+
+ err := cleanupBackupHistoryStandbySyncRemoteTempAfterError(primaryErr, "sdw-standby", "gpadmin", "/standby/.tmp")
+ finished := time.Now()
+
+ Expect(err).To(HaveOccurred())
+ Expect(errors.Is(err, primaryErr)).To(BeTrue())
+ Expect(errors.Is(err, cleanupCommandErr)).To(BeTrue())
+ Expect(errors.Is(err, context.DeadlineExceeded)).To(BeTrue())
+ Expect(err.Error()).To(ContainSubstring("install failed"))
+ Expect(err.Error()).To(ContainSubstring("additionally failed to clean up remote temp file"))
+ Expect(err.Error()).To(ContainSubstring("cleanup failed"))
+ Expect(*commandCalls).To(HaveLen(1))
+ Expect((*commandCalls)[0].name).To(Equal("ssh"))
+ Expect((*commandCalls)[0].ctxErr).ToNot(HaveOccurred())
+ Expect((*commandCalls)[0].hasDeadline).To(BeTrue())
+ Expect((*commandCalls)[0].deadline).To(BeTemporally(">=", start.Add(backupHistoryStandbySyncCleanupTimeout)))
+ Expect((*commandCalls)[0].deadline).To(BeTemporally("<=", finished.Add(backupHistoryStandbySyncCleanupTimeout)))
+ })
+
+ It("logs disabled automatic sync without invoking discovery", func() {
+ stdout, _, _ := testhelper.SetupTestLogger()
+ syncCalls := 0
+ backupHistoryStandbySync = func() (string, error) {
+ syncCalls++
+ return "", errors.New("sync should not run")
+ }
+
+ skipReason, err := syncBackupHistoryToStandbyBestEffort(true)
+
+ Expect(err).ToNot(HaveOccurred())
+ Expect(skipReason).To(Equal("disabled by --" + options.NO_HISTORY_SYNC_STANDBY))
+ Expect(syncCalls).To(Equal(0))
+ Expect(string(stdout.Contents())).To(ContainSubstring("Skipping history db sync to standby coordinator: disabled by --" + options.NO_HISTORY_SYNC_STANDBY))
+ })
+
+ It("warns automatic sync failures without exiting", func() {
+ stdout, _, _ := testhelper.SetupTestLogger()
+ originalErrorCode := gplog.GetErrorCode()
+ DeferCleanup(gplog.SetErrorCode, originalErrorCode)
+ gplog.SetErrorCode(0)
+ backupHistoryStandbySync = func() (string, error) {
+ return "", fmt.Errorf("standby history sync transport timed out after 300 seconds: %w", context.DeadlineExceeded)
+ }
+
+ _, err := syncBackupHistoryToStandbyBestEffort(false)
+
+ Expect(err).To(HaveOccurred())
+ Expect(errors.Is(err, context.DeadlineExceeded)).To(BeTrue())
+ Expect(string(stdout.Contents())).To(ContainSubstring("History db sync to standby coordinator failed; standby history may be stale: standby history sync transport timed out after 300 seconds"))
+ Expect(gplog.GetErrorCode()).To(Equal(0))
+ })
+
+ It("runs automatic sync only after successful cleanup history update for successful backups", func() {
+ calls := 0
+ disabledValues := make([]bool, 0)
+ originalBestEffort := backupHistoryStandbySync
+ backupHistoryStandbySync = func() (string, error) {
+ calls++
+ return "", nil
+ }
+ defer func() {
+ backupHistoryStandbySync = originalBestEffort
+ }()
+ backupHistoryStandbySyncCommandExec = func(ctx context.Context, name string, args ...string) backupHistoryStandbySyncCommand {
+ return backupHistoryStandbySyncFakeCommand{}
+ }
+
+ syncBackupHistoryToStandbyAfterCleanup(false, true)
+ Expect(calls).To(Equal(1))
+
+ syncBackupHistoryToStandbyAfterCleanup(true, true)
+ syncBackupHistoryToStandbyAfterCleanup(false, false)
+ Expect(cmdFlags.Set(options.NO_HISTORY, "true")).To(Succeed())
+ syncBackupHistoryToStandbyAfterCleanup(false, true)
+ Expect(calls).To(Equal(1))
+
+ Expect(cmdFlags.Set(options.NO_HISTORY, "false")).To(Succeed())
+ Expect(cmdFlags.Set(options.NO_HISTORY_SYNC_STANDBY, "true")).To(Succeed())
+ backupHistoryStandbySync = func() (string, error) {
+ calls++
+ disabledValues = append(disabledValues, true)
+ return "", nil
+ }
+ syncBackupHistoryToStandbyAfterCleanup(false, true)
+ Expect(calls).To(Equal(1))
+ Expect(disabledValues).To(BeEmpty())
+ })
+})
+
+func requireBackupHistoryStandbySyncRsync3() string {
+ GinkgoHelper()
+
+ rsyncPath, err := exec.LookPath("rsync")
+ if err != nil {
+ Skip("rsync is not installed")
+ return ""
+ }
+ output, err := exec.Command(rsyncPath, "--version").CombinedOutput()
+ if err != nil {
+ Skip(fmt.Sprintf("cannot determine rsync version: %v", err))
+ return ""
+ }
+ match := regexp.MustCompile(`(?m)^rsync\s+version\s+([0-9]+)\.`).FindStringSubmatch(string(output))
+ if len(match) != 2 {
+ Skip("cannot parse rsync version")
+ return ""
+ }
+ majorVersion, err := strconv.Atoi(match[1])
+ if err != nil || majorVersion < 3 {
+ Skip("rsync 3.0.0 or later is required")
+ return ""
+ }
+ return rsyncPath
+}
+
+func createBackupHistoryStandbySyncSQLiteDB(path string) {
+ db, err := sql.Open("sqlite3", backupHistoryStandbySyncSQLiteURI(path, "rwc"))
+ Expect(err).ToNot(HaveOccurred())
+ defer db.Close()
+
+ _, err = db.Exec("CREATE TABLE sync_test (id INTEGER PRIMARY KEY, value TEXT)")
+ Expect(err).ToNot(HaveOccurred())
+ _, err = db.Exec("INSERT INTO sync_test (value) VALUES ('present')")
+ Expect(err).ToNot(HaveOccurred())
+}
+
+func setupBackupHistoryStandbySyncConnection() sqlmock.Sqlmock {
+ sqlDB, mock, err := sqlmock.New()
+ Expect(err).ToNot(HaveOccurred())
+ connectionPool = &dbconn.DBConn{
+ ConnPool: []*sqlx.DB{sqlx.NewDb(sqlDB, "sqlmock")},
+ NumConns: 1,
+ Tx: []*sqlx.Tx{nil},
+ }
+ return mock
+}
+
+func expectBackupHistoryStandbySyncStandby(mock sqlmock.Sqlmock, host, dataDir string) {
+ mock.ExpectQuery(regexp.QuoteMeta(backupHistoryStandbySyncStandbySQL)).
+ WillReturnRows(sqlmock.NewRows([]string{"hostname", "datadir"}).AddRow(host, dataDir))
+}
+
+func setBackupHistoryStandbySyncCommands(responses []backupHistoryStandbySyncCommandResponse) *[]backupHistoryStandbySyncCommandCall {
+ calls := make([]backupHistoryStandbySyncCommandCall, 0)
+ backupHistoryStandbySyncCommandExec = func(ctx context.Context, name string, args ...string) backupHistoryStandbySyncCommand {
+ deadline, hasDeadline := ctx.Deadline()
+ calls = append(calls, backupHistoryStandbySyncCommandCall{
+ ctx: ctx,
+ ctxErr: ctx.Err(),
+ deadline: deadline,
+ hasDeadline: hasDeadline,
+ name: name,
+ args: append([]string{}, args...),
+ })
+ response := backupHistoryStandbySyncCommandResponse{}
+ if len(calls) <= len(responses) {
+ response = responses[len(calls)-1]
+ }
+ return backupHistoryStandbySyncFakeCommand{output: response.output, err: response.err}
+ }
+ return &calls
+}
diff --git a/backup/validate.go b/backup/validate.go
index e3169d6..2902d1a 100644
--- a/backup/validate.go
+++ b/backup/validate.go
@@ -165,6 +165,10 @@
}
func validateFlagValues() {
+ timeoutSeconds := MustGetFlagInt(options.HISTORY_SYNC_STANDBY_TIMEOUT)
+ if timeoutSeconds <= 0 || timeoutSeconds > maxHistorySyncStandbyTimeoutSeconds {
+ gplog.Fatal(errors.Errorf("--%s must be between 1 and %d seconds", options.HISTORY_SYNC_STANDBY_TIMEOUT, maxHistorySyncStandbyTimeoutSeconds), "")
+ }
err := utils.ValidateFullPath(MustGetFlagString(options.BACKUP_DIR))
gplog.FatalOnError(err)
err = utils.ValidateFullPath(MustGetFlagString(options.PLUGIN_CONFIG))
diff --git a/backup/validate_test.go b/backup/validate_test.go
index 29e200c..8267825 100644
--- a/backup/validate_test.go
+++ b/backup/validate_test.go
@@ -215,6 +215,10 @@
}
},
Entry("--backup-dir combo", "--backup-dir /tmp --plugin-config /tmp/config", false),
+ Entry("standby history sync timeout must be positive", "--history-sync-standby-timeout 0", false),
+ Entry("standby history sync timeout must not be negative", "--history-sync-standby-timeout -1", false),
+ Entry("standby history sync timeout accepts one day", "--history-sync-standby-timeout 86400", true),
+ Entry("standby history sync timeout must not exceed one day", "--history-sync-standby-timeout 86401", false),
/*
* Below are all the different filter combinations
diff --git a/end_to_end/end_to_end_suite_test.go b/end_to_end/end_to_end_suite_test.go
index cbbfc90..75d60a3 100644
--- a/end_to_end/end_to_end_suite_test.go
+++ b/end_to_end/end_to_end_suite_test.go
@@ -90,17 +90,37 @@
* to allow checking its output.
*/
func gpbackup(gpbackupPath string, backupHelperPath string, args ...string) []byte {
+ return runGpbackup(gpbackupPath, backupHelperPath, true, args...)
+}
+
+func gpbackupWithHistoryStandbySync(gpbackupPath string, backupHelperPath string, args ...string) []byte {
+ return runGpbackup(gpbackupPath, backupHelperPath, false, args...)
+}
+
+func runGpbackup(gpbackupPath string, backupHelperPath string, disableHistoryStandbySync bool, args ...string) []byte {
if useOldBackupVersion {
_ = os.Chdir("..")
command := exec.Command("make", "install", fmt.Sprintf("helper_path=%s", backupHelperPath))
mustRunCommand(command)
_ = os.Chdir("end_to_end")
}
+ if disableHistoryStandbySync && !useOldBackupVersion && !hasCommandArgument(args, "--no-history-sync-standby") {
+ args = append(args, "--no-history-sync-standby")
+ }
args = append([]string{"--verbose", "--dbname", "testdb"}, args...)
command := exec.Command(gpbackupPath, args...)
return mustRunCommand(command)
}
+func hasCommandArgument(args []string, expected string) bool {
+ for _, arg := range args {
+ if arg == expected {
+ return true
+ }
+ }
+ return false
+}
+
func gprestore(gprestorePath string, restoreHelperPath string, timestamp string, args ...string) []byte {
if useOldBackupVersion {
_ = os.Chdir("..")
@@ -469,15 +489,37 @@
// gpbackman helpers
func gpbackman(args ...string) []byte {
+ return runGpbackman(true, args...)
+}
+
+func gpbackmanWithHistoryStandbySync(args ...string) []byte {
+ return runGpbackman(false, args...)
+}
+
+func runGpbackman(disableHistoryStandbySync bool, args ...string) []byte {
+ args = gpbackmanArgsWithHistoryStandbySyncPolicy(disableHistoryStandbySync, args)
command := exec.Command(gpbackmanPath, args...)
return mustRunCommand(command)
}
func gpbackmanWithError(args ...string) ([]byte, error) {
+ args = gpbackmanArgsWithHistoryStandbySyncPolicy(true, args)
command := exec.Command(gpbackmanPath, args...)
return command.CombinedOutput()
}
+func gpbackmanArgsWithHistoryStandbySyncPolicy(disabled bool, args []string) []string {
+ if !disabled || len(args) == 0 || hasCommandArgument(args, "--no-history-sync-standby") {
+ return args
+ }
+ switch args[0] {
+ case "backup-delete", "backup-clean", "history-clean":
+ return append(args, "--no-history-sync-standby")
+ default:
+ return args
+ }
+}
+
func getHistoryDBPathForCluster() string {
mdd := backupCluster.GetDirForContent(-1)
return path.Join(mdd, "gpbackup_history.db")
diff --git a/end_to_end/history_standby_sync_test.go b/end_to_end/history_standby_sync_test.go
new file mode 100644
index 0000000..794847c
--- /dev/null
+++ b/end_to_end/history_standby_sync_test.go
@@ -0,0 +1,272 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package end_to_end_test
+
+import (
+ "bytes"
+ "database/sql"
+ "fmt"
+ "net/url"
+ "os"
+ "os/exec"
+ stdpath "path/filepath"
+ "strings"
+ "time"
+
+ . "github.com/onsi/ginkgo/v2"
+ . "github.com/onsi/gomega"
+)
+
+const upStandbyCoordinatorQuery = `
+ SELECT hostname, datadir
+ FROM gp_segment_configuration
+ WHERE content = -1 AND role = 'm' AND status = 'u'`
+
+type standbyCoordinatorTarget struct {
+ Hostname string `db:"hostname"`
+ DataDir string `db:"datadir"`
+}
+
+type historyLogicalRow struct {
+ Timestamp string
+ Status string
+ DateDeleted string
+}
+
+func discoverUpStandbyCoordinator() standbyCoordinatorTarget {
+ var targets []standbyCoordinatorTarget
+ err := backupConn.Select(&targets, upStandbyCoordinatorQuery)
+ Expect(err).ToNot(HaveOccurred())
+ if len(targets) == 0 {
+ Skip("standby history sync requires an up standby coordinator")
+ }
+ Expect(targets).To(HaveLen(1), "expected exactly one up standby coordinator")
+ return targets[0]
+}
+
+func quoteRemoteShellPath(value string) string {
+ return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
+}
+
+func copyStandbyHistoryDB(target standbyCoordinatorTarget) string {
+ tempDir, err := os.MkdirTemp("", "gpbackup-history-standby-e2e-")
+ Expect(err).ToNot(HaveOccurred())
+ DeferCleanup(func() {
+ Expect(os.RemoveAll(tempDir)).To(Succeed())
+ })
+
+ localPath := stdpath.Join(tempDir, "gpbackup_history.db")
+ localFile, err := os.OpenFile(localPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600)
+ Expect(err).ToNot(HaveOccurred())
+
+ remotePath := stdpath.Join(target.DataDir, "gpbackup_history.db")
+ remoteCommand := fmt.Sprintf("cat -- %s", quoteRemoteShellPath(remotePath))
+ command := exec.Command(
+ "ssh",
+ "-o", "StrictHostKeyChecking=no",
+ "-o", "ConnectTimeout=30",
+ target.Hostname,
+ remoteCommand,
+ )
+ command.Stdout = localFile
+ var stderr bytes.Buffer
+ command.Stderr = &stderr
+
+ runErr := command.Run()
+ closeErr := localFile.Close()
+ Expect(runErr).ToNot(HaveOccurred(), "copy %s:%s: %s", target.Hostname, remotePath, strings.TrimSpace(stderr.String()))
+ Expect(closeErr).ToNot(HaveOccurred())
+ return localPath
+}
+
+func preserveStandbyHistoryDB(target standbyCoordinatorTarget) {
+ remotePath := stdpath.Join(target.DataDir, "gpbackup_history.db")
+ savedPath := fmt.Sprintf("%s.end-to-end-%d-%d", remotePath, os.Getpid(), time.Now().UnixNano())
+ quotedRemotePath := quoteRemoteShellPath(remotePath)
+ quotedSavedPath := quoteRemoteShellPath(savedPath)
+ saveCommand := fmt.Sprintf(
+ "if test -f %s; then test ! -e %s && test ! -L %s && cp -p -- %s %s && printf present; elif test ! -e %s && test ! -L %s; then printf absent; else exit 1; fi",
+ quotedRemotePath,
+ quotedSavedPath,
+ quotedSavedPath,
+ quotedRemotePath,
+ quotedSavedPath,
+ quotedRemotePath,
+ quotedRemotePath,
+ )
+ state := strings.TrimSpace(string(runStandbyHistorySSHCommand(target, saveCommand)))
+ Expect(state).To(Or(Equal("present"), Equal("absent")))
+
+ DeferCleanup(func() {
+ var restoreCommand string
+ if state == "present" {
+ restoreCommand = fmt.Sprintf(
+ "test -f %s && test ! -d %s && mv -f -- %s %s",
+ quotedSavedPath,
+ quotedRemotePath,
+ quotedSavedPath,
+ quotedRemotePath,
+ )
+ } else {
+ restoreCommand = fmt.Sprintf(
+ "test ! -d %s && rm -f -- %s %s",
+ quotedRemotePath,
+ quotedRemotePath,
+ quotedSavedPath,
+ )
+ }
+ runStandbyHistorySSHCommand(target, restoreCommand)
+ })
+}
+
+func runStandbyHistorySSHCommand(target standbyCoordinatorTarget, remoteCommand string) []byte {
+ command := exec.Command(
+ "ssh",
+ "-o", "StrictHostKeyChecking=no",
+ "-o", "ConnectTimeout=30",
+ target.Hostname,
+ remoteCommand,
+ )
+ var stderr bytes.Buffer
+ command.Stderr = &stderr
+ output, err := command.Output()
+ Expect(err).ToNot(HaveOccurred(), "run standby history command on %s: %s", target.Hostname, strings.TrimSpace(stderr.String()))
+ return output
+}
+
+func readHistoryLogicalRows(historyDBPath string) []historyLogicalRow {
+ dsn := (&url.URL{Scheme: "file", Path: historyDBPath}).String() + "?mode=ro"
+ db, err := sql.Open("sqlite3", dsn)
+ Expect(err).ToNot(HaveOccurred())
+ defer db.Close()
+
+ var quickCheck string
+ err = db.QueryRow("PRAGMA quick_check").Scan(&quickCheck)
+ Expect(err).ToNot(HaveOccurred())
+ Expect(quickCheck).To(Equal("ok"))
+
+ rows, err := db.Query(`
+ SELECT timestamp, status, date_deleted
+ FROM backups
+ ORDER BY timestamp`)
+ Expect(err).ToNot(HaveOccurred())
+ defer rows.Close()
+
+ logicalRows := make([]historyLogicalRow, 0)
+ for rows.Next() {
+ var row historyLogicalRow
+ Expect(rows.Scan(&row.Timestamp, &row.Status, &row.DateDeleted)).To(Succeed())
+ logicalRows = append(logicalRows, row)
+ }
+ Expect(rows.Err()).ToNot(HaveOccurred())
+ return logicalRows
+}
+
+func findHistoryLogicalRow(rows []historyLogicalRow, timestamp string) historyLogicalRow {
+ for _, row := range rows {
+ if row.Timestamp == timestamp {
+ return row
+ }
+ }
+ Fail(fmt.Sprintf("history row %s was not found", timestamp))
+ return historyLogicalRow{}
+}
+
+var _ = Describe("history database standby sync", func() {
+ var (
+ primaryHistoryDB string
+ standbyTarget standbyCoordinatorTarget
+ )
+
+ BeforeEach(func() {
+ if useOldBackupVersion {
+ Skip("standby history sync is not applicable in old backup version mode")
+ }
+ end_to_end_setup()
+ standbyTarget = discoverUpStandbyCoordinator()
+ preserveStandbyHistoryDB(standbyTarget)
+ primaryHistoryDB = getHistoryDBPathForCluster()
+ })
+
+ AfterEach(func() {
+ end_to_end_teardown()
+ })
+
+ It("keeps standby history logically consistent across automatic, disabled, explicit, and mutation sync", func() {
+ baselineOutput := gpbackupWithHistoryStandbySync(
+ gpbackupPath,
+ backupHelperPath,
+ "--backup-dir", backupDir,
+ )
+ baselineTimestamp := getBackupTimestamp(string(baselineOutput))
+ Expect(baselineTimestamp).ToNot(BeEmpty())
+
+ primaryBaseline := readHistoryLogicalRows(primaryHistoryDB)
+ standbyBaseline := readHistoryLogicalRows(copyStandbyHistoryDB(standbyTarget))
+ Expect(standbyBaseline).To(Equal(primaryBaseline))
+ Expect(findHistoryLogicalRow(standbyBaseline, baselineTimestamp).Status).To(Equal("Success"))
+
+ disabledOutput := gpbackup(
+ gpbackupPath,
+ backupHelperPath,
+ "--backup-dir", backupDir,
+ "--no-history-sync-standby",
+ )
+ disabledTimestamp := getBackupTimestamp(string(disabledOutput))
+ Expect(disabledTimestamp).ToNot(BeEmpty())
+
+ primaryAfterDisabledSync := readHistoryLogicalRows(primaryHistoryDB)
+ Expect(findHistoryLogicalRow(primaryAfterDisabledSync, disabledTimestamp).Status).To(Equal("Success"))
+ standbyAfterDisabledSync := readHistoryLogicalRows(copyStandbyHistoryDB(standbyTarget))
+ Expect(standbyAfterDisabledSync).To(Equal(standbyBaseline))
+
+ historySyncCommand := exec.Command(
+ gpbackmanPath,
+ "history-sync",
+ "--auto-load-history-db",
+ )
+ historySyncCommand.Env = append(
+ os.Environ(),
+ fmt.Sprintf("COORDINATOR_DATA_DIRECTORY=%s", stdpath.Dir(primaryHistoryDB)),
+ )
+ mustRunCommand(historySyncCommand)
+
+ primaryAfterExplicitSync := readHistoryLogicalRows(primaryHistoryDB)
+ standbyAfterExplicitSync := readHistoryLogicalRows(copyStandbyHistoryDB(standbyTarget))
+ Expect(standbyAfterExplicitSync).To(Equal(primaryAfterExplicitSync))
+ Expect(findHistoryLogicalRow(standbyAfterExplicitSync, disabledTimestamp).Status).To(Equal("Success"))
+
+ gpbackmanWithHistoryStandbySync(
+ "backup-delete",
+ "--history-db", primaryHistoryDB,
+ "--timestamp", baselineTimestamp,
+ "--backup-dir", backupDir,
+ )
+
+ primaryAfterDelete := readHistoryLogicalRows(primaryHistoryDB)
+ deletedRow := findHistoryLogicalRow(primaryAfterDelete, baselineTimestamp)
+ Expect(deletedRow.DateDeleted).ToNot(BeEmpty())
+ Expect(deletedRow.DateDeleted).ToNot(Equal("In progress"))
+
+ standbyAfterDelete := readHistoryLogicalRows(copyStandbyHistoryDB(standbyTarget))
+ Expect(standbyAfterDelete).To(Equal(primaryAfterDelete))
+ Expect(findHistoryLogicalRow(standbyAfterDelete, baselineTimestamp).DateDeleted).To(Equal(deletedRow.DateDeleted))
+ })
+})
diff --git a/gpbackman/COMMANDS.md b/gpbackman/COMMANDS.md
index d9e43d4..715536b 100644
--- a/gpbackman/COMMANDS.md
+++ b/gpbackman/COMMANDS.md
@@ -17,6 +17,7 @@
under the License.
-->
+- [Standby history DB sync](#standby-history-db-sync)
- [Delete all existing backups older than the specified time condition (`backup-clean`)](#delete-all-existing-backups-older-than-the-specified-time-condition-backup-clean)
- [Examples](#examples)
- [Delete all backups from local storage older than the specified time condition](#delete-all-backups-from-local-storage-older-than-the-specified-time-condition)
@@ -31,17 +32,33 @@
- [Examples](#examples-3)
- [Delete information about deleted backups from history database older than n days](#delete-information-about-deleted-backups-from-history-database-older-than-n-days)
- [Delete information about deleted backups from history database older than timestamp](#delete-information-about-deleted-backups-from-history-database-older-than-timestamp)
-- [Display the report for a specific backup (`report-info`)](#display-the-report-for-a-specific-backup-report-info)
+- [Sync the history database to the standby coordinator (`history-sync`)](#sync-the-history-database-to-the-standby-coordinator-history-sync)
- [Examples](#examples-4)
+- [Display the report for a specific backup (`report-info`)](#display-the-report-for-a-specific-backup-report-info)
+ - [Examples](#examples-5)
- [Display the backup report from local storage](#display-the-backup-report-from-local-storage)
- [Display the backup report using storage plugin](#display-the-backup-report-using-storage-plugin)
+# Standby history DB sync
+
+The explicit `history-sync` command synchronizes the cluster `gpbackup_history.db` to an up standby coordinator. It does not have successful skips: an unavailable standby, an ineligible source, or a discovery, snapshot, validation, SSH, rsync, or cleanup error is reported as an error and returns a non-zero exit status.
+
+The source must resolve to the cluster history database at `<primary coordinator data directory>/gpbackup_history.db`. Select it with `--history-db`, or use `--auto-load-history-db` when `$COORDINATOR_DATA_DIRECTORY` points to the primary coordinator data directory. Custom history databases and the default working-directory database are not eligible for explicit synchronization.
+
+After a successful `backup-delete`, `backup-clean`, or `history-clean`, gpBackMan also attempts this synchronization automatically. Automatic sync is best-effort: `--no-history-sync-standby` produces an info-level skip, while no up standby and ineligible sources are debug-only skips; sync failures are warnings and do not change the successful primary command result. Read-only commands do not trigger automatic sync.
+
+The `history-sync`, `backup-delete`, `backup-clean`, and `history-clean` commands accept `--history-sync-standby-timeout SECONDS`. The default is 300 seconds. `SECONDS` must be an integer from 1 to 86400 seconds; `0`, `86401`, fractions, and duration strings such as `5m` are rejected. The 24-hour upper bound guards against accidentally oversized values; a longer timeout is not meaningful for this synchronization. This value is one shared budget for `rsync` and remote install, not a separate timeout for each command. The timeout starts only after snapshot validation and does not include standby discovery, `VACUUM INTO`, or `PRAGMA quick_check`. Remote cleanup after a transport failure uses a separate fixed timeout of 120 seconds. Read-only commands do not accept this option.
+
+`rsync` 3.0.0 or later must be installed on both the host running gpBackMan and the standby coordinator. The current OS user must have non-interactive SSH access to the standby host.
+
+Only `gpbackup_history.db` is synchronized. Report files, backup data, and other backup artifacts are not synchronized.
+
# Delete all existing backups older than the specified time condition (`backup-clean`)
Available options for `backup-clean` command and their description:
```bash
./gpbackman backup-clean -h
-elete all existing backups older than the specified time condition.
+Delete all existing backups older than the specified time condition.
To delete backup sets older than the given timestamp, use the --before-timestamp option.
To delete backup sets older than the given number of days, use the --older-than-day option.
@@ -72,7 +89,7 @@
The gpbackup_history.db file location can be set using the --history-db option.
Can be specified only once. The full path to the file is required.
-If the --history-db option is not specified, the history database is looked for in the current directory. Pass `--auto-load-history-db` to resolve it from `$COORDINATOR_DATA_DIRECTORY` instead.
+If the --history-db option is not specified, the history database is looked for in the current directory. To resolve it from $COORDINATOR_DATA_DIRECTORY instead, pass the --auto-load-history-db flag.
Usage:
gpbackman backup-clean [flags]
@@ -83,11 +100,14 @@
--before-timestamp string delete backup sets older than the given timestamp
--cascade delete all dependent backups
-h, --help help for backup-clean
+ --history-sync-standby-timeout int shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400 (default 300)
+ --no-history-sync-standby skip automatic gpbackup_history.db sync to standby coordinator after this command
--older-than-days uint delete backup sets older than the given number of days
--parallel-processes int the number of parallel processes to delete local backups (default 1)
--plugin-config string the full path to plugin config file
Global Flags:
+ --auto-load-history-db resolve gpbackup_history.db from $COORDINATOR_DATA_DIRECTORY when --history-db is unset
--history-db string full path to the gpbackup_history.db file
--log-file string full path to log file directory, if not specified, the log file will be created in the $HOME/gpAdminLogs directory
--log-level-console string level for console logging (error, info, debug, verbose) (default "info")
@@ -97,16 +117,16 @@
## Examples
### Delete all backups from local storage older than the specified time condition
-Delete specific backup :
+Delete backups older than a timestamp:
```bash
./gpbackman backup-clean \
--before-timestamp 20240701100000 \
--cascade
```
-Delete specific backup with specifying the number of parallel processes:
+Delete backups older than a number of days with multiple parallel processes:
```bash
-./gpbackman backup-delete \
+./gpbackman backup-clean \
--older-than-days 7 \
--parallel-processes 5
```
@@ -158,7 +178,7 @@
The gpbackup_history.db file location can be set using the --history-db option.
Can be specified only once. The full path to the file is required.
-If the --history-db option is not specified, the history database is looked for in the current directory. Pass `--auto-load-history-db` to resolve it from `$COORDINATOR_DATA_DIRECTORY` instead.
+If the --history-db option is not specified, the history database is looked for in the current directory. To resolve it from $COORDINATOR_DATA_DIRECTORY instead, pass the --auto-load-history-db flag.
Usage:
gpbackman backup-delete [flags]
@@ -168,12 +188,15 @@
--cascade delete all dependent backups for the specified backup timestamp
--force try to delete, even if the backup already mark as deleted
-h, --help help for backup-delete
+ --history-sync-standby-timeout int shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400 (default 300)
--ignore-errors ignore errors when deleting backups
+ --no-history-sync-standby skip automatic gpbackup_history.db sync to standby coordinator after this command
--parallel-processes int the number of parallel processes to delete local backups (default 1)
--plugin-config string the full path to plugin config file
--timestamp stringArray the backup timestamp for deleting, could be specified multiple times
Global Flags:
+ --auto-load-history-db resolve gpbackup_history.db from $COORDINATOR_DATA_DIRECTORY when --history-db is unset
--history-db string full path to the gpbackup_history.db file
--log-file string full path to log file directory, if not specified, the log file will be created in the $HOME/gpAdminLogs directory
--log-level-console string level for console logging (error, info, debug, verbose) (default "info")
@@ -256,7 +279,7 @@
The gpbackup_history.db file location can be set using the --history-db option.
Can be specified only once. The full path to the file is required.
-If the --history-db option is not specified, the history database is looked for in the current directory. Pass `--auto-load-history-db` to resolve it from `$COORDINATOR_DATA_DIRECTORY` instead.
+If the --history-db option is not specified, the history database is looked for in the current directory. To resolve it from $COORDINATOR_DATA_DIRECTORY instead, pass the --auto-load-history-db flag.
Usage:
gpbackman backup-info [flags]
@@ -273,6 +296,7 @@
--type string backup type filter (full, incremental, data-only, metadata-only)
Global Flags:
+ --auto-load-history-db resolve gpbackup_history.db from $COORDINATOR_DATA_DIRECTORY when --history-db is unset
--history-db string full path to the gpbackup_history.db file
--log-file string full path to log file directory, if not specified, the log file will be created in the $HOME/gpAdminLogs directory
--log-level-console string level for console logging (error, info, debug, verbose) (default "info")
@@ -455,7 +479,7 @@
The gpbackup_history.db file location can be set using the --history-db option.
Can be specified only once. The full path to the file is required.
-If the --history-db option is not specified, the history database is looked for in the current directory. Pass `--auto-load-history-db` to resolve it from `$COORDINATOR_DATA_DIRECTORY` instead.
+If the --history-db option is not specified, the history database is looked for in the current directory. To resolve it from $COORDINATOR_DATA_DIRECTORY instead, pass the --auto-load-history-db flag.
Usage:
gpbackman history-clean [flags]
@@ -463,9 +487,12 @@
Flags:
--before-timestamp string delete information about backups older than the given timestamp
-h, --help help for history-clean
+ --history-sync-standby-timeout int shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400 (default 300)
+ --no-history-sync-standby skip automatic gpbackup_history.db sync to standby coordinator after this command
--older-than-days uint delete information about backups older than the given number of days
Global Flags:
+ --auto-load-history-db resolve gpbackup_history.db from $COORDINATOR_DATA_DIRECTORY when --history-db is unset
--history-db string full path to the gpbackup_history.db file
--log-file string full path to log file directory, if not specified, the log file will be created in the $HOME/gpAdminLogs directory
--log-level-console string level for console logging (error, info, debug, verbose) (default "info")
@@ -477,14 +504,56 @@
Delete information about deleted backups from history database older than 7 days:
```bash
./gpbackman history-clean \
- --older-than-days 7 \
+ --older-than-days 7
```
### Delete information about deleted backups from history database older than timestamp
Delete information about deleted backups from history database older than timestamp `20240101100000`:
```bash
./gpbackman history-clean \
- --before-timestamp 20240101100000 \
+ --before-timestamp 20240101100000
+```
+
+# Sync the history database to the standby coordinator (`history-sync`)
+
+Available options for `history-sync` command and their description:
+
+```bash
+./gpbackman history-sync -h
+Sync the gpbackup_history.db file to the standby coordinator.
+
+The command uses the cluster history database from --history-db, or from
+$COORDINATOR_DATA_DIRECTORY when --auto-load-history-db is set. It succeeds
+only after the standby file is replaced atomically with a verified snapshot.
+
+Usage:
+ gpbackman history-sync [flags]
+
+Flags:
+ -h, --help help for history-sync
+ --history-sync-standby-timeout int shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400 (default 300)
+
+Global Flags:
+ --auto-load-history-db resolve gpbackup_history.db from $COORDINATOR_DATA_DIRECTORY when --history-db is unset
+ --history-db string full path to the gpbackup_history.db file
+ --log-file string full path to log file directory, if not specified, the log file will be created in the $HOME/gpAdminLogs directory
+ --log-level-console string level for console logging (error, info, debug, verbose) (default "info")
+ --log-level-file string level for file logging (error, info, debug, verbose) (default "info")
+```
+
+## Examples
+
+Synchronize an explicitly selected cluster history database:
+
+```bash
+./gpbackman history-sync \
+ --history-db "$COORDINATOR_DATA_DIRECTORY/gpbackup_history.db"
+```
+
+Resolve the cluster history database from the coordinator environment and synchronize it:
+
+```bash
+./gpbackman history-sync --auto-load-history-db
```
# Display the report for a specific backup (`report-info`)
@@ -492,7 +561,7 @@
Available options for `report-info` command and their description:
```bash
-./gpbackman.go report-info -h
+./gpbackman report-info -h
Display the report for a specific backup.
The --timestamp option must be specified.
@@ -524,7 +593,7 @@
The gpbackup_history.db file location can be set using the --history-db option.
Can be specified only once. The full path to the file is required.
-If the --history-db option is not specified, the history database is looked for in the current directory. Pass `--auto-load-history-db` to resolve it from `$COORDINATOR_DATA_DIRECTORY` instead.
+If the --history-db option is not specified, the history database is looked for in the current directory. To resolve it from $COORDINATOR_DATA_DIRECTORY instead, pass the --auto-load-history-db flag.
Usage:
gpbackman report-info [flags]
@@ -537,6 +606,7 @@
--timestamp string the backup timestamp for report displaying
Global Flags:
+ --auto-load-history-db resolve gpbackup_history.db from $COORDINATOR_DATA_DIRECTORY when --history-db is unset
--history-db string full path to the gpbackup_history.db file
--log-file string full path to log file directory, if not specified, the log file will be created in the $HOME/gpAdminLogs directory
--log-level-console string level for console logging (error, info, debug, verbose) (default "info")
@@ -546,17 +616,17 @@
## Examples
### Display the backup report from local storage
-With specifying backup directory path:
+Without specifying a backup directory path:
```bash
./gpbackman report-info \
- --timestamp 20230809232817 \
- --backup-dir /some/path
+ --timestamp 20230809232817
```
With specifying backup directory path:
```bash
./gpbackman report-info \
--timestamp 20230809232817 \
+ --backup-dir /some/path
```
### Display the backup report using storage plugin
@@ -570,7 +640,7 @@
For other plugins:
```bash
-./gpbackman report-infodoc \
+./gpbackman report-info \
--timestamp 20230725101959 \
--plugin-config /tmp/gpbackup_plugin_config.yaml \
--plugin-report-file-path /some/path/to/report
diff --git a/gpbackman/README.md b/gpbackman/README.md
index d9bd9ae..2c9b4af 100644
--- a/gpbackman/README.md
+++ b/gpbackman/README.md
@@ -29,6 +29,8 @@
* delete existing backups from local storage or using storage plugins;
* delete all existing backups from local storage or using storage plugins older than the specified time condition;
* clean deleted backups from the history database;
+* manually synchronize the cluster `gpbackup_history.db` to the standby coordinator;
+* automatically synchronize the cluster `gpbackup_history.db` after successful backup deletion and history cleanup.
## Commands
### Introduction
@@ -49,11 +51,12 @@
completion Generate the autocompletion script for the specified shell
help Help about any command
history-clean Clean deleted backups from the history database
+ history-sync Sync the history database to the standby coordinator
report-info Display the report for a specific backup
Flags:
- -h, --help help for gpbackman
--auto-load-history-db resolve gpbackup_history.db from $COORDINATOR_DATA_DIRECTORY when --history-db is unset
+ -h, --help help for gpbackman
--history-db string full path to the gpbackup_history.db file
--log-file string full path to log file directory, if not specified, the log file will be created in the $HOME/gpAdminLogs directory
--log-level-console string level for console logging (error, info, debug, verbose) (default "info")
@@ -63,6 +66,33 @@
Use "gpbackman [command] --help" for more information about a command.
```
+### Standby history DB sync
+
+Run `history-sync` to explicitly synchronize the cluster `gpbackup_history.db` to an up standby coordinator. The source must resolve to `<primary coordinator data directory>/gpbackup_history.db`; a custom database or the default working-directory database is not eligible. Explicit sync treats every non-sync outcome as an error and exits non-zero.
+
+For the usual cluster setup, resolve the source from the coordinator data directory:
+
+```bash
+./gpbackman history-sync --auto-load-history-db
+```
+
+After a successful `backup-delete`, `backup-clean`, or `history-clean`, gpBackMan also attempts the same synchronization automatically. Automatic sync is best-effort: ineligible source paths and no standby are debug-only skips, while sync failures are warnings and do not change the successful primary command result. Pass `--no-history-sync-standby` to those mutation commands to disable automatic sync.
+
+Configure the sync timeout with `--history-sync-standby-timeout SECONDS` on
+`history-sync`, `backup-delete`, `backup-clean`, and `history-clean`. The
+default is 300 seconds; the supported range is 1 to 86400 seconds. The timeout
+is one shared budget for `rsync` and remote install. It starts after snapshot
+validation. Standby discovery and SQLite snapshot creation and validation
+(`VACUUM INTO` and `PRAGMA quick_check`) are outside this budget. If a
+transport step fails, remote cleanup of the temporary file uses its own fixed
+120-second timeout, independent of `--history-sync-standby-timeout`.
+
+`rsync` 3.0.0 or later must be installed on both the host running gpBackMan
+and the standby coordinator. The current OS user must have non-interactive SSH
+access to the standby host.
+
+Only `gpbackup_history.db` is synchronized. Report files, backup data, and other backup artifacts are not synchronized.
+
### Detail info about commands
Description of each command:
@@ -70,9 +100,9 @@
* [Delete a specific existing backup (`backup-delete`)](./COMMANDS.md#delete-a-specific-existing-backup-backup-delete)
* [Display information about backups (`backup-info`)](./COMMANDS.md#display-information-about-backups-backup-info)
* [Clean deleted backups from the history database (`history-clean`)](./COMMANDS.md#clean-deleted-backups-from-the-history-database-history-clean)
+* [Sync the history database to the standby coordinator (`history-sync`)](./COMMANDS.md#sync-the-history-database-to-the-standby-coordinator-history-sync)
* [Display the report for a specific backup (`report-info`)](./COMMANDS.md#display-the-report-for-a-specific-backup-report-info)
## About
gpBackMan is part of the Apache Cloudberry Backup (Incubating) toolset. It is based on the original [gpbackman](https://github.com/woblerr/gpbackman) project.
-
diff --git a/gpbackman/cmd/backup_clean.go b/gpbackman/cmd/backup_clean.go
index 46c9548..5435c2c 100644
--- a/gpbackman/cmd/backup_clean.go
+++ b/gpbackman/cmd/backup_clean.go
@@ -34,13 +34,14 @@
// Flags for the gpbackman backup-clean command (backupCleanCmd)
var (
- backupCleanBeforeTimestamp string
- backupCleanAfterTimestamp string
- backupCleanPluginConfigFile string
- backupCleanBackupDir string
- backupCleanOlderThanDays uint
- backupCleanParallelProcesses int
- backupCleanCascade bool
+ backupCleanBeforeTimestamp string
+ backupCleanAfterTimestamp string
+ backupCleanPluginConfigFile string
+ backupCleanBackupDir string
+ backupCleanOlderThanDays uint
+ backupCleanParallelProcesses int
+ backupCleanCascade bool
+ backupCleanNoHistorySyncStandby bool
)
var backupCleanCmd = &cobra.Command{
@@ -130,6 +131,18 @@
1,
"the number of parallel processes to delete local backups",
)
+ backupCleanCmd.Flags().BoolVar(
+ &backupCleanNoHistorySyncStandby,
+ noHistorySyncStandbyFlagName,
+ false,
+ "skip automatic gpbackup_history.db sync to standby coordinator after this command",
+ )
+ backupCleanCmd.Flags().IntVar(
+ &historyStandbySyncTimeoutSeconds,
+ historySyncStandbyTimeoutFlagName,
+ historySyncStandbyTimeoutDefault,
+ "shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400",
+ )
backupCleanCmd.MarkFlagsMutuallyExclusive(beforeTimestampFlagName, olderThanDaysFlagName, afterTimestampFlagName)
}
@@ -198,10 +211,7 @@
func doCleanBackup() {
logHeadersDebug()
- err := cleanBackup()
- if err != nil {
- execOSExit(exitErrorCode)
- }
+ runHistoryMutationWithStandbySync(cleanBackup, backupCleanNoHistorySyncStandby)
}
func cleanBackup() error {
diff --git a/gpbackman/cmd/backup_delete.go b/gpbackman/cmd/backup_delete.go
index 4eb525a..edae2e8 100644
--- a/gpbackman/cmd/backup_delete.go
+++ b/gpbackman/cmd/backup_delete.go
@@ -41,13 +41,14 @@
// Flags for the gpbackman backup-delete command (backupDeleteCmd)
var (
- backupDeleteTimestamp []string
- backupDeletePluginConfigFile string
- backupDeleteBackupDir string
- backupDeleteCascade bool
- backupDeleteForce bool
- backupDeleteIgnoreErrors bool
- backupDeleteParallelProcesses int
+ backupDeleteTimestamp []string
+ backupDeletePluginConfigFile string
+ backupDeleteBackupDir string
+ backupDeleteCascade bool
+ backupDeleteForce bool
+ backupDeleteIgnoreErrors bool
+ backupDeleteNoHistorySyncStandby bool
+ backupDeleteParallelProcesses int
)
var backupDeleteCmd = &cobra.Command{
Use: "backup-delete",
@@ -139,6 +140,18 @@
false,
"ignore errors when deleting backups",
)
+ backupDeleteCmd.Flags().BoolVar(
+ &backupDeleteNoHistorySyncStandby,
+ noHistorySyncStandbyFlagName,
+ false,
+ "skip automatic gpbackup_history.db sync to standby coordinator after this command",
+ )
+ backupDeleteCmd.Flags().IntVar(
+ &historyStandbySyncTimeoutSeconds,
+ historySyncStandbyTimeoutFlagName,
+ historySyncStandbyTimeoutDefault,
+ "shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400",
+ )
_ = backupDeleteCmd.MarkPersistentFlagRequired(timestampFlagName)
}
@@ -198,10 +211,7 @@
func doDeleteBackup() {
logHeadersDebug()
- err := deleteBackup()
- if err != nil {
- execOSExit(exitErrorCode)
- }
+ runHistoryMutationWithStandbySync(deleteBackup, backupDeleteNoHistorySyncStandby)
}
func deleteBackup() error {
diff --git a/gpbackman/cmd/constants.go b/gpbackman/cmd/constants.go
index 03b13d3..3d00a61 100644
--- a/gpbackman/cmd/constants.go
+++ b/gpbackman/cmd/constants.go
@@ -34,34 +34,37 @@
historyDBNameConst = historyFileNameBaseConst + historyFileDBSuffixConst
// Flags.
- historyDBFlagName = "history-db"
- autoLoadHistoryDBFlagName = "auto-load-history-db"
- logFileFlagName = "log-file"
- logLevelConsoleFlagName = "log-level-console"
- logLevelFileFlagName = "log-level-file"
- timestampFlagName = "timestamp"
- pluginConfigFileFlagName = "plugin-config"
- reportFilePluginPathFlagName = "plugin-report-file-path"
- deletedFlagName = "deleted"
- failedFlagName = "failed"
- cascadeFlagName = "cascade"
- forceFlagName = "force"
- olderThanDaysFlagName = "older-than-days"
- beforeTimestampFlagName = "before-timestamp"
- afterTimestampFlagName = "after-timestamp"
- typeFlagName = "type"
- tableFlagName = "table"
- schemaFlagName = "schema"
- excludeFlagName = "exclude"
- backupDirFlagName = "backup-dir"
- parallelProcessesFlagName = "parallel-processes"
- ignoreErrorsFlagName = "ignore-errors"
- detailFlagName = "detail"
+ historyDBFlagName = "history-db"
+ autoLoadHistoryDBFlagName = "auto-load-history-db"
+ logFileFlagName = "log-file"
+ logLevelConsoleFlagName = "log-level-console"
+ logLevelFileFlagName = "log-level-file"
+ timestampFlagName = "timestamp"
+ pluginConfigFileFlagName = "plugin-config"
+ reportFilePluginPathFlagName = "plugin-report-file-path"
+ deletedFlagName = "deleted"
+ failedFlagName = "failed"
+ cascadeFlagName = "cascade"
+ forceFlagName = "force"
+ olderThanDaysFlagName = "older-than-days"
+ beforeTimestampFlagName = "before-timestamp"
+ afterTimestampFlagName = "after-timestamp"
+ typeFlagName = "type"
+ tableFlagName = "table"
+ schemaFlagName = "schema"
+ excludeFlagName = "exclude"
+ backupDirFlagName = "backup-dir"
+ parallelProcessesFlagName = "parallel-processes"
+ ignoreErrorsFlagName = "ignore-errors"
+ noHistorySyncStandbyFlagName = "no-history-sync-standby"
+ historySyncStandbyTimeoutFlagName = "history-sync-standby-timeout"
+ detailFlagName = "detail"
exitErrorCode = 1
// Default for checking the existence of the file.
- checkFileExistsConst = true
+ checkFileExistsConst = true
+ historySyncStandbyTimeoutDefault = 300
// Batch size for deleting from sqlite3.
// This is to prevent problem with sqlite3.
diff --git a/gpbackman/cmd/history_clean.go b/gpbackman/cmd/history_clean.go
index 4236642..b801233 100644
--- a/gpbackman/cmd/history_clean.go
+++ b/gpbackman/cmd/history_clean.go
@@ -32,8 +32,9 @@
// Flags for the gpbackman history-clean command (historyCleanCmd)
var (
- historyCleanBeforeTimestamp string
- historyCleanOlderThanDays uint
+ historyCleanBeforeTimestamp string
+ historyCleanOlderThanDays uint
+ historyCleanNoHistorySyncStandby bool
)
var historyCleanCmd = &cobra.Command{
@@ -73,6 +74,18 @@
"",
"delete information about backups older than the given timestamp",
)
+ historyCleanCmd.Flags().BoolVar(
+ &historyCleanNoHistorySyncStandby,
+ noHistorySyncStandbyFlagName,
+ false,
+ "skip automatic gpbackup_history.db sync to standby coordinator after this command",
+ )
+ historyCleanCmd.Flags().IntVar(
+ &historyStandbySyncTimeoutSeconds,
+ historySyncStandbyTimeoutFlagName,
+ historySyncStandbyTimeoutDefault,
+ "shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400",
+ )
historyCleanCmd.MarkFlagsMutuallyExclusive(beforeTimestampFlagName, olderThanDaysFlagName)
}
@@ -99,10 +112,7 @@
func doCleanHistory() {
logHeadersDebug()
- err := cleanHistory()
- if err != nil {
- execOSExit(exitErrorCode)
- }
+ runHistoryMutationWithStandbySync(cleanHistory, historyCleanNoHistorySyncStandby)
}
func cleanHistory() error {
diff --git a/gpbackman/cmd/history_standby_sync.go b/gpbackman/cmd/history_standby_sync.go
new file mode 100644
index 0000000..06589b5
--- /dev/null
+++ b/gpbackman/cmd/history_standby_sync.go
@@ -0,0 +1,513 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package cmd
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "net/url"
+ "os"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/apache/cloudberry-backup/gpbackman/gpbckpconfig"
+ "github.com/apache/cloudberry-backup/gpbackman/textmsg"
+ "github.com/apache/cloudberry-go-libs/gplog"
+ "github.com/apache/cloudberry-go-libs/operating"
+ "github.com/jmoiron/sqlx"
+ _ "github.com/mattn/go-sqlite3"
+ "github.com/nightlyone/lockfile"
+)
+
+const (
+ historyStandbySyncSSHOptions = "ssh -o BatchMode=yes -o StrictHostKeyChecking=no -o ConnectTimeout=30"
+ historyStandbySyncTempDirPattern = "gpbackman-history-standby-sync-%s-%d-*"
+ // Leave enough time for the 30-second SSH connection timeout and remote removal
+ // while keeping failure cleanup bounded.
+ historyStandbySyncCleanupTimeout = 120 * time.Second
+ // Cap the timeout at one day to catch accidentally oversized CLI values.
+ // A longer transport deadline is not meaningful for standby history synchronization.
+ historySyncStandbyTimeoutMax = int(24 * time.Hour / time.Second)
+)
+
+type historyStandbySyncResult struct {
+ skipReason string
+ err error
+}
+
+type historyStandbySyncTarget struct {
+ sourceDBPath string
+ sourceMode os.FileMode
+ standbyHost string
+ standbyDataDir string
+ standbyHistoryDBPath string
+}
+
+var (
+ historyStandbySync = syncHistoryStandby
+ historyStandbySyncOpenClusterConn = gpbckpconfig.NewClusterLocalClusterDefaultConn
+ historyStandbySyncOpenSQLite = sql.Open
+ historyStandbySyncMkdirTemp = os.MkdirTemp
+ historyStandbySyncRemoveAll = os.RemoveAll
+ historyStandbySyncNow = time.Now
+ historyStandbySyncPID = os.Getpid
+ historyStandbySyncTimeoutSeconds = historySyncStandbyTimeoutDefault
+ historyStandbySyncContextWithTimeout = context.WithTimeout
+ historyStandbySyncCurrentUser = func() (string, error) {
+ currentUser, err := operating.System.CurrentUser()
+ if err != nil {
+ return "", err
+ }
+ return currentUser.Username, nil
+ }
+ historyStandbySyncRunSSHCommand = runHistoryStandbySyncSSHCommand
+)
+
+func syncHistoryStandbyBestEffort(disabled bool) historyStandbySyncResult {
+ if disabled {
+ result := historyStandbySyncResult{skipReason: "disabled by --" + noHistorySyncStandbyFlagName}
+ gplog.Info("%s", textmsg.InfoTextHistoryStandbySyncSkip(result.skipReason))
+ return result
+ }
+
+ result := historyStandbySync()
+ if result.err != nil {
+ gplog.Warn("%s", textmsg.WarnTextHistoryStandbySyncFailed(result.err))
+ return result
+ }
+ if result.skipReason != "" {
+ gplog.Debug("%s", textmsg.InfoTextHistoryStandbySyncSkip(result.skipReason))
+ }
+ return result
+}
+
+func syncHistoryStandbyStrict() error {
+ result := historyStandbySync()
+ if result.err != nil {
+ return result.err
+ }
+ if result.skipReason != "" {
+ return textmsg.ErrorHistoryStandbySyncSkippedError(result.skipReason)
+ }
+ return nil
+}
+
+func syncHistoryStandby() historyStandbySyncResult {
+ sourceDBPath, skipReason := getHistoryStandbySyncSourceDBPath()
+ if skipReason != "" {
+ return historyStandbySyncResult{skipReason: skipReason}
+ }
+
+ target, skipReason, err := discoverHistoryStandbySyncTarget(sourceDBPath)
+ if err != nil {
+ return historyStandbySyncResult{err: err}
+ }
+ if skipReason != "" {
+ return historyStandbySyncResult{skipReason: skipReason}
+ }
+
+ userName, err := historyStandbySyncCurrentUser()
+ if err != nil {
+ return historyStandbySyncResult{err: fmt.Errorf("resolve current OS user for standby history sync: %w", err)}
+ }
+
+ gplog.Info("%s", textmsg.InfoTextHistoryStandbySyncStart(target.sourceDBPath))
+ err = withHistoryStandbySyncLock(target.sourceDBPath, func() error {
+ return withHistoryStandbySyncSnapshot(target.sourceDBPath, target.sourceMode, func(snapshotPath string) error {
+ ctx, cancel := historyStandbySyncContextWithTimeout(context.Background(), time.Duration(historyStandbySyncTimeoutSeconds)*time.Second)
+ defer cancel()
+
+ transportErr := syncHistoryStandbySnapshotToStandby(ctx, target, userName, snapshotPath)
+ if transportErr != nil && errors.Is(ctx.Err(), context.DeadlineExceeded) {
+ return fmt.Errorf("standby history sync transport timed out after %d seconds: %w", historyStandbySyncTimeoutSeconds, transportErr)
+ }
+ return transportErr
+ })
+ })
+ if err != nil {
+ return historyStandbySyncResult{err: err}
+ }
+ gplog.Info("%s", textmsg.InfoTextHistoryStandbySyncSuccess(target.standbyHost, target.standbyHistoryDBPath))
+ return historyStandbySyncResult{}
+}
+
+func getHistoryStandbySyncSourceDBPath() (string, string) {
+ sourceDBPath := getHistoryDBPath(rootHistoryDB, rootAutoLoadHistoryDB)
+ if rootHistoryDB == "" && !rootAutoLoadHistoryDB {
+ return sourceDBPath, "using default working-directory history db"
+ }
+ if rootHistoryDB == "" && rootAutoLoadHistoryDB && sourceDBPath == historyDBNameConst {
+ return sourceDBPath, "--auto-load-history-db did not resolve the cluster history db"
+ }
+ return sourceDBPath, ""
+}
+
+func discoverHistoryStandbySyncTarget(sourceDBPath string) (target *historyStandbySyncTarget, skipReason string, retErr error) {
+ db, err := historyStandbySyncOpenClusterConn()
+ if err != nil {
+ return nil, "", fmt.Errorf("connect to local cluster for standby history sync discovery: %w", err)
+ }
+ defer func() {
+ if closeErr := db.Close(); closeErr != nil {
+ retErr = errors.Join(retErr, fmt.Errorf("close local cluster connection for standby history sync discovery: %w", closeErr))
+ }
+ }()
+
+ primaryDataDir, err := queryHistoryStandbySyncPrimaryDataDir(db)
+ if err != nil {
+ return nil, "", fmt.Errorf("query primary coordinator datadir for standby history sync discovery: %w", err)
+ }
+
+ canonicalSourceDBPath, sourceInfo, err := canonicalHistoryStandbySyncSource(sourceDBPath)
+ if err != nil {
+ return nil, "", err
+ }
+ canonicalPrimaryHistoryDBPath, err := canonicalHistoryStandbySyncPath(filepath.Join(primaryDataDir, historyDBNameConst))
+ if err != nil {
+ return nil, "", fmt.Errorf("resolve canonical primary history db path for standby sync: %w", err)
+ }
+ if canonicalSourceDBPath != canonicalPrimaryHistoryDBPath {
+ return nil, fmt.Sprintf("source history db %s is not cluster history db %s", canonicalSourceDBPath, canonicalPrimaryHistoryDBPath), nil
+ }
+
+ standbyConfig, err := queryHistoryStandbySyncStandby(db)
+ if err != nil {
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, "no up standby coordinator found", nil
+ }
+ return nil, "", fmt.Errorf("query up standby coordinator for standby history sync discovery: %w", err)
+ }
+
+ target = &historyStandbySyncTarget{
+ sourceDBPath: canonicalSourceDBPath,
+ sourceMode: sourceInfo.Mode().Perm(),
+ standbyHost: standbyConfig.Hostname,
+ standbyDataDir: standbyConfig.DataDir,
+ standbyHistoryDBPath: filepath.Join(standbyConfig.DataDir, historyDBNameConst),
+ }
+ gplog.Debug("Discovered standby history sync target: source=%s standby=%s:%s", target.sourceDBPath, target.standbyHost, target.standbyHistoryDBPath)
+ return target, "", nil
+}
+
+func queryHistoryStandbySyncPrimaryDataDir(db *sqlx.DB) (string, error) {
+ return gpbckpconfig.QueryPrimaryCoordinatorDataDir(db)
+}
+
+func queryHistoryStandbySyncStandby(db *sqlx.DB) (gpbckpconfig.StandbyCoordinator, error) {
+ return gpbckpconfig.QueryUpStandbyCoordinator(db)
+}
+
+func canonicalHistoryStandbySyncSource(sourceDBPath string) (string, os.FileInfo, error) {
+ canonicalSourceDBPath, err := canonicalHistoryStandbySyncPath(sourceDBPath)
+ if err != nil {
+ return "", nil, fmt.Errorf("resolve canonical source history db path for standby sync: %w", err)
+ }
+ sourceInfo, err := os.Stat(canonicalSourceDBPath)
+ if err != nil {
+ return "", nil, fmt.Errorf("stat source history db for standby sync: %w", err)
+ }
+ if !sourceInfo.Mode().IsRegular() {
+ return "", nil, fmt.Errorf("source history db for standby sync is not a regular file: %s", canonicalSourceDBPath)
+ }
+ return canonicalSourceDBPath, sourceInfo, nil
+}
+
+func canonicalHistoryStandbySyncPath(path string) (string, error) {
+ absolutePath, err := filepath.Abs(filepath.Clean(path))
+ if err != nil {
+ return "", err
+ }
+ canonicalPath, err := filepath.EvalSymlinks(absolutePath)
+ if err != nil {
+ return "", err
+ }
+ return filepath.Clean(canonicalPath), nil
+}
+
+func withHistoryStandbySyncLock(sourceDBPath string, syncFn func() error) error {
+ lockPath := historyStandbySyncLockPath(sourceDBPath)
+ sourceLock, err := lockfile.New(lockPath)
+ if err != nil {
+ return fmt.Errorf("create standby history sync lock %s: %w", lockPath, err)
+ }
+ if err := sourceLock.TryLock(); err != nil {
+ return fmt.Errorf("lock standby history sync source %s: %w", sourceDBPath, err)
+ }
+
+ syncErr := syncFn()
+ unlockErr := sourceLock.Unlock()
+ if syncErr != nil {
+ if unlockErr != nil {
+ return fmt.Errorf("%w; additionally failed to release standby history sync lock %s: %v", syncErr, lockPath, unlockErr)
+ }
+ return syncErr
+ }
+ if unlockErr != nil {
+ return fmt.Errorf("release standby history sync lock %s: %w", lockPath, unlockErr)
+ }
+ return nil
+}
+
+func historyStandbySyncLockPath(sourceDBPath string) string {
+ return sourceDBPath + ".sync.lock"
+}
+
+func withHistoryStandbySyncSnapshot(sourceDBPath string, sourceMode os.FileMode, syncFn func(string) error) (retErr error) {
+ snapshotPath, tempDir, err := createHistoryStandbySyncSnapshot(sourceDBPath, sourceMode)
+ if tempDir != "" {
+ defer func() {
+ retErr = errors.Join(retErr, cleanupHistoryStandbySyncTempDir(tempDir))
+ }()
+ }
+ if err != nil {
+ return err
+ }
+ return syncFn(snapshotPath)
+}
+
+func createHistoryStandbySyncSnapshot(sourceDBPath string, sourceMode os.FileMode) (string, string, error) {
+ tempDirPattern := fmt.Sprintf(
+ historyStandbySyncTempDirPattern,
+ historyStandbySyncNow().UTC().Format("20060102150405"),
+ historyStandbySyncPID(),
+ )
+ tempDir, err := historyStandbySyncMkdirTemp("", tempDirPattern)
+ if err != nil {
+ return "", "", fmt.Errorf("create local standby history sync temp directory: %w", err)
+ }
+ snapshotPath := filepath.Join(tempDir, historyDBNameConst)
+ if err := vacuumHistoryStandbySyncSnapshot(sourceDBPath, snapshotPath); err != nil {
+ return "", "", errors.Join(err, cleanupHistoryStandbySyncTempDir(tempDir))
+ }
+ if err := os.Chmod(snapshotPath, sourceMode); err != nil {
+ return "", "", errors.Join(
+ fmt.Errorf("set standby history sync snapshot permissions from source history db: %w", err),
+ cleanupHistoryStandbySyncTempDir(tempDir),
+ )
+ }
+ if err := validateHistoryStandbySyncSnapshot(snapshotPath); err != nil {
+ return "", "", errors.Join(err, cleanupHistoryStandbySyncTempDir(tempDir))
+ }
+ return snapshotPath, tempDir, nil
+}
+
+func vacuumHistoryStandbySyncSnapshot(sourceDBPath, snapshotPath string) (retErr error) {
+ sourceDB, err := historyStandbySyncOpenSQLite("sqlite3", historyStandbySyncSQLiteURI(sourceDBPath, "ro"))
+ if err != nil {
+ return fmt.Errorf("open source history db for standby sync snapshot: %w", err)
+ }
+ defer func() {
+ if closeErr := sourceDB.Close(); closeErr != nil {
+ retErr = errors.Join(retErr, fmt.Errorf("close source history db for standby sync snapshot: %w", closeErr))
+ }
+ }()
+
+ if _, err := sourceDB.Exec("VACUUM main INTO ?", snapshotPath); err != nil {
+ return fmt.Errorf("create standby history sync snapshot with VACUUM INTO: %w", err)
+ }
+ return nil
+}
+
+func validateHistoryStandbySyncSnapshot(snapshotPath string) error {
+ results, err := runHistoryStandbySyncQuickCheck(snapshotPath)
+ if err != nil {
+ return err
+ }
+ if len(results) != 1 || results[0] != "ok" {
+ return fmt.Errorf("validate standby history sync snapshot quick_check: expected single ok result, got %v", results)
+ }
+ return nil
+}
+
+func runHistoryStandbySyncQuickCheck(snapshotPath string) (results []string, retErr error) {
+ snapshotDB, err := historyStandbySyncOpenSQLite("sqlite3", historyStandbySyncSQLiteURI(snapshotPath, "ro"))
+ if err != nil {
+ return nil, fmt.Errorf("open standby history sync snapshot read-only: %w", err)
+ }
+ defer func() {
+ if closeErr := snapshotDB.Close(); closeErr != nil {
+ retErr = errors.Join(retErr, fmt.Errorf("close standby history sync snapshot: %w", closeErr))
+ }
+ }()
+
+ rows, err := snapshotDB.Query("PRAGMA quick_check")
+ if err != nil {
+ return nil, fmt.Errorf("run PRAGMA quick_check on standby history sync snapshot: %w", err)
+ }
+ defer func() {
+ if closeErr := rows.Close(); closeErr != nil {
+ retErr = errors.Join(retErr, fmt.Errorf("close standby history sync quick_check rows: %w", closeErr))
+ }
+ }()
+
+ results = make([]string, 0)
+ for rows.Next() {
+ var result string
+ if err := rows.Scan(&result); err != nil {
+ return nil, fmt.Errorf("scan PRAGMA quick_check result for standby history sync snapshot: %w", err)
+ }
+ results = append(results, result)
+ }
+ if err := rows.Err(); err != nil {
+ return nil, fmt.Errorf("read PRAGMA quick_check results for standby history sync snapshot: %w", err)
+ }
+ return results, nil
+}
+
+func cleanupHistoryStandbySyncTempDir(tempDir string) error {
+ if err := historyStandbySyncRemoveAll(tempDir); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return fmt.Errorf("remove local standby history sync temp directory %s: %w", tempDir, err)
+ }
+ return nil
+}
+
+func syncHistoryStandbySnapshotToStandby(ctx context.Context, target *historyStandbySyncTarget, userName, snapshotPath string) error {
+ remoteTempPath := newHistoryStandbySyncRemoteTempPath(target.standbyDataDir, snapshotPath)
+ if err := rsyncHistoryStandbySyncSnapshot(ctx, snapshotPath, target.standbyHost, userName, remoteTempPath); err != nil {
+ return cleanupHistoryStandbySyncRemoteTempAfterError(err, target.standbyHost, userName, remoteTempPath)
+ }
+ if err := installHistoryStandbySyncSnapshotOnStandby(ctx, target, userName, remoteTempPath); err != nil {
+ return cleanupHistoryStandbySyncRemoteTempAfterError(err, target.standbyHost, userName, remoteTempPath)
+ }
+ return nil
+}
+
+func newHistoryStandbySyncRemoteTempPath(standbyDataDir, snapshotPath string) string {
+ return filepath.Join(standbyDataDir, fmt.Sprintf(".%s.%s.tmp", historyDBNameConst, filepath.Base(filepath.Dir(snapshotPath))))
+}
+
+func rsyncHistoryStandbySyncSnapshot(ctx context.Context, snapshotPath, standbyHost, userName, remoteTempPath string) error {
+ args := buildHistoryStandbySyncRsyncArgs(snapshotPath, standbyHost, userName, remoteTempPath)
+ gplog.Debug("Transfer history db snapshot to standby coordinator: %s -> %s:%s", snapshotPath, standbyHost, remoteTempPath)
+ output, err := execCombinedOutputCommand(ctx, "rsync", args...).CombinedOutput()
+ if ctxErr := ctx.Err(); ctxErr != nil && err != nil {
+ err = ctxErr
+ }
+ if err != nil {
+ return fmt.Errorf("rsync standby history snapshot to %s:%s failed: %w%s", standbyHost, remoteTempPath, err, formatHistoryStandbySyncCommandOutput(output))
+ }
+ return nil
+}
+
+func buildHistoryStandbySyncRsyncArgs(snapshotPath, standbyHost, userName, remoteTempPath string) []string {
+ return []string{
+ "-p",
+ "-s",
+ "-e",
+ historyStandbySyncSSHOptions,
+ "--",
+ snapshotPath,
+ fmt.Sprintf("%s@%s:%s", userName, standbyHost, remoteTempPath),
+ }
+}
+
+func installHistoryStandbySyncSnapshotOnStandby(ctx context.Context, target *historyStandbySyncTarget, userName, remoteTempPath string) error {
+ command := buildHistoryStandbySyncRemoteInstallCommand(remoteTempPath, target.standbyHistoryDBPath)
+ gplog.Debug("Install history db snapshot on standby coordinator: %s:%s", target.standbyHost, target.standbyHistoryDBPath)
+ output, err := historyStandbySyncRunSSHCommand(ctx, command, target.standbyHost, userName)
+ if err != nil {
+ return fmt.Errorf("install standby history snapshot on %s:%s failed: %w%s", target.standbyHost, target.standbyHistoryDBPath, err, formatHistoryStandbySyncCommandOutput(output))
+ }
+ return nil
+}
+
+func buildHistoryStandbySyncRemoteInstallCommand(remoteTempPath, standbyHistoryDBPath string) string {
+ quotedTempPath := shellQuoteHistoryStandbySyncPath(remoteTempPath)
+ quotedHistoryDBPath := shellQuoteHistoryStandbySyncPath(standbyHistoryDBPath)
+ return fmt.Sprintf(
+ "test -f %s && if test -e %s; then chown --reference=%s -- %s && chmod --reference=%s -- %s; fi && mv -f -- %s %s",
+ quotedTempPath,
+ quotedHistoryDBPath,
+ quotedHistoryDBPath,
+ quotedTempPath,
+ quotedHistoryDBPath,
+ quotedTempPath,
+ quotedTempPath,
+ quotedHistoryDBPath,
+ )
+}
+
+func cleanupHistoryStandbySyncRemoteTempAfterError(primaryErr error, standbyHost, userName, remoteTempPath string) error {
+ cleanupCtx, cancel := context.WithTimeout(context.Background(), historyStandbySyncCleanupTimeout)
+ defer cancel()
+
+ if cleanupErr := cleanupHistoryStandbySyncRemoteTemp(cleanupCtx, standbyHost, userName, remoteTempPath); cleanupErr != nil {
+ return fmt.Errorf("%w; additionally failed to clean up remote temp file: %w", primaryErr, cleanupErr)
+ }
+ return primaryErr
+}
+
+func cleanupHistoryStandbySyncRemoteTemp(ctx context.Context, standbyHost, userName, remoteTempPath string) error {
+ command := buildHistoryStandbySyncRemoteCleanupCommand(remoteTempPath)
+ gplog.Debug("Clean up remote standby history sync temp file: %s:%s", standbyHost, remoteTempPath)
+ output, err := historyStandbySyncRunSSHCommand(ctx, command, standbyHost, userName)
+ if err != nil {
+ return fmt.Errorf("clean up remote standby history sync temp file %s:%s failed: %w%s", standbyHost, remoteTempPath, err, formatHistoryStandbySyncCommandOutput(output))
+ }
+ return nil
+}
+
+func buildHistoryStandbySyncRemoteCleanupCommand(remoteTempPath string) string {
+ return fmt.Sprintf("rm -f -- %s", shellQuoteHistoryStandbySyncPath(remoteTempPath))
+}
+
+func runHistoryStandbySyncSSHCommand(ctx context.Context, remoteCommand, standbyHost, userName string) ([]byte, error) {
+ output, err := execCombinedOutputCommand(
+ ctx,
+ "ssh",
+ "-o",
+ "BatchMode=yes",
+ "-o",
+ "StrictHostKeyChecking=no",
+ "-o",
+ "ConnectTimeout=30",
+ fmt.Sprintf("%s@%s", userName, standbyHost),
+ remoteCommand,
+ ).CombinedOutput()
+ if ctxErr := ctx.Err(); ctxErr != nil && err != nil {
+ return output, ctxErr
+ }
+ return output, err
+}
+
+func historyStandbySyncSQLiteURI(dbPath, mode string) string {
+ query := url.Values{}
+ query.Set("mode", mode)
+ dbURI := url.URL{Scheme: "file", Path: dbPath, RawQuery: query.Encode()}
+ return dbURI.String()
+}
+
+func shellQuoteHistoryStandbySyncPath(value string) string {
+ if value == "" {
+ return "''"
+ }
+ return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
+}
+
+func formatHistoryStandbySyncCommandOutput(output []byte) string {
+ trimmedOutput := strings.TrimSpace(string(output))
+ if trimmedOutput == "" {
+ return ""
+ }
+ return ": " + trimmedOutput
+}
diff --git a/gpbackman/cmd/history_standby_sync_test.go b/gpbackman/cmd/history_standby_sync_test.go
new file mode 100644
index 0000000..822063e
--- /dev/null
+++ b/gpbackman/cmd/history_standby_sync_test.go
@@ -0,0 +1,735 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package cmd
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "regexp"
+ "time"
+
+ "github.com/DATA-DOG/go-sqlmock"
+ "github.com/apache/cloudberry-go-libs/testhelper"
+ "github.com/jmoiron/sqlx"
+ "github.com/nightlyone/lockfile"
+
+ . "github.com/onsi/ginkgo/v2"
+ . "github.com/onsi/gomega"
+)
+
+const (
+ historyStandbySyncPrimarySQL = "SELECT datadir FROM gp_segment_configuration WHERE content = -1 AND role = 'p' AND status = 'u';"
+ historyStandbySyncStandbySQL = "SELECT hostname, datadir FROM gp_segment_configuration WHERE content = -1 AND role = 'm' AND status = 'u';"
+)
+
+type historyStandbySyncCommandCall struct {
+ ctx context.Context
+ ctxErr error
+ deadline time.Time
+ hasDeadline bool
+ name string
+ args []string
+}
+
+type historyStandbySyncCommandResponse struct {
+ output []byte
+ err error
+}
+
+type historyStandbySyncFakeCommand struct {
+ output []byte
+ err error
+}
+
+func (c historyStandbySyncFakeCommand) CombinedOutput() ([]byte, error) {
+ return c.output, c.err
+}
+
+var _ = Describe("history standby sync", func() {
+ var (
+ originalHistoryStandbySync func() historyStandbySyncResult
+ originalOpenClusterConn func() (*sqlx.DB, error)
+ originalOpenSQLite func(string, string) (*sql.DB, error)
+ originalMkdirTemp func(string, string) (string, error)
+ originalRemoveAll func(string) error
+ originalNow func() time.Time
+ originalPID func() int
+ originalCurrentUser func() (string, error)
+ originalRunSSHCommand func(context.Context, string, string, string) ([]byte, error)
+ originalExecCombinedOutputCommand func(context.Context, string, ...string) combinedOutputCommand
+ originalContextWithTimeout func(context.Context, time.Duration) (context.Context, context.CancelFunc)
+ originalTimeoutSeconds int
+ savedRootHistoryDB string
+ savedRootAutoLoadHistoryDB bool
+ savedHistoryStandbySyncEnvironment map[string]string
+ savedHistoryStandbySyncEnvironmentPresent map[string]bool
+ )
+
+ BeforeEach(func() {
+ testhelper.SetupTestLogger()
+ originalHistoryStandbySync = historyStandbySync
+ originalOpenClusterConn = historyStandbySyncOpenClusterConn
+ originalOpenSQLite = historyStandbySyncOpenSQLite
+ originalMkdirTemp = historyStandbySyncMkdirTemp
+ originalRemoveAll = historyStandbySyncRemoveAll
+ originalNow = historyStandbySyncNow
+ originalPID = historyStandbySyncPID
+ originalCurrentUser = historyStandbySyncCurrentUser
+ originalRunSSHCommand = historyStandbySyncRunSSHCommand
+ originalExecCombinedOutputCommand = execCombinedOutputCommand
+ originalContextWithTimeout = historyStandbySyncContextWithTimeout
+ originalTimeoutSeconds = historyStandbySyncTimeoutSeconds
+ savedRootHistoryDB = rootHistoryDB
+ savedRootAutoLoadHistoryDB = rootAutoLoadHistoryDB
+ savedHistoryStandbySyncEnvironment = make(map[string]string)
+ savedHistoryStandbySyncEnvironmentPresent = make(map[string]bool)
+ for _, name := range append(historyDBEnvVars, "PGDATABASE") {
+ value, ok := os.LookupEnv(name)
+ savedHistoryStandbySyncEnvironment[name] = value
+ savedHistoryStandbySyncEnvironmentPresent[name] = ok
+ Expect(os.Unsetenv(name)).To(Succeed())
+ }
+
+ rootHistoryDB = ""
+ rootAutoLoadHistoryDB = false
+ historyStandbySync = syncHistoryStandby
+ historyStandbySyncOpenClusterConn = func() (*sqlx.DB, error) {
+ return nil, errors.New("cluster connection was not expected")
+ }
+ historyStandbySyncOpenSQLite = sql.Open
+ historyStandbySyncMkdirTemp = os.MkdirTemp
+ historyStandbySyncRemoveAll = os.RemoveAll
+ historyStandbySyncNow = func() time.Time {
+ return time.Date(2026, 7, 28, 16, 0, 0, 0, time.UTC)
+ }
+ historyStandbySyncPID = func() int {
+ return 4242
+ }
+ historyStandbySyncCurrentUser = func() (string, error) {
+ return "gpadmin", nil
+ }
+ historyStandbySyncRunSSHCommand = runHistoryStandbySyncSSHCommand
+ historyStandbySyncTimeoutSeconds = historySyncStandbyTimeoutDefault
+ historyStandbySyncContextWithTimeout = context.WithTimeout
+ execCombinedOutputCommand = func(ctx context.Context, name string, args ...string) combinedOutputCommand {
+ return historyStandbySyncFakeCommand{}
+ }
+ })
+
+ AfterEach(func() {
+ historyStandbySync = originalHistoryStandbySync
+ historyStandbySyncOpenClusterConn = originalOpenClusterConn
+ historyStandbySyncOpenSQLite = originalOpenSQLite
+ historyStandbySyncMkdirTemp = originalMkdirTemp
+ historyStandbySyncRemoveAll = originalRemoveAll
+ historyStandbySyncNow = originalNow
+ historyStandbySyncPID = originalPID
+ historyStandbySyncCurrentUser = originalCurrentUser
+ historyStandbySyncRunSSHCommand = originalRunSSHCommand
+ execCombinedOutputCommand = originalExecCombinedOutputCommand
+ historyStandbySyncContextWithTimeout = originalContextWithTimeout
+ historyStandbySyncTimeoutSeconds = originalTimeoutSeconds
+ rootHistoryDB = savedRootHistoryDB
+ rootAutoLoadHistoryDB = savedRootAutoLoadHistoryDB
+ for name, value := range savedHistoryStandbySyncEnvironment {
+ if savedHistoryStandbySyncEnvironmentPresent[name] {
+ Expect(os.Setenv(name, value)).To(Succeed())
+ } else {
+ Expect(os.Unsetenv(name)).To(Succeed())
+ }
+ }
+ })
+
+ It("skips default and unresolved auto-loaded history db sources before discovery", func() {
+ sourceDBPath, skipReason := getHistoryStandbySyncSourceDBPath()
+ Expect(sourceDBPath).To(Equal(historyDBNameConst))
+ Expect(skipReason).To(Equal("using default working-directory history db"))
+
+ rootAutoLoadHistoryDB = true
+ sourceDBPath, skipReason = getHistoryStandbySyncSourceDBPath()
+ Expect(sourceDBPath).To(Equal(historyDBNameConst))
+ Expect(skipReason).To(Equal("--auto-load-history-db did not resolve the cluster history db"))
+ })
+
+ It("uses auto-load history db path resolved from coordinator data directory", func() {
+ rootAutoLoadHistoryDB = true
+ Expect(os.Setenv("COORDINATOR_DATA_DIRECTORY", "/coordinator/data")).To(Succeed())
+
+ sourceDBPath, skipReason := getHistoryStandbySyncSourceDBPath()
+
+ Expect(sourceDBPath).To(Equal(filepath.Join("/coordinator/data", historyDBNameConst)))
+ Expect(skipReason).To(BeEmpty())
+ })
+
+ It("creates a verified snapshot with source contents and permissions", func() {
+ tmpDir := GinkgoT().TempDir()
+ sourceDBPath := filepath.Join(tmpDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ Expect(os.Chmod(sourceDBPath, 0o640)).To(Succeed())
+
+ snapshotPath, tempDir, err := createHistoryStandbySyncSnapshot(sourceDBPath, 0o640)
+ Expect(err).ToNot(HaveOccurred())
+ defer cleanupHistoryStandbySyncTempDir(tempDir)
+
+ Expect(snapshotPath).To(Equal(filepath.Join(tempDir, historyDBNameConst)))
+ snapshotInfo, err := os.Stat(snapshotPath)
+ Expect(err).ToNot(HaveOccurred())
+ Expect(snapshotInfo.Mode().Perm()).To(Equal(os.FileMode(0o640)))
+
+ snapshotDB, err := sql.Open("sqlite3", historyStandbySyncSQLiteURI(snapshotPath, "ro"))
+ Expect(err).ToNot(HaveOccurred())
+ defer snapshotDB.Close()
+ var value string
+ Expect(snapshotDB.QueryRow("SELECT value FROM sync_test WHERE id = 1").Scan(&value)).To(Succeed())
+ Expect(value).To(Equal("present"))
+ Expect(validateHistoryStandbySyncSnapshot(snapshotPath)).To(Succeed())
+ })
+
+ It("rejects corrupted SQLite sources before transport and removes the temp directory", func() {
+ tmpDir := GinkgoT().TempDir()
+ sourceDBPath := filepath.Join(tmpDir, historyDBNameConst)
+ Expect(os.WriteFile(sourceDBPath, []byte("not sqlite"), 0o600)).To(Succeed())
+ snapshotDir := filepath.Join(tmpDir, "snapshot")
+ historyStandbySyncMkdirTemp = func(dir, pattern string) (string, error) {
+ Expect(os.Mkdir(snapshotDir, 0o700)).To(Succeed())
+ return snapshotDir, nil
+ }
+
+ _, tempDir, err := createHistoryStandbySyncSnapshot(sourceDBPath, 0o600)
+
+ Expect(err).To(HaveOccurred())
+ Expect(err.Error()).To(ContainSubstring("VACUUM INTO"))
+ Expect(tempDir).To(BeEmpty())
+ _, statErr := os.Stat(snapshotDir)
+ Expect(errors.Is(statErr, os.ErrNotExist)).To(BeTrue())
+ })
+
+ It("returns SQLite close errors from snapshot validation", func() {
+ sqlDB, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+ Expect(err).ToNot(HaveOccurred())
+ closeErr := errors.New("close failed")
+ mock.ExpectQuery("PRAGMA quick_check").
+ WillReturnRows(sqlmock.NewRows([]string{"quick_check"}).AddRow("ok"))
+ mock.ExpectClose().WillReturnError(closeErr)
+ historyStandbySyncOpenSQLite = func(driverName, dataSourceName string) (*sql.DB, error) {
+ Expect(driverName).To(Equal("sqlite3"))
+ return sqlDB, nil
+ }
+
+ results, err := runHistoryStandbySyncQuickCheck("/tmp/snapshot.db")
+
+ Expect(results).To(Equal([]string{"ok"}))
+ Expect(errors.Is(err, closeErr)).To(BeTrue())
+ Expect(err.Error()).To(ContainSubstring("close standby history sync snapshot"))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("returns local cleanup errors after success and joins them with sync errors", func() {
+ sourceDBPath := filepath.Join(GinkgoT().TempDir(), historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ cleanupErr := errors.New("cleanup failed")
+ historyStandbySyncMkdirTemp = func(dir, pattern string) (string, error) {
+ return GinkgoT().TempDir(), nil
+ }
+ historyStandbySyncRemoveAll = func(path string) error {
+ return cleanupErr
+ }
+
+ err := withHistoryStandbySyncSnapshot(sourceDBPath, 0o600, func(string) error {
+ return nil
+ })
+ Expect(errors.Is(err, cleanupErr)).To(BeTrue())
+
+ syncErr := errors.New("sync failed")
+ err = withHistoryStandbySyncSnapshot(sourceDBPath, 0o600, func(string) error {
+ return syncErr
+ })
+ Expect(errors.Is(err, syncErr)).To(BeTrue())
+ Expect(errors.Is(err, cleanupErr)).To(BeTrue())
+ })
+
+ It("joins snapshot creation and local cleanup errors", func() {
+ sourceDBPath := filepath.Join(GinkgoT().TempDir(), historyDBNameConst)
+ Expect(os.WriteFile(sourceDBPath, []byte("not sqlite"), 0o600)).To(Succeed())
+ cleanupErr := errors.New("cleanup failed")
+ historyStandbySyncRemoveAll = func(path string) error {
+ return cleanupErr
+ }
+
+ _, tempDir, err := createHistoryStandbySyncSnapshot(sourceDBPath, 0o600)
+
+ Expect(tempDir).To(BeEmpty())
+ Expect(err.Error()).To(ContainSubstring("VACUUM INTO"))
+ Expect(errors.Is(err, cleanupErr)).To(BeTrue())
+ })
+
+ It("returns discovery connection close errors", func() {
+ primaryDataDir := GinkgoT().TempDir()
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ sqlDB, mock, err := sqlmock.New()
+ Expect(err).ToNot(HaveOccurred())
+ mock.ExpectQuery(regexp.QuoteMeta(historyStandbySyncPrimarySQL)).
+ WillReturnRows(sqlmock.NewRows([]string{"datadir"}).AddRow(primaryDataDir))
+ mock.ExpectQuery(regexp.QuoteMeta(historyStandbySyncStandbySQL)).
+ WillReturnRows(sqlmock.NewRows([]string{"hostname", "datadir"}).AddRow("sdw-standby", "/data/standby"))
+ closeErr := errors.New("close failed")
+ mock.ExpectClose().WillReturnError(closeErr)
+ historyStandbySyncOpenClusterConn = func() (*sqlx.DB, error) {
+ return sqlx.NewDb(sqlDB, "sqlmock"), nil
+ }
+
+ target, skipReason, err := discoverHistoryStandbySyncTarget(sourceDBPath)
+
+ Expect(target).ToNot(BeNil())
+ Expect(skipReason).To(BeEmpty())
+ Expect(errors.Is(err, closeErr)).To(BeTrue())
+ Expect(err.Error()).To(ContainSubstring("close local cluster connection for standby history sync discovery"))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("canonicalizes symlink sources and uses the shared lock path suffix", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ // Resolve any symlinks in the OS temp dir itself (e.g. macOS /var -> /private/var)
+ // so the expected path matches the canonicalization performed by the code under test.
+ canonicalPrimaryDataDir, err := filepath.EvalSymlinks(primaryDataDir)
+ Expect(err).ToNot(HaveOccurred())
+ primaryDataDir = canonicalPrimaryDataDir
+ realSourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(realSourceDBPath)
+ linkSourceDBPath := filepath.Join(tmpDir, "history-link.db")
+ Expect(os.Symlink(realSourceDBPath, linkSourceDBPath)).To(Succeed())
+
+ canonicalSourceDBPath, _, err := canonicalHistoryStandbySyncSource(linkSourceDBPath)
+
+ Expect(err).ToNot(HaveOccurred())
+ Expect(canonicalSourceDBPath).To(Equal(realSourceDBPath))
+ Expect(historyStandbySyncLockPath(canonicalSourceDBPath)).To(Equal(realSourceDBPath + ".sync.lock"))
+ })
+
+ It("rejects custom history db paths after primary datadir discovery", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ customDataDir := filepath.Join(tmpDir, "custom")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ Expect(os.Mkdir(customDataDir, 0o700)).To(Succeed())
+ createHistoryStandbySyncSQLiteDB(filepath.Join(primaryDataDir, historyDBNameConst))
+ customSourceDBPath := filepath.Join(customDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(customSourceDBPath)
+ mock := setupHistoryStandbySyncClusterConn(primaryDataDir, "", nil, false)
+
+ target, skipReason, err := discoverHistoryStandbySyncTarget(customSourceDBPath)
+
+ Expect(err).ToNot(HaveOccurred())
+ Expect(target).To(BeNil())
+ Expect(skipReason).To(ContainSubstring("is not cluster history db"))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("rejects non-regular source history db files", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ Expect(os.MkdirAll(sourceDBPath, 0o700)).To(Succeed())
+ mock := setupHistoryStandbySyncClusterConn(primaryDataDir, "", nil, false)
+
+ target, skipReason, err := discoverHistoryStandbySyncTarget(sourceDBPath)
+
+ Expect(target).To(BeNil())
+ Expect(skipReason).To(BeEmpty())
+ Expect(err).To(HaveOccurred())
+ Expect(err.Error()).To(ContainSubstring("not a regular file"))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("skips when no up standby coordinator exists", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ rootHistoryDB = sourceDBPath
+ mock := setupHistoryStandbySyncClusterConn(primaryDataDir, "", sql.ErrNoRows, true)
+ commandCalls := setHistoryStandbySyncCommands(nil)
+
+ result := syncHistoryStandby()
+
+ Expect(result.err).ToNot(HaveOccurred())
+ Expect(result.skipReason).To(Equal("no up standby coordinator found"))
+ Expect(*commandCalls).To(BeEmpty())
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("orchestrates discovery, lock, snapshot, rsync transport, atomic install, and local cleanup", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby data")
+ snapshotDir := filepath.Join(tmpDir, "snapshot dir")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ rootHistoryDB = sourceDBPath
+ mock := setupHistoryStandbySyncClusterConn(primaryDataDir, standbyDataDir, nil, true)
+ historyStandbySyncMkdirTemp = func(dir, pattern string) (string, error) {
+ Expect(dir).To(Equal(""))
+ Expect(pattern).To(Equal("gpbackman-history-standby-sync-20260728160000-4242-*"))
+ Expect(os.Mkdir(snapshotDir, 0o700)).To(Succeed())
+ return snapshotDir, nil
+ }
+ commandCalls := setHistoryStandbySyncCommands([]historyStandbySyncCommandResponse{{}, {}})
+ historyStandbySyncTimeoutSeconds = 600
+ start := time.Now()
+
+ result := syncHistoryStandby()
+ finished := time.Now()
+
+ Expect(result.err).ToNot(HaveOccurred())
+ Expect(result.skipReason).To(BeEmpty())
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ Expect(*commandCalls).To(HaveLen(2))
+ snapshotPath := filepath.Join(snapshotDir, historyDBNameConst)
+ remoteTempPath := newHistoryStandbySyncRemoteTempPath(standbyDataDir, snapshotPath)
+ Expect((*commandCalls)[0].name).To(Equal("rsync"))
+ Expect((*commandCalls)[0].args).To(Equal(buildHistoryStandbySyncRsyncArgs(snapshotPath, "sdw-standby", "gpadmin", remoteTempPath)))
+ Expect((*commandCalls)[1].name).To(Equal("ssh"))
+ Expect((*commandCalls)[1].ctx).To(BeIdenticalTo((*commandCalls)[0].ctx))
+ deadline, ok := (*commandCalls)[0].ctx.Deadline()
+ Expect(ok).To(BeTrue())
+ Expect(deadline).To(BeTemporally(">=", start.Add(600*time.Second)))
+ Expect(deadline).To(BeTemporally("<=", finished.Add(600*time.Second)))
+ Expect((*commandCalls)[1].args).To(Equal([]string{
+ "-o",
+ "BatchMode=yes",
+ "-o",
+ "StrictHostKeyChecking=no",
+ "-o",
+ "ConnectTimeout=30",
+ "gpadmin@sdw-standby",
+ buildHistoryStandbySyncRemoteInstallCommand(remoteTempPath, filepath.Join(standbyDataDir, historyDBNameConst)),
+ }))
+ _, err := os.Stat(snapshotDir)
+ Expect(errors.Is(err, os.ErrNotExist)).To(BeTrue())
+ })
+
+ It("returns the rsync stage, configured seconds, and DeadlineExceeded without waiting", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ rootHistoryDB = sourceDBPath
+ mock := setupHistoryStandbySyncClusterConn(primaryDataDir, standbyDataDir, nil, true)
+ historyStandbySyncTimeoutSeconds = 600
+ historyStandbySyncContextWithTimeout = func(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) {
+ Expect(timeout).To(Equal(600 * time.Second))
+ return context.WithDeadline(parent, time.Now().Add(-time.Second))
+ }
+ commandCalls := setHistoryStandbySyncCommands([]historyStandbySyncCommandResponse{
+ {err: context.DeadlineExceeded},
+ {},
+ })
+ start := time.Now()
+
+ result := syncHistoryStandby()
+ finished := time.Now()
+
+ Expect(result.err).To(HaveOccurred())
+ Expect(errors.Is(result.err, context.DeadlineExceeded)).To(BeTrue())
+ Expect(result.err.Error()).To(ContainSubstring("rsync standby history snapshot"))
+ Expect(result.err.Error()).To(ContainSubstring("timed out after 600 seconds"))
+ Expect(*commandCalls).To(HaveLen(2))
+ Expect((*commandCalls)[0].ctxErr).To(Equal(context.DeadlineExceeded))
+ Expect((*commandCalls)[1].ctx).ToNot(BeIdenticalTo((*commandCalls)[0].ctx))
+ Expect((*commandCalls)[1].ctxErr).ToNot(HaveOccurred())
+ Expect((*commandCalls)[1].hasDeadline).To(BeTrue())
+ Expect((*commandCalls)[1].deadline).To(BeTemporally(">=", start.Add(historyStandbySyncCleanupTimeout)))
+ Expect((*commandCalls)[1].deadline).To(BeTemporally("<=", finished.Add(historyStandbySyncCleanupTimeout)))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("uses the default cluster discovery connection for PGDATABASE resolution", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ rootHistoryDB = sourceDBPath
+ Expect(os.Setenv("PGDATABASE", "template1")).To(Succeed())
+ openCalls := 0
+ mock := setupHistoryStandbySyncClusterConnWithHook(primaryDataDir, filepath.Join(tmpDir, "standby"), nil, true, func(db *sqlx.DB) (*sqlx.DB, error) {
+ openCalls++
+ Expect(os.Getenv("PGDATABASE")).To(Equal("template1"))
+ return db, nil
+ })
+ setHistoryStandbySyncCommands([]historyStandbySyncCommandResponse{{}, {}})
+
+ result := syncHistoryStandby()
+
+ Expect(result.err).ToNot(HaveOccurred())
+ Expect(openCalls).To(Equal(1))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("releases the source lock after transport errors", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ rootHistoryDB = sourceDBPath
+ mock := setupHistoryStandbySyncClusterConn(primaryDataDir, standbyDataDir, nil, true)
+ setHistoryStandbySyncCommands([]historyStandbySyncCommandResponse{
+ {output: []byte("rsync failed"), err: errors.New("exit status 1")},
+ {},
+ })
+
+ result := syncHistoryStandby()
+
+ Expect(result.err).To(HaveOccurred())
+ Expect(result.err.Error()).To(ContainSubstring("rsync standby history snapshot"))
+ Expect(result.err.Error()).To(ContainSubstring("rsync failed"))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+
+ sourceLock, err := lockfile.New(historyStandbySyncLockPath(sourceDBPath))
+ Expect(err).ToNot(HaveOccurred())
+ Expect(sourceLock.TryLock()).To(Succeed())
+ Expect(sourceLock.Unlock()).To(Succeed())
+ })
+
+ It("returns lock contention as an error without creating a snapshot", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ rootHistoryDB = sourceDBPath
+ mock := setupHistoryStandbySyncClusterConn(primaryDataDir, standbyDataDir, nil, true)
+ lockPath := historyStandbySyncLockPath(sourceDBPath)
+ Expect(os.WriteFile(lockPath, []byte(fmt.Sprintf("%d\n", os.Getppid())), 0o600)).To(Succeed())
+ defer os.Remove(lockPath)
+ mkdirTempCalls := 0
+ historyStandbySyncMkdirTemp = func(dir, pattern string) (string, error) {
+ mkdirTempCalls++
+ return "", errors.New("snapshot should not be created")
+ }
+
+ result := syncHistoryStandby()
+
+ Expect(result.err).To(HaveOccurred())
+ Expect(result.err.Error()).To(ContainSubstring("lock standby history sync source"))
+ Expect(mkdirTempCalls).To(Equal(0))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("protects rsync paths and quotes remote shell paths", func() {
+ remoteTempPath := "/data dir/standby's/.gpbackup_history.db.tmp"
+ destPath := "/data dir/standby's/gpbackup_history.db"
+ Expect(historyStandbySyncSSHOptions).To(ContainSubstring("BatchMode=yes"))
+
+ Expect(buildHistoryStandbySyncRsyncArgs("/tmp/snapshot", "sdw-standby", "gpadmin", remoteTempPath)).To(Equal([]string{
+ "-p",
+ "-s",
+ "-e",
+ historyStandbySyncSSHOptions,
+ "--",
+ "/tmp/snapshot",
+ "gpadmin@sdw-standby:" + remoteTempPath,
+ }))
+ installCommand := buildHistoryStandbySyncRemoteInstallCommand(remoteTempPath, destPath)
+ Expect(installCommand).To(ContainSubstring("test -f " + shellQuoteHistoryStandbySyncPath(remoteTempPath)))
+ Expect(installCommand).To(ContainSubstring("chown --reference=" + shellQuoteHistoryStandbySyncPath(destPath) + " -- " + shellQuoteHistoryStandbySyncPath(remoteTempPath)))
+ Expect(installCommand).To(ContainSubstring("chmod --reference=" + shellQuoteHistoryStandbySyncPath(destPath) + " -- " + shellQuoteHistoryStandbySyncPath(remoteTempPath)))
+ Expect(installCommand).To(ContainSubstring("mv -f -- " + shellQuoteHistoryStandbySyncPath(remoteTempPath) + " " + shellQuoteHistoryStandbySyncPath(destPath)))
+ Expect(buildHistoryStandbySyncRemoteCleanupCommand(remoteTempPath)).To(Equal("rm -f -- " + shellQuoteHistoryStandbySyncPath(remoteTempPath)))
+ })
+
+ It("cleans the concrete remote temp file after install errors", func() {
+ tmpDir := GinkgoT().TempDir()
+ primaryDataDir := filepath.Join(tmpDir, "primary")
+ standbyDataDir := filepath.Join(tmpDir, "standby")
+ Expect(os.Mkdir(primaryDataDir, 0o700)).To(Succeed())
+ sourceDBPath := filepath.Join(primaryDataDir, historyDBNameConst)
+ createHistoryStandbySyncSQLiteDB(sourceDBPath)
+ rootHistoryDB = sourceDBPath
+ mock := setupHistoryStandbySyncClusterConn(primaryDataDir, standbyDataDir, nil, true)
+ commandCalls := setHistoryStandbySyncCommands([]historyStandbySyncCommandResponse{
+ {},
+ {output: []byte("install failed"), err: errors.New("exit status 1")},
+ {},
+ })
+
+ result := syncHistoryStandby()
+
+ Expect(result.err).To(HaveOccurred())
+ Expect(result.err.Error()).To(ContainSubstring("install standby history snapshot"))
+ Expect(result.err.Error()).To(ContainSubstring("install failed"))
+ Expect(*commandCalls).To(HaveLen(3))
+ Expect((*commandCalls)[2].name).To(Equal("ssh"))
+ Expect((*commandCalls)[2].args[len((*commandCalls)[2].args)-1]).To(HavePrefix("rm -f -- "))
+ Expect((*commandCalls)[2].ctx).ToNot(BeIdenticalTo((*commandCalls)[1].ctx))
+ Expect((*commandCalls)[2].ctxErr).ToNot(HaveOccurred())
+ Expect((*commandCalls)[2].hasDeadline).To(BeTrue())
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("chains remote cleanup errors onto the primary transport error", func() {
+ cleanupCommandErr := fmt.Errorf("cleanup timeout: %w", context.DeadlineExceeded)
+ var cleanupDeadline time.Time
+ historyStandbySyncRunSSHCommand = func(ctx context.Context, remoteCommand, standbyHost, userName string) ([]byte, error) {
+ Expect(ctx.Err()).ToNot(HaveOccurred())
+ var ok bool
+ cleanupDeadline, ok = ctx.Deadline()
+ Expect(ok).To(BeTrue())
+ Expect(remoteCommand).To(Equal(buildHistoryStandbySyncRemoteCleanupCommand("/standby/.tmp")))
+ Expect(standbyHost).To(Equal("sdw-standby"))
+ Expect(userName).To(Equal("gpadmin"))
+ return []byte("cleanup failed"), cleanupCommandErr
+ }
+ primaryErr := errors.New("install failed")
+ start := time.Now()
+
+ err := cleanupHistoryStandbySyncRemoteTempAfterError(primaryErr, "sdw-standby", "gpadmin", "/standby/.tmp")
+ finished := time.Now()
+
+ Expect(err).To(HaveOccurred())
+ Expect(errors.Is(err, primaryErr)).To(BeTrue())
+ Expect(errors.Is(err, cleanupCommandErr)).To(BeTrue())
+ Expect(errors.Is(err, context.DeadlineExceeded)).To(BeTrue())
+ Expect(err.Error()).To(ContainSubstring("install failed"))
+ Expect(err.Error()).To(ContainSubstring("additionally failed to clean up remote temp file"))
+ Expect(err.Error()).To(ContainSubstring("cleanup failed"))
+ Expect(cleanupDeadline).To(BeTemporally(">=", start.Add(historyStandbySyncCleanupTimeout)))
+ Expect(cleanupDeadline).To(BeTemporally("<=", finished.Add(historyStandbySyncCleanupTimeout)))
+ })
+
+ It("keeps automatic sync best-effort while strict sync treats skips as errors", func() {
+ stdout, _, _ := testhelper.SetupTestLogger()
+ syncCalls := 0
+ cleanupErr := errors.New("remove local standby history sync temp directory: cleanup failed")
+ historyStandbySync = func() historyStandbySyncResult {
+ syncCalls++
+ return historyStandbySyncResult{err: cleanupErr}
+ }
+
+ result := syncHistoryStandbyBestEffort(false)
+ Expect(errors.Is(result.err, cleanupErr)).To(BeTrue())
+ Expect(syncCalls).To(Equal(1))
+ Expect(string(stdout.Contents())).To(ContainSubstring("History db sync to standby coordinator failed; standby history may be stale: remove local standby history sync temp directory: cleanup failed"))
+
+ err := syncHistoryStandbyStrict()
+ Expect(errors.Is(err, cleanupErr)).To(BeTrue())
+
+ historyStandbySync = func() historyStandbySyncResult {
+ return historyStandbySyncResult{skipReason: "no up standby coordinator found"}
+ }
+ err = syncHistoryStandbyStrict()
+ Expect(err).To(HaveOccurred())
+ Expect(err.Error()).To(ContainSubstring("history db sync to standby coordinator skipped: no up standby coordinator found"))
+ })
+
+ It("skips disabled automatic sync without invoking discovery", func() {
+ syncCalls := 0
+ historyStandbySync = func() historyStandbySyncResult {
+ syncCalls++
+ return historyStandbySyncResult{err: errors.New("sync should not run")}
+ }
+
+ result := syncHistoryStandbyBestEffort(true)
+
+ Expect(result.err).ToNot(HaveOccurred())
+ Expect(result.skipReason).To(Equal("disabled by --" + noHistorySyncStandbyFlagName))
+ Expect(syncCalls).To(Equal(0))
+ })
+})
+
+func createHistoryStandbySyncSQLiteDB(path string) {
+ Expect(os.MkdirAll(filepath.Dir(path), 0o700)).To(Succeed())
+ db, err := sql.Open("sqlite3", historyStandbySyncSQLiteURI(path, "rwc"))
+ Expect(err).ToNot(HaveOccurred())
+ defer db.Close()
+
+ _, err = db.Exec("CREATE TABLE sync_test (id INTEGER PRIMARY KEY, value TEXT)")
+ Expect(err).ToNot(HaveOccurred())
+ _, err = db.Exec("INSERT INTO sync_test (value) VALUES ('present')")
+ Expect(err).ToNot(HaveOccurred())
+}
+
+func setupHistoryStandbySyncClusterConn(primaryDataDir, standbyDataDir string, standbyErr error, expectStandby bool) sqlmock.Sqlmock {
+ return setupHistoryStandbySyncClusterConnWithHook(primaryDataDir, standbyDataDir, standbyErr, expectStandby, func(db *sqlx.DB) (*sqlx.DB, error) {
+ return db, nil
+ })
+}
+
+func setupHistoryStandbySyncClusterConnWithHook(
+ primaryDataDir string,
+ standbyDataDir string,
+ standbyErr error,
+ expectStandby bool,
+ hook func(*sqlx.DB) (*sqlx.DB, error),
+) sqlmock.Sqlmock {
+ sqlDB, mock, err := sqlmock.New()
+ Expect(err).ToNot(HaveOccurred())
+ db := sqlx.NewDb(sqlDB, "sqlmock")
+ mock.ExpectQuery(regexp.QuoteMeta(historyStandbySyncPrimarySQL)).
+ WillReturnRows(sqlmock.NewRows([]string{"datadir"}).AddRow(primaryDataDir))
+ if expectStandby {
+ if standbyErr != nil {
+ mock.ExpectQuery(regexp.QuoteMeta(historyStandbySyncStandbySQL)).WillReturnError(standbyErr)
+ } else {
+ mock.ExpectQuery(regexp.QuoteMeta(historyStandbySyncStandbySQL)).
+ WillReturnRows(sqlmock.NewRows([]string{"hostname", "datadir"}).AddRow("sdw-standby", standbyDataDir))
+ }
+ }
+ mock.ExpectClose()
+ historyStandbySyncOpenClusterConn = func() (*sqlx.DB, error) {
+ return hook(db)
+ }
+ return mock
+}
+
+func setHistoryStandbySyncCommands(responses []historyStandbySyncCommandResponse) *[]historyStandbySyncCommandCall {
+ calls := make([]historyStandbySyncCommandCall, 0)
+ execCombinedOutputCommand = func(ctx context.Context, name string, args ...string) combinedOutputCommand {
+ deadline, hasDeadline := ctx.Deadline()
+ calls = append(calls, historyStandbySyncCommandCall{
+ ctx: ctx,
+ ctxErr: ctx.Err(),
+ deadline: deadline,
+ hasDeadline: hasDeadline,
+ name: name,
+ args: append([]string{}, args...),
+ })
+ response := historyStandbySyncCommandResponse{}
+ if len(calls) <= len(responses) {
+ response = responses[len(calls)-1]
+ }
+ return historyStandbySyncFakeCommand{output: response.output, err: response.err}
+ }
+ return &calls
+}
diff --git a/gpbackman/cmd/history_sync.go b/gpbackman/cmd/history_sync.go
new file mode 100644
index 0000000..dbbf14d
--- /dev/null
+++ b/gpbackman/cmd/history_sync.go
@@ -0,0 +1,60 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package cmd
+
+import (
+ "github.com/apache/cloudberry-go-libs/gplog"
+ "github.com/spf13/cobra"
+
+ "github.com/apache/cloudberry-backup/gpbackman/textmsg"
+)
+
+var historySyncCmd = &cobra.Command{
+ Use: "history-sync",
+ Short: "Sync the history database to the standby coordinator",
+ Long: `Sync the gpbackup_history.db file to the standby coordinator.
+
+The command uses the cluster history database from --history-db, or from
+$COORDINATOR_DATA_DIRECTORY when --auto-load-history-db is set. It succeeds
+only after the standby file is replaced atomically with a verified snapshot.`,
+ Args: cobra.NoArgs,
+ Run: func(cmd *cobra.Command, args []string) {
+ doRootFlagValidation(cmd.Flags(), checkFileExistsConst)
+ doHistorySync()
+ },
+}
+
+func init() {
+ rootCmd.AddCommand(historySyncCmd)
+ historySyncCmd.Flags().IntVar(
+ &historyStandbySyncTimeoutSeconds,
+ historySyncStandbyTimeoutFlagName,
+ historySyncStandbyTimeoutDefault,
+ "shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400",
+ )
+}
+
+func doHistorySync() {
+ logHeadersDebug()
+ if err := syncHistoryStandbyStrict(); err != nil {
+ gplog.Error("%s", textmsg.ErrorTextUnableSyncHistoryDBToStandby(err))
+ execOSExit(exitErrorCode)
+ }
+}
diff --git a/gpbackman/cmd/history_sync_test.go b/gpbackman/cmd/history_sync_test.go
new file mode 100644
index 0000000..acebdda
--- /dev/null
+++ b/gpbackman/cmd/history_sync_test.go
@@ -0,0 +1,320 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package cmd
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "os"
+
+ "github.com/apache/cloudberry-go-libs/testhelper"
+ "github.com/spf13/cobra"
+ "github.com/spf13/pflag"
+
+ . "github.com/onsi/ginkgo/v2"
+ . "github.com/onsi/gomega"
+)
+
+var _ = Describe("history sync command", func() {
+ Describe("command registration", func() {
+ AfterEach(func() {
+ rootCmd.SetOut(os.Stdout)
+ rootCmd.SetErr(os.Stderr)
+ })
+
+ It("shows history-sync in root help without exposing the automatic disable flag", func() {
+ var output bytes.Buffer
+ rootCmd.SetOut(&output)
+ rootCmd.SetErr(&output)
+
+ Expect(rootCmd.Help()).To(Succeed())
+
+ help := output.String()
+ Expect(help).To(ContainSubstring("history-sync"))
+ Expect(help).ToNot(ContainSubstring(noHistorySyncStandbyFlagName))
+ })
+
+ It("registers no-history-sync-standby only on mutation commands", func() {
+ mutationCommands := map[string]bool{
+ "backup-delete": true,
+ "backup-clean": true,
+ "history-clean": true,
+ }
+
+ for _, command := range rootCmd.Commands() {
+ flag := command.Flags().Lookup(noHistorySyncStandbyFlagName)
+ if mutationCommands[command.Name()] {
+ Expect(flag).ToNot(BeNil(), command.Name())
+ Expect(flag.DefValue).To(Equal("false"), command.Name())
+ continue
+ }
+ Expect(flag).To(BeNil(), command.Name())
+ }
+ Expect(rootCmd.PersistentFlags().Lookup(noHistorySyncStandbyFlagName)).To(BeNil())
+ })
+
+ It("registers history-sync-standby-timeout only on sync-capable commands", func() {
+ syncCommands := map[string]bool{
+ "history-sync": true,
+ "backup-delete": true,
+ "backup-clean": true,
+ "history-clean": true,
+ }
+
+ for _, command := range rootCmd.Commands() {
+ flag := command.Flags().Lookup(historySyncStandbyTimeoutFlagName)
+ if syncCommands[command.Name()] {
+ Expect(flag).ToNot(BeNil(), command.Name())
+ Expect(flag.DefValue).To(Equal("300"), command.Name())
+ continue
+ }
+ Expect(flag).To(BeNil(), command.Name())
+ }
+ Expect(rootCmd.PersistentFlags().Lookup(historySyncStandbyTimeoutFlagName)).To(BeNil())
+ })
+
+ DescribeTable("rejects non-integer timeout values",
+ func(value string) {
+ flag := historySyncCmd.Flags().Lookup(historySyncStandbyTimeoutFlagName)
+ Expect(flag).ToNot(BeNil())
+ Expect(flag.Value.Set(value)).ToNot(Succeed())
+ },
+ Entry("fractional", "1.5"),
+ Entry("duration", "5m"),
+ )
+
+ It("accepts a custom timeout in seconds", func() {
+ originalTimeout := historyStandbySyncTimeoutSeconds
+ flag := historySyncCmd.Flags().Lookup(historySyncStandbyTimeoutFlagName)
+ originalChanged := flag.Changed
+ DeferCleanup(func() {
+ historyStandbySyncTimeoutSeconds = originalTimeout
+ flag.Changed = originalChanged
+ })
+
+ Expect(historySyncCmd.Flags().Set(historySyncStandbyTimeoutFlagName, "600")).To(Succeed())
+ Expect(historyStandbySyncTimeoutSeconds).To(Equal(600))
+ })
+
+ It("keeps history-sync strict with inherited global flags and its timeout flag", func() {
+ Expect(commandByName("history-sync")).To(Equal(historySyncCmd))
+ Expect(historySyncCmd.Args(historySyncCmd, []string{"unexpected"})).To(HaveOccurred())
+ Expect(flagNames(historySyncCmd.LocalFlags())).To(Equal([]string{historySyncStandbyTimeoutFlagName}))
+ Expect(historySyncCmd.Flags().Lookup(noHistorySyncStandbyFlagName)).To(BeNil())
+ for _, flagName := range []string{
+ historyDBFlagName,
+ autoLoadHistoryDBFlagName,
+ logFileFlagName,
+ logLevelConsoleFlagName,
+ logLevelFileFlagName,
+ } {
+ Expect(historySyncCmd.Flag(flagName)).ToNot(BeNil(), flagName)
+ }
+ })
+ })
+
+ Describe("strict history sync execution", func() {
+ var (
+ originalHistoryStandbySync func() historyStandbySyncResult
+ originalExecOSExit func(int)
+ savedRootHistoryDB string
+ savedRootAutoLoadHistoryDB bool
+ savedPGPassword string
+ savedPGPasswordPresent bool
+ exitCodes []int
+ )
+
+ BeforeEach(func() {
+ testhelper.SetupTestLogger()
+ originalHistoryStandbySync = historyStandbySync
+ originalExecOSExit = execOSExit
+ savedRootHistoryDB = rootHistoryDB
+ savedRootAutoLoadHistoryDB = rootAutoLoadHistoryDB
+ savedPGPassword, savedPGPasswordPresent = os.LookupEnv("PGPASSWORD")
+ Expect(os.Setenv("PGPASSWORD", "do-not-log-this-password")).To(Succeed())
+ exitCodes = make([]int, 0)
+ execOSExit = func(code int) {
+ exitCodes = append(exitCodes, code)
+ }
+ rootHistoryDB = ""
+ rootAutoLoadHistoryDB = false
+ })
+
+ AfterEach(func() {
+ historyStandbySync = originalHistoryStandbySync
+ execOSExit = originalExecOSExit
+ rootHistoryDB = savedRootHistoryDB
+ rootAutoLoadHistoryDB = savedRootAutoLoadHistoryDB
+ if savedPGPasswordPresent {
+ Expect(os.Setenv("PGPASSWORD", savedPGPassword)).To(Succeed())
+ } else {
+ Expect(os.Unsetenv("PGPASSWORD")).To(Succeed())
+ }
+ })
+
+ It("exits zero only when strict sync succeeds", func() {
+ historyStandbySync = func() historyStandbySyncResult {
+ return historyStandbySyncResult{}
+ }
+
+ doHistorySync()
+
+ Expect(exitCodes).To(BeEmpty())
+ })
+
+ DescribeTable("rejects a standby history sync timeout outside the supported range", func(timeoutSeconds int) {
+ originalTimeout := historyStandbySyncTimeoutSeconds
+ DeferCleanup(func() {
+ historyStandbySyncTimeoutSeconds = originalTimeout
+ })
+ historyStandbySyncTimeoutSeconds = timeoutSeconds
+
+ doRootFlagValidation(historySyncCmd.Flags(), false)
+
+ Expect(exitCodes).To(Equal([]int{exitErrorCode}))
+ },
+ Entry("zero", 0),
+ Entry("negative", -1),
+ Entry("more than one day", 86401),
+ )
+
+ It("treats the default working-directory source as a strict error", func() {
+ stdout, stderr, _ := testhelper.SetupTestLogger()
+ historyStandbySync = syncHistoryStandby
+
+ doHistorySync()
+
+ logOutput := string(stdout.Contents()) + string(stderr.Contents())
+ Expect(exitCodes).To(Equal([]int{exitErrorCode}))
+ Expect(logOutput).To(ContainSubstring("Unable to sync history db to standby coordinator"))
+ Expect(logOutput).To(ContainSubstring("using default working-directory history db"))
+ Expect(logOutput).ToNot(ContainSubstring("do-not-log-this-password"))
+ })
+
+ It("exits with one for strict skip and stage errors", func() {
+ tests := []struct {
+ name string
+ result historyStandbySyncResult
+ want string
+ }{
+ {
+ name: "custom source",
+ result: historyStandbySyncResult{skipReason: "source history db /custom/gpbackup_history.db is not cluster history db /primary/gpbackup_history.db"},
+ want: "source history db /custom/gpbackup_history.db is not cluster history db /primary/gpbackup_history.db",
+ },
+ {
+ name: "no standby",
+ result: historyStandbySyncResult{skipReason: "no up standby coordinator found"},
+ want: "no up standby coordinator found",
+ },
+ {
+ name: "busy lock",
+ result: historyStandbySyncResult{err: errors.New("lock standby history sync source /primary/gpbackup_history.db: already locked")},
+ want: "lock standby history sync source /primary/gpbackup_history.db",
+ },
+ {
+ name: "stage error",
+ result: historyStandbySyncResult{err: errors.New("validate standby history sync snapshot quick_check failed")},
+ want: "validate standby history sync snapshot quick_check failed",
+ },
+ {
+ name: "transport timeout",
+ result: historyStandbySyncResult{err: fmt.Errorf("rsync standby history snapshot timed out after 300 seconds: %w", context.DeadlineExceeded)},
+ want: "rsync standby history snapshot timed out after 300 seconds",
+ },
+ }
+
+ for _, tt := range tests {
+ stdout, stderr, _ := testhelper.SetupTestLogger()
+ exitCodes = make([]int, 0)
+ result := tt.result
+ historyStandbySync = func() historyStandbySyncResult {
+ return result
+ }
+
+ doHistorySync()
+
+ logOutput := string(stdout.Contents()) + string(stderr.Contents())
+ Expect(exitCodes).To(Equal([]int{exitErrorCode}), tt.name)
+ Expect(logOutput).To(ContainSubstring("Unable to sync history db to standby coordinator"), tt.name)
+ Expect(logOutput).To(ContainSubstring(tt.want), tt.name)
+ Expect(logOutput).ToNot(ContainSubstring("do-not-log-this-password"), tt.name)
+ }
+ })
+ })
+
+ Describe("mutation command automatic sync hooks", func() {
+ var (
+ originalRunHistoryMutationWithStandbySync func(func() error, bool)
+ savedBackupDeleteNoHistorySyncStandby bool
+ savedBackupCleanNoHistorySyncStandby bool
+ savedHistoryCleanNoHistorySyncStandby bool
+ )
+
+ BeforeEach(func() {
+ originalRunHistoryMutationWithStandbySync = runHistoryMutationWithStandbySync
+ savedBackupDeleteNoHistorySyncStandby = backupDeleteNoHistorySyncStandby
+ savedBackupCleanNoHistorySyncStandby = backupCleanNoHistorySyncStandby
+ savedHistoryCleanNoHistorySyncStandby = historyCleanNoHistorySyncStandby
+ })
+
+ AfterEach(func() {
+ runHistoryMutationWithStandbySync = originalRunHistoryMutationWithStandbySync
+ backupDeleteNoHistorySyncStandby = savedBackupDeleteNoHistorySyncStandby
+ backupCleanNoHistorySyncStandby = savedBackupCleanNoHistorySyncStandby
+ historyCleanNoHistorySyncStandby = savedHistoryCleanNoHistorySyncStandby
+ })
+
+ It("wraps all mutation commands and passes their disable flag values", func() {
+ disabledValues := make([]bool, 0)
+ runHistoryMutationWithStandbySync = func(work func() error, disabled bool) {
+ disabledValues = append(disabledValues, disabled)
+ }
+ backupDeleteNoHistorySyncStandby = true
+ backupCleanNoHistorySyncStandby = false
+ historyCleanNoHistorySyncStandby = true
+
+ doDeleteBackup()
+ doCleanBackup()
+ doCleanHistory()
+
+ Expect(disabledValues).To(Equal([]bool{true, false, true}))
+ })
+ })
+})
+
+func commandByName(name string) *cobra.Command {
+ for _, command := range rootCmd.Commands() {
+ if command.Name() == name {
+ return command
+ }
+ }
+ return nil
+}
+
+func flagNames(flags *pflag.FlagSet) []string {
+ names := make([]string, 0)
+ flags.VisitAll(func(flag *pflag.Flag) {
+ names = append(names, flag.Name)
+ })
+ return names
+}
diff --git a/gpbackman/cmd/root.go b/gpbackman/cmd/root.go
index 79156ce..b49ddce 100644
--- a/gpbackman/cmd/root.go
+++ b/gpbackman/cmd/root.go
@@ -21,6 +21,7 @@
import (
"fmt"
+ "strconv"
"github.com/apache/cloudberry-backup/gpbackman/gpbckpconfig"
"github.com/apache/cloudberry-backup/gpbackman/textmsg"
@@ -92,6 +93,16 @@
// These flag checks are applied for all commands:
func doRootFlagValidation(flags *pflag.FlagSet, checkFileExists bool) {
var err error
+ if flags.Lookup(historySyncStandbyTimeoutFlagName) != nil {
+ timeoutSeconds, timeoutErr := flags.GetInt(historySyncStandbyTimeoutFlagName)
+ if timeoutErr == nil && (timeoutSeconds <= 0 || timeoutSeconds > historySyncStandbyTimeoutMax) {
+ timeoutErr = fmt.Errorf("must be between 1 and %d seconds", historySyncStandbyTimeoutMax)
+ }
+ if timeoutErr != nil {
+ gplog.Error("%s", textmsg.ErrorTextUnableValidateFlag(strconv.Itoa(timeoutSeconds), historySyncStandbyTimeoutFlagName, timeoutErr))
+ execOSExit(exitErrorCode)
+ }
+ }
// If history-db flag is specified and full path.
// The existence of the file is checked by condition from each specific command.
// Not all commands require a history db file to exist.
diff --git a/gpbackman/cmd/wrappers.go b/gpbackman/cmd/wrappers.go
index c36a525..9086c6a 100644
--- a/gpbackman/cmd/wrappers.go
+++ b/gpbackman/cmd/wrappers.go
@@ -20,9 +20,11 @@
package cmd
import (
+ "context"
"database/sql"
"fmt"
"os"
+ "os/exec"
"path/filepath"
"strings"
@@ -36,6 +38,14 @@
var execOSExit = os.Exit
+type combinedOutputCommand interface {
+ CombinedOutput() ([]byte, error)
+}
+
+var execCombinedOutputCommand = func(ctx context.Context, name string, args ...string) combinedOutputCommand {
+ return exec.CommandContext(ctx, name, args...)
+}
+
func logHeadersDebug() {
gplog.Debug("Start %s version %s", commandName, getVersion())
gplog.Debug("Use console log level: %s", rootLogLevelConsole)
@@ -122,6 +132,14 @@
return fmt.Sprintf("%02d:%02d:%02d", hours, minutes, seconds)
}
+var runHistoryMutationWithStandbySync = func(work func() error, disabled bool) {
+ if err := work(); err != nil {
+ execOSExit(exitErrorCode)
+ return
+ }
+ _ = syncHistoryStandbyBestEffort(disabled)
+}
+
// The backup can be used in one of the cases for local and plugin backups:
// - backup is active
// - backup is not active, but the --force flag is set.
diff --git a/gpbackman/cmd/wrappers_test.go b/gpbackman/cmd/wrappers_test.go
index 10c0d0c..12342ae 100644
--- a/gpbackman/cmd/wrappers_test.go
+++ b/gpbackman/cmd/wrappers_test.go
@@ -20,6 +20,7 @@
package cmd
import (
+ "errors"
"fmt"
"os"
"path/filepath"
@@ -100,6 +101,106 @@
})
})
+ Describe("runHistoryMutationWithStandbySync", func() {
+ var (
+ originalHistoryStandbySync func() historyStandbySyncResult
+ originalExecOSExit func(int)
+ )
+
+ BeforeEach(func() {
+ originalHistoryStandbySync = historyStandbySync
+ originalExecOSExit = execOSExit
+ })
+
+ AfterEach(func() {
+ historyStandbySync = originalHistoryStandbySync
+ execOSExit = originalExecOSExit
+ })
+
+ It("runs standby sync after successful work returns", func() {
+ calls := make([]string, 0)
+ historyStandbySync = func() historyStandbySyncResult {
+ calls = append(calls, "sync")
+ return historyStandbySyncResult{}
+ }
+
+ runHistoryMutationWithStandbySync(func() error {
+ calls = append(calls, "work")
+ return nil
+ }, false)
+
+ Expect(calls).To(Equal([]string{"work", "sync"}))
+ })
+
+ It("runs standby sync after deferred work cleanup completes", func() {
+ calls := make([]string, 0)
+ historyStandbySync = func() historyStandbySyncResult {
+ calls = append(calls, "sync")
+ return historyStandbySyncResult{}
+ }
+
+ runHistoryMutationWithStandbySync(func() error {
+ calls = append(calls, "work")
+ defer func() {
+ calls = append(calls, "close")
+ }()
+ return nil
+ }, false)
+
+ Expect(calls).To(Equal([]string{"work", "close", "sync"}))
+ })
+
+ It("does not run standby sync after work errors", func() {
+ syncCalls := 0
+ exitCalls := 0
+ historyStandbySync = func() historyStandbySyncResult {
+ syncCalls++
+ return historyStandbySyncResult{}
+ }
+ execOSExit = func(code int) {
+ exitCalls++
+ Expect(code).To(Equal(exitErrorCode))
+ }
+
+ runHistoryMutationWithStandbySync(func() error {
+ return errors.New("work failed")
+ }, false)
+
+ Expect(syncCalls).To(Equal(0))
+ Expect(exitCalls).To(Equal(1))
+ })
+
+ It("keeps the command exit code successful when automatic sync fails", func() {
+ exitCalls := 0
+ historyStandbySync = func() historyStandbySyncResult {
+ return historyStandbySyncResult{err: errors.New("transport failed")}
+ }
+ execOSExit = func(code int) {
+ exitCalls++
+ }
+
+ runHistoryMutationWithStandbySync(func() error {
+ return nil
+ }, false)
+
+ Expect(exitCalls).To(Equal(0))
+ })
+
+ It("honors the disabled automatic policy after successful work", func() {
+ syncCalls := 0
+ historyStandbySync = func() historyStandbySyncResult {
+ syncCalls++
+ return historyStandbySyncResult{}
+ }
+
+ runHistoryMutationWithStandbySync(func() error {
+ return nil
+ }, true)
+
+ Expect(syncCalls).To(Equal(0))
+ })
+ })
+
Describe("checkCompatibleFlags", func() {
It("does not return error when no flags changed", func() {
flags := pflag.NewFlagSet("test", pflag.ContinueOnError)
diff --git a/gpbackman/gpbckpconfig/cluster.go b/gpbackman/gpbckpconfig/cluster.go
index 93f6f7c..b664da2 100644
--- a/gpbackman/gpbckpconfig/cluster.go
+++ b/gpbackman/gpbckpconfig/cluster.go
@@ -35,6 +35,20 @@
DataDir string
}
+// StandbyCoordinator stores the standby coordinator connection target.
+type StandbyCoordinator struct {
+ Hostname string `db:"hostname"`
+ DataDir string `db:"datadir"`
+}
+
+const (
+ defaultClusterDatabase = "postgres"
+ primaryCoordinatorDataDirSQL = "SELECT datadir FROM gp_segment_configuration WHERE content = -1 AND role = 'p' AND status = 'u';"
+ upStandbyCoordinatorSQL = "SELECT hostname, datadir FROM gp_segment_configuration WHERE content = -1 AND role = 'm' AND status = 'u';"
+)
+
+var connectLocalCluster = sqlx.Connect
+
// NewClusterLocalClusterConn creates a new connection to the local postgres database.
// Returns an error if the connection could not be established.
func NewClusterLocalClusterConn(dbName string) (*sqlx.DB, error) {
@@ -55,7 +69,26 @@
port = 5432
}
connStr := fmt.Sprintf("postgres://%s@%s:%d/%s?sslmode=disable&connect_timeout=60", username, host, port, dbName)
- return sqlx.Connect("postgres", connStr)
+ return connectLocalCluster("postgres", connStr)
+}
+
+// NewClusterLocalClusterDefaultConn creates a local cluster connection using PGDATABASE or postgres.
+func NewClusterLocalClusterDefaultConn() (*sqlx.DB, error) {
+ dbName := operating.System.Getenv("PGDATABASE")
+ if dbName == "" {
+ dbName = defaultClusterDatabase
+ }
+ return NewClusterLocalClusterConn(dbName)
+}
+
+// QueryPrimaryCoordinatorDataDir queries the up primary coordinator data directory.
+func QueryPrimaryCoordinatorDataDir(conn *sqlx.DB) (string, error) {
+ return ExecuteQueryLocalClusterConn[string](conn, primaryCoordinatorDataDirSQL)
+}
+
+// QueryUpStandbyCoordinator queries the up standby coordinator from the local cluster catalog.
+func QueryUpStandbyCoordinator(conn *sqlx.DB) (StandbyCoordinator, error) {
+ return ExecuteQueryLocalClusterConn[StandbyCoordinator](conn, upStandbyCoordinatorSQL)
}
// ExecuteQueryLocalClusterConn executes a query on the local cluster connection and returns the result.
@@ -72,6 +105,7 @@
// The function supports the following types for T:
// - string: The result will be a single string value.
// - []SegmentConfig: The result will be a slice of SegmentConfig structs.
+// - StandbyCoordinator: The result will be a standby coordinator struct.
//
// If the type T is not supported, the function returns an error indicating the unsupported type.
func ExecuteQueryLocalClusterConn[T any](conn *sqlx.DB, query string) (T, error) {
@@ -91,6 +125,13 @@
return result, err
}
result = any(segConfigs).(T)
+ case StandbyCoordinator:
+ var standbyCoordinator StandbyCoordinator
+ err := conn.Get(&standbyCoordinator, query)
+ if err != nil {
+ return result, err
+ }
+ result = any(standbyCoordinator).(T)
default:
return result, fmt.Errorf("unsupported type")
}
diff --git a/gpbackman/gpbckpconfig/cluster_test.go b/gpbackman/gpbckpconfig/cluster_test.go
new file mode 100644
index 0000000..bf6954b
--- /dev/null
+++ b/gpbackman/gpbckpconfig/cluster_test.go
@@ -0,0 +1,230 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package gpbckpconfig
+
+import (
+ "database/sql"
+ "errors"
+ "os"
+ "regexp"
+
+ "github.com/DATA-DOG/go-sqlmock"
+ "github.com/jmoiron/sqlx"
+ . "github.com/onsi/ginkgo/v2"
+ . "github.com/onsi/gomega"
+)
+
+type savedEnvValue struct {
+ value string
+ ok bool
+}
+
+var _ = Describe("cluster tests", func() {
+ var (
+ originalConnect func(string, string) (*sqlx.DB, error)
+ savedEnv map[string]savedEnvValue
+ )
+
+ BeforeEach(func() {
+ originalConnect = connectLocalCluster
+ savedEnv = saveClusterEnv("PGDATABASE", "PGUSER", "PGHOST", "PGPORT")
+ })
+
+ AfterEach(func() {
+ connectLocalCluster = originalConnect
+ restoreClusterEnv(savedEnv)
+ })
+
+ Describe("NewClusterLocalClusterDefaultConn", func() {
+ It("uses PGDATABASE when it is set", func() {
+ setClusterEnv(map[string]string{
+ "PGDATABASE": "template1",
+ "PGUSER": "backup_user",
+ "PGHOST": "coordinator",
+ "PGPORT": "15432",
+ })
+ sqlDB, mock, err := sqlmock.New()
+ Expect(err).NotTo(HaveOccurred())
+ defer sqlDB.Close()
+ var gotDriver string
+ var gotConnStr string
+ connectLocalCluster = func(driverName, dataSourceName string) (*sqlx.DB, error) {
+ gotDriver = driverName
+ gotConnStr = dataSourceName
+ return sqlx.NewDb(sqlDB, "sqlmock"), nil
+ }
+
+ db, err := NewClusterLocalClusterDefaultConn()
+
+ Expect(err).NotTo(HaveOccurred())
+ mock.ExpectClose()
+ Expect(db.Close()).To(Succeed())
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ Expect(gotDriver).To(Equal("postgres"))
+ Expect(gotConnStr).To(Equal("postgres://backup_user@coordinator:15432/template1?sslmode=disable&connect_timeout=60"))
+ })
+
+ It("falls back to postgres when PGDATABASE is not set", func() {
+ setClusterEnv(map[string]string{
+ "PGUSER": "backup_user",
+ "PGHOST": "coordinator",
+ "PGPORT": "15432",
+ })
+ sqlDB, mock, err := sqlmock.New()
+ Expect(err).NotTo(HaveOccurred())
+ defer sqlDB.Close()
+ var gotConnStr string
+ connectLocalCluster = func(driverName, dataSourceName string) (*sqlx.DB, error) {
+ gotConnStr = dataSourceName
+ return sqlx.NewDb(sqlDB, "sqlmock"), nil
+ }
+
+ db, err := NewClusterLocalClusterDefaultConn()
+
+ Expect(err).NotTo(HaveOccurred())
+ mock.ExpectClose()
+ Expect(db.Close()).To(Succeed())
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ Expect(gotConnStr).To(Equal("postgres://backup_user@coordinator:15432/postgres?sslmode=disable&connect_timeout=60"))
+ })
+
+ It("returns connection errors", func() {
+ setClusterEnv(map[string]string{
+ "PGUSER": "backup_user",
+ "PGHOST": "coordinator",
+ "PGPORT": "15432",
+ })
+ connectErr := errors.New("connection failed")
+ connectLocalCluster = func(driverName, dataSourceName string) (*sqlx.DB, error) {
+ return nil, connectErr
+ }
+
+ db, err := NewClusterLocalClusterDefaultConn()
+
+ Expect(db).To(BeNil())
+ Expect(err).To(MatchError(connectErr))
+ })
+ })
+
+ Describe("QueryPrimaryCoordinatorDataDir", func() {
+ It("returns the primary coordinator data directory", func() {
+ db, mock := newClusterSQLMock()
+ defer db.Close()
+ mock.ExpectQuery(regexp.QuoteMeta(primaryCoordinatorDataDirSQL)).
+ WillReturnRows(sqlmock.NewRows([]string{"datadir"}).AddRow("/data/primary"))
+
+ dataDir, err := QueryPrimaryCoordinatorDataDir(db)
+
+ Expect(err).NotTo(HaveOccurred())
+ Expect(dataDir).To(Equal("/data/primary"))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("returns query errors", func() {
+ db, mock := newClusterSQLMock()
+ defer db.Close()
+ queryErr := errors.New("query failed")
+ mock.ExpectQuery(regexp.QuoteMeta(primaryCoordinatorDataDirSQL)).WillReturnError(queryErr)
+
+ dataDir, err := QueryPrimaryCoordinatorDataDir(db)
+
+ Expect(dataDir).To(BeEmpty())
+ Expect(err).To(MatchError(queryErr))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+ })
+
+ Describe("QueryUpStandbyCoordinator", func() {
+ It("returns the up standby coordinator", func() {
+ db, mock := newClusterSQLMock()
+ defer db.Close()
+ mock.ExpectQuery(regexp.QuoteMeta(upStandbyCoordinatorSQL)).
+ WillReturnRows(sqlmock.NewRows([]string{"hostname", "datadir"}).AddRow("standby-host", "/data/standby"))
+
+ standbyCoordinator, err := QueryUpStandbyCoordinator(db)
+
+ Expect(err).NotTo(HaveOccurred())
+ Expect(standbyCoordinator).To(Equal(StandbyCoordinator{
+ Hostname: "standby-host",
+ DataDir: "/data/standby",
+ }))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("returns sql.ErrNoRows when no up standby is present", func() {
+ db, mock := newClusterSQLMock()
+ defer db.Close()
+ mock.ExpectQuery(regexp.QuoteMeta(upStandbyCoordinatorSQL)).
+ WillReturnRows(sqlmock.NewRows([]string{"hostname", "datadir"}))
+
+ standbyCoordinator, err := QueryUpStandbyCoordinator(db)
+
+ Expect(standbyCoordinator).To(Equal(StandbyCoordinator{}))
+ Expect(errors.Is(err, sql.ErrNoRows)).To(BeTrue())
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+
+ It("returns query errors", func() {
+ db, mock := newClusterSQLMock()
+ defer db.Close()
+ queryErr := errors.New("query failed")
+ mock.ExpectQuery(regexp.QuoteMeta(upStandbyCoordinatorSQL)).WillReturnError(queryErr)
+
+ standbyCoordinator, err := QueryUpStandbyCoordinator(db)
+
+ Expect(standbyCoordinator).To(Equal(StandbyCoordinator{}))
+ Expect(err).To(MatchError(queryErr))
+ Expect(mock.ExpectationsWereMet()).To(Succeed())
+ })
+ })
+
+})
+
+func newClusterSQLMock() (*sqlx.DB, sqlmock.Sqlmock) {
+ sqlDB, mock, err := sqlmock.New()
+ Expect(err).NotTo(HaveOccurred())
+ return sqlx.NewDb(sqlDB, "sqlmock"), mock
+}
+
+func saveClusterEnv(names ...string) map[string]savedEnvValue {
+ saved := make(map[string]savedEnvValue, len(names))
+ for _, name := range names {
+ value, ok := os.LookupEnv(name)
+ saved[name] = savedEnvValue{value: value, ok: ok}
+ _ = os.Unsetenv(name)
+ }
+ return saved
+}
+
+func restoreClusterEnv(saved map[string]savedEnvValue) {
+ for name, envValue := range saved {
+ if envValue.ok {
+ _ = os.Setenv(name, envValue.value)
+ continue
+ }
+ _ = os.Unsetenv(name)
+ }
+}
+
+func setClusterEnv(values map[string]string) {
+ for name, value := range values {
+ _ = os.Setenv(name, value)
+ }
+}
diff --git a/gpbackman/textmsg/error.go b/gpbackman/textmsg/error.go
index 4659933..fc6475e 100644
--- a/gpbackman/textmsg/error.go
+++ b/gpbackman/textmsg/error.go
@@ -147,6 +147,10 @@
return fmt.Sprintf("Unable to clean db. Error: %v", err)
}
+func ErrorTextUnableSyncHistoryDBToStandby(err error) string {
+ return fmt.Sprintf("Unable to sync history db to standby coordinator. Error: %v", err)
+}
+
func ErrorTextUnableDeletePluginBackup(backupName string, err error) string {
return fmt.Sprintf("Unable to delete plugin backup %s. Error: %v", backupName, err)
}
@@ -259,6 +263,10 @@
return fmt.Errorf("invalid input value: %s", value)
}
+func ErrorHistoryStandbySyncSkippedError(reason string) error {
+ return fmt.Errorf("history db sync to standby coordinator skipped: %s", reason)
+}
+
// Error that is returned when backup has specific delete status.
func ErrorSetBackupDeleteStatus(backupName, status string) error {
diff --git a/gpbackman/textmsg/error_test.go b/gpbackman/textmsg/error_test.go
index b022f50..cd86e48 100644
--- a/gpbackman/textmsg/error_test.go
+++ b/gpbackman/textmsg/error_test.go
@@ -42,6 +42,7 @@
{"ErrorTextUnableCheckPath", ErrorTextUnableCheckPath, "Unable to check path. Error: test error"},
{"ErrorTextUnableDeleteLocalBackup", ErrorTextUnableDeleteLocalBackup, "Unable to delete local backup. Error: test error"},
{"ErrorTextUnableCleanDB", ErrorTextUnableCleanDB, "Unable to clean db. Error: test error"},
+ {"ErrorTextUnableSyncHistoryDBToStandby", ErrorTextUnableSyncHistoryDBToStandby, "Unable to sync history db to standby coordinator. Error: test error"},
}
for _, tt := range tests {
Expect(tt.function(testError)).To(Equal(tt.want), tt.name)
@@ -161,4 +162,15 @@
}
})
})
+
+ Describe("history standby sync errors", func() {
+ It("returns accepted text without environment values", func() {
+ err := ErrorHistoryStandbySyncSkippedError("no up standby coordinator found")
+
+ Expect(err).To(HaveOccurred())
+ Expect(err.Error()).To(Equal("history db sync to standby coordinator skipped: no up standby coordinator found"))
+ Expect(err.Error()).ToNot(ContainSubstring("PGPASSWORD"))
+ Expect(err.Error()).ToNot(ContainSubstring("secret"))
+ })
+ })
})
diff --git a/gpbackman/textmsg/info.go b/gpbackman/textmsg/info.go
index dafe140..f8a1f7e 100644
--- a/gpbackman/textmsg/info.go
+++ b/gpbackman/textmsg/info.go
@@ -71,3 +71,15 @@
func InfoTextNothingToDo() string {
return "Nothing to do"
}
+
+func InfoTextHistoryStandbySyncStart(sourceDBPath string) string {
+ return fmt.Sprintf("Sync history db to standby coordinator: %s", sourceDBPath)
+}
+
+func InfoTextHistoryStandbySyncSuccess(standbyHost, standbyHistoryDBPath string) string {
+ return fmt.Sprintf("History db sync to standby coordinator succeeded: %s:%s", standbyHost, standbyHistoryDBPath)
+}
+
+func InfoTextHistoryStandbySyncSkip(reason string) string {
+ return fmt.Sprintf("Skipping history db sync to standby coordinator: %s", reason)
+}
diff --git a/gpbackman/textmsg/info_test.go b/gpbackman/textmsg/info_test.go
index 34d7c5c..a7438d3 100644
--- a/gpbackman/textmsg/info_test.go
+++ b/gpbackman/textmsg/info_test.go
@@ -38,6 +38,8 @@
{"InfoTextBackupAlreadyDeleted", "TestBackup", InfoTextBackupAlreadyDeleted, "Backup TestBackup has already been deleted"},
{"InfoTextBackupDirPath", "/test/path", InfoTextBackupDirPath, "Path to backup directory: /test/path"},
{"InfoTextSegmentPrefix", "TestValue", InfoTextSegmentPrefix, "Segment Prefix: TestValue"},
+ {"InfoTextHistoryStandbySyncStart", "/data/gpbackup_history.db", InfoTextHistoryStandbySyncStart, "Sync history db to standby coordinator: /data/gpbackup_history.db"},
+ {"InfoTextHistoryStandbySyncSkip", "no up standby coordinator found", InfoTextHistoryStandbySyncSkip, "Skipping history db sync to standby coordinator: no up standby coordinator found"},
}
for _, tt := range tests {
Expect(tt.function(tt.value)).To(Equal(tt.want), tt.name)
@@ -55,6 +57,7 @@
want string
}{
{"InfoTextBackupStatus", "TestBackup", "In Progress", InfoTextBackupStatus, "Backup TestBackup has status: In Progress"},
+ {"InfoTextHistoryStandbySyncSuccess", "sdw-standby", "/standby/gpbackup_history.db", InfoTextHistoryStandbySyncSuccess, "History db sync to standby coordinator succeeded: sdw-standby:/standby/gpbackup_history.db"},
}
for _, tt := range tests {
Expect(tt.function(tt.value1, tt.value2)).To(Equal(tt.want), tt.name)
diff --git a/gpbackman/textmsg/warn.go b/gpbackman/textmsg/warn.go
index 97dc366..eb90663 100644
--- a/gpbackman/textmsg/warn.go
+++ b/gpbackman/textmsg/warn.go
@@ -24,3 +24,7 @@
func WarnTextBackupUnableGetReport(backupName string) string {
return fmt.Sprintf("Unable to get report for backup %s. Check if backup is active", backupName)
}
+
+func WarnTextHistoryStandbySyncFailed(err error) string {
+ return fmt.Sprintf("History db sync to standby coordinator failed; standby history may be stale: %v", err)
+}
diff --git a/gpbackman/textmsg/warn_test.go b/gpbackman/textmsg/warn_test.go
index 775d9c5..50077e5 100644
--- a/gpbackman/textmsg/warn_test.go
+++ b/gpbackman/textmsg/warn_test.go
@@ -20,6 +20,8 @@
package textmsg
import (
+ "errors"
+
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
@@ -40,4 +42,14 @@
}
})
})
+
+ Describe("warn text functions with error only", func() {
+ It("returns correct warn text without environment values", func() {
+ text := WarnTextHistoryStandbySyncFailed(errors.New("transport failed"))
+
+ Expect(text).To(Equal("History db sync to standby coordinator failed; standby history may be stale: transport failed"))
+ Expect(text).ToNot(ContainSubstring("PGPASSWORD"))
+ Expect(text).ToNot(ContainSubstring("secret"))
+ })
+ })
})
diff --git a/options/flag.go b/options/flag.go
index 3618a25..f79a23a 100644
--- a/options/flag.go
+++ b/options/flag.go
@@ -14,46 +14,49 @@
)
const (
- BACKUP_DIR = "backup-dir"
- COMPRESSION_TYPE = "compression-type"
- COMPRESSION_LEVEL = "compression-level"
- DATA_ONLY = "data-only"
- DBNAME = "dbname"
- DEBUG = "debug"
- EXCLUDE_RELATION = "exclude-table"
- EXCLUDE_RELATION_FILE = "exclude-table-file"
- EXCLUDE_SCHEMA = "exclude-schema"
- EXCLUDE_SCHEMA_FILE = "exclude-schema-file"
- FROM_TIMESTAMP = "from-timestamp"
- INCLUDE_RELATION = "include-table"
- INCLUDE_RELATION_FILE = "include-table-file"
- INCLUDE_SCHEMA = "include-schema"
- INCLUDE_SCHEMA_FILE = "include-schema-file"
- INCREMENTAL = "incremental"
- JOBS = "jobs"
- LEAF_PARTITION_DATA = "leaf-partition-data"
- METADATA_ONLY = "metadata-only"
- NO_COMPRESSION = "no-compression"
- NO_HISTORY = "no-history"
- PLUGIN_CONFIG = "plugin-config"
- QUIET = "quiet"
- SINGLE_DATA_FILE = "single-data-file"
- COPY_QUEUE_SIZE = "copy-queue-size"
- VERBOSE = "verbose"
- WITH_STATS = "with-stats"
- CREATE_DB = "create-db"
- ON_ERROR_CONTINUE = "on-error-continue"
- REDIRECT_DB = "redirect-db"
- RUN_ANALYZE = "run-analyze"
- SINGLE_BACKUP_DIR = "single-backup-dir"
- TIMESTAMP = "timestamp"
- WITH_GLOBALS = "with-globals"
- REDIRECT_SCHEMA = "redirect-schema"
- TRUNCATE_TABLE = "truncate-table"
- WITHOUT_GLOBALS = "without-globals"
- RESIZE_CLUSTER = "resize-cluster"
- NO_INHERITS = "no-inherits"
- REPORT_DIR = "report-dir"
+ BACKUP_DIR = "backup-dir"
+ COMPRESSION_TYPE = "compression-type"
+ COMPRESSION_LEVEL = "compression-level"
+ DATA_ONLY = "data-only"
+ DBNAME = "dbname"
+ DEBUG = "debug"
+ EXCLUDE_RELATION = "exclude-table"
+ EXCLUDE_RELATION_FILE = "exclude-table-file"
+ EXCLUDE_SCHEMA = "exclude-schema"
+ EXCLUDE_SCHEMA_FILE = "exclude-schema-file"
+ FROM_TIMESTAMP = "from-timestamp"
+ HISTORY_SYNC_STANDBY_TIMEOUT = "history-sync-standby-timeout"
+ INCLUDE_RELATION = "include-table"
+ INCLUDE_RELATION_FILE = "include-table-file"
+ INCLUDE_SCHEMA = "include-schema"
+ INCLUDE_SCHEMA_FILE = "include-schema-file"
+ INCREMENTAL = "incremental"
+ JOBS = "jobs"
+ LEAF_PARTITION_DATA = "leaf-partition-data"
+ METADATA_ONLY = "metadata-only"
+ NO_COMPRESSION = "no-compression"
+ NO_HISTORY = "no-history"
+ NO_HISTORY_SYNC_STANDBY = "no-history-sync-standby"
+ PLUGIN_CONFIG = "plugin-config"
+ QUIET = "quiet"
+ SINGLE_DATA_FILE = "single-data-file"
+ COPY_QUEUE_SIZE = "copy-queue-size"
+ VERBOSE = "verbose"
+ WITH_STATS = "with-stats"
+ CREATE_DB = "create-db"
+ ON_ERROR_CONTINUE = "on-error-continue"
+ REDIRECT_DB = "redirect-db"
+ RUN_ANALYZE = "run-analyze"
+ SINGLE_BACKUP_DIR = "single-backup-dir"
+ TIMESTAMP = "timestamp"
+ WITH_GLOBALS = "with-globals"
+ REDIRECT_SCHEMA = "redirect-schema"
+ TRUNCATE_TABLE = "truncate-table"
+ WITHOUT_GLOBALS = "without-globals"
+ RESIZE_CLUSTER = "resize-cluster"
+ NO_INHERITS = "no-inherits"
+ REPORT_DIR = "report-dir"
+ DEFAULT_HISTORY_SYNC_STANDBY_TIMEOUT = 300
)
func SetBackupFlagDefaults(flagSet *pflag.FlagSet) {
@@ -74,11 +77,13 @@
flagSet.StringArray(INCLUDE_RELATION, []string{}, "Back up only the specified table(s). --include-table can be specified multiple times.")
flagSet.String(INCLUDE_RELATION_FILE, "", "A file containing a list of fully-qualified tables to be included in the backup")
flagSet.Bool(INCREMENTAL, false, "Only back up data for AO tables that have been modified since the last backup")
+ flagSet.Int(HISTORY_SYNC_STANDBY_TIMEOUT, DEFAULT_HISTORY_SYNC_STANDBY_TIMEOUT, "Shared rsync and remote install timeout in seconds; must be an integer between 1 and 86400")
flagSet.Int(JOBS, 1, "The number of parallel connections to use when backing up data")
flagSet.Bool(LEAF_PARTITION_DATA, false, "For partition tables, create one data file per leaf partition instead of one data file for the whole table")
flagSet.Bool(METADATA_ONLY, false, "Only back up metadata, do not back up data")
flagSet.Bool(NO_COMPRESSION, false, "Skip compression of data files")
flagSet.Bool(NO_HISTORY, false, "Do not write a backup entry to the gpbackup_history database")
+ flagSet.Bool(NO_HISTORY_SYNC_STANDBY, false, "Do not sync gpbackup_history.db to the standby coordinator")
flagSet.String(PLUGIN_CONFIG, "", "The configuration file to use for a plugin")
flagSet.Bool("version", false, "Print version number and exit")
flagSet.Bool(QUIET, false, "Suppress non-warning, non-error log messages")
diff --git a/options/flag_test.go b/options/flag_test.go
index b9ecb27..57e5199 100644
--- a/options/flag_test.go
+++ b/options/flag_test.go
@@ -57,5 +57,58 @@
Expect(result).To(Equal([]string{"-s", "some_argument"}))
})
})
+ Context("SetBackupFlagDefaults", func() {
+ It("registers history-sync-standby-timeout with a 300 second default", func() {
+ flagSet := pflag.NewFlagSet("gpbackup", pflag.ContinueOnError)
+ options.SetBackupFlagDefaults(flagSet)
+
+ flag := flagSet.Lookup(options.HISTORY_SYNC_STANDBY_TIMEOUT)
+ Expect(flag).ToNot(BeNil())
+ Expect(flag.DefValue).To(Equal("300"))
+ value, err := flagSet.GetInt(options.HISTORY_SYNC_STANDBY_TIMEOUT)
+ Expect(err).ToNot(HaveOccurred())
+ Expect(value).To(Equal(300))
+ })
+
+ It("accepts a custom standby history sync timeout in seconds", func() {
+ flagSet := pflag.NewFlagSet("gpbackup", pflag.ContinueOnError)
+ options.SetBackupFlagDefaults(flagSet)
+
+ Expect(flagSet.Parse([]string{"--" + options.HISTORY_SYNC_STANDBY_TIMEOUT, "600"})).To(Succeed())
+ value, err := flagSet.GetInt(options.HISTORY_SYNC_STANDBY_TIMEOUT)
+ Expect(err).ToNot(HaveOccurred())
+ Expect(value).To(Equal(600))
+ })
+
+ DescribeTable("rejects non-integer standby history sync timeout values",
+ func(value string) {
+ flagSet := pflag.NewFlagSet("gpbackup", pflag.ContinueOnError)
+ options.SetBackupFlagDefaults(flagSet)
+
+ Expect(flagSet.Parse([]string{"--" + options.HISTORY_SYNC_STANDBY_TIMEOUT, value})).ToNot(Succeed())
+ },
+ Entry("fractional", "1.5"),
+ Entry("duration", "5m"),
+ )
+
+ It("registers no-history-sync-standby for gpbackup with a false default", func() {
+ flagSet := pflag.NewFlagSet("gpbackup", pflag.ContinueOnError)
+ options.SetBackupFlagDefaults(flagSet)
+
+ flag := flagSet.Lookup(options.NO_HISTORY_SYNC_STANDBY)
+ Expect(flag).ToNot(BeNil())
+ value, err := flagSet.GetBool(options.NO_HISTORY_SYNC_STANDBY)
+ Expect(err).ToNot(HaveOccurred())
+ Expect(value).To(BeFalse())
+ })
+
+ It("does not register no-history-sync-standby for gprestore", func() {
+ flagSet := pflag.NewFlagSet("gprestore", pflag.ContinueOnError)
+ options.SetRestoreFlagDefaults(flagSet)
+
+ Expect(flagSet.Lookup(options.NO_HISTORY_SYNC_STANDBY)).To(BeNil())
+ Expect(flagSet.Lookup(options.HISTORY_SYNC_STANDBY_TIMEOUT)).To(BeNil())
+ })
+ })
})
})
