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
<namespace>/<name>/ 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 <androndo@gmail.com>
This commit is contained in:
Andrey Kolkov authored and Myasnikov Daniil committed 2026-11-04 13:04:22 +00:00
1 parent de60488a2c
commit f222e52863
6 files changed
+119 -48

No files matched your search

+20 -12
View File
@@ -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
}
@@ -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
}
@@ -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
@@ -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, `<?xml version="1.0" encoding="UTF-8"?><LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/"></LocationConstraint>`)
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, `<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<Name>bkt</Name><Prefix>%[1]s</Prefix><KeyCount>2</KeyCount><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated>
<Contents><Key>%[1]sbackup.info</Key><Size>7</Size><ETag>"a"</ETag><LastModified>2026-10-21T10:01:00.000Z</LastModified></Contents>
<Contents><Key>%[1]sdata.tar.gz</Key><Size>4096</Size><ETag>"b"</ETag><LastModified>2026-10-21T10:01:00.000Z</LastModified></Contents>
</ListBucketResult>`, 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, `<?xml version="1.0" encoding="UTF-8"?><LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/"></LocationConstraint>`)
return
}
if seen != nil {
seen(req)
}
fmt.Fprintf(w, `<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<Name>bkt</Name><Prefix>%[1]s</Prefix><KeyCount>2</KeyCount><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated>
<Contents><Key>%[1]sbackup.info</Key><Size>7</Size><ETag>"a"</ETag><LastModified>2026-10-21T10:01:00.000Z</LastModified></Contents>
<Contents><Key>%[1]sdata.tar.gz</Key><Size>4096</Size><ETag>"b"</ETag><LastModified>2026-10-21T10:01:00.000Z</LastModified></Contents>
</ListBucketResult>`, 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
@@ -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
}
@@ -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, `<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<Name>bkt</Name><Prefix>tenant/db/</Prefix><KeyCount>1</KeyCount><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated>
<Contents><Key>%s</Key><Size>31337</Size><ETag>"x"</ETag><LastModified>2026-10-21T10:01:00.000Z</LastModified></Contents>
<CommonPrefixes><Prefix>tenant/db/srv/</Prefix></CommonPrefixes>
</ListBucketResult>`, 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)