Skip to content

Commit a1e7fc0

Browse files
Slachclaude
andcommitted
Spread rebalanced parts across same-filesystem JBOD disks
least_used picked the disk with the largest free space from one system.disks snapshot. JBOD disks on the same filesystem report almost equal free space, and the difference is statvfs noise from concurrent writes, so all parts of a download could land on a single disk. TestForceRebalance flaked on CI this way (25.3, no files on hdd2). Local disks whose free space is within 1% of total space of the least used one are now treated as equally used and get parts round-robin. Disks with really different free space keep least_used behavior. Also surface find output in TestForceRebalance assertions, the CI log showed only "exit status 1". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 78f849d commit a1e7fc0

3 files changed

Lines changed: 71 additions & 10 deletions

File tree

‎pkg/backup/download.go‎

Lines changed: 25 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -497,6 +497,7 @@ func (b *Backuper) saveLocalBackupMetadata(backupName string, backupMetadata met
497497

498498
func (b *Backuper) reBalanceTablesMetadataIfDiskNotExists(tableMetadataAfterDownload ListOfTables, disks []clickhouse.Disk, remoteBackup storage.Backup) error {
499499
var disksByStoragePolicyAndType map[string]map[string][]clickhouse.Disk
500+
assignedParts := map[string]int{}
500501
filterDisksByTypeAndStoragePolicies := func(disk string, diskType string, disks []clickhouse.Disk, remoteBackup storage.Backup, t metadata.TableMetadata) (string, []clickhouse.Disk, error) {
501502
_, ok := remoteBackup.DiskTypes[disk]
502503
if !ok {
@@ -568,6 +569,7 @@ func (b *Backuper) reBalanceTablesMetadataIfDiskNotExists(tableMetadataAfterDown
568569
if t.Parts[disk][j].RebalancedDisk != "" {
569570
existingDisk := t.Parts[disk][j].RebalancedDisk
570571
rebalancedDisks[existingDisk] = struct{}{}
572+
assignedParts[existingDisk]++
571573
isRebalanced = true
572574
// Rebuild RebalancedFiles map for already-rebalanced parts
573575
if t.Files != nil && len(t.Files[disk]) > 0 {
@@ -582,7 +584,7 @@ func (b *Backuper) reBalanceTablesMetadataIfDiskNotExists(tableMetadataAfterDown
582584
}
583585
continue
584586
}
585-
isObjectDisk, downloadDisk, newFreeSpace, reBalanceErr := b.getDownloadDiskForNonExistsDisk(diskType, filteredDisks, partSize)
587+
isObjectDisk, downloadDisk, newFreeSpace, reBalanceErr := b.getDownloadDiskForNonExistsDisk(diskType, filteredDisks, partSize, assignedParts)
586588
if reBalanceErr != nil {
587589
return errors.Wrap(reBalanceErr, "getDownloadDiskForNonExistsDisk")
588590
}
@@ -2417,23 +2419,38 @@ func (b *Backuper) splitDisksByTypeAndStoragePolicy(disks []clickhouse.Disk) map
24172419
// getDownloadDiskForNonExistsDisk - https://github.com/Altinity/clickhouse-backup/issues/561
24182420
// allow to Restore to new server with different storage policy, different disk count,
24192421
// implements `least_used` for normal disk and `random` for Object disks
2420-
func (b *Backuper) getDownloadDiskForNonExistsDisk(notExistsDiskType string, filteredDisks []clickhouse.Disk, partSize uint64) (bool, string, uint64, error) {
2422+
// local disks whose free space differs from the least used one by less than 1% of total space are treated as
2423+
// equally used (JBOD disks on the same filesystem, the difference is statvfs noise from concurrent writes),
2424+
// parts are distributed across them round-robin by assignedParts
2425+
func (b *Backuper) getDownloadDiskForNonExistsDisk(notExistsDiskType string, filteredDisks []clickhouse.Disk, partSize uint64, assignedParts map[string]int) (bool, string, uint64, error) {
24212426
// random for non-local disks
24222427
if notExistsDiskType != "local" {
24232428
roundRobinIdx := rand.Intn(len(filteredDisks))
24242429
return true, filteredDisks[roundRobinIdx].Name, 0, nil
24252430
}
24262431
// least_used for local
2427-
freeSpace := partSize
2432+
maxFreeSpace := uint64(0)
2433+
maxTotalSpace := uint64(0)
2434+
for _, d := range filteredDisks {
2435+
if d.FreeSpace > partSize && d.FreeSpace > maxFreeSpace {
2436+
maxFreeSpace = d.FreeSpace
2437+
}
2438+
maxTotalSpace = max(maxTotalSpace, d.TotalSpace)
2439+
}
2440+
if maxFreeSpace == 0 {
2441+
return false, "", 0, errors.Errorf("%s free space, not found in system.disks with `local` type", utils.FormatBytes(partSize))
2442+
}
2443+
tolerance := maxTotalSpace / 100
24282444
leastUsedIdx := -1
24292445
for idx, d := range filteredDisks {
2430-
if filteredDisks[idx].FreeSpace > freeSpace {
2431-
freeSpace = d.FreeSpace
2446+
if d.FreeSpace <= partSize || d.FreeSpace+tolerance < maxFreeSpace {
2447+
continue
2448+
}
2449+
if leastUsedIdx < 0 || assignedParts[d.Name] < assignedParts[filteredDisks[leastUsedIdx].Name] ||
2450+
(assignedParts[d.Name] == assignedParts[filteredDisks[leastUsedIdx].Name] && d.FreeSpace > filteredDisks[leastUsedIdx].FreeSpace) {
24322451
leastUsedIdx = idx
24332452
}
24342453
}
2435-
if leastUsedIdx < 0 {
2436-
return false, "", 0, errors.Errorf("%s free space, not found in system.disks with `local` type", utils.FormatBytes(partSize))
2437-
}
2454+
assignedParts[filteredDisks[leastUsedIdx].Name]++
24382455
return false, filteredDisks[leastUsedIdx].Name, filteredDisks[leastUsedIdx].FreeSpace - partSize, nil
24392456
}

‎pkg/backup/download_test.go‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -234,6 +234,50 @@ func TestReBalanceTablesMetadataIfDiskNotExists_Parts_NoErrors(t *testing.T) {
234234
}
235235
}
236236

237+
// JBOD disks on the same filesystem report almost the same free space, the difference is statvfs noise
238+
// from concurrent writes, parts shall be distributed across all of them instead of landing on one disk
239+
func TestReBalanceTablesMetadataIfDiskNotExists_SameFilesystemDisks(t *testing.T) {
240+
fb := Backuper{
241+
DiskToPathMap: map[string]string{"default": "/var/lib/clickhouse"},
242+
cfg: &config.Config{ClickHouse: config.ClickHouseConfig{ForceRebalance: true}},
243+
}
244+
const totalSpace = 100 * 1024 * 1024 * 1024
245+
sameFSDisks := []clickhouse.Disk{
246+
{Name: "hdd1", Path: "/hdd1_data", Type: "local", FreeSpace: 50*1024*1024*1024 + 10*1024*1024, TotalSpace: totalSpace, StoragePolicies: []string{"default"}},
247+
{Name: "hdd2", Path: "/hdd2_data", Type: "local", FreeSpace: 50 * 1024 * 1024 * 1024, TotalSpace: totalSpace, StoragePolicies: []string{"default"}},
248+
}
249+
table := metadata.TableMetadata{
250+
Parts: map[string][]metadata.Part{
251+
"default": {{Name: "202401_1_1_0"}, {Name: "202402_2_2_0"}, {Name: "202403_3_3_0"}, {Name: "202404_4_4_0"}},
252+
},
253+
Database: "default",
254+
Table: "test",
255+
Query: "CREATE TABLE default.test(id UInt64) ENGINE=MergeTree() ORDER BY id",
256+
TotalBytes: 4 * 8192,
257+
LocalFile: "/dev/null",
258+
}
259+
backup := storage.Backup{BackupMetadata: metadata.BackupMetadata{BackupName: "test", DataFormat: "directory", DiskTypes: map[string]string{"default": "local"}}}
260+
require.NoError(t, fb.reBalanceTablesMetadataIfDiskNotExists(ListOfTables{&table}, sameFSDisks, backup))
261+
partsByDisk := map[string]int{}
262+
for _, p := range table.Parts["default"] {
263+
partsByDisk[p.RebalancedDisk]++
264+
}
265+
assert.Equal(t, map[string]int{"hdd1": 2, "hdd2": 2}, partsByDisk)
266+
267+
// disks with really different free space keep `least_used` behavior
268+
differentDisks := []clickhouse.Disk{
269+
{Name: "hdd1", Path: "/hdd1_data", Type: "local", FreeSpace: 80 * 1024 * 1024 * 1024, TotalSpace: totalSpace, StoragePolicies: []string{"default"}},
270+
{Name: "hdd2", Path: "/hdd2_data", Type: "local", FreeSpace: 20 * 1024 * 1024 * 1024, TotalSpace: totalSpace, StoragePolicies: []string{"default"}},
271+
}
272+
for j := range table.Parts["default"] {
273+
table.Parts["default"][j].RebalancedDisk = ""
274+
}
275+
require.NoError(t, fb.reBalanceTablesMetadataIfDiskNotExists(ListOfTables{&table}, differentDisks, backup))
276+
for _, p := range table.Parts["default"] {
277+
assert.Equal(t, "hdd1", p.RebalancedDisk)
278+
}
279+
}
280+
237281
func TestReBalanceTablesMetadataIfDiskNotExists_CheckErrors(t *testing.T) {
238282
invalidRemoteBackup := remoteBackup
239283
invalidRemoteBackup.DataFormat = DirectoryFormat

‎test/integration/forceRebalance_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -107,9 +107,9 @@ XML
107107

108108
// Step 6: Verify backup data landed on both hdd1 and hdd2
109109
hdd1Out, err := env.DockerExecOut("clickhouse-backup", "find", "/hdd1_data/backup/"+backupName+"/shadow", "-type", "f")
110-
r.NoError(err)
110+
r.NoError(err, hdd1Out)
111111
hdd2Out, err := env.DockerExecOut("clickhouse-backup", "find", "/hdd2_data/backup/"+backupName+"/shadow", "-type", "f")
112-
r.NoError(err)
112+
r.NoError(err, hdd2Out)
113113
log.Debug().Msgf("hdd1 files:\n%s", hdd1Out)
114114
log.Debug().Msgf("hdd2 files:\n%s", hdd2Out)
115115
r.NotEmpty(hdd1Out, "hdd1 should contain some backup data")

0 commit comments

Comments
 (0)