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.

283 lines
9.5 KiB

6 years ago
7 months ago
7 months ago
7 months ago
7 months ago
3 years ago
6 years ago
3 years ago
  1. package topology
  2. import (
  3. "encoding/json"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  6. "math/rand"
  7. "sync"
  8. "time"
  9. "google.golang.org/grpc"
  10. "github.com/seaweedfs/seaweedfs/weed/glog"
  11. "github.com/seaweedfs/seaweedfs/weed/storage"
  12. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  13. "github.com/seaweedfs/seaweedfs/weed/storage/super_block"
  14. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  15. )
  16. /*
  17. This package is created to resolve these replica placement issues:
  18. 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
  19. 2. in time of tight storage, how to reduce replica level
  20. 3. optimizing for hot data on faster disk, cold data on cheaper storage,
  21. 4. volume allocation for each bucket
  22. */
  23. type VolumeGrowRequest struct {
  24. Option *VolumeGrowOption
  25. Count uint32
  26. }
  27. type volumeGrowthStrategy struct {
  28. Copy1Count uint32
  29. Copy2Count uint32
  30. Copy3Count uint32
  31. CopyOtherCount uint32
  32. Threshold float64
  33. }
  34. var (
  35. VolumeGrowStrategy = volumeGrowthStrategy{
  36. Copy1Count: 7,
  37. Copy2Count: 6,
  38. Copy3Count: 3,
  39. CopyOtherCount: 1,
  40. Threshold: 0.9,
  41. }
  42. )
  43. type VolumeGrowOption struct {
  44. Collection string `json:"collection,omitempty"`
  45. ReplicaPlacement *super_block.ReplicaPlacement `json:"replication,omitempty"`
  46. Ttl *needle.TTL `json:"ttl,omitempty"`
  47. DiskType types.DiskType `json:"disk,omitempty"`
  48. Preallocate int64 `json:"preallocate,omitempty"`
  49. DataCenter string `json:"dataCenter,omitempty"`
  50. Rack string `json:"rack,omitempty"`
  51. DataNode string `json:"dataNode,omitempty"`
  52. MemoryMapMaxSizeMb uint32 `json:"memoryMapMaxSizeMb,omitempty"`
  53. }
  54. type VolumeGrowth struct {
  55. accessLock sync.Mutex
  56. }
  57. func (o *VolumeGrowOption) String() string {
  58. blob, _ := json.Marshal(o)
  59. return string(blob)
  60. }
  61. func NewDefaultVolumeGrowth() *VolumeGrowth {
  62. return &VolumeGrowth{}
  63. }
  64. // one replication type may need rp.GetCopyCount() actual volumes
  65. // given copyCount, how many logical volumes to create
  66. func (vg *VolumeGrowth) findVolumeCount(copyCount int) (count uint32) {
  67. switch copyCount {
  68. case 1:
  69. count = VolumeGrowStrategy.Copy1Count
  70. case 2:
  71. count = VolumeGrowStrategy.Copy2Count
  72. case 3:
  73. count = VolumeGrowStrategy.Copy3Count
  74. default:
  75. count = VolumeGrowStrategy.CopyOtherCount
  76. }
  77. return
  78. }
  79. func (vg *VolumeGrowth) AutomaticGrowByType(option *VolumeGrowOption, grpcDialOption grpc.DialOption, topo *Topology, targetCount uint32) (result []*master_pb.VolumeLocation, err error) {
  80. if targetCount == 0 {
  81. targetCount = vg.findVolumeCount(option.ReplicaPlacement.GetCopyCount())
  82. }
  83. result, err = vg.GrowByCountAndType(grpcDialOption, targetCount, option, topo)
  84. if len(result) > 0 && len(result)%option.ReplicaPlacement.GetCopyCount() == 0 {
  85. return result, nil
  86. }
  87. return result, err
  88. }
  89. func (vg *VolumeGrowth) GrowByCountAndType(grpcDialOption grpc.DialOption, targetCount uint32, option *VolumeGrowOption, topo *Topology) (result []*master_pb.VolumeLocation, err error) {
  90. vg.accessLock.Lock()
  91. defer vg.accessLock.Unlock()
  92. for i := uint32(0); i < targetCount; i++ {
  93. if res, e := vg.findAndGrow(grpcDialOption, topo, option); e == nil {
  94. result = append(result, res...)
  95. } else {
  96. glog.V(0).Infof("create %d volume, created %d: %v", targetCount, len(result), e)
  97. return result, e
  98. }
  99. }
  100. return
  101. }
  102. func (vg *VolumeGrowth) findAndGrow(grpcDialOption grpc.DialOption, topo *Topology, option *VolumeGrowOption) (result []*master_pb.VolumeLocation, err error) {
  103. servers, e := vg.findEmptySlotsForOneVolume(topo, option)
  104. if e != nil {
  105. return nil, e
  106. }
  107. vid, raftErr := topo.NextVolumeId()
  108. if raftErr != nil {
  109. return nil, raftErr
  110. }
  111. if err = vg.grow(grpcDialOption, topo, vid, option, servers...); err == nil {
  112. for _, server := range servers {
  113. result = append(result, &master_pb.VolumeLocation{
  114. Url: server.Url(),
  115. PublicUrl: server.PublicUrl,
  116. DataCenter: server.GetDataCenterId(),
  117. GrpcPort: uint32(server.GrpcPort),
  118. NewVids: []uint32{uint32(vid)},
  119. })
  120. }
  121. }
  122. return
  123. }
  124. // 1. find the main data node
  125. // 1.1 collect all data nodes that have 1 slots
  126. // 2.2 collect all racks that have rp.SameRackCount+1
  127. // 2.2 collect all data centers that have DiffRackCount+rp.SameRackCount+1
  128. // 2. find rest data nodes
  129. func (vg *VolumeGrowth) findEmptySlotsForOneVolume(topo *Topology, option *VolumeGrowOption) (servers []*DataNode, err error) {
  130. //find main datacenter and other data centers
  131. rp := option.ReplicaPlacement
  132. mainDataCenter, otherDataCenters, dc_err := topo.PickNodesByWeight(rp.DiffDataCenterCount+1, option, func(node Node) error {
  133. if option.DataCenter != "" && node.IsDataCenter() && node.Id() != NodeId(option.DataCenter) {
  134. return fmt.Errorf("Not matching preferred data center:%s", option.DataCenter)
  135. }
  136. if len(node.Children()) < rp.DiffRackCount+1 {
  137. return fmt.Errorf("Only has %d racks, not enough for %d.", len(node.Children()), rp.DiffRackCount+1)
  138. }
  139. if node.AvailableSpaceFor(option) < int64(rp.DiffRackCount+rp.SameRackCount+1) {
  140. return fmt.Errorf("Free:%d < Expected:%d", node.AvailableSpaceFor(option), rp.DiffRackCount+rp.SameRackCount+1)
  141. }
  142. possibleRacksCount := 0
  143. for _, rack := range node.Children() {
  144. possibleDataNodesCount := 0
  145. for _, n := range rack.Children() {
  146. if n.AvailableSpaceFor(option) >= 1 {
  147. possibleDataNodesCount++
  148. }
  149. }
  150. if possibleDataNodesCount >= rp.SameRackCount+1 {
  151. possibleRacksCount++
  152. }
  153. }
  154. if possibleRacksCount < rp.DiffRackCount+1 {
  155. 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)
  156. }
  157. return nil
  158. })
  159. if dc_err != nil {
  160. return nil, dc_err
  161. }
  162. //find main rack and other racks
  163. mainRack, otherRacks, rackErr := mainDataCenter.(*DataCenter).PickNodesByWeight(rp.DiffRackCount+1, option, func(node Node) error {
  164. if option.Rack != "" && node.IsRack() && node.Id() != NodeId(option.Rack) {
  165. return fmt.Errorf("Not matching preferred rack:%s", option.Rack)
  166. }
  167. if node.AvailableSpaceFor(option) < int64(rp.SameRackCount+1) {
  168. return fmt.Errorf("Free:%d < Expected:%d", node.AvailableSpaceFor(option), rp.SameRackCount+1)
  169. }
  170. if len(node.Children()) < rp.SameRackCount+1 {
  171. // a bit faster way to test free racks
  172. return fmt.Errorf("Only has %d data nodes, not enough for %d.", len(node.Children()), rp.SameRackCount+1)
  173. }
  174. possibleDataNodesCount := 0
  175. for _, n := range node.Children() {
  176. if n.AvailableSpaceFor(option) >= 1 {
  177. possibleDataNodesCount++
  178. }
  179. }
  180. if possibleDataNodesCount < rp.SameRackCount+1 {
  181. return fmt.Errorf("Only has %d data nodes with a slot, not enough for %d.", possibleDataNodesCount, rp.SameRackCount+1)
  182. }
  183. return nil
  184. })
  185. if rackErr != nil {
  186. return nil, rackErr
  187. }
  188. //find main server and other servers
  189. mainServer, otherServers, serverErr := mainRack.(*Rack).PickNodesByWeight(rp.SameRackCount+1, option, func(node Node) error {
  190. if option.DataNode != "" && node.IsDataNode() && node.Id() != NodeId(option.DataNode) {
  191. return fmt.Errorf("Not matching preferred data node:%s", option.DataNode)
  192. }
  193. if node.AvailableSpaceFor(option) < 1 {
  194. return fmt.Errorf("Free:%d < Expected:%d", node.AvailableSpaceFor(option), 1)
  195. }
  196. return nil
  197. })
  198. if serverErr != nil {
  199. return nil, serverErr
  200. }
  201. servers = append(servers, mainServer.(*DataNode))
  202. for _, server := range otherServers {
  203. servers = append(servers, server.(*DataNode))
  204. }
  205. for _, rack := range otherRacks {
  206. r := rand.Int63n(rack.AvailableSpaceFor(option))
  207. if server, e := rack.ReserveOneVolume(r, option); e == nil {
  208. servers = append(servers, server)
  209. } else {
  210. return servers, e
  211. }
  212. }
  213. for _, datacenter := range otherDataCenters {
  214. r := rand.Int63n(datacenter.AvailableSpaceFor(option))
  215. if server, e := datacenter.ReserveOneVolume(r, option); e == nil {
  216. servers = append(servers, server)
  217. } else {
  218. return servers, e
  219. }
  220. }
  221. return
  222. }
  223. func (vg *VolumeGrowth) grow(grpcDialOption grpc.DialOption, topo *Topology, vid needle.VolumeId, option *VolumeGrowOption, servers ...*DataNode) (growErr error) {
  224. var createdVolumes []storage.VolumeInfo
  225. for _, server := range servers {
  226. if err := AllocateVolume(server, grpcDialOption, vid, option); err == nil {
  227. createdVolumes = append(createdVolumes, storage.VolumeInfo{
  228. Id: vid,
  229. Size: 0,
  230. Collection: option.Collection,
  231. ReplicaPlacement: option.ReplicaPlacement,
  232. Ttl: option.Ttl,
  233. Version: needle.CurrentVersion,
  234. DiskType: option.DiskType.String(),
  235. ModifiedAtSecond: time.Now().Unix(),
  236. })
  237. glog.V(0).Infof("Created Volume %d on %s", vid, server.NodeImpl.String())
  238. } else {
  239. glog.Warningf("Failed to assign volume %d on %s: %v", vid, server.NodeImpl.String(), err)
  240. growErr = fmt.Errorf("failed to assign volume %d on %s: %v", vid, server.NodeImpl.String(), err)
  241. break
  242. }
  243. }
  244. if growErr == nil {
  245. for i, vi := range createdVolumes {
  246. server := servers[i]
  247. server.AddOrUpdateVolume(vi)
  248. topo.RegisterVolumeLayout(vi, server)
  249. glog.V(0).Infof("Registered Volume %d on %s", vid, server.NodeImpl.String())
  250. }
  251. } else {
  252. // cleaning up created volume replicas
  253. for i, vi := range createdVolumes {
  254. server := servers[i]
  255. if err := DeleteVolume(server, grpcDialOption, vi.Id); err != nil {
  256. glog.Warningf("Failed to clean up volume %d on %s", vid, server.NodeImpl.String())
  257. }
  258. }
  259. }
  260. return growErr
  261. }