diff --git a/core/cmd/backup/run.go b/core/cmd/backup/run.go index 0b06522f..6e4bd13e 100644 --- a/core/cmd/backup/run.go +++ b/core/cmd/backup/run.go @@ -158,6 +158,7 @@ func runBackup(cmd *cobra.Command, _ []string) error { EndWal: metadata.EndWAL, SegmentSize: metadata.SegmentSize, SendToTier2: tier2, + EnqueueWithoutWals: !waitWALs, Tier2RetentionPolicy: marshalTier2RetentionPolicy(cmd.Context(), &configuration), Tier2CompressionPolicy: marshalTier2CompressionPolicy(cmd.Context(), &configuration), }) @@ -167,14 +168,21 @@ func runBackup(cmd *cobra.Command, _ []string) error { backupfailure.RepositoryError.ExitCode) } - if waitWALs && len(result.GetMissingWalFiles()) > 0 { + if len(result.GetMissingWalFiles()) > 0 { + if waitWALs { + contextLogger.Info( + "Detected missing WAL files, waiting for 5 seconds", + "missingWALFiles", result.GetMissingWalFiles(), + ) + time.Sleep(5 * time.Second) + + continue + } + contextLogger.Info( - "Detected missing WAL files, waiting for 5 seconds", + "Detected missing WAL files, not waiting for them", "missingWALFiles", result.GetMissingWalFiles(), ) - time.Sleep(5 * time.Second) - - continue } if result.GetTier2Schedule() { diff --git a/core/internal/grpc/klio_wal.pb.go b/core/internal/grpc/klio_wal.pb.go index 9cc51d6d..cb429dbe 100644 --- a/core/internal/grpc/klio_wal.pb.go +++ b/core/internal/grpc/klio_wal.pb.go @@ -730,8 +730,13 @@ type CloseBackupRequest struct { Tier2RetentionPolicy string `protobuf:"bytes,9,opt,name=tier2_retention_policy,json=tier2RetentionPolicy,proto3" json:"tier2_retention_policy,omitempty"` // When present, set the tier2 compression policy to the specified JSON-serialized policy. Tier2CompressionPolicy string `protobuf:"bytes,10,opt,name=tier2_compression_policy,json=tier2CompressionPolicy,proto3" json:"tier2_compression_policy,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Enqueue the post-backup task even if WAL files are still missing. A + // client that leaves this unset calls CloseBackup again until no WAL file + // is missing, and the server defers the task until that call. A client can + // request an immediate enqueue of the post-backup task by setting this to true. + EnqueueWithoutWals bool `protobuf:"varint,11,opt,name=enqueue_without_wals,json=enqueueWithoutWals,proto3" json:"enqueue_without_wals,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *CloseBackupRequest) Reset() { @@ -827,6 +832,13 @@ func (x *CloseBackupRequest) GetTier2CompressionPolicy() string { return "" } +func (x *CloseBackupRequest) GetEnqueueWithoutWals() bool { + if x != nil { + return x.EnqueueWithoutWals + } + return false +} + // This is sent by the WAL server in response to a CloseBackupRequest // message. type CloseBackupResult struct { @@ -930,7 +942,7 @@ const file_proto_klio_wal_proto_rawDesc = "" + "\fStartWALFile\x12!\n" + "\fklio_version\x18\x01 \x01(\x04R\vklioVersion\x12\x1f\n" + "\vfile_length\x18\x02 \x01(\x04R\n" + - "fileLength\"\xe1\x02\n" + + "fileLength\"\x93\x03\n" + "\x12CloseBackupRequest\x12!\n" + "\fcluster_name\x18\x01 \x01(\tR\vclusterName\x12\x1f\n" + "\vbackup_name\x18\x03 \x01(\tR\n" + @@ -942,7 +954,8 @@ const file_proto_klio_wal_proto_rawDesc = "" + "\rsend_to_tier2\x18\b \x01(\bR\vsendToTier2\x124\n" + "\x16tier2_retention_policy\x18\t \x01(\tR\x14tier2RetentionPolicy\x128\n" + "\x18tier2_compression_policy\x18\n" + - " \x01(\tR\x16tier2CompressionPolicy\"f\n" + + " \x01(\tR\x16tier2CompressionPolicy\x120\n" + + "\x14enqueue_without_wals\x18\v \x01(\bR\x12enqueueWithoutWals\"f\n" + "\x11CloseBackupResult\x12%\n" + "\x0etier2_schedule\x18\x01 \x01(\bR\rtier2Schedule\x12*\n" + "\x11missing_wal_files\x18\x02 \x03(\tR\x0fmissingWalFiles2\xd8\x03\n" + diff --git a/core/internal/server/walserver/backup.go b/core/internal/server/walserver/backup.go index fefe4d85..8cbbca86 100644 --- a/core/internal/server/walserver/backup.go +++ b/core/internal/server/walserver/backup.go @@ -45,7 +45,11 @@ func (w *Implementation) CloseBackup( return nil, err } - if len(missingWALFiles) > 0 { + // By default a client calls CloseBackup again until no WAL is missing, so + // the task is deferred to that call. A client may not wait for the WALs to be archived, + // and can set EnqueueWithoutWals to true in the request + // to force the backup to be enqueued even if some WALs are missing. + if len(missingWALFiles) > 0 && !request.GetEnqueueWithoutWals() { return &grpc.CloseBackupResult{ Tier2Schedule: false, MissingWalFiles: missingWALFiles, @@ -67,7 +71,7 @@ func (w *Implementation) CloseBackup( return &grpc.CloseBackupResult{ Tier2Schedule: w.queue != nil && request.GetSendToTier2(), - MissingWalFiles: nil, + MissingWalFiles: missingWALFiles, }, nil } diff --git a/core/internal/server/walserver/backup_test.go b/core/internal/server/walserver/backup_test.go new file mode 100644 index 00000000..e1845437 --- /dev/null +++ b/core/internal/server/walserver/backup_test.go @@ -0,0 +1,153 @@ +/* +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" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/spf13/afero" + "github.com/stretchr/testify/require" + + "github.com/cloudnative-pg/klio/core/internal/grpc" + "github.com/cloudnative-pg/klio/core/internal/queue" + "github.com/cloudnative-pg/klio/core/internal/repository" +) + +const ( + testClusterName = "cluster-example" + testSegmentSize = 16 * 1024 * 1024 + testStartWAL = "000000010000000000000001" + testEndWAL = "000000010000000000000002" +) + +// newCloseBackupServer builds a WAL server whose repository holds only the +// backup's first WAL segment, so the last one is always reported missing, +// and whose queue is backed by an embedded NATS server. It returns the +// server and the queue connection to consume the enqueued tasks from. +func newCloseBackupServer(t *testing.T) (*Implementation, *queue.Conn) { + t.Helper() + + ns, err := server.NewServer(&server.Options{ + Host: "127.0.0.1", + Port: -1, + JetStream: true, + StoreDir: t.TempDir(), + }) + require.NoError(t, err) + go ns.Start() + require.True(t, ns.ReadyForConnections(4*time.Second), "NATS server not ready") + t.Cleanup(ns.Shutdown) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(nc.Close) + + q, err := queue.New(context.Background(), nc) + require.NoError(t, err) + + fs := afero.NewMemMapFs() + repoOpts := repository.Options{FS: fs, Password: "test-password"} + require.NoError(t, repository.Initialize(repoOpts)) + conn, err := repository.Open(repoOpts) + require.NoError(t, err) + t.Cleanup(conn.Close) + + // Only the first segment has been archived. + walPath := path.Join(testClusterName, testStartWAL[:16], testStartWAL) + require.NoError(t, afero.WriteFile(fs, walPath, []byte("wal"), 0o600)) + + return New(Options{Connection: conn, Queue: q}), q +} + +// receiveBackupTask consumes one backup task from the queue, or returns nil +// when none arrives within the timeout. +func receiveBackupTask(t *testing.T, q *queue.Conn, timeout time.Duration) *queue.BackupTask { + t.Helper() + + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + received := make(chan *queue.BackupTask, 1) + go func() { + _ = q.ConsumeBackupReceivedMessages(ctx, func(_ context.Context, task *queue.BackupTask) error { + received <- task + cancel() + + return nil + }) + }() + + select { + case task := <-received: + return task + case <-ctx.Done(): + return nil + } +} + +func newCloseBackupRequest(enqueueWithoutWALs bool) *grpc.CloseBackupRequest { + return &grpc.CloseBackupRequest{ + ClusterName: testClusterName, + BackupName: "backup-1", + Timeline: 1, + StartWal: testStartWAL, + EndWal: testEndWAL, + SegmentSize: testSegmentSize, + SendToTier2: true, + EnqueueWithoutWals: enqueueWithoutWALs, + } +} + +// TestCloseBackupMissingWALsWithoutWaitEnqueuesTask covers a backup taken +// with a client that does not wait for the last WAL to be archived, so the post-backup +// task must be enqueued on this single call or the backup is never relayed. +func TestCloseBackupMissingWALsWithoutWaitEnqueuesTask(t *testing.T) { + impl, q := newCloseBackupServer(t) + + result, err := impl.CloseBackup(context.Background(), newCloseBackupRequest(true)) + require.NoError(t, err) + require.Equal(t, []string{testEndWAL}, result.GetMissingWalFiles()) + require.True(t, result.GetTier2Schedule()) + + task := receiveBackupTask(t, q, 5*time.Second) + require.NotNil(t, task, "the backup task must be enqueued even if WALs are missing") + require.Equal(t, testClusterName, task.ClusterName) + require.True(t, task.SendToTier2) +} + +// TestCloseBackupMissingWALsWithWaitDefersTask covers a backup taken with a client +// that waits for the last WAL to be archived: the client retries CloseBackup until no WAL is missing, so the +// task must not be enqueued before then. +func TestCloseBackupMissingWALsWithWaitDefersTask(t *testing.T) { + impl, q := newCloseBackupServer(t) + + result, err := impl.CloseBackup(context.Background(), newCloseBackupRequest(false)) + require.NoError(t, err) + require.Equal(t, []string{testEndWAL}, result.GetMissingWalFiles()) + require.False(t, result.GetTier2Schedule()) + + require.Nil(t, receiveBackupTask(t, q, time.Second), + "no backup task must be enqueued while the client is still waiting for WALs") +} diff --git a/core/proto/klio_wal.proto b/core/proto/klio_wal.proto index 2d936e0c..e2f013e0 100644 --- a/core/proto/klio_wal.proto +++ b/core/proto/klio_wal.proto @@ -164,6 +164,12 @@ message CloseBackupRequest { // When present, set the tier2 compression policy to the specified JSON-serialized policy. string tier2_compression_policy = 10; + + // Enqueue the post-backup task even if WAL files are still missing. A + // client that leaves this unset calls CloseBackup again until no WAL file + // is missing, and the server defers the task until that call. A client can + // request an immediate enqueue of the post-backup task by setting this to true. + bool enqueue_without_wals = 11; } // This is sent by the WAL server in response to a CloseBackupRequest diff --git a/documentation/web/docs/developer/_protocol.md b/documentation/web/docs/developer/_protocol.md index 45aeaabe..dc375548 100644 --- a/documentation/web/docs/developer/_protocol.md +++ b/documentation/web/docs/developer/_protocol.md @@ -352,6 +352,7 @@ been completed. | send_to_tier2 | [bool](#bool) | | Require this backup to be sent to tier2. | | tier2_retention_policy | [string](#string) | | When present, set the tier2 retention policy to the specified JSON-serialized policy. | | tier2_compression_policy | [string](#string) | | When present, set the tier2 compression policy to the specified JSON-serialized policy. | +| enqueue_without_wals | [bool](#bool) | | Enqueue the post-backup task even if WAL files are still missing. A client that leaves this unset calls CloseBackup again until no WAL file is missing, and the server defers the task until that call. A client can request an immediate enqueue of the post-backup task by setting this to true. | diff --git a/documentation/web/docs/developer/running-e2e-tests.md b/documentation/web/docs/developer/running-e2e-tests.md index 9c7edc40..8ee02c12 100644 --- a/documentation/web/docs/developer/running-e2e-tests.md +++ b/documentation/web/docs/developer/running-e2e-tests.md @@ -121,8 +121,9 @@ The E2E tests are located in `operator/test/e2e/` and include: - `RecoverReplicaCluster`: replica cluster creation from backup - **`tablespace_recovery_test.go`** - Recovery preserving PostgreSQL tablespaces (`RecoverClusterWithTablespaces`) -- **`tier2_recovery_test.go`** - Recovery from tier2 S3 storage - (`RecoverClusterFromTier2`) +- **`tier2_recovery_test.go`** - Recovery from tier2 S3 storage of a + backup taken on a standby of a two-instance cluster, which must reach + tier2 without waiting for its last WAL (`RecoverClusterFromTier2`) - **`tier2_pitr_test.go`** - Point-in-time recovery from tier2 storage (`RecoverClusterFromTier2Pitr`) - **`tier2_recovery_common_test.go`** - Shared tier2 recovery helpers used diff --git a/operator/test/e2e/tier2_recovery_test.go b/operator/test/e2e/tier2_recovery_test.go index 46b72c77..ffdbff09 100644 --- a/operator/test/e2e/tier2_recovery_test.go +++ b/operator/test/e2e/tier2_recovery_test.go @@ -40,6 +40,7 @@ import ( machineryConditions "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/conditions" machineryFeatures "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/features" "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/namespaces" + "github.com/cloudnative-pg/klio/operator/test/machinery/pkg/postgres" ) // tier2RecoveryScenario contains all resources needed for tier2 recovery testing. @@ -168,6 +169,18 @@ func (s *tier2RecoveryScenario) Teardown( return ctx } +// switchWALOnSource forces a WAL switch on the source primary after the +// backup. A backup taken on a standby cannot switch the WAL itself, so +// without this the segment holding its end LSN would never be completed on +// an idle cluster, and the recovery would wait for it forever. +func (s *tier2RecoveryScenario) switchWALOnSource( + ctx context.Context, + _ *cnpgv1.Cluster, + r *resources.Resources, +) error { + return postgres.CheckpointAndSwitchWal(ctx, r, &s.sourcePrimaryPod) +} + // deployRecoveryServer creates the second Klio Server after tier2 replication. func (s *tier2RecoveryScenario) deployRecoveryServer( ctx context.Context, @@ -220,21 +233,27 @@ func NewTier2RecoveryFeatureConfig( } recoveryConfig := machineryFeatures.RecoveryFeatureConfig{ - Name: name, - Setup: scenario.Setup, - Teardown: scenario.Teardown, - SourcePrimaryPod: &scenario.sourcePrimaryPod, - Backup: res.Backup, - RecoveryCluster: res.RecoveryCluster, - MutateRecoveryCluster: []machineryFeatures.RecoveryClusterMutateFunc{scenario.deployRecoveryServer}, - BackupTimeout: 5 * time.Minute, + Name: name, + Setup: scenario.Setup, + Teardown: scenario.Teardown, + SourcePrimaryPod: &scenario.sourcePrimaryPod, + Backup: res.Backup, + RecoveryCluster: res.RecoveryCluster, + MutateRecoveryCluster: []machineryFeatures.RecoveryClusterMutateFunc{ + scenario.switchWALOnSource, + scenario.deployRecoveryServer, + }, + BackupTimeout: 5 * time.Minute, } return recoveryConfig } // RecoverClusterFromTier2 returns a RecoveryFeature for tier2 recovery testing. +// The source cluster has two instances so that the backup, with the default +// prefer-standby target, is taken on the standby: such a backup does not wait +// for its last WAL segment and must still be relayed to tier2. func RecoverClusterFromTier2(namespace string) *machineryFeatures.RecoveryFeature { return machineryFeatures.NewRecoveryFeature( - NewTier2RecoveryFeatureConfig("RecoverClusterFromTier2", 1, namespace)) + NewTier2RecoveryFeatureConfig("RecoverClusterFromTier2", 2, namespace)) }