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.

234 lines
8.0 KiB

6 years ago
5 years ago
6 years ago
6 years ago
  1. package topology
  2. import (
  3. "fmt"
  4. "github.com/chrislusf/seaweedfs/weed/storage/types"
  5. "math/rand"
  6. "sync"
  7. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  8. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  9. "github.com/chrislusf/seaweedfs/weed/util"
  10. "google.golang.org/grpc"
  11. "github.com/chrislusf/seaweedfs/weed/glog"
  12. "github.com/chrislusf/seaweedfs/weed/storage"
  13. )
  14. /*
  15. This package is created to resolve these replica placement issues:
  16. 1. growth factor for each replica level, e.g., add 10 volumes for 1 copy, 20 volumes for 2 copies, 30 volumes for 3 copies
  17. 2. in time of tight storage, how to reduce replica level
  18. 3. optimizing for hot data on faster disk, cold data on cheaper storage,
  19. 4. volume allocation for each bucket
  20. */
  21. type VolumeGrowOption struct {
  22. Collection string
  23. ReplicaPlacement *super_block.ReplicaPlacement
  24. Ttl *needle.TTL
  25. DiskType types.DiskType
  26. Prealloacte int64
  27. DataCenter string
  28. Rack string
  29. DataNode string
  30. MemoryMapMaxSizeMb uint32
  31. }
  32. type VolumeGrowth struct {
  33. accessLock sync.Mutex
  34. }
  35. func (o *VolumeGrowOption) String() string {
  36. return fmt.Sprintf("Collection:%s, ReplicaPlacement:%v, Ttl:%v, DataCenter:%s, Rack:%s, DataNode:%s", o.Collection, o.ReplicaPlacement, o.Ttl, o.DataCenter, o.Rack, o.DataNode)
  37. }
  38. func NewDefaultVolumeGrowth() *VolumeGrowth {
  39. return &VolumeGrowth{}
  40. }
  41. // one replication type may need rp.GetCopyCount() actual volumes
  42. // given copyCount, how many logical volumes to create
  43. func (vg *VolumeGrowth) findVolumeCount(copyCount int) (count int) {
  44. v := util.GetViper()
  45. v.SetDefault("master.volume_growth.copy_1", 7)
  46. v.SetDefault("master.volume_growth.copy_2", 6)
  47. v.SetDefault("master.volume_growth.copy_3", 3)
  48. v.SetDefault("master.volume_growth.copy_other", 1)
  49. switch copyCount {
  50. case 1:
  51. count = v.GetInt("master.volume_growth.copy_1")
  52. case 2:
  53. count = v.GetInt("master.volume_growth.copy_2")
  54. case 3:
  55. count = v.GetInt("master.volume_growth.copy_3")
  56. default:
  57. count = v.GetInt("master.volume_growth.copy_other")
  58. }
  59. return
  60. }
  61. func (vg *VolumeGrowth) AutomaticGrowByType(option *VolumeGrowOption, grpcDialOption grpc.DialOption, topo *Topology, targetCount int) (count int, err error) {
  62. if targetCount == 0 {
  63. targetCount = vg.findVolumeCount(option.ReplicaPlacement.GetCopyCount())
  64. }
  65. count, err = vg.GrowByCountAndType(grpcDialOption, targetCount, option, topo)
  66. if count > 0 && count%option.ReplicaPlacement.GetCopyCount() == 0 {
  67. return count, nil
  68. }
  69. return count, err
  70. }
  71. func (vg *VolumeGrowth) GrowByCountAndType(grpcDialOption grpc.DialOption, targetCount int, option *VolumeGrowOption, topo *Topology) (counter int, err error) {
  72. vg.accessLock.Lock()
  73. defer vg.accessLock.Unlock()
  74. for i := 0; i < targetCount; i++ {
  75. if c, e := vg.findAndGrow(grpcDialOption, topo, option); e == nil {
  76. counter += c
  77. } else {
  78. glog.V(0).Infof("create %d volume, created %d: %v", targetCount, counter, e)
  79. return counter, e
  80. }
  81. }
  82. return
  83. }
  84. func (vg *VolumeGrowth) findAndGrow(grpcDialOption grpc.DialOption, topo *Topology, option *VolumeGrowOption) (int, error) {
  85. servers, e := vg.findEmptySlotsForOneVolume(topo, option)
  86. if e != nil {
  87. return 0, e
  88. }
  89. vid, raftErr := topo.NextVolumeId()
  90. if raftErr != nil {
  91. return 0, raftErr
  92. }
  93. err := vg.grow(grpcDialOption, topo, vid, option, servers...)
  94. return len(servers), err
  95. }
  96. // 1. find the main data node
  97. // 1.1 collect all data nodes that have 1 slots
  98. // 2.2 collect all racks that have rp.SameRackCount+1
  99. // 2.2 collect all data centers that have DiffRackCount+rp.SameRackCount+1
  100. // 2. find rest data nodes
  101. func (vg *VolumeGrowth) findEmptySlotsForOneVolume(topo *Topology, option *VolumeGrowOption) (servers []*DataNode, err error) {
  102. //find main datacenter and other data centers
  103. rp := option.ReplicaPlacement
  104. mainDataCenter, otherDataCenters, dc_err := topo.PickNodesByWeight(rp.DiffDataCenterCount+1, option, func(node Node) error {
  105. if option.DataCenter != "" && node.IsDataCenter() && node.Id() != NodeId(option.DataCenter) {
  106. return fmt.Errorf("Not matching preferred data center:%s", option.DataCenter)
  107. }
  108. if len(node.Children()) < rp.DiffRackCount+1 {
  109. return fmt.Errorf("Only has %d racks, not enough for %d.", len(node.Children()), rp.DiffRackCount+1)
  110. }
  111. if node.AvailableSpaceFor(option) < int64(rp.DiffRackCount+rp.SameRackCount+1) {
  112. return fmt.Errorf("Free:%d < Expected:%d", node.AvailableSpaceFor(option), rp.DiffRackCount+rp.SameRackCount+1)
  113. }
  114. possibleRacksCount := 0
  115. for _, rack := range node.Children() {
  116. possibleDataNodesCount := 0
  117. for _, n := range rack.Children() {
  118. if n.AvailableSpaceFor(option) >= 1 {
  119. possibleDataNodesCount++
  120. }
  121. }
  122. if possibleDataNodesCount >= rp.SameRackCount+1 {
  123. possibleRacksCount++
  124. }
  125. }
  126. if possibleRacksCount < rp.DiffRackCount+1 {
  127. return fmt.Errorf("Only has %d racks with more than %d free data nodes, not enough for %d.", possibleRacksCount, rp.SameRackCount+1, rp.DiffRackCount+1)
  128. }
  129. return nil
  130. })
  131. if dc_err != nil {
  132. return nil, dc_err
  133. }
  134. //find main rack and other racks
  135. mainRack, otherRacks, rackErr := mainDataCenter.(*DataCenter).PickNodesByWeight(rp.DiffRackCount+1, option, func(node Node) error {
  136. if option.Rack != "" && node.IsRack() && node.Id() != NodeId(option.Rack) {
  137. return fmt.Errorf("Not matching preferred rack:%s", option.Rack)
  138. }
  139. if node.AvailableSpaceFor(option) < int64(rp.SameRackCount+1) {
  140. return fmt.Errorf("Free:%d < Expected:%d", node.AvailableSpaceFor(option), rp.SameRackCount+1)
  141. }
  142. if len(node.Children()) < rp.SameRackCount+1 {
  143. // a bit faster way to test free racks
  144. return fmt.Errorf("Only has %d data nodes, not enough for %d.", len(node.Children()), rp.SameRackCount+1)
  145. }
  146. possibleDataNodesCount := 0
  147. for _, n := range node.Children() {
  148. if n.AvailableSpaceFor(option) >= 1 {
  149. possibleDataNodesCount++
  150. }
  151. }
  152. if possibleDataNodesCount < rp.SameRackCount+1 {
  153. return fmt.Errorf("Only has %d data nodes with a slot, not enough for %d.", possibleDataNodesCount, rp.SameRackCount+1)
  154. }
  155. return nil
  156. })
  157. if rackErr != nil {
  158. return nil, rackErr
  159. }
  160. //find main rack and other racks
  161. mainServer, otherServers, serverErr := mainRack.(*Rack).PickNodesByWeight(rp.SameRackCount+1, option, func(node Node) error {
  162. if option.DataNode != "" && node.IsDataNode() && node.Id() != NodeId(option.DataNode) {
  163. return fmt.Errorf("Not matching preferred data node:%s", option.DataNode)
  164. }
  165. if node.AvailableSpaceFor(option) < 1 {
  166. return fmt.Errorf("Free:%d < Expected:%d", node.AvailableSpaceFor(option), 1)
  167. }
  168. return nil
  169. })
  170. if serverErr != nil {
  171. return nil, serverErr
  172. }
  173. servers = append(servers, mainServer.(*DataNode))
  174. for _, server := range otherServers {
  175. servers = append(servers, server.(*DataNode))
  176. }
  177. for _, rack := range otherRacks {
  178. r := rand.Int63n(rack.AvailableSpaceFor(option))
  179. if server, e := rack.ReserveOneVolume(r, option); e == nil {
  180. servers = append(servers, server)
  181. } else {
  182. return servers, e
  183. }
  184. }
  185. for _, datacenter := range otherDataCenters {
  186. r := rand.Int63n(datacenter.AvailableSpaceFor(option))
  187. if server, e := datacenter.ReserveOneVolume(r, option); e == nil {
  188. servers = append(servers, server)
  189. } else {
  190. return servers, e
  191. }
  192. }
  193. return
  194. }
  195. func (vg *VolumeGrowth) grow(grpcDialOption grpc.DialOption, topo *Topology, vid needle.VolumeId, option *VolumeGrowOption, servers ...*DataNode) error {
  196. for _, server := range servers {
  197. if err := AllocateVolume(server, grpcDialOption, vid, option); err == nil {
  198. vi := storage.VolumeInfo{
  199. Id: vid,
  200. Size: 0,
  201. Collection: option.Collection,
  202. ReplicaPlacement: option.ReplicaPlacement,
  203. Ttl: option.Ttl,
  204. Version: needle.CurrentVersion,
  205. DiskType: string(option.DiskType),
  206. }
  207. server.AddOrUpdateVolume(vi)
  208. topo.RegisterVolumeLayout(vi, server)
  209. glog.V(0).Infoln("Created Volume", vid, "on", server.NodeImpl.String())
  210. } else {
  211. glog.V(0).Infoln("Failed to assign volume", vid, "to", servers, "error", err)
  212. return fmt.Errorf("Failed to assign %d: %v", vid, err)
  213. }
  214. }
  215. return nil
  216. }