diff --git a/docs/examples/backup-example.yaml b/docs/examples/backup-example.yaml
new file mode 100644
index 0000000..5508254
--- /dev/null
+++ b/docs/examples/backup-example.yaml
@@ -0,0 +1,12 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Backup
+metadata:
+  name: backup-example
+spec:
+  method: plugin
+
+  cluster:
+    name: cluster-example
+
+  pluginConfiguration:
+    name: barman-cloud.cloudnative-pg.io
\ No newline at end of file
diff --git a/internal/cnpgi/instance/backup.go b/internal/cnpgi/instance/backup.go
index 30b579a..075397d 100644
--- a/internal/cnpgi/instance/backup.go
+++ b/internal/cnpgi/instance/backup.go
@@ -2,31 +2,46 @@ package instance
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"strconv"
+	"time"
 
 	barmanBackup "github.com/cloudnative-pg/barman-cloud/pkg/backup"
 	barmanCapabilities "github.com/cloudnative-pg/barman-cloud/pkg/capabilities"
 	barmanCredentials "github.com/cloudnative-pg/barman-cloud/pkg/credentials"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/postgres"
-	"github.com/cloudnative-pg/cnpg-i-machinery/pkg/pluginhelper/decoder"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 	"github.com/cloudnative-pg/cnpg-i/pkg/backup"
 	"github.com/cloudnative-pg/machinery/pkg/fileutils"
 	"github.com/cloudnative-pg/machinery/pkg/log"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
 )
 
 // BackupServiceImplementation is the implementation
 // of the Backup CNPG capability
 type BackupServiceImplementation struct {
-	Client       client.Client
-	InstanceName string
+	BarmanObjectKey  client.ObjectKey
+	ClusterObjectKey client.ObjectKey
+	Client           client.Client
+	InstanceName     string
 	backup.UnimplementedBackupServer
 }
 
+// This is an implementation of the barman executor
+// that always instruct the barman library to use the
+// "--name" option for backups. We don't support old
+// Barman versions that do not implement that option.
+type barmanCloudExecutor struct{}
+
+func (barmanCloudExecutor) ShouldForceLegacyBackup() bool {
+	return false
+}
+
 // GetCapabilities implements the BackupService interface
 func (b BackupServiceImplementation) GetCapabilities(
 	_ context.Context, _ *backup.BackupCapabilitiesRequest,
@@ -47,15 +62,12 @@ func (b BackupServiceImplementation) GetCapabilities(
 // Backup implements the Backup interface
 func (b BackupServiceImplementation) Backup(
 	ctx context.Context,
-	req *backup.BackupRequest,
+	_ *backup.BackupRequest,
 ) (*backup.BackupResult, error) {
 	contextLogger := log.FromContext(ctx)
-	backupObj, err := decoder.DecodeBackup(req.BackupDefinition)
-	if err != nil {
-		return nil, err
-	}
-	cluster, err := decoder.DecodeClusterJSON(req.ClusterDefinition)
-	if err != nil {
+
+	var objectStore barmancloudv1.ObjectStore
+	if err := b.Client.Get(ctx, b.BarmanObjectKey, &objectStore); err != nil {
 		return nil, err
 	}
 
@@ -69,33 +81,43 @@ func (b BackupServiceImplementation) Backup(
 		return nil, err
 	}
 	backupCmd := barmanBackup.NewBackupCommand(
-		cluster.Spec.Backup.BarmanObjectStore,
+		&objectStore.Spec.Configuration,
 		capabilities,
 	)
-	env := os.Environ()
-	env, err = barmanCredentials.EnvSetBackupCloudCredentials(
+
+	// We need to connect to PostgreSQL and to do that we need
+	// PGHOST (and the like) to be available
+	osEnvironment := os.Environ()
+	caBundleEnvironment := getRestoreCABundleEnv(&objectStore.Spec.Configuration)
+	env, err := barmanCredentials.EnvSetBackupCloudCredentials(
 		ctx,
 		b.Client,
-		cluster.Namespace,
-		cluster.Spec.Backup.BarmanObjectStore,
-		env)
+		objectStore.Namespace,
+		&objectStore.Spec.Configuration,
+		mergeEnv(osEnvironment, caBundleEnvironment))
 	if err != nil {
 		return nil, err
 	}
 
+	backupName := fmt.Sprintf("backup-%v", utils.ToCompactISO8601(time.Now()))
+
 	if err = backupCmd.Take(
 		ctx,
-		backupObj.Status.BackupName,
-		backupObj.Status.ServerName,
+		backupName,
+		b.InstanceName,
 		env,
-		cluster,
+		barmanCloudExecutor{},
 		postgres.BackupTemporaryDirectory,
 	); err != nil {
 		return nil, err
 	}
 
 	executedBackupInfo, err := backupCmd.GetExecutedBackupInfo(
-		ctx, backupObj.Status.BackupName, backupObj.Status.ServerName, cluster, env)
+		ctx,
+		backupName,
+		b.InstanceName,
+		barmanCloudExecutor{},
+		env)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/cnpgi/instance/start.go b/internal/cnpgi/instance/start.go
index 3299ed4..34cb89a 100644
--- a/internal/cnpgi/instance/start.go
+++ b/internal/cnpgi/instance/start.go
@@ -36,8 +36,10 @@ func (c *CNPGI) Start(ctx context.Context) error {
 			PGWALPath:        c.PGWALPath,
 		})
 		backup.RegisterBackupServer(server, BackupServiceImplementation{
-			Client:       c.Client,
-			InstanceName: c.InstanceName,
+			Client:           c.Client,
+			BarmanObjectKey:  c.BarmanObjectKey,
+			ClusterObjectKey: c.ClusterObjectKey,
+			InstanceName:     c.InstanceName,
 		})
 		return nil
 	}
diff --git a/internal/cnpgi/instance/wal.go b/internal/cnpgi/instance/wal.go
index 25eb19c..cd9e76d 100644
--- a/internal/cnpgi/instance/wal.go
+++ b/internal/cnpgi/instance/wal.go
@@ -124,9 +124,8 @@ func (w WALServiceImplementation) Restore(
 	contextLogger := log.FromContext(ctx)
 	startTime := time.Now()
 
-	var cluster *cnpgv1.Cluster
-
-	if err := w.Client.Get(ctx, w.ClusterObjectKey, cluster); err != nil {
+	var cluster cnpgv1.Cluster
+	if err := w.Client.Get(ctx, w.ClusterObjectKey, &cluster); err != nil {
 		return nil, err
 	}
 
@@ -152,7 +151,7 @@ func (w WALServiceImplementation) Restore(
 	if err != nil {
 		return nil, fmt.Errorf("while getting recover credentials: %w", err)
 	}
-	mergeEnv(env, credentialsEnv)
+	env = mergeEnv(env, credentialsEnv)
 
 	options, err := barmanCommand.CloudWalRestoreOptions(ctx, barmanConfiguration, objectStore.Name)
 	if err != nil {
@@ -178,7 +177,7 @@ func (w WALServiceImplementation) Restore(
 	}
 
 	// We skip this step if streaming connection is not available
-	if isStreamingAvailable(cluster, w.InstanceName) {
+	if isStreamingAvailable(&cluster, w.InstanceName) {
 		if err := checkEndOfWALStreamFlag(walRestorer); err != nil {
 			return nil, err
 		}
@@ -213,7 +212,7 @@ func (w WALServiceImplementation) Restore(
 
 	// We skip this step if streaming connection is not available
 	endOfWALStream := isEndOfWALStream(walStatus)
-	if isStreamingAvailable(cluster, w.InstanceName) && endOfWALStream {
+	if isStreamingAvailable(&cluster, w.InstanceName) && endOfWALStream {
 		contextLogger.Info(
 			"Set end-of-wal-stream flag as one of the WAL files to be prefetched was not found")
 
@@ -262,18 +261,29 @@ func (w WALServiceImplementation) SetFirstRequired(
 }
 
 // mergeEnv merges all the values inside incomingEnv into env.
-func mergeEnv(env []string, incomingEnv []string) {
+func mergeEnv(env []string, incomingEnv []string) []string {
+	result := make([]string, len(env), len(env)+len(incomingEnv))
+	copy(result, env)
+
 	for _, incomingItem := range incomingEnv {
 		incomingKV := strings.SplitAfterN(incomingItem, "=", 2)
 		if len(incomingKV) != 2 {
 			continue
 		}
-		for idx, item := range env {
+
+		found := false
+		for idx, item := range result {
 			if strings.HasPrefix(item, incomingKV[0]) {
-				env[idx] = incomingItem
+				result[idx] = incomingItem
+				found = true
 			}
 		}
+		if !found {
+			result = append(result, incomingItem)
+		}
 	}
+
+	return result
 }
 
 // TODO: refactor.
diff --git a/scripts/cleanup.sh b/scripts/cleanup.sh
new file mode 100755
index 0000000..f76d63b
--- /dev/null
+++ b/scripts/cleanup.sh
@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+
+set -eu
+
+cd "$(dirname "$0")/.." || exit
+
+kubectl delete clusters --all
+kubectl delete backups --all
+kubectl exec -ti mc -- mc rm -r --force minio/backups
\ No newline at end of file
diff --git a/docs/minio/minio-delete.sh b/scripts/minio-delete.sh
old mode 100644
new mode 100755
similarity index 100%
rename from docs/minio/minio-delete.sh
rename to scripts/minio-delete.sh
diff --git a/scripts/run.sh b/scripts/run.sh
index 730bc53..0dceefb 100755
--- a/scripts/run.sh
+++ b/scripts/run.sh
@@ -8,15 +8,19 @@ if [ -f .env ]; then
     source .env
 fi
 
+
+MYTMPDIR="$(mktemp -d)"
+trap 'rm -rf -- "$MYTMPDIR"' EXIT
+
 current_context=$(kubectl config view --raw -o json | jq -r '."current-context"' | sed "s/kind-//")
 operator_image=$(KIND_CLUSTER_NAME="$current_context" KO_DOCKER_REPO=kind.local ko build -BP ./cmd/operator)
 instance_image=$(KIND_CLUSTER_NAME="$current_context" KO_DOCKER_REPO=kind.local KO_DEFAULTBASEIMAGE="ghcr.io/cloudnative-pg/postgresql:17.0" ko build -BP ./cmd/instance)
 
+# Now we deploy the plugin inside the `cnpg-system` workspace
 (
-  cd kubernetes;
+  cp -r kubernetes config "$MYTMPDIR"
+  cd "$MYTMPDIR/kubernetes"
   kustomize edit set image "plugin-barman-cloud=$operator_image"
   kustomize edit set secret plugin-barman-cloud "--from-literal=SIDECAR_IMAGE=$instance_image"
+  kubectl apply -k .
 )
-
-# Now we deploy the plugin inside the `cnpg-system` workspace
-kubectl apply -k kubernetes/