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)