Run backup command for cloud-based repos in the backup job.

Author: dsessler7

internal/controller/postgrescluster/instance.go

@@ -1395,10 +1395,8 @@ func addPGBackRestToInstancePodSpec(
 	ctx context.Context, cluster *v1beta1.PostgresCluster,
 	instanceCertificates *corev1.Secret, instancePod *corev1.PodSpec,
 ) {
-	if pgbackrest.RepoHostVolumeDefined(cluster) {
-		pgbackrest.AddServerToInstancePod(ctx, cluster, instancePod,
-			instanceCertificates.Name)
-	}
+	pgbackrest.AddServerToInstancePod(ctx, cluster, instancePod,
+		instanceCertificates.Name)
 
 	pgbackrest.AddConfigToInstancePod(cluster, instancePod)
 }

internal/controller/postgrescluster/pgbackrest.go

@@ -23,7 +23,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/labels"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -776,11 +775,6 @@ func generateBackupJobSpecIntent(ctx context.Context, postgresCluster *v1beta1.P
 	repo v1beta1.PGBackRestRepo, serviceAccountName string,
 	labels, annotations map[string]string, opts ...string) (*batchv1.JobSpec, error) {
 
-	selector, containerName, err := getPGBackRestExecSelector(postgresCluster, repo)
-	if err != nil {
-		return nil, errors.WithStack(err)
-	}
-
 	repoIndex := regexRepoIndex.FindString(repo.Name)
 	cmdOpts := []string{
 		"--stanza=" + pgbackrest.DefaultStanzaName,
@@ -794,21 +788,31 @@ func generateBackupJobSpecIntent(ctx context.Context, postgresCluster *v1beta1.P
 	cmdOpts = append(cmdOpts, opts...)
 
 	container := corev1.Container{
-		Command: []string{"/opt/crunchy/bin/pgbackrest"},
-		Env: []corev1.EnvVar{
-			{Name: "COMMAND", Value: "backup"},
-			{Name: "COMMAND_OPTS", Value: strings.Join(cmdOpts, " ")},
-			{Name: "COMPARE_HASH", Value: "true"},
-			{Name: "CONTAINER", Value: containerName},
-			{Name: "NAMESPACE", Value: postgresCluster.GetNamespace()},
-			{Name: "SELECTOR", Value: selector.String()},
-		},
 		Image:           config.PGBackRestContainerImage(postgresCluster),
 		ImagePullPolicy: postgresCluster.Spec.ImagePullPolicy,
 		Name:            naming.PGBackRestRepoContainerName,
 		SecurityContext: initialize.RestrictedSecurityContext(),
 	}
 
+	// If the repo that we are backing up to is a local volume, we will configure
+	// the job to use the pgbackrest go binary to exec into the repo host and run
+	// the backup. If the repo is a cloud-based repo, we will run the pgbackrest
+	// backup command directly in the job pod.
+	if repo.Volume != nil {
+		container.Command = []string{"/opt/crunchy/bin/pgbackrest"}
+		container.Env = []corev1.EnvVar{
+			{Name: "COMMAND", Value: "backup"},
+			{Name: "COMMAND_OPTS", Value: strings.Join(cmdOpts, " ")},
+			{Name: "COMPARE_HASH", Value: "true"},
+			{Name: "CONTAINER", Value: naming.PGBackRestRepoContainerName},
+			{Name: "NAMESPACE", Value: postgresCluster.GetNamespace()},
+			{Name: "SELECTOR", Value: naming.PGBackRestDedicatedSelector(postgresCluster.GetName()).String()},
+		}
+	} else {
+		container.Command = []string{"/bin/pgbackrest", "backup"}
+		container.Command = append(container.Command, cmdOpts...)
+	}
+
 	if postgresCluster.Spec.Backups.PGBackRest.Jobs != nil {
 		container.Resources = postgresCluster.Spec.Backups.PGBackRest.Jobs.Resources
 	}
@@ -862,10 +866,13 @@ func generateBackupJobSpecIntent(ctx context.Context, postgresCluster *v1beta1.P
 	jobSpec.Template.Spec.ImagePullSecrets = postgresCluster.Spec.ImagePullSecrets
 
 	// add pgBackRest configs to template
-	if containerName == naming.PGBackRestRepoContainerName {
+	if repo.Volume != nil {
 		pgbackrest.AddConfigToRepoPod(postgresCluster, &jobSpec.Template.Spec)
 	} else {
-		pgbackrest.AddConfigToInstancePod(postgresCluster, &jobSpec.Template.Spec)
+		// If we are doing a cloud repo backup, we need to give pgbackrest proper permissions
+		// to read certificate files
+		jobSpec.Template.Spec.SecurityContext = postgres.PodSecurityContext(postgresCluster)
+		pgbackrest.AddConfigToCloudBackupJob(postgresCluster, &jobSpec.Template)
 	}
 
 	return jobSpec, nil
@@ -2033,8 +2040,6 @@ func (r *Reconciler) reconcilePGBackRestConfig(ctx context.Context,
 	repoHostName, configHash, serviceName, serviceNamespace string,
 	instanceNames []string) error {
 
-	log := logging.FromContext(ctx).WithValues("reconcileResource", "repoConfig")
-
 	backrestConfig, err := pgbackrest.CreatePGBackRestConfigMapIntent(ctx, postgresCluster, repoHostName,
 		configHash, serviceName, serviceNamespace, instanceNames)
 	if err != nil {
@@ -2048,12 +2053,6 @@ func (r *Reconciler) reconcilePGBackRestConfig(ctx context.Context,
 		return errors.WithStack(err)
 	}
 
-	repoHostConfigured := pgbackrest.RepoHostVolumeDefined(postgresCluster)
-	if !repoHostConfigured {
-		log.V(1).Info("skipping SSH reconciliation, no repo hosts configured")
-		return nil
-	}
-
 	return nil
 }
 
@@ -2547,11 +2546,15 @@ func (r *Reconciler) reconcileReplicaCreateBackup(ctx context.Context,
 		replicaRepoReady = (condition.Status == metav1.ConditionTrue)
 	}
 
-	// get pod name and container name as needed to exec into the proper pod and create
-	// the pgBackRest backup
-	_, containerName, err := getPGBackRestExecSelector(postgresCluster, replicaCreateRepo)
-	if err != nil {
-		return errors.WithStack(err)
+	// TODO: Since we now only exec into the repo host when backing up to a local volume and
+	// run the backup in the job pod when backing up to a cloud-based repo, we should consider
+	// using a different value than the container name for the "pgbackrest-config" annotation
+	// that we attach to these backups
+	var containerName string
+	if replicaCreateRepo.Volume != nil {
+		containerName = naming.PGBackRestRepoContainerName
+	} else {
+		containerName = naming.ContainerDatabase
 	}
 
 	// determine if the dedicated repository host is ready using the repo host ready status
@@ -2603,10 +2606,10 @@ func (r *Reconciler) reconcileReplicaCreateBackup(ctx context.Context,
 		}
 	}
 
-	dedicatedEnabled := pgbackrest.RepoHostVolumeDefined(postgresCluster)
 	// return if no job has been created and the replica repo or the dedicated
 	// repo host is not ready
-	if job == nil && ((dedicatedEnabled && !dedicatedRepoReady) || !replicaRepoReady) {
+	if job == nil && ((pgbackrest.RepoHostVolumeDefined(postgresCluster) && !dedicatedRepoReady) ||
+		!replicaRepoReady) {
 		return nil
 	}
 
@@ -2817,27 +2820,6 @@ func (r *Reconciler) reconcileStanzaCreate(ctx context.Context,
 	return false, nil
 }
 
-// getPGBackRestExecSelector returns a selector and container name that allows the proper
-// Pod (along with a specific container within it) to be found within the Kubernetes
-// cluster as needed to exec into the container and run a pgBackRest command.
-func getPGBackRestExecSelector(postgresCluster *v1beta1.PostgresCluster,
-	repo v1beta1.PGBackRestRepo) (labels.Selector, string, error) {
-
-	var err error
-	var podSelector labels.Selector
-	var containerName string
-
-	if repo.Volume != nil {
-		podSelector = naming.PGBackRestDedicatedSelector(postgresCluster.GetName())
-		containerName = naming.PGBackRestRepoContainerName
-	} else {
-		podSelector, err = naming.AsSelector(naming.ClusterPrimary(postgresCluster.GetName()))
-		containerName = naming.ContainerDatabase
-	}
-
-	return podSelector, containerName, err
-}
-
 // getRepoHostStatus is responsible for returning the pgBackRest status for the
 // provided pgBackRest repository host
 func getRepoHostStatus(repoHost *appsv1.StatefulSet) *v1beta1.RepoHostStatus {

internal/controller/postgrescluster/pgbackrest_test.go

@@ -887,52 +887,6 @@ func TestReconcileStanzaCreate(t *testing.T) {
 	}
 }
 
-func TestGetPGBackRestExecSelector(t *testing.T) {
-
-	testCases := []struct {
-		cluster           *v1beta1.PostgresCluster
-		repo              v1beta1.PGBackRestRepo
-		desc              string
-		expectedSelector  string
-		expectedContainer string
-	}{{
-		desc: "volume repo defined dedicated repo host enabled",
-		cluster: &v1beta1.PostgresCluster{
-			ObjectMeta: metav1.ObjectMeta{Name: "hippo"},
-		},
-		repo: v1beta1.PGBackRestRepo{
-			Name:   "repo1",
-			Volume: &v1beta1.RepoPVC{},
-		},
-		expectedSelector: "postgres-operator.crunchydata.com/cluster=hippo," +
-			"postgres-operator.crunchydata.com/pgbackrest=," +
-			"postgres-operator.crunchydata.com/pgbackrest-dedicated=",
-		expectedContainer: "pgbackrest",
-	}, {
-		desc: "cloud repo defined no repo host enabled",
-		cluster: &v1beta1.PostgresCluster{
-			ObjectMeta: metav1.ObjectMeta{Name: "hippo"},
-		},
-		repo: v1beta1.PGBackRestRepo{
-			Name: "repo1",
-			S3:   &v1beta1.RepoS3{},
-		},
-		expectedSelector: "postgres-operator.crunchydata.com/cluster=hippo," +
-			"postgres-operator.crunchydata.com/instance," +
-			"postgres-operator.crunchydata.com/role=master",
-		expectedContainer: "database",
-	}}
-
-	for _, tc := range testCases {
-		t.Run(tc.desc, func(t *testing.T) {
-			selector, container, err := getPGBackRestExecSelector(tc.cluster, tc.repo)
-			assert.NilError(t, err)
-			assert.Assert(t, selector.String() == tc.expectedSelector)
-			assert.Assert(t, container == tc.expectedContainer)
-		})
-	}
-}
-
 func TestReconcileReplicaCreateBackup(t *testing.T) {
 	// Garbage collector cleans up test resources before the test completes
 	if strings.EqualFold(os.Getenv("USE_EXISTING_CLUSTER"), "true") {

internal/pgbackrest/config.go

@@ -38,6 +38,10 @@ const (
 	// repository host
 	CMRepoKey = "pgbackrest_repo.conf"
 
+	// CMCloudRepoKey is the name of the pgBackRest configuration file used by backup jobs
+	// for cloud repos
+	CMCloudRepoKey = "pgbackrest_cloud.conf"
+
 	// configDirectory is the pgBackRest configuration directory.
 	configDirectory = "/etc/pgbackrest/conf.d"
 
@@ -69,6 +73,7 @@ const (
 // pgbackrest_job.conf is used by certain jobs, such as stanza create and backup
 // pgbackrest_primary.conf is used by the primary database pod
 // pgbackrest_repo.conf is used by the pgBackRest repository pod
+// pgbackrest_cloud.conf is used by cloud repo backup jobs
 func CreatePGBackRestConfigMapIntent(ctx context.Context, postgresCluster *v1beta1.PostgresCluster,
 	repoHostName, configHash, serviceName, serviceNamespace string,
 	instanceNames []string) (*corev1.ConfigMap, error) {
@@ -96,7 +101,6 @@ func CreatePGBackRestConfigMapIntent(ctx context.Context, postgresCluster *v1bet
 	// create an empty map for the config data
 	initialize.Map(&cm.Data)
 
-	addDedicatedHost := RepoHostVolumeDefined(postgresCluster)
 	pgdataDir := postgres.DataDirectory(postgresCluster)
 	// Port will always be populated, since the API will set a default of 5432 if not provided
 	pgPort := *postgresCluster.Spec.Port
@@ -113,12 +117,10 @@ func CreatePGBackRestConfigMapIntent(ctx context.Context, postgresCluster *v1bet
 	// PostgreSQL instances that have not rolled out expect to mount a server
 	// config file. Always populate that file so those volumes stay valid and
 	// Kubernetes propagates their contents to those pods.
-	cm.Data[serverConfigMapKey] = ""
-
-	if addDedicatedHost && repoHostName != "" {
-		cm.Data[serverConfigMapKey] = iniGeneratedWarning +
-			serverConfig(postgresCluster).String()
+	cm.Data[serverConfigMapKey] = iniGeneratedWarning +
+		serverConfig(postgresCluster).String()
 
+	if RepoHostVolumeDefined(postgresCluster) && repoHostName != "" {
 		cm.Data[CMRepoKey] = iniGeneratedWarning +
 			populateRepoHostConfigurationMap(
 				serviceName, serviceNamespace,
@@ -129,8 +131,7 @@ func CreatePGBackRestConfigMapIntent(ctx context.Context, postgresCluster *v1bet
 				postgresCluster.Spec.Backups.PGBackRest.Global,
 			).String()
 
-		if RepoHostVolumeDefined(postgresCluster) &&
-			collector.OpenTelemetryLogsOrMetricsEnabled(ctx, postgresCluster) {
+		if collector.OpenTelemetryLogsOrMetricsEnabled(ctx, postgresCluster) {
 
 			err = collector.AddToConfigMap(ctx, collector.NewConfigForPgBackrestRepoHostPod(
 				ctx,
@@ -156,6 +157,18 @@ func CreatePGBackRestConfigMapIntent(ctx context.Context, postgresCluster *v1bet
 		}
 	}
 
+	if CloudRepoDefined(postgresCluster) {
+		cm.Data[CMCloudRepoKey] = iniGeneratedWarning +
+			populateCloudRepoConfigurationMap(
+				serviceName, serviceNamespace, pgdataDir,
+				config.FetchKeyCommand(&postgresCluster.Spec),
+				strconv.Itoa(postgresCluster.Spec.PostgresVersion),
+				pgPort, instanceNames,
+				postgresCluster.Spec.Backups.PGBackRest.Repos,
+				postgresCluster.Spec.Backups.PGBackRest.Global,
+			).String()
+	}
+
 	cm.Data[ConfigHashKey] = configHash
 
 	return cm, err
@@ -504,6 +517,64 @@ func populateRepoHostConfigurationMap(
 	}
 }
 
+func populateCloudRepoConfigurationMap(
+	serviceName, serviceNamespace, pgdataDir,
+	fetchKeyCommand, postgresVersion string,
+	pgPort int32, pgHosts []string, repos []v1beta1.PGBackRestRepo,
+	globalConfig map[string]string,
+) iniSectionSet {
+
+	global := iniMultiSet{}
+	stanza := iniMultiSet{}
+
+	for _, repo := range repos {
+		if repo.Volume != nil {
+			continue
+		}
+
+		global.Set(repo.Name+"-path", defaultRepo1Path+repo.Name)
+
+		for option, val := range getExternalRepoConfigs(repo) {
+			global.Set(option, val)
+		}
+	}
+
+	global.Set("log-level-file", "off")
+
+	for option, val := range globalConfig {
+		global.Set(option, val)
+	}
+
+	// set the configs for all PG hosts
+	for i, pgHost := range pgHosts {
+		// TODO(cbandy): pass a FQDN in already.
+		pgHostFQDN := pgHost + "-0." +
+			serviceName + "." + serviceNamespace + ".svc." +
+			naming.KubernetesClusterDomain(context.Background())
+
+		stanza.Set(fmt.Sprintf("pg%d-host", i+1), pgHostFQDN)
+		stanza.Set(fmt.Sprintf("pg%d-host-type", i+1), "tls")
+		stanza.Set(fmt.Sprintf("pg%d-host-ca-file", i+1), certAuthorityAbsolutePath)
+		stanza.Set(fmt.Sprintf("pg%d-host-cert-file", i+1), certClientAbsolutePath)
+		stanza.Set(fmt.Sprintf("pg%d-host-key-file", i+1), certClientPrivateKeyAbsolutePath)
+
+		stanza.Set(fmt.Sprintf("pg%d-path", i+1), pgdataDir)
+		stanza.Set(fmt.Sprintf("pg%d-port", i+1), fmt.Sprint(pgPort))
+		stanza.Set(fmt.Sprintf("pg%d-socket-path", i+1), postgres.SocketDirectory)
+
+		if fetchKeyCommand != "" {
+			stanza.Set("archive-header-check", "n")
+			stanza.Set("page-header-check", "n")
+			stanza.Set("pg-version-force", postgresVersion)
+		}
+	}
+
+	return iniSectionSet{
+		"global":          global,
+		DefaultStanzaName: stanza,
+	}
+}
+
 // getExternalRepoConfigs returns a map containing the configuration settings for an external
 // pgBackRest repository as defined in the PostgresCluster spec
 func getExternalRepoConfigs(repo v1beta1.PGBackRestRepo) map[string]string {