feat: serve restore hooks from the instance sidecar

internal/cnpgi/common/common.go

@@ -52,6 +52,11 @@ const (
 	// BarmanEndpointCACertificateFileName is the name of the file in which the barman endpoint
 	// CA certificate is stored.
 	BarmanEndpointCACertificateFileName = "barman-ca.crt"
+
+	// PgWalVolumePgWalPath is the path of the pg_wal directory inside the WAL volume,
+	// used when a separate WAL storage is configured. During a restore the pg_wal
+	// directory is moved here and symlinked back into PGDATA.
+	PgWalVolumePgWalPath = "/var/lib/postgresql/wal/pg_wal"
 )
 
 // GetRestoreCABundleEnv gets the enveronment variables to be used when custom

internal/cnpgi/instance/identity.go

@@ -70,6 +70,13 @@ func (i IdentityImplementation) GetPluginCapabilities(
 					},
 				},
 			},
+			{
+				Type: &identity.PluginCapability_Service_{
+					Service: &identity.PluginCapability_Service{
+						Type: identity.PluginCapability_Service_TYPE_RESTORE_JOB,
+					},
+				},
+			},
 		},
 	}, nil
 }

internal/cnpgi/instance/identity_test.go

@@ -0,0 +1,52 @@
+/*
+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 instance
+
+import (
+	"github.com/cloudnative-pg/cnpg-i/pkg/identity"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("IdentityImplementation", func() {
+	Describe("GetPluginCapabilities", func() {
+		It("declares the WAL, backup, metrics and restore-job services", func(ctx SpecContext) {
+			impl := IdentityImplementation{}
+			response, err := impl.GetPluginCapabilities(ctx, &identity.GetPluginCapabilitiesRequest{})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(response).NotTo(BeNil())
+
+			var serviceTypes []identity.PluginCapability_Service_Type
+			for _, capability := range response.GetCapabilities() {
+				serviceTypes = append(serviceTypes, capability.GetService().GetType())
+			}
+
+			// The instance sidecar now runs the phase-0 restore in-process, so it must
+			// advertise TYPE_RESTORE_JOB alongside the services it already served.
+			Expect(serviceTypes).To(ConsistOf(
+				identity.PluginCapability_Service_TYPE_WAL_SERVICE,
+				identity.PluginCapability_Service_TYPE_BACKUP_SERVICE,
+				identity.PluginCapability_Service_TYPE_METRICS,
+				identity.PluginCapability_Service_TYPE_RESTORE_JOB,
+			))
+		})
+	})
+})

internal/cnpgi/instance/start.go

@@ -25,11 +25,13 @@ import (
 	"github.com/cloudnative-pg/cnpg-i-machinery/pkg/pluginhelper/http"
 	"github.com/cloudnative-pg/cnpg-i/pkg/backup"
 	"github.com/cloudnative-pg/cnpg-i/pkg/metrics"
+	restore "github.com/cloudnative-pg/cnpg-i/pkg/restore/job"
 	"github.com/cloudnative-pg/cnpg-i/pkg/wal"
 	"google.golang.org/grpc"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/common"
+	barmanrestore "github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/restore"
 )
 
 // CNPGI is the implementation of the PostgreSQL sidecar
@@ -60,6 +62,15 @@ func (c *CNPGI) Start(ctx context.Context) error {
 		metrics.RegisterMetricsServer(server, &metricsImpl{
 			Client: c.Client,
 		})
+		// The instance pod runs the phase-0 bootstrap in-process (no separate
+		// recovery Job), so the same sidecar must answer the Restore RPC that
+		// initializes PGDATA from the object store before PostgreSQL starts.
+		restore.RegisterRestoreJobHooksServer(server, &barmanrestore.JobHookImpl{
+			Client:               c.Client,
+			SpoolDirectory:       c.SpoolDirectory,
+			PgDataPath:           c.PGDataPath,
+			PgWalFolderToSymlink: common.PgWalVolumePgWalPath,
+		})
 		common.AddHealthCheck(server)
 		return nil
 	}

internal/cnpgi/operator/config/config.go

@@ -102,6 +102,15 @@ func (config *PluginConfiguration) GetReplicaSourceBarmanObjectKey() types.Names
 	}
 }
 
+// HasAnyBarmanObjectStore returns true if the configuration references at least
+// one barman object store, be it for backup/archiving, recovery, or as a
+// replica source.
+func (config *PluginConfiguration) HasAnyBarmanObjectStore() bool {
+	return len(config.BarmanObjectName) > 0 ||
+		len(config.RecoveryBarmanObjectName) > 0 ||
+		len(config.ReplicaSourceBarmanObjectName) > 0
+}
+
 // GetReferredBarmanObjectsKey gets the list of barman objects referred by this
 // plugin configuration
 func (config *PluginConfiguration) GetReferredBarmanObjectsKey() []types.NamespacedName {
@@ -263,9 +272,7 @@ func getReplicaSourcePlugin(cluster *cnpgv1.Cluster) *cnpgv1.PluginConfiguration
 func (config *PluginConfiguration) Validate() error {
 	err := NewConfigurationError()
 
-	if len(config.BarmanObjectName) == 0 &&
-		len(config.RecoveryBarmanObjectName) == 0 &&
-		len(config.ReplicaSourceBarmanObjectName) == 0 {
+	if !config.HasAnyBarmanObjectStore() {
 		return err.WithMessage("no reference to barmanObjectName have been included")
 	}
 

internal/cnpgi/operator/lifecycle.go

@@ -322,6 +322,38 @@ func (impl LifecycleImplementation) collectAdditionalInstanceArgs(
 	return nil, nil
 }
 
+// shouldInjectBarmanSidecar decides whether an instance pod needs the
+// plugin-barman-cloud sidecar.
+//
+// A cluster doing backup/archiving or serving as a replica source needs the
+// sidecar in every instance pod for as long as the cluster exists, so those
+// two cases always inject it. A recovery-only cluster (only
+// RecoveryBarmanObjectName set, mirroring what pluginConfiguration.Validate()
+// accepts) only ever needs the sidecar for its one-time bootstrap restore, so
+// it's gated on cluster.IsInitialized() instead.
+//
+// Once the cluster finishes initializing, this makes the operator's own
+// drift-check (checkPodSpecIsOutdated) see the running pod's spec as outdated
+// and roll it out to drop the sidecar. That's deliberately accepted rather
+// than engineered around: it's one deterministic rollout using the same
+// machinery the operator already uses for every other pod-spec change (a
+// switchover if a replica is available, an in-place restart otherwise), not a
+// new or fragile risk.
+func shouldInjectBarmanSidecar(
+	cluster *cnpgv1.Cluster,
+	pluginConfiguration *config.PluginConfiguration,
+) bool {
+	if len(pluginConfiguration.BarmanObjectName) != 0 || len(pluginConfiguration.ReplicaSourceBarmanObjectName) != 0 {
+		return true
+	}
+
+	if len(pluginConfiguration.RecoveryBarmanObjectName) == 0 {
+		return false
+	}
+
+	return !cluster.IsInitialized()
+}
+
 func reconcileInstancePod(
 	ctx context.Context,
 	cluster *cnpgv1.Cluster,
@@ -339,8 +371,7 @@ func reconcileInstancePod(
 
 	mutatedPod := pod.DeepCopy()
 
-	if len(pluginConfiguration.BarmanObjectName) != 0 ||
-		len(pluginConfiguration.ReplicaSourceBarmanObjectName) != 0 {
+	if shouldInjectBarmanSidecar(cluster, pluginConfiguration) {
 		if err := reconcilePodSpec(
 			cluster,
 			&mutatedPod.Spec,
@@ -353,7 +384,7 @@ func reconcileInstancePod(
 			return nil, fmt.Errorf("while reconciling pod spec for pod: %w", err)
 		}
 	} else {
-		contextLogger.Debug("No need to mutate instance with no backup & archiving configuration")
+		contextLogger.Debug("No need to mutate instance, sidecar not required for this configuration and pod")
 	}
 
 	patch, err := object.CreatePatch(mutatedPod, pod)

internal/cnpgi/operator/lifecycle_test.go

@@ -242,6 +242,70 @@ var _ = Describe("LifecycleImplementation", func() {
 				HaveKey("value")))
 		})
 
+		It("injects the sidecar for a recovery-only cluster", func(ctx SpecContext) {
+			recoveryOnlyConfig := &config.PluginConfiguration{
+				RecoveryBarmanObjectName: "minio-store-recovery",
+			}
+			pod := &corev1.Pod{
+				TypeMeta:   podTypeMeta,
+				ObjectMeta: metav1.ObjectMeta{Name: "test-pod"},
+				Spec:       corev1.PodSpec{Containers: []corev1.Container{{Name: "postgres"}}},
+			}
+			podJSON, _ := json.Marshal(pod)
+			request := &lifecycle.OperatorLifecycleRequest{
+				ObjectDefinition: podJSON,
+			}
+
+			response, err := reconcileInstancePod(ctx, cluster, request, recoveryOnlyConfig, sidecarConfiguration{})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(response).NotTo(BeNil())
+			Expect(response.JsonPatch).NotTo(BeEmpty())
+			var patch []map[string]interface{}
+			Expect(json.Unmarshal(response.JsonPatch, &patch)).To(Succeed())
+			Expect(patch).To(ContainElement(HaveKeyWithValue("path", "/spec/initContainers")))
+		})
+
+		It("does not inject the sidecar for a recovery-only cluster that has "+
+			"already completed its initial bootstrap", func(ctx SpecContext) {
+			recoveryOnlyConfig := &config.PluginConfiguration{
+				RecoveryBarmanObjectName: "minio-store-recovery",
+			}
+			cluster.Status.LatestGeneratedNode = 1
+			pod := &corev1.Pod{
+				TypeMeta:   podTypeMeta,
+				ObjectMeta: metav1.ObjectMeta{Name: "test-pod"},
+				Spec:       corev1.PodSpec{Containers: []corev1.Container{{Name: "postgres"}}},
+			}
+			podJSON, _ := json.Marshal(pod)
+			request := &lifecycle.OperatorLifecycleRequest{
+				ObjectDefinition: podJSON,
+			}
+
+			response, err := reconcileInstancePod(ctx, cluster, request, recoveryOnlyConfig, sidecarConfiguration{})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(response).NotTo(BeNil())
+			Expect(response.JsonPatch).To(BeEmpty())
+		})
+
+		It("does not mutate the pod when no object store is configured", func(ctx SpecContext) {
+			emptyConfig := &config.PluginConfiguration{}
+			pod := &corev1.Pod{
+				TypeMeta:   podTypeMeta,
+				ObjectMeta: metav1.ObjectMeta{Name: "test-pod"},
+				Spec:       corev1.PodSpec{Containers: []corev1.Container{{Name: "postgres"}}},
+			}
+			podJSON, _ := json.Marshal(pod)
+			request := &lifecycle.OperatorLifecycleRequest{
+				ObjectDefinition: podJSON,
+			}
+
+			response, err := reconcileInstancePod(ctx, cluster, request, emptyConfig, sidecarConfiguration{})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(response).NotTo(BeNil())
+			// An empty patch means the pod was left untouched: no sidecar injected.
+			Expect(response.JsonPatch).To(BeEmpty())
+		})
+
 		It("returns an error for invalid pod definition", func(ctx SpecContext) {
 			request := &lifecycle.OperatorLifecycleRequest{
 				ObjectDefinition: []byte("invalid-json"),

internal/cnpgi/restore/start.go

@@ -44,9 +44,6 @@ type CNPGI struct {
 
 // Start starts the GRPC service
 func (c *CNPGI) Start(ctx context.Context) error {
-	// PgWalVolumePgWalPath is the path of pg_wal directory inside the WAL volume when present
-	const PgWalVolumePgWalPath = "/var/lib/postgresql/wal/pg_wal"
-
 	enrich := func(server *grpc.Server) error {
 		wal.RegisterWALServer(server, common.WALServiceImplementation{
 			InstanceName:   c.InstanceName,
@@ -60,7 +57,7 @@ func (c *CNPGI) Start(ctx context.Context) error {
 			Client:               c.Client,
 			SpoolDirectory:       c.SpoolDirectory,
 			PgDataPath:           c.PGDataPath,
-			PgWalFolderToSymlink: PgWalVolumePgWalPath,
+			PgWalFolderToSymlink: common.PgWalVolumePgWalPath,
 		})
 
 		common.AddHealthCheck(server)