feat: parallel WAL archiving

This patch adds a parallel WAL archiving feature, similar to the one
that Barman has for its archive_command. This new feature can be
enabled via the `maxParallel` option of the WAL archiving configuration.

Every time the archive_command is invoked by PostgreSQL, we pre-archive
a maximum of `maxParallel` WAL files chosen from the ready ones. The
archival status is kept inside a spool folder.

Co-authored-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Co-authored-by: Philippe Scorsolini <philippe.scorsolini@enterprisedb.com>
Co-authored-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Co-authored-by: Gabriele Bartolini <gabriele.bartolini@enterprisedb.com>

Author: 5 people

api/v1/cluster_types.go

@@ -808,6 +808,15 @@ type WalBackupConfiguration struct {
 	// `AES256` and `aws:kms`
 	// +kubebuilder:validation:Enum=AES256;"aws:kms"
 	Encryption EncryptionType `json:"encryption,omitempty"`
+
+	// Number of WAL files to be either archived in parallel (when the
+	// PostgreSQL instance is archiving to a backup object store) or
+	// restored in parallel (when a PostgreSQL standby is fetching WAL
+	// files from a recovery object store). If not specified, WAL files
+	// will be processed one at a time. It accepts a positive integer as a
+	// value - with 1 being the minimum accepted value.
+	// +kubebuilder:validation:Minimum=1
+	MaxParallel int `json:"maxParallel,omitempty"`
 }
 
 // DataBackupConfiguration is the configuration of the backup of

config/crd/bases/postgresql.k8s.enterprisedb.io_clusters.yaml

@@ -935,6 +935,16 @@ spec:
                             - AES256
                             - aws:kms
                             type: string
+                          maxParallel:
+                            description: Number of WAL files to be either archived
+                              in parallel (when the PostgreSQL instance is archiving
+                              to a backup object store) or restored in parallel (when
+                              a PostgreSQL standby is fetching WAL files from a recovery
+                              object store). If not specified, WAL files will be processed
+                              one at a time. It accepts a positive integer as a value
+                              - with 1 being the minimum accepted value.
+                            minimum: 1
+                            type: integer
                         type: object
                     required:
                     - destinationPath
@@ -1359,6 +1369,17 @@ spec:
                               - AES256
                               - aws:kms
                               type: string
+                            maxParallel:
+                              description: Number of WAL files to be either archived
+                                in parallel (when the PostgreSQL instance is archiving
+                                to a backup object store) or restored in parallel
+                                (when a PostgreSQL standby is fetching WAL files from
+                                a recovery object store). If not specified, WAL files
+                                will be processed one at a time. It accepts a positive
+                                integer as a value - with 1 being the minimum accepted
+                                value.
+                              minimum: 1
+                              type: integer
                           type: object
                       required:
                       - destinationPath

docs/src/api_reference.md

@@ -767,8 +767,9 @@ Name               | Description
 
 WalBackupConfiguration is the configuration of the backup of the WAL stream
 
-Name        | Description                                                                                                                                                                             | Type           
------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------
-`compression` | Compress a WAL file before sending it to the object store. Available options are empty string (no compression, default), `gzip` or `bzip2`.                                             | CompressionType
-`encryption ` | Whenever to force the encryption of files (if the bucket is not already configured for that). Allowed options are empty string (use the bucket policy, default), `AES256` and `aws:kms` | EncryptionType 
+Name        | Description                                                                                                                                                                                                                                                                                                                                                                         | Type           
+----------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------
+`compression` | Compress a WAL file before sending it to the object store. Available options are empty string (no compression, default), `gzip` or `bzip2`.                                                                                                                                                                                                                                         | CompressionType
+`encryption ` | Whenever to force the encryption of files (if the bucket is not already configured for that). Allowed options are empty string (use the bucket policy, default), `AES256` and `aws:kms`                                                                                                                                                                                             | EncryptionType 
+`maxParallel` | Number of WAL files to be either archived in parallel (when the PostgreSQL instance is archiving to a backup object store) or restored in parallel (when a PostgreSQL standby is fetching WAL files from a recovery object store). If not specified, WAL files will be processed one at a time. It accepts a positive integer as a value - with 1 being the minimum accepted value. | int            
 

docs/src/backup_recovery.md

@@ -458,6 +458,37 @@ spec:
 You can configure the encryption directly in your bucket, and the operator
 will use it unless you override it in the cluster configuration.
 
+PostgreSQL implements a sequential archiving scheme, where the
+`archive_command` will be executed sequentially for every WAL
+segment to be archived.
+
+When the bandwidth between the PostgreSQL instance and the object
+store allows archiving more than one WAL file in parallel, you
+can use the parallel WAL archiving feature of the instance manager
+like in the following example:
+
+```yaml
+apiVersion: postgresql.k8s.enterprisedb.io/v1
+kind: Cluster
+[...]
+spec:
+  backup:
+    barmanObjectStore:
+      [...]
+      wal:
+        compression: gzip
+        maxParallel: 8
+        encryption: AES256
+```
+
+In the previous example, the instance manager optimizes the WAL
+archiving process by archiving in parallel at most eight ready
+WALs, including the one requested by PostgreSQL.
+
+When PostgreSQL will request the archiving of a WAL that has
+already been archived by the instance manager as an optimization,
+that archival request will be just dismissed with a positive status.
+
 ## Recovery
 
 Cluster restores are not performed "in-place" on an existing cluster.

