feat: enrich the Backup status with volume snapshots (cloudnative-pg#2838)

This patch enhances the status section of a volume snapshot Backup
with Postgres metadata to match the status of an object store backup.

Closes cloudnative-pg#2813

Signed-off-by: Jaime Silvela <jaime.silvela@enterprisedb.com>
Signed-off-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Co-authored-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Co-authored-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>

Author: 3 people

controllers/backup_controller.go

@@ -25,10 +25,12 @@ import (
 	storagesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/retry"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -322,6 +324,8 @@ func (r *BackupReconciler) startSnapshotBackup(
 		backup.Status.SetAsStarted(targetPod, apiv1.BackupMethodVolumeSnapshot)
 		// given that we use only kubernetes resources we can use the backup name as ID
 		backup.Status.BackupID = backup.Name
+		backup.Status.BackupName = backup.Name
+		backup.Status.StartedAt = ptr.To(metav1.Now())
 		if err := postgres.PatchBackupStatusAndRetry(ctx, r.Client, backup); err != nil {
 			return nil, err
 		}
@@ -375,10 +379,35 @@ func (r *BackupReconciler) startSnapshotBackup(
 	}
 
 	backup.Status.BackupSnapshotStatus.SetSnapshotList(snapshots)
+	if err := backupStatusFromSnapshots(snapshots, &backup.Status); err != nil {
+		contextLogger.Error(err, "while enriching the backup status")
+	}
 
 	return nil, postgres.PatchBackupStatusAndRetry(ctx, r.Client, backup)
 }
 
+// backupStatusFromSnapshots adds fields to the backup status based on the snapshots
+func backupStatusFromSnapshots(
+	snapshots volumesnapshot.Slice,
+	backupStatus *apiv1.BackupStatus,
+) error {
+	_, lastCreation := snapshots.GetSnapshotsInterval()
+	backupStatus.StoppedAt = ptr.To(lastCreation)
+	controldata, err := snapshots.GetControldata()
+	if err != nil {
+		return err
+	}
+	pairs := utils.ParsePgControldataOutput(controldata)
+
+	// the begin/end WAL and LSN are the same, since the instance was fenced
+	// for the snapshot
+	backupStatus.BeginWal = pairs["Latest checkpoint's REDO WAL file"]
+	backupStatus.EndWal = pairs["Latest checkpoint's REDO WAL file"]
+	backupStatus.BeginLSN = pairs["Latest checkpoint's REDO location"]
+	backupStatus.EndLSN = pairs["Latest checkpoint's REDO location"]
+	return nil
+}
+
 // isErrorRetryable detects is an error is retryable or not
 func isErrorRetryable(err error) bool {
 	return apierrs.IsServerTimeout(err) || apierrs.IsConflict(err) || apierrs.IsInternalError(err)

pkg/management/postgres/restore.go

@@ -25,7 +25,6 @@ import (
 	"os"
 	"os/exec"
 	"path"
-	"regexp"
 	"strings"
 	"time"
 
@@ -68,13 +67,12 @@ var (
 		Steps: math.MaxInt32,
 	}
 
-	enforcedParametersRegex          = regexp.MustCompile(`(?P<PARAM>[a-z_]+) setting:\s+(?P<VALUE>[a-z0-9]+)`)
 	pgControldataSettingsToParamsMap = map[string]string{
-		"max_connections":      "max_connections",
-		"max_wal_senders":      "max_wal_senders",
-		"max_worker_processes": "max_worker_processes",
-		"max_prepared_xacts":   "max_prepared_transactions",
-		"max_locks_per_xact":   "max_locks_per_transaction",
+		"max_connections setting":      "max_connections",
+		"max_wal_senders setting":      "max_wal_senders",
+		"max_worker_processes setting": "max_worker_processes",
+		"max_prepared_xacts setting":   "max_prepared_transactions",
+		"max_locks_per_xact setting":   "max_locks_per_transaction",
 	}
 )
 
@@ -641,13 +639,9 @@ func GetEnforcedParametersThroughPgControldata(pgData string) (map[string]string
 	log.Debug("pg_controldata stdout", "stdout", stdoutBuffer.String())
 
 	enforcedParams := map[string]string{}
-	for _, line := range strings.Split(stdoutBuffer.String(), "\n") {
-		matches := enforcedParametersRegex.FindStringSubmatch(line)
-		if len(matches) < 3 {
-			continue
-		}
-		if param, ok := pgControldataSettingsToParamsMap[matches[1]]; ok {
-			enforcedParams[param] = matches[2]
+	for key, value := range utils.ParsePgControldataOutput(stdoutBuffer.String()) {
+		if param, ok := pgControldataSettingsToParamsMap[key]; ok {
+			enforcedParams[param] = value
 		}
 	}
 	return enforcedParams, nil

pkg/reconciler/backup/volumesnapshot/reconciler.go

@@ -174,6 +174,8 @@ func (se *Reconciler) ensurePodIsFenced(
 	backup *apiv1.Backup,
 	targetPodName string,
 ) error {
+	contextLogger := log.FromContext(ctx)
+
 	fencedInstances, err := utils.GetFencedInstances(cluster.Annotations)
 	if err != nil {
 		return fmt.Errorf("could not check if cluster is fenced: %v", err)
@@ -188,21 +190,27 @@ func (se *Reconciler) ensurePodIsFenced(
 		return errors.New("cannot execute volume snapshot on a cluster that has fenced instances")
 	}
 
-	// The list of fenced instances is empty, so we need to request
-	// fencing for the target pod
-	se.recorder.Eventf(backup, "Normal", "FencePod",
-		"Requesting fencing for Pod %v", targetPodName)
-
-	if err := resources.ApplyFenceFunc(
+	err = resources.ApplyFenceFunc(
 		ctx,
 		se.cli,
 		cluster.Name,
 		cluster.Namespace,
 		targetPodName,
 		utils.AddFencedInstance,
-	); !errors.Is(err, utils.ErrorServerAlreadyFenced) {
+	)
+	if errors.Is(err, utils.ErrorServerAlreadyFenced) {
+		return nil
+	}
+	if err != nil {
 		return err
 	}
+
+	// The list of fenced instances is empty, so we need to request
+	// fencing for the target pod
+	contextLogger.Info("Fencing Pod", "podName", targetPodName)
+	se.recorder.Eventf(backup, "Normal", "FencePod",
+		"Fencing Pod %v", targetPodName)
+
 	return nil
 }
 
@@ -214,23 +222,28 @@ func (se *Reconciler) EnsurePodIsUnfenced(
 	targetPod *corev1.Pod,
 ) error {
 	contextLogger := log.FromContext(ctx)
-	contextLogger.Info("Unfencing Pod")
 
-	if err := resources.ApplyFenceFunc(
+	err := resources.ApplyFenceFunc(
 		ctx,
 		se.cli,
 		cluster.Name,
 		cluster.Namespace,
 		targetPod.Name,
 		utils.RemoveFencedInstance,
-	); err != nil {
+	)
+	if errors.Is(err, utils.ErrorServerAlreadyUnfenced) {
+		return nil
+	}
+	if err != nil {
 		return err
 	}
 
 	// The list of fenced instances is empty, so we need to request
 	// fencing for the target pod
+	contextLogger.Info("Unfencing Pod", "podName", targetPod.Name)
 	se.recorder.Eventf(backup, "Normal", "UnfencePod",
-		"Un-fencing Pod %v", targetPod.Name)
+		"Unfencing Pod %v", targetPod.Name)
+
 	return nil
 }
 

pkg/reconciler/backup/volumesnapshot/resources.go

@@ -19,8 +19,10 @@ package volumesnapshot
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	storagesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
@@ -57,20 +59,55 @@ type volumeSnapshotError struct {
 
 // Error implements the error interface
 func (err volumeSnapshotError) Error() string {
-	if err.InternalError.Message != nil {
+	if err.InternalError.Message == nil {
 		return "non specified volume snapshot error"
 	}
 	return *err.InternalError.Message
 }
 
+// Slice represents a slice of []storagesnapshotv1.VolumeSnapshot
+type Slice []storagesnapshotv1.VolumeSnapshot
+
+// GetSnapshotsInterval gets the earliest and latest creation times from a list of VolumeSnapshots
+func (s Slice) GetSnapshotsInterval() (metav1.Time, metav1.Time) {
+	var firstCreation, lastCreation metav1.Time
+	for idx := range s {
+		volumeSnapshot := &s[idx]
+		if firstCreation.IsZero() || lastCreation.IsZero() {
+			firstCreation = volumeSnapshot.CreationTimestamp
+			lastCreation = volumeSnapshot.CreationTimestamp
+			continue
+		}
+		if volumeSnapshot.CreationTimestamp.Before(&firstCreation) {
+			firstCreation = volumeSnapshot.CreationTimestamp
+		}
+		if lastCreation.Before(&volumeSnapshot.CreationTimestamp) {
+			lastCreation = volumeSnapshot.CreationTimestamp
+		}
+	}
+	return firstCreation, lastCreation
+}
+
+// GetControldata retrieves the pg_controldata stored as an annotation in VolumeSnapshots
+func (s Slice) GetControldata() (string, error) {
+	for _, volumeSnapshot := range s {
+		pgControlData, ok := volumeSnapshot.Annotations[utils.PgControldataAnnotationName]
+		if !ok {
+			continue
+		}
+		return pgControlData, nil
+	}
+	return "", fmt.Errorf("could not retrieve pg_controldata from any snapshot")
+}
+
 // GetBackupVolumeSnapshots extracts the list of volume snapshots related
 // to a backup name
 func GetBackupVolumeSnapshots(
 	ctx context.Context,
 	cli client.Client,
 	namespace string,
 	backupLabelName string,
-) ([]storagesnapshotv1.VolumeSnapshot, error) {
+) (Slice, error) {
 	var list storagesnapshotv1.VolumeSnapshotList
 
 	if err := cli.List(

pkg/utils/parser.go

@@ -0,0 +1,33 @@
+/*
+Copyright The CloudNativePG Contributors
+
+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.
+*/
+
+package utils
+
+import "strings"
+
+// ParsePgControldataOutput parses a pg_controldata output into a map of key-value pairs
+func ParsePgControldataOutput(data string) map[string]string {
+	pairs := make(map[string]string)
+	lines := strings.Split(data, "\n")
+	for _, line := range lines {
+		frags := strings.Split(line, ":")
+		if len(frags) != 2 {
+			continue
+		}
+		pairs[strings.TrimSpace(frags[0])] = strings.TrimSpace(frags[1])
+	}
+	return pairs
+}

tests/e2e/volume_snapshot_test.go

@@ -380,6 +380,15 @@ var _ = Describe("Verify Volume Snapshot",
 					AssertBackupConditionInClusterStatus(namespace, clusterToBackupName)
 				})
 
+				By("checking that the backup status is correctly populated", func() {
+					Expect(backup.Status.BeginWal).ToNot(BeEmpty())
+					Expect(backup.Status.EndWal).ToNot(BeEmpty())
+					Expect(backup.Status.BeginLSN).ToNot(BeEmpty())
+					Expect(backup.Status.EndLSN).ToNot(BeEmpty())
+					Expect(backup.Status.StoppedAt).ToNot(BeNil())
+					Expect(backup.Status.StartedAt).ToNot(BeNil())
+				})
+
 				var clusterToBackup *apiv1.Cluster
 
 				By("fetching the created cluster", func() {