s3api_object_handlers.go 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222
  1. package s3api
  2. import (
  3. "bytes"
  4. "fmt"
  5. "io"
  6. "net/http"
  7. "net/url"
  8. "strings"
  9. "time"
  10. "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
  11. "github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
  12. "github.com/seaweedfs/seaweedfs/weed/util/mem"
  13. "github.com/seaweedfs/seaweedfs/weed/glog"
  14. "github.com/seaweedfs/seaweedfs/weed/util"
  15. )
  16. func mimeDetect(r *http.Request, dataReader io.Reader) io.ReadCloser {
  17. mimeBuffer := make([]byte, 512)
  18. size, _ := dataReader.Read(mimeBuffer)
  19. if size > 0 {
  20. r.Header.Set("Content-Type", http.DetectContentType(mimeBuffer[:size]))
  21. return io.NopCloser(io.MultiReader(bytes.NewReader(mimeBuffer[:size]), dataReader))
  22. }
  23. return io.NopCloser(dataReader)
  24. }
  25. func urlEscapeObject(object string) string {
  26. t := urlPathEscape(removeDuplicateSlashes(object))
  27. if strings.HasPrefix(t, "/") {
  28. return t
  29. }
  30. return "/" + t
  31. }
  32. func urlPathEscape(object string) string {
  33. var escapedParts []string
  34. for _, part := range strings.Split(object, "/") {
  35. escapedParts = append(escapedParts, url.PathEscape(part))
  36. }
  37. return strings.Join(escapedParts, "/")
  38. }
  39. func removeDuplicateSlashes(object string) string {
  40. result := strings.Builder{}
  41. result.Grow(len(object))
  42. isLastSlash := false
  43. for _, r := range object {
  44. switch r {
  45. case '/':
  46. if !isLastSlash {
  47. result.WriteRune(r)
  48. }
  49. isLastSlash = true
  50. default:
  51. result.WriteRune(r)
  52. isLastSlash = false
  53. }
  54. }
  55. return result.String()
  56. }
  57. func (s3a *S3ApiServer) toFilerUrl(bucket, object string) string {
  58. object = urlPathEscape(removeDuplicateSlashes(object))
  59. destUrl := fmt.Sprintf("http://%s%s/%s%s",
  60. s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, object)
  61. return destUrl
  62. }
  63. func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
  64. bucket, object := s3_constants.GetBucketAndObject(r)
  65. glog.V(3).Infof("GetObjectHandler %s %s", bucket, object)
  66. if strings.HasSuffix(r.URL.Path, "/") {
  67. s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
  68. return
  69. }
  70. destUrl := s3a.toFilerUrl(bucket, object)
  71. s3a.proxyToFiler(w, r, destUrl, false, passThroughResponse)
  72. }
  73. func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
  74. bucket, object := s3_constants.GetBucketAndObject(r)
  75. glog.V(3).Infof("HeadObjectHandler %s %s", bucket, object)
  76. destUrl := s3a.toFilerUrl(bucket, object)
  77. s3a.proxyToFiler(w, r, destUrl, false, passThroughResponse)
  78. }
  79. func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, isWrite bool, responseFn func(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int)) {
  80. glog.V(3).Infof("s3 proxying %s to %s", r.Method, destUrl)
  81. start := time.Now()
  82. proxyReq, err := http.NewRequest(r.Method, destUrl, r.Body)
  83. if err != nil {
  84. glog.Errorf("NewRequest %s: %v", destUrl, err)
  85. s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
  86. return
  87. }
  88. proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
  89. for k, v := range r.URL.Query() {
  90. if _, ok := s3_constants.PassThroughHeaders[strings.ToLower(k)]; ok {
  91. proxyReq.Header[k] = v
  92. }
  93. if k == "partNumber" {
  94. proxyReq.Header[s3_constants.SeaweedFSPartNumber] = v
  95. }
  96. }
  97. for header, values := range r.Header {
  98. proxyReq.Header[header] = values
  99. }
  100. // ensure that the Authorization header is overriding any previous
  101. // Authorization header which might be already present in proxyReq
  102. s3a.maybeAddFilerJwtAuthorization(proxyReq, isWrite)
  103. resp, postErr := s3a.client.Do(proxyReq)
  104. if postErr != nil {
  105. glog.Errorf("post to filer: %v", postErr)
  106. s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
  107. return
  108. }
  109. defer util.CloseResponse(resp)
  110. if resp.StatusCode == http.StatusPreconditionFailed {
  111. s3err.WriteErrorResponse(w, r, s3err.ErrPreconditionFailed)
  112. return
  113. }
  114. if resp.StatusCode == http.StatusRequestedRangeNotSatisfiable {
  115. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRange)
  116. return
  117. }
  118. if r.Method == "DELETE" {
  119. if resp.StatusCode == http.StatusNotFound {
  120. // this is normal
  121. responseStatusCode := responseFn(resp, w)
  122. s3err.PostLog(r, responseStatusCode, s3err.ErrNone)
  123. return
  124. }
  125. }
  126. if resp.StatusCode == http.StatusNotFound {
  127. s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
  128. return
  129. }
  130. TimeToFirstByte(r.Method, start, r)
  131. if resp.Header.Get(s3_constants.SeaweedFSIsDirectoryKey) == "true" {
  132. responseStatusCode := responseFn(resp, w)
  133. s3err.PostLog(r, responseStatusCode, s3err.ErrNone)
  134. return
  135. }
  136. if resp.StatusCode == http.StatusInternalServerError {
  137. s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
  138. return
  139. }
  140. // when HEAD a directory, it should be reported as no such key
  141. // https://github.com/seaweedfs/seaweedfs/issues/3457
  142. if resp.ContentLength == -1 && resp.StatusCode != http.StatusNotModified {
  143. s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
  144. return
  145. }
  146. if resp.StatusCode == http.StatusBadRequest {
  147. resp_body, _ := io.ReadAll(resp.Body)
  148. switch string(resp_body) {
  149. case "InvalidPart":
  150. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
  151. default:
  152. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)
  153. }
  154. resp.Body.Close()
  155. return
  156. }
  157. setUserMetadataKeyToLowercase(resp)
  158. responseStatusCode := responseFn(resp, w)
  159. s3err.PostLog(r, responseStatusCode, s3err.ErrNone)
  160. }
  161. func setUserMetadataKeyToLowercase(resp *http.Response) {
  162. for key, value := range resp.Header {
  163. if strings.HasPrefix(key, s3_constants.AmzUserMetaPrefix) {
  164. resp.Header[strings.ToLower(key)] = value
  165. delete(resp.Header, key)
  166. }
  167. }
  168. }
  169. func passThroughResponse(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int) {
  170. for k, v := range proxyResponse.Header {
  171. w.Header()[k] = v
  172. }
  173. if proxyResponse.Header.Get("Content-Range") != "" && proxyResponse.StatusCode == 200 {
  174. w.WriteHeader(http.StatusPartialContent)
  175. statusCode = http.StatusPartialContent
  176. } else {
  177. statusCode = proxyResponse.StatusCode
  178. }
  179. w.WriteHeader(statusCode)
  180. buf := mem.Allocate(128 * 1024)
  181. defer mem.Free(buf)
  182. if n, err := io.CopyBuffer(w, proxyResponse.Body, buf); err != nil {
  183. glog.V(1).Infof("passthrough response read %d bytes: %v", n, err)
  184. }
  185. return statusCode
  186. }