filer_server_handlers_read.go 9.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302
  1. package weed_server
  2. import (
  3. "bytes"
  4. "context"
  5. "encoding/base64"
  6. "encoding/hex"
  7. "errors"
  8. "fmt"
  9. "io"
  10. "math"
  11. "mime"
  12. "net/http"
  13. "path/filepath"
  14. "strconv"
  15. "strings"
  16. "time"
  17. "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
  18. "github.com/seaweedfs/seaweedfs/weed/security"
  19. "github.com/seaweedfs/seaweedfs/weed/util/mem"
  20. "github.com/seaweedfs/seaweedfs/weed/filer"
  21. "github.com/seaweedfs/seaweedfs/weed/glog"
  22. "github.com/seaweedfs/seaweedfs/weed/images"
  23. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  24. "github.com/seaweedfs/seaweedfs/weed/stats"
  25. "github.com/seaweedfs/seaweedfs/weed/util"
  26. )
  27. // Validates the preconditions. Returns true if GET/HEAD operation should not proceed.
  28. // Preconditions supported are:
  29. //
  30. // If-Modified-Since
  31. // If-Unmodified-Since
  32. // If-Match
  33. // If-None-Match
  34. func checkPreconditions(w http.ResponseWriter, r *http.Request, entry *filer.Entry) bool {
  35. etag := filer.ETagEntry(entry)
  36. /// When more than one conditional request header field is present in a
  37. /// request, the order in which the fields are evaluated becomes
  38. /// important. In practice, the fields defined in this document are
  39. /// consistently implemented in a single, logical order, since "lost
  40. /// update" preconditions have more strict requirements than cache
  41. /// validation, a validated cache is more efficient than a partial
  42. /// response, and entity tags are presumed to be more accurate than date
  43. /// validators. https://tools.ietf.org/html/rfc7232#section-5
  44. if entry.Attr.Mtime.IsZero() {
  45. return false
  46. }
  47. w.Header().Set("Last-Modified", entry.Attr.Mtime.UTC().Format(http.TimeFormat))
  48. ifMatchETagHeader := r.Header.Get("If-Match")
  49. ifUnmodifiedSinceHeader := r.Header.Get("If-Unmodified-Since")
  50. if ifMatchETagHeader != "" {
  51. if util.CanonicalizeETag(etag) != util.CanonicalizeETag(ifMatchETagHeader) {
  52. w.WriteHeader(http.StatusPreconditionFailed)
  53. return true
  54. }
  55. } else if ifUnmodifiedSinceHeader != "" {
  56. if t, parseError := time.Parse(http.TimeFormat, ifUnmodifiedSinceHeader); parseError == nil {
  57. if t.Before(entry.Attr.Mtime) {
  58. w.WriteHeader(http.StatusPreconditionFailed)
  59. return true
  60. }
  61. }
  62. }
  63. ifNoneMatchETagHeader := r.Header.Get("If-None-Match")
  64. ifModifiedSinceHeader := r.Header.Get("If-Modified-Since")
  65. if ifNoneMatchETagHeader != "" {
  66. if util.CanonicalizeETag(etag) == util.CanonicalizeETag(ifNoneMatchETagHeader) {
  67. SetEtag(w, etag)
  68. w.WriteHeader(http.StatusNotModified)
  69. return true
  70. }
  71. } else if ifModifiedSinceHeader != "" {
  72. if t, parseError := time.Parse(http.TimeFormat, ifModifiedSinceHeader); parseError == nil {
  73. if !t.Before(entry.Attr.Mtime) {
  74. SetEtag(w, etag)
  75. w.WriteHeader(http.StatusNotModified)
  76. return true
  77. }
  78. }
  79. }
  80. return false
  81. }
  82. func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) {
  83. path := r.URL.Path
  84. isForDirectory := strings.HasSuffix(path, "/")
  85. if isForDirectory && len(path) > 1 {
  86. path = path[:len(path)-1]
  87. }
  88. entry, err := fs.filer.FindEntry(context.Background(), util.FullPath(path))
  89. if err != nil {
  90. if path == "/" {
  91. fs.listDirectoryHandler(w, r)
  92. return
  93. }
  94. if err == filer_pb.ErrNotFound {
  95. glog.V(2).Infof("Not found %s: %v", path, err)
  96. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadNotFound).Inc()
  97. w.WriteHeader(http.StatusNotFound)
  98. } else {
  99. glog.Errorf("Internal %s: %v", path, err)
  100. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadInternal).Inc()
  101. w.WriteHeader(http.StatusInternalServerError)
  102. }
  103. return
  104. }
  105. query := r.URL.Query()
  106. if entry.IsDirectory() {
  107. if fs.option.DisableDirListing {
  108. w.WriteHeader(http.StatusForbidden)
  109. return
  110. }
  111. if query.Get("metadata") == "true" {
  112. writeJsonQuiet(w, r, http.StatusOK, entry)
  113. return
  114. }
  115. if entry.Attr.Mime == "" || (entry.Attr.Mime == s3_constants.FolderMimeType && r.Header.Get(s3_constants.AmzIdentityId) == "") {
  116. // Don't return directory meta if config value is set to true
  117. if fs.option.ExposeDirectoryData == false {
  118. writeJsonError(w, r, http.StatusForbidden, errors.New("directory listing is disabled"))
  119. return
  120. }
  121. // return index of directory for non s3 gateway
  122. fs.listDirectoryHandler(w, r)
  123. return
  124. }
  125. // inform S3 API this is a user created directory key object
  126. w.Header().Set(s3_constants.SeaweedFSIsDirectoryKey, "true")
  127. }
  128. if isForDirectory && entry.Attr.Mime != s3_constants.FolderMimeType {
  129. w.WriteHeader(http.StatusNotFound)
  130. return
  131. }
  132. if query.Get("metadata") == "true" {
  133. if query.Get("resolveManifest") == "true" {
  134. if entry.Chunks, _, err = filer.ResolveChunkManifest(
  135. fs.filer.MasterClient.GetLookupFileIdFunction(),
  136. entry.GetChunks(), 0, math.MaxInt64); err != nil {
  137. err = fmt.Errorf("failed to resolve chunk manifest, err: %s", err.Error())
  138. writeJsonError(w, r, http.StatusInternalServerError, err)
  139. return
  140. }
  141. }
  142. writeJsonQuiet(w, r, http.StatusOK, entry)
  143. return
  144. }
  145. if checkPreconditions(w, r, entry) {
  146. return
  147. }
  148. var etag string
  149. if partNumber, errNum := strconv.Atoi(r.Header.Get(s3_constants.SeaweedFSPartNumber)); errNum == nil {
  150. if len(entry.Chunks) < partNumber {
  151. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadChunk).Inc()
  152. w.WriteHeader(http.StatusBadRequest)
  153. w.Write([]byte("InvalidPart"))
  154. return
  155. }
  156. w.Header().Set(s3_constants.AmzMpPartsCount, strconv.Itoa(len(entry.Chunks)))
  157. partChunk := entry.GetChunks()[partNumber-1]
  158. md5, _ := base64.StdEncoding.DecodeString(partChunk.ETag)
  159. etag = hex.EncodeToString(md5)
  160. r.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", partChunk.Offset, uint64(partChunk.Offset)+partChunk.Size-1))
  161. } else {
  162. etag = filer.ETagEntry(entry)
  163. }
  164. w.Header().Set("Accept-Ranges", "bytes")
  165. // mime type
  166. mimeType := entry.Attr.Mime
  167. if mimeType == "" {
  168. if ext := filepath.Ext(entry.Name()); ext != "" {
  169. mimeType = mime.TypeByExtension(ext)
  170. }
  171. }
  172. if mimeType != "" {
  173. w.Header().Set("Content-Type", mimeType)
  174. } else {
  175. w.Header().Set("Content-Type", "application/octet-stream")
  176. }
  177. // print out the header from extended properties
  178. for k, v := range entry.Extended {
  179. if !strings.HasPrefix(k, "xattr-") {
  180. // "xattr-" prefix is set in filesys.XATTR_PREFIX
  181. w.Header().Set(k, string(v))
  182. }
  183. }
  184. //Seaweed custom header are not visible to Vue or javascript
  185. seaweedHeaders := []string{}
  186. for header := range w.Header() {
  187. if strings.HasPrefix(header, "Seaweed-") {
  188. seaweedHeaders = append(seaweedHeaders, header)
  189. }
  190. }
  191. seaweedHeaders = append(seaweedHeaders, "Content-Disposition")
  192. w.Header().Set("Access-Control-Expose-Headers", strings.Join(seaweedHeaders, ","))
  193. //set tag count
  194. tagCount := 0
  195. for k := range entry.Extended {
  196. if strings.HasPrefix(k, s3_constants.AmzObjectTagging+"-") {
  197. tagCount++
  198. }
  199. }
  200. if tagCount > 0 {
  201. w.Header().Set(s3_constants.AmzTagCount, strconv.Itoa(tagCount))
  202. }
  203. SetEtag(w, etag)
  204. filename := entry.Name()
  205. AdjustPassthroughHeaders(w, r, filename)
  206. totalSize := int64(entry.Size())
  207. if r.Method == http.MethodHead {
  208. w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
  209. return
  210. }
  211. if rangeReq := r.Header.Get("Range"); rangeReq == "" {
  212. ext := filepath.Ext(filename)
  213. if len(ext) > 0 {
  214. ext = strings.ToLower(ext)
  215. }
  216. width, height, mode, shouldResize := shouldResizeImages(ext, r)
  217. if shouldResize {
  218. data := mem.Allocate(int(totalSize))
  219. defer mem.Free(data)
  220. err := filer.ReadAll(data, fs.filer.MasterClient, entry.GetChunks())
  221. if err != nil {
  222. glog.Errorf("failed to read %s: %v", path, err)
  223. w.WriteHeader(http.StatusInternalServerError)
  224. return
  225. }
  226. rs, _, _ := images.Resized(ext, bytes.NewReader(data), width, height, mode)
  227. io.Copy(w, rs)
  228. return
  229. }
  230. }
  231. ProcessRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
  232. if offset+size <= int64(len(entry.Content)) {
  233. return func(writer io.Writer) error {
  234. _, err := writer.Write(entry.Content[offset : offset+size])
  235. if err != nil {
  236. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorWriteEntry).Inc()
  237. glog.Errorf("failed to write entry content: %v", err)
  238. }
  239. return err
  240. }, nil
  241. }
  242. chunks := entry.GetChunks()
  243. if entry.IsInRemoteOnly() {
  244. dir, name := entry.FullPath.DirAndName()
  245. if resp, err := fs.CacheRemoteObjectToLocalCluster(context.Background(), &filer_pb.CacheRemoteObjectToLocalClusterRequest{
  246. Directory: dir,
  247. Name: name,
  248. }); err != nil {
  249. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadCache).Inc()
  250. glog.Errorf("CacheRemoteObjectToLocalCluster %s: %v", entry.FullPath, err)
  251. return nil, fmt.Errorf("cache %s: %v", entry.FullPath, err)
  252. } else {
  253. chunks = resp.Entry.GetChunks()
  254. }
  255. }
  256. streamFn, err := filer.PrepareStreamContentWithThrottler(fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, chunks, offset, size, fs.option.DownloadMaxBytesPs)
  257. if err != nil {
  258. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
  259. glog.Errorf("failed to prepare stream content %s: %v", r.URL, err)
  260. return nil, err
  261. }
  262. return func(writer io.Writer) error {
  263. err := streamFn(writer)
  264. if err != nil {
  265. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
  266. glog.Errorf("failed to stream content %s: %v", r.URL, err)
  267. }
  268. return err
  269. }, nil
  270. })
  271. }
  272. func (fs *FilerServer) maybeGetVolumeReadJwtAuthorizationToken(fileId string) string {
  273. return string(security.GenJwtForVolumeServer(fs.volumeGuard.ReadSigningKey, fs.volumeGuard.ReadExpiresAfterSec, fileId))
  274. }