mirror of
https://github.com/cozystack/cozystack.git
synced 2026-10-30 00:27:04 +09:00
Aleksei Sviridkin
e983911bae
The tenant chart now renders its ResourceQuota and the default LimitRange only for keys other than the gpu.cozystack.io/* GPU budgets, so a tenant with only GPU budgets has no LimitRange. The comments and the system memory limits page that explain sidecar sizing and system defaults by that LimitRange said any declared resourceQuotas brings it. Assisted-by: LLM Signed-off-by: Aleksei Sviridkin <f@lex.la>
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 non-GPU resourceQuotas
|
|
// 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 without non-GPU resourceQuotas 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
|
|
}
|