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 ef82226..42d2d0f 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,9 +137,30 @@ func (w WALServiceImplementation) Archive( return &wal.WALArchiveResult{}, nil } - // Check if we're ok to archive in the desired destination + // Check that the destination repository is reachable and its stanza exists. err = arch.CheckWalArchiveDestination(ctx, &archive.Spec.Configuration, configuration.Stanza, envArchive) - if err != nil { + 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 } 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/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) 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/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..2fd1fe1 --- /dev/null +++ b/test/e2e/internal/tests/walarchive/fixtures.go @@ -0,0 +1,117 @@ +/* +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" +) + +// 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 +} + +// 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), + } +} + +// 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, + }, + }, + } +} 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_stanza.go b/test/e2e/internal/tests/walarchive/wal_archive_stanza.go new file mode 100644 index 0000000..9a7aa2b --- /dev/null +++ b/test/e2e/internal/tests/walarchive/wal_archive_stanza.go @@ -0,0 +1,207 @@ +/* +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 a backup", + 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 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) + }) +}) + +// 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()) +} + +// 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, + 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 +}