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.

213 lines
6.7 KiB

6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
  1. package s3api
  2. import (
  3. "context"
  4. "encoding/xml"
  5. "fmt"
  6. "path/filepath"
  7. "strconv"
  8. "strings"
  9. "time"
  10. "github.com/aws/aws-sdk-go/aws"
  11. "github.com/aws/aws-sdk-go/service/s3"
  12. "github.com/chrislusf/seaweedfs/weed/filer2"
  13. "github.com/chrislusf/seaweedfs/weed/glog"
  14. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  15. "github.com/satori/go.uuid"
  16. )
  17. type InitiateMultipartUploadResult struct {
  18. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ InitiateMultipartUploadResult"`
  19. s3.CreateMultipartUploadOutput
  20. }
  21. func (s3a *S3ApiServer) createMultipartUpload(ctx context.Context, input *s3.CreateMultipartUploadInput) (output *InitiateMultipartUploadResult, code ErrorCode) {
  22. uploadId, _ := uuid.NewV4()
  23. uploadIdString := uploadId.String()
  24. if err := s3a.mkdir(ctx, s3a.genUploadsFolder(*input.Bucket), uploadIdString, func(entry *filer_pb.Entry) {
  25. if entry.Extended == nil {
  26. entry.Extended = make(map[string][]byte)
  27. }
  28. entry.Extended["key"] = []byte(*input.Key)
  29. }); err != nil {
  30. glog.Errorf("NewMultipartUpload error: %v", err)
  31. return nil, ErrInternalError
  32. }
  33. output = &InitiateMultipartUploadResult{
  34. CreateMultipartUploadOutput: s3.CreateMultipartUploadOutput{
  35. Bucket: input.Bucket,
  36. Key: objectKey(input.Key),
  37. UploadId: aws.String(uploadIdString),
  38. },
  39. }
  40. return
  41. }
  42. type CompleteMultipartUploadResult struct {
  43. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CompleteMultipartUploadResult"`
  44. s3.CompleteMultipartUploadOutput
  45. }
  46. func (s3a *S3ApiServer) completeMultipartUpload(ctx context.Context, input *s3.CompleteMultipartUploadInput) (output *CompleteMultipartUploadResult, code ErrorCode) {
  47. uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
  48. entries, err := s3a.list(ctx, uploadDirectory, "", "", false, 0)
  49. if err != nil {
  50. glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
  51. return nil, ErrNoSuchUpload
  52. }
  53. var finalParts []*filer_pb.FileChunk
  54. var offset int64
  55. for _, entry := range entries {
  56. if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
  57. for _, chunk := range entry.Chunks {
  58. p := &filer_pb.FileChunk{
  59. FileId: chunk.GetFileIdString(),
  60. Offset: offset,
  61. Size: chunk.Size,
  62. Mtime: chunk.Mtime,
  63. ETag: chunk.ETag,
  64. }
  65. finalParts = append(finalParts, p)
  66. offset += int64(chunk.Size)
  67. }
  68. }
  69. }
  70. entryName := filepath.Base(*input.Key)
  71. dirName := filepath.Dir(*input.Key)
  72. if dirName == "." {
  73. dirName = ""
  74. }
  75. if strings.HasPrefix(dirName, "/") {
  76. dirName = dirName[1:]
  77. }
  78. dirName = fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, *input.Bucket, dirName)
  79. err = s3a.mkFile(ctx, dirName, entryName, finalParts)
  80. if err != nil {
  81. glog.Errorf("completeMultipartUpload %s/%s error: %v", dirName, entryName, err)
  82. return nil, ErrInternalError
  83. }
  84. output = &CompleteMultipartUploadResult{
  85. CompleteMultipartUploadOutput: s3.CompleteMultipartUploadOutput{
  86. Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer, dirName, entryName)),
  87. Bucket: input.Bucket,
  88. ETag: aws.String("\"" + filer2.ETag(finalParts) + "\""),
  89. Key: objectKey(input.Key),
  90. },
  91. }
  92. if err = s3a.rm(ctx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, false, true); err != nil {
  93. glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
  94. }
  95. return
  96. }
  97. func (s3a *S3ApiServer) abortMultipartUpload(ctx context.Context, input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code ErrorCode) {
  98. exists, err := s3a.exists(ctx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true)
  99. if err != nil {
  100. glog.V(1).Infof("bucket %s abort upload %s: %v", *input.Bucket, *input.UploadId, err)
  101. return nil, ErrNoSuchUpload
  102. }
  103. if exists {
  104. err = s3a.rm(ctx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true, true)
  105. }
  106. if err != nil {
  107. glog.V(1).Infof("bucket %s remove upload %s: %v", *input.Bucket, *input.UploadId, err)
  108. return nil, ErrInternalError
  109. }
  110. return &s3.AbortMultipartUploadOutput{}, ErrNone
  111. }
  112. type ListMultipartUploadsResult struct {
  113. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListMultipartUploadsResult"`
  114. s3.ListMultipartUploadsOutput
  115. }
  116. func (s3a *S3ApiServer) listMultipartUploads(ctx context.Context, input *s3.ListMultipartUploadsInput) (output *ListMultipartUploadsResult, code ErrorCode) {
  117. output = &ListMultipartUploadsResult{
  118. ListMultipartUploadsOutput: s3.ListMultipartUploadsOutput{
  119. Bucket: input.Bucket,
  120. Delimiter: input.Delimiter,
  121. EncodingType: input.EncodingType,
  122. KeyMarker: input.KeyMarker,
  123. MaxUploads: input.MaxUploads,
  124. Prefix: input.Prefix,
  125. },
  126. }
  127. entries, err := s3a.list(ctx, s3a.genUploadsFolder(*input.Bucket), *input.Prefix, *input.KeyMarker, true, int(*input.MaxUploads))
  128. if err != nil {
  129. glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err)
  130. return
  131. }
  132. for _, entry := range entries {
  133. if entry.Extended != nil {
  134. key := entry.Extended["key"]
  135. output.Uploads = append(output.Uploads, &s3.MultipartUpload{
  136. Key: objectKey(aws.String(string(key))),
  137. UploadId: aws.String(entry.Name),
  138. })
  139. }
  140. }
  141. return
  142. }
  143. type ListPartsResult struct {
  144. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListPartsResult"`
  145. s3.ListPartsOutput
  146. }
  147. func (s3a *S3ApiServer) listObjectParts(ctx context.Context, input *s3.ListPartsInput) (output *ListPartsResult, code ErrorCode) {
  148. output = &ListPartsResult{
  149. ListPartsOutput: s3.ListPartsOutput{
  150. Bucket: input.Bucket,
  151. Key: objectKey(input.Key),
  152. UploadId: input.UploadId,
  153. MaxParts: input.MaxParts, // the maximum number of parts to return.
  154. PartNumberMarker: input.PartNumberMarker, // the part number starts after this, exclusive
  155. },
  156. }
  157. entries, err := s3a.list(ctx, s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId,
  158. "", fmt.Sprintf("%04d.part", *input.PartNumberMarker), false, int(*input.MaxParts))
  159. if err != nil {
  160. glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err)
  161. return nil, ErrNoSuchUpload
  162. }
  163. for _, entry := range entries {
  164. if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
  165. partNumberString := entry.Name[:len(entry.Name)-len(".part")]
  166. partNumber, err := strconv.Atoi(partNumberString)
  167. if err != nil {
  168. glog.Errorf("listObjectParts %s %s parse %s: %v", *input.Bucket, *input.UploadId, entry.Name, err)
  169. continue
  170. }
  171. output.Parts = append(output.Parts, &s3.Part{
  172. PartNumber: aws.Int64(int64(partNumber)),
  173. LastModified: aws.Time(time.Unix(entry.Attributes.Mtime, 0)),
  174. Size: aws.Int64(int64(filer2.TotalSize(entry.Chunks))),
  175. ETag: aws.String("\"" + filer2.ETag(entry.Chunks) + "\""),
  176. })
  177. }
  178. }
  179. return
  180. }