diff --git a/internal/controller/cluster_create.go b/internal/controller/cluster_create.go index d698983c3..088af7ed8 100644 --- a/internal/controller/cluster_create.go +++ b/internal/controller/cluster_create.go @@ -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], ) @@ -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) diff --git a/pkg/reconciler/persistentvolumeclaim/storagesource.go b/pkg/reconciler/persistentvolumeclaim/storagesource.go index 75a64b8fc..a4ead3983 100644 --- a/pkg/reconciler/persistentvolumeclaim/storagesource.go +++ b/pkg/reconciler/persistentvolumeclaim/storagesource.go @@ -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" @@ -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 { diff --git a/pkg/reconciler/persistentvolumeclaim/storagesource_test.go b/pkg/reconciler/persistentvolumeclaim/storagesource_test.go index 934a6c8d6..412f2c28a 100644 --- a/pkg/reconciler/persistentvolumeclaim/storagesource_test.go +++ b/pkg/reconciler/persistentvolumeclaim/storagesource_test.go @@ -20,15 +20,7 @@ SPDX-License-Identifier: Apache-2.0 package persistentvolumeclaim import ( - "context" - "time" - - corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/utils/ptr" - apiv1 "github.com/xataio/xata-cnpg/api/v1" - "github.com/xataio/xata-cnpg/pkg/utils" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -52,395 +44,3 @@ var _ = Describe("Storage configuration", func() { Expect(err).ToNot(HaveOccurred()) }) }) - -var _ = Describe("Storage source", func() { - pgDataSnapshotVolumeName := "pgdata-snapshot" - pgWalSnapshotVolumeName := "pgwal-snapshot" - clusterWithBootstrapSnapshot := &apiv1.Cluster{ - Spec: apiv1.ClusterSpec{ - StorageConfiguration: apiv1.StorageConfiguration{}, - WalStorage: &apiv1.StorageConfiguration{}, - Bootstrap: &apiv1.BootstrapConfiguration{ - Recovery: &apiv1.BootstrapRecovery{ - VolumeSnapshots: &apiv1.DataSource{ - Storage: corev1.TypedLocalObjectReference{ - Name: pgDataSnapshotVolumeName, - Kind: apiv1.VolumeSnapshotKind, - APIGroup: ptr.To("snapshot.storage.k8s.io"), - }, - WalStorage: &corev1.TypedLocalObjectReference{ - Name: pgWalSnapshotVolumeName, - Kind: apiv1.VolumeSnapshotKind, - APIGroup: ptr.To("snapshot.storage.k8s.io"), - }, - }, - }, - }, - Backup: &apiv1.BackupConfiguration{ - BarmanObjectStore: &apiv1.BarmanObjectStoreConfiguration{ - DestinationPath: "s3://test", - }, - }, - }, - } - - clusterWithBackupSection := &apiv1.Cluster{ - Spec: apiv1.ClusterSpec{ - StorageConfiguration: apiv1.StorageConfiguration{}, - WalStorage: &apiv1.StorageConfiguration{}, - Backup: &apiv1.BackupConfiguration{ - BarmanObjectStore: &apiv1.BarmanObjectStoreConfiguration{ - DestinationPath: "s3://test", - }, - }, - }, - } - - clusterWithPluginOnly := &apiv1.Cluster{ - Spec: apiv1.ClusterSpec{ - StorageConfiguration: apiv1.StorageConfiguration{}, - WalStorage: &apiv1.StorageConfiguration{}, - Backup: nil, - Plugins: []apiv1.PluginConfiguration{ - { - Name: "test-wal-archiver", - IsWALArchiver: ptr.To(true), - }, - }, - }, - } - - backupList := apiv1.BackupList{ - Items: []apiv1.Backup{ - { - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(time.Now()), - Name: "completed-backup", - }, - Spec: apiv1.BackupSpec{ - Method: apiv1.BackupMethodVolumeSnapshot, - }, - Status: apiv1.BackupStatus{ - Phase: apiv1.BackupPhaseCompleted, - BackupSnapshotStatus: apiv1.BackupSnapshotStatus{ - Elements: []apiv1.BackupSnapshotElementStatus{ - { - Name: "completed-backup", - Type: string(utils.PVCRolePgData), - }, - }, - }, - }, - }, - }, - } - - When("bootstrapping from a VolumeSnapshot", func() { - When("we don't have backups", func() { - When("there's no source WAL archive", func() { - It("should return the correct source when choosing pgdata", func(ctx context.Context) { - source, err := NewPgDataCalculator().GetSource(GetCandidateStorageSourceForReplica( - ctx, clusterWithBootstrapSnapshot, apiv1.BackupList{})) - Expect(err).ToNot(HaveOccurred()) - Expect(source).ToNot(BeNil()) - Expect(source.Name).To(Equal(pgDataSnapshotVolumeName)) - }) - - It("should return the correct source when choosing pgwal", func(ctx context.Context) { - source, err := NewPgWalCalculator().GetSource(GetCandidateStorageSourceForReplica( - ctx, clusterWithBootstrapSnapshot, apiv1.BackupList{})) - Expect(err).ToNot(HaveOccurred()) - Expect(source).ToNot(BeNil()) - Expect(source.Name).To(Equal(pgWalSnapshotVolumeName)) - }) - }) - - When("there's a source WAL archive", func() { - It("should return an empty storage source", func(ctx context.Context) { - clusterSourceWALArchive := clusterWithBootstrapSnapshot.DeepCopy() - clusterSourceWALArchive.Spec.Bootstrap.Recovery.Source = "test" - source, err := NewPgDataCalculator().GetSource(GetCandidateStorageSourceForReplica( - ctx, - clusterSourceWALArchive, - apiv1.BackupList{}, - )) - Expect(err).ToNot(HaveOccurred()) - Expect(source).To(BeNil()) - }) - }) - }) - - When("we have backups", func() { - It("should return the correct backup", func(ctx context.Context) { - source, err := NewPgDataCalculator().GetSource(GetCandidateStorageSourceForReplica( - ctx, - clusterWithBootstrapSnapshot, - backupList, - )) - Expect(err).ToNot(HaveOccurred()) - Expect(source).ToNot(BeNil()) - Expect(source.Name).To(Equal("completed-backup")) - }) - }) - }) - - When("not bootstrapping from a VolumeSnapshot with no backups", func() { - It("should return an empty storage source", func(ctx context.Context) { - source, err := NewPgDataCalculator().GetSource(GetCandidateStorageSourceForReplica( - ctx, - clusterWithBackupSection, - apiv1.BackupList{}, - )) - Expect(err).ToNot(HaveOccurred()) - Expect(source).To(BeNil()) - }) - }) - - When("not bootstrapping from a VolumeSnapshot with backups", func() { - It("should return the backup as storage source", func(ctx context.Context) { - source, err := NewPgDataCalculator().GetSource(GetCandidateStorageSourceForReplica( - ctx, - clusterWithBackupSection, - backupList, - )) - Expect(err).ToNot(HaveOccurred()) - Expect(source).ToNot(BeNil()) - Expect(source.Name).To(Equal("completed-backup")) - }) - - It("should return the backup as storage source when WAL archiving is via plugin only", func(ctx context.Context) { - source, err := NewPgDataCalculator().GetSource(GetCandidateStorageSourceForReplica( - ctx, - clusterWithPluginOnly, - backupList, - )) - Expect(err).ToNot(HaveOccurred()) - Expect(source).ToNot(BeNil()) - Expect(source.Name).To(Equal("completed-backup")) - }) - }) - - When("there's no WAL archiving", func() { - It("should return an empty storage source", func(ctx context.Context) { - clusterNoWalArchiving := clusterWithBackupSection.DeepCopy() - clusterNoWalArchiving.Spec.Backup = nil - - source, err := NewPgDataCalculator().GetSource(GetCandidateStorageSourceForReplica( - ctx, - clusterNoWalArchiving, - backupList, - )) - Expect(err).ToNot(HaveOccurred()) - Expect(source).To(BeNil()) - }) - }) -}) - -var _ = Describe("candidate backups", func() { - now := time.Now() - completedBackup := apiv1.Backup{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(now), - Name: "completed-backup", - }, - Spec: apiv1.BackupSpec{ - Method: apiv1.BackupMethodVolumeSnapshot, - }, - Status: apiv1.BackupStatus{ - Phase: apiv1.BackupPhaseCompleted, - BackupSnapshotStatus: apiv1.BackupSnapshotStatus{ - Elements: []apiv1.BackupSnapshotElementStatus{ - { - Name: "completed-backup", - Type: string(utils.PVCRolePgData), - }, - }, - }, - }, - } - - oldCompletedBackup := apiv1.Backup{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(now.Add(-5 * time.Hour)), - Name: "old-backup", - }, - Spec: apiv1.BackupSpec{ - Method: apiv1.BackupMethodVolumeSnapshot, - }, - Status: apiv1.BackupStatus{ - Phase: apiv1.BackupPhaseCompleted, - BackupSnapshotStatus: apiv1.BackupSnapshotStatus{ - Elements: []apiv1.BackupSnapshotElementStatus{ - { - Name: "bad-name", - Type: string(utils.PVCRolePgData), - }, - }, - }, - }, - } - - nonCompletedBackup := apiv1.Backup{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(now), - Name: "non-completed-backup", - }, - Spec: apiv1.BackupSpec{ - Method: apiv1.BackupMethodVolumeSnapshot, - }, - Status: apiv1.BackupStatus{}, - } - - objectStoreBackup := apiv1.Backup{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(now), - Name: "object-store-backup", - }, - Spec: apiv1.BackupSpec{ - Method: apiv1.BackupMethodBarmanObjectStore, - }, - Status: apiv1.BackupStatus{ - Phase: apiv1.BackupPhaseCompleted, - }, - } - - It("takes the most recent candidate backup as source", func(ctx context.Context) { - backupList := apiv1.BackupList{ - Items: []apiv1.Backup{ - objectStoreBackup, - nonCompletedBackup, - oldCompletedBackup, - completedBackup, - }, - } - backupList.SortByReverseCreationTime() - - cluster := &apiv1.Cluster{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(time.Now().Add(-1 * time.Hour)), - }, - } - source := getCandidateSourceFromBackupList(ctx, cluster, backupList) - Expect(source).ToNot(BeNil()) - Expect(source.DataSource.Name).To(Equal("completed-backup")) - }) - - It("will refuse to use automatically use snapshots if they are older than the Cluster", func(ctx context.Context) { - backupList := apiv1.BackupList{ - Items: []apiv1.Backup{ - objectStoreBackup, - nonCompletedBackup, - oldCompletedBackup, - completedBackup, - }, - } - backupList.SortByReverseCreationTime() - - cluster := &apiv1.Cluster{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(time.Now().Add(1 * time.Hour)), - }, - } - source := getCandidateSourceFromBackupList(ctx, cluster, backupList) - Expect(source).To(BeNil()) - }) -}) - -var _ = Describe("major version filtering in candidate backup selection", func() { - makeCompletedSnapshotBackup := func(name string) apiv1.Backup { - return apiv1.Backup{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(time.Now()), - Name: name, - }, - Spec: apiv1.BackupSpec{ - Method: apiv1.BackupMethodVolumeSnapshot, - }, - Status: apiv1.BackupStatus{ - Phase: apiv1.BackupPhaseCompleted, - BackupSnapshotStatus: apiv1.BackupSnapshotStatus{ - Elements: []apiv1.BackupSnapshotElementStatus{ - { - Name: "completed-backup", - Type: string(utils.PVCRolePgData), - }, - }, - }, - }, - } - } - - It("does not filter by major version when PGDataImageInfo is nil", func(ctx context.Context) { - backup := makeCompletedSnapshotBackup("no-major-version") - // Explicitly leave MajorVersion=0 and no annotation to simulate missing version - backupList := apiv1.BackupList{Items: []apiv1.Backup{backup}} - - cluster := &apiv1.Cluster{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(time.Now().Add(-1 * time.Hour)), - }, - Spec: apiv1.ClusterSpec{ - Backup: &apiv1.BackupConfiguration{ - BarmanObjectStore: &apiv1.BarmanObjectStoreConfiguration{ - DestinationPath: "s3://test", - }, - }, - }, - // Status.PGDataImageInfo is nil here on purpose - } - - source := getCandidateSourceFromBackupList(ctx, cluster, backupList) - Expect(source).ToNot(BeNil()) - Expect(source.DataSource.Name).To(Equal("completed-backup")) - }) - - It("skips backup when PGDataImageInfo is set and backup major version is missing", func(ctx context.Context) { - backup := makeCompletedSnapshotBackup("missing-major-version") - // MajorVersion=0 and no annotation, will be rejected when PGDataImageInfo != nil - backupList := apiv1.BackupList{Items: []apiv1.Backup{backup}} - - cluster := &apiv1.Cluster{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(time.Now().Add(-1 * time.Hour)), - }, - Spec: apiv1.ClusterSpec{ - Backup: &apiv1.BackupConfiguration{ - BarmanObjectStore: &apiv1.BarmanObjectStoreConfiguration{ - DestinationPath: "s3://test", - }, - }, - }, - Status: apiv1.ClusterStatus{ - PGDataImageInfo: &apiv1.ImageInfo{}, - }, - } - - source := getCandidateSourceFromBackupList(ctx, cluster, backupList) - Expect(source).To(BeNil()) - }) - - It("skips backup when PGDataImageInfo is set and major version mismatches", func(ctx context.Context) { - backup := makeCompletedSnapshotBackup("mismatching-major-version") - backup.Status.MajorVersion = 99 // intentionally mismatching - backupList := apiv1.BackupList{Items: []apiv1.Backup{backup}} - - cluster := &apiv1.Cluster{ - ObjectMeta: metav1.ObjectMeta{ - CreationTimestamp: metav1.NewTime(time.Now().Add(-1 * time.Hour)), - }, - Spec: apiv1.ClusterSpec{ - Backup: &apiv1.BackupConfiguration{ - BarmanObjectStore: &apiv1.BarmanObjectStoreConfiguration{ - DestinationPath: "s3://test", - }, - }, - }, - Status: apiv1.ClusterStatus{ - PGDataImageInfo: &apiv1.ImageInfo{}, - }, - } - - source := getCandidateSourceFromBackupList(ctx, cluster, backupList) - Expect(source).To(BeNil()) - }) -}) diff --git a/pkg/specs/jobs.go b/pkg/specs/jobs.go index 45542cc6d..ff42f3ee0 100644 --- a/pkg/specs/jobs.go +++ b/pkg/specs/jobs.go @@ -279,23 +279,6 @@ func JoinReplicaInstance(cluster apiv1.Cluster, nodeSerial int) *batchv1.Job { return CreatePrimaryJob(cluster, nodeSerial, jobRoleJoin, initCommand) } -// RestoreReplicaInstance creates a new PostgreSQL replica starting from a volume snapshot backup -func RestoreReplicaInstance(cluster apiv1.Cluster, nodeSerial int) *batchv1.Job { - commonFlags := buildCommonInitJobFlags(cluster) - initCommand := make([]string, 0, 4+len(commonFlags)) - initCommand = append(initCommand, - "/controller/manager", - "instance", - "restoresnapshot", - "--immediate", - ) - - initCommand = append(initCommand, commonFlags...) - - job := CreatePrimaryJob(cluster, nodeSerial, jobRoleSnapshotRecovery, initCommand) - return job -} - func buildCommonInitJobFlags(cluster apiv1.Cluster) []string { var flags []string diff --git a/tests/e2e/volume_snapshot_test.go b/tests/e2e/volume_snapshot_test.go index 477b71262..c399adce1 100644 --- a/tests/e2e/volume_snapshot_test.go +++ b/tests/e2e/volume_snapshot_test.go @@ -34,7 +34,6 @@ import ( k8client "sigs.k8s.io/controller-runtime/pkg/client" apiv1 "github.com/xataio/xata-cnpg/api/v1" - "github.com/xataio/xata-cnpg/pkg/specs" "github.com/xataio/xata-cnpg/pkg/utils" "github.com/xataio/xata-cnpg/tests" "github.com/xataio/xata-cnpg/tests/utils/backups" @@ -800,85 +799,6 @@ var _ = Describe("Verify Volume Snapshot", }) }) - It("should scale up the cluster with volume snapshot", func() { - // insert some data after the snapshot is taken, we want to verify the data exists in - // the new pod when cluster scaled up - By("inserting more test data and creating WALs on the cluster snapshotted", func() { - forward, conn, err := postgres.ForwardPSQLConnection( - env.Ctx, - env.Client, - env.Interface, - env.RestClientConfig, - namespace, - clusterToSnapshotName, - postgres.AppDBName, - apiv1.ApplicationUserSecretSuffix, - ) - defer func() { - _ = conn.Close() - forward.Close() - }() - Expect(err).ToNot(HaveOccurred()) - // Insert 2 more rows which we expect not to be present at the end of the recovery - insertRecordIntoTable(tableName, 5, conn) - insertRecordIntoTable(tableName, 6, conn) - - // Close and archive the current WAL file - AssertArchiveWalOnMinio(namespace, clusterToSnapshotName, clusterToSnapshotName) - }) - - // reuse the snapshot taken from the clusterToSnapshot cluster - By("fetching the volume snapshots", func() { - snapshotList, err := getSnapshots(backupTaken.Name, clusterToSnapshotName, namespace) - Expect(err).ToNot(HaveOccurred()) - Expect(snapshotList.Items).To(HaveLen(len(backupTaken.Status.BackupSnapshotStatus.Elements))) - - envVars := storage.EnvVarsForSnapshots{ - DataSnapshot: snapshotDataEnv, - WalSnapshot: snapshotWalEnv, - } - err = storage.SetSnapshotNameAsEnv(&snapshotList, backupTaken, envVars) - Expect(err).ToNot(HaveOccurred()) - }) - - By("scale up the cluster", func() { - err := clusterutils.ScaleSize(env.Ctx, env.Client, namespace, clusterToSnapshotName, 3) - Expect(err).ToNot(HaveOccurred()) - }) - - By("checking the cluster is working", func() { - // Setting up a cluster with three pods is slow, usually 200-600s - AssertClusterIsReady(namespace, clusterToSnapshotName, testTimeouts[timeouts.ClusterIsReady], env) - }) - - By("checking the new replicas have been created using the snapshot", func() { - pvcList, err := storage.GetPVCList(env.Ctx, env.Client, namespace) - Expect(err).ToNot(HaveOccurred()) - for _, pvc := range pvcList.Items { - if pvc.Labels[utils.ClusterInstanceRoleLabelName] == specs.ClusterRoleLabelReplica && - pvc.Labels[utils.ClusterLabelName] == clusterToSnapshotName { - Expect(pvc.Spec.DataSource.Kind).To(Equal(apiv1.VolumeSnapshotKind)) - Expect(pvc.Spec.DataSourceRef.Kind).To(Equal(apiv1.VolumeSnapshotKind)) - } - } - }) - - // we need to verify the streaming replica continue works - By("verifying the correct data exists in the new pod of the scaled cluster", func() { - podList, err := clusterutils.GetReplicas(env.Ctx, env.Client, namespace, - clusterToSnapshotName) - Expect(err).ToNot(HaveOccurred()) - Expect(podList.Items).To(HaveLen(2)) - tableLocator := TableLocator{ - Namespace: namespace, - ClusterName: clusterToSnapshotName, - DatabaseName: postgres.AppDBName, - TableName: tableName, - } - AssertDataExpectedCount(env, tableLocator, 6) - }) - }) - It("should clean up unused backup connections", func() { By("setting a non-existing snapshotClass", func() { updateClusterSnapshotClass(namespace, clusterToSnapshotName, "wrongSnapshotClass")