You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

192 lines
5.8 KiB

  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  7. "github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
  8. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  9. "io"
  10. "sort"
  11. )
  12. func init() {
  13. Commands = append(Commands, &commandVolumeServerEvacuate{})
  14. }
  15. type commandVolumeServerEvacuate struct {
  16. }
  17. func (c *commandVolumeServerEvacuate) Name() string {
  18. return "volumeServer.evacuate"
  19. }
  20. func (c *commandVolumeServerEvacuate) Help() string {
  21. return `move out all data on a volume server
  22. volumeServer.evacuate -node <host:port>
  23. This command moves all data away from the volume server.
  24. The volumes on the volume servers will be redistributed.
  25. Usually this is used to prepare to shutdown or upgrade the volume server.
  26. `
  27. }
  28. func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  29. if err = commandEnv.confirmIsLocked(); err != nil {
  30. return
  31. }
  32. vsEvacuateCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  33. volumeServer := vsEvacuateCommand.String("node", "", "<host>:<port> of the volume server")
  34. applyChange := vsEvacuateCommand.Bool("force", false, "actually apply the changes")
  35. if err = vsEvacuateCommand.Parse(args); err != nil {
  36. return nil
  37. }
  38. if *volumeServer == "" {
  39. return fmt.Errorf("need to specify volume server by -node=<host>:<port>")
  40. }
  41. return volumeServerEvacuate(commandEnv, *volumeServer, *applyChange, writer)
  42. }
  43. func volumeServerEvacuate(commandEnv *CommandEnv, volumeServer string, applyChange bool, writer io.Writer) (err error) {
  44. // 1. confirm the volume server is part of the cluster
  45. // 2. collect all other volume servers, sort by empty slots
  46. // 3. move to any other volume server as long as it satisfy the replication requirements
  47. // list all the volumes
  48. var resp *master_pb.VolumeListResponse
  49. err = commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
  50. resp, err = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
  51. return err
  52. })
  53. if err != nil {
  54. return err
  55. }
  56. if err := evacuateNormalVolumes(commandEnv, resp, volumeServer, applyChange); err != nil {
  57. return err
  58. }
  59. if err := evacuateEcVolumes(commandEnv, resp, volumeServer, applyChange); err != nil {
  60. return err
  61. }
  62. return nil
  63. }
  64. func evacuateNormalVolumes(commandEnv *CommandEnv, resp *master_pb.VolumeListResponse, volumeServer string, applyChange bool) error {
  65. // find this volume server
  66. volumeServers := collectVolumeServersByDc(resp.TopologyInfo, "")
  67. thisNode, otherNodes := nodesOtherThan(volumeServers, volumeServer)
  68. if thisNode == nil {
  69. return fmt.Errorf("%s is not found in this cluster", volumeServer)
  70. }
  71. // move away normal volumes
  72. volumeReplicas, _ := collectVolumeReplicaLocations(resp)
  73. for _, vol := range thisNode.info.VolumeInfos {
  74. hasMoved, err := moveAwayOneNormalVolume(commandEnv, volumeReplicas, vol, thisNode, otherNodes, applyChange)
  75. if err != nil {
  76. return fmt.Errorf("move away volume %d from %s: %v", vol.Id, volumeServer, err)
  77. }
  78. if !hasMoved {
  79. return fmt.Errorf("failed to move volume %d from %s", vol.Id, volumeServer)
  80. }
  81. }
  82. return nil
  83. }
  84. func evacuateEcVolumes(commandEnv *CommandEnv, resp *master_pb.VolumeListResponse, volumeServer string, applyChange bool) error {
  85. // find this ec volume server
  86. ecNodes, _ := collectEcVolumeServersByDc(resp.TopologyInfo, "")
  87. thisNode, otherNodes := ecNodesOtherThan(ecNodes, volumeServer)
  88. if thisNode == nil {
  89. return fmt.Errorf("%s is not found in this cluster", volumeServer)
  90. }
  91. // move away ec volumes
  92. for _, ecShardInfo := range thisNode.info.EcShardInfos {
  93. hasMoved, err := moveAwayOneEcVolume(commandEnv, ecShardInfo, thisNode, otherNodes, applyChange)
  94. if err != nil {
  95. return fmt.Errorf("move away volume %d from %s: %v", ecShardInfo.Id, volumeServer, err)
  96. }
  97. if !hasMoved {
  98. return fmt.Errorf("failed to move ec volume %d from %s", ecShardInfo.Id, volumeServer)
  99. }
  100. }
  101. return nil
  102. }
  103. func moveAwayOneEcVolume(commandEnv *CommandEnv, ecShardInfo *master_pb.VolumeEcShardInformationMessage, thisNode *EcNode, otherNodes []*EcNode, applyChange bool) (hasMoved bool, err error) {
  104. for _, shardId := range erasure_coding.ShardBits(ecShardInfo.EcIndexBits).ShardIds() {
  105. sort.Slice(otherNodes, func(i, j int) bool {
  106. return otherNodes[i].localShardIdCount(ecShardInfo.Id) < otherNodes[j].localShardIdCount(ecShardInfo.Id)
  107. })
  108. for i := 0; i < len(otherNodes); i++ {
  109. emptyNode := otherNodes[i]
  110. err = moveMountedShardToEcNode(commandEnv, thisNode, ecShardInfo.Collection, needle.VolumeId(ecShardInfo.Id), shardId, emptyNode, applyChange)
  111. if err != nil {
  112. return
  113. } else {
  114. hasMoved = true
  115. break
  116. }
  117. }
  118. if !hasMoved {
  119. return
  120. }
  121. }
  122. return
  123. }
  124. func moveAwayOneNormalVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, vol *master_pb.VolumeInformationMessage, thisNode *Node, otherNodes []*Node, applyChange bool) (hasMoved bool, err error) {
  125. sort.Slice(otherNodes, func(i, j int) bool {
  126. return otherNodes[i].localVolumeRatio() < otherNodes[j].localVolumeRatio()
  127. })
  128. for i := 0; i < len(otherNodes); i++ {
  129. emptyNode := otherNodes[i]
  130. hasMoved, err = maybeMoveOneVolume(commandEnv, volumeReplicas, thisNode, vol, emptyNode, applyChange)
  131. if err != nil {
  132. return
  133. }
  134. if hasMoved {
  135. break
  136. }
  137. }
  138. return
  139. }
  140. func nodesOtherThan(volumeServers []*Node, thisServer string) (thisNode *Node, otherNodes []*Node) {
  141. for _, node := range volumeServers {
  142. if node.info.Id == thisServer {
  143. thisNode = node
  144. continue
  145. }
  146. otherNodes = append(otherNodes, node)
  147. }
  148. return
  149. }
  150. func ecNodesOtherThan(volumeServers []*EcNode, thisServer string) (thisNode *EcNode, otherNodes []*EcNode) {
  151. for _, node := range volumeServers {
  152. if node.info.Id == thisServer {
  153. thisNode = node
  154. continue
  155. }
  156. otherNodes = append(otherNodes, node)
  157. }
  158. return
  159. }