diff --git a/core/internal/client/sendwal/receiver.go b/core/internal/client/sendwal/receiver.go index 245d5a68..5a6fe9c5 100644 --- a/core/internal/client/sendwal/receiver.go +++ b/core/internal/client/sendwal/receiver.go @@ -31,6 +31,7 @@ import ( "github.com/cloudnative-pg/machinery/pkg/log" "github.com/cloudnative-pg/machinery/pkg/types" "github.com/jackc/pglogrepl" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgproto3" "go.opentelemetry.io/otel/attribute" @@ -202,19 +203,59 @@ func (s *Process) getReplicationStartPointFromClient( return slotResult.RestartLSN, nil } - // 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. + // Neither the Klio server nor the replication slot have a resume point. + // This usually happens when we are running against this PostgreSQL instance + // for the first time. // - // This usually happens when we are running against this - // PostgreSQL instance for the first time. + // We start from the redo point of the latest checkpoint (on a standby, the + // latest restartpoint) rather than from the current flush position. That + // redo point is the earliest LSN a later pg_backup_start on this instance + // can report as a backup start, so the WAL a backup needs is always within + // what we archive to tier1. This matters on a standby, where pg_backup_start + // reports the last restartpoint, which lags the flush position: streaming + // from the flush position would leave the segments in between permanently + // out of tier1. + // Failing rather than falling back to the flush position: the fallback + // reinstates the gap permanently, since once the slot and the server hold a + // resume point past it, no later run comes back to it. + redoStart, err := s.getCheckpointRedoStartLSN(ctx, segmentSize) + if err != nil { + return 0, err + } + contextLogger.Debug( - "Current flush LSN", + "Checkpoint redo LSN", + "redoStart", redoStart, "xlogFlushPos", xlogFlushPos, "segmentSize", segmentSize, ) - return getStartWALLSN(xlogFlushPos, segmentSize), nil + return redoStart, nil +} + +// getCheckpointRedoStartLSN returns the start of the WAL file that contains the +// redo point of the latest checkpoint (or restartpoint, on a standby). It opens +// a regular (non-replication) connection because pg_control_checkpoint() cannot +// be queried on the physical replication connection used for streaming. +func (s *Process) getCheckpointRedoStartLSN( + ctx context.Context, + segmentSize uint64, +) (pglogrepl.LSN, error) { + conn, err := pgx.Connect(ctx, s.config.Source.StandardDSN) + if err != nil { + return 0, fmt.Errorf("while connecting to PostgreSQL: %w", err) + } + defer func() { + _ = conn.Close(ctx) + }() + + var redoLSN uint64 + row := conn.QueryRow(ctx, "SELECT redo_lsn - '0/0' FROM pg_control_checkpoint()") + if err := row.Scan(&redoLSN); err != nil { + return 0, fmt.Errorf("while reading the checkpoint redo LSN: %w", err) + } + + return getStartWALLSN(pglogrepl.LSN(redoLSN), segmentSize), nil } type walCoordinate struct { diff --git a/core/internal/repository/wals.go b/core/internal/repository/wals.go index 10997f1c..f3c01db5 100644 --- a/core/internal/repository/wals.go +++ b/core/internal/repository/wals.go @@ -132,3 +132,79 @@ func (c *Connection) GetLatestWALFileForCluster( return lastWal, nil } + +// GetEarliestWALFileForCluster gets the earliest archived WAL segment for a +// certain cluster, or an empty string when the archive holds none. +// +// Only complete WAL segments are considered. The archive also stores the +// in-flight `.partial` file, backup labels and history files, and a name such +// as "000000010000000000000005.partial" would otherwise be reported as older +// than the very segment "000000010000000000000005" that is being written into +// it. +// +// This is the earliest WAL that currently survives in the archive, not the +// earliest one ever archived: the retention removes older segments, and +// `klio reset-lsn` can leave a gap behind. +func (c *Connection) GetEarliestWALFileForCluster( + ctx context.Context, + clusterName string, +) (string, error) { + logger := log.FromContext(ctx) + + readClusterDir, err := afero.ReadDir(c.fs, clusterName) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return "", nil + } + + logger.Error( + err, + "while reading cluster directory", + "clusterName", clusterName, + ) + + return "", fmt.Errorf("while reading cluster directory: %w", err) + } + + // afero.ReadDir sorts its result by name, and a WAL directory sorts in the + // same order as the segments it holds. The earliest directories may hold no + // complete segment at all: the retention skips files carrying an extension, + // so an orphan `.partial` keeps a directory alive. The scan therefore + // continues until a directory yields a segment. + for _, entry := range readClusterDir { + if !entry.IsDir() { + continue + } + + earliestWal, err := c.getEarliestWALFileInDirectory(ctx, path.Join(clusterName, entry.Name())) + if err != nil { + return "", err + } + + if earliestWal != "" { + return earliestWal, nil + } + } + + return "", nil +} + +// getEarliestWALFileInDirectory gets the earliest complete WAL segment held by +// the passed WAL archive directory, or an empty string when it holds none. +func (c *Connection) getEarliestWALFileInDirectory(ctx context.Context, directory string) (string, error) { + readWalDirectory, err := afero.ReadDir(c.fs, directory) + if err != nil { + log.FromContext(ctx).Error(err, "while reading directory", "directory", directory) + return "", fmt.Errorf("while reading WAL directory: %w", err) + } + + for _, entry := range readWalDirectory { + if entry.IsDir() || len(entry.Name()) != expectedWalFileNameLength { + continue + } + + return entry.Name(), nil + } + + return "", nil +} diff --git a/core/internal/repository/wals_test.go b/core/internal/repository/wals_test.go index 7d765c31..4b89d076 100644 --- a/core/internal/repository/wals_test.go +++ b/core/internal/repository/wals_test.go @@ -116,3 +116,94 @@ func TestGetLatestWALFileForCluster(t *testing.T) { require.NoError(t, err) assert.Empty(t, latestWal) } + +func TestGetEarliestWALFileForCluster(t *testing.T) { + opts := Options{ + FS: afero.NewMemMapFs(), + Password: "test-password", + } + require.NoError(t, Initialize(opts)) + + conn, err := Open(opts) + require.NoError(t, err) + require.NotNil(t, conn) + defer conn.Close() + + tests := []struct { + name string + clusterName string + // walDirs maps each WAL archive directory to the files it holds. A + // directory with no files is still created. + walDirs map[string][]string + expected string + }{ + { + name: "non-existent cluster", + clusterName: "non-existent-cluster", + expected: "", + }, + { + name: "several WAL files returns the smallest", + clusterName: "test-cluster", + walDirs: map[string][]string{ + "0000000100000000": { + "00000001000000000000000A", + "00000001000000000000000B", + "00000001000000000000000C", + }, + }, + expected: "00000001000000000000000A", + }, + { + name: "empty cluster directory", + clusterName: "empty-cluster", + walDirs: map[string][]string{"0000000100000000": {}}, + expected: "", + }, + { + name: "in-flight partial is not a segment", + clusterName: "partial-only-cluster", + walDirs: map[string][]string{ + "0000000100000000": {"000000010000000000000005.partial"}, + }, + expected: "", + }, + { + name: "backup label is not a segment", + clusterName: "label-only-cluster", + walDirs: map[string][]string{ + "0000000100000000": {"000000010000000000000004.00000028.backup"}, + }, + expected: "", + }, + { + name: "scan continues past a directory holding no segment", + clusterName: "partial-then-segments-cluster", + walDirs: map[string][]string{ + "0000000100000000": {"000000010000000000000005.partial"}, + "0000000100000001": { + "000000010000000100000002", + "000000010000000100000003", + }, + }, + expected: "000000010000000100000002", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + for walDir, walNames := range tc.walDirs { + require.NoError(t, opts.FS.MkdirAll(path.Join(tc.clusterName, walDir), 0o750)) + for _, walName := range walNames { + file, err := opts.FS.Create(path.Join(tc.clusterName, walDir, walName)) + require.NoError(t, err) + require.NoError(t, file.Close()) + } + } + + earliestWal, err := conn.GetEarliestWALFileForCluster(context.Background(), tc.clusterName) + require.NoError(t, err) + assert.Equal(t, tc.expected, earliestWal) + }) + } +} diff --git a/core/internal/server/walserver/backup.go b/core/internal/server/walserver/backup.go index 76645ae1..24084bdf 100644 --- a/core/internal/server/walserver/backup.go +++ b/core/internal/server/walserver/backup.go @@ -32,6 +32,7 @@ import ( "github.com/cloudnative-pg/klio/core/internal/grpc" "github.com/cloudnative-pg/klio/core/internal/kopia" "github.com/cloudnative-pg/klio/core/internal/queue" + "github.com/cloudnative-pg/klio/core/internal/repository" ) // CloseBackup implements the CloseBackup GRPC call. @@ -39,6 +40,10 @@ func (w *Implementation) CloseBackup( ctx context.Context, request *grpc.CloseBackupRequest, ) (*grpc.CloseBackupResult, error) { + if err := repository.ValidatePathComponent(request.GetClusterName()); err != nil { + return nil, status.Errorf(codes.InvalidArgument, "invalid cluster name: %v", err.Error()) + } + // Step 1: verify if the WALs have been archived missingWALFiles, err := w.checkWALFiles(request) if err != nil { @@ -46,6 +51,28 @@ func (w *Implementation) CloseBackup( } if len(missingWALFiles) > 0 { + // If a required WAL predates the earliest segment the archive holds, it + // can never be archived: this cluster started streaming from a later + // point, and nothing will go back to fill the gap. Fail the backup + // instead of letting the client wait for a WAL that will never arrive. + // + // checkWALFiles walks a single timeline by ascending position, so the + // missing list is already sorted and only its first entry can be the + // oldest required segment. + earliestWAL, err := w.conn.GetEarliestWALFileForCluster(ctx, request.GetClusterName()) + if err != nil { + return nil, status.Errorf(codes.Internal, "while reading earliest archived WAL: %v", err.Error()) + } + if earliestWAL != "" && missingWALFiles[0] < earliestWAL { + return nil, status.Errorf( + codes.FailedPrecondition, + "backup requires WAL %q which predates the earliest archived WAL %q and can never be "+ + "archived: the backup ran on an instance whose last checkpoint precedes the point "+ + "the WAL stream started from. Retry the backup targeting the primary, or wait for a "+ + "checkpoint to be replayed on this instance", + missingWALFiles[0], earliestWAL) + } + return &grpc.CloseBackupResult{ Tier2Schedule: false, MissingWalFiles: missingWALFiles, diff --git a/core/internal/server/walserver/backup_test.go b/core/internal/server/walserver/backup_test.go new file mode 100644 index 00000000..5b56bab7 --- /dev/null +++ b/core/internal/server/walserver/backup_test.go @@ -0,0 +1,146 @@ +/* +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 walserver + +import ( + "context" + "path" + "testing" + + "github.com/spf13/afero" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/cloudnative-pg/klio/core/internal/grpc" + "github.com/cloudnative-pg/klio/core/internal/repository" +) + +const closeBackupSegmentSize = 16 * 1024 * 1024 + +// newTestImplementation returns a WAL server backed by an in-memory repository +// pre-populated with the given WAL files for a single cluster. +func newTestImplementation(t *testing.T, clusterName string, walFiles []string) *Implementation { + t.Helper() + + opts := repository.Options{ + FS: afero.NewMemMapFs(), + Password: "test-password", + } + require.NoError(t, repository.Initialize(opts)) + + conn, err := repository.Open(opts) + require.NoError(t, err) + t.Cleanup(conn.Close) + + for _, walName := range walFiles { + walDir := path.Join(clusterName, walName[0:16]) + require.NoError(t, opts.FS.MkdirAll(walDir, 0o750)) + file, err := opts.FS.Create(path.Join(walDir, walName)) + require.NoError(t, err) + require.NoError(t, file.Close()) + } + + return New(Options{Connection: conn}) +} + +// TestCloseBackupFailsOnPermanentlyMissingWAL verifies that CloseBackup returns +// a terminal error when a required WAL predates the earliest archived WAL and +// can therefore never be archived. +func TestCloseBackupFailsOnPermanentlyMissingWAL(t *testing.T) { + const clusterName = "test-cluster" + + // The archive starts at segment 05: segments 03 and 04 required by the + // backup will never appear. + impl := newTestImplementation(t, clusterName, []string{ + "000000010000000000000005", + "000000010000000000000006", + "000000010000000000000007", + }) + + result, err := impl.CloseBackup(context.Background(), &grpc.CloseBackupRequest{ + ClusterName: clusterName, + Timeline: 1, + StartWal: "000000010000000000000003", + EndWal: "000000010000000000000007", + SegmentSize: closeBackupSegmentSize, + }) + + require.Error(t, err) + require.Nil(t, result) + + s, ok := status.FromError(err) + require.True(t, ok, "expected a gRPC status error") + assert.Equal(t, codes.FailedPrecondition, s.Code()) +} + +// TestCloseBackupWaitsForRecentMissingWAL verifies that CloseBackup keeps +// reporting a not-yet-archived WAL as missing (so the client waits) when that +// WAL does not predate the earliest archived WAL. +func TestCloseBackupWaitsForRecentMissingWAL(t *testing.T) { + const clusterName = "test-cluster" + + // Segment 06 is not archived yet, but it does not predate the earliest + // archived WAL (03): it can still arrive. + impl := newTestImplementation(t, clusterName, []string{ + "000000010000000000000003", + "000000010000000000000004", + "000000010000000000000005", + "000000010000000000000007", + }) + + result, err := impl.CloseBackup(context.Background(), &grpc.CloseBackupRequest{ + ClusterName: clusterName, + Timeline: 1, + StartWal: "000000010000000000000003", + EndWal: "000000010000000000000007", + SegmentSize: closeBackupSegmentSize, + }) + + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, []string{"000000010000000000000006"}, result.GetMissingWalFiles()) +} + +// TestCloseBackupWaitsForWALStillBeingStreamed verifies that the in-flight +// `.partial` file the WAL writer creates for the segment it is receiving does +// not make that same segment look permanently un-archivable. This is the state +// a freshly created cluster is in when its first backup closes. +func TestCloseBackupWaitsForWALStillBeingStreamed(t *testing.T) { + const clusterName = "test-cluster" + + // Nothing is archived yet: segment 05 is still being received. + impl := newTestImplementation(t, clusterName, []string{ + "000000010000000000000005.partial", + }) + + result, err := impl.CloseBackup(context.Background(), &grpc.CloseBackupRequest{ + ClusterName: clusterName, + Timeline: 1, + StartWal: "000000010000000000000005", + EndWal: "000000010000000000000005", + SegmentSize: closeBackupSegmentSize, + }) + + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, []string{"000000010000000000000005"}, result.GetMissingWalFiles()) +} diff --git a/documentation/web/docs/developer/running-e2e-tests.md b/documentation/web/docs/developer/running-e2e-tests.md index 5c173203..a1e83b9e 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 00000000..aba11731 --- /dev/null +++ b/operator/test/e2e/backup_from_replica_cluster_test.go @@ -0,0 +1,280 @@ +/* +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. On a replica cluster the WAL +// streamer of the designated primary starts archiving from the current flush +// position, while pg_backup_start on the underlying standby reports the older +// last-restartpoint LSN: the WAL segments in between must still end up in tier1 +// or the backup waits for WAL files that never arrive. +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, replays past its last restartpoint: this is the state in + // which the streamer starts ahead of what pg_backup_start reports. + _, 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: before the fix it loops forever on missing WAL files. + 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 b0b41e72..28958988 100644 --- a/operator/test/e2e/main_test.go +++ b/operator/test/e2e/main_test.go @@ -50,6 +50,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)))