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.

252 lines
6.3 KiB

6 years ago
6 years ago
4 years ago
6 years ago
6 years ago
6 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/raft"
  6. "reflect"
  7. "sync"
  8. "time"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  11. "github.com/chrislusf/seaweedfs/weed/security"
  12. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  13. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  14. "github.com/chrislusf/seaweedfs/weed/storage/types"
  15. "github.com/chrislusf/seaweedfs/weed/topology"
  16. )
  17. func (ms *MasterServer) ProcessGrowRequest() {
  18. go func() {
  19. filter := sync.Map{}
  20. for {
  21. req, ok := <-ms.vgCh
  22. if !ok {
  23. break
  24. }
  25. if !ms.Topo.IsLeader() {
  26. //discard buffered requests
  27. time.Sleep(time.Second * 1)
  28. continue
  29. }
  30. // filter out identical requests being processed
  31. found := false
  32. filter.Range(func(k, v interface{}) bool {
  33. if reflect.DeepEqual(k, req) {
  34. found = true
  35. }
  36. return !found
  37. })
  38. // not atomic but it's okay
  39. if !found && ms.shouldVolumeGrow(req.Option) {
  40. filter.Store(req, nil)
  41. // we have lock called inside vg
  42. go func() {
  43. glog.V(1).Infoln("starting automatic volume grow")
  44. start := time.Now()
  45. _, err := ms.vg.AutomaticGrowByType(req.Option, ms.grpcDialOption, ms.Topo, req.Count)
  46. glog.V(1).Infoln("finished automatic volume grow, cost ", time.Now().Sub(start))
  47. if req.ErrCh != nil {
  48. req.ErrCh <- err
  49. close(req.ErrCh)
  50. }
  51. filter.Delete(req)
  52. }()
  53. } else {
  54. glog.V(4).Infoln("discard volume grow request")
  55. }
  56. }
  57. }()
  58. }
  59. func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) {
  60. if !ms.Topo.IsLeader() {
  61. return nil, raft.NotLeaderError
  62. }
  63. resp := &master_pb.LookupVolumeResponse{}
  64. volumeLocations := ms.lookupVolumeId(req.VolumeIds, req.Collection)
  65. for _, result := range volumeLocations {
  66. var locations []*master_pb.Location
  67. for _, loc := range result.Locations {
  68. locations = append(locations, &master_pb.Location{
  69. Url: loc.Url,
  70. PublicUrl: loc.PublicUrl,
  71. })
  72. }
  73. resp.VolumeIdLocations = append(resp.VolumeIdLocations, &master_pb.LookupVolumeResponse_VolumeIdLocation{
  74. VolumeId: result.VolumeId,
  75. Locations: locations,
  76. Error: result.Error,
  77. })
  78. }
  79. return resp, nil
  80. }
  81. func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest) (*master_pb.AssignResponse, error) {
  82. if !ms.Topo.IsLeader() {
  83. return nil, raft.NotLeaderError
  84. }
  85. if req.Count == 0 {
  86. req.Count = 1
  87. }
  88. if req.Replication == "" {
  89. req.Replication = ms.option.DefaultReplicaPlacement
  90. }
  91. replicaPlacement, err := super_block.NewReplicaPlacementFromString(req.Replication)
  92. if err != nil {
  93. return nil, err
  94. }
  95. ttl, err := needle.ReadTTL(req.Ttl)
  96. if err != nil {
  97. return nil, err
  98. }
  99. diskType := types.ToDiskType(req.DiskType)
  100. option := &topology.VolumeGrowOption{
  101. Collection: req.Collection,
  102. ReplicaPlacement: replicaPlacement,
  103. Ttl: ttl,
  104. DiskType: diskType,
  105. Preallocate: ms.preallocateSize,
  106. DataCenter: req.DataCenter,
  107. Rack: req.Rack,
  108. DataNode: req.DataNode,
  109. MemoryMapMaxSizeMb: req.MemoryMapMaxSizeMb,
  110. }
  111. if ms.shouldVolumeGrow(option) {
  112. if ms.Topo.AvailableSpaceFor(option) <= 0 {
  113. return nil, fmt.Errorf("no free volumes left for " + option.String())
  114. }
  115. ms.vgCh <- &topology.VolumeGrowRequest{
  116. Option: option,
  117. Count: int(req.WritableVolumeCount),
  118. }
  119. }
  120. var (
  121. lastErr error
  122. maxTimeout = time.Second * 10
  123. startTime = time.Now()
  124. )
  125. for time.Now().Sub(startTime) < maxTimeout {
  126. fid, count, dn, err := ms.Topo.PickForWrite(req.Count, option)
  127. if err == nil {
  128. return &master_pb.AssignResponse{
  129. Fid: fid,
  130. Url: dn.Url(),
  131. PublicUrl: dn.PublicUrl,
  132. Count: count,
  133. Auth: string(security.GenJwt(ms.guard.SigningKey, ms.guard.ExpiresAfterSec, fid)),
  134. }, nil
  135. }
  136. //glog.V(4).Infoln("waiting for volume growing...")
  137. lastErr = err
  138. time.Sleep(200 * time.Millisecond)
  139. }
  140. return nil, lastErr
  141. }
  142. func (ms *MasterServer) Statistics(ctx context.Context, req *master_pb.StatisticsRequest) (*master_pb.StatisticsResponse, error) {
  143. if !ms.Topo.IsLeader() {
  144. return nil, raft.NotLeaderError
  145. }
  146. if req.Replication == "" {
  147. req.Replication = ms.option.DefaultReplicaPlacement
  148. }
  149. replicaPlacement, err := super_block.NewReplicaPlacementFromString(req.Replication)
  150. if err != nil {
  151. return nil, err
  152. }
  153. ttl, err := needle.ReadTTL(req.Ttl)
  154. if err != nil {
  155. return nil, err
  156. }
  157. volumeLayout := ms.Topo.GetVolumeLayout(req.Collection, replicaPlacement, ttl, types.ToDiskType(req.DiskType))
  158. stats := volumeLayout.Stats()
  159. resp := &master_pb.StatisticsResponse{
  160. TotalSize: stats.TotalSize,
  161. UsedSize: stats.UsedSize,
  162. FileCount: stats.FileCount,
  163. }
  164. return resp, nil
  165. }
  166. func (ms *MasterServer) VolumeList(ctx context.Context, req *master_pb.VolumeListRequest) (*master_pb.VolumeListResponse, error) {
  167. if !ms.Topo.IsLeader() {
  168. return nil, raft.NotLeaderError
  169. }
  170. resp := &master_pb.VolumeListResponse{
  171. TopologyInfo: ms.Topo.ToTopologyInfo(),
  172. VolumeSizeLimitMb: uint64(ms.option.VolumeSizeLimitMB),
  173. }
  174. return resp, nil
  175. }
  176. func (ms *MasterServer) LookupEcVolume(ctx context.Context, req *master_pb.LookupEcVolumeRequest) (*master_pb.LookupEcVolumeResponse, error) {
  177. if !ms.Topo.IsLeader() {
  178. return nil, raft.NotLeaderError
  179. }
  180. resp := &master_pb.LookupEcVolumeResponse{}
  181. ecLocations, found := ms.Topo.LookupEcShards(needle.VolumeId(req.VolumeId))
  182. if !found {
  183. return resp, fmt.Errorf("ec volume %d not found", req.VolumeId)
  184. }
  185. resp.VolumeId = req.VolumeId
  186. for shardId, shardLocations := range ecLocations.Locations {
  187. var locations []*master_pb.Location
  188. for _, dn := range shardLocations {
  189. locations = append(locations, &master_pb.Location{
  190. Url: string(dn.Id()),
  191. PublicUrl: dn.PublicUrl,
  192. })
  193. }
  194. resp.ShardIdLocations = append(resp.ShardIdLocations, &master_pb.LookupEcVolumeResponse_EcShardIdLocation{
  195. ShardId: uint32(shardId),
  196. Locations: locations,
  197. })
  198. }
  199. return resp, nil
  200. }
  201. func (ms *MasterServer) VacuumVolume(ctx context.Context, req *master_pb.VacuumVolumeRequest) (*master_pb.VacuumVolumeResponse, error) {
  202. if !ms.Topo.IsLeader() {
  203. return nil, raft.NotLeaderError
  204. }
  205. resp := &master_pb.VacuumVolumeResponse{}
  206. ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.preallocateSize)
  207. return resp, nil
  208. }