gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

volume server evacuate from rack

Browse source
This commit is contained in:
Konstantin Lebedev 2022-07-12 11:33:08 +05:00
parent 4236c36599
commit 087fa1347f
1 changed files with 56 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

103
weed/shell/command_volume_server_evacuate.go
View file

@ -19,6 +19,7 @@ func init() {
type commandVolumeServerEvacuate struct { type commandVolumeServerEvacuate struct {
targetServer string targetServer string
volumeRack string
} }
func (c *commandVolumeServerEvacuate) Name() string { func (c *commandVolumeServerEvacuate) Name() string {
@ -47,6 +48,7 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
vsEvacuateCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) vsEvacuateCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
volumeServer := vsEvacuateCommand.String("node", "", "<host>:<port> of the volume server") volumeServer := vsEvacuateCommand.String("node", "", "<host>:<port> of the volume server")
volumeRack := vsEvacuateCommand.String("rack", "", "rack for then volume servers")
targetServer := vsEvacuateCommand.String("target", "", "<host>:<port> of target volume") targetServer := vsEvacuateCommand.String("target", "", "<host>:<port> of target volume")
skipNonMoveable := vsEvacuateCommand.Bool("skipNonMoveable", false, "skip volumes that can not be moved") skipNonMoveable := vsEvacuateCommand.Bool("skipNonMoveable", false, "skip volumes that can not be moved")
applyChange := vsEvacuateCommand.Bool("force", false, "actually apply the changes") applyChange := vsEvacuateCommand.Bool("force", false, "actually apply the changes")
@ -66,6 +68,9 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
if *targetServer != "" { if *targetServer != "" {
c.targetServer = *targetServer c.targetServer = *targetServer
} }
if *volumeRack != "" {
c.volumeRack = *volumeRack
}
for i := 0; i < *retryCount+1; i++ { for i := 0; i < *retryCount+1; i++ {
if err = c.volumeServerEvacuate(commandEnv, *volumeServer, *skipNonMoveable, *applyChange, writer); err == nil { if err = c.volumeServerEvacuate(commandEnv, *volumeServer, *skipNonMoveable, *applyChange, writer); err == nil {
return nil return nil
@ -102,41 +107,31 @@ func (c *commandVolumeServerEvacuate) volumeServerEvacuate(commandEnv *CommandEn
func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error { func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error {
// find this volume server // find this volume server
volumeServers := collectVolumeServersByDc(topologyInfo, "") volumeServers := collectVolumeServersByDc(topologyInfo, "")
thisNode, otherNodes := nodesOtherThan(volumeServers, volumeServer) thisNodes, otherNodes := c.nodesOtherThan(volumeServers, volumeServer)
if thisNode == nil { if len(thisNodes) == 0 {
return fmt.Errorf("%s is not found in this cluster", volumeServer) return fmt.Errorf("%s is not found in this cluster", volumeServer)
} }
if c.targetServer != "" {
targetServerFound := false
for _, otherNode := range otherNodes {
if otherNode.info.Id == c.targetServer {
otherNodes = []*Node{otherNode}
targetServerFound = true
break
}
}
if !targetServerFound {
return fmt.Errorf("target %s is not found in this cluster", c.targetServer)
}
}
// move away normal volumes // move away normal volumes
volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo) for _, thisNode := range thisNodes {
for _, diskInfo := range thisNode.info.DiskInfos { for _, diskInfo := range thisNode.info.DiskInfos {
for _, vol := range diskInfo.VolumeInfos { volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo)
hasMoved, err := c.moveAwayOneNormalVolume(commandEnv, volumeReplicas, vol, thisNode, otherNodes, applyChange) for _, vol := range diskInfo.VolumeInfos {
if err != nil { hasMoved, err := c.moveAwayOneNormalVolume(commandEnv, volumeReplicas, vol, thisNode, otherNodes, applyChange)
fmt.Fprintf(writer, "move away volume %d from %s: %v", vol.Id, volumeServer, err) if err != nil {
} fmt.Fprintf(writer, "move away volume %d from %s: %v", vol.Id, volumeServer, err)
if !hasMoved { }
if skipNonMoveable { if !hasMoved {
replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(vol.ReplicaPlacement)) if skipNonMoveable {
fmt.Fprintf(writer, "skipping non moveable volume %d replication:%s\n", vol.Id, replicaPlacement.String()) replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(vol.ReplicaPlacement))
} else { fmt.Fprintf(writer, "skipping non moveable volume %d replication:%s\n", vol.Id, replicaPlacement.String())
return fmt.Errorf("failed to move volume %d from %s", vol.Id, volumeServer) } else {
return fmt.Errorf("failed to move volume %d from %s", vol.Id, volumeServer)
}
} }
} }
} }
} }
return nil return nil
} }
@ -144,23 +139,25 @@ func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandE
func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error { func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeServer string, skipNonMoveable, applyChange bool, writer io.Writer) error {
// find this ec volume server // find this ec volume server
ecNodes, _ := collectEcVolumeServersByDc(topologyInfo, "") ecNodes, _ := collectEcVolumeServersByDc(topologyInfo, "")
thisNode, otherNodes := ecNodesOtherThan(ecNodes, volumeServer) thisNodes, otherNodes := c.ecNodesOtherThan(ecNodes, volumeServer)
if thisNode == nil { if len(thisNodes) == 0 {
return fmt.Errorf("%s is not found in this cluster\n", volumeServer) return fmt.Errorf("%s is not found in this cluster\n", volumeServer)
} }
// move away ec volumes // move away ec volumes
for _, diskInfo := range thisNode.info.DiskInfos { for _, thisNode := range thisNodes {
for _, ecShardInfo := range diskInfo.EcShardInfos { for _, diskInfo := range thisNode.info.DiskInfos {
hasMoved, err := c.moveAwayOneEcVolume(commandEnv, ecShardInfo, thisNode, otherNodes, applyChange) for _, ecShardInfo := range diskInfo.EcShardInfos {
if err != nil { hasMoved, err := c.moveAwayOneEcVolume(commandEnv, ecShardInfo, thisNode, otherNodes, applyChange)
fmt.Fprintf(writer, "move away volume %d from %s: %v", ecShardInfo.Id, volumeServer, err) if err != nil {
} fmt.Fprintf(writer, "move away volume %d from %s: %v", ecShardInfo.Id, volumeServer, err)
if !hasMoved { }
if skipNonMoveable { if !hasMoved {
fmt.Fprintf(writer, "failed to move away ec volume %d from %s\n", ecShardInfo.Id, volumeServer) if skipNonMoveable {
} else { fmt.Fprintf(writer, "failed to move away ec volume %d from %s\n", ecShardInfo.Id, volumeServer)
return fmt.Errorf("failed to move away ec volume %d from %s", ecShardInfo.Id, volumeServer) } else {
return fmt.Errorf("failed to move away ec volume %d from %s", ecShardInfo.Id, volumeServer)
}
} }
} }
} }
@ -220,10 +217,16 @@ func (c *commandVolumeServerEvacuate) moveAwayOneNormalVolume(commandEnv *Comman
return return
} }
func nodesOtherThan(volumeServers []*Node, thisServer string) (thisNode *Node, otherNodes []*Node) { func (c *commandVolumeServerEvacuate) nodesOtherThan(volumeServers []*Node, thisServer string) (thisNodes []*Node, otherNodes []*Node) {
for _, node := range volumeServers { for _, node := range volumeServers {
if node.info.Id == thisServer { if node.info.Id == thisServer || (c.volumeRack != "" && node.rack == c.volumeRack) {
thisNode = node thisNodes = append(thisNodes, node)
continue
}
if c.volumeRack != "" && c.volumeRack == node.rack {
continue
}
if c.targetServer != "" && c.targetServer != node.info.Id {
continue continue
} }
otherNodes = append(otherNodes, node) otherNodes = append(otherNodes, node)
@ -231,10 +234,16 @@ func nodesOtherThan(volumeServers []*Node, thisServer string) (thisNode *Node, o
return return
} }
func ecNodesOtherThan(volumeServers []*EcNode, thisServer string) (thisNode *EcNode, otherNodes []*EcNode) { func (c *commandVolumeServerEvacuate) ecNodesOtherThan(volumeServers []*EcNode, thisServer string) (thisNodes []*EcNode, otherNodes []*EcNode) {
for _, node := range volumeServers { for _, node := range volumeServers {
if node.info.Id == thisServer { if node.info.Id == thisServer || (c.volumeRack != "" && string(node.rack) == c.volumeRack) {
thisNode = node thisNodes = append(thisNodes, node)
continue
}
if c.volumeRack != "" && c.volumeRack == string(node.rack) {
continue
}
if c.targetServer != "" && c.targetServer != node.info.Id {
continue continue
} }
otherNodes = append(otherNodes, node) otherNodes = append(otherNodes, node)