api-put-object.go 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258
  1. /*
  2. * Minio Go Library for Amazon S3 Compatible Cloud Storage
  3. * Copyright 2015-2017 Minio, Inc.
  4. *
  5. * Licensed under the Apache License, Version 2.0 (the "License");
  6. * you may not use this file except in compliance with the License.
  7. * You may obtain a copy of the License at
  8. *
  9. * http://www.apache.org/licenses/LICENSE-2.0
  10. *
  11. * Unless required by applicable law or agreed to in writing, software
  12. * distributed under the License is distributed on an "AS IS" BASIS,
  13. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. * See the License for the specific language governing permissions and
  15. * limitations under the License.
  16. */
  17. package minio
  18. import (
  19. "bytes"
  20. "context"
  21. "fmt"
  22. "io"
  23. "net/http"
  24. "runtime/debug"
  25. "sort"
  26. "github.com/minio/minio-go/pkg/encrypt"
  27. "github.com/minio/minio-go/pkg/s3utils"
  28. )
  29. // PutObjectOptions represents options specified by user for PutObject call
  30. type PutObjectOptions struct {
  31. UserMetadata map[string]string
  32. Progress io.Reader
  33. ContentType string
  34. ContentEncoding string
  35. ContentDisposition string
  36. CacheControl string
  37. EncryptMaterials encrypt.Materials
  38. NumThreads uint
  39. StorageClass string
  40. }
  41. // getNumThreads - gets the number of threads to be used in the multipart
  42. // put object operation
  43. func (opts PutObjectOptions) getNumThreads() (numThreads int) {
  44. if opts.NumThreads > 0 {
  45. numThreads = int(opts.NumThreads)
  46. } else {
  47. numThreads = totalWorkers
  48. }
  49. return
  50. }
  51. // Header - constructs the headers from metadata entered by user in
  52. // PutObjectOptions struct
  53. func (opts PutObjectOptions) Header() (header http.Header) {
  54. header = make(http.Header)
  55. if opts.ContentType != "" {
  56. header["Content-Type"] = []string{opts.ContentType}
  57. } else {
  58. header["Content-Type"] = []string{"application/octet-stream"}
  59. }
  60. if opts.ContentEncoding != "" {
  61. header["Content-Encoding"] = []string{opts.ContentEncoding}
  62. }
  63. if opts.ContentDisposition != "" {
  64. header["Content-Disposition"] = []string{opts.ContentDisposition}
  65. }
  66. if opts.CacheControl != "" {
  67. header["Cache-Control"] = []string{opts.CacheControl}
  68. }
  69. if opts.EncryptMaterials != nil {
  70. header[amzHeaderIV] = []string{opts.EncryptMaterials.GetIV()}
  71. header[amzHeaderKey] = []string{opts.EncryptMaterials.GetKey()}
  72. header[amzHeaderMatDesc] = []string{opts.EncryptMaterials.GetDesc()}
  73. }
  74. if opts.StorageClass != "" {
  75. header[amzStorageClass] = []string{opts.StorageClass}
  76. }
  77. for k, v := range opts.UserMetadata {
  78. if !isAmzHeader(k) && !isStandardHeader(k) && !isSSEHeader(k) && !isStorageClassHeader(k) {
  79. header["X-Amz-Meta-"+k] = []string{v}
  80. } else {
  81. header[k] = []string{v}
  82. }
  83. }
  84. return
  85. }
  86. // validate() checks if the UserMetadata map has standard headers or client side
  87. // encryption headers and raises an error if so.
  88. func (opts PutObjectOptions) validate() (err error) {
  89. for k := range opts.UserMetadata {
  90. if isStandardHeader(k) || isCSEHeader(k) || isStorageClassHeader(k) {
  91. return ErrInvalidArgument(k + " unsupported request parameter for user defined metadata from minio-go")
  92. }
  93. }
  94. return nil
  95. }
  96. // completedParts is a collection of parts sortable by their part numbers.
  97. // used for sorting the uploaded parts before completing the multipart request.
  98. type completedParts []CompletePart
  99. func (a completedParts) Len() int { return len(a) }
  100. func (a completedParts) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
  101. func (a completedParts) Less(i, j int) bool { return a[i].PartNumber < a[j].PartNumber }
  102. // PutObject creates an object in a bucket.
  103. //
  104. // You must have WRITE permissions on a bucket to create an object.
  105. //
  106. // - For size smaller than 64MiB PutObject automatically does a
  107. // single atomic Put operation.
  108. // - For size larger than 64MiB PutObject automatically does a
  109. // multipart Put operation.
  110. // - For size input as -1 PutObject does a multipart Put operation
  111. // until input stream reaches EOF. Maximum object size that can
  112. // be uploaded through this operation will be 5TiB.
  113. func (c Client) PutObject(bucketName, objectName string, reader io.Reader, objectSize int64,
  114. opts PutObjectOptions) (n int64, err error) {
  115. return c.PutObjectWithContext(context.Background(), bucketName, objectName, reader, objectSize, opts)
  116. }
  117. func (c Client) putObjectCommon(ctx context.Context, bucketName, objectName string, reader io.Reader, size int64, opts PutObjectOptions) (n int64, err error) {
  118. // Check for largest object size allowed.
  119. if size > int64(maxMultipartPutObjectSize) {
  120. return 0, ErrEntityTooLarge(size, maxMultipartPutObjectSize, bucketName, objectName)
  121. }
  122. // NOTE: Streaming signature is not supported by GCS.
  123. if s3utils.IsGoogleEndpoint(c.endpointURL) {
  124. // Do not compute MD5 for Google Cloud Storage.
  125. return c.putObjectNoChecksum(ctx, bucketName, objectName, reader, size, opts)
  126. }
  127. if c.overrideSignerType.IsV2() {
  128. if size >= 0 && size < minPartSize {
  129. return c.putObjectNoChecksum(ctx, bucketName, objectName, reader, size, opts)
  130. }
  131. return c.putObjectMultipart(ctx, bucketName, objectName, reader, size, opts)
  132. }
  133. if size < 0 {
  134. return c.putObjectMultipartStreamNoLength(ctx, bucketName, objectName, reader, opts)
  135. }
  136. if size < minPartSize {
  137. return c.putObjectNoChecksum(ctx, bucketName, objectName, reader, size, opts)
  138. }
  139. // For all sizes greater than 64MiB do multipart.
  140. return c.putObjectMultipartStream(ctx, bucketName, objectName, reader, size, opts)
  141. }
  142. func (c Client) putObjectMultipartStreamNoLength(ctx context.Context, bucketName, objectName string, reader io.Reader, opts PutObjectOptions) (n int64, err error) {
  143. // Input validation.
  144. if err = s3utils.CheckValidBucketName(bucketName); err != nil {
  145. return 0, err
  146. }
  147. if err = s3utils.CheckValidObjectName(objectName); err != nil {
  148. return 0, err
  149. }
  150. // Total data read and written to server. should be equal to
  151. // 'size' at the end of the call.
  152. var totalUploadedSize int64
  153. // Complete multipart upload.
  154. var complMultipartUpload completeMultipartUpload
  155. // Calculate the optimal parts info for a given size.
  156. totalPartsCount, partSize, _, err := optimalPartInfo(-1)
  157. if err != nil {
  158. return 0, err
  159. }
  160. // Initiate a new multipart upload.
  161. uploadID, err := c.newUploadID(ctx, bucketName, objectName, opts)
  162. if err != nil {
  163. return 0, err
  164. }
  165. defer func() {
  166. if err != nil {
  167. c.abortMultipartUpload(ctx, bucketName, objectName, uploadID)
  168. }
  169. }()
  170. // Part number always starts with '1'.
  171. partNumber := 1
  172. // Initialize parts uploaded map.
  173. partsInfo := make(map[int]ObjectPart)
  174. // Create a buffer.
  175. buf := make([]byte, partSize)
  176. defer debug.FreeOSMemory()
  177. for partNumber <= totalPartsCount {
  178. length, rErr := io.ReadFull(reader, buf)
  179. if rErr == io.EOF && partNumber > 1 {
  180. break
  181. }
  182. if rErr != nil && rErr != io.ErrUnexpectedEOF {
  183. return 0, rErr
  184. }
  185. // Update progress reader appropriately to the latest offset
  186. // as we read from the source.
  187. rd := newHook(bytes.NewReader(buf[:length]), opts.Progress)
  188. // Proceed to upload the part.
  189. var objPart ObjectPart
  190. objPart, err = c.uploadPart(ctx, bucketName, objectName, uploadID, rd, partNumber,
  191. "", "", int64(length), opts.UserMetadata)
  192. if err != nil {
  193. return totalUploadedSize, err
  194. }
  195. // Save successfully uploaded part metadata.
  196. partsInfo[partNumber] = objPart
  197. // Save successfully uploaded size.
  198. totalUploadedSize += int64(length)
  199. // Increment part number.
  200. partNumber++
  201. // For unknown size, Read EOF we break away.
  202. // We do not have to upload till totalPartsCount.
  203. if rErr == io.EOF {
  204. break
  205. }
  206. }
  207. // Loop over total uploaded parts to save them in
  208. // Parts array before completing the multipart request.
  209. for i := 1; i < partNumber; i++ {
  210. part, ok := partsInfo[i]
  211. if !ok {
  212. return 0, ErrInvalidArgument(fmt.Sprintf("Missing part number %d", i))
  213. }
  214. complMultipartUpload.Parts = append(complMultipartUpload.Parts, CompletePart{
  215. ETag: part.ETag,
  216. PartNumber: part.PartNumber,
  217. })
  218. }
  219. // Sort all completed parts.
  220. sort.Sort(completedParts(complMultipartUpload.Parts))
  221. if _, err = c.completeMultipartUpload(ctx, bucketName, objectName, uploadID, complMultipartUpload); err != nil {
  222. return totalUploadedSize, err
  223. }
  224. // Return final size.
  225. return totalUploadedSize, nil
  226. }