Merge 97675a7685 into 0c687cbbf1

This PR contains the following updates:

| Package | Type | Update | Change |
|---|---|---|---|
|
[github.com/cloudnative-pg/barman-cloud](https://redirect.github.com/cloudnative-pg/barman-cloud)
| require | digest | `53ffbf0` → `8d9302e` |

Signed-off-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>

.github/workflows/ci.yml

@@ -32,7 +32,7 @@ jobs:
           df -h
           echo "-----------------------------------------------------"
       - name: Checkout
-        uses: actions/checkout@v6
+        uses: actions/checkout@v7.0.0
         # We need the full history for the commitlint task
         with:
           fetch-depth: 0

.github/workflows/publish-docs.yml

@@ -25,7 +25,7 @@ jobs:
       contents: read
     steps:
       - name: Checkout
-        uses: actions/checkout@v6
+        uses: actions/checkout@v7.0.0
 
       - name: Setup Node
         uses: actions/setup-node@v6

.github/workflows/release-please.yml

@@ -19,7 +19,7 @@ jobs:
       #   We'll use the cli until there's a fix for
       #   https://github.com/googleapis/release-please/issues/2280.
       - name: Checkout
-        uses: actions/checkout@v6
+        uses: actions/checkout@v7.0.0
       - name: Install node
         uses: actions/setup-node@v6
         with:

.github/workflows/release-publish.yml

@@ -13,7 +13,7 @@ jobs:
       contents: write
     steps:
       - name: Checkout
-        uses: actions/checkout@v6
+        uses: actions/checkout@v7.0.0
       - name: Install QEMU static binaries
         uses: docker/setup-qemu-action@v4
       - name: Install Task

Makefile

@@ -159,7 +159,7 @@ GOLANGCI_LINT = $(LOCALBIN)/golangci-lint
 
 ## Tool Versions
 KUSTOMIZE_VERSION ?= v5.4.3
-CONTROLLER_TOOLS_VERSION ?= v0.16.1
+CONTROLLER_TOOLS_VERSION ?= v0.19.0
 ENVTEST_VERSION ?= release-0.19
 GOLANGCI_LINT_VERSION ?= v1.64.8
 

Taskfile.yml

@@ -46,7 +46,7 @@ tasks:
       - wordlist-ordered
     env:
       # renovate: datasource=git-refs depName=spellcheck lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_SPELLCHECK_SHA: 0a3b65e4cad754d9abdecbe5cf8d443dba63477c
+      DAGGER_SPELLCHECK_SHA: 679d5056358d3cd838ec226333decff673602937
     cmds:
       - >
         GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/spellcheck@${DAGGER_SPELLCHECK_SHA}
@@ -60,7 +60,7 @@ tasks:
     desc: Check for conventional commits
     env:
       # renovate: datasource=git-refs depName=commitlint lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_COMMITLINT_SHA: 0a3b65e4cad754d9abdecbe5cf8d443dba63477c
+      DAGGER_COMMITLINT_SHA: 679d5056358d3cd838ec226333decff673602937
     cmds:
       - >
         GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/commitlint@${DAGGER_COMMITLINT_SHA}
@@ -74,7 +74,7 @@ tasks:
       - wordlist-ordered
     env:
       # renovate: datasource=git-refs depName=uncommitted lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_UNCOMMITTED_SHA: 0a3b65e4cad754d9abdecbe5cf8d443dba63477c
+      DAGGER_UNCOMMITTED_SHA: 679d5056358d3cd838ec226333decff673602937
     cmds:
       - GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/uncommitted@${DAGGER_UNCOMMITTED_SHA} check-uncommitted --source . stdout
     sources:
@@ -86,7 +86,7 @@ tasks:
       - controller-gen
     env:
       # renovate: datasource=git-refs depName=crd-gen-refs lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_CRDGENREF_SHA: 0a3b65e4cad754d9abdecbe5cf8d443dba63477c
+      DAGGER_CRDGENREF_SHA: 679d5056358d3cd838ec226333decff673602937
       # renovate: datasource=go depName=github.com/elastic/crd-ref-docs
       CRDREFDOCS_VERSION: v0.3.0
     cmds:
@@ -381,7 +381,7 @@ tasks:
     run: once
     env:
       # renovate: datasource=git-refs depName=controller-gen lookupName=https://github.com/cloudnative-pg/daggerverse currentValue=main
-      DAGGER_CONTROLLER_GEN_SHA: 0a3b65e4cad754d9abdecbe5cf8d443dba63477c
+      DAGGER_CONTROLLER_GEN_SHA: 679d5056358d3cd838ec226333decff673602937
     cmds:
       - >
         GITHUB_REF= dagger -s call -m github.com/cloudnative-pg/daggerverse/controller-gen@${DAGGER_CONTROLLER_GEN_SHA}

api/v1/objectstore_types.go

@@ -94,6 +94,9 @@ type RecoveryWindow struct {
 
 	// The last failed backup time
 	LastFailedBackupTime *metav1.Time `json:"lastFailedBackupTime,omitempty"`
+
+	// The last time a WAL file was successfully archived by this plugin
+	LastArchivedWALTime *metav1.Time `json:"lastArchivedWALTime,omitempty"`
 }
 
 // +kubebuilder:object:root=true

api/v1/zz_generated.deepcopy.go

@@ -169,6 +169,10 @@ func (in *RecoveryWindow) DeepCopyInto(out *RecoveryWindow) {
 		in, out := &in.LastFailedBackupTime, &out.LastFailedBackupTime
 		*out = (*in).DeepCopy()
 	}
+	if in.LastArchivedWALTime != nil {
+		in, out := &in.LastArchivedWALTime, &out.LastArchivedWALTime
+		*out = (*in).DeepCopy()
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RecoveryWindow.

config/crd/bases/barmancloud.cnpg.io_objectstores.yaml

@@ -696,6 +696,11 @@ spec:
                         restored.
                       format: date-time
                       type: string
+                    lastArchivedWALTime:
+                      description: The last time a WAL file was successfully archived
+                        by this plugin
+                      format: date-time
+                      type: string
                     lastFailedBackupTime:
                       description: The last failed backup time
                       format: date-time

go.mod

@@ -5,7 +5,7 @@ go 1.26.3
 require (
 	github.com/cert-manager/cert-manager v1.20.2
 	github.com/cloudnative-pg/api v1.29.1
-	github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260618100956-53ffbf09b567
+	github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260619084538-8d9302e232e7
 	github.com/cloudnative-pg/cloudnative-pg v1.29.1
 	github.com/cloudnative-pg/cnpg-i v0.5.0
 	github.com/cloudnative-pg/cnpg-i-machinery v0.4.2

go.sum

@@ -20,8 +20,8 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cloudnative-pg/api v1.29.1 h1:FWr8S7EQeOfdhYXyr2cof8wUXLARZiiQt5Qa6ltED7w=
 github.com/cloudnative-pg/api v1.29.1/go.mod h1:QtWF3yzSvIfORMHaSkPAk/o3bhCJwEHJgN3riyRiz3o=
-github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260618100956-53ffbf09b567 h1:vqyRgqs1qDhOO2ZUEx3X4PKJGRnfwXN5xU5KSdjrnps=
-github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260618100956-53ffbf09b567/go.mod h1:vmW7blcg0BECd/9YKqn7NdiF7jz9BlXCWJcGJpzzBCg=
+github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260619084538-8d9302e232e7 h1:lNWqibnwlJesDrwAukg8gQF4Hpys3jx31LaXZFoP9vA=
+github.com/cloudnative-pg/barman-cloud v0.5.2-0.20260619084538-8d9302e232e7/go.mod h1:vmW7blcg0BECd/9YKqn7NdiF7jz9BlXCWJcGJpzzBCg=
 github.com/cloudnative-pg/cloudnative-pg v1.29.1 h1:ZNEt1TMlnQKXI1kho2UqQuqdfvIvjGln4kN7C1lsmGA=
 github.com/cloudnative-pg/cloudnative-pg v1.29.1/go.mod h1:Sbgx9jVmkle4/gR2U5JHrzDd74sRPOBHDtPkvncg5v8=
 github.com/cloudnative-pg/cnpg-i v0.5.0 h1:/TOzpNT6cwNgrpftTtrnLKdoHgMwd+88vZgXjlVgXeE=

hack/examples/cluster-replica-streaming.yaml

@@ -0,0 +1,36 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-replica
+spec:
+  instances: 3
+  bootstrap:
+    pg_basebackup:
+      source: source
+  replica:
+    enabled: true
+    source: source
+  externalClusters:
+  - name: source
+    connectionParameters:
+      host: cluster-example-rw.default.svc
+      user: streaming_replica
+      sslmode: verify-full
+      dbname: postgres
+    sslKey:
+      name: cluster-example-replication
+      key: tls.key
+    sslCert:
+      name: cluster-example-replication
+      key: tls.crt
+    sslRootCert:
+      name: cluster-example-ca
+      key: ca.crt
+    plugin:
+      name: barman-cloud.cloudnative-pg.io
+      parameters:
+        barmanObjectName: minio-store
+        serverName: cluster-example
+  storage:
+    size: 1Gi
+

internal/cnpgi/common/wal.go

@@ -25,6 +25,7 @@ import (
 	"fmt"
 	"os"
 	"path"
+	"sync"
 	"time"
 
 	"github.com/cloudnative-pg/barman-cloud/pkg/archiver"
@@ -38,7 +39,10 @@ import (
 	walUtils "github.com/cloudnative-pg/machinery/pkg/fileutils/wals"
 	"github.com/cloudnative-pg/machinery/pkg/log"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/retry"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
@@ -67,6 +71,11 @@ func (e *SpoolManagementError) Unwrap() error {
 	return e.err
 }
 
+const (
+	// walStatusUpdateThrottle is the minimum time between status updates for WAL archiving
+	walStatusUpdateThrottle = 5 * time.Minute
+)
+
 // WALServiceImplementation is the implementation of the WAL Service
 type WALServiceImplementation struct {
 	wal.UnimplementedWALServer
@@ -75,6 +84,9 @@ type WALServiceImplementation struct {
 	SpoolDirectory string
 	PGDataPath     string
 	PGWALPath      string
+	// LastStatusUpdate tracks the last time we updated the status for each ObjectStore+ServerName
+	// Key format: "namespace/objectStoreName/serverName"
+	LastStatusUpdate *sync.Map
 }
 
 // GetCapabilities implements the WALService interface
@@ -102,6 +114,37 @@ func (w WALServiceImplementation) GetCapabilities(
 	}, nil
 }
 
+// shouldUpdateStatus checks if we should update the status based on the throttle.
+// It returns true if walStatusUpdateThrottle minutes have passed since the last update, or if this is the first update.
+func (w WALServiceImplementation) shouldUpdateStatus(objectStoreKey client.ObjectKey, serverName string) bool {
+	if w.LastStatusUpdate == nil {
+		return true
+	}
+
+	key := fmt.Sprintf("%s/%s", objectStoreKey.String(), serverName)
+	lastUpdate, ok := w.LastStatusUpdate.Load(key)
+	if !ok {
+		return true
+	}
+
+	lastUpdateTime, ok := lastUpdate.(time.Time)
+	if !ok {
+		return true
+	}
+
+	return time.Since(lastUpdateTime) >= walStatusUpdateThrottle
+}
+
+// recordStatusUpdate records that we just updated the status for a given ObjectStore and server.
+func (w WALServiceImplementation) recordStatusUpdate(objectStoreKey client.ObjectKey, serverName string) {
+	if w.LastStatusUpdate == nil {
+		return
+	}
+
+	key := fmt.Sprintf("%s/%s", objectStoreKey.String(), serverName)
+	w.LastStatusUpdate.Store(key, time.Now())
+}
+
 // Archive implements the WALService interface
 func (w WALServiceImplementation) Archive(
 	ctx context.Context,
@@ -220,6 +263,28 @@ func (w WALServiceImplementation) Archive(
 		}
 	}
 
+	// Update the last archived WAL time in the ObjectStore status
+	// Only update if walStatusUpdateThrottle minutes have passed since the last update to avoid hitting the API server too often
+	objectStoreKey := configuration.GetBarmanObjectKey()
+	if w.shouldUpdateStatus(objectStoreKey, configuration.ServerName) {
+		contextLogger.Debug("Updating last archived WAL time", "serverName", configuration.ServerName)
+		if err := setLastArchivedWALTime(
+			ctx,
+			w.Client,
+			objectStoreKey,
+			configuration.ServerName,
+			time.Now(),
+		); err != nil {
+			// Log the error but don't fail the archive operation
+			contextLogger.Error(err, "Error updating last archived WAL time in ObjectStore status")
+		} else {
+			contextLogger.Debug("Successfully updated last archived WAL time")
+			w.recordStatusUpdate(objectStoreKey, configuration.ServerName)
+		}
+	} else {
+		contextLogger.Debug("Skipping status update due to throttle", "serverName", configuration.ServerName)
+	}
+
 	return &wal.WALArchiveResult{}, nil
 }
 
@@ -502,3 +567,30 @@ func isEndOfWALStream(results []barmanRestorer.Result) bool {
 
 	return false
 }
+
+// SetLastArchivedWALTime sets the last archived WAL time in the
+// passed object store, for the passed server name.
+func setLastArchivedWALTime(
+	ctx context.Context,
+	c client.Client,
+	objectStoreKey client.ObjectKey,
+	serverName string,
+	lastArchivedWALTime time.Time,
+) error {
+	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+		var objectStore barmancloudv1.ObjectStore
+
+		if err := c.Get(ctx, objectStoreKey, &objectStore); err != nil {
+			return err
+		}
+		recoveryWindow := objectStore.Status.ServerRecoveryWindow[serverName]
+		recoveryWindow.LastArchivedWALTime = ptr.To(metav1.NewTime(lastArchivedWALTime))
+
+		if objectStore.Status.ServerRecoveryWindow == nil {
+			objectStore.Status.ServerRecoveryWindow = make(map[string]barmancloudv1.RecoveryWindow)
+		}
+		objectStore.Status.ServerRecoveryWindow[serverName] = recoveryWindow
+
+		return c.Status().Update(ctx, &objectStore)
+	})
+}

internal/cnpgi/instance/metrics.go

@@ -51,6 +51,7 @@ var (
 	firstRecoverabilityPointMetricName     = buildFqName("first_recoverability_point")
 	lastAvailableBackupTimestampMetricName = buildFqName("last_available_backup_timestamp")
 	lastFailedBackupTimestampMetricName    = buildFqName("last_failed_backup_timestamp")
+	lastArchivedWALTimestampMetricName     = buildFqName("last_archived_wal_timestamp")
 )
 
 func (m metricsImpl) GetCapabilities(
@@ -97,6 +98,11 @@ func (m metricsImpl) Define(
 				Help:      "The last failed backup as a unix timestamp",
 				ValueType: &metrics.MetricType{Type: metrics.MetricType_TYPE_GAUGE},
 			},
+			{
+				FqName:    lastArchivedWALTimestampMetricName,
+				Help:      "The last archived WAL timestamp as a unix timestamp",
+				ValueType: &metrics.MetricType{Type: metrics.MetricType_TYPE_GAUGE},
+			},
 		},
 	}, nil
 }
