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.

275 lines
8.1 KiB

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