From 2b2176357761ac32a1215f19eeaf4e5cbde224a5 Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Thu, 22 Oct 2026 18:18:51 +0400 Subject: [PATCH] fix(backups): do not fail a backup on a transient retry-check read The check for a Backup an earlier reconcile already created returned its read error, and the drivers turn any error from that path into a terminally Failed BackupJob. An apiserver timeout right after the operator backup completed therefore lost a backup whose data was safely in S3. The check only saves a probe, and the Create that follows already handles an existing Backup, so log the error and go on to Create. A trailing slash on the S3 endpoint is not a path prefix; the operators accept it, so do not refuse it. Pin the uncached AlreadyExists refetch, the upper slack of the MariaDB window and the probe timeout with tests. Assisted-by: LLM Signed-off-by: Andrey Kolkov --- internal/backupcontroller/artifact_size.go | 30 ++++--- .../backupcontroller/artifact_size_test.go | 85 +++++++++++++++++-- .../cnpg_artifact_size_test.go | 28 ++++++ .../cnpgstrategy_controller.go | 4 +- .../mariadb_artifact_size_test.go | 61 +++++++++++++ .../mariadbstrategy_controller.go | 4 +- 6 files changed, 190 insertions(+), 22 deletions(-) diff --git a/internal/backupcontroller/artifact_size.go b/internal/backupcontroller/artifact_size.go index 517fe07a6..182ea0b58 100644 --- a/internal/backupcontroller/artifact_size.go +++ b/internal/backupcontroller/artifact_size.go @@ -22,8 +22,8 @@ import ( // makes to measure Backup.status.artifact.sizeBytes. The probe runs inside // the reconcile that creates the Backup, and the BackupJob controller has a // single worker, so an unreachable endpoint stalls every other BackupJob for -// this long. -const artifactSizeProbeTimeout = 10 * time.Second +// this long. A variable so tests can shorten it. +var artifactSizeProbeTimeout = 10 * time.Second type s3ObjectInfo struct { Key string @@ -63,8 +63,10 @@ type minioObjectLister struct { func newMinioObjectLister(conn s3Connection) (s3ObjectLister, error) { // An endpoint behind a path prefix cannot be addressed: minio-go has no // such option, and the SigV4 signature covers the request path, so the - // prefix cannot be added below the client either. - if strings.Contains(conn.Endpoint, "/") { + // prefix cannot be added below the client either. A bare trailing slash + // is not a prefix, and the operators accept it. + endpoint := strings.TrimSuffix(conn.Endpoint, "/") + if strings.Contains(endpoint, "/") { return nil, fmt.Errorf("endpoint %q has a path, which the S3 client cannot address", conn.Endpoint) } opts := &minio.Options{ @@ -90,7 +92,7 @@ func newMinioObjectLister(conn s3Connection) (s3ObjectLister, error) { tr.TLSClientConfig.RootCAs = pool opts.Transport = tr } - client, err := minio.New(conn.Endpoint, opts) + client, err := minio.New(endpoint, opts) if err != nil { return nil, err } @@ -121,18 +123,20 @@ func collectObjects(ctx context.Context, objects <-chan minio.ObjectInfo) ([]s3O // existingBackup returns the Backup a previous reconcile already created // for the job, so a retry skips the size probe whose result it would -// discard. Returns (nil, nil) when there is none. It reads past the cache: -// a retry typically follows a Create made moments earlier, which the cache -// may not show yet. -func (r *BackupJobReconciler) existingBackup(ctx context.Context, namespace, name string) (*backupsv1alpha1.Backup, error) { +// discard. It reads past the cache: a retry typically follows a Create made +// moments earlier, which the cache may not show yet. It returns nil both +// when there is no Backup and when the read fails: it only saves a probe, +// and the Create that follows handles an existing Backup on its own, so a +// transient read error must not fail a backup that succeeded. +func (r *BackupJobReconciler) existingBackup(ctx context.Context, namespace, name string) *backupsv1alpha1.Backup { existing := &backupsv1alpha1.Backup{} if err := r.apiReader().Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, existing); err != nil { - if apierrors.IsNotFound(err) { - return nil, nil + if !apierrors.IsNotFound(err) { + getLogger(ctx).Info("cannot check for an existing Backup; creating it", "backup", name, "error", err) } - return nil, err + return nil } - return existing, nil + return existing } func (r *BackupJobReconciler) secretValue(ctx context.Context, namespace, name, key string) (string, error) { diff --git a/internal/backupcontroller/artifact_size_test.go b/internal/backupcontroller/artifact_size_test.go index f24079f8b..0e66968b5 100644 --- a/internal/backupcontroller/artifact_size_test.go +++ b/internal/backupcontroller/artifact_size_test.go @@ -5,12 +5,15 @@ import ( "context" "errors" "testing" + "time" "github.com/minio/minio-go/v7" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" backupsv1alpha1 "github.com/cozystack/cozystack/api/backups/v1alpha1" ) @@ -87,8 +90,10 @@ func TestNewMinioObjectLister_RejectsEndpointPath(t *testing.T) { if _, err := newMinioObjectLister(s3Connection{Endpoint: "s3.example.org/prefix", Secure: true}); err == nil { t.Error("an endpoint with a path was accepted") } - if _, err := newMinioObjectLister(s3Connection{Endpoint: "s3.example.org:9000", Secure: true}); err != nil { - t.Errorf("host:port endpoint rejected: %v", err) + for _, ok := range []string{"s3.example.org:9000", "s3.example.org/"} { + if _, err := newMinioObjectLister(s3Connection{Endpoint: ok, Secure: true}); err != nil { + t.Errorf("endpoint %q rejected: %v", ok, err) + } } } @@ -101,8 +106,78 @@ func TestExistingBackup_ReadsPastTheCache(t *testing.T) { Client: newMariaDBStrategyTestClient(t), APIReader: newMariaDBStrategyTestClient(t, backup), } - got, err := r.existingBackup(context.Background(), "tenant", "bj") - if err != nil || got == nil { - t.Fatalf("existingBackup = %v, %v; want the Backup the API server holds", got, err) + if got := r.existingBackup(context.Background(), "tenant", "bj"); got == nil { + t.Fatal("existingBackup = nil; want the Backup the API server holds") + } +} + +// staleCache wraps a fake client so that Get reports NotFound for Backups, +// the way an informer cache does right after another writer created one. +func staleCache(t *testing.T, c client.Client) client.Client { + t.Helper() + return interceptor.NewClient(c.(client.WithWatch), interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*backupsv1alpha1.Backup); ok { + return apierrors.NewNotFound(backupsv1alpha1.GroupVersion.WithResource("backups").GroupResource(), key.Name) + } + return cl.Get(ctx, key, obj, opts...) + }, + }) +} + +// notFoundOnce wraps a fake client whose first Backup Get reports NotFound: +// the retry check runs before the Backup exists, the refetch after Create +// collides with it. +func notFoundOnce(t *testing.T, c client.Client) client.Client { + t.Helper() + missed := false + return interceptor.NewClient(c.(client.WithWatch), interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*backupsv1alpha1.Backup); ok && !missed { + missed = true + return apierrors.NewNotFound(backupsv1alpha1.GroupVersion.WithResource("backups").GroupResource(), key.Name) + } + return cl.Get(ctx, key, obj, opts...) + }, + }) +} + +func TestExistingBackup_ReadErrorIsNotFatal(t *testing.T) { + failing := interceptor.NewClient(newMariaDBStrategyTestClient(t).(client.WithWatch), interceptor.Funcs{ + Get: func(context.Context, client.WithWatch, client.ObjectKey, client.Object, ...client.GetOption) error { + return errors.New("etcdserver: request timed out") + }, + }) + r := &BackupJobReconciler{Client: newMariaDBStrategyTestClient(t), APIReader: failing} + if got := r.existingBackup(context.Background(), "tenant", "bj"); got != nil { + t.Errorf("existingBackup = %#v, want nil on a read error", got) + } +} + +// blockingLister answers nothing until the probe's context is done, the way +// a black-holed endpoint does. +type blockingLister struct{} + +func (blockingLister) ListObjects(ctx context.Context, _, _ string, _ bool) ([]s3ObjectInfo, error) { + <-ctx.Done() + return nil, ctx.Err() +} + +// withShortProbeTimeout shortens the probe timeout and fails the test if fn +// is not back well before the default timeout would have expired. +func withShortProbeTimeout(t *testing.T, fn func()) { + t.Helper() + saved := artifactSizeProbeTimeout + artifactSizeProbeTimeout = 50 * time.Millisecond + defer func() { artifactSizeProbeTimeout = saved }() + done := make(chan struct{}) + go func() { + fn() + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("the probe did not give up at its timeout") } } diff --git a/internal/backupcontroller/cnpg_artifact_size_test.go b/internal/backupcontroller/cnpg_artifact_size_test.go index 13b647bdd..70bb2a485 100644 --- a/internal/backupcontroller/cnpg_artifact_size_test.go +++ b/internal/backupcontroller/cnpg_artifact_size_test.go @@ -83,6 +83,16 @@ func TestCNPGArtifactSizeBytes(t *testing.T) { } }) + t.Run("a hanging endpoint is given up at the probe timeout", func(t *testing.T) { + r := reconcilerWithLister(t, &fakeS3Lister{}, s3CredsSecret("tenant", "creds")) + r.newS3ObjectLister = func(s3Connection) (s3ObjectLister, error) { return blockingLister{}, nil } + withShortProbeTimeout(t, func() { + if got := r.cnpgArtifactSizeBytes(context.Background(), "tenant", newSizedBarmanObjectStore(), "srv", "20261021T100000"); got != 0 { + t.Errorf("size = %d, want 0", got) + } + }) + }) + t.Run("an empty endpointURL means AWS", func(t *testing.T) { store := newSizedBarmanObjectStore() store.EndpointURL = "" @@ -306,3 +316,21 @@ func TestCreateCNPGBackupArtifact_ExistingBackupSkipsProbe(t *testing.T) { t.Errorf("S3 listed %d times for a Backup that already exists", l.calls) } } + +func TestCreateCNPGBackupArtifact_AlreadyExistsRefetchReadsPastTheCache(t *testing.T) { + apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup + j := &backupsv1alpha1.BackupJob{ + ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj-race"}, + Spec: backupsv1alpha1.BackupJobSpec{ + ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: "pg"}, + }, + } + backup := &backupsv1alpha1.Backup{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj-race"}} + store := newCNPGStrategyTestClient(t, backup) + r := &BackupJobReconciler{Client: staleCache(t, store), APIReader: notFoundOnce(t, store)} + rendered := &strategyv1alpha1.CNPGTemplate{BarmanObjectStore: strategyv1alpha1.BarmanObjectStoreTemplate{DestinationPath: "s3://b/"}} + got, err := r.createCNPGBackupArtifact(context.Background(), j, &ResolvedBackupConfig{}, &cnpgtypes.Backup{ObjectMeta: metav1.ObjectMeta{Name: "cnpg-bk"}}, "postgres-pg", "srv", "postgresql", rendered, newPostgresApp("pg", "tenant")) + if err != nil || got == nil || got.Name != "bj-race" { + t.Fatalf("createCNPGBackupArtifact = %v, %v; want the Backup that already exists", got, err) + } +} diff --git a/internal/backupcontroller/cnpgstrategy_controller.go b/internal/backupcontroller/cnpgstrategy_controller.go index 5dabfa66c..712dd8713 100644 --- a/internal/backupcontroller/cnpgstrategy_controller.go +++ b/internal/backupcontroller/cnpgstrategy_controller.go @@ -815,8 +815,8 @@ func (r *BackupJobReconciler) createCNPGBackupArtifact( rendered *strategyv1alpha1.CNPGTemplate, sourceApp *postgresapp.Postgres, ) (*backupsv1alpha1.Backup, error) { - if existing, err := r.existingBackup(ctx, j.Namespace, j.Name); err != nil || existing != nil { - return existing, err + if existing := r.existingBackup(ctx, j.Namespace, j.Name); existing != nil { + return existing, nil } takenAt := metav1.Now() if cnpgBackup.Status.StartedAt != nil && !cnpgBackup.Status.StartedAt.IsZero() { diff --git a/internal/backupcontroller/mariadb_artifact_size_test.go b/internal/backupcontroller/mariadb_artifact_size_test.go index 6f9df56e7..91f555bbd 100644 --- a/internal/backupcontroller/mariadb_artifact_size_test.go +++ b/internal/backupcontroller/mariadb_artifact_size_test.go @@ -13,6 +13,8 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" strategyv1alpha1 "github.com/cozystack/cozystack/api/backups/strategy/v1alpha1" backupsv1alpha1 "github.com/cozystack/cozystack/api/backups/v1alpha1" @@ -151,6 +153,25 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) { } }) + t.Run("a dump uploaded just after Complete is inside the slack", func(t *testing.T) { + l := &fakeS3Lister{objects: []s3ObjectInfo{ + {Key: dumpKey(prefix, completed.Add(30*time.Second), "gzip"), Size: 5, LastModified: completed.Add(30 * time.Second)}, + }} + if got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 5 { + t.Errorf("size = %d, want 5", got) + } + }) + + t.Run("a hanging endpoint is given up at the probe timeout", func(t *testing.T) { + r := reconcilerWithLister(t, &fakeS3Lister{}, s3CredsSecret("tenant", "creds")) + r.newS3ObjectLister = func(s3Connection) (s3ObjectLister, error) { return blockingLister{}, nil } + withShortProbeTimeout(t, func() { + if got := r.mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 { + t.Errorf("size = %d, want 0", got) + } + }) + }) + t.Run("two dumps in the window leave size unset", func(t *testing.T) { l := &fakeS3Lister{objects: []s3ObjectInfo{ {Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 1, LastModified: created.Add(time.Minute)}, @@ -298,3 +319,43 @@ func TestCreateMariaDBBackupArtifact_ExistingBackupSkipsProbe(t *testing.T) { t.Errorf("Artifact = %#v, want the existing one", got.Status.Artifact) } } + +func TestCreateMariaDBBackupArtifact_TransientRetryReadStillCreates(t *testing.T) { + apps := mariadbapp.GroupName + job := &backupsv1alpha1.BackupJob{ + ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj-transient"}, + Spec: backupsv1alpha1.BackupJobSpec{ + ApplicationRef: corev1.TypedLocalObjectReference{Kind: "MariaDB", Name: "src", APIGroup: &apps}, + }, + } + r := reconcilerWithLister(t, &fakeS3Lister{}, job) + r.APIReader = interceptor.NewClient(newMariaDBStrategyTestClient(t).(client.WithWatch), interceptor.Funcs{ + Get: func(context.Context, client.WithWatch, client.ObjectKey, client.Object, ...client.GetOption) error { + return errors.New("etcdserver: request timed out") + }, + }) + got, err := r.createMariaDBBackupArtifact(context.Background(), job, &ResolvedBackupConfig{}, &mariadbtypes.Backup{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "op"}}, newRenderedMariaDBTemplate()) + if err != nil || got == nil || got.Name != "bj-transient" { + t.Fatalf("createMariaDBBackupArtifact = %v, %v; a transient read must not fail the backup", got, err) + } + if err := r.Get(context.Background(), client.ObjectKeyFromObject(got), &backupsv1alpha1.Backup{}); err != nil { + t.Errorf("the Backup was not created: %v", err) + } +} + +func TestCreateMariaDBBackupArtifact_AlreadyExistsRefetchReadsPastTheCache(t *testing.T) { + apps := mariadbapp.GroupName + job := &backupsv1alpha1.BackupJob{ + ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj-race"}, + Spec: backupsv1alpha1.BackupJobSpec{ + ApplicationRef: corev1.TypedLocalObjectReference{Kind: "MariaDB", Name: "src", APIGroup: &apps}, + }, + } + backup := &backupsv1alpha1.Backup{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj-race"}} + store := newMariaDBStrategyTestClient(t, job, backup) + r := &BackupJobReconciler{Client: staleCache(t, store), APIReader: notFoundOnce(t, store), Scheme: store.Scheme()} + got, err := r.createMariaDBBackupArtifact(context.Background(), job, &ResolvedBackupConfig{}, &mariadbtypes.Backup{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "op"}}, &strategyv1alpha1.MariaDBTemplate{}) + if err != nil || got == nil || got.Name != "bj-race" { + t.Fatalf("createMariaDBBackupArtifact = %v, %v; want the Backup that already exists", got, err) + } +} diff --git a/internal/backupcontroller/mariadbstrategy_controller.go b/internal/backupcontroller/mariadbstrategy_controller.go index f6abed32d..8db41fb7f 100644 --- a/internal/backupcontroller/mariadbstrategy_controller.go +++ b/internal/backupcontroller/mariadbstrategy_controller.go @@ -365,8 +365,8 @@ func (r *BackupJobReconciler) createMariaDBBackupArtifact( mdbBackup *mariadbtypes.Backup, rendered *strategyv1alpha1.MariaDBTemplate, ) (*backupsv1alpha1.Backup, error) { - if existing, err := r.existingBackup(ctx, j.Namespace, j.Name); err != nil || existing != nil { - return existing, err + if existing := r.existingBackup(ctx, j.Namespace, j.Name); existing != nil { + return existing, nil } takenAt := metav1.Now() if cond := apimeta.FindStatusCondition(mdbBackup.Status.Conditions, mariadbtypes.ConditionTypeComplete); cond != nil && !cond.LastTransitionTime.IsZero() {