@@ -136,6 +142,10 @@ func (m metricsImpl) Collect(
 					FqName: lastFailedBackupTimestampMetricName,
 					Value:  0,
 				},
+				{
+					FqName: lastArchivedWALTimestampMetricName,
+					Value:  0,
+				},
 			},
 		}, nil
 	}
@@ -143,6 +153,7 @@ func (m metricsImpl) Collect(
 	var firstRecoverabilityPoint float64
 	var lastAvailableBackup float64
 	var lastFailedBackup float64
+	var lastArchivedWAL float64
 	if x.FirstRecoverabilityPoint != nil {
 		firstRecoverabilityPoint = float64(x.FirstRecoverabilityPoint.Unix())
 	}
@@ -152,6 +163,9 @@ func (m metricsImpl) Collect(
 	if x.LastFailedBackupTime != nil {
 		lastFailedBackup = float64(x.LastFailedBackupTime.Unix())
 	}
+	if x.LastArchivedWALTime != nil {
+		lastArchivedWAL = float64(x.LastArchivedWALTime.Unix())
+	}
 
 	return &metrics.CollectMetricsResult{
 		Metrics: []*metrics.CollectMetric{
@@ -167,6 +181,10 @@ func (m metricsImpl) Collect(
 				FqName: lastFailedBackupTimestampMetricName,
 				Value:  lastFailedBackup,
 			},
+			{
+				FqName: lastArchivedWALTimestampMetricName,
+				Value:  lastArchivedWAL,
+			},
 		},
 	}, nil
 }

internal/cnpgi/instance/metrics_test.go

@@ -22,10 +22,11 @@ package instance
 import (
 	"context"
 	"encoding/json"
+	"time"
+
 	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
 	"k8s.io/utils/ptr"
-	"time"
 
 	"github.com/cloudnative-pg/cnpg-i/pkg/metrics"
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
@@ -117,7 +118,7 @@ var _ = Describe("Metrics Collect method", func() {
 		res, err := m.Collect(ctx, req)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(res).ToNot(BeNil())
-		Expect(res.Metrics).To(HaveLen(3))
+		Expect(res.Metrics).To(HaveLen(4))
 
 		// Verify the metrics
 		metricsMap := make(map[string]float64)
@@ -131,6 +132,13 @@ var _ = Describe("Metrics Collect method", func() {
 
 		expectedLastBackup, _ := metricsMap[lastAvailableBackupTimestampMetricName]
 		Expect(expectedLastBackup).To(BeNumerically("~", float64(objectStore.Status.ServerRecoveryWindow[clusterName].LastSuccessfulBackupTime.Unix()), 1))
+
+		// Check that unset timestamps are 0
+		expectedLastFailedBackup, _ := metricsMap[lastFailedBackupTimestampMetricName]
+		Expect(expectedLastFailedBackup).To(BeZero())
+
+		expectedLastArchivedWAL, _ := metricsMap[lastArchivedWALTimestampMetricName]
+		Expect(expectedLastArchivedWAL).To(BeZero())
 	})
 
 	It("should return an error if the object store is not found", func() {

internal/cnpgi/instance/start.go

@@ -21,6 +21,7 @@ package instance
 
 import (
 	"context"
+	"sync"
 
 	"github.com/cloudnative-pg/cnpg-i-machinery/pkg/pluginhelper/http"
 	"github.com/cloudnative-pg/cnpg-i/pkg/backup"
@@ -52,6 +53,7 @@ func (c *CNPGI) Start(ctx context.Context) error {
 			SpoolDirectory:   c.SpoolDirectory,
 			PGDataPath:       c.PGDataPath,
 			PGWALPath:        c.PGWALPath,
+			LastStatusUpdate: &sync.Map{},
 		})
 		backup.RegisterBackupServer(server, BackupServiceImplementation{
 			Client:       c.Client,

internal/cnpgi/operator/config/config.go

@@ -263,7 +263,9 @@ func getReplicaSourcePlugin(cluster *cnpgv1.Cluster) *cnpgv1.PluginConfiguration
 func (config *PluginConfiguration) Validate() error {
 	err := NewConfigurationError()
 
-	if len(config.BarmanObjectName) == 0 && len(config.RecoveryBarmanObjectName) == 0 {
+	if len(config.BarmanObjectName) == 0 &&
+		len(config.RecoveryBarmanObjectName) == 0 &&
+		len(config.ReplicaSourceBarmanObjectName) == 0 {
 		return err.WithMessage("no reference to barmanObjectName have been included")
 	}
 

internal/cnpgi/operator/config/config_test.go

@@ -0,0 +1,126 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+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.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package config
+
+import (
+	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
+)
+
+var _ = Describe("PluginConfiguration.Validate", func() {
+	It("fails when no barman object name is set", func() {
+		cfg := &PluginConfiguration{}
+		Expect(cfg.Validate()).To(HaveOccurred())
+	})
+
+	It("passes when only BarmanObjectName is set (backup/archive)", func() {
+		cfg := &PluginConfiguration{BarmanObjectName: "my-store"}
+		Expect(cfg.Validate()).To(Succeed())
+	})
+
+	It("passes when only RecoveryBarmanObjectName is set (recovery bootstrap)", func() {
+		cfg := &PluginConfiguration{RecoveryBarmanObjectName: "my-store"}
+		Expect(cfg.Validate()).To(Succeed())
+	})
+
+	It("passes when only ReplicaSourceBarmanObjectName is set (pg_basebackup replica cluster)", func() {
+		cfg := &PluginConfiguration{ReplicaSourceBarmanObjectName: "my-store"}
+		Expect(cfg.Validate()).To(Succeed())
+	})
+})
+
+var _ = Describe("NewFromCluster", func() {
+	enabled := true
+
+	It("derives the replica source object store for a pg_basebackup replica cluster", func() {
+		cluster := &cnpgv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{Name: "cluster-replica", Namespace: "test-ns"},
+			Spec: cnpgv1.ClusterSpec{
+				Bootstrap: &cnpgv1.BootstrapConfiguration{
+					PgBaseBackup: &cnpgv1.BootstrapPgBaseBackup{Source: "source"},
+				},
+				ReplicaCluster: &cnpgv1.ReplicaClusterConfiguration{
+					Enabled: &enabled,
+					Source:  "source",
+				},
+				ExternalClusters: []cnpgv1.ExternalCluster{
+					{
+						Name: "source",
+						PluginConfiguration: &cnpgv1.PluginConfiguration{
+							Name: metadata.PluginName,
+							Parameters: map[string]string{
+								"barmanObjectName": "minio-store",
+								"serverName":       "cluster-example",
+							},
+						},
+					},
+				},
+			},
+		}
+
+		cfg := NewFromCluster(cluster)
+
+		// The replica source object store is derived from the external cluster plugin,
+		// while the backup/archive and recovery object stores remain empty: this is the
+		// distinguishing trait of a pg_basebackup replica cluster (a recovery-bootstrapped
+		// replica would also populate RecoveryBarmanObjectName).
+		Expect(cfg.ReplicaSourceBarmanObjectName).To(Equal("minio-store"))
+		Expect(cfg.ReplicaSourceServerName).To(Equal("cluster-example"))
+		Expect(cfg.BarmanObjectName).To(BeEmpty())
+		Expect(cfg.RecoveryBarmanObjectName).To(BeEmpty())
+
+		// Validate must accept it, otherwise the lifecycle hook skips sidecar injection.
+		Expect(cfg.Validate()).To(Succeed())
+	})
+
+	It("ignores a replica source backed by a different plugin", func() {
+		cluster := &cnpgv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{Name: "cluster-replica", Namespace: "test-ns"},
+			Spec: cnpgv1.ClusterSpec{
+				Bootstrap: &cnpgv1.BootstrapConfiguration{
+					PgBaseBackup: &cnpgv1.BootstrapPgBaseBackup{Source: "source"},
+				},
+				ReplicaCluster: &cnpgv1.ReplicaClusterConfiguration{
+					Enabled: &enabled,
+					Source:  "source",
+				},
+				ExternalClusters: []cnpgv1.ExternalCluster{
+					{
+						Name: "source",
+						PluginConfiguration: &cnpgv1.PluginConfiguration{
+							Name:       "some-other-plugin.cloudnative-pg.io",
+							Parameters: map[string]string{"barmanObjectName": "minio-store"},
+						},
+					},
+				},
+			},
+		}
+
+		cfg := NewFromCluster(cluster)
+
+		Expect(cfg.ReplicaSourceBarmanObjectName).To(BeEmpty())
+		Expect(cfg.Validate()).NotTo(Succeed())
+	})
+})

internal/cnpgi/operator/config/suite_test.go

@@ -0,0 +1,32 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+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.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package config
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestConfig(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Config Suite")
+}

internal/cnpgi/operator/lifecycle.go

@@ -273,8 +273,10 @@ func (impl LifecycleImplementation) collectAdditionalInstanceArgs(
 		return args
 	}
 
-	// Prefer the cluster object store (backup/archive). If not set, fallback to the recovery object store.
-	// If neither is configured, no additional args are provided.
+	// Only one object store provides the sidecar arguments, with precedence:
+	// BarmanObjectName (backup/archive) > RecoveryBarmanObjectName (recovery
+	// bootstrap) > ReplicaSourceBarmanObjectName (pg_basebackup replica).
+	// If none is configured, no additional args are provided.
 	if len(pluginConfiguration.BarmanObjectName) > 0 {
 		var barmanObjectStore barmancloudv1.ObjectStore
 		if err := impl.Client.Get(ctx, pluginConfiguration.GetBarmanObjectKey(), &barmanObjectStore); err != nil {
@@ -303,6 +305,20 @@ func (impl LifecycleImplementation) collectAdditionalInstanceArgs(
 		return args, nil
 	}
 
+	if len(pluginConfiguration.ReplicaSourceBarmanObjectName) > 0 {
+		key := pluginConfiguration.GetReplicaSourceBarmanObjectKey()
+		var barmanObjectStore barmancloudv1.ObjectStore
+		if err := impl.Client.Get(ctx, key, &barmanObjectStore); err != nil {
+			return nil, fmt.Errorf("while getting replica source barman object store %s: %w", key.String(), err)
+		}
+		args := barmanObjectStore.Spec.InstanceSidecarConfiguration.AdditionalContainerArgs
+		args = append(
+			args,
+			collectTypedAdditionalArgs(&barmanObjectStore)...,
+		)
+		return args, nil
+	}
+
 	return nil, nil
 }
 

internal/cnpgi/operator/lifecycle_resources.go

@@ -48,6 +48,9 @@ func (impl LifecycleImplementation) collectSidecarResourcesForPod(
 	ctx context.Context,
 	configuration *config.PluginConfiguration,
 ) (corev1.ResourceRequirements, error) {
+	// Only one object store provides the sidecar resources, with precedence:
+	// BarmanObjectName (backup/archive) > RecoveryBarmanObjectName (recovery
+	// bootstrap) > ReplicaSourceBarmanObjectName (pg_basebackup replica).
 	if len(configuration.BarmanObjectName) > 0 {
 		// On a replica cluster that also archives, the designated primary
 		// will use both the replica source object store and the object store
@@ -64,10 +67,10 @@ func (impl LifecycleImplementation) collectSidecarResourcesForPod(
 	}
 
 	if len(configuration.RecoveryBarmanObjectName) > 0 {
-		// On a replica cluster that doesn't archive, the designated primary
-		// uses only the replica source object store.
-		// In this case, we use the replica source object store for configuring
-		// the resources of the sidecar container.
+		// On a cluster recovering from an object store (including log-shipping
+		// replica clusters, where the recovery and replica source object stores
+		// coincide), we use the recovery object store for configuring the
+		// resources of the sidecar container.
 		var barmanObjectStore barmancloudv1.ObjectStore
 		if err := impl.Client.Get(ctx, configuration.GetRecoveryBarmanObjectKey(), &barmanObjectStore); err != nil {
 			return corev1.ResourceRequirements{}, err
@@ -76,5 +79,18 @@ func (impl LifecycleImplementation) collectSidecarResourcesForPod(
 		return barmanObjectStore.Spec.InstanceSidecarConfiguration.Resources, nil
 	}
 
+	if len(configuration.ReplicaSourceBarmanObjectName) > 0 {
+		// On a replica cluster bootstrapped via pg_basebackup, the designated
+		// primary uses only the replica source object store.
+		// In this case, we use the replica source object store for configuring
+		// the resources of the sidecar container.
+		var barmanObjectStore barmancloudv1.ObjectStore
+		if err := impl.Client.Get(ctx, configuration.GetReplicaSourceBarmanObjectKey(), &barmanObjectStore); err != nil {
+			return corev1.ResourceRequirements{}, err
+		}
+
+		return barmanObjectStore.Spec.InstanceSidecarConfiguration.Resources, nil
+	}
+
 	return corev1.ResourceRequirements{}, nil
 }

internal/cnpgi/operator/lifecycle_test.go

@@ -28,6 +28,7 @@ import (
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
@@ -293,6 +294,24 @@ var _ = Describe("LifecycleImplementation", func() {
 			Expect(args).To(Equal(recoveryArgs))
 		})
 
+		It("falls back to replica source object store when primary and recovery not set", func(ctx SpecContext) {
+			ns := "test-ns"
+			cluster := &cnpgv1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: ns}}
+			pc := &config.PluginConfiguration{
+				Cluster:                       cluster,
+				ReplicaSourceBarmanObjectName: "replica-source-store",
+			}
+			replicaArgs := []string{"--replica-a", "--replica-b"}
+			cli := buildClientFunc(
+				makeStoreFunc(ns, pc.ReplicaSourceBarmanObjectName, replicaArgs),
+			).Build()
+
+			impl := LifecycleImplementation{Client: cli}
+			args, err := impl.collectAdditionalInstanceArgs(ctx, pc)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(args).To(Equal(replicaArgs))
+		})
+
 		It("returns nil when neither object name is configured", func(ctx SpecContext) {
 			ns := "test-ns"
 			cluster := &cnpgv1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: ns}}
@@ -386,6 +405,117 @@ var _ = Describe("LifecycleImplementation", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(args).To(Equal([]string{"--log-level=info"}))
 		})
+
+		It("includes --log-level from replica source object store when only replica source set", func(ctx SpecContext) {
+			ns := "test-ns"
+			cluster := &cnpgv1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: ns}}
+			pc := &config.PluginConfiguration{
+				Cluster:                       cluster,
+				ReplicaSourceBarmanObjectName: "replica-source-store",
+			}
+			store := &barmancloudv1.ObjectStore{
+				TypeMeta:   metav1.TypeMeta{Kind: "ObjectStore", APIVersion: barmancloudv1.GroupVersion.String()},
+				ObjectMeta: metav1.ObjectMeta{Name: pc.ReplicaSourceBarmanObjectName, Namespace: ns},
+				Spec: barmancloudv1.ObjectStoreSpec{
+					InstanceSidecarConfiguration: barmancloudv1.InstanceSidecarConfiguration{
+						LogLevel: "warning",
+					},
+				},
+			}
+			s := runtime.NewScheme()
+			barmancloudv1.AddKnownTypes(s)
+			cli := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(store).Build()
+
+			impl := LifecycleImplementation{Client: cli}
+			args, err := impl.collectAdditionalInstanceArgs(ctx, pc)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(args).To(Equal([]string{"--log-level=warning"}))
+		})
+	})
+
+	Describe("collectSidecarResourcesForPod", func() {
+		makeStoreWithResourcesFunc := func(ns, name string, res corev1.ResourceRequirements) *barmancloudv1.ObjectStore {
+			return &barmancloudv1.ObjectStore{
+				TypeMeta:   metav1.TypeMeta{Kind: "ObjectStore", APIVersion: barmancloudv1.GroupVersion.String()},
+				ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns},
+				Spec: barmancloudv1.ObjectStoreSpec{
+					InstanceSidecarConfiguration: barmancloudv1.InstanceSidecarConfiguration{
+						Resources: res,
+					},
+				},
+			}
+		}
+
+		It("uses the cluster object store resources when set", func(ctx SpecContext) {
+			ns := "test-ns"
+			cluster := &cnpgv1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: ns}}
+			pc := &config.PluginConfiguration{Cluster: cluster, BarmanObjectName: "primary-store"}
+			res := corev1.ResourceRequirements{
+				Requests: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("64Mi")},
+			}
+			cli := buildClientFunc(makeStoreWithResourcesFunc(ns, pc.BarmanObjectName, res)).Build()
+
+			impl := LifecycleImplementation{Client: cli}
+			got, err := impl.collectSidecarResourcesForPod(ctx, pc)
+			Expect(err).NotTo(HaveOccurred())
+			gotMem := got.Requests[corev1.ResourceMemory]
+			Expect(gotMem.String()).To(Equal("64Mi"))
+		})
+
+		It("uses the recovery object store resources when only recovery is set", func(ctx SpecContext) {
+			ns := "test-ns"
+			cluster := &cnpgv1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: ns}}
+			pc := &config.PluginConfiguration{Cluster: cluster, RecoveryBarmanObjectName: "recovery-store"}
+			res := corev1.ResourceRequirements{
+				Requests: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("128Mi")},
+			}
+			cli := buildClientFunc(makeStoreWithResourcesFunc(ns, pc.RecoveryBarmanObjectName, res)).Build()
+
+			impl := LifecycleImplementation{Client: cli}
+			got, err := impl.collectSidecarResourcesForPod(ctx, pc)
+			Expect(err).NotTo(HaveOccurred())
+			gotMem := got.Requests[corev1.ResourceMemory]
+			Expect(gotMem.String()).To(Equal("128Mi"))
+		})
+
+		It("uses the replica source object store resources when only replica source is set", func(ctx SpecContext) {
+			ns := "test-ns"
+			cluster := &cnpgv1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: ns}}
+			pc := &config.PluginConfiguration{Cluster: cluster, ReplicaSourceBarmanObjectName: "replica-source-store"}
+			res := corev1.ResourceRequirements{
+				Requests: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("256Mi")},
+			}
+			cli := buildClientFunc(makeStoreWithResourcesFunc(ns, pc.ReplicaSourceBarmanObjectName, res)).Build()
+
+			impl := LifecycleImplementation{Client: cli}
+			got, err := impl.collectSidecarResourcesForPod(ctx, pc)
+			Expect(err).NotTo(HaveOccurred())
+			gotMem := got.Requests[corev1.ResourceMemory]
+			Expect(gotMem.String()).To(Equal("256Mi"))
+		})
+
+		It("returns empty resources when no object store is configured", func(ctx SpecContext) {
+			ns := "test-ns"
+			cluster := &cnpgv1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: ns}}
+			pc := &config.PluginConfiguration{Cluster: cluster}
+			cli := buildClientFunc().Build()
+
+			impl := LifecycleImplementation{Client: cli}
+			got, err := impl.collectSidecarResourcesForPod(ctx, pc)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(got).To(Equal(corev1.ResourceRequirements{}))
+		})
+
+		It("returns an error if the replica source object store cannot be retrieved", func(ctx SpecContext) {
+			ns := "test-ns"
+			cluster := &cnpgv1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: ns}}
+			pc := &config.PluginConfiguration{Cluster: cluster, ReplicaSourceBarmanObjectName: "missing-store"}
+			cli := buildClientFunc().Build()
+
+			impl := LifecycleImplementation{Client: cli}
+			_, err := impl.collectSidecarResourcesForPod(ctx, pc)
+			Expect(err).To(HaveOccurred())
+		})
 	})
 })
 

internal/cnpgi/restore/start.go

@@ -22,6 +22,7 @@ package restore
 import (
 	"context"
 	"path"
+	"sync"
 
 	"github.com/cloudnative-pg/cnpg-i-machinery/pkg/pluginhelper/http"
 	restore "github.com/cloudnative-pg/cnpg-i/pkg/restore/job"
@@ -54,6 +55,7 @@ func (c *CNPGI) Start(ctx context.Context) error {
 			SpoolDirectory:   c.SpoolDirectory,
 			PGDataPath:       c.PGDataPath,
 			PGWALPath:        path.Join(c.PGDataPath, "pg_wal"),
+			LastStatusUpdate: &sync.Map{},
 		})
 
 		restore.RegisterRestoreJobHooksServer(server, &JobHookImpl{

web/yarn.lock

@@ -8396,9 +8396,9 @@ send@~0.19.0, send@~0.19.1:
     statuses "~2.0.2"
 
 serialize-javascript@>=7.0.5, serialize-javascript@^6.0.0, serialize-javascript@^6.0.1:
-  version "7.0.5"
-  resolved "https://registry.yarnpkg.com/serialize-javascript/-/serialize-javascript-7.0.5.tgz#c798cc0552ffbb08981914a42a8756e339d0d5b1"
-  integrity sha512-F4LcB0UqUl1zErq+1nYEEzSHJnIwb3AF2XWB94b+afhrekOUijwooAYqFyRbjYkm2PAKBabx6oYv/xDxNi8IBw==
+  version "7.0.6"
+  resolved "https://registry.yarnpkg.com/serialize-javascript/-/serialize-javascript-7.0.6.tgz#f2f20c8af0757e4d8fa329d0210636da0682ddef"
+  integrity sha512-ATTK5Q4gFVg0YDp1my2vqygyvhcklD/UV5GIlYHooGTn/NogJqIzpetkD6E5kmuVULqz/S9inUL25XcAgDRJQg==
 
 serve-handler@^6.1.7:
   version "6.1.7"