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/"