Files
cozystack/internal/backupcontroller/cnpgstrategy_controller_test.go
Aleksei Sviridkin f203a47afe fix(backups): hold a Postgres backup until the plugin rollout is over (#4674)
<!-- Thank you for making a contribution! Here are some tips for you:
- Use Conventional Commits for the PR title: `type(scope): description`
- Types: feat, fix, docs, style, refactor, perf, test, build, ci, chore
- Scopes are not an exhaustive list — pick the most specific scope for
the change and extend the list when a genuinely new area appears.
Examples:
- System components: dashboard, platform, operator, cilium, kube-ovn,
linstor, fluxcd, cluster-api
- Managed apps: postgres, mariadb, redis, kafka, clickhouse,
virtual-machine, kubernetes
- Development and maintenance: api, hack, tests, ci, docs, maintenance
- Breaking changes: append `!` after type/scope (`feat(api)!: ...`) or
add a `BREAKING CHANGE:` footer
- If it's a work in progress, consider creating this PR as a draft.
- Don't hesistate to ask for opinion and review in the community chats,
even if it's still a draft.
- Add the label `kind/backport` if it's a bugfix that needs to be
backported to a previous version.
-->

## What this PR does

The first BackupJob on a Postgres app whose `cnpg.io/Cluster` has never
carried the barman-cloud plugin — a fresh platform-flow app, or one an
in-place restore re-rendered — attaches the plugin and creates the
`cnpg.io/Backup` in the same reconcile. Attaching the plugin changes the
instance pod spec, so CNPG rolls every instance (`Pod rollout required:
volumes: element plugins has been added`). The Backup then lands on an
instance without the plugin sidecar (`requested plugin is not available:
barman-cloud.cloudnative-pg.io`) or on one CNPG restarts under it
(`instance manager was restarted during backup`). CNPG does not retry a
failed Backup, so the BackupJob stays Running in
`CNPGBackupTransientFailure` until `cnpgDefaultBackupDeadline` and
fails. The first backup of each such Postgres is always lost; this was
observed on a v1.6.3 cluster and reproduced against `main`.

The driver now holds the Backup back until that rollout is over. Before
attaching the plugin to a Cluster that has none, it records
`PluginRolledOut=False` on the BackupJob, with `Ready=False` reason
`WaitingForPluginRollout` for the user, and requeues; the next pass
attaches the plugin. The Backup is created once CNPG reports the Cluster
in its healthy phase and every instance pod (`cnpg.io/cluster`,
`cnpg.io/podRole=instance`) was created after the condition and is
Ready; the condition then turns True and `Ready` is cleared, as before.
Recording the condition first means a lost status write leaves the
plugin unattached for the next pass instead of a Cluster that looks
ready while it rolls. A Cluster that already runs the plugin is not
rolled by the driver and its Backup is created on the first pass, as
today. The wait is bounded by the existing `cnpgDefaultBackupDeadline`,
after which the BackupJob fails with the phase or pod CNPG is stuck on.

Why this signal, checked against the CNPG 1.29 source and the 1.30
Cluster CRD vendored here: the Cluster status has no
`observedGeneration` and its `Ready` condition is derived from
`status.phase`, so right after the attach the phase still reads healthy
from before until CNPG reconciles the new spec. `status.pluginStatus`
does not help either, since CNPG also lists plugins referenced only from
`externalClusters` there, which is exactly the state of a Cluster a
restore re-rendered. A pod spec change is never applied in place (only a
pending Postgres restart is), so the instances are recreated and pods
newer than the attach are the ones running the sidecar. This reads pod
metadata and CNPG's documented labels only, not the plugin's container
or volume names, and needs no RBAC change: the controller already lists
and watches pods.

Tests: `TestReconcileCNPG_WaitsForPluginRolloutBeforeBackup` drives a
Cluster without the plugin through the attach, a stale healthy phase
over the old pod, a rolling phase over the recreated pod, and the
healthy phase, and asserts no Backup exists until the last step.
`TestReconcileCNPG_AttachedPluginDoesNotDelayBackup` covers a Cluster
that already runs the plugin, and
`TestReconcileCNPG_PluginRolloutWaitIsBounded` the deadline. With the
gate disabled, the first and third fail; with only the phase checked and
not the pods, the first fails at the attach pass.

### Screenshots

<!-- REQUIRED for UI changes: attach screenshots or screen recordings
demonstrating
the visual impact of your changes. PRs with UI changes without
screenshots will not be merged. -->

Not applicable: no UI change.

### Downstream repositories

<!--
Cozystack is upstream for repositories that are NOT kept in sync with it
automatically, and nothing in CI compares the two sides, so a change
here can break them silently.

Which change forces what, and which file to touch, is the trigger map in
docs/agents/contributing.md under "Downstream Repositories". Walk it
before you tick anything.

IF YOU ARE AN AI AGENT OR AN LLM, THIS IS PART OF THE TASK, NOT A FORM
TO FILL IN:

1. Walk the trigger map against the actual diff of this PR, file by
file. Do not decide from the PR title or from your memory of what you
changed.
2. For every repository you tick, open the follow-up PR there and link
it on the same line. A ticked box with no link claims work that does not
exist, which is worse than an empty box. Search that repository for an
open PR or issue covering it first, and link that instead of filing a
duplicate.
3. When the follow-up is out of scope, or needs a decision that is not
yours to make, open an issue in that repository instead, link that, and
say so in the PR body. Do not push a speculative PR into another
repository just to fill a line here.
4. Do not tick "No downstream repository is affected" to make the
checklist go away. If you are unsure, leave every box empty and say so
in the PR body, so a human decides.
-->

The diff touches only `internal/backupcontroller` Go code: no package,
values, chart, CRD, `hack/` tooling, telemetry metric or contributor
convention. Walked against the trigger map, it reaches none of the
repositories below, and the website documents no workaround for the lost
first backup that this would make obsolete.

- [x] No downstream repository is affected by this change
- [ ] [cozystack/website](https://github.com/cozystack/website) -
follow-up:
- [ ]
[cozystack/terraform-provider-cozystack](https://github.com/cozystack/terraform-provider-cozystack)
- follow-up:
- [ ]
[cozystack/ansible-cozystack](https://github.com/cozystack/ansible-cozystack)
- follow-up:
- [ ] [cozystack/ccp](https://github.com/cozystack/ccp) - follow-up:
- [ ] [cozystack/talm](https://github.com/cozystack/talm) - follow-up:
- [ ] [cozystack/cozyhr](https://github.com/cozystack/cozyhr) -
follow-up:
- [ ] [cozystack/cozy-proxy](https://github.com/cozystack/cozy-proxy) -
follow-up:
- [ ]
[cozystack/cozystack-telemetry-server](https://github.com/cozystack/cozystack-telemetry-server)
- follow-up:
- [ ]
[cozystack/external-apps-example](https://github.com/cozystack/external-apps-example)
- follow-up:
- [ ] [cozystack/examples](https://github.com/cozystack/examples) -
follow-up:
- [ ] [cozystack/community](https://github.com/cozystack/community) -
follow-up:

### Release note

<!--  Write a release note:
- Explain what has changed internally and for users.
- Start with the same `type(scope):` prefix as in the PR title
- Follow the guidelines at
https://github.com/kubernetes/community/blob/master/contributors/guide/release-notes.md.
-->

```release-note
fix(backups): the CNPG backup driver no longer loses the first backup of a Postgres app whose Cluster had no barman-cloud plugin yet (a new platform-flow app, or one restored in place). It now waits for CNPG to finish rolling the instances onto the plugin before creating the cnpg.io Backup, and reports the wait on the BackupJob as Ready=False with reason WaitingForPluginRollout.
```


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Reliability**
* Backups wait for the backup plugin rollout to complete and cluster
instances to become healthy before starting.
* If rollout does not finish before the backup deadline, the backup job
fails instead of waiting indefinitely.
* Backups can start without waiting for a new rollout when the plugin is
already attached.
* **Backup Configuration**
* When no server name is configured, backups use the cluster name and
application UID, giving recreated applications distinct archive
prefixes.
* Existing server names are preserved when the plugin is already
attached.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
2026-10-08 13:11:46 +03:00

4220 lines
178 KiB
Go

// SPDX-License-Identifier: Apache-2.0
package backupcontroller
import (
"context"
"fmt"
"strings"
"testing"
"time"
apiequality "k8s.io/apimachinery/pkg/api/equality"
apierrors "k8s.io/apimachinery/pkg/api/errors"
apimeta "k8s.io/apimachinery/pkg/api/meta"
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"
"k8s.io/client-go/dynamic"
dynamicfake "k8s.io/client-go/dynamic/fake"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client"
clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
corev1 "k8s.io/api/core/v1"
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"
)
func TestCNPGClusterNameForApp(t *testing.T) {
got := cnpgClusterNameForApp("foo")
if got != "postgres-foo" {
t.Fatalf("cnpgClusterNameForApp(%q) = %q, want postgres-foo", "foo", got)
}
}
func TestBuildBarmanObjectStore_AllFields(t *testing.T) {
jobs := int32(4)
tmpl := strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://bucket/path/",
EndpointURL: "http://s3:9000",
RetentionPolicy: "30d",
S3Credentials: &strategyv1alpha1.S3CredentialsTemplate{
SecretRef: corev1.LocalObjectReference{Name: "creds"},
},
Wal: &strategyv1alpha1.BarmanWalTemplate{Compression: "gzip"},
Data: &strategyv1alpha1.BarmanDataTemplate{Compression: "gzip", Jobs: &jobs},
}
got := buildBarmanObjectStore(tmpl, "my-server")
// retentionPolicy lives at spec.backup.retentionPolicy in CNPG; it is
// emitted by applyClusterBarmanObjectStore and intentionally absent here.
want := &cnpgtypes.BarmanObjectStoreConfiguration{
DestinationPath: "s3://bucket/path/",
EndpointURL: "http://s3:9000",
ServerName: "my-server",
S3Credentials: &cnpgtypes.S3Credentials{
AccessKeyID: &cnpgtypes.SecretKeySelector{Name: "creds", Key: defaultS3AccessKeyIDKey},
SecretAccessKey: &cnpgtypes.SecretKeySelector{Name: "creds", Key: defaultS3SecretAccessKeyKey},
},
Wal: &cnpgtypes.WalBackupConfiguration{Compression: "gzip"},
Data: &cnpgtypes.DataBackupConfiguration{Compression: "gzip", Jobs: new(int32(4))},
}
if !apiequality.Semantic.DeepEqual(got, want) {
t.Fatalf("buildBarmanObjectStore mismatch\n got: %#v\nwant: %#v", got, want)
}
}
func TestBuildBarmanObjectStore_Minimal(t *testing.T) {
tmpl := strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://only/",
}
got := buildBarmanObjectStore(tmpl, "fallback")
want := &cnpgtypes.BarmanObjectStoreConfiguration{
DestinationPath: "s3://only/",
ServerName: "fallback",
}
if !apiequality.Semantic.DeepEqual(got, want) {
t.Fatalf("minimal buildBarmanObjectStore mismatch\n got: %#v\nwant: %#v", got, want)
}
}
func TestBuildBarmanObjectStore_CustomCredentialKeys(t *testing.T) {
tmpl := strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://x/",
S3Credentials: &strategyv1alpha1.S3CredentialsTemplate{
SecretRef: corev1.LocalObjectReference{Name: "creds"},
AccessKeyIDKey: "AKID",
SecretAccessKeyKey: "SKID",
},
}
got := buildBarmanObjectStore(tmpl, "s")
if got.S3Credentials.AccessKeyID.Key != "AKID" {
t.Fatalf("expected custom access key id key AKID, got %q", got.S3Credentials.AccessKeyID.Key)
}
if got.S3Credentials.SecretAccessKey.Key != "SKID" {
t.Fatalf("expected custom secret access key key SKID, got %q", got.S3Credentials.SecretAccessKey.Key)
}
}
// TestBuildBarmanObjectStore_EndpointCA covers the TLS-with-self-signed-CA
// path: a strategy that names a Secret holding the CA bundle gets translated
// into the matching cnpgtypes.SecretKeySelector on the live Cluster, defaulting
// the key name to "ca.crt" when not specified.
func TestBuildBarmanObjectStore_EndpointCA(t *testing.T) {
t.Run("default key", func(t *testing.T) {
tmpl := strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://b/",
EndpointCA: &strategyv1alpha1.EndpointCARef{
SecretRef: corev1.LocalObjectReference{Name: "trust-bundle"},
},
}
got := buildBarmanObjectStore(tmpl, "s")
if got.EndpointCA == nil || got.EndpointCA.Name != "trust-bundle" || got.EndpointCA.Key != "ca.crt" {
t.Fatalf("EndpointCA mismatch: %#v", got.EndpointCA)
}
})
t.Run("custom key", func(t *testing.T) {
tmpl := strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://b/",
EndpointCA: &strategyv1alpha1.EndpointCARef{
SecretRef: corev1.LocalObjectReference{Name: "trust-bundle"},
Key: "tls.crt",
},
}
got := buildBarmanObjectStore(tmpl, "s")
if got.EndpointCA == nil || got.EndpointCA.Key != "tls.crt" {
t.Fatalf("EndpointCA key not honored: %#v", got.EndpointCA)
}
})
t.Run("nil block stays nil", func(t *testing.T) {
tmpl := strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://b/"}
got := buildBarmanObjectStore(tmpl, "s")
if got.EndpointCA != nil {
t.Fatalf("EndpointCA must be nil when not configured; got %#v", got.EndpointCA)
}
})
t.Run("empty SecretRef.Name treated as not configured", func(t *testing.T) {
tmpl := strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://b/",
EndpointCA: &strategyv1alpha1.EndpointCARef{Key: "ca.crt"},
}
got := buildBarmanObjectStore(tmpl, "s")
if got.EndpointCA != nil {
t.Fatalf("expected nil EndpointCA when SecretRef.Name is empty; got %#v", got.EndpointCA)
}
})
}
func TestRenderCNPGTemplate_TemplatingApplicationName(t *testing.T) {
tmpl := strategyv1alpha1.CNPGTemplate{
ServerName: "{{ .Application.metadata.name }}",
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://bucket/{{ .Application.metadata.name }}/",
S3Credentials: &strategyv1alpha1.S3CredentialsTemplate{
SecretRef: corev1.LocalObjectReference{Name: "{{ .Application.metadata.name }}-creds"},
},
},
}
app := newPostgresApp("pg-src", "tenant")
got, err := renderCNPGTemplate(tmpl, app, nil)
if err != nil {
t.Fatalf("renderCNPGTemplate: %v", err)
}
if got.ServerName != "pg-src" {
t.Errorf("ServerName not templated: got %q", got.ServerName)
}
if got.BarmanObjectStore.DestinationPath != "s3://bucket/pg-src/" {
t.Errorf("DestinationPath not templated: got %q", got.BarmanObjectStore.DestinationPath)
}
if got.BarmanObjectStore.S3Credentials.SecretRef.Name != "pg-src-creds" {
t.Errorf("SecretRef.Name not templated: got %q", got.BarmanObjectStore.S3Credentials.SecretRef.Name)
}
}
// TestRenderCNPGTemplate_RoundTripsParametersThroughSnapshot locks in the
// fix for review blocker 1 (#4387688473): a strategy template that uses
// {{ .Parameters.foo }} for a Secret reference / endpointCA / key name
// must yield the same rendered value at restore time as it did at backup
// time. Before the fix, the restore reconciler called renderCNPGTemplate
// with parameters=nil, so any .Parameters reference rendered "<no value>"
// and the patched target Postgres app got broken Secret references that
// CNPG would later reject.
//
// Verifies the contract end-to-end: render-backup -> marshal snapshot ->
// unmarshal snapshot -> render-restore. Also asserts the buggy path
// (parameters=nil at restore) produces a recognisable broken value, so
// a regression that drops the persistence side gets caught here too.
func TestRenderCNPGTemplate_RoundTripsParametersThroughSnapshot(t *testing.T) {
tmpl := strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://bucket/{{ .Application.metadata.name }}/",
S3Credentials: &strategyv1alpha1.S3CredentialsTemplate{
SecretRef: corev1.LocalObjectReference{Name: "{{ .Parameters.credsSecret }}"},
},
EndpointCA: &strategyv1alpha1.EndpointCARef{
SecretRef: corev1.LocalObjectReference{Name: "{{ .Parameters.caSecret }}"},
Key: "ca.crt",
},
},
}
app := newPostgresApp("pg-src", "tenant")
parameters := map[string]string{
"credsSecret": "tenant-shared-creds",
"caSecret": "tenant-shared-ca",
}
// Backup-time render: real values flow through.
rendBackup, err := renderCNPGTemplate(tmpl, app, parameters)
if err != nil {
t.Fatalf("backup-time render: %v", err)
}
if rendBackup.BarmanObjectStore.S3Credentials.SecretRef.Name != "tenant-shared-creds" {
t.Fatalf("backup-time SecretRef.Name: got %q want %q",
rendBackup.BarmanObjectStore.S3Credentials.SecretRef.Name, "tenant-shared-creds")
}
if rendBackup.BarmanObjectStore.EndpointCA.SecretRef.Name != "tenant-shared-ca" {
t.Fatalf("backup-time EndpointCA.SecretRef.Name: got %q want %q",
rendBackup.BarmanObjectStore.EndpointCA.SecretRef.Name, "tenant-shared-ca")
}
// Round-trip parameters through Backup.status.underlyingResources.
raw, err := marshalCNPGBackupSnapshot(app, parameters, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
bk := &backupsv1alpha1.Backup{Status: backupsv1alpha1.BackupStatus{UnderlyingResources: raw}}
snap, err := unmarshalCNPGBackupSnapshot(bk)
if err != nil {
t.Fatalf("unmarshal snapshot: %v", err)
}
gotParams := snap.Parameters
// Restore-time render with retrieved parameters: same values out.
rendRestore, err := renderCNPGTemplate(tmpl, app, gotParams)
if err != nil {
t.Fatalf("restore-time render: %v", err)
}
if rendRestore.BarmanObjectStore.S3Credentials.SecretRef.Name != "tenant-shared-creds" {
t.Errorf("restore-time SecretRef.Name: got %q want %q",
rendRestore.BarmanObjectStore.S3Credentials.SecretRef.Name, "tenant-shared-creds")
}
if rendRestore.BarmanObjectStore.EndpointCA.SecretRef.Name != "tenant-shared-ca" {
t.Errorf("restore-time EndpointCA.SecretRef.Name: got %q want %q",
rendRestore.BarmanObjectStore.EndpointCA.SecretRef.Name, "tenant-shared-ca")
}
// Buggy path: rendering at restore with parameters=nil produces
// "<no value>". Catching this here means a future change that
// silently drops parameters off the snapshot fails this test before
// it ships a broken patch to a real Postgres app.
rendBroken, err := renderCNPGTemplate(tmpl, app, nil)
if err != nil {
t.Fatalf("nil-params render: %v", err)
}
if rendBroken.BarmanObjectStore.S3Credentials.SecretRef.Name != "<no value>" {
t.Errorf("expected broken render for nil params (so the test catches regressions), got %q",
rendBroken.BarmanObjectStore.S3Credentials.SecretRef.Name)
}
}
func TestParseCNPGRestoreOptions(t *testing.T) {
cases := []struct {
name string
raw string
wantRecoveryTime string
wantTimeout int64
wantErr bool
}{
{"nil blob", "", "", 0, false},
{"missing fields", `{"foo":"bar"}`, "", 0, false},
{"recovery time only", `{"recoveryTime":"2026-05-01T12:00:00Z"}`, "2026-05-01T12:00:00Z", 0, false},
{"timeout only", `{"restoreTimeoutSeconds":7200}`, "", 7200, false},
{"both", `{"recoveryTime":"2026-05-01T12:00:00Z","restoreTimeoutSeconds":3600}`, "2026-05-01T12:00:00Z", 3600, false},
// A malformed blob must surface a parse error so callers can log it.
// Returning silently used to mask "why didn't recoveryTime apply" for
// tenants debugging restores; see Blocker 5 in the branch review.
{"malformed", `not-json`, "", 0, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
var ext *runtime.RawExtension
if tc.raw != "" {
ext = &runtime.RawExtension{Raw: []byte(tc.raw)}
}
got, err := parseCNPGRestoreOptions(ext)
if tc.wantErr {
if err == nil {
t.Fatalf("expected parse error, got nil")
}
return
}
if err != nil {
t.Fatalf("unexpected parse error: %v", err)
}
if got.RecoveryTime != tc.wantRecoveryTime {
t.Errorf("recoveryTime: got %q want %q", got.RecoveryTime, tc.wantRecoveryTime)
}
if got.RestoreTimeoutSeconds != tc.wantTimeout {
t.Errorf("restoreTimeoutSeconds: got %d want %d", got.RestoreTimeoutSeconds, tc.wantTimeout)
}
})
}
}
func TestEffectiveRestoreDeadline(t *testing.T) {
cases := []struct {
name string
opts CNPGRestoreOptions
want time.Duration
}{
{"unset falls back to default", CNPGRestoreOptions{}, cnpgDefaultRestoreDeadline},
{"zero falls back to default", CNPGRestoreOptions{RestoreTimeoutSeconds: 0}, cnpgDefaultRestoreDeadline},
{"negative falls back to default", CNPGRestoreOptions{RestoreTimeoutSeconds: -5}, cnpgDefaultRestoreDeadline},
{"positive override", CNPGRestoreOptions{RestoreTimeoutSeconds: 7200}, 2 * time.Hour},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := tc.opts.effectiveRestoreDeadline()
if got != tc.want {
t.Errorf("got %s want %s", got, tc.want)
}
})
}
}
// TestEffectiveBootstrapDisableGrace pins that the post-convergence
// bootstrap-disable window is a FIXED cnpgBootstrapDisableGrace, independent of
// restoreTimeoutSeconds: that knob bounds the recovery wait, whereas this window
// absorbs a control-plane blip whose duration is unrelated to it. A short timeout
// must not shrink the window (a blip would then falsely fail a converged restore
// and expose the healthy Cluster to a resubmit's purge-guard), and a long timeout
// must not inflate it.
func TestEffectiveBootstrapDisableGrace(t *testing.T) {
// The grace is a FIXED window, independent of restoreTimeoutSeconds: a short,
// a long and an unset recovery timeout all yield the same control-plane-blip
// window. Re-coupling it to the deadline turns these red.
cases := []struct {
name string
opts CNPGRestoreOptions
}{
{"unset", CNPGRestoreOptions{}},
{"short timeout", CNPGRestoreOptions{RestoreTimeoutSeconds: 60}},
{"long timeout", CNPGRestoreOptions{RestoreTimeoutSeconds: 7200}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := tc.opts.effectiveBootstrapDisableGrace()
if got != cnpgBootstrapDisableGrace {
t.Errorf("grace %s should be the fixed %s regardless of restoreTimeoutSeconds", got, cnpgBootstrapDisableGrace)
}
})
}
}
// TestEffectiveWALArchiveDeadline mirrors TestEffectiveRestoreDeadline for
// the dedicated WAL-archive gate cap. The two deadlines exist independently
// so a big-DB restore can extend RestoreTimeoutSeconds without simultaneously
// loosening the gate that catches a stuck archive_command.
func TestEffectiveWALArchiveDeadline(t *testing.T) {
cases := []struct {
name string
opts CNPGRestoreOptions
want time.Duration
}{
{"unset falls back to default", CNPGRestoreOptions{}, cnpgWALArchiveDeadline},
{"zero falls back to default", CNPGRestoreOptions{WALArchiveTimeoutSeconds: 0}, cnpgWALArchiveDeadline},
{"negative falls back to default", CNPGRestoreOptions{WALArchiveTimeoutSeconds: -5}, cnpgWALArchiveDeadline},
{"positive override", CNPGRestoreOptions{WALArchiveTimeoutSeconds: 600}, 10 * time.Minute},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := tc.opts.effectiveWALArchiveDeadline()
if got != tc.want {
t.Errorf("got %s want %s", got, tc.want)
}
})
}
}
// TestValidateCNPGApplicationRef locks in the API-group guard. Without it a
// non-Cozystack ref like other.example.com/Postgres slipped past the Kind
// check and hit the hard-wired apps.cozystack.io typed client, reconciling
// against the wrong CRD entirely.
func TestValidateCNPGApplicationRef(t *testing.T) {
cozyGroup := postgresapp.GroupName
otherGroup := "other.example.com"
cases := []struct {
name string
ref corev1.TypedLocalObjectReference
wantErr bool
}{
{
name: "kind=Postgres, apiGroup=apps.cozystack.io: accepted",
ref: corev1.TypedLocalObjectReference{APIGroup: &cozyGroup, Kind: postgresAppKind, Name: "pg"},
},
{
name: "kind=Postgres, nil apiGroup: accepted (default applies upstream)",
ref: corev1.TypedLocalObjectReference{Kind: postgresAppKind, Name: "pg"},
},
{
name: "wrong kind: rejected",
ref: corev1.TypedLocalObjectReference{APIGroup: &cozyGroup, Kind: "MySQL", Name: "pg"},
wantErr: true,
},
{
name: "right kind, wrong apiGroup: rejected",
ref: corev1.TypedLocalObjectReference{APIGroup: &otherGroup, Kind: postgresAppKind, Name: "pg"},
wantErr: true,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
err := validateCNPGApplicationRef(tc.ref)
if tc.wantErr && err == nil {
t.Fatalf("expected validation error, got nil")
}
if !tc.wantErr && err != nil {
t.Fatalf("unexpected validation error: %v", err)
}
})
}
}
func TestResolveCNPGRestoreTarget(t *testing.T) {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
src := &backupsv1alpha1.Backup{}
src.Namespace = "tenant-foo"
src.Spec.ApplicationRef.APIGroup = &apiGroup
src.Spec.ApplicationRef.Kind = postgresAppKind
src.Spec.ApplicationRef.Name = "pg-src"
r := &RestoreJobReconciler{}
t.Run("source app when targetApplicationRef nil", func(t *testing.T) {
rj := &backupsv1alpha1.RestoreJob{}
got := r.resolveCNPGRestoreTarget(rj, src)
if got.AppName != "pg-src" || got.Namespace != "tenant-foo" || got.Kind != postgresAppKind {
t.Fatalf("unexpected target: %+v", got)
}
})
t.Run("override app name when targetApplicationRef differs", func(t *testing.T) {
rj := &backupsv1alpha1.RestoreJob{}
rj.Spec.TargetApplicationRef = newTypedRef(apiGroup, postgresAppKind, "pg-target")
got := r.resolveCNPGRestoreTarget(rj, src)
if got.AppName != "pg-target" {
t.Fatalf("unexpected app name: %q", got.AppName)
}
})
t.Run("keep source app name when targetApplicationRef matches source", func(t *testing.T) {
rj := &backupsv1alpha1.RestoreJob{}
rj.Spec.TargetApplicationRef = newTypedRef(apiGroup, postgresAppKind, "pg-src")
got := r.resolveCNPGRestoreTarget(rj, src)
if got.AppName != "pg-src" {
t.Fatalf("unexpected app name: %q", got.AppName)
}
})
}
func TestCNPGBackupPhase(t *testing.T) {
cases := []struct {
name string
backup *cnpgtypes.Backup
wantPhase string
wantMsg string
}{
{"nil", nil, "", ""},
{"completed", &cnpgtypes.Backup{Status: cnpgtypes.BackupStatus{Phase: "completed"}}, "completed", ""},
{"failed-with-error", &cnpgtypes.Backup{Status: cnpgtypes.BackupStatus{Phase: "failed", Error: "boom"}}, "failed", "boom"},
{"empty status", &cnpgtypes.Backup{}, "", ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
phase, msg := cnpgBackupPhase(tc.backup)
if phase != tc.wantPhase || msg != tc.wantMsg {
t.Fatalf("got (%q,%q), want (%q,%q)", phase, msg, tc.wantPhase, tc.wantMsg)
}
})
}
}
// newTypedRef is a small helper to build a TypedLocalObjectReference without
// the test having to bother with apiGroup pointer plumbing.
func newTypedRef(apiGroup, kind, name string) *corev1.TypedLocalObjectReference {
return &corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: kind, Name: name}
}
// newPostgresApp builds a typed Postgres CR for tests.
func newPostgresApp(name, namespace string) *postgresapp.Postgres {
return &postgresapp.Postgres{
TypeMeta: metav1.TypeMeta{
APIVersion: postgresapp.GroupVersion.String(),
Kind: postgresapp.Kind,
},
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: namespace,
},
}
}
// TestBuildPostgresAppRestorePatch_ForwardsSecretRef_NotCleartext locks in
// the security-critical contract that S3 credentials are forwarded to the
// chart as a Secret reference, not pasted into the CR .spec as cleartext.
// Regressing this would expose access keys via etcd, audit logs, and
// tenant-readable kubectl get -o yaml.
func TestBuildPostgresAppRestorePatch_ForwardsSecretRef_NotCleartext(t *testing.T) {
app := newPostgresApp("pg-target", "tenant-foo")
creds := &strategyv1alpha1.S3CredentialsTemplate{
SecretRef: corev1.LocalObjectReference{Name: "pg-target-cnpg-backup-creds"},
}
patched := buildPostgresAppRestorePatch(
app,
"pg-src",
"postgres-pg-target-restore-deadbeef",
"s3://bucket/pg-src/",
"https://s3.example",
"",
"",
creds,
nil,
nil, nil,
)
if got := patched.Spec.Backup.S3CredentialsSecret.Name; got != "pg-target-cnpg-backup-creds" {
t.Errorf("s3CredentialsSecret.name: got %q want %q", got, "pg-target-cnpg-backup-creds")
}
// Cleartext key fields must NOT be set. Their presence on the CR .spec is
// the bug this test guards against.
if got := patched.Spec.Backup.S3AccessKey; got != "" {
t.Errorf("spec.backup.s3AccessKey must not be set on the patched app; got %q", got)
}
if got := patched.Spec.Backup.S3SecretKey; got != "" {
t.Errorf("spec.backup.s3SecretKey must not be set on the patched app; got %q", got)
}
}
// TestBuildPostgresAppRestorePatch_ForwardsCustomKeyOverrides verifies the
// strategy template's optional key-name overrides round-trip into the chart
// values so non-default Secret layouts work end-to-end.
func TestBuildPostgresAppRestorePatch_ForwardsCustomKeyOverrides(t *testing.T) {
app := newPostgresApp("pg", "tenant")
creds := &strategyv1alpha1.S3CredentialsTemplate{
SecretRef: corev1.LocalObjectReference{Name: "creds"},
AccessKeyIDKey: "ACCESS_KEY",
SecretAccessKeyKey: "SECRET_KEY",
}
patched := buildPostgresAppRestorePatch(app, "src", "pg-new", "s3://b/", "", "", "", creds, nil, nil, nil)
want := postgresapp.S3CredentialsSecret{
Name: "creds",
AccessKeyIDKey: "ACCESS_KEY",
SecretAccessKeyKey: "SECRET_KEY",
}
if got := patched.Spec.Backup.S3CredentialsSecret; got != want {
t.Fatalf("s3CredentialsSecret mismatch\n got: %#v\nwant: %#v", got, want)
}
}
// TestBuildPostgresAppRestorePatch_NoSecretRefIsSkipped verifies the
// patcher does not stamp an empty s3CredentialsSecret block when the strategy
// did not declare credentials. Otherwise the chart would render an empty
// Secret name and helm install would fail.
func TestBuildPostgresAppRestorePatch_NoSecretRefIsSkipped(t *testing.T) {
app := newPostgresApp("pg", "tenant")
patched := buildPostgresAppRestorePatch(app, "src", "pg-new", "s3://b/", "", "", "", nil, nil, nil, nil)
if got := patched.Spec.Backup.S3CredentialsSecret; got != (postgresapp.S3CredentialsSecret{}) {
t.Errorf("spec.backup.s3CredentialsSecret must be zero when credsRef is nil; got %#v", got)
}
}
// TestCNPGPurgeNeeded locks in the dual-guard against re-purging the
// freshly-recovered Cluster on a status-update failure. The controller used
// to rely solely on a Status condition: if the post-purge Status().Update
// raced or failed, the next reconcile would re-purge the just-restored
// Cluster, destroying recovery progress. Cross-checking that the live
// Cluster is a *freshly-recovered* one (bootstrap.recovery + created after
// the job started) makes that second purge a no-op - while still purging a
// stale recovery Cluster left over from an earlier completed restore, so a
// repeat in-place restore is not a silent no-op.
func TestCNPGPurgeNeeded(t *testing.T) {
cases := []struct {
name string
purgedCondition bool
liveClusterFreshlyRecovered bool
want bool
}{
{
name: "fresh restore, old cluster still in place: purge",
purgedCondition: false,
liveClusterFreshlyRecovered: false,
want: true,
},
{
name: "purge already recorded: skip",
purgedCondition: true,
liveClusterFreshlyRecovered: false,
want: false,
},
{
name: "live cluster freshly recovered by this restore (status write raced): skip",
purgedCondition: false,
liveClusterFreshlyRecovered: true,
want: false,
},
{
name: "stale recovery cluster from a previous completed restore: purge (repeat-restore fix)",
purgedCondition: false,
liveClusterFreshlyRecovered: false,
want: true,
},
{
name: "both true: skip (post-purge steady state)",
purgedCondition: true,
liveClusterFreshlyRecovered: true,
want: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := cnpgPurgeNeeded(tc.purgedCondition, tc.liveClusterFreshlyRecovered)
if got != tc.want {
t.Errorf("got %v want %v", got, tc.want)
}
})
}
}
// TestCNPGClusterFreshlyRecovered locks in the signal that separates a
// recovery Cluster THIS restore just produced from one left over by an
// earlier completed restore. The repeat-in-place-restore bug was exactly a
// stale recovery Cluster (created before StartedAt) being mistaken for a
// freshly-recovered one and skipping the purge.
func TestCNPGClusterFreshlyRecovered(t *testing.T) {
started := metav1.NewTime(time.Now())
after := metav1.NewTime(started.Add(time.Minute))
before := metav1.NewTime(started.Add(-time.Hour))
cases := []struct {
name string
hasRecovery bool
createdAt *metav1.Time
startedAt *metav1.Time
want bool
}{
{
name: "no recovery bootstrap: not fresh",
hasRecovery: false,
createdAt: &after,
startedAt: &started,
want: false,
},
{
name: "recovery cluster created after start (our own purge re-render): fresh",
hasRecovery: true,
createdAt: &after,
startedAt: &started,
want: true,
},
{
name: "recovery cluster created before start (leftover from previous restore): not fresh",
hasRecovery: true,
createdAt: &before,
startedAt: &started,
want: false,
},
{
name: "recovery cluster created exactly at start: not fresh (conservative tie -> purge)",
hasRecovery: true,
createdAt: &started,
startedAt: &started,
want: false,
},
{
name: "missing cluster creation timestamp: not fresh",
hasRecovery: true,
createdAt: nil,
startedAt: &started,
want: false,
},
{
name: "missing job start timestamp: not fresh",
hasRecovery: true,
createdAt: &after,
startedAt: nil,
want: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := cnpgClusterFreshlyRecovered(tc.hasRecovery, tc.createdAt, tc.startedAt)
if got != tc.want {
t.Errorf("got %v want %v", got, tc.want)
}
})
}
}
// TestBuildPostgresAppRestorePatch_ReplacesTargetUsersAndDatabases locks
// in the authoritative-from-source semantics: target's stale users and
// databases must be wiped and replaced with source's exact map. The
// recovered cluster carries source's role catalog and data; if target's
// pre-restore drift survives, the chart's init-job either tries to
// re-create roles against the wrong data or resurrects a role a previous
// tenant configuration had dropped.
func TestBuildPostgresAppRestorePatch_ReplacesTargetUsersAndDatabases(t *testing.T) {
app := newPostgresApp("pg-target", "tenant")
// Target had pre-existing users/databases (e.g. from a previous restore
// or operator drift). Replace must wipe them.
app.Spec.Users = map[string]postgresapp.User{
"stale-target-user": {Replication: true},
}
app.Spec.Databases = map[string]postgresapp.Database{
"stale-target-db": {Extensions: []string{"pgcrypto"}},
}
sourceUsers := map[string]postgresapp.User{
"app": {Replication: true},
}
sourceDatabases := map[string]postgresapp.Database{
"appdb": {Extensions: []string{"hstore"}},
}
patched := buildPostgresAppRestorePatch(app, "src", "pg-new", "s3://b/", "", "", "", nil, nil, sourceDatabases, sourceUsers)
if _, ok := patched.Spec.Users["stale-target-user"]; ok {
t.Errorf("stale target user survived restore; replace semantics regressed")
}
if u, ok := patched.Spec.Users["app"]; !ok {
t.Errorf("source user was not propagated onto target")
} else if !u.Replication {
t.Errorf("source user attributes not propagated; got %#v", u)
}
if _, ok := patched.Spec.Databases["stale-target-db"]; ok {
t.Errorf("stale target database survived restore; replace semantics regressed")
}
if _, ok := patched.Spec.Databases["appdb"]; !ok {
t.Errorf("source database was not propagated onto target")
}
}
// TestBuildPostgresAppRestorePatch_ScrubsStaleBackupSettings guards the
// stale-state hygiene called out by coderabbitai on
// internal/backupcontroller/cnpgstrategy_controller.go:597-604: a re-restore
// into the same target must not inherit a previous restore's recoveryTime,
// endpointURL, S3 credentials Secret reference, or inline cleartext S3 keys.
func TestBuildPostgresAppRestorePatch_ScrubsStaleBackupSettings(t *testing.T) {
app := newPostgresApp("pg-target", "tenant")
app.Spec.Bootstrap.RecoveryTime = "2025-01-01T00:00:00Z"
app.Spec.Backup.UseSystemBucket = true
app.Spec.Backup.EndpointURL = "https://stale.example"
app.Spec.Backup.S3AccessKey = "stale-ak"
app.Spec.Backup.S3SecretKey = "stale-sk"
app.Spec.Backup.S3CredentialsSecret = postgresapp.S3CredentialsSecret{
Name: "stale-creds", AccessKeyIDKey: "OLD_ID", SecretAccessKeyKey: "OLD_KEY",
}
app.Spec.Backup.EndpointCA = postgresapp.EndpointCA{Name: "stale-ca", Key: "stale.crt"}
// Restore with no recoveryTime, no endpointURL, no creds, no CA -
// everything stale on the target must be wiped.
patched := buildPostgresAppRestorePatch(app, "src", "pg-new", "s3://b/", "", "", "", nil, nil, nil, nil)
if got := patched.Spec.Bootstrap.RecoveryTime; got != "" {
t.Errorf("stale recoveryTime survived; got %q", got)
}
if got := patched.Spec.Backup.EndpointURL; got != "" {
t.Errorf("stale endpointURL survived; got %q", got)
}
if got := patched.Spec.Backup.S3AccessKey; got != "" {
t.Errorf("inline s3AccessKey was not scrubbed; got %q", got)
}
if got := patched.Spec.Backup.S3SecretKey; got != "" {
t.Errorf("inline s3SecretKey was not scrubbed; got %q", got)
}
if patched.Spec.Backup.S3CredentialsSecret != (postgresapp.S3CredentialsSecret{}) {
t.Errorf("stale s3CredentialsSecret survived; got %+v", patched.Spec.Backup.S3CredentialsSecret)
}
if patched.Spec.Backup.EndpointCA != (postgresapp.EndpointCA{}) {
t.Errorf("stale endpointCA survived; got %+v", patched.Spec.Backup.EndpointCA)
}
// A system-bucket source must be flipped to the explicit-creds path, or the
// chart's bootstrap.enabled + useSystemBucket guard wedges the restore.
if patched.Spec.Backup.UseSystemBucket {
t.Errorf("useSystemBucket must be cleared on restore so the bootstrap guard passes")
}
}
// TestBuildPostgresAppRestorePatch_ForwardsEndpointCA verifies that a
// strategy template with an EndpointCA reference flows through to the
// target app's spec.backup.endpointCA. Without this the chart's
// externalClusters[].barmanObjectStore.endpointCA stays empty and the
// recovery init job panics in CNPG's instance-manager when the S3
// endpoint serves a self-signed certificate (cozystack seaweedfs-s3 does).
func TestBuildPostgresAppRestorePatch_ForwardsEndpointCA(t *testing.T) {
app := newPostgresApp("pg", "tenant")
caRef := &strategyv1alpha1.EndpointCARef{
SecretRef: corev1.LocalObjectReference{Name: "pg-cnpg-backup-ca"},
Key: "ca.crt",
}
patched := buildPostgresAppRestorePatch(app, "src", "pg-new", "s3://b/", "", "", "", nil, caRef, nil, nil)
want := postgresapp.EndpointCA{Name: "pg-cnpg-backup-ca", Key: "ca.crt"}
if got := patched.Spec.Backup.EndpointCA; got != want {
t.Fatalf("endpointCA mismatch\n got: %#v\nwant: %#v", got, want)
}
}
// TestBuildPostgresAppRestorePatch_ArchiveServerNameDiffersFromSource guards the
// in-place restore fix: the restored cluster archives under bootstrap.newServerName,
// distinct from the recovery source (bootstrap.serverName / oldName).
func TestBuildPostgresAppRestorePatch_ArchiveServerNameDiffersFromSource(t *testing.T) {
app := newPostgresApp("pg", "tenant")
// An in-place restore is the worst case: source serverName equals the
// target cluster's own name.
const source = "postgres-pg"
patched := buildPostgresAppRestorePatch(app, source, "postgres-pg-restore-deadbeef", "s3://b/", "", "", "", nil, nil, nil, nil)
if got := patched.Spec.Bootstrap.ServerName; got != source {
t.Errorf("recovery serverName must stay the source; got %q want %q", got, source)
}
if got := patched.Spec.Bootstrap.OldName; got != source {
t.Errorf("recovery oldName must stay the source; got %q want %q", got, source)
}
if got := patched.Spec.Bootstrap.NewServerName; got == "" || got == source {
t.Errorf("archive newServerName must be non-empty and differ from the source %q; got %q", source, got)
}
}
// TestRestoredServerName pins the archive-prefix generator: deterministic per
// RestoreJob UID, distinct across UIDs, and never equal to the plain cluster
// name (which is the source serverName on an in-place restore).
func TestRestoredServerName(t *testing.T) {
const cluster = "postgres-pg"
a1 := restoredServerName(cluster, types.UID("3c2ee41b-346f-4ae2-98d2-5d629c96f744"))
a2 := restoredServerName(cluster, types.UID("3c2ee41b-346f-4ae2-98d2-5d629c96f744"))
b := restoredServerName(cluster, types.UID("9eb8feed-7d49-467e-b719-d1896c0602d5"))
if a1 != a2 {
t.Errorf("not deterministic for a fixed UID: %q vs %q", a1, a2)
}
if a1 == b {
t.Errorf("distinct UIDs must yield distinct serverNames; both %q", a1)
}
if a1 == cluster {
t.Errorf("restored serverName must never equal the plain cluster name %q", cluster)
}
if !strings.HasPrefix(a1, cluster+"-restore-") {
t.Errorf("restored serverName %q must carry the %q-restore- prefix", a1, cluster)
}
}
// TestMarshalUnmarshalCNPGBackupSnapshot round-trips the source-spec
// snapshot persisted in Backup.status.underlyingResources, including the
// BackupClassStrategy parameters that were in effect at backup time -
// without persisting those a strategy that uses {{ .Parameters.foo }} for
// a Secret reference / endpointCA / key name would re-render at restore
// time with parameters=nil and produce a broken patch.
func TestMarshalUnmarshalCNPGBackupSnapshot(t *testing.T) {
src := newPostgresApp("pg-src", "tenant")
src.Spec.Databases = map[string]postgresapp.Database{
"app": {Extensions: []string{"pgcrypto"}},
}
src.Spec.Users = map[string]postgresapp.User{
"app": {Replication: true},
}
parameters := map[string]string{
"credsSecret": "tenant-shared-creds",
"region": "eu-west-1",
}
raw, err := marshalCNPGBackupSnapshot(src, parameters, "")
if err != nil {
t.Fatalf("marshal: %v", err)
}
if raw == nil {
t.Fatalf("expected non-nil RawExtension")
}
bk := &backupsv1alpha1.Backup{Status: backupsv1alpha1.BackupStatus{UnderlyingResources: raw}}
snap, err := unmarshalCNPGBackupSnapshot(bk)
if err != nil {
t.Fatalf("unmarshal: %v", err)
}
dbs, users, params := snap.Databases, snap.Users, snap.Parameters
if dbs["app"].Extensions[0] != "pgcrypto" {
t.Errorf("databases round-trip mismatch: %#v", dbs)
}
if !users["app"].Replication {
t.Errorf("users round-trip mismatch: %#v", users)
}
if params["credsSecret"] != "tenant-shared-creds" {
t.Errorf("parameters round-trip mismatch: got %#v", params)
}
if params["region"] != "eu-west-1" {
t.Errorf("parameters round-trip mismatch: got %#v", params)
}
}
// TestMarshalUnmarshalCNPGBackupSnapshot_BackwardCompat covers a Backup
// taken by an older controller version (no parameters block in the
// snapshot JSON). The unmarshal path must treat that as "no parameter
// overrides" rather than failing - the strategy template might have used
// .Parameters at the time, and we cannot reconstruct values we never
// persisted, so the caller proceeds with parameters=nil and lets the
// rendered output fall back to whatever the template defaults emit.
func TestMarshalUnmarshalCNPGBackupSnapshot_BackwardCompat(t *testing.T) {
legacyRaw := []byte(`{"kind":"CNPGBackupSnapshot","apiVersion":"backups.cozystack.io/v1alpha1","databases":{"app":{}},"users":{"app":{}}}`)
bk := &backupsv1alpha1.Backup{Status: backupsv1alpha1.BackupStatus{UnderlyingResources: &runtime.RawExtension{Raw: legacyRaw}}}
snap, err := unmarshalCNPGBackupSnapshot(bk)
if err != nil {
t.Fatalf("unmarshal legacy snapshot: %v", err)
}
if len(snap.Parameters) != 0 {
t.Errorf("legacy snapshot must yield empty parameters, got %#v", snap.Parameters)
}
// Flavors did not exist when this snapshot was written: every cluster
// ran postgresql, and the restore flavor check must read it that way.
if got := cnpgFlavor(snap.Flavor); got != "postgresql" {
t.Errorf("legacy snapshot must resolve to the postgresql flavor, got %q", got)
}
}
// TestUnmarshalCNPGBackupSnapshot_RejectsMissingOrWrongKind locks in the
// fail-loud contract: a Backup without a snapshot must NOT silently
// proceed - it would let the chart's init-job DROP recovered roles. Same
// for a Backup whose snapshot carries the wrong Kind (foreign payload).
func TestUnmarshalCNPGBackupSnapshot_RejectsMissingOrWrongKind(t *testing.T) {
cases := []struct {
name string
ur *runtime.RawExtension
}{
{"nil underlyingResources", nil},
{"empty raw", &runtime.RawExtension{Raw: []byte{}}},
{"wrong kind", &runtime.RawExtension{Raw: []byte(`{"kind":"VMInstanceResources","apiVersion":"backups.cozystack.io/v1alpha1"}`)}},
{"malformed json", &runtime.RawExtension{Raw: []byte(`not-json`)}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
bk := &backupsv1alpha1.Backup{Status: backupsv1alpha1.BackupStatus{UnderlyingResources: tc.ur}}
_, err := unmarshalCNPGBackupSnapshot(bk)
if err == nil {
t.Fatalf("expected error, got nil")
}
})
}
}
// TestCNPGBackupDeadlineExceeded locks in the wall-clock cap that prevents a
// perpetually-failed cnpg.io/Backup from pinning the BackupJob in Running
// forever. Without this guard the Plan-controller queue would back up
// behind a single broken backup.
func TestCNPGBackupDeadlineExceeded(t *testing.T) {
cases := []struct {
name string
startedAt *metav1.Time
want bool
}{
{
name: "nil startedAt does not trip the gate",
startedAt: nil,
want: false,
},
{
name: "fresh start is well under the deadline",
startedAt: &metav1.Time{Time: time.Now()},
want: false,
},
{
name: "just under deadline does not trip",
startedAt: &metav1.Time{Time: time.Now().Add(-(cnpgDefaultBackupDeadline - time.Minute))},
want: false,
},
{
name: "past deadline trips the gate",
startedAt: &metav1.Time{Time: time.Now().Add(-(cnpgDefaultBackupDeadline + time.Minute))},
want: true,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := cnpgBackupDeadlineExceeded(tc.startedAt)
if got != tc.want {
t.Errorf("got %v want %v", got, tc.want)
}
})
}
}
// TestCreateCNPGBackupArtifact_AlreadyExistsReturnsExisting locks in the
// idempotency guard for createCNPGBackupArtifact: an AlreadyExists collision
// (typical when the previous reconcile created the artifact and then raced
// on the BackupJob status update) must return the existing artifact rather
// than propagate the error and flip a completed run to Failed.
func TestCreateCNPGBackupArtifact_AlreadyExistsReturnsExisting(t *testing.T) {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
preExisting := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: "pg"},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "strat"},
DriverMetadata: map[string]string{"marker": "first-reconcile"},
},
}
c := newCNPGStrategyTestClient(t, preExisting)
r := &BackupJobReconciler{Client: c}
j := &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
Spec: backupsv1alpha1.BackupJobSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: "pg"},
},
}
resolved := &ResolvedBackupConfig{
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "strat"},
}
cnpgBk := &cnpgtypes.Backup{ObjectMeta: metav1.ObjectMeta{Name: "cnpg-bk"}}
rendered := &strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://b/"},
}
sourceApp := newPostgresApp("pg", "tenant")
got, err := r.createCNPGBackupArtifact(context.Background(), j, resolved, cnpgBk, "postgres-pg", "postgres-pg", "postgresql", rendered, sourceApp)
if err != nil {
t.Fatalf("expected AlreadyExists to be swallowed, got error %v", err)
}
if got == nil {
t.Fatalf("expected existing Backup returned, got nil")
}
if got.Spec.DriverMetadata["marker"] != "first-reconcile" {
t.Errorf("expected the existing object back, got %#v", got.Spec.DriverMetadata)
}
}
// TestApplyClusterPluginBackup_NotFoundOnMissingCluster locks in the
// precondition on applyClusterPluginBackup: when the live cnpg.io Cluster has
// not yet been rendered by the chart, the driver must surface NotFound (so the
// caller can retry-with-backoff) instead of issuing a doomed SSA Apply that
// would be rejected by the API server for missing required Cluster fields.
func TestApplyClusterPluginBackup_NotFoundOnMissingCluster(t *testing.T) {
c := newCNPGStrategyTestClient(t)
r := &BackupJobReconciler{Client: c, Interface: cnpgDynamicFor(t)}
tmpl := &strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://bucket/x/",
},
}
_, err := r.applyClusterPluginBackup(context.Background(), "tenant", "postgres-missing", tmpl, "postgres-missing", true)
if err == nil {
t.Fatalf("expected NotFound error, got nil")
}
if !apierrors.IsNotFound(err) {
t.Fatalf("expected IsNotFound error, got %v", err)
}
}
// TestRecoveryBootstrapClusterState_TerminatingCluster locks in the
// DeletionTimestamp guard. Without it, a Cluster CR mid-deletion (after
// purgeExistingCluster fired r.Delete but cnpg.io's finalizers haven't
// drained yet) would get treated as "still has bootstrap" by the
// reconciler. The caller would then either skip the next purge or wait
// for healthy on a CR that's about to disappear; in either case Helm
// might SSA-merge the chart's bootstrap.recovery onto the terminating
// CR and cnpg-operator's bootstrap-immutability check would drop the
// change, leaving the cluster on the original initdb spec.
func TestRecoveryBootstrapClusterState_TerminatingCluster(t *testing.T) {
now := metav1.Now()
t.Run("terminating cluster reports not-yet-recovered", func(t *testing.T) {
cluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{
Namespace: "tenant",
Name: "postgres-app",
DeletionTimestamp: &now,
Finalizers: []string{"cnpg.io/cluster"}, // satisfies finalizer requirement for DeletionTimestamp
},
Spec: cnpgtypes.ClusterSpec{
Bootstrap: &cnpgtypes.BootstrapConfiguration{
Recovery: &cnpgtypes.RecoverySource{Source: "pg-src"},
},
},
}
c := newCNPGStrategyTestClient(t, cluster)
r := &RestoreJobReconciler{Client: c}
got, createdAt, err := r.recoveryBootstrapClusterState(context.Background(), "tenant", "postgres-app")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got {
t.Fatalf("expected false for terminating cluster, got true")
}
if createdAt != nil {
t.Fatalf("expected nil createdAt for terminating cluster, got %v", createdAt)
}
})
t.Run("live recovery cluster reports recovered with creation timestamp", func(t *testing.T) {
created := metav1.NewTime(time.Now().Add(-time.Hour))
cluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "postgres-app", CreationTimestamp: created},
Spec: cnpgtypes.ClusterSpec{
Bootstrap: &cnpgtypes.BootstrapConfiguration{
Recovery: &cnpgtypes.RecoverySource{Source: "pg-src"},
},
},
}
c := newCNPGStrategyTestClient(t, cluster)
r := &RestoreJobReconciler{Client: c}
got, createdAt, err := r.recoveryBootstrapClusterState(context.Background(), "tenant", "postgres-app")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !got {
t.Fatalf("expected true for live recovery cluster, got false")
}
if createdAt == nil {
t.Fatalf("expected non-nil createdAt for live recovery cluster")
}
})
t.Run("missing cluster reports not-yet", func(t *testing.T) {
c := newCNPGStrategyTestClient(t)
r := &RestoreJobReconciler{Client: c}
got, createdAt, err := r.recoveryBootstrapClusterState(context.Background(), "tenant", "postgres-app")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got {
t.Fatalf("expected false for missing cluster, got true")
}
if createdAt != nil {
t.Fatalf("expected nil createdAt for missing cluster, got %v", createdAt)
}
})
}
// TestCNPGBackupWALArchived locks in the gate that defers the destructive
// in-place purge until the underlying cnpg.io/Backup is in a state that
// guarantees its closing WAL is in object storage. CNPG only flips
// Backup.status.phase=completed (and writes endWal) after barman-cloud-
// backup has confirmed the upload, so checking the two together is what
// the gate uses; an earlier version of this gate compared
// Cluster.status.lastArchivedWAL to Backup.status.endWal but
// lastArchivedWAL is not present on CNPG 1.27.x cluster status, and the
// gate silently treated the missing field as the empty string and never
// cleared.
func TestCNPGBackupWALArchived(t *testing.T) {
const (
ns = "tenant"
cluster = "postgres-pg-src"
backupID = "pg-src-adhoc"
endWal = "000000010000000000000004"
)
mkPlatformBackup := func() *backupsv1alpha1.Backup {
return &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: backupID},
Spec: backupsv1alpha1.BackupSpec{
DriverMetadata: map[string]string{
cnpgBackupNameKey: backupID,
cnpgClusterNameKey: cluster,
},
},
}
}
mkCNPGBackup := func(phase, end string) *cnpgtypes.Backup {
return &cnpgtypes.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: backupID},
Spec: cnpgtypes.BackupSpec{Cluster: cnpgtypes.ClusterReference{Name: cluster}},
Status: cnpgtypes.BackupStatus{Phase: phase, EndWal: end},
}
}
t.Run("blocks while underlying cnpg.io/Backup is still running", func(t *testing.T) {
c := newCNPGStrategyTestClient(t, mkCNPGBackup("running", ""))
r := &RestoreJobReconciler{Client: c}
ready, msg, err := r.cnpgBackupWALArchived(context.Background(), mkPlatformBackup())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if ready {
t.Fatalf("expected not-ready, got ready (%s)", msg)
}
if msg == "" {
t.Errorf("expected human-readable message, got empty")
}
})
t.Run("blocks when underlying cnpg.io/Backup is in a failure state", func(t *testing.T) {
// Failure state must NOT clear the gate even when endWal somehow has
// a value - the failure could mean barman is mid-retry and the
// closing WAL upload never landed.
c := newCNPGStrategyTestClient(t, mkCNPGBackup("failed", endWal))
r := &RestoreJobReconciler{Client: c}
ready, _, err := r.cnpgBackupWALArchived(context.Background(), mkPlatformBackup())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if ready {
t.Fatalf("expected not-ready for phase=failed")
}
})
t.Run("clears when phase=completed and endWal is set", func(t *testing.T) {
c := newCNPGStrategyTestClient(t, mkCNPGBackup(cnpgBackupPhaseComplete, endWal))
r := &RestoreJobReconciler{Client: c}
ready, _, err := r.cnpgBackupWALArchived(context.Background(), mkPlatformBackup())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !ready {
t.Fatalf("expected ready when phase=completed and endWal is set")
}
})
t.Run("blocks when phase=completed but endWal is still empty", func(t *testing.T) {
// Defensive: if a future CNPG version flips phase to completed
// before persisting endWal (or a buggy build does), we want the
// gate to stay closed rather than proceeding to a destructive
// purge against a backup whose closing WAL we cannot prove is
// in object storage.
c := newCNPGStrategyTestClient(t, mkCNPGBackup(cnpgBackupPhaseComplete, ""))
r := &RestoreJobReconciler{Client: c}
ready, _, err := r.cnpgBackupWALArchived(context.Background(), mkPlatformBackup())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if ready {
t.Fatalf("expected not-ready when endWal is empty")
}
})
t.Run("conservative on missing cnpg.io/Backup (treat as ready, retention case)", func(t *testing.T) {
c := newCNPGStrategyTestClient(t)
r := &RestoreJobReconciler{Client: c}
ready, _, err := r.cnpgBackupWALArchived(context.Background(), mkPlatformBackup())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !ready {
t.Fatalf("expected ready when underlying cnpg.io/Backup is gone (retention)")
}
})
t.Run("blocks when driverMetadata is missing the backup name", func(t *testing.T) {
c := newCNPGStrategyTestClient(t)
r := &RestoreJobReconciler{Client: c}
platform := mkPlatformBackup()
delete(platform.Spec.DriverMetadata, cnpgBackupNameKey)
ready, _, err := r.cnpgBackupWALArchived(context.Background(), platform)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if ready {
t.Fatalf("expected not-ready when driverMetadata is missing the backup name")
}
})
}
// TestSetCNPGRestoreHRSuspended locks in the toggle helper used by the
// destructive purge to suspend helm-controller before mutating the
// Postgres app spec, and to resume it once the live Cluster + PVCs are
// gone. Without this guard helm-controller observes the bootstrap.enabled=
// true flip while the live Cluster still has bootstrap.initdb, attempts an
// upgrade that CNPG admission rejects, and rolls back into a fresh initdb
// cluster that races our purge.
func TestSetCNPGRestoreHRSuspended(t *testing.T) {
const (
ns = "tenant"
name = "postgres-app"
)
mkHR := func(suspended *bool) *unstructured.Unstructured {
hr := &unstructured.Unstructured{}
hr.SetGroupVersionKind(schema.GroupVersionKind{Group: "helm.toolkit.fluxcd.io", Version: "v2", Kind: "HelmRelease"})
hr.SetNamespace(ns)
hr.SetName(name)
spec := map[string]any{}
if suspended != nil {
spec["suspend"] = *suspended
}
_ = unstructured.SetNestedMap(hr.Object, spec, "spec")
return hr
}
suspendedField := func(t *testing.T, dyn dynamic.Interface) bool {
t.Helper()
hr, err := dyn.Resource(helmReleaseGVR).Namespace(ns).Get(context.Background(), name, metav1.GetOptions{})
if err != nil {
t.Fatalf("get HR: %v", err)
}
got, _, _ := unstructured.NestedBool(hr.Object, "spec", "suspend")
return got
}
t.Run("flips spec.suspend false -> true", func(t *testing.T) {
falsePtr := false
dyn := dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t), mkHR(&falsePtr))
r := &RestoreJobReconciler{Interface: dyn}
if err := r.setCNPGRestoreHRSuspended(context.Background(), ns, name, true); err != nil {
t.Fatalf("suspend: %v", err)
}
if !suspendedField(t, dyn) {
t.Fatalf("expected spec.suspend=true after suspend call")
}
})
t.Run("flips spec.suspend true -> false", func(t *testing.T) {
truePtr := true
dyn := dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t), mkHR(&truePtr))
r := &RestoreJobReconciler{Interface: dyn}
if err := r.setCNPGRestoreHRSuspended(context.Background(), ns, name, false); err != nil {
t.Fatalf("resume: %v", err)
}
if suspendedField(t, dyn) {
t.Fatalf("expected spec.suspend=false after resume call")
}
})
t.Run("idempotent when already at desired state", func(t *testing.T) {
// Calling suspend on an already-suspended HR (or resume on an
// already-resumed one) must be a no-op so retries from a
// flapping reconcile loop don't churn HR resourceVersions.
truePtr := true
dyn := dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t), mkHR(&truePtr))
r := &RestoreJobReconciler{Interface: dyn}
if err := r.setCNPGRestoreHRSuspended(context.Background(), ns, name, true); err != nil {
t.Fatalf("idempotent suspend: %v", err)
}
if !suspendedField(t, dyn) {
t.Fatalf("expected spec.suspend to stay true")
}
})
t.Run("missing HR is a no-op", func(t *testing.T) {
// A target Postgres app whose chart-rendered HR has not landed
// yet (or has been deleted) must not fail the destructive flow:
// the controller proceeds with patch+purge and the chart will
// later create a fresh HR from the patched values.
dyn := dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t))
r := &RestoreJobReconciler{Interface: dyn}
if err := r.setCNPGRestoreHRSuspended(context.Background(), ns, name, true); err != nil {
t.Fatalf("missing HR: %v", err)
}
})
}
// TestCNPGClusterFullyGone locks in the wait-for-purge helper that gates
// HR resume until the live Cluster + its PVCs have actually disappeared.
// Returning early on a cluster that is mid-deletion (DeletionTimestamp
// set, finalizers still draining) lets helm race the still-terminating CR
// and we end up back in the "Only one bootstrap" admission-rejection loop.
func TestCNPGClusterFullyGone(t *testing.T) {
const (
ns = "tenant"
name = "postgres-app"
)
now := metav1.Now()
t.Run("cluster + PVCs absent -> gone=true", func(t *testing.T) {
c := newCNPGStrategyTestClient(t)
r := &RestoreJobReconciler{Client: c}
gone, err := r.cnpgClusterFullyGone(context.Background(), ns, name)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !gone {
t.Fatalf("expected gone=true with no Cluster and no PVCs")
}
})
t.Run("live cluster -> gone=false", func(t *testing.T) {
cluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: name},
}
c := newCNPGStrategyTestClient(t, cluster)
r := &RestoreJobReconciler{Client: c}
gone, err := r.cnpgClusterFullyGone(context.Background(), ns, name)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if gone {
t.Fatalf("expected gone=false when cluster still exists")
}
})
t.Run("terminating cluster (DeletionTimestamp) -> gone=false", func(t *testing.T) {
cluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{
Namespace: ns,
Name: name,
DeletionTimestamp: &now,
Finalizers: []string{"cnpg.io/cluster"},
},
}
c := newCNPGStrategyTestClient(t, cluster)
r := &RestoreJobReconciler{Client: c}
gone, err := r.cnpgClusterFullyGone(context.Background(), ns, name)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if gone {
t.Fatalf("expected gone=false for terminating cluster")
}
})
t.Run("cluster gone but PVC remains -> gone=false", func(t *testing.T) {
// CNPG's PVCs are deleted alongside the Cluster but storage
// drivers (linstor, ceph) can hold them in a Terminating state
// for tens of seconds. Resuming the HR while a PVC with the
// cluster label still exists lets helm re-render and immediately
// re-attach to a leftover volume.
pvc := &corev1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{
Namespace: ns,
Name: name + "-1",
Labels: map[string]string{cnpgClusterLabel: name},
},
}
c := newCNPGStrategyTestClient(t, pvc)
r := &RestoreJobReconciler{Client: c}
gone, err := r.cnpgClusterFullyGone(context.Background(), ns, name)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if gone {
t.Fatalf("expected gone=false when a labelled PVC is still around")
}
})
}
func TestLogIndicatesRecoveryTargetUnreachable(t *testing.T) {
cases := []struct {
name string
log string
want bool
}{
{
name: "empty log",
log: "",
want: false,
},
{
name: "healthy replay progress",
log: "LOG: restored log file \"000000010000000000000005\" from archive\nLOG: consistent recovery state reached",
want: false,
},
{
name: "transient recovery-pod crash (API unreachable) does not match",
log: `{"level":"error","msg":"while building the manager","error":"failed to get server groups: Get \"https://10.96.0.1:443/api\": dial tcp 10.96.0.1:443: i/o timeout"}`,
want: false,
},
{
name: "target unreachable FATAL matches",
log: "LOG: redo done at 0/50000A8\nFATAL: recovery ended before configured recovery target was reached\nLOG: startup process exited with exit code 1",
want: true,
},
{
name: "target unreachable FATAL embedded in JSON log line matches",
log: `{"level":"info","record":{"error_severity":"FATAL","message":"recovery ended before configured recovery target was reached"}}`,
want: true,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := logIndicatesRecoveryTargetUnreachable(tc.log); got != tc.want {
t.Fatalf("logIndicatesRecoveryTargetUnreachable(%q) = %v, want %v", tc.log, got, tc.want)
}
})
}
}
// TestRecoveryUnreachableFromLogs: any recovery log carrying the unreachable-
// target FATAL classifies the (already deadline-expired) restore as
// target-unreachable; logs without it do not.
func TestRecoveryUnreachableFromLogs(t *testing.T) {
fatal := "LOG: redo done\nFATAL: " + cnpgRecoveryTargetUnreachableLog
healthy := "LOG: restored log file from archive\nLOG: consistent recovery state reached"
cases := []struct {
name string
logs []string
want bool
}{
{"no logs", nil, false},
{"only healthy/progress logs", []string{healthy, healthy}, false},
{"a FATAL among others", []string{healthy, fatal}, true},
{"all FATAL", []string{fatal, fatal}, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := recoveryUnreachableFromLogs(tc.logs); got != tc.want {
t.Fatalf("recoveryUnreachableFromLogs = %v, want %v", got, tc.want)
}
})
}
}
// TestRecoveryPodsToInspect pins the filter/order/cap contract of the pod
// selection the fail-fast guard relies on: only full-recovery pods, newest
// first, capped at cnpgRecoveryMaxInspectPods. A regression here (wrong sort
// direction, off-by-one cap, broken container filter) would silently degrade
// the fail-fast back into a 30-minute deadline hang.
func TestRecoveryPodsToInspect(t *testing.T) {
recoveryPod := func(name string, ageSeconds int64) corev1.Pod {
return corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: name,
CreationTimestamp: metav1.NewTime(time.Unix(ageSeconds, 0)),
},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: cnpgRecoveryContainerName}}},
}
}
t.Run("no pods -> empty", func(t *testing.T) {
if got := recoveryPodsToInspect(nil); len(got) != 0 {
t.Fatalf("expected empty, got %d", len(got))
}
})
t.Run("skips non-recovery pods (no full-recovery container)", func(t *testing.T) {
pods := []corev1.Pod{
recoveryPod("rec-1", 100),
{
ObjectMeta: metav1.ObjectMeta{Name: "primary-newest", CreationTimestamp: metav1.NewTime(time.Unix(999, 0))},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "postgres"}}},
},
}
got := recoveryPodsToInspect(pods)
if len(got) != 1 || got[0].Name != "rec-1" {
t.Fatalf("expected only the full-recovery pod, got %v", names(got))
}
})
t.Run("newest-first order and cap at cnpgRecoveryMaxInspectPods", func(t *testing.T) {
// Seed two more recovery pods than the cap, out of order, plus a
// non-recovery pod that is the newest of all (must be excluded). The
// result must be exactly the cap-many newest recovery pods, newest-first.
n := cnpgRecoveryMaxInspectPods + 2
var pods []corev1.Pod
for i := range n {
// age i*100 so higher i == newer.
pods = append(pods, recoveryPod(fmt.Sprintf("rec-%d", i), int64(i*100)))
}
pods = append(pods, corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "primary", CreationTimestamp: metav1.NewTime(time.Unix(9_999_999, 0))},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "postgres"}}},
})
got := recoveryPodsToInspect(pods)
want := make([]string, 0, cnpgRecoveryMaxInspectPods)
for i := range cnpgRecoveryMaxInspectPods {
want = append(want, fmt.Sprintf("rec-%d", n-1-i)) // newest-first
}
if gotNames := names(got); !equalStrings(gotNames, want) {
t.Fatalf("expected %v, got %v", want, gotNames)
}
})
}
func names(pods []*corev1.Pod) []string {
out := make([]string, len(pods))
for i, p := range pods {
out[i] = p.Name
}
return out
}
func equalStrings(a, b []string) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}
// recoveryTargetUnreachable is nil-safe when no Clientset is wired, so the
// deadline stays the backstop instead of the driver panicking.
func TestRecoveryTargetUnreachable_NilClientsetReportsFalse(t *testing.T) {
r := &RestoreJobReconciler{Client: newCNPGStrategyTestClient(t)}
unreachable, pod, forbidden, err := r.recoveryTargetUnreachable(context.Background(), "tenant", "postgres-app")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if unreachable || pod != "" || forbidden {
t.Fatalf("expected (false, \"\", false) with no Clientset, got (%v, %q, %v)", unreachable, pod, forbidden)
}
}
// TestRecoveryTargetUnreachable_ClassifiesFromLogs drives the log-based
// classification through an injected reader (the fake Clientset's GetLogs
// can't return specific content or a Forbidden), covering the three outcomes
// the deadline path branches on: the unreachable-target FATAL is present
// (-> unreachable, newest pod), it is absent (-> not unreachable), and the log
// read is Forbidden (-> forbidden flagged so the caller surfaces the missing
// pods/log RBAC instead of failing silently-generic).
func TestRecoveryTargetUnreachable_ClassifiesFromLogs(t *testing.T) {
const (
ns = "tenant"
cluster = "postgres-app"
)
recPod := func(name string, ageSeconds int64) client.Object {
return &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Namespace: ns,
Name: name,
Labels: map[string]string{cnpgClusterLabel: cluster},
CreationTimestamp: metav1.NewTime(time.Unix(ageSeconds, 0)),
},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: cnpgRecoveryContainerName}}},
}
}
seed := func() []client.Object { return []client.Object{recPod("rec-old", 100), recPod("rec-new", 200)} }
fatalLog := "LOG: redo done\nFATAL: " + cnpgRecoveryTargetUnreachableLog
t.Run("FATAL present -> unreachable, reports newest pod", func(t *testing.T) {
r := &RestoreJobReconciler{
Client: newCNPGStrategyTestClient(t, seed()...),
readPodLog: func(context.Context, string, string, string) (string, error) { return fatalLog, nil },
}
un, pod, forb, err := r.recoveryTargetUnreachable(context.Background(), ns, cluster)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !un || forb || pod != "rec-new" {
t.Fatalf("got unreachable=%v pod=%q forbidden=%v, want (true, rec-new, false)", un, pod, forb)
}
})
t.Run("no FATAL -> not unreachable", func(t *testing.T) {
r := &RestoreJobReconciler{
Client: newCNPGStrategyTestClient(t, seed()...),
readPodLog: func(context.Context, string, string, string) (string, error) {
return "LOG: consistent recovery state reached", nil
},
}
un, _, forb, err := r.recoveryTargetUnreachable(context.Background(), ns, cluster)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if un || forb {
t.Fatalf("got unreachable=%v forbidden=%v, want (false, false)", un, forb)
}
})
t.Run("Forbidden log read -> forbidden flagged, not unreachable", func(t *testing.T) {
r := &RestoreJobReconciler{
Client: newCNPGStrategyTestClient(t, seed()...),
readPodLog: func(context.Context, string, string, string) (string, error) {
return "", apierrors.NewForbidden(schema.GroupResource{Resource: "pods"}, "rec-new", fmt.Errorf("pods/log grant missing"))
},
}
un, _, forb, err := r.recoveryTargetUnreachable(context.Background(), ns, cluster)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if un || !forb {
t.Fatalf("got unreachable=%v forbidden=%v, want (false, true)", un, forb)
}
})
}
// TestApplyClusterPluginBackup_ReadsTheLiveClusterNotTheCache seeds the cached
// and the live client with the Cluster before and after an in-place restore
// re-rendered it, and asserts the driver takes both the serverName and the
// ObjectStore owner UID from the live one.
func TestApplyClusterPluginBackup_ReadsTheLiveClusterNotTheCache(t *testing.T) {
archiver := true
purged := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "postgres-app", UID: "uid-before-restore"},
}
purged.Spec.Plugins = []cnpgtypes.PluginConfiguration{{
Name: cnpgtypes.PluginName,
IsWALArchiver: &archiver,
Parameters: map[string]string{barmanObjectNameParam: "postgres-app", barmanServerNameParam: "postgres-app"},
}}
rebootstrapped := purged.DeepCopy()
rebootstrapped.UID = "uid-after-restore"
rebootstrapped.Spec.Plugins[0].Parameters[barmanServerNameParam] = "postgres-app-restore-0123456789abcdef"
c := newCNPGStrategyTestClient(t, purged)
r := &BackupJobReconciler{Client: c, Interface: cnpgDynamicFor(t, rebootstrapped)}
tmpl := &strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/x/"},
}
got, err := r.applyClusterPluginBackup(context.Background(), "tenant", "postgres-app", tmpl, "strategy-name", true)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if want := "postgres-app-restore-0123456789abcdef"; got.serverName != want {
t.Fatalf("effective serverName = %q, want %q: the restored cluster would archive onto the source's prefix", got.serverName, want)
}
store := &cnpgtypes.ObjectStore{}
if err := c.Get(context.Background(), client.ObjectKey{Namespace: "tenant", Name: "postgres-app"}, store); err != nil {
t.Fatalf("get ObjectStore after apply: %v", err)
}
owners := store.GetOwnerReferences()
if len(owners) != 1 {
t.Fatalf("ObjectStore owner references = %v, want exactly the live Cluster", owners)
}
if got := string(owners[0].UID); got != "uid-after-restore" {
t.Fatalf("ObjectStore owned by UID %q, want the live Cluster's: it would be garbage-collected immediately", got)
}
}
// TestReconcileCNPG_RecordsTheLiveClusterFlavor: the chart refuses a flavor
// change on a live Cluster, but the Postgres app keeps the refused value. The
// Backup must record the flavor the Cluster actually runs, or a restore would
// accept a target of the wrong image family and refuse the right one.
func TestReconcileCNPG_RecordsTheLiveClusterFlavor(t *testing.T) {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
const ns, appName = "tenant", "pg"
clusterName := cnpgClusterNameForApp(appName)
cases := []struct {
name string
liveImage string
specFlavor string
want string
}{
{name: "app switched to postgis on a postgresql cluster", liveImage: "ghcr.io/cloudnative-pg/postgresql:17.5", specFlavor: "postgis", want: "postgresql"},
{name: "app switched to postgresql on a postgis cluster", liveImage: "ghcr.io/cloudnative-pg/postgis:17-3.5", specFlavor: "", want: "postgis"},
{name: "cluster on the CNPG default image", liveImage: "", specFlavor: "postgis", want: "postgresql"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
startedAt := metav1.Now()
job := &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bj"},
Spec: backupsv1alpha1.BackupJobSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
},
Status: backupsv1alpha1.BackupJobStatus{StartedAt: &startedAt},
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{
Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
},
},
}
app := newPostgresApp(appName, ns)
app.Spec.Flavor = tc.specFlavor
cluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, UID: "cluster-uid"},
Spec: cnpgtypes.ClusterSpec{ImageName: tc.liveImage},
}
cnpgBackup := &cnpgtypes.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bj-cnpg", Labels: map[string]string{
backupsv1alpha1.OwningJobNameLabel: job.Name,
backupsv1alpha1.OwningJobNamespaceLabel: job.Namespace,
}},
Spec: cnpgtypes.BackupSpec{Cluster: cnpgtypes.ClusterReference{Name: clusterName}},
Status: cnpgtypes.BackupStatus{Phase: cnpgBackupPhaseComplete},
}
c := newCNPGStrategyTestClient(t, job, strategy, app, cluster, cnpgBackup)
r := &BackupJobReconciler{Client: c, Interface: cnpgDynamicFor(t, cluster)}
resolved := &ResolvedBackupConfig{
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: strategy.Name},
}
j := &backupsv1alpha1.BackupJob{}
if err := c.Get(context.Background(), client.ObjectKeyFromObject(job), j); err != nil {
t.Fatalf("get BackupJob: %v", err)
}
if _, err := r.reconcileCNPG(context.Background(), j, resolved); err != nil {
t.Fatalf("reconcileCNPG: %v", err)
}
if j.Status.Phase != backupsv1alpha1.BackupJobPhaseSucceeded {
t.Fatalf("BackupJob phase = %q (%s), want Succeeded", j.Status.Phase, j.Status.Message)
}
backup := &backupsv1alpha1.Backup{}
if err := c.Get(context.Background(), client.ObjectKey{Namespace: ns, Name: job.Name}, backup); err != nil {
t.Fatalf("get Backup artifact: %v", err)
}
snap, err := unmarshalCNPGBackupSnapshot(backup)
if err != nil {
t.Fatalf("unmarshal snapshot: %v", err)
}
if got := cnpgFlavor(snap.Flavor); got != tc.want {
t.Fatalf("Backup records flavor %q, want the live Cluster's %q", got, tc.want)
}
})
}
}
// TestCNPGImageFlavor pins the image-to-flavor mapping to the chart's
// postgres.flavorGuard: only the image name counts, never the host, tag or
// digest.
func TestCNPGImageFlavor(t *testing.T) {
cases := map[string]string{
"": "postgresql",
"ghcr.io/cloudnative-pg/postgresql:18.1": "postgresql",
"ghcr.io/cloudnative-pg/postgis:18-3.6": "postgis",
"mirror.local:5000/cloudnative-pg/postgis:18-3.6@sha256:0123": "postgis",
"postgis@sha256:0123": "postgis",
}
for image, want := range cases {
if got := cnpgImageFlavor(image); got != want {
t.Errorf("cnpgImageFlavor(%q) = %q, want %q", image, got, want)
}
}
}
// cnpgDynamicFor builds the dynamic client applyClusterPluginBackup reads
// Clusters through, seeded independently of the typed client.
func cnpgDynamicFor(t *testing.T, clusters ...*cnpgtypes.Cluster) dynamic.Interface {
t.Helper()
s := runtime.NewScheme()
s.AddKnownTypeWithName(cnpgtypes.GroupVersion.WithKind("Cluster"), &unstructured.Unstructured{})
s.AddKnownTypeWithName(cnpgtypes.GroupVersion.WithKind("ClusterList"), &unstructured.UnstructuredList{})
s.AddKnownTypeWithName(cnpgtypes.BarmanGroupVersion.WithKind("ObjectStore"), &unstructured.Unstructured{})
s.AddKnownTypeWithName(cnpgtypes.BarmanGroupVersion.WithKind("ObjectStoreList"), &unstructured.UnstructuredList{})
objs := make([]runtime.Object, 0, len(clusters))
for _, c := range clusters {
raw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(c)
if err != nil {
t.Fatalf("converting Cluster %s: %v", c.Name, err)
}
u := &unstructured.Unstructured{Object: raw}
u.SetAPIVersion(cnpgtypes.GroupVersion.String())
u.SetKind("Cluster")
objs = append(objs, u)
}
return dynamicfake.NewSimpleDynamicClient(s, objs...)
}
// testCNPGScheme returns a runtime.Scheme that knows the unstructured
// HelmRelease GVK used by the dynamic-client tests above.
func testCNPGScheme(t *testing.T) *runtime.Scheme {
t.Helper()
s := runtime.NewScheme()
_ = scheme.AddToScheme(s)
s.AddKnownTypeWithName(schema.GroupVersionKind{Group: "helm.toolkit.fluxcd.io", Version: "v2", Kind: "HelmRelease"}, &unstructured.Unstructured{})
s.AddKnownTypeWithName(schema.GroupVersionKind{Group: "helm.toolkit.fluxcd.io", Version: "v2", Kind: "HelmReleaseList"}, &unstructured.UnstructuredList{})
return s
}
// TestApplyClusterPluginBackup_PatchesExistingCluster confirms the happy path:
// when the Cluster exists the driver SSA-applies an ObjectStore CR carrying the
// The serverName is the S3 WAL-archive path prefix; changing it on a live
// Cluster splits the archive across two prefixes and breaks the eventual
// restore with "WAL not found". A Cluster that already has the barman-cloud
// plugin attached must therefore keep its effective serverName — the explicit
// parameter when present, its own name (the plugin's default) when the plugin
// is attached without one — and only a Cluster with no plugin at all takes the
// strategy template's value.
func TestApplyClusterPluginBackup_PreservesLiveServerName(t *testing.T) {
explicit := true
cases := []struct {
name string
plugins []cnpgtypes.PluginConfiguration
wantServer string
}{
{
name: "explicit live serverName wins over the strategy's",
plugins: []cnpgtypes.PluginConfiguration{{
Name: cnpgtypes.PluginName,
IsWALArchiver: &explicit,
Parameters: map[string]string{barmanObjectNameParam: "postgres-app", barmanServerNameParam: "postgres-app"},
}},
wantServer: "postgres-app",
},
{
name: "plugin attached without serverName defaults to the cluster name",
plugins: []cnpgtypes.PluginConfiguration{{
Name: cnpgtypes.PluginName,
IsWALArchiver: &explicit,
Parameters: map[string]string{barmanObjectNameParam: "postgres-app"},
}},
wantServer: "postgres-app",
},
{
name: "no plugin attached: the strategy's serverName applies",
plugins: nil,
wantServer: "strategy-name",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
cluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "postgres-app", UID: "cluster-uid-123"},
}
cluster.Spec.Plugins = tc.plugins
c := newCNPGStrategyTestClient(t, cluster)
r := &BackupJobReconciler{Client: c, Interface: cnpgDynamicFor(t, cluster)}
tmpl := &strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/x/"},
}
got, err := r.applyClusterPluginBackup(context.Background(), "tenant", "postgres-app", tmpl, "strategy-name", true)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got.serverName != tc.wantServer {
t.Fatalf("effective serverName = %q, want %q", got.serverName, tc.wantServer)
}
patched := &cnpgtypes.Cluster{}
if err := c.Get(context.Background(), client.ObjectKey{Namespace: "tenant", Name: "postgres-app"}, patched); err != nil {
t.Fatalf("get Cluster after apply: %v", err)
}
if len(patched.Spec.Plugins) != 1 {
t.Fatalf("expected exactly one plugin entry, got %+v", patched.Spec.Plugins)
}
if s := patched.Spec.Plugins[0].Parameters[barmanServerNameParam]; s != tc.wantServer {
t.Fatalf("patched plugin serverName = %q, want %q", s, tc.wantServer)
}
})
}
}
// defaultStrategyServerName is the serverName the cozy-default-cnpg Strategy
// renders; tests/strategy_cnpg_test.yaml in the backupstrategy-controller chart
// pins the chart to this same string.
const defaultStrategyServerName = "{{ .Application.metadata.namespace }}-{{ .Application.metadata.name }}{{ with .Application.metadata.uid }}-{{ . }}{{ end }}"
func TestRenderCNPGTemplate_DefaultServerName(t *testing.T) {
tmpl := strategyv1alpha1.CNPGTemplate{ServerName: defaultStrategyServerName}
for _, tc := range []struct {
name string
uid types.UID
want string
}{
{name: "uid present", uid: "8f14e45f-ceea-467a-9575-6f0d6e2c1a3b", want: "tenant-pg-8f14e45f-ceea-467a-9575-6f0d6e2c1a3b"},
{name: "uid absent", want: "tenant-pg"},
} {
t.Run(tc.name, func(t *testing.T) {
app := newPostgresApp("pg", "tenant")
app.UID = tc.uid
got, err := renderCNPGTemplate(tmpl, app, nil)
if err != nil {
t.Fatalf("renderCNPGTemplate: %v", err)
}
if got.ServerName != tc.want {
t.Fatalf("serverName = %q, want %q", got.ServerName, tc.want)
}
})
}
}
// TestReconcileCNPG_RecreatedAppArchivesUnderAFreshPrefix deletes a Postgres
// and recreates it under the same name, then runs a BackupJob against each.
// The destinationPath is keyed by name, so only the serverName keeps the new
// Cluster from writing its WALs and base backups into the old one's archive;
// the empty-archive check that would refuse it is gone by the time the first
// BackupJob attaches the plugin. A Cluster that already archives keeps its
// serverName, so existing archives stay continuous.
func TestReconcileCNPG_RecreatedAppArchivesUnderAFreshPrefix(t *testing.T) {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
archiver := true
backUp := func(t *testing.T, strategyServerName string, appUID types.UID, plugins []cnpgtypes.PluginConfiguration) string {
t.Helper()
started := metav1.NewTime(time.Now())
job := &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
Spec: backupsv1alpha1.BackupJobSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: "pg"},
},
Status: backupsv1alpha1.BackupJobStatus{StartedAt: &started},
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cozy-default-cnpg"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
ServerName: strategyServerName,
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://bucket/{{ .Application.metadata.namespace }}/{{ .Application.metadata.name }}/",
},
}},
}
app := newPostgresApp("pg", "tenant")
app.UID = appUID
cluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "postgres-pg", UID: types.UID("cluster-" + appUID)},
}
cluster.Spec.Plugins = plugins
c := newCNPGStrategyTestClient(t, job, strategy, app, cluster)
r := &BackupJobReconciler{Client: c, Interface: cnpgDynamicFor(t, cluster)}
resolved := &ResolvedBackupConfig{StrategyRef: corev1.TypedLocalObjectReference{
APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: strategy.Name,
}}
// A Cluster without the plugin takes two passes: the first only
// records that the driver is about to roll it, the second attaches.
patched := &cnpgtypes.Cluster{}
for range 2 {
j := &backupsv1alpha1.BackupJob{}
if err := c.Get(context.Background(), client.ObjectKeyFromObject(job), j); err != nil {
t.Fatalf("get BackupJob: %v", err)
}
if _, err := r.reconcileCNPG(context.Background(), j, resolved); err != nil {
t.Fatalf("reconcileCNPG: %v", err)
}
if j.Status.Phase == backupsv1alpha1.BackupJobPhaseFailed {
t.Fatalf("BackupJob failed: %+v", j.Status.Conditions)
}
if err := c.Get(context.Background(), client.ObjectKeyFromObject(cluster), patched); err != nil {
t.Fatalf("get Cluster: %v", err)
}
if got := currentBarmanServerName(patched); got != "" {
return got
}
}
t.Fatalf("BackupJob left the barman-cloud plugin unattached: %+v", patched.Spec.Plugins)
return ""
}
for _, tc := range []struct {
name string
strategyServerName string
}{
{name: "default strategy", strategyServerName: defaultStrategyServerName},
{name: "strategy without serverName", strategyServerName: ""},
} {
t.Run(tc.name, func(t *testing.T) {
first := backUp(t, tc.strategyServerName, "3c2ee41b-346f-4ae2-98d2-5d629c96f744", nil)
second := backUp(t, tc.strategyServerName, "9eb8feed-7d49-467e-b719-d1896c0602d5", nil)
if first == second {
t.Fatalf("both instances of tenant/pg archive under serverName %q: the recreated one writes into its predecessor's archive", first)
}
for _, name := range []string{first, second} {
if strings.Contains(name, "<no value>") {
t.Fatalf("serverName %q rendered a missing key", name)
}
}
const legacy = "tenant-pg"
kept := backUp(t, tc.strategyServerName, "9eb8feed-7d49-467e-b719-d1896c0602d5", []cnpgtypes.PluginConfiguration{{
Name: cnpgtypes.PluginName,
IsWALArchiver: &archiver,
Parameters: map[string]string{barmanObjectNameParam: "postgres-pg", barmanServerNameParam: legacy},
}})
if kept != legacy {
t.Fatalf("a Cluster already archiving under %q moved to %q, splitting its archive", legacy, kept)
}
})
}
}
// barman configuration and SSA-patches the Cluster's spec.plugins to reference
// it via the barman-cloud plugin (the plugin replaces the deprecated native
// spec.backup.barmanObjectStore). serverName must land on the Cluster plugin
// parameter, never inside the ObjectStore configuration (the plugin forbids it).
func TestApplyClusterPluginBackup_PatchesExistingCluster(t *testing.T) {
cluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "postgres-app", UID: "cluster-uid-123"},
}
c := newCNPGStrategyTestClient(t, cluster)
r := &BackupJobReconciler{Client: c, Interface: cnpgDynamicFor(t, cluster)}
tmpl := &strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://bucket/x/",
RetentionPolicy: "30d",
},
}
if _, err := r.applyClusterPluginBackup(context.Background(), "tenant", "postgres-app", tmpl, "tenant-app", true); err != nil {
t.Fatalf("unexpected error: %v", err)
}
// The ObjectStore CR carries destinationPath + retentionPolicy and no serverName.
store := &cnpgtypes.ObjectStore{}
if err := c.Get(context.Background(), client.ObjectKey{Namespace: "tenant", Name: "postgres-app"}, store); err != nil {
t.Fatalf("get ObjectStore after apply: %v", err)
}
if store.Spec.Configuration.DestinationPath != "s3://bucket/x/" {
t.Errorf("ObjectStore destinationPath: got %q", store.Spec.Configuration.DestinationPath)
}
if store.Spec.RetentionPolicy != "30d" {
t.Errorf("ObjectStore retentionPolicy: got %q", store.Spec.RetentionPolicy)
}
// The ObjectStore must pin the barman-cloud sidecar's S3 request-checksum
// policy to when_required, or uploads to non-AWS S3 gateways (Ceph RGW, the
// platform's own SeaweedFS system bucket) fail with an x-amz-content-sha256
// InvalidArgument. The chart does not render this ObjectStore in the
// platform-managed useSystemBucket=true flow, so this Go path is the only
// place that sets it there.
sc := store.Spec.InstanceSidecarConfiguration
if sc == nil {
t.Fatal("ObjectStore has no instanceSidecarConfiguration")
}
if len(sc.Env) != 1 ||
sc.Env[0].Name != "AWS_REQUEST_CHECKSUM_CALCULATION" || sc.Env[0].Value != "when_required" {
t.Errorf("ObjectStore instanceSidecarConfiguration.env: got %+v, want [AWS_REQUEST_CHECKSUM_CALCULATION=when_required]", sc)
}
if sc.Resources.Requests.Memory().Value() != 256*1024*1024 ||
sc.Resources.Limits.Memory().Value() != 1024*1024*1024 ||
sc.Resources.Requests.Cpu().MilliValue() != 100 {
t.Errorf("ObjectStore leaves Barman subject to undersized tenant resource defaults: %+v", sc.Resources)
}
if _, ok := sc.Resources.Limits[corev1.ResourceCPU]; ok {
t.Errorf("ObjectStore caps sidecar CPU, which throttles WAL archiving instead of failing it: %+v", sc.Resources.Limits)
}
// The ObjectStore must be owner-referenced to the Cluster so Kubernetes GC
// removes it when the Cluster is deleted (no orphan in the platform flow,
// where the chart does not render this ObjectStore).
if len(store.OwnerReferences) != 1 {
t.Fatalf("expected exactly one ownerReference on the ObjectStore, got %+v", store.OwnerReferences)
}
own := store.OwnerReferences[0]
if own.Kind != "Cluster" || own.Name != "postgres-app" || own.UID != "cluster-uid-123" {
t.Errorf("ObjectStore ownerReference: got kind=%q name=%q uid=%q, want Cluster/postgres-app/cluster-uid-123", own.Kind, own.Name, own.UID)
}
if store.Spec.Configuration.ServerName != "" {
t.Errorf("ObjectStore configuration must not carry serverName (plugin forbids it); got %q", store.Spec.Configuration.ServerName)
}
// The Cluster references the ObjectStore through spec.plugins, with
// serverName on the plugin parameter and isWALArchiver=true.
got := &cnpgtypes.Cluster{}
if err := c.Get(context.Background(), client.ObjectKeyFromObject(cluster), got); err != nil {
t.Fatalf("get cluster after patch: %v", err)
}
if len(got.Spec.Plugins) != 1 {
t.Fatalf("expected exactly one plugin entry, got %+v", got.Spec.Plugins)
}
p := got.Spec.Plugins[0]
if p.Name != cnpgtypes.PluginName {
t.Errorf("plugin name: got %q, want %q", p.Name, cnpgtypes.PluginName)
}
if p.IsWALArchiver == nil || !*p.IsWALArchiver {
t.Errorf("expected isWALArchiver=true, got %v", p.IsWALArchiver)
}
if got := p.Parameters[barmanObjectNameParam]; got != "postgres-app" {
t.Errorf("plugin barmanObjectName: got %q, want %q", got, "postgres-app")
}
if got := p.Parameters[barmanServerNameParam]; got != "tenant-app" {
t.Errorf("plugin serverName: got %q, want %q", got, "tenant-app")
}
}
// TestEnsureCNPGBackup_UsesPluginMethod locks in that driver-initiated backups
// go through the barman-cloud plugin. spec.method / spec.pluginConfiguration is
// the single field that routes the run: if it regressed to the default
// (barmanObjectStore) method, CNPG would run the legacy method against a Cluster
// that no longer has one and every platform backup would fail.
func TestEnsureCNPGBackup_UsesPluginMethod(t *testing.T) {
j := &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
}
c := newCNPGStrategyTestClient(t, j)
r := &BackupJobReconciler{Client: c}
bk, err := r.ensureCNPGBackup(context.Background(), j, "postgres-app")
if err != nil {
t.Fatalf("ensureCNPGBackup: %v", err)
}
if bk.Spec.Method != cnpgtypes.BackupMethodPlugin {
t.Errorf("Backup spec.method: got %q, want %q", bk.Spec.Method, cnpgtypes.BackupMethodPlugin)
}
if bk.Spec.PluginConfiguration == nil || bk.Spec.PluginConfiguration.Name != cnpgtypes.PluginName {
t.Errorf("Backup spec.pluginConfiguration: got %+v, want name=%q", bk.Spec.PluginConfiguration, cnpgtypes.PluginName)
}
if bk.Spec.Cluster.Name != "postgres-app" {
t.Errorf("Backup spec.cluster.name: got %q, want postgres-app", bk.Spec.Cluster.Name)
}
}
// TestReconcileCNPG_StartedAtPreservedAcrossStaleCache locks in the bug fix
// for the StartedAt race called out in the branch review: a second reconcile
// with a stale local copy (which observes StartedAt==nil) must NOT advance
// the persisted timestamp the first reconcile already wrote. Without the
// refetch-before-write guard, the wall-clock deadline gate slides forward
// on every poll, defeating the purpose of cnpgDefaultBackupDeadline.
func TestReconcileCNPG_StartedAtPreservedAcrossStaleCache(t *testing.T) {
// metav1.Time loses sub-second precision on JSON round-trip; truncate
// upfront so the equality check below isn't fooled by nanosecond drift.
original := metav1.NewTime(time.Now().Add(-10 * time.Minute).Truncate(time.Second))
stored := &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
Status: backupsv1alpha1.BackupJobStatus{
StartedAt: &original,
},
}
c := newCNPGStrategyTestClient(t, stored)
r := &BackupJobReconciler{Client: c}
// Stale local copy as the reconcile would have observed it: StartedAt==nil.
stale := stored.DeepCopy()
stale.Status.StartedAt = nil
// Inline the refetch-before-write block from reconcileCNPG. (The full
// reconciler path requires more wiring than this lightweight test
// builds; the block under test is the bug-fix surface, not the
// surrounding decode/template logic.)
fresh := &backupsv1alpha1.BackupJob{}
if err := r.Get(context.Background(), client.ObjectKeyFromObject(stored), fresh); err != nil {
t.Fatalf("get fresh: %v", err)
}
if fresh.Status.StartedAt != nil {
stale.Status.StartedAt = fresh.Status.StartedAt
} else {
base := fresh.DeepCopy()
now := metav1.Now()
fresh.Status.StartedAt = &now
if err := r.Status().Patch(context.Background(), fresh, client.MergeFrom(base)); err != nil {
t.Fatalf("patch: %v", err)
}
stale.Status.StartedAt = fresh.Status.StartedAt
}
got := &backupsv1alpha1.BackupJob{}
if err := c.Get(context.Background(), client.ObjectKeyFromObject(stored), got); err != nil {
t.Fatalf("get after fix: %v", err)
}
if got.Status.StartedAt == nil || !got.Status.StartedAt.Equal(&original) {
t.Errorf("persisted StartedAt was advanced by stale-cache reconcile; got %v, want %v",
got.Status.StartedAt, original)
}
if stale.Status.StartedAt == nil || !stale.Status.StartedAt.Equal(&original) {
t.Errorf("local copy did not pick up persisted StartedAt; got %v, want %v",
stale.Status.StartedAt, original)
}
}
// TestReconcileCNPG_StartedAtReturnsEarlyToAvoidStaleResourceVersion locks
// in the early-return after the StartedAt patch. Before the fix, reconcileCNPG
// patched StartedAt onto a refetched copy and then continued the same call
// using the local `j` whose ResourceVersion was stale, so any subsequent
// r.Status().Update on `j` (e.g. the markBackupJobFailed path or the
// ClusterNotReady condition write) carried a pre-patch RV and produced a
// Conflict on the very first poll. The fix returns RequeueAfter immediately
// after the StartedAt patch so the next reconcile reads `j` cleanly.
func TestReconcileCNPG_StartedAtReturnsEarlyToAvoidStaleResourceVersion(t *testing.T) {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
stored := &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
Spec: backupsv1alpha1.BackupJobSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: "pg"},
},
}
c := newCNPGStrategyTestClient(t, stored)
r := &BackupJobReconciler{Client: c}
j := stored.DeepCopy()
resolved := &ResolvedBackupConfig{
StrategyRef: corev1.TypedLocalObjectReference{
APIGroup: &strategyGroup,
Kind: strategyv1alpha1.CNPGStrategyKind,
Name: "missing-strategy",
},
}
result, err := r.reconcileCNPG(context.Background(), j, resolved)
if err != nil {
t.Fatalf("reconcileCNPG: unexpected error %v", err)
}
if result.RequeueAfter == 0 {
t.Fatalf("expected RequeueAfter > 0 after StartedAt patch, got %+v", result)
}
persisted := &backupsv1alpha1.BackupJob{}
if err := c.Get(context.Background(), client.ObjectKeyFromObject(stored), persisted); err != nil {
t.Fatalf("get persisted: %v", err)
}
if persisted.Status.StartedAt == nil {
t.Errorf("expected StartedAt persisted, got nil")
}
// Without the early return, reconcileCNPG would have proceeded past the
// StartedAt block, found the strategy missing, and called
// markBackupJobFailed - persisting Phase=Failed (and, depending on RV
// enforcement, returning Conflict). With the fix, Phase stays empty and
// the next reconcile picks up cleanly with the post-patch RV.
if persisted.Status.Phase != "" {
t.Errorf("expected Phase unchanged after StartedAt early return, got %q", persisted.Status.Phase)
}
}
// cnpgRolloutFixture drives reconcileCNPG against one Postgres app whose
// Cluster and instance pods a test swaps between passes, standing in for
// CNPG rolling the instances.
type cnpgRolloutFixture struct {
t *testing.T
c client.Client
r *BackupJobReconciler
resolved *ResolvedBackupConfig
}
const (
rolloutNS = "tenant"
rolloutApp = "app"
rolloutCluster = "postgres-app"
)
func newCNPGRolloutFixture(t *testing.T, j *backupsv1alpha1.BackupJob, cluster *cnpgtypes.Cluster, pods ...*corev1.Pod) *cnpgRolloutFixture {
t.Helper()
strategyGroup := strategyv1alpha1.GroupVersion.Group
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
}},
}
objs := []client.Object{j, strategy, newPostgresApp(rolloutApp, rolloutNS), cluster}
for _, p := range pods {
objs = append(objs, p)
}
c := newCNPGStrategyTestClient(t, objs...)
return &cnpgRolloutFixture{
t: t,
c: c,
r: &BackupJobReconciler{Client: c, Interface: cnpgDynamicFor(t, cluster)},
resolved: &ResolvedBackupConfig{StrategyRef: corev1.TypedLocalObjectReference{
APIGroup: &strategyGroup,
Kind: strategyv1alpha1.CNPGStrategyKind,
Name: strategy.Name,
}},
}
}
func newRolloutBackupJob(startedAt time.Time) *backupsv1alpha1.BackupJob {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
started := metav1.NewTime(startedAt)
return &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: rolloutNS, Name: "bj"},
Spec: backupsv1alpha1.BackupJobSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: rolloutApp},
},
Status: backupsv1alpha1.BackupJobStatus{StartedAt: &started},
}
}
func newRolloutCluster(phase string, plugins ...cnpgtypes.PluginConfiguration) *cnpgtypes.Cluster {
return &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: rolloutNS, Name: rolloutCluster, UID: "cluster-uid"},
Spec: cnpgtypes.ClusterSpec{Plugins: plugins},
Status: cnpgtypes.ClusterStatus{Phase: phase},
}
}
func newRolloutInstancePod(created time.Time) *corev1.Pod {
return &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Namespace: rolloutNS,
Name: rolloutCluster + "-1",
CreationTimestamp: metav1.NewTime(created),
Labels: map[string]string{cnpgClusterLabel: rolloutCluster, cnpgPodRoleLabel: cnpgPodRoleInstance},
},
Status: corev1.PodStatus{Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}},
}
}
// pass runs one reconcile and returns the persisted BackupJob and the
// cnpg.io/Backups that exist afterwards.
func (f *cnpgRolloutFixture) pass() (*backupsv1alpha1.BackupJob, []cnpgtypes.Backup) {
f.t.Helper()
ctx := context.Background()
key := client.ObjectKey{Namespace: rolloutNS, Name: "bj"}
j := &backupsv1alpha1.BackupJob{}
if err := f.c.Get(ctx, key, j); err != nil {
f.t.Fatalf("get BackupJob: %v", err)
}
if _, err := f.r.reconcileCNPG(ctx, j, f.resolved); err != nil {
f.t.Fatalf("reconcileCNPG: %v", err)
}
got := &backupsv1alpha1.BackupJob{}
if err := f.c.Get(ctx, key, got); err != nil {
f.t.Fatalf("get BackupJob after reconcile: %v", err)
}
backups := &cnpgtypes.BackupList{}
if err := f.c.List(ctx, backups, client.InNamespace(rolloutNS)); err != nil {
f.t.Fatalf("list cnpg.io/Backups: %v", err)
}
return got, backups.Items
}
// observeCNPG makes the next live read see cluster, as CNPG has written it.
func (f *cnpgRolloutFixture) observeCNPG(cluster *cnpgtypes.Cluster) {
f.r.Interface = cnpgDynamicFor(f.t, cluster)
}
func (f *cnpgRolloutFixture) replacePod(old, recreated *corev1.Pod) {
f.t.Helper()
if err := f.c.Delete(context.Background(), old); err != nil {
f.t.Fatalf("delete pod: %v", err)
}
if err := f.c.Create(context.Background(), recreated); err != nil {
f.t.Fatalf("create pod: %v", err)
}
}
// TestReconcileCNPG_WaitsForPluginRolloutBeforeBackup covers the first backup
// of a Cluster that had no barman-cloud plugin. Attaching it makes CNPG roll
// every instance, and a cnpg.io/Backup created before the rollout is over
// fails for good ("requested plugin is not available"). No Backup may exist
// until the instances have been recreated and CNPG reports the Cluster
// healthy again, including in the window where CNPG has not reconciled the
// new spec yet and the phase still reads healthy from before.
func TestReconcileCNPG_WaitsForPluginRolloutBeforeBackup(t *testing.T) {
oldPod := newRolloutInstancePod(time.Now().Add(-time.Hour).Truncate(time.Second))
f := newCNPGRolloutFixture(t, newRolloutBackupJob(time.Now().Add(-time.Minute)),
newRolloutCluster(cnpgClusterHealthyPhase), oldPod)
j, backups := f.pass()
if len(backups) != 0 {
t.Fatalf("a cnpg.io/Backup was created before the plugin was even attached: %+v", backups)
}
if cond := apimeta.FindStatusCondition(j.Status.Conditions, backupCondPluginRolledOut); cond == nil || cond.Status != metav1.ConditionFalse {
t.Fatalf("%s condition = %+v, want False before the plugin is attached", backupCondPluginRolledOut, cond)
}
if cond := apimeta.FindStatusCondition(j.Status.Conditions, "Ready"); cond == nil || cond.Reason != "WaitingForPluginRollout" {
t.Fatalf("Ready condition = %+v, want reason WaitingForPluginRollout", cond)
}
attachedAt := apimeta.FindStatusCondition(j.Status.Conditions, backupCondPluginRolledOut).LastTransitionTime
_, backups = f.pass()
if len(backups) != 0 {
t.Fatalf("a cnpg.io/Backup was created in the pass that attached the plugin: %+v", backups)
}
attached := &cnpgtypes.Cluster{}
if err := f.c.Get(context.Background(), client.ObjectKey{Namespace: rolloutNS, Name: rolloutCluster}, attached); err != nil {
t.Fatalf("get Cluster: %v", err)
}
if currentBarmanServerName(attached) == "" {
t.Fatalf("the barman-cloud plugin was never attached: %+v", attached.Spec.Plugins)
}
// CNPG has not picked the new spec up yet: its phase is the one from
// before the attach.
f.observeCNPG(newRolloutCluster(cnpgClusterHealthyPhase, attached.Spec.Plugins...))
j, backups = f.pass()
if len(backups) != 0 {
t.Fatalf("a cnpg.io/Backup was created on a stale healthy phase, against instances without the plugin: %+v", backups)
}
if cond := apimeta.FindStatusCondition(j.Status.Conditions, "Ready"); cond == nil || !strings.Contains(cond.Message, "predates") {
t.Fatalf("Ready condition = %+v, want the instance pod that predates the plugin in its message", cond)
}
rolling := newRolloutCluster("Upgrading cluster", attached.Spec.Plugins...)
rolling.Status.PhaseReason = "Restarting instance postgres-app-1, because: volumes: element plugins has been added"
f.observeCNPG(rolling)
recreated := newRolloutInstancePod(attachedAt.Add(time.Minute))
f.replacePod(oldPod, recreated)
j, backups = f.pass()
if len(backups) != 0 {
t.Fatalf("a cnpg.io/Backup was created while CNPG was rolling the instances: %+v", backups)
}
if cond := apimeta.FindStatusCondition(j.Status.Conditions, "Ready"); cond == nil || !strings.Contains(cond.Message, "Upgrading cluster") {
t.Fatalf("Ready condition = %+v, want the rolling phase in its message", cond)
}
f.observeCNPG(newRolloutCluster(cnpgClusterHealthyPhase, attached.Spec.Plugins...))
j, backups = f.pass()
if len(backups) != 1 {
t.Fatalf("got %d cnpg.io/Backups once the rollout is over, want 1", len(backups))
}
if j.Status.Phase != backupsv1alpha1.BackupJobPhaseRunning {
t.Errorf("BackupJob phase = %q, want Running", j.Status.Phase)
}
if !apimeta.IsStatusConditionTrue(j.Status.Conditions, backupCondPluginRolledOut) {
t.Errorf("%s condition = %+v, want True", backupCondPluginRolledOut, apimeta.FindStatusCondition(j.Status.Conditions, backupCondPluginRolledOut))
}
if cond := apimeta.FindStatusCondition(j.Status.Conditions, "Ready"); cond != nil {
t.Errorf("Ready condition = %+v left over from the wait while the backup runs", cond)
}
}
// TestReconcileCNPG_AttachedPluginDoesNotDelayBackup: a Cluster that already
// runs the plugin is not rolled by the driver, so its Backup is created on the
// first pass, whatever the age of its pods.
func TestReconcileCNPG_AttachedPluginDoesNotDelayBackup(t *testing.T) {
f := newCNPGRolloutFixture(t, newRolloutBackupJob(time.Now().Add(-time.Minute)),
newRolloutCluster(cnpgClusterHealthyPhase, buildBarmanPlugin(rolloutCluster, rolloutCluster)),
newRolloutInstancePod(time.Now().Add(-time.Hour)))
j, backups := f.pass()
if len(backups) != 1 {
t.Fatalf("got %d cnpg.io/Backups on the first pass, want 1", len(backups))
}
if cond := apimeta.FindStatusCondition(j.Status.Conditions, backupCondPluginRolledOut); cond != nil {
t.Errorf("%s condition = %+v on a Cluster the driver did not roll", backupCondPluginRolledOut, cond)
}
}
// TestReconcileCNPG_PluginRolloutWaitIsBounded: a rollout that never ends
// fails the BackupJob at the backup deadline instead of holding it Running.
func TestReconcileCNPG_PluginRolloutWaitIsBounded(t *testing.T) {
startedAt := time.Now().Add(-cnpgDefaultBackupDeadline - time.Minute)
j := newRolloutBackupJob(startedAt)
j.Status.Conditions = []metav1.Condition{{
Type: backupCondPluginRolledOut,
Status: metav1.ConditionFalse,
Reason: "WaitingForPluginRollout",
LastTransitionTime: metav1.NewTime(startedAt),
}}
f := newCNPGRolloutFixture(t, j,
newRolloutCluster("Waiting for user action", buildBarmanPlugin(rolloutCluster, rolloutCluster)),
newRolloutInstancePod(startedAt.Add(-time.Hour)))
got, backups := f.pass()
if len(backups) != 0 {
t.Fatalf("a cnpg.io/Backup was created against an unfinished rollout: %+v", backups)
}
if got.Status.Phase != backupsv1alpha1.BackupJobPhaseFailed {
t.Fatalf("BackupJob phase = %q, want Failed past the deadline", got.Status.Phase)
}
if !strings.Contains(got.Status.Message, "Waiting for user action") {
t.Errorf("BackupJob message = %q, want the phase CNPG is stuck in", got.Status.Message)
}
}
// TestReconcileCNPGRestore_MissingStrategyFailsClosedAfterDeadline covers the
// restore-path blocker: a RestoreJob whose referenced CNPG Strategy CR never
// appears must NOT requeue forever. requeueRestoreStrategyNotReady gates
// terminal-vs-transient on Status.StartedAt, and the "fail closed" guarantee
// only holds if every driver stamps StartedAt before it can reach the Strategy
// NotFound branch. reconcileCNPGRestore stamps StartedAt and returns early on
// the first reconcile (well before the strategy lookup), so the deadline clock
// starts even when the job begins with StartedAt==nil; once the deadline
// elapses the next reconcile flips the RestoreJob to Phase=Failed.
//
// The existing TestRequeueRestoreStrategyNotReady_BoundedByDeadline only
// exercises the helper in isolation with StartedAt pre-populated, so it cannot
// catch a driver that reaches the NotFound branch without ever stamping
// StartedAt. This test proves the guarantee end to end at the driver call site.
func TestReconcileCNPGRestore_MissingStrategyFailsClosedAfterDeadline(t *testing.T) {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: "pg"},
StrategyRef: corev1.TypedLocalObjectReference{
APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "missing-strategy",
},
},
}
// Status.StartedAt deliberately nil: the driver itself must start the clock.
rj := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: backup.Name}},
}
// No CNPG Strategy CR seeded - the NotFound branch is the point.
c := newCNPGStrategyTestClient(t, backup, rj)
r := &RestoreJobReconciler{Client: c}
ctx := context.Background()
// Pass 1: StartedAt is nil, so the driver stamps it and requeues. It must
// NOT fail yet - the bootstrap grace window has only just opened.
res, err := r.reconcileCNPGRestore(ctx, rj, backup)
if err != nil {
t.Fatalf("pass 1 reconcileCNPGRestore: %v", err)
}
if res.RequeueAfter == 0 {
t.Fatalf("pass 1 must requeue after stamping StartedAt, got %+v", res)
}
persisted := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKeyFromObject(rj), persisted); err != nil {
t.Fatalf("get after pass 1: %v", err)
}
if persisted.Status.StartedAt == nil {
t.Fatalf("driver did not stamp StartedAt before the Strategy NotFound branch; " +
"the deadline clock would never start and the RestoreJob would requeue forever")
}
if persisted.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("must not fail within the grace window, got Phase=%q", persisted.Status.Phase)
}
// Simulate StrategyNotReadyDeadline of wall-clock elapsing by backdating the
// persisted StartedAt past the deadline.
stale := metav1.NewTime(persisted.Status.StartedAt.Add(-StrategyNotReadyDeadline - time.Minute))
base := persisted.DeepCopy()
persisted.Status.StartedAt = &stale
if err := c.Status().Patch(ctx, persisted, client.MergeFrom(base)); err != nil {
t.Fatalf("backdate StartedAt: %v", err)
}
// Pass 2: StartedAt is now past the deadline and the Strategy CR still does
// not exist, so the driver call site must fail closed.
reloaded := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKeyFromObject(rj), reloaded); err != nil {
t.Fatalf("get before pass 2: %v", err)
}
res, err = r.reconcileCNPGRestore(ctx, reloaded, backup)
if err != nil {
t.Fatalf("pass 2 reconcileCNPGRestore: %v", err)
}
if res.RequeueAfter != 0 {
t.Fatalf("terminal failure must not requeue, got %+v", res)
}
final := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKeyFromObject(rj), final); err != nil {
t.Fatalf("get after pass 2: %v", err)
}
if final.Status.Phase != backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("expected Phase=Failed once the deadline elapsed, got %q", final.Status.Phase)
}
if !strings.Contains(final.Status.Message, "missing-strategy") {
t.Errorf("failure message should name the missing strategy, got %q", final.Status.Message)
}
}
// TestReconcileCNPGRestore_RepeatInPlacePurgesStaleRecoveryCluster is the
// reconcile-level regression test for #3311. The pure-function tests
// (TestCNPGPurgeNeeded / TestCNPGClusterFreshlyRecovered) lock in the truth
// table, but the bug lived at the call site: a repeat in-place restore fed the
// purge decision the wrong signal and skipped the destructive purge, so the
// job reported Succeeded against untouched data. This drives the real
// reconcileCNPGRestore path with a fake client and asserts the destructive
// purge actually fires for a stale leftover recovery Cluster - and, in the
// mirror case, that a Cluster this restore just re-created is NOT re-purged
// (the status-write-race protection the guard was originally built for).
func TestReconcileCNPGRestore_RepeatInPlacePurgesStaleRecoveryCluster(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
// startedAt anchors the discriminator: a stale leftover Cluster predates
// it, a freshly-recovered one postdates it.
startedAt := metav1.NewTime(time.Now())
mkBackupArtifact := func(t *testing.T) *backupsv1alpha1.Backup {
t.Helper()
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
return &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{
Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
},
},
}
// A completed cnpg.io/Backup with endWal set clears the WAL-archive gate
// so the reconcile can reach the purge step.
cnpgBackup := &cnpgtypes.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: cnpgBkName},
Spec: cnpgtypes.BackupSpec{Cluster: cnpgtypes.ClusterReference{Name: clusterName}},
Status: cnpgtypes.BackupStatus{Phase: cnpgBackupPhaseComplete, EndWal: "000000010000000000000003"},
}
mkRecoveryCluster := func(created metav1.Time) *cnpgtypes.Cluster {
return &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: created},
Spec: cnpgtypes.ClusterSpec{
Bootstrap: &cnpgtypes.BootstrapConfiguration{
Recovery: &cnpgtypes.RecoverySource{Source: appName},
},
},
}
}
mkClusterPVC := func() *corev1.PersistentVolumeClaim {
return &corev1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName + "-1", Labels: map[string]string{cnpgClusterLabel: clusterName}},
}
}
mkRestoreJob := func() *backupsv1alpha1.RestoreJob {
sa := startedAt
return &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
Status: backupsv1alpha1.RestoreJobStatus{StartedAt: &sa, Phase: backupsv1alpha1.RestoreJobPhaseRunning},
}
}
t.Run("stale leftover recovery cluster from a prior restore is purged", func(t *testing.T) {
backup := mkBackupArtifact(t)
// creationTimestamp an hour before StartedAt: leftover from a prior restore.
stale := metav1.NewTime(startedAt.Add(-time.Hour))
c := newCNPGStrategyTestClient(t, backup, mkRestoreJob(), strategy, cnpgBackup,
newPostgresApp(appName, ns), mkRecoveryCluster(stale), mkClusterPVC())
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: record.NewFakeRecorder(10)}
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, rj); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
if _, err := r.reconcileCNPGRestore(ctx, rj, backup); err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
// The stale Cluster (and its labelled PVC) must have been purged.
err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: clusterName}, &cnpgtypes.Cluster{})
if !apierrors.IsNotFound(err) {
t.Fatalf("expected stale Cluster to be purged (NotFound), got err=%v", err)
}
pvcs := &corev1.PersistentVolumeClaimList{}
if err := c.List(ctx, pvcs, client.InNamespace(ns), client.MatchingLabels{cnpgClusterLabel: clusterName}); err != nil {
t.Fatalf("list PVCs: %v", err)
}
if len(pvcs.Items) != 0 {
t.Fatalf("expected cluster PVCs purged, still have %d", len(pvcs.Items))
}
})
t.Run("freshly-recovered cluster from this restore is not re-purged", func(t *testing.T) {
backup := mkBackupArtifact(t)
// creationTimestamp a minute after StartedAt: this restore's own re-render.
fresh := metav1.NewTime(startedAt.Add(time.Minute))
c := newCNPGStrategyTestClient(t, backup, mkRestoreJob(), strategy, cnpgBackup,
newPostgresApp(appName, ns), mkRecoveryCluster(fresh))
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: record.NewFakeRecorder(10)}
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, rj); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
if _, err := r.reconcileCNPGRestore(ctx, rj, backup); err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
// The freshly-recovered Cluster must survive - re-purging it would
// destroy the recovery this restore just started (the status-write-race
// protection the guard was built for).
got := &cnpgtypes.Cluster{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: clusterName}, got); err != nil {
t.Fatalf("expected freshly-recovered Cluster to survive, got err=%v", err)
}
if !got.DeletionTimestamp.IsZero() {
t.Fatalf("freshly-recovered Cluster must not be marked for deletion")
}
})
}
// TestReconcileCNPGRestore_RefusesCrossFlavorRestore: a physical backup only
// restores onto the image family it was taken from. The postgresql and
// postgis images ship different glibc versions, and postgis objects need the
// postgis libraries, so a cross-flavor RestoreJob must fail before the purge
// deletes the target's Cluster and PVCs. A snapshot without a flavor predates
// flavors and therefore comes from a postgresql cluster.
func TestReconcileCNPGRestore_RefusesCrossFlavorRestore(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
startedAt := metav1.NewTime(time.Now())
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{
Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
},
},
}
cnpgBackup := &cnpgtypes.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: cnpgBkName},
Spec: cnpgtypes.BackupSpec{Cluster: cnpgtypes.ClusterReference{Name: clusterName}},
Status: cnpgtypes.BackupStatus{Phase: cnpgBackupPhaseComplete, EndWal: "000000010000000000000003"},
}
cases := []struct {
name string
sourceFlavor string
targetFlavor string
wantFailed bool
wantInMessage string
}{
{name: "postgis backup into a postgresql app", sourceFlavor: "postgis", targetFlavor: "", wantFailed: true, wantInMessage: "taken from a postgis cluster"},
{name: "pre-flavor backup into a postgis app", sourceFlavor: "", targetFlavor: "postgis", wantFailed: true, wantInMessage: "taken from a postgresql cluster"},
{name: "postgis backup into a postgis app", sourceFlavor: "postgis", targetFlavor: "postgis", wantFailed: false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, tc.sourceFlavor)
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
target := newPostgresApp(appName, ns)
target.Spec.Flavor = tc.targetFlavor
sa := startedAt
restoreJob := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
Status: backupsv1alpha1.RestoreJobStatus{StartedAt: &sa, Phase: backupsv1alpha1.RestoreJobPhaseRunning},
}
liveCluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: metav1.NewTime(startedAt.Add(-time.Hour))},
}
pvc := &corev1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName + "-1", Labels: map[string]string{cnpgClusterLabel: clusterName}},
}
c := newCNPGStrategyTestClient(t, backup, restoreJob, strategy, cnpgBackup, target, liveCluster, pvc)
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: record.NewFakeRecorder(10)}
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, rj); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
if _, err := r.reconcileCNPGRestore(ctx, rj, backup); err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
got := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, got); err != nil {
t.Fatalf("get RestoreJob: %v", err)
}
clusterErr := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: clusterName}, &cnpgtypes.Cluster{})
app, err := r.getPostgresApp(ctx, ns, appName)
if err != nil {
t.Fatalf("get target app: %v", err)
}
if !tc.wantFailed {
if got.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("same-flavor restore must proceed, got Failed: %s", got.Status.Message)
}
if !apierrors.IsNotFound(clusterErr) {
t.Fatalf("same-flavor restore must purge the target Cluster, got err=%v", clusterErr)
}
return
}
if got.Status.Phase != backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("cross-flavor restore must fail, got phase %q", got.Status.Phase)
}
if !strings.Contains(got.Status.Message, tc.wantInMessage) || !strings.Contains(got.Status.Message, "pg_dump") {
t.Errorf("failure message must name the source flavor and point to a logical dump, got %q", got.Status.Message)
}
if clusterErr != nil {
t.Fatalf("cross-flavor restore must leave the target Cluster in place, got err=%v", clusterErr)
}
pvcs := &corev1.PersistentVolumeClaimList{}
if err := c.List(ctx, pvcs, client.InNamespace(ns), client.MatchingLabels{cnpgClusterLabel: clusterName}); err != nil {
t.Fatalf("list PVCs: %v", err)
}
if len(pvcs.Items) != 1 {
t.Fatalf("cross-flavor restore must leave the target PVCs in place, have %d", len(pvcs.Items))
}
if app.Spec.Bootstrap.Enabled {
t.Fatalf("cross-flavor restore must not patch the target app for recovery")
}
})
}
}
// TestReconcileCNPGRestore_HealthyClusterSucceeds drives the reconcile to its
// success branch: with the target already purged (TargetPurged=True) and the
// recovery Cluster reporting healthy, the RestoreJob must reach phase
// Succeeded with Ready=True AND the new RecoveryConverged=True condition.
func TestReconcileCNPGRestore_HealthyClusterSucceeds(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
startedAt := metav1.NewTime(time.Now())
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
}},
}
// Healthy recovery Cluster created after StartedAt (this restore's own).
healthyCluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: metav1.NewTime(startedAt.Add(time.Minute))},
Spec: cnpgtypes.ClusterSpec{Bootstrap: &cnpgtypes.BootstrapConfiguration{Recovery: &cnpgtypes.RecoverySource{Source: appName}}},
Status: cnpgtypes.ClusterStatus{Phase: cnpgClusterHealthyPhase},
}
sa := startedAt
restoreJob := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
Status: backupsv1alpha1.RestoreJobStatus{
StartedAt: &sa,
Phase: backupsv1alpha1.RestoreJobPhaseRunning,
// Purge already done for this restore, so reconcile skips straight
// to the wait-for-healthy / success path. A prior reconcile hit a
// transient disable error and recorded BootstrapDisablePending=True
// within the grace window; the disable now succeeds, so the terminal
// success write must clear that pending condition rather than leave it
// standing True next to a Succeeded RestoreJob.
Conditions: []metav1.Condition{
{
Type: restoreCondTargetPurged, Status: metav1.ConditionTrue, Reason: "ClusterPurged",
LastTransitionTime: startedAt, Message: "purged",
},
{
Type: restoreCondBootstrapDisablePending, Status: metav1.ConditionTrue, Reason: "Retrying",
LastTransitionTime: startedAt, Message: "clearing spec.bootstrap.enabled is retrying after a transient error",
},
},
},
}
c := newCNPGStrategyTestClient(t, backup, restoreJob, strategy, newPostgresApp(appName, ns), healthyCluster)
rec := record.NewFakeRecorder(10)
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: rec}
// A healthy recovery now converges in two reconciles: the first latches
// RecoveryConverged durably and requeues, the second disables bootstrap and
// marks Succeeded (see reconcileCNPGRestore's terminal block).
got := reconcileCNPGRestoreToTerminal(t, ctx, r, c, backup, ns, "rj")
if got.Status.Phase != backupsv1alpha1.RestoreJobPhaseSucceeded {
t.Fatalf("expected phase Succeeded, got %q (msg=%q)", got.Status.Phase, got.Status.Message)
}
if c := apimeta.FindStatusCondition(got.Status.Conditions, "Ready"); c == nil || c.Status != metav1.ConditionTrue {
t.Fatalf("expected Ready=True, got %+v", c)
}
if c := apimeta.FindStatusCondition(got.Status.Conditions, restoreCondRecoveryConverged); c == nil || c.Status != metav1.ConditionTrue {
t.Fatalf("expected RecoveryConverged=True, got %+v", c)
}
// The terminal write must also emit the pending-convergence Event, so the
// credential handoff is discoverable in `kubectl get events`, not only in
// .status. Dropping the r.Recorder.Eventf call leaves this channel empty.
select {
case ev := <-rec.Events:
if !strings.Contains(ev, "CredentialsConvergencePending") {
t.Fatalf("expected a CredentialsConvergencePending event, got %q", ev)
}
default:
t.Fatalf("expected a CredentialsConvergencePending event, got none")
}
// The pending credential handoff is carried as the REASON on Ready=True (not a
// standalone never-True condition): a durable, readable marker that does not
// look like a standing failure. Changing the reason back to a bare
// "RestoreCompleted" reddens this.
if c := apimeta.FindStatusCondition(got.Status.Conditions, "Ready"); c == nil || c.Reason != "RestoreCompletedCredentialsPending" {
t.Fatalf("expected Ready reason RestoreCompletedCredentialsPending, got %+v", c)
}
// The seeded BootstrapDisablePending=True must be closed out on success, not
// left standing next to a Succeeded RestoreJob. Dropping the terminal
// clear-block reddens this.
if c := apimeta.FindStatusCondition(got.Status.Conditions, restoreCondBootstrapDisablePending); c == nil || c.Status != metav1.ConditionFalse || c.Reason != "Cleared" {
t.Fatalf("expected BootstrapDisablePending cleared to False/Cleared, got %+v", c)
}
}
// TestReconcileCNPGRestore_HealthyDisablesBootstrap is the regression guard for
// the restore-credentials fix: patchPostgresAppForRestore leaves the target app
// with bootstrap.enabled=true, and while it stays true the chart skips the
// init-job that reconciles the generated <release>-credentials Secret onto the
// recovered roles. The recovered roles carry the source's password hashes, the
// freshly generated Secret does not match them, so without this flip every
// application-user login against the restored target fails. On convergence the
// controller must clear bootstrap.enabled so the next HelmRelease upgrade runs
// that init-job.
func TestReconcileCNPGRestore_HealthyDisablesBootstrap(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
startedAt := metav1.NewTime(time.Now().Add(-time.Hour))
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
}},
}
healthyCluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: metav1.NewTime(startedAt.Add(time.Minute))},
Spec: cnpgtypes.ClusterSpec{Bootstrap: &cnpgtypes.BootstrapConfiguration{Recovery: &cnpgtypes.RecoverySource{Source: appName}}},
Status: cnpgtypes.ClusterStatus{Phase: cnpgClusterHealthyPhase},
}
// The target app is mid-restore: patchPostgresAppForRestore already set
// bootstrap.enabled=true (oldName is required alongside it by the schema).
app := newPostgresApp(appName, ns)
app.Spec.Bootstrap.Enabled = true
app.Spec.Bootstrap.OldName = appName
sa := startedAt
restoreJob := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
Status: backupsv1alpha1.RestoreJobStatus{
StartedAt: &sa,
Phase: backupsv1alpha1.RestoreJobPhaseRunning,
Conditions: []metav1.Condition{{
Type: restoreCondTargetPurged, Status: metav1.ConditionTrue, Reason: "ClusterPurged",
LastTransitionTime: startedAt, Message: "purged",
}},
},
}
c := newCNPGStrategyTestClient(t, backup, restoreJob, strategy, app, healthyCluster)
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: record.NewFakeRecorder(10)}
got := reconcileCNPGRestoreToTerminal(t, ctx, r, c, backup, ns, "rj")
if got.Status.Phase != backupsv1alpha1.RestoreJobPhaseSucceeded {
t.Fatalf("expected phase Succeeded, got %q (msg=%q)", got.Status.Phase, got.Status.Message)
}
gotApp := &postgresapp.Postgres{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: appName}, gotApp); err != nil {
t.Fatalf("get target app after reconcile: %v", err)
}
if gotApp.Spec.Bootstrap.Enabled {
t.Fatalf("expected bootstrap.enabled cleared after recovery converged, still true")
}
// oldName must survive: the app schema requires it whenever the bootstrap
// block is present, and clearing it would fail admission on the next apply.
if gotApp.Spec.Bootstrap.OldName != appName {
t.Fatalf("expected bootstrap.oldName preserved, got %q", gotApp.Spec.Bootstrap.OldName)
}
}
// TestReconcileCNPGRestore_BootstrapDisableFailsPastDeadline is the regression
// guard for the still-open risk on the disable-bootstrap terminal step: recovery
// has converged (RecoveryConverged=True latched) but clearing bootstrap.enabled
// keeps failing - the app was deleted/renamed mid-restore, or a GitOps source
// re-asserts it. Without a deadline bound the terminal block requeues forever and
// the RestoreJob sits Running with no Failed/Ready=False. Past the restore
// deadline it must terminate as Failed with a BootstrapDisableFailed reason.
func TestReconcileCNPGRestore_BootstrapDisableFailsPastDeadline(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
// Well past the 30m default deadline.
startedAt := metav1.NewTime(time.Now().Add(-72 * time.Hour))
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
}},
}
healthyCluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: metav1.NewTime(startedAt.Add(time.Minute))},
Spec: cnpgtypes.ClusterSpec{Bootstrap: &cnpgtypes.BootstrapConfiguration{Recovery: &cnpgtypes.RecoverySource{Source: appName}}},
Status: cnpgtypes.ClusterStatus{Phase: cnpgClusterHealthyPhase},
}
app := newPostgresApp(appName, ns)
app.Spec.Bootstrap.Enabled = true
app.Spec.Bootstrap.OldName = appName
sa := startedAt
restoreJob := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
Status: backupsv1alpha1.RestoreJobStatus{
StartedAt: &sa,
Phase: backupsv1alpha1.RestoreJobPhaseRunning,
Conditions: []metav1.Condition{
{Type: restoreCondTargetPurged, Status: metav1.ConditionTrue, Reason: "ClusterPurged", LastTransitionTime: startedAt, Message: "purged"},
{Type: restoreCondRecoveryConverged, Status: metav1.ConditionTrue, Reason: "RecoveryConverged", LastTransitionTime: startedAt, Message: "converged"},
},
},
}
// A fake client that fails Patch on the Postgres app, so
// disablePostgresAppBootstrap can never clear bootstrap.enabled.
s := runtime.NewScheme()
_ = scheme.AddToScheme(s)
_ = backupsv1alpha1.AddToScheme(s)
_ = strategyv1alpha1.AddToScheme(s)
_ = cnpgtypes.AddToScheme(s)
_ = postgresapp.AddToScheme(s)
c := clientfake.NewClientBuilder().
WithScheme(s).
WithObjects(backup, restoreJob, strategy, app, healthyCluster).
WithStatusSubresource(&backupsv1alpha1.BackupJob{}, &backupsv1alpha1.RestoreJob{}, &backupsv1alpha1.Backup{}).
WithInterceptorFuncs(interceptor.Funcs{
Patch: func(ctx context.Context, cl client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
if _, ok := obj.(*postgresapp.Postgres); ok {
return fmt.Errorf("injected bootstrap-disable failure")
}
return cl.Patch(ctx, obj, patch, opts...)
},
}).
Build()
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: record.NewFakeRecorder(10)}
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, rj); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
if _, err := r.reconcileCNPGRestore(ctx, rj, backup); err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
got := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, got); err != nil {
t.Fatalf("get RestoreJob after reconcile: %v", err)
}
if got.Status.Phase != backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("expected phase Failed when bootstrap-disable keeps failing past the deadline, got %q", got.Status.Phase)
}
if cond := apimeta.FindStatusCondition(got.Status.Conditions, "Ready"); cond == nil || cond.Reason != "BootstrapDisableFailed" {
t.Fatalf("expected Ready=False reason BootstrapDisableFailed, got %+v", cond)
}
}
// TestReconcileCNPGRestore_BootstrapDisableTransientErrorRequeues guards the
// grace window on the disable step: the health check wins BEFORE the restore
// deadline on purpose, so recovery may legitimately have spent the whole
// deadline by the time bootstrap is cleared. The disable step must therefore be
// bounded from RecoveryConverged's LastTransitionTime, not from StartedAt - or a
// single transient error on the first disable attempt (StartedAt already past
// the deadline) would terminate the restore Failed, which on a resubmit lets the
// purge-guard delete the healthy restored Cluster. A transient failure right
// after convergence must requeue, not fail.
func TestReconcileCNPGRestore_BootstrapDisableTransientErrorRequeues(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
// StartedAt is well past the 30m restore deadline: a StartedAt-based bound
// would fail immediately; convergence just happened, so the disable window is
// still open and a transient error must requeue.
startedAt := metav1.NewTime(time.Now().Add(-72 * time.Hour))
convergedAt := metav1.Now()
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
}},
}
healthyCluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: metav1.NewTime(startedAt.Add(time.Minute))},
Spec: cnpgtypes.ClusterSpec{Bootstrap: &cnpgtypes.BootstrapConfiguration{Recovery: &cnpgtypes.RecoverySource{Source: appName}}},
Status: cnpgtypes.ClusterStatus{Phase: cnpgClusterHealthyPhase},
}
app := newPostgresApp(appName, ns)
app.Spec.Bootstrap.Enabled = true
app.Spec.Bootstrap.OldName = appName
sa := startedAt
restoreJob := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
Status: backupsv1alpha1.RestoreJobStatus{
StartedAt: &sa,
Phase: backupsv1alpha1.RestoreJobPhaseRunning,
Conditions: []metav1.Condition{
{Type: restoreCondTargetPurged, Status: metav1.ConditionTrue, Reason: "ClusterPurged", LastTransitionTime: startedAt, Message: "purged"},
{Type: restoreCondRecoveryConverged, Status: metav1.ConditionTrue, Reason: "RecoveryConverged", LastTransitionTime: convergedAt, Message: "converged"},
},
},
}
s := runtime.NewScheme()
_ = scheme.AddToScheme(s)
_ = backupsv1alpha1.AddToScheme(s)
_ = strategyv1alpha1.AddToScheme(s)
_ = cnpgtypes.AddToScheme(s)
_ = postgresapp.AddToScheme(s)
c := clientfake.NewClientBuilder().
WithScheme(s).
WithObjects(backup, restoreJob, strategy, app, healthyCluster).
WithStatusSubresource(&backupsv1alpha1.BackupJob{}, &backupsv1alpha1.RestoreJob{}, &backupsv1alpha1.Backup{}).
WithInterceptorFuncs(interceptor.Funcs{
Patch: func(ctx context.Context, cl client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
if _, ok := obj.(*postgresapp.Postgres); ok {
return fmt.Errorf("injected transient bootstrap-disable failure")
}
return cl.Patch(ctx, obj, patch, opts...)
},
}).
Build()
rec := record.NewFakeRecorder(10)
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: rec}
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, rj); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
res, err := r.reconcileCNPGRestore(ctx, rj, backup)
if err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
if res.RequeueAfter <= 0 {
t.Fatalf("expected a requeue after a transient disable failure, got %+v", res)
}
got := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, got); err != nil {
t.Fatalf("get RestoreJob after reconcile: %v", err)
}
if got.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("a transient disable failure within the post-convergence window must requeue, not Fail; got phase %q", got.Status.Phase)
}
// The window must not be silent: the requeue records WHY on .status (a durable
// condition) and announces it once via an Event. Dropping either leaves the
// RestoreJob non-terminal for up to the whole grace window with no explanation.
if c := apimeta.FindStatusCondition(got.Status.Conditions, restoreCondBootstrapDisablePending); c == nil || c.Status != metav1.ConditionTrue {
t.Fatalf("expected BootstrapDisablePending=True during the window, got %+v", c)
}
select {
case ev := <-rec.Events:
if !strings.Contains(ev, "BootstrapDisablePending") {
t.Fatalf("expected a BootstrapDisablePending event, got %q", ev)
}
default:
t.Fatalf("expected a BootstrapDisablePending event, got none")
}
}
// TestReconcileCNPGRestore_BootstrapDisableShortTimeoutHonorsGraceFloor is the
// regression guard for the grace-window floor. A tenant sets
// restoreTimeoutSeconds:60 to fail fast on a stuck recovery; recovery actually
// converged 90s ago and the disable step is now hitting a transient error.
// Without the floor the disable grace collapses to 60s (the recovery timeout),
// so 90s-since-convergence terminates the restore Failed on a blip - and a
// resubmit's purge-guard would then delete the healthy restored Cluster. With
// the floor (cnpgBootstrapDisableGrace) the window stays open and the
// transient error requeues. Reverting effectiveBootstrapDisableGrace back to
// effectiveRestoreDeadline turns this red.
func TestReconcileCNPGRestore_BootstrapDisableShortTimeoutHonorsGraceFloor(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
startedAt := metav1.NewTime(time.Now().Add(-72 * time.Hour))
// Converged 90s ago: past the tenant's 60s restoreTimeoutSeconds but well
// within the 5m post-convergence floor.
convergedAt := metav1.NewTime(time.Now().Add(-90 * time.Second))
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
}},
}
healthyCluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: metav1.NewTime(startedAt.Add(time.Minute))},
Spec: cnpgtypes.ClusterSpec{Bootstrap: &cnpgtypes.BootstrapConfiguration{Recovery: &cnpgtypes.RecoverySource{Source: appName}}},
Status: cnpgtypes.ClusterStatus{Phase: cnpgClusterHealthyPhase},
}
app := newPostgresApp(appName, ns)
app.Spec.Bootstrap.Enabled = true
app.Spec.Bootstrap.OldName = appName
sa := startedAt
restoreJob := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{
BackupRef: corev1.LocalObjectReference{Name: "bk"},
Options: &runtime.RawExtension{Raw: []byte(`{"restoreTimeoutSeconds":60}`)},
},
Status: backupsv1alpha1.RestoreJobStatus{
StartedAt: &sa,
Phase: backupsv1alpha1.RestoreJobPhaseRunning,
Conditions: []metav1.Condition{
{Type: restoreCondTargetPurged, Status: metav1.ConditionTrue, Reason: "ClusterPurged", LastTransitionTime: startedAt, Message: "purged"},
{Type: restoreCondRecoveryConverged, Status: metav1.ConditionTrue, Reason: "RecoveryConverged", LastTransitionTime: convergedAt, Message: "converged"},
},
},
}
s := runtime.NewScheme()
_ = scheme.AddToScheme(s)
_ = backupsv1alpha1.AddToScheme(s)
_ = strategyv1alpha1.AddToScheme(s)
_ = cnpgtypes.AddToScheme(s)
_ = postgresapp.AddToScheme(s)
c := clientfake.NewClientBuilder().
WithScheme(s).
WithObjects(backup, restoreJob, strategy, app, healthyCluster).
WithStatusSubresource(&backupsv1alpha1.BackupJob{}, &backupsv1alpha1.RestoreJob{}, &backupsv1alpha1.Backup{}).
WithInterceptorFuncs(interceptor.Funcs{
Patch: func(ctx context.Context, cl client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
if _, ok := obj.(*postgresapp.Postgres); ok {
return fmt.Errorf("injected transient bootstrap-disable failure")
}
return cl.Patch(ctx, obj, patch, opts...)
},
}).
Build()
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: record.NewFakeRecorder(10)}
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, rj); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
res, err := r.reconcileCNPGRestore(ctx, rj, backup)
if err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
if res.RequeueAfter <= 0 {
t.Fatalf("expected a requeue within the floored grace window, got %+v", res)
}
got := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, got); err != nil {
t.Fatalf("get RestoreJob after reconcile: %v", err)
}
if got.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("a short restoreTimeoutSeconds must not shrink the disable grace below the floor; a blip 90s after convergence must requeue, got phase %q", got.Status.Phase)
}
}
// TestReconcileCNPGRestore_HealthyPastDeadlineSucceeds is the regression guard
// for the health-before-deadline ordering: a recovery that has reached a
// healthy Cluster must be reported Succeeded even when the reconcile observing
// it lands AFTER the restore deadline has elapsed (e.g. a controller restart
// spanning the window). StartedAt is 72h in the past (well past the 30m default
// deadline) and a recoveryTime is set, so if the deadline branch ran first it
// would classify the restore Failed/RecoveryTargetUnreachable - a wrong verdict
// that can escalate to data loss on resubmit. The injected readPodLog fails the
// test if the classification path is entered at all, proving health short-circuits.
func TestReconcileCNPGRestore_HealthyPastDeadlineSucceeds(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
// Far past the 30m default deadline: this is the reconcile-gap case.
startedAt := metav1.NewTime(time.Now().Add(-72 * time.Hour))
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
}},
}
healthyCluster := &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: metav1.NewTime(startedAt.Add(time.Minute))},
Spec: cnpgtypes.ClusterSpec{Bootstrap: &cnpgtypes.BootstrapConfiguration{Recovery: &cnpgtypes.RecoverySource{Source: appName}}},
Status: cnpgtypes.ClusterStatus{Phase: cnpgClusterHealthyPhase},
}
sa := startedAt
restoreJob := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{
BackupRef: corev1.LocalObjectReference{Name: "bk"},
// recoveryTime set so the (wrongly-ordered) deadline branch would classify.
Options: &runtime.RawExtension{Raw: []byte(`{"recoveryTime":"2026-01-01T00:00:00Z"}`)},
},
Status: backupsv1alpha1.RestoreJobStatus{
StartedAt: &sa,
Phase: backupsv1alpha1.RestoreJobPhaseRunning,
Conditions: []metav1.Condition{{
Type: restoreCondTargetPurged, Status: metav1.ConditionTrue, Reason: "ClusterPurged",
LastTransitionTime: startedAt, Message: "purged",
}},
},
}
c := newCNPGStrategyTestClient(t, backup, restoreJob, strategy, newPostgresApp(appName, ns), healthyCluster)
r := &RestoreJobReconciler{
Client: c,
Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)),
Recorder: record.NewFakeRecorder(10),
// If the deadline/classification branch runs before the health check,
// it reaches the log reader - which must never happen for a healthy cluster.
readPodLog: func(context.Context, string, string, string) (string, error) {
t.Fatalf("classification (log read) ran for an already-healthy Cluster: health check must precede the deadline branch")
return "", nil
},
}
// Two reconciles to terminal (latch, then disable-bootstrap + Succeeded);
// the injected readPodLog still fails the test if the deadline/classification
// branch runs on either pass, proving health short-circuits regardless.
got := reconcileCNPGRestoreToTerminal(t, ctx, r, c, backup, ns, "rj")
if got.Status.Phase != backupsv1alpha1.RestoreJobPhaseSucceeded {
t.Fatalf("expected phase Succeeded (healthy cluster past deadline), got %q (msg=%q)", got.Status.Phase, got.Status.Message)
}
if cond := apimeta.FindStatusCondition(got.Status.Conditions, "Ready"); cond == nil || cond.Status != metav1.ConditionTrue {
t.Fatalf("expected Ready=True, got %+v", cond)
}
if cond := apimeta.FindStatusCondition(got.Status.Conditions, restoreCondRecoveryConverged); cond == nil || cond.Status != metav1.ConditionTrue {
t.Fatalf("expected RecoveryConverged=True, got %+v", cond)
}
}
// TestReconcileCNPGRestore_DeadlineClassifiesRecoveryTargetUnreachable drives the
// full reconcileCNPGRestore deadline branch (the classification call site), which
// the pure-helper tests (TestRecoveryTargetUnreachable_*) do not exercise:
// swapping the RecoveryTargetUnreachable reason back to the generic RestoreFailed
// at that call site leaves them green. Each sub-case backdates StartedAt past the
// restore deadline with a recoveryTime set (so classification runs) and asserts
// the three outcomes the branch distinguishes: FATAL found, FATAL absent, and a
// Forbidden log read (missing pods/log RBAC).
func TestReconcileCNPGRestore_DeadlineClassifiesRecoveryTargetUnreachable(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
// StartedAt far past cnpgDefaultRestoreDeadline (30m) so the first reconcile
// enters the deadline-classification branch.
startedAt := metav1.NewTime(time.Now().Add(-72 * time.Hour))
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
mkBackup := func() *backupsv1alpha1.Backup {
return &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
},
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
}},
}
// Recovery Cluster created after StartedAt (this restore's own -> not
// re-purged), so the reconcile flows past the purge block to the deadline
// check. Phase left empty (still wedged): the deadline check precedes the
// healthy check, so it fires regardless.
mkCluster := func() *cnpgtypes.Cluster {
return &cnpgtypes.Cluster{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: clusterName, CreationTimestamp: metav1.NewTime(startedAt.Add(time.Minute))},
Spec: cnpgtypes.ClusterSpec{Bootstrap: &cnpgtypes.BootstrapConfiguration{Recovery: &cnpgtypes.RecoverySource{Source: appName}}},
}
}
// A bootstrap-recovery pod for the cluster; recoveryTargetUnreachable lists
// these and reads each through the readPodLog seam.
mkPod := func() *corev1.Pod {
return &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Namespace: ns,
Name: clusterName + "-full-recovery-abcde",
Labels: map[string]string{cnpgClusterLabel: clusterName},
CreationTimestamp: metav1.NewTime(startedAt.Add(2 * time.Minute)),
},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: cnpgRecoveryContainerName}}},
}
}
mkRestoreJob := func() *backupsv1alpha1.RestoreJob {
sa := startedAt
return &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{
BackupRef: corev1.LocalObjectReference{Name: "bk"},
// recoveryTime present -> the reconcile runs the classification.
Options: &runtime.RawExtension{Raw: []byte(`{"recoveryTime":"2026-01-01T00:00:00Z"}`)},
},
Status: backupsv1alpha1.RestoreJobStatus{
StartedAt: &sa,
Phase: backupsv1alpha1.RestoreJobPhaseRunning,
// Purge already done -> reconcile reaches the deadline check.
Conditions: []metav1.Condition{{
Type: restoreCondTargetPurged, Status: metav1.ConditionTrue, Reason: "ClusterPurged",
LastTransitionTime: startedAt, Message: "purged",
}},
},
}
}
fatalLog := "LOG: redo done\nFATAL: " + cnpgRecoveryTargetUnreachableLog
newReconciler := func(c client.Client, readLog func(context.Context, string, string, string) (string, error)) *RestoreJobReconciler {
return &RestoreJobReconciler{
Client: c,
Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)),
Recorder: record.NewFakeRecorder(10),
readPodLog: readLog,
}
}
reconcileToFailed := func(t *testing.T, r *RestoreJobReconciler, c client.Client, backup *backupsv1alpha1.Backup) *backupsv1alpha1.RestoreJob {
t.Helper()
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, rj); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
res, err := r.reconcileCNPGRestore(ctx, rj, backup)
if err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
if res.RequeueAfter != 0 {
t.Fatalf("terminal deadline failure must not requeue, got %+v", res)
}
got := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, got); err != nil {
t.Fatalf("get RestoreJob after reconcile: %v", err)
}
if got.Status.Phase != backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("expected Phase=Failed after deadline, got %q (msg=%q)", got.Status.Phase, got.Status.Message)
}
return got
}
t.Run("recovery-target-unreachable FATAL -> RecoveryTargetUnreachable reason", func(t *testing.T) {
backup := mkBackup()
c := newCNPGStrategyTestClient(t, backup, mkRestoreJob(), strategy, newPostgresApp(appName, ns), mkCluster(), mkPod())
r := newReconciler(c, func(context.Context, string, string, string) (string, error) { return fatalLog, nil })
got := reconcileToFailed(t, r, c, backup)
// The mutation the review flagged: the call site must set reason
// RecoveryTargetUnreachable, not the generic RestoreFailed.
conv := apimeta.FindStatusCondition(got.Status.Conditions, restoreCondRecoveryConverged)
if conv == nil || conv.Status != metav1.ConditionFalse || conv.Reason != "RecoveryTargetUnreachable" {
t.Fatalf("expected RecoveryConverged=False reason=RecoveryTargetUnreachable, got %+v", conv)
}
if ready := apimeta.FindStatusCondition(got.Status.Conditions, "Ready"); ready == nil || ready.Reason != "RecoveryTargetUnreachable" {
t.Fatalf("expected Ready reason=RecoveryTargetUnreachable, got %+v", ready)
}
if !strings.Contains(got.Status.Message, "past the latest WAL archived") {
t.Errorf("message should explain the unreachable target, got %q", got.Status.Message)
}
})
t.Run("no FATAL -> generic deadline failure, not RecoveryTargetUnreachable", func(t *testing.T) {
backup := mkBackup()
c := newCNPGStrategyTestClient(t, backup, mkRestoreJob(), strategy, newPostgresApp(appName, ns), mkCluster(), mkPod())
r := newReconciler(c, func(context.Context, string, string, string) (string, error) {
return "LOG: consistent recovery state reached", nil
})
got := reconcileToFailed(t, r, c, backup)
if conv := apimeta.FindStatusCondition(got.Status.Conditions, restoreCondRecoveryConverged); conv != nil {
t.Fatalf("did not expect a RecoveryConverged condition on a generic timeout, got %+v", conv)
}
if ready := apimeta.FindStatusCondition(got.Status.Conditions, "Ready"); ready != nil && ready.Reason == "RecoveryTargetUnreachable" {
t.Fatalf("generic timeout must NOT be classified RecoveryTargetUnreachable, got %+v", ready)
}
if !strings.Contains(got.Status.Message, "deadline") {
t.Errorf("message should mention the deadline, got %q", got.Status.Message)
}
if strings.Contains(got.Status.Message, "pods/log RBAC") {
t.Errorf("a readable log must not report a Forbidden RBAC note, got %q", got.Status.Message)
}
})
t.Run("Forbidden log read -> generic failure with missing-RBAC note", func(t *testing.T) {
backup := mkBackup()
c := newCNPGStrategyTestClient(t, backup, mkRestoreJob(), strategy, newPostgresApp(appName, ns), mkCluster(), mkPod())
r := newReconciler(c, func(context.Context, string, string, string) (string, error) {
return "", apierrors.NewForbidden(schema.GroupResource{Resource: "pods"}, "rec", fmt.Errorf("pods/log grant missing"))
})
got := reconcileToFailed(t, r, c, backup)
if conv := apimeta.FindStatusCondition(got.Status.Conditions, restoreCondRecoveryConverged); conv != nil && conv.Reason == "RecoveryTargetUnreachable" {
t.Fatalf("a Forbidden log read must not yield a RecoveryTargetUnreachable classification, got %+v", conv)
}
if !strings.Contains(got.Status.Message, "pods/log RBAC") {
t.Errorf("Forbidden classification should surface the missing pods/log RBAC grant, got %q", got.Status.Message)
}
})
}
// newCNPGStrategyTestClient returns a fake client.Client wired up with
// the schemes the CNPG-strategy reconciler needs.
func newCNPGStrategyTestClient(t *testing.T, objs ...client.Object) client.Client {
t.Helper()
s := runtime.NewScheme()
_ = scheme.AddToScheme(s)
_ = backupsv1alpha1.AddToScheme(s)
_ = strategyv1alpha1.AddToScheme(s)
_ = cnpgtypes.AddToScheme(s)
_ = postgresapp.AddToScheme(s)
return clientfake.NewClientBuilder().
WithScheme(s).
WithObjects(objs...).
WithStatusSubresource(&backupsv1alpha1.BackupJob{}, &backupsv1alpha1.RestoreJob{}, &backupsv1alpha1.Backup{}).
Build()
}
// reconcileCNPGRestoreToTerminal drives reconcileCNPGRestore until the
// RestoreJob reaches a terminal phase (Succeeded/Failed) or a small iteration
// bound is hit. The success path is multi-step - it latches RecoveryConverged
// on one reconcile and disables bootstrap + marks Succeeded on the next - so a
// single reconcile does not reach the verdict. Re-fetching between passes
// mirrors how the controller-runtime requeue would re-enter Reconcile.
func reconcileCNPGRestoreToTerminal(t *testing.T, ctx context.Context, r *RestoreJobReconciler, c client.Client, backup *backupsv1alpha1.Backup, ns, name string) *backupsv1alpha1.RestoreJob {
t.Helper()
got := &backupsv1alpha1.RestoreJob{}
for i := 0; i < 8; i++ {
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: name}, rj); err != nil {
t.Fatalf("get RestoreJob: %v", err)
}
if _, err := r.reconcileCNPGRestore(ctx, rj, backup); err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: name}, got); err != nil {
t.Fatalf("get RestoreJob after reconcile: %v", err)
}
if got.Status.Phase == backupsv1alpha1.RestoreJobPhaseSucceeded ||
got.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
return got
}
}
t.Fatalf("RestoreJob did not reach a terminal phase; last phase %q", got.Status.Phase)
return got
}
// TestDisablePostgresAppBootstrap_Idempotent pins the early-return in
// disablePostgresAppBootstrap. When the app already has bootstrap.enabled=false
// the method must return without issuing a Patch: the already-disabled subtest
// fails any Patch on the Postgres app, so removing the `if !Enabled { return nil }`
// guard turns the intended no-op into a failed Patch and reddens it. When the flag
// is true the method must patch it to false.
func TestDisablePostgresAppBootstrap_Idempotent(t *testing.T) {
const (
ns = "tenant"
appName = "app"
)
ctx := context.Background()
s := runtime.NewScheme()
_ = scheme.AddToScheme(s)
_ = postgresapp.AddToScheme(s)
t.Run("already disabled is a no-op that issues no Patch", func(t *testing.T) {
app := newPostgresApp(appName, ns)
app.Spec.Bootstrap.Enabled = false
c := clientfake.NewClientBuilder().
WithScheme(s).
WithObjects(app).
WithInterceptorFuncs(interceptor.Funcs{
Patch: func(ctx context.Context, cl client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
if _, ok := obj.(*postgresapp.Postgres); ok {
return fmt.Errorf("Patch must not be issued when bootstrap is already disabled")
}
return cl.Patch(ctx, obj, patch, opts...)
},
}).
Build()
r := &RestoreJobReconciler{Client: c}
if err := r.disablePostgresAppBootstrap(ctx, ns, appName); err != nil {
t.Fatalf("expected a no-op returning nil, got %v", err)
}
})
t.Run("enabled is patched to false", func(t *testing.T) {
app := newPostgresApp(appName, ns)
app.Spec.Bootstrap.Enabled = true
c := clientfake.NewClientBuilder().WithScheme(s).WithObjects(app).Build()
r := &RestoreJobReconciler{Client: c}
if err := r.disablePostgresAppBootstrap(ctx, ns, appName); err != nil {
t.Fatalf("disable: %v", err)
}
got := &postgresapp.Postgres{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: appName}, got); err != nil {
t.Fatalf("get app after disable: %v", err)
}
if got.Spec.Bootstrap.Enabled {
t.Fatalf("expected bootstrap.enabled=false after disable, still true")
}
})
}
// TestCNPGRestoreBackupID pins when a restore names the base backup it starts
// from. Without a backupID the barman-cloud plugin picks the newest backup of
// the stream (no recoveryTime) or the newest one ending by recoveryTime, on
// any timeline, which is not necessarily the backup the RestoreJob names.
func TestCNPGRestoreBackupID(t *testing.T) {
stopped := metav1.NewTime(time.Date(2026, 9, 21, 6, 27, 23, 0, time.UTC))
cases := []struct {
name string
backupID string
stoppedAt *metav1.Time
recoveryTime string
want string
}{
{"no recoveryTime pins the requested backup", "20260921T062713", &stopped, "", "20260921T062713"},
{"no recoveryTime pins it even without its end", "20260921T062713", nil, "", "20260921T062713"},
{"recoveryTime a second past the end pins it", "20260921T062713", &stopped, "2026-09-21T06:27:24Z", "20260921T062713"},
{"recoveryTime well after the end pins it", "20260921T062713", &stopped, "2026-09-22T10:00:00+02:00", "20260921T062713"},
{"recoveryTime within the truncated second leaves the choice to the plugin", "20260921T062713", &stopped, "2026-09-21T06:27:23.5Z", ""},
{"recoveryTime before the end leaves the choice to the plugin", "20260921T062713", &stopped, "2026-09-21T05:00:00Z", ""},
{"unknown end with a recoveryTime leaves the choice to the plugin", "20260921T062713", nil, "2026-09-22T10:00:00Z", ""},
{"unparseable recoveryTime leaves the choice to the plugin", "20260921T062713", &stopped, "2026-09-22 10:00:00", ""},
{"unknown backupID pins nothing", "", &stopped, "", ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
source := &cnpgtypes.Backup{Status: cnpgtypes.BackupStatus{BackupID: tc.backupID, StoppedAt: tc.stoppedAt}}
if got := cnpgRestoreBackupID(source, tc.recoveryTime); got != tc.want {
t.Errorf("cnpgRestoreBackupID(%q, %v, %q) = %q, want %q", tc.backupID, tc.stoppedAt, tc.recoveryTime, got, tc.want)
}
})
}
t.Run("a cnpg.io/Backup that is gone pins nothing", func(t *testing.T) {
if got := cnpgRestoreBackupID(nil, ""); got != "" {
t.Errorf("cnpgRestoreBackupID(nil, \"\") = %q, want none", got)
}
})
}
// TestBuildPostgresAppRestorePatch_ReplacesBackupID guards against a
// re-restore into the same target starting from the backup a previous
// restore pinned.
func TestBuildPostgresAppRestorePatch_ReplacesBackupID(t *testing.T) {
app := newPostgresApp("pg", "tenant")
app.Spec.Bootstrap.BackupID = "20260101T000000"
if got := buildPostgresAppRestorePatch(app, "src", "pg-new", "s3://b/", "", "", "20260921T062713", nil, nil, nil, nil).Spec.Bootstrap.BackupID; got != "20260921T062713" {
t.Errorf("bootstrap.backupID: got %q want %q", got, "20260921T062713")
}
if got := buildPostgresAppRestorePatch(app, "src", "pg-new", "s3://b/", "", "", "", nil, nil, nil, nil).Spec.Bootstrap.BackupID; got != "" {
t.Errorf("bootstrap.backupID from a previous restore must be cleared; got %q", got)
}
}
// TestReconcileCNPGRestore_PinsTheRequestedBackup drives a restore through the
// purge step and reads back what the target Postgres app is patched with: the
// barman backupId of the backup the RestoreJob names, read from its
// cnpg.io/Backup. Retention deletes that cnpg.io/Backup together with the base
// backup, and pinning an id the catalog has lost fails the recovery after the
// purge, so a cnpg.io/Backup that is gone leaves the choice to the plugin.
func TestReconcileCNPGRestore_PinsTheRequestedBackup(t *testing.T) {
const (
ns = "tenant"
appName = "app"
clusterName = "postgres-app"
cnpgBkName = "cnpgbk"
backupID = "20260921T062713"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
startedAt := metav1.NewTime(time.Now())
stoppedAt := metav1.NewTime(time.Date(2026, 9, 21, 6, 27, 23, 0, time.UTC))
mkBackupArtifact := func(t *testing.T) *backupsv1alpha1.Backup {
t.Helper()
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(appName, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
driverMD := map[string]string{
cnpgServerNameKey: appName,
cnpgDestinationPathKey: "s3://bucket/" + appName + "/",
cnpgBackupNameKey: cnpgBkName,
}
return &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: appName},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: driverMD,
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
}
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{
Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
},
},
}
mkCNPGBackup := func(id string) *cnpgtypes.Backup {
return &cnpgtypes.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: cnpgBkName},
Spec: cnpgtypes.BackupSpec{Cluster: cnpgtypes.ClusterReference{Name: clusterName}},
Status: cnpgtypes.BackupStatus{
Phase: cnpgBackupPhaseComplete, EndWal: "000000010000000000000003",
BackupID: id, StoppedAt: &stoppedAt,
},
}
}
mkRestoreJob := func(recoveryTime string) *backupsv1alpha1.RestoreJob {
sa := startedAt
rj := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
Status: backupsv1alpha1.RestoreJobStatus{StartedAt: &sa, Phase: backupsv1alpha1.RestoreJobPhaseRunning},
}
if recoveryTime != "" {
rj.Spec.Options = &runtime.RawExtension{Raw: []byte(`{"recoveryTime":"` + recoveryTime + `"}`)}
}
return rj
}
restoredBootstrap := func(t *testing.T, cnpgBackup *cnpgtypes.Backup, recoveryTime string) (postgresapp.Bootstrap, *record.FakeRecorder) {
t.Helper()
backup := mkBackupArtifact(t)
objs := []client.Object{backup, mkRestoreJob(recoveryTime), strategy, newPostgresApp(appName, ns)}
if cnpgBackup != nil {
objs = append(objs, cnpgBackup)
}
c := newCNPGStrategyTestClient(t, objs...)
recorder := record.NewFakeRecorder(10)
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: recorder}
rj := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, rj); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
if _, err := r.reconcileCNPGRestore(ctx, rj, backup); err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
app := &postgresapp.Postgres{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: appName}, app); err != nil {
t.Fatalf("get patched Postgres app: %v", err)
}
if !app.Spec.Bootstrap.Enabled {
t.Fatalf("the reconcile did not reach the restore patch")
}
return app.Spec.Bootstrap, recorder
}
t.Run("without recoveryTime the requested backup is pinned", func(t *testing.T) {
if b, _ := restoredBootstrap(t, mkCNPGBackup(backupID), ""); b.BackupID != backupID {
t.Errorf("bootstrap.backupID: got %q want %q", b.BackupID, backupID)
}
})
t.Run("a recoveryTime after the backup ended pins it", func(t *testing.T) {
b, _ := restoredBootstrap(t, mkCNPGBackup(backupID), "2026-09-22T10:00:00Z")
if b.BackupID != backupID || b.RecoveryTime != "2026-09-22T10:00:00Z" {
t.Errorf("bootstrap: got backupID %q recoveryTime %q, want %q and the requested instant", b.BackupID, b.RecoveryTime, backupID)
}
})
t.Run("a recoveryTime before the backup ended leaves the choice to the plugin", func(t *testing.T) {
if b, _ := restoredBootstrap(t, mkCNPGBackup(backupID), "2026-09-21T05:00:00Z"); b.BackupID != "" {
t.Errorf("bootstrap.backupID: got %q, want none: the backup cannot serve an instant before its end", b.BackupID)
}
})
for _, recoveryTime := range []string{"", "2026-09-22T10:00:00Z"} {
t.Run("a cnpg.io/Backup removed by retention leaves the choice to the plugin, recoveryTime="+recoveryTime, func(t *testing.T) {
b, recorder := restoredBootstrap(t, nil, recoveryTime)
if b.BackupID != "" {
t.Errorf("bootstrap.backupID: got %q, want none: the catalog no longer holds that base backup", b.BackupID)
}
select {
case ev := <-recorder.Events:
if !strings.Contains(ev, "Warning BaseBackupGone") {
t.Errorf("event: got %q, want a BaseBackupGone warning", ev)
}
default:
t.Errorf("no event: a restore that cannot start from the backup it names must say so")
}
})
}
}
func TestCreateCNPGBackupArtifact_RecordsMajorVersion(t *testing.T) {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
for _, tc := range []struct {
major int
want string
}{{major: 18, want: "18"}, {major: 0, want: ""}} {
t.Run(fmt.Sprintf("majorVersion=%d", tc.major), func(t *testing.T) {
r := &BackupJobReconciler{Client: newCNPGStrategyTestClient(t)}
j := &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
Spec: backupsv1alpha1.BackupJobSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: "pg"},
},
}
resolved := &ResolvedBackupConfig{
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "strat"},
}
cnpgBk := &cnpgtypes.Backup{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-bk"},
Status: cnpgtypes.BackupStatus{MajorVersion: tc.major},
}
rendered := &strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://b/"},
}
got, err := r.createCNPGBackupArtifact(context.Background(), j, resolved, cnpgBk, "postgres-pg", "postgres-pg", "postgresql", rendered, newPostgresApp("pg", "tenant"))
if err != nil {
t.Fatalf("createCNPGBackupArtifact: %v", err)
}
if v, ok := got.Spec.DriverMetadata[cnpgMajorVersionKey]; v != tc.want || ok != (tc.want != "") {
t.Errorf("driverMetadata[%s]: got %q (present=%v), want %q", cnpgMajorVersionKey, v, ok, tc.want)
}
})
}
}
// TestReconcileCNPGRestore_RefusesCrossMajorRestore drives the restore up to
// the purge with the backup and the target on different PostgreSQL majors.
// The purge used to run regardless, and the recovery could then never start:
// the target was left empty.
func TestReconcileCNPGRestore_RefusesCrossMajorRestore(t *testing.T) {
const (
ns = "tenant"
sourceApp = "app"
copyApp = "copy"
cnpgBkName = "cnpgbk"
)
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
ctx := context.Background()
startedAt := metav1.NewTime(time.Now())
strategy := &strategyv1alpha1.CNPG{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-strategy"},
Spec: strategyv1alpha1.CNPGSpec{
Template: strategyv1alpha1.CNPGTemplate{
BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://bucket/"},
},
},
}
for _, tc := range []struct {
name string
// recorded is the major the artifact's driverMetadata carries, "" for
// an artifact taken before the driver recorded it.
recorded string
// cnpgMajor is the cnpg.io/Backup's status.majorVersion; -1 for a
// cnpg.io/Backup that retention has removed.
cnpgMajor int
targetVersion string
toCopy bool
wantRefused bool
}{
{name: "v18 backup into a v16 copy", recorded: "18", cnpgMajor: 18, targetVersion: "v16", toCopy: true, wantRefused: true},
{name: "in place into an app upgraded since its v16 backup", recorded: "16", cnpgMajor: 16, targetVersion: "v18", wantRefused: true},
{name: "artifact without the record falls back to the cnpg.io/Backup", cnpgMajor: 18, targetVersion: "v16", toCopy: true, wantRefused: true},
{name: "same major restores", recorded: "16", cnpgMajor: 16, targetVersion: "v16", toCopy: true},
{name: "unknown backup major restores", cnpgMajor: -1, targetVersion: "v16", toCopy: true},
{name: "cnpg.io/Backup without majorVersion restores", cnpgMajor: 0, targetVersion: "v16"},
{name: "unknown target version restores", recorded: "18", cnpgMajor: 18, targetVersion: ""},
} {
t.Run(tc.name, func(t *testing.T) {
targetApp := sourceApp
if tc.toCopy {
targetApp = copyApp
}
targetCluster := cnpgClusterNameForApp(targetApp)
snap, err := marshalCNPGBackupSnapshot(newPostgresApp(sourceApp, ns), nil, "")
if err != nil {
t.Fatalf("marshal snapshot: %v", err)
}
driverMD := map[string]string{
cnpgServerNameKey: sourceApp,
cnpgDestinationPathKey: "s3://bucket/" + sourceApp + "/",
cnpgBackupNameKey: cnpgBkName,
}
if tc.recorded != "" {
driverMD[cnpgMajorVersionKey] = tc.recorded
}
backup := &backupsv1alpha1.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "bk"},
Spec: backupsv1alpha1.BackupSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: sourceApp},
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "cnpg-strategy"},
DriverMetadata: driverMD,
},
Status: backupsv1alpha1.BackupStatus{UnderlyingResources: snap},
}
rj := &backupsv1alpha1.RestoreJob{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "rj"},
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
Status: backupsv1alpha1.RestoreJobStatus{StartedAt: &startedAt, Phase: backupsv1alpha1.RestoreJobPhaseRunning},
}
if tc.toCopy {
rj.Spec.TargetApplicationRef = &corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: copyApp}
}
app := newPostgresApp(targetApp, ns)
app.Spec.Version = tc.targetVersion
objs := []client.Object{backup, rj, strategy, app,
&cnpgtypes.Cluster{ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: targetCluster}},
&corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{
Namespace: ns, Name: targetCluster + "-1", Labels: map[string]string{cnpgClusterLabel: targetCluster},
}},
}
if tc.cnpgMajor >= 0 {
objs = append(objs, &cnpgtypes.Backup{
ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: cnpgBkName},
Spec: cnpgtypes.BackupSpec{Cluster: cnpgtypes.ClusterReference{Name: cnpgClusterNameForApp(sourceApp)}},
Status: cnpgtypes.BackupStatus{
Phase: cnpgBackupPhaseComplete, EndWal: "000000010000000000000003", MajorVersion: tc.cnpgMajor,
},
})
}
s := runtime.NewScheme()
_ = scheme.AddToScheme(s)
_ = backupsv1alpha1.AddToScheme(s)
_ = strategyv1alpha1.AddToScheme(s)
_ = cnpgtypes.AddToScheme(s)
_ = postgresapp.AddToScheme(s)
var deleted []string
c := clientfake.NewClientBuilder().
WithScheme(s).
WithObjects(objs...).
WithStatusSubresource(&backupsv1alpha1.BackupJob{}, &backupsv1alpha1.RestoreJob{}, &backupsv1alpha1.Backup{}).
WithInterceptorFuncs(interceptor.Funcs{
Delete: func(ctx context.Context, cl client.WithWatch, obj client.Object, opts ...client.DeleteOption) error {
deleted = append(deleted, fmt.Sprintf("%T %s", obj, obj.GetName()))
return cl.Delete(ctx, obj, opts...)
},
}).
Build()
r := &RestoreJobReconciler{Client: c, Interface: dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t)), Recorder: record.NewFakeRecorder(10)}
live := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, live); err != nil {
t.Fatalf("get seeded RestoreJob: %v", err)
}
if _, err := r.reconcileCNPGRestore(ctx, live, backup); err != nil {
t.Fatalf("reconcileCNPGRestore: %v", err)
}
got := &backupsv1alpha1.RestoreJob{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: "rj"}, got); err != nil {
t.Fatalf("get RestoreJob after reconcile: %v", err)
}
patched := &postgresapp.Postgres{}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: targetApp}, patched); err != nil {
t.Fatalf("get target Postgres app: %v", err)
}
if !tc.wantRefused {
if got.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
t.Fatalf("restore failed: %s", got.Status.Message)
}
if len(deleted) == 0 || !patched.Spec.Bootstrap.Enabled {
t.Fatalf("restore did not reach the purge: deleted=%v bootstrap.enabled=%v", deleted, patched.Spec.Bootstrap.Enabled)
}
return
}
if got.Status.Phase != backupsv1alpha1.RestoreJobPhaseFailed {
t.Errorf("expected phase Failed, got %q", got.Status.Phase)
}
if cond := apimeta.FindStatusCondition(got.Status.Conditions, "Ready"); cond == nil || cond.Reason != "MajorVersionMismatch" {
t.Errorf("expected Ready=False reason MajorVersionMismatch, got %+v", cond)
}
for _, major := range []string{"PostgreSQL 16", "PostgreSQL 18"} {
if !strings.Contains(got.Status.Message, major) {
t.Errorf("message must name both majors, %q missing from %q", major, got.Status.Message)
}
}
if len(deleted) != 0 {
t.Errorf("a refused restore must delete nothing, deleted %v", deleted)
}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: targetCluster}, &cnpgtypes.Cluster{}); err != nil {
t.Errorf("target Cluster must survive: %v", err)
}
if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: targetCluster + "-1"}, &corev1.PersistentVolumeClaim{}); err != nil {
t.Errorf("target PVC must survive: %v", err)
}
if patched.Spec.Bootstrap.Enabled {
t.Errorf("a refused restore must not patch the target app")
}
})
}
}