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.

383 lines
12 KiB

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