mirror of
https://github.com/cozystack/cozystack.git
synced 2026-10-27 15:17:13 +09:00
Andrey Kolkov
6a75883013
A recovery Cluster without the restored-server-name annotation fell back to the creation-time comparison. On the current chart the annotation is absent when bootstrap.newServerName is empty, i.e. a manually configured recovery: re-rendered during the WAL-archive wait it is created after StartedAt, the purge was skipped and the restore could report success on the manual recovery source. Use the creation-time fallback only when the application's bootstrap.newServerName already carries this restore's server name, which keeps it for Clusters rendered by a chart that predates the annotation. Assisted-By: LLM Signed-off-by: Andrey Kolkov <androndo@gmail.com>
2389 lines
111 KiB
Go
2389 lines
111 KiB
Go
// 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 (<cluster>-<n>-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://.../<serverName>/ 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 `<no value>` 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
|
|
// <target>-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 <release>-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 `<no value>` 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
|
|
}
|