internal/cmd/manager/walarchive/cmd.go

@@ -8,18 +8,30 @@ Copyright (C) 2019-2021 EnterpriseDB Corporation.
 package walarchive
 
 import (
+	"context"
+	"errors"
 	"fmt"
-	"os/exec"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+	"time"
 
 	"github.com/spf13/cobra"
 
 	apiv1 "github.com/EnterpriseDB/cloud-native-postgresql/api/v1"
 	"github.com/EnterpriseDB/cloud-native-postgresql/internal/management/cache"
 	cacheClient "github.com/EnterpriseDB/cloud-native-postgresql/internal/management/cache/client"
 	"github.com/EnterpriseDB/cloud-native-postgresql/pkg/management/barman"
-	barmanCapabilities "github.com/EnterpriseDB/cloud-native-postgresql/pkg/management/barman/capabilities"
-	"github.com/EnterpriseDB/cloud-native-postgresql/pkg/management/execlog"
+	"github.com/EnterpriseDB/cloud-native-postgresql/pkg/management/barman/archiver"
 	"github.com/EnterpriseDB/cloud-native-postgresql/pkg/management/log"
+	"github.com/EnterpriseDB/cloud-native-postgresql/pkg/postgres"
+)
+
+const (
+	// SpoolDirectory is the directory where we spool the WAL files that
+	// were pre-archived in parallel
+	SpoolDirectory = postgres.ScratchDataDirectory + "/wal-archive-spool"
 )
 
 // NewCmd creates the new cobra command
