// SPDX-License-Identifier: Apache-2.0 package backupcontroller import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "slices" "sort" "strconv" "strings" "time" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" apimeta "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" strategyv1alpha1 "github.com/cozystack/cozystack/api/backups/strategy/v1alpha1" backupsv1alpha1 "github.com/cozystack/cozystack/api/backups/v1alpha1" "github.com/cozystack/cozystack/internal/backupcontroller/cnpgtypes" "github.com/cozystack/cozystack/internal/backupcontroller/postgresapp" "github.com/cozystack/cozystack/internal/template" ) // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- const ( cnpgFieldManager = "cozystack-cnpg-backup-driver" cnpgClusterLabel = "cnpg.io/cluster" cnpgBackupPhaseComplete = "completed" cnpgBackupPhaseFailed = "failed" // barmanObjectNameParam / barmanServerNameParam are the barman-cloud plugin // parameter keys on a Cluster's spec.plugins entry. barmanObjectName points // at the ObjectStore CR; serverName is the per-server folder under // destinationPath (the plugin forbids serverName inside the ObjectStore). barmanObjectNameParam = "barmanObjectName" barmanServerNameParam = "serverName" postgresAppKind = "Postgres" postgresAppPrefix = "postgres-" defaultS3AccessKeyIDKey = "AWS_ACCESS_KEY_ID" defaultS3SecretAccessKeyKey = "AWS_SECRET_ACCESS_KEY" // defaultEndpointCAKey mirrors the convention every Kubernetes-issued // CA Secret follows (cert-manager, kube-system, COSI providers, etc.). defaultEndpointCAKey = "ca.crt" // Driver metadata keys persisted on Cozystack Backup artifacts. cnpgBackupNameKey = "cnpg.io/backup-name" cnpgServerNameKey = "cnpg.io/server-name" cnpgDestinationPathKey = "cnpg.io/destination-path" cnpgEndpointURLKey = "cnpg.io/endpoint-url" cnpgClusterNameKey = "cnpg.io/cluster-name" cnpgS3SecretRefKey = "cnpg.io/s3-secret-ref" cnpgMajorVersionKey = "cnpg.io/major-version" // Annotation the postgres chart renders on the Cluster from // bootstrap.newServerName. cnpgRestoredServerNameAnnotation = "postgres.cozystack.io/restored-server-name" // Polling cadence for the CNPG backup/restore lifecycle. Mirrors the // Velero strategy's defaults so behaviour is uniform across drivers. cnpgPollInterval = 5 * time.Second // Condition Type recorded on a RestoreJob once we've purged the target // Cluster + PVCs for it. Subsequent reconciles skip the purge step. restoreCondTargetPurged = "TargetPurged" // Condition Type recorded on a RestoreJob while we wait for the source // Cluster's LastArchivedWAL to reach the Backup's EndWal. Without this // gate the in-place purge can race the source primary's archive_command // and ship the destructive Cluster+PVC delete before the WALs needed to // replay the backup land in object storage - recovery pods then loop // with "WAL not found" and the source data is gone with the PVC. restoreCondWALArchiveReady = "WALArchiveReady" // Condition Type recorded on a RestoreJob when the chart-rendered // recovery Cluster keeps failing to converge - i.e. its bootstrap- // recovery pods exit non-zero and CNPG recreates them in a loop. Set to // False with reason RecoveryTargetUnreachable when a recoveryTime is in // play (the usual cause: a point-in-time target past the latest archived // WAL) or RecoveryFailed otherwise. Lets a tenant see *why* the restore // is failing before the full restore deadline elapses. restoreCondRecoveryConverged = "RecoveryConverged" // True while clearing spec.bootstrap.enabled keeps failing transiently, before // the grace window is exhausted. Without it the RestoreJob sits non-terminal // for up to the whole window with nothing on .status explaining the wait. restoreCondBootstrapDisablePending = "BootstrapDisablePending" // Default deadline on the time a RestoreJob can spend waiting for the // target Cluster to reach a healthy state. Tenants override this via // RestoreJob.spec.options.restoreTimeoutSeconds when the source DB is // large enough that 30 minutes isn't enough. cnpgDefaultRestoreDeadline = 30 * time.Minute // Fixed grace window the post-convergence bootstrap-disable step keeps // requeueing over a transient error before it terminates the restore Failed. // It is deliberately NOT tied to restoreTimeoutSeconds: that knob bounds the // *recovery* wait (a tenant sets it short to fail fast on a stuck PITR // target), whereas this window absorbs a transient control-plane blip between // convergence and clearing bootstrap.enabled - a blip whose duration has // nothing to do with how long recovery is allowed to take or how big the // database is. A fixed value rides out those blips without letting a short // timeout shrink it or a large one inflate it. It is sized at 30m, near the old // max(restoreDeadline, floor) it replaced, because the disable Patch travels // through the aggregated cozystack-api apiserver and the expensive direction // here is a FALSE Failed (a resubmit's purge-guard then deletes the healthy // restored Cluster + PVCs, per the branch below) - and cozystack-api can itself // be unavailable for several minutes during a platform upgrade, so a 5m window // would fail restores that a slightly longer one rides out. cnpgBootstrapDisableGrace = 30 * time.Minute // Wall-clock cap on how long the WALArchiveReady gate can stay False // before the RestoreJob is marked Failed. The gate fires before the // destructive purge to confirm the backup's endWal is in object // storage; if the source's lastArchivedWAL never advances, WAL // archiving on the source is broken (no barman-cloud plugin wired on // the Cluster, S3 outage, bad credentials, etc.) - waiting longer won't // fix it. This is independent of cnpgDefaultRestoreDeadline because // archive lag and recovery time scale with completely different // inputs: the gate clears in seconds for any healthy cluster // regardless of DB size, while pg_basebackup download + WAL replay // time scales linearly with the data and is what restoreTimeoutSeconds // covers. Failing fast here surfaces the real misconfiguration to // the user instead of burning the whole 30-minute restore budget on // a problem that won't self-resolve. cnpgWALArchiveDeadline = 3 * time.Minute // Both constants below couple to CNPG/PostgreSQL internals. Verified // against CNPG 1.28.1 + barman-cloud plugin (postgresql:18.1) on a test // cluster. The negative e2e in examples/backups/postgres/run-all.sh // (step 46: a recoveryTime past the archive must fail with reason // RecoveryTargetUnreachable) is what catches a drift in either string on // a future CNPG/PostgreSQL bump - if it regresses, this fail-fast silently // degrades back to hanging until the restore deadline. // Container name of a CNPG bootstrap-recovery pod (--full-recovery). // The driver reads this container's log to classify a stuck recovery. cnpgRecoveryContainerName = "full-recovery" // cnpgRecoveryTargetUnreachableLog is the PostgreSQL FATAL a recovery // instance logs when it replays every available WAL without reaching the // configured recovery target and then gives up - the definitive signal // that a point-in-time recoveryTime is past the latest archived WAL. // Matching this exact string (rather than counting failed pods) is what // keeps the fail-fast from misfiring on transient recovery-pod crashes // (node blips, brief API-server unreachability) that CNPG recovers from: // those never emit this message, and the recovery goes on to converge. cnpgRecoveryTargetUnreachableLog = "recovery ended before configured recovery target was reached" // Number of tail lines of the recovery container log the driver scans for // the unreachable-target signature. The message is emitted once near the // end of the replay, so a small tail is enough and keeps the read cheap. cnpgRecoveryLogTailLines = 200 // Max bootstrap-recovery pods whose log the driver reads when classifying a // deadline-expired restore. Bounds the log reads even if many failed // attempts have piled up; the newest few are enough since every attempt // against an unreachable target reproduces the same FATAL. cnpgRecoveryMaxInspectPods = 5 // Cap on the wall-clock time a BackupJob can spend observing a // cnpg.io/Backup stuck in phase=failed before the driver gives up and // marks the BackupJob Failed. CNPG can transition out of "failed" on // internal retries (e.g. transient instance-manager restart), so we // don't fail on the first observation. But without this cap, a // permanently-broken Backup would pin the BackupJob in Running forever, // blocking later runs and wedging the Plan-controller queue. cnpgDefaultBackupDeadline = 30 * time.Minute cnpgClusterHealthyPhase = "Cluster in healthy state" // Condition Type recorded on a BackupJob that has to attach the // barman-cloud plugin to a Cluster carrying none: a fresh platform-flow // app, or one an in-place restore re-rendered. Attaching it changes the // instance pod spec (the plugin injects its sidecar), so CNPG rolls every // instance, and a cnpg.io/Backup created meanwhile fails with "requested // plugin is not available" or "instance manager was restarted during // backup" and never recovers. False holds the Backup back; its // LastTransitionTime, written before the plugin is attached, is what the // instance pods must postdate (see cnpgPluginRolloutPending). backupCondPluginRolledOut = "PluginRolledOut" cnpgPodRoleLabel = "cnpg.io/podRole" cnpgPodRoleInstance = "instance" ) // cnpgClusterNameForApp returns the cnpg.io Cluster name for a Postgres // application instance. The mapping mirrors the postgres ApplicationDefinition // (release.prefix=postgres-). func cnpgClusterNameForApp(appName string) string { return postgresAppPrefix + appName } // defaultCNPGServerName is the WAL-archive serverName of a Cluster whose // strategy names none. It is keyed by the application UID because the // destinationPath is keyed by name only: a Postgres deleted and recreated under // the same name would otherwise archive into its predecessor's prefix, and // barman-cloud's empty-archive check cannot catch it, since CNPG removes // .check-empty-wal-archive once ContinuousArchiving turns True, which happens // on the first no-op archive_command, long before a BackupJob attaches the // plugin. func defaultCNPGServerName(clusterName string, appUID types.UID) string { if appUID == "" { return clusterName } return clusterName + "-" + string(appUID) } // validateCNPGApplicationRef rejects ApplicationRefs that name a Kind/APIGroup // the CNPG driver does not own. The driver assumes apps.cozystack.io/Postgres; // without this gate a ref like other.example.com/Postgres would be accepted // by the Kind check alone and then reconciled against the wrong CRD via the // hard-wired apps.cozystack.io typed client. func validateCNPGApplicationRef(ref corev1.TypedLocalObjectReference) error { if ref.Kind != postgresAppKind { return fmt.Errorf("CNPG strategy supports applicationRef.kind=%q, got %q", postgresAppKind, ref.Kind) } apiGroup := "" if ref.APIGroup != nil { apiGroup = *ref.APIGroup } if apiGroup != "" && apiGroup != postgresapp.GroupName { return fmt.Errorf("CNPG strategy supports applicationRef.apiGroup=%q, got %q", postgresapp.GroupName, apiGroup) } return nil } // --------------------------------------------------------------------------- // BackupJob path // --------------------------------------------------------------------------- func (r *BackupJobReconciler) reconcileCNPG(ctx context.Context, j *backupsv1alpha1.BackupJob, resolved *ResolvedBackupConfig) (ctrl.Result, error) { logger := getLogger(ctx) logger.Debug("reconciling CNPG strategy", "backupjob", j.Name, "phase", j.Status.Phase) if j.Status.Phase == backupsv1alpha1.BackupJobPhaseSucceeded || j.Status.Phase == backupsv1alpha1.BackupJobPhaseFailed { return ctrl.Result{}, nil } if err := validateCNPGApplicationRef(j.Spec.ApplicationRef); err != nil { return r.markBackupJobFailed(ctx, j, err.Error()) } if j.Status.StartedAt == nil { // Refetch the latest persisted state before writing StartedAt: // without this, a stale informer cache that returns StartedAt==nil // would let us write a fresh timestamp on top of one we already // persisted on a previous reconcile, sliding the deadline gate // forward on every poll. MergeFrom on the freshly-fetched object // then makes the write idempotent under concurrent edits to other // status subfields. fresh := &backupsv1alpha1.BackupJob{} if err := r.Get(ctx, types.NamespacedName{Namespace: j.Namespace, Name: j.Name}, fresh); err != nil { return ctrl.Result{}, err } if fresh.Status.StartedAt != nil { j.Status.StartedAt = fresh.Status.StartedAt } else { base := fresh.DeepCopy() now := metav1.Now() fresh.Status.StartedAt = &now if err := r.Status().Patch(ctx, fresh, client.MergeFrom(base)); err != nil { return ctrl.Result{}, err } // Return so the next reconcile re-Gets j with the post-patch // ResourceVersion; copying StartedAt back into the local j here // would leave subsequent r.Status().Update calls in the same // reconcile carrying the pre-patch ResourceVersion and failing // with Conflict. Mirrors the RestoreJob StartedAt path below. return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } } strategy := &strategyv1alpha1.CNPG{} if err := r.Get(ctx, client.ObjectKey{Name: resolved.StrategyRef.Name}, strategy); err != nil { if apierrors.IsNotFound(err) { return r.requeueStrategyNotReady(ctx, j, resolved.StrategyRef.Name) } return ctrl.Result{}, err } app, err := r.getPostgresApp(ctx, j.Namespace, j.Spec.ApplicationRef.Name) if err != nil { if apierrors.IsNotFound(err) { return r.markBackupJobFailed(ctx, j, fmt.Sprintf("Postgres application not found: %s/%s", j.Namespace, j.Spec.ApplicationRef.Name)) } return ctrl.Result{}, err } rendered, err := renderCNPGTemplate(strategy.Spec.Template, app, resolved.Parameters) if err != nil { return r.markBackupJobFailed(ctx, j, fmt.Sprintf("failed to template CNPG strategy: %v", err)) } clusterName := cnpgClusterNameForApp(j.Spec.ApplicationRef.Name) serverName := rendered.ServerName if serverName == "" { serverName = defaultCNPGServerName(clusterName, app.UID) } existingBackup, err := r.findCNPGBackupForJob(ctx, j) if err != nil { return ctrl.Result{}, err } rollout := apimeta.FindStatusCondition(j.Status.Conditions, backupCondPluginRolledOut) awaitingRollout := existingBackup == nil && rollout != nil && rollout.Status == metav1.ConditionFalse attachment, err := r.applyClusterPluginBackup(ctx, j.Namespace, clusterName, rendered, serverName, existingBackup != nil || awaitingRollout) if err != nil { if apierrors.IsNotFound(err) { // HelmRelease has not yet rendered the Cluster (fresh app, or // the operator restart wiped its informer cache). Surface the // situation as a transient condition and back off; failing the // BackupJob would force the tenant to recreate it once the // chart catches up. apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{ Type: "Ready", Status: metav1.ConditionFalse, Reason: "ClusterNotReady", Message: fmt.Sprintf("waiting for cnpg.io/Cluster %s/%s to exist", j.Namespace, clusterName), }) if err := r.Status().Update(ctx, j); err != nil { return ctrl.Result{}, err } return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } return r.markBackupJobFailed(ctx, j, fmt.Sprintf("failed to attach barman-cloud plugin to Cluster: %v", err)) } if attachment.serverName != serverName { logger.Info("preserving the live Cluster's barman serverName over the strategy template's", "cluster", clusterName, "live", attachment.serverName, "strategy", serverName) serverName = attachment.serverName } rolledOut := false if existingBackup == nil && (attachment.missing || awaitingRollout) { if !awaitingRollout { // Latch before attaching: if this write is lost, the next pass // still finds the plugin missing instead of a Cluster that looks // ready to back up while CNPG rolls it. message := fmt.Sprintf("attaching the barman-cloud plugin to cnpg.io/Cluster %s/%s, which rolls its instances", j.Namespace, clusterName) apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{ Type: backupCondPluginRolledOut, Status: metav1.ConditionFalse, Reason: "WaitingForPluginRollout", Message: message, }) apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{ Type: "Ready", Status: metav1.ConditionFalse, Reason: "WaitingForPluginRollout", Message: message, }) if err := r.Status().Update(ctx, j); err != nil { return ctrl.Result{}, err } return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } pending, err := r.cnpgPluginRolloutPending(ctx, j.Namespace, clusterName, attachment, rollout.LastTransitionTime) if err != nil { return ctrl.Result{}, err } if pending != "" { if cnpgBackupDeadlineExceeded(j.Status.StartedAt) { return r.markBackupJobFailed(ctx, j, fmt.Sprintf( "barman-cloud plugin rollout did not finish within %s: %s", cnpgDefaultBackupDeadline, pending)) } if apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{ Type: "Ready", Status: metav1.ConditionFalse, Reason: "WaitingForPluginRollout", Message: pending, }) { if err := r.Status().Update(ctx, j); err != nil { return ctrl.Result{}, err } } return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{ Type: backupCondPluginRolledOut, Status: metav1.ConditionTrue, Reason: "PluginRolledOut", Message: fmt.Sprintf("every instance of cnpg.io/Cluster %s/%s restarted with the barman-cloud plugin", j.Namespace, clusterName), }) apimeta.RemoveStatusCondition(&j.Status.Conditions, "Ready") rolledOut = true } // The chart refuses a flavor change on a live Cluster, but the app keeps // the refused value: the Backup records the image family the data was // written by, which is what a restore has to match. if specFlavor := cnpgFlavor(app.Spec.Flavor); specFlavor != attachment.flavor { logger.Info("recording the live Cluster's image flavor over the application's", "cluster", clusterName, "live", attachment.flavor, "application", specFlavor) } cnpgBackup, err := r.ensureCNPGBackup(ctx, j, clusterName) if err != nil { return r.markBackupJobFailed(ctx, j, fmt.Sprintf("failed to ensure cnpg.io/Backup: %v", err)) } if j.Status.Phase != backupsv1alpha1.BackupJobPhaseRunning || rolledOut { j.Status.Phase = backupsv1alpha1.BackupJobPhaseRunning if err := r.Status().Update(ctx, j); err != nil { return ctrl.Result{}, err } } phase, message := cnpgBackupPhase(cnpgBackup) logger.Debug("cnpg.io/Backup status", "phase", phase, "message", message) switch phase { case cnpgBackupPhaseComplete: if j.Status.BackupRef != nil { return ctrl.Result{}, nil } artifact, err := r.createCNPGBackupArtifact(ctx, j, resolved, cnpgBackup, clusterName, serverName, attachment.flavor, rendered, app) if err != nil { return r.markBackupJobFailed(ctx, j, fmt.Sprintf("failed to create Backup artifact: %v", err)) } now := metav1.Now() j.Status.BackupRef = &corev1.LocalObjectReference{Name: artifact.Name} j.Status.CompletedAt = &now j.Status.Phase = backupsv1alpha1.BackupJobPhaseSucceeded apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{ Type: "Ready", Status: metav1.ConditionTrue, Reason: "BackupCompleted", Message: "cnpg.io Backup completed", }) if err := r.Status().Update(ctx, j); err != nil { return ctrl.Result{}, err } return ctrl.Result{}, nil case cnpgBackupPhaseFailed: // CNPG can transition the same Backup CR back out of "failed" when it // retries internally (e.g., after a transient instance-manager // restart), so we don't fail the BackupJob on the first observation. // But without a cap a permanently-broken Backup would pin the // BackupJob in Running forever; once StartedAt + cnpgDefaultBackupDeadline // has elapsed and the cnpg.io/Backup is still failed, we give up. if cnpgBackupDeadlineExceeded(j.Status.StartedAt) { final := message if final == "" { final = fmt.Sprintf( "cnpg.io Backup remained in phase=failed past %s deadline", cnpgDefaultBackupDeadline) } return r.markBackupJobFailed(ctx, j, final) } if message == "" { message = "cnpg.io Backup reported phase=failed; awaiting recovery or retry" } apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{ Type: "Ready", Status: metav1.ConditionFalse, Reason: "CNPGBackupTransientFailure", Message: message, }) if err := r.Status().Update(ctx, j); err != nil { return ctrl.Result{}, err } return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil default: return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } } // cnpgBackupDeadlineExceeded reports whether enough wall-clock time has // elapsed since the BackupJob started that we should stop polling a stuck // cnpg.io/Backup and fail the run. Returns false when StartedAt is nil so // the very first reconcile (which sets StartedAt) does not trip the gate. func cnpgBackupDeadlineExceeded(startedAt *metav1.Time) bool { if startedAt == nil { return false } return time.Since(startedAt.Time) > cnpgDefaultBackupDeadline } // cnpgPurgeNeeded decides whether the controller should delete the live // CNPG Cluster + PVCs as part of a CNPG restore. Two conditions skip the // destructive step: // 1. The RestoreJob already records that we've purged // (restoreCondTargetPurged=True). Normal idempotent path. // 2. The live Cluster is a freshly-recovered one that THIS restore's own // purge + chart re-render produced (see cnpgClusterFreshlyRecovered). // This only happens when an earlier reconcile purged successfully but the // status-condition write failed; the chart has since re-rendered the // Cluster with our restore-shaped values. Re-purging here would delete // the Cluster CNPG is actively bootstrapping from S3. // // A live Cluster that carries spec.bootstrap.recovery but predates this // RestoreJob is NOT a skip signal: it is leftover from an earlier, already- // completed restore and still holds the old data. Skipping the purge on it // (the previous behaviour, which keyed only on "has recovery bootstrap") made // a repeat in-place restore a silent no-op - the job reported Succeeded while // the PVC, disk, and data were never touched. func cnpgPurgeNeeded(purgedCondition, liveClusterFreshlyRecovered bool) bool { if purgedCondition { return false } return !liveClusterFreshlyRecovered } // cnpgClusterFreshlyRecovered reports whether a live cnpg.io Cluster that // carries spec.bootstrap.recovery was produced by THIS RestoreJob's own purge // + chart re-render rather than rendered for some other restore. // // Only the former is safe to skip re-purging. Any other recovery Cluster holds // another restore's data and must be purged for this restore to re-bootstrap // from its backup. That includes a Cluster a concurrent RestoreJob produced, // and one the chart re-renders from an earlier restore's values after being // deleted while this restore waits for the WAL archive: both are created after // this job's StartedAt, so a timestamp alone cannot tell them apart. // // The chart annotates the Cluster with bootstrap.newServerName, which the // restore patch sets to restoredServerName(cluster, RestoreJob UID). The value // lives on the object the chart renders, not in RestoreJob status, so it // survives the failed status write the skip exists for, and it names the // restore that rendered the Cluster. When the annotation is present, the // Cluster is fresh only if it carries wantMark. // // A Cluster without the annotation was rendered either by a chart that // predates it or from values with no bootstrap.newServerName (a manually // configured recovery). It is fresh only if appMark, the application's current // bootstrap.newServerName, is wantMark - this restore has patched the values - // and it was created strictly after StartedAt, which separates a Cluster from // an earlier completed restore from this restore's own re-render. // // Returns false when freshness cannot be determined (no recovery bootstrap, or // a missing timestamp on the fallback path): the caller then purges, which is // the safe default. func cnpgClusterFreshlyRecovered(state recoveryClusterState, wantMark, appMark string, restoreStartedAt *metav1.Time) bool { if !state.hasRecovery { return false } if state.restoreMark != "" { return state.restoreMark == wantMark } if appMark != wantMark || state.createdAt == nil || restoreStartedAt == nil { return false } return state.createdAt.After(restoreStartedAt.Time) } // cnpgClusterGVR addresses Clusters on the dynamic client, for reads that must // bypass the manager's cache. var cnpgClusterGVR = schema.GroupVersionResource{ Group: cnpgtypes.GroupName, Version: cnpgtypes.Version, Resource: "clusters", } // applyClusterPluginBackup wires the templated strategy onto the live CNPG // Cluster through the barman-cloud plugin: it SSA-applies an ObjectStore CR // carrying the S3/barman configuration and SSA-patches the Cluster's // spec.plugins to reference it. This replaces the deprecated native // spec.backup.barmanObjectStore path (removed from the `standard` image // variant in CNPG 1.29). The driver owns both objects via its own field // manager so the chart - which only emits them for the non-platform flow - // does not contend. // // Returns an apierrors.IsNotFound error when the Cluster has not yet been // rendered by the HelmRelease. The Cluster SSA on its own would fail with a // hard validation error (CNPG's Cluster CRD has many required fields the // driver does not set), so we surface the precondition explicitly and let // the caller treat it as a retryable wait. // applyClusterPluginBackup returns the serverName the Cluster effectively // archives under. A live Cluster that already has the barman-cloud plugin // attached (chart-rendered, or a previous BackupJob) keeps its current // serverName even when the strategy template names a different one: the // serverName is the S3 path prefix of the WAL archive, and flipping it // mid-stream splits the archive across two prefixes — WALs around the flip // land under the old prefix while the base backup indexes under the new one, // and the eventual restore fails with "WAL not found". // // It also returns the flavor of the image the live Cluster runs, read from // the same object. // // With attach unset, a Cluster that has no barman-cloud plugin yet is left // untouched and only reported missing, so the caller can record that it is // about to roll the Cluster before it does. func (r *BackupJobReconciler) applyClusterPluginBackup(ctx context.Context, namespace, clusterName string, t *strategyv1alpha1.CNPGTemplate, serverName string, attach bool) (cnpgPluginAttachment, error) { // Live read: the cached Cluster can predate a restore re-render, and the // serverName, the flavor and the UID below must come from the current // object. existing := &cnpgtypes.Cluster{} live, err := r.Resource(cnpgClusterGVR).Namespace(namespace).Get(ctx, clusterName, metav1.GetOptions{}) if err != nil { return cnpgPluginAttachment{}, err } if err := runtime.DefaultUnstructuredConverter.FromUnstructured(live.Object, existing); err != nil { return cnpgPluginAttachment{}, fmt.Errorf("decode Cluster %s/%s: %w", namespace, clusterName, err) } attachment := cnpgPluginAttachment{ serverName: serverName, flavor: cnpgImageFlavor(existing.Spec.ImageName), clusterStatus: existing.Status, } if live := currentBarmanServerName(existing); live != "" { attachment.serverName = live } else { attachment.missing = true if !attach { return attachment, nil } } serverName = attachment.serverName // The ObjectStore is named after the Cluster (distinct kind, same // namespace). serverName is deliberately left off the ObjectStore // configuration - the plugin forbids it there and takes it from the // Cluster plugin parameter instead - so backups keep landing under the // same s3://...// path the native barmanObjectStore used. objStoreName := clusterName objStore := newObjectStorePatch(namespace, objStoreName) // Own the ObjectStore by the live Cluster so Kubernetes garbage-collects it // when the Cluster is deleted (tenant app teardown, or the RestoreJob purge). // The platform flow does not chart-render this ObjectStore, so without an // owner reference it would otherwise be orphaned in the tenant namespace. // Owner and owned are in the same namespace, as GC requires. controller := true objStore.OwnerReferences = []metav1.OwnerReference{{ APIVersion: cnpgtypes.GroupVersion.String(), Kind: "Cluster", Name: existing.Name, UID: existing.UID, Controller: &controller, }} objStore.Spec = cnpgtypes.ObjectStoreSpec{ Configuration: *buildBarmanObjectStore(t.BarmanObjectStore, ""), RetentionPolicy: t.BarmanObjectStore.RetentionPolicy, InstanceSidecarConfiguration: barmanSidecarConfiguration(), } if err := r.Patch(ctx, objStore, client.Apply, client.FieldOwner(cnpgFieldManager), client.ForceOwnership); err != nil { return cnpgPluginAttachment{}, fmt.Errorf("apply ObjectStore %s/%s: %w", namespace, objStoreName, err) } patch := newCNPGClusterPatch(namespace, clusterName) patch.Spec.Plugins = []cnpgtypes.PluginConfiguration{buildBarmanPlugin(objStoreName, serverName)} if err := r.Patch(ctx, patch, client.Apply, client.FieldOwner(cnpgFieldManager), client.ForceOwnership); err != nil { return cnpgPluginAttachment{}, err } return attachment, nil } // cnpgPluginAttachment is what applyClusterPluginBackup read off the live // Cluster before applying the plugin. type cnpgPluginAttachment struct { // serverName is the serverName the Cluster effectively archives under. serverName string // flavor is the image flavor the live Cluster runs. flavor string // missing reports a Cluster that carried no barman-cloud plugin. missing bool clusterStatus cnpgtypes.ClusterStatus } // cnpgPluginRolloutPending returns why the rollout that attaching the // barman-cloud plugin started is not over yet, or "" once it is: CNPG reports // the Cluster healthy and every instance pod was created after attachedAt, // and is Ready. // // The phase alone cannot tell: the Cluster status carries no // observedGeneration and its Ready condition is derived from the phase, so // until CNPG reconciles the new spec the phase still reads healthy from // before. Nor can status.pluginStatus, which also lists plugins referenced // only from externalClusters, as on a Cluster an in-place restore // re-rendered. A pod spec change cannot be applied in place, so the instance // pods are recreated, and pods newer than the attach are the ones that run // the plugin sidecar. func (r *BackupJobReconciler) cnpgPluginRolloutPending(ctx context.Context, namespace, clusterName string, attachment cnpgPluginAttachment, attachedAt metav1.Time) (string, error) { if phase := attachment.clusterStatus.Phase; phase != cnpgClusterHealthyPhase { pending := fmt.Sprintf("cnpg.io/Cluster %s/%s is in phase %q", namespace, clusterName, phase) if reason := attachment.clusterStatus.PhaseReason; reason != "" { pending += ": " + reason } return pending, nil } pods := &corev1.PodList{} if err := r.List(ctx, pods, client.InNamespace(namespace), client.MatchingLabels{ cnpgClusterLabel: clusterName, cnpgPodRoleLabel: cnpgPodRoleInstance, }); err != nil { return "", err } if len(pods.Items) == 0 { return fmt.Sprintf("cnpg.io/Cluster %s/%s has no instance pod", namespace, clusterName), nil } for i := range pods.Items { pod := &pods.Items[i] if !pod.CreationTimestamp.After(attachedAt.Time) { return fmt.Sprintf("instance pod %s/%s predates the barman-cloud plugin", namespace, pod.Name), nil } if pod.DeletionTimestamp != nil || !podReady(pod) { return fmt.Sprintf("instance pod %s/%s is not ready", namespace, pod.Name), nil } } return "", nil } func podReady(pod *corev1.Pod) bool { for _, c := range pod.Status.Conditions { if c.Type == corev1.PodReady { return c.Status == corev1.ConditionTrue } } return false } // currentBarmanServerName returns the serverName the Cluster's barman-cloud // plugin currently archives under: the explicit plugin parameter when set, the // Cluster's own name when the plugin is attached without one (the plugin's // documented default), and "" when the plugin is not attached at all. func currentBarmanServerName(c *cnpgtypes.Cluster) string { for _, p := range c.Spec.Plugins { if p.Name != cnpgtypes.PluginName { continue } if name := p.Parameters[barmanServerNameParam]; name != "" { return name } return c.Name } return "" } // ensureCNPGBackup creates a one-shot postgresql.cnpg.io/Backup CR labelled // with the BackupJob, or returns the existing one if a previous reconcile // already created it. Idempotency relies on the OwningJob labels. func (r *BackupJobReconciler) ensureCNPGBackup(ctx context.Context, j *backupsv1alpha1.BackupJob, clusterName string) (*cnpgtypes.Backup, error) { existing, err := r.findCNPGBackupForJob(ctx, j) if err != nil { return nil, err } if existing != nil { return existing, nil } obj := &cnpgtypes.Backup{ ObjectMeta: metav1.ObjectMeta{ Namespace: j.Namespace, GenerateName: fmt.Sprintf("%s-", j.Name), Labels: map[string]string{ backupsv1alpha1.OwningJobNameLabel: j.Name, backupsv1alpha1.OwningJobNamespaceLabel: j.Namespace, cnpgClusterLabel: clusterName, }, }, Spec: cnpgtypes.BackupSpec{ Method: cnpgtypes.BackupMethodPlugin, Cluster: cnpgtypes.ClusterReference{Name: clusterName}, PluginConfiguration: &cnpgtypes.BackupPluginConfiguration{Name: cnpgtypes.PluginName}, }, } if err := r.Create(ctx, obj); err != nil { return nil, err } return obj, nil } // findCNPGBackupForJob returns the cnpg.io/Backup labelled with the // BackupJob's OwningJob{Name,Namespace}, if any. Returns (nil, nil) when // no match is found. func (r *BackupJobReconciler) findCNPGBackupForJob(ctx context.Context, j *backupsv1alpha1.BackupJob) (*cnpgtypes.Backup, error) { list := &cnpgtypes.BackupList{} if err := r.List(ctx, list, client.InNamespace(j.Namespace), client.MatchingLabels{ backupsv1alpha1.OwningJobNameLabel: j.Name, backupsv1alpha1.OwningJobNamespaceLabel: j.Namespace, }, ); err != nil { return nil, err } if len(list.Items) == 0 { return nil, nil } return &list.Items[0], nil } // createCNPGBackupArtifact materialises a Cozystack Backup resource carrying // the metadata callers need to drive a future restore. The source app's // spec.databases / spec.users are snapshotted into Status.UnderlyingResources // so a future RestoreJob can mirror them onto the target without re-reading // the source CR (which may have been deleted by the time the restore fires). func (r *BackupJobReconciler) createCNPGBackupArtifact( ctx context.Context, j *backupsv1alpha1.BackupJob, resolved *ResolvedBackupConfig, cnpgBackup *cnpgtypes.Backup, clusterName, serverName, flavor string, rendered *strategyv1alpha1.CNPGTemplate, sourceApp *postgresapp.Postgres, ) (*backupsv1alpha1.Backup, error) { takenAt := metav1.Now() if cnpgBackup.Status.StartedAt != nil && !cnpgBackup.Status.StartedAt.IsZero() { takenAt = *cnpgBackup.Status.StartedAt } driverMD := map[string]string{ cnpgBackupNameKey: cnpgBackup.Name, cnpgServerNameKey: serverName, cnpgDestinationPathKey: rendered.BarmanObjectStore.DestinationPath, cnpgEndpointURLKey: rendered.BarmanObjectStore.EndpointURL, cnpgClusterNameKey: clusterName, } if rendered.BarmanObjectStore.S3Credentials != nil { driverMD[cnpgS3SecretRefKey] = rendered.BarmanObjectStore.S3Credentials.SecretRef.Name } // Copied rather than read back at restore time: barman retention deletes // the cnpg.io/Backup while the artifact can still be restored from. if cnpgBackup.Status.MajorVersion > 0 { driverMD[cnpgMajorVersionKey] = strconv.Itoa(cnpgBackup.Status.MajorVersion) } underlyingResources, err := marshalCNPGBackupSnapshot(sourceApp, resolved.Parameters, flavor) if err != nil { return nil, fmt.Errorf("encode source snapshot for Backup.status.underlyingResources: %w", err) } backup := &backupsv1alpha1.Backup{ ObjectMeta: metav1.ObjectMeta{ Name: j.Name, Namespace: j.Namespace, }, Spec: backupsv1alpha1.BackupSpec{ ApplicationRef: j.Spec.ApplicationRef, StrategyRef: resolved.StrategyRef, TakenAt: takenAt, DriverMetadata: driverMD, }, Status: backupsv1alpha1.BackupStatus{ Phase: backupsv1alpha1.BackupPhaseReady, Artifact: &backupsv1alpha1.BackupArtifact{ URI: fmt.Sprintf("cnpg://%s/%s", serverName, cnpgBackup.Name), }, UnderlyingResources: underlyingResources, }, } if j.Spec.PlanRef != nil { backup.Spec.PlanRef = j.Spec.PlanRef } if err := r.Create(ctx, backup); err != nil { // AlreadyExists means a previous reconcile created the artifact and // then raced on the BackupJob Status().Update. Returning the error // would flip the next reconcile to Failed even though the artifact // is in place. Fetch and return the existing object instead. if !apierrors.IsAlreadyExists(err) { return nil, err } existing := &backupsv1alpha1.Backup{} if getErr := r.Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name}, existing); getErr != nil { return nil, getErr } return existing, nil } return backup, nil } // --------------------------------------------------------------------------- // RestoreJob path // --------------------------------------------------------------------------- func (r *RestoreJobReconciler) reconcileCNPGRestore(ctx context.Context, restoreJob *backupsv1alpha1.RestoreJob, backup *backupsv1alpha1.Backup) (ctrl.Result, error) { logger := getLogger(ctx) logger.Debug("reconciling CNPG restore", "restorejob", restoreJob.Name, "backup", backup.Name) if err := validateCNPGApplicationRef(backup.Spec.ApplicationRef); err != nil { return r.markRestoreJobFailed(ctx, restoreJob, err.Error()) } if restoreJob.Status.StartedAt == nil { // Refetch before writing StartedAt: a stale informer cache that // returns StartedAt==nil after we already persisted it on a // previous reconcile would otherwise let us slide the timestamp // forward, advancing the deadline gate on every poll. fresh := &backupsv1alpha1.RestoreJob{} if err := r.Get(ctx, types.NamespacedName{Namespace: restoreJob.Namespace, Name: restoreJob.Name}, fresh); err != nil { return ctrl.Result{}, err } if fresh.Status.StartedAt != nil { restoreJob.Status.StartedAt = fresh.Status.StartedAt return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } base := fresh.DeepCopy() now := metav1.Now() fresh.Status.StartedAt = &now fresh.Status.Phase = backupsv1alpha1.RestoreJobPhaseRunning if err := r.Status().Patch(ctx, fresh, client.MergeFrom(base)); err != nil { return ctrl.Result{}, err } restoreJob.Status.StartedAt = fresh.Status.StartedAt restoreJob.Status.Phase = fresh.Status.Phase return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } target := r.resolveCNPGRestoreTarget(restoreJob, backup) if target.Kind != postgresAppKind { return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf( "target applicationRef.kind=%q is not supported by the CNPG driver", target.Kind)) } if target.APIGroup != "" && target.APIGroup != postgresapp.GroupName { return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf( "target applicationRef.apiGroup=%q is not supported by the CNPG driver", target.APIGroup)) } strategy := &strategyv1alpha1.CNPG{} if err := r.Get(ctx, client.ObjectKey{Name: backup.Spec.StrategyRef.Name}, strategy); err != nil { if apierrors.IsNotFound(err) { return r.requeueRestoreStrategyNotReady(ctx, restoreJob, backup.Spec.StrategyRef.Name) } return ctrl.Result{}, err } targetApp, err := r.getPostgresApp(ctx, target.Namespace, target.AppName) if err != nil { if apierrors.IsNotFound(err) { return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf( "target Postgres application not found: %s/%s (deploy it before requesting a copy restore)", target.Namespace, target.AppName)) } return ctrl.Result{}, err } sourceServerName := backup.Spec.DriverMetadata[cnpgServerNameKey] sourceDestinationPath := backup.Spec.DriverMetadata[cnpgDestinationPathKey] sourceEndpointURL := backup.Spec.DriverMetadata[cnpgEndpointURLKey] if sourceServerName == "" || sourceDestinationPath == "" { return r.markRestoreJobFailed(ctx, restoreJob, "Backup driverMetadata is missing required CNPG fields") } options, err := parseCNPGRestoreOptions(restoreJob.Spec.Options) if err != nil { // Don't fail the RestoreJob on a parse error - we want behaviour to // stay permissive against future field additions in the typed // CNPGRestoreOptions struct. But log the error and surface a // transient condition so a tenant who wonders why their // recoveryTime didn't apply has a breadcrumb to follow. logger.Info("malformed restoreJob.spec.options; falling back to defaults", "error", err) r.Recorder.Eventf(restoreJob, corev1.EventTypeWarning, "MalformedOptions", "spec.options is not valid JSON; falling back to defaults: %v", err) } // The chart's init-job runs post-install and DROPs any database/role on // the recovered cluster that isn't declared in the target app spec, so // the patched target MUST carry source's spec.databases / spec.users. // We read those from the Backup artifact's status.underlyingResources // snapshot taken at backup time - never from the live source app: // (a) the source app may have been deleted before the restore fires, // and we still need the snapshot to drive a safe restore; // (b) source spec drift between backup time and restore time would // otherwise silently re-shape the recovered roles/databases. // // The snapshot also carries the BackupClassStrategy parameters that // were in effect at backup time (see cnpgBackupSnapshot doc). We feed // them back into renderCNPGTemplate below so a strategy that uses // `{{ .Parameters.foo }}` for a secret reference / endpointCA / key // name renders the same value at restore time as it did at backup // time. Without this, restore-time rendering with parameters=nil // silently produced `` and patched the target Postgres app // with broken Secret references. snapshot, err := unmarshalCNPGBackupSnapshot(backup) if err != nil { return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf( "Backup %s/%s carries no usable source-spec snapshot in status.underlyingResources: %v "+ "(re-take the backup with a controller version that persists source spec)", backup.Namespace, backup.Name, err)) } sourceDatabases, sourceUsers, sourceParameters := snapshot.Databases, snapshot.Users, snapshot.Parameters // Checked before the destructive purge below, which would otherwise wipe // the target only for the recovery to come up on the wrong image family. if src, dst := cnpgFlavor(snapshot.Flavor), cnpgFlavor(targetApp.Spec.Flavor); src != dst { return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf( "Backup %s/%s was taken from a %s cluster and cannot be restored into %s/%s, which runs the %s flavor: "+ "a physical restore must keep the image family. Restore into a Postgres app with flavor %s, "+ "or move the data with a logical dump (pg_dump / pg_restore)", backup.Namespace, backup.Name, src, target.Namespace, target.AppName, dst, src)) } rendered, err := renderCNPGTemplate(strategy.Spec.Template, targetApp, sourceParameters) if err != nil { return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf("failed to template CNPG strategy: %v", err)) } clusterName := cnpgClusterNameForApp(target.AppName) // CNPG forbids changing bootstrap on an initialized Cluster ("Only one // bootstrap" CRD validation), so the only way to make it consume // bootstrap.recovery is to delete the live Cluster + PVCs. The chart // re-renders with the patched values producing a fresh Cluster. The same // applies to in-place AND to-copy: in to-copy the target Postgres app // was likely already initdb-bootstrapped, so we must purge it too. // // Idempotency: once we've purged for this RestoreJob, record it on a // status condition so subsequent reconciles don't loop on the purge // step every time the new Cluster comes up with bootstrap.recovery. // Without this guard, a retried RestoreJob targeting an already-restored // app would skip the purge and report success against stale data. // // Status writes can race (informer-stale conflicts, transient etcd // errors). If Status().Update fails after a successful purge, the next // reconcile would re-enter this block with purgedCondition=false. To // avoid re-purging the freshly-recovered Cluster in that case, we also // check the live Cluster for bootstrap.recovery: if present, the chart // has already re-rendered after a previous purge, and we must NOT delete // it again. clusterState, err := r.recoveryBootstrapClusterState(ctx, target.Namespace, clusterName) if err != nil { return ctrl.Result{}, err } purgedCondition := apimeta.IsStatusConditionTrue(restoreJob.Status.Conditions, restoreCondTargetPurged) // Only skip the destructive purge for a recovery Cluster THIS restore just // produced. A recovery Cluster rendered for any other restore still holds // that restore's data, so it must be purged - otherwise a repeat in-place // restore silently no-ops. freshlyRecovered := cnpgClusterFreshlyRecovered(clusterState, restoredServerName(clusterName, restoreJob.UID), targetApp.Spec.Bootstrap.NewServerName, restoreJob.Status.StartedAt) if cnpgPurgeNeeded(purgedCondition, freshlyRecovered) { // A physical restore only opens on the major version that wrote the // data files. Across majors the recovery pods loop on "database files // are incompatible with server" until the deadline, with the target // already purged. spec.version is what the chart renders the // recovered Cluster from, so it is compared rather than the live // Cluster's image. An unknown side keeps the restore going: refusing // would break every backup taken before the version was recorded. backupMajor, err := r.cnpgBackupMajorVersion(ctx, backup) if err != nil { return ctrl.Result{}, err } targetMajor := postgresAppMajorVersion(targetApp.Spec.Version) switch { case backupMajor == 0 || targetMajor == 0: logger.Info("PostgreSQL major version unknown; restoring without the version check", "backup", backup.Name, "backupMajorVersion", backupMajor, "targetVersion", targetApp.Spec.Version) case backupMajor != targetMajor: return r.markRestoreJobFailedReason(ctx, restoreJob, "MajorVersionMismatch", fmt.Sprintf( "Backup %s/%s was taken on PostgreSQL %d and cannot be restored into %s/%s, which runs PostgreSQL %d (spec.version=%q): "+ "a physical restore needs the same major version; the target was left untouched", backup.Namespace, backup.Name, backupMajor, target.Namespace, target.AppName, targetMajor, targetApp.Spec.Version)) } // Gate the destructive flow on the source cluster having shipped // the backup's required WALs to object storage. archive_command runs // on the source primary; once we delete the Cluster + PVCs, any // unshipped WALs on the primary's pg_wal are gone for good and the // recovery driver fails with "WAL not found" against the empty // archive. // // The gate runs BEFORE patchPostgresAppForRestore: flipping // bootstrap.enabled on the Postgres app triggers Flux to upgrade // the HelmRelease, and CNPG admission rejects the live Cluster's // bootstrap.initdb -> bootstrap.recovery swap with "Only one // bootstrap". helm-controller then drops into an UpgradeFailed / // RollbackFailed loop whose rollbacks strip the SSA-applied // spec.plugins (and its ObjectStore) - which stops WAL archiving // and pins lastArchivedWAL at "" forever. By gating before the // patch, the HelmRelease stays untouched while we wait, the // source primary keeps archiving, and the chart's failure-loop // race only happens once we already have everything we need. walReady, walMessage, err := r.cnpgBackupWALArchived(ctx, backup) if err != nil { return ctrl.Result{}, err } if !walReady { apimeta.SetStatusCondition(&restoreJob.Status.Conditions, metav1.Condition{ Type: restoreCondWALArchiveReady, Status: metav1.ConditionFalse, Reason: "WaitingForArchive", Message: walMessage, }) if err := r.Status().Update(ctx, restoreJob); err != nil { return ctrl.Result{}, err } // Fail fast if the gate has been False past the dedicated // archive deadline. apimeta.SetStatusCondition only // touches LastTransitionTime when Status changes, so once // the in-memory slice has been updated by the call above // it carries the same persisted timestamp that an operator // inspecting the CR would see - the elapsed window // measures from the first observation, and a gate that // briefly cleared then re-failed gets a fresh budget. walDeadline := options.effectiveWALArchiveDeadline() if cond := apimeta.FindStatusCondition(restoreJob.Status.Conditions, restoreCondWALArchiveReady); cond != nil && time.Since(cond.LastTransitionTime.Time) > walDeadline { return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf( "WAL archive did not catch up to the backup endWal within %s: %s "+ "(check that the source cnpg.io/Cluster has the barman-cloud plugin wired - spec.plugins referencing an ObjectStore - and is shipping WALs to object storage; override via spec.options.walArchiveTimeoutSeconds)", walDeadline, walMessage)) } return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } if !apimeta.IsStatusConditionTrue(restoreJob.Status.Conditions, restoreCondWALArchiveReady) { apimeta.SetStatusCondition(&restoreJob.Status.Conditions, metav1.Condition{ Type: restoreCondWALArchiveReady, Status: metav1.ConditionTrue, Reason: "ArchiveCaughtUp", Message: walMessage, }) } // Suspend the target's HelmRelease before mutating the Postgres // app spec. helm-controller observes Postgres CR changes // asynchronously; if it sees bootstrap.enabled=true while the // live Cluster still has bootstrap.initdb, it tries an upgrade // that CNPG admission rejects with "Only one bootstrap method // can be specified at a time", drops into an UpgradeFailed/ // RollbackFailed loop, and the rollback resurrects an initdb // cluster that races our purge. With the HR suspended Helm stops // reconciling, we patch + purge in peace, and only resume once // the Cluster + PVCs are fully gone so the next render lands // bootstrap.recovery on an empty namespace. sourceCNPGBackup, err := r.cnpgSourceBackup(ctx, backup) if err != nil { return ctrl.Result{}, err } if sourceCNPGBackup == nil { r.Recorder.Eventf(restoreJob, corev1.EventTypeWarning, "BaseBackupGone", "cnpg.io/Backup %s/%s no longer exists, so its base backup has likely been removed by retention; the barman-cloud plugin picks the base backup from the catalog", backup.Namespace, backup.Spec.DriverMetadata[cnpgBackupNameKey]) } backupID := cnpgRestoreBackupID(sourceCNPGBackup, options.RecoveryTime) if backupID == "" { logger.Info("restore does not pin its base backup; the barman-cloud plugin picks one from the catalog", "backup", backup.Name, "recoveryTime", options.RecoveryTime) } hrName := postgresAppPrefix + target.AppName if err := r.setCNPGRestoreHRSuspended(ctx, target.Namespace, hrName, true); err != nil { return ctrl.Result{}, err } newServerName := restoredServerName(clusterName, restoreJob.UID) if newServerName == sourceServerName { // Practically unreachable (see restoredServerName), but if a digest // ever collides the restored cluster would archive onto the source's // prefix and CNPG would fail with "Expected empty archive" - log so // that recurrence is diagnosable as a hash collision, not a mystery. logger.Info("restored WAL-archive serverName collided with the recovery source", "cluster", clusterName, "serverName", newServerName) } if err := r.patchPostgresAppForRestore(ctx, targetApp, sourceServerName, newServerName, sourceDestinationPath, sourceEndpointURL, options.RecoveryTime, backupID, rendered.BarmanObjectStore.S3Credentials, rendered.BarmanObjectStore.EndpointCA, sourceDatabases, sourceUsers); err != nil { // Resume HR before terminal failure so an operator deleting // the failed RestoreJob does not leave the HR stuck. _ = r.setCNPGRestoreHRSuspended(ctx, target.Namespace, hrName, false) return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf("failed to patch Postgres app spec: %v", err)) } if err := r.purgeExistingCluster(ctx, target.Namespace, clusterName); err != nil { _ = r.setCNPGRestoreHRSuspended(ctx, target.Namespace, hrName, false) return r.markRestoreJobFailed(ctx, restoreJob, fmt.Sprintf("failed to purge existing cluster: %v", err)) } // Wait for the Cluster + its PVCs to actually disappear before // resuming the HR. r.Delete returns when DeletionTimestamp lands // but CNPG's finalizers + storage drain take additional time; // resuming earlier lets helm race the still-terminating cluster // and we end up in the same admission-rejection loop the // suspend was meant to prevent. gone, err := r.cnpgClusterFullyGone(ctx, target.Namespace, clusterName) if err != nil { return ctrl.Result{}, err } if !gone { return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } } // Idempotently resume the HR. Runs both at the bottom of the // destructive block (cluster + PVCs gone -> chart-render is now // safe) and on every subsequent reconcile after TargetPurged is // set, so a controller crash between "cluster gone" and "HR resumed" // gets repaired on the next pass instead of stranding the HR. hrName := postgresAppPrefix + target.AppName if err := r.setCNPGRestoreHRSuspended(ctx, target.Namespace, hrName, false); err != nil { return ctrl.Result{}, err } if !purgedCondition { apimeta.SetStatusCondition(&restoreJob.Status.Conditions, metav1.Condition{ Type: restoreCondTargetPurged, Status: metav1.ConditionTrue, Reason: "ClusterPurged", Message: fmt.Sprintf("Cluster %s/%s deleted; awaiting chart-rendered re-bootstrap", target.Namespace, clusterName), }) if err := r.Status().Update(ctx, restoreJob); err != nil { return ctrl.Result{}, err } return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } // Health wins unconditionally, and is checked BEFORE the deadline. A // recovery that has reached a healthy Cluster is a success no matter how // late the reconcile observes it: a reconcile gap longer than // restoreTimeoutSeconds (a controller restart or a stalled workqueue that // spans the window) must not flip an already-converged restore to Failed. // Getting the order wrong is not just a cosmetic mislabel - a false Failed // can be resubmitted, and the next RestoreJob's purge guard would then // delete the already-recovered Cluster + PVCs to start over. if clusterState.hasRecovery { healthy, herr := r.cnpgClusterHealthy(ctx, target.Namespace, clusterName) if herr != nil { return ctrl.Result{}, herr } if healthy && !apimeta.IsStatusConditionTrue(restoreJob.Status.Conditions, restoreCondRecoveryConverged) { // Latch convergence durably BEFORE the terminal step below disables // bootstrap. Disabling bootstrap re-renders the Cluster without // spec.bootstrap.recovery, so hasRecovery never reports true again; // a crash between that flip and the terminal status write must not // drop an already-converged restore into the deadline path and fail // it. Persisting the condition here, on its own Status().Update, // makes the success verdict survive independent of the live // cluster's current bootstrap state. // // RecoveryConverged=True is also symmetric with the False the // unreachable-target path records, so .status.conditions tells the // whole story rather than only ever showing the condition on failure. apimeta.SetStatusCondition(&restoreJob.Status.Conditions, metav1.Condition{ Type: restoreCondRecoveryConverged, Status: metav1.ConditionTrue, Reason: "RecoveryConverged", Message: fmt.Sprintf("target cnpg.io Cluster %s/%s reached a healthy state", target.Namespace, clusterName), }) if err := r.Status().Update(ctx, restoreJob); err != nil { return ctrl.Result{}, err } return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } } // Terminal step, reachable regardless of the live cluster's current // bootstrap state once convergence has latched above. Recovery restores the // source's role catalog and password hashes, but the freshly reconciled // -credentials Secret advertises chart-generated passwords that do // not match those hashes. While bootstrap.enabled stays true the chart skips // the init-job that would reconcile them (see init-job.yaml), so every // application-user login against the target keeps failing. Disable bootstrap // so the post-upgrade init-job runs ALTER ROLE ... WITH PASSWORD for each // spec.users entry, converging the recovered roles onto the generated // Secret. Idempotent, so a crash before the Succeeded write below just // re-runs it on the next reconcile. if apimeta.IsStatusConditionTrue(restoreJob.Status.Conditions, restoreCondRecoveryConverged) { if err := r.disablePostgresAppBootstrap(ctx, target.Namespace, target.AppName); err != nil { // Recovery already converged - the data is restored and reachable as // the CNPG superuser; only clearing bootstrap.enabled (the trigger for // the init-job that reconciles the generated passwords onto the // recovered roles) is failing. Give this step its OWN grace window, // measured from when recovery converged (RecoveryConverged's // LastTransitionTime), NOT from the restore StartedAt: the health // check wins BEFORE the restore deadline on purpose (a large-DB // recovery legitimately outlives it - see the health-before-deadline // ordering above), so a StartedAt-based bound is often already past // the deadline the instant this step first runs and would fail it on // the FIRST transient error (an apiserver restart, a webhook timeout). // That is exactly the false Failed the ordering exists to prevent: a // resubmit's purge-guard would then delete the healthy restored // Cluster + PVCs. Only a failure that persists across the whole window // AFTER convergence terminates as Failed; a transient one requeues. // The window is a fixed cnpgBootstrapDisableGrace, independent of // restoreTimeoutSeconds, sized for a control-plane blip. grace := options.effectiveBootstrapDisableGrace() if cond := apimeta.FindStatusCondition(restoreJob.Status.Conditions, restoreCondRecoveryConverged); cond != nil && time.Since(cond.LastTransitionTime.Time) > grace { return r.markRestoreJobFailedReason(ctx, restoreJob, "BootstrapDisableFailed", fmt.Sprintf( "recovery converged but clearing spec.bootstrap.enabled on Postgres app %s/%s kept failing for %s after convergence, so the restored copy's application credentials will not converge on their own: %v. "+ "The data is restored and reachable as the CNPG superuser; clear bootstrap.enabled on the app - and stop any GitOps source from re-asserting it - so the init-job reconciles the passwords.", target.Namespace, target.AppName, grace, err)) } // Still inside the window. Record WHY the RestoreJob is non-terminal on // .status so it does not sit silent for up to the whole grace window, and // emit the Event once (on the transition into pending), not on every poll. firstFailure := !apimeta.IsStatusConditionTrue(restoreJob.Status.Conditions, restoreCondBootstrapDisablePending) apimeta.SetStatusCondition(&restoreJob.Status.Conditions, metav1.Condition{ Type: restoreCondBootstrapDisablePending, Status: metav1.ConditionTrue, Reason: "Retrying", Message: fmt.Sprintf( "recovery converged; clearing spec.bootstrap.enabled on Postgres app %s/%s is retrying after a transient error and will fail the restore if it does not clear within %s of convergence: %v", target.Namespace, target.AppName, grace, err), }) if updErr := r.Status().Update(ctx, restoreJob); updErr != nil { return ctrl.Result{}, updErr } if firstFailure { r.Recorder.Eventf(restoreJob, corev1.EventTypeWarning, "BootstrapDisablePending", "recovery converged but clearing spec.bootstrap.enabled on Postgres app %s/%s failed; retrying up to %s after convergence: %v", target.Namespace, target.AppName, grace, err) } return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } now := metav1.Now() restoreJob.Status.CompletedAt = &now restoreJob.Status.Phase = backupsv1alpha1.RestoreJobPhaseSucceeded // The DATA is restored and the Cluster is healthy - that is what this // terminal Succeeded asserts. Clearing bootstrap only STARTS the credential // convergence: the chart's post-upgrade init-job runs ALTER ROLE on the next // HelmRelease reconcile, which this controller does not wait for. Carry that // pending handoff as the REASON on Ready rather than as a standalone // CredentialsConverged condition: this controller never observes the // convergence, so such a condition would be structurally False on every // success - indistinguishable from a standing failure to a generic // conditions view - and answer nothing. A reason on Ready=True says the same // thing without pretending to a lifecycle it cannot complete; the message // and the Event below carry the how-to-confirm. apimeta.SetStatusCondition(&restoreJob.Status.Conditions, metav1.Condition{ Type: "Ready", Status: metav1.ConditionTrue, Reason: "RestoreCompletedCredentialsPending", Message: "target cnpg.io Cluster reached a healthy state and spec.bootstrap.enabled was cleared; the app's post-upgrade init-job reconciles the generated passwords onto the recovered roles on the next HelmRelease reconcile (this RestoreJob does not wait for it - confirm by logging in as an application user)", }) // If the disable retried within the window, close out that pending condition. if apimeta.FindStatusCondition(restoreJob.Status.Conditions, restoreCondBootstrapDisablePending) != nil { apimeta.SetStatusCondition(&restoreJob.Status.Conditions, metav1.Condition{ Type: restoreCondBootstrapDisablePending, Status: metav1.ConditionFalse, Reason: "Cleared", Message: "spec.bootstrap.enabled was cleared", }) } // The status message above is honest that credential convergence is still // pending; back it with an Event so the handoff is discoverable in // `kubectl describe`/`get events`, not only by reading .status. This // RestoreJob's contract is the DATA restore (healthy Cluster + bootstrap // cleared); the ALTER ROLE convergence belongs to the chart's post-upgrade // init-job, and gating this terminal write on that Helm-hook Job would couple // two controllers and wedge the restore whenever a GitOps source re-asserts // bootstrap.enabled. Announce the pending convergence instead of blocking on // it. r.Recorder.Eventf(restoreJob, corev1.EventTypeNormal, "CredentialsConvergencePending", "Data restored and spec.bootstrap.enabled cleared on Postgres app %s/%s; application credentials converge when the chart's post-upgrade init-job runs ALTER ROLE on the next HelmRelease reconcile. This RestoreJob does not wait for it - verify with an application login.", target.Namespace, target.AppName) if err := r.Status().Update(ctx, restoreJob); err != nil { return ctrl.Result{}, err } return ctrl.Result{}, nil } // Not (yet) healthy. Now the restore deadline is the authority for "this // restore is stuck". A recovery that CAN converge does so within the // deadline and is caught by the health check above; only a genuinely wedged // restore (most often a recoveryTime past the latest archived WAL) burns the // whole window. We do NOT fail earlier off the recovery pod's FATAL: a valid // near-now target hits that same "recovery ended before ... target ... // reached" FATAL transiently - and repeatedly, on a slow archiver - before // it converges, so an earlier trip would wrongly reject a recoverable // restore. The FATAL is used only to explain a failure the deadline has // already declared. Tenants who want a fast rejection set a short // spec.options.restoreTimeoutSeconds. deadline := options.effectiveRestoreDeadline() if restoreJob.Status.StartedAt != nil && time.Since(restoreJob.Status.StartedAt.Time) > deadline { classificationForbidden := false if options.RecoveryTime != "" { unreachable, pod, forbidden, uerr := r.recoveryTargetUnreachable(ctx, target.Namespace, clusterName) if uerr != nil { return ctrl.Result{}, uerr } classificationForbidden = forbidden if unreachable { msg := fmt.Sprintf( "point-in-time recovery target %s is past the latest WAL archived to object storage: over the %s restore window recovery kept replaying all available WAL without reaching it and PostgreSQL gave up (%q; most recent recovery pod: %s). "+ "Pick a recoveryTime inside the recoverable window - see the PITR docs on discovering the earliest/latest restorable time.", options.RecoveryTime, deadline, cnpgRecoveryTargetUnreachableLog, pod) apimeta.SetStatusCondition(&restoreJob.Status.Conditions, metav1.Condition{ Type: restoreCondRecoveryConverged, Status: metav1.ConditionFalse, Reason: "RecoveryTargetUnreachable", Message: msg, }) return r.markRestoreJobFailedReason(ctx, restoreJob, "RecoveryTargetUnreachable", msg) } } msg := fmt.Sprintf( "RestoreJob exceeded %s deadline before target Cluster reached a healthy state (override via spec.options.restoreTimeoutSeconds)", deadline) if options.RecoveryTime != "" { msg += fmt.Sprintf("; spec.options.recoveryTime %s may be past the recoverable window, or the source may be large enough to need a longer restoreTimeoutSeconds", options.RecoveryTime) } if classificationForbidden { // The RecoveryTargetUnreachable classification needs to read the // recovery pod's log; the controller was denied that (missing // pods/log RBAC grant), so it could not tell an unreachable target // apart from a slow one. Make the misconfiguration visible rather // than hiding it behind a bare generic timeout. msg += ". NOTE: could not read recovery pod logs to classify this failure - the controller's pods/log RBAC grant appears to be missing, so a RecoveryTargetUnreachable diagnosis was unavailable" r.Recorder.Eventf(restoreJob, corev1.EventTypeWarning, "RecoveryClassificationForbidden", "cannot read recovery pod logs (pods/log RBAC grant missing); unable to diagnose whether recoveryTime %s is past the recoverable window", options.RecoveryTime) } return r.markRestoreJobFailed(ctx, restoreJob, msg) } // Within the deadline and not yet healthy (or the recovery Cluster is not // chart-rendered yet): keep waiting. return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil } // cnpgRestoreTarget captures the resolved target for a CNPG restore. The // reconcile path treats in-place and to-copy identically (same purge, same // patch, same recovery flow — see the "in-place AND to-copy" comment above // the purge step), so the struct does not record which mode was resolved; // callers infer it from AppName != backup.Spec.ApplicationRef.Name. type cnpgRestoreTarget struct { Namespace string AppName string Kind string APIGroup string } func (r *RestoreJobReconciler) resolveCNPGRestoreTarget(restoreJob *backupsv1alpha1.RestoreJob, backup *backupsv1alpha1.Backup) cnpgRestoreTarget { t := cnpgRestoreTarget{ Namespace: backup.Namespace, AppName: backup.Spec.ApplicationRef.Name, Kind: backup.Spec.ApplicationRef.Kind, } if backup.Spec.ApplicationRef.APIGroup != nil { t.APIGroup = *backup.Spec.ApplicationRef.APIGroup } if restoreJob.Spec.TargetApplicationRef != nil { if restoreJob.Spec.TargetApplicationRef.Name != "" { t.AppName = restoreJob.Spec.TargetApplicationRef.Name } if restoreJob.Spec.TargetApplicationRef.Kind != "" { t.Kind = restoreJob.Spec.TargetApplicationRef.Kind } if restoreJob.Spec.TargetApplicationRef.APIGroup != nil { t.APIGroup = *restoreJob.Spec.TargetApplicationRef.APIGroup } } return t } // patchPostgresAppForRestore writes the restore-related fields into the // target Postgres app instance spec. The chart already exposes these knobs; // once the HelmRelease re-renders, the cnpg.io Cluster picks up // bootstrap.recovery and externalClusters[].plugin referencing the // chart-rendered recovery ObjectStore. // // Credentials are forwarded as a Secret reference (spec.backup.s3CredentialsSecret), // not as cleartext keys. The controller never reads the Secret itself; the // chart wires the named Secret into the recovery ObjectStore's // spec.configuration.s3Credentials. // This keeps S3 access keys out of the Postgres CR .spec, etcd, audit logs, // and any tenant-readable copies. // // Mirrors source's spec.databases and spec.users so the chart's post-install // init-job (which DROPs any database/role not in chart values) doesn't // nuke the restored data right after recovery. Source keys are overlaid on // top of target keys: the restored cluster's data wins, but target-only // databases/users (added intentionally on the target app) are preserved. func (r *RestoreJobReconciler) patchPostgresAppForRestore( ctx context.Context, app *postgresapp.Postgres, sourceServerName, newServerName, sourceDestinationPath, sourceEndpointURL, recoveryTime, backupID string, credsRef *strategyv1alpha1.S3CredentialsTemplate, caRef *strategyv1alpha1.EndpointCARef, sourceDatabases map[string]postgresapp.Database, sourceUsers map[string]postgresapp.User, ) error { patched := buildPostgresAppRestorePatch(app, sourceServerName, newServerName, sourceDestinationPath, sourceEndpointURL, recoveryTime, backupID, credsRef, caRef, sourceDatabases, sourceUsers) return r.Patch(ctx, patched, client.MergeFrom(app), client.FieldOwner(cnpgFieldManager)) } // disablePostgresAppBootstrap clears spec.bootstrap.enabled on the target // Postgres app once recovery has converged. patchPostgresAppForRestore set it // true to make the chart render bootstrap.recovery; nothing else ever turns it // back off, and while it stays true the chart skips its post-upgrade init-job // (see init-job.yaml). That init-job is the only actor that reconciles the // chart-generated -credentials Secret onto the live roles via // ALTER ROLE ... WITH PASSWORD. The recovered roles carry the source's password // hashes, the freshly generated Secret does not match them, so until the flag // flips every application-user login against the restored target fails while // the Secret advertises a password that was never applied. Flipping it lets the // init-job re-converge the roles onto the Secret on the next HelmRelease // upgrade. Idempotent: a no-op once already disabled, and re-GETs the live app // so a concurrent tenant edit is merged rather than clobbered. func (r *RestoreJobReconciler) disablePostgresAppBootstrap(ctx context.Context, namespace, appName string) error { app, err := r.getPostgresApp(ctx, namespace, appName) if err != nil { return err } if !app.Spec.Bootstrap.Enabled { return nil } patched := app.DeepCopy() patched.Spec.Bootstrap.Enabled = false return r.Patch(ctx, patched, client.MergeFrom(app), client.FieldOwner(cnpgFieldManager)) } // restoredServerName returns the WAL-archive serverName a restored cluster // writes to. Keyed by the RestoreJob UID, it is unique per restore and never // equals a source serverName, so the re-bootstrapped cluster archives into an // empty prefix and barman-cloud-check-wal-archive passes even on an in-place // restore (where the target cluster name equals the source's). func restoredServerName(clusterName string, uid types.UID) string { // 64 bits of the digest: enough that two RestoreJob UIDs colliding on the // same cluster is not a practical concern, while keeping the prefix short. sum := sha256.Sum256([]byte(uid)) return fmt.Sprintf("%s-restore-%s", clusterName, hex.EncodeToString(sum[:])[:16]) } // buildPostgresAppRestorePatch returns a deep-copied Postgres application // instance with restore-related fields overlaid. Pure: no cluster I/O. The // wrapper above invokes the Kubernetes Patch. // // Stale state hygiene: app.DeepCopy() carries any prior-restore values // (recoveryTime from a different PITR, an old s3CredentialsSecret, an // inline s3AccessKey/s3SecretKey, an outdated databases/users map). Each of // those is cleared before the new values are applied so a re-restore into // the same target does not replay or leak old configuration. Databases // and users are authoritative-from-source: the recovered cluster carries // the source's catalog and roles, so the chart's init-job (which DROPs // anything not in spec) must see the source's exact map. func buildPostgresAppRestorePatch( app *postgresapp.Postgres, sourceServerName, newServerName, sourceDestinationPath, sourceEndpointURL, recoveryTime, backupID string, credsRef *strategyv1alpha1.S3CredentialsTemplate, caRef *strategyv1alpha1.EndpointCARef, sourceDatabases map[string]postgresapp.Database, sourceUsers map[string]postgresapp.User, ) *postgresapp.Postgres { patched := app.DeepCopy() patched.Spec.Bootstrap.Enabled = true patched.Spec.Bootstrap.OldName = sourceServerName patched.Spec.Bootstrap.ServerName = sourceServerName // Archive under a fresh prefix distinct from the recovery source, or an // in-place restore fails barman-cloud-check-wal-archive ("Expected empty archive"). patched.Spec.Bootstrap.NewServerName = newServerName patched.Spec.Bootstrap.RecoveryTime = recoveryTime patched.Spec.Bootstrap.BackupID = backupID patched.Spec.Backup.DestinationPath = sourceDestinationPath patched.Spec.Backup.EndpointURL = sourceEndpointURL // The restore populates explicit S3 coordinates (destinationPath / // s3CredentialsSecret / endpointCA below), so the app is no longer on the // system-bucket path. Clear useSystemBucket: the chart rejects // bootstrap.enabled together with useSystemBucket (an unset-coordinates // guard), and a system-bucket app restored in place would otherwise wedge // its HelmRelease in that render error instead of re-bootstrapping. patched.Spec.Backup.UseSystemBucket = false // Switching to s3CredentialsSecret means inline keys must not survive // on the CR .spec; otherwise tenants who switch credential modes leave // cleartext keys behind in etcd and audit logs. patched.Spec.Backup.S3AccessKey = "" patched.Spec.Backup.S3SecretKey = "" patched.Spec.Backup.S3CredentialsSecret = postgresapp.S3CredentialsSecret{} if credsRef != nil && credsRef.SecretRef.Name != "" { patched.Spec.Backup.S3CredentialsSecret = postgresapp.S3CredentialsSecret{ Name: credsRef.SecretRef.Name, AccessKeyIDKey: credsRef.AccessKeyIDKey, SecretAccessKeyKey: credsRef.SecretAccessKeyKey, } } // endpointCA flows into both the chart's backup ObjectStore // (spec.configuration.endpointCA) and the recovery ObjectStore referenced // from externalClusters[].plugin. The recovery path specifically needs it - // without a trusted CA the cnpg-instance-manager panics in // InitInfo.loadBackup when it can't verify the seaweedfs cert. patched.Spec.Backup.EndpointCA = postgresapp.EndpointCA{} if caRef != nil && caRef.SecretRef.Name != "" { patched.Spec.Backup.EndpointCA = postgresapp.EndpointCA{ Name: caRef.SecretRef.Name, Key: caRef.Key, } } // Replace, do not merge: the recovered cluster's data and role catalog // match source's spec exactly. Merging would let stale entries from a // prior restore (or operator-added drift) ride along and the chart's // init-job would either re-create those roles against source's data or // fail to drop ones the source doesn't have. patched.Spec.Databases = sourceDatabases patched.Spec.Users = sourceUsers return patched } // cnpgBackupSnapshot is the postgresql.cnpg.io-specific payload persisted in // Backup.status.underlyingResources at backup time. It snapshots the source // app's spec.databases and spec.users so a future RestoreJob can mirror them // onto the target without re-reading the live source app (which may have // been deleted by the time the restore fires), and the BackupClassStrategy // parameters that were in effect when the strategy template rendered. The // strategy template's CNPGSpec documents `.Parameters` as a supported input // alongside `.Application`; without persisting those values on the artifact // the restore reconciler would re-render with parameters=nil and any // strategy that used `{{ .Parameters.foo }}` for a secret reference, key // name, endpointCA, etc. would render `` and produce a broken // patch on the target Postgres app. // // Note that .Parameters carries tenant-supplied configuration knobs (region // names, prefixes, secret references), never cleartext credentials - // credentials are always referenced through `secretRef`, not inlined, and // the Secret content stays in etcd's encryption-at-rest store. // // Flavor records the source's image family. A physical backup only restores // safely onto the same family: the two ship different glibc versions, which // invalidates collatable indexes, and postgis objects need the postgis // libraries. type cnpgBackupSnapshot struct { Kind string `json:"kind"` APIVersion string `json:"apiVersion"` Databases map[string]postgresapp.Database `json:"databases,omitempty"` Users map[string]postgresapp.User `json:"users,omitempty"` Parameters map[string]string `json:"parameters,omitempty"` Flavor string `json:"flavor,omitempty"` } const cnpgBackupSnapshotKind = "CNPGBackupSnapshot" // cnpgBackupSnapshotAPIVersion is the apiVersion stamped onto the snapshot. // Borrows the Cozystack backups group so the field is self-typed within the // existing API surface. var cnpgBackupSnapshotAPIVersion = backupsv1alpha1.GroupVersion.String() // marshalCNPGBackupSnapshot serializes the source app's spec.databases / // spec.users plus the BackupClassStrategy parameters into a // runtime.RawExtension suitable for Backup.Status.UnderlyingResources. // flavor is the live Cluster's, not app.Spec.Flavor: the chart refuses a // flavor change, but the app keeps the refused value. func marshalCNPGBackupSnapshot(app *postgresapp.Postgres, parameters map[string]string, flavor string) (*runtime.RawExtension, error) { if app == nil { return nil, nil } snap := cnpgBackupSnapshot{ Kind: cnpgBackupSnapshotKind, APIVersion: cnpgBackupSnapshotAPIVersion, Databases: app.Spec.Databases, Users: app.Spec.Users, Parameters: parameters, Flavor: flavor, } raw, err := json.Marshal(snap) if err != nil { return nil, err } return &runtime.RawExtension{Raw: raw}, nil } // unmarshalCNPGBackupSnapshot reads the snapshot persisted at backup time // from Backup.status.underlyingResources. Returns an error when the field // is missing or carries the wrong Kind/APIVersion - in either case the // caller must fail the restore rather than silently disable database/user // mirroring (which would let the chart's init-job DROP recovered roles). // // Parameters can legitimately be empty (a strategy that uses no // `{{ .Parameters.foo }}` references stores nil here, and a Backup taken // by a controller version that pre-dates parameter persistence has no // parameters block at all). The caller must treat a nil/empty map the // same as "no parameter overrides" rather than failing. Flavor is empty on // a Backup taken before flavors existed, when every cluster ran postgresql; // compare it through cnpgFlavor. func unmarshalCNPGBackupSnapshot(backup *backupsv1alpha1.Backup) (*cnpgBackupSnapshot, error) { if backup == nil || backup.Status.UnderlyingResources == nil || len(backup.Status.UnderlyingResources.Raw) == 0 { return nil, fmt.Errorf("status.underlyingResources is empty") } var snap cnpgBackupSnapshot if err := json.Unmarshal(backup.Status.UnderlyingResources.Raw, &snap); err != nil { return nil, fmt.Errorf("decode snapshot: %w", err) } if snap.Kind != cnpgBackupSnapshotKind { return nil, fmt.Errorf("unexpected snapshot kind %q (want %q)", snap.Kind, cnpgBackupSnapshotKind) } return &snap, nil } // cnpgFlavor resolves an unset flavor to the chart default. func cnpgFlavor(flavor string) string { if flavor == "" { return "postgresql" } return flavor } // cnpgImageFlavor names the flavor of a Cluster's spec.imageName the way the // chart's postgres.flavorGuard compares images: the last path segment without // tag or digest, so a registry mirror rewriting the host still matches. The // chart names each flavor after its image; an empty imageName is CNPG's // default postgresql image. func cnpgImageFlavor(imageName string) string { name := imageName[strings.LastIndex(imageName, "/")+1:] if i := strings.IndexAny(name, ":@"); i >= 0 { name = name[:i] } return cnpgFlavor(name) } // purgeExistingCluster deletes the live cnpg.io Cluster and its PVCs so the // chart-rendered Cluster can re-bootstrap from S3. Used only by the in-place // restore variant. func (r *RestoreJobReconciler) purgeExistingCluster(ctx context.Context, namespace, clusterName string) error { logger := getLogger(ctx) cluster := &cnpgtypes.Cluster{ ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: clusterName}, } if err := r.Delete(ctx, cluster); err != nil && !apierrors.IsNotFound(err) { return fmt.Errorf("delete cnpg Cluster %s/%s: %w", namespace, clusterName, err) } logger.Debug("deleted cnpg Cluster (or already absent)", "namespace", namespace, "name", clusterName) pvcList := &corev1.PersistentVolumeClaimList{} if err := r.List(ctx, pvcList, client.InNamespace(namespace), client.MatchingLabels{cnpgClusterLabel: clusterName}, ); err != nil { return fmt.Errorf("list PVCs for cluster %s/%s: %w", namespace, clusterName, err) } for i := range pvcList.Items { pvc := &pvcList.Items[i] if err := r.Delete(ctx, pvc); err != nil && !apierrors.IsNotFound(err) { return fmt.Errorf("delete PVC %s/%s: %w", pvc.Namespace, pvc.Name, err) } } logger.Debug("deleted cnpg cluster PVCs", "count", len(pvcList.Items)) return nil } // setCNPGRestoreHRSuspended toggles spec.suspend on the target Postgres // app's HelmRelease. The destructive purge flow needs the HR suspended // before the controller patches Postgres app spec - otherwise // helm-controller observes the change while the live Cluster still has // bootstrap.initdb, attempts an upgrade that CNPG admission rejects with // "Only one bootstrap method can be specified at a time", and drops into // an UpgradeFailed/RollbackFailed loop. Resume only after the live // Cluster + PVCs are fully gone so the next chart render lands // bootstrap.recovery on a clean slate. Idempotent. func (r *RestoreJobReconciler) setCNPGRestoreHRSuspended(ctx context.Context, namespace, name string, suspend bool) error { hr, err := r.Resource(helmReleaseGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{}) if err != nil { if apierrors.IsNotFound(err) { return nil } return fmt.Errorf("get HelmRelease %s/%s: %w", namespace, name, err) } currentSuspend, _, _ := unstructured.NestedBool(hr.Object, "spec", "suspend") if currentSuspend == suspend { return nil } if err := unstructured.SetNestedField(hr.Object, suspend, "spec", "suspend"); err != nil { return fmt.Errorf("set spec.suspend on HelmRelease %s/%s: %w", namespace, name, err) } if _, err := r.Resource(helmReleaseGVR).Namespace(namespace).Update(ctx, hr, metav1.UpdateOptions{}); err != nil { return fmt.Errorf("update HelmRelease %s/%s: %w", namespace, name, err) } return nil } // cnpgClusterFullyGone returns true when the named cnpg.io/Cluster and // every PVC labelled with that cluster have actually disappeared from the // API. The destructive purge issues r.Delete calls that mark // DeletionTimestamp but rely on CNPG's finalizers + the storage backend // to drain - returning before that drain finishes lets the next helm // reconcile race the still-terminating Cluster, and we end up right back // in the "Only one bootstrap" loop the suspend was meant to break. func (r *RestoreJobReconciler) cnpgClusterFullyGone(ctx context.Context, namespace, clusterName string) (bool, error) { cluster := &cnpgtypes.Cluster{} switch err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: clusterName}, cluster); { case err == nil: return false, nil case !apierrors.IsNotFound(err): return false, fmt.Errorf("get cnpg Cluster %s/%s: %w", namespace, clusterName, err) } pvcList := &corev1.PersistentVolumeClaimList{} if err := r.List(ctx, pvcList, client.InNamespace(namespace), client.MatchingLabels{cnpgClusterLabel: clusterName}, ); err != nil { return false, fmt.Errorf("list PVCs for cluster %s/%s: %w", namespace, clusterName, err) } return len(pvcList.Items) == 0, nil } // cnpgBackupWALArchived returns true once the underlying cnpg.io/Backup is // in a state that guarantees the closing WAL of the backup has been shipped // to object storage - i.e. the backup is restorable. CNPG's barman-cloud- // backup driver only flips Backup.status.phase to "completed" after it has // confirmed the post-stop WAL is archived, and only writes endWal when the // upload finished, so checking those two fields together is the most direct // signal we can get without re-implementing the archive verification ourselves. // // Why not Cluster.status.lastArchivedWAL: that field is not present on // CNPG 1.27.x cluster status (the available archive-related fields are // firstRecoverabilityPoint, firstRecoverabilityPointByMethod, // lastSuccessfulBackup{,ByMethod}, lastFailedBackup). An earlier version // of this gate read lastArchivedWAL and silently treated the missing field // as the empty string, so the gate never cleared even when archive was // healthy and the backup had completed. The Backup-CR signal is the same // invariant CNPG itself relies on internally and matches what tenants see // in `kubectl get backups.postgresql.cnpg.io` output. // // Conservative on missing data: // - underlying cnpg.io/Backup gone (e.g., retention policy ran): treat // as ready and let recovery surface a clearer error if the WALs are // actually missing - retention drops backups whose WAL chain is no // longer needed, so refusing to proceed would deadlock the restore. // - cnpg.io/Backup carries an empty EndWal: treat as not-ready - we // don't have a fix point to compare against, and recovery without a // stable end WAL is unsafe. // - cnpg.io/Backup has phase != "completed" (still running, mid-retry, // or stuck in failed): treat as not-ready - the closing WAL may not // be archived yet. func (r *RestoreJobReconciler) cnpgBackupWALArchived(ctx context.Context, backup *backupsv1alpha1.Backup) (bool, string, error) { backupName := backup.Spec.DriverMetadata[cnpgBackupNameKey] if backupName == "" { return false, "Backup driverMetadata is missing cnpg.io/backup-name", nil } cnpgBackup := &cnpgtypes.Backup{} if err := r.Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backupName}, cnpgBackup); err != nil { if apierrors.IsNotFound(err) { return true, fmt.Sprintf("cnpg.io/Backup %s/%s no longer exists; skipping archive gate", backup.Namespace, backupName), nil } return false, "", err } if phase := cnpgBackup.Status.Phase; phase != cnpgBackupPhaseComplete { return false, fmt.Sprintf("waiting for cnpg.io/Backup %s/%s to complete (current phase=%q)", backup.Namespace, backupName, phase), nil } if cnpgBackup.Status.EndWal == "" { return false, fmt.Sprintf("cnpg.io/Backup %s/%s has phase=completed but has not recorded status.endWal yet", backup.Namespace, backupName), nil } return true, fmt.Sprintf("cnpg.io/Backup %s/%s phase=completed, endWal=%q", backup.Namespace, backupName, cnpgBackup.Status.EndWal), nil } // cnpgSourceBackup returns the cnpg.io/Backup behind a Backup artifact, or // nil once it is gone. The driver deletes it only together with its // artifact, while plugin-barman-cloud's retention deletes it together with // the base backup it describes (deleteBackupsNotInCatalog), so an artifact // that outlives its cnpg.io/Backup names a base backup the catalog has lost. func (r *RestoreJobReconciler) cnpgSourceBackup(ctx context.Context, backup *backupsv1alpha1.Backup) (*cnpgtypes.Backup, error) { name := backup.Spec.DriverMetadata[cnpgBackupNameKey] if name == "" { return nil, nil } cnpgBackup := &cnpgtypes.Backup{} if err := r.Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: name}, cnpgBackup); err != nil { if apierrors.IsNotFound(err) { return nil, nil } return nil, err } return cnpgBackup, nil } // cnpgBackupMajorVersion returns the PostgreSQL major a Backup artifact was // taken on, or 0 when it is unknown. An artifact created before the driver // recorded it falls back to the cnpg.io/Backup, as long as retention has not // removed it. func (r *RestoreJobReconciler) cnpgBackupMajorVersion(ctx context.Context, backup *backupsv1alpha1.Backup) (int, error) { if v, err := strconv.Atoi(backup.Spec.DriverMetadata[cnpgMajorVersionKey]); err == nil && v > 0 { return v, nil } source, err := r.cnpgSourceBackup(ctx, backup) if err != nil || source == nil { return 0, err } return source.Status.MajorVersion, nil } // postgresAppMajorVersion parses a Postgres app's spec.version ("v16"), or // returns 0 when it is empty or not in that form. func postgresAppMajorVersion(version string) int { v, err := strconv.Atoi(strings.TrimPrefix(version, "v")) if err != nil || v <= 0 { return 0 } return v } // cnpgRestoreBackupID decides whether a restore pins the base backup it // starts from, and returns the backupID to set in // bootstrap.recovery.recoveryTarget, or "" to leave the choice to the // barman-cloud plugin. // // Left to itself the plugin starts from the newest backup in the catalog // when no recoveryTime is set, whichever backup the RestoreJob names, and // otherwise from the newest one ending at or before recoveryTime, on any // timeline (barman-cloud pkg/catalog FindBackupInfo). On a stream holding // several timelines either can start from another branch of the history // and bring back other data than the backup asked for. // // A source that is gone (nil) is not pinned: its base backup is no longer // in the catalog, and pinning it would fail the recovery after the target // has been purged. // // A backup cannot serve a recoveryTime before it ended: recovery would stop // before it is consistent. Such a request keeps the plugin's choice, which // can find an older backup. stoppedAt is truncated to the second while the // catalog compares barman's end time to the microsecond, so a backup is // pinned only once recoveryTime is a full second past stoppedAt. func cnpgRestoreBackupID(source *cnpgtypes.Backup, recoveryTime string) string { if source == nil || source.Status.BackupID == "" { return "" } if recoveryTime == "" { return source.Status.BackupID } stoppedAt := source.Status.StoppedAt if stoppedAt == nil || stoppedAt.IsZero() { return "" } target, err := time.Parse(time.RFC3339Nano, recoveryTime) if err != nil || target.Before(stoppedAt.Add(time.Second)) { return "" } return source.Status.BackupID } // cnpgClusterHealthy returns true once the named cnpg.io Cluster reports its // healthy phase. Treats a missing Cluster as not-yet-healthy. func (r *RestoreJobReconciler) cnpgClusterHealthy(ctx context.Context, namespace, clusterName string) (bool, error) { cluster := &cnpgtypes.Cluster{} if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: clusterName}, cluster); err != nil { if apierrors.IsNotFound(err) { return false, nil } return false, err } return cluster.Status.Phase == cnpgClusterHealthyPhase, nil } // logIndicatesRecoveryTargetUnreachable reports whether a recovery container // log contains the PostgreSQL FATAL that a recovery instance emits when it // exhausts the WAL archive before reaching the configured recovery target. // Pure string match so it is unit-testable without a live cluster. func logIndicatesRecoveryTargetUnreachable(recoveryLog string) bool { return strings.Contains(recoveryLog, cnpgRecoveryTargetUnreachableLog) } // recoveryUnreachableFromLogs reports whether any recovery-pod log carries the // unreachable-target FATAL. It is consulted only once the restore deadline has // elapsed (see reconcileCNPGRestore), so a single match is enough: recovery has // already had the whole window to converge, and this only classifies *why* it // did not. Pure so the decision is unit-testable without a live cluster. func recoveryUnreachableFromLogs(logs []string) bool { return slices.ContainsFunc(logs, logIndicatesRecoveryTargetUnreachable) } // recoveryTargetUnreachable inspects the target Cluster's bootstrap-recovery // pods and reports whether recovery has definitively failed because the // point-in-time target is past the last archived WAL. It reads the recovery // container's log and matches the PostgreSQL "recovery ended before ..." // FATAL - a signal unique to an unreachable target, so a transient recovery- // pod crash (which never emits it) does NOT trip the guard and the recovery // is left to converge or hit the restore deadline. Returns the inspected pod // name for the operator-facing message. Nil-safe: with no Clientset wired it // reports false so the deadline remains the backstop. // // Every recovery attempt for an unreachable target reproduces the same FATAL, // so scanning the most recent pods (running or Failed, newest first) reliably // catches it across reconciles without depending on any single pod surviving. func (r *RestoreJobReconciler) recoveryTargetUnreachable(ctx context.Context, namespace, clusterName string) (unreachable bool, pod string, forbidden bool, err error) { readLog := r.readPodLog if readLog == nil { readLog = r.readPodContainerLog } // No way to read logs (no Clientset wired and no injected reader): report // not-unreachable so the deadline stays the backstop. if r.Clientset == nil && r.readPodLog == nil { return false, "", false, nil } podList := &corev1.PodList{} if err := r.List(ctx, podList, client.InNamespace(namespace), client.MatchingLabels{cnpgClusterLabel: clusterName}, ); err != nil { return false, "", false, fmt.Errorf("list pods for cluster %s/%s: %w", namespace, clusterName, err) } logger := getLogger(ctx) logs := make([]string, 0, cnpgRecoveryMaxInspectPods) fatalPod := "" // newest FATAL-ing pod (pods are inspected newest-first) forbiddenSeen := false for _, p := range recoveryPodsToInspect(podList.Items) { recoveryLog, readErr := readLog(ctx, p.Namespace, p.Name, cnpgRecoveryContainerName) if readErr != nil { // A log read can fail transiently (pod being GC'd, apiserver // blip); don't fail the RestoreJob on it - skip this pod. But a // Forbidden is NOT transient: it means the controller lacks the // pods/log RBAC grant, which disables the RecoveryTargetUnreachable // classification. Flag it so the caller can surface it (Warning // Event + message) instead of failing with a bare generic reason. if apierrors.IsForbidden(readErr) { forbiddenSeen = true logger.Info("cannot read recovery pod log to classify a stuck point-in-time restore - the pods/log RBAC grant is missing, so the RecoveryTargetUnreachable classification is unavailable and this restore falls back to the restore deadline", "pod", p.Name, "error", readErr) } else { logger.Debug("skipping recovery pod log read", "pod", p.Name, "error", readErr) } continue } logs = append(logs, recoveryLog) if fatalPod == "" && logIndicatesRecoveryTargetUnreachable(recoveryLog) { fatalPod = p.Name } } return recoveryUnreachableFromLogs(logs), fatalPod, forbiddenSeen, nil } // recoveryPodsToInspect narrows a Cluster's pod list to the bootstrap-recovery // pods worth reading a log from: it keeps only pods carrying the full-recovery // container, orders them newest-first (the current/most-recent recovery // attempt is the most likely to hold a fresh, complete replay log), and caps // the result at cnpgRecoveryMaxInspectPods to bound the per-reconcile log // reads. Pure and deterministic so the filter/order/cap contract - the part // that would silently regress the fail-fast into a deadline hang if it drifted // - is unit-testable without a live cluster. func recoveryPodsToInspect(pods []corev1.Pod) []*corev1.Pod { out := make([]*corev1.Pod, 0, len(pods)) for i := range pods { p := &pods[i] for _, c := range p.Spec.Containers { if c.Name == cnpgRecoveryContainerName { out = append(out, p) break } } } sort.Slice(out, func(i, j int) bool { return out[i].CreationTimestamp.After(out[j].CreationTimestamp.Time) }) if len(out) > cnpgRecoveryMaxInspectPods { out = out[:cnpgRecoveryMaxInspectPods] } return out } // readPodContainerLog returns the tail of a pod container's log via the // clientset (the controller-runtime cache client cannot read the log // subresource). func (r *RestoreJobReconciler) readPodContainerLog(ctx context.Context, namespace, podName, container string) (string, error) { tail := int64(cnpgRecoveryLogTailLines) req := r.Clientset.CoreV1().Pods(namespace).GetLogs(podName, &corev1.PodLogOptions{ Container: container, TailLines: &tail, }) stream, err := req.Stream(ctx) if err != nil { return "", err } defer func() { _ = stream.Close() }() // read-only log stream; nothing depends on the close error data, err := io.ReadAll(stream) if err != nil { return "", err } return string(data), nil } // recoveryBootstrapClusterState fetches the live cnpg.io Cluster and reports // whether its spec.bootstrap.recovery is populated - the signal that the chart // has re-rendered with our restore-shaped values and the operator is using the // recovery bootstrap path - together with the Cluster's creationTimestamp. The // caller pairs the timestamp with the RestoreJob's StartedAt (via // cnpgClusterFreshlyRecovered) to tell a Cluster this restore just produced // apart from one left over by an earlier completed restore. Treats a missing // Cluster as "not yet" rather than an error so the caller can keep polling // while HelmRelease catches up. // // A Cluster carrying DeletionTimestamp is also treated as "not yet": that is // the in-flight purge case, where r.Delete has fired but cnpg.io's // finalizers have not yet drained. If we returned true here, the caller // would skip the next purge step and start waiting for "healthy" against a // CR that is about to disappear; if we returned false (would re-purge) // without the DeletionTimestamp guard, an SSA Apply from Helm could // race the original delete and merge bootstrap.recovery into the still- // terminating CR - cnpg-operator's bootstrap-immutability check then // drops the change and the cluster ends up with the original initdb spec. // Holding here forces the caller to requeue until the old CR is fully GC'd // and the chart re-creates a fresh one. func (r *RestoreJobReconciler) recoveryBootstrapClusterState(ctx context.Context, namespace, clusterName string) (recoveryClusterState, error) { cluster := &cnpgtypes.Cluster{} if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: clusterName}, cluster); err != nil { if apierrors.IsNotFound(err) { return recoveryClusterState{}, nil } return recoveryClusterState{}, err } if !cluster.DeletionTimestamp.IsZero() { return recoveryClusterState{}, nil } if cluster.Spec.Bootstrap == nil || cluster.Spec.Bootstrap.Recovery == nil { return recoveryClusterState{}, nil } created := cluster.CreationTimestamp return recoveryClusterState{ hasRecovery: true, createdAt: &created, restoreMark: cluster.Annotations[cnpgRestoredServerNameAnnotation], }, nil } // recoveryClusterState is what the purge decision reads off the live Cluster. // The zero value is a Cluster that is absent, terminating, or not bootstrapped // from a recovery. type recoveryClusterState struct { hasRecovery bool createdAt *metav1.Time // restoreMark is cnpgRestoredServerNameAnnotation, or "" on a Cluster the // chart rendered before it set the annotation. restoreMark string } // --------------------------------------------------------------------------- // Helpers shared between Backup and Restore // --------------------------------------------------------------------------- // renderCNPGTemplate templates the strategy template against a context // containing the live application object and the BackupClass parameters. // Reuses the same templating helper as the Velero strategy. The Application // is exposed as JSON-tagged map so user templates keep working with paths // like {{ .Application.metadata.name }}. func renderCNPGTemplate(t strategyv1alpha1.CNPGTemplate, app *postgresapp.Postgres, parameters map[string]string) (*strategyv1alpha1.CNPGTemplate, error) { appAsMap, err := toJSONMap(app) if err != nil { return nil, fmt.Errorf("encode application for templating: %w", err) } templateContext := map[string]any{ "Application": appAsMap, "Parameters": parameters, } return template.Template(&t, templateContext) } // toJSONMap converts a typed object to a generic map via JSON tags. Used // so user-authored go-templates continue to address fields by their JSON // names (e.g. .Application.metadata.name) without leaking the Go struct // hierarchy to user-facing strategy templates. func toJSONMap(obj any) (map[string]any, error) { raw, err := json.Marshal(obj) if err != nil { return nil, err } out := map[string]any{} if err := json.Unmarshal(raw, &out); err != nil { return nil, err } return out, nil } // getPostgresApp fetches the apps.cozystack.io Postgres instance via the // shared typed client. The Postgres scheme is registered in main.go so the // controller-runtime cache serves it directly. func (r *BackupJobReconciler) getPostgresApp(ctx context.Context, namespace, name string) (*postgresapp.Postgres, error) { app := &postgresapp.Postgres{} if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, app); err != nil { return nil, err } return app, nil } func (r *RestoreJobReconciler) getPostgresApp(ctx context.Context, namespace, name string) (*postgresapp.Postgres, error) { app := &postgresapp.Postgres{} if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, app); err != nil { return nil, err } return app, nil } // newCNPGClusterPatch returns an empty typed Cluster object addressed by // (namespace, name), ready to receive spec mutations. TypeMeta is set so // the SSA Apply path can identify the kind without consulting the scheme. func newCNPGClusterPatch(namespace, name string) *cnpgtypes.Cluster { return &cnpgtypes.Cluster{ TypeMeta: metav1.TypeMeta{ APIVersion: cnpgtypes.GroupVersion.String(), Kind: "Cluster", }, ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, } } // newObjectStorePatch returns an empty typed ObjectStore addressed by // (namespace, name), TypeMeta set so the SSA Apply path can identify the kind. func newObjectStorePatch(namespace, name string) *cnpgtypes.ObjectStore { return &cnpgtypes.ObjectStore{ TypeMeta: metav1.TypeMeta{ APIVersion: cnpgtypes.BarmanGroupVersion.String(), Kind: "ObjectStore", }, ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, } } // buildBarmanPlugin returns the Cluster spec.plugins entry that routes // backup/WAL/recovery through the barman-cloud plugin, referencing the named // ObjectStore and (optionally) the per-server folder. func buildBarmanPlugin(objectStoreName, serverName string) cnpgtypes.PluginConfiguration { isWALArchiver := true params := map[string]string{barmanObjectNameParam: objectStoreName} if serverName != "" { params[barmanServerNameParam] = serverName } return cnpgtypes.PluginConfiguration{ Name: cnpgtypes.PluginName, IsWALArchiver: &isWALArchiver, Parameters: params, } } // barmanSidecarConfiguration carries the two settings the barman-cloud sidecar // cannot get right on its own. // // Resources: the plugin injects the sidecar with no resources. This ObjectStore // lands in the application's namespace, and a tenant with resourceQuotas set // carries a LimitRange (packages/apps/tenant/templates/quota.yaml) that defaults // every container to 128Mi; the sidecar is OOMKilled mid-backup under it, // leaving the ObjectStore healthy and the backup failed. Measured cgroup // high-water mark on a 38 MB database is 254 MiB, so 128Mi cannot hold it. A // tenant that leaves resourceQuotas empty has no LimitRange, and there the // sidecar had no requests and no limit at all. 256Mi holds the measured working // set and 1Gi is four times the measurement. No CPU limit is set, matching the // entityOperator precedent in packages/apps/kafka: a throttled sidecar stalls // WAL archiving instead of failing it. // // Checksum: since botocore ~1.36 the default (when_supported) attaches a // flexible checksum to every PutObject, which non-AWS S3-compatible backends // (Ceph RGW, and the platform's own default SeaweedFS system bucket) reject // with "x-amz-content-sha256 must be UNSIGNED-PAYLOAD, ...". Compute a checksum // only when required; AWS S3 accepts that too, so it is a safe default // everywhere. // // This mirrors the chart-rendered ObjectStores, which take both fields from // cozy-lib.barman.sidecarConfiguration (packages/{apps/postgres, // system/keycloak}/templates/db.yaml), and the etcd-operator fix (#342). func barmanSidecarConfiguration() *cnpgtypes.InstanceSidecarConfiguration { return &cnpgtypes.InstanceSidecarConfiguration{ Resources: corev1.ResourceRequirements{ Requests: corev1.ResourceList{ corev1.ResourceCPU: resource.MustParse("100m"), corev1.ResourceMemory: resource.MustParse("256Mi"), }, Limits: corev1.ResourceList{ corev1.ResourceMemory: resource.MustParse("1Gi"), }, }, Env: []cnpgtypes.EnvVar{{ Name: "AWS_REQUEST_CHECKSUM_CALCULATION", Value: "when_required", }}, } } // buildBarmanObjectStore translates the typed strategy template into the // barman configuration shape shared by the deprecated // spec.backup.barmanObjectStore and the plugin's ObjectStore.spec.configuration. // Pass serverName="" when building an ObjectStore (the plugin forbids // serverName there and takes it from the Cluster plugin parameter). func buildBarmanObjectStore(t strategyv1alpha1.BarmanObjectStoreTemplate, serverName string) *cnpgtypes.BarmanObjectStoreConfiguration { out := &cnpgtypes.BarmanObjectStoreConfiguration{ DestinationPath: t.DestinationPath, ServerName: serverName, EndpointURL: t.EndpointURL, } if t.EndpointCA != nil && t.EndpointCA.SecretRef.Name != "" { caKey := t.EndpointCA.Key if caKey == "" { caKey = defaultEndpointCAKey } out.EndpointCA = &cnpgtypes.SecretKeySelector{ Name: t.EndpointCA.SecretRef.Name, Key: caKey, } } if t.S3Credentials != nil { accessKeyKey := t.S3Credentials.AccessKeyIDKey if accessKeyKey == "" { accessKeyKey = defaultS3AccessKeyIDKey } secretKeyKey := t.S3Credentials.SecretAccessKeyKey if secretKeyKey == "" { secretKeyKey = defaultS3SecretAccessKeyKey } out.S3Credentials = &cnpgtypes.S3Credentials{ AccessKeyID: &cnpgtypes.SecretKeySelector{ Name: t.S3Credentials.SecretRef.Name, Key: accessKeyKey, }, SecretAccessKey: &cnpgtypes.SecretKeySelector{ Name: t.S3Credentials.SecretRef.Name, Key: secretKeyKey, }, } } if t.Wal != nil && (t.Wal.Compression != "" || t.Wal.Encryption != "") { out.Wal = &cnpgtypes.WalBackupConfiguration{ Compression: t.Wal.Compression, Encryption: t.Wal.Encryption, } } if t.Data != nil && (t.Data.Compression != "" || t.Data.Encryption != "" || t.Data.Jobs != nil) { out.Data = &cnpgtypes.DataBackupConfiguration{ Compression: t.Data.Compression, Encryption: t.Data.Encryption, Jobs: t.Data.Jobs, } } return out } // cnpgBackupPhase extracts the lowercase phase + message from a // postgresql.cnpg.io/Backup status block. func cnpgBackupPhase(b *cnpgtypes.Backup) (string, string) { if b == nil { return "", "" } return b.Status.Phase, b.Status.Error } // CNPGRestoreOptions is the typed shape of RestoreJob.Spec.Options for the // CNPG driver. Mirrors the Velero strategy's RestoreOptions pattern - one // shared opaque blob, parsed lazily at the dispatch boundary. type CNPGRestoreOptions struct { // RecoveryTime is an optional RFC3339 timestamp the chart maps onto // CNPG's bootstrap.recovery.recoveryTarget.targetTime. Empty means // recover to the end of the latest WAL in the archive (chart default). // +optional RecoveryTime string `json:"recoveryTime,omitempty"` // RestoreTimeoutSeconds caps the time the driver waits for the target // Cluster to reach the healthy phase before it marks the RestoreJob // Failed. Zero or unset falls back to cnpgDefaultRestoreDeadline. // +optional RestoreTimeoutSeconds int64 `json:"restoreTimeoutSeconds,omitempty"` // WALArchiveTimeoutSeconds caps the time the driver waits for the // source Cluster's lastArchivedWAL to catch up to the backup's endWal // before it marks the RestoreJob Failed. Zero or unset falls back to // cnpgWALArchiveDeadline. Independent from RestoreTimeoutSeconds so // big-DB restores keep their full recovery budget while a stuck // archive_command surfaces fast - the gate clears in seconds for any // healthy cluster regardless of DB size, so a long timeout here is // almost always wrong. // +optional WALArchiveTimeoutSeconds int64 `json:"walArchiveTimeoutSeconds,omitempty"` } // parseCNPGRestoreOptions decodes RestoreJob.Spec.Options into the typed // shape. Returns the zero value plus a parse error when the blob is malformed // so the caller can surface it. Callers keep behaviour permissive against // future field additions by ignoring the error and proceeding with the zero // value, but they have the option to log it for tenant-debuggability. func parseCNPGRestoreOptions(opts *runtime.RawExtension) (CNPGRestoreOptions, error) { var out CNPGRestoreOptions if opts == nil || len(opts.Raw) == 0 { return out, nil } if err := json.Unmarshal(opts.Raw, &out); err != nil { return CNPGRestoreOptions{}, fmt.Errorf("decode restoreJob.spec.options: %w", err) } return out, nil } // effectiveRestoreDeadline returns the configured deadline, applying the // driver's default when the option is unset or non-positive. func (o CNPGRestoreOptions) effectiveRestoreDeadline() time.Duration { if o.RestoreTimeoutSeconds > 0 { return time.Duration(o.RestoreTimeoutSeconds) * time.Second } return cnpgDefaultRestoreDeadline } // effectiveBootstrapDisableGrace returns the fixed window the post-convergence // bootstrap-disable step keeps requeueing over a transient error before it // terminates the restore Failed. It is independent of restoreTimeoutSeconds: // that knob bounds the recovery wait, which is unrelated to how long a // control-plane blip takes to clear (see cnpgBootstrapDisableGrace). func (o CNPGRestoreOptions) effectiveBootstrapDisableGrace() time.Duration { return cnpgBootstrapDisableGrace } // effectiveWALArchiveDeadline returns the configured WAL-archive gate // deadline, applying the driver's default when the option is unset or // non-positive. func (o CNPGRestoreOptions) effectiveWALArchiveDeadline() time.Duration { if o.WALArchiveTimeoutSeconds > 0 { return time.Duration(o.WALArchiveTimeoutSeconds) * time.Second } return cnpgWALArchiveDeadline }