fix(backups): report sizeBytes for CNPG backup artifacts

CNPG backups arrived without Backup.status.artifact.sizeBytes:
cnpg.io/Backup status carries no size, and the data lives in barman's
object-store layout rather than in a single object.

barman-cloud uploads every base backup under
<destinationPath>/<serverName>/base/<backupID>/ (backup.info plus the
data and tablespace tarballs, possibly split into parts). On completion
the driver now lists that prefix with the credentials and endpoint CA of
the rendered barmanObjectStore and records the total. The WAL archive is
not counted: it is one stream shared by every base backup of the server.
The field stays unset when the backup ID is unknown, the bucket cannot be
read, or nothing is stored under the prefix.

The S3 client and Secret lookup are now shared with the MariaDB driver.

Refs #4316

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 d4ad5612d3
commit de60488a2c
8 files changed
+564 -144

No files matched your search

+102
View File
@@ -0,0 +1,102 @@
// SPDX-License-Identifier: Apache-2.0
package backupcontroller
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"net/http"
"time"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
)
// 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
type s3ObjectInfo struct {
Key string
Size int64
}
type s3ObjectLister interface {
ListObjects(ctx context.Context, bucket, prefix string) ([]s3ObjectInfo, error)
}
// s3Connection is what a driver resolves from its rendered strategy to
// read the bucket with the same identity the backup was written with.
type s3Connection struct {
// Endpoint is host[:port], without a scheme.
Endpoint string
Secure bool
Region string
AccessKey string
SecretKey string
SessionToken string
// CAPEM is trusted on top of the system pool when set.
CAPEM []byte
}
func (r *BackupJobReconciler) s3Lister(conn s3Connection) (s3ObjectLister, error) {
if r.newS3ObjectLister != nil {
return r.newS3ObjectLister(conn)
}
return newMinioObjectLister(conn)
}
type minioObjectLister struct {
client *minio.Client
}
func newMinioObjectLister(conn s3Connection) (s3ObjectLister, error) {
opts := &minio.Options{
Creds: credentials.NewStaticV4(conn.AccessKey, conn.SecretKey, conn.SessionToken),
Region: conn.Region,
Secure: conn.Secure,
}
if conn.Secure && len(conn.CAPEM) > 0 {
pool, err := x509.SystemCertPool()
if 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}}
}
client, err := minio.New(conn.Endpoint, opts)
if err != nil {
return nil, err
}
return &minioObjectLister{client: client}, nil
}
func (l *minioObjectLister) ListObjects(ctx context.Context, bucket, prefix string) ([]s3ObjectInfo, error) {
var out []s3ObjectInfo
for obj := range l.client.ListObjects(ctx, bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: true}) {
if obj.Err != nil {
return nil, obj.Err
}
out = append(out, s3ObjectInfo{Key: obj.Key, Size: obj.Size})
}
return out, nil
}
func (r *BackupJobReconciler) secretValue(ctx context.Context, namespace, name, key string) (string, error) {
secret := &corev1.Secret{}
if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, secret); err != nil {
return "", err
}
v, ok := secret.Data[key]
if !ok {
return "", fmt.Errorf("Secret %s/%s has no key %q", namespace, name, key)
}
return string(v), nil
}
@@ -0,0 +1,49 @@
// SPDX-License-Identifier: Apache-2.0
package backupcontroller
import (
"context"
"testing"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"sigs.k8s.io/controller-runtime/pkg/client"
)
type fakeS3Lister struct {
objects []s3ObjectInfo
err error
conn s3Connection
bucket, prefix string
}
func (f *fakeS3Lister) ListObjects(_ context.Context, bucket, prefix string) ([]s3ObjectInfo, error) {
f.bucket, f.prefix = bucket, prefix
return f.objects, f.err
}
func s3CredsSecret(namespace, name string) *corev1.Secret {
return &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name},
Data: map[string][]byte{
"AWS_ACCESS_KEY_ID": []byte("AKID"),
"AWS_SECRET_ACCESS_KEY": []byte("secret"),
},
}
}
// reconcilerWithLister returns a reconciler whose S3 reads go to l, backed
// by a fake client seeded with objs (typically the credentials Secret).
func reconcilerWithLister(t *testing.T, l *fakeS3Lister, objs ...client.Object) *BackupJobReconciler {
t.Helper()
c := newMariaDBStrategyTestClient(t, objs...)
return &BackupJobReconciler{
Client: c,
Scheme: c.Scheme(),
newS3ObjectLister: func(conn s3Connection) (s3ObjectLister, error) {
l.conn = conn
return l, nil
},
}
}
@@ -42,7 +42,7 @@ type BackupJobReconciler struct {
// newS3ObjectLister overrides the S3 client used to measure artifact
// sizes; nil means a real minio client. Set by unit tests.
newS3ObjectLister func(ctx context.Context, namespace string, s3 *strategyv1alpha1.MariaDBS3Template) (s3ObjectLister, error)
newS3ObjectLister func(conn s3Connection) (s3ObjectLister, error)
}
func (r *BackupJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -0,0 +1,128 @@
// SPDX-License-Identifier: Apache-2.0
package backupcontroller
import (
"context"
"errors"
"fmt"
"net/url"
"strings"
strategyv1alpha1 "github.com/cozystack/cozystack/api/backups/strategy/v1alpha1"
)
// cnpgArtifactSizeBytes sums the objects barman-cloud uploaded for one base
// backup, <destinationPath>/<serverName>/base/<backupID>/. The WAL archive
// is not counted: it is one continuous stream shared by every base backup of
// the server, so no part of it belongs to a single Backup. Returns 0 -
// leaving Artifact.SizeBytes absent - when the backup ID is unknown, the
// bucket cannot be read, or nothing is stored under the prefix.
func (r *BackupJobReconciler) cnpgArtifactSizeBytes(
ctx context.Context,
namespace string,
store *strategyv1alpha1.BarmanObjectStoreTemplate,
serverName, backupID string,
) int64 {
logger := getLogger(ctx)
if backupID == "" {
return 0
}
ctx, cancel := context.WithTimeout(ctx, artifactSizeProbeTimeout)
defer cancel()
bucket, path, err := parseBarmanS3DestinationPath(store.DestinationPath)
if err != nil {
logger.Debug("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)
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)
return 0
}
prefix := barmanBaseBackupPrefix(path, serverName, backupID)
objects, err := lister.ListObjects(ctx, bucket, prefix)
if err != nil {
logger.Debug("cannot list CNPG base backup objects to measure size", "backupID", backupID, "error", err)
return 0
}
var total int64
for _, obj := range objects {
total += obj.Size
}
if total == 0 {
logger.Debug("no CNPG base backup objects found; leaving size unset", "backupID", backupID, "prefix", prefix)
}
return total
}
func (r *BackupJobReconciler) cnpgS3Connection(ctx context.Context, namespace string, store *strategyv1alpha1.BarmanObjectStoreTemplate) (s3Connection, error) {
var conn s3Connection
if store.S3Credentials == nil {
return conn, errors.New("barmanObjectStore has no s3Credentials")
}
if store.EndpointURL == "" {
conn.Endpoint, conn.Secure = "s3.amazonaws.com", true
} else {
u, err := url.Parse(store.EndpointURL)
if err != nil || u.Host == "" {
return conn, fmt.Errorf("invalid endpointURL %q", store.EndpointURL)
}
conn.Endpoint, conn.Secure = u.Host, u.Scheme == "https"
}
creds := store.S3Credentials
accessKeyKey, secretKeyKey := creds.AccessKeyIDKey, creds.SecretAccessKeyKey
if accessKeyKey == "" {
accessKeyKey = defaultS3AccessKeyIDKey
}
if secretKeyKey == "" {
secretKeyKey = defaultS3SecretAccessKeyKey
}
var err error
if conn.AccessKey, err = r.secretValue(ctx, namespace, creds.SecretRef.Name, accessKeyKey); err != nil {
return conn, err
}
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 {
key := ca.Key
if key == "" {
key = defaultEndpointCAKey
}
pem, err := r.secretValue(ctx, namespace, ca.SecretRef.Name, key)
if err != nil {
return conn, err
}
conn.CAPEM = []byte(pem)
}
return conn, nil
}
// parseBarmanS3DestinationPath splits s3://bucket/path into the bucket and
// the key path barman-cloud prefixes every object with.
func parseBarmanS3DestinationPath(destinationPath string) (bucket, path string, err error) {
u, err := url.Parse(destinationPath)
if err != nil || u.Scheme != "s3" || u.Host == "" {
return "", "", fmt.Errorf("destinationPath %q is not an s3:// URL", destinationPath)
}
return u.Host, strings.Trim(u.Path, "/"), nil
}
// barmanBaseBackupPrefix mirrors barman-cloud's
// os.path.join(path, server_name, "base", backup_id), with a trailing slash
// so one backup ID never matches another that extends it.
func barmanBaseBackupPrefix(path, serverName, backupID string) string {
parts := make([]string, 0, 4)
if path != "" {
parts = append(parts, path)
}
parts = append(parts, serverName, "base", backupID)
return strings.Join(parts, "/") + "/"
}
@@ -0,0 +1,229 @@
// SPDX-License-Identifier: Apache-2.0
package backupcontroller
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"testing"
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/cnpgtypes"
)
func newSizedBarmanObjectStore() *strategyv1alpha1.BarmanObjectStoreTemplate {
return &strategyv1alpha1.BarmanObjectStoreTemplate{
DestinationPath: "s3://bkt/tenant/pg/",
EndpointURL: "https://s3.example:8443",
S3Credentials: &strategyv1alpha1.S3CredentialsTemplate{
SecretRef: corev1.LocalObjectReference{Name: "creds"},
},
}
}
func TestBarmanBaseBackupPrefix(t *testing.T) {
cases := []struct{ path, want string }{
{"tenant/pg", "tenant/pg/srv/base/20261021T100000/"},
{"", "srv/base/20261021T100000/"},
}
for _, tc := range cases {
if got := barmanBaseBackupPrefix(tc.path, "srv", "20261021T100000"); got != tc.want {
t.Errorf("barmanBaseBackupPrefix(%q) = %q, want %q", tc.path, got, tc.want)
}
}
}
func TestParseBarmanS3DestinationPath(t *testing.T) {
cases := []struct {
in, bucket, path string
ok bool
}{
{"s3://bkt/tenant/pg/", "bkt", "tenant/pg", true},
{"s3://bkt", "bkt", "", true},
{"s3://bkt/", "bkt", "", true},
{"gs://bkt/x/", "", "", false},
{"bkt/x/", "", "", false},
}
for _, tc := range cases {
bucket, path, err := parseBarmanS3DestinationPath(tc.in)
if (err == nil) != tc.ok || bucket != tc.bucket || path != tc.path {
t.Errorf("parseBarmanS3DestinationPath(%q) = %q, %q, %v", tc.in, bucket, path, err)
}
}
}
func TestCNPGArtifactSizeBytes(t *testing.T) {
const prefix = "tenant/pg/srv/base/20261021T100000/"
t.Run("sums every object of the base backup", func(t *testing.T) {
l := &fakeS3Lister{objects: []s3ObjectInfo{
{Key: prefix + "backup.info", Size: 1},
{Key: prefix + "data_0000.tar.gz", Size: 1000},
{Key: prefix + "data_0001.tar.gz", Size: 200},
{Key: prefix + "16385.tar.gz", Size: 30},
}}
r := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds"))
got := r.cnpgArtifactSizeBytes(context.Background(), "tenant", newSizedBarmanObjectStore(), "srv", "20261021T100000")
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.conn.Endpoint != "s3.example:8443" || !l.conn.Secure || l.conn.AccessKey != "AKID" || l.conn.SecretKey != "secret" {
t.Errorf("connection = %+v", l.conn)
}
})
t.Run("honours custom credential keys and the endpoint CA", func(t *testing.T) {
store := newSizedBarmanObjectStore()
store.S3Credentials.AccessKeyIDKey = "id"
store.S3Credentials.SecretAccessKeyKey = "key"
store.EndpointCA = &strategyv1alpha1.EndpointCARef{SecretRef: corev1.LocalObjectReference{Name: "ca"}}
creds := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "creds"},
Data: map[string][]byte{"id": []byte("ID2"), "key": []byte("KEY2")},
}
ca := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "ca"},
Data: map[string][]byte{"ca.crt": []byte("PEM")},
}
l := &fakeS3Lister{objects: []s3ObjectInfo{{Key: prefix + "data.tar", Size: 5}}}
r := reconcilerWithLister(t, l, creds, ca)
if got := r.cnpgArtifactSizeBytes(context.Background(), "tenant", store, "srv", "20261021T100000"); got != 5 {
t.Errorf("size = %d, want 5", got)
}
if l.conn.AccessKey != "ID2" || l.conn.SecretKey != "KEY2" || string(l.conn.CAPEM) != "PEM" {
t.Errorf("connection = %+v", l.conn)
}
})
unset := map[string]struct {
store func() *strategyv1alpha1.BarmanObjectStoreTemplate
backupID string
lister *fakeS3Lister
noSecret bool
}{
"unknown backup ID": {
store: newSizedBarmanObjectStore, lister: &fakeS3Lister{objects: []s3ObjectInfo{{Key: "x", Size: 1}}},
},
"no objects under the prefix": {
store: newSizedBarmanObjectStore, backupID: "id", lister: &fakeS3Lister{},
},
"list error": {
store: newSizedBarmanObjectStore, backupID: "id", lister: &fakeS3Lister{err: errors.New("AccessDenied")},
},
"missing credentials Secret": {
store: newSizedBarmanObjectStore, backupID: "id", noSecret: true,
lister: &fakeS3Lister{objects: []s3ObjectInfo{{Key: "x", Size: 1}}},
},
"no s3Credentials": {
store: func() *strategyv1alpha1.BarmanObjectStoreTemplate {
s := newSizedBarmanObjectStore()
s.S3Credentials = nil
return s
},
backupID: "id", lister: &fakeS3Lister{objects: []s3ObjectInfo{{Key: "x", Size: 1}}},
},
"non-s3 destination": {
store: func() *strategyv1alpha1.BarmanObjectStoreTemplate {
s := newSizedBarmanObjectStore()
s.DestinationPath = "azure://container/x"
return s
},
backupID: "id", lister: &fakeS3Lister{objects: []s3ObjectInfo{{Key: "x", Size: 1}}},
},
}
for name, tc := range unset {
t.Run(name+" leaves size unset", func(t *testing.T) {
var r *BackupJobReconciler
if tc.noSecret {
r = reconcilerWithLister(t, tc.lister)
} else {
r = reconcilerWithLister(t, tc.lister, s3CredsSecret("tenant", "creds"))
}
if got := r.cnpgArtifactSizeBytes(context.Background(), "tenant", tc.store(), "srv", tc.backupID); got != 0 {
t.Errorf("size = %d, want 0", got)
}
})
}
}
// TestCNPGArtifactSizeBytes_MinioClient drives the real minio client against
// a stub ListObjectsV2 endpoint, so the endpoint URL parsing, Secret wiring
// 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
}
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)
}))
defer srv.Close()
c := newMariaDBStrategyTestClient(t, s3CredsSecret("tenant", "creds"))
r := &BackupJobReconciler{Client: c, Scheme: c.Scheme()}
store := newSizedBarmanObjectStore()
store.EndpointURL = srv.URL
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 !strings.Contains(gotAuth, "Credential=AKID/") {
t.Errorf("request not signed with the Secret's access key: %q", gotAuth)
}
}
func TestCreateCNPGBackupArtifact_SizeBytes(t *testing.T) {
apiGroup := backupsv1alpha1.DefaultApplicationAPIGroup
strategyGroup := strategyv1alpha1.GroupVersion.Group
j := &backupsv1alpha1.BackupJob{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj-size"},
Spec: backupsv1alpha1.BackupJobSpec{
ApplicationRef: corev1.TypedLocalObjectReference{APIGroup: &apiGroup, Kind: postgresAppKind, Name: "pg"},
},
}
resolved := &ResolvedBackupConfig{
StrategyRef: corev1.TypedLocalObjectReference{APIGroup: &strategyGroup, Kind: strategyv1alpha1.CNPGStrategyKind, Name: "strat"},
}
cnpgBk := &cnpgtypes.Backup{
ObjectMeta: metav1.ObjectMeta{Name: "cnpg-bk"},
Status: cnpgtypes.BackupStatus{BackupID: "20261021T100000"},
}
rendered := &strategyv1alpha1.CNPGTemplate{BarmanObjectStore: *newSizedBarmanObjectStore()}
l := &fakeS3Lister{objects: []s3ObjectInfo{{Key: "tenant/pg/srv/base/20261021T100000/data.tar.gz", Size: 777}}}
r := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds"))
r.Client = newCNPGStrategyTestClient(t, s3CredsSecret("tenant", "creds"))
got, err := r.createCNPGBackupArtifact(context.Background(), j, resolved, cnpgBk, "postgres-pg", "srv", "postgresql", rendered, newPostgresApp("pg", "tenant"))
if err != nil {
t.Fatalf("createCNPGBackupArtifact: %v", err)
}
if got.Status.Artifact == nil || got.Status.Artifact.SizeBytes != 777 {
t.Fatalf("Artifact = %#v, want SizeBytes 777", got.Status.Artifact)
}
if l.prefix != "tenant/pg/srv/base/20261021T100000/" {
t.Errorf("listed prefix %q does not use the effective serverName", l.prefix)
}
}
@@ -855,7 +855,8 @@ func (r *BackupJobReconciler) createCNPGBackupArtifact(
Status: backupsv1alpha1.BackupStatus{
Phase: backupsv1alpha1.BackupPhaseReady,
Artifact: &backupsv1alpha1.BackupArtifact{
URI: fmt.Sprintf("cnpg://%s/%s", serverName, cnpgBackup.Name),
URI: fmt.Sprintf("cnpg://%s/%s", serverName, cnpgBackup.Name),
SizeBytes: r.cnpgArtifactSizeBytes(ctx, j.Namespace, &rendered.BarmanObjectStore, serverName, cnpgBackup.Status.BackupID),
},
UnderlyingResources: underlyingResources,
},
@@ -3,40 +3,18 @@ package backupcontroller
import (
"context"
"crypto/tls"
"crypto/x509"
"fmt"
"net/http"
"strings"
"time"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
strategyv1alpha1 "github.com/cozystack/cozystack/api/backups/strategy/v1alpha1"
"github.com/cozystack/cozystack/internal/backupcontroller/mariadbtypes"
)
const (
mariadbSizeProbeTimeout = 30 * time.Second
// Slack around the [created, completed] window of the operator Backup
// when matching the timestamp mariadb-operator embeds in the dump file
// name: that timestamp comes from the backup Job pod's clock, not the
// apiserver's, and is truncated to seconds.
mariadbSizeProbeClockSkew = time.Minute
)
type s3ObjectInfo struct {
Key string
Size int64
}
type s3ObjectLister interface {
ListObjects(ctx context.Context, bucket, prefix string) ([]s3ObjectInfo, error)
}
// Slack around the [created, completed] window of the operator Backup when
// matching the timestamp mariadb-operator embeds in the dump file name: that
// timestamp comes from the backup Job pod's clock, not the apiserver's, and
// is truncated to seconds.
const mariadbSizeProbeClockSkew = time.Minute
// mariadbArtifactSizeBytes measures the dump mariadb-operator uploaded for
// mdbBackup. The strategy prefix is shared by every backup of the
@@ -53,14 +31,15 @@ func (r *BackupJobReconciler) mariadbArtifactSizeBytes(
completedAt time.Time,
) int64 {
logger := getLogger(ctx)
ctx, cancel := context.WithTimeout(ctx, mariadbSizeProbeTimeout)
ctx, cancel := context.WithTimeout(ctx, artifactSizeProbeTimeout)
defer cancel()
newLister := r.newS3ObjectLister
if newLister == nil {
newLister = r.newMariaDBMinioLister
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)
return 0
}
lister, err := newLister(ctx, namespace, s3)
lister, err := r.s3Lister(conn)
if err != nil {
logger.Debug("cannot build S3 client to measure MariaDB backup size", "backup", mdbBackup.Name, "error", err)
return 0
@@ -90,6 +69,37 @@ func (r *BackupJobReconciler) mariadbArtifactSizeBytes(
return matched[0].Size
}
// mariadbS3Connection resolves the Secrets and TLS settings the operator
// Backup was rendered with, so the bucket is read with the tenant's
// credentials rather than the controller's.
func (r *BackupJobReconciler) mariadbS3Connection(ctx context.Context, namespace string, s3 *strategyv1alpha1.MariaDBS3Template) (s3Connection, error) {
conn := s3Connection{
Endpoint: s3.Endpoint,
Region: s3.Region,
Secure: s3.TLS != nil && s3.TLS.Enabled,
}
var err error
if conn.AccessKey, err = r.secretValue(ctx, namespace, s3.AccessKeyIDSecretKeyRef.Name, s3.AccessKeyIDSecretKeyRef.Key); err != nil {
return conn, err
}
if conn.SecretKey, err = r.secretValue(ctx, namespace, s3.SecretAccessKeySecretKeyRef.Name, s3.SecretAccessKeySecretKeyRef.Key); err != nil {
return conn, err
}
if ref := s3.SessionTokenSecretKeyRef; ref != nil {
if conn.SessionToken, err = r.secretValue(ctx, namespace, ref.Name, ref.Key); err != nil {
return conn, err
}
}
if conn.Secure && s3.TLS.CASecretKeyRef != nil {
ca, err := r.secretValue(ctx, namespace, s3.TLS.CASecretKeyRef.Name, s3.TLS.CASecretKeyRef.Key)
if err != nil {
return conn, err
}
conn.CAPEM = []byte(ca)
}
return conn, nil
}
// mariadbOperatorS3Prefix mirrors how mariadb-operator normalises
// spec.storage.s3.prefix before prepending it to object keys.
func mariadbOperatorS3Prefix(prefix string) string {
@@ -119,75 +129,3 @@ func parseMariaDBLogicalBackupFile(name string) (time.Time, bool) {
}
return ts, true
}
type minioObjectLister struct {
client *minio.Client
}
func (l *minioObjectLister) ListObjects(ctx context.Context, bucket, prefix string) ([]s3ObjectInfo, error) {
var out []s3ObjectInfo
for obj := range l.client.ListObjects(ctx, bucket, minio.ListObjectsOptions{Prefix: prefix}) {
if obj.Err != nil {
return nil, obj.Err
}
out = append(out, s3ObjectInfo{Key: obj.Key, Size: obj.Size})
}
return out, nil
}
// newMariaDBMinioLister builds an S3 client from the same Secrets and TLS
// settings the operator Backup was rendered with, so the controller reads
// the bucket with the tenant's credentials rather than its own.
func (r *BackupJobReconciler) newMariaDBMinioLister(ctx context.Context, namespace string, s3 *strategyv1alpha1.MariaDBS3Template) (s3ObjectLister, error) {
accessKey, err := r.mariadbSecretValue(ctx, namespace, s3.AccessKeyIDSecretKeyRef)
if err != nil {
return nil, err
}
secretKey, err := r.mariadbSecretValue(ctx, namespace, s3.SecretAccessKeySecretKeyRef)
if err != nil {
return nil, err
}
var sessionToken string
if s3.SessionTokenSecretKeyRef != nil {
if sessionToken, err = r.mariadbSecretValue(ctx, namespace, *s3.SessionTokenSecretKeyRef); err != nil {
return nil, err
}
}
opts := &minio.Options{
Creds: credentials.NewStaticV4(accessKey, secretKey, sessionToken),
Region: s3.Region,
Secure: s3.TLS != nil && s3.TLS.Enabled,
}
if opts.Secure && s3.TLS.CASecretKeyRef != nil {
caPEM, err := r.mariadbSecretValue(ctx, namespace, *s3.TLS.CASecretKeyRef)
if err != nil {
return nil, err
}
pool, err := x509.SystemCertPool()
if err != nil {
pool = x509.NewCertPool()
}
if !pool.AppendCertsFromPEM([]byte(caPEM)) {
return nil, fmt.Errorf("no PEM certificates in Secret %s/%s key %q", namespace, s3.TLS.CASecretKeyRef.Name, s3.TLS.CASecretKeyRef.Key)
}
opts.Transport = &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}}
}
client, err := minio.New(s3.Endpoint, opts)
if err != nil {
return nil, err
}
return &minioObjectLister{client: client}, nil
}
func (r *BackupJobReconciler) mariadbSecretValue(ctx context.Context, namespace string, ref strategyv1alpha1.MariaDBSecretKeySelector) (string, error) {
secret := &corev1.Secret{}
if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: ref.Name}, secret); err != nil {
return "", err
}
v, ok := secret.Data[ref.Key]
if !ok {
return "", fmt.Errorf("Secret %s/%s has no key %q", namespace, ref.Name, ref.Key)
}
return string(v), nil
}
@@ -14,32 +14,11 @@ 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"
)
type fakeS3Lister struct {
objects []s3ObjectInfo
err error
bucket, prefix string
}
func (f *fakeS3Lister) ListObjects(_ context.Context, bucket, prefix string) ([]s3ObjectInfo, error) {
f.bucket, f.prefix = bucket, prefix
return f.objects, f.err
}
func reconcilerWithLister(l *fakeS3Lister) *BackupJobReconciler {
return &BackupJobReconciler{
newS3ObjectLister: func(context.Context, string, *strategyv1alpha1.MariaDBS3Template) (s3ObjectLister, error) {
return l, nil
},
}
}
func dumpKey(prefix string, ts time.Time, compression string) string {
if compression == "" {
return fmt.Sprintf("%sbackup.%s.sql", prefix, ts.UTC().Format(time.RFC3339))
@@ -93,10 +72,13 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 4242},
{Key: prefix + "0-backup-target.txt", Size: 7},
}}
got := reconcilerWithLister(l).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed)
got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed)
if got != 4242 {
t.Errorf("size = %d, want 4242", got)
}
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)
}
@@ -106,7 +88,7 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
l := &fakeS3Lister{objects: []s3ObjectInfo{
{Key: dumpKey(prefix, created.Add(-30*time.Second), ""), Size: 9},
}}
if got := reconcilerWithLister(l).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 9 {
if got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 9 {
t.Errorf("size = %d, want 9", got)
}
})
@@ -116,7 +98,7 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 1},
{Key: dumpKey(prefix, created.Add(2*time.Minute), "gzip"), Size: 2},
}}
if got := reconcilerWithLister(l).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 {
if got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 {
t.Errorf("size = %d, want 0 (ambiguous)", got)
}
})
@@ -125,14 +107,14 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
l := &fakeS3Lister{objects: []s3ObjectInfo{
{Key: dumpKey(prefix, completed.Add(time.Hour), "gzip"), Size: 1},
}}
if got := reconcilerWithLister(l).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 {
if got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 {
t.Errorf("size = %d, want 0", got)
}
})
t.Run("list error leaves size unset", func(t *testing.T) {
l := &fakeS3Lister{err: errors.New("AccessDenied")}
if got := reconcilerWithLister(l).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 {
if got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 {
t.Errorf("size = %d, want 0", got)
}
})
@@ -165,14 +147,7 @@ func TestMariaDBArtifactSizeBytes_MinioClient(t *testing.T) {
}))
defer srv.Close()
creds := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "creds"},
Data: map[string][]byte{
"AWS_ACCESS_KEY_ID": []byte("AKID"),
"AWS_SECRET_ACCESS_KEY": []byte("secret"),
},
}
c := newMariaDBStrategyTestClient(t, creds)
c := newMariaDBStrategyTestClient(t, s3CredsSecret("tenant", "creds"))
r := &BackupJobReconciler{Client: c, Scheme: c.Scheme()}
s3 := newRenderedMariaDBTemplate().Storage.S3
@@ -216,11 +191,9 @@ func TestCreateMariaDBBackupArtifact_SizeBytes(t *testing.T) {
ApplicationRef: corev1.TypedLocalObjectReference{Kind: "MariaDB", Name: "src", APIGroup: &apps},
},
}
c := newMariaDBStrategyTestClient(t, job)
r := reconcilerWithLister(&fakeS3Lister{objects: []s3ObjectInfo{
r := reconcilerWithLister(t, &fakeS3Lister{objects: []s3ObjectInfo{
{Key: dumpKey("src/", completed.Add(-time.Minute), "gzip"), Size: 2048},
}})
r.Client, r.Scheme = c, c.Scheme()
}}, job, s3CredsSecret("tenant", "creds"))
rendered := newRenderedMariaDBTemplate()
rendered.Storage.S3.Prefix = "src/"