Skip to content

Commit d67bbf2

Browse files
scooby87scooby87
committed
fix(postgres): time the plugin rollout from the write that attached it
The PluginRolledOut latch is recorded before the Cluster write that attaches the barman-cloud plugin, and its time was used both as the start of the rollout deadline and as the age every instance pod must exceed to count as rolled. Now that a failing write is retried for up to the backup deadline, the attach can come long after the latch: the deadline was then spent before the rollout began, and a pod recreated between the latch and the attach, without the plugin sidecar, passed as rolled, so a Backup was created that CloudNativePG fails for good. The latch now carries its own reason, and the write that finds the plugin missing and attaches it re-stamps the condition, once, with the time it was attempted. A plugin someone else attached after the latch, another backup job or a restore, keeps the latch time: the rollout it started can be over by the time this job writes, and its instances would otherwise never count as rolled. Signed-off-by: Alexey Artamonov <[email protected]> Assisted-by: LLM
1 parent 048b350 commit d67bbf2

2 files changed

Lines changed: 140 additions & 0 deletions

File tree

‎internal/backupcontroller/cnpgstrategy_controller.go‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,11 @@ const (
197197
// instance pods must postdate (see cnpgPluginRolloutPending).
198198
backupCondPluginRolledOut = "PluginRolledOut"
199199

200+
// Reason of the PluginRolledOut latch recorded before the write that
201+
// attaches the plugin. The write that attaches it replaces the latch with
202+
// its own time, which retries can put long after the latch.
203+
cnpgReasonAttachingPlugin = "AttachingPlugin"
204+
200205
// Condition Type recorded while writing the WAL archive destination keeps
201206
// failing. Its LastTransitionTime marks the first failure, so the retry
202207
// budget is not spent by time the job waited for another backup of the
@@ -356,6 +361,7 @@ func (r *BackupJobReconciler) reconcileCNPG(ctx context.Context, j *backupsv1alp
356361
}
357362
}
358363

364+
attemptedAt := metav1.Now()
359365
attachment, err := r.applyClusterPluginBackup(ctx, j.Namespace, clusterName, rendered, serverName, existingBackup != nil || awaitingRollout, writeDestination)
360366
if err != nil {
361367
if apierrors.IsNotFound(err) {
@@ -419,9 +425,34 @@ func (r *BackupJobReconciler) reconcileCNPG(ctx context.Context, j *backupsv1alp
419425
apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{
420426
Type: backupCondPluginRolledOut,
421427
Status: metav1.ConditionFalse,
428+
Reason: cnpgReasonAttachingPlugin,
429+
Message: message,
430+
})
431+
apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{
432+
Type: "Ready",
433+
Status: metav1.ConditionFalse,
422434
Reason: "WaitingForPluginRollout",
423435
Message: message,
424436
})
437+
if err := r.Status().Update(ctx, j); err != nil {
438+
return ctrl.Result{}, err
439+
}
440+
return ctrl.Result{RequeueAfter: cnpgPollInterval}, nil
441+
}
442+
if rollout.Reason == cnpgReasonAttachingPlugin && attachment.missing {
443+
// Instances recreated between the latch and this write run
444+
// without the plugin, and the rollout deadline starts here. A
445+
// plugin someone else attached since the latch keeps it: the
446+
// rollout that attach started may already be over.
447+
message := fmt.Sprintf("attached the barman-cloud plugin to cnpg.io/Cluster %s/%s, waiting for its instances to roll", j.Namespace, clusterName)
448+
apimeta.RemoveStatusCondition(&j.Status.Conditions, backupCondPluginRolledOut)
449+
apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{
450+
Type: backupCondPluginRolledOut,
451+
Status: metav1.ConditionFalse,
452+
Reason: "WaitingForPluginRollout",
453+
Message: message,
454+
LastTransitionTime: attemptedAt,
455+
})
425456
apimeta.SetStatusCondition(&j.Status.Conditions, metav1.Condition{
426457
Type: "Ready",
427458
Status: metav1.ConditionFalse,

‎internal/backupcontroller/cnpgstrategy_controller_test.go‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2814,6 +2814,115 @@ func TestReconcileCNPG_PluginRolloutDeadlineCountsFromAttach(t *testing.T) {
28142814
}
28152815
}
28162816

2817+
// The rollout is measured from the write that attached the plugin, not from
2818+
// the latch recorded before it: retries of a failing write can lie between
2819+
// the two, and an instance recreated in that time runs without the plugin.
2820+
func TestReconcileCNPG_PluginRolloutCountsFromTheSuccessfulAttach(t *testing.T) {
2821+
cases := []struct {
2822+
name string
2823+
// podBetween recreates the instance after the latch and before the
2824+
// attach; otherwise it predates both.
2825+
podBetween bool
2826+
}{
2827+
{name: "an instance recreated before the attach still has to roll", podBetween: true},
2828+
{name: "retries before the attach do not spend the rollout deadline"},
2829+
}
2830+
for _, tc := range cases {
2831+
t.Run(tc.name, func(t *testing.T) {
2832+
expired := time.Now().Add(-(cnpgDefaultBackupDeadline + time.Minute)).Truncate(time.Second)
2833+
oldPod := newRolloutInstancePod(expired.Add(-time.Hour))
2834+
f := newCNPGRolloutFixture(t, newRolloutBackupJob(expired.Add(-time.Hour)),
2835+
newRolloutCluster(cnpgClusterHealthyPhase), oldPod)
2836+
2837+
j, _ := f.pass() // latch
2838+
latch := apimeta.FindStatusCondition(j.Status.Conditions, backupCondPluginRolledOut)
2839+
if latch == nil {
2840+
t.Fatalf("no %s latch: %+v", backupCondPluginRolledOut, j.Status.Conditions)
2841+
}
2842+
latch.LastTransitionTime = metav1.NewTime(expired)
2843+
if err := f.c.Status().Update(context.Background(), j); err != nil {
2844+
t.Fatalf("age the latch: %v", err)
2845+
}
2846+
2847+
failing := true
2848+
f.r.Client = interceptor.NewClient(f.c.(client.WithWatch), interceptor.Funcs{
2849+
Patch: func(ctx context.Context, cl client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
2850+
if _, ok := obj.(*cnpgtypes.Cluster); ok && failing {
2851+
return apierrors.NewInternalError(errors.New(`failed calling webhook "mcluster.cnpg.io"`))
2852+
}
2853+
return cl.Patch(ctx, obj, patch, opts...)
2854+
},
2855+
})
2856+
if j, _ = f.pass(); j.Status.Phase == backupsv1alpha1.BackupJobPhaseFailed {
2857+
t.Fatalf("a failed attach failed the job: %s", j.Status.Message)
2858+
}
2859+
if tc.podBetween {
2860+
f.replacePod(oldPod, newRolloutInstancePod(expired.Add(time.Minute)))
2861+
}
2862+
2863+
failing = false
2864+
attachAt := time.Now().Truncate(time.Second)
2865+
j, backups := f.pass()
2866+
if len(backups) != 0 {
2867+
t.Fatalf("a cnpg.io/Backup was created in the pass that attached the plugin: %+v", backups)
2868+
}
2869+
rollout := apimeta.FindStatusCondition(j.Status.Conditions, backupCondPluginRolledOut)
2870+
if rollout == nil || rollout.Status != metav1.ConditionFalse || rollout.LastTransitionTime.Time.Before(attachAt) {
2871+
t.Fatalf("%s = %+v, want False since the attach at %s", backupCondPluginRolledOut, rollout, attachAt)
2872+
}
2873+
2874+
attached := &cnpgtypes.Cluster{}
2875+
if err := f.c.Get(context.Background(), client.ObjectKey{Namespace: rolloutNS, Name: rolloutCluster}, attached); err != nil {
2876+
t.Fatalf("get Cluster: %v", err)
2877+
}
2878+
f.observeCNPG(newRolloutCluster(cnpgClusterHealthyPhase, attached.Spec.Plugins...))
2879+
j, backups = f.pass()
2880+
if j.Status.Phase == backupsv1alpha1.BackupJobPhaseFailed {
2881+
t.Fatalf("a rollout that began with the attach failed: %s", j.Status.Message)
2882+
}
2883+
if len(backups) != 0 {
2884+
t.Fatalf("a cnpg.io/Backup was created against an instance older than the attach: %+v", backups)
2885+
}
2886+
if ready := apimeta.FindStatusCondition(j.Status.Conditions, "Ready"); ready == nil || !strings.Contains(ready.Message, "predates") {
2887+
t.Errorf("Ready = %+v, want the instance that predates the plugin", ready)
2888+
}
2889+
})
2890+
}
2891+
}
2892+
2893+
// A plugin attached by someone else after the latch, another backup job or
2894+
// a restore, started the rollout before this job's write, which only finds
2895+
// it there: instances that rolled since the latch carry the plugin.
2896+
func TestReconcileCNPG_PluginAttachedByAnotherWriterKeepsTheLatch(t *testing.T) {
2897+
oldPod := newRolloutInstancePod(time.Now().Add(-time.Hour).Truncate(time.Second))
2898+
f := newCNPGRolloutFixture(t, newRolloutBackupJob(time.Now().Add(-time.Hour)),
2899+
newRolloutCluster(cnpgClusterHealthyPhase), oldPod)
2900+
2901+
j, _ := f.pass() // latch
2902+
latch := apimeta.FindStatusCondition(j.Status.Conditions, backupCondPluginRolledOut)
2903+
if latch == nil {
2904+
t.Fatalf("no %s latch: %+v", backupCondPluginRolledOut, j.Status.Conditions)
2905+
}
2906+
latchedAt := time.Now().Add(-10 * time.Minute).Truncate(time.Second)
2907+
latch.LastTransitionTime = metav1.NewTime(latchedAt)
2908+
if err := f.c.Status().Update(context.Background(), j); err != nil {
2909+
t.Fatalf("age the latch: %v", err)
2910+
}
2911+
f.observeCNPG(newRolloutCluster(cnpgClusterHealthyPhase, buildBarmanPlugin(rolloutCluster, rolloutCluster)))
2912+
f.replacePod(oldPod, newRolloutInstancePod(latchedAt.Add(5*time.Minute)))
2913+
2914+
var backups []cnpgtypes.Backup
2915+
for range 3 {
2916+
if j, backups = f.pass(); len(backups) != 0 || j.Status.Phase == backupsv1alpha1.BackupJobPhaseFailed {
2917+
break
2918+
}
2919+
}
2920+
if len(backups) != 1 {
2921+
t.Fatalf("got %d cnpg.io/Backups against instances that rolled with the plugin; Ready = %+v",
2922+
len(backups), apimeta.FindStatusCondition(j.Status.Conditions, "Ready"))
2923+
}
2924+
}
2925+
28172926
// Waiting for another backup resets the retry budget of the destination
28182927
// write, so the queue does not spend it.
28192928
func TestReconcileCNPG_QueueResetsDestinationRetryBudget(t *testing.T) {

0 commit comments

Comments
 (0)