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 +}