Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

collect disk types

pull/1804/head
Chris Lu 4 years ago
parent
861f8fab9b
commit
876dbe3d26
1 changed files with 27 additions and 9 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 36
      weed/shell/command_volume_balance.go

36
weed/shell/command_volume_balance.go
View File

@ -86,6 +86,7 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
volumeServers := collectVolumeServersByDc(resp.TopologyInfo, *dc) volumeServers := collectVolumeServersByDc(resp.TopologyInfo, *dc)
volumeReplicas, _ := collectVolumeReplicaLocations(resp) volumeReplicas, _ := collectVolumeReplicaLocations(resp)
diskTypes := collectVolumeDiskTypes(resp.TopologyInfo)
if *collection == "EACH_COLLECTION" { if *collection == "EACH_COLLECTION" {
collections, err := ListCollectionNames(commandEnv, true, false) collections, err := ListCollectionNames(commandEnv, true, false)
@ -93,16 +94,16 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
return err return err
} }
for _, c := range collections { for _, c := range collections {
if err = balanceVolumeServers(commandEnv, volumeReplicas, volumeServers, resp.VolumeSizeLimitMb*1024*1024, c, *applyBalancing); err != nil {
if err = balanceVolumeServers(commandEnv, diskTypes, volumeReplicas, volumeServers, resp.VolumeSizeLimitMb*1024*1024, c, *applyBalancing); err != nil {
return err return err
} }
} }
} else if *collection == "ALL_COLLECTIONS" { } else if *collection == "ALL_COLLECTIONS" {
if err = balanceVolumeServers(commandEnv, volumeReplicas, volumeServers, resp.VolumeSizeLimitMb*1024*1024, "ALL_COLLECTIONS", *applyBalancing); err != nil {
if err = balanceVolumeServers(commandEnv, diskTypes, volumeReplicas, volumeServers, resp.VolumeSizeLimitMb*1024*1024, "ALL_COLLECTIONS", *applyBalancing); err != nil {
return err return err
} }
} else { } else {
if err = balanceVolumeServers(commandEnv, volumeReplicas, volumeServers, resp.VolumeSizeLimitMb*1024*1024, *collection, *applyBalancing); err != nil {
if err = balanceVolumeServers(commandEnv, diskTypes, volumeReplicas, volumeServers, resp.VolumeSizeLimitMb*1024*1024, *collection, *applyBalancing); err != nil {
return err return err
} }
} }
@ -110,9 +111,9 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
return nil return nil
} }
func balanceVolumeServers(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, volumeSizeLimit uint64, collection string, applyBalancing bool) error {
func balanceVolumeServers(commandEnv *CommandEnv, diskTypes []storage.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, volumeSizeLimit uint64, collection string, applyBalancing bool) error {
for _, diskType := range []storage.DiskType{storage.HardDriveType, storage.SsdType} {
for _, diskType := range diskTypes {
if err := balanceVolumeServersByDiskType(commandEnv, diskType, volumeReplicas, nodes, volumeSizeLimit, collection, applyBalancing); err != nil { if err := balanceVolumeServersByDiskType(commandEnv, diskType, volumeReplicas, nodes, volumeSizeLimit, collection, applyBalancing); err != nil {
return err return err
} }
@ -123,8 +124,7 @@ func balanceVolumeServers(commandEnv *CommandEnv, volumeReplicas map[uint32][]*V
func balanceVolumeServersByDiskType(commandEnv *CommandEnv, diskType storage.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, volumeSizeLimit uint64, collection string, applyBalancing bool) error { func balanceVolumeServersByDiskType(commandEnv *CommandEnv, diskType storage.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, volumeSizeLimit uint64, collection string, applyBalancing bool) error {
// balance writable hdd volumes
// fmt.Fprintf(os.Stdout, "\nbalance collection %s writable hdd volumes\n", collection)
// balance writable volumes
for _, n := range nodes { for _, n := range nodes {
n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool { n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
if collection != "ALL_COLLECTIONS" { if collection != "ALL_COLLECTIONS" {
@ -139,8 +139,7 @@ func balanceVolumeServersByDiskType(commandEnv *CommandEnv, diskType storage.Dis
return err return err
} }
// balance readable hdd volumes
// fmt.Fprintf(os.Stdout, "\nbalance collection %s readable hdd volumes\n", collection)
// balance readable volumes
for _, n := range nodes { for _, n := range nodes {
n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool { n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
if collection != "ALL_COLLECTIONS" { if collection != "ALL_COLLECTIONS" {
@ -176,6 +175,25 @@ func collectVolumeServersByDc(t *master_pb.TopologyInfo, selectedDataCenter stri
return return
} }
func collectVolumeDiskTypes(t *master_pb.TopologyInfo) (diskTypes []storage.DiskType) {
knownTypes := make(map[string]bool)
for _, dc := range t.DataCenterInfos {
for _, r := range dc.RackInfos {
for _, dn := range r.DataNodeInfos {
for _, vi := range dn.VolumeInfos {
if _, found := knownTypes[vi.DiskType]; !found {
knownTypes[vi.DiskType] = true
}
}
}
}
}
for diskType, _ := range knownTypes {
diskTypes = append(diskTypes, storage.ToDiskType(diskType))
}
return
}
type Node struct { type Node struct {
info *master_pb.DataNodeInfo info *master_pb.DataNodeInfo
selectedVolumes map[uint32]*master_pb.VolumeInformationMessage selectedVolumes map[uint32]*master_pb.VolumeInformationMessage

Write Preview
Loading…
Cancel
Save