Added async restore in place

Author: adshin21

pkg/cluster/cluster.go

@@ -3,6 +3,7 @@ package cluster
 // Postgres CustomResourceDefinition object i.e. Spilo
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"fmt"
@@ -431,6 +432,30 @@ func (c *Cluster) Create() (err error) {
 		c.logger.Errorf("could not list resources: %v", err)
 	}
 
+
+	if err := c.updatePITRResources(); err != nil {
+		return fmt.Errorf("could not update pitr resources: %v", err)
+	}
+	return nil
+}
+
+// update the label to finished for PITR for the given config map
+func (c *Cluster) updatePITRResources() error {
+	cmName := fmt.Sprintf(PitrConfigMapNameTemplate, c.Name)
+	cmNamespace := c.Namespace
+	patchPayload := map[string]any{
+		"metadata": map[string]any{
+			"labels": map[string]string{
+				PitrStateLabelKey: PitrStateLabelValueFinished,
+			},
+		},
+	}
+
+	data, _ := json.Marshal(patchPayload)
+	if _, err := c.KubeClient.ConfigMaps(cmNamespace).Patch(context.TODO(), cmName, types.MergePatchType, data, metav1.PatchOptions{}, ""); err != nil {
+		c.logger.Errorf("restore-in-place: error updating config map label to final state: %v", err)
+		return err
+	}
 	return nil
 }
 
@@ -1200,6 +1225,15 @@ func syncResources(a, b *v1.ResourceRequirements) bool {
 	return false
 }
 
+const (
+	PitrStateLabelKey         = "postgres-operator.zalando.org/pitr-state"
+	PitrStateLabelValuePending    = "pending"
+	PitrStateLabelValueInProgress = "in-progress"
+	PitrStateLabelValueFinished = "finished"
+	PitrConfigMapNameTemplate = "pitr-state-%s"
+	PitrSpecDataKey           = "spec"
+)
+
 // Delete deletes the cluster and cleans up all objects associated with it (including statefulsets).
 // The deletion order here is somewhat significant, because Patroni, when running with the Kubernetes
 // DCS, reuses the master's endpoint to store the leader related metadata. If we remove the endpoint
