Skip to content

Commit 6ac8e07

Browse files
scooby87scooby87
committed
fix(postgres): read cnpg Backups from the apiserver before writing the destination
The queue that keeps two BackupJobs of one application from writing the WAL archive destination at once listed cnpg.io/Backups from the informer cache. The controller runs one reconcile at a time under leader election, so a Backup created by one job's pass already exists when the next pass starts, but the cache can still miss it. The next job then took the queue as empty, rewrote the destination under the running backup and left that backup's recorded destination naming a place its WAL no longer went. A job could likewise miss its own fresh Backup, write the destination again and create a second Backup. The queue now lists Backups from the apiserver, and a job that finds no Backup of its own in the cache checks the apiserver before it writes or creates one. Only a leader handover, when two processes reconcile for a moment, can still let two jobs through. The pass that creates the Backup no longer looks it up a second time right before the create. Nothing can create it in between, and once the lookup can reach the apiserver, a failed second lookup would come after the destination write and fail the job instead of retrying. Assisted-by: LLM Signed-off-by: Alexey Artamonov <[email protected]>
1 parent 5a5bced commit 6ac8e07

2 files changed

Lines changed: 150 additions & 15 deletions

File tree

‎internal/backupcontroller/cnpgstrategy_controller.go‎

Lines changed: 23 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -531,7 +531,7 @@ func (r *BackupJobReconciler) reconcileCNPG(ctx context.Context, j *backupsv1alp
531531
"cluster", clusterName, "live", attachment.flavor, "application", specFlavor)
532532
}
533533

534-
cnpgBackup, err := r.ensureCNPGBackup(ctx, j, clusterName, cnpgDestinationOf(rendered))
534+
cnpgBackup, err := r.ensureCNPGBackup(ctx, j, clusterName, existingBackup, cnpgDestinationOf(rendered))
535535
if err != nil {
536536
return r.markBackupJobFailed(ctx, j, fmt.Sprintf("failed to ensure cnpg.io/Backup: %v", err))
537537
}
@@ -876,18 +876,16 @@ func currentBarmanServerName(c *cnpgtypes.Cluster) string {
876876
}
877877

878878
// ensureCNPGBackup creates a one-shot postgresql.cnpg.io/Backup CR labelled
879-
// with the BackupJob, or returns the existing one if a previous reconcile
880-
// already created it. Idempotency relies on the OwningJob labels.
879+
// with the BackupJob, or returns existing, the job's Backup as looked up at
880+
// the start of the pass. It does not look again: a reconcile of the same
881+
// job cannot run in between, and a failed lookup here would come after the
882+
// destination write and fail the job.
881883
//
882884
// written is the destination this pass wrote to the ObjectStore. A pass
883885
// reaches the create only after that write: a job without a Backup writes
884886
// the destination on every pass, and a pass that wrote nothing, or wrote it
885887
// and waits for the plugin rollout, returns before this point.
886-
func (r *BackupJobReconciler) ensureCNPGBackup(ctx context.Context, j *backupsv1alpha1.BackupJob, clusterName string, written cnpgDestination) (*cnpgtypes.Backup, error) {
887-
existing, err := r.findCNPGBackupForJob(ctx, j)
888-
if err != nil {
889-
return nil, err
890-
}
888+
func (r *BackupJobReconciler) ensureCNPGBackup(ctx context.Context, j *backupsv1alpha1.BackupJob, clusterName string, existing *cnpgtypes.Backup, written cnpgDestination) (*cnpgtypes.Backup, error) {
891889
if existing != nil {
892890
return existing, nil
893891
}
@@ -923,17 +921,27 @@ func (r *BackupJobReconciler) ensureCNPGBackup(ctx context.Context, j *backupsv1
923921
// findCNPGBackupForJob returns the cnpg.io/Backup labelled with the
924922
// BackupJob's OwningJob{Name,Namespace}, if any. Returns (nil, nil) when
925923
// no match is found.
924+
//
925+
// A miss in the cache is checked against the apiserver: the cache can lag
926+
// behind a Backup the previous pass created, and taking that miss as no
927+
// Backup would write the destination under it and create a second one.
926928
func (r *BackupJobReconciler) findCNPGBackupForJob(ctx context.Context, j *backupsv1alpha1.BackupJob) (*cnpgtypes.Backup, error) {
927-
list := &cnpgtypes.BackupList{}
928-
if err := r.List(ctx, list,
929+
opts := []client.ListOption{
929930
client.InNamespace(j.Namespace),
930931
client.MatchingLabels{
931932
backupsv1alpha1.OwningJobNameLabel: j.Name,
932933
backupsv1alpha1.OwningJobNamespaceLabel: j.Namespace,
933934
},
934-
); err != nil {
935+
}
936+
list := &cnpgtypes.BackupList{}
937+
if err := r.List(ctx, list, opts...); err != nil {
935938
return nil, err
936939
}
940+
if len(list.Items) == 0 {
941+
if err := r.apiReader().List(ctx, list, opts...); err != nil {
942+
return nil, err
943+
}
944+
}
937945
if len(list.Items) == 0 {
938946
return nil, nil
939947
}
@@ -944,9 +952,12 @@ func (r *BackupJobReconciler) findCNPGBackupForJob(ctx context.Context, j *backu
944952
// running and whose cnpg.io/Backup of clusterName has not finished, or "" when
945953
// there is none. A Backup left behind by a finished or deleted job does not
946954
// hold the queue.
955+
//
956+
// The Backups are read from the apiserver, not the cache, which can still
957+
// miss a Backup another job's earlier pass created.
947958
func (r *BackupJobReconciler) otherCNPGBackupInFlight(ctx context.Context, j *backupsv1alpha1.BackupJob, clusterName string) (string, error) {
948959
list := &cnpgtypes.BackupList{}
949-
if err := r.List(ctx, list, client.InNamespace(j.Namespace)); err != nil {
960+
if err := r.apiReader().List(ctx, list, client.InNamespace(j.Namespace)); err != nil {
950961
return "", err
951962
}
952963
var holders []string

‎internal/backupcontroller/cnpgstrategy_controller_test.go‎

Lines changed: 127 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"context"
66
"errors"
77
"fmt"
8+
"slices"
89
"strings"
910
"testing"
1011
"time"
@@ -2207,7 +2208,7 @@ func TestEnsureCNPGBackup_UsesPluginMethod(t *testing.T) {
22072208
c := newCNPGStrategyTestClient(t, j)
22082209
r := &BackupJobReconciler{Client: c}
22092210

2210-
bk, err := r.ensureCNPGBackup(context.Background(), j, "postgres-app", cnpgDestination{DestinationPath: "s3://bucket/"})
2211+
bk, err := r.ensureCNPGBackup(context.Background(), j, "postgres-app", nil, cnpgDestination{DestinationPath: "s3://bucket/"})
22112212
if err != nil {
22122213
t.Fatalf("ensureCNPGBackup: %v", err)
22132214
}
@@ -2567,8 +2568,8 @@ func (f *cnpgRolloutFixture) seedOtherBackupJob(name string, phase backupsv1alph
25672568
func TestReconcileCNPG_RunningBackupDoesNotReapplyObjectStore(t *testing.T) {
25682569
f := newAttachedObjectStoreFixture(t, time.Now(), "s3://other/")
25692570
f.seedCNPGBackup("bj-cnpg", "bj", rolloutCluster, "running")
2570-
// Two jobs can pass the queue together, since it reads the cache; once
2571-
// both have a Backup neither may wait on the other.
2571+
// Two jobs can still pass the queue together, across a leader handover;
2572+
// once both have a Backup neither may wait on the other.
25722573
f.seedOtherBackupJob("bj-other", backupsv1alpha1.BackupJobPhaseRunning)
25732574
f.seedCNPGBackup("bj-other-cnpg", "bj-other", rolloutCluster, "running")
25742575
writes := 0
@@ -5655,3 +5656,126 @@ func TestReconcileCNPG_ArtifactOfABackupWithoutARecord(t *testing.T) {
56555656
t.Fatalf("artifact = %+v, want one built from the strategy", artifact)
56565657
}
56575658
}
5659+
5660+
// cacheLaggingBehind makes the reconciler's cached client miss the named
5661+
// cnpg.io/Backups, as an informer does right after they are created, while
5662+
// its uncached reader sees them.
5663+
func (f *cnpgRolloutFixture) cacheLaggingBehind(hidden ...string) {
5664+
f.r.APIReader = f.c
5665+
f.r.Client = interceptor.NewClient(f.c.(client.WithWatch), interceptor.Funcs{
5666+
List: func(ctx context.Context, cl client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
5667+
if err := cl.List(ctx, list, opts...); err != nil {
5668+
return err
5669+
}
5670+
if backups, ok := list.(*cnpgtypes.BackupList); ok {
5671+
backups.Items = slices.DeleteFunc(backups.Items, func(b cnpgtypes.Backup) bool {
5672+
return slices.Contains(hidden, b.Name)
5673+
})
5674+
}
5675+
return nil
5676+
},
5677+
})
5678+
}
5679+
5680+
// Reconciles run one at a time, so a Backup created by one pass exists by
5681+
// the next; the queue has to see it even when the cache does not yet, or
5682+
// the next job rewrites the destination under it and the record on that
5683+
// Backup no longer names where its WAL goes.
5684+
func TestReconcileCNPG_QueueSeesABackupTheCacheHasNotCaught(t *testing.T) {
5685+
f := newAttachedObjectStoreFixture(t, time.Now(), "s3://other/")
5686+
f.seedOtherBackupJob("bj-other", backupsv1alpha1.BackupJobPhaseRunning)
5687+
f.seedCNPGBackup("bj-other-cnpg", "bj-other", rolloutCluster, "running")
5688+
f.cacheLaggingBehind("bj-other-cnpg")
5689+
5690+
j, backups := f.pass()
5691+
if ready := apimeta.FindStatusCondition(j.Status.Conditions, "Ready"); ready == nil || ready.Reason != "WaitingForOtherBackup" {
5692+
t.Fatalf("Ready = %+v, want WaitingForOtherBackup", ready)
5693+
}
5694+
if got := f.liveDestination(); got != "s3://other/" {
5695+
t.Fatalf("destination = %q, rewritten under the other job's Backup", got)
5696+
}
5697+
if len(backups) != 1 {
5698+
t.Fatalf("got %d cnpg.io/Backups, want only the other job's", len(backups))
5699+
}
5700+
}
5701+
5702+
// The job's own Backup, missing from the cache right after it was created,
5703+
// is neither created again nor written under.
5704+
func TestReconcileCNPG_OwnBackupTheCacheHasNotCaught(t *testing.T) {
5705+
f := newAttachedObjectStoreFixture(t, time.Now(), "s3://other/")
5706+
f.seedCNPGBackup("bj-cnpg", "bj", rolloutCluster, "running")
5707+
f.cacheLaggingBehind("bj-cnpg")
5708+
5709+
_, backups := f.pass()
5710+
if len(backups) != 1 {
5711+
t.Fatalf("got %d cnpg.io/Backups, want the job's one", len(backups))
5712+
}
5713+
if got := f.liveDestination(); got != "s3://other/" {
5714+
t.Fatalf("destination = %q, rewritten under the job's own Backup", got)
5715+
}
5716+
}
5717+
5718+
// failingBackupLists makes the uncached reader fail the cnpg.io/Backup lists
5719+
// for which fail, given whether the list selects the job's own Backup by
5720+
// label and how many such lists came before, returns true.
5721+
func (f *cnpgRolloutFixture) failingBackupLists(fail func(own bool, n int) bool) {
5722+
counts := map[bool]int{}
5723+
f.r.APIReader = interceptor.NewClient(f.c.(client.WithWatch), interceptor.Funcs{
5724+
List: func(ctx context.Context, cl client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
5725+
if _, ok := list.(*cnpgtypes.BackupList); ok {
5726+
o := &client.ListOptions{}
5727+
o.ApplyOptions(opts)
5728+
own := o.LabelSelector != nil && !o.LabelSelector.Empty()
5729+
counts[own]++
5730+
if fail(own, counts[own]) {
5731+
return apierrors.NewServiceUnavailable("apiserver unavailable")
5732+
}
5733+
}
5734+
return cl.List(ctx, list, opts...)
5735+
},
5736+
})
5737+
}
5738+
5739+
// A failed apiserver read of the Backups is retried, not taken as the job
5740+
// failing: nothing has been written yet that the retry would not redo.
5741+
func TestReconcileCNPG_BackupListErrorIsReturned(t *testing.T) {
5742+
for name, own := range map[string]bool{"own Backup": true, "queue": false} {
5743+
t.Run(name, func(t *testing.T) {
5744+
f := newAttachedObjectStoreFixture(t, time.Now(), "s3://other/")
5745+
f.failingBackupLists(func(o bool, _ int) bool { return o == own })
5746+
j := &backupsv1alpha1.BackupJob{}
5747+
if err := f.c.Get(context.Background(), client.ObjectKey{Namespace: rolloutNS, Name: "bj"}, j); err != nil {
5748+
t.Fatalf("get BackupJob: %v", err)
5749+
}
5750+
if _, err := f.r.reconcileCNPG(context.Background(), j, f.resolved); err == nil {
5751+
t.Fatalf("reconcileCNPG swallowed the Backup list error")
5752+
}
5753+
got := &backupsv1alpha1.BackupJob{}
5754+
if err := f.c.Get(context.Background(), client.ObjectKey{Namespace: rolloutNS, Name: "bj"}, got); err != nil {
5755+
t.Fatalf("get BackupJob: %v", err)
5756+
}
5757+
if got.Status.Phase == backupsv1alpha1.BackupJobPhaseFailed {
5758+
t.Fatalf("BackupJob failed on a Backup list error: %s", got.Status.Message)
5759+
}
5760+
if dest := f.liveDestination(); dest != "s3://other/" {
5761+
t.Fatalf("destination = %q, written although the Backups could not be read", dest)
5762+
}
5763+
})
5764+
}
5765+
}
5766+
5767+
// The pass that writes the destination looks its own Backup up once, at the
5768+
// start: a second lookup before the create could only fail, after the write,
5769+
// and fail the job with it.
5770+
func TestReconcileCNPG_LooksUpItsOwnBackupOncePerPass(t *testing.T) {
5771+
f := newAttachedObjectStoreFixture(t, time.Now(), "s3://other/")
5772+
f.failingBackupLists(func(own bool, n int) bool { return own && n > 1 })
5773+
5774+
j, backups := f.pass()
5775+
if j.Status.Phase == backupsv1alpha1.BackupJobPhaseFailed {
5776+
t.Fatalf("BackupJob failed: %s", j.Status.Message)
5777+
}
5778+
if len(backups) != 1 {
5779+
t.Fatalf("got %d cnpg.io/Backups, want 1", len(backups))
5780+
}
5781+
}

0 commit comments

Comments
 (0)