Repository: cloudnative-pg/cloudnative-pg

Issue:
Kubernetes Volume Snapshots support
Currently, CloudNativePG supports object stores only for continuous backup, requiring that both WAL files and base backups reside in the same **backup object store**.

However, this technique is not adequate for very large database scenarios, especially on the recovery side with high values of RTO following a (rare) full disaster.

This epic ticket proposes the introduction in CloudNativePG of:

- Hot (online) physical base backups for Postgres databases through [Kubernetes native volume snapshots](https://kubernetes.io/docs/concepts/storage/volume-snapshots/)
- Full recovery and Point In Time Recovery (PITR) through Kubernetes native volume snapshots

Both capabilities need to be declarative.

Volume snapshotting was first introduced in [Kubernetes 1.12 as alpha](https://kubernetes.io/blog/2018/10/09/introducing-volume-snapshot-alpha-for-kubernetes/), promoted to [beta in 1.17](https://kubernetes.io/blog/2019/12/09/kubernetes-1-17-feature-cis-volume-snapshot-beta/), and [moved to GA in 1.20](https://kubernetes.io/blog/2020/12/10/kubernetes-1.20-volume-snapshot-moves-to-ga/). It’s now stable and widely available, and provides 3 custom resource definitions: `VolumeSnapshot`, `VolumeSnapshotContent` and `VolumeSnapshotClass`.

Thanks to the transparent support for both incremental and differential copy that volume snapshots provide through the underlying storage classes, this feature has two main benefits:

- make CloudNativePG excel in the Very Large DataBase (VLDB) context
- make CloudNativePG more “Kubernetes native”, by leveraging the Kubernetes API for volume snapshotting to be used both for backup and recovery operations, including across Kubernetes clusters, by delegating the complexity and management to the underlying storage classes

**Hot backup**'s been the default in PostgreSQL for 20 years now, and one of the core capabilities that guaranteed success to Postgres. It enables taking physical base backups from any instance in the Postgres cluster (either primary or standby) without shutting down the server, and it requires WAL archiving (currently only on object stores).

As part of this activity, we should also address cold (offline) physical base backups and "database snapshot" recovery situations. Opposed to hot backup, **cold backup** is a technique that enables the operator to also work in the worst case scenario where a WAL archive is not present and “standalone consistent database snapshots” are accepted: this technique basically fences a replica, takes a snapshot and restarts it, without disrupting the primary.

Volume snapshots can have the potential to become the main base backup method in the public cloud, as well as in some enterprise on-premise environments.

The main general idea of this feature is to extend the existing API of the operator for “Backup”, “ScheduledBackup” and “Cluster” CRDs to consider another way of taking physical base backups on top of the existing one based on object stores with Barman Cloud, as well as of recovering, in a declarative way.


Hints from issue comments:
@gbartolini: The initial release of volume snapshots (version 1.21.0) only supported cold backups, which required fencing of the instance. This limitation has been waived starting with version 1.21.1. Given the minimal impact of the change on the code, maintainers have decided to backport this feature immediately instead of waiting for version 1.22.0 to be out, and make online backups the default behavior on volume snapshots too. If you are planning to rely instead on cold backups, make sure you follow the instructions below.@krishnaGajabi: Does Cloudnative-PG leverage group volume snapshot?

@gbartolini  How does cloudnative pg take simulatenous volume snapshot if database is spread across multiple storage devices?. This is required to have crash consistent data.



Solution patch:
diff --git a/.github/pg_versions.json b/.github/pg_versions.json
index bda120a6..489f5cb3 100644
--- a/.github/pg_versions.json
+++ b/.github/pg_versions.json
@@ -1,7 +1,7 @@
 {
   "16": [
     "16.0",
-    "16.0-6"
+    "16.0-8"
   ],
   "15": [
     "15.4",
diff --git a/.wordlist-en-custom.txt b/.wordlist-en-custom.txt
index 4157ed81..d4f2b37d 100644
--- a/.wordlist-en-custom.txt
+++ b/.wordlist-en-custom.txt
@@ -19,6 +19,7 @@ Azurite
 BDR
 BackupConfiguration
 BackupFrom
+BackupLabelFile
 BackupList
 BackupMethod
 BackupPhase
@@ -226,6 +227,7 @@ ObjectMeta
 OngoingBackupStatus
 OngoingBackups
 OngoingSnapshotBackups
+OnlineConfiguration
 OnlineUpdateEnabled
 OnlineUpgrading
 OpenShift
@@ -369,6 +371,7 @@ TCP
 TLS
 TOC
 TODO
+TablespaceMapFile
 TemporaryData
 TimelineId
 TopologyKey
@@ -455,6 +458,7 @@ backport
 backported
 backupID
 backupId
+backupLabelFile
 backupName
 backupOwnerReference
 backupRetentionPolicy
@@ -838,6 +842,7 @@ oc
 ol
 olm
 ongoingBackups
+onlineConfiguration
 onlineUpdateEnabled
 openldap
 openshift
@@ -1072,6 +1077,7 @@ systemd
 sysv
 tAc
 tablespace
+tablespaceMapFile
 tablespaces
 targetImmediate
 targetLSN
@@ -1131,6 +1137,7 @@ volumeMounts
 volumeSnapshot
 volumeSnapshots
 volumesnapshot
+waitForArchive
 wal
 walClassName
 walSegmentSize
diff --git a/api/v1/backup_types.go b/api/v1/backup_types.go
index 221e2e2f..c469b0d0 100644
--- a/api/v1/backup_types.go
+++ b/api/v1/backup_types.go
@@ -185,6 +185,14 @@ type BackupStatus struct {
 	// +optional
 	CommandError string `json:"commandError,omitempty"`
 
+	// Backup label file content as returned by Postgres in case of online (hot) backups
+	// +optional
+	BackupLabelFile []byte `json:"backupLabelFile,omitempty"`
+
+	// Tablespace map file content as returned by Postgres in case of online (hot) backups
+	// +optional
+	TablespaceMapFile []byte `json:"tablespaceMapFile,omitempty"`
+
 	// Information to identify the instance where the backup has been taken from
 	// +optional
 	InstanceID *InstanceID `json:"instanceID,omitempty"`
@@ -196,6 +204,9 @@ type BackupStatus struct {
 	// The backup method being used
 	// +optional
 	Method BackupMethod `json:"method,omitempty"`
+
+	// Whether the backup was online/hot (`true`) or offline/cold (`false`)
+	Online *bool `json:"online,omitempty"`
 }
 
 // InstanceID contains the information to identify an instance
@@ -292,6 +303,16 @@ func (backupStatus *BackupStatus) IsDone() bool {
 	return backupStatus.Phase == BackupPhaseCompleted || backupStatus.Phase == BackupPhaseFailed
 }
 
+// GetOnline tells whether this backup was taken while the database
+// was up
+func (backupStatus *BackupStatus) GetOnline() bool {
+	if backupStatus.Online == nil {
+		return false
+	}
+
+	return *backupStatus.Online
+}
+
 // IsCompletedVolumeSnapshot checks if a backup is completed using the volume snapshot method.
 // It returns true if the backup's method is BackupMethodVolumeSnapshot and its status phase is BackupPhaseCompleted.
 // Otherwise, it returns false.
diff --git a/api/v1/backup_webhook.go b/api/v1/backup_webhook.go
index 6572fae6..c417ffb8 100644
--- a/api/v1/backup_webhook.go
+++ b/api/v1/backup_webhook.go
@@ -82,7 +82,7 @@ func (r *Backup) validate() field.ErrorList {
 	var result field.ErrorList
 
 	if r.Spec.Method == BackupMethodVolumeSnapshot && !utils.HaveVolumeSnapshot() {
-		return append(result, field.Invalid(
+		result = append(result, field.Invalid(
 			field.NewPath("spec", "method"),
 			r.Spec.Method,
 			"Cannot use volumeSnapshot backup method due to missing "+
diff --git a/api/v1/cluster_types.go b/api/v1/cluster_types.go
index 4ba248ae..55cee1d8 100644
--- a/api/v1/cluster_types.go
+++ b/api/v1/cluster_types.go
@@ -149,6 +149,51 @@ type VolumeSnapshotConfiguration struct {
 	// +kubebuilder:validation:Enum:=none;cluster;backup
 	// +kubebuilder:default:=none
 	SnapshotOwnerReference SnapshotOwnerReference `json:"snapshotOwnerReference,omitempty"`
+
+	// Whether the default type of backup with volume snapshots is
+	// online/hot (`true`, default) or offline/cold (`false`)
+	// +optional
+	// +kubebuilder:default:=true
+	Online *bool `json:"online,omitempty"`
+
+	// Configuration parameters to control the online/hot backup with volume snapshots
+	// +kubebuilder:default:={waitForArchive:true,immediateCheckpoint:false}
+	// +optional
+	OnlineConfiguration OnlineConfiguration `json:"onlineConfiguration,omitempty"`
+}
+
+// GetOnline tells whether this volume snapshot configuration allows
+// online backups
+func (configuration *VolumeSnapshotConfiguration) GetOnline() bool {
+	if configuration.Online == nil {
+		return true
+	}
+
+	return *configuration.Online
+}
+
+// OnlineConfiguration contains the configuration parameters for the online volume snapshot
+type OnlineConfiguration struct {
+	// If false, the function will return immediately after the backup is completed,
+	// without waiting for WAL to be archived.
+	// This behavior is only useful with backup software that independently monitors WAL archiving.
+	// Otherwise, WAL required to make the backup consistent might be missing and make the backup useless.
+	// By default, or when this parameter is true, pg_backup_stop will wait for WAL to be archived when archiving is
+	// enabled.
+	// On a standby, this means that it will wait only when archive_mode = always.
+	// If write activity on the primary is low, it may be useful to run pg_switch_wal on the primary in order to trigger
+	// an immediate segment switch.
+	// +kubebuilder:default:=true
+	// +optional
+	WaitForArchive bool `json:"waitForArchive,omitempty"`
+
+	// Control whether the I/O workload for the backup initial checkpoint will
+	// be limited, according to the `checkpoint_completion_target` setting on
+	// the PostgreSQL server. If set to true, an immediate checkpoint will be
+	// used, meaning PostgreSQL will complete the checkpoint as soon as
+	// possible. `false` by default.
+	// +optional
+	ImmediateCheckpoint bool `json:"immediateCheckpoint,omitempty"`
 }
 
 // ClusterSpec defines the desired state of Cluster
diff --git a/api/v1/zz_generated.deepcopy.go b/api/v1/zz_generated.deepcopy.go
index acdd98bd..ed7b2a97 100644
--- a/api/v1/zz_generated.deepcopy.go
+++ b/api/v1/zz_generated.deepcopy.go
@@ -285,12 +285,27 @@ func (in *BackupStatus) DeepCopyInto(out *BackupStatus) {
 		in, out := &in.StoppedAt, &out.StoppedAt
 		*out = (*in).DeepCopy()
 	}
+	if in.BackupLabelFile != nil {
+		in, out := &in.BackupLabelFile, &out.BackupLabelFile
+		*out = make([]byte, len(*in))
+		copy(*out, *in)
+	}
+	if in.TablespaceMapFile != nil {
+		in, out := &in.TablespaceMapFile, &out.TablespaceMapFile
+		*out = make([]byte, len(*in))
+		copy(*out, *in)
+	}
 	if in.InstanceID != nil {
 		in, out := &in.InstanceID, &out.InstanceID
 		*out = new(InstanceID)
 		**out = **in
 	}
 	in.BackupSnapshotStatus.DeepCopyInto(&out.BackupSnapshotStatus)
+	if in.Online != nil {
+		in, out := &in.Online, &out.Online
+		*out = new(bool)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupStatus.
@@ -1343,6 +1358,21 @@ func (in *NodeMaintenanceWindow) DeepCopy() *NodeMaintenanceWindow {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *OnlineConfiguration) DeepCopyInto(out *OnlineConfiguration) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OnlineConfiguration.
+func (in *OnlineConfiguration) DeepCopy() *OnlineConfiguration {
+	if in == nil {
+		return nil
+	}
+	out := new(OnlineConfiguration)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *PasswordState) DeepCopyInto(out *PasswordState) {
 	*out = *in
@@ -2154,6 +2184,12 @@ func (in *VolumeSnapshotConfiguration) DeepCopyInto(out *VolumeSnapshotConfigura
 			(*out)[key] = val
 		}
 	}
+	if in.Online != nil {
+		in, out := &in.Online, &out.Online
+		*out = new(bool)
+		**out = **in
+	}
+	out.OnlineConfiguration = in.OnlineConfiguration
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new VolumeSnapshotConfiguration.
diff --git a/config/crd/bases/postgresql.cnpg.io_backups.yaml b/config/crd/bases/postgresql.cnpg.io_backups.yaml
index 3e8b7e73..ab668c2c 100644
--- a/config/crd/bases/postgresql.cnpg.io_backups.yaml
+++ b/config/crd/bases/postgresql.cnpg.io_backups.yaml
@@ -150,6 +150,11 @@ spec:
               backupId:
                 description: The ID of the Barman backup
                 type: string
+              backupLabelFile:
+                description: Backup label file content as returned by Postgres in
+                  case of online (hot) backups
+                format: byte
+                type: string
               backupName:
                 description: The Name of the Barman backup
                 type: string
@@ -238,6 +243,10 @@ spec:
               method:
                 description: The backup method being used
                 type: string
+              online:
+                description: Whether the backup was online/hot (`true`) or offline/cold
+                  (`false`)
+                type: boolean
               phase:
                 description: The last backup status
                 type: string
@@ -337,6 +346,11 @@ spec:
                 description: When the backup was terminated
                 format: date-time
                 type: string
+              tablespaceMapFile:
+                description: Tablespace map file content as returned by Postgres in
+                  case of online (hot) backups
+                format: byte
+                type: string
             type: object
         required:
         - metadata
diff --git a/config/crd/bases/postgresql.cnpg.io_clusters.yaml b/config/crd/bases/postgresql.cnpg.io_clusters.yaml
index c7ecba9e..4665ea38 100644
--- a/config/crd/bases/postgresql.cnpg.io_clusters.yaml
+++ b/config/crd/bases/postgresql.cnpg.io_clusters.yaml
@@ -1263,6 +1263,43 @@ spec:
                         description: Labels are key-value pairs that will be added
                           to .metadata.labels snapshot resources.
                         type: object
+                      online:
+                        default: true
+                        description: Whether the default type of backup with volume
+                          snapshots is online/hot (`true`, default) or offline/cold
+                          (`false`)
+                        type: boolean
+                      onlineConfiguration:
+                        default:
+                          immediateCheckpoint: false
+                          waitForArchive: true
+                        description: Configuration parameters to control the online/hot
+                          backup with volume snapshots
+                        properties:
+                          immediateCheckpoint:
+                            description: Control whether the I/O workload for the
+                              backup initial checkpoint will be limited, according
+                              to the `checkpoint_completion_target` setting on the
+                              PostgreSQL server. If set to true, an immediate checkpoint
+                              will be used, meaning PostgreSQL will complete the checkpoint
+                              as soon as possible. `false` by default.
+                            type: boolean
+                          waitForArchive:
+                            default: true
+                            description: If false, the function will return immediately
+                              after the backup is completed, without waiting for WAL
+                              to be archived. This behavior is only useful with backup
+                              software that independently monitors WAL archiving.
+                              Otherwise, WAL required to make the backup consistent
+                              might be missing and make the backup useless. By default,
+                              or when this parameter is true, pg_backup_stop will
+                              wait for WAL to be archived when archiving is enabled.
+                              On a standby, this means that it will wait only when
+                              archive_mode = always. If write activity on the primary
+                              is low, it may be useful to run pg_switch_wal on the
+                              primary in order to trigger an immediate segment switch.
+                            type: boolean
+                        type: object
                       snapshotOwnerReference:
                         default: none
                         description: SnapshotOwnerReference indicates the type of
diff --git a/controllers/backup_controller.go b/controllers/backup_controller.go
index e55295fd..90f49689 100644
--- a/controllers/backup_controller.go
+++ b/controllers/backup_controller.go
@@ -373,7 +373,6 @@ func (r *BackupReconciler) reconcileSnapshotBackup(
 
 	executor := volumesnapshot.
 		NewExecutorBuilder(r.Client, r.Recorder).
-		FenceInstance(true).
 		Build()
 
 	res, err := executor.Execute(ctx, cluster, backup, targetPod, pvcs)
@@ -403,22 +402,7 @@ func (r *BackupReconciler) reconcileSnapshotBackup(
 		contextLogger.Error(err, "Can't update the cluster with the completed snapshot backup data")
 	}
 
-	backup.Status.SetAsCompleted()
-	snapshots, err := volumesnapshot.GetBackupVolumeSnapshots(ctx, r.Client, backup.Namespace, backup.Name)
-	if err != nil {
-		return nil, err
-	}
-
-	backup.Status.BackupSnapshotStatus.SetSnapshotElements(snapshots)
-	if err := backupStatusFromSnapshots(snapshots, &backup.Status); err != nil {
-		contextLogger.Error(err, "while enriching the backup status")
-	}
-
-	if err := annotateSnapshotsWithBackupData(ctx, r.Client, snapshots, &backup.Status); err != nil {
-		contextLogger.Error(err, "while enriching the snapshots's status")
-	}
-
-	return nil, postgres.PatchBackupStatusAndRetry(ctx, r.Client, backup)
+	return nil, nil
 }
 
 func (r *BackupReconciler) getSnapshotTargetPod(
@@ -450,49 +434,6 @@ func (r *BackupReconciler) getSnapshotTargetPod(
 	return targetPod, nil
 }
 
-// AnnotateSnapshots adds labels and annotations to the snapshots using the backup
-// status to facilitate access
-func annotateSnapshotsWithBackupData(
-	ctx context.Context,
-	cli client.Client,
-	snapshots volumesnapshot.Slice,
-	backupStatus *apiv1.BackupStatus,
-) error {
-	contextLogger := log.FromContext(ctx)
-	for idx := range snapshots {
-		snapshot := &snapshots[idx]
-		oldSnapshot := snapshot.DeepCopy()
-		snapshot.Annotations[utils.BackupStartTimeAnnotationName] = backupStatus.StartedAt.Format(time.RFC3339)
-		snapshot.Annotations[utils.BackupEndTimeAnnotationName] = backupStatus.StoppedAt.Format(time.RFC3339)
-		if err := cli.Patch(ctx, snapshot, client.MergeFrom(oldSnapshot)); err != nil {
-			contextLogger.Error(err, "while updating volume snapshot from backup object",
-				"snapshot", snapshot.Name)
-			return err
-		}
-	}
-	return nil
-}
-
-// backupStatusFromSnapshots adds fields to the backup status based on the snapshots
-func backupStatusFromSnapshots(
-	snapshots volumesnapshot.Slice,
-	backupStatus *apiv1.BackupStatus,
-) error {
-	controldata, err := snapshots.GetControldata()
-	if err != nil {
-		return err
-	}
-	pairs := utils.ParsePgControldataOutput(controldata)
-
-	// the begin/end WAL and LSN are the same, since the instance was fenced
-	// for the snapshot
-	backupStatus.BeginWal = pairs["Latest checkpoint's REDO WAL file"]
-	backupStatus.EndWal = pairs["Latest checkpoint's REDO WAL file"]
-	backupStatus.BeginLSN = pairs["Latest checkpoint's REDO location"]
-	backupStatus.EndLSN = pairs["Latest checkpoint's REDO location"]
-	return nil
-}
-
 // isErrorRetryable detects is an error is retryable or not
 func isErrorRetryable(err error) bool {
 	return apierrs.IsServerTimeout(err) || apierrs.IsConflict(err) || apierrs.IsInternalError(err)
diff --git a/controllers/cluster_create.go b/controllers/cluster_create.go
index e825eb1d..73a6829d 100644
--- a/controllers/cluster_create.go
+++ b/controllers/cluster_create.go
@@ -22,6 +22,7 @@ import (
 	"reflect"
 	"time"
 
+	volumesnapshot "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	"github.com/sethvargo/go-password/password"
 	"golang.org/x/exp/slices"
@@ -989,9 +990,16 @@ func (r *ClusterReconciler) createPrimaryInstance(
 			}
 		}
 
-		if cluster.Spec.Bootstrap.Recovery.VolumeSnapshots != nil {
+		volumeSnapshotsRecovery := cluster.Spec.Bootstrap.Recovery.VolumeSnapshots
+		if volumeSnapshotsRecovery != nil {
+			var snapshot volumesnapshot.VolumeSnapshot
+			if err := r.Client.Get(ctx,
+				types.NamespacedName{Name: volumeSnapshotsRecovery.Storage.Name, Namespace: cluster.Namespace},
+				&snapshot); err != nil {
+				return ctrl.Result{}, err
+			}
 			r.Recorder.Event(cluster, "Normal", "CreatingInstance", "Primary instance (from volumeSnapshots)")
-			job = specs.CreatePrimaryJobViaRestoreSnapshot(*cluster, nodeSerial, backup)
+			job = specs.CreatePrimaryJobViaRestoreSnapshot(*cluster, nodeSerial, snapshot, backup)
 			break
 		}
 
diff --git a/docs/src/backup.md b/docs/src/backup.md
index bb554aa0..dc181331 100644
--- a/docs/src/backup.md
+++ b/docs/src/backup.md
@@ -115,9 +115,9 @@ In CloudNativePG, object store based backups:
 VolumeSnapshots instead:
 
 - don't require the WAL archive, although in production it is always recommended
-- support cold backup only (currently)
 - support incremental copy, depending on the underlying storage classes
 - support differential copy, depending on the underlying storage classes
+- also support cold backup
 
 Which one to use depends on your specific requirements and environment,
 including:
@@ -141,24 +141,21 @@ available methods for storing physical base backups.
 |-----------------------------------|:------------:|:--------------------:|
 | **WAL archiving**                 |   Required   |    Recommended (1)   |
 | **Cold backup**                   |      𐄂       |           ✓          |
-| **Hot backup**                    |       ✓      |        𐄂  (2)        |
-| **Incremental copy**              |      𐄂       |         ✓  (3)       |
-| **Differential copy**             |      𐄂       |         ✓  (3)       |
+| **Hot backup**                    |       ✓      |           ✓          |
+| **Incremental copy**              |      𐄂       |         ✓  (2)       |
+| **Differential copy**             |      𐄂       |         ✓  (2)       |
 | **Backup from a standby**         |       ✓      |           ✓          |
-| **Snapshot recovery**             |    𐄂 (4)     |           ✓          |
+| **Snapshot recovery**             |    𐄂 (3)     |           ✓          |
 | **Point In Time Recovery (PITR)** |       ✓      | Requires WAL archive |
 | **Underlying technology**         | Barman Cloud |   Kubernetes API     |
 
 
 > See the explanation below for the notes in the above table:
 >
-> 1. WAL archive must be on an object store
-> 2. Hot backup is not available yet for volume snapshots, and it is planned
-> for version 1.22; however, considering that cold backup is taken by fencing
-> temporarily a standby, the operation does not induce any downtime for your
-> write applications
-> 3. If supported by the underlying storage classes of the PostgreSQL volumes
-> 4. Snapshot recovery can be emulated using the `bootstrap.recovery.recoveryTarget.targetImmediate` option
+> 1. WAL archive must be on an object store at the moment
+> 2. If supported by the underlying storage classes of the PostgreSQL volumes
+> 3. Snapshot recovery can be emulated using the
+>    `bootstrap.recovery.recoveryTarget.targetImmediate` option
 
 ## Scheduled backups
 
diff --git a/docs/src/backup_volumesnapshot.md b/docs/src/backup_volumesnapshot.md
index 42df6596..a95becad 100644
--- a/docs/src/backup_volumesnapshot.md
+++ b/docs/src/backup_volumesnapshot.md
@@ -1,24 +1,20 @@
 # Backup on volume snapshots
 
-!!! Important
-    The current implementation of volume snapshots in CloudNativePG
-    supports [cold backup](backup.md#cold-and-hot-backups) only.
-    Hot backup with direct support of
-    [PostgreSQL's low level API for base backups](https://www.postgresql.org/docs/current/continuous-archiving.html#BACKUP-LOWLEVEL-BASE-BACKUP)
-    will be added in version 1.22. Having said this, the current implementation
-    is suitable for production HA environments, as it will by default work on
-    the most aligned standby without impacting the primary.
+!!! Warning
+    The initial release of volume snapshots (version 1.21.0) only supported
+    cold backups, which required fencing of the instance. This limitation
+    has been waived starting with version 1.21.1. Given the minimal impact of
+    the change on the code, maintainers have decided to backport this feature
+    immediately instead of waiting for version 1.22.0 to be out, and make online
+    backups the default behavior on volume snapshots too. If you are planning
+    to rely instead on cold backups, make sure you follow the instructions below.
 
 !!! Warning
     As noted in the [backup document](backup.md), a cold snapshot explicitly
     set to target the primary will result in the primary being fenced for
     the duration of the backup, rendering the cluster read-only during that
-    time.
-
-!!! Warning
-    A volume snapshot backup requires fencing the target instance. For safety,
-    in a cluster already containing fenced instances, a cold snapshot would be
-    rejected.
+    For safety, in a cluster already containing fenced instances, a cold
+    snapshot is rejected.
 
 CloudNativePG is one of the first known cases of database operators that
 directly leverages the Kubernetes native Volume Snapshot API for both
@@ -126,6 +122,62 @@ defined in your PostgreSQL cluster.
 Once a cluster is defined for volume snapshot backups, you need to define
 a `ScheduledBackup` resource that requests such backups on a periodic basis.
 
+## Hot and cold backups
+
+By default, CloudNativePG requests an online/hot backup on volume snapshots, using the
+[PostgreSQL defaults of the low-level API for base backups](https://www.postgresql.org/docs/current/continuous-archiving.html#BACKUP-LOWLEVEL-BASE-BACKUP):
+
+- it doesn't request an immediate checkpoint when starting the backup procedure
+- it waits for the WAL archiver to archive the last segment of the backup when
+  terminating the backup procedure
+
+!!! Important
+    The default values are suitable for most production environments. Hot
+    backups are consistent and can be used to perform snapshot recovery, as we
+    ensure WAL retention from the start of the backup through a temporary
+    replication slot. However, our recommendation is to rely on cold backups for
+    that purpose.
+
+You can explicitly change the default behavior through the following options in
+the `.spec.backup.volumeSnapshot` stanza of the `Cluster` resource:
+
+- `online`: accepting `true` (default) or `false` as a value
+- `onlineConfiguration.immediateCheckpoint`: whether you want to request an
+  immediate checkpoint before you start the backup procedure or not;
+  technically, it corresponds to the `fast` argument you pass to the
+  `pg_backup_start`/`pg_start_backup()` function in PostgreSQL, accepting
+  `true` (default) or `false`
+- `onlineConfiguration.waitForArchive`: whether you want to wait for the
+  archiver to process the last segment of the backup or not; technically, it
+  corresponds to the `wait_for_archive` argument you pass to the
+  `pg_backup_stop`/`pg_stop_backup()` function in PostgreSQL, accepting `true`
+  (default) or `false`
+
+If you want to change the default behavior of your Postgres cluster to take
+cold backups by default, all you need to do is add the `online: false` option
+to your manifest, as follows:
+
+```yaml
+  # ...
+  backup:
+    volumeSnapshot:
+       online: false
+       # ...
+```
+
+If you are instead requesting an immediate checkpoint as the default behavior,
+you can add this section:
+
+```yaml
+  # ...
+  backup:
+    volumeSnapshot:
+       online: true
+       onlineConfiguration:
+         immediateCheckpoint: true
+       # ...
+```
+
 ## Persistence of volume snapshot objects
 
 By default, `VolumeSnapshot` objects created by CloudNativePG are retained after
diff --git a/docs/src/cloudnative-pg.v1.md b/docs/src/cloudnative-pg.v1.md
index 20cae263..efa3b122 100644
--- a/docs/src/cloudnative-pg.v1.md
+++ b/docs/src/cloudnative-pg.v1.md
@@ -684,6 +684,20 @@ parameter is omitted</p>
    <p>The backup command output in case of error</p>
 </td>
 </tr>
+<tr><td><code>backupLabelFile</code><br/>
+<i>[]byte</i>
+</td>
+<td>
+   <p>Backup label file content as returned by Postgres in case of online (hot) backups</p>
+</td>
+</tr>
+<tr><td><code>tablespaceMapFile</code><br/>
+<i>[]byte</i>
+</td>
+<td>
+   <p>Tablespace map file content as returned by Postgres in case of online (hot) backups</p>
+</td>
+</tr>
 <tr><td><code>instanceID</code><br/>
 <a href="#postgresql-cnpg-io-v1-InstanceID"><i>InstanceID</i></a>
 </td>
@@ -705,6 +719,13 @@ parameter is omitted</p>
    <p>The backup method being used</p>
 </td>
 </tr>
+<tr><td><code>online</code> <B>[Required]</B><br/>
+<i>bool</i>
+</td>
+<td>
+   <p>Whether the backup was online/hot (<code>true</code>) or offline/cold (<code>false</code>)</p>
+</td>
+</tr>
 </tbody>
 </table>
 
@@ -2846,6 +2867,49 @@ up again) or not (recreate it elsewhere - when <code>instances</code> &gt;1)</p>
 </tbody>
 </table>
 
+## OnlineConfiguration     {#postgresql-cnpg-io-v1-OnlineConfiguration}
+
+
+**Appears in:**
+
+- [VolumeSnapshotConfiguration](#postgresql-cnpg-io-v1-VolumeSnapshotConfiguration)
+
+
+<p>OnlineConfiguration contains the configuration parameters for the online volume snapshot</p>
+
+
+<table class="table">
+<thead><tr><th width="30%">Field</th><th>Description</th></tr></thead>
+<tbody>
+<tr><td><code>waitForArchive</code><br/>
+<i>bool</i>
+</td>
+<td>
+   <p>If false, the function will return immediately after the backup is completed,
+without waiting for WAL to be archived.
+This behavior is only useful with backup software that independently monitors WAL archiving.
+Otherwise, WAL required to make the backup consistent might be missing and make the backup useless.
+By default, or when this parameter is true, pg_backup_stop will wait for WAL to be archived when archiving is
+enabled.
+On a standby, this means that it will wait only when archive_mode = always.
+If write activity on the primary is low, it may be useful to run pg_switch_wal on the primary in order to trigger
+an immediate segment switch.</p>
+</td>
+</tr>
+<tr><td><code>immediateCheckpoint</code><br/>
+<i>bool</i>
+</td>
+<td>
+   <p>Control whether the I/O workload for the backup initial checkpoint will
+be limited, according to the <code>checkpoint_completion_target</code> setting on
+the PostgreSQL server. If set to true, an immediate checkpoint will be
+used, meaning PostgreSQL will complete the checkpoint as soon as
+possible. <code>false</code> by default.</p>
+</td>
+</tr>
+</tbody>
+</table>
+
 ## PasswordState     {#postgresql-cnpg-io-v1-PasswordState}
 
 
@@ -4317,6 +4381,21 @@ It is the default class for the other types if no specific class is present</p>
    <p>SnapshotOwnerReference indicates the type of owner reference the snapshot should have</p>
 </td>
 </tr>
+<tr><td><code>online</code><br/>
+<i>bool</i>
+</td>
+<td>
+   <p>Whether the default type of backup with volume snapshots is
+online/hot (<code>true</code>, default) or offline/cold (<code>false</code>)</p>
+</td>
+</tr>
+<tr><td><code>onlineConfiguration</code><br/>
+<a href="#postgresql-cnpg-io-v1-OnlineConfiguration"><i>OnlineConfiguration</i></a>
+</td>
+<td>
+   <p>Configuration parameters to control the online/hot backup with volume snapshots</p>
+</td>
+</tr>
 </tbody>
 </table>
 
diff --git a/docs/src/operator_capability_levels.md b/docs/src/operator_capability_levels.md
index 7d958d7e..be22c3bd 100644
--- a/docs/src/operator_capability_levels.md
+++ b/docs/src/operator_capability_levels.md
@@ -360,10 +360,9 @@ alongside WAL files.
 Both `barman-cloud-wal-restore` and `barman-cloud-backup` are distributed in
 the application container image under GNU GPL 3 terms.
 
-Object store backups are taken while PostgreSQL is up and running (hot
-backups). Volume Snapshot backups are currently taken after PostgreSQL has
-been shut down (cold backups). This will change in the future with support
-for online Volume Snapshot backups.
+Object store backups and Volume Snapshot backups are taken while PostgreSQL is
+up and running (hot backups). Volume snapshots also support taking consistent
+database snapshots with cold backups.
 
 ### Backups from a standby
 
diff --git a/docs/src/recovery.md b/docs/src/recovery.md
index d8aa48e4..64ce8191 100644
--- a/docs/src/recovery.md
+++ b/docs/src/recovery.md
@@ -48,14 +48,11 @@ compatible, from a physical replication standpoint, with the original one.
 
 For recovery using volume snapshots:
 
-- take a consistent cold backup of the Postgres cluster from a standby through
-  the `kubectl cnpg backup` command (see the [plugin document](kubectl-plugin.md#requesting-a-new-base-backup)
-  for reference), which creates the necessary `VolumeSnapshot` objects (two if
-  you have a separate volume for WALs, one if you don't) - recover from the above
-  *VolumeSnapshot* objects through the `volumeSnapshots` option in the
-  `.spec.bootstrap.recovery` stanza, as described in
-  ["Recovery from `VolumeSnapshot` objects"](#recovery-from-volumesnapshot-objects)
-  below
+- using a consistent set of `VolumeSnapshot` objects that all belong to the
+  same backup, and identified by the same `cnpg.io/cluster` and
+  `cnpg.io/backupName` labels, then recovering through the `volumeSnapshots`
+  option in the `.spec.bootstrap.recovery` stanza, as described in
+  ["Recovery from `VolumeSnapshot` objects"](#recovery-from-volumesnapshot-objects) below
 
 ## Recovery from an object store
 
@@ -118,6 +115,7 @@ spec:
     to synchronize them, resulting in a slower process depending on the size
     of the database. This limitation will be lifted in the future when support
     for online backups will be introduced.
+
 CloudNativePG can create a new cluster from a `VolumeSnapshot` of a PVC of an
 existing `Cluster` that's been taken using the declarative API for
 [volume snapshot backups](backup_volumesnapshot.md).
diff --git a/internal/cmd/manager/instance/restoresnapshot/cmd.go b/internal/cmd/manager/instance/restoresnapshot/cmd.go
index 9c32ed1b..a0f40461 100644
--- a/internal/cmd/manager/instance/restoresnapshot/cmd.go
+++ b/internal/cmd/manager/instance/restoresnapshot/cmd.go
@@ -19,6 +19,7 @@ package restoresnapshot
 
 import (
 	"context"
+	"encoding/base64"
 	"os"
 
 	"github.com/spf13/cobra"
@@ -33,11 +34,15 @@ import (
 
 // NewCmd creates the "restoresnapshot" subcommand
 func NewCmd() *cobra.Command {
-	var clusterName string
-	var namespace string
-	var pgData string
-	var pgWal string
-	var immediate bool
+	var (
+		clusterName   string
+		namespace     string
+		pgData        string
+		pgWal         string
+		backupLabel   string
+		tablespaceMap string
+		immediate     bool
+	)
 
 	cmd := &cobra.Command{
 		Use:           "restoresnapshot [flags]",
@@ -58,6 +63,22 @@ func NewCmd() *cobra.Command {
 				PgWal:       pgWal,
 			}
 
+			if backupLabel != "" {
+				res, err := base64.StdEncoding.DecodeString(backupLabel)
+				if err != nil {
+					return err
+				}
+				info.BackupLabelFile = res
+			}
+
+			if tablespaceMap != "" {
+				res, err := base64.StdEncoding.DecodeString(tablespaceMap)
+				if err != nil {
+					return err
+				}
+				info.TablespaceMapFile = res
+			}
+
 			err := execute(ctx, info, immediate)
 			if err != nil {
 				log.Error(err, "Error while recovering Volume Snapshot backup")
@@ -79,6 +100,8 @@ func NewCmd() *cobra.Command {
 		"the cluster")
 	cmd.Flags().StringVar(&pgData, "pg-data", os.Getenv("PGDATA"), "The PGDATA to be restored")
 	cmd.Flags().StringVar(&pgWal, "pg-wal", "", "The PGWAL to be restored")
+	cmd.Flags().StringVar(&backupLabel, "backuplabel", "", "The restore backup_label file content")
+	cmd.Flags().StringVar(&tablespaceMap, "tablespacemap", "", "The restore tablespace_map file content")
 	cmd.Flags().BoolVar(&immediate, "immediate", false, "Do not start PostgreSQL but just recover the snapshot")
 
 	return cmd
diff --git a/pkg/management/postgres/backup.go b/pkg/management/postgres/backup.go
index cf27d87b..c5ff5adf 100644
--- a/pkg/management/postgres/backup.go
+++ b/pkg/management/postgres/backup.go
@@ -432,6 +432,7 @@ func (b *BackupCommand) backupMaintenance(ctx context.Context) {
 
 // PatchBackupStatusAndRetry updates a certain backup's status in the k8s database,
 // retries when error occurs
+// TODO: this method does not belong here, it should be moved to api/v1/backup_types.go
 func PatchBackupStatusAndRetry(
 	ctx context.Context,
 	cli client.Client,
diff --git a/pkg/management/postgres/constants/constants.go b/pkg/management/postgres/constants/constants.go
index 361dd6e8..d15ac205 100644
--- a/pkg/management/postgres/constants/constants.go
+++ b/pkg/management/postgres/constants/constants.go
@@ -37,6 +37,12 @@ const (
 	// the user name maps
 	PostgresqlIdentFile = "pg_ident.conf"
 
+	// BackupLabelFile holds the content of BackupLabelFile. Used during a restore from a hot backup.
+	BackupLabelFile = "backup_label"
+
+	// TablespaceMapFile holds the content of TablespaceMapFile. Used during a restore from a hot backup.
+	TablespaceMapFile = "tablespace_map"
+
 	// InitdbName is the name of the command to initialize a PostgreSQL database
 	InitdbName = "initdb"
 
diff --git a/pkg/management/postgres/initdb.go b/pkg/management/postgres/initdb.go
index 3d404900..8eef521d 100644
--- a/pkg/management/postgres/initdb.go
+++ b/pkg/management/postgres/initdb.go
@@ -95,6 +95,12 @@ type InitInfo struct {
 	// PostInitApplicationSQLRefsFolder is the folder which contains a bunch
 	// of SQL files to be executed just after having configured a new instance
 	PostInitApplicationSQLRefsFolder string
+
+	// BackupLabelFile holds the content returned by pg_stop_backup. Needed for a hot backup restore
+	BackupLabelFile []byte
+
+	// TablespaceMapFile holds the content returned by pg_stop_backup. Needed for a hot backup restore
+	TablespaceMapFile []byte
 }
 
 // VerifyPGData verifies if the passed configuration is OK, otherwise it returns an error
diff --git a/pkg/management/postgres/restore.go b/pkg/management/postgres/restore.go
index b65a1a05..d9788d72 100644
--- a/pkg/management/postgres/restore.go
+++ b/pkg/management/postgres/restore.go
@@ -25,6 +25,7 @@ import (
 	"os"
 	"os/exec"
 	"path"
+	"path/filepath"
 	"strings"
 	"time"
 
@@ -105,6 +106,20 @@ func (info InitInfo) RestoreSnapshot(ctx context.Context, cli client.Client, imm
 	contextLogger.Info("Recovering from volume snapshot",
 		"sourceName", cluster.Spec.Bootstrap.Recovery.Source)
 
+	if len(info.BackupLabelFile) > 0 {
+		filePath := filepath.Join(info.PgData, constants.BackupLabelFile)
+		if _, err := fileutils.WriteFileAtomic(filePath, info.BackupLabelFile, 0o666); err != nil {
+			return err
+		}
+	}
+
+	if len(info.TablespaceMapFile) > 0 {
+		filePath := filepath.Join(info.PgData, constants.TablespaceMapFile)
+		if _, err := fileutils.WriteFileAtomic(filePath, info.TablespaceMapFile, 0o666); err != nil {
+			return err
+		}
+	}
+
 	backup, env, err := info.createBackupObjectForSnapshotRestore(ctx, cli, cluster)
 	if err != nil {
 		return err
diff --git a/pkg/management/postgres/webserver/backup_client.go b/pkg/management/postgres/webserver/backup_client.go
new file mode 100644
index 00000000..0bac258e
--- /dev/null
+++ b/pkg/management/postgres/webserver/backup_client.go
@@ -0,0 +1,134 @@
+/*
+Copyright The CloudNativePG Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package webserver
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/log"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/url"
+)
+
+// BackupClient a client to interact with the instance backup endpoints
+type BackupClient struct {
+	cli *http.Client
+}
+
+// NewBackupClient creates a client capable of interacting with the instance backup endpoints
+func NewBackupClient() *BackupClient {
+	const connectionTimeout = 2 * time.Second
+	const requestTimeout = 30 * time.Second
+
+	// We want a connection timeout to prevent waiting for the default
+	// TCP connection timeout (30 seconds) on lost SYN packets
+	timeoutClient := &http.Client{
+		Transport: &http.Transport{
+			DialContext: (&net.Dialer{
+				Timeout: connectionTimeout,
+			}).DialContext,
+		},
+		Timeout: requestTimeout,
+	}
+	return &BackupClient{cli: timeoutClient}
+}
+
+// Status the current status of the backup. Returns empty BackupResultData struct if it is not running.
+func (c *BackupClient) Status(ctx context.Context, podIP string) (*BackupResultData, error) {
+	httpURL := url.Build(podIP, url.PathPgModeBackup, url.StatusPort)
+	req, err := http.NewRequestWithContext(ctx, "GET", httpURL, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	return executeRequest[BackupResultData](ctx, c.cli, req)
+}
+
+// Start runs the pg_start_backup
+func (c *BackupClient) Start(
+	ctx context.Context,
+	podIP string,
+	sbq StartBackupRequest,
+) (*struct{}, error) {
+	httpURL := url.Build(podIP, url.PathPgModeBackup, url.StatusPort)
+
+	// Marshalling the payload to JSON
+	jsonBody, err := json.Marshal(sbq)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal payload: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "POST", httpURL, bytes.NewReader(jsonBody))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	return executeRequest[struct{}](ctx, c.cli, req)
+}
+
+// Stop runs the pg_stop_backup
+func (c *BackupClient) Stop(ctx context.Context, podIP string) error {
+	httpURL := url.Build(podIP, url.PathPgModeBackup, url.StatusPort)
+	req, err := http.NewRequestWithContext(ctx, "DELETE", httpURL, nil)
+	if err != nil {
+		return err
+	}
+	_, err = executeRequest[BackupResultData](ctx, c.cli, req)
+	return err
+}
+
+func executeRequest[T any](ctx context.Context, cli *http.Client, req *http.Request) (*T, error) {
+	contextLogger := log.FromContext(ctx)
+
+	resp, err := cli.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("while execution a request: %w", err)
+	}
+
+	defer func() {
+		if err := resp.Body.Close(); err != nil {
+			contextLogger.Error(err, "while closing body")
+		}
+	}()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("while reading the body: %w", err)
+	}
+
+	if resp.StatusCode == 500 {
+		return nil, fmt.Errorf("encountered an internal server error status code 500 with body: %s", string(body))
+	}
+
+	var result Response[T]
+	if err := json.Unmarshal(body, &result); err != nil {
+		return nil, fmt.Errorf("while unmarshalling the body, body: %s err: %w", string(body), err)
+	}
+	if result.Error != nil {
+		return nil, fmt.Errorf("body contained an error code: %s and message: %s",
+			result.Error.Code, result.Error.Message)
+	}
+
+	return result.Data, nil
+}
diff --git a/pkg/management/postgres/webserver/backup_connection.go b/pkg/management/postgres/webserver/backup_connection.go
new file mode 100644
index 00000000..82ad1d09
--- /dev/null
+++ b/pkg/management/postgres/webserver/backup_connection.go
@@ -0,0 +1,151 @@
+/*
+Copyright The CloudNativePG Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package webserver
+
+import (
+	"context"
+	"database/sql"
+	"regexp"
+
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/log"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/utils"
+	postgresUtils "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres"
+)
+
+// BackupResultData is the result of executing pg_start_backup and pg_stop_backup
+type BackupResultData struct {
+	BeginLSN   postgresUtils.LSN     `json:"beginLSN,omitempty"`
+	EndLSN     postgresUtils.LSN     `json:"endLSN,omitempty"`
+	LabelFile  []byte                `json:"labelFile,omitempty"`
+	SpcmapFile []byte                `json:"spcmapFile,omitempty"`
+	BackupName string                `json:"backupName,omitempty"`
+	Phase      BackupConnectionPhase `json:"phase,omitempty"`
+}
+
+// BackupConnectionPhase a connection phase of the backup
+type BackupConnectionPhase string
+
+// A backup phase
+const (
+	Starting  BackupConnectionPhase = "starting"
+	Started   BackupConnectionPhase = "started"
+	Closing   BackupConnectionPhase = "closing"
+	Completed BackupConnectionPhase = "completed"
+)
+
+// replicationSlotInvalidCharacters matches every character that is
+// not valid in a replication slot name
+var replicationSlotInvalidCharacters = regexp.MustCompile(`[^a-z0-9_]`)
+
+type backupConnection struct {
+	immediateCheckpoint  bool
+	waitForArchive       bool
+	conn                 *sql.Conn
+	postgresMajorVersion uint64
+	data                 BackupResultData
+	err                  error
+}
+
+func newBackupConnection(
+	ctx context.Context,
+	instance *postgres.Instance,
+	backupName string,
+	immediateCheckpoint bool,
+	waitForArchive bool,
+) (*backupConnection, error) {
+	superUserDB, err := instance.GetSuperUserDB()
+	if err != nil {
+		return nil, err
+	}
+
+	vers, err := utils.GetPgVersion(superUserDB)
+	if err != nil {
+		return nil, err
+	}
+
+	// the context is used only while obtaining the connection
+	conn, err := superUserDB.Conn(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &backupConnection{
+		immediateCheckpoint:  immediateCheckpoint,
+		waitForArchive:       waitForArchive,
+		conn:                 conn,
+		postgresMajorVersion: vers.Major,
+		data:                 BackupResultData{BackupName: backupName},
+	}, nil
+}
+
+func (bc *backupConnection) startBackup(ctx context.Context) {
+	if bc == nil {
+		return
+	}
+	bc.data.Phase = Starting
+
+	// TODO: refactor with the same logic of GetSlotNameFromInstanceName in the api package
+	slotName := replicationSlotInvalidCharacters.ReplaceAllString(bc.data.BackupName, "_")
+	if _, bc.err = bc.conn.ExecContext(
+		ctx,
+		"SELECT pg_create_physical_replication_slot(slot_name => $1, immediately_reserve => true, temporary => true)",
+		slotName,
+	); bc.err != nil {
+		return
+	}
+
+	var row *sql.Row
+	if bc.postgresMajorVersion < 15 {
+		row = bc.conn.QueryRowContext(ctx, "SELECT pg_start_backup($1, $2, false);", bc.data.BackupName,
+			bc.immediateCheckpoint)
+	} else {
+		row = bc.conn.QueryRowContext(ctx, "SELECT pg_backup_start(label => $1, fast => $2);", bc.data.BackupName,
+			bc.immediateCheckpoint)
+	}
+
+	bc.err = row.Scan(&bc.data.BeginLSN)
+	bc.data.Phase = Started
+}
+
+func (bc *backupConnection) stopBackup(ctx context.Context) {
+	if bc == nil {
+		return
+	}
+
+	if bc.err != nil {
+		return
+	}
+	bc.data.Phase = Closing
+
+	contextLogger := log.FromContext(ctx)
+
+	var row *sql.Row
+	if bc.postgresMajorVersion < 15 {
+		row = bc.conn.QueryRowContext(ctx,
+			"SELECT lsn, labelfile, spcmapfile FROM pg_stop_backup(false, $1);", bc.waitForArchive)
+	} else {
+		row = bc.conn.QueryRowContext(ctx,
+			"SELECT lsn, labelfile, spcmapfile FROM pg_backup_stop(wait_for_archive => $1);", bc.waitForArchive)
+	}
+
+	bc.err = row.Scan(&bc.data.EndLSN, &bc.data.LabelFile, &bc.data.SpcmapFile)
+	bc.data.Phase = Completed
+	if err := bc.conn.Close(); err != nil {
+		contextLogger.Error(err, "while closing backup connection")
+	}
+}
diff --git a/pkg/management/postgres/webserver/remote.go b/pkg/management/postgres/webserver/remote.go
index 23779015..87cf48b1 100644
--- a/pkg/management/postgres/webserver/remote.go
+++ b/pkg/management/postgres/webserver/remote.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"time"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -33,8 +34,17 @@ import (
 )
 
 type remoteWebserverEndpoints struct {
-	typedClient client.Client
-	instance    *postgres.Instance
+	typedClient   client.Client
+	instance      *postgres.Instance
+	currentBackup *backupConnection
+}
+
+// StartBackupRequest the required data to execute the pg_start_backup
+type StartBackupRequest struct {
+	ImmediateCheckpoint bool   `json:"immediateCheckpoint"`
+	WaitForArchive      bool   `json:"waitForArchive"`
+	BackupName          string `json:"backupName"`
+	Force               bool   `json:"force,omitempty"`
 }
 
 // NewRemoteWebServer returns a webserver that allows connection from external clients
@@ -52,7 +62,10 @@ func NewRemoteWebServer(
 		typedClient: typedClient,
 		instance:    instance,
 	}
+	go endpoints.keepBackupAliveConn()
+
 	serveMux := http.NewServeMux()
+	serveMux.HandleFunc(url.PathPgModeBackup, endpoints.backup)
 	serveMux.HandleFunc(url.PathHealth, endpoints.isServerHealthy)
 	serveMux.HandleFunc(url.PathReady, endpoints.isServerReady)
 	serveMux.HandleFunc(url.PathPgStatus, endpoints.pgStatus)
@@ -190,3 +203,79 @@ func (ws *remoteWebserverEndpoints) updateInstanceManager(
 		_, _ = fmt.Fprint(w, "OK")
 	}
 }
+
+func (ws *remoteWebserverEndpoints) backup(w http.ResponseWriter, req *http.Request) {
+	log.Trace("request method", "method", req.Method)
+
+	switch req.Method {
+	case http.MethodGet:
+		if ws.currentBackup == nil {
+			sendDataJSONResponse(w, 200, struct{}{})
+			return
+		}
+		if ws.currentBackup.err != nil {
+			sendBadRequestJSONResponse(w, "ERROR_WHILE_PROCESSING", ws.currentBackup.err.Error())
+			return
+		}
+		sendDataJSONResponse(w, 200, ws.currentBackup.data)
+
+	case http.MethodPost:
+		var p StartBackupRequest
+		err := json.NewDecoder(req.Body).Decode(&p)
+		if err != nil {
+			sendBadRequestJSONResponse(w, "FAILED_TO_PARSE_REQUEST", "Failed to parse request body")
+			return
+		}
+		defer func() {
+			if err := req.Body.Close(); err != nil {
+				log.Error(err, "while closing the body")
+			}
+		}()
+		if ws.currentBackup != nil {
+			if !p.Force {
+				sendBadRequestJSONResponse(w, "PROCESS_ALREADY_RUNNING", "")
+				return
+			}
+			ws.currentBackup.stopBackup(req.Context())
+		}
+		ws.currentBackup, err = newBackupConnection(
+			req.Context(),
+			ws.instance,
+			p.BackupName,
+			p.ImmediateCheckpoint,
+			p.WaitForArchive,
+		)
+		if err != nil {
+			sendBadRequestJSONResponse(w, "INITIALIZING_CONNECTION", err.Error())
+			return
+		}
+		go ws.currentBackup.startBackup(context.Background())
+		sendDataJSONResponse(w, 200, struct{}{})
+
+	case http.MethodDelete:
+		if ws.currentBackup == nil {
+			sendBadRequestJSONResponse(w, "NO_ONGOING_BACKUP", "")
+			return
+		}
+
+		if ws.currentBackup.data.Phase != Started {
+			sendBadRequestJSONResponse(w, "CANNOT_CLOSE_NOT_STARTED", "")
+			return
+		}
+
+		go ws.currentBackup.stopBackup(context.Background())
+		sendDataJSONResponse(w, 200, struct{}{})
+	}
+}
+
+// TODO: no need to active ping, we are connected locally
+func (ws *remoteWebserverEndpoints) keepBackupAliveConn() {
+	for {
+		if ws.currentBackup != nil && ws.currentBackup.conn != nil &&
+			ws.currentBackup.err == nil && ws.currentBackup.data.Phase != Completed {
+			log.Trace("keeping current backup connection alive")
+			_ = ws.currentBackup.conn.PingContext(context.Background())
+		}
+		time.Sleep(3 * time.Second)
+	}
+}
diff --git a/pkg/management/postgres/webserver/webserver.go b/pkg/management/postgres/webserver/webserver.go
index c9e23881..7801241f 100644
--- a/pkg/management/postgres/webserver/webserver.go
+++ b/pkg/management/postgres/webserver/webserver.go
@@ -18,6 +18,7 @@ package webserver
 
 import (
 	"context"
+	"encoding/json"
 	"net/http"
 	"time"
 
@@ -32,6 +33,22 @@ const (
 	DefaultReadHeaderTimeout = 3 * time.Second
 )
 
+// Error an error response from http webserver
+type Error struct {
+	// One of a server-defined set of error codes
+	Code string `json:"code"`
+	// A human-readable representation of the error.
+	Message string `json:"message"`
+	// An array of details about specific errors that led to this reported error.
+	Details []Error `json:"details,omitempty"`
+}
+
+// Response a response from the http webserver
+type Response[T interface{}] struct {
+	Data  *T     `json:"data,omitempty"`
+	Error *Error `json:"error,omitempty"`
+}
+
 // Webserver contains a server that interacts with postgres instance
 type Webserver struct {
 	// instance is the PostgreSQL instance to be collected
@@ -76,3 +93,28 @@ func (ws *Webserver) Start(ctx context.Context) error {
 
 	return nil
 }
+
+// sendJSONResponse sends a generic JSON response.
+func sendJSONResponse[T any](w http.ResponseWriter, statusCode int, data Response[T]) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(statusCode)
+
+	if err := json.NewEncoder(w).Encode(data); err != nil {
+		http.Error(w, "Failed to write response", http.StatusInternalServerError)
+	}
+}
+
+func sendBadRequestJSONResponse(w http.ResponseWriter, errorCode string, message string) {
+	sendJSONResponse(w, http.StatusBadRequest, Response[any]{
+		Error: &Error{
+			Code:    errorCode,
+			Message: message,
+		},
+	})
+}
+
+func sendDataJSONResponse[T interface{}](w http.ResponseWriter, statusCode int, data T) {
+	sendJSONResponse(w, statusCode, Response[T]{
+		Data: &data,
+	})
+}
diff --git a/pkg/management/url/url.go b/pkg/management/url/url.go
index 8f2d8d2b..a8429fe6 100644
--- a/pkg/management/url/url.go
+++ b/pkg/management/url/url.go
@@ -46,6 +46,9 @@ const (
 	// PathPgBackup is the URL path for PostgreSQL Backup
 	PathPgBackup string = "/pg/backup"
 
+	// PathPgModeBackup is the URL path to interact with pg_start_backup and pg_stop_backup
+	PathPgModeBackup string = "/pg/mode/backup"
+
 	// PathMetrics is the URL path for Metrics
 	PathMetrics string = "/metrics"
 
diff --git a/pkg/reconciler/backup/volumesnapshot/reconciler.go b/pkg/reconciler/backup/volumesnapshot/reconciler.go
index 16418793..9d6f9954 100644
--- a/pkg/reconciler/backup/volumesnapshot/reconciler.go
+++ b/pkg/reconciler/backup/volumesnapshot/reconciler.go
@@ -18,6 +18,7 @@ package volumesnapshot
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -29,12 +30,15 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/utils/ptr"
 	"k8s.io/utils/strings/slices"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/log"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/webserver"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/resources"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/resources/instance"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
@@ -43,9 +47,9 @@ import (
 // Reconciler is an object capable of executing a volume snapshot on a running cluster
 type Reconciler struct {
 	cli                  client.Client
-	shouldFence          bool
 	recorder             record.EventRecorder
 	instanceStatusClient *instance.StatusClient
+	backupClient         *webserver.BackupClient
 }
 
 // ExecutorBuilder is a struct capable of creating a Reconciler
@@ -63,16 +67,11 @@ func NewExecutorBuilder(
 			cli:                  cli,
 			recorder:             recorder,
 			instanceStatusClient: instance.NewStatusClient(),
+			backupClient:         webserver.NewBackupClient(),
 		},
 	}
 }
 
-// FenceInstance instructs if the Reconciler should fence or not the instance while taking the snapshot
-func (e *ExecutorBuilder) FenceInstance(fence bool) *ExecutorBuilder {
-	e.executor.shouldFence = fence
-	return e
-}
-
 // Build returns the Reconciler instance
 func (e *ExecutorBuilder) Build() *Reconciler {
 	return &e.executor
@@ -120,8 +119,7 @@ func (se *Reconciler) enrichSnapshot(
 	vs.Labels[utils.BackupDateLabelName] = time.Now().Format("20060102")
 	vs.Labels[utils.BackupMonthLabelName] = time.Now().Format("200601")
 	vs.Labels[utils.BackupYearLabelName] = strconv.Itoa(time.Now().Year())
-	// TODO: once we have online volumesnapshot backups, this should change
-	vs.Annotations[utils.IsOnlineBackupLabelName] = "false"
+	vs.Annotations[utils.IsOnlineBackupLabelName] = strconv.FormatBool(backup.Status.GetOnline())
 
 	rawCluster, err := json.Marshal(cluster)
 	if err != nil {
@@ -134,6 +132,8 @@ func (se *Reconciler) enrichSnapshot(
 }
 
 // Execute the volume snapshot of the given cluster instance
+// TODO: remove the nolint
+// nolint: gocognit
 func (se *Reconciler) Execute(
 	ctx context.Context,
 	cluster *apiv1.Cluster,
@@ -147,14 +147,15 @@ func (se *Reconciler) Execute(
 
 	contextLogger := log.FromContext(ctx).WithValues("podName", targetPod.Name)
 
+	volumeSnapshotConfig := cluster.Spec.Backup.VolumeSnapshot
 	// Step 0: check if the snapshots have been created already
-	volumeSnapshots, err := GetBackupVolumeSnapshots(ctx, se.cli, cluster.Namespace, backup.Name)
+	volumeSnapshots, err := getBackupVolumeSnapshots(ctx, se.cli, cluster.Namespace, backup.Name)
 	if err != nil {
 		return nil, err
 	}
 
-	// Step 1: fencing
-	if len(volumeSnapshots) == 0 && se.shouldFence {
+	// Step 1: preparation of snapshot backup
+	if len(volumeSnapshots) == 0 && !volumeSnapshotConfig.GetOnline() {
 		contextLogger.Debug("Checking pre-requisites")
 		if err := se.ensurePodIsFenced(ctx, cluster, backup, targetPod.Name); err != nil {
 			return nil, err
@@ -164,6 +165,27 @@ func (se *Reconciler) Execute(
 			return res, err
 		}
 	}
+	if len(volumeSnapshots) == 0 && volumeSnapshotConfig.GetOnline() {
+		status, err := se.backupClient.Status(ctx, targetPod.Status.PodIP)
+		if err != nil {
+			return nil, fmt.Errorf("while getting status: %w", err)
+		}
+
+		switch status.Phase {
+		case webserver.Starting:
+			return &ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+		case "":
+			req := webserver.StartBackupRequest{
+				ImmediateCheckpoint: volumeSnapshotConfig.OnlineConfiguration.ImmediateCheckpoint,
+				WaitForArchive:      volumeSnapshotConfig.OnlineConfiguration.WaitForArchive,
+				BackupName:          backup.Name,
+				Force:               true,
+			}
+			if _, err := se.backupClient.Start(ctx, targetPod.Status.PodIP, req); err != nil {
+				return nil, fmt.Errorf("while trying to start the backup: %w", err)
+			}
+		}
+	}
 
 	// Step 2: create snapshot
 	if len(volumeSnapshots) == 0 {
@@ -187,7 +209,108 @@ func (se *Reconciler) Execute(
 		return nil, err
 	}
 
-	return nil, nil
+	// TODO: remove the nolint
+	// nolint: nestif
+	if volumeSnapshotConfig.GetOnline() {
+		status, err := se.backupClient.Status(ctx, targetPod.Status.PodIP)
+		if err != nil {
+			return nil, fmt.Errorf("while getting status: %w", err)
+		}
+
+		if status.Phase == webserver.Started {
+			if err := se.backupClient.Stop(ctx, targetPod.Status.PodIP); err != nil {
+				return nil, fmt.Errorf("while stopping the backup client: %w", err)
+			}
+			return &ctrl.Result{RequeueAfter: time.Second * 5}, nil
+		}
+
+		if status.Phase != webserver.Completed {
+			return &ctrl.Result{RequeueAfter: time.Second}, nil
+		}
+
+		backup.Status.BeginLSN = string(status.BeginLSN)
+		backup.Status.EndLSN = string(status.EndLSN)
+		backup.Status.TablespaceMapFile = status.SpcmapFile
+		backup.Status.BackupLabelFile = status.LabelFile
+	}
+
+	backup.Status.SetAsCompleted()
+	backup.Status.Online = ptr.To(volumeSnapshotConfig.GetOnline())
+	snapshots, err := getBackupVolumeSnapshots(ctx, se.cli, backup.Namespace, backup.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	backup.Status.BackupSnapshotStatus.SetSnapshotElements(snapshots)
+	if err := backupStatusFromSnapshots(snapshots, &backup.Status); err != nil {
+		contextLogger.Error(err, "while enriching the backup status")
+	}
+
+	if err := annotateSnapshotsWithBackupData(ctx, se.cli, snapshots, &backup.Status); err != nil {
+		contextLogger.Error(err, "while enriching the snapshots's status")
+		return &ctrl.Result{RequeueAfter: 1 * time.Second}, nil
+	}
+
+	return nil, postgres.PatchBackupStatusAndRetry(ctx, se.cli, backup)
+}
+
+// AnnotateSnapshots adds labels and annotations to the snapshots using the backup
+// status to facilitate access
+func annotateSnapshotsWithBackupData(
+	ctx context.Context,
+	cli client.Client,
+	snapshots slice,
+	backupStatus *apiv1.BackupStatus,
+) error {
+	contextLogger := log.FromContext(ctx)
+	for idx := range snapshots {
+		snapshot := &snapshots[idx]
+		oldSnapshot := snapshot.DeepCopy()
+		snapshot.Annotations[utils.BackupStartTimeAnnotationName] = backupStatus.StartedAt.Format(time.RFC3339)
+		snapshot.Annotations[utils.BackupEndTimeAnnotationName] = backupStatus.StoppedAt.Format(time.RFC3339)
+
+		if len(backupStatus.BackupLabelFile) > 0 {
+			snapshot.Annotations[utils.BackupLabelFileAnnotationName] = base64.StdEncoding.EncodeToString(
+				backupStatus.BackupLabelFile)
+		}
+
+		if len(backupStatus.TablespaceMapFile) > 0 {
+			snapshot.Annotations[utils.BackupTablespaceMapFileAnnotationName] = base64.StdEncoding.EncodeToString(
+				backupStatus.TablespaceMapFile)
+		}
+
+		if err := cli.Patch(ctx, snapshot, client.MergeFrom(oldSnapshot)); err != nil {
+			contextLogger.Error(err, "while updating volume snapshot from backup object",
+				"snapshot", snapshot.Name)
+			return err
+		}
+	}
+	return nil
+}
+
+// backupStatusFromSnapshots adds fields to the backup status based on the snapshots
+func backupStatusFromSnapshots(
+	snapshots slice,
+	backupStatus *apiv1.BackupStatus,
+) error {
+	controldata, err := snapshots.getControldata()
+	if err != nil {
+		return err
+	}
+	pairs := utils.ParsePgControldataOutput(controldata)
+
+	// TODO: calculate wal in case of online backup
+	// the begin/end WAL and LSN are the same, since the instance was fenced
+	// for the snapshot
+	backupStatus.BeginWal = pairs["Latest checkpoint's REDO WAL file"]
+	backupStatus.EndWal = pairs["Latest checkpoint's REDO WAL file"]
+
+	if !backupStatus.GetOnline() {
+		backupStatus.BeginLSN = pairs["Latest checkpoint's REDO location"]
+		backupStatus.EndLSN = pairs["Latest checkpoint's REDO location"]
+	}
+
+	return nil
 }
 
 // ensurePodIsFenced checks if the preconditions for the execution of this step are
diff --git a/pkg/reconciler/backup/volumesnapshot/resources.go b/pkg/reconciler/backup/volumesnapshot/resources.go
index 5e35b5e1..5da1e540 100644
--- a/pkg/reconciler/backup/volumesnapshot/resources.go
+++ b/pkg/reconciler/backup/volumesnapshot/resources.go
@@ -18,7 +18,6 @@ package volumesnapshot
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	storagesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
@@ -27,10 +26,6 @@ import (
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 )
 
-// ErrUnexpectedFencedInstances is raised when there are more than one Pod fenced or when
-// there are no Pod fenced at all.
-var ErrUnexpectedFencedInstances = errors.New("expected one and only one instance to be fenced")
-
 // volumeSnapshotInfo host information about a volume snapshot
 type volumeSnapshotInfo struct {
 	// Error contains the raised error when the volume snapshot terminated
@@ -64,11 +59,11 @@ func (err volumeSnapshotError) Error() string {
 	return *err.InternalError.Message
 }
 
-// Slice represents a slice of []storagesnapshotv1.VolumeSnapshot
-type Slice []storagesnapshotv1.VolumeSnapshot
+// slice represents a slice of []storagesnapshotv1.VolumeSnapshot
+type slice []storagesnapshotv1.VolumeSnapshot
 
-// GetControldata retrieves the pg_controldata stored as an annotation in VolumeSnapshots
-func (s Slice) GetControldata() (string, error) {
+// getControldata retrieves the pg_controldata stored as an annotation in VolumeSnapshots
+func (s slice) getControldata() (string, error) {
 	for _, volumeSnapshot := range s {
 		pgControlData, ok := volumeSnapshot.Annotations[utils.PgControldataAnnotationName]
 		if !ok {
@@ -79,14 +74,14 @@ func (s Slice) GetControldata() (string, error) {
 	return "", fmt.Errorf("could not retrieve pg_controldata from any snapshot")
 }
 
-// GetBackupVolumeSnapshots extracts the list of volume snapshots related
+// getBackupVolumeSnapshots extracts the list of volume snapshots related
 // to a backup name
-func GetBackupVolumeSnapshots(
+func getBackupVolumeSnapshots(
 	ctx context.Context,
 	cli client.Client,
 	namespace string,
 	backupLabelName string,
-) (Slice, error) {
+) (slice, error) {
 	var list storagesnapshotv1.VolumeSnapshotList
 
 	if err := cli.List(
diff --git a/pkg/specs/jobs.go b/pkg/specs/jobs.go
index 9a311d21..f699f130 100644
--- a/pkg/specs/jobs.go
+++ b/pkg/specs/jobs.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/kballard/go-shellquote"
+	storagesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -136,13 +137,28 @@ func buildInitDBFlags(cluster apiv1.Cluster) (initCommand []string) {
 }
 
 // CreatePrimaryJobViaRestoreSnapshot creates a new primary instance in a Pod, restoring from a volumeSnapshot
-func CreatePrimaryJobViaRestoreSnapshot(cluster apiv1.Cluster, nodeSerial int, backup *apiv1.Backup) *batchv1.Job {
+func CreatePrimaryJobViaRestoreSnapshot(
+	cluster apiv1.Cluster,
+	nodeSerial int,
+	snapshot storagesnapshotv1.VolumeSnapshot,
+	backup *apiv1.Backup,
+) *batchv1.Job {
 	initCommand := []string{
 		"/controller/manager",
 		"instance",
 		"restoresnapshot",
 	}
 
+	if snapshot.Annotations[utils.BackupLabelFileAnnotationName] != "" {
+		flag := fmt.Sprintf("--backuplabel=%s", snapshot.Annotations[utils.BackupLabelFileAnnotationName])
+		initCommand = append(initCommand, flag)
+	}
+
+	if snapshot.Annotations[utils.BackupTablespaceMapFileAnnotationName] != "" {
+		flag := fmt.Sprintf("--tablespacemap=%s", snapshot.Annotations[utils.BackupTablespaceMapFileAnnotationName])
+		initCommand = append(initCommand, flag)
+	}
+
 	initCommand = append(initCommand, buildCommonInitJobFlags(cluster)...)
 
 	job := createPrimaryJob(cluster, nodeSerial, jobRoleSnapshotRecovery, initCommand)
diff --git a/pkg/utils/labels_annotations.go b/pkg/utils/labels_annotations.go
index 547e80fa..56f4db14 100644
--- a/pkg/utils/labels_annotations.go
+++ b/pkg/utils/labels_annotations.go
@@ -180,6 +180,12 @@ const (
 	// BackupEndTimeAnnotationName is the name of the annotation where a backup's end time is kept
 	BackupEndTimeAnnotationName = MetadataNamespace + "/backupEndTime"
 
+	// BackupLabelFileAnnotationName is the name of the annotation where the `backup_label` file is kept
+	BackupLabelFileAnnotationName = MetadataNamespace + "/backupLabelFile"
+
+	// BackupTablespaceMapFileAnnotationName is the name of the annotation where the `tablespace_map` file is kept
+	BackupTablespaceMapFileAnnotationName = MetadataNamespace + "/backupTablespaceMapFile"
+
 	// SnapshotStartTimeAnnotationName is the name of the annotation where a snapshot's start time is kept
 	SnapshotStartTimeAnnotationName = MetadataNamespace + "/snapshotStartTime"
 



Test patch:
diff --git a/controllers/backup_controller_test.go b/controllers/backup_controller_test.go
index b7084b9e..adea7859 100644
--- a/controllers/backup_controller_test.go
+++ b/controllers/backup_controller_test.go
@@ -18,19 +18,11 @@ package controllers
 
 import (
 	"context"
-	"time"
 
-	volumesnapshot "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/utils/ptr"
-	k8client "sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
-	schemeBuilder "github.com/cloudnative-pg/cloudnative-pg/internal/scheme"
-	volumesnapshotbackup "github.com/cloudnative-pg/cloudnative-pg/pkg/reconciler/backup/volumesnapshot"
-	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -255,40 +247,3 @@ var _ = Describe("backup_controller volumeSnapshot unit tests", func() {
 		})
 	})
 })
-
-var _ = Describe("annotateSnapshotsWithBackupData", func() {
-	var (
-		fakeClient   k8client.Client
-		snapshots    volumesnapshot.VolumeSnapshotList
-		backupStatus *apiv1.BackupStatus
-		startedAt    metav1.Time
-		stoppedAt    metav1.Time
-	)
-
-	BeforeEach(func() {
-		snapshots = volumesnapshot.VolumeSnapshotList{
-			Items: volumesnapshotbackup.Slice{
-				{ObjectMeta: metav1.ObjectMeta{Name: "snapshot-1", Annotations: map[string]string{"avoid": "nil"}}},
-				{ObjectMeta: metav1.ObjectMeta{Name: "snapshot-2", Annotations: map[string]string{"avoid": "nil"}}},
-			},
-		}
-		startedAt = metav1.Now()
-		stoppedAt = metav1.NewTime(time.Now().Add(time.Hour))
-		backupStatus = &apiv1.BackupStatus{
-			StartedAt: ptr.To(startedAt),
-			StoppedAt: ptr.To(stoppedAt),
-		}
-		fakeClient = fake.NewClientBuilder().WithScheme(schemeBuilder.BuildWithAllKnownScheme()).
-			WithLists(&snapshots).Build()
-	})
-
-	It("should update all snapshots with backup time annotations", func(ctx context.Context) {
-		err := annotateSnapshotsWithBackupData(ctx, fakeClient, snapshots.Items, backupStatus)
-		Expect(err).ToNot(HaveOccurred())
-
-		for _, snapshot := range snapshots.Items {
-			Expect(snapshot.Annotations[utils.BackupStartTimeAnnotationName]).To(BeEquivalentTo(startedAt.Format(time.RFC3339)))
-			Expect(snapshot.Annotations[utils.BackupEndTimeAnnotationName]).To(BeEquivalentTo(stoppedAt.Format(time.RFC3339)))
-		}
-	})
-})
diff --git a/pkg/reconciler/backup/volumesnapshot/reconciler_test.go b/pkg/reconciler/backup/volumesnapshot/reconciler_test.go
index 01181aa9..0d427294 100644
--- a/pkg/reconciler/backup/volumesnapshot/reconciler_test.go
+++ b/pkg/reconciler/backup/volumesnapshot/reconciler_test.go
@@ -17,8 +17,9 @@ limitations under the License.
 package volumesnapshot
 
 import (
+	"context"
 	"fmt"
-	"strconv"
+	"time"
 
 	storagesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	v1 "k8s.io/api/core/v1"
@@ -26,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/ptr"
+	k8client "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
@@ -66,14 +68,13 @@ var _ = Describe("Volumesnapshot reconciler", func() {
 	const (
 		namespace   = "test-namespace"
 		clusterName = "clusterName"
-		backupName  = "theBakcup"
+		backupName  = "theBackup"
 	)
 	var (
-		cluster       *apiv1.Cluster
-		targetPod     *v1.Pod
-		pvcs          []v1.PersistentVolumeClaim
-		backup        *apiv1.Backup
-		backupCounter = 1
+		cluster   *apiv1.Cluster
+		targetPod *v1.Pod
+		pvcs      []v1.PersistentVolumeClaim
+		backup    *apiv1.Backup
 	)
 
 	BeforeEach(func() {
@@ -88,6 +89,7 @@ var _ = Describe("Volumesnapshot reconciler", func() {
 				Backup: &apiv1.BackupConfiguration{
 					VolumeSnapshot: &apiv1.VolumeSnapshotConfiguration{
 						ClassName: "csi-hostpath-snapclass",
+						Online:    ptr.To(false),
 					},
 				},
 			},
@@ -118,24 +120,30 @@ var _ = Describe("Volumesnapshot reconciler", func() {
 				},
 			},
 		}
+		startedAt := metav1.Now()
+		stoppedAt := metav1.NewTime(time.Now().Add(time.Hour))
+
 		backup = &apiv1.Backup{
 			ObjectMeta: metav1.ObjectMeta{
 				Namespace: namespace,
-				Name:      backupName + strconv.Itoa(backupCounter),
+				Name:      backupName,
+			},
+			Status: apiv1.BackupStatus{
+				StartedAt: ptr.To(startedAt),
+				StoppedAt: ptr.To(stoppedAt),
 			},
 		}
-		backupCounter++
 	})
 
 	It("should fence the target pod when there are no volumesnapshots", func(ctx SpecContext) {
 		mockClient := fake.NewClientBuilder().
 			WithScheme(scheme.BuildWithAllKnownScheme()).
-			WithObjects(cluster, targetPod).
+			WithObjects(backup, cluster, targetPod).
 			Build()
+
 		fakeRecorder := record.NewFakeRecorder(3)
 
 		executor := NewExecutorBuilder(mockClient, fakeRecorder).
-			FenceInstance(true).
 			Build()
 
 		result, err := executor.Execute(ctx, cluster, backup, targetPod, pvcs)
@@ -162,8 +170,9 @@ var _ = Describe("Volumesnapshot reconciler", func() {
 			Items: []storagesnapshotv1.VolumeSnapshot{
 				{
 					ObjectMeta: metav1.ObjectMeta{
-						Namespace: namespace,
-						Name:      backup.Name,
+						Namespace:   namespace,
+						Name:        backup.Name,
+						Annotations: map[string]string{},
 						Labels: map[string]string{
 							utils.BackupNameLabelName: backup.Name,
 						},
@@ -171,8 +180,9 @@ var _ = Describe("Volumesnapshot reconciler", func() {
 				},
 				{
 					ObjectMeta: metav1.ObjectMeta{
-						Namespace: namespace,
-						Name:      backup.Name + "-wal",
+						Namespace:   namespace,
+						Name:        backup.Name + "-wal",
+						Annotations: map[string]string{},
 						Labels: map[string]string{
 							utils.BackupNameLabelName: backup.Name,
 						},
@@ -184,12 +194,12 @@ var _ = Describe("Volumesnapshot reconciler", func() {
 		mockClient := fake.NewClientBuilder().
 			WithScheme(scheme.BuildWithAllKnownScheme()).
 			WithObjects(cluster, targetPod).
+			WithStatusSubresource(backup).
 			WithLists(&snapshots).
 			Build()
 		fakeRecorder := record.NewFakeRecorder(3)
 
 		executor := NewExecutorBuilder(mockClient, fakeRecorder).
-			FenceInstance(true).
 			Build()
 
 		result, err := executor.Execute(ctx, cluster, backup, targetPod, pvcs)
@@ -251,13 +261,13 @@ var _ = Describe("Volumesnapshot reconciler", func() {
 
 		mockClient := fake.NewClientBuilder().
 			WithScheme(scheme.BuildWithAllKnownScheme()).
-			WithObjects(cluster, targetPod).
+			WithObjects(cluster, targetPod, backup).
+			WithStatusSubresource(backup).
 			WithLists(&snapshots).
 			Build()
 		fakeRecorder := record.NewFakeRecorder(3)
 
 		executor := NewExecutorBuilder(mockClient, fakeRecorder).
-			FenceInstance(true).
 			Build()
 
 		result, err := executor.Execute(ctx, cluster, backup, targetPod, pvcs)
@@ -339,3 +349,40 @@ var _ = Describe("transferLabelsToAnnotations", func() {
 		Expect(labels).To(HaveKeyWithValue("nonMatchingLabel2", exampleValueTwo))
 	})
 })
+
+var _ = Describe("annotateSnapshotsWithBackupData", func() {
+	var (
+		fakeClient   k8client.Client
+		snapshots    storagesnapshotv1.VolumeSnapshotList
+		backupStatus *apiv1.BackupStatus
+		startedAt    metav1.Time
+		stoppedAt    metav1.Time
+	)
+
+	BeforeEach(func() {
+		snapshots = storagesnapshotv1.VolumeSnapshotList{
+			Items: slice{
+				{ObjectMeta: metav1.ObjectMeta{Name: "snapshot-1", Annotations: map[string]string{"avoid": "nil"}}},
+				{ObjectMeta: metav1.ObjectMeta{Name: "snapshot-2", Annotations: map[string]string{"avoid": "nil"}}},
+			},
+		}
+		startedAt = metav1.Now()
+		stoppedAt = metav1.NewTime(time.Now().Add(time.Hour))
+		backupStatus = &apiv1.BackupStatus{
+			StartedAt: ptr.To(startedAt),
+			StoppedAt: ptr.To(stoppedAt),
+		}
+		fakeClient = fake.NewClientBuilder().WithScheme(scheme.BuildWithAllKnownScheme()).
+			WithLists(&snapshots).Build()
+	})
+
+	It("should update all snapshots with backup time annotations", func(ctx context.Context) {
+		err := annotateSnapshotsWithBackupData(ctx, fakeClient, snapshots.Items, backupStatus)
+		Expect(err).ToNot(HaveOccurred())
+
+		for _, snapshot := range snapshots.Items {
+			Expect(snapshot.Annotations[utils.BackupStartTimeAnnotationName]).To(BeEquivalentTo(startedAt.Format(time.RFC3339)))
+			Expect(snapshot.Annotations[utils.BackupEndTimeAnnotationName]).To(BeEquivalentTo(stoppedAt.Format(time.RFC3339)))
+		}
+	})
+})
diff --git a/tests/e2e/fixtures/volume_snapshot/cluster-pvc-hot-restore.yaml.template b/tests/e2e/fixtures/volume_snapshot/cluster-pvc-hot-restore.yaml.template
new file mode 100644
index 00000000..1ca5e636
--- /dev/null
+++ b/tests/e2e/fixtures/volume_snapshot/cluster-pvc-hot-restore.yaml.template
@@ -0,0 +1,46 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-pvc-hot-recovery
+spec:
+  instances: 2
+  primaryUpdateStrategy: unsupervised
+  imageName: "${POSTGRES_IMG}"
+
+  # Persistent storage configuration
+  storage:
+    storageClass: ${E2E_CSI_STORAGE_CLASS}
+    size: 1Gi
+  walStorage:
+    storageClass: ${E2E_CSI_STORAGE_CLASS}
+    size: 1Gi
+
+  bootstrap:
+    recovery:
+      source: cluster-pvc-snapshot
+      volumeSnapshots:
+        storage:
+          name: ${SNAPSHOT_PITR_PGDATA}
+          kind: VolumeSnapshot
+          apiGroup: snapshot.storage.k8s.io
+        walStorage:
+          name: ${SNAPSHOT_PITR_PGWAL}
+          kind: VolumeSnapshot
+          apiGroup: snapshot.storage.k8s.io
+
+
+  externalClusters:
+    - name: cluster-pvc-snapshot
+      barmanObjectStore:
+        destinationPath: s3://cluster-backups/
+        endpointURL: https://minio-service:9000
+        endpointCA:
+          key: ca.crt
+          name: minio-server-ca-secret
+        s3Credentials:
+          accessKeyId:
+            name: backup-storage-creds
+            key: ID
+          secretAccessKey:
+            name: backup-storage-creds
+            key: KEY
diff --git a/tests/e2e/fixtures/volume_snapshot/cluster-pvc-hot-snapshot.yaml.template b/tests/e2e/fixtures/volume_snapshot/cluster-pvc-hot-snapshot.yaml.template
new file mode 100644
index 00000000..a42eb96f
--- /dev/null
+++ b/tests/e2e/fixtures/volume_snapshot/cluster-pvc-hot-snapshot.yaml.template
@@ -0,0 +1,46 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-pvc-snapshot
+spec:
+  instances: 2
+  primaryUpdateStrategy: unsupervised
+  imageName: "${POSTGRES_IMG}"
+
+  # Persistent storage configuration
+  storage:
+    storageClass: ${E2E_CSI_STORAGE_CLASS}
+    size: 1Gi
+  walStorage:
+    storageClass: ${E2E_CSI_STORAGE_CLASS}
+    size: 1Gi
+
+  backup:
+    volumeSnapshot:
+      className: ${E2E_DEFAULT_VOLUMESNAPSHOT_CLASS}
+      online: true
+      onlineConfiguration:
+        immediateCheckpoint: true
+        waitForArchive: true
+    barmanObjectStore:
+        destinationPath: s3://cluster-backups/
+        endpointURL: https://minio-service:9000
+        endpointCA:
+          key: ca.crt
+          name: minio-server-ca-secret
+        s3Credentials:
+          accessKeyId:
+            name: backup-storage-creds
+            key: ID
+          secretAccessKey:
+            name: backup-storage-creds
+            key: KEY
+        wal:
+          compression: gzip
+        data:
+          immediateCheckpoint: true
+        tags:
+          retention: "30days"
+        historyTags:
+          retention: "30days"
+    retentionPolicy: "30d"
diff --git a/tests/e2e/replica_mode_cluster_test.go b/tests/e2e/replica_mode_cluster_test.go
index f9c032ed..acc75f43 100644
--- a/tests/e2e/replica_mode_cluster_test.go
+++ b/tests/e2e/replica_mode_cluster_test.go
@@ -306,7 +306,7 @@ var _ = Describe("Replica Mode", Label(tests.LabelReplication), func() {
 				Expect(err).ToNot(HaveOccurred())
 				Expect(snapshotList.Items).To(HaveLen(len(backup.Status.BackupSnapshotStatus.Elements)))
 
-				err = testUtils.SetSnapshotNameAsEnv(&snapshotList, snapshotDataEnv, snapshotWalEnv)
+				err = testUtils.SetSnapshotNameAsEnv(&snapshotList, backup, snapshotDataEnv, snapshotWalEnv)
 				Expect(err).ToNot(HaveOccurred())
 			})
 
diff --git a/tests/e2e/volume_snapshot_test.go b/tests/e2e/volume_snapshot_test.go
index 309afdd1..96e5a50a 100644
--- a/tests/e2e/volume_snapshot_test.go
+++ b/tests/e2e/volume_snapshot_test.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	volumesnapshot "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	k8client "sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -41,11 +42,28 @@ import (
 // with different storage providers in different k8s environments
 var _ = Describe("Verify Volume Snapshot",
 	Label(tests.LabelBackupRestore, tests.LabelStorage, tests.LabelSnapshot), func() {
+		getSnapshots := func(
+			backupName string,
+			clusterName string,
+			namespace string,
+		) (volumesnapshot.VolumeSnapshotList, error) {
+			var snapshotList volumesnapshot.VolumeSnapshotList
+			err := env.Client.List(env.Ctx, &snapshotList, k8client.InNamespace(namespace),
+				k8client.MatchingLabels{
+					utils.ClusterLabelName:    clusterName,
+					utils.BackupNameLabelName: backupName,
+				})
+			if err != nil {
+				return snapshotList, err
+			}
+
+			return snapshotList, nil
+		}
+
 		// Initializing a global namespace variable to be used in each test case
 		var namespace string
 
 		Context("using the kubectl cnpg plugin", Ordered, func() {
-			// test env constants
 			const (
 				sampleFile      = fixturesDir + "/volume_snapshot/cluster-volume-snapshot.yaml.template"
 				namespacePrefix = "volume-snapshot"
@@ -254,14 +272,11 @@ var _ = Describe("Verify Volume Snapshot",
 				})
 
 				By("fetching the volume snapshots", func() {
-					snapshotList := volumesnapshot.VolumeSnapshotList{}
-					err := env.Client.List(env.Ctx, &snapshotList, k8client.MatchingLabels{
-						utils.ClusterLabelName: clusterToSnapshotName,
-					})
+					snapshotList, err := getSnapshots(backup.Name, clusterToSnapshotName, namespace)
 					Expect(err).ToNot(HaveOccurred())
 					Expect(snapshotList.Items).To(HaveLen(len(backup.Status.BackupSnapshotStatus.Elements)))
 
-					err = testUtils.SetSnapshotNameAsEnv(&snapshotList, snapshotDataEnv, snapshotWalEnv)
+					err = testUtils.SetSnapshotNameAsEnv(&snapshotList, backup, snapshotDataEnv, snapshotWalEnv)
 					Expect(err).ToNot(HaveOccurred())
 				})
 
@@ -328,16 +343,13 @@ var _ = Describe("Verify Volume Snapshot",
 			var clusterToBackupName string
 
 			getAndVerifySnapshots := func(
-				backupName string,
 				clusterToBackup *apiv1.Cluster,
 				backup apiv1.Backup,
 			) volumesnapshot.VolumeSnapshotList {
 				snapshotList := volumesnapshot.VolumeSnapshotList{}
 				By("fetching the volume snapshots", func() {
-					err := env.Client.List(env.Ctx, &snapshotList, k8client.MatchingLabels{
-						utils.ClusterLabelName:    clusterToBackupName,
-						utils.BackupNameLabelName: backupName,
-					})
+					var err error
+					snapshotList, err = getSnapshots(backup.Name, clusterToBackup.Name, backup.Namespace)
 					Expect(err).ToNot(HaveOccurred())
 					Expect(snapshotList.Items).To(HaveLen(len(backup.Status.BackupSnapshotStatus.Elements)))
 				})
@@ -432,8 +444,8 @@ var _ = Describe("Verify Volume Snapshot",
 					Expect(err).ToNot(HaveOccurred())
 				})
 
-				snapshotList := getAndVerifySnapshots(backupName, clusterToBackup, backup)
-				err = testUtils.SetSnapshotNameAsEnv(&snapshotList, snapshotDataEnv, snapshotWalEnv)
+				snapshotList := getAndVerifySnapshots(clusterToBackup, backup)
+				err = testUtils.SetSnapshotNameAsEnv(&snapshotList, &backup, snapshotDataEnv, snapshotWalEnv)
 				Expect(err).ToNot(HaveOccurred())
 
 				clusterToRestoreName, err := env.GetResourceNameFromYAML(clusterToRestoreFilePath)
@@ -482,7 +494,7 @@ var _ = Describe("Verify Volume Snapshot",
 					Expect(err).ToNot(HaveOccurred())
 				})
 
-				_ = getAndVerifySnapshots(backupName, clusterToBackup, backup)
+				_ = getAndVerifySnapshots(clusterToBackup, backup)
 
 				By("ensuring cluster resumes after snapshot", func() {
 					AssertClusterIsReady(namespace, clusterToBackupName, testTimeouts[testUtils.ClusterIsReadyQuick], env)
@@ -546,11 +558,189 @@ var _ = Describe("Verify Volume Snapshot",
 					Expect(err).ToNot(HaveOccurred())
 				})
 
-				_ = getAndVerifySnapshots(backupName, clusterToBackup, backup)
+				_ = getAndVerifySnapshots(clusterToBackup, backup)
 
 				By("ensuring cluster resumes after snapshot", func() {
 					AssertClusterIsReady(namespace, clusterToBackupName, testTimeouts[testUtils.ClusterIsReadyQuick], env)
 				})
 			})
 		})
+
+		Context("Declarative Hot Backup", Ordered, func() {
+			// test env constants
+			const (
+				namespacePrefix = "volume-snapshot-recovery"
+				level           = tests.High
+				filesDir        = fixturesDir + "/volume_snapshot"
+				snapshotDataEnv = "SNAPSHOT_PITR_PGDATA"
+				snapshotWalEnv  = "SNAPSHOT_PITR_PGWAL"
+			)
+			// minio constants
+			const (
+				minioCaSecName  = "minio-server-ca-secret"
+				minioTLSSecName = "minio-server-tls-secret"
+			)
+			// file constants
+			const (
+				clusterToSnapshot = filesDir + "/cluster-pvc-hot-snapshot.yaml.template"
+			)
+
+			var clusterToSnapshotName string
+			BeforeAll(func() {
+				if testLevelEnv.Depth < int(level) {
+					Skip("Test depth is lower than the amount requested for this test")
+				}
+
+				if !(IsLocal() || IsGKE()) {
+					Skip("This test is only executed on gke, openshift and local")
+				}
+
+				var err error
+				clusterToSnapshotName, err = env.GetResourceNameFromYAML(clusterToSnapshot)
+				Expect(err).ToNot(HaveOccurred())
+
+				namespace, err = env.CreateUniqueNamespace(namespacePrefix)
+				Expect(err).ToNot(HaveOccurred())
+
+				DeferCleanup(func() error {
+					if CurrentSpecReport().Failed() {
+						env.DumpNamespaceObjects(namespace, "out/"+CurrentSpecReport().LeafNodeText+".log")
+					}
+					return env.DeleteNamespace(namespace)
+				})
+
+				By("creating ca and tls certificate secrets", func() {
+					// create CA certificates
+					_, caPair, err := testUtils.CreateSecretCA(namespace, clusterToSnapshotName, minioCaSecName, true, env)
+					Expect(err).ToNot(HaveOccurred())
+
+					// sign and create secret using CA certificate and key
+					serverPair, err := caPair.CreateAndSignPair("minio-service", certs.CertTypeServer,
+						[]string{"minio-service.internal.mydomain.net, minio-service.default.svc, minio-service.default,"},
+					)
+					Expect(err).ToNot(HaveOccurred())
+					serverSecret := serverPair.GenerateCertificateSecret(namespace, minioTLSSecName)
+					err = env.Client.Create(env.Ctx, serverSecret)
+					Expect(err).ToNot(HaveOccurred())
+				})
+
+				By("creating the credentials for minio", func() {
+					AssertStorageCredentialsAreCreated(namespace, "backup-storage-creds", "minio", "minio123")
+				})
+
+				By("setting up minio", func() {
+					setup, err := testUtils.MinioSSLSetup(namespace)
+					Expect(err).ToNot(HaveOccurred())
+					err = testUtils.InstallMinio(env, setup, uint(testTimeouts[testUtils.MinioInstallation]))
+					Expect(err).ToNot(HaveOccurred())
+				})
+
+				// Create the minio client pod and wait for it to be ready.
+				// We'll use it to check if everything is archived correctly
+				By("setting up minio client pod", func() {
+					minioClient := testUtils.MinioSSLClient(namespace)
+					err := testUtils.PodCreateAndWaitForReady(env, &minioClient, 240)
+					Expect(err).ToNot(HaveOccurred())
+				})
+
+				By("creating the cluster to snapshot", func() {
+					AssertCreateCluster(namespace, clusterToSnapshotName, clusterToSnapshot, env)
+				})
+
+				By("verify test connectivity to minio using barman-cloud-wal-archive script", func() {
+					primaryPod, err := env.GetClusterPrimary(namespace, clusterToSnapshotName)
+					Expect(err).ToNot(HaveOccurred())
+					Eventually(func() (bool, error) {
+						connectionStatus, err := testUtils.MinioTestConnectivityUsingBarmanCloudWalArchive(
+							namespace, clusterToSnapshotName, primaryPod.GetName(), "minio", "minio123")
+						if err != nil {
+							return false, err
+						}
+						return connectionStatus, nil
+					}, 60).Should(BeTrue())
+				})
+			})
+
+			It("should execute a backup with online set to true", func() {
+				const (
+					tableName                  = "online_test"
+					clusterSnapshotRestoreFile = filesDir + "/cluster-pvc-hot-restore.yaml.template"
+				)
+
+				DeferCleanup(func() error {
+					if err := os.Unsetenv(snapshotDataEnv); err != nil {
+						return err
+					}
+
+					return os.Unsetenv(snapshotWalEnv)
+				})
+
+				By("inserting test data and creating WALs on the cluster to be snapshotted", func() {
+					// Create a "test" table with values 1,2
+					AssertCreateTestData(namespace, clusterToSnapshotName, tableName, psqlClientPod)
+
+					// Insert 2 more rows which we expect not to be present at the end of the recovery
+					insertRecordIntoTable(namespace, clusterToSnapshotName, tableName, 3, psqlClientPod)
+					insertRecordIntoTable(namespace, clusterToSnapshotName, tableName, 4, psqlClientPod)
+
+					// Close and archive the current WAL file
+					AssertArchiveWalOnMinio(namespace, clusterToSnapshotName, clusterToSnapshotName)
+				})
+
+				var backup *apiv1.Backup
+				By("creating a snapshot and waiting until it's completed", func() {
+					var err error
+					backupName := fmt.Sprintf("%s-online", clusterToSnapshotName)
+					backup, err = testUtils.CreateBackup(
+						apiv1.Backup{
+							ObjectMeta: metav1.ObjectMeta{
+								Namespace: namespace,
+								Name:      backupName,
+							},
+							Spec: apiv1.BackupSpec{
+								Target:  apiv1.BackupTargetPrimary,
+								Method:  apiv1.BackupMethodVolumeSnapshot,
+								Cluster: apiv1.LocalObjectReference{Name: clusterToSnapshotName},
+							},
+						},
+						env,
+					)
+					Expect(err).ToNot(HaveOccurred())
+
+					Eventually(func(g Gomega) {
+						err = env.Client.Get(env.Ctx, types.NamespacedName{
+							Namespace: namespace,
+							Name:      backupName,
+						}, backup)
+						g.Expect(err).ToNot(HaveOccurred())
+						g.Expect(backup.Status.BackupSnapshotStatus.Elements).To(HaveLen(2))
+						g.Expect(backup.Status.Phase).To(BeEquivalentTo(apiv1.BackupPhaseCompleted))
+						g.Expect(backup.Status.BackupLabelFile).ToNot(BeEmpty())
+					}, testTimeouts[testUtils.VolumeSnapshotIsReady]).Should(Succeed())
+				})
+
+				By("fetching the volume snapshots", func() {
+					snapshotList, err := getSnapshots(backup.Name, clusterToSnapshotName, namespace)
+					Expect(err).ToNot(HaveOccurred())
+					Expect(snapshotList.Items).To(HaveLen(len(backup.Status.BackupSnapshotStatus.Elements)))
+
+					err = testUtils.SetSnapshotNameAsEnv(&snapshotList, backup, snapshotDataEnv, snapshotWalEnv)
+					Expect(err).ToNot(HaveOccurred())
+				})
+
+				clusterToRestoreName, err := env.GetResourceNameFromYAML(clusterSnapshotRestoreFile)
+				Expect(err).ToNot(HaveOccurred())
+
+				By("creating the cluster to be restored through snapshot and PITR", func() {
+					AssertCreateCluster(namespace, clusterToRestoreName, clusterSnapshotRestoreFile, env)
+					AssertClusterIsReady(namespace, clusterToRestoreName, testTimeouts[testUtils.ClusterIsReadySlow], env)
+				})
+
+				By("verifying the correct data exists in the restored cluster", func() {
+					restoredPrimary, err := env.GetClusterPrimary(namespace, clusterToRestoreName)
+					Expect(err).ToNot(HaveOccurred())
+					AssertDataExpectedCount(namespace, clusterToRestoreName, tableName, 4, restoredPrimary)
+				})
+			})
+		})
 	})
diff --git a/tests/utils/backup.go b/tests/utils/backup.go
index d1585131..78047589 100644
--- a/tests/utils/backup.go
+++ b/tests/utils/backup.go
@@ -513,6 +513,8 @@ func CreateOnDemandBackupViaKubectlPlugin(
 }
 
 // CreateOnDemandBackup creates a Backup resource for a given cluster name
+// Deprecated: Use CreateBackup.
+// TODO: eradicate
 func CreateOnDemandBackup(
 	namespace,
 	clusterName,
@@ -551,6 +553,22 @@ func CreateOnDemandBackup(
 	return backup, nil
 }
 
+// CreateBackup creates a Backup resource for a given cluster name
+func CreateBackup(
+	targetBackup apiv1.Backup,
+	env *TestingEnvironment,
+) (*apiv1.Backup, error) {
+	obj, err := CreateObject(env, &targetBackup)
+	if err != nil {
+		return nil, err
+	}
+	backup, ok := obj.(*apiv1.Backup)
+	if !ok {
+		return nil, fmt.Errorf("created object is not of Backup type: %T %v", obj, obj)
+	}
+	return backup, nil
+}
+
 // GetVolumeSnapshot gets a VolumeSnapshot given name and namespace
 func (env TestingEnvironment) GetVolumeSnapshot(
 	namespace,
diff --git a/tests/utils/storage.go b/tests/utils/storage.go
index ca68d176..b6da1290 100644
--- a/tests/utils/storage.go
+++ b/tests/utils/storage.go
@@ -25,6 +25,7 @@ import (
 	storagev1 "k8s.io/api/storage/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 )
 
@@ -93,11 +94,12 @@ func ObjectMatchesAnnotations(
 // as env variables
 func SetSnapshotNameAsEnv(
 	snapshotList *volumesnapshot.VolumeSnapshotList,
+	backup *apiv1.Backup,
 	dataSnapshotName,
 	walSnapshotName string,
 ) error {
-	if len(snapshotList.Items) > 2 {
-		return fmt.Errorf("cannot handle a snapshotList with more than 2 snapshots")
+	if len(snapshotList.Items) != len(backup.Status.BackupSnapshotStatus.Elements) {
+		return fmt.Errorf("could not find the expected number of snapshots")
 	}
 
 	for _, item := range snapshotList.Items {