Implement AppendObject() API (#2082)

Author: harshavardhana

api-append-object.go

@@ -0,0 +1,226 @@
+/*
+ * MinIO Go Library for Amazon S3 Compatible Cloud Storage
+ * Copyright 2015-2025 MinIO, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package minio
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+
+	"github.com/minio/minio-go/v7/pkg/s3utils"
+)
+
+// AppendObjectOptions https://docs.aws.amazon.com/AmazonS3/latest/userguide/directory-buckets-objects-append.html
+type AppendObjectOptions struct {
+	// Provide a progress reader to indicate the current append() progress.
+	Progress io.Reader
+	// ChunkSize indicates the maximum append() size,
+	// it is useful when you want to control how much data
+	// per append() you are interested in sending to server
+	// while keeping the input io.Reader of a longer length.
+	ChunkSize uint64
+	// Aggressively disable sha256 payload, it is automatically
+	// turned-off for TLS supporting endpoints, useful in benchmarks
+	// where you are interested in the peak() numbers.
+	DisableContentSha256 bool
+
+	customHeaders http.Header
+	checksumType  ChecksumType
+}
+
+// Header returns the custom header for AppendObject API
+func (opts AppendObjectOptions) Header() (header http.Header) {
+	header = make(http.Header)
+	for k, v := range opts.customHeaders {
+		header[k] = v
+	}
+	return header
+}
+
+func (opts *AppendObjectOptions) setWriteOffset(offset int64) {
+	if len(opts.customHeaders) == 0 {
+		opts.customHeaders = make(http.Header)
+	}
+	opts.customHeaders["x-amz-write-offset-bytes"] = []string{strconv.FormatInt(offset, 10)}
+}
+
+func (opts *AppendObjectOptions) setChecksumParams(info ObjectInfo) {
+	if len(opts.customHeaders) == 0 {
+		opts.customHeaders = make(http.Header)
+	}
+	fullObject := info.ChecksumMode == ChecksumFullObjectMode.String()
+	switch {
+	case info.ChecksumCRC32 != "":
+		if fullObject {
+			opts.checksumType = ChecksumFullObjectCRC32
+		}
+	case info.ChecksumCRC32C != "":
+		if fullObject {
+			opts.checksumType = ChecksumFullObjectCRC32C
+		}
+	case info.ChecksumCRC64NVME != "":
+		// CRC64NVME only has a full object variant
+		// so it does not carry any special full object
+		// modifier
+		opts.checksumType = ChecksumCRC64NVME
+	}
+}
+
+func (opts AppendObjectOptions) validate(c *Client) (err error) {
+	if opts.ChunkSize > maxPartSize {
+		return errInvalidArgument("Append chunkSize cannot be larger than max part size allowed")
+	}
+	switch {
+	case !c.trailingHeaderSupport:
+		return errInvalidArgument("AppendObject() requires Client with TrailingHeaders enabled")
+	case c.overrideSignerType.IsV2():
+		return errInvalidArgument("AppendObject() cannot be used with v2 signatures")
+	case s3utils.IsGoogleEndpoint(*c.endpointURL):
+		return errInvalidArgument("AppendObject() cannot be used with GCS endpoints")
+	}
+
+	return nil
+}
+
+// appendObjectDo - executes the append object http operation.
+// NOTE: You must have WRITE permissions on a bucket to add an object to it.
+func (c *Client) appendObjectDo(ctx context.Context, bucketName, objectName string, reader io.Reader, size int64, opts AppendObjectOptions) (UploadInfo, error) {
+	// Input validation.
+	if err := s3utils.CheckValidBucketName(bucketName); err != nil {
+		return UploadInfo{}, err
+	}
+	if err := s3utils.CheckValidObjectName(objectName); err != nil {
+		return UploadInfo{}, err
+	}
+
+	// Set headers.
+	customHeader := opts.Header()
+
+	// Populate request metadata.
+	reqMetadata := requestMetadata{
+		bucketName:    bucketName,
+		objectName:    objectName,
+		customHeader:  customHeader,
+		contentBody:   reader,
+		contentLength: size,
+		streamSha256:  !opts.DisableContentSha256,
+	}
+
+	if opts.checksumType.IsSet() {
+		reqMetadata.addCrc = &opts.checksumType
+	}
+
+	// Execute PUT an objectName.
+	resp, err := c.executeMethod(ctx, http.MethodPut, reqMetadata)
+	defer closeResponse(resp)
+	if err != nil {
+		return UploadInfo{}, err
+	}
+	if resp != nil {
+		if resp.StatusCode != http.StatusOK {
+			return UploadInfo{}, httpRespToErrorResponse(resp, bucketName, objectName)
+		}
+	}
+
+	h := resp.Header
+
+	// When AppendObject() is used, S3 Express will return final object size as x-amz-object-size
+	if amzSize := h.Get("x-amz-object-size"); amzSize != "" {
+		size, err = strconv.ParseInt(amzSize, 10, 64)
+		if err != nil {
+			return UploadInfo{}, err
+		}
+	}
+
+	return UploadInfo{
+		Bucket: bucketName,
+		Key:    objectName,
+		ETag:   trimEtag(h.Get("ETag")),
+		Size:   size,
+
+		// Checksum values
+		ChecksumCRC32:     h.Get(ChecksumCRC32.Key()),
+		ChecksumCRC32C:    h.Get(ChecksumCRC32C.Key()),
+		ChecksumSHA1:      h.Get(ChecksumSHA1.Key()),
+		ChecksumSHA256:    h.Get(ChecksumSHA256.Key()),
+		ChecksumCRC64NVME: h.Get(ChecksumCRC64NVME.Key()),
+		ChecksumMode:      h.Get(ChecksumFullObjectMode.Key()),
+	}, nil
+}
+
+// AppendObject - S3 Express Zone https://docs.aws.amazon.com/AmazonS3/latest/userguide/directory-buckets-objects-append.html
+func (c *Client) AppendObject(ctx context.Context, bucketName, objectName string, reader io.Reader, objectSize int64,
+	opts AppendObjectOptions,
+) (info UploadInfo, err error) {
+	if objectSize < 0 && opts.ChunkSize == 0 {
+		return UploadInfo{}, errors.New("object size must be provided when no chunk size is provided")
+	}
+
+	if err = opts.validate(c); err != nil {
+		return UploadInfo{}, err
+	}
+
+	oinfo, err := c.StatObject(ctx, bucketName, objectName, StatObjectOptions{Checksum: true})
+	if err != nil {
+		return UploadInfo{}, err
+	}
+	if oinfo.ChecksumMode != ChecksumFullObjectMode.String() {
+		return UploadInfo{}, fmt.Errorf("append API is not allowed on objects that are not full_object checksum type: %s", oinfo.ChecksumMode)
+	}
+	opts.setChecksumParams(oinfo)   // set the appropriate checksum params based on the existing object checksum metadata.
+	opts.setWriteOffset(oinfo.Size) // First append must set the current object size as the offset.
+
+	if opts.ChunkSize > 0 {
+		finalObjSize := int64(-1)
+		if objectSize > 0 {
+			finalObjSize = info.Size + objectSize
+		}
+		totalPartsCount, partSize, lastPartSize, err := OptimalPartInfo(finalObjSize, opts.ChunkSize)
+		if err != nil {
+			return UploadInfo{}, err
+		}
+		buf := make([]byte, partSize)
+		var partNumber int
+		for partNumber = 1; partNumber <= totalPartsCount; partNumber++ {
+			// Proceed to upload the part.
+			if partNumber == totalPartsCount {
+				partSize = lastPartSize
+			}
+			n, err := readFull(reader, buf)
+			if err != nil {
+				return info, err
+			}
+			if n != int(partSize) {
+				return info, io.ErrUnexpectedEOF
+			}
+			rd := newHook(bytes.NewReader(buf[:n]), opts.Progress)
+			uinfo, err := c.appendObjectDo(ctx, bucketName, objectName, rd, partSize, opts)
+			if err != nil {
+				return info, err
+			}
+			opts.setWriteOffset(uinfo.Size)
+		}
+	}
+
+	rd := newHook(reader, opts.Progress)
+	return c.appendObjectDo(ctx, bucketName, objectName, rd, objectSize, opts)
+}

api-datatypes.go

@@ -148,6 +148,7 @@ type UploadInfo struct {
 	ChecksumSHA1      string
 	ChecksumSHA256    string
 	ChecksumCRC64NVME string
+	ChecksumMode      string
 }
 
 // RestoreInfo contains information of the restore operation of an archived object
@@ -223,6 +224,7 @@ type ObjectInfo struct {
 	ChecksumSHA1      string
 	ChecksumSHA256    string
 	ChecksumCRC64NVME string
+	ChecksumMode      string
 
 	Internal *struct {
 		K int // Data blocks

api-put-object-multipart.go

@@ -457,5 +457,6 @@ func (c *Client) completeMultipartUpload(ctx context.Context, bucketName, object
 		ChecksumCRC32:     completeMultipartUploadResult.ChecksumCRC32,
 		ChecksumCRC32C:    completeMultipartUploadResult.ChecksumCRC32C,
 		ChecksumCRC64NVME: completeMultipartUploadResult.ChecksumCRC64NVME,
+		ChecksumMode:      completeMultipartUploadResult.ChecksumType,
 	}, nil
 }

api-put-object-streaming.go

@@ -805,5 +805,6 @@ func (c *Client) putObjectDo(ctx context.Context, bucketName, objectName string,
 		ChecksumSHA1:      h.Get(ChecksumSHA1.Key()),
 		ChecksumSHA256:    h.Get(ChecksumSHA256.Key()),
 		ChecksumCRC64NVME: h.Get(ChecksumCRC64NVME.Key()),
+		ChecksumMode:      h.Get(ChecksumFullObjectMode.Key()),
 	}, nil
 }

api-s3-datatypes.go

@@ -366,6 +366,7 @@ type completeMultipartUploadResult struct {
 	ChecksumSHA1      string
 	ChecksumSHA256    string
 	ChecksumCRC64NVME string
+	ChecksumType      string
 }
 
 // CompletePart sub container lists individual part numbers and their

checksum.go

@@ -34,6 +34,43 @@ import (
 	"github.com/minio/crc64nvme"
 )
 
+// ChecksumMode contains information about the checksum mode on the object
+type ChecksumMode uint32
+
+const (
+	// ChecksumFullObjectMode Full object checksum `csumCombine(csum1, csum2...)...), csumN...)`
+	ChecksumFullObjectMode ChecksumMode = 1 << iota
+
+	// ChecksumCompositeMode Composite checksum `csum([csum1 + csum2 ... + csumN])`
+	ChecksumCompositeMode
+
+	// Keep after all valid checksums
+	checksumLastMode
+
+	// checksumModeMask is a mask for valid checksum mode types.
+	checksumModeMask = checksumLastMode - 1
+)
+
+// Is returns if c is all of t.
+func (c ChecksumMode) Is(t ChecksumMode) bool {
+	return c&t == t
+}
+
+// Key returns the header key.
+func (c ChecksumMode) Key() string {
+	return amzChecksumMode
+}
+
+func (c ChecksumMode) String() string {
+	switch c & checksumModeMask {
+	case ChecksumFullObjectMode:
+		return "FULL_OBJECT"
+	case ChecksumCompositeMode:
+		return "COMPOSITE"
+	}
+	return ""
+}
+
 // ChecksumType contains information about the checksum type.
 type ChecksumType uint32
 
@@ -75,6 +112,7 @@ const (
 	amzChecksumSHA1      = "x-amz-checksum-sha1"
 	amzChecksumSHA256    = "x-amz-checksum-sha256"
 	amzChecksumCRC64NVME = "x-amz-checksum-crc64nvme"
+	amzChecksumMode      = "x-amz-checksum-type"
 )
 
 // Base returns the base type, without modifiers.
@@ -397,7 +435,7 @@ func addAutoChecksumHeaders(opts *PutObjectOptions) {
 	}
 	opts.UserMetadata["X-Amz-Checksum-Algorithm"] = opts.AutoChecksum.String()
 	if opts.AutoChecksum.FullObjectRequested() {
-		opts.UserMetadata["X-Amz-Checksum-Type"] = "FULL_OBJECT"
+		opts.UserMetadata[amzChecksumMode] = ChecksumFullObjectMode.String()
 	}
 }
 
@@ -414,7 +452,10 @@ func applyAutoChecksum(opts *PutObjectOptions, allParts []ObjectPart) {
 	} else if opts.AutoChecksum.CanMergeCRC() {
 		crc, err := opts.AutoChecksum.FullObjectChecksum(allParts)
 		if err == nil {
-			opts.UserMetadata = map[string]string{opts.AutoChecksum.KeyCapitalized(): crc.Encoded(), "X-Amz-Checksum-Type": "FULL_OBJECT"}
+			opts.UserMetadata = map[string]string{
+				opts.AutoChecksum.KeyCapitalized(): crc.Encoded(),
+				amzChecksumMode:                    ChecksumFullObjectMode.String(),
+			}
 		}
 	}
 }

hook-reader.go

@@ -20,25 +20,20 @@ package minio
 import (
 	"fmt"
 	"io"
-	"sync"
 )
 
 // hookReader hooks additional reader in the source stream. It is
 // useful for making progress bars. Second reader is appropriately
 // notified about the exact number of bytes read from the primary
 // source on each Read operation.
 type hookReader struct {
-	mu     sync.RWMutex
 	source io.Reader
 	hook   io.Reader
 }
 
 // Seek implements io.Seeker. Seeks source first, and if necessary
 // seeks hook if Seek method is appropriately found.
 func (hr *hookReader) Seek(offset int64, whence int) (n int64, err error) {
-	hr.mu.Lock()
-	defer hr.mu.Unlock()
-
 	// Verify for source has embedded Seeker, use it.
 	sourceSeeker, ok := hr.source.(io.Seeker)
 	if ok {
@@ -70,9 +65,6 @@ func (hr *hookReader) Seek(offset int64, whence int) (n int64, err error) {
 // value 'n' number of bytes are reported through the hook. Returns
 // error for all non io.EOF conditions.
 func (hr *hookReader) Read(b []byte) (n int, err error) {
-	hr.mu.RLock()
-	defer hr.mu.RUnlock()
-
 	n, err = hr.source.Read(b)
 	if err != nil && err != io.EOF {
 		return n, err
@@ -92,7 +84,7 @@ func (hr *hookReader) Read(b []byte) (n int, err error) {
 // reports the data read from the source to the hook.
 func newHook(source, hook io.Reader) io.Reader {
 	if hook == nil {
-		return &hookReader{source: source}
+		return source
 	}
 	return &hookReader{
 		source: source,

utils.go

@@ -390,6 +390,7 @@ func ToObjectInfo(bucketName, objectName string, h http.Header) (ObjectInfo, err
 		ChecksumSHA1:      h.Get(ChecksumSHA1.Key()),
 		ChecksumSHA256:    h.Get(ChecksumSHA256.Key()),
 		ChecksumCRC64NVME: h.Get(ChecksumCRC64NVME.Key()),
+		ChecksumMode:      h.Get(ChecksumFullObjectMode.Key()),
 	}, nil
 }