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 <androndo@gmail.com>

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) {

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")
 	}
 }

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)
+	}
+}

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() {

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)
+	}
+}

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() {