@@ -30,7 +42,9 @@ func NewCmd() *cobra.Command {
 		Args:          cobra.ExactArgs(1),
 		RunE: func(cobraCmd *cobra.Command, args []string) error {
 			contextLog := log.WithName("wal-archive")
-			err := run(contextLog, args)
+			ctx := log.IntoContext(cobraCmd.Context(), contextLog)
+
+			err := run(ctx, args)
 			if err != nil {
 				contextLog.Error(err, "failed to run wal-archive command")
 				return err
@@ -42,15 +56,15 @@ func NewCmd() *cobra.Command {
 	return &cmd
 }
 
-func run(contextLog log.Logger, args []string) error {
+func run(ctx context.Context, args []string) error {
+	startTime := time.Now()
+	contextLog := log.FromContext(ctx)
 	walName := args[0]
 
 	var cluster *apiv1.Cluster
 	var err error
 
-	cluster, err = cacheClient.GetCluster()
-	if err != nil {
-		contextLog.Error(err, "Error while getting cluster from cache")
+	if cluster, err = cacheClient.GetCluster(); err != nil {
 		return fmt.Errorf("failed to get cluster: %w", err)
 	}
 
@@ -64,53 +78,135 @@ func run(contextLog log.Logger, args []string) error {
 		return nil
 	}
 
-	options, err := barmanCloudWalArchiveOptions(*cluster, cluster.Name, walName)
-	if err != nil {
-		contextLog.Error(err, "while getting barman-cloud-wal-archive options")
-		return err
+	maxParallel := 1
+	if cluster.Spec.Backup.BarmanObjectStore.Wal != nil {
+		maxParallel = cluster.Spec.Backup.BarmanObjectStore.Wal.MaxParallel
 	}
 
-	env, err := cacheClient.GetEnv(cache.WALArchiveKey)
+	// Get environment from cache
+	var env []string
+	env, err = cacheClient.GetEnv(cache.WALArchiveKey)
 	if err != nil {
-		contextLog.Error(err, "Error while getting environment from cache")
 		return fmt.Errorf("failed to get envs: %w", err)
 	}
 
-	contextLog.Trace("Executing "+barmanCapabilities.BarmanCloudWalArchive,
-		"walName", walName,
-		"currentPrimary", cluster.Status.CurrentPrimary,
-		"targetPrimary", cluster.Status.TargetPrimary,
-		"options", options,
-	)
-
-	barmanCloudWalArchiveCmd := exec.Command(barmanCapabilities.BarmanCloudWalArchive, options...) // #nosec G204
-	barmanCloudWalArchiveCmd.Env = env
+	// Create the archiver
+	var walArchiver *archiver.WALArchiver
+	if walArchiver, err = archiver.New(cluster, env, SpoolDirectory); err != nil {
+		return fmt.Errorf("while creating the archiver: %w", err)
+	}
 
-	err = execlog.RunStreaming(barmanCloudWalArchiveCmd, barmanCapabilities.BarmanCloudWalArchive)
+	// Step 1: check if this WAL file has not been already archived
+	var isDeletedFromSpool bool
+	isDeletedFromSpool, err = walArchiver.DeleteFromSpool(walName)
 	if err != nil {
-		contextLog.Error(err, "Error invoking "+barmanCapabilities.BarmanCloudWalArchive,
+		return fmt.Errorf("while testing the existence of the WAL file in the spool directory: %w", err)
+	}
+	if isDeletedFromSpool {
+		contextLog.Info("Archived WAL file (parallel)",
 			"walName", walName,
 			"currentPrimary", cluster.Status.CurrentPrimary,
-			"targetPrimary", cluster.Status.TargetPrimary,
-			"options", options,
-			"exitCode", barmanCloudWalArchiveCmd.ProcessState.ExitCode(),
-		)
-		return fmt.Errorf("unexpected failure invoking %s: %w", barmanCapabilities.BarmanCloudWalArchive, err)
+			"targetPrimary", cluster.Status.TargetPrimary)
+		return nil
+	}
+
+	// Step 3: gather the WAL files names to archive
+	walFilesList := gatherWALFilesToArchive(ctx, walName, maxParallel)
+
+	options, err := barmanCloudWalArchiveOptions(cluster, cluster.Name)
+	if err != nil {
+		log.Error(err, "while getting barman-cloud-wal-archive options")
+		return err
 	}
 
-	contextLog.Info("Archived WAL file",
-		"walName", walName,
-		"currentPrimary", cluster.Status.CurrentPrimary,
-		"targetPrimary", cluster.Status.TargetPrimary,
-	)
+	// Step 4: archive the WAL files in parallel
+	uploadStartTime := time.Now()
+	walStatus := walArchiver.ArchiveList(walFilesList, options)
+	if len(walStatus) > 1 {
+		contextLog.Info("Completed archive command (parallel)",
+			"walsCount", len(walStatus),
+			"startTime", startTime,
+			"uploadStartTime", uploadStartTime,
+			"uploadTotalTime", time.Since(uploadStartTime),
+			"totalTime", time.Since(startTime))
+	}
+
+	// We return only the first error to PostgreSQL, because the first error
+	// is the one raised by the file that PostgreSQL has requested to archive.
+	// The other errors are related to WAL files that were pre-archived as
+	// a performance optimization and are just logged
+	return walStatus[0].Err
+}
+
+// gatherWALFilesToArchive reads from the archived status the list of WAL files
+// that can be archived in parallel way.
+// `requestedWALFile` is the name of the file whose archiving was requested by
+// PostgreSQL, and that file is always the first of the list and is always included.
+// `parallel` is the maximum number of WALs that we can archive in parallel
+func gatherWALFilesToArchive(ctx context.Context, requestedWALFile string, parallel int) (walList []string) {
+	contextLog := log.FromContext(ctx)
+	pgWalDirectory := path.Join(os.Getenv("PGDATA"), "pg_wal")
+	archiveStatusPath := path.Join(pgWalDirectory, "archive_status")
+	noMoreWALFilesNeeded := errors.New("no more files needed")
+
+	// slightly more optimized, but equivalent to:
+	// walList = []string{requestedWALFile}
+	walList = make([]string, 1, 1+parallel)
+	walList[0] = requestedWALFile
+
+	err := filepath.WalkDir(archiveStatusPath, func(path string, d os.DirEntry, err error) error {
+		// If err is set, it means the current path is a directory and the readdir raised an error
+		// The only available option here is to skip the path and log the error.
+		if err != nil {
+			contextLog.Error(err, "failed reading path", "path", path)
+			return filepath.SkipDir
+		}
+
+		if len(walList) >= parallel {
+			return noMoreWALFilesNeeded
+		}
+
+		// We don't process directories beside the archive status path
+		if d.IsDir() {
+			// We want to proceed exploring the archive status folder
+			if path == archiveStatusPath {
+				return nil
+			}
+
+			return filepath.SkipDir
+		}
+
+		// We only process ready files
+		if !strings.HasSuffix(path, ".ready") {
+			return nil
+		}
+
+		walFileName := strings.TrimSuffix(filepath.Base(path), ".ready")
+
+		// We are already archiving the requested WAL file,
+		// and we need to avoid archiving it twice.
+		// requestedWALFile is usually "pg_wal/wal_file_name" and
+		// we compare it with the path we read
+		if strings.HasSuffix(requestedWALFile, walFileName) {
+			return nil
+		}
+
+		walList = append(walList, filepath.Join("pg_wal", walFileName))
+		return nil
+	})
+
+	// In this point err must be nil or noMoreWALFilesNeeded, if it is something different
+	// there is a programming error
+	if err != nil && err != noMoreWALFilesNeeded {
+		contextLog.Error(err, "unexpected error while reading the list of WAL files to archive")
+	}
 
-	return nil
+	return walList
 }
 
 func barmanCloudWalArchiveOptions(
-	cluster apiv1.Cluster,
+	cluster *apiv1.Cluster,
 	clusterName string,
-	walName string,
 ) ([]string, error) {
 	configuration := cluster.Spec.Backup.BarmanObjectStore
 
@@ -147,7 +243,6 @@ func barmanCloudWalArchiveOptions(
 	options = append(
 		options,
 		configuration.DestinationPath,
-		serverName,
-		walName)
+		serverName)
 	return options, nil
 }