mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-12-24 17:47:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

[shell] only apply the balancing for writable volumes (#6346)

Browse Source
This commit is contained in:
Konstantin Lebedev 2024-12-13 11:10:00 +02:00 committed by GitHub
parent 9987a65e8a
commit 0a4b1909a2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 26 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
weed/shell/command_volume_balance.go
View File

@ -24,6 +24,10 @@ func init() {
} }
type commandVolumeBalance struct { type commandVolumeBalance struct {
volumeSizeLimitMb uint64
commandEnv *CommandEnv
writable bool
applyBalancing bool
} }
func (c *commandVolumeBalance) Name() string { func (c *commandVolumeBalance) Name() string {
@ -76,12 +80,13 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter") dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter")
racks := balanceCommand.String("racks", "", "only apply the balancing for this racks") racks := balanceCommand.String("racks", "", "only apply the balancing for this racks")
nodes := balanceCommand.String("nodes", "", "only apply the balancing for this nodes") nodes := balanceCommand.String("nodes", "", "only apply the balancing for this nodes")
c.writable = *(balanceCommand.Bool("writable", false, "only apply the balancing for writable volumes"))
noLock := balanceCommand.Bool("noLock", false, "do not lock the admin shell at one's own risk") noLock := balanceCommand.Bool("noLock", false, "do not lock the admin shell at one's own risk")
applyBalancing := balanceCommand.Bool("force", false, "apply the balancing plan.") c.applyBalancing = *(balanceCommand.Bool("force", false, "apply the balancing plan."))
if err = balanceCommand.Parse(args); err != nil { if err = balanceCommand.Parse(args); err != nil {
return nil return nil
} }
infoAboutSimulationMode(writer, *applyBalancing, "-force") infoAboutSimulationMode(writer, c.applyBalancing, "-force")
if *noLock { if *noLock {
commandEnv.noLock = true commandEnv.noLock = true
@ -90,9 +95,11 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
return return
} }
} }
c.commandEnv = commandEnv
// collect topology information // collect topology information
topologyInfo, _, err := collectTopologyInfo(commandEnv, 5*time.Second) var topologyInfo *master_pb.TopologyInfo
topologyInfo, c.volumeSizeLimitMb, err = collectTopologyInfo(commandEnv, 5*time.Second)
if err != nil { if err != nil {
return err return err
} }
@ -106,13 +113,13 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
if err != nil { if err != nil {
return err return err
} }
for _, c := range collections { for _, col := range collections {
if err = balanceVolumeServers(commandEnv, diskTypes, volumeReplicas, volumeServers, c, *applyBalancing); err != nil { if err = c.balanceVolumeServers(diskTypes, volumeReplicas, volumeServers, col); err != nil {
return err return err
} }
} }
} else { } else {
if err = balanceVolumeServers(commandEnv, diskTypes, volumeReplicas, volumeServers, *collection, *applyBalancing); err != nil { if err = c.balanceVolumeServers(diskTypes, volumeReplicas, volumeServers, *collection); err != nil {
return err return err
} }
} }
@ -120,10 +127,10 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
return nil return nil
} }
func balanceVolumeServers(commandEnv *CommandEnv, diskTypes []types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, collection string, applyBalancing bool) error { func (c *commandVolumeBalance) balanceVolumeServers(diskTypes []types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, collection string) error {
for _, diskType := range diskTypes { for _, diskType := range diskTypes {
if err := balanceVolumeServersByDiskType(commandEnv, diskType, volumeReplicas, nodes, collection, applyBalancing); err != nil { if err := c.balanceVolumeServersByDiskType(diskType, volumeReplicas, nodes, collection); err != nil {
return err return err
} }
} }
@ -131,7 +138,7 @@ func balanceVolumeServers(commandEnv *CommandEnv, diskTypes []types.DiskType, vo
} }
func balanceVolumeServersByDiskType(commandEnv *CommandEnv, diskType types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, collection string, applyBalancing bool) error { func (c *commandVolumeBalance) balanceVolumeServersByDiskType(diskType types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, collection string) error {
for _, n := range nodes { for _, n := range nodes {
n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool { n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
@ -140,10 +147,16 @@ func balanceVolumeServersByDiskType(commandEnv *CommandEnv, diskType types.DiskT
return false return false
} }
} }
return v.DiskType == string(diskType) if v.DiskType != string(diskType) {
return false
}
if c.writable && v.Size > c.volumeSizeLimitMb {
return false
}
return true
}) })
} }
if err := balanceSelectedVolume(commandEnv, diskType, volumeReplicas, nodes, sortWritableVolumes, applyBalancing); err != nil { if err := balanceSelectedVolume(c.commandEnv, diskType, volumeReplicas, nodes, sortWritableVolumes, c.applyBalancing); err != nil {
return err return err
} }

4
weed/shell/command_volume_balance_test.go
View File

@ -255,8 +255,8 @@ func TestBalance(t *testing.T) {
volumeServers := collectVolumeServersByDcRackNode(topologyInfo, "", "", "") volumeServers := collectVolumeServersByDcRackNode(topologyInfo, "", "", "")
volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo) volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo)
diskTypes := collectVolumeDiskTypes(topologyInfo) diskTypes := collectVolumeDiskTypes(topologyInfo)
c := &commandVolumeBalance{}
if err := balanceVolumeServers(nil, diskTypes, volumeReplicas, volumeServers, "ALL_COLLECTIONS", false); err != nil { if err := c.balanceVolumeServers(diskTypes, volumeReplicas, volumeServers, "ALL_COLLECTIONS"); err != nil {
t.Errorf("balance: %v", err) t.Errorf("balance: %v", err)
} }