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.

189 lines
5.6 KiB

3 years ago
8 months ago
  1. package weed_server
  2. import (
  3. "fmt"
  4. "github.com/seaweedfs/seaweedfs/weed/glog"
  5. "net/http"
  6. "strconv"
  7. "strings"
  8. "time"
  9. "github.com/seaweedfs/seaweedfs/weed/operation"
  10. "github.com/seaweedfs/seaweedfs/weed/security"
  11. "github.com/seaweedfs/seaweedfs/weed/stats"
  12. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  13. "github.com/seaweedfs/seaweedfs/weed/topology"
  14. )
  15. func (ms *MasterServer) lookupVolumeId(vids []string, collection string) (volumeLocations map[string]operation.LookupResult) {
  16. volumeLocations = make(map[string]operation.LookupResult)
  17. for _, vid := range vids {
  18. commaSep := strings.Index(vid, ",")
  19. if commaSep > 0 {
  20. vid = vid[0:commaSep]
  21. }
  22. if _, ok := volumeLocations[vid]; ok {
  23. continue
  24. }
  25. volumeLocations[vid] = ms.findVolumeLocation(collection, vid)
  26. }
  27. return
  28. }
  29. // If "fileId" is provided, this returns the fileId location and a JWT to update or delete the file.
  30. // If "volumeId" is provided, this only returns the volumeId location
  31. func (ms *MasterServer) dirLookupHandler(w http.ResponseWriter, r *http.Request) {
  32. vid := r.FormValue("volumeId")
  33. if vid != "" {
  34. // backward compatible
  35. commaSep := strings.Index(vid, ",")
  36. if commaSep > 0 {
  37. vid = vid[0:commaSep]
  38. }
  39. }
  40. fileId := r.FormValue("fileId")
  41. if fileId != "" {
  42. commaSep := strings.Index(fileId, ",")
  43. if commaSep > 0 {
  44. vid = fileId[0:commaSep]
  45. }
  46. }
  47. collection := r.FormValue("collection") // optional, but can be faster if too many collections
  48. location := ms.findVolumeLocation(collection, vid)
  49. httpStatus := http.StatusOK
  50. if location.Error != "" || location.Locations == nil {
  51. httpStatus = http.StatusNotFound
  52. } else {
  53. forRead := r.FormValue("read")
  54. isRead := forRead == "yes"
  55. ms.maybeAddJwtAuthorization(w, fileId, !isRead)
  56. }
  57. writeJsonQuiet(w, r, httpStatus, location)
  58. }
  59. // findVolumeLocation finds the volume location from master topo if it is leader,
  60. // or from master client if not leader
  61. func (ms *MasterServer) findVolumeLocation(collection, vid string) operation.LookupResult {
  62. var locations []operation.Location
  63. var err error
  64. if ms.Topo.IsLeader() {
  65. volumeId, newVolumeIdErr := needle.NewVolumeId(vid)
  66. if newVolumeIdErr != nil {
  67. err = fmt.Errorf("Unknown volume id %s", vid)
  68. } else {
  69. machines := ms.Topo.Lookup(collection, volumeId)
  70. for _, loc := range machines {
  71. locations = append(locations, operation.Location{
  72. Url: loc.Url(), PublicUrl: loc.PublicUrl, DataCenter: loc.GetDataCenterId(),
  73. })
  74. }
  75. }
  76. } else {
  77. machines, getVidLocationsErr := ms.MasterClient.GetVidLocations(vid)
  78. for _, loc := range machines {
  79. locations = append(locations, operation.Location{
  80. Url: loc.Url, PublicUrl: loc.PublicUrl, DataCenter: loc.DataCenter,
  81. })
  82. }
  83. err = getVidLocationsErr
  84. }
  85. if len(locations) == 0 && err == nil {
  86. err = fmt.Errorf("volume id %s not found", vid)
  87. }
  88. ret := operation.LookupResult{
  89. VolumeOrFileId: vid,
  90. Locations: locations,
  91. }
  92. if err != nil {
  93. ret.Error = err.Error()
  94. }
  95. return ret
  96. }
  97. func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) {
  98. stats.AssignRequest()
  99. requestedCount, e := strconv.ParseUint(r.FormValue("count"), 10, 64)
  100. if e != nil || requestedCount == 0 {
  101. requestedCount = 1
  102. }
  103. writableVolumeCount, e := strconv.ParseUint(r.FormValue("writableVolumeCount"), 10, 32)
  104. if e != nil {
  105. writableVolumeCount = 0
  106. }
  107. option, err := ms.getVolumeGrowOption(r)
  108. if err != nil {
  109. writeJsonQuiet(w, r, http.StatusNotAcceptable, operation.AssignResult{Error: err.Error()})
  110. return
  111. }
  112. vl := ms.Topo.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType)
  113. var (
  114. lastErr error
  115. maxTimeout = time.Second * 10
  116. startTime = time.Now()
  117. )
  118. if !ms.Topo.DataCenterExists(option.DataCenter) {
  119. writeJsonQuiet(w, r, http.StatusBadRequest, operation.AssignResult{
  120. Error: fmt.Sprintf("data center %v not found in topology", option.DataCenter),
  121. })
  122. return
  123. }
  124. for time.Now().Sub(startTime) < maxTimeout {
  125. fid, count, dnList, shouldGrow, err := ms.Topo.PickForWrite(requestedCount, option, vl)
  126. if shouldGrow && !vl.HasGrowRequest() {
  127. // if picked volume is almost full, trigger a volume-grow request
  128. glog.V(0).Infof("dirAssign volume growth %v from %v", option.String(), r.RemoteAddr)
  129. if ms.Topo.AvailableSpaceFor(option) <= 0 {
  130. writeJsonQuiet(w, r, http.StatusNotFound, operation.AssignResult{Error: "No free volumes left for " + option.String()})
  131. return
  132. }
  133. vl.AddGrowRequest()
  134. ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{
  135. Option: option,
  136. Count: uint32(writableVolumeCount),
  137. }
  138. }
  139. if err != nil {
  140. // glog.Warningf("PickForWrite %+v: %v", req, err)
  141. lastErr = err
  142. time.Sleep(200 * time.Millisecond)
  143. continue
  144. } else {
  145. ms.maybeAddJwtAuthorization(w, fid, true)
  146. dn := dnList.Head()
  147. if dn == nil {
  148. continue
  149. }
  150. writeJsonQuiet(w, r, http.StatusOK, operation.AssignResult{Fid: fid, Url: dn.Url(), PublicUrl: dn.PublicUrl, Count: count})
  151. return
  152. }
  153. }
  154. if lastErr != nil {
  155. writeJsonQuiet(w, r, http.StatusNotAcceptable, operation.AssignResult{Error: lastErr.Error()})
  156. } else {
  157. writeJsonQuiet(w, r, http.StatusRequestTimeout, operation.AssignResult{Error: "request timeout"})
  158. }
  159. }
  160. func (ms *MasterServer) maybeAddJwtAuthorization(w http.ResponseWriter, fileId string, isWrite bool) {
  161. if fileId == "" {
  162. return
  163. }
  164. var encodedJwt security.EncodedJwt
  165. if isWrite {
  166. encodedJwt = security.GenJwtForVolumeServer(ms.guard.SigningKey, ms.guard.ExpiresAfterSec, fileId)
  167. } else {
  168. encodedJwt = security.GenJwtForVolumeServer(ms.guard.ReadSigningKey, ms.guard.ReadExpiresAfterSec, fileId)
  169. }
  170. if encodedJwt == "" {
  171. return
  172. }
  173. w.Header().Set("Authorization", "BEARER "+string(encodedJwt))
  174. }