diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index 245d5a687..4dcf4da74 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/internal/client/sendwal/receiver.go @@ -157,6 +157,13 @@ func (s *Process) Start(ctx context.Context) error { "systemID", identifyData.SystemID, ) + // The slot must exist, with its WAL reserved, before we read its restart + // LSN as the start point below: otherwise the WAL between here and the + // slot's eventual creation could be recycled before we ever read it. + if err := s.ensureReplicationSlotExists(ctx, conn); err != nil { + return err + } + // Negotiate the starting point with the server point, err := s.getReplicationStartPoint(ctx, conn, identifyData, walSegmentSize) if err != nil { @@ -173,10 +180,6 @@ func (s *Process) Start(ctx context.Context) error { contextLogger.Debug("Some timeline history files could not be processed", "innerErr", histErr.Error()) } - if err := s.ensureReplicationSlotExists(ctx, conn); err != nil { - return err - } - return s.startReplication(ctx, conn, point, walSegmentSize) } @@ -205,9 +208,6 @@ func (s *Process) getReplicationStartPointFromClient( // If nor the Klio server nor the replication slot are set, // we use the XLOG flush position, taking care of // starting streaming from the beginning of the WAL file. - // - // This usually happens when we are running against this - // PostgreSQL instance for the first time. contextLogger.Debug( "Current flush LSN", "xlogFlushPos", xlogFlushPos, @@ -291,6 +291,11 @@ func getStartWALLSN(xlogFlushPos pglogrepl.LSN, segmentSize uint64) pglogrepl.LS return pglogrepl.LSN(uint64(xlogFlushPos) & ^(segmentSize - 1)) } +// ensureReplicationSlotExists creates the Klio physical replication slot if +// it does not exist yet, using RESERVE_WAL so its restart LSN, and the WAL +// from it, are reserved immediately at creation instead of at the first +// replication connection. Without it, WAL between slot creation and that +// first connection is free to be recycled before Klio ever streams it. func (s *Process) ensureReplicationSlotExists( ctx context.Context, conn *pgconn.PgConn, @@ -308,16 +313,11 @@ func (s *Process) ensureReplicationSlotExists( return nil } - replicationSlotResult, err := pglogrepl.CreateReplicationSlot( - ctx, - conn, - s.config.Source.Slot, - "", // output plugin name: this is meaningful only for logical replication - pglogrepl.CreateReplicationSlotOptions{ - Temporary: false, - Mode: pglogrepl.PhysicalReplication, - }, - ) + // pglogrepl.CreateReplicationSlotOptions has no RESERVE_WAL field, so the + // command is built and parsed manually here. + sql := fmt.Sprintf("CREATE_REPLICATION_SLOT %s PHYSICAL RESERVE_WAL", s.config.Source.Slot) + + replicationSlotResult, err := pglogrepl.ParseCreateReplicationSlot(conn.Exec(ctx, sql)) if err != nil { return fmt.Errorf("while creating temporary replication slot: %w", err) } diff --git a/documentation/web/docs/developer/running-e2e-tests.md b/documentation/web/docs/developer/running-e2e-tests.md index b63f1565a..9c7edc40f 100644 --- a/documentation/web/docs/developer/running-e2e-tests.md +++ b/documentation/web/docs/developer/running-e2e-tests.md @@ -106,6 +106,10 @@ The E2E tests are located in `operator/test/e2e/` and include: - `BackupFromPrimary`: backup from a single-instance cluster - `BackupFromStandby`: backup from a standby in a multi-instance cluster +- **`backup_from_replica_cluster_test.go`** - Immediate backup from a + freshly-created replica cluster: verifies the backup completes even + when the WAL streamer and `pg_backup_start` disagree on the starting + WAL (`BackupFromReplicaCluster`) - **`maintenance_test.go`** - Server-side post-backup maintenance on a tier1-only deployment: verifies the backup queue consumer applies tier1 WAL retention after a backup even when tier2 is not configured diff --git a/operator/test/e2e/backup_from_replica_cluster_test.go b/operator/test/e2e/backup_from_replica_cluster_test.go new file mode 100644 index 000000000..a4f415be7 --- /dev/null +++ b/operator/test/e2e/backup_from_replica_cluster_test.go @@ -0,0 +1,275 @@ +/* +Copyright © contributors to CloudNativePG, established as +CloudNativePG a Series of LF Projects, LLC. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package e2e + +import ( + "context" + "testing" + "time" + + certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1" + cnpgv1 "github.com/cloudnative-pg/api/pkg/api/v1" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/e2e-framework/klient/k8s/resources" + "sigs.k8s.io/e2e-framework/klient/wait" + "sigs.k8s.io/e2e-framework/pkg/envconf" + "sigs.k8s.io/e2e-framework/pkg/types" + + kliov1alpha1 "github.com/cloudnative-pg/klio/operator/api/v1alpha1" + "github.com/cloudnative-pg/klio/operator/internal/klioconfig" + machineryConditions "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/conditions" + "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/postgres" + "github.com/cloudnative-pg/klio/operator/test/utils/templates/certificates" + "github.com/cloudnative-pg/klio/operator/test/utils/templates/cnpg" + "github.com/cloudnative-pg/klio/operator/test/utils/templates/klio" + "github.com/cloudnative-pg/klio/operator/test/utils/templates/secrets" +) + +// ReplicaClusterBackupFeature verifies that an immediate backup taken from a +// freshly-created replica cluster completes. +type ReplicaClusterBackupFeature struct { + scenario *commonBackupRestoreScenario + + // sourceBackup is the base backup of the source cluster the replica + // bootstraps from. + sourceBackup *cnpgv1.Backup + // replicaCluster is the replica cluster archiving to its own tier1. + replicaCluster *cnpgv1.Cluster + // replicaUserCertificate authenticates the replica cluster against the + // Klio server under its own cluster name. + replicaUserCertificate *certmanagerv1.Certificate + // replicaPluginConfiguration wires the replica cluster to its own tier1. + replicaPluginConfiguration *kliov1alpha1.PluginConfiguration + // replicaBackup is the immediate backup taken from the replica cluster. + replicaBackup *cnpgv1.Backup + + sourceBackupTimeout time.Duration + recoveryTimeout time.Duration + // replicaBackupTimeout bounds the wait for the replica backup so a + // never-arriving WAL fails the test instead of hanging. + replicaBackupTimeout time.Duration + checkInterval time.Duration +} + +// BackupFromReplicaCluster builds the "immediate backup from a replica cluster" +// feature: it backs up a source cluster, bootstraps a replica cluster that +// streams from it and archives to its own tier1, then takes an immediate backup +// of the replica cluster and asserts it completes. +func BackupFromReplicaCluster(namespace string) *ReplicaClusterBackupFeature { + const ( + sourceClusterName = "test-cluster-source" + replicaClusterName = "test-cluster-replica" + sourceExternalName = "source-cluster" + ) + + namespaceObj := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: namespace}, + } + + issuer := certificates.GetSelfSignedIssuerObject("selfsigned-issuer", namespace) + certificate := certificates.GetCertificateObject("test", namespace, []string{klioServerName}, issuer) + + caCertificate := certificates.GetCACertificateObject("test-ca", namespace, issuer) + caIssuer := certificates.GetCAIssuerObject("test-ca-issuer", namespace, caCertificate.Spec.SecretName) + + sourceCluster := cnpg.GetCnpgClusterObject(sourceClusterName, namespace, 1, + "klio-plugin-configuration", + cnpg.ClusterTemplateOptions{StorageClass: testCfg.StorageClass}) + // Switch WAL frequently so the segments required by a backup are archived + // promptly, and keep enough WAL around for the replica streamer to resume + // from an older position. + sourceCluster.Spec.PostgresConfiguration.Parameters = map[string]string{ + "archive_timeout": "30s", + "wal_keep_size": "512MB", + } + + sourceUserCertificate := certificates.GetUserCertificateObject( + "klio-user", namespace, "klio-user@"+sourceClusterName, caIssuer) + sourcePluginConfiguration := klio.GetPluginConfigurationObject( + "klio-plugin-configuration", + namespace, + klio.PluginConfigurationTemplateOptions{ + ServerCertificate: certificate, + ClientCertificate: sourceUserCertificate, + ClusterName: sourceClusterName, + }, + ) + // The replica reads the source's tier1 (same server, same cluster name) to + // bootstrap and to stream from it. + sourceExternalPluginConfiguration := sourcePluginConfiguration.DeepCopy() + sourceExternalPluginConfiguration.Name = "klio-plugin-configuration-source" + + ageSecrets := secrets.GetKlioAgeEncryptionSecrets("encryption", namespace, "testencryptionpassword123") + klioServer := klio.GetServerObject( + klioServerName, + namespace, + klio.ServerTemplateOptions{ + Image: testCfg.ServerImage, + StorageClass: testCfg.StorageClass, + TLSSecretName: certificate.Spec.SecretName, + ClientCASecretName: caCertificate.Spec.SecretName, + Encryption: klio.EncryptionOptions{ + EncryptionKeySecretName: ageSecrets.EncryptionKeySecret.Name, + EncryptionKeyFileName: "encryption-key.age", + IdentitySecretName: ageSecrets.IdentitySecret.Name, + IdentityFileName: "identity.txt", + }, + }, + ) + + sourceBackup := cnpg.GetCnpgBackupObject("test-backup-source", namespace, + cnpgv1.BackupTargetPrimary, sourceCluster) + + // The replica cluster archives to its own tier1 (its own cluster name and + // client certificate). + replicaUserCertificate := certificates.GetUserCertificateObject( + "klio-user-replica", namespace, "klio-user@"+replicaClusterName, caIssuer) + replicaPluginConfiguration := klio.GetPluginConfigurationObject( + "klio-plugin-configuration-replica", + namespace, + klio.PluginConfigurationTemplateOptions{ + ServerCertificate: certificate, + ClientCertificate: replicaUserCertificate, + ClusterName: replicaClusterName, + }, + ) + + replicaCluster := sourceCluster.DeepCopy() + replicaCluster.Name = replicaClusterName + replicaCluster.Spec.Plugins[0].Parameters[klioconfig.PluginConfigurationRefParam] = replicaPluginConfiguration.Name + replicaCluster.Spec.Bootstrap = &cnpgv1.BootstrapConfiguration{ + Recovery: &cnpgv1.BootstrapRecovery{ + Source: sourceExternalName, + }, + } + replicaCluster.Spec.ReplicaCluster = &cnpgv1.ReplicaClusterConfiguration{ + Source: sourceExternalName, + Enabled: new(true), + } + replicaCluster.Spec.ExternalClusters = []cnpgv1.ExternalCluster{{ + Name: sourceExternalName, + PluginConfiguration: &cnpgv1.PluginConfiguration{ + Name: "klio.cnpg.io", + Enabled: new(true), + Parameters: map[string]string{ + klioconfig.PluginConfigurationRefParam: sourceExternalPluginConfiguration.Name, + }, + }, + }} + + replicaBackup := cnpg.GetCnpgBackupObject("test-backup-replica", namespace, + cnpgv1.BackupTargetPrimary, replicaCluster) + + scenario := &commonBackupRestoreScenario{ + namespace: namespaceObj, + cnpgCluster: sourceCluster, + userCertificate: sourceUserCertificate, + encryptionSecret: ageSecrets.EncryptionKeySecret, + identitySecret: ageSecrets.IdentitySecret, + issuer: issuer, + caIssuer: caIssuer, + caCertificate: caCertificate, + certificate: certificate, + klioServer: klioServer, + klioPluginConfigurationSource: sourcePluginConfiguration, + klioPluginConfigurationRecovery: sourceExternalPluginConfiguration, + name: "BackupFromReplicaCluster", + } + + return &ReplicaClusterBackupFeature{ + scenario: scenario, + sourceBackup: sourceBackup, + replicaCluster: replicaCluster, + replicaUserCertificate: replicaUserCertificate, + replicaPluginConfiguration: replicaPluginConfiguration, + replicaBackup: replicaBackup, + sourceBackupTimeout: 2 * time.Minute, + recoveryTimeout: 5 * time.Minute, + replicaBackupTimeout: 3 * time.Minute, + checkInterval: 10 * time.Second, + } +} + +// Name returns the feature name. +func (f *ReplicaClusterBackupFeature) Name() string { + return f.scenario.name +} + +// Setup creates the source cluster, the Klio server and the source-side plugin +// configurations, and waits for them to be ready. +func (f *ReplicaClusterBackupFeature) Setup() types.StepFunc { + return f.scenario.Setup +} + +// Run backs up the source cluster, bootstraps the replica cluster, then takes an +// immediate backup of the replica cluster and asserts it completes. +func (f *ReplicaClusterBackupFeature) Run() types.StepFunc { + return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + t.Helper() + t.Log("Running backup-from-replica-cluster feature test") + r, err := resources.New(cfg.Client().RESTConfig()) + require.NoError(t, err, "failed to create resources client") + + // Take a base backup of the source cluster so the replica can bootstrap. + require.NoError(t, r.Create(ctx, f.sourceBackup), "failed to create source backup") + require.NoError(t, wait.For( + machineryConditions.BackupIsCompleted(r, f.sourceBackup), + wait.WithTimeout(f.sourceBackupTimeout), + wait.WithInterval(f.checkInterval), + ), "source backup not completed") + + // Advance the source WAL (without a checkpoint) so the replica, once + // bootstrapped, has the final WAL and isn't stuck waiting for it. + _, err = postgres.ExecPostgresQuery(ctx, r, &f.scenario.sourcePrimaryPod, "postgres", + "CREATE TABLE numbers AS SELECT generate_series(1, 1000) AS x; "+ + "SELECT pg_switch_wal(); SELECT pg_switch_wal();") + require.NoError(t, err, "failed to advance source WAL") + + // Create the replica-side archiving resources and the replica cluster. + require.NoError(t, r.Create(ctx, f.replicaUserCertificate), + "failed to create replica user certificate") + require.NoError(t, r.Create(ctx, f.replicaPluginConfiguration), + "failed to create replica plugin configuration") + require.NoError(t, r.Create(ctx, f.replicaCluster), "failed to create replica cluster") + require.NoError(t, wait.For( + machineryConditions.ClusterIsReady(r, f.replicaCluster), + wait.WithTimeout(f.recoveryTimeout), + wait.WithInterval(f.checkInterval), + ), "replica cluster not ready") + + // The immediate backup of the freshly-created replica cluster must + // complete. + require.NoError(t, r.Create(ctx, f.replicaBackup), "failed to create replica backup") + require.NoError(t, wait.For( + machineryConditions.BackupIsCompleted(r, f.replicaBackup), + wait.WithTimeout(f.replicaBackupTimeout), + wait.WithInterval(f.checkInterval), + ), "replica cluster backup not completed") + + return ctx + } +} + +// Teardown removes the resources created for the feature. +func (f *ReplicaClusterBackupFeature) Teardown() types.StepFunc { + return f.scenario.Teardown +} diff --git a/operator/test/e2e/main_test.go b/operator/test/e2e/main_test.go index b05680ea5..b5a80ff22 100644 --- a/operator/test/e2e/main_test.go +++ b/operator/test/e2e/main_test.go @@ -51,6 +51,7 @@ func TestMain(m *testing.M) { runner.RegisterFeature(BackupFromPrimary(envconf.RandomName("backup-from-primary", 32))) runner.RegisterFeature(BackupFromStandby(envconf.RandomName("backup-from-standby", 32))) + runner.RegisterFeature(BackupFromReplicaCluster(envconf.RandomName("backup-from-replica-cluster", 32))) runner.RegisterFeature(Tier1ServerSideMaintenance(envconf.RandomName("tier1-maintenance", 32))) runner.RegisterFeature(RecoverClusterFromBackupID(envconf.RandomName("recovery-from-backup-id", 32))) runner.RegisterFeature(RecoverClusterFromLatestBackup(envconf.RandomName("recovery-from-latest-backup", 32)))