volume_server_handlers_read.go 9.2 KB

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