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.

320 lines
7.9 KiB

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