feat: support custom params and parallelism for restore (#6)

All other basic operations already support custom flags.
Even the restore itself is supposed to already support them but, as opposed to Barman, pgBackRest has two separate commands used during the restore:
* restore (responsible for downloading and applying the main backup),
* archive-get (responsible for PITR part - retrieving necessary WAL files later than the backup itself).

Currently only the archive-get part is customizable.

Especially painful is the inability to configure parallelism as that means the main backup can take very long time to download, hence a dedicated field for it (just like for backups).

Signed-off-by: Szymon Soloch <[email protected]>

Author: Agalin

config/crd/bases/pgbackrest.cnpg.opera.com_archives.yaml

@@ -283,6 +283,31 @@ spec:
                       - destinationPath
                       type: object
                     type: array
+                  restore:
+                    description: The configuration to be used to restore the data
+                      files
+                    properties:
+                      additionalCommandArgs:
+                        description: |-
+                          AdditionalCommandArgs represents additional arguments that can be appended
+                          to the 'pgbackrest restore' command-line invocation. These arguments
+                          provide flexibility to customize the restore process further according to
+                          specific requirements or configurations.
+
+                          Note:
+                          It's essential to ensure that the provided arguments are valid and supported
+                          by the 'pgbackrest restore' command, to avoid potential errors or unintended
+                          behavior during execution.
+                        items:
+                          type: string
+                        type: array
+                      jobs:
+                        description: The number of parallel jobs to be used to download
+                          the main backup.
+                        format: int32
+                        minimum: 1
+                        type: integer
+                    type: object
                   stanza:
                     description: |-
                       Pgbackrest stanza (name used in the archive store), the cluster name is used if

internal/cnpgi/common/wal.go

@@ -244,7 +244,7 @@ func (w WALServiceImplementation) restoreFromPgbackrestArchive(
 
 	// Create the restorer
 	var walRestorer *pgbackrestRestorer.WALRestorer
-	if walRestorer, err = pgbackrestRestorer.New(ctx, env, w.SpoolDirectory); err != nil {
+	if walRestorer, err = pgbackrestRestorer.NewWALRestorer(ctx, env, w.SpoolDirectory); err != nil {
 		return fmt.Errorf("while creating the restorer: %w", err)
 	}
 

internal/cnpgi/operator/reconciler.go

@@ -184,7 +184,7 @@ func (r ReconcilerImplementation) ensureRole(
 		"namespace", newRole.Namespace,
 		"rules", newRole.Rules,
 	)
-
+	// TODO: Something is wrong here, we also should check for ownership.
 	return r.Client.Patch(ctx, newRole, client.MergeFrom(&role))
 }
 

internal/cnpgi/restore/restore.go

@@ -19,18 +19,15 @@ package restore
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"os"
-	"os/exec"
 	"path"
 	"time"
 
 	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/postgres"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 	restore "github.com/cloudnative-pg/cnpg-i/pkg/restore/job"
-	"github.com/cloudnative-pg/machinery/pkg/execlog"
 	"github.com/cloudnative-pg/machinery/pkg/fileutils"
 	"github.com/cloudnative-pg/machinery/pkg/log"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -153,63 +150,12 @@ func (impl JobHookImpl) restoreDataDir(
 	env []string,
 	pgbackrestConfiguration *pgbackrestApi.PgbackrestConfiguration,
 ) error {
-	var options []string
-
-	if backup.Status.EndpointURL != "" {
-		options = append(options, "--endpoint-url", backup.Status.EndpointURL)
-	}
-
-	options = append(
-		options,
-		"--stanza", backup.Status.ServerName,
-		"--lock-path",
-		"/controller/tmp/pgbackrest")
-
-	options, err := pgbackrestCommand.AppendCloudProviderOptionsFromConfiguration(ctx, options, pgbackrestConfiguration)
-	if err != nil {
-		return err
-	}
-
-	options, err = pgbackrestCommand.AppendLogOptionsFromConfiguration(ctx, options, pgbackrestConfiguration)
-	if err != nil {
-		return err
-	}
-
-	options, err = pgbackrestCommand.AppendStanzaOptionsFromConfiguration(
-		ctx,
-		options,
+	restoreCmd := pgbackrestRestorer.NewRestoreCommand(
 		pgbackrestConfiguration,
 		impl.PgDataPath,
-		false,
 	)
-	if err != nil {
-		return err
-	}
-
-	options = append(
-		options,
-		"restore",
-		"--set",
-		backup.Status.BackupID,
-	)
-
-	log.Info("Starting pgbackrest restore",
-		"options", options)
 
-	cmd := exec.Command("pgbackrest", options...) // #nosec G204
-	cmd.Env = env
-	err = execlog.RunStreaming(cmd, "pgbackrest restore")
-	if err != nil {
-		var exitError *exec.ExitError
-		if errors.As(err, &exitError) {
-			err = pgbackrestCommand.UnmarshalPgbackrestRestoreExitCode(ctx, exitError.ExitCode())
-		}
-
-		log.Error(err, "Can't restore backup")
-		return err
-	}
-	log.Info("Restore completed")
-	return nil
+	return restoreCmd.Restore(ctx, backup.Status.BackupID, backup.Status.ServerName, env)
 }
 
 // TODO: Likely doesn't make sense for pgbackrest. Might be tricky to implement properly.
@@ -234,7 +180,7 @@ func (impl JobHookImpl) ensureArchiveContainsLastCheckpointRedoWAL(
 		return err
 	}
 
-	rest, err := pgbackrestRestorer.New(
+	rest, err := pgbackrestRestorer.NewWALRestorer(
 		ctx,
 		env,
 		impl.SpoolDirectory,

internal/pgbackrest/api/config.go

@@ -236,6 +236,28 @@ type DataBackupConfiguration struct {
 	AdditionalCommandArgs []string `json:"additionalCommandArgs,omitempty"`
 }
 
+// DataRestoreConfiguration is the configuration of the main backup restore process
+// (pgbackrest restore call) which is then followed by a series of WAL restore
+// (pgbackrest archive-get) calls using the WalBackupConfiguration
+type DataRestoreConfiguration struct {
+	// The number of parallel jobs to be used to download the main backup.
+	// +kubebuilder:validation:Minimum=1
+	// +optional
+	Jobs *int32 `json:"jobs,omitempty"`
+
+	// AdditionalCommandArgs represents additional arguments that can be appended
+	// to the 'pgbackrest restore' command-line invocation. These arguments
+	// provide flexibility to customize the restore process further according to
+	// specific requirements or configurations.
+	//
+	// Note:
+	// It's essential to ensure that the provided arguments are valid and supported
+	// by the 'pgbackrest restore' command, to avoid potential errors or unintended
+	// behavior during execution.
+	// +optional
+	AdditionalCommandArgs []string `json:"additionalCommandArgs,omitempty"`
+}
+
 // PgbackrestRepository contains configuration of a single Pgbackrest backup target
 // repository, including all data needed to properly connect and authenticate with
 // a selected object store.
@@ -298,6 +320,10 @@ type PgbackrestConfiguration struct {
 	// +optional
 	Data *DataBackupConfiguration `json:"data,omitempty"`
 
+	// The configuration to be used to restore the data files
+	// +optional
+	Restore *DataRestoreConfiguration `json:"restore,omitempty"`
+
 	// Compress a WAL file before sending it to the object store. Available
 	// options are empty string (no compression, default), `gz`, `bz2`, `lz4` or 'zst'.
 	// +kubebuilder:validation:Enum=gz;bz2;lz4;zst
@@ -316,8 +342,16 @@ func (credentials PgbackrestCredentials) ArePopulated() bool {
 	return credentials.AWS != nil
 }
 
-// AppendRestoreAdditionalCommandArgs adds custom arguments as pgbackrest restore command-line options
-func (cfg *WalBackupConfiguration) AppendRestoreAdditionalCommandArgs(options []string) []string {
+// AppendAdditionalRestoreCommandArgs adds custom arguments as pgbackrest restore command-line options
+func (cfg *DataRestoreConfiguration) AppendAdditionalRestoreCommandArgs(options []string) []string {
+	if cfg == nil || len(cfg.AdditionalCommandArgs) == 0 {
+		return options
+	}
+	return appendAdditionalCommandArgs(cfg.AdditionalCommandArgs, options)
+}
+
+// AppendAdditionalArchiveGetCommandArgs adds custom arguments as pgbackrest archive-get command-line options
+func (cfg *WalBackupConfiguration) AppendAdditionalArchiveGetCommandArgs(options []string) []string {
 	if cfg == nil || len(cfg.RestoreAdditionalCommandArgs) == 0 {
 		return options
 	}
@@ -342,16 +376,16 @@ func appendAdditionalCommandArgs(additionalCommandArgs []string, options []strin
 	return options
 }
 
-// AppendAdditionalCommandArgs adds custom arguments as pgbackrest backup command-line options
-func (cfg *DataBackupConfiguration) AppendAdditionalCommandArgs(options []string) []string {
+// AppendAdditionalBackupCommandArgs adds custom arguments as pgbackrest backup command-line options
+func (cfg *DataBackupConfiguration) AppendAdditionalBackupCommandArgs(options []string) []string {
 	if cfg == nil || len(cfg.AdditionalCommandArgs) == 0 {
 		return options
 	}
 	return appendAdditionalCommandArgs(cfg.AdditionalCommandArgs, options)
 }
 
-// AppendArchiveAdditionalCommandArgs adds custom arguments as pgbackrest archive-push command-line options
-func (cfg *WalBackupConfiguration) AppendArchiveAdditionalCommandArgs(options []string) []string {
+// AppendAdditionalArchivePushCommandArgs adds custom arguments as pgbackrest archive-push command-line options
+func (cfg *WalBackupConfiguration) AppendAdditionalArchivePushCommandArgs(options []string) []string {
 	if cfg == nil || len(cfg.ArchiveAdditionalCommandArgs) == 0 {
 		return options
 	}

internal/pgbackrest/api/config_test.go

@@ -22,7 +22,7 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-var _ = Describe("DataBackupConfiguration.AppendAdditionalCommandArgs", func() {
+var _ = Describe("DataBackupConfiguration.AppendAdditionalBackupCommandArgs", func() {
 	var options []string
 	var config DataBackupConfiguration
 	BeforeEach(func() {
@@ -33,35 +33,79 @@ var _ = Describe("DataBackupConfiguration.AppendAdditionalCommandArgs", func() {
 	})
 
 	It("should append additional command args to the options", func() {
-		updatedOptions := config.AppendAdditionalCommandArgs(options)
+		updatedOptions := config.AppendAdditionalBackupCommandArgs(options)
 		Expect(updatedOptions).To(Equal([]string{"--option1", "--option2", "--option3", "--option4"}))
 	})
 
 	It("should return the original options if there are no additional command args", func() {
 		config.AdditionalCommandArgs = nil
-		updatedOptions := config.AppendAdditionalCommandArgs(options)
+		updatedOptions := config.AppendAdditionalBackupCommandArgs(options)
 		Expect(updatedOptions).To(Equal(options))
 	})
 })
 
-var _ = Describe("WalBackupConfiguration.AppendAdditionalCommandArgs", func() {
+var _ = Describe("DataBackupConfiguration.AppendAdditionalRestoreCommandArgs", func() {
 	var options []string
-	var config DataBackupConfiguration
+	var config DataRestoreConfiguration
 	BeforeEach(func() {
 		options = []string{"--option1", "--option2"}
-		config = DataBackupConfiguration{
+		config = DataRestoreConfiguration{
 			AdditionalCommandArgs: []string{"--option3", "--option4"},
 		}
 	})
 
 	It("should append additional command args to the options", func() {
-		updatedOptions := config.AppendAdditionalCommandArgs(options)
+		updatedOptions := config.AppendAdditionalRestoreCommandArgs(options)
 		Expect(updatedOptions).To(Equal([]string{"--option1", "--option2", "--option3", "--option4"}))
 	})
 
 	It("should return the original options if there are no additional command args", func() {
 		config.AdditionalCommandArgs = nil
-		updatedOptions := config.AppendAdditionalCommandArgs(options)
+		updatedOptions := config.AppendAdditionalRestoreCommandArgs(options)
+		Expect(updatedOptions).To(Equal(options))
+	})
+})
+
+var _ = Describe("WalBackupConfiguration.AppendAdditionalArchivePushCommandArgs", func() {
+	var options []string
+	var config WalBackupConfiguration
+	BeforeEach(func() {
+		options = []string{"--option1", "--option2"}
+		config = WalBackupConfiguration{
+			ArchiveAdditionalCommandArgs: []string{"--option3", "--option4"},
+		}
+	})
+
+	It("should append additional command args to the options", func() {
+		updatedOptions := config.AppendAdditionalArchivePushCommandArgs(options)
+		Expect(updatedOptions).To(Equal([]string{"--option1", "--option2", "--option3", "--option4"}))
+	})
+
+	It("should return the original options if there are no additional command args", func() {
+		config.ArchiveAdditionalCommandArgs = nil
+		updatedOptions := config.AppendAdditionalArchivePushCommandArgs(options)
+		Expect(updatedOptions).To(Equal(options))
+	})
+})
+
+var _ = Describe("WalBackupConfiguration.AppendAdditionalArchiveGetCommandArgs", func() {
+	var options []string
+	var config WalBackupConfiguration
+	BeforeEach(func() {
+		options = []string{"--option1", "--option2"}
+		config = WalBackupConfiguration{
+			RestoreAdditionalCommandArgs: []string{"--option3", "--option4"},
+		}
+	})
+
+	It("should append additional command args to the options", func() {
+		updatedOptions := config.AppendAdditionalArchiveGetCommandArgs(options)
+		Expect(updatedOptions).To(Equal([]string{"--option1", "--option2", "--option3", "--option4"}))
+	})
+
+	It("should return the original options if there are no additional command args", func() {
+		config.RestoreAdditionalCommandArgs = nil
+		updatedOptions := config.AppendAdditionalArchiveGetCommandArgs(options)
 		Expect(updatedOptions).To(Equal(options))
 	})
 })

internal/pgbackrest/api/zz_generated.deepcopy.go

@@ -129,7 +129,7 @@ func (archiver *WALArchiver) PgbackrestWalArchiveOptions(
 	// }
 
 	if configuration.Wal != nil {
-		options = configuration.Wal.AppendArchiveAdditionalCommandArgs(options)
+		options = configuration.Wal.AppendAdditionalArchivePushCommandArgs(options)
 	}
 
 	options, err = pgbackrestCommand.AppendCloudProviderOptionsFromConfiguration(ctx, options, configuration)

internal/pgbackrest/archiver/command.go

@@ -88,7 +88,7 @@ func (b *Command) GetDataConfiguration(
 			strconv.Itoa(int(*b.configuration.Data.Jobs)))
 	}
 
-	return b.configuration.Data.AppendAdditionalCommandArgs(options), nil
+	return b.configuration.Data.AppendAdditionalBackupCommandArgs(options), nil
 }
 
 // GetPgbackrestBackupOptions extract the list of command line options to be used with

internal/pgbackrest/backup/backup.go

@@ -51,7 +51,7 @@ func CloudWalRestoreOptions(
 		"--stanza",
 		stanza)
 
-	options = configuration.Wal.AppendRestoreAdditionalCommandArgs(options)
+	options = configuration.Wal.AppendAdditionalArchiveGetCommandArgs(options)
 
 	return options, nil
 }