filer_server_handlers_read.go 9.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300
  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. // Don't return directory meta if config value is set to true
  113. if fs.option.ExposeDirectoryData == false {
  114. writeJsonError(w, r, http.StatusForbidden, errors.New("directory listing is disabled"))
  115. return
  116. }
  117. }
  118. if entry.Attr.Mime == "" || (entry.Attr.Mime == s3_constants.FolderMimeType && r.Header.Get(s3_constants.AmzIdentityId) == "") {
  119. // return index of directory for non s3 gateway
  120. fs.listDirectoryHandler(w, r)
  121. return
  122. }
  123. // inform S3 API this is a user created directory key object
  124. w.Header().Set(s3_constants.SeaweedFSIsDirectoryKey, "true")
  125. }
  126. if isForDirectory && entry.Attr.Mime != s3_constants.FolderMimeType {
  127. w.WriteHeader(http.StatusNotFound)
  128. return
  129. }
  130. if query.Get("metadata") == "true" {
  131. if query.Get("resolveManifest") == "true" {
  132. if entry.Chunks, _, err = filer.ResolveChunkManifest(
  133. fs.filer.MasterClient.GetLookupFileIdFunction(),
  134. entry.GetChunks(), 0, math.MaxInt64); err != nil {
  135. err = fmt.Errorf("failed to resolve chunk manifest, err: %s", err.Error())
  136. writeJsonError(w, r, http.StatusInternalServerError, err)
  137. return
  138. }
  139. }
  140. writeJsonQuiet(w, r, http.StatusOK, entry)
  141. return
  142. }
  143. if checkPreconditions(w, r, entry) {
  144. return
  145. }
  146. var etag string
  147. if partNumber, errNum := strconv.Atoi(r.Header.Get(s3_constants.SeaweedFSPartNumber)); errNum == nil {
  148. if len(entry.Chunks) < partNumber {
  149. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadChunk).Inc()
  150. w.WriteHeader(http.StatusBadRequest)
  151. w.Write([]byte("InvalidPart"))
  152. return
  153. }
  154. w.Header().Set(s3_constants.AmzMpPartsCount, strconv.Itoa(len(entry.Chunks)))
  155. partChunk := entry.GetChunks()[partNumber-1]
  156. md5, _ := base64.StdEncoding.DecodeString(partChunk.ETag)
  157. etag = hex.EncodeToString(md5)
  158. r.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", partChunk.Offset, uint64(partChunk.Offset)+partChunk.Size-1))
  159. } else {
  160. etag = filer.ETagEntry(entry)
  161. }
  162. w.Header().Set("Accept-Ranges", "bytes")
  163. // mime type
  164. mimeType := entry.Attr.Mime
  165. if mimeType == "" {
  166. if ext := filepath.Ext(entry.Name()); ext != "" {
  167. mimeType = mime.TypeByExtension(ext)
  168. }
  169. }
  170. if mimeType != "" {
  171. w.Header().Set("Content-Type", mimeType)
  172. } else {
  173. w.Header().Set("Content-Type", "application/octet-stream")
  174. }
  175. // print out the header from extended properties
  176. for k, v := range entry.Extended {
  177. if !strings.HasPrefix(k, "xattr-") {
  178. // "xattr-" prefix is set in filesys.XATTR_PREFIX
  179. w.Header().Set(k, string(v))
  180. }
  181. }
  182. //Seaweed custom header are not visible to Vue or javascript
  183. seaweedHeaders := []string{}
  184. for header := range w.Header() {
  185. if strings.HasPrefix(header, "Seaweed-") {
  186. seaweedHeaders = append(seaweedHeaders, header)
  187. }
  188. }
  189. seaweedHeaders = append(seaweedHeaders, "Content-Disposition")
  190. w.Header().Set("Access-Control-Expose-Headers", strings.Join(seaweedHeaders, ","))
  191. //set tag count
  192. tagCount := 0
  193. for k := range entry.Extended {
  194. if strings.HasPrefix(k, s3_constants.AmzObjectTagging+"-") {
  195. tagCount++
  196. }
  197. }
  198. if tagCount > 0 {
  199. w.Header().Set(s3_constants.AmzTagCount, strconv.Itoa(tagCount))
  200. }
  201. setEtag(w, etag)
  202. filename := entry.Name()
  203. adjustPassthroughHeaders(w, r, filename)
  204. totalSize := int64(entry.Size())
  205. if r.Method == "HEAD" {
  206. w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
  207. return
  208. }
  209. if rangeReq := r.Header.Get("Range"); rangeReq == "" {
  210. ext := filepath.Ext(filename)
  211. if len(ext) > 0 {
  212. ext = strings.ToLower(ext)
  213. }
  214. width, height, mode, shouldResize := shouldResizeImages(ext, r)
  215. if shouldResize {
  216. data := mem.Allocate(int(totalSize))
  217. defer mem.Free(data)
  218. err := filer.ReadAll(data, fs.filer.MasterClient, entry.GetChunks())
  219. if err != nil {
  220. glog.Errorf("failed to read %s: %v", path, err)
  221. w.WriteHeader(http.StatusInternalServerError)
  222. return
  223. }
  224. rs, _, _ := images.Resized(ext, bytes.NewReader(data), width, height, mode)
  225. io.Copy(w, rs)
  226. return
  227. }
  228. }
  229. processRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
  230. if offset+size <= int64(len(entry.Content)) {
  231. return func(writer io.Writer) error {
  232. _, err := writer.Write(entry.Content[offset : offset+size])
  233. if err != nil {
  234. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorWriteEntry).Inc()
  235. glog.Errorf("failed to write entry content: %v", err)
  236. }
  237. return err
  238. }, nil
  239. }
  240. chunks := entry.GetChunks()
  241. if entry.IsInRemoteOnly() {
  242. dir, name := entry.FullPath.DirAndName()
  243. if resp, err := fs.CacheRemoteObjectToLocalCluster(context.Background(), &filer_pb.CacheRemoteObjectToLocalClusterRequest{
  244. Directory: dir,
  245. Name: name,
  246. }); err != nil {
  247. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadCache).Inc()
  248. glog.Errorf("CacheRemoteObjectToLocalCluster %s: %v", entry.FullPath, err)
  249. return nil, fmt.Errorf("cache %s: %v", entry.FullPath, err)
  250. } else {
  251. chunks = resp.Entry.GetChunks()
  252. }
  253. }
  254. streamFn, err := filer.PrepareStreamContentWithThrottler(fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, chunks, offset, size, fs.option.DownloadMaxBytesPs)
  255. if err != nil {
  256. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
  257. glog.Errorf("failed to prepare stream content %s: %v", r.URL, err)
  258. return nil, err
  259. }
  260. return func(writer io.Writer) error {
  261. err := streamFn(writer)
  262. if err != nil {
  263. stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
  264. glog.Errorf("failed to stream content %s: %v", r.URL, err)
  265. }
  266. return err
  267. }, nil
  268. })
  269. }
  270. func (fs *FilerServer) maybeGetVolumeReadJwtAuthorizationToken(fileId string) string {
  271. return string(security.GenJwtForVolumeServer(fs.volumeGuard.ReadSigningKey, fs.volumeGuard.ReadExpiresAfterSec, fileId))
  272. }