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.

173 lines
5.9 KiB

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