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.

266 lines
7.5 KiB

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