fix: honour the object store requested by a Backup resource

internal/cnpgi/instance/backup.go

@@ -81,6 +81,8 @@ func (b BackupServiceImplementation) Backup(
 		return nil, err
 	}
 
+	configuration.ApplyBackupParameters(request.GetParameters())
+
 	var objectStore barmancloudv1.ObjectStore
 	if err := b.Client.Get(ctx, configuration.GetBarmanObjectKey(), &objectStore); err != nil {
 		contextLogger.Error(err, "while getting object store", "key", configuration.GetRecoveryBarmanObjectKey())

internal/cnpgi/operator/config/config.go

@@ -76,6 +76,12 @@ type PluginConfiguration struct {
 
 	ReplicaSourceBarmanObjectName string
 	ReplicaSourceServerName       string
+
+	// AdditionalBarmanObjectNames lists the object stores that a Backup
+	// resource is allowed to request on top of the ones used by the cluster
+	// itself. Nothing is written to them unless a Backup asks for one, but
+	// they take part in the RBAC and in the certificates of the instances.
+	AdditionalBarmanObjectNames []string
 }
 
 // GetBarmanObjectKey gets the namespaced name of the barman object
@@ -86,6 +92,24 @@ func (config *PluginConfiguration) GetBarmanObjectKey() types.NamespacedName {
 	}
 }
 
+// ApplyBackupParameters overrides the object store selection with the
+// parameters of the Backup resource. The operator relays them in the
+// BackupRequest, and without this a Backup asking for a different object
+// store is silently written to the cluster one.
+func (config *PluginConfiguration) ApplyBackupParameters(parameters map[string]string) {
+	if len(parameters) == 0 {
+		return
+	}
+
+	if value := parameters["barmanObjectName"]; len(value) > 0 {
+		config.BarmanObjectName = value
+	}
+
+	if value := parameters["serverName"]; len(value) > 0 {
+		config.ServerName = value
+	}
+}
+
 // GetRecoveryBarmanObjectKey gets the namespaced name of the recovery barman object
 func (config *PluginConfiguration) GetRecoveryBarmanObjectKey() types.NamespacedName {
 	return types.NamespacedName{
@@ -115,8 +139,11 @@ func (config *PluginConfiguration) GetReferredBarmanObjectsKey() []types.Namespa
 	if len(config.ReplicaSourceBarmanObjectName) > 0 {
 		objectNames.Put(config.ReplicaSourceBarmanObjectName)
 	}
+	for _, name := range config.AdditionalBarmanObjectNames {
+		objectNames.Put(name)
+	}
 
-	result := make([]types.NamespacedName, 0, 3)
+	result := make([]types.NamespacedName, 0, 4)
 	for _, name := range objectNames.ToSortedList() {
 		result = append(result, types.NamespacedName{
 			Name:      name,
@@ -179,6 +206,8 @@ func NewFromCluster(cluster *cnpgv1.Cluster) *PluginConfiguration {
 		// used for the backup/archive
 		BarmanObjectName: helper.Parameters["barmanObjectName"],
 		ServerName:       serverName,
+		// reachable by a Backup resource requesting them explicitly
+		AdditionalBarmanObjectNames: parseObjectNameList(helper.Parameters["additionalBarmanObjectNames"]),
 		// used for restore and wal_restore during backup recovery
 		RecoveryServerName:       recoveryServerName,
 		RecoveryBarmanObjectName: recoveryBarmanObjectName,
@@ -190,6 +219,23 @@ func NewFromCluster(cluster *cnpgv1.Cluster) *PluginConfiguration {
 	return result
 }
 
+// parseObjectNameList splits a comma separated list of object store names,
+// dropping the empty entries
+func parseObjectNameList(value string) []string {
+	if len(value) == 0 {
+		return nil
+	}
+
+	var result []string
+	for _, name := range strings.Split(value, ",") {
+		if name = strings.TrimSpace(name); len(name) > 0 {
+			result = append(result, name)
+		}
+	}
+
+	return result
+}
+
 func getRecoveryParameters(cluster *cnpgv1.Cluster) map[string]string {
 	recoveryPluginConfiguration := getRecoverySourcePlugin(cluster)
 	if recoveryPluginConfiguration == nil {

internal/cnpgi/operator/config/config_test.go

@@ -124,3 +124,79 @@ var _ = Describe("NewFromCluster", func() {
 		Expect(cfg.Validate()).NotTo(Succeed())
 	})
 })
+
+var _ = Describe("PluginConfiguration.ApplyBackupParameters", func() {
+	newConfiguration := func(parameters map[string]string) *PluginConfiguration {
+		return NewFromCluster(&cnpgv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{Name: "cluster-example", Namespace: "test-ns"},
+			Spec: cnpgv1.ClusterSpec{
+				Plugins: []cnpgv1.PluginConfiguration{
+					{Name: metadata.PluginName, Parameters: parameters},
+				},
+			},
+		})
+	}
+
+	It("sends the backup to the object store requested by the Backup resource", func() {
+		cfg := newConfiguration(map[string]string{"barmanObjectName": "minio-store"})
+
+		cfg.ApplyBackupParameters(map[string]string{"barmanObjectName": "archive-store"})
+
+		Expect(cfg.GetBarmanObjectKey().Name).To(Equal("archive-store"))
+	})
+
+	It("overrides the server name too", func() {
+		cfg := newConfiguration(map[string]string{"barmanObjectName": "minio-store"})
+
+		cfg.ApplyBackupParameters(map[string]string{"serverName": "another-name"})
+
+		Expect(cfg.ServerName).To(Equal("another-name"))
+		Expect(cfg.GetBarmanObjectKey().Name).To(Equal("minio-store"))
+	})
+
+	It("keeps the cluster object store when the Backup carries no parameters", func() {
+		cfg := newConfiguration(map[string]string{"barmanObjectName": "minio-store"})
+
+		cfg.ApplyBackupParameters(nil)
+
+		Expect(cfg.GetBarmanObjectKey().Name).To(Equal("minio-store"))
+		Expect(cfg.ServerName).To(Equal("cluster-example"))
+	})
+})
+
+var _ = Describe("Additional object stores", func() {
+	newConfiguration := func(parameters map[string]string) *PluginConfiguration {
+		return NewFromCluster(&cnpgv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{Name: "cluster-example", Namespace: "test-ns"},
+			Spec: cnpgv1.ClusterSpec{
+				Plugins: []cnpgv1.PluginConfiguration{
+					{Name: metadata.PluginName, Parameters: parameters},
+				},
+			},
+		})
+	}
+
+	It("are referred to, so that they are covered by RBAC and certificates", func() {
+		cfg := newConfiguration(map[string]string{
+			"barmanObjectName":            "minio-store",
+			"additionalBarmanObjectNames": "archive-store, monthly-store ,",
+		})
+
+		names := make([]string, 0, 3)
+		for _, key := range cfg.GetReferredBarmanObjectsKey() {
+			Expect(key.Namespace).To(Equal("test-ns"))
+			names = append(names, key.Name)
+		}
+
+		Expect(names).To(ConsistOf("minio-store", "archive-store", "monthly-store"))
+	})
+
+	It("do not receive anything on their own", func() {
+		cfg := newConfiguration(map[string]string{
+			"barmanObjectName":            "minio-store",
+			"additionalBarmanObjectNames": "archive-store",
+		})
+
+		Expect(cfg.GetBarmanObjectKey().Name).To(Equal("minio-store"))
+	})
+})