Skip to content

Commit eafa441

Browse files
mattia-eleuterimattia-eleuteri
committed
fix(backups): create no driver object for a job that already finished
The BackupJob and RestoreJob reconcilers decide a job is still active from the informer cache, and the cache can trail their own writes. A driver's first pass stamps StartedAt, which requeues the job, and may then fail the job in the same pass on a transient error, such as a 503 from the aggregated apps.cozystack.io API while it reads the application. The requeued pass can start before the informer has seen Failed: it retries the step that failed and, once the error has cleared, creates the driver object anyway. Its own status write then conflicts, the pass after it sees Failed and skips the job, and a Velero Backup is left running that no Backup artifact will reference and nothing tracks. Against a real apiserver and the manager's cache, with one 503 on the first application read of each job, 5 to 10 BackupJobs out of 40 ended Failed with a Velero Backup. A restore pass that starts on such a stale copy does worse, since the step it retries can be destructive: suspending the VM's HelmRelease and halting the VM before creating a Velero Restore, or patching the Postgres application and purging its cluster. Both reconcilers now re-read the job through the manager's uncached reader when the cached copy says it is still active, and skip it when the apiserver holds it as finished. The check sits ahead of the dispatch, so it covers every driver, and a job the cache already reports as finished costs no extra request. Assisted-by: LLM Signed-off-by: Mattia Eleuteri <[email protected]>
1 parent 6addef4 commit eafa441

3 files changed

Lines changed: 246 additions & 3 deletions

File tree

‎internal/backupcontroller/backupjob_controller.go‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,10 @@ type BackupJobReconciler struct {
3535
Scheme *runtime.Scheme
3636
Recorder record.EventRecorder
3737
CredentialsConfig BackupCredentialsConfig
38+
// APIReader is the manager's uncached reader, through which Reconcile
39+
// re-reads the BackupJob before dispatching to a driver. Wired in
40+
// SetupWithManager.
41+
APIReader client.Reader
3842
}
3943

4044
func (r *BackupJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -62,6 +66,24 @@ func (r *BackupJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
6266
return ctrl.Result{}, nil
6367
}
6468

69+
// The cache can trail this reconciler's own writes. A pass that stamps
70+
// StartedAt and then fails the job has already requeued it through the
71+
// first write, so the next pass can start before the informer has seen
72+
// Failed, retry the step that failed, and create a driver object for a
73+
// job that is over and that nothing will track. Re-read the job past the
74+
// cache; the check above only spares finished jobs the extra request.
75+
if err := r.apiReader().Get(ctx, req.NamespacedName, j); err != nil {
76+
if apierrors.IsNotFound(err) {
77+
return ctrl.Result{}, nil
78+
}
79+
logger.Error(err, "failed to re-read BackupJob")
80+
return ctrl.Result{}, err
81+
}
82+
if j.Status.Phase == backupsv1alpha1.BackupJobPhaseSucceeded || j.Status.Phase == backupsv1alpha1.BackupJobPhaseFailed {
83+
logger.V(1).Info("BackupJob terminal on the apiserver ahead of the cache, skipping", "phase", j.Status.Phase)
84+
return ctrl.Result{}, nil
85+
}
86+
6587
// Normalize ApplicationRef (default apiGroup if not specified)
6688
normalizedAppRef := NormalizeApplicationRef(j.Spec.ApplicationRef)
6789

@@ -230,6 +252,7 @@ func (r *BackupJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
230252
return err
231253
}
232254

255+
r.APIReader = mgr.GetAPIReader()
233256
cfg := mgr.GetConfig()
234257
var err error
235258
if r.Interface, err = dynamic.NewForConfig(cfg); err != nil {
@@ -247,6 +270,16 @@ func (r *BackupJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
247270
Complete(r)
248271
}
249272

273+
// apiReader returns the uncached reader. Reconcilers built by hand in unit
274+
// tests have none and fall back to their fake Client, which has no cache to
275+
// bypass.
276+
func (r *BackupJobReconciler) apiReader() client.Reader {
277+
if r.APIReader != nil {
278+
return r.APIReader
279+
}
280+
return r.Client
281+
}
282+
250283
// handleProjectionError classifies a credentials-projection error as
251284
// transient (requeue) or terminal (mark Failed). Transient errors record
252285
// a Ready=False condition without setting Phase=Failed so the BackupJob

‎internal/backupcontroller/restorejob_controller.go‎

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,9 +49,10 @@ type RestoreJobReconciler struct {
4949
// (missing pods/log RBAC) branch - is unit-testable without a live cluster.
5050
readPodLog func(ctx context.Context, namespace, podName, container string) (string, error)
5151
CredentialsConfig BackupCredentialsConfig
52-
// APIReader is the manager's uncached reader. The keepOriginalPVC rename
53-
// reads back objects it has just written, which the cache may not have
54-
// seen yet. Wired in SetupWithManager.
52+
// APIReader is the manager's uncached reader. Reconcile re-reads the
53+
// RestoreJob through it before dispatching to a driver, and the
54+
// keepOriginalPVC rename reads back objects it has just written, which
55+
// the cache may not have seen yet. Wired in SetupWithManager.
5556
APIReader client.Reader
5657
}
5758

@@ -104,6 +105,22 @@ func (r *RestoreJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
104105
return ctrl.Result{}, nil
105106
}
106107

108+
// Re-read past the cache for the reason BackupJobReconciler.Reconcile
109+
// gives. Here the step retried for a job already over can be destructive:
110+
// halting the VM, renaming its PVCs, deleting the Postgres cluster.
111+
if err := r.apiReader().Get(ctx, req.NamespacedName, restoreJob); err != nil {
112+
if apierrors.IsNotFound(err) {
113+
return ctrl.Result{}, nil
114+
}
115+
logger.Error(err, "failed to re-read RestoreJob")
116+
return ctrl.Result{}, err
117+
}
118+
if restoreJob.Status.Phase == backupsv1alpha1.RestoreJobPhaseSucceeded ||
119+
restoreJob.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
120+
logger.V(1).Info("RestoreJob terminal on the apiserver ahead of the cache, skipping", "phase", restoreJob.Status.Phase)
121+
return ctrl.Result{}, nil
122+
}
123+
107124
// Step 1: Fetch the referenced Backup
108125
backup := &backupsv1alpha1.Backup{}
109126
backupKey := types.NamespacedName{Namespace: req.Namespace, Name: restoreJob.Spec.BackupRef.Name}

‎internal/backupcontroller/stale_cache_test.go‎

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
package backupcontroller
3+
4+
import (
5+
"context"
6+
"testing"
7+
"time"
8+
9+
corev1 "k8s.io/api/core/v1"
10+
"k8s.io/apimachinery/pkg/api/meta"
11+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
12+
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
13+
"k8s.io/apimachinery/pkg/runtime"
14+
"k8s.io/apimachinery/pkg/runtime/schema"
15+
dynamicfake "k8s.io/client-go/dynamic/fake"
16+
"k8s.io/client-go/kubernetes/scheme"
17+
"k8s.io/client-go/tools/record"
18+
"sigs.k8s.io/controller-runtime/pkg/client"
19+
clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
20+
"sigs.k8s.io/controller-runtime/pkg/reconcile"
21+
22+
strategyv1alpha1 "github.com/cozystack/cozystack/api/backups/strategy/v1alpha1"
23+
backupsv1alpha1 "github.com/cozystack/cozystack/api/backups/v1alpha1"
24+
velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
25+
)
26+
27+
// staleCacheScheme registers what both the cache and the apiserver stand-in
28+
// hold.
29+
func staleCacheScheme() *runtime.Scheme {
30+
s := runtime.NewScheme()
31+
_ = scheme.AddToScheme(s)
32+
_ = backupsv1alpha1.AddToScheme(s)
33+
_ = strategyv1alpha1.AddToScheme(s)
34+
_ = velerov1.AddToScheme(s)
35+
return s
36+
}
37+
38+
// apiServerHolding stands in for the apiserver: it holds job as the cache
39+
// does, except for its phase.
40+
func apiServerHolding(s *runtime.Scheme, job client.Object, phase string) client.Reader {
41+
onAPI := job.DeepCopyObject().(client.Object)
42+
switch j := onAPI.(type) {
43+
case *backupsv1alpha1.BackupJob:
44+
j.Status.Phase = backupsv1alpha1.BackupJobPhase(phase)
45+
case *backupsv1alpha1.RestoreJob:
46+
j.Status.Phase = backupsv1alpha1.RestoreJobPhase(phase)
47+
}
48+
onAPI.SetResourceVersion("")
49+
return clientfake.NewClientBuilder().WithScheme(s).WithObjects(onAPI).Build()
50+
}
51+
52+
func countIn(t *testing.T, c client.Client, list client.ObjectList) int {
53+
t.Helper()
54+
if err := c.List(context.Background(), list); err != nil {
55+
t.Fatalf("list %T: %v", list, err)
56+
}
57+
return meta.LenList(list)
58+
}
59+
60+
// TestReconcile_DoesNotRetryAJobItAlreadyFailed reproduces a pass that
61+
// starts on a cache trailing the controller's own writes: the previous pass
62+
// stamped StartedAt, which requeued the job, then failed it on a transient
63+
// error. The cache holds the job with StartedAt and no phase; the apiserver
64+
// holds it Failed. Before the re-read, the pass retried the Velero Backup
65+
// creation and the job ended Failed with a Velero Backup nothing tracks.
66+
func TestReconcile_DoesNotRetryAJobItAlreadyFailed(t *testing.T) {
67+
appsGroup := "apps.cozystack.io"
68+
strategyGroup := strategyv1alpha1.GroupVersion.Group
69+
started := metav1.NewTime(time.Now().Add(-time.Second).Truncate(time.Second))
70+
71+
run := func(t *testing.T, apiPhase backupsv1alpha1.BackupJobPhase) (created int, cached backupsv1alpha1.BackupJobPhase) {
72+
s := staleCacheScheme()
73+
bc := &backupsv1alpha1.BackupClass{
74+
ObjectMeta: metav1.ObjectMeta{Name: "bc"},
75+
Spec: backupsv1alpha1.BackupClassSpec{Strategies: []backupsv1alpha1.BackupClassStrategy{{
76+
Application: backupsv1alpha1.ApplicationSelector{APIGroup: &appsGroup, Kind: "Bucket"},
77+
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.VeleroStrategyKind, Name: "velero"},
78+
}}},
79+
}
80+
j := &backupsv1alpha1.BackupJob{
81+
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
82+
Spec: backupsv1alpha1.BackupJobSpec{
83+
BackupClassName: "bc",
84+
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &appsGroup, Kind: "Bucket", Name: "app"},
85+
},
86+
Status: backupsv1alpha1.BackupJobStatus{StartedAt: &started},
87+
}
88+
strategy := &strategyv1alpha1.Velero{ObjectMeta: metav1.ObjectMeta{Name: "velero"}}
89+
c := clientfake.NewClientBuilder().WithScheme(s).WithObjects(j, bc, strategy).
90+
WithStatusSubresource(&backupsv1alpha1.BackupJob{}).Build()
91+
gvr := schema.GroupVersionResource{Group: appsGroup, Version: "v1alpha1", Resource: "buckets"}
92+
app := &unstructured.Unstructured{}
93+
app.SetGroupVersionKind(gvr.GroupVersion().WithKind("Bucket"))
94+
app.SetNamespace("tenant")
95+
app.SetName("app")
96+
r := &BackupJobReconciler{
97+
Client: c,
98+
APIReader: apiServerHolding(s, j, string(apiPhase)),
99+
Interface: dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{gvr: "BucketList"}, app),
100+
RESTMapper: &mockRESTMapper{mapping: &meta.RESTMapping{
101+
Resource: gvr,
102+
GroupVersionKind: gvr.GroupVersion().WithKind("Bucket"),
103+
Scope: meta.RESTScopeNamespace,
104+
}},
105+
Scheme: s,
106+
Recorder: record.NewFakeRecorder(10),
107+
}
108+
if _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: client.ObjectKeyFromObject(j)}); err != nil {
109+
t.Fatalf("Reconcile: %v", err)
110+
}
111+
got := &backupsv1alpha1.BackupJob{}
112+
if err := c.Get(context.Background(), client.ObjectKeyFromObject(j), got); err != nil {
113+
t.Fatalf("get BackupJob: %v", err)
114+
}
115+
return countIn(t, c, &velerov1.BackupList{}), got.Status.Phase
116+
}
117+
118+
if created, _ := run(t, ""); created != 1 {
119+
t.Fatalf("job still active on the apiserver: %d Velero Backups created, want 1", created)
120+
}
121+
created, cached := run(t, backupsv1alpha1.BackupJobPhaseFailed)
122+
if created != 0 {
123+
t.Errorf("job Failed on the apiserver: %d Velero Backups created, want 0", created)
124+
}
125+
if cached != "" {
126+
t.Errorf("the pass wrote phase %q over a job it should have left alone", cached)
127+
}
128+
}
129+
130+
// TestReconcileRestore_DoesNotRetryAJobItAlreadyFailed is the restore-side
131+
// counterpart. The retried step is destructive there: before creating the
132+
// Velero Restore, the pass suspends the VM's HelmRelease and halts the VM.
133+
func TestReconcileRestore_DoesNotRetryAJobItAlreadyFailed(t *testing.T) {
134+
const ns = "tenant"
135+
strategyGroup := strategyv1alpha1.GroupVersion.Group
136+
hrName := vmNamePrefix + "vm"
137+
started := metav1.NewTime(time.Now().Add(-time.Second).Truncate(time.Second))
138+
139+
run := func(t *testing.T, apiPhase backupsv1alpha1.RestoreJobPhase) (restores int, hrSuspended bool, cached backupsv1alpha1.RestoreJobPhase) {
140+
s := staleCacheScheme()
141+
backup := &backupsv1alpha1.Backup{
142+
ObjectMeta: metav1.ObjectMeta{Name: "bk", Namespace: ns},
143+
Spec: backupsv1alpha1.BackupSpec{
144+
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: new("apps.cozystack.io"), Kind: vmInstanceKind, Name: "vm"},
145+
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.VeleroStrategyKind, Name: "velero"},
146+
DriverMetadata: map[string]string{veleroBackupNameMetadataKey: "vb"},
147+
},
148+
}
149+
rj := &backupsv1alpha1.RestoreJob{
150+
ObjectMeta: metav1.ObjectMeta{Name: "rj", Namespace: ns, Finalizers: []string{restoreJobFinalizer}},
151+
Spec: backupsv1alpha1.RestoreJobSpec{BackupRef: corev1.LocalObjectReference{Name: "bk"}},
152+
Status: backupsv1alpha1.RestoreJobStatus{StartedAt: &started, Phase: backupsv1alpha1.RestoreJobPhaseRunning},
153+
}
154+
strategy := &strategyv1alpha1.Velero{ObjectMeta: metav1.ObjectMeta{Name: "velero"}}
155+
c := clientfake.NewClientBuilder().WithScheme(s).WithObjects(rj, backup, strategy).
156+
WithStatusSubresource(&backupsv1alpha1.RestoreJob{}).Build()
157+
dyn := dynamicfake.NewSimpleDynamicClient(testCNPGScheme(t), makeUnstructuredHelmRelease(hrName, ns, nil))
158+
r := &RestoreJobReconciler{
159+
Client: c,
160+
APIReader: apiServerHolding(s, rj, string(apiPhase)),
161+
Interface: dyn,
162+
Scheme: s,
163+
Recorder: record.NewFakeRecorder(100),
164+
}
165+
if _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: client.ObjectKeyFromObject(rj)}); err != nil {
166+
t.Fatalf("Reconcile: %v", err)
167+
}
168+
hr, err := dyn.Resource(helmReleaseGVR).Namespace(ns).Get(context.Background(), hrName, metav1.GetOptions{})
169+
if err != nil {
170+
t.Fatalf("get HelmRelease: %v", err)
171+
}
172+
hrSuspended, _, _ = unstructured.NestedBool(hr.Object, "spec", "suspend")
173+
got := &backupsv1alpha1.RestoreJob{}
174+
if err := c.Get(context.Background(), client.ObjectKeyFromObject(rj), got); err != nil {
175+
t.Fatalf("get RestoreJob: %v", err)
176+
}
177+
return countIn(t, c, &velerov1.RestoreList{}), hrSuspended, got.Status.Phase
178+
}
179+
180+
if restores, suspended, _ := run(t, backupsv1alpha1.RestoreJobPhaseRunning); restores != 1 || !suspended {
181+
t.Fatalf("job still active on the apiserver: %d Velero Restores, HelmRelease suspended=%v; want 1 and true", restores, suspended)
182+
}
183+
restores, suspended, cached := run(t, backupsv1alpha1.RestoreJobPhaseFailed)
184+
if suspended {
185+
t.Error("job Failed on the apiserver: the pass suspended the VM's HelmRelease")
186+
}
187+
if restores != 0 {
188+
t.Errorf("job Failed on the apiserver: %d Velero Restores created, want 0", restores)
189+
}
190+
if cached != backupsv1alpha1.RestoreJobPhaseRunning {
191+
t.Errorf("the pass wrote phase %q over a job it should have left alone", cached)
192+
}
193+
}

0 commit comments

Comments
 (0)