From f222e52863ff2432329c3971c8a3d1882a6e379d Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Wed, 21 Oct 2026 12:12:11 +0400 Subject: [PATCH] fix(backups): harden the artifact size probe The probe runs inside the reconcile that creates the Backup, and the BackupJob controller has a single worker, so its timeout delays every other BackupJob while an endpoint hangs. Lower it from 30s to 10s. List the MariaDB prefix non-recursively, as mariadb-operator does. The default MariaDB prefix and the default CNPG destinationPath are the same // path, so a recursive listing of a MariaDB backed up next to a same-named Postgres walked the whole WAL archive and ran out of time. The CNPG base-backup prefix is still listed recursively. Build the custom-CA transport from minio.DefaultTransport so it keeps the proxy-from-environment and dial/TLS timeout settings a bare http.Transport drops, and cover it with a test against a TLS endpoint with a self-signed certificate. Log the reasons the size could not be measured (credentials, client, listing, empty CNPG prefix) at info. The chart runs the controller at info, which hid them before. The expected ambiguity of several MariaDB dumps in the window stays at debug. Skip an endpointCA without a Secret name, as buildBarmanObjectStore does. Assisted-by: LLM Signed-off-by: Andrey Kolkov --- internal/backupcontroller/artifact_size.go | 32 +++++---- .../backupcontroller/artifact_size_test.go | 5 +- .../backupcontroller/cnpg_artifact_size.go | 14 ++-- .../cnpg_artifact_size_test.go | 71 ++++++++++++++----- .../backupcontroller/mariadb_artifact_size.go | 11 +-- .../mariadb_artifact_size_test.go | 34 +++++++-- 6 files changed, 119 insertions(+), 48 deletions(-) diff --git a/internal/backupcontroller/artifact_size.go b/internal/backupcontroller/artifact_size.go index 6ca2ad3ea..0c6d52f19 100644 --- a/internal/backupcontroller/artifact_size.go +++ b/internal/backupcontroller/artifact_size.go @@ -3,11 +3,9 @@ package backupcontroller import ( "context" - "crypto/tls" "crypto/x509" "errors" "fmt" - "net/http" "time" "github.com/minio/minio-go/v7" @@ -17,9 +15,11 @@ import ( ) // artifactSizeProbeTimeout bounds the object-store round trips a driver -// makes to measure Backup.status.artifact.sizeBytes; the probe runs inside -// the reconcile that creates the Backup. -const artifactSizeProbeTimeout = 30 * time.Second +// 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 type s3ObjectInfo struct { Key string @@ -27,7 +27,7 @@ type s3ObjectInfo struct { } type s3ObjectLister interface { - ListObjects(ctx context.Context, bucket, prefix string) ([]s3ObjectInfo, error) + ListObjects(ctx context.Context, bucket, prefix string, recursive bool) ([]s3ObjectInfo, error) } // s3Connection is what a driver resolves from its rendered strategy to @@ -62,14 +62,22 @@ func newMinioObjectLister(conn s3Connection) (s3ObjectLister, error) { Secure: conn.Secure, } if conn.Secure && len(conn.CAPEM) > 0 { - pool, err := x509.SystemCertPool() + // Start from minio's transport to keep its proxy and timeout settings. + tr, err := minio.DefaultTransport(true) if err != nil { - pool = x509.NewCertPool() + return nil, err + } + pool := tr.TLSClientConfig.RootCAs + if pool == nil { + if pool, err = x509.SystemCertPool(); err != nil { + pool = x509.NewCertPool() + } } if !pool.AppendCertsFromPEM(conn.CAPEM) { return nil, errors.New("no PEM certificates in the endpoint CA bundle") } - opts.Transport = &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}} + tr.TLSClientConfig.RootCAs = pool + opts.Transport = tr } client, err := minio.New(conn.Endpoint, opts) if err != nil { @@ -78,9 +86,9 @@ func newMinioObjectLister(conn s3Connection) (s3ObjectLister, error) { return &minioObjectLister{client: client}, nil } -func (l *minioObjectLister) ListObjects(ctx context.Context, bucket, prefix string) ([]s3ObjectInfo, error) { +func (l *minioObjectLister) ListObjects(ctx context.Context, bucket, prefix string, recursive bool) ([]s3ObjectInfo, error) { var out []s3ObjectInfo - for obj := range l.client.ListObjects(ctx, bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: true}) { + for obj := range l.client.ListObjects(ctx, bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: recursive}) { if obj.Err != nil { return nil, obj.Err } @@ -96,7 +104,7 @@ func (r *BackupJobReconciler) secretValue(ctx context.Context, namespace, name, } v, ok := secret.Data[key] if !ok { - return "", fmt.Errorf("Secret %s/%s has no key %q", namespace, name, key) + return "", fmt.Errorf("secret %s/%s has no key %q", namespace, name, key) } return string(v), nil } diff --git a/internal/backupcontroller/artifact_size_test.go b/internal/backupcontroller/artifact_size_test.go index 6e389cf66..c0a957852 100644 --- a/internal/backupcontroller/artifact_size_test.go +++ b/internal/backupcontroller/artifact_size_test.go @@ -16,10 +16,11 @@ type fakeS3Lister struct { conn s3Connection bucket, prefix string + recursive bool } -func (f *fakeS3Lister) ListObjects(_ context.Context, bucket, prefix string) ([]s3ObjectInfo, error) { - f.bucket, f.prefix = bucket, prefix +func (f *fakeS3Lister) ListObjects(_ context.Context, bucket, prefix string, recursive bool) ([]s3ObjectInfo, error) { + f.bucket, f.prefix, f.recursive = bucket, prefix, recursive return f.objects, f.err } diff --git a/internal/backupcontroller/cnpg_artifact_size.go b/internal/backupcontroller/cnpg_artifact_size.go index bff2d5b39..61b00d5e9 100644 --- a/internal/backupcontroller/cnpg_artifact_size.go +++ b/internal/backupcontroller/cnpg_artifact_size.go @@ -32,23 +32,23 @@ func (r *BackupJobReconciler) cnpgArtifactSizeBytes( bucket, path, err := parseBarmanS3DestinationPath(store.DestinationPath) if err != nil { - logger.Debug("cannot measure CNPG backup size", "backupID", backupID, "error", err) + logger.Info("cannot measure CNPG backup size", "backupID", backupID, "error", err) return 0 } conn, err := r.cnpgS3Connection(ctx, namespace, store) if err != nil { - logger.Debug("cannot resolve S3 credentials to measure CNPG backup size", "backupID", backupID, "error", err) + logger.Info("cannot resolve S3 credentials to measure CNPG backup size", "backupID", backupID, "error", err) return 0 } lister, err := r.s3Lister(conn) if err != nil { - logger.Debug("cannot build S3 client to measure CNPG backup size", "backupID", backupID, "error", err) + logger.Info("cannot build S3 client to measure CNPG backup size", "backupID", backupID, "error", err) return 0 } prefix := barmanBaseBackupPrefix(path, serverName, backupID) - objects, err := lister.ListObjects(ctx, bucket, prefix) + objects, err := lister.ListObjects(ctx, bucket, prefix, true) if err != nil { - logger.Debug("cannot list CNPG base backup objects to measure size", "backupID", backupID, "error", err) + logger.Info("cannot list CNPG base backup objects to measure size", "backupID", backupID, "error", err) return 0 } var total int64 @@ -56,7 +56,7 @@ func (r *BackupJobReconciler) cnpgArtifactSizeBytes( total += obj.Size } if total == 0 { - logger.Debug("no CNPG base backup objects found; leaving size unset", "backupID", backupID, "prefix", prefix) + logger.Info("no CNPG base backup objects found; leaving size unset", "backupID", backupID, "prefix", prefix) } return total } @@ -91,7 +91,7 @@ func (r *BackupJobReconciler) cnpgS3Connection(ctx context.Context, namespace st if conn.SecretKey, err = r.secretValue(ctx, namespace, creds.SecretRef.Name, secretKeyKey); err != nil { return conn, err } - if ca := store.EndpointCA; ca != nil && conn.Secure { + if ca := store.EndpointCA; ca != nil && ca.SecretRef.Name != "" && conn.Secure { key := ca.Key if key == "" { key = defaultEndpointCAKey diff --git a/internal/backupcontroller/cnpg_artifact_size_test.go b/internal/backupcontroller/cnpg_artifact_size_test.go index 29f09253b..81d850f64 100644 --- a/internal/backupcontroller/cnpg_artifact_size_test.go +++ b/internal/backupcontroller/cnpg_artifact_size_test.go @@ -3,6 +3,7 @@ package backupcontroller import ( "context" + "encoding/pem" "errors" "fmt" "net/http" @@ -74,8 +75,8 @@ func TestCNPGArtifactSizeBytes(t *testing.T) { if got != 1231 { t.Errorf("size = %d, want 1231", got) } - if l.bucket != "bkt" || l.prefix != prefix { - t.Errorf("listed %q/%q, want bkt/%q", l.bucket, l.prefix, prefix) + if l.bucket != "bkt" || l.prefix != prefix || !l.recursive { + t.Errorf("listed %q/%q recursive=%v, want bkt/%q recursive", l.bucket, l.prefix, l.recursive, prefix) } if l.conn.Endpoint != "s3.example:8443" || !l.conn.Secure || l.conn.AccessKey != "AKID" || l.conn.SecretKey != "secret" { t.Errorf("connection = %+v", l.conn) @@ -161,21 +162,10 @@ func TestCNPGArtifactSizeBytes(t *testing.T) { // and response parsing are exercised together. func TestCNPGArtifactSizeBytes_MinioClient(t *testing.T) { const prefix = "tenant/pg/srv/base/20261021T100000/" - var gotPath, gotPrefix, gotAuth string - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { - if _, ok := req.URL.Query()["location"]; ok { - w.Header().Set("Content-Type", "application/xml") - fmt.Fprint(w, ``) - return - } + var gotPath, gotPrefix, gotDelimiter, gotAuth string + srv := httptest.NewServer(cnpgListStub(prefix, func(req *http.Request) { gotPath, gotPrefix, gotAuth = req.URL.Path, req.URL.Query().Get("prefix"), req.Header.Get("Authorization") - w.Header().Set("Content-Type", "application/xml") - fmt.Fprintf(w, ` - - bkt%[1]s21000false - %[1]sbackup.info7"a"2026-10-21T10:01:00.000Z - %[1]sdata.tar.gz4096"b"2026-10-21T10:01:00.000Z -`, prefix) + gotDelimiter = req.URL.Query().Get("delimiter") })) defer srv.Close() @@ -187,14 +177,59 @@ func TestCNPGArtifactSizeBytes_MinioClient(t *testing.T) { if got := r.cnpgArtifactSizeBytes(context.Background(), "tenant", store, "srv", "20261021T100000"); got != 4103 { t.Fatalf("size = %d, want 4103", got) } - if gotPath != "/bkt/" || gotPrefix != prefix { - t.Errorf("request path=%q prefix=%q, want /bkt/ and %q", gotPath, gotPrefix, prefix) + if gotPath != "/bkt/" || gotPrefix != prefix || gotDelimiter != "" { + t.Errorf("request path=%q prefix=%q delimiter=%q, want /bkt/, %q and a recursive listing", gotPath, gotPrefix, gotDelimiter, prefix) } if !strings.Contains(gotAuth, "Credential=AKID/") { t.Errorf("request not signed with the Secret's access key: %q", gotAuth) } } +func cnpgListStub(prefix string, seen func(*http.Request)) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + w.Header().Set("Content-Type", "application/xml") + if _, ok := req.URL.Query()["location"]; ok { + fmt.Fprint(w, ``) + return + } + if seen != nil { + seen(req) + } + fmt.Fprintf(w, ` + + bkt%[1]s21000false + %[1]sbackup.info7"a"2026-10-21T10:01:00.000Z + %[1]sdata.tar.gz4096"b"2026-10-21T10:01:00.000Z +`, prefix) + } +} + +// TestCNPGArtifactSizeBytes_EndpointCA serves the listing over TLS with a +// self-signed certificate: the probe must trust it only through endpointCA. +func TestCNPGArtifactSizeBytes_EndpointCA(t *testing.T) { + const prefix = "tenant/pg/srv/base/20261021T100000/" + srv := httptest.NewTLSServer(cnpgListStub(prefix, nil)) + defer srv.Close() + caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srv.Certificate().Raw}) + + ca := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "s3-ca"}, + Data: map[string][]byte{"ca.crt": caPEM}, + } + c := newMariaDBStrategyTestClient(t, s3CredsSecret("tenant", "creds"), ca) + r := &BackupJobReconciler{Client: c, Scheme: c.Scheme()} + store := newSizedBarmanObjectStore() + store.EndpointURL = srv.URL + + if got := r.cnpgArtifactSizeBytes(context.Background(), "tenant", store, "srv", "20261021T100000"); got != 0 { + t.Errorf("without endpointCA: size = %d, want 0 (untrusted certificate)", got) + } + store.EndpointCA = &strategyv1alpha1.EndpointCARef{SecretRef: corev1.LocalObjectReference{Name: "s3-ca"}} + if got := r.cnpgArtifactSizeBytes(context.Background(), "tenant", store, "srv", "20261021T100000"); got != 4103 { + t.Errorf("with endpointCA: size = %d, want 4103", got) + } +} + func TestCreateCNPGBackupArtifact_SizeBytes(t *testing.T) { apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup strategyGroup := strategyv1alpha1.GroupVersion.Group diff --git a/internal/backupcontroller/mariadb_artifact_size.go b/internal/backupcontroller/mariadb_artifact_size.go index 20f68a454..b13e7ff1f 100644 --- a/internal/backupcontroller/mariadb_artifact_size.go +++ b/internal/backupcontroller/mariadb_artifact_size.go @@ -36,18 +36,21 @@ func (r *BackupJobReconciler) mariadbArtifactSizeBytes( conn, err := r.mariadbS3Connection(ctx, namespace, s3) if err != nil { - logger.Debug("cannot resolve S3 credentials to measure MariaDB backup size", "backup", mdbBackup.Name, "error", err) + logger.Info("cannot resolve S3 credentials to measure MariaDB backup size", "backup", mdbBackup.Name, "error", err) return 0 } lister, err := r.s3Lister(conn) if err != nil { - logger.Debug("cannot build S3 client to measure MariaDB backup size", "backup", mdbBackup.Name, "error", err) + logger.Info("cannot build S3 client to measure MariaDB backup size", "backup", mdbBackup.Name, "error", err) return 0 } + // Non-recursive, as mariadb-operator lists it: the default prefix is also + // the default CNPG destinationPath, so an app of each kind with the same + // name shares it and a recursive listing would walk the WAL archive. prefix := mariadbOperatorS3Prefix(s3.Prefix) - objects, err := lister.ListObjects(ctx, s3.Bucket, prefix) + objects, err := lister.ListObjects(ctx, s3.Bucket, prefix, false) if err != nil { - logger.Debug("cannot list MariaDB backup objects to measure size", "backup", mdbBackup.Name, "error", err) + logger.Info("cannot list MariaDB backup objects to measure size", "backup", mdbBackup.Name, "error", err) return 0 } diff --git a/internal/backupcontroller/mariadb_artifact_size_test.go b/internal/backupcontroller/mariadb_artifact_size_test.go index 214c7274c..0260cb14e 100644 --- a/internal/backupcontroller/mariadb_artifact_size_test.go +++ b/internal/backupcontroller/mariadb_artifact_size_test.go @@ -14,6 +14,7 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + strategyv1alpha1 "github.com/cozystack/cozystack/api/backups/strategy/v1alpha1" backupsv1alpha1 "github.com/cozystack/cozystack/api/backups/v1alpha1" "github.com/cozystack/cozystack/internal/backupcontroller/mariadbapp" "github.com/cozystack/cozystack/internal/backupcontroller/mariadbtypes" @@ -79,8 +80,29 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) { if l.conn.AccessKey != "AKID" || l.conn.Endpoint != "s3.example:9000" { t.Errorf("connection = %+v, want the Secret's key and the rendered endpoint", l.conn) } - if l.bucket != "bkt" || l.prefix != prefix { - t.Errorf("listed %q/%q, want bkt/%q", l.bucket, l.prefix, prefix) + if l.bucket != "bkt" || l.prefix != prefix || l.recursive { + t.Errorf("listed %q/%q recursive=%v, want bkt/%q non-recursive", l.bucket, l.prefix, l.recursive, prefix) + } + }) + + t.Run("reads the endpoint CA from its Secret", func(t *testing.T) { + tlsS3 := newRenderedMariaDBTemplate().Storage.S3 + tlsS3.Prefix = "tenant/db" + tlsS3.TLS = &strategyv1alpha1.MariaDBS3TLS{ + Enabled: true, + CASecretKeyRef: &strategyv1alpha1.MariaDBSecretKeySelector{Name: "ca", Key: "ca.crt"}, + } + ca := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "ca"}, + Data: map[string][]byte{"ca.crt": []byte("PEM")}, + } + l := &fakeS3Lister{objects: []s3ObjectInfo{{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 3}}} + r := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds"), ca) + if got := r.mariadbArtifactSizeBytes(context.Background(), "tenant", tlsS3, mdbBackup, completed); got != 3 { + t.Errorf("size = %d, want 3", got) + } + if !l.conn.Secure || string(l.conn.CAPEM) != "PEM" { + t.Errorf("connection = %+v, want TLS with the Secret's CA", l.conn) } }) @@ -135,14 +157,16 @@ func TestMariaDBArtifactSizeBytes_MinioClient(t *testing.T) { created := time.Date(2026, 10, 21, 10, 0, 0, 0, time.UTC) key := dumpKey("tenant/db/", created.Add(time.Minute), "gzip") - var gotPath, gotPrefix, gotAuth string + var gotPath, gotPrefix, gotDelimiter, gotAuth string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { gotPath, gotPrefix, gotAuth = req.URL.Path, req.URL.Query().Get("prefix"), req.Header.Get("Authorization") + gotDelimiter = req.URL.Query().Get("delimiter") w.Header().Set("Content-Type", "application/xml") fmt.Fprintf(w, ` bkttenant/db/11000false %s31337"x"2026-10-21T10:01:00.000Z + tenant/db/srv/ `, key) })) defer srv.Close() @@ -161,8 +185,8 @@ func TestMariaDBArtifactSizeBytes_MinioClient(t *testing.T) { if got := r.mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, created.Add(3*time.Minute)); got != 31337 { t.Fatalf("size = %d, want 31337", got) } - if gotPath != "/bkt/" || gotPrefix != "tenant/db/" { - t.Errorf("request path=%q prefix=%q, want /bkt/ and tenant/db/", gotPath, gotPrefix) + if gotPath != "/bkt/" || gotPrefix != "tenant/db/" || gotDelimiter != "/" { + t.Errorf("request path=%q prefix=%q delimiter=%q, want /bkt/, tenant/db/ and a non-recursive listing", gotPath, gotPrefix, gotDelimiter) } if !strings.Contains(gotAuth, "Credential=AKID/") { t.Errorf("request not signed with the Secret's access key: %q", gotAuth)