mirror of
https://github.com/cloudnative-pg/plugin-barman-cloud.git
synced 2026-09-06 15:02:21 +02:00
Compare commits
6 Commits
f847c3d5d6
...
572cb59c05
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
572cb59c05 | ||
|
|
ce8179442b | ||
|
|
65b028003e | ||
|
|
7bcb74b987 | ||
|
|
1d777435d0 | ||
|
|
fb3f0f9dab |
30
.github/workflows/ci.yml
vendored
30
.github/workflows/ci.yml
vendored
@ -6,12 +6,15 @@ on:
|
||||
|
||||
permissions: read-all
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.ref }}
|
||||
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
|
||||
|
||||
jobs:
|
||||
ci:
|
||||
runs-on: ${{ vars.CI_RUNNERS || 'ubuntu-latest' }}
|
||||
permissions:
|
||||
packages: write
|
||||
contents: write
|
||||
contents: read
|
||||
steps:
|
||||
- name: Cleanup Disk
|
||||
if: vars.CI_RUNNERS == '' || vars.CI_RUNNERS == 'ubuntu-latest'
|
||||
@ -39,8 +42,6 @@ jobs:
|
||||
with:
|
||||
fetch-depth: 0
|
||||
ref: ${{ github.event.pull_request.head.sha }}
|
||||
- name: Install QEMU static binaries
|
||||
uses: docker/setup-qemu-action@v4
|
||||
- name: Install Task
|
||||
uses: arduino/setup-task@v3.0.0
|
||||
- name: Install Dagger
|
||||
@ -52,6 +53,27 @@ jobs:
|
||||
- name: Run CI task
|
||||
run: |
|
||||
task ci
|
||||
|
||||
publish:
|
||||
runs-on: ${{ vars.CI_RUNNERS || 'ubuntu-latest' }}
|
||||
permissions:
|
||||
contents: read
|
||||
packages: write
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v7.0.1
|
||||
with:
|
||||
ref: ${{ github.event.pull_request.head.sha }}
|
||||
- name: Install QEMU static binaries
|
||||
uses: docker/setup-qemu-action@v4
|
||||
- name: Install Task
|
||||
uses: arduino/setup-task@v3.0.0
|
||||
- name: Install Dagger
|
||||
env:
|
||||
# renovate: datasource=github-tags depName=dagger/dagger versioning=semver
|
||||
DAGGER_VERSION: 0.21.7
|
||||
run: |
|
||||
curl -L https://dl.dagger.io/dagger/install.sh | BIN_DIR=$HOME/.local/bin sh
|
||||
- name: Write manifest
|
||||
run: |
|
||||
task manifest
|
||||
|
||||
14
go.mod
14
go.mod
@ -16,10 +16,10 @@ require (
|
||||
github.com/spf13/viper v1.21.0
|
||||
google.golang.org/grpc v1.82.1
|
||||
gopkg.in/yaml.v3 v3.0.1
|
||||
k8s.io/api v0.36.2
|
||||
k8s.io/apiextensions-apiserver v0.36.2
|
||||
k8s.io/apimachinery v0.36.2
|
||||
k8s.io/client-go v0.36.2
|
||||
k8s.io/api v0.36.3
|
||||
k8s.io/apiextensions-apiserver v0.36.3
|
||||
k8s.io/apimachinery v0.36.3
|
||||
k8s.io/client-go v0.36.3
|
||||
k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3
|
||||
sigs.k8s.io/controller-runtime v0.24.1
|
||||
sigs.k8s.io/kustomize/api v0.21.1
|
||||
@ -125,11 +125,11 @@ require (
|
||||
google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect
|
||||
gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
|
||||
gopkg.in/inf.v0 v0.9.1 // indirect
|
||||
k8s.io/apiserver v0.36.2 // indirect
|
||||
k8s.io/component-base v0.36.2 // indirect
|
||||
k8s.io/apiserver v0.36.3 // indirect
|
||||
k8s.io/component-base v0.36.3 // indirect
|
||||
k8s.io/klog/v2 v2.140.0 // indirect
|
||||
k8s.io/kube-openapi v0.0.0-20260603220949-865597e52e25 // indirect
|
||||
k8s.io/streaming v0.36.2 // indirect
|
||||
k8s.io/streaming v0.36.3 // indirect
|
||||
sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.34.0 // indirect
|
||||
sigs.k8s.io/gateway-api v1.6.0 // indirect
|
||||
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
|
||||
|
||||
28
go.sum
28
go.sum
@ -308,24 +308,24 @@ gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
k8s.io/api v0.36.2 h1:TF6YDLIzKfccK7cq9YpTcGX8TJmEkHVRv78DM51fRYY=
|
||||
k8s.io/api v0.36.2/go.mod h1:F4LbMO4brjZYh7yFkXWhynSvtB7YauxV4c+HHkNRGNg=
|
||||
k8s.io/apiextensions-apiserver v0.36.2 h1:3O5gqOj/dt2XWWbpMe+TXWpE9yU6pjM/tXxtHHJT/K4=
|
||||
k8s.io/apiextensions-apiserver v0.36.2/go.mod h1:cL1tBWe8XSaP1H30iWKGo7hf6iAUUUJPEU70dskmAnA=
|
||||
k8s.io/apimachinery v0.36.2 h1:0PE/W/WNy1UX61NLbXY5TMbJ6UwLL6E6lAPkYrKFxbQ=
|
||||
k8s.io/apimachinery v0.36.2/go.mod h1:fvf/HOLXq9RId0rnDIbN1OEBvHXdQbLMM8nu0LcBUf4=
|
||||
k8s.io/apiserver v0.36.2 h1:6vMnkmHZPeBloNkHUhmZYq7Ylv8WIB8xjyEl+eSt26E=
|
||||
k8s.io/apiserver v0.36.2/go.mod h1:9PoQ2ikCytrZyZg11mGhLEF5m8Rgsb5FJmYJ4Wvnl1k=
|
||||
k8s.io/client-go v0.36.2 h1:bfgxmFKc9CgqsgX4xKLAAdmTQlWee7Ob/HlDOrJ5TBI=
|
||||
k8s.io/client-go v0.36.2/go.mod h1:1vgO4OAlfPnoLcb+Rze2GF5rAr14w8qjrYMoyXJzQj0=
|
||||
k8s.io/component-base v0.36.2 h1:Z0VH80O7Ng0HDZnZj3WRR3urEGa0kTwmO8CwEwjVK1w=
|
||||
k8s.io/component-base v0.36.2/go.mod h1:mGfFOA7Gwpdm1VW2cwSQYbiDIlz8GD2WGwH88QSeCyA=
|
||||
k8s.io/api v0.36.3 h1:NxB+05W2UGqXWFXcLO0RB5cnqnUPP5v5sVlaOH0Iz4w=
|
||||
k8s.io/api v0.36.3/go.mod h1:JzLQKqRHC5+I8RVj/lS3lCg0mg6nWI9Fo/Sk3ElxHzg=
|
||||
k8s.io/apiextensions-apiserver v0.36.3 h1:dPmOAPhwTtqb1bTxbFPsy18KHPhktQeO3WUPXunZIB0=
|
||||
k8s.io/apiextensions-apiserver v0.36.3/go.mod h1:KTXFqgXiuw2pRoL+Wpmttqc+up9Xt/GohadPWeLLOa4=
|
||||
k8s.io/apimachinery v0.36.3 h1:PkzMRBRG8joFD8EhCuQAtNPvJlxb82FwplP26HIzvAM=
|
||||
k8s.io/apimachinery v0.36.3/go.mod h1:cTSjBWgPe/6CQyBKzY/hDIRWCQQQeK0mfLbml0UYFHE=
|
||||
k8s.io/apiserver v0.36.3 h1:MGSg2SkdfuytiDEcRylT5mQFmmSsbx90XFUO67Y4bsQ=
|
||||
k8s.io/apiserver v0.36.3/go.mod h1:fVH7zv9EUNUA7Fl7LtDKh8aB9W7u1VQPSGtWV5SjUxg=
|
||||
k8s.io/client-go v0.36.3 h1:M4JdVzXxYcZk4fGpfDdYnxSwhLKWCFoQsHW6t+z8Hfg=
|
||||
k8s.io/client-go v0.36.3/go.mod h1:gcPwr0c87vjjG6HB6pWEqOeuYVoXSsREjzux2j6GF30=
|
||||
k8s.io/component-base v0.36.3 h1:vc/UFvPCkW0irPz84LAodAL1j3f4xktPM6dDJIEheAY=
|
||||
k8s.io/component-base v0.36.3/go.mod h1:hZbNFG+gCMl9EbykDGEu73feKP9/Cq6JsV4pTo9GTO8=
|
||||
k8s.io/klog/v2 v2.140.0 h1:Tf+J3AH7xnUzZyVVXhTgGhEKnFqye14aadWv7bzXdzc=
|
||||
k8s.io/klog/v2 v2.140.0/go.mod h1:o+/RWfJ6PwpnFn7OyAG3QnO47BFsymfEfrz6XyYSSp0=
|
||||
k8s.io/kube-openapi v0.0.0-20260603220949-865597e52e25 h1:mPMaPMpBij2V1Wv/fR+HW124vVGXXvOSS9ver/9yjWs=
|
||||
k8s.io/kube-openapi v0.0.0-20260603220949-865597e52e25/go.mod h1:V/QaCUYDa+0QpcHhVVc5l99Uz56wEMEXBSj9oCDkNDY=
|
||||
k8s.io/streaming v0.36.2 h1:NSKthPPg9UFSKsRauVJUVGH2Dvn8fhKmY4qrMkw/p98=
|
||||
k8s.io/streaming v0.36.2/go.mod h1:z6fV3D+NVkoeqRMtWwlUZK6U17SY/LqNzOxWL6GyR/s=
|
||||
k8s.io/streaming v0.36.3 h1:9rAaqBk0C0Pc7+/fqGekj07NV+/Xrew58p647A0JT8w=
|
||||
k8s.io/streaming v0.36.3/go.mod h1:z6fV3D+NVkoeqRMtWwlUZK6U17SY/LqNzOxWL6GyR/s=
|
||||
k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3 h1:jVkFFVfXdXP74B/zbO3hM3hpSFD0xvhQ5U686DPurkE=
|
||||
k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3/go.mod h1:M2s5JB1lIYP3jzZdorPLHXIPJzt9vv2muW5a6L9DtNM=
|
||||
sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.34.0 h1:hSfpvjjTQXQY2Fol2CS0QHMNs/WI1MOSGzCm1KhM5ec=
|
||||
|
||||
@ -27,6 +27,7 @@ import (
|
||||
"path"
|
||||
"time"
|
||||
|
||||
barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api"
|
||||
"github.com/cloudnative-pg/barman-cloud/pkg/archiver"
|
||||
barmanCommand "github.com/cloudnative-pg/barman-cloud/pkg/command"
|
||||
barmanCredentials "github.com/cloudnative-pg/barman-cloud/pkg/credentials"
|
||||
@ -155,13 +156,17 @@ func (w WALServiceImplementation) Archive(
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Step 2: Check if the archive location is safe to perform archiving
|
||||
checkFileExisting, err := fileutils.FileExists(emptyWalArchiveFile)
|
||||
// Step 2: Check if the archive location is safe to perform archiving.
|
||||
checkEmptyWalArchive, err := resolveArchiveEmptyWalArchiveCheck(
|
||||
request.CheckEmptyWalArchive,
|
||||
configuration.Cluster,
|
||||
emptyWalArchiveFile,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("while checking for empty wal archive check file %q: %w", emptyWalArchiveFile, err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if utils.IsEmptyWalArchiveCheckEnabled(&configuration.Cluster.ObjectMeta) && checkFileExisting {
|
||||
if checkEmptyWalArchive {
|
||||
if err := CheckBackupDestination(
|
||||
ctx,
|
||||
&objectStore.Spec.Configuration,
|
||||
@ -223,6 +228,31 @@ func (w WALServiceImplementation) Archive(
|
||||
return &wal.WALArchiveResult{}, nil
|
||||
}
|
||||
|
||||
// resolveArchiveEmptyWalArchiveCheck reports whether the WAL archive
|
||||
// destination must be verified before archiving this segment.
|
||||
//
|
||||
// The operator owns the marker file's lifecycle, so when it sets the decision
|
||||
// (non-nil) that value already accounts for the marker and is obeyed as-is. A
|
||||
// nil value comes from an operator that predates this field, so we fall back to
|
||||
// the previous logic: the Cluster annotation combined with the on-disk marker
|
||||
// file.
|
||||
func resolveArchiveEmptyWalArchiveCheck(
|
||||
operatorDecision *bool,
|
||||
cluster *cnpgv1.Cluster,
|
||||
markerFilePath string,
|
||||
) (bool, error) {
|
||||
if operatorDecision != nil {
|
||||
return *operatorDecision, nil
|
||||
}
|
||||
|
||||
markerFilePresent, err := fileutils.FileExists(markerFilePath)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("while checking for empty wal archive check file %q: %w", markerFilePath, err)
|
||||
}
|
||||
|
||||
return utils.IsEmptyWalArchiveCheckEnabled(&cluster.ObjectMeta) && markerFilePresent, nil
|
||||
}
|
||||
|
||||
// Restore implements the WALService interface
|
||||
func (w WALServiceImplementation) Restore(
|
||||
ctx context.Context,
|
||||
@ -249,9 +279,11 @@ func (w WALServiceImplementation) Restore(
|
||||
"Restoring WAL file",
|
||||
"objectStore", objectStore.Name,
|
||||
"serverName", serverName,
|
||||
"walName", walName)
|
||||
"walName", walName,
|
||||
"mode", request.GetMode())
|
||||
return &wal.WALRestoreResult{}, w.restoreFromBarmanObjectStore(
|
||||
ctx, configuration.Cluster, &objectStore, serverName, walName, destinationPath)
|
||||
ctx, configuration.Cluster, &objectStore, serverName, walName, destinationPath,
|
||||
request.GetMode() == wal.WALRestoreRequest_MODE_REWIND)
|
||||
}
|
||||
|
||||
// resolveRestoreObjectStore selects the object store and server name to use when
|
||||
@ -288,6 +320,7 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
|
||||
serverName string,
|
||||
walName string,
|
||||
destinationPath string,
|
||||
rewindMode bool,
|
||||
) error {
|
||||
contextLogger := log.FromContext(ctx)
|
||||
startTime := time.Now()
|
||||
@ -319,6 +352,20 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
|
||||
return fmt.Errorf("while creating the restorer: %w", err)
|
||||
}
|
||||
|
||||
// A flag left over from a normal-recovery invocation that ran before this pod
|
||||
// was demoted must not survive into a pg_rewind restore: the flag machinery
|
||||
// does not apply while restoring on behalf of pg_rewind (see
|
||||
// shouldUseEndOfWALStreamFlag below), so it is never checked here, but left
|
||||
// untouched it would resurface and wrongly abort the first normal-recovery
|
||||
// invocation that runs once the rewind is done. This runs unconditionally,
|
||||
// before the spool short-circuit in Step 1, so a request for a WAL file that
|
||||
// happens to already be staged in the spool cannot skip the clear.
|
||||
if rewindMode {
|
||||
if err := clearEndOfWALStreamFlag(walRestorer); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Step 1: check if this WAL file is not already in the spool
|
||||
var wasInSpool bool
|
||||
if wasInSpool, err = walRestorer.RestoreFromSpool(walName, destinationPath); err != nil {
|
||||
@ -331,8 +378,10 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
|
||||
return nil
|
||||
}
|
||||
|
||||
// We skip this step if streaming connection is not available
|
||||
if isStreamingAvailable(cluster, w.InstanceName) {
|
||||
// Step 2: return error if the end-of-wal-stream flag is set.
|
||||
// We skip this step if the flag machinery does not apply to this invocation
|
||||
useEndOfWALStreamFlag := shouldUseEndOfWALStreamFlag(cluster, w.InstanceName, rewindMode)
|
||||
if useEndOfWALStreamFlag {
|
||||
if err := checkEndOfWALStreamFlag(walRestorer); err != nil {
|
||||
return err
|
||||
}
|
||||
@ -340,10 +389,7 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
|
||||
|
||||
// Step 3: gather the WAL files names to restore. If the required file isn't a regular WAL, we download it directly.
|
||||
var walFilesList []string
|
||||
maxParallel := 1
|
||||
if barmanConfiguration.Wal != nil && barmanConfiguration.Wal.MaxParallel > 1 {
|
||||
maxParallel = barmanConfiguration.Wal.MaxParallel
|
||||
}
|
||||
maxParallel := maxWALFilesPerInvocation(barmanConfiguration, rewindMode)
|
||||
if IsWALFile(walName) {
|
||||
// If this is a regular WAL file, we try to prefetch
|
||||
if walFilesList, err = gatherWALFilesToRestore(walName, maxParallel); err != nil {
|
||||
@ -365,9 +411,9 @@ func (w WALServiceImplementation) restoreFromBarmanObjectStore(
|
||||
return classifyWALRestoreError(walStatus[0].WalName, walStatus[0].Err)
|
||||
}
|
||||
|
||||
// We skip this step if streaming connection is not available
|
||||
// We skip this step if the flag machinery does not apply to this invocation
|
||||
endOfWALStream := isEndOfWALStream(walStatus)
|
||||
if isStreamingAvailable(cluster, w.InstanceName) && endOfWALStream {
|
||||
if useEndOfWALStreamFlag && endOfWALStream {
|
||||
contextLogger.Info(
|
||||
"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")
|
||||
|
||||
@ -415,6 +461,38 @@ func (w WALServiceImplementation) SetFirstRequired(
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
// maxWALFilesPerInvocation returns how many WAL files a single restore
|
||||
// invocation is allowed to fetch, the requested one included. Prefetching is
|
||||
// disabled when restoring on behalf of pg_rewind, which walks the WAL
|
||||
// backward: the following segments would never be requested, and past the end
|
||||
// of the timeline they do not even exist
|
||||
func maxWALFilesPerInvocation(barmanConfiguration *barmanapi.BarmanObjectStoreConfiguration, rewindMode bool) int {
|
||||
if rewindMode {
|
||||
return 1
|
||||
}
|
||||
|
||||
if barmanConfiguration.Wal != nil && barmanConfiguration.Wal.MaxParallel > 1 {
|
||||
return barmanConfiguration.Wal.MaxParallel
|
||||
}
|
||||
|
||||
return 1
|
||||
}
|
||||
|
||||
// shouldUseEndOfWALStreamFlag returns true when the end-of-wal-stream flag
|
||||
// machinery applies to the current invocation. The flag makes the following
|
||||
// invocation fail, so that PostgreSQL stops polling the WAL archive and
|
||||
// switches to streaming replication. It does not apply when no streaming
|
||||
// connection is available, nor when restoring on behalf of pg_rewind:
|
||||
// pg_rewind cannot fall back to streaming replication, and a stale flag would
|
||||
// make it abort on a segment that is available in the archive
|
||||
func shouldUseEndOfWALStreamFlag(cluster *cnpgv1.Cluster, podName string, rewindMode bool) bool {
|
||||
if rewindMode {
|
||||
return false
|
||||
}
|
||||
|
||||
return isStreamingAvailable(cluster, podName)
|
||||
}
|
||||
|
||||
// isStreamingAvailable checks if this pod can replicate via streaming connection.
|
||||
func isStreamingAvailable(cluster *cnpgv1.Cluster, podName string) bool {
|
||||
if cluster == nil {
|
||||
@ -491,6 +569,22 @@ func checkEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// clearEndOfWALStreamFlag removes the end-of-wal-stream flag, if present, without
|
||||
// treating it as an error. It is used instead of checkEndOfWALStreamFlag when
|
||||
// restoring on behalf of pg_rewind, which must not abort on a flag it did not
|
||||
// set itself.
|
||||
func clearEndOfWALStreamFlag(walRestorer *barmanRestorer.WALRestorer) error {
|
||||
contain, err := walRestorer.IsEndOfWALStream()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if contain {
|
||||
return walRestorer.ResetEndOfWalStream()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// isEndOfWALStream returns true if one of the downloads has returned
|
||||
// a file-not-found error.
|
||||
func isEndOfWALStream(results []barmanRestorer.Result) bool {
|
||||
|
||||
@ -20,12 +20,21 @@ SPDX-License-Identifier: Apache-2.0
|
||||
package common
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api"
|
||||
barmanRestorer "github.com/cloudnative-pg/barman-cloud/pkg/restorer"
|
||||
cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/utils/ptr"
|
||||
|
||||
"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
|
||||
"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/config"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
|
||||
"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/config"
|
||||
)
|
||||
|
||||
var _ = Describe("resolveRestoreObjectStore", func() {
|
||||
@ -96,3 +105,142 @@ var _ = Describe("resolveRestoreObjectStore", func() {
|
||||
"cluster-server", "cluster-store"),
|
||||
)
|
||||
})
|
||||
|
||||
var _ = Describe("maxWALFilesPerInvocation", func() {
|
||||
configWithMaxParallel := func(maxParallel int) *barmanapi.BarmanObjectStoreConfiguration {
|
||||
return &barmanapi.BarmanObjectStoreConfiguration{
|
||||
Wal: &barmanapi.WalBackupConfiguration{MaxParallel: maxParallel},
|
||||
}
|
||||
}
|
||||
|
||||
DescribeTable(
|
||||
"computes how many WAL files a single invocation may fetch",
|
||||
func(cfg *barmanapi.BarmanObjectStoreConfiguration, rewindMode bool, want int) {
|
||||
Expect(maxWALFilesPerInvocation(cfg, rewindMode)).To(Equal(want))
|
||||
},
|
||||
|
||||
Entry("no WAL configuration", &barmanapi.BarmanObjectStoreConfiguration{}, false, 1),
|
||||
Entry("parallel restore configured", configWithMaxParallel(8), false, 8),
|
||||
|
||||
// pg_rewind walks the timeline backwards: prefetching must stay off no
|
||||
// matter what the object store configuration asks for
|
||||
Entry("rewind mode overrides the configured parallelism", configWithMaxParallel(8), true, 1),
|
||||
)
|
||||
})
|
||||
|
||||
var _ = Describe("shouldUseEndOfWALStreamFlag", func() {
|
||||
clusterWithPrimary := func(currentPrimary string) *cnpgv1.Cluster {
|
||||
return &cnpgv1.Cluster{
|
||||
Status: cnpgv1.ClusterStatus{CurrentPrimary: currentPrimary},
|
||||
}
|
||||
}
|
||||
|
||||
DescribeTable(
|
||||
"decides whether the end-of-wal-stream flag machinery applies",
|
||||
func(cluster *cnpgv1.Cluster, podName string, rewindMode bool, want bool) {
|
||||
Expect(shouldUseEndOfWALStreamFlag(cluster, podName, rewindMode)).To(Equal(want))
|
||||
},
|
||||
|
||||
Entry("replica with streaming available", clusterWithPrimary("cluster-1"), "cluster-2", false, true),
|
||||
Entry("primary cannot stream from anyone", clusterWithPrimary("cluster-1"), "cluster-1", false, false),
|
||||
|
||||
// pg_rewind cannot fall back to streaming replication: the flag machinery
|
||||
// must stay off even where a standby would use it
|
||||
Entry("rewind mode wins over streaming availability", clusterWithPrimary("cluster-1"), "cluster-2", true, false),
|
||||
)
|
||||
})
|
||||
|
||||
var _ = Describe("clearEndOfWALStreamFlag", func() {
|
||||
newRestorer := func() *barmanRestorer.WALRestorer {
|
||||
restorer, err := barmanRestorer.New(context.Background(), nil, GinkgoT().TempDir())
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
return restorer
|
||||
}
|
||||
|
||||
It("is a no-op when the flag is not set", func() {
|
||||
restorer := newRestorer()
|
||||
|
||||
Expect(clearEndOfWALStreamFlag(restorer)).To(Succeed())
|
||||
|
||||
isEOS, err := restorer.IsEndOfWALStream()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(isEOS).To(BeFalse())
|
||||
})
|
||||
|
||||
// Regression guard: a flag left over from a normal-recovery invocation that
|
||||
// ran before this pod was demoted must not survive a pg_rewind restore, or
|
||||
// it would resurface and wrongly abort the first normal-recovery invocation
|
||||
// that runs once the rewind is done.
|
||||
It("removes a pre-existing flag without returning an error", func() {
|
||||
restorer := newRestorer()
|
||||
Expect(restorer.SetEndOfWALStream()).To(Succeed())
|
||||
|
||||
Expect(clearEndOfWALStreamFlag(restorer)).To(Succeed())
|
||||
|
||||
isEOS, err := restorer.IsEndOfWALStream()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(isEOS).To(BeFalse())
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("resolveArchiveEmptyWalArchiveCheck", func() {
|
||||
// skipAnnotation mirrors the unexported constant in cloudnative-pg's
|
||||
// pkg/utils; hard-coding the literal makes a divergence surface as a
|
||||
// failing test rather than silently disabling the check.
|
||||
const skipAnnotation = "cnpg.io/skipEmptyWalArchiveCheck"
|
||||
|
||||
clusterWith := func(annotationValue *string) *cnpgv1.Cluster {
|
||||
cluster := &cnpgv1.Cluster{}
|
||||
if annotationValue != nil {
|
||||
cluster.Annotations = map[string]string{skipAnnotation: *annotationValue}
|
||||
}
|
||||
return cluster
|
||||
}
|
||||
|
||||
// markerPath returns the marker file path inside a fresh temp dir,
|
||||
// creating the file there when present is true.
|
||||
markerPath := func(present bool) string {
|
||||
filePath := filepath.Join(GinkgoT().TempDir(), metadata.CheckEmptyWalArchiveFile)
|
||||
if present {
|
||||
Expect(os.WriteFile(filePath, []byte{}, 0o600)).To(Succeed())
|
||||
}
|
||||
return filePath
|
||||
}
|
||||
|
||||
When("the operator sets the decision", func() {
|
||||
It("obeys true, ignoring the annotation and the marker file", func() {
|
||||
// annotation would skip the check and the marker is absent, yet the
|
||||
// operator's explicit true must still win.
|
||||
got, err := resolveArchiveEmptyWalArchiveCheck(
|
||||
ptr.To(true), clusterWith(ptr.To("enabled")), markerPath(false))
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(got).To(BeTrue())
|
||||
})
|
||||
|
||||
It("obeys false, ignoring the annotation and the marker file", func() {
|
||||
// annotation would keep the check on and the marker is present, yet the
|
||||
// operator's explicit false must still win.
|
||||
got, err := resolveArchiveEmptyWalArchiveCheck(
|
||||
ptr.To(false), clusterWith(nil), markerPath(true))
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(got).To(BeFalse())
|
||||
})
|
||||
})
|
||||
|
||||
When("the operator predates the field (nil decision)", func() {
|
||||
DescribeTable(
|
||||
"falls back to the annotation combined with the marker file",
|
||||
func(annotationValue *string, markerPresent bool, expected bool) {
|
||||
got, err := resolveArchiveEmptyWalArchiveCheck(
|
||||
nil, clusterWith(annotationValue), markerPath(markerPresent))
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
Expect(got).To(Equal(expected))
|
||||
},
|
||||
Entry("no annotation and marker present: check runs", nil, true, true),
|
||||
Entry("no annotation and marker absent: check skipped", nil, false, false),
|
||||
Entry("opt-out annotation and marker present: check skipped", ptr.To("enabled"), true, false),
|
||||
Entry("unrelated annotation value and marker present: check runs", ptr.To("something-else"), true, true),
|
||||
Entry("empty annotation value and marker present: check runs", ptr.To(""), true, true),
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
@ -113,6 +113,7 @@ func (impl JobHookImpl) Restore(
|
||||
configuration.Cluster,
|
||||
&targetObjectStore.Spec.Configuration,
|
||||
targetObjectStore.Name,
|
||||
req.CheckEmptyWalArchive,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@ -250,6 +251,7 @@ func (impl *JobHookImpl) checkBackupDestination(
|
||||
cluster *cnpgv1.Cluster,
|
||||
barmanConfiguration *cnpgv1.BarmanObjectStoreConfiguration,
|
||||
objectStoreName string,
|
||||
operatorCheckEmptyWalArchive *bool,
|
||||
) error {
|
||||
// Get environment from cache
|
||||
env, err := barmanCredentials.EnvSetCloudCredentialsAndCertificates(ctx,
|
||||
@ -288,14 +290,26 @@ func (impl *JobHookImpl) checkBackupDestination(
|
||||
}
|
||||
}
|
||||
|
||||
// Check if we're ok to archive in the desired destination
|
||||
if utils.IsEmptyWalArchiveCheckEnabled(&cluster.ObjectMeta) {
|
||||
if resolveRestoreEmptyWalArchiveCheck(operatorCheckEmptyWalArchive, cluster) {
|
||||
return common.CheckBackupDestination(ctx, barmanConfiguration, walArchiver, serverName)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// resolveRestoreEmptyWalArchiveCheck reports whether the destination must be
|
||||
// verified before restoring. When the operator sets the decision (non-nil) it
|
||||
// is obeyed as-is; a nil value comes from an operator that predates this field,
|
||||
// so we fall back to the Cluster annotation. Unlike archiving, restore is a
|
||||
// one-shot operation that has never been gated on the first-archive marker
|
||||
// file, so the annotation is the only fallback needed.
|
||||
func resolveRestoreEmptyWalArchiveCheck(operatorDecision *bool, cluster *cnpgv1.Cluster) bool {
|
||||
if operatorDecision != nil {
|
||||
return *operatorDecision
|
||||
}
|
||||
return utils.IsEmptyWalArchiveCheckEnabled(&cluster.ObjectMeta)
|
||||
}
|
||||
|
||||
// restoreCustomWalDir moves the current pg_wal data to the specified custom wal dir and applies the symlink
|
||||
// returns indicating if any changes were made and any error encountered in the process
|
||||
func (impl JobHookImpl) restoreCustomWalDir(ctx context.Context) (bool, error) {
|
||||
|
||||
66
internal/cnpgi/restore/restore_test.go
Normal file
66
internal/cnpgi/restore/restore_test.go
Normal file
@ -0,0 +1,66 @@
|
||||
/*
|
||||
Copyright © contributors to CloudNativePG, established as
|
||||
CloudNativePG a Series of LF Projects, LLC.
|
||||
|
||||
Licensed 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.
|
||||
|
||||
SPDX-License-Identifier: Apache-2.0
|
||||
*/
|
||||
|
||||
package restore
|
||||
|
||||
import (
|
||||
cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
|
||||
"k8s.io/utils/ptr"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("resolveRestoreEmptyWalArchiveCheck", func() {
|
||||
// skipAnnotation mirrors the unexported constant in cloudnative-pg's
|
||||
// pkg/utils; hard-coding the literal makes a divergence surface as a
|
||||
// failing test rather than silently disabling the check.
|
||||
const skipAnnotation = "cnpg.io/skipEmptyWalArchiveCheck"
|
||||
|
||||
clusterWith := func(annotationValue *string) *cnpgv1.Cluster {
|
||||
cluster := &cnpgv1.Cluster{}
|
||||
if annotationValue != nil {
|
||||
cluster.Annotations = map[string]string{skipAnnotation: *annotationValue}
|
||||
}
|
||||
return cluster
|
||||
}
|
||||
|
||||
When("the operator sets the decision", func() {
|
||||
It("obeys true even when the annotation would skip the check", func() {
|
||||
Expect(resolveRestoreEmptyWalArchiveCheck(ptr.To(true), clusterWith(ptr.To("enabled")))).To(BeTrue())
|
||||
})
|
||||
|
||||
It("obeys false even when the annotation would keep the check on", func() {
|
||||
Expect(resolveRestoreEmptyWalArchiveCheck(ptr.To(false), clusterWith(nil))).To(BeFalse())
|
||||
})
|
||||
})
|
||||
|
||||
When("the operator predates the field (nil decision)", func() {
|
||||
DescribeTable(
|
||||
"falls back to the Cluster annotation, never to a marker file",
|
||||
func(annotationValue *string, expected bool) {
|
||||
Expect(resolveRestoreEmptyWalArchiveCheck(nil, clusterWith(annotationValue))).To(Equal(expected))
|
||||
},
|
||||
Entry("no annotation: check runs", nil, true),
|
||||
Entry("opt-out annotation: check skipped", ptr.To("enabled"), false),
|
||||
Entry("unrelated annotation value: check runs", ptr.To("something-else"), true),
|
||||
Entry("empty annotation value: check runs", ptr.To(""), true),
|
||||
)
|
||||
})
|
||||
})
|
||||
32
internal/cnpgi/restore/suite_test.go
Normal file
32
internal/cnpgi/restore/suite_test.go
Normal file
@ -0,0 +1,32 @@
|
||||
/*
|
||||
Copyright © contributors to CloudNativePG, established as
|
||||
CloudNativePG a Series of LF Projects, LLC.
|
||||
|
||||
Licensed 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.
|
||||
|
||||
SPDX-License-Identifier: Apache-2.0
|
||||
*/
|
||||
|
||||
package restore
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
func TestRestore(t *testing.T) {
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "Restore job hook test suite")
|
||||
}
|
||||
@ -121,8 +121,8 @@ func newS3ClientDeployment(namespace string) *appsv1.Deployment {
|
||||
{
|
||||
Name: s3ClientName,
|
||||
// renovate: datasource=docker depName=amazon/aws-cli versioning=docker
|
||||
// Version: 2.36.3
|
||||
Image: "docker.io/amazon/aws-cli@sha256:bdd02067a00c354684086071b475955c54caa7bd88b851aac99a51326fe19652",
|
||||
// Version: 2.36.6
|
||||
Image: "docker.io/amazon/aws-cli@sha256:a9d9b72a496fd434172bef0a1cac0bf3812986184f9d3b3cbff65d6999e7c4f8",
|
||||
Command: []string{"sleep", "infinity"},
|
||||
Env: []corev1.EnvVar{
|
||||
{
|
||||
|
||||
@ -7970,9 +7970,9 @@ rc@1.2.8:
|
||||
strip-json-comments "~2.0.1"
|
||||
|
||||
react-dom@^19.0.0:
|
||||
version "19.2.7"
|
||||
resolved "https://registry.yarnpkg.com/react-dom/-/react-dom-19.2.7.tgz#0450dc9ae9ddbff76ef196401cd8b8c7fb466ccc"
|
||||
integrity sha512-t0BRVXvbiE/o20Hfw669rLbMCDWtYZLvmJigy2f0MxsXF+71pxhR3xOkspmsO8h3ZlNzyibAmtCa3l4lYKk6gQ==
|
||||
version "19.2.8"
|
||||
resolved "https://registry.yarnpkg.com/react-dom/-/react-dom-19.2.8.tgz#3b46b9eeda877cdff2cf13d2770fff4ae36c2ec2"
|
||||
integrity sha512-rVprimfGBG3DR+Tq0IQG2DT5PxKth1WIGDmj5yPmlzr4YBe7uyE+Du4oVqTDXZSHGGGXRtTJEGSSePyQCMBglQ==
|
||||
dependencies:
|
||||
scheduler "^0.27.0"
|
||||
|
||||
@ -8052,9 +8052,9 @@ react-router@5.3.4, react-router@^5.3.4:
|
||||
tiny-warning "^1.0.0"
|
||||
|
||||
react@^19.0.0:
|
||||
version "19.2.7"
|
||||
resolved "https://registry.yarnpkg.com/react/-/react-19.2.7.tgz#1f47a1bfc06f8ec885752c6f4af14369a9f8260b"
|
||||
integrity sha512-HNe9WslTbXmFK8o8cmwgAeJFSBvt1bPdHCVKtaaV+WlAN36mpT4hcRpwbf3fY56ar2oIXzsBpOAiIRHAdY0OlQ==
|
||||
version "19.2.8"
|
||||
resolved "https://registry.yarnpkg.com/react/-/react-19.2.8.tgz#a80663dbb58d69c6fe3fd291d3cb324e8a7dff2d"
|
||||
integrity sha512-PWaYA1L/q9u2u7xYQi+Y3L3Yfnie7XyLeaJICV1MGD6LprsBxcAqGjYyr0eY3p+QdsA+x/Irkt4Qif8D63+Sbw==
|
||||
|
||||
readable-stream@^2.0.1:
|
||||
version "2.3.8"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user