mirror of
https://github.com/cozystack/cozystack.git
synced 2026-11-09 22:13:18 +09:00
fix(backups): attribute MariaDB dumps on two clocks and read retries uncached
The MariaDB probe identified this run's dump by the timestamp in its file name, which comes from the backup pod's clock. With that clock skewed past the slack, this run's dump falls outside the window while a neighbouring run's dump can be the single match left, and its size is reported as this backup's. A dump now has to fall inside the window by both its name and its LastModified, which the object store stamps, so a single skewed clock leaves the size unset instead. The name parser also accepts the layout later mariadb-operator releases write for compressed dumps, backup.<ts>.sql.<ext>, and checks the compression token in either layout, so an operator upgrade does not silently stop the measurement. The retry check and the AlreadyExists refetch read the Backup through the cache. A retry usually follows a Create made moments earlier; a cache that does not show it yet turned the refetch into NotFound and failed a backup that had succeeded. Read it from the API server instead. Assisted-by: LLM Signed-off-by: Andrey Kolkov <androndo@gmail.com>
This commit is contained in:
1 parent
3b86f57803
commit
a3d854739d
7 files changed
+94
-26
No files matched your search
@@ -26,8 +26,9 @@ import (
|
||||
const artifactSizeProbeTimeout = 10 * time.Second
|
||||
|
||||
type s3ObjectInfo struct {
|
||||
Key string
|
||||
Size int64
|
||||
Key string
|
||||
Size int64
|
||||
LastModified time.Time
|
||||
}
|
||||
|
||||
type s3ObjectLister interface {
|
||||
@@ -110,7 +111,7 @@ func collectObjects(ctx context.Context, objects <-chan minio.ObjectInfo) ([]s3O
|
||||
if obj.Err != nil {
|
||||
return nil, obj.Err
|
||||
}
|
||||
out = append(out, s3ObjectInfo{Key: obj.Key, Size: obj.Size})
|
||||
out = append(out, s3ObjectInfo{Key: obj.Key, Size: obj.Size, LastModified: obj.LastModified})
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
@@ -120,10 +121,12 @@ func collectObjects(ctx context.Context, objects <-chan minio.ObjectInfo) ([]s3O
|
||||
|
||||
// existingBackup returns the Backup a previous reconcile already created
|
||||
// for the job, so a retry skips the size probe whose result it would
|
||||
// discard. Returns (nil, nil) when there is none.
|
||||
// discard. Returns (nil, nil) when there is none. It reads past the cache:
|
||||
// a retry typically follows a Create made moments earlier, which the cache
|
||||
// may not show yet.
|
||||
func (r *BackupJobReconciler) existingBackup(ctx context.Context, namespace, name string) (*backupsv1alpha1.Backup, error) {
|
||||
existing := &backupsv1alpha1.Backup{}
|
||||
if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, existing); err != nil {
|
||||
if err := r.apiReader().Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, existing); err != nil {
|
||||
if apierrors.IsNotFound(err) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
@@ -11,6 +11,8 @@ import (
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
|
||||
backupsv1alpha1 "github.com/cozystack/cozystack/api/backups/v1alpha1"
|
||||
)
|
||||
|
||||
type fakeS3Lister struct {
|
||||
@@ -89,3 +91,18 @@ func TestNewMinioObjectLister_RejectsEndpointPath(t *testing.T) {
|
||||
t.Errorf("host:port endpoint rejected: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExistingBackup_ReadsPastTheCache pins that the retry check sees a
|
||||
// Backup the cache does not show yet: the previous reconcile created it
|
||||
// moments earlier.
|
||||
func TestExistingBackup_ReadsPastTheCache(t *testing.T) {
|
||||
backup := &backupsv1alpha1.Backup{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"}}
|
||||
r := &BackupJobReconciler{
|
||||
Client: newMariaDBStrategyTestClient(t),
|
||||
APIReader: newMariaDBStrategyTestClient(t, backup),
|
||||
}
|
||||
got, err := r.existingBackup(context.Background(), "tenant", "bj")
|
||||
if err != nil || got == nil {
|
||||
t.Fatalf("existingBackup = %v, %v; want the Backup the API server holds", got, err)
|
||||
}
|
||||
}
|
||||
@@ -83,6 +83,19 @@ func TestCNPGArtifactSizeBytes(t *testing.T) {
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("an empty endpointURL means AWS", func(t *testing.T) {
|
||||
store := newSizedBarmanObjectStore()
|
||||
store.EndpointURL = ""
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{{Key: prefix + "data.tar", Size: 5}}}
|
||||
r := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds"))
|
||||
if got := r.cnpgArtifactSizeBytes(context.Background(), "tenant", store, "srv", "20261021T100000"); got != 5 {
|
||||
t.Errorf("size = %d, want 5", got)
|
||||
}
|
||||
if l.conn.Endpoint != "s3.amazonaws.com" || !l.conn.Secure {
|
||||
t.Errorf("connection = %+v, want s3.amazonaws.com over TLS", l.conn)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("honours custom credential keys and the endpoint CA", func(t *testing.T) {
|
||||
store := newSizedBarmanObjectStore()
|
||||
store.S3Credentials.AccessKeyIDKey = "id"
|
||||
|
||||
@@ -876,7 +876,7 @@ func (r *BackupJobReconciler) createCNPGBackupArtifact(
|
||||
return nil, err
|
||||
}
|
||||
existing := &backupsv1alpha1.Backup{}
|
||||
if getErr := r.Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name}, existing); getErr != nil {
|
||||
if getErr := r.apiReader().Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name}, existing); getErr != nil {
|
||||
return nil, getErr
|
||||
}
|
||||
return existing, nil
|
||||
|
||||
@@ -11,15 +11,17 @@ import (
|
||||
)
|
||||
|
||||
// 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.
|
||||
// matching a dump: the timestamp in its file name comes from the backup Job
|
||||
// pod's clock and is truncated to seconds, and its LastModified comes from
|
||||
// the object store's clock, neither of which is the apiserver's.
|
||||
const mariadbSizeProbeClockSkew = time.Minute
|
||||
|
||||
// mariadbArtifactSizeBytes measures the dump mariadb-operator uploaded for
|
||||
// mdbBackup. The strategy prefix is shared by every backup of the
|
||||
// application, so the dump is identified by the timestamp in its file name
|
||||
// falling inside the operator Backup's lifetime. Returns 0 - leaving
|
||||
// application, so the dump is identified by both the timestamp in its file
|
||||
// name and its LastModified falling inside the operator Backup's lifetime:
|
||||
// the two come from different clocks, so a skewed pod clock alone cannot
|
||||
// pull a neighbouring run's dump into the window. Returns 0 - leaving
|
||||
// Artifact.SizeBytes absent - when the object store cannot be read or the
|
||||
// window does not match exactly one dump: an unknown size must not be
|
||||
// reported as a measured one.
|
||||
@@ -59,7 +61,7 @@ func (r *BackupJobReconciler) mariadbArtifactSizeBytes(
|
||||
var matched []s3ObjectInfo
|
||||
for _, obj := range objects {
|
||||
ts, ok := parseMariaDBLogicalBackupFile(strings.TrimPrefix(obj.Key, prefix))
|
||||
if !ok || ts.Before(lower) || ts.After(upper) {
|
||||
if !ok || !inWindow(ts, lower, upper) || !inWindow(obj.LastModified, lower, upper) {
|
||||
continue
|
||||
}
|
||||
matched = append(matched, obj)
|
||||
@@ -115,15 +117,30 @@ func mariadbOperatorS3Prefix(prefix string) string {
|
||||
return prefix
|
||||
}
|
||||
|
||||
func inWindow(t, lower, upper time.Time) bool {
|
||||
return !t.Before(lower) && !t.After(upper)
|
||||
}
|
||||
|
||||
// Compression tokens mariadb-operator puts in a dump file name: the
|
||||
// algorithm before .sql in v25.10 (backup.<ts>.gzip.sql), the extension
|
||||
// after it in later releases (backup.<ts>.sql.gz).
|
||||
var (
|
||||
mariadbDumpAlgorithms = map[string]bool{"none": true, "bzip2": true, "gzip": true, "zstd": true}
|
||||
mariadbDumpExtensions = map[string]bool{"bz2": true, "gz": true, "zst": true}
|
||||
)
|
||||
|
||||
// parseMariaDBLogicalBackupFile accepts the logical-backup file names
|
||||
// mariadb-operator writes, backup.<RFC3339>.sql and
|
||||
// backup.<RFC3339>.<compression>.sql, and returns the embedded timestamp.
|
||||
// mariadb-operator writes and returns the embedded timestamp.
|
||||
func parseMariaDBLogicalBackupFile(name string) (time.Time, bool) {
|
||||
parts := strings.Split(name, ".")
|
||||
if len(parts) != 3 && len(parts) != 4 {
|
||||
if parts[0] != "backup" {
|
||||
return time.Time{}, false
|
||||
}
|
||||
if parts[0] != "backup" || parts[len(parts)-1] != "sql" {
|
||||
switch {
|
||||
case len(parts) == 3 && parts[2] == "sql":
|
||||
case len(parts) == 4 && parts[2] == "sql" && mariadbDumpExtensions[parts[3]]:
|
||||
case len(parts) == 4 && parts[3] == "sql" && mariadbDumpAlgorithms[parts[2]]:
|
||||
default:
|
||||
return time.Time{}, false
|
||||
}
|
||||
ts, err := time.Parse(time.RFC3339, parts[1])
|
||||
|
||||
@@ -33,6 +33,11 @@ func TestParseMariaDBLogicalBackupFile(t *testing.T) {
|
||||
"backup.2026-10-21T10:00:00Z.gzip.sql": true,
|
||||
"backup.2026-10-21T10:00:00+04:00.sql": true,
|
||||
"backup.2026-10-21T10:00:00Z.sql.tmp": false,
|
||||
"backup.2026-10-21T10:00:00Z.sql.gz": true,
|
||||
"backup.2026-10-21T10:00:00Z.sql.zst": true,
|
||||
"backup.2026-10-21T10:00:00Z.zstd.sql": true,
|
||||
"backup.2026-10-21T10:00:00Z.xz.sql": false,
|
||||
"backup.2026-10-21T10:00:00Z.sql.gz.tmp": false,
|
||||
"physicalbackup-20261021100000.xb": false,
|
||||
"backup.not-a-date.sql": false,
|
||||
"sub/backup.2026-10-21T10:00:00Z.sql": false,
|
||||
@@ -69,8 +74,8 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
|
||||
|
||||
t.Run("single dump inside the backup window", func(t *testing.T) {
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{
|
||||
{Key: dumpKey(prefix, created.Add(-24*time.Hour), "gzip"), Size: 111},
|
||||
{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 4242},
|
||||
{Key: dumpKey(prefix, created.Add(-24*time.Hour), "gzip"), Size: 111, LastModified: created.Add(-24 * time.Hour)},
|
||||
{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 4242, LastModified: created.Add(time.Minute)},
|
||||
{Key: prefix + "0-backup-target.txt", Size: 7},
|
||||
}}
|
||||
got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed)
|
||||
@@ -96,7 +101,7 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
|
||||
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "ca"},
|
||||
Data: map[string][]byte{"ca.crt": []byte("PEM")},
|
||||
}
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 3}}}
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 3, LastModified: created.Add(time.Minute)}}}
|
||||
r := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds"), ca)
|
||||
if got := r.mariadbArtifactSizeBytes(context.Background(), "tenant", tlsS3, mdbBackup, completed); got != 3 {
|
||||
t.Errorf("size = %d, want 3", got)
|
||||
@@ -114,7 +119,7 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
|
||||
ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "sts"},
|
||||
Data: map[string][]byte{"token": []byte("TOKEN")},
|
||||
}
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 4}}}
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 4, LastModified: created.Add(time.Minute)}}}
|
||||
r := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds"), sts)
|
||||
if got := r.mariadbArtifactSizeBytes(context.Background(), "tenant", stsS3, mdbBackup, completed); got != 4 {
|
||||
t.Errorf("size = %d, want 4", got)
|
||||
@@ -126,17 +131,30 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
|
||||
|
||||
t.Run("pod clock skew within tolerance", func(t *testing.T) {
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{
|
||||
{Key: dumpKey(prefix, created.Add(-30*time.Second), ""), Size: 9},
|
||||
{Key: dumpKey(prefix, created.Add(-30*time.Second), ""), Size: 9, LastModified: created.Add(time.Minute)},
|
||||
}}
|
||||
if got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 9 {
|
||||
t.Errorf("size = %d, want 9", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a skewed pod clock does not attribute a neighbouring run's dump", func(t *testing.T) {
|
||||
// The pod clock runs 5m behind. This run's dump is named outside the
|
||||
// window; the previous run's dump, uploaded before this run started,
|
||||
// is named inside it.
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{
|
||||
{Key: dumpKey(prefix, created.Add(-4*time.Minute), "gzip"), Size: 7, LastModified: created.Add(time.Minute)},
|
||||
{Key: dumpKey(prefix, created.Add(30*time.Second), "gzip"), Size: 666, LastModified: created.Add(-4 * time.Minute)},
|
||||
}}
|
||||
if got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 {
|
||||
t.Errorf("size = %d, want 0: no dump is inside the window on both clocks", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("two dumps in the window leave size unset", func(t *testing.T) {
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{
|
||||
{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 1},
|
||||
{Key: dumpKey(prefix, created.Add(2*time.Minute), "gzip"), Size: 2},
|
||||
{Key: dumpKey(prefix, created.Add(time.Minute), "gzip"), Size: 1, LastModified: created.Add(time.Minute)},
|
||||
{Key: dumpKey(prefix, created.Add(2*time.Minute), "gzip"), Size: 2, LastModified: created.Add(2 * time.Minute)},
|
||||
}}
|
||||
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)
|
||||
@@ -145,7 +163,7 @@ func TestMariaDBArtifactSizeBytes(t *testing.T) {
|
||||
|
||||
t.Run("no dump in the window leaves size unset", func(t *testing.T) {
|
||||
l := &fakeS3Lister{objects: []s3ObjectInfo{
|
||||
{Key: dumpKey(prefix, completed.Add(time.Hour), "gzip"), Size: 1},
|
||||
{Key: dumpKey(prefix, completed.Add(time.Hour), "gzip"), Size: 1, LastModified: completed.Add(time.Hour)},
|
||||
}}
|
||||
if got := reconcilerWithLister(t, l, s3CredsSecret("tenant", "creds")).mariadbArtifactSizeBytes(context.Background(), "tenant", s3, mdbBackup, completed); got != 0 {
|
||||
t.Errorf("size = %d, want 0", got)
|
||||
@@ -234,7 +252,7 @@ func TestCreateMariaDBBackupArtifact_SizeBytes(t *testing.T) {
|
||||
},
|
||||
}
|
||||
r := reconcilerWithLister(t, &fakeS3Lister{objects: []s3ObjectInfo{
|
||||
{Key: dumpKey("src/", completed.Add(-time.Minute), "gzip"), Size: 2048},
|
||||
{Key: dumpKey("src/", completed.Add(-time.Minute), "gzip"), Size: 2048, LastModified: completed.Add(-time.Minute)},
|
||||
}}, job, s3CredsSecret("tenant", "creds"))
|
||||
rendered := newRenderedMariaDBTemplate()
|
||||
rendered.Storage.S3.Prefix = "src/"
|
||||
|
||||
@@ -421,7 +421,7 @@ func (r *BackupJobReconciler) createMariaDBBackupArtifact(
|
||||
return nil, err
|
||||
}
|
||||
existing := &backupsv1alpha1.Backup{}
|
||||
if getErr := r.Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name}, existing); getErr != nil {
|
||||
if getErr := r.apiReader().Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name}, existing); getErr != nil {
|
||||
return nil, getErr
|
||||
}
|
||||
return existing, nil
|
||||
|
||||
Reference in new issue
Block a user