@@ -1211,7 +1245,23 @@ func (c *Cluster) Delete() error {
 	defer c.mu.Unlock()
 	c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Delete", "Started deletion of cluster resources")
 
-	isRestoreInPlace := c.Annotations["postgres-operator.zalando.org/action"] == "restore-in-place"
+
+	cmName := fmt.Sprintf(PitrConfigMapNameTemplate, c.Name)
+
+	isRestoreInPlace := false
+	cm, err := c.KubeClient.ConfigMaps(c.Namespace).Get(context.TODO(), cmName, metav1.GetOptions{})
+	if err != nil {
+		c.logger.Debugf("restore-in-place: Error while fetching config map: %s before deletion", cmName)
+	}
+
+	if cm != nil {
+		if val, ok := cm.Labels[PitrStateLabelKey]; ok {
+			if val == PitrStateLabelValuePending {
+				isRestoreInPlace = true
+			}
+		}
+	}
+
 	c.logger.Debugf("restore-in-place: Deleting the cluster, verifying whether resotore-in-place is true or not: %+v\n", isRestoreInPlace)
 	if err := c.deleteStreams(); err != nil {
 		anyErrors = true

pkg/cluster/resources.go

@@ -10,6 +10,7 @@ import (
 	batchv1 "k8s.io/api/batch/v1"
 	v1 "k8s.io/api/core/v1"
 	policyv1 "k8s.io/api/policy/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 
@@ -301,6 +302,21 @@ func (c *Cluster) createService(role PostgresRole) (*v1.Service, error) {
 	c.setProcessName("creating %v service", role)
 
 	serviceSpec := c.generateService(role, &c.Spec)
+
+	// check if the service already exists in case of pitr
+	svc, err := c.KubeClient.Services(serviceSpec.Namespace).Get(context.TODO(), serviceSpec.Name, metav1.GetOptions{})
+
+	// service already exists
+	if err == nil {
+		c.Services[role] = svc
+		return svc, nil
+	}
+
+	if !errors.IsNotFound(err) {
+		return nil, err
+	}
+
+	// at last create the service
 	service, err := c.KubeClient.Services(serviceSpec.Namespace).Create(context.TODO(), serviceSpec, metav1.CreateOptions{})
 	if err != nil {
 		return nil, err

pkg/controller/postgresql.go

@@ -16,7 +16,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
 
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
@@ -27,6 +26,11 @@ import (
 	"github.com/zalando/postgres-operator/pkg/util/ringlog"
 )
 
+const (
+	restoreAnnotationKey      = "postgres-operator.zalando.org/action"
+	restoreAnnotationValue    = "restore-in-place"
+)
+
 func (c *Controller) clusterResync(stopCh <-chan struct{}, wg *sync.WaitGroup) {
 	defer wg.Done()
 	ticker := time.NewTicker(c.opConfig.ResyncPeriod)
@@ -37,6 +41,9 @@ func (c *Controller) clusterResync(stopCh <-chan struct{}, wg *sync.WaitGroup) {
 			if err := c.clusterListAndSync(); err != nil {
 				c.logger.Errorf("could not list clusters: %v", err)
 			}
+			if err := c.processPendingRestores(); err != nil {
+				c.logger.Errorf("could not process pending restores: %v", err)
+			}
 		case <-stopCh:
 			return
 		}
@@ -542,7 +549,7 @@ func (c *Controller) postgresqlUpdate(prev, cur interface{}) {
 	pgNew := c.postgresqlCheck(cur)
 	if pgOld != nil && pgNew != nil {
 
-		if pgNew.Annotations["postgres-operator.zalando.org/action"] == "restore-in-place" {
+		if pgNew.Annotations[restoreAnnotationKey] == restoreAnnotationValue {
 			c.logger.Debugf("restore-in-place: postgresqlUpdate called for cluster %q", pgNew.Name)
 			c.handlerRestoreInPlace(pgOld, pgNew)
 			return
@@ -604,117 +611,157 @@ func (c *Controller) validateRestoreInPlace(pgOld, pgNew *acidv1.Postgresql) err
 	return nil
 }
 
-// waitForOldResourcesTermination waits until the postgresql CR and its StatefulSet are terminated
-func (c *Controller) waitForOldResourcesTermination(pgOld *acidv1.Postgresql, statefulSetName string) error {
-	c.logger.Debugf("restore-in-place: Waiting for old CR %q and StatefulSet %q to be fully terminated", pgOld.Name, statefulSetName)
-
-	err := wait.PollUntilContextTimeout(context.TODO(), 2*time.Second, 5*time.Minute, true, func(ctx context.Context) (bool, error) {
-		// Check for CR
-		_, crErr := c.KubeClient.AcidV1ClientSet.AcidV1().Postgresqls(pgOld.Namespace).Get(ctx, pgOld.Name, metav1.GetOptions{})
-		crGone := errors.IsNotFound(crErr)
-		if crErr != nil && !crGone {
-			c.logger.Errorf("restore-in-place: Error while waiting for CR deletion: %v", crErr)
-			return false, crErr // A real error occurred
-		}
-
-		// Check for StatefulSet
-		_, stsErr := c.KubeClient.StatefulSets(pgOld.Namespace).Get(ctx, statefulSetName, metav1.GetOptions{})
-		stsGone := errors.IsNotFound(stsErr)
-		if stsErr != nil && !stsGone {
-			c.logger.Errorf("restore-in-place: Error while waiting for StatefulSet deletion: %v", stsErr)
-			return false, stsErr // A real error occurred
-		}
-
-		if crGone && stsGone {
-			c.logger.Debugf("restore-in-place: Both old CR and StatefulSet are fully terminated.")
-			return true, nil
-		}
-
-		if !crGone {
-			c.logger.Infof("restore-in-place: still waiting for postgresql CR %q to be deleted", pgOld.Name)
-		}
-		if !stsGone {
-			c.logger.Infof("restore-in-place: still waiting for StatefulSet %q to be deleted", statefulSetName)
-		}
-
-		return false, nil // Not done yet, continue polling.
-	})
-
-	if err != nil {
-		return fmt.Errorf("error while waiting for old resources to be deleted: %v", err)
-	}
 
-	c.logger.Debugf("restore-in-place: Finished waiting for old resource deletion.")
-	return nil
-}
 
-// handlerRestoreInPlace is to handle the resotre in place, it does few operatons
-// 1. Verifies the parameters required for restoring in place
-// 2. Removes the old CR if it exists, wait for it, if not present check the err that it is a k8sNotfound error and continue
-// 3. Wait for the successful removal of statefulsets, if not present check the err that it is a k8sNotfound error and continue
-// 4. Create a new CR with the latest details, while keeping few metadata about restore
+// handlerRestoreInPlace starts an asynchronous point-in-time-restore.
+// It creates a ConfigMap to store the state and then deletes the old Postgresql CR.
 func (c *Controller) handlerRestoreInPlace(pgOld, pgNew *acidv1.Postgresql) {
-	c.logger.Infof("restore-in-place: starting restore-in-place for cluster %q", pgNew.Name)
+	c.logger.Infof("restore-in-place: starting asynchronous restore-in-place for cluster %q", pgNew.Name)
 
 	if err := c.validateRestoreInPlace(pgOld, pgNew); err != nil {
 		c.logger.Errorf("restore-in-place: validation failed for cluster %q: %v", pgNew.Name, err)
 		return
 	}
 
+	// Prepare new spec for the restored cluster
+	c.logger.Debugf("restore-in-place: preparing new postgresql spec for cluster %q", pgNew.Name)
 	newPgSpec := pgNew.DeepCopy()
-	delete(newPgSpec.Annotations, "postgres-operator.zalando.org/action")
+	delete(newPgSpec.Annotations, restoreAnnotationKey)
 	newPgSpec.ResourceVersion = ""
 	newPgSpec.UID = ""
-	c.logger.Debugf("restore-in-place: newPgSpec after removing annotation: %+v", newPgSpec)
 
-	statefulSetName := pgOld.Name // Capture StatefulSet name, it's the same as the cluster name
+	specData, err := json.Marshal(newPgSpec)
+	if err != nil {
+		c.logger.Errorf("restore-in-place: could not marshal new postgresql spec for cluster %q: %v", newPgSpec.Name, err)
+		return
+	}
+
+	// Create or update ConfigMap to store restore state
+	cmName := fmt.Sprintf(cluster.PitrConfigMapNameTemplate, newPgSpec.Name)
+	c.logger.Debugf("restore-in-place: creating or updating state ConfigMap %q for cluster %q", cmName, newPgSpec.Name)
+	cm := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      cmName,
+			Namespace: newPgSpec.Namespace,
+			Labels: map[string]string{
+				cluster.PitrStateLabelKey: cluster.PitrStateLabelValuePending,
+			},
+		},
+		Data: map[string]string{
+			cluster.PitrSpecDataKey: string(specData),
+		},
+	}
+
+	// Check if ConfigMap already exists
+	_, err = c.KubeClient.ConfigMaps(cm.Namespace).Get(context.TODO(), cm.Name, metav1.GetOptions{})
+	if err != nil {
+		if errors.IsNotFound(err) {
+			_, err = c.KubeClient.ConfigMaps(cm.Namespace).Create(context.TODO(), cm, metav1.CreateOptions{})
+		}
+	} else {
+		// If for some reason CM exists, update it
+		_, err = c.KubeClient.ConfigMaps(cm.Namespace).Update(context.TODO(), cm, metav1.UpdateOptions{})
+	}
+
+	if err != nil {
+		c.logger.Errorf("restore-in-place: could not create or update state ConfigMap %q for cluster %q: %v", cmName, newPgSpec.Name, err)
+		return
+	}
+	c.logger.Infof("restore-in-place: state ConfigMap %q created for cluster %q", cmName, newPgSpec.Name)
 
-	// Initiate CR deletion first, as requested
-	c.logger.Debugf("restore-in-place: Attempting direct API deletion of postgresql CR %q", pgOld.Name)
-	err := c.KubeClient.AcidV1ClientSet.AcidV1().Postgresqls(pgOld.Namespace).Delete(context.TODO(), pgOld.Name, metav1.DeleteOptions{})
+	// Delete old postgresql CR to trigger cleanup and UID change
+	c.logger.Debugf("restore-in-place: attempting deletion of postgresql CR %q", pgOld.Name)
+	err = c.KubeClient.Postgresqls(pgOld.Namespace).Delete(context.TODO(), pgOld.Name, metav1.DeleteOptions{})
 	if err != nil && !errors.IsNotFound(err) {
-		c.logger.Errorf("restore-in-place: could not delete postgresql CR via API: %v", err)
-		return // Stop if there's a critical error deleting the CR
+		c.logger.Errorf("restore-in-place: could not delete postgresql CR %q: %v", pgOld.Name, err)
+		// Consider deleting the ConfigMap here to allow a retry
+		return
+	}
+	c.logger.Infof("restore-in-place: initiated deletion of postgresql CR %q", pgOld.Name)
+}
+
+// processPendingRestores handles the re-creation part of the asynchronous point-in-time-restore.
+// It is called periodically and checks for ConfigMaps that signal a pending or in-progress restore.
+func (c *Controller) processPendingRestores() error {
+	c.logger.Debug("restore-in-place: checking for pending restores")
+
+	namespace := c.opConfig.WatchedNamespace
+	if namespace == "" {
+		namespace = v1.NamespaceAll
 	}
-	c.logger.Debugf("restore-in-place: Direct API deletion of postgresql CR for %q initiated (or CR was already not found).", pgOld.Name)
 
-	// Then, initiate cluster sub-resource deletion if the cluster object is in memory
-	clusterName := util.NameFromMeta(pgOld.ObjectMeta)
-	c.clustersMu.RLock()
-	cl, clusterFound := c.clusters[clusterName]
-	c.clustersMu.RUnlock()
+	// Process "pending" restores: wait for deletion and move to "in-progress"
+	pendingOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", cluster.PitrStateLabelKey, cluster.PitrStateLabelValuePending)}
+	pendingCmList, err := c.KubeClient.ConfigMaps(namespace).List(context.TODO(), pendingOpts)
+	if err != nil {
+		return fmt.Errorf("restore-in-place: could not list pending restore ConfigMaps: %v", err)
+	}
+	if len(pendingCmList.Items) > 0 {
+		c.logger.Debugf("restore-in-place: found %d pending restore(s) to process", len(pendingCmList.Items))
+	}
 
-	if clusterFound {
-		c.logger.Debugf("restore-in-place: Cluster object found in memory. Calling cluster.Delete() for %q", clusterName)
-		if cl.Annotations == nil {
-			cl.Annotations = make(map[string]string)
+	for _, cm := range pendingCmList.Items {
+		c.logger.Debugf("restore-in-place: processing pending ConfigMap %q", cm.Name)
+		clusterName := strings.TrimPrefix(cm.Name, "pitr-state-")
+
+		_, err := c.KubeClient.Postgresqls(cm.Namespace).Get(context.TODO(), clusterName, metav1.GetOptions{})
+		if err == nil {
+			c.logger.Infof("restore-in-place: pending restore for cluster %q is waiting for old Postgresql CR to be deleted", clusterName)
+			continue
 		}
-		cl.Annotations["postgres-operator.zalando.org/action"] = "restore-in-place" // User requested to keep this
-		if err := cl.Delete(); err != nil {
-			// Log error but continue to ensure we wait for termination
-			c.logger.Errorf("restore-in-place: error during cluster.Delete() for %q: %v. Proceeding to wait for termination.", clusterName, err)
+		if !errors.IsNotFound(err) {
+			c.logger.Errorf("restore-in-place: could not check for existence of Postgresql CR %q: %v", clusterName, err)
+			continue
 		}
-		c.logger.Debugf("restore-in-place: cluster.Delete() returned for %q", clusterName)
-	} else {
-		c.logger.Warningf("restore-in-place: cluster %q not found in controller's map. Relying on CR deletion to trigger cleanup.", clusterName)
-	}
 
-	if err := c.waitForOldResourcesTermination(pgOld, statefulSetName); err != nil {
-		c.logger.Errorf("restore-in-place: %v", err)
-		return
+		c.logger.Infof("restore-in-place: old Postgresql CR %q is deleted, moving restore to 'in-progress'", clusterName)
+		cm.Labels[cluster.PitrStateLabelKey] = cluster.PitrStateLabelValueInProgress
+		if _, err := c.KubeClient.ConfigMaps(cm.Namespace).Update(context.TODO(), &cm, metav1.UpdateOptions{}); err != nil {
+			c.logger.Errorf("restore-in-place: could not update ConfigMap %q to 'in-progress': %v", cm.Name, err)
+		}
 	}
 
-	// Create a new CR with the latest details
-	c.logger.Debugf("restore-in-place: Creating new postgresql CR %q", newPgSpec.Name)
-	_, err = c.KubeClient.AcidV1ClientSet.AcidV1().Postgresqls(newPgSpec.Namespace).Create(context.TODO(), newPgSpec, metav1.CreateOptions{})
+	// Process "in-progress" restores: re-create the CR and clean up
+	inProgressOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", cluster.PitrStateLabelKey, cluster.PitrStateLabelValueInProgress)}
+	inProgressCmList, err := c.KubeClient.ConfigMaps(namespace).List(context.TODO(), inProgressOpts)
 	if err != nil {
-		c.logger.Errorf("restore-in-place: could not create postgresql CR for restore-in-place: %v", err)
-		// If the new CR cannot be created, the user needs to intervene.
-		return
+		return fmt.Errorf("restore-in-place: could not list in-progress restore ConfigMaps: %v", err)
+	}
+	if len(inProgressCmList.Items) > 0 {
+		c.logger.Debugf("restore-in-place: found %d in-progress restore(s) to process", len(inProgressCmList.Items))
+	}
+
+	for _, cm := range inProgressCmList.Items {
+		c.logger.Infof("restore-in-place: processing in-progress restore for ConfigMap %q", cm.Name)
+
+		c.logger.Debugf("restore-in-place: unmarshalling spec from ConfigMap %q", cm.Name)
+		var newPgSpec acidv1.Postgresql
+		if err := json.Unmarshal([]byte(cm.Data[cluster.PitrSpecDataKey]), &newPgSpec); err != nil {
+			c.logger.Errorf("restore-in-place: could not unmarshal postgresql spec from ConfigMap %q: %v", cm.Name, err)
+			continue
+		}
+
+		c.logger.Debugf("restore-in-place: creating new Postgresql CR %q from ConfigMap spec", newPgSpec.Name)
+		_, err := c.KubeClient.Postgresqls(newPgSpec.Namespace).Create(context.TODO(), &newPgSpec, metav1.CreateOptions{})
+		if err != nil {
+			if errors.IsAlreadyExists(err) {
+				c.logger.Infof("restore-in-place: Postgresql CR %q already exists, cleaning up restore ConfigMap", newPgSpec.Name)
+				// fallthrough to delete
+			} else {
+				c.logger.Errorf("restore-in-place: could not re-create Postgresql CR %q for restore: %v", newPgSpec.Name, err)
+				continue // Retry on next cycle
+			}
+		} else {
+			c.logger.Infof("restore-in-place: successfully re-created Postgresql CR %q to complete restore", newPgSpec.Name)
+		}
+
+		// c.logger.Debugf("restore-in-place: deleting successfully used restore ConfigMap %q", cm.Name)
+		// if err := c.KubeClient.ConfigMaps(cm.Namespace).Delete(context.TODO(), cm.Name, metav1.DeleteOptions{}); err != nil {
+		// 	c.logger.Errorf("restore-in-place: could not delete state ConfigMap %q: %v", cm.Name, err)
+		// }
 	}
-	c.logger.Debugf("restore-in-place: New postgresql CR %q created", newPgSpec.Name)
 
-	c.logger.Infof("restore-in-place: for cluster %q triggered successfully", pgNew.Name)
+	return nil
 }
 
 /*