common.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414
  1. package weed_server
  2. import (
  3. "bufio"
  4. "bytes"
  5. "encoding/json"
  6. "errors"
  7. "fmt"
  8. "io"
  9. "io/fs"
  10. "mime/multipart"
  11. "net/http"
  12. "net/url"
  13. "path/filepath"
  14. "strconv"
  15. "strings"
  16. "sync"
  17. "time"
  18. "github.com/seaweedfs/seaweedfs/weed/filer"
  19. "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
  20. "google.golang.org/grpc"
  21. "github.com/seaweedfs/seaweedfs/weed/glog"
  22. "github.com/seaweedfs/seaweedfs/weed/operation"
  23. "github.com/seaweedfs/seaweedfs/weed/stats"
  24. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  25. "github.com/seaweedfs/seaweedfs/weed/util"
  26. "github.com/gorilla/mux"
  27. )
  28. var serverStats *stats.ServerStats
  29. var startTime = time.Now()
  30. var writePool = sync.Pool{New: func() interface{} {
  31. return bufio.NewWriterSize(nil, 128*1024)
  32. },
  33. }
  34. func init() {
  35. serverStats = stats.NewServerStats()
  36. go serverStats.Start()
  37. }
  38. // bodyAllowedForStatus is a copy of http.bodyAllowedForStatus non-exported function.
  39. func bodyAllowedForStatus(status int) bool {
  40. switch {
  41. case status >= 100 && status <= 199:
  42. return false
  43. case status == http.StatusNoContent:
  44. return false
  45. case status == http.StatusNotModified:
  46. return false
  47. }
  48. return true
  49. }
  50. func writeJson(w http.ResponseWriter, r *http.Request, httpStatus int, obj interface{}) (err error) {
  51. if !bodyAllowedForStatus(httpStatus) {
  52. return
  53. }
  54. var bytes []byte
  55. if obj != nil {
  56. if r.FormValue("pretty") != "" {
  57. bytes, err = json.MarshalIndent(obj, "", " ")
  58. } else {
  59. bytes, err = json.Marshal(obj)
  60. }
  61. }
  62. if err != nil {
  63. return
  64. }
  65. if httpStatus >= 400 {
  66. glog.V(0).Infof("response method:%s URL:%s with httpStatus:%d and JSON:%s",
  67. r.Method, r.URL.String(), httpStatus, string(bytes))
  68. }
  69. callback := r.FormValue("callback")
  70. if callback == "" {
  71. w.Header().Set("Content-Type", "application/json")
  72. w.WriteHeader(httpStatus)
  73. if httpStatus == http.StatusNotModified {
  74. return
  75. }
  76. _, err = w.Write(bytes)
  77. } else {
  78. w.Header().Set("Content-Type", "application/javascript")
  79. w.WriteHeader(httpStatus)
  80. if httpStatus == http.StatusNotModified {
  81. return
  82. }
  83. if _, err = w.Write([]uint8(callback)); err != nil {
  84. return
  85. }
  86. if _, err = w.Write([]uint8("(")); err != nil {
  87. return
  88. }
  89. fmt.Fprint(w, string(bytes))
  90. if _, err = w.Write([]uint8(")")); err != nil {
  91. return
  92. }
  93. }
  94. return
  95. }
  96. // wrapper for writeJson - just logs errors
  97. func writeJsonQuiet(w http.ResponseWriter, r *http.Request, httpStatus int, obj interface{}) {
  98. if err := writeJson(w, r, httpStatus, obj); err != nil {
  99. glog.V(0).Infof("error writing JSON status %s %d: %v", r.URL, httpStatus, err)
  100. glog.V(1).Infof("JSON content: %+v", obj)
  101. }
  102. }
  103. func writeJsonError(w http.ResponseWriter, r *http.Request, httpStatus int, err error) {
  104. m := make(map[string]interface{})
  105. m["error"] = err.Error()
  106. glog.V(1).Infof("error JSON response status %d: %s", httpStatus, m["error"])
  107. writeJsonQuiet(w, r, httpStatus, m)
  108. }
  109. func debug(params ...interface{}) {
  110. glog.V(4).Infoln(params...)
  111. }
  112. func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption) {
  113. m := make(map[string]interface{})
  114. if r.Method != "POST" {
  115. writeJsonError(w, r, http.StatusMethodNotAllowed, errors.New("Only submit via POST!"))
  116. return
  117. }
  118. debug("parsing upload file...")
  119. bytesBuffer := bufPool.Get().(*bytes.Buffer)
  120. defer bufPool.Put(bytesBuffer)
  121. pu, pe := needle.ParseUpload(r, 256*1024*1024, bytesBuffer)
  122. if pe != nil {
  123. writeJsonError(w, r, http.StatusBadRequest, pe)
  124. return
  125. }
  126. debug("assigning file id for", pu.FileName)
  127. r.ParseForm()
  128. count := uint64(1)
  129. if r.FormValue("count") != "" {
  130. count, pe = strconv.ParseUint(r.FormValue("count"), 10, 32)
  131. if pe != nil {
  132. writeJsonError(w, r, http.StatusBadRequest, pe)
  133. return
  134. }
  135. }
  136. ar := &operation.VolumeAssignRequest{
  137. Count: count,
  138. DataCenter: r.FormValue("dataCenter"),
  139. Rack: r.FormValue("rack"),
  140. Replication: r.FormValue("replication"),
  141. Collection: r.FormValue("collection"),
  142. Ttl: r.FormValue("ttl"),
  143. DiskType: r.FormValue("disk"),
  144. }
  145. assignResult, ae := operation.Assign(masterFn, grpcDialOption, ar)
  146. if ae != nil {
  147. writeJsonError(w, r, http.StatusInternalServerError, ae)
  148. return
  149. }
  150. url := "http://" + assignResult.Url + "/" + assignResult.Fid
  151. if pu.ModifiedTime != 0 {
  152. url = url + "?ts=" + strconv.FormatUint(pu.ModifiedTime, 10)
  153. }
  154. debug("upload file to store", url)
  155. uploadOption := &operation.UploadOption{
  156. UploadUrl: url,
  157. Filename: pu.FileName,
  158. Cipher: false,
  159. IsInputCompressed: pu.IsGzipped,
  160. MimeType: pu.MimeType,
  161. PairMap: pu.PairMap,
  162. Jwt: assignResult.Auth,
  163. }
  164. uploadResult, err := operation.UploadData(pu.Data, uploadOption)
  165. if err != nil {
  166. writeJsonError(w, r, http.StatusInternalServerError, err)
  167. return
  168. }
  169. m["fileName"] = pu.FileName
  170. m["fid"] = assignResult.Fid
  171. m["fileUrl"] = assignResult.PublicUrl + "/" + assignResult.Fid
  172. m["size"] = pu.OriginalDataSize
  173. m["eTag"] = uploadResult.ETag
  174. writeJsonQuiet(w, r, http.StatusCreated, m)
  175. return
  176. }
  177. func parseURLPath(path string) (vid, fid, filename, ext string, isVolumeIdOnly bool) {
  178. switch strings.Count(path, "/") {
  179. case 3:
  180. parts := strings.Split(path, "/")
  181. vid, fid, filename = parts[1], parts[2], parts[3]
  182. ext = filepath.Ext(filename)
  183. case 2:
  184. parts := strings.Split(path, "/")
  185. vid, fid = parts[1], parts[2]
  186. dotIndex := strings.LastIndex(fid, ".")
  187. if dotIndex > 0 {
  188. ext = fid[dotIndex:]
  189. fid = fid[0:dotIndex]
  190. }
  191. default:
  192. sepIndex := strings.LastIndex(path, "/")
  193. commaIndex := strings.LastIndex(path[sepIndex:], ",")
  194. if commaIndex <= 0 {
  195. vid, isVolumeIdOnly = path[sepIndex+1:], true
  196. return
  197. }
  198. dotIndex := strings.LastIndex(path[sepIndex:], ".")
  199. vid = path[sepIndex+1 : commaIndex]
  200. fid = path[commaIndex+1:]
  201. ext = ""
  202. if dotIndex > 0 {
  203. fid = path[commaIndex+1 : dotIndex]
  204. ext = path[dotIndex:]
  205. }
  206. }
  207. return
  208. }
  209. func statsHealthHandler(w http.ResponseWriter, r *http.Request) {
  210. m := make(map[string]interface{})
  211. m["Version"] = util.Version()
  212. writeJsonQuiet(w, r, http.StatusOK, m)
  213. }
  214. func statsCounterHandler(w http.ResponseWriter, r *http.Request) {
  215. m := make(map[string]interface{})
  216. m["Version"] = util.Version()
  217. m["Counters"] = serverStats
  218. writeJsonQuiet(w, r, http.StatusOK, m)
  219. }
  220. func statsMemoryHandler(w http.ResponseWriter, r *http.Request) {
  221. m := make(map[string]interface{})
  222. m["Version"] = util.Version()
  223. m["Memory"] = stats.MemStat()
  224. writeJsonQuiet(w, r, http.StatusOK, m)
  225. }
  226. var StaticFS fs.FS
  227. func handleStaticResources(defaultMux *http.ServeMux) {
  228. defaultMux.Handle("/favicon.ico", http.FileServer(http.FS(StaticFS)))
  229. defaultMux.Handle("/seaweedfsstatic/", http.StripPrefix("/seaweedfsstatic", http.FileServer(http.FS(StaticFS))))
  230. }
  231. func handleStaticResources2(r *mux.Router) {
  232. r.Handle("/favicon.ico", http.FileServer(http.FS(StaticFS)))
  233. r.PathPrefix("/seaweedfsstatic/").Handler(http.StripPrefix("/seaweedfsstatic", http.FileServer(http.FS(StaticFS))))
  234. }
  235. func adjustPassthroughHeaders(w http.ResponseWriter, r *http.Request, filename string) {
  236. for header, values := range r.Header {
  237. if normalizedHeader, ok := s3_constants.PassThroughHeaders[strings.ToLower(header)]; ok {
  238. w.Header()[normalizedHeader] = values
  239. }
  240. }
  241. adjustHeaderContentDisposition(w, r, filename)
  242. }
  243. func adjustHeaderContentDisposition(w http.ResponseWriter, r *http.Request, filename string) {
  244. if contentDisposition := w.Header().Get("Content-Disposition"); contentDisposition != "" {
  245. return
  246. }
  247. if filename != "" {
  248. filename = url.QueryEscape(filename)
  249. contentDisposition := "inline"
  250. if r.FormValue("dl") != "" {
  251. if dl, _ := strconv.ParseBool(r.FormValue("dl")); dl {
  252. contentDisposition = "attachment"
  253. }
  254. }
  255. w.Header().Set("Content-Disposition", contentDisposition+`; filename="`+fileNameEscaper.Replace(filename)+`"`)
  256. }
  257. }
  258. func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64, mimeType string, prepareWriteFn func(offset int64, size int64) (filer.DoStreamContent, error)) error {
  259. rangeReq := r.Header.Get("Range")
  260. bufferedWriter := writePool.Get().(*bufio.Writer)
  261. bufferedWriter.Reset(w)
  262. defer func() {
  263. bufferedWriter.Flush()
  264. writePool.Put(bufferedWriter)
  265. }()
  266. if rangeReq == "" {
  267. w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
  268. writeFn, err := prepareWriteFn(0, totalSize)
  269. if err != nil {
  270. glog.Errorf("processRangeRequest: %v", err)
  271. http.Error(w, err.Error(), http.StatusInternalServerError)
  272. return fmt.Errorf("processRangeRequest: %v", err)
  273. }
  274. if err = writeFn(bufferedWriter); err != nil {
  275. glog.Errorf("processRangeRequest: %v", err)
  276. http.Error(w, err.Error(), http.StatusInternalServerError)
  277. return fmt.Errorf("processRangeRequest: %v", err)
  278. }
  279. return nil
  280. }
  281. //the rest is dealing with partial content request
  282. //mostly copy from src/pkg/net/http/fs.go
  283. ranges, err := parseRange(rangeReq, totalSize)
  284. if err != nil {
  285. glog.Errorf("processRangeRequest headers: %+v err: %v", w.Header(), err)
  286. http.Error(w, err.Error(), http.StatusRequestedRangeNotSatisfiable)
  287. return fmt.Errorf("processRangeRequest header: %v", err)
  288. }
  289. if sumRangesSize(ranges) > totalSize {
  290. // The total number of bytes in all the ranges
  291. // is larger than the size of the file by
  292. // itself, so this is probably an attack, or a
  293. // dumb client. Ignore the range request.
  294. return nil
  295. }
  296. if len(ranges) == 0 {
  297. return nil
  298. }
  299. if len(ranges) == 1 {
  300. // RFC 2616, Section 14.16:
  301. // "When an HTTP message includes the content of a single
  302. // range (for example, a response to a request for a
  303. // single range, or to a request for a set of ranges
  304. // that overlap without any holes), this content is
  305. // transmitted with a Content-Range header, and a
  306. // Content-Length header showing the number of bytes
  307. // actually transferred.
  308. // ...
  309. // A response to a request for a single range MUST NOT
  310. // be sent using the multipart/byteranges media type."
  311. ra := ranges[0]
  312. w.Header().Set("Content-Length", strconv.FormatInt(ra.length, 10))
  313. w.Header().Set("Content-Range", ra.contentRange(totalSize))
  314. writeFn, err := prepareWriteFn(ra.start, ra.length)
  315. if err != nil {
  316. glog.Errorf("processRangeRequest range[0]: %+v err: %v", w.Header(), err)
  317. http.Error(w, err.Error(), http.StatusInternalServerError)
  318. return fmt.Errorf("processRangeRequest: %v", err)
  319. }
  320. w.WriteHeader(http.StatusPartialContent)
  321. err = writeFn(bufferedWriter)
  322. if err != nil {
  323. glog.Errorf("processRangeRequest range[0]: %+v err: %v", w.Header(), err)
  324. http.Error(w, err.Error(), http.StatusInternalServerError)
  325. return fmt.Errorf("processRangeRequest range[0]: %v", err)
  326. }
  327. return nil
  328. }
  329. // process multiple ranges
  330. writeFnByRange := make(map[int](func(writer io.Writer) error))
  331. for i, ra := range ranges {
  332. if ra.start > totalSize {
  333. http.Error(w, "Out of Range", http.StatusRequestedRangeNotSatisfiable)
  334. return fmt.Errorf("out of range: %v", err)
  335. }
  336. writeFn, err := prepareWriteFn(ra.start, ra.length)
  337. if err != nil {
  338. glog.Errorf("processRangeRequest range[%d] err: %v", i, err)
  339. http.Error(w, "Internal Error", http.StatusInternalServerError)
  340. return fmt.Errorf("processRangeRequest range[%d] err: %v", i, err)
  341. }
  342. writeFnByRange[i] = writeFn
  343. }
  344. sendSize := rangesMIMESize(ranges, mimeType, totalSize)
  345. pr, pw := io.Pipe()
  346. mw := multipart.NewWriter(pw)
  347. w.Header().Set("Content-Type", "multipart/byteranges; boundary="+mw.Boundary())
  348. sendContent := pr
  349. defer pr.Close() // cause writing goroutine to fail and exit if CopyN doesn't finish.
  350. go func() {
  351. for i, ra := range ranges {
  352. part, e := mw.CreatePart(ra.mimeHeader(mimeType, totalSize))
  353. if e != nil {
  354. pw.CloseWithError(e)
  355. return
  356. }
  357. writeFn := writeFnByRange[i]
  358. if writeFn == nil {
  359. pw.CloseWithError(e)
  360. return
  361. }
  362. if e = writeFn(part); e != nil {
  363. pw.CloseWithError(e)
  364. return
  365. }
  366. }
  367. mw.Close()
  368. pw.Close()
  369. }()
  370. if w.Header().Get("Content-Encoding") == "" {
  371. w.Header().Set("Content-Length", strconv.FormatInt(sendSize, 10))
  372. }
  373. w.WriteHeader(http.StatusPartialContent)
  374. if _, err := io.CopyN(bufferedWriter, sendContent, sendSize); err != nil {
  375. glog.Errorf("processRangeRequest err: %v", err)
  376. http.Error(w, "Internal Error", http.StatusInternalServerError)
  377. return fmt.Errorf("processRangeRequest err: %v", err)
  378. }
  379. return nil
  380. }