Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 1 addition & 17 deletions internal/controller/cluster_create.go
Original file line number Diff line number Diff line change
Expand Up @@ -1242,27 +1242,11 @@ func (r *ClusterReconciler) joinReplicaInstance(
) (ctrl.Result, error) {
contextLogger := log.FromContext(ctx)

var backupList apiv1.BackupList
if err := r.List(ctx, &backupList,
client.MatchingFields{clusterNameField: cluster.Name},
client.InNamespace(cluster.Namespace),
); err != nil {
contextLogger.Error(err, "Error while getting backup list, when bootstrapping a new replica")
return ctrl.Result{}, err
}

job := specs.JoinReplicaInstance(*cluster, nodeSerial)

// If we can bootstrap this replica from a pre-existing source, we do it
storageSource := persistentvolumeclaim.GetCandidateStorageSourceForReplica(ctx, cluster, backupList)
if storageSource != nil {
job = specs.RestoreReplicaInstance(*cluster, nodeSerial)
}

contextLogger.Info("Creating new Job",
"job", job.Name,
"primary", false,
"storageSource", storageSource,
"role", job.Spec.Template.Labels[utils.JobRoleLabelName],
)

Expand Down Expand Up @@ -1304,7 +1288,7 @@ func (r *ClusterReconciler) joinReplicaInstance(
ctx,
r.Client,
cluster,
storageSource,
nil,
nodeSerial,
); err != nil {
return ctrl.Result{}, fmt.Errorf("cannot create replica instance PVCs: %w", err)
Expand Down
125 changes: 0 additions & 125 deletions pkg/reconciler/persistentvolumeclaim/storagesource.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,6 @@ SPDX-License-Identifier: Apache-2.0
package persistentvolumeclaim

import (
"context"

"github.com/cloudnative-pg/machinery/pkg/log"
volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/utils/ptr"
Expand Down Expand Up @@ -56,128 +53,6 @@ func GetCandidateStorageSourceForPrimary(
return getCandidateSourceFromClusterDefinition(cluster)
}

// GetCandidateStorageSourceForReplica gets the candidate storage source
// to be used to create a replica PVC
func GetCandidateStorageSourceForReplica(
ctx context.Context,
cluster *apiv1.Cluster,
backupList apiv1.BackupList,
) *StorageSource {
// We can't use a Backup to create a replica when:
//
// 1. we don't have WAL archiving, because the backup may be old
// and the primary may not have the WAL files needed for the
// new replica to be in-sync
//
// 2. we need two different WAL object stores, because we cannot
// access them at the same time. This can happen when we have:
//
// - the object store where we upload the WAL files
// i.e. `.spec.backup.barmanObjectStore`
//
// - the object store where were we aed WAL files to create the
// bootstrap primary instance
// i.e. `.spec.externalClusters[i].barmanObjectStore` and
// `.spec.bootstrap.recovery.source`
//
// This is true only for the backup that was used to bootstrap
// the cluster itself. Other backups are fine because the required
// WALs have been archived in the cluster object store.

// Unless WAL archiving is active (via BarmanObjectStore or a WAL-archiver plugin),
// we can't recover a replica from a backup
walArchivingActive := (cluster.Spec.Backup != nil && cluster.Spec.Backup.BarmanObjectStore != nil) ||
cluster.GetEnabledWALArchivePluginName() != ""
if !walArchivingActive {
return nil
}

if result := getCandidateSourceFromBackupList(
ctx,
cluster,
backupList,
); result != nil {
return result
}

// We support one and only one object store, see comment at the beginning
// of this function
if cluster.Spec.Bootstrap != nil &&
cluster.Spec.Bootstrap.Recovery != nil &&
len(cluster.Spec.Bootstrap.Recovery.Source) > 0 {
return nil
}

// Try using the backup the Cluster has been bootstrapped from
return getCandidateSourceFromClusterDefinition(cluster)
}

// getCandidateSourceFromBackupList gets a candidate storage source
// given a backup list
func getCandidateSourceFromBackupList(
ctx context.Context,
cluster *apiv1.Cluster,
backupList apiv1.BackupList,
) *StorageSource {
contextLogger := log.FromContext(ctx)

majorVersion, err := cluster.GetPostgresqlMajorVersion()
if err != nil {
contextLogger.Warning(
"unable to determine cluster major version; skipping backup as a recovery source",
"error", err.Error(),
)
return nil
}

isCorrectMajorVersion := func(backup *apiv1.Backup) bool {
// If we don't have image info, we can't determine the cluster version reliably; skip enforcement
if cluster.Status.PGDataImageInfo == nil {
return true
}

backupMajorVersion := backup.Status.MajorVersion
if backupMajorVersion == 0 {
contextLogger.Warning(
"majorVersion on backup status is not populated, cannot use it as a recovery source.",
)
return false
}

return majorVersion == backupMajorVersion
}

backupList.SortByReverseCreationTime()
for idx := range backupList.Items {
backup := &backupList.Items[idx]

if !backup.IsCompletedVolumeSnapshot() {
contextLogger.Trace("skipping backup, not a valid storage source candidate")
continue
}

if backup.CreationTimestamp.Before(&cluster.CreationTimestamp) {
contextLogger.Info(
"skipping backup as a potential recovery storage source candidate because it was created before the Cluster object",
)
continue
}

if !isCorrectMajorVersion(backup) {
contextLogger.Info(
"skipping backup as a potential recovery storage source candidate because of major version mismatch",
)
continue
}

contextLogger.Debug("found a backup that is a valid storage source candidate")

return getCandidateSourceFromBackup(backup)
}

return nil
}

func getCandidateSourceFromBackup(backup *apiv1.Backup) *StorageSource {
var result StorageSource
for _, element := range backup.Status.BackupSnapshotStatus.Elements {
Expand Down
Loading
Loading