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.

413 lines
11 KiB

6 years ago
7 years ago
7 years ago
7 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 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
4 years ago
4 years ago
5 years ago
5 years ago
4 years ago
4 years ago
4 years ago
5 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 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. "github.com/chrislusf/seaweedfs/weed/filer"
  8. "io"
  9. "io/ioutil"
  10. "net/http"
  11. "net/url"
  12. "sort"
  13. "strings"
  14. "github.com/chrislusf/seaweedfs/weed/s3api/s3err"
  15. "github.com/gorilla/mux"
  16. "github.com/chrislusf/seaweedfs/weed/glog"
  17. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  18. weed_server "github.com/chrislusf/seaweedfs/weed/server"
  19. "github.com/chrislusf/seaweedfs/weed/util"
  20. )
  21. var (
  22. client *http.Client
  23. )
  24. func init() {
  25. client = &http.Client{Transport: &http.Transport{
  26. MaxIdleConns: 1024,
  27. MaxIdleConnsPerHost: 1024,
  28. }}
  29. }
  30. func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
  31. // http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
  32. bucket, object := getBucketAndObject(r)
  33. _, err := validateContentMd5(r.Header)
  34. if err != nil {
  35. writeErrorResponse(w, s3err.ErrInvalidDigest, r.URL)
  36. return
  37. }
  38. dataReader := r.Body
  39. if s3a.iam.isEnabled() {
  40. rAuthType := getRequestAuthType(r)
  41. var s3ErrCode s3err.ErrorCode
  42. switch rAuthType {
  43. case authTypeStreamingSigned:
  44. dataReader, s3ErrCode = s3a.iam.newSignV4ChunkedReader(r)
  45. case authTypeSignedV2, authTypePresignedV2:
  46. _, s3ErrCode = s3a.iam.isReqAuthenticatedV2(r)
  47. case authTypePresigned, authTypeSigned:
  48. _, s3ErrCode = s3a.iam.reqSignatureV4Verify(r)
  49. }
  50. if s3ErrCode != s3err.ErrNone {
  51. writeErrorResponse(w, s3ErrCode, r.URL)
  52. return
  53. }
  54. }
  55. defer dataReader.Close()
  56. if strings.HasSuffix(object, "/") {
  57. if err := s3a.mkdir(s3a.option.BucketsPath, bucket+object, nil); err != nil {
  58. writeErrorResponse(w, s3err.ErrInternalError, r.URL)
  59. return
  60. }
  61. } else {
  62. uploadUrl := fmt.Sprintf("http://%s%s/%s%s", s3a.option.Filer, s3a.option.BucketsPath, bucket, urlPathEscape(object))
  63. etag, errCode := s3a.putToFiler(r, uploadUrl, dataReader)
  64. if errCode != s3err.ErrNone {
  65. writeErrorResponse(w, errCode, r.URL)
  66. return
  67. }
  68. setEtag(w, etag)
  69. }
  70. writeSuccessResponseEmpty(w)
  71. }
  72. func urlPathEscape(object string) string {
  73. var escapedParts []string
  74. for _, part := range strings.Split(object, "/") {
  75. escapedParts = append(escapedParts, url.PathEscape(part))
  76. }
  77. return strings.Join(escapedParts, "/")
  78. }
  79. func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
  80. bucket, object := getBucketAndObject(r)
  81. if strings.HasSuffix(r.URL.Path, "/") {
  82. writeErrorResponse(w, s3err.ErrNotImplemented, r.URL)
  83. return
  84. }
  85. destUrl := fmt.Sprintf("http://%s%s/%s%s",
  86. s3a.option.Filer, s3a.option.BucketsPath, bucket, urlPathEscape(object))
  87. s3a.proxyToFiler(w, r, destUrl, passThroughResponse)
  88. }
  89. func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
  90. bucket, object := getBucketAndObject(r)
  91. destUrl := fmt.Sprintf("http://%s%s/%s%s",
  92. s3a.option.Filer, s3a.option.BucketsPath, bucket, urlPathEscape(object))
  93. s3a.proxyToFiler(w, r, destUrl, passThroughResponse)
  94. }
  95. func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Request) {
  96. bucket, object := getBucketAndObject(r)
  97. destUrl := fmt.Sprintf("http://%s%s/%s%s?recursive=true",
  98. s3a.option.Filer, s3a.option.BucketsPath, bucket, urlPathEscape(object))
  99. s3a.proxyToFiler(w, r, destUrl, func(proxyResponse *http.Response, w http.ResponseWriter) {
  100. for k, v := range proxyResponse.Header {
  101. w.Header()[k] = v
  102. }
  103. w.WriteHeader(http.StatusNoContent)
  104. })
  105. }
  106. // / ObjectIdentifier carries key name for the object to delete.
  107. type ObjectIdentifier struct {
  108. ObjectName string `xml:"Key"`
  109. }
  110. // DeleteObjectsRequest - xml carrying the object key names which needs to be deleted.
  111. type DeleteObjectsRequest struct {
  112. // Element to enable quiet mode for the request
  113. Quiet bool
  114. // List of objects to be deleted
  115. Objects []ObjectIdentifier `xml:"Object"`
  116. }
  117. // DeleteError structure.
  118. type DeleteError struct {
  119. Code string
  120. Message string
  121. Key string
  122. }
  123. // DeleteObjectsResponse container for multiple object deletes.
  124. type DeleteObjectsResponse struct {
  125. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ DeleteResult" json:"-"`
  126. // Collection of all deleted objects
  127. DeletedObjects []ObjectIdentifier `xml:"Deleted,omitempty"`
  128. // Collection of errors deleting certain objects.
  129. Errors []DeleteError `xml:"Error,omitempty"`
  130. }
  131. // DeleteMultipleObjectsHandler - Delete multiple objects
  132. func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Request) {
  133. bucket, _ := getBucketAndObject(r)
  134. deleteXMLBytes, err := ioutil.ReadAll(r.Body)
  135. if err != nil {
  136. writeErrorResponse(w, s3err.ErrInternalError, r.URL)
  137. return
  138. }
  139. deleteObjects := &DeleteObjectsRequest{}
  140. if err := xml.Unmarshal(deleteXMLBytes, deleteObjects); err != nil {
  141. writeErrorResponse(w, s3err.ErrMalformedXML, r.URL)
  142. return
  143. }
  144. var deletedObjects []ObjectIdentifier
  145. var deleteErrors []DeleteError
  146. directoriesWithDeletion := make(map[string]int)
  147. s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  148. // delete file entries
  149. for _, object := range deleteObjects.Objects {
  150. lastSeparator := strings.LastIndex(object.ObjectName, "/")
  151. parentDirectoryPath, entryName, isDeleteData, isRecursive := "", object.ObjectName, true, false
  152. if lastSeparator > 0 && lastSeparator+1 < len(object.ObjectName) {
  153. entryName = object.ObjectName[lastSeparator+1:]
  154. parentDirectoryPath = "/" + object.ObjectName[:lastSeparator]
  155. }
  156. parentDirectoryPath = fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, parentDirectoryPath)
  157. err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
  158. if err == nil {
  159. directoriesWithDeletion[parentDirectoryPath]++
  160. deletedObjects = append(deletedObjects, object)
  161. } else if strings.Contains(err.Error(), filer.MsgFailDelNonEmptyFolder) {
  162. deletedObjects = append(deletedObjects, object)
  163. } else {
  164. delete(directoriesWithDeletion, parentDirectoryPath)
  165. deleteErrors = append(deleteErrors, DeleteError{
  166. Code: "",
  167. Message: err.Error(),
  168. Key: object.ObjectName,
  169. })
  170. }
  171. }
  172. // purge empty folders, only checking folders with deletions
  173. for len(directoriesWithDeletion) > 0 {
  174. directoriesWithDeletion = s3a.doDeleteEmptyDirectories(client, directoriesWithDeletion)
  175. }
  176. return nil
  177. })
  178. deleteResp := DeleteObjectsResponse{}
  179. if !deleteObjects.Quiet {
  180. deleteResp.DeletedObjects = deletedObjects
  181. }
  182. deleteResp.Errors = deleteErrors
  183. writeSuccessResponseXML(w, encodeResponse(deleteResp))
  184. }
  185. func (s3a *S3ApiServer) doDeleteEmptyDirectories(client filer_pb.SeaweedFilerClient, directoriesWithDeletion map[string]int) (newDirectoriesWithDeletion map[string]int) {
  186. var allDirs []string
  187. for dir, _ := range directoriesWithDeletion {
  188. allDirs = append(allDirs, dir)
  189. }
  190. sort.Slice(allDirs, func(i, j int) bool {
  191. return len(allDirs[i]) > len(allDirs[j])
  192. })
  193. newDirectoriesWithDeletion = make(map[string]int)
  194. for _, dir := range allDirs {
  195. parentDir, dirName := util.FullPath(dir).DirAndName()
  196. if parentDir == s3a.option.BucketsPath {
  197. continue
  198. }
  199. if err := doDeleteEntry(client, parentDir, dirName, false, false); err != nil {
  200. glog.V(4).Infof("directory %s has %d deletion but still not empty: %v", dir, directoriesWithDeletion[dir], err)
  201. } else {
  202. newDirectoriesWithDeletion[parentDir]++
  203. }
  204. }
  205. return
  206. }
  207. var passThroughHeaders = []string{
  208. "response-cache-control",
  209. "response-content-disposition",
  210. "response-content-encoding",
  211. "response-content-language",
  212. "response-content-type",
  213. "response-expires",
  214. }
  215. func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, responseFn func(proxyResponse *http.Response, w http.ResponseWriter)) {
  216. glog.V(2).Infof("s3 proxying %s to %s", r.Method, destUrl)
  217. proxyReq, err := http.NewRequest(r.Method, destUrl, r.Body)
  218. if err != nil {
  219. glog.Errorf("NewRequest %s: %v", destUrl, err)
  220. writeErrorResponse(w, s3err.ErrInternalError, r.URL)
  221. return
  222. }
  223. proxyReq.Header.Set("Host", s3a.option.Filer)
  224. proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
  225. for header, values := range r.Header {
  226. // handle s3 related headers
  227. passed := false
  228. for _, h := range passThroughHeaders {
  229. if strings.ToLower(header) == h && len(values) > 0 {
  230. proxyReq.Header.Add(header[len("response-"):], values[0])
  231. passed = true
  232. break
  233. }
  234. }
  235. if passed {
  236. continue
  237. }
  238. // handle other headers
  239. for _, value := range values {
  240. proxyReq.Header.Add(header, value)
  241. }
  242. }
  243. resp, postErr := client.Do(proxyReq)
  244. if postErr != nil {
  245. glog.Errorf("post to filer: %v", postErr)
  246. writeErrorResponse(w, s3err.ErrInternalError, r.URL)
  247. return
  248. }
  249. defer util.CloseResponse(resp)
  250. if (resp.ContentLength == -1 || resp.StatusCode == 404) && resp.StatusCode != 304 {
  251. if r.Method != "DELETE" {
  252. writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
  253. return
  254. }
  255. }
  256. responseFn(resp, w)
  257. }
  258. func passThroughResponse(proxyResponse *http.Response, w http.ResponseWriter) {
  259. for k, v := range proxyResponse.Header {
  260. w.Header()[k] = v
  261. }
  262. if proxyResponse.Header.Get("Content-Range") != "" && proxyResponse.StatusCode == 200 {
  263. w.WriteHeader(http.StatusPartialContent)
  264. } else {
  265. w.WriteHeader(proxyResponse.StatusCode)
  266. }
  267. io.Copy(w, proxyResponse.Body)
  268. }
  269. func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader io.Reader) (etag string, code s3err.ErrorCode) {
  270. hash := md5.New()
  271. var body = io.TeeReader(dataReader, hash)
  272. proxyReq, err := http.NewRequest("PUT", uploadUrl, body)
  273. if err != nil {
  274. glog.Errorf("NewRequest %s: %v", uploadUrl, err)
  275. return "", s3err.ErrInternalError
  276. }
  277. proxyReq.Header.Set("Host", s3a.option.Filer)
  278. proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
  279. for header, values := range r.Header {
  280. for _, value := range values {
  281. proxyReq.Header.Add(header, value)
  282. }
  283. }
  284. resp, postErr := client.Do(proxyReq)
  285. if postErr != nil {
  286. glog.Errorf("post to filer: %v", postErr)
  287. return "", s3err.ErrInternalError
  288. }
  289. defer resp.Body.Close()
  290. etag = fmt.Sprintf("%x", hash.Sum(nil))
  291. resp_body, ra_err := ioutil.ReadAll(resp.Body)
  292. if ra_err != nil {
  293. glog.Errorf("upload to filer response read %d: %v", resp.StatusCode, ra_err)
  294. return etag, s3err.ErrInternalError
  295. }
  296. var ret weed_server.FilerPostResult
  297. unmarshal_err := json.Unmarshal(resp_body, &ret)
  298. if unmarshal_err != nil {
  299. glog.Errorf("failing to read upload to %s : %v", uploadUrl, string(resp_body))
  300. return "", s3err.ErrInternalError
  301. }
  302. if ret.Error != "" {
  303. glog.Errorf("upload to filer error: %v", ret.Error)
  304. return "", filerErrorToS3Error(ret.Error)
  305. }
  306. return etag, s3err.ErrNone
  307. }
  308. func setEtag(w http.ResponseWriter, etag string) {
  309. if etag != "" {
  310. if strings.HasPrefix(etag, "\"") {
  311. w.Header().Set("ETag", etag)
  312. } else {
  313. w.Header().Set("ETag", "\""+etag+"\"")
  314. }
  315. }
  316. }
  317. func getBucketAndObject(r *http.Request) (bucket, object string) {
  318. vars := mux.Vars(r)
  319. bucket = vars["bucket"]
  320. object = vars["object"]
  321. if !strings.HasPrefix(object, "/") {
  322. object = "/" + object
  323. }
  324. return
  325. }
  326. func filerErrorToS3Error(errString string) s3err.ErrorCode {
  327. if strings.HasPrefix(errString, "existing ") && strings.HasSuffix(errString, "is a directory") {
  328. return s3err.ErrExistingObjectIsDirectory
  329. }
  330. return s3err.ErrInternalError
  331. }