Skip to content

fix(walrestore): use timeline-aware barman-cloud end-of-WAL markers #993

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ardentperf wants to merge 1 commit into
base: main
Choose a base branch
from
+16 −30
Open

fix(walrestore): use timeline-aware barman-cloud end-of-WAL markers #993

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 16 additions & 30 deletions internal/cnpgi/common/wal.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ package common

import (
"context"
"errors"
"fmt"
"os"
"path"
Expand Down Expand Up @@ -331,9 +330,14 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
return nil
}

// Step 2: return error if the end-of-wal-stream flag is set.
// PostgreSQL treats a restore_command failure as "WAL not available from
// archive" and can then try streaming via primary_conninfo. When prefetch has
// already observed archive exhaustion for this timeline, return one failure to
// let PostgreSQL switch sources instead of repeatedly probing the archive.
// We skip this step if streaming connection is not available
if isStreamingAvailable(cluster, w.InstanceName) {
if err := checkEndOfWALStreamFlag(walRestorer); err != nil {
if err := checkEndOfWALStreamFlag(walRestorer, walName); err != nil {
return err
}
}
Expand Down Expand Up @@ -366,15 +370,15 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
}

// We skip this step if streaming connection is not available
endOfWALStream := isEndOfWALStream(walStatus)
if isStreamingAvailable(cluster, w.InstanceName) && endOfWALStream {
contextLogger.Info(
"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")

err = walRestorer.SetEndOfWALStream()
if isStreamingAvailable(cluster, w.InstanceName) {
endOfWALStream, err := walRestorer.SetEndOfWALStreamFromResults(walStatus)
if err != nil {
return err
}
if endOfWALStream {
contextLogger.Info(
"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")
}
}

successfulWalRestore := 0
Expand Down Expand Up @@ -473,32 +477,14 @@ func gatherWALFilesToRestore(walName string, parallel int) (walList []string, er
return walList, err
}

// checkEndOfWALStreamFlag returns ErrEndOfWALStreamReached if the flag is set in the restorer.
func checkEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer) error {
contain, err := walRestorer.IsEndOfWALStream()
// checkEndOfWALStreamFlag returns ErrEndOfWALStreamReached if the flag is set for the requested WAL timeline.
func checkEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer, walName string) error {
contains, err := walRestorer.ConsumeEndOfWALStreamForWAL(walName)
if err != nil {
return err
}

if contain {
err := walRestorer.ResetEndOfWalStream()
if err != nil {
return err
}

if contains {
return ErrEndOfWALStreamReached
}
return nil
}

// isEndOfWALStream returns true if one of the downloads has returned
// a file-not-found error.
func isEndOfWALStream(results []barmanRestorer.Result) bool {
for _, result := range results {
if errors.Is(result.Err, barmanRestorer.ErrWALNotFound) {
return true
}
}

return false
}