command_volume_balance.go 13 KB

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