mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
volume server evacuate to target server
This commit is contained in:
parent
b6471ecd75
commit
ba0e3ce5fa
|
@ -47,7 +47,7 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
|
|||
|
||||
vsEvacuateCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
|
||||
volumeServer := vsEvacuateCommand.String("node", "", "<host>:<port> of the volume server")
|
||||
c.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")
|
||||
applyChange := vsEvacuateCommand.Bool("force", false, "actually apply the changes")
|
||||
retryCount := vsEvacuateCommand.Int("retry", 0, "how many times to retry")
|
||||
|
@ -63,6 +63,9 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
|
|||
if *volumeServer == "" {
|
||||
return fmt.Errorf("need to specify volume server by -node=<host>:<port>")
|
||||
}
|
||||
if *targetServer != "" {
|
||||
c.targetServer = *targetServer
|
||||
}
|
||||
for i := 0; i < *retryCount+1; i++ {
|
||||
if err = c.volumeServerEvacuate(commandEnv, *volumeServer, *skipNonMoveable, *applyChange, writer); err == nil {
|
||||
return nil
|
||||
|
@ -103,14 +106,27 @@ func (c *commandVolumeServerEvacuate) evacuateNormalVolumes(commandEnv *CommandE
|
|||
if thisNode == nil {
|
||||
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
|
||||
volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo)
|
||||
for _, diskInfo := range thisNode.info.DiskInfos {
|
||||
for _, vol := range diskInfo.VolumeInfos {
|
||||
hasMoved, err := moveAwayOneNormalVolume(commandEnv, volumeReplicas, vol, thisNode, otherNodes, applyChange)
|
||||
hasMoved, err := c.moveAwayOneNormalVolume(commandEnv, volumeReplicas, vol, thisNode, otherNodes, applyChange)
|
||||
if err != nil {
|
||||
return fmt.Errorf("move away volume %d from %s: %v", vol.Id, volumeServer, err)
|
||||
fmt.Fprintf(writer, "move away volume %d from %s: %v", vol.Id, volumeServer, err)
|
||||
}
|
||||
if !hasMoved {
|
||||
if skipNonMoveable {
|
||||
|
@ -138,7 +154,7 @@ func (c *commandVolumeServerEvacuate) evacuateEcVolumes(commandEnv *CommandEnv,
|
|||
for _, ecShardInfo := range diskInfo.EcShardInfos {
|
||||
hasMoved, err := c.moveAwayOneEcVolume(commandEnv, ecShardInfo, thisNode, otherNodes, applyChange)
|
||||
if err != nil {
|
||||
return fmt.Errorf("move away volume %d from %s: %v", ecShardInfo.Id, volumeServer, err)
|
||||
fmt.Fprintf(writer, "move away volume %d from %s: %v", ecShardInfo.Id, volumeServer, err)
|
||||
}
|
||||
if !hasMoved {
|
||||
if skipNonMoveable {
|
||||
|
@ -160,9 +176,6 @@ func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv
|
|||
})
|
||||
for i := 0; i < len(otherNodes); i++ {
|
||||
emptyNode := otherNodes[i]
|
||||
if c.targetServer != "" && c.targetServer != emptyNode.info.Id {
|
||||
continue
|
||||
}
|
||||
collectionPrefix := ""
|
||||
if ecShardInfo.Collection != "" {
|
||||
collectionPrefix = ecShardInfo.Collection + "_"
|
||||
|
@ -184,7 +197,7 @@ func (c *commandVolumeServerEvacuate) moveAwayOneEcVolume(commandEnv *CommandEnv
|
|||
return
|
||||
}
|
||||
|
||||
func moveAwayOneNormalVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, vol *master_pb.VolumeInformationMessage, thisNode *Node, otherNodes []*Node, applyChange bool) (hasMoved bool, err error) {
|
||||
func (c *commandVolumeServerEvacuate) moveAwayOneNormalVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, vol *master_pb.VolumeInformationMessage, thisNode *Node, otherNodes []*Node, applyChange bool) (hasMoved bool, err error) {
|
||||
fn := capacityByFreeVolumeCount(types.ToDiskType(vol.DiskType))
|
||||
for _, n := range otherNodes {
|
||||
n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
|
||||
|
|
Loading…
Reference in a new issue