command_volume_balance.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402
  1. package shell
  2. import (
  3. "flag"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  6. "github.com/chrislusf/seaweedfs/weed/storage/types"
  7. "io"
  8. "os"
  9. "sort"
  10. "time"
  11. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  12. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  13. )
  14. func init() {
  15. Commands = append(Commands, &commandVolumeBalance{})
  16. }
  17. type commandVolumeBalance struct {
  18. }
  19. func (c *commandVolumeBalance) Name() string {
  20. return "volume.balance"
  21. }
  22. func (c *commandVolumeBalance) Help() string {
  23. return `balance all volumes among volume servers
  24. volume.balance [-collection ALL|EACH_COLLECTION|<collection_name>] [-force] [-dataCenter=<data_center_name>]
  25. Algorithm:
  26. For each type of volume server (different max volume count limit){
  27. for each collection {
  28. balanceWritableVolumes()
  29. balanceReadOnlyVolumes()
  30. }
  31. }
  32. func balanceWritableVolumes(){
  33. idealWritableVolumeRatio = totalWritableVolumes / totalNumberOfMaxVolumes
  34. for hasMovedOneVolume {
  35. sort all volume servers ordered by the localWritableVolumeRatio = localWritableVolumes to localVolumeMax
  36. pick the volume server B with the highest localWritableVolumeRatio y
  37. for any the volume server A with the number of writable volumes x + 1 <= idealWritableVolumeRatio * localVolumeMax {
  38. if y > localWritableVolumeRatio {
  39. if B has a writable volume id v that A does not have, and satisfy v replication requirements {
  40. move writable volume v from A to B
  41. }
  42. }
  43. }
  44. }
  45. }
  46. func balanceReadOnlyVolumes(){
  47. //similar to balanceWritableVolumes
  48. }
  49. `
  50. }
  51. func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  52. if err = commandEnv.confirmIsLocked(); err != nil {
  53. return
  54. }
  55. balanceCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  56. collection := balanceCommand.String("collection", "EACH_COLLECTION", "collection name, or use \"ALL_COLLECTIONS\" across collections, \"EACH_COLLECTION\" for each collection")
  57. dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter")
  58. applyBalancing := balanceCommand.Bool("force", false, "apply the balancing plan.")
  59. if err = balanceCommand.Parse(args); err != nil {
  60. return nil
  61. }
  62. // collect topology information
  63. topologyInfo, volumeSizeLimitMb, err := collectTopologyInfo(commandEnv)
  64. if err != nil {
  65. return err
  66. }
  67. volumeServers := collectVolumeServersByDc(topologyInfo, *dc)
  68. volumeReplicas, _ := collectVolumeReplicaLocations(topologyInfo)
  69. diskTypes := collectVolumeDiskTypes(topologyInfo)
  70. if *collection == "EACH_COLLECTION" {
  71. collections, err := ListCollectionNames(commandEnv, true, false)
  72. if err != nil {
  73. return err
  74. }
  75. for _, c := range collections {
  76. if err = balanceVolumeServers(commandEnv, diskTypes, volumeReplicas, volumeServers, volumeSizeLimitMb*1024*1024, c, *applyBalancing); err != nil {
  77. return err
  78. }
  79. }
  80. } else if *collection == "ALL_COLLECTIONS" {
  81. if err = balanceVolumeServers(commandEnv, diskTypes, volumeReplicas, volumeServers, volumeSizeLimitMb*1024*1024, "ALL_COLLECTIONS", *applyBalancing); err != nil {
  82. return err
  83. }
  84. } else {
  85. if err = balanceVolumeServers(commandEnv, diskTypes, volumeReplicas, volumeServers, volumeSizeLimitMb*1024*1024, *collection, *applyBalancing); err != nil {
  86. return err
  87. }
  88. }
  89. return nil
  90. }
  91. func balanceVolumeServers(commandEnv *CommandEnv, diskTypes []types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, volumeSizeLimit uint64, collection string, applyBalancing bool) error {
  92. for _, diskType := range diskTypes {
  93. if err := balanceVolumeServersByDiskType(commandEnv, diskType, volumeReplicas, nodes, volumeSizeLimit, collection, applyBalancing); err != nil {
  94. return err
  95. }
  96. }
  97. return nil
  98. }
  99. func balanceVolumeServersByDiskType(commandEnv *CommandEnv, diskType types.DiskType, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, volumeSizeLimit uint64, collection string, applyBalancing bool) error {
  100. // balance writable volumes
  101. for _, n := range nodes {
  102. n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
  103. if collection != "ALL_COLLECTIONS" {
  104. if v.Collection != collection {
  105. return false
  106. }
  107. }
  108. return v.DiskType == string(diskType) && (!v.ReadOnly && v.Size < volumeSizeLimit)
  109. })
  110. }
  111. if err := balanceSelectedVolume(commandEnv, volumeReplicas, nodes, capacityByMaxVolumeCount(diskType), sortWritableVolumes, applyBalancing); err != nil {
  112. return err
  113. }
  114. // balance readable volumes
  115. for _, n := range nodes {
  116. n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool {
  117. if collection != "ALL_COLLECTIONS" {
  118. if v.Collection != collection {
  119. return false
  120. }
  121. }
  122. return v.DiskType == string(diskType) && (v.ReadOnly || v.Size >= volumeSizeLimit)
  123. })
  124. }
  125. if err := balanceSelectedVolume(commandEnv, volumeReplicas, nodes, capacityByMaxVolumeCount(diskType), sortReadOnlyVolumes, applyBalancing); err != nil {
  126. return err
  127. }
  128. return nil
  129. }
  130. func collectVolumeServersByDc(t *master_pb.TopologyInfo, selectedDataCenter string) (nodes []*Node) {
  131. for _, dc := range t.DataCenterInfos {
  132. if selectedDataCenter != "" && dc.Id != selectedDataCenter {
  133. continue
  134. }
  135. for _, r := range dc.RackInfos {
  136. for _, dn := range r.DataNodeInfos {
  137. nodes = append(nodes, &Node{
  138. info: dn,
  139. dc: dc.Id,
  140. rack: r.Id,
  141. })
  142. }
  143. }
  144. }
  145. return
  146. }
  147. func collectVolumeDiskTypes(t *master_pb.TopologyInfo) (diskTypes []types.DiskType) {
  148. knownTypes := make(map[string]bool)
  149. for _, dc := range t.DataCenterInfos {
  150. for _, r := range dc.RackInfos {
  151. for _, dn := range r.DataNodeInfos {
  152. for diskType, _ := range dn.DiskInfos {
  153. if _, found := knownTypes[diskType]; !found {
  154. knownTypes[diskType] = true
  155. }
  156. }
  157. }
  158. }
  159. }
  160. for diskType, _ := range knownTypes {
  161. diskTypes = append(diskTypes, types.ToDiskType(diskType))
  162. }
  163. return
  164. }
  165. type Node struct {
  166. info *master_pb.DataNodeInfo
  167. selectedVolumes map[uint32]*master_pb.VolumeInformationMessage
  168. dc string
  169. rack string
  170. }
  171. type CapacityFunc func(*master_pb.DataNodeInfo) int
  172. func capacityByMaxVolumeCount(diskType types.DiskType) CapacityFunc {
  173. return func(info *master_pb.DataNodeInfo) int {
  174. diskInfo, found := info.DiskInfos[string(diskType)]
  175. if !found {
  176. return 0
  177. }
  178. return int(diskInfo.MaxVolumeCount)
  179. }
  180. }
  181. func capacityByFreeVolumeCount(diskType types.DiskType) CapacityFunc {
  182. return func(info *master_pb.DataNodeInfo) int {
  183. diskInfo, found := info.DiskInfos[string(diskType)]
  184. if !found {
  185. return 0
  186. }
  187. return int(diskInfo.MaxVolumeCount - diskInfo.VolumeCount)
  188. }
  189. }
  190. func (n *Node) localVolumeRatio(capacityFunc CapacityFunc) float64 {
  191. return divide(len(n.selectedVolumes), capacityFunc(n.info))
  192. }
  193. func (n *Node) localVolumeNextRatio(capacityFunc CapacityFunc) float64 {
  194. return divide(len(n.selectedVolumes)+1, capacityFunc(n.info))
  195. }
  196. func (n *Node) selectVolumes(fn func(v *master_pb.VolumeInformationMessage) bool) {
  197. n.selectedVolumes = make(map[uint32]*master_pb.VolumeInformationMessage)
  198. for _, diskInfo := range n.info.DiskInfos {
  199. for _, v := range diskInfo.VolumeInfos {
  200. if fn(v) {
  201. n.selectedVolumes[v.Id] = v
  202. }
  203. }
  204. }
  205. }
  206. func sortWritableVolumes(volumes []*master_pb.VolumeInformationMessage) {
  207. sort.Slice(volumes, func(i, j int) bool {
  208. return volumes[i].Size < volumes[j].Size
  209. })
  210. }
  211. func sortReadOnlyVolumes(volumes []*master_pb.VolumeInformationMessage) {
  212. sort.Slice(volumes, func(i, j int) bool {
  213. return volumes[i].Id < volumes[j].Id
  214. })
  215. }
  216. func balanceSelectedVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, nodes []*Node, capacityFunc CapacityFunc, sortCandidatesFn func(volumes []*master_pb.VolumeInformationMessage), applyBalancing bool) (err error) {
  217. selectedVolumeCount, volumeMaxCount := 0, 0
  218. var nodesWithCapacity []*Node
  219. for _, dn := range nodes {
  220. selectedVolumeCount += len(dn.selectedVolumes)
  221. capacity := capacityFunc(dn.info)
  222. if capacity > 0 {
  223. nodesWithCapacity = append(nodesWithCapacity, dn)
  224. }
  225. volumeMaxCount += capacity
  226. }
  227. idealVolumeRatio := divide(selectedVolumeCount, volumeMaxCount)
  228. hasMoved := true
  229. // fmt.Fprintf(os.Stdout, " total %d volumes, max %d volumes, idealVolumeRatio %f\n", selectedVolumeCount, volumeMaxCount, idealVolumeRatio)
  230. for hasMoved {
  231. hasMoved = false
  232. sort.Slice(nodesWithCapacity, func(i, j int) bool {
  233. return nodesWithCapacity[i].localVolumeRatio(capacityFunc) < nodesWithCapacity[j].localVolumeRatio(capacityFunc)
  234. })
  235. fullNode := nodesWithCapacity[len(nodesWithCapacity)-1]
  236. var candidateVolumes []*master_pb.VolumeInformationMessage
  237. for _, v := range fullNode.selectedVolumes {
  238. candidateVolumes = append(candidateVolumes, v)
  239. }
  240. sortCandidatesFn(candidateVolumes)
  241. for i := 0; i < len(nodesWithCapacity)-1; i++ {
  242. emptyNode := nodesWithCapacity[i]
  243. if !(fullNode.localVolumeRatio(capacityFunc) > idealVolumeRatio && emptyNode.localVolumeNextRatio(capacityFunc) <= idealVolumeRatio) {
  244. // no more volume servers with empty slots
  245. break
  246. }
  247. hasMoved, err = attemptToMoveOneVolume(commandEnv, volumeReplicas, fullNode, candidateVolumes, emptyNode, applyBalancing)
  248. if err != nil {
  249. return
  250. }
  251. if hasMoved {
  252. // moved one volume
  253. break
  254. }
  255. }
  256. }
  257. return nil
  258. }
  259. func attemptToMoveOneVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, fullNode *Node, candidateVolumes []*master_pb.VolumeInformationMessage, emptyNode *Node, applyBalancing bool) (hasMoved bool, err error) {
  260. for _, v := range candidateVolumes {
  261. hasMoved, err = maybeMoveOneVolume(commandEnv, volumeReplicas, fullNode, v, emptyNode, applyBalancing)
  262. if err != nil {
  263. return
  264. }
  265. if hasMoved {
  266. break
  267. }
  268. }
  269. return
  270. }
  271. func maybeMoveOneVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][]*VolumeReplica, fullNode *Node, candidateVolume *master_pb.VolumeInformationMessage, emptyNode *Node, applyChange bool) (hasMoved bool, err error) {
  272. if candidateVolume.ReplicaPlacement > 0 {
  273. replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(candidateVolume.ReplicaPlacement))
  274. if !isGoodMove(replicaPlacement, volumeReplicas[candidateVolume.Id], fullNode, emptyNode) {
  275. return false, nil
  276. }
  277. }
  278. if _, found := emptyNode.selectedVolumes[candidateVolume.Id]; !found {
  279. if err = moveVolume(commandEnv, candidateVolume, fullNode, emptyNode, applyChange); err == nil {
  280. adjustAfterMove(candidateVolume, volumeReplicas, fullNode, emptyNode)
  281. return true, nil
  282. } else {
  283. return
  284. }
  285. }
  286. return
  287. }
  288. func moveVolume(commandEnv *CommandEnv, v *master_pb.VolumeInformationMessage, fullNode *Node, emptyNode *Node, applyChange bool) error {
  289. collectionPrefix := v.Collection + "_"
  290. if v.Collection == "" {
  291. collectionPrefix = ""
  292. }
  293. fmt.Fprintf(os.Stdout, " moving %s volume %s%d %s => %s\n", v.DiskType, collectionPrefix, v.Id, fullNode.info.Id, emptyNode.info.Id)
  294. if applyChange {
  295. return LiveMoveVolume(commandEnv.option.GrpcDialOption, needle.VolumeId(v.Id), fullNode.info.Id, emptyNode.info.Id, 5*time.Second, v.DiskType)
  296. }
  297. return nil
  298. }
  299. func isGoodMove(placement *super_block.ReplicaPlacement, existingReplicas []*VolumeReplica, sourceNode, targetNode *Node) bool {
  300. for _, replica := range existingReplicas {
  301. if replica.location.dataNode.Id == targetNode.info.Id &&
  302. replica.location.rack == targetNode.rack &&
  303. replica.location.dc == targetNode.dc {
  304. // never move to existing nodes
  305. return false
  306. }
  307. }
  308. dcs, racks := make(map[string]bool), make(map[string]int)
  309. for _, replica := range existingReplicas {
  310. if replica.location.dataNode.Id != sourceNode.info.Id {
  311. dcs[replica.location.DataCenter()] = true
  312. racks[replica.location.Rack()]++
  313. }
  314. }
  315. dcs[targetNode.dc] = true
  316. racks[fmt.Sprintf("%s %s", targetNode.dc, targetNode.rack)]++
  317. if len(dcs) != placement.DiffDataCenterCount+1 {
  318. return false
  319. }
  320. if len(racks) != placement.DiffRackCount+placement.DiffDataCenterCount+1 {
  321. return false
  322. }
  323. for _, sameRackCount := range racks {
  324. if sameRackCount != placement.SameRackCount+1 {
  325. return false
  326. }
  327. }
  328. return true
  329. }
  330. func adjustAfterMove(v *master_pb.VolumeInformationMessage, volumeReplicas map[uint32][]*VolumeReplica, fullNode *Node, emptyNode *Node) {
  331. delete(fullNode.selectedVolumes, v.Id)
  332. if emptyNode.selectedVolumes != nil {
  333. emptyNode.selectedVolumes[v.Id] = v
  334. }
  335. existingReplicas := volumeReplicas[v.Id]
  336. for _, replica := range existingReplicas {
  337. if replica.location.dataNode.Id == fullNode.info.Id &&
  338. replica.location.rack == fullNode.rack &&
  339. replica.location.dc == fullNode.dc {
  340. replica.location.dc = emptyNode.dc
  341. replica.location.rack = emptyNode.rack
  342. replica.location.dataNode = emptyNode.info
  343. return
  344. }
  345. }
  346. }