command_volume_balance.go 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253
  1. package shell
  2. import (
  3. "context"
  4. "flag"
  5. "fmt"
  6. "io"
  7. "os"
  8. "sort"
  9. "time"
  10. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  11. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  12. )
  13. func init() {
  14. Commands = append(Commands, &commandVolumeBalance{})
  15. }
  16. type commandVolumeBalance struct {
  17. }
  18. func (c *commandVolumeBalance) Name() string {
  19. return "volume.balance"
  20. }
  21. func (c *commandVolumeBalance) Help() string {
  22. return `balance all volumes among volume servers
  23. volume.balance [-collection ALL|EACH_COLLECTION|<collection_name>] [-force] [-dataCenter=<data_center_name>]
  24. Algorithm:
  25. For each type of volume server (different max volume count limit){
  26. for each collection {
  27. balanceWritableVolumes()
  28. balanceReadOnlyVolumes()
  29. }
  30. }
  31. func balanceWritableVolumes(){
  32. idealWritableVolumes = totalWritableVolumes / numVolumeServers
  33. for hasMovedOneVolume {
  34. sort all volume servers ordered by the number of local writable volumes
  35. pick the volume server A with the lowest number of writable volumes x
  36. pick the volume server B with the highest number of writable volumes y
  37. if y > idealWritableVolumes and x +1 <= idealWritableVolumes {
  38. if B has a writable volume id v that A does not have {
  39. move writable volume v from A to B
  40. }
  41. }
  42. }
  43. }
  44. func balanceReadOnlyVolumes(){
  45. //similar to balanceWritableVolumes
  46. }
  47. `
  48. }
  49. func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  50. balanceCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  51. collection := balanceCommand.String("collection", "EACH_COLLECTION", "collection name, or use \"ALL_COLLECTIONS\" across collections, \"EACH_COLLECTION\" for each collection")
  52. dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter")
  53. applyBalancing := balanceCommand.Bool("force", false, "apply the balancing plan.")
  54. if err = balanceCommand.Parse(args); err != nil {
  55. return nil
  56. }
  57. var resp *master_pb.VolumeListResponse
  58. err = commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
  59. resp, err = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
  60. return err
  61. })
  62. if err != nil {
  63. return err
  64. }
  65. typeToNodes := collectVolumeServersByType(resp.TopologyInfo, *dc)
  66. for maxVolumeCount, volumeServers := range typeToNodes {
  67. if len(volumeServers) < 2 {
  68. fmt.Printf("only 1 node is configured max %d volumes, skipping balancing\n", maxVolumeCount)
  69. continue
  70. }
  71. if *collection == "EACH_COLLECTION" {
  72. collections, err := ListCollectionNames(commandEnv, true, false)
  73. if err != nil {
  74. return err
  75. }
  76. for _, c := range collections {
  77. if err = balanceVolumeServers(commandEnv, volumeServers, resp.VolumeSizeLimitMb*1024*1024, c, *applyBalancing); err != nil {
  78. return err
  79. }
  80. }
  81. } else if *collection == "ALL_COLLECTIONS" {
  82. if err = balanceVolumeServers(commandEnv, volumeServers, resp.VolumeSizeLimitMb*1024*1024, "ALL_COLLECTIONS", *applyBalancing); err != nil {
  83. return err
  84. }
  85. } else {
  86. if err = balanceVolumeServers(commandEnv, volumeServers, resp.VolumeSizeLimitMb*1024*1024, *collection, *applyBalancing); err != nil {
  87. return err
  88. }
  89. }
  90. }
  91. return nil
  92. }
  93. func balanceVolumeServers(commandEnv *CommandEnv, nodes []*Node, volumeSizeLimit uint64, collection string, applyBalancing bool) error {
  94. // balance writable volumes
  95. for _, n := range nodes {
  96. n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
  97. if collection != "ALL_COLLECTIONS" {
  98. if v.Collection != collection {
  99. return false
  100. }
  101. }
  102. return !v.ReadOnly && v.Size < volumeSizeLimit
  103. })
  104. }
  105. if err := balanceSelectedVolume(commandEnv, nodes, sortWritableVolumes, applyBalancing); err != nil {
  106. return err
  107. }
  108. // balance readable volumes
  109. for _, n := range nodes {
  110. n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
  111. if collection != "ALL_COLLECTIONS" {
  112. if v.Collection != collection {
  113. return false
  114. }
  115. }
  116. return v.ReadOnly || v.Size >= volumeSizeLimit
  117. })
  118. }
  119. if err := balanceSelectedVolume(commandEnv, nodes, sortReadOnlyVolumes, applyBalancing); err != nil {
  120. return err
  121. }
  122. return nil
  123. }
  124. func collectVolumeServersByType(t *master_pb.TopologyInfo, selectedDataCenter string) (typeToNodes map[uint64][]*Node) {
  125. typeToNodes = make(map[uint64][]*Node)
  126. for _, dc := range t.DataCenterInfos {
  127. if selectedDataCenter != "" && dc.Id != selectedDataCenter {
  128. continue
  129. }
  130. for _, r := range dc.RackInfos {
  131. for _, dn := range r.DataNodeInfos {
  132. typeToNodes[dn.MaxVolumeCount] = append(typeToNodes[dn.MaxVolumeCount], &Node{
  133. info: dn,
  134. dc: dc.Id,
  135. rack: r.Id,
  136. })
  137. }
  138. }
  139. }
  140. return
  141. }
  142. type Node struct {
  143. info *master_pb.DataNodeInfo
  144. selectedVolumes map[uint32]*master_pb.VolumeInformationMessage
  145. dc string
  146. rack string
  147. }
  148. func sortWritableVolumes(volumes []*master_pb.VolumeInformationMessage) {
  149. sort.Slice(volumes, func(i, j int) bool {
  150. return volumes[i].Size < volumes[j].Size
  151. })
  152. }
  153. func sortReadOnlyVolumes(volumes []*master_pb.VolumeInformationMessage) {
  154. sort.Slice(volumes, func(i, j int) bool {
  155. return volumes[i].Id < volumes[j].Id
  156. })
  157. }
  158. func balanceSelectedVolume(commandEnv *CommandEnv, nodes []*Node, sortCandidatesFn func(volumes []*master_pb.VolumeInformationMessage), applyBalancing bool) error {
  159. selectedVolumeCount := 0
  160. for _, dn := range nodes {
  161. selectedVolumeCount += len(dn.selectedVolumes)
  162. }
  163. idealSelectedVolumes := ceilDivide(selectedVolumeCount, len(nodes))
  164. hasMove := true
  165. for hasMove {
  166. hasMove = false
  167. sort.Slice(nodes, func(i, j int) bool {
  168. // TODO sort by free volume slots???
  169. return len(nodes[i].selectedVolumes) < len(nodes[j].selectedVolumes)
  170. })
  171. emptyNode, fullNode := nodes[0], nodes[len(nodes)-1]
  172. if len(fullNode.selectedVolumes) > idealSelectedVolumes && len(emptyNode.selectedVolumes)+1 <= idealSelectedVolumes {
  173. // sort the volumes to move
  174. var candidateVolumes []*master_pb.VolumeInformationMessage
  175. for _, v := range fullNode.selectedVolumes {
  176. candidateVolumes = append(candidateVolumes, v)
  177. }
  178. sortCandidatesFn(candidateVolumes)
  179. for _, v := range candidateVolumes {
  180. if v.ReplicaPlacement > 0 {
  181. if fullNode.dc != emptyNode.dc && fullNode.rack != emptyNode.rack {
  182. // TODO this logic is too simple, but should work most of the time
  183. // Need a correct algorithm to handle all different cases
  184. continue
  185. }
  186. }
  187. if _, found := emptyNode.selectedVolumes[v.Id]; !found {
  188. if err := moveVolume(commandEnv, v, fullNode, emptyNode, applyBalancing); err == nil {
  189. delete(fullNode.selectedVolumes, v.Id)
  190. emptyNode.selectedVolumes[v.Id] = v
  191. hasMove = true
  192. break
  193. } else {
  194. return err
  195. }
  196. }
  197. }
  198. }
  199. }
  200. return nil
  201. }
  202. func moveVolume(commandEnv *CommandEnv, v *master_pb.VolumeInformationMessage, fullNode *Node, emptyNode *Node, applyBalancing bool) error {
  203. collectionPrefix := v.Collection + "_"
  204. if v.Collection == "" {
  205. collectionPrefix = ""
  206. }
  207. fmt.Fprintf(os.Stdout, "moving volume %s%d %s => %s\n", collectionPrefix, v.Id, fullNode.info.Id, emptyNode.info.Id)
  208. if applyBalancing {
  209. return LiveMoveVolume(commandEnv.option.GrpcDialOption, needle.VolumeId(v.Id), fullNode.info.Id, emptyNode.info.Id, 5*time.Second)
  210. }
  211. return nil
  212. }
  213. func (node *Node) selectVolumes(fn func(v *master_pb.VolumeInformationMessage) bool) {
  214. node.selectedVolumes = make(map[uint32]*master_pb.VolumeInformationMessage)
  215. for _, v := range node.info.VolumeInfos {
  216. if fn(v) {
  217. node.selectedVolumes[v.Id] = v
  218. }
  219. }
  220. }