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.

267 lines
6.9 KiB

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