From de60488a2cb9c25500af07ce13dbc467dfd72c98 Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Wed, 21 Oct 2026 11:03:59 +0400 Subject: [PATCH] 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 //base// (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 --- internal/backupcontroller/artifact_size.go | 102 ++++++++ .../backupcontroller/artifact_size_test.go | 49 ++++ .../backupcontroller/backupjob_controller.go | 2 +- .../backupcontroller/cnpg_artifact_size.go | 128 ++++++++++ .../cnpg_artifact_size_test.go | 229 ++++++++++++++++++ .../cnpgstrategy_controller.go | 3 +- .../backupcontroller/mariadb_artifact_size.go | 146 ++++------- .../mariadb_artifact_size_test.go | 49 +--- 8 files changed, 564 insertions(+), 144 deletions(-) create mode 100644 internal/backupcontroller/artifact_size.go create mode 100644 internal/backupcontroller/artifact_size_test.go create mode 100644 internal/backupcontroller/cnpg_artifact_size.go create mode 100644 internal/backupcontroller/cnpg_artifact_size_test.go diff --git a/internal/backupcontroller/artifact_size.go b/internal/backupcontroller/artifact_size.go new file mode 100644 index 000000000..6ca2ad3ea --- /dev/null +++ b/internal/backupcontroller/artifact_size.go @@ -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 +} diff --git a/internal/backupcontroller/artifact_size_test.go b/internal/backupcontroller/artifact_size_test.go new file mode 100644 index 000000000..6e389cf66 --- /dev/null +++ b/internal/backupcontroller/artifact_size_test.go @@ -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 + }, + } +} diff --git a/internal/backupcontroller/backupjob_controller.go b/internal/backupcontroller/backupjob_controller.go index f329eed4d..b86278649 100644 --- a/internal/backupcontroller/backupjob_controller.go +++ b/internal/backupcontroller/backupjob_controller.go @@ -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) { diff --git a/internal/backupcontroller/cnpg_artifact_size.go b/internal/backupcontroller/cnpg_artifact_size.go new file mode 100644 index 000000000..bff2d5b39 --- /dev/null +++ b/internal/backupcontroller/cnpg_artifact_size.go @@ -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, //base//. 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, "/") + "/" +} diff --git a/internal/backupcontroller/cnpg_artifact_size_test.go b/internal/backupcontroller/cnpg_artifact_size_test.go new file mode 100644 index 000000000..29f09253b --- /dev/null +++ b/internal/backupcontroller/cnpg_artifact_size_test.go @@ -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, ``) + 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, ` + + 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) + })) + 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) + } +} diff --git a/internal/backupcontroller/cnpgstrategy_controller.go b/internal/backupcontroller/cnpgstrategy_controller.go index 99e6b03f6..3b62ce6e1 100644 --- a/internal/backupcontroller/cnpgstrategy_controller.go +++ b/internal/backupcontroller/cnpgstrategy_controller.go @@ -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, }, diff --git a/internal/backupcontroller/mariadb_artifact_size.go b/internal/backupcontroller/mariadb_artifact_size.go index b968a98d4..20f68a454 100644 --- a/internal/backupcontroller/mariadb_artifact_size.go +++ b/internal/backupcontroller/mariadb_artifact_size.go @@ -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 -} diff --git a/internal/backupcontroller/mariadb_artifact_size_test.go b/internal/backupcontroller/mariadb_artifact_size_test.go index f7e450a26..214c7274c 100644 --- a/internal/backupcontroller/mariadb_artifact_size_test.go +++ b/internal/backupcontroller/mariadb_artifact_size_test.go @@ -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/"