volume_growth.go 7.2 KB

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