feat: implement recovery window

Signed-off-by: Leonardo Cecchi <[email protected]>

Author: leonardoce

api/v1/objectstore_types.go

@@ -54,8 +54,21 @@ type ObjectStoreSpec struct {
 
 // ObjectStoreStatus defines the observed state of ObjectStore.
 type ObjectStoreStatus struct {
-	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
-	// Important: Run "make" to regenerate code after modifying this file
+	// ServerRecoveryWindow maps each server to its recovery window
+	ServerRecoveryWindow map[string]RecoveryWindow `json:"serverRecoveryWindow,omitempty"`
+}
+
+// RecoveryWindow represents the time span between the first
+// recoverability point and the last successful backup of a PostgreSQL
+// server, defining the period during which data can be restored.
+type RecoveryWindow struct {
+	// The first recoverability point in a PostgreSQL server refers to
+	// the earliest point in time to which the database can be
+	// restored.
+	FirstRecoverabilityPoint *metav1.Time `json:"firstRecoverabilityPoint,omitempty"`
+
+	// The last successful backup time
+	LastSuccessfulBackupTime *metav1.Time `json:"lastSuccussfulBackupTime,omitempty"`
 }
 
 // +kubebuilder:object:root=true

api/v1/zz_generated.deepcopy.go

@@ -520,6 +520,29 @@ spec:
             type: object
           status:
             description: ObjectStoreStatus defines the observed state of ObjectStore.
+            properties:
+              serverRecoveryWindow:
+                additionalProperties:
+                  description: |-
+                    RecoveryWindow represents the time span between the first
+                    recoverability point and the last successful backup of a PostgreSQL
+                    server, defining the period during which data can be restored.
+                  properties:
+                    firstRecoverabilityPoint:
+                      description: |-
+                        The first recoverability point in a PostgreSQL server refers to
+                        the earliest point in time to which the database can be
+                        restored.
+                      format: date-time
+                      type: string
+                    lastSuccussfulBackupTime:
+                      description: The last successful backup time
+                      format: date-time
+                      type: string
+                  type: object
+                description: ServerRecoveryWindow maps each server to its recovery
+                  window
+                type: object
             type: object
         required:
         - metadata

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

@@ -7,11 +7,14 @@ import (
 	"slices"
 	"time"
 
+	"github.com/cloudnative-pg/barman-cloud/pkg/catalog"
 	barmanCommand "github.com/cloudnative-pg/barman-cloud/pkg/command"
 	barmanCredentials "github.com/cloudnative-pg/barman-cloud/pkg/credentials"
 	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
@@ -75,8 +78,6 @@ func (c *RetentionPolicyRunnable) Start(ctx context.Context) error {
 // cycle enforces the retention policies. On success, it returns the amount
 // of time to wait to the next check.
 func (c *RetentionPolicyRunnable) cycle(ctx context.Context) (time.Duration, error) {
-	contextLogger := log.FromContext(ctx)
-
 	var cluster cnpgv1.Cluster
 	var barmanObjectStore barmancloudv1.ObjectStore
 
@@ -89,67 +90,118 @@ func (c *RetentionPolicyRunnable) cycle(ctx context.Context) (time.Duration, err
 		return 0, err
 	}
 
-	retentionPolicy := barmanObjectStore.Spec.RetentionPolicy
+	if err := c.applyRetentionPolicy(ctx, &cluster, &barmanObjectStore); err != nil {
+		return 0, err
+	}
+
 	nextCheckInterval := time.Second * time.Duration(
 		barmanObjectStore.Spec.InstanceSidecarConfiguration.RetentionPolicyIntervalSeconds)
+	return nextCheckInterval, nil
+}
 
+// applyRetentionPolicy applies the retention policy to the object
+// store and deletes the stale Kubernetes backup objects.
+func (c *RetentionPolicyRunnable) applyRetentionPolicy(
+	ctx context.Context,
+	cluster *cnpgv1.Cluster,
+	objectStore *barmancloudv1.ObjectStore,
+) error {
+	contextLogger := log.FromContext(ctx)
+
+	configuration := config.NewFromCluster(cluster)
+
+	retentionPolicy := objectStore.Spec.RetentionPolicy
 	if len(retentionPolicy) == 0 {
 		contextLogger.Info("Skipping retention policy enforcement, no retention policy specified")
-		return nextCheckInterval, nil
+		return nil
 	}
 	if cluster.Status.CurrentPrimary != c.PodName {
 		contextLogger.Info(
-			"Skipping retention policy enforcement, detected standby",
+			"Skipping retention policy enforcement, not the current primary",
 			"currentPrimary", cluster.Status.CurrentPrimary, "podName", c.PodName)
-		return nextCheckInterval, nil
+		return nil
 	}
 
-	contextLogger.Info("Enforcing retention policies",
+	contextLogger.Info("Applying backup retention policy",
 		"retentionPolicy", retentionPolicy)
 
 	osEnvironment := os.Environ()
-	caBundleEnvironment := common.GetRestoreCABundleEnv(&barmanObjectStore.Spec.Configuration)
+	caBundleEnvironment := common.GetRestoreCABundleEnv(&objectStore.Spec.Configuration)
 	env, err := barmanCredentials.EnvSetBackupCloudCredentials(
 		ctx,
 		c.Client,
-		barmanObjectStore.Namespace,
-		&barmanObjectStore.Spec.Configuration,
+		objectStore.Namespace,
+		&objectStore.Spec.Configuration,
 		common.MergeEnv(osEnvironment, caBundleEnvironment))
 	if err != nil {
 		contextLogger.Error(err, "while setting backup cloud credentials")
-		return 0, err
+		return err
 	}
 
 	if err := barmanCommand.DeleteBackupsByPolicy(
 		ctx,
-		&barmanObjectStore.Spec.Configuration,
+		&objectStore.Spec.Configuration,
 		configuration.ServerName,
 		env,
 		retentionPolicy,
 	); err != nil {
 		contextLogger.Error(err, "while enforcing retention policies")
-		c.Recorder.Event(&cluster, "Warning", "RetentionPolicyFailed", "Retention policy failed")
-		return 0, err
+		c.Recorder.Event(cluster, "Warning", "RetentionPolicyFailed", "Retention policy failed")
+		return err
 	}
 
 	backupList, err := barmanCommand.GetBackupList(
 		ctx,
-		&barmanObjectStore.Spec.Configuration,
+		&objectStore.Spec.Configuration,
 		configuration.ServerName,
 		env,
 	)
 	if err != nil {
 		contextLogger.Error(err, "while reading the backup list")
-		return 0, err
+		return err
 	}
 
-	if err := deleteBackupsNotInCatalog(ctx, c.Client, &cluster, backupList.GetBackupIDs()); err != nil {
+	if err := deleteBackupsNotInCatalog(ctx, c.Client, cluster, backupList.GetBackupIDs()); err != nil {
 		contextLogger.Error(err, "while deleting Backups not present in the catalog")
-		return 0, err
+		return err
 	}
 
-	// TODO(leonardoce): update first point of recoverability
-	return nextCheckInterval, nil
+	return c.updateRecoveryWindow(ctx, backupList, objectStore, configuration.ServerName)
+}
+
+// updateRecoveryWindow updates the recovery window inside the object
+// store status subresource
+func (c *RetentionPolicyRunnable) updateRecoveryWindow(
+	ctx context.Context,
+	backupList *catalog.Catalog,
+	objectStore *barmancloudv1.ObjectStore,
+	serverName string,
+) error {
+	// Set the recovery window inside the barman object store object
+	convertTime := func(t *time.Time) *metav1.Time {
+		if t == nil {
+			return nil
+		}
+
+		return ptr.To(metav1.NewTime(*t))
+	}
+
+	firstRecoverabilityPoint := backupList.GetFirstRecoverabilityPoint()
+	lastSuccessfulBackupTime := backupList.GetLastSuccessfulBackupTime()
+	recoveryWindow := barmancloudv1.RecoveryWindow{
+		FirstRecoverabilityPoint: convertTime(firstRecoverabilityPoint),
+		LastSuccessfulBackupTime: convertTime(lastSuccessfulBackupTime),
+	}
+
+	if objectStore.Status.ServerRecoveryWindow == nil {
+		objectStore.Status.ServerRecoveryWindow = make(map[string]barmancloudv1.RecoveryWindow)
+	}
+	objectStore.Status.ServerRecoveryWindow[serverName] = recoveryWindow
+	if err := c.Client.Status().Update(ctx, objectStore); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // deleteBackupsNotInCatalog deletes all Backup objects pointing to the given cluster that are not

internal/cnpgi/instance/retention.go

@@ -309,7 +309,7 @@ func reconcilePodSpec(
 		},
 		{
 			Name:  "CUSTOM_CNPG_GROUP",
-			Value: "postgresql.cnpg.io",
+			Value: "postgresql.k8s.enterprisedb.io",
 		},
 		{
 			Name:  "CUSTOM_CNPG_VERSION",

internal/cnpgi/operator/lifecycle.go

@@ -35,35 +35,49 @@ func BuildRole(
 		}
 	}
 
-	role.Rules = append(role.Rules, rbacv1.PolicyRule{
-		APIGroups: []string{
-			"barmancloud.cnpg.io",
-		},
-		Verbs: []string{
-			"get",
-			"watch",
-			"list",
-		},
-		Resources: []string{
-			"objectstores",
-		},
-		ResourceNames: barmanObjectsSet.ToSortedList(),
-	})
-
-	role.Rules = append(role.Rules, rbacv1.PolicyRule{
-		APIGroups: []string{
-			"",
+	role.Rules = append(
+		role.Rules,
+		rbacv1.PolicyRule{
+			APIGroups: []string{
+				"barmancloud.cnpg.io",
+			},
+			Verbs: []string{
+				"get",
+				"watch",
+				"list",
+			},
+			Resources: []string{
+				"objectstores",
+			},
+			ResourceNames: barmanObjectsSet.ToSortedList(),
 		},
-		Resources: []string{
-			"secrets",
+		rbacv1.PolicyRule{
+			APIGroups: []string{
+				"barmancloud.cnpg.io",
+			},
+			Verbs: []string{
+				"update",
+			},
+			Resources: []string{
+				"objectstores/status",
+			},
+			ResourceNames: barmanObjectsSet.ToSortedList(),
 		},
-		Verbs: []string{
-			"get",
-			"watch",
-			"list",
+		rbacv1.PolicyRule{
+			APIGroups: []string{
+				"",
+			},
+			Resources: []string{
+				"secrets",
+			},
+			Verbs: []string{
+				"get",
+				"watch",
+				"list",
+			},
+			ResourceNames: secretsSet.ToSortedList(),
 		},
-		ResourceNames: secretsSet.ToSortedList(),
-	})
+	)
 
 	return role
 }