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.

362 lines
10 KiB

9 years ago
9 years ago
6 years ago
6 years ago
5 years ago
5 years ago
5 years ago
6 years ago
4 years ago
2 years ago
2 years ago
6 years ago
2 years ago
4 years ago
5 years ago
4 years ago
5 years ago
9 years ago
5 years ago
5 years ago
5 years ago
  1. package weed_server
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  8. "github.com/seaweedfs/seaweedfs/weed/util/mem"
  9. "io"
  10. "mime"
  11. "net/http"
  12. "net/url"
  13. "path/filepath"
  14. "strconv"
  15. "strings"
  16. "sync/atomic"
  17. "time"
  18. "github.com/seaweedfs/seaweedfs/weed/glog"
  19. "github.com/seaweedfs/seaweedfs/weed/images"
  20. "github.com/seaweedfs/seaweedfs/weed/operation"
  21. "github.com/seaweedfs/seaweedfs/weed/storage"
  22. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  23. "github.com/seaweedfs/seaweedfs/weed/util"
  24. )
  25. var fileNameEscaper = strings.NewReplacer(`\`, `\\`, `"`, `\"`)
  26. func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) {
  27. n := new(needle.Needle)
  28. vid, fid, filename, ext, _ := parseURLPath(r.URL.Path)
  29. if !vs.maybeCheckJwtAuthorization(r, vid, fid, false) {
  30. writeJsonError(w, r, http.StatusUnauthorized, errors.New("wrong jwt"))
  31. return
  32. }
  33. volumeId, err := needle.NewVolumeId(vid)
  34. if err != nil {
  35. glog.V(2).Infof("parsing vid %s: %v", r.URL.Path, err)
  36. w.WriteHeader(http.StatusBadRequest)
  37. return
  38. }
  39. err = n.ParsePath(fid)
  40. if err != nil {
  41. glog.V(2).Infof("parsing fid %s: %v", r.URL.Path, err)
  42. w.WriteHeader(http.StatusBadRequest)
  43. return
  44. }
  45. // glog.V(4).Infoln("volume", volumeId, "reading", n)
  46. hasVolume := vs.store.HasVolume(volumeId)
  47. _, hasEcVolume := vs.store.FindEcVolume(volumeId)
  48. if !hasVolume && !hasEcVolume {
  49. if vs.ReadMode == "local" {
  50. glog.V(0).Infoln("volume is not local:", err, r.URL.Path)
  51. w.WriteHeader(http.StatusNotFound)
  52. return
  53. }
  54. lookupResult, err := operation.LookupVolumeId(vs.GetMaster, vs.grpcDialOption, volumeId.String())
  55. glog.V(2).Infoln("volume", volumeId, "found on", lookupResult, "error", err)
  56. if err != nil || len(lookupResult.Locations) <= 0 {
  57. glog.V(0).Infoln("lookup error:", err, r.URL.Path)
  58. w.WriteHeader(http.StatusNotFound)
  59. return
  60. }
  61. if vs.ReadMode == "proxy" {
  62. // proxy client request to target server
  63. u, _ := url.Parse(util.NormalizeUrl(lookupResult.Locations[0].Url))
  64. r.URL.Host = u.Host
  65. r.URL.Scheme = u.Scheme
  66. request, err := http.NewRequest("GET", r.URL.String(), nil)
  67. if err != nil {
  68. glog.V(0).Infof("failed to instance http request of url %s: %v", r.URL.String(), err)
  69. w.WriteHeader(http.StatusInternalServerError)
  70. return
  71. }
  72. for k, vv := range r.Header {
  73. for _, v := range vv {
  74. request.Header.Add(k, v)
  75. }
  76. }
  77. response, err := client.Do(request)
  78. if err != nil {
  79. glog.V(0).Infof("request remote url %s: %v", r.URL.String(), err)
  80. w.WriteHeader(http.StatusInternalServerError)
  81. return
  82. }
  83. defer util.CloseResponse(response)
  84. // proxy target response to client
  85. for k, vv := range response.Header {
  86. for _, v := range vv {
  87. w.Header().Add(k, v)
  88. }
  89. }
  90. w.WriteHeader(response.StatusCode)
  91. buf := mem.Allocate(128 * 1024)
  92. defer mem.Free(buf)
  93. io.CopyBuffer(w, response.Body, buf)
  94. return
  95. } else {
  96. // redirect
  97. u, _ := url.Parse(util.NormalizeUrl(lookupResult.Locations[0].PublicUrl))
  98. u.Path = fmt.Sprintf("%s/%s,%s", u.Path, vid, fid)
  99. arg := url.Values{}
  100. if c := r.FormValue("collection"); c != "" {
  101. arg.Set("collection", c)
  102. }
  103. u.RawQuery = arg.Encode()
  104. http.Redirect(w, r, u.String(), http.StatusMovedPermanently)
  105. return
  106. }
  107. }
  108. cookie := n.Cookie
  109. readOption := &storage.ReadOption{
  110. ReadDeleted: r.FormValue("readDeleted") == "true",
  111. }
  112. var count int
  113. var memoryCost types.Size
  114. readOption.AttemptMetaOnly, readOption.MustMetaOnly = shouldAttemptStreamWrite(hasVolume, ext, r)
  115. onReadSizeFn := func(size types.Size) {
  116. memoryCost = size
  117. atomic.AddInt64(&vs.inFlightDownloadDataSize, int64(memoryCost))
  118. }
  119. if hasVolume {
  120. count, err = vs.store.ReadVolumeNeedle(volumeId, n, readOption, onReadSizeFn)
  121. } else if hasEcVolume {
  122. count, err = vs.store.ReadEcShardNeedle(volumeId, n, onReadSizeFn)
  123. }
  124. defer func() {
  125. atomic.AddInt64(&vs.inFlightDownloadDataSize, -int64(memoryCost))
  126. vs.inFlightDownloadDataLimitCond.Signal()
  127. }()
  128. if err != nil && err != storage.ErrorDeleted && hasVolume {
  129. glog.V(4).Infof("read needle: %v", err)
  130. // start to fix it from other replicas, if not deleted and hasVolume and is not a replicated request
  131. }
  132. // glog.V(4).Infoln("read bytes", count, "error", err)
  133. if err != nil || count < 0 {
  134. glog.V(3).Infof("read %s isNormalVolume %v error: %v", r.URL.Path, hasVolume, err)
  135. if err == storage.ErrorNotFound || err == storage.ErrorDeleted {
  136. w.WriteHeader(http.StatusNotFound)
  137. } else {
  138. w.WriteHeader(http.StatusInternalServerError)
  139. }
  140. return
  141. }
  142. if n.Cookie != cookie {
  143. glog.V(0).Infof("request %s with cookie:%x expected:%x from %s agent %s", r.URL.Path, cookie, n.Cookie, r.RemoteAddr, r.UserAgent())
  144. w.WriteHeader(http.StatusNotFound)
  145. return
  146. }
  147. if n.LastModified != 0 {
  148. w.Header().Set("Last-Modified", time.Unix(int64(n.LastModified), 0).UTC().Format(http.TimeFormat))
  149. if r.Header.Get("If-Modified-Since") != "" {
  150. if t, parseError := time.Parse(http.TimeFormat, r.Header.Get("If-Modified-Since")); parseError == nil {
  151. if t.Unix() >= int64(n.LastModified) {
  152. w.WriteHeader(http.StatusNotModified)
  153. return
  154. }
  155. }
  156. }
  157. }
  158. if inm := r.Header.Get("If-None-Match"); inm == "\""+n.Etag()+"\"" {
  159. w.WriteHeader(http.StatusNotModified)
  160. return
  161. }
  162. setEtag(w, n.Etag())
  163. if n.HasPairs() {
  164. pairMap := make(map[string]string)
  165. err = json.Unmarshal(n.Pairs, &pairMap)
  166. if err != nil {
  167. glog.V(0).Infoln("Unmarshal pairs error:", err)
  168. }
  169. for k, v := range pairMap {
  170. w.Header().Set(k, v)
  171. }
  172. }
  173. if vs.tryHandleChunkedFile(n, filename, ext, w, r) {
  174. return
  175. }
  176. if n.NameSize > 0 && filename == "" {
  177. filename = string(n.Name)
  178. if ext == "" {
  179. ext = filepath.Ext(filename)
  180. }
  181. }
  182. mtype := ""
  183. if n.MimeSize > 0 {
  184. mt := string(n.Mime)
  185. if !strings.HasPrefix(mt, "application/octet-stream") {
  186. mtype = mt
  187. }
  188. }
  189. if n.IsCompressed() {
  190. if _, _, _, shouldResize := shouldResizeImages(ext, r); shouldResize {
  191. if n.Data, err = util.DecompressData(n.Data); err != nil {
  192. glog.V(0).Infoln("ungzip error:", err, r.URL.Path)
  193. }
  194. // } else if strings.Contains(r.Header.Get("Accept-Encoding"), "zstd") && util.IsZstdContent(n.Data) {
  195. // w.Header().Set("Content-Encoding", "zstd")
  196. } else if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") && util.IsGzippedContent(n.Data) {
  197. w.Header().Set("Content-Encoding", "gzip")
  198. } else {
  199. if n.Data, err = util.DecompressData(n.Data); err != nil {
  200. glog.V(0).Infoln("uncompress error:", err, r.URL.Path)
  201. }
  202. }
  203. }
  204. if !readOption.IsMetaOnly {
  205. rs := conditionallyResizeImages(bytes.NewReader(n.Data), ext, r)
  206. if e := writeResponseContent(filename, mtype, rs, w, r); e != nil {
  207. glog.V(2).Infoln("response write error:", e)
  208. }
  209. } else {
  210. vs.streamWriteResponseContent(filename, mtype, volumeId, n, w, r, readOption)
  211. }
  212. }
  213. func shouldAttemptStreamWrite(hasLocalVolume bool, ext string, r *http.Request) (shouldAttempt bool, mustMetaOnly bool) {
  214. if !hasLocalVolume {
  215. return false, false
  216. }
  217. if len(ext) > 0 {
  218. ext = strings.ToLower(ext)
  219. }
  220. if r.Method == "HEAD" {
  221. return true, true
  222. }
  223. _, _, _, shouldResize := shouldResizeImages(ext, r)
  224. if shouldResize {
  225. return false, false
  226. }
  227. return true, false
  228. }
  229. func (vs *VolumeServer) tryHandleChunkedFile(n *needle.Needle, fileName string, ext string, w http.ResponseWriter, r *http.Request) (processed bool) {
  230. if !n.IsChunkedManifest() || r.URL.Query().Get("cm") == "false" {
  231. return false
  232. }
  233. chunkManifest, e := operation.LoadChunkManifest(n.Data, n.IsCompressed())
  234. if e != nil {
  235. glog.V(0).Infof("load chunked manifest (%s) error: %v", r.URL.Path, e)
  236. return false
  237. }
  238. if fileName == "" && chunkManifest.Name != "" {
  239. fileName = chunkManifest.Name
  240. }
  241. if ext == "" {
  242. ext = filepath.Ext(fileName)
  243. }
  244. mType := ""
  245. if chunkManifest.Mime != "" {
  246. mt := chunkManifest.Mime
  247. if !strings.HasPrefix(mt, "application/octet-stream") {
  248. mType = mt
  249. }
  250. }
  251. w.Header().Set("X-File-Store", "chunked")
  252. chunkedFileReader := operation.NewChunkedFileReader(chunkManifest.Chunks, vs.GetMaster(), vs.grpcDialOption)
  253. defer chunkedFileReader.Close()
  254. rs := conditionallyResizeImages(chunkedFileReader, ext, r)
  255. if e := writeResponseContent(fileName, mType, rs, w, r); e != nil {
  256. glog.V(2).Infoln("response write error:", e)
  257. }
  258. return true
  259. }
  260. func conditionallyResizeImages(originalDataReaderSeeker io.ReadSeeker, ext string, r *http.Request) io.ReadSeeker {
  261. rs := originalDataReaderSeeker
  262. if len(ext) > 0 {
  263. ext = strings.ToLower(ext)
  264. }
  265. width, height, mode, shouldResize := shouldResizeImages(ext, r)
  266. if shouldResize {
  267. rs, _, _ = images.Resized(ext, originalDataReaderSeeker, width, height, mode)
  268. }
  269. return rs
  270. }
  271. func shouldResizeImages(ext string, r *http.Request) (width, height int, mode string, shouldResize bool) {
  272. if ext == ".png" || ext == ".jpg" || ext == ".jpeg" || ext == ".gif" || ext == ".webp" {
  273. if r.FormValue("width") != "" {
  274. width, _ = strconv.Atoi(r.FormValue("width"))
  275. }
  276. if r.FormValue("height") != "" {
  277. height, _ = strconv.Atoi(r.FormValue("height"))
  278. }
  279. }
  280. mode = r.FormValue("mode")
  281. shouldResize = width > 0 || height > 0
  282. return
  283. }
  284. func writeResponseContent(filename, mimeType string, rs io.ReadSeeker, w http.ResponseWriter, r *http.Request) error {
  285. totalSize, e := rs.Seek(0, 2)
  286. if mimeType == "" {
  287. if ext := filepath.Ext(filename); ext != "" {
  288. mimeType = mime.TypeByExtension(ext)
  289. }
  290. }
  291. if mimeType != "" {
  292. w.Header().Set("Content-Type", mimeType)
  293. }
  294. w.Header().Set("Accept-Ranges", "bytes")
  295. adjustPassthroughHeaders(w, r, filename)
  296. if r.Method == "HEAD" {
  297. w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
  298. return nil
  299. }
  300. return processRangeRequest(r, w, totalSize, mimeType, func(writer io.Writer, offset int64, size int64) error {
  301. if _, e = rs.Seek(offset, 0); e != nil {
  302. return e
  303. }
  304. _, e = io.CopyN(writer, rs, size)
  305. return e
  306. })
  307. }
  308. func (vs *VolumeServer) streamWriteResponseContent(filename string, mimeType string, volumeId needle.VolumeId, n *needle.Needle, w http.ResponseWriter, r *http.Request, readOption *storage.ReadOption) {
  309. totalSize := int64(n.DataSize)
  310. if mimeType == "" {
  311. if ext := filepath.Ext(filename); ext != "" {
  312. mimeType = mime.TypeByExtension(ext)
  313. }
  314. }
  315. if mimeType != "" {
  316. w.Header().Set("Content-Type", mimeType)
  317. }
  318. w.Header().Set("Accept-Ranges", "bytes")
  319. adjustPassthroughHeaders(w, r, filename)
  320. if r.Method == "HEAD" {
  321. w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
  322. return
  323. }
  324. processRangeRequest(r, w, totalSize, mimeType, func(writer io.Writer, offset int64, size int64) error {
  325. return vs.store.ReadVolumeNeedleDataInto(volumeId, n, readOption, writer, offset, size)
  326. })
  327. }