From daac9901783ba78b0449f5a7fcbfaffe1c53dfc1 Mon Sep 17 00:00:00 2001 From: Vasiliy Fakunin <61789920+melancholictheory@users.noreply.github.com> Date: Tue, 28 Jul 2026 18:50:56 +0300 Subject: [PATCH 1/3] feat: create pgbackrest stanza during WAL archiving when missing Stanza creation previously ran only inside the Backup RPC, so on a fresh cluster (or after a major upgrade changes the repository path) WAL archiving stayed broken until the first backup happened to run: archive-push kept failing with "has a stanza-create been performed?", and a rejoining replica could hang on archive recovery. The WAL archive path now inspects the "pgbackrest info" status and, when the stanza is missing (status code 1), runs the existing idempotent CreatePgbackrestStanza before archive-push. This executes on the primary as soon as its sidecar is up and relies on PostgreSQL's own archive_command retry, so archiving comes up without waiting for a backup. Creation is best-effort and only triggers when the stanza is genuinely absent, so it does not contend with a running backup for the stanza lock. The stanza check lives entirely in the WAL archive path and does not touch CheckWalArchiveDestination, which the restore path uses for a read-only check. Refs #60, #18, #42. Signed-off-by: Vasiliy Fakunin <61789920+melancholictheory@users.noreply.github.com> --- internal/cnpgi/common/wal.go | 29 +++++++++++++++++++-- internal/pgbackrest/catalog/catalog.go | 20 ++++++++++++++ internal/pgbackrest/catalog/catalog_test.go | 23 ++++++++++++++++ 3 files changed, 70 insertions(+), 2 deletions(-) diff --git a/internal/cnpgi/common/wal.go b/internal/cnpgi/common/wal.go index ef82226..c8c2c22 100644 --- a/internal/cnpgi/common/wal.go +++ b/internal/cnpgi/common/wal.go @@ -35,6 +35,7 @@ import ( "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/metadata" "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/cnpgi/operator/config" "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/archiver" + pgbackrestBackup "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/backup" pgbackrestCommand "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/command" pgbackrestCredentials "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/credentials" pgbackrestRestorer "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/restorer" @@ -136,13 +137,37 @@ func (w WALServiceImplementation) Archive( return &wal.WALArchiveResult{}, nil } - // Check if we're ok to archive in the desired destination - err = arch.CheckWalArchiveDestination(ctx, &archive.Spec.Configuration, configuration.Stanza, envArchive) + // Check that the destination repository is reachable and inspect the stanza status. + // This is the same "pgbackrest info" call that CheckWalArchiveDestination performs + // for the restore path, but here we keep the returned catalog so we can create the + // stanza below when it is missing. + destinationCatalog, err := pgbackrestCommand.GetBackupList(ctx, &archive.Spec.Configuration, configuration.Stanza, envArchive) if err != nil { log.Error(err, "while checking if pgbackrest repo can be used for archival") return nil, err } + // On a fresh cluster, or after a major upgrade changes the repository path, the + // stanza does not exist yet and archive-push cannot succeed until it is created. + // Stanza creation otherwise runs only during the first backup, so WAL archiving + // stays broken until a backup happens to run. Create it here instead: this path + // executes on the primary as soon as its sidecar is up, and PostgreSQL retries + // archiving on its own. stanza-create is idempotent, and we only reach it when the + // stanza is genuinely missing (no backup has created it yet), so it does not + // contend with a running backup for the stanza lock. + if destinationCatalog.StanzaMissing() { + backupCmd := pgbackrestBackup.NewBackupCommand(&archive.Spec.Configuration, nil, w.PGDataPath) + if stanzaErr := backupCmd.CreatePgbackrestStanza(ctx, configuration.Stanza, envArchive); stanzaErr != nil { + // Best-effort: log and continue. archive-push below reports the real + // outcome, and PostgreSQL retries the WAL if the stanza is still missing. + contextLogger.Warning("could not auto-create pgbackrest stanza; WAL archiving will retry", + "stanza", configuration.Stanza, "err", stanzaErr.Error()) + } else { + contextLogger.Info("created pgbackrest stanza so WAL archiving can start", + "stanza", configuration.Stanza) + } + } + options, err := arch.PgbackrestWalArchiveOptions(ctx, &archive.Spec.Configuration, configuration.Stanza) if err != nil { return nil, err diff --git a/internal/pgbackrest/catalog/catalog.go b/internal/pgbackrest/catalog/catalog.go index 27ce405..6c8171e 100644 --- a/internal/pgbackrest/catalog/catalog.go +++ b/internal/pgbackrest/catalog/catalog.go @@ -335,6 +335,26 @@ type Catalog struct { Stanza string `json:"name"` Databases []PgbackrestBackupDatabase `json:"db"` Encryption string `json:"cipher"` + Status PgbackrestStanzaStatus `json:"status"` +} + +// StanzaStatusCodeMissing is the "pgbackrest info" stanza status code returned when +// the stanza has not been created in the repository yet (its archive.info is absent). +// While the stanza is in this state WAL archiving cannot succeed until +// "pgbackrest stanza-create" has been run. +const StanzaStatusCodeMissing = 1 + +// PgbackrestStanzaStatus is the "status" object that "pgbackrest info" reports for a +// stanza. Only the fields we act on are parsed; the rest of the object is ignored. +type PgbackrestStanzaStatus struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// StanzaMissing reports whether "pgbackrest info" says the stanza has not been created +// in the repository yet, i.e. WAL archiving cannot work until stanza-create is run. +func (catalog *Catalog) StanzaMissing() bool { + return catalog.Status.Code == StanzaStatusCodeMissing } // NewSingleBackupCatalogFromPgbackrestInfo parses the output of pgbackrest info diff --git a/internal/pgbackrest/catalog/catalog_test.go b/internal/pgbackrest/catalog/catalog_test.go index 50d5b89..0cb8919 100644 --- a/internal/pgbackrest/catalog/catalog_test.go +++ b/internal/pgbackrest/catalog/catalog_test.go @@ -144,6 +144,29 @@ var _ = Describe("pgbackrest info parsing", func() { Equal(time.Date(2025, 3, 31, 14, 20, 41, 0, time.UTC))) }) + It("reports the stanza as present when the status code is zero", func() { + result, err := NewCatalogFromPgbackrestInfo(pgbackrestInfoOutput) + Expect(err).ToNot(HaveOccurred()) + Expect(result.StanzaMissing()).To(BeFalse()) + }) + + It("reports the stanza as missing when the status code is 1", func() { + const missingStanzaOutput = `[ + { + "archive": [], + "backup": [], + "cipher": "none", + "db": [], + "name": "cluster-example-pgbackrest", + "status": { "code": 1, "message": "missing stanza path" } + } +]` + result, err := NewCatalogFromPgbackrestInfo(missingStanzaOutput) + Expect(err).ToNot(HaveOccurred()) + Expect(result.StanzaMissing()).To(BeTrue()) + Expect(result.Status.Message).To(Equal("missing stanza path")) + }) + // It("can find the closest backup info when there is one", func() { // recoveryTarget := &v1.RecoveryTarget{TargetTime: time.Now().Format("2006-01-02 15:04:04")} // closestBackupInfo, err := catalog.FindBackupInfo(recoveryTarget) From d7e397defbf88bd8caece8797dac4b3e9d2f429c Mon Sep 17 00:00:00 2001 From: Samarth Verma Date: Tue, 14 Jul 2026 09:19:29 -0400 Subject: [PATCH 2/3] test(e2e): add WAL archive without prior backup test Covers lazy stanza creation on first WAL archive, successful archiving before any backup, then backup and restore. Log assertion matches #121. Co-authored-by: Cursor Signed-off-by: Vasiliy Fakunin <61789920+melancholictheory@users.noreply.github.com> --- test/e2e/e2e_suite_test.go | 1 + test/e2e/internal/logs/logs.go | 25 +- test/e2e/internal/tests/walarchive/doc.go | 20 ++ .../e2e/internal/tests/walarchive/fixtures.go | 172 ++++++++++++ .../tests/walarchive/wal_archive_no_backup.go | 251 ++++++++++++++++++ 5 files changed, 454 insertions(+), 15 deletions(-) create mode 100644 test/e2e/internal/tests/walarchive/doc.go create mode 100644 test/e2e/internal/tests/walarchive/fixtures.go create mode 100644 test/e2e/internal/tests/walarchive/wal_archive_no_backup.go diff --git a/test/e2e/e2e_suite_test.go b/test/e2e/e2e_suite_test.go index 65108ba..8fa63d4 100644 --- a/test/e2e/e2e_suite_test.go +++ b/test/e2e/e2e_suite_test.go @@ -36,6 +36,7 @@ import ( _ "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/tests/backup" _ "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/tests/parallelarchive" _ "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/tests/replicacluster" + _ "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/tests/walarchive" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" diff --git a/test/e2e/internal/logs/logs.go b/test/e2e/internal/logs/logs.go index ae2465c..1a105cd 100644 --- a/test/e2e/internal/logs/logs.go +++ b/test/e2e/internal/logs/logs.go @@ -84,29 +84,24 @@ func GetPodContainerLogs( // FindArchiveBatches finds "WAL archive batch prepared" log entries and returns the parsed data. // Each returned map contains the structured log fields. func FindArchiveBatches(logEntries []map[string]any) []map[string]any { - var batches []map[string]any - - for _, logEntry := range logEntries { - // Check if this is a "WAL archive batch prepared" message - if msg, ok := logEntry["msg"].(string); ok && msg == "WAL archive batch prepared" { - batches = append(batches, logEntry) - } - } - - return batches + return FindLogEntriesByMessage(logEntries, "WAL archive batch prepared") } // FindArchiveBatchCompletions finds "WAL archive batch completed" log entries and returns the parsed data. // Each returned map contains the structured log fields. func FindArchiveBatchCompletions(logEntries []map[string]any) []map[string]any { - var batches []map[string]any + return FindLogEntriesByMessage(logEntries, "WAL archive batch completed") +} + +// FindLogEntriesByMessage returns the log entries whose "msg" field equals the given message. +func FindLogEntriesByMessage(logEntries []map[string]any, message string) []map[string]any { + var matches []map[string]any for _, logEntry := range logEntries { - // Check if this is a "WAL archive batch completed" message - if msg, ok := logEntry["msg"].(string); ok && msg == "WAL archive batch completed" { - batches = append(batches, logEntry) + if msg, ok := logEntry["msg"].(string); ok && msg == message { + matches = append(matches, logEntry) } } - return batches + return matches } diff --git a/test/e2e/internal/tests/walarchive/doc.go b/test/e2e/internal/tests/walarchive/doc.go new file mode 100644 index 0000000..ff20bd4 --- /dev/null +++ b/test/e2e/internal/tests/walarchive/doc.go @@ -0,0 +1,20 @@ +/* +Copyright 2025, Opera Norway AS + +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. +*/ + +// Package walarchive contains e2e tests verifying that WAL archiving works +// without taking a pgbackrest backup first, by creating the stanza lazily on +// the first WAL archive. +package walarchive diff --git a/test/e2e/internal/tests/walarchive/fixtures.go b/test/e2e/internal/tests/walarchive/fixtures.go new file mode 100644 index 0000000..e4971d1 --- /dev/null +++ b/test/e2e/internal/tests/walarchive/fixtures.go @@ -0,0 +1,172 @@ +/* +Copyright 2025, Opera Norway AS + +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. +*/ + +package walarchive + +import ( + cloudnativepgv1 "github.com/cloudnative-pg/api/pkg/api/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + pluginPgbackrestV1 "github.com/operasoftware/cnpg-plugin-pgbackrest/api/v1" + "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/objectstore" +) + +const ( + minio = "minio" + // size is the size of the PVCs for the object store and the cluster instances. + size = "1Gi" + pluginName = "pgbackrest.cnpg.opera.com" + srcClusterName = "wal-archive-source" + archiveName = "wal-archive" + backupName = "wal-archive-backup" + restoreClusterName = "wal-archive-restore" +) + +// walArchiveTestResources contains the resources needed to test WAL archiving +// without a prior backup, plus backup and restore to prove those paths still work. +type walArchiveTestResources struct { + ObjectStoreResources *objectstore.Resources + Archive *pluginPgbackrestV1.Archive + Cluster *cloudnativepgv1.Cluster + Backup *cloudnativepgv1.Backup + RestoreCluster *cloudnativepgv1.Cluster +} + +// createWalArchiveTestResources builds all resources for the WAL archiving test. +func createWalArchiveTestResources(namespace string) walArchiveTestResources { + return walArchiveTestResources{ + ObjectStoreResources: objectstore.NewMinioObjectStoreResources(namespace, minio), + // maxParallel=1 so every WAL is archived in its own batch, which keeps the + // assertions on batch completions unambiguous. + Archive: objectstore.NewMinioArchive(namespace, archiveName, minio, 1), + Cluster: newClusterWithPlugin(namespace, srcClusterName), + Backup: newPluginBackup(namespace, backupName, srcClusterName), + RestoreCluster: newRestoreCluster(namespace, restoreClusterName), + } +} + +// newClusterWithPlugin creates a cluster that only enables the plugin for WAL +// archiving. Crucially it defines no bootstrap and no backup, so the stanza must +// be created lazily on the first WAL archive. +func newClusterWithPlugin(namespace, name string) *cloudnativepgv1.Cluster { + return &cloudnativepgv1.Cluster{ + TypeMeta: metav1.TypeMeta{ + Kind: "Cluster", + APIVersion: "postgresql.cnpg.io/v1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + }, + Spec: cloudnativepgv1.ClusterSpec{ + Instances: 2, + ImagePullPolicy: corev1.PullAlways, + Plugins: []cloudnativepgv1.PluginConfiguration{ + { + Name: pluginName, + Parameters: map[string]string{ + "pgbackrestObjectName": archiveName, + }, + }, + }, + PostgresConfiguration: cloudnativepgv1.PostgresConfiguration{ + Parameters: map[string]string{ + "log_min_messages": "DEBUG4", + }, + }, + StorageConfiguration: cloudnativepgv1.StorageConfiguration{ + Size: size, + }, + }, + } +} + +// newPluginBackup creates a plugin backup targeting the primary of the given cluster. +func newPluginBackup(namespace, name, clusterName string) *cloudnativepgv1.Backup { + return &cloudnativepgv1.Backup{ + TypeMeta: metav1.TypeMeta{ + Kind: "Backup", + APIVersion: "postgresql.cnpg.io/v1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + }, + Spec: cloudnativepgv1.BackupSpec{ + Cluster: cloudnativepgv1.LocalObjectReference{ + Name: clusterName, + }, + Method: "plugin", + Target: "primary", + PluginConfiguration: &cloudnativepgv1.BackupPluginConfiguration{ + Name: pluginName, + }, + }, + } +} + +// newRestoreCluster creates a cluster that bootstraps by recovering from the +// source cluster's archive, while also archiving its own WALs to the same store. +func newRestoreCluster(namespace, name string) *cloudnativepgv1.Cluster { + return &cloudnativepgv1.Cluster{ + TypeMeta: metav1.TypeMeta{ + Kind: "Cluster", + APIVersion: "postgresql.cnpg.io/v1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + }, + Spec: cloudnativepgv1.ClusterSpec{ + Instances: 2, + ImagePullPolicy: corev1.PullAlways, + Bootstrap: &cloudnativepgv1.BootstrapConfiguration{ + Recovery: &cloudnativepgv1.BootstrapRecovery{ + Source: "source", + }, + }, + Plugins: []cloudnativepgv1.PluginConfiguration{ + { + Name: pluginName, + Parameters: map[string]string{ + "pgbackrestObjectName": archiveName, + }, + }, + }, + PostgresConfiguration: cloudnativepgv1.PostgresConfiguration{ + Parameters: map[string]string{ + "log_min_messages": "DEBUG4", + }, + }, + ExternalClusters: []cloudnativepgv1.ExternalCluster{ + { + Name: "source", + PluginConfiguration: &cloudnativepgv1.PluginConfiguration{ + Name: pluginName, + Parameters: map[string]string{ + "pgbackrestObjectName": archiveName, + "stanza": srcClusterName, + }, + }, + }, + }, + StorageConfiguration: cloudnativepgv1.StorageConfiguration{ + Size: size, + }, + }, + } +} diff --git a/test/e2e/internal/tests/walarchive/wal_archive_no_backup.go b/test/e2e/internal/tests/walarchive/wal_archive_no_backup.go new file mode 100644 index 0000000..9b06da0 --- /dev/null +++ b/test/e2e/internal/tests/walarchive/wal_archive_no_backup.go @@ -0,0 +1,251 @@ +/* +Copyright 2025, Opera Norway AS + +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. +*/ + +package walarchive + +import ( + "fmt" + "strings" + "time" + + v1 "github.com/cloudnative-pg/api/pkg/api/v1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "sigs.k8s.io/controller-runtime/pkg/client" + + internalClient "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/client" + internalCluster "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/cluster" + "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/command" + internalLogs "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/logs" + nmsp "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/namespace" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// lazyStanzaLogMessage is the message emitted by the WAL archive handler when it +// creates the pgbackrest stanza on the fly because it did not exist yet. +const lazyStanzaLogMessage = "created pgbackrest stanza so WAL archiving can start" + +var _ = Describe("WAL archiving without a prior backup", func() { + var namespace *corev1.Namespace + var cl client.Client + + BeforeEach(func(ctx SpecContext) { + var err error + cl, _, err = internalClient.NewClient() + Expect(err).NotTo(HaveOccurred()) + namespace, err = nmsp.CreateUniqueNamespace(ctx, cl, "wal-archive-no-backup") + Expect(err).NotTo(HaveOccurred()) + }) + + AfterEach(func(ctx SpecContext) { + Expect(cl.Delete(ctx, namespace)).To(Succeed()) + }) + + It("creates the stanza lazily, archives WALs before any backup, and still supports backup and restore", + func(ctx SpecContext) { + testResources := createWalArchiveTestResources(namespace.Name) + + By("starting the object store deployment") + Expect(testResources.ObjectStoreResources.Create(ctx, cl)).To(Succeed()) + + By("creating the Archive") + Expect(cl.Create(ctx, testResources.Archive)).To(Succeed()) + + By("creating a CloudNativePG cluster that only enables WAL archiving (no backup)") + cluster := testResources.Cluster + Expect(cl.Create(ctx, cluster)).To(Succeed()) + + By("waiting for the cluster to be ready") + waitForClusterReady(ctx, cl, cluster) + + clientSet, cfg, err := internalClient.NewClientSet() + Expect(err).NotTo(HaveOccurred()) + + primaryPod := fmt.Sprintf("%s-1", cluster.Name) + + By("adding data to PostgreSQL") + execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "CREATE TABLE wal_test (id int, data text);") + + By("generating WAL files WITHOUT creating any backup first") + // Each pg_switch_wal() forces a WAL segment to be archived. On a fresh + // cluster the stanza does not exist yet, so the first archive must create + // it lazily. Before the fix, this would fail indefinitely until a backup + // was taken. + for i := 0; i < 5; i++ { + execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, + fmt.Sprintf("INSERT INTO wal_test VALUES (%d, 'data-%d'); SELECT pg_switch_wal();", i, i)) + time.Sleep(500 * time.Millisecond) + } + + By("verifying the stanza was created lazily and WALs were archived successfully") + Eventually(func(g Gomega) { + logs := getSidecarLogs(ctx, g, clientSet, cluster.Namespace, primaryPod) + + lazyStanzaEntries := internalLogs.FindLogEntriesByMessage(logs, lazyStanzaLogMessage) + g.Expect(lazyStanzaEntries).NotTo(BeEmpty(), + "the sidecar should have logged that it created the stanza on the first WAL archive") + + completedBatches := internalLogs.FindArchiveBatchCompletions(logs) + g.Expect(completedBatches).NotTo(BeEmpty(), + "there should be at least one completed WAL archive batch") + + g.Expect(hasSuccessfulArchiveBatch(completedBatches)).To(BeTrue(), + "at least one WAL archive batch should have completed with a successful, error-free archive") + }).WithTimeout(4 * time.Minute).WithPolling(10 * time.Second).Should(Succeed()) + + By("taking a backup now to prove the backup path still works after lazy stanza creation") + backup := testResources.Backup + Expect(cl.Create(ctx, backup)).To(Succeed()) + waitForBackupCompleted(ctx, cl, backup) + + By("adding data after the backup so restore has WALs to replay") + execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "INSERT INTO wal_test VALUES (99, 'after-backup');") + // The committed row lives in the current WAL segment. Capture it, force a + // switch so it becomes archivable, then wait until the archiver has actually + // uploaded it. Otherwise the restored cluster may finish recovery before the + // segment reaches the object store and the post-backup row would be lost. + postBackupWAL := queryPsqlOutput(ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "SELECT pg_walfile_name(pg_current_wal_lsn());") + execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, "SELECT pg_switch_wal();") + + By("waiting for the post-backup WAL segment to be archived") + Eventually(func(g Gomega) { + lastArchived := queryPsqlOutputG(g, ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "SELECT COALESCE(last_archived_wal, '') FROM pg_stat_archiver;") + g.Expect(lastArchived).NotTo(BeEmpty(), "no WAL has been archived yet") + g.Expect(lastArchived >= postBackupWAL).To(BeTrue(), + fmt.Sprintf("archived WAL %q has not yet reached the post-backup segment %q", + lastArchived, postBackupWAL)) + }).WithTimeout(2 * time.Minute).WithPolling(5 * time.Second).Should(Succeed()) + + By("restoring into a new cluster from the archive") + restore := testResources.RestoreCluster + Expect(cl.Create(ctx, restore)).To(Succeed()) + + By("waiting for the restored cluster to be ready") + waitForClusterReady(ctx, cl, restore) + + By("verifying the restored data is present") + output := queryPsqlOutput(ctx, clientSet, cfg, restore.Namespace, + fmt.Sprintf("%s-1", restore.Name), "SELECT count(*) FROM wal_test;") + Expect(output).To(Equal("6"), + "restored cluster should contain the 5 pre-backup rows plus the 1 post-backup row") + }) +}) + +// waitForClusterReady blocks until the given cluster reports a ready status. +func waitForClusterReady(ctx SpecContext, cl client.Client, cluster *v1.Cluster) { + Eventually(func(g Gomega) { + g.Expect(cl.Get(ctx, + types.NamespacedName{Name: cluster.Name, Namespace: cluster.Namespace}, + cluster)).To(Succeed()) + g.Expect(internalCluster.IsReady(*cluster)).To(BeTrue()) + }).WithTimeout(10 * time.Minute).WithPolling(10 * time.Second).Should(Succeed()) +} + +// waitForBackupCompleted blocks until the given backup reaches the completed phase. +func waitForBackupCompleted(ctx SpecContext, cl client.Client, backup *v1.Backup) { + Eventually(func(g Gomega) { + g.Expect(cl.Get(ctx, + types.NamespacedName{Name: backup.Name, Namespace: backup.Namespace}, + backup)).To(Succeed()) + g.Expect(backup.Status.Phase).To(BeEquivalentTo(v1.BackupPhaseCompleted)) + }).Within(3 * time.Minute).WithPolling(5 * time.Second).Should(Succeed()) +} + +// execPsql runs a SQL statement in the postgres container of the given pod. +func execPsql( + ctx SpecContext, + clientSet *kubernetes.Clientset, + cfg *rest.Config, + namespace, podName, sql string, +) { + _, _, err := command.ExecuteInContainer(ctx, + *clientSet, + cfg, + command.ContainerLocator{ + NamespaceName: namespace, + PodName: podName, + ContainerName: "postgres", + }, + nil, + []string{"psql", "-tAc", sql}) + Expect(err).NotTo(HaveOccurred()) +} + +// queryPsqlOutput runs a SQL query and returns its trimmed stdout, failing the spec on error. +func queryPsqlOutput( + ctx SpecContext, + clientSet *kubernetes.Clientset, + cfg *rest.Config, + namespace, podName, sql string, +) string { + return queryPsqlOutputG(Default, ctx, clientSet, cfg, namespace, podName, sql) +} + +// queryPsqlOutputG is like queryPsqlOutput but reports failures to the provided +// Gomega, so it can be used safely inside an Eventually block. +func queryPsqlOutputG( + g Gomega, + ctx SpecContext, + clientSet *kubernetes.Clientset, + cfg *rest.Config, + namespace, podName, sql string, +) string { + out, _, err := command.ExecuteInContainer(ctx, + *clientSet, + cfg, + command.ContainerLocator{ + NamespaceName: namespace, + PodName: podName, + ContainerName: "postgres", + }, + nil, + []string{"psql", "-tAc", sql}) + g.Expect(err).NotTo(HaveOccurred()) + return strings.TrimSpace(out) +} + +// getSidecarLogs retrieves the parsed JSON logs of the plugin-pgbackrest sidecar. +func getSidecarLogs( + ctx SpecContext, + g Gomega, + clientSet *kubernetes.Clientset, + namespace, podName string, +) []map[string]any { + logs, err := internalLogs.GetPodContainerLogs(ctx, clientSet, namespace, podName, "plugin-pgbackrest", nil) + g.Expect(err).NotTo(HaveOccurred()) + return logs +} + +// hasSuccessfulArchiveBatch reports whether any completed batch archived at least +// one WAL file with no failures. +func hasSuccessfulArchiveBatch(completedBatches []map[string]any) bool { + for _, batch := range completedBatches { + successful, okS := batch["successfulArchives"].(float64) + failed, okF := batch["failedArchives"].(float64) + if okS && okF && successful >= 1 && failed == 0 { + return true + } + } + return false +} From 090af96ad28c292df9a8e8ec44b9b824795714f8 Mon Sep 17 00:00:00 2001 From: Vasiliy Fakunin <61789920+melancholictheory@users.noreply.github.com> Date: Mon, 3 Aug 2026 22:06:20 +0300 Subject: [PATCH 3/3] feat!: make stanza creation configurable via Archive createStanza Add a createStanza policy to the Archive configuration that controls when the pgBackRest stanza is created: - OnFirstArchive (default): create it on the first WAL archive if missing, so archiving works without a prior backup. - OnBackup: create it only when a backup runs (the previous behavior). - Disabled: never create it automatically; the stanza is managed out of band. The WAL archive path creates the stanza only under OnFirstArchive; the backup path creates it unless Disabled. This keeps archiving working out of the box while giving a backward-compatible opt-out, as discussed in #60. The WAL archive handler relies on CheckWalArchiveDestination returning a typed ErrStanzaMissing rather than inspecting the info catalog itself, which keeps the restore path's intent clear. The e2e coverage exercises both the default and the OnBackup policy. BREAKING CHANGE: WAL archiving now creates the pgBackRest stanza on the first WAL archive by default (createStanza=OnFirstArchive) instead of only during the first backup. Set createStanza=OnBackup to keep the previous behavior. Refs #60. Signed-off-by: Vasiliy Fakunin <61789920+melancholictheory@users.noreply.github.com> --- .../pgbackrest.cnpg.opera.com_archives.yaml | 12 ++ internal/cnpgi/common/wal.go | 51 ++++---- internal/cnpgi/instance/backup.go | 11 +- internal/cnpgi/restore/restore.go | 9 +- internal/pgbackrest/api/config.go | 46 +++++++ internal/pgbackrest/archiver/archiver.go | 26 ++-- manifest.yaml | 12 ++ .../e2e/internal/tests/walarchive/fixtures.go | 75 ++--------- .../tests/walarchive/wal_archive_on_backup.go | 116 ++++++++++++++++++ ...ive_no_backup.go => wal_archive_stanza.go} | 52 +------- 10 files changed, 258 insertions(+), 152 deletions(-) create mode 100644 test/e2e/internal/tests/walarchive/wal_archive_on_backup.go rename test/e2e/internal/tests/walarchive/{wal_archive_no_backup.go => wal_archive_stanza.go} (74%) diff --git a/config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml b/config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml index 112f29e..0e6b2df 100644 --- a/config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml +++ b/config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml @@ -53,6 +53,18 @@ spec: - lz4 - zst type: string + createStanza: + default: OnFirstArchive + description: |- + CreateStanza controls when the pgBackRest stanza is created. `OnFirstArchive` + (default) creates it on the first WAL archive if it does not exist yet, so + archiving works without a prior backup. `OnBackup` creates it only when a backup + runs. `Disabled` never creates it automatically. + enum: + - OnFirstArchive + - OnBackup + - Disabled + type: string data: description: |- The configuration to be used to backup the data files diff --git a/internal/cnpgi/common/wal.go b/internal/cnpgi/common/wal.go index c8c2c22..42d2d0f 100644 --- a/internal/cnpgi/common/wal.go +++ b/internal/cnpgi/common/wal.go @@ -137,37 +137,34 @@ func (w WALServiceImplementation) Archive( return &wal.WALArchiveResult{}, nil } - // Check that the destination repository is reachable and inspect the stanza status. - // This is the same "pgbackrest info" call that CheckWalArchiveDestination performs - // for the restore path, but here we keep the returned catalog so we can create the - // stanza below when it is missing. - destinationCatalog, err := pgbackrestCommand.GetBackupList(ctx, &archive.Spec.Configuration, configuration.Stanza, envArchive) - if err != nil { + // Check that the destination repository is reachable and its stanza exists. + err = arch.CheckWalArchiveDestination(ctx, &archive.Spec.Configuration, configuration.Stanza, envArchive) + switch { + case errors.Is(err, archiver.ErrStanzaMissing): + // On a fresh cluster, or after a major upgrade changes the repository path, the + // stanza does not exist yet and archive-push cannot succeed until it is created. + // When the Archive opts into it (createStanza=OnFirstArchive, the default), create + // it here instead of waiting for the first backup: this runs on the primary as soon + // as its sidecar is up, and PostgreSQL retries archiving on its own. stanza-create + // is idempotent, and we only reach it when the stanza is genuinely missing, so it + // does not contend with a running backup for the stanza lock. + if archive.Spec.Configuration.ShouldCreateStanzaOnArchive() { + backupCmd := pgbackrestBackup.NewBackupCommand(&archive.Spec.Configuration, nil, w.PGDataPath) + if stanzaErr := backupCmd.CreatePgbackrestStanza(ctx, configuration.Stanza, envArchive); stanzaErr != nil { + // Best-effort: log and continue. archive-push below reports the real + // outcome, and PostgreSQL retries the WAL if the stanza is still missing. + contextLogger.Warning("could not auto-create pgbackrest stanza; WAL archiving will retry", + "stanza", configuration.Stanza, "err", stanzaErr.Error()) + } else { + contextLogger.Info("created pgbackrest stanza so WAL archiving can start", + "stanza", configuration.Stanza) + } + } + case err != nil: log.Error(err, "while checking if pgbackrest repo can be used for archival") return nil, err } - // On a fresh cluster, or after a major upgrade changes the repository path, the - // stanza does not exist yet and archive-push cannot succeed until it is created. - // Stanza creation otherwise runs only during the first backup, so WAL archiving - // stays broken until a backup happens to run. Create it here instead: this path - // executes on the primary as soon as its sidecar is up, and PostgreSQL retries - // archiving on its own. stanza-create is idempotent, and we only reach it when the - // stanza is genuinely missing (no backup has created it yet), so it does not - // contend with a running backup for the stanza lock. - if destinationCatalog.StanzaMissing() { - backupCmd := pgbackrestBackup.NewBackupCommand(&archive.Spec.Configuration, nil, w.PGDataPath) - if stanzaErr := backupCmd.CreatePgbackrestStanza(ctx, configuration.Stanza, envArchive); stanzaErr != nil { - // Best-effort: log and continue. archive-push below reports the real - // outcome, and PostgreSQL retries the WAL if the stanza is still missing. - contextLogger.Warning("could not auto-create pgbackrest stanza; WAL archiving will retry", - "stanza", configuration.Stanza, "err", stanzaErr.Error()) - } else { - contextLogger.Info("created pgbackrest stanza so WAL archiving can start", - "stanza", configuration.Stanza) - } - } - options, err := arch.PgbackrestWalArchiveOptions(ctx, &archive.Spec.Configuration, configuration.Stanza) if err != nil { return nil, err diff --git a/internal/cnpgi/instance/backup.go b/internal/cnpgi/instance/backup.go index b06f34b..98431d6 100644 --- a/internal/cnpgi/instance/backup.go +++ b/internal/cnpgi/instance/backup.go @@ -118,10 +118,13 @@ func (b BackupServiceImplementation) Backup( return nil, err } - err = backupCmd.CreatePgbackrestStanza(ctx, configuration.Stanza, env) - if err != nil { - contextLogger.Error(err, "while initializing pgbackrest stanza") - return nil, err + // Create the stanza unless the Archive disables it (createStanza=Disabled), in which + // case it is expected to be managed out of band. + if archive.Spec.Configuration.ShouldCreateStanzaOnBackup() { + if err = backupCmd.CreatePgbackrestStanza(ctx, configuration.Stanza, env); err != nil { + contextLogger.Error(err, "while initializing pgbackrest stanza") + return nil, err + } } backupName := fmt.Sprintf("backup-%v", pgTime.ToCompactISO8601(time.Now())) diff --git a/internal/cnpgi/restore/restore.go b/internal/cnpgi/restore/restore.go index d3a16b7..15010c3 100644 --- a/internal/cnpgi/restore/restore.go +++ b/internal/cnpgi/restore/restore.go @@ -19,6 +19,7 @@ package restore import ( "context" + "errors" "fmt" "os" "path" @@ -247,7 +248,13 @@ func (impl *JobHookImpl) checkBackupDestination( } if utils.IsEmptyWalArchiveCheckEnabled(&cluster.ObjectMeta) { - return walArchiver.CheckWalArchiveDestination(ctx, pgbackrestConfiguration, stanza, env) + err := walArchiver.CheckWalArchiveDestination(ctx, pgbackrestConfiguration, stanza, env) + if errors.Is(err, pgbackrestArchiver.ErrStanzaMissing) { + // A reachable but not-yet-created stanza is a legitimate empty destination for + // a cluster about to start archiving; treat it as OK. + return nil + } + return err } return nil diff --git a/internal/pgbackrest/api/config.go b/internal/pgbackrest/api/config.go index 38b87d5..338e791 100644 --- a/internal/pgbackrest/api/config.go +++ b/internal/pgbackrest/api/config.go @@ -320,6 +320,23 @@ type PgbackrestRepository struct { Retention *PgbackrestRetention `json:"retention,omitempty"` } +// StanzaCreatePolicy controls when the pgBackRest stanza is created. +// +kubebuilder:validation:Enum=OnFirstArchive;OnBackup;Disabled +type StanzaCreatePolicy string + +const ( + // StanzaCreateOnFirstArchive creates the stanza the first time a WAL is archived, + // if it does not exist yet, so archiving works without waiting for a backup. + StanzaCreateOnFirstArchive StanzaCreatePolicy = "OnFirstArchive" + + // StanzaCreateOnBackup creates the stanza only when a backup runs (the legacy behavior). + StanzaCreateOnBackup StanzaCreatePolicy = "OnBackup" + + // StanzaCreateDisabled never creates the stanza automatically; it must be created + // out of band. + StanzaCreateDisabled StanzaCreatePolicy = "Disabled" +) + // PgbackrestConfiguration is the configuration of all pgBackRest operations type PgbackrestConfiguration struct { Repositories []PgbackrestRepository `json:"repositories"` @@ -351,6 +368,35 @@ type PgbackrestConfiguration struct { // this parameter is omitted // +optional Stanza string `json:"stanza,omitempty"` + + // CreateStanza controls when the pgBackRest stanza is created. `OnFirstArchive` + // (default) creates it on the first WAL archive if it does not exist yet, so + // archiving works without a prior backup. `OnBackup` creates it only when a backup + // runs. `Disabled` never creates it automatically. + // +kubebuilder:validation:Enum=OnFirstArchive;OnBackup;Disabled + // +kubebuilder:default=OnFirstArchive + // +optional + CreateStanza StanzaCreatePolicy `json:"createStanza,omitempty"` +} + +// GetCreateStanzaPolicy returns the configured stanza creation policy, defaulting to +// OnFirstArchive when unset. +func (c *PgbackrestConfiguration) GetCreateStanzaPolicy() StanzaCreatePolicy { + if c.CreateStanza == "" { + return StanzaCreateOnFirstArchive + } + return c.CreateStanza +} + +// ShouldCreateStanzaOnArchive reports whether the WAL archive path should create the +// stanza when it is missing. +func (c *PgbackrestConfiguration) ShouldCreateStanzaOnArchive() bool { + return c.GetCreateStanzaPolicy() == StanzaCreateOnFirstArchive +} + +// ShouldCreateStanzaOnBackup reports whether the backup path should create the stanza. +func (c *PgbackrestConfiguration) ShouldCreateStanzaOnBackup() bool { + return c.GetCreateStanzaPolicy() != StanzaCreateDisabled } // ArePopulated checks if the passed set of credentials contains diff --git a/internal/pgbackrest/archiver/archiver.go b/internal/pgbackrest/archiver/archiver.go index efb3df4..aaac3cc 100644 --- a/internal/pgbackrest/archiver/archiver.go +++ b/internal/pgbackrest/archiver/archiver.go @@ -20,6 +20,7 @@ package archiver import ( "context" + "errors" "fmt" "time" @@ -128,19 +129,30 @@ func (archiver *WALArchiver) ArchiveList( return result } -// CheckWalArchiveDestination checks if the destination archive is ready to perform -// archiving, i.e. if proper stanzas exist. +// ErrStanzaMissing is returned by CheckWalArchiveDestination when the destination +// repository is reachable but the pgBackRest stanza has not been created yet. Callers +// can react to it (create the stanza, or treat the destination as empty) instead of +// inspecting the "pgbackrest info" output themselves. +var ErrStanzaMissing = errors.New("pgbackrest stanza has not been created") + +// CheckWalArchiveDestination checks that the destination archive is reachable and its +// stanza exists. It runs "pgbackrest info" (which, unlike stanza-create, does not take +// the stanza lock) and returns ErrStanzaMissing when the repository is reachable but +// the stanza has not been created yet. func (archiver *WALArchiver) CheckWalArchiveDestination( ctx context.Context, configuration *pgbackrestApi.PgbackrestConfiguration, stanza string, env []string, ) error { - // Probably the easiest way to check if stanza exists is to run "pgbackrest info". - // It's possible to use stanza-create instead but it requires the lock file - // which makes it unusable during backups. - _, err := pgbackrestCommand.GetBackupList(ctx, configuration, stanza, env) - return err + destinationCatalog, err := pgbackrestCommand.GetBackupList(ctx, configuration, stanza, env) + if err != nil { + return err + } + if destinationCatalog.StanzaMissing() { + return ErrStanzaMissing + } + return nil } // PgbackrestCheckWalArchiveOptions create the options needed for the `pgbackrest check` diff --git a/manifest.yaml b/manifest.yaml index 43d03a9..cc0bd4d 100644 --- a/manifest.yaml +++ b/manifest.yaml @@ -52,6 +52,18 @@ spec: - lz4 - zst type: string + createStanza: + default: OnFirstArchive + description: |- + CreateStanza controls when the pgBackRest stanza is created. `OnFirstArchive` + (default) creates it on the first WAL archive if it does not exist yet, so + archiving works without a prior backup. `OnBackup` creates it only when a backup + runs. `Disabled` never creates it automatically. + enum: + - OnFirstArchive + - OnBackup + - Disabled + type: string data: description: |- The configuration to be used to backup the data files diff --git a/test/e2e/internal/tests/walarchive/fixtures.go b/test/e2e/internal/tests/walarchive/fixtures.go index e4971d1..2fd1fe1 100644 --- a/test/e2e/internal/tests/walarchive/fixtures.go +++ b/test/e2e/internal/tests/walarchive/fixtures.go @@ -28,22 +28,20 @@ import ( const ( minio = "minio" // size is the size of the PVCs for the object store and the cluster instances. - size = "1Gi" - pluginName = "pgbackrest.cnpg.opera.com" - srcClusterName = "wal-archive-source" - archiveName = "wal-archive" - backupName = "wal-archive-backup" - restoreClusterName = "wal-archive-restore" + size = "1Gi" + pluginName = "pgbackrest.cnpg.opera.com" + srcClusterName = "wal-archive-source" + archiveName = "wal-archive" + backupName = "wal-archive-backup" ) -// walArchiveTestResources contains the resources needed to test WAL archiving -// without a prior backup, plus backup and restore to prove those paths still work. +// walArchiveTestResources contains the resources needed to test WAL archiving, +// with a backup to prove that path still works. type walArchiveTestResources struct { ObjectStoreResources *objectstore.Resources Archive *pluginPgbackrestV1.Archive Cluster *cloudnativepgv1.Cluster Backup *cloudnativepgv1.Backup - RestoreCluster *cloudnativepgv1.Cluster } // createWalArchiveTestResources builds all resources for the WAL archiving test. @@ -52,10 +50,9 @@ func createWalArchiveTestResources(namespace string) walArchiveTestResources { ObjectStoreResources: objectstore.NewMinioObjectStoreResources(namespace, minio), // maxParallel=1 so every WAL is archived in its own batch, which keeps the // assertions on batch completions unambiguous. - Archive: objectstore.NewMinioArchive(namespace, archiveName, minio, 1), - Cluster: newClusterWithPlugin(namespace, srcClusterName), - Backup: newPluginBackup(namespace, backupName, srcClusterName), - RestoreCluster: newRestoreCluster(namespace, restoreClusterName), + Archive: objectstore.NewMinioArchive(namespace, archiveName, minio, 1), + Cluster: newClusterWithPlugin(namespace, srcClusterName), + Backup: newPluginBackup(namespace, backupName, srcClusterName), } } @@ -118,55 +115,3 @@ func newPluginBackup(namespace, name, clusterName string) *cloudnativepgv1.Backu }, } } - -// newRestoreCluster creates a cluster that bootstraps by recovering from the -// source cluster's archive, while also archiving its own WALs to the same store. -func newRestoreCluster(namespace, name string) *cloudnativepgv1.Cluster { - return &cloudnativepgv1.Cluster{ - TypeMeta: metav1.TypeMeta{ - Kind: "Cluster", - APIVersion: "postgresql.cnpg.io/v1", - }, - ObjectMeta: metav1.ObjectMeta{ - Name: name, - Namespace: namespace, - }, - Spec: cloudnativepgv1.ClusterSpec{ - Instances: 2, - ImagePullPolicy: corev1.PullAlways, - Bootstrap: &cloudnativepgv1.BootstrapConfiguration{ - Recovery: &cloudnativepgv1.BootstrapRecovery{ - Source: "source", - }, - }, - Plugins: []cloudnativepgv1.PluginConfiguration{ - { - Name: pluginName, - Parameters: map[string]string{ - "pgbackrestObjectName": archiveName, - }, - }, - }, - PostgresConfiguration: cloudnativepgv1.PostgresConfiguration{ - Parameters: map[string]string{ - "log_min_messages": "DEBUG4", - }, - }, - ExternalClusters: []cloudnativepgv1.ExternalCluster{ - { - Name: "source", - PluginConfiguration: &cloudnativepgv1.PluginConfiguration{ - Name: pluginName, - Parameters: map[string]string{ - "pgbackrestObjectName": archiveName, - "stanza": srcClusterName, - }, - }, - }, - }, - StorageConfiguration: cloudnativepgv1.StorageConfiguration{ - Size: size, - }, - }, - } -} diff --git a/test/e2e/internal/tests/walarchive/wal_archive_on_backup.go b/test/e2e/internal/tests/walarchive/wal_archive_on_backup.go new file mode 100644 index 0000000..5dc47fe --- /dev/null +++ b/test/e2e/internal/tests/walarchive/wal_archive_on_backup.go @@ -0,0 +1,116 @@ +/* +Copyright 2025, Opera Norway AS + +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. +*/ + +package walarchive + +import ( + "fmt" + "time" + + corev1 "k8s.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + pgbackrestApi "github.com/operasoftware/cnpg-plugin-pgbackrest/internal/pgbackrest/api" + internalClient "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/client" + internalLogs "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/logs" + nmsp "github.com/operasoftware/cnpg-plugin-pgbackrest/test/e2e/internal/namespace" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("WAL archiving with createStanza set to OnBackup", func() { + var namespace *corev1.Namespace + var cl client.Client + + BeforeEach(func(ctx SpecContext) { + var err error + cl, _, err = internalClient.NewClient() + Expect(err).NotTo(HaveOccurred()) + namespace, err = nmsp.CreateUniqueNamespace(ctx, cl, "wal-archive-on-backup") + Expect(err).NotTo(HaveOccurred()) + }) + + AfterEach(func(ctx SpecContext) { + Expect(cl.Delete(ctx, namespace)).To(Succeed()) + }) + + It("does not create the stanza on WAL archive, so archiving fails until a backup creates it", + func(ctx SpecContext) { + testResources := createWalArchiveTestResources(namespace.Name) + // Opt out of lazy stanza creation. With OnBackup the WAL archive path must not + // create the stanza, so archiving stays broken until a backup runs. + testResources.Archive.Spec.Configuration.CreateStanza = pgbackrestApi.StanzaCreateOnBackup + + By("starting the object store deployment") + Expect(testResources.ObjectStoreResources.Create(ctx, cl)).To(Succeed()) + + By("creating the Archive with createStanza=OnBackup") + Expect(cl.Create(ctx, testResources.Archive)).To(Succeed()) + + By("creating a CloudNativePG cluster that only enables WAL archiving (no backup)") + cluster := testResources.Cluster + Expect(cl.Create(ctx, cluster)).To(Succeed()) + + By("waiting for the cluster to be ready") + waitForClusterReady(ctx, cl, cluster) + + clientSet, cfg, err := internalClient.NewClientSet() + Expect(err).NotTo(HaveOccurred()) + + primaryPod := fmt.Sprintf("%s-1", cluster.Name) + + By("adding data and forcing WAL switches without any backup") + execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "CREATE TABLE wal_test (id int, data text);") + for i := 0; i < 5; i++ { + execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, + fmt.Sprintf("INSERT INTO wal_test VALUES (%d, 'data-%d'); SELECT pg_switch_wal();", i, i)) + time.Sleep(500 * time.Millisecond) + } + + By("verifying WAL archiving fails because the stanza was not created on archive") + Eventually(func(g Gomega) { + failed := queryPsqlOutputG(g, ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "SELECT failed_count FROM pg_stat_archiver;") + g.Expect(failed).NotTo(Equal("0"), + "archive-push should be failing while the stanza does not exist") + + lastArchived := queryPsqlOutputG(g, ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "SELECT COALESCE(last_archived_wal, '') FROM pg_stat_archiver;") + g.Expect(lastArchived).To(BeEmpty(), "no WAL should be archived before a backup under OnBackup") + + logs := getSidecarLogs(ctx, g, clientSet, cluster.Namespace, primaryPod) + g.Expect(internalLogs.FindLogEntriesByMessage(logs, lazyStanzaLogMessage)).To(BeEmpty(), + "the sidecar must not create the stanza on WAL archive when createStanza=OnBackup") + }).WithTimeout(3 * time.Minute).WithPolling(10 * time.Second).Should(Succeed()) + + By("taking a backup, which creates the stanza under OnBackup") + backup := testResources.Backup + Expect(cl.Create(ctx, backup)).To(Succeed()) + waitForBackupCompleted(ctx, cl, backup) + + By("verifying WAL archiving recovers once the backup created the stanza") + execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "INSERT INTO wal_test VALUES (99, 'after-backup'); SELECT pg_switch_wal();") + Eventually(func(g Gomega) { + lastArchived := queryPsqlOutputG(g, ctx, clientSet, cfg, cluster.Namespace, primaryPod, + "SELECT COALESCE(last_archived_wal, '') FROM pg_stat_archiver;") + g.Expect(lastArchived).NotTo(BeEmpty(), + "WAL archiving should succeed after the backup created the stanza") + }).WithTimeout(2 * time.Minute).WithPolling(5 * time.Second).Should(Succeed()) + }) +}) diff --git a/test/e2e/internal/tests/walarchive/wal_archive_no_backup.go b/test/e2e/internal/tests/walarchive/wal_archive_stanza.go similarity index 74% rename from test/e2e/internal/tests/walarchive/wal_archive_no_backup.go rename to test/e2e/internal/tests/walarchive/wal_archive_stanza.go index 9b06da0..9a7aa2b 100644 --- a/test/e2e/internal/tests/walarchive/wal_archive_no_backup.go +++ b/test/e2e/internal/tests/walarchive/wal_archive_stanza.go @@ -58,7 +58,7 @@ var _ = Describe("WAL archiving without a prior backup", func() { Expect(cl.Delete(ctx, namespace)).To(Succeed()) }) - It("creates the stanza lazily, archives WALs before any backup, and still supports backup and restore", + It("creates the stanza lazily, archives WALs before any backup, and still supports a backup", func(ctx SpecContext) { testResources := createWalArchiveTestResources(namespace.Name) @@ -111,44 +111,10 @@ var _ = Describe("WAL archiving without a prior backup", func() { "at least one WAL archive batch should have completed with a successful, error-free archive") }).WithTimeout(4 * time.Minute).WithPolling(10 * time.Second).Should(Succeed()) - By("taking a backup now to prove the backup path still works after lazy stanza creation") + By("taking a backup to prove the backup path still works after lazy stanza creation") backup := testResources.Backup Expect(cl.Create(ctx, backup)).To(Succeed()) waitForBackupCompleted(ctx, cl, backup) - - By("adding data after the backup so restore has WALs to replay") - execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, - "INSERT INTO wal_test VALUES (99, 'after-backup');") - // The committed row lives in the current WAL segment. Capture it, force a - // switch so it becomes archivable, then wait until the archiver has actually - // uploaded it. Otherwise the restored cluster may finish recovery before the - // segment reaches the object store and the post-backup row would be lost. - postBackupWAL := queryPsqlOutput(ctx, clientSet, cfg, cluster.Namespace, primaryPod, - "SELECT pg_walfile_name(pg_current_wal_lsn());") - execPsql(ctx, clientSet, cfg, cluster.Namespace, primaryPod, "SELECT pg_switch_wal();") - - By("waiting for the post-backup WAL segment to be archived") - Eventually(func(g Gomega) { - lastArchived := queryPsqlOutputG(g, ctx, clientSet, cfg, cluster.Namespace, primaryPod, - "SELECT COALESCE(last_archived_wal, '') FROM pg_stat_archiver;") - g.Expect(lastArchived).NotTo(BeEmpty(), "no WAL has been archived yet") - g.Expect(lastArchived >= postBackupWAL).To(BeTrue(), - fmt.Sprintf("archived WAL %q has not yet reached the post-backup segment %q", - lastArchived, postBackupWAL)) - }).WithTimeout(2 * time.Minute).WithPolling(5 * time.Second).Should(Succeed()) - - By("restoring into a new cluster from the archive") - restore := testResources.RestoreCluster - Expect(cl.Create(ctx, restore)).To(Succeed()) - - By("waiting for the restored cluster to be ready") - waitForClusterReady(ctx, cl, restore) - - By("verifying the restored data is present") - output := queryPsqlOutput(ctx, clientSet, cfg, restore.Namespace, - fmt.Sprintf("%s-1", restore.Name), "SELECT count(*) FROM wal_test;") - Expect(output).To(Equal("6"), - "restored cluster should contain the 5 pre-backup rows plus the 1 post-backup row") }) }) @@ -192,18 +158,8 @@ func execPsql( Expect(err).NotTo(HaveOccurred()) } -// queryPsqlOutput runs a SQL query and returns its trimmed stdout, failing the spec on error. -func queryPsqlOutput( - ctx SpecContext, - clientSet *kubernetes.Clientset, - cfg *rest.Config, - namespace, podName, sql string, -) string { - return queryPsqlOutputG(Default, ctx, clientSet, cfg, namespace, podName, sql) -} - -// queryPsqlOutputG is like queryPsqlOutput but reports failures to the provided -// Gomega, so it can be used safely inside an Eventually block. +// queryPsqlOutputG runs a SQL query and returns its trimmed stdout, reporting failures +// to the provided Gomega so it can be used safely inside an Eventually block. func queryPsqlOutputG( g Gomega, ctx SpecContext,