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.

325 lines
8.2 KiB

6 years ago
7 years ago
7 years ago
7 years ago
4 years ago
4 years ago
6 years ago
4 years ago
6 years ago
4 years ago
4 years ago
4 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
4 years ago
  1. package s3api
  2. import (
  3. "crypto/md5"
  4. "encoding/json"
  5. "encoding/xml"
  6. "fmt"
  7. "io"
  8. "io/ioutil"
  9. "net/http"
  10. "strings"
  11. "github.com/gorilla/mux"
  12. "github.com/chrislusf/seaweedfs/weed/glog"
  13. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  14. weed_server "github.com/chrislusf/seaweedfs/weed/server"
  15. "github.com/chrislusf/seaweedfs/weed/util"
  16. )
  17. var (
  18. client *http.Client
  19. )
  20. func init() {
  21. client = &http.Client{Transport: &http.Transport{
  22. MaxIdleConnsPerHost: 1024,
  23. }}
  24. }
  25. func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
  26. // http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
  27. bucket, object := getBucketAndObject(r)
  28. _, err := validateContentMd5(r.Header)
  29. if err != nil {
  30. writeErrorResponse(w, ErrInvalidDigest, r.URL)
  31. return
  32. }
  33. rAuthType := getRequestAuthType(r)
  34. dataReader := r.Body
  35. var s3ErrCode ErrorCode
  36. switch rAuthType {
  37. case authTypeStreamingSigned:
  38. dataReader, s3ErrCode = s3a.iam.newSignV4ChunkedReader(r)
  39. case authTypeSignedV2, authTypePresignedV2:
  40. _, s3ErrCode = s3a.iam.isReqAuthenticatedV2(r)
  41. case authTypePresigned, authTypeSigned:
  42. _, s3ErrCode = s3a.iam.reqSignatureV4Verify(r)
  43. }
  44. if s3ErrCode != ErrNone {
  45. writeErrorResponse(w, s3ErrCode, r.URL)
  46. return
  47. }
  48. defer dataReader.Close()
  49. if strings.HasSuffix(object, "/") {
  50. if err := s3a.mkdir(s3a.option.BucketsPath, bucket+object, nil); err != nil {
  51. writeErrorResponse(w, ErrInternalError, r.URL)
  52. return
  53. }
  54. } else {
  55. uploadUrl := fmt.Sprintf("http://%s%s/%s%s", s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
  56. etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
  57. if errCode != ErrNone {
  58. writeErrorResponse(w, errCode, r.URL)
  59. return
  60. }
  61. setEtag(w, etag)
  62. }
  63. writeSuccessResponseEmpty(w)
  64. }
  65. func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
  66. bucket, object := getBucketAndObject(r)
  67. if strings.HasSuffix(r.URL.Path, "/") {
  68. writeErrorResponse(w, ErrNotImplemented, r.URL)
  69. return
  70. }
  71. destUrl := fmt.Sprintf("http://%s%s/%s%s",
  72. s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
  73. s3a.proxyToFiler(w, r, destUrl, passThroughResponse)
  74. }
  75. func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
  76. bucket, object := getBucketAndObject(r)
  77. destUrl := fmt.Sprintf("http://%s%s/%s%s",
  78. s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
  79. s3a.proxyToFiler(w, r, destUrl, passThroughResponse)
  80. }
  81. func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Request) {
  82. bucket, object := getBucketAndObject(r)
  83. destUrl := fmt.Sprintf("http://%s%s/%s%s?recursive=true",
  84. s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
  85. s3a.proxyToFiler(w, r, destUrl, func(proxyResponse *http.Response, w http.ResponseWriter) {
  86. for k, v := range proxyResponse.Header {
  87. w.Header()[k] = v
  88. }
  89. w.WriteHeader(http.StatusNoContent)
  90. })
  91. }
  92. // / ObjectIdentifier carries key name for the object to delete.
  93. type ObjectIdentifier struct {
  94. ObjectName string `xml:"Key"`
  95. }
  96. // DeleteObjectsRequest - xml carrying the object key names which needs to be deleted.
  97. type DeleteObjectsRequest struct {
  98. // Element to enable quiet mode for the request
  99. Quiet bool
  100. // List of objects to be deleted
  101. Objects []ObjectIdentifier `xml:"Object"`
  102. }
  103. // DeleteError structure.
  104. type DeleteError struct {
  105. Code string
  106. Message string
  107. Key string
  108. }
  109. // DeleteObjectsResponse container for multiple object deletes.
  110. type DeleteObjectsResponse struct {
  111. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ DeleteResult" json:"-"`
  112. // Collection of all deleted objects
  113. DeletedObjects []ObjectIdentifier `xml:"Deleted,omitempty"`
  114. // Collection of errors deleting certain objects.
  115. Errors []DeleteError `xml:"Error,omitempty"`
  116. }
  117. // DeleteMultipleObjectsHandler - Delete multiple objects
  118. func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Request) {
  119. bucket, _ := getBucketAndObject(r)
  120. deleteXMLBytes, err := ioutil.ReadAll(r.Body)
  121. if err != nil {
  122. writeErrorResponse(w, ErrInternalError, r.URL)
  123. return
  124. }
  125. deleteObjects := &DeleteObjectsRequest{}
  126. if err := xml.Unmarshal(deleteXMLBytes, deleteObjects); err != nil {
  127. writeErrorResponse(w, ErrMalformedXML, r.URL)
  128. return
  129. }
  130. var deletedObjects []ObjectIdentifier
  131. var deleteErrors []DeleteError
  132. s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  133. for _, object := range deleteObjects.Objects {
  134. lastSeparator := strings.LastIndex(object.ObjectName, "/")
  135. parentDirectoryPath, entryName, isDeleteData, isRecursive := "/", object.ObjectName, true, true
  136. if lastSeparator > 0 && lastSeparator+1 < len(object.ObjectName) {
  137. entryName = object.ObjectName[lastSeparator+1:]
  138. parentDirectoryPath = "/" + object.ObjectName[:lastSeparator]
  139. }
  140. parentDirectoryPath = fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, parentDirectoryPath)
  141. err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
  142. if err == nil {
  143. deletedObjects = append(deletedObjects, object)
  144. } else {
  145. deleteErrors = append(deleteErrors, DeleteError{
  146. Code: "",
  147. Message: err.Error(),
  148. Key: object.ObjectName,
  149. })
  150. }
  151. }
  152. return nil
  153. })
  154. deleteResp := DeleteObjectsResponse{}
  155. if !deleteObjects.Quiet {
  156. deleteResp.DeletedObjects = deletedObjects
  157. }
  158. deleteResp.Errors = deleteErrors
  159. writeSuccessResponseXML(w, encodeResponse(deleteResp))
  160. }
  161. func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, responseFn func(proxyResponse *http.Response, w http.ResponseWriter)) {
  162. glog.V(2).Infof("s3 proxying %s to %s", r.Method, destUrl)
  163. proxyReq, err := http.NewRequest(r.Method, destUrl, r.Body)
  164. if err != nil {
  165. glog.Errorf("NewRequest %s: %v", destUrl, err)
  166. writeErrorResponse(w, ErrInternalError, r.URL)
  167. return
  168. }
  169. proxyReq.Header.Set("Host", s3a.option.Filer)
  170. proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
  171. for header, values := range r.Header {
  172. for _, value := range values {
  173. proxyReq.Header.Add(header, value)
  174. }
  175. }
  176. resp, postErr := client.Do(proxyReq)
  177. if resp.ContentLength == -1 {
  178. writeErrorResponse(w, ErrNoSuchKey, r.URL)
  179. return
  180. }
  181. if postErr != nil {
  182. glog.Errorf("post to filer: %v", postErr)
  183. writeErrorResponse(w, ErrInternalError, r.URL)
  184. return
  185. }
  186. defer util.CloseResponse(resp)
  187. responseFn(resp, w)
  188. }
  189. func passThroughResponse(proxyResponse *http.Response, w http.ResponseWriter) {
  190. for k, v := range proxyResponse.Header {
  191. w.Header()[k] = v
  192. }
  193. w.WriteHeader(proxyResponse.StatusCode)
  194. io.Copy(w, proxyResponse.Body)
  195. }
  196. func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader io.Reader) (etag string, code ErrorCode) {
  197. hash := md5.New()
  198. var body = io.TeeReader(dataReader, hash)
  199. proxyReq, err := http.NewRequest("PUT", uploadUrl, body)
  200. if err != nil {
  201. glog.Errorf("NewRequest %s: %v", uploadUrl, err)
  202. return "", ErrInternalError
  203. }
  204. proxyReq.Header.Set("Host", s3a.option.Filer)
  205. proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
  206. for header, values := range r.Header {
  207. for _, value := range values {
  208. proxyReq.Header.Add(header, value)
  209. }
  210. }
  211. resp, postErr := client.Do(proxyReq)
  212. if postErr != nil {
  213. glog.Errorf("post to filer: %v", postErr)
  214. return "", ErrInternalError
  215. }
  216. defer resp.Body.Close()
  217. etag = fmt.Sprintf("%x", hash.Sum(nil))
  218. resp_body, ra_err := ioutil.ReadAll(resp.Body)
  219. if ra_err != nil {
  220. glog.Errorf("upload to filer response read: %v", ra_err)
  221. return etag, ErrInternalError
  222. }
  223. var ret weed_server.FilerPostResult
  224. unmarshal_err := json.Unmarshal(resp_body, &ret)
  225. if unmarshal_err != nil {
  226. glog.Errorf("failing to read upload to %s : %v", uploadUrl, string(resp_body))
  227. return "", ErrInternalError
  228. }
  229. if ret.Error != "" {
  230. glog.Errorf("upload to filer error: %v", ret.Error)
  231. return "", ErrInternalError
  232. }
  233. return etag, ErrNone
  234. }
  235. func setEtag(w http.ResponseWriter, etag string) {
  236. if etag != "" {
  237. if strings.HasPrefix(etag, "\"") {
  238. w.Header().Set("ETag", etag)
  239. } else {
  240. w.Header().Set("ETag", "\""+etag+"\"")
  241. }
  242. }
  243. }
  244. func getBucketAndObject(r *http.Request) (bucket, object string) {
  245. vars := mux.Vars(r)
  246. bucket = vars["bucket"]
  247. object = vars["object"]
  248. if !strings.HasPrefix(object, "/") {
  249. object = "/" + object
  250. }
  251. return
  252. }