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.

176 lines
6.1 KiB

3 years ago
6 years ago
5 years ago
6 years ago
3 years ago
6 years ago
6 years ago
6 years ago
4 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/pb"
  6. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  7. "math/rand"
  8. "net/http"
  9. "strconv"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/operation"
  12. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  13. "github.com/chrislusf/seaweedfs/weed/storage/backend/memory_map"
  14. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  15. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  16. "github.com/chrislusf/seaweedfs/weed/storage/types"
  17. "github.com/chrislusf/seaweedfs/weed/topology"
  18. "github.com/chrislusf/seaweedfs/weed/util"
  19. )
  20. func (ms *MasterServer) collectionDeleteHandler(w http.ResponseWriter, r *http.Request) {
  21. collectionName := r.FormValue("collection")
  22. collection, ok := ms.Topo.FindCollection(collectionName)
  23. if !ok {
  24. writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("collection %s does not exist", collectionName))
  25. return
  26. }
  27. for _, server := range collection.ListVolumeServers() {
  28. err := operation.WithVolumeServerClient(false, server.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  29. _, deleteErr := client.DeleteCollection(context.Background(), &volume_server_pb.DeleteCollectionRequest{
  30. Collection: collection.Name,
  31. })
  32. return deleteErr
  33. })
  34. if err != nil {
  35. writeJsonError(w, r, http.StatusInternalServerError, err)
  36. return
  37. }
  38. }
  39. ms.Topo.DeleteCollection(collectionName)
  40. w.WriteHeader(http.StatusNoContent)
  41. return
  42. }
  43. func (ms *MasterServer) dirStatusHandler(w http.ResponseWriter, r *http.Request) {
  44. m := make(map[string]interface{})
  45. m["Version"] = util.Version()
  46. m["Topology"] = ms.Topo.ToMap()
  47. writeJsonQuiet(w, r, http.StatusOK, m)
  48. }
  49. func (ms *MasterServer) volumeVacuumHandler(w http.ResponseWriter, r *http.Request) {
  50. gcString := r.FormValue("garbageThreshold")
  51. gcThreshold := ms.option.GarbageThreshold
  52. if gcString != "" {
  53. var err error
  54. gcThreshold, err = strconv.ParseFloat(gcString, 32)
  55. if err != nil {
  56. glog.V(0).Infof("garbageThreshold %s is not a valid float number: %v", gcString, err)
  57. writeJsonError(w, r, http.StatusNotAcceptable, fmt.Errorf("garbageThreshold %s is not a valid float number", gcString))
  58. return
  59. }
  60. }
  61. // glog.Infoln("garbageThreshold =", gcThreshold)
  62. ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, 0, "", ms.preallocateSize)
  63. ms.dirStatusHandler(w, r)
  64. }
  65. func (ms *MasterServer) volumeGrowHandler(w http.ResponseWriter, r *http.Request) {
  66. count := 0
  67. option, err := ms.getVolumeGrowOption(r)
  68. if err != nil {
  69. writeJsonError(w, r, http.StatusNotAcceptable, err)
  70. return
  71. }
  72. glog.V(0).Infof("volumeGrowHandler received %v from %v", option.String(), r.RemoteAddr)
  73. if count, err = strconv.Atoi(r.FormValue("count")); err == nil {
  74. if ms.Topo.AvailableSpaceFor(option) < int64(count*option.ReplicaPlacement.GetCopyCount()) {
  75. err = fmt.Errorf("only %d volumes left, not enough for %d", ms.Topo.AvailableSpaceFor(option), count*option.ReplicaPlacement.GetCopyCount())
  76. } else {
  77. var newVidLocations []*master_pb.VolumeLocation
  78. newVidLocations, err = ms.vg.GrowByCountAndType(ms.grpcDialOption, count, option, ms.Topo)
  79. count = len(newVidLocations)
  80. }
  81. } else {
  82. err = fmt.Errorf("can not parse parameter count %s", r.FormValue("count"))
  83. }
  84. if err != nil {
  85. writeJsonError(w, r, http.StatusNotAcceptable, err)
  86. } else {
  87. writeJsonQuiet(w, r, http.StatusOK, map[string]interface{}{"count": count})
  88. }
  89. }
  90. func (ms *MasterServer) volumeStatusHandler(w http.ResponseWriter, r *http.Request) {
  91. m := make(map[string]interface{})
  92. m["Version"] = util.Version()
  93. m["Volumes"] = ms.Topo.ToVolumeMap()
  94. writeJsonQuiet(w, r, http.StatusOK, m)
  95. }
  96. func (ms *MasterServer) redirectHandler(w http.ResponseWriter, r *http.Request) {
  97. vid, _, _, _, _ := parseURLPath(r.URL.Path)
  98. collection := r.FormValue("collection")
  99. location := ms.findVolumeLocation(collection, vid)
  100. if location.Error == "" {
  101. loc := location.Locations[rand.Intn(len(location.Locations))]
  102. var url string
  103. if r.URL.RawQuery != "" {
  104. url = util.NormalizeUrl(loc.PublicUrl) + r.URL.Path + "?" + r.URL.RawQuery
  105. } else {
  106. url = util.NormalizeUrl(loc.PublicUrl) + r.URL.Path
  107. }
  108. http.Redirect(w, r, url, http.StatusPermanentRedirect)
  109. } else {
  110. writeJsonError(w, r, http.StatusNotFound, fmt.Errorf("volume id %s not found: %s", vid, location.Error))
  111. }
  112. }
  113. func (ms *MasterServer) submitFromMasterServerHandler(w http.ResponseWriter, r *http.Request) {
  114. if ms.Topo.IsLeader() {
  115. submitForClientHandler(w, r, func() pb.ServerAddress { return ms.option.Master }, ms.grpcDialOption)
  116. } else {
  117. masterUrl, err := ms.Topo.Leader()
  118. if err != nil {
  119. writeJsonError(w, r, http.StatusInternalServerError, err)
  120. } else {
  121. submitForClientHandler(w, r, func() pb.ServerAddress { return masterUrl }, ms.grpcDialOption)
  122. }
  123. }
  124. }
  125. func (ms *MasterServer) getVolumeGrowOption(r *http.Request) (*topology.VolumeGrowOption, error) {
  126. replicationString := r.FormValue("replication")
  127. if replicationString == "" {
  128. replicationString = ms.option.DefaultReplicaPlacement
  129. }
  130. replicaPlacement, err := super_block.NewReplicaPlacementFromString(replicationString)
  131. if err != nil {
  132. return nil, err
  133. }
  134. ttl, err := needle.ReadTTL(r.FormValue("ttl"))
  135. if err != nil {
  136. return nil, err
  137. }
  138. memoryMapMaxSizeMb, err := memory_map.ReadMemoryMapMaxSizeMb(r.FormValue("memoryMapMaxSizeMb"))
  139. if err != nil {
  140. return nil, err
  141. }
  142. diskType := types.ToDiskType(r.FormValue("disk"))
  143. preallocate := ms.preallocateSize
  144. if r.FormValue("preallocate") != "" {
  145. preallocate, err = strconv.ParseInt(r.FormValue("preallocate"), 10, 64)
  146. if err != nil {
  147. return nil, fmt.Errorf("Failed to parse int64 preallocate = %s: %v", r.FormValue("preallocate"), err)
  148. }
  149. }
  150. volumeGrowOption := &topology.VolumeGrowOption{
  151. Collection: r.FormValue("collection"),
  152. ReplicaPlacement: replicaPlacement,
  153. Ttl: ttl,
  154. DiskType: diskType,
  155. Preallocate: preallocate,
  156. DataCenter: r.FormValue("dataCenter"),
  157. Rack: r.FormValue("rack"),
  158. DataNode: r.FormValue("dataNode"),
  159. MemoryMapMaxSizeMb: memoryMapMaxSizeMb,
  160. }
  161. return volumeGrowOption, nil
  162. }