filer_multipart.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366
  1. package s3api
  2. import (
  3. "encoding/hex"
  4. "encoding/xml"
  5. "fmt"
  6. "github.com/google/uuid"
  7. "github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
  8. "golang.org/x/exp/slices"
  9. "math"
  10. "path/filepath"
  11. "sort"
  12. "strconv"
  13. "strings"
  14. "time"
  15. "github.com/aws/aws-sdk-go/aws"
  16. "github.com/aws/aws-sdk-go/service/s3"
  17. "github.com/seaweedfs/seaweedfs/weed/filer"
  18. "github.com/seaweedfs/seaweedfs/weed/glog"
  19. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  20. )
  21. type InitiateMultipartUploadResult struct {
  22. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ InitiateMultipartUploadResult"`
  23. s3.CreateMultipartUploadOutput
  24. }
  25. func (s3a *S3ApiServer) createMultipartUpload(input *s3.CreateMultipartUploadInput) (output *InitiateMultipartUploadResult, code s3err.ErrorCode) {
  26. glog.V(2).Infof("createMultipartUpload input %v", input)
  27. uploadIdString := s3a.generateUploadID(*input.Key)
  28. uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.New().String(), "-", "")
  29. if err := s3a.mkdir(s3a.genUploadsFolder(*input.Bucket), uploadIdString, func(entry *filer_pb.Entry) {
  30. if entry.Extended == nil {
  31. entry.Extended = make(map[string][]byte)
  32. }
  33. entry.Extended["key"] = []byte(*input.Key)
  34. for k, v := range input.Metadata {
  35. entry.Extended[k] = []byte(*v)
  36. }
  37. if input.ContentType != nil {
  38. entry.Attributes.Mime = *input.ContentType
  39. }
  40. }); err != nil {
  41. glog.Errorf("NewMultipartUpload error: %v", err)
  42. return nil, s3err.ErrInternalError
  43. }
  44. output = &InitiateMultipartUploadResult{
  45. CreateMultipartUploadOutput: s3.CreateMultipartUploadOutput{
  46. Bucket: input.Bucket,
  47. Key: objectKey(input.Key),
  48. UploadId: aws.String(uploadIdString),
  49. },
  50. }
  51. return
  52. }
  53. type CompleteMultipartUploadResult struct {
  54. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CompleteMultipartUploadResult"`
  55. s3.CompleteMultipartUploadOutput
  56. }
  57. func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) {
  58. glog.V(2).Infof("completeMultipartUpload input %v", input)
  59. completedParts := parts.Parts
  60. slices.SortFunc(completedParts, func(a, b CompletedPart) int {
  61. return a.PartNumber - b.PartNumber
  62. })
  63. uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
  64. entries, _, err := s3a.list(uploadDirectory, "", "", false, maxPartsList)
  65. if err != nil || len(entries) == 0 {
  66. glog.Errorf("completeMultipartUpload %s %s error: %v, entries:%d", *input.Bucket, *input.UploadId, err, len(entries))
  67. return nil, s3err.ErrNoSuchUpload
  68. }
  69. pentry, err := s3a.getEntry(s3a.genUploadsFolder(*input.Bucket), *input.UploadId)
  70. if err != nil {
  71. glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
  72. return nil, s3err.ErrNoSuchUpload
  73. }
  74. // check whether completedParts is more than received parts
  75. {
  76. partNumbers := make(map[int]struct{}, len(entries))
  77. for _, entry := range entries {
  78. if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
  79. partNumberString := entry.Name[:len(entry.Name)-len(".part")]
  80. partNumber, err := strconv.Atoi(partNumberString)
  81. if err == nil {
  82. partNumbers[partNumber] = struct{}{}
  83. }
  84. }
  85. }
  86. for _, part := range completedParts {
  87. if _, found := partNumbers[part.PartNumber]; !found {
  88. return nil, s3err.ErrInvalidPart
  89. }
  90. }
  91. }
  92. mime := pentry.Attributes.Mime
  93. var finalParts []*filer_pb.FileChunk
  94. var offset int64
  95. for _, entry := range entries {
  96. if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
  97. partETag, found := findByPartNumber(entry.Name, completedParts)
  98. if !found {
  99. continue
  100. }
  101. entryETag := hex.EncodeToString(entry.Attributes.GetMd5())
  102. if partETag != "" && len(partETag) == 32 && entryETag != "" && entryETag != partETag {
  103. glog.Errorf("completeMultipartUpload %s ETag mismatch chunk: %s part: %s", entry.Name, entryETag, partETag)
  104. return nil, s3err.ErrInvalidPart
  105. }
  106. for _, chunk := range entry.GetChunks() {
  107. p := &filer_pb.FileChunk{
  108. FileId: chunk.GetFileIdString(),
  109. Offset: offset,
  110. Size: chunk.Size,
  111. ModifiedTsNs: chunk.ModifiedTsNs,
  112. CipherKey: chunk.CipherKey,
  113. ETag: chunk.ETag,
  114. }
  115. finalParts = append(finalParts, p)
  116. offset += int64(chunk.Size)
  117. }
  118. }
  119. }
  120. entryName := filepath.Base(*input.Key)
  121. dirName := filepath.ToSlash(filepath.Dir(*input.Key))
  122. if dirName == "." {
  123. dirName = ""
  124. }
  125. if strings.HasPrefix(dirName, "/") {
  126. dirName = dirName[1:]
  127. }
  128. dirName = fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, *input.Bucket, dirName)
  129. // remove suffix '/'
  130. if strings.HasSuffix(dirName, "/") {
  131. dirName = dirName[:len(dirName)-1]
  132. }
  133. err = s3a.mkFile(dirName, entryName, finalParts, func(entry *filer_pb.Entry) {
  134. if entry.Extended == nil {
  135. entry.Extended = make(map[string][]byte)
  136. }
  137. for k, v := range pentry.Extended {
  138. if k != "key" {
  139. entry.Extended[k] = v
  140. }
  141. }
  142. if pentry.Attributes.Mime != "" {
  143. entry.Attributes.Mime = pentry.Attributes.Mime
  144. } else if mime != "" {
  145. entry.Attributes.Mime = mime
  146. }
  147. entry.Attributes.FileSize = uint64(offset)
  148. })
  149. if err != nil {
  150. glog.Errorf("completeMultipartUpload %s/%s error: %v", dirName, entryName, err)
  151. return nil, s3err.ErrInternalError
  152. }
  153. output = &CompleteMultipartUploadResult{
  154. CompleteMultipartUploadOutput: s3.CompleteMultipartUploadOutput{
  155. Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer.ToHttpAddress(), urlEscapeObject(dirName), urlPathEscape(entryName))),
  156. Bucket: input.Bucket,
  157. ETag: aws.String("\"" + filer.ETagChunks(finalParts) + "\""),
  158. Key: objectKey(input.Key),
  159. },
  160. }
  161. if err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
  162. glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
  163. }
  164. return
  165. }
  166. func findByPartNumber(fileName string, parts []CompletedPart) (etag string, found bool) {
  167. partNumber, formatErr := strconv.Atoi(fileName[:4])
  168. if formatErr != nil {
  169. return
  170. }
  171. x := sort.Search(len(parts), func(i int) bool {
  172. return parts[i].PartNumber >= partNumber
  173. })
  174. if x >= len(parts) {
  175. return
  176. }
  177. if parts[x].PartNumber != partNumber {
  178. return
  179. }
  180. y := 0
  181. for i, part := range parts[x:] {
  182. if part.PartNumber == partNumber {
  183. y = i
  184. } else {
  185. break
  186. }
  187. }
  188. return parts[x+y].ETag, true
  189. }
  190. func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code s3err.ErrorCode) {
  191. glog.V(2).Infof("abortMultipartUpload input %v", input)
  192. exists, err := s3a.exists(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true)
  193. if err != nil {
  194. glog.V(1).Infof("bucket %s abort upload %s: %v", *input.Bucket, *input.UploadId, err)
  195. return nil, s3err.ErrNoSuchUpload
  196. }
  197. if exists {
  198. err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true)
  199. }
  200. if err != nil {
  201. glog.V(1).Infof("bucket %s remove upload %s: %v", *input.Bucket, *input.UploadId, err)
  202. return nil, s3err.ErrInternalError
  203. }
  204. return &s3.AbortMultipartUploadOutput{}, s3err.ErrNone
  205. }
  206. type ListMultipartUploadsResult struct {
  207. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListMultipartUploadsResult"`
  208. // copied from s3.ListMultipartUploadsOutput, the Uploads is not converting to <Upload></Upload>
  209. Bucket *string `type:"string"`
  210. Delimiter *string `type:"string"`
  211. EncodingType *string `type:"string" enum:"EncodingType"`
  212. IsTruncated *bool `type:"boolean"`
  213. KeyMarker *string `type:"string"`
  214. MaxUploads *int64 `type:"integer"`
  215. NextKeyMarker *string `type:"string"`
  216. NextUploadIdMarker *string `type:"string"`
  217. Prefix *string `type:"string"`
  218. UploadIdMarker *string `type:"string"`
  219. Upload []*s3.MultipartUpload `locationName:"Upload" type:"list" flattened:"true"`
  220. }
  221. func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput) (output *ListMultipartUploadsResult, code s3err.ErrorCode) {
  222. // https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListMultipartUploads.html
  223. glog.V(2).Infof("listMultipartUploads input %v", input)
  224. output = &ListMultipartUploadsResult{
  225. Bucket: input.Bucket,
  226. Delimiter: input.Delimiter,
  227. EncodingType: input.EncodingType,
  228. KeyMarker: input.KeyMarker,
  229. MaxUploads: input.MaxUploads,
  230. Prefix: input.Prefix,
  231. IsTruncated: aws.Bool(false),
  232. }
  233. entries, _, err := s3a.list(s3a.genUploadsFolder(*input.Bucket), "", *input.UploadIdMarker, false, math.MaxInt32)
  234. if err != nil {
  235. glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err)
  236. return
  237. }
  238. uploadsCount := int64(0)
  239. for _, entry := range entries {
  240. if entry.Extended != nil {
  241. key := string(entry.Extended["key"])
  242. if *input.KeyMarker != "" && *input.KeyMarker != key {
  243. continue
  244. }
  245. if *input.Prefix != "" && !strings.HasPrefix(key, *input.Prefix) {
  246. continue
  247. }
  248. output.Upload = append(output.Upload, &s3.MultipartUpload{
  249. Key: objectKey(aws.String(key)),
  250. UploadId: aws.String(entry.Name),
  251. })
  252. uploadsCount += 1
  253. }
  254. if uploadsCount >= *input.MaxUploads {
  255. output.IsTruncated = aws.Bool(true)
  256. output.NextUploadIdMarker = aws.String(entry.Name)
  257. break
  258. }
  259. }
  260. return
  261. }
  262. type ListPartsResult struct {
  263. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListPartsResult"`
  264. // copied from s3.ListPartsOutput, the Parts is not converting to <Part></Part>
  265. Bucket *string `type:"string"`
  266. IsTruncated *bool `type:"boolean"`
  267. Key *string `min:"1" type:"string"`
  268. MaxParts *int64 `type:"integer"`
  269. NextPartNumberMarker *int64 `type:"integer"`
  270. PartNumberMarker *int64 `type:"integer"`
  271. Part []*s3.Part `locationName:"Part" type:"list" flattened:"true"`
  272. StorageClass *string `type:"string" enum:"StorageClass"`
  273. UploadId *string `type:"string"`
  274. }
  275. func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListPartsResult, code s3err.ErrorCode) {
  276. // https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListParts.html
  277. glog.V(2).Infof("listObjectParts input %v", input)
  278. output = &ListPartsResult{
  279. Bucket: input.Bucket,
  280. Key: objectKey(input.Key),
  281. UploadId: input.UploadId,
  282. MaxParts: input.MaxParts, // the maximum number of parts to return.
  283. PartNumberMarker: input.PartNumberMarker, // the part number starts after this, exclusive
  284. StorageClass: aws.String("STANDARD"),
  285. }
  286. entries, isLast, err := s3a.list(s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId, "", fmt.Sprintf("%04d.part", *input.PartNumberMarker), false, uint32(*input.MaxParts))
  287. if err != nil {
  288. glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err)
  289. return nil, s3err.ErrNoSuchUpload
  290. }
  291. // Note: The upload directory is sort of a marker of the existence of an multipart upload request.
  292. // So can not just delete empty upload folders.
  293. output.IsTruncated = aws.Bool(!isLast)
  294. for _, entry := range entries {
  295. if strings.HasSuffix(entry.Name, ".part") && !entry.IsDirectory {
  296. partNumberString := entry.Name[:len(entry.Name)-len(".part")]
  297. partNumber, err := strconv.Atoi(partNumberString)
  298. if err != nil {
  299. glog.Errorf("listObjectParts %s %s parse %s: %v", *input.Bucket, *input.UploadId, entry.Name, err)
  300. continue
  301. }
  302. output.Part = append(output.Part, &s3.Part{
  303. PartNumber: aws.Int64(int64(partNumber)),
  304. LastModified: aws.Time(time.Unix(entry.Attributes.Mtime, 0).UTC()),
  305. Size: aws.Int64(int64(filer.FileSize(entry))),
  306. ETag: aws.String("\"" + filer.ETag(entry) + "\""),
  307. })
  308. if !isLast {
  309. output.NextPartNumberMarker = aws.Int64(int64(partNumber))
  310. }
  311. }
  312. }
  313. return
  314. }