feat: serve restore hooks from the instance sidecar

The bootstrap that used to run in a dedicated recovery Job now happens
in-process inside the instance pod before PostgreSQL starts, so the sidecar
shipped in that pod must answer the same Restore RPC the operator sends over
the plugin sockets. The instance mode now registers the restore job hooks and
advertises the restore-job service capability so the operator can reach them.
A cluster that only bootstraps from an object store, without continued
archiving, previously received no sidecar in its instance pods; the injection
condition is widened to match what the plugin configuration already considers
valid, so those clusters get the sidecar too.

Signed-off-by: Armando Ruocco <armando.ruocco@enterprisedb.com>

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/lifecycle.go

@@ -339,7 +339,12 @@ func reconcileInstancePod(
 
 	mutatedPod := pod.DeepCopy()
 
+	// A recovery-only cluster (only RecoveryBarmanObjectName set) still needs the
+	// sidecar in its instance pods: the phase-0 bootstrap restore and the WAL
+	// replay that follows both run inside the instance and rely on it. This
+	// condition therefore mirrors what pluginConfiguration.Validate() accepts.
 	if len(pluginConfiguration.BarmanObjectName) != 0 ||
+		len(pluginConfiguration.RecoveryBarmanObjectName) != 0 ||
 		len(pluginConfiguration.ReplicaSourceBarmanObjectName) != 0 {
 		if err := reconcilePodSpec(
 			cluster,
@@ -353,7 +358,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 with no barman object store configuration")
 	}
 
 	patch, err := object.CreatePatch(mutatedPod, pod)

internal/cnpgi/operator/lifecycle_test.go

@@ -242,6 +242,48 @@ 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 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)