mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 18:15:51 +00:00
2291 lines
69 KiB
Go
2291 lines
69 KiB
Go
// Copyright 2019 The CubeFS Authors.
|
|
//
|
|
// 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 objectnode
|
|
|
|
import (
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/xml"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"regexp"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"github.com/cubefs/cubefs/blobstore/common/trace"
|
|
"github.com/cubefs/cubefs/proto"
|
|
"github.com/cubefs/cubefs/util/log"
|
|
)
|
|
|
|
var (
|
|
rangeRegexp = regexp.MustCompile(`^bytes=(\d)*-(\d)*$`)
|
|
MaxKeyLength = 750
|
|
)
|
|
|
|
// Get object
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html
|
|
func (o *ObjectNode) getObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("getObjectHandler: load volume fail: requestID(%v) volume(%v) path(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// parse http range option
|
|
var (
|
|
revRange bool
|
|
isRangeRead bool
|
|
rangeLower uint64
|
|
rangeUpper uint64
|
|
partSize uint64
|
|
partCount uint64
|
|
)
|
|
rangeOpt := strings.TrimSpace(r.Header.Get(Range))
|
|
if len(rangeOpt) > 0 {
|
|
if !rangeRegexp.MatchString(rangeOpt) {
|
|
log.LogErrorf("getObjectHandler: invalid range header: requestID(%v) volume(%v) path(%v) rangeOpt(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), rangeOpt)
|
|
errorCode = InvalidRange
|
|
return
|
|
}
|
|
hyphenIndex := strings.Index(rangeOpt, "-")
|
|
if hyphenIndex < 0 {
|
|
log.LogErrorf("getObjectHandler: invalid range header: requestID(%v) volume(%v) path(%v) rangeOpt(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), rangeOpt)
|
|
errorCode = InvalidRange
|
|
return
|
|
}
|
|
lowerPart := rangeOpt[len("bytes="):hyphenIndex]
|
|
upperPart := ""
|
|
if hyphenIndex+1 < len(rangeOpt) {
|
|
// bytes=-5
|
|
if hyphenIndex == len("bytes=") { // suffix range opt
|
|
revRange = true
|
|
rangeUpper = 1<<64 - 1
|
|
lowerPart = rangeOpt[hyphenIndex+1:]
|
|
} else { // bytes=1-10
|
|
upperPart = rangeOpt[hyphenIndex+1:]
|
|
}
|
|
} else if hyphenIndex+1 == len(rangeOpt) { // bytes=1-
|
|
rangeUpper = 1<<64 - 1
|
|
}
|
|
if len(lowerPart) > 0 {
|
|
if rangeLower, err = strconv.ParseUint(lowerPart, 10, 64); err != nil {
|
|
log.LogErrorf("getObjectHandler: parse range lower fail: requestID(%v) volume(%v) path(%v) rangeOpt(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), rangeOpt, err)
|
|
errorCode = InvalidRange
|
|
return
|
|
}
|
|
}
|
|
if len(upperPart) > 0 {
|
|
if rangeUpper, err = strconv.ParseUint(upperPart, 10, 64); err != nil {
|
|
log.LogErrorf("getObjectHandler: parse range upper fail: requestID(%v) volume(%v) path(%v) rangeOpt(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), rangeOpt, err)
|
|
errorCode = InvalidRange
|
|
return
|
|
}
|
|
}
|
|
if rangeUpper < rangeLower {
|
|
// upper enabled and lower than lower side
|
|
log.LogErrorf("getObjectHandler: invalid range header: requestID(%v) volume(%v) path(%v) rangeOpt(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), rangeOpt)
|
|
errorCode = InvalidRange
|
|
return
|
|
}
|
|
isRangeRead = true
|
|
}
|
|
|
|
responseCacheControl := r.URL.Query().Get(ParamResponseCacheControl)
|
|
if len(responseCacheControl) > 0 && !ValidateCacheControl(responseCacheControl) {
|
|
errorCode = InvalidCacheArgument
|
|
return
|
|
}
|
|
responseExpires := r.URL.Query().Get(ParamResponseExpires)
|
|
if len(responseExpires) > 0 && !ValidateCacheExpires(responseExpires) {
|
|
errorCode = InvalidCacheArgument
|
|
return
|
|
}
|
|
responseContentType := r.URL.Query().Get(ParamResponseContentType)
|
|
responseContentDisposition := r.URL.Query().Get(ParamResponseContentDisposition)
|
|
|
|
// get object meta
|
|
start := time.Now()
|
|
fileInfo, xattr, err := vol.ObjectMeta(param.Object())
|
|
span.AppendTrackLog("meta.r", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectHandler: get file meta fail: requestId(%v) volume(%v) path(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), param.Object(), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
|
|
// header condition check
|
|
errorCode = CheckConditionInHeader(r, fileInfo)
|
|
if errorCode != nil {
|
|
return
|
|
}
|
|
|
|
// validate and fix range
|
|
if isRangeRead && rangeUpper > uint64(fileInfo.Size)-1 {
|
|
rangeUpper = uint64(fileInfo.Size) - 1
|
|
if revRange && rangeLower > 0 {
|
|
rangeLower = rangeUpper + 1 - rangeLower
|
|
}
|
|
}
|
|
|
|
// compute content length
|
|
contentLength := uint64(fileInfo.Size)
|
|
if isRangeRead {
|
|
contentLength = rangeUpper - rangeLower + 1
|
|
}
|
|
|
|
// get object tagging size
|
|
ossTaggingData := xattr.Get(XAttrKeyOSSTagging)
|
|
output, _ := ParseTagging(string(ossTaggingData))
|
|
if output != nil && len(output.TagSet) > 0 {
|
|
w.Header().Set(XAmzTaggingCount, strconv.Itoa(len(output.TagSet)))
|
|
}
|
|
|
|
// set response header for GetObject
|
|
w.Header().Set(AcceptRanges, ValueAcceptRanges)
|
|
w.Header().Set(LastModified, formatTimeRFC1123(fileInfo.ModifyTime))
|
|
if len(responseContentType) > 0 {
|
|
w.Header().Set(ContentType, responseContentType)
|
|
} else if len(fileInfo.MIMEType) > 0 {
|
|
w.Header().Set(ContentType, fileInfo.MIMEType)
|
|
} else {
|
|
w.Header().Set(ContentType, ValueContentTypeStream)
|
|
}
|
|
if len(responseContentDisposition) > 0 {
|
|
w.Header().Set(ContentDisposition, responseContentDisposition)
|
|
} else if len(fileInfo.Disposition) > 0 {
|
|
w.Header().Set(ContentDisposition, fileInfo.Disposition)
|
|
}
|
|
if len(responseCacheControl) > 0 {
|
|
w.Header().Set(CacheControl, responseCacheControl)
|
|
} else if len(fileInfo.CacheControl) > 0 {
|
|
w.Header().Set(CacheControl, fileInfo.CacheControl)
|
|
}
|
|
if len(responseExpires) > 0 {
|
|
w.Header().Set(Expires, responseExpires)
|
|
} else if len(fileInfo.Expires) > 0 {
|
|
w.Header().Set(Expires, fileInfo.Expires)
|
|
}
|
|
if len(fileInfo.RetainUntilDate) > 0 {
|
|
w.Header().Set(XAmzObjectLockMode, ComplianceMode)
|
|
w.Header().Set(XAmzObjectLockRetainUntilDate, fileInfo.RetainUntilDate)
|
|
}
|
|
|
|
// check request is whether contain param : partNumber
|
|
partNumber := r.URL.Query().Get(ParamPartNumber)
|
|
if len(partNumber) > 0 && fileInfo.Size >= MinParallelDownloadFileSize {
|
|
partNumberInt, err := strconv.ParseUint(partNumber, 10, 64)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectHandler: parse partNumber(%v) fail: requestID(%v) volume(%v) path(%v) err(%v)",
|
|
partNumber, GetRequestID(r), param.Bucket(), param.Object(), err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
partSize, partCount, rangeLower, rangeUpper = parsePartInfo(partNumberInt, uint64(fileInfo.Size))
|
|
log.LogDebugf("getObjectHandler: partNumber(%v) fileSize(%v) parsed: partSize(%d) partCount(%d) rangeLower(%d) rangeUpper(%d)",
|
|
partNumberInt, fileInfo.Size, partSize, partCount, rangeLower, rangeUpper)
|
|
if partNumberInt > partCount {
|
|
log.LogErrorf("getObjectHandler: partNumber(%d) > partCount(%d): requestID(%v) volume(%v) path(%v)",
|
|
partNumberInt, partCount, GetRequestID(r), param.Bucket(), param.Object())
|
|
errorCode = NoSuchKey
|
|
return
|
|
}
|
|
// Header : Accept-Range, Content-Length, Content-Range, ETag, x-amz-mp-parts-count
|
|
w.Header().Set(ContentLength, strconv.Itoa(int(partSize)))
|
|
w.Header().Set(ContentRange, fmt.Sprintf("bytes %d-%d/%d", rangeLower, rangeUpper, fileInfo.Size))
|
|
w.Header().Set(XAmzMpPartsCount, strconv.Itoa(int(partCount)))
|
|
if len(fileInfo.ETag) > 0 && !strings.Contains(fileInfo.ETag, "-") {
|
|
w.Header()[ETag] = []string{fmt.Sprintf("%s-%d", fileInfo.ETag, partCount)}
|
|
}
|
|
} else {
|
|
w.Header().Set(ContentLength, strconv.FormatUint(contentLength, 10))
|
|
if len(fileInfo.ETag) > 0 {
|
|
w.Header()[ETag] = []string{wrapUnescapedQuot(fileInfo.ETag)}
|
|
}
|
|
if isRangeRead {
|
|
w.Header().Set(ContentRange, fmt.Sprintf("bytes %d-%d/%d", rangeLower, rangeUpper, fileInfo.Size))
|
|
}
|
|
}
|
|
|
|
// User-defined metadata
|
|
for name, value := range fileInfo.Metadata {
|
|
w.Header().Set(XAmzMetaPrefix+name, value)
|
|
}
|
|
|
|
if fileInfo.Mode.IsDir() {
|
|
return
|
|
}
|
|
|
|
// get object content
|
|
offset := rangeLower
|
|
size, err := safeConvertInt64ToUint64(fileInfo.Size)
|
|
fileSize := size
|
|
if err != nil {
|
|
return
|
|
}
|
|
if isRangeRead || len(partNumber) > 0 {
|
|
size = rangeUpper - rangeLower + 1
|
|
w.WriteHeader(http.StatusPartialContent)
|
|
}
|
|
|
|
// Flow Control
|
|
var writer io.Writer
|
|
if size > DefaultFlowLimitSize {
|
|
writer = rateLimit.GetResponseWriter(vol.owner, param.apiName, w)
|
|
} else {
|
|
writer = w
|
|
}
|
|
|
|
// read file
|
|
start = time.Now()
|
|
err = vol.readFile(fileInfo.Inode, fileSize, param.Object(), writer, offset, size, fileInfo.StorageClass)
|
|
span.AppendTrackLog("file.r", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectHandler: read file fail: requestID(%v) volume(%v) path(%v) offset(%v) size(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), offset, size, err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
}
|
|
|
|
func CheckConditionInHeader(r *http.Request, fileInfo *FSFileInfo) *ErrorCode {
|
|
// parse request header
|
|
match := r.Header.Get(IfMatch)
|
|
noneMatch := r.Header.Get(IfNoneMatch)
|
|
modified := r.Header.Get(IfModifiedSince)
|
|
unmodified := r.Header.Get(IfUnmodifiedSince)
|
|
// Checking precondition: If-Match
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html#API_GetObject_RequestSyntax
|
|
if match != "" {
|
|
if matchEag := strings.Trim(match, "\""); matchEag != fileInfo.ETag {
|
|
log.LogDebugf("getObjectHandler: object eTag(%s) not match If-Match header value(%s), requestId(%v)",
|
|
fileInfo.ETag, matchEag, GetRequestID(r))
|
|
return PreconditionFailed
|
|
|
|
}
|
|
}
|
|
// Checking precondition: If-Modified-Since
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html#API_GetObject_RequestSyntax
|
|
if modified != "" {
|
|
fileModTime := fileInfo.ModifyTime
|
|
modifiedTime, err := parseTimeRFC1123(modified)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectHandler: parse RFC1123 time fail: requestID(%v) err(%v)", GetRequestID(r), err)
|
|
return InvalidArgument
|
|
}
|
|
if !fileModTime.After(modifiedTime) {
|
|
log.LogInfof("getObjectHandler: file modified time not after than specified time: requestID(%v)", GetRequestID(r))
|
|
return NotModified
|
|
}
|
|
}
|
|
// Checking precondition: If-None-Match
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html#API_GetObject_RequestSyntax
|
|
if noneMatch != "" {
|
|
if noneMatchEtag := strings.Trim(noneMatch, "\""); noneMatchEtag == fileInfo.ETag {
|
|
log.LogErrorf("getObjectHandler: object eTag(%s) match If-None-Match header value(%s), requestId(%v)",
|
|
fileInfo.ETag, noneMatchEtag, GetRequestID(r))
|
|
return NotModified
|
|
}
|
|
}
|
|
// Checking precondition: If-Unmodified-Since
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html#API_GetObject_RequestSyntax
|
|
if unmodified != "" && match == "" {
|
|
fileModTime := fileInfo.ModifyTime
|
|
modifiedTime, err := parseTimeRFC1123(unmodified)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectHandler: parse RFC1123 time fail: requestID(%v) err(%v)", GetRequestID(r), err)
|
|
return InvalidArgument
|
|
}
|
|
if fileModTime.After(modifiedTime) {
|
|
log.LogInfof("getObjectHandler: file modified time after than specified time: requestID(%v)", GetRequestID(r))
|
|
return PreconditionFailed
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Head object
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html
|
|
func (o *ObjectNode) headObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
// check args
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("headObjectHandler: load volume fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// get object meta
|
|
start := time.Now()
|
|
fileInfo, _, err := vol.ObjectMeta(param.Object())
|
|
span.AppendTrackLog("meta.r", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("headObjectHandler: get file meta fail: requestId(%v) volume(%v) path(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), param.Object(), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
|
|
// parse request header
|
|
match := r.Header.Get(IfMatch)
|
|
noneMatch := r.Header.Get(IfNoneMatch)
|
|
modified := r.Header.Get(IfModifiedSince)
|
|
unmodified := r.Header.Get(IfUnmodifiedSince)
|
|
|
|
// Checking precondition: If-Match
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html#API_HeadObject_RequestSyntax
|
|
if match != "" {
|
|
if matchEag := strings.Trim(match, "\""); matchEag != fileInfo.ETag {
|
|
log.LogDebugf("headObjectHandler: object eTag(%s) not match If-Match header value(%s), requestId(%v)",
|
|
fileInfo.ETag, matchEag, GetRequestID(r))
|
|
errorCode = PreconditionFailed
|
|
return
|
|
}
|
|
}
|
|
// Checking precondition: If-Modified-Since
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html#API_HeadObject_RequestSyntax
|
|
if modified != "" {
|
|
fileModTime := fileInfo.ModifyTime
|
|
modifiedTime, err := parseTimeRFC1123(modified)
|
|
if err != nil {
|
|
log.LogDebugf("headObjectHandler: parse RFC1123 time fail: requestID(%v) err(%v)", GetRequestID(r), err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
if !fileModTime.After(modifiedTime) {
|
|
log.LogDebugf("headObjectHandler: file modified time not after than specified time: requestID(%v)", GetRequestID(r))
|
|
errorCode = NotModified
|
|
return
|
|
}
|
|
}
|
|
// Checking precondition: If-None-Match
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html#API_HeadObject_RequestSyntax
|
|
if noneMatch != "" {
|
|
if noneMatchEtag := strings.Trim(noneMatch, "\""); noneMatchEtag == fileInfo.ETag {
|
|
log.LogDebugf("headObjectHandler: object eTag(%s) match If-None-Match header value(%s), requestId(%v)",
|
|
fileInfo.ETag, noneMatchEtag, GetRequestID(r))
|
|
errorCode = NotModified
|
|
return
|
|
}
|
|
}
|
|
// Checking precondition: If-Unmodified-Since
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html#API_HeadObject_RequestSyntax
|
|
if unmodified != "" && match == "" {
|
|
fileModTime := fileInfo.ModifyTime
|
|
modifiedTime, err := parseTimeRFC1123(unmodified)
|
|
if err != nil {
|
|
log.LogDebugf("headObjectHandler: parse RFC1123 time fail: requestID(%v) err(%v)", GetRequestID(r), err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
if fileModTime.After(modifiedTime) {
|
|
log.LogDebugf("headObjectHandler: file modified time after than specified time: requestID(%v)", GetRequestID(r))
|
|
errorCode = PreconditionFailed
|
|
return
|
|
}
|
|
}
|
|
|
|
// set response header
|
|
w.Header().Set(AcceptRanges, ValueAcceptRanges)
|
|
w.Header().Set(LastModified, formatTimeRFC1123(fileInfo.ModifyTime))
|
|
w.Header().Set(ContentMD5, EmptyContentMD5String)
|
|
if len(fileInfo.MIMEType) > 0 {
|
|
w.Header().Set(ContentType, fileInfo.MIMEType)
|
|
} else {
|
|
w.Header().Set(ContentType, ValueContentTypeStream)
|
|
}
|
|
if len(fileInfo.Disposition) > 0 {
|
|
w.Header().Set(ContentDisposition, fileInfo.Disposition)
|
|
}
|
|
if len(fileInfo.CacheControl) > 0 {
|
|
w.Header().Set(CacheControl, fileInfo.CacheControl)
|
|
}
|
|
if len(fileInfo.Expires) > 0 {
|
|
w.Header().Set(Expires, fileInfo.Expires)
|
|
}
|
|
if len(fileInfo.RetainUntilDate) > 0 {
|
|
w.Header().Set(XAmzObjectLockMode, ComplianceMode)
|
|
w.Header().Set(XAmzObjectLockRetainUntilDate, fileInfo.RetainUntilDate)
|
|
}
|
|
|
|
// check request is whether contain param : partNumber
|
|
partNumber := r.URL.Query().Get(ParamPartNumber)
|
|
if len(partNumber) > 0 && fileInfo.Size >= MinParallelDownloadFileSize {
|
|
partNumberInt, err := strconv.ParseUint(partNumber, 10, 64)
|
|
if err != nil {
|
|
log.LogErrorf("headObjectHandler: parse param partNumber(%s) fail: requestID(%v) err(%v)", partNumber, GetRequestID(r), err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
partSize, partCount, rangeLower, rangeUpper := parsePartInfo(partNumberInt, uint64(fileInfo.Size))
|
|
log.LogDebugf("headObjectHandler: parsed partSize(%d), partCount(%d), rangeLower(%d), rangeUpper(%d)", partSize, partCount, rangeLower, rangeUpper)
|
|
if partNumberInt > partCount {
|
|
log.LogErrorf("headObjectHandler: param partNumber(%d) is more then partCount(%d): requestID(%v)", partNumberInt, partCount, GetRequestID(r))
|
|
errorCode = NoSuchKey
|
|
return
|
|
}
|
|
w.Header().Set(ContentLength, strconv.Itoa(int(partSize)))
|
|
w.Header().Set(ContentRange, fmt.Sprintf("bytes %d-%d/%d", rangeLower, rangeUpper, fileInfo.Size))
|
|
w.Header().Set(XAmzMpPartsCount, strconv.Itoa(int(partCount)))
|
|
if len(fileInfo.ETag) > 0 && !strings.Contains(fileInfo.ETag, "-") {
|
|
w.Header()[ETag] = []string{fmt.Sprintf("%s-%d", fileInfo.ETag, partCount)}
|
|
}
|
|
} else {
|
|
w.Header().Set(ContentLength, strconv.Itoa(int(fileInfo.Size)))
|
|
if len(fileInfo.ETag) > 0 {
|
|
w.Header()[ETag] = []string{wrapUnescapedQuot(fileInfo.ETag)}
|
|
}
|
|
}
|
|
|
|
// User-defined metadata
|
|
for name, value := range fileInfo.Metadata {
|
|
w.Header().Set(XAmzMetaPrefix+name, value)
|
|
}
|
|
}
|
|
|
|
// Delete objects (multiple objects)
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html
|
|
func (o *ObjectNode) deleteObjectsHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
|
|
if _, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("deleteObjectsHandler: load volume fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
requestMD5 := r.Header.Get(ContentMD5)
|
|
if requestMD5 == "" {
|
|
errorCode = MissingContentMD5
|
|
return
|
|
}
|
|
|
|
_, errorCode = VerifyContentLength(r, BodyLimit)
|
|
if errorCode != nil {
|
|
return
|
|
}
|
|
bytes, err := io.ReadAll(r.Body)
|
|
if err != nil {
|
|
log.LogErrorf("deleteObjectsHandler: read request body fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
errorCode = UnexpectedContent
|
|
return
|
|
}
|
|
if requestMD5 != GetMD5(bytes) {
|
|
errorCode = BadDigest
|
|
return
|
|
}
|
|
|
|
deleteReq := DeleteRequest{}
|
|
err = UnmarshalXMLEntity(bytes, &deleteReq)
|
|
if err != nil {
|
|
log.LogErrorf("deleteObjectsHandler: unmarshal xml fail: requestID(%v) volume(%v) request(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), string(bytes), err)
|
|
errorCode = MalformedXML
|
|
return
|
|
}
|
|
if len(deleteReq.Objects) > 1000 {
|
|
errorCode = EntityTooLarge
|
|
return
|
|
}
|
|
|
|
// Sort the key values in reverse order.
|
|
// The purpose of this is to delete the child leaf first and then the parent node.
|
|
// Source:
|
|
// 0. backup/
|
|
// 1. backup/20200101.bak
|
|
// 2. backup/20200102.bak
|
|
// Result:
|
|
// 0. backup/20200102.bak
|
|
// 1. backup/20200101.bak
|
|
// 2. backup/
|
|
sort.SliceStable(deleteReq.Objects, func(i, j int) bool {
|
|
return deleteReq.Objects[i].Key > deleteReq.Objects[j].Key
|
|
})
|
|
|
|
vol, acl, policy, err := o.loadBucketMeta(param.Bucket())
|
|
if err != nil {
|
|
log.LogErrorf("deleteObjectsHandler: load bucket metadata fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
userInfo, err := o.getUserInfoByAccessKeyV2(param.AccessKey())
|
|
if err != nil {
|
|
log.LogErrorf("deleteObjectsHandler: get userinfo fail: requestID(%v) volume(%v) accessKey(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.AccessKey(), err)
|
|
return
|
|
}
|
|
|
|
allowByAcl := false
|
|
if acl == nil && userInfo.UserID == vol.owner {
|
|
allowByAcl = true
|
|
}
|
|
if acl != nil && acl.IsAllowed(userInfo.UserID, param.Action()) {
|
|
allowByAcl = true
|
|
}
|
|
|
|
deletedObjects := make([]Deleted, 0, len(deleteReq.Objects))
|
|
deletedErrors := make([]Error, 0)
|
|
start := time.Now()
|
|
for _, object := range deleteReq.Objects {
|
|
result := POLICY_UNKNOW
|
|
if policy != nil && !policy.IsEmpty() {
|
|
conditionCheck := map[string]string{
|
|
SOURCEIP: param.sourceIP,
|
|
REFERER: param.r.Referer(),
|
|
KEYNAME: object.Key,
|
|
HOST: param.r.Host,
|
|
}
|
|
result = policy.IsAllowed(param, userInfo.UserID, vol.owner, conditionCheck)
|
|
}
|
|
if result == POLICY_DENY || (result == POLICY_UNKNOW && !allowByAcl) {
|
|
deletedErrors = append(deletedErrors, Error{
|
|
Key: object.Key,
|
|
Code: "AccessDenied",
|
|
Message: "Not Allowed By Policy",
|
|
})
|
|
continue
|
|
}
|
|
log.LogWarnf("deleteObjectsHandler: delete path: requestID(%v) remote(%v) volume(%v) path(%v)",
|
|
GetRequestID(r), getRequestIP(r), vol.Name(), object.Key)
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, DELETE_OBJECT); err != nil {
|
|
return
|
|
}
|
|
if err1 := vol.DeletePath(object.Key); err1 != nil {
|
|
log.LogErrorf("deleteObjectsHandler: delete object failed: requestID(%v) volume(%v) path(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), object.Key, err1)
|
|
if !strings.Contains(err1.Error(), AccessDenied.ErrorMessage) {
|
|
deletedErrors = append(deletedErrors, Error{Key: object.Key, Code: "InternalError", Message: err1.Error()})
|
|
} else {
|
|
deletedErrors = append(deletedErrors, Error{Key: object.Key, Code: "AccessDenied", Message: err1.Error()})
|
|
}
|
|
} else {
|
|
deletedObjects = append(deletedObjects, Deleted{Key: object.Key})
|
|
}
|
|
rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
}
|
|
span.AppendTrackLog("files.d", start, err)
|
|
|
|
deleteResult := DeleteResult{
|
|
Deleted: deletedObjects,
|
|
Error: deletedErrors,
|
|
}
|
|
response, err1 := MarshalXMLEntity(deleteResult)
|
|
if err1 != nil {
|
|
log.LogErrorf("deleteObjectsHandler: xml marshal fail: requestID(%v) volume(%v) result(%+v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), deleteResult, err1)
|
|
}
|
|
|
|
writeSuccessResponseXML(w, response)
|
|
}
|
|
|
|
func extractSrcBucketKey(r *http.Request) (srcBucketId, srcKey, versionId string, err error) {
|
|
copySource := r.Header.Get(XAmzCopySource)
|
|
copySource, err = url.QueryUnescape(copySource)
|
|
if err != nil {
|
|
return "", "", "", InvalidArgument
|
|
}
|
|
// path could be /bucket/key or bucket/key
|
|
copySource = strings.TrimPrefix(copySource, "/")
|
|
elements := strings.SplitN(copySource, "?versionId=", 2)
|
|
if len(elements) == 1 {
|
|
versionId = ""
|
|
} else {
|
|
versionId = elements[1]
|
|
}
|
|
path := strings.SplitN(elements[0], "/", 2)
|
|
if len(path) == 1 {
|
|
return "", "", "", InvalidArgument
|
|
}
|
|
srcBucketId, srcKey = path[0], path[1]
|
|
if srcBucketId == "" || srcKey == "" {
|
|
return "", "", "", InvalidArgument
|
|
}
|
|
return
|
|
}
|
|
|
|
// Copy object
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html .
|
|
func (o *ObjectNode) copyObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
if len(param.Object()) > MaxKeyLength {
|
|
errorCode = KeyTooLong
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("copyObjectHandler: load volume fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// client can reset these system metadata: Content-Type, Content-Disposition
|
|
contentType := r.Header.Get(ContentType)
|
|
contentDisposition := r.Header.Get(ContentDisposition)
|
|
cacheControl := r.Header.Get(CacheControl)
|
|
if len(cacheControl) > 0 && !ValidateCacheControl(cacheControl) {
|
|
errorCode = InvalidCacheArgument
|
|
return
|
|
}
|
|
expires := r.Header.Get(Expires)
|
|
if len(expires) > 0 && !ValidateCacheExpires(expires) {
|
|
errorCode = InvalidCacheArgument
|
|
return
|
|
}
|
|
|
|
// metadata directive, direct object node use source file metadata or recreate metadata for target file
|
|
metadataDirective := r.Header.Get(XAmzMetadataDirective)
|
|
// metadata directive default value is COPY
|
|
if len(metadataDirective) == 0 {
|
|
metadataDirective = MetadataDirectiveCopy
|
|
}
|
|
if metadataDirective != MetadataDirectiveCopy && metadataDirective != MetadataDirectiveReplace {
|
|
log.LogErrorf("copyObjectHandler: x-amz-metadata-directive invalid: requestID(%v) volume(%v) x-amz-metadata-directive(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), metadataDirective, err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
// parse x-amz-copy-source header
|
|
sourceBucket, sourceObject, _, err := extractSrcBucketKey(r)
|
|
if err != nil {
|
|
log.LogErrorf("copyObjectHandler: copySource(%v) argument invalid: requestID(%v) volume(%v) err(%v)",
|
|
r.Header.Get(XAmzCopySource), GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
// check ACL
|
|
userInfo, err := o.getUserInfoByAccessKeyV2(param.AccessKey())
|
|
if err != nil {
|
|
log.LogErrorf("copyObjectHandler: get user info fail: requestID(%v) volume(%v) accessKey(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.AccessKey(), err)
|
|
return
|
|
}
|
|
acl, err := ParseACL(r, userInfo.UserID, false, vol.GetOwner() != userInfo.UserID)
|
|
if err != nil {
|
|
log.LogErrorf("copyObjectHandler: parse acl fail: requestID(%v) volume(%v) acl(%+v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), acl, err)
|
|
return
|
|
}
|
|
|
|
// get src object meta
|
|
var sourceVol *Volume
|
|
if sourceVol, err = o.getVol(sourceBucket); err != nil {
|
|
log.LogErrorf("copyObjectHandler: load source volume fail: requestID(%v) srcVolume(%v) err(%v)",
|
|
GetRequestID(r), sourceBucket, err)
|
|
return
|
|
}
|
|
|
|
// get object meta
|
|
start := time.Now()
|
|
fileInfo, _, err := sourceVol.ObjectMeta(sourceObject)
|
|
span.AppendTrackLog("meta.r", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("copyObjectHandler: get object meta fail: requestID(%v) srcVolume(%v) srcObject(%v) err(%v)",
|
|
GetRequestID(r), sourceBucket, sourceObject, err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
if fileInfo.Size > SinglePutLimit {
|
|
errorCode = EntityTooLarge
|
|
return
|
|
}
|
|
if fileInfo.Size > SinglePutLimit {
|
|
errorCode = EntityTooLarge
|
|
return
|
|
}
|
|
|
|
// get header
|
|
copyMatch := r.Header.Get(XAmzCopySourceIfMatch)
|
|
noneMatch := r.Header.Get(XAmzCopySourceIfNoneMatch)
|
|
modified := r.Header.Get(XAmzCopySourceIfModifiedSince)
|
|
unModified := r.Header.Get(XAmzCopySourceIfUnmodifiedSince)
|
|
|
|
// response 412
|
|
if modified != "" {
|
|
fileModTime := fileInfo.ModifyTime
|
|
modifiedTime, err := parseTimeRFC1123(modified)
|
|
if err != nil {
|
|
log.LogErrorf("copyObjectHandler: parse RFC1123 time fail: requestID(%v) err(%v)", GetRequestID(r), err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
if fileModTime.Before(modifiedTime) {
|
|
log.LogInfof("copyObjectHandler: file modified time not after than specified time: requestID(%v)", GetRequestID(r))
|
|
errorCode = PreconditionFailed
|
|
return
|
|
}
|
|
}
|
|
if unModified != "" {
|
|
fileModTime := fileInfo.ModifyTime
|
|
unmodifiedTime, err := parseTimeRFC1123(unModified)
|
|
if err != nil {
|
|
log.LogErrorf("copyObjectHandler: parse RFC1123 time fail: requestID(%v) err(%v)", GetRequestID(r), err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
if fileModTime.After(unmodifiedTime) {
|
|
log.LogInfof("copyObjectHandler: file modified time not before than specified time: requestID(%v)", GetRequestID(r))
|
|
errorCode = PreconditionFailed
|
|
return
|
|
}
|
|
}
|
|
if copyMatch != "" && fileInfo.ETag != copyMatch {
|
|
log.LogInfof("copyObjectHandler: eTag mismatched with specified: requestID(%v)", GetRequestID(r))
|
|
errorCode = PreconditionFailed
|
|
return
|
|
}
|
|
if noneMatch != "" && fileInfo.ETag == noneMatch {
|
|
log.LogInfof("copyObjectHandler: eTag same with specified: requestID(%v)", GetRequestID(r))
|
|
errorCode = PreconditionFailed
|
|
return
|
|
}
|
|
|
|
// ObjectLock Config
|
|
objetLock, err := vol.metaLoader.loadObjectLock()
|
|
if err != nil {
|
|
log.LogErrorf("copyObjectHandler: load volume objetLock: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
// parse user-defined metadata
|
|
metadata := ParseUserDefinedMetadata(r.Header)
|
|
|
|
// copy file
|
|
opt := &PutFileOption{
|
|
MIMEType: contentType,
|
|
Disposition: contentDisposition,
|
|
Metadata: metadata,
|
|
CacheControl: cacheControl,
|
|
Expires: expires,
|
|
ACL: acl,
|
|
ObjectLock: objetLock,
|
|
}
|
|
start = time.Now()
|
|
fsFileInfo, err := vol.CopyFile(sourceVol, sourceObject, param.Object(), metadataDirective, opt)
|
|
span.AppendTrackLog("file.c", start, err)
|
|
if err != nil && err != syscall.EINVAL && err != syscall.EFBIG {
|
|
log.LogErrorf("copyObjectHandler: Volume copy file fail: requestID(%v) Volume(%v) source(%v) target(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), sourceObject, param.Object(), err)
|
|
return
|
|
}
|
|
if err == syscall.EINVAL {
|
|
log.LogErrorf("copyObjectHandler: target file existed, and mode conflict: requestID(%v) Volume(%v) source(%v) target(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), sourceObject, param.Object(), err)
|
|
errorCode = ObjectModeConflict
|
|
return
|
|
}
|
|
if err == syscall.EFBIG {
|
|
log.LogErrorf("copyObjectHandler: source file size greater than 5GB: requestID(%v) Volume(%v) source(%v) target(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), sourceObject, param.Object(), err)
|
|
errorCode = CopySourceSizeTooLarge
|
|
return
|
|
}
|
|
|
|
copyResult := CopyResult{
|
|
ETag: "\"" + fsFileInfo.ETag + "\"",
|
|
LastModified: formatTimeISO(fsFileInfo.ModifyTime),
|
|
}
|
|
response, ierr := MarshalXMLEntity(copyResult)
|
|
if ierr != nil {
|
|
log.LogErrorf("copyObjectHandler: marshal xml result fail: requestID(%v) result(%v) err(%v)",
|
|
GetRequestID(r), copyResult, ierr)
|
|
}
|
|
|
|
writeSuccessResponseXML(w, response)
|
|
}
|
|
|
|
// List objects v1
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjects.html
|
|
func (o *ObjectNode) getBucketV1Handler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("getBucketV1Handler: load volume fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// get options
|
|
marker := r.URL.Query().Get(ParamMarker)
|
|
prefix := r.URL.Query().Get(ParamPrefix)
|
|
maxKeys := r.URL.Query().Get(ParamMaxKeys)
|
|
delimiter := r.URL.Query().Get(ParamPartDelimiter)
|
|
encodingType := r.URL.Query().Get(ParamEncodingType)
|
|
|
|
var maxKeysInt uint64
|
|
if maxKeys != "" {
|
|
maxKeysInt, err = strconv.ParseUint(maxKeys, 10, 16)
|
|
if err != nil {
|
|
log.LogErrorf("getBucketV1Handler: parse max key fail: requestID(%v) volume(%v) maxKeys(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), maxKeys, err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
if maxKeysInt > MaxKeys {
|
|
maxKeysInt = MaxKeys
|
|
}
|
|
} else {
|
|
maxKeysInt = uint64(MaxKeys)
|
|
}
|
|
|
|
// Validate encoding type option
|
|
if encodingType != "" && encodingType != "url" {
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
|
|
if marker != "" && prefix != "" && !strings.HasPrefix(marker, prefix) {
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
|
|
// list files
|
|
option := &ListFilesV1Option{
|
|
Prefix: prefix,
|
|
Delimiter: delimiter,
|
|
Marker: marker,
|
|
MaxKeys: maxKeysInt,
|
|
OnlyObject: true,
|
|
}
|
|
start := time.Now()
|
|
result, err := vol.ListFilesV1(option)
|
|
span.AppendTrackLog("file.l", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("getBucketV1Handler: list files fail: requestID(%v) volume(%v) option(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), option, err)
|
|
return
|
|
}
|
|
// The result of next list request should not include nextMarker.
|
|
if result.Truncated && len(result.Files) != 0 {
|
|
result.NextMarker = result.Files[len(result.Files)-1].Path
|
|
}
|
|
|
|
// get owner
|
|
bucketOwner := NewBucketOwner(vol)
|
|
log.LogDebugf("Owner: %v", bucketOwner)
|
|
contents := make([]*Content, 0, len(result.Files))
|
|
for _, file := range result.Files {
|
|
if file.Mode == 0 {
|
|
// Invalid file mode, which means that the inode of the file may not exist.
|
|
// Record and filter out the file.
|
|
log.LogWarnf("getBucketV1Handler: invalid file found: requestID(%v) volume(%v) path(%v) inode(%v)",
|
|
GetRequestID(r), vol.Name(), file.Path, file.Inode)
|
|
continue
|
|
}
|
|
content := &Content{
|
|
Key: encodeKey(file.Path, encodingType),
|
|
LastModified: formatTimeISO(file.ModifyTime),
|
|
ETag: wrapUnescapedQuot(file.ETag),
|
|
Size: int(file.Size),
|
|
StorageClass: StorageClassStandard,
|
|
Owner: bucketOwner,
|
|
}
|
|
contents = append(contents, content)
|
|
}
|
|
|
|
commonPrefixes := make([]*CommonPrefix, 0)
|
|
for _, prefix := range result.CommonPrefixes {
|
|
commonPrefix := &CommonPrefix{
|
|
Prefix: prefix,
|
|
}
|
|
commonPrefixes = append(commonPrefixes, commonPrefix)
|
|
}
|
|
|
|
listBucketResult := &ListBucketResult{
|
|
Bucket: param.Bucket(),
|
|
Prefix: prefix,
|
|
Marker: marker,
|
|
MaxKeys: int(maxKeysInt),
|
|
Delimiter: delimiter,
|
|
IsTruncated: result.Truncated,
|
|
NextMarker: result.NextMarker,
|
|
Contents: contents,
|
|
CommonPrefixes: commonPrefixes,
|
|
}
|
|
response, err := MarshalXMLEntity(listBucketResult)
|
|
if err != nil {
|
|
log.LogErrorf("getBucketV1Handler: xml marshal result fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), err)
|
|
return
|
|
}
|
|
|
|
writeSuccessResponseXML(w, response)
|
|
}
|
|
|
|
// List objects version 2
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html
|
|
func (o *ObjectNode) getBucketV2Handler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("getBucketV2Handler: load volume fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// get options
|
|
prefix := r.URL.Query().Get(ParamPrefix)
|
|
maxKeys := r.URL.Query().Get(ParamMaxKeys)
|
|
delimiter := r.URL.Query().Get(ParamPartDelimiter)
|
|
contToken := r.URL.Query().Get(ParamContToken)
|
|
fetchOwner := r.URL.Query().Get(ParamFetchOwner)
|
|
startAfter := r.URL.Query().Get(ParamStartAfter)
|
|
encodingType := r.URL.Query().Get(ParamEncodingType)
|
|
|
|
var maxKeysInt uint64
|
|
if maxKeys != "" {
|
|
maxKeysInt, err = strconv.ParseUint(maxKeys, 10, 16)
|
|
if err != nil {
|
|
log.LogErrorf("getBucketV2Handler: parse max keys fail: requestID(%v) volume(%v) maxKeys(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), maxKeys, err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
if maxKeysInt > MaxKeys {
|
|
maxKeysInt = MaxKeys
|
|
}
|
|
} else {
|
|
maxKeysInt = MaxKeys
|
|
}
|
|
|
|
var fetchOwnerBool bool
|
|
if fetchOwner != "" {
|
|
fetchOwnerBool, err = strconv.ParseBool(fetchOwner)
|
|
if err != nil {
|
|
log.LogErrorf("getBucketV2Handler: parse fetch owner fail: requestID(%v) volume(%v) fetchOwner(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), fetchOwner, err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
} else {
|
|
fetchOwnerBool = false
|
|
}
|
|
|
|
// Validate encoding type option
|
|
if encodingType != "" && encodingType != "url" {
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
|
|
var marker string
|
|
if contToken != "" {
|
|
marker = contToken
|
|
} else {
|
|
marker = startAfter
|
|
}
|
|
if marker != "" && prefix != "" && !strings.HasPrefix(marker, prefix) {
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
|
|
// list files
|
|
option := &ListFilesV2Option{
|
|
Delimiter: delimiter,
|
|
MaxKeys: maxKeysInt,
|
|
Prefix: prefix,
|
|
ContToken: contToken,
|
|
FetchOwner: fetchOwnerBool,
|
|
StartAfter: startAfter,
|
|
}
|
|
start := time.Now()
|
|
result, err := vol.ListFilesV2(option)
|
|
span.AppendTrackLog("file.l", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("getBucketV2Handler: list files fail, requestID(%v) volume(%v) option(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), option, err)
|
|
return
|
|
}
|
|
// The result of next list request should not include continuationToken.
|
|
if result.Truncated && len(result.Files) > 0 {
|
|
result.NextToken = result.Files[len(result.Files)-1].Path
|
|
}
|
|
|
|
// get owner
|
|
var bucketOwner *BucketOwner
|
|
if fetchOwnerBool {
|
|
bucketOwner = NewBucketOwner(vol)
|
|
}
|
|
|
|
contents := make([]*Content, 0)
|
|
if len(result.Files) > 0 {
|
|
for _, file := range result.Files {
|
|
if file.Mode == 0 {
|
|
// Invalid file mode, which means that the inode of the file may not exist.
|
|
// Record and filter out the file.
|
|
log.LogWarnf("getBucketV2Handler: invalid file found: requestID(%v) volume(%v) path(%v) inode(%v)",
|
|
GetRequestID(r), vol.Name(), file.Path, file.Inode)
|
|
result.KeyCount--
|
|
continue
|
|
}
|
|
content := &Content{
|
|
Key: encodeKey(file.Path, encodingType),
|
|
LastModified: formatTimeISO(file.ModifyTime),
|
|
ETag: wrapUnescapedQuot(file.ETag),
|
|
Size: int(file.Size),
|
|
StorageClass: StorageClassStandard,
|
|
Owner: bucketOwner,
|
|
}
|
|
contents = append(contents, content)
|
|
}
|
|
}
|
|
|
|
commonPrefixes := make([]*CommonPrefix, 0)
|
|
for _, prefix := range result.CommonPrefixes {
|
|
commonPrefix := &CommonPrefix{
|
|
Prefix: prefix,
|
|
}
|
|
commonPrefixes = append(commonPrefixes, commonPrefix)
|
|
}
|
|
|
|
listBucketResult := ListBucketResultV2{
|
|
Name: param.Bucket(),
|
|
Prefix: prefix,
|
|
Token: contToken,
|
|
NextToken: result.NextToken,
|
|
KeyCount: result.KeyCount,
|
|
MaxKeys: maxKeysInt,
|
|
Delimiter: delimiter,
|
|
IsTruncated: result.Truncated,
|
|
Contents: contents,
|
|
CommonPrefixes: commonPrefixes,
|
|
}
|
|
response, err := MarshalXMLEntity(listBucketResult)
|
|
if err != nil {
|
|
log.LogErrorf("getBucketV2Handler: xml marshal result fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), err)
|
|
return
|
|
}
|
|
|
|
writeSuccessResponseXML(w, response)
|
|
}
|
|
|
|
// Put object
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html
|
|
func (o *ObjectNode) putObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
if len(param.Object()) > MaxKeyLength {
|
|
errorCode = KeyTooLong
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("putObjectHandler: load volume fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
// Get request MD5, if request MD5 is not empty, compute and verify it.
|
|
requestMD5 := r.Header.Get(ContentMD5)
|
|
if requestMD5 != "" {
|
|
decoded, err := base64.StdEncoding.DecodeString(requestMD5)
|
|
if err != nil {
|
|
errorCode = InvalidDigest
|
|
return
|
|
}
|
|
requestMD5 = hex.EncodeToString(decoded)
|
|
}
|
|
|
|
// ObjectLock Config
|
|
objetLock, err := vol.metaLoader.loadObjectLock()
|
|
if err != nil {
|
|
log.LogErrorf("putObjectHandler: load volume objetLock: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
if objetLock != nil && objetLock.ToRetention() != nil && requestMD5 == "" {
|
|
errorCode = NoContentMd5HeaderErr
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// Check 'x-amz-tagging' header
|
|
// Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html#API_PutObject_RequestSyntax
|
|
var tagging *Tagging
|
|
if xAmxTagging := r.Header.Get(XAmzTagging); xAmxTagging != "" {
|
|
if tagging, err = ParseTagging(xAmxTagging); err != nil {
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
if err = tagging.Validate(); err != nil {
|
|
log.LogErrorf("putObjectHandler: tagging validate fail: requestID(%v) volume(%v) path(%v) tagging(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), param.Object(), tagging, err)
|
|
return
|
|
}
|
|
}
|
|
|
|
var userInfo *proto.UserInfo
|
|
if userInfo, err = o.getUserInfoByAccessKeyV2(param.AccessKey()); err != nil {
|
|
log.LogErrorf("putObjectHandler: get user info fail: requestID(%v) volume(%v) accessKey(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.AccessKey(), err)
|
|
return
|
|
}
|
|
|
|
// Check ACL
|
|
acl, err := ParseACL(r, userInfo.UserID, false, vol.GetOwner() != userInfo.UserID)
|
|
if err != nil {
|
|
log.LogErrorf("putObjectHandler: parse acl fail: requestID(%v) volume(%v) path(%v) acl(%+v) err(%v)",
|
|
GetRequestID(r), vol.Name(), param.Object(), acl, err)
|
|
return
|
|
}
|
|
|
|
// Verify ContentLength
|
|
length := GetContentLength(r)
|
|
if length > SinglePutLimit {
|
|
errorCode = EntityTooLarge
|
|
return
|
|
}
|
|
if length < 0 {
|
|
errorCode = MissingContentLength
|
|
return
|
|
}
|
|
|
|
// Get the requested content-type.
|
|
// In addition to being used to manage data types, it is used to distinguish
|
|
// whether the request is to create a directory.
|
|
contentType := r.Header.Get(ContentType)
|
|
// Get request header : content-disposition
|
|
contentDisposition := r.Header.Get(ContentDisposition)
|
|
// Get request header : Cache-Control
|
|
cacheControl := r.Header.Get(CacheControl)
|
|
if len(cacheControl) > 0 && !ValidateCacheControl(cacheControl) {
|
|
errorCode = InvalidCacheArgument
|
|
return
|
|
}
|
|
// Get request header : Expires
|
|
expires := r.Header.Get(Expires)
|
|
if len(expires) > 0 && !ValidateCacheExpires(expires) {
|
|
errorCode = InvalidCacheArgument
|
|
return
|
|
}
|
|
// Checking user-defined metadata
|
|
metadata := ParseUserDefinedMetadata(r.Header)
|
|
// Audit file write
|
|
log.LogInfof("Audit: put object: requestID(%v) remote(%v) volume(%v) path(%v) type(%v)",
|
|
GetRequestID(r), getRequestIP(r), vol.Name(), param.Object(), contentType)
|
|
|
|
// Flow Control
|
|
var reader io.Reader
|
|
if length > DefaultFlowLimitSize {
|
|
reader = rateLimit.GetReader(vol.owner, param.apiName, r.Body)
|
|
} else {
|
|
reader = r.Body
|
|
}
|
|
|
|
// Put Object
|
|
opt := &PutFileOption{
|
|
MIMEType: contentType,
|
|
Disposition: contentDisposition,
|
|
Tagging: tagging,
|
|
Metadata: metadata,
|
|
CacheControl: cacheControl,
|
|
Expires: expires,
|
|
ACL: acl,
|
|
ObjectLock: objetLock,
|
|
}
|
|
start := time.Now()
|
|
fsFileInfo, err := vol.PutObject(param.Object(), reader, opt)
|
|
span.AppendTrackLog("file.w", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("putObjectHandler: put object fail: requestId(%v) volume(%v) path(%v) remote(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), param.Object(), getRequestIP(r), err)
|
|
err = handlePutObjectErr(err)
|
|
return
|
|
}
|
|
|
|
// check content MD5
|
|
if requestMD5 != "" && requestMD5 != fsFileInfo.ETag {
|
|
log.LogErrorf("putObjectHandler: MD5 validate fail: requestID(%v) volume(%v) path(%v) requestMD5(%v) serverMD5(%v)",
|
|
GetRequestID(r), vol.Name(), param.Object(), requestMD5, fsFileInfo.ETag)
|
|
errorCode = BadDigest
|
|
return
|
|
}
|
|
|
|
// set response header
|
|
w.Header()[ETag] = []string{wrapUnescapedQuot(fsFileInfo.ETag)}
|
|
}
|
|
|
|
// Post object
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectPOST.html
|
|
func (o *ObjectNode) postObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("postObjectHandler: load volume fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
var userInfo *proto.UserInfo
|
|
if userInfo, err = o.getUserInfoByAccessKeyV2(param.AccessKey()); err != nil {
|
|
log.LogErrorf("postObjectHandler: get user info fail: requestID(%v) volume(%v) accessKey(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.AccessKey(), err)
|
|
return
|
|
}
|
|
|
|
// qps and concurrency limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// parse the request form
|
|
formReq := NewFormRequest(r)
|
|
if err = formReq.ParseMultipartForm(); err != nil {
|
|
log.LogErrorf("postObjectHandler: parse form fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%v)", errorCode.ErrorMessage, err)
|
|
return
|
|
}
|
|
|
|
// content-md5 check if specified in the request
|
|
requestMD5 := formReq.MultipartFormValue(ContentMD5)
|
|
if requestMD5 != "" {
|
|
decoded, err := base64.StdEncoding.DecodeString(requestMD5)
|
|
if err != nil {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "Invalid Content-MD5")
|
|
return
|
|
}
|
|
requestMD5 = hex.EncodeToString(decoded)
|
|
}
|
|
|
|
// object lock check
|
|
objetLock, err := vol.metaLoader.loadObjectLock()
|
|
if err != nil {
|
|
log.LogErrorf("postObjectHandler: load volume objetLock fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
if objetLock != nil && objetLock.ToRetention() != nil && requestMD5 == "" {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "No Content-MD5 with Object Lock")
|
|
return
|
|
}
|
|
|
|
// other form fields check
|
|
key := formReq.MultipartFormValue("key")
|
|
if key == "" {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "Missing key")
|
|
return
|
|
}
|
|
key = strings.Replace(key, "${filename}", formReq.FileName(), -1)
|
|
if !utf8.ValidString(key) || len(key) > MaxKeyLength {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "Invalid utf8 string or the key is too long")
|
|
return
|
|
}
|
|
|
|
var aclInfo *AccessControlPolicy
|
|
if acl := formReq.MultipartFormValue("acl"); acl != "" {
|
|
if aclInfo, err = ParseCannedAcl(acl, userInfo.UserID); err != nil {
|
|
log.LogErrorf("postObjectHandler: parse canned acl fail: requestID(%v) volume(%v) acl(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), acl, err)
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%v)", errorCode.ErrorMessage, err)
|
|
return
|
|
}
|
|
}
|
|
|
|
var tagging *Tagging
|
|
if taggingRaw := formReq.MultipartFormValue("tagging"); taggingRaw != "" {
|
|
tagging = NewTagging()
|
|
if err = xml.Unmarshal([]byte(taggingRaw), tagging); err != nil {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "Invalid tagging")
|
|
return
|
|
}
|
|
if err = tagging.Validate(); err != nil {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%v)", errorCode.ErrorMessage, err)
|
|
return
|
|
}
|
|
}
|
|
|
|
successStatus := formReq.MultipartFormValue("success_action_status")
|
|
successRedirect := formReq.MultipartFormValue("success_action_redirect")
|
|
var successRedirectURL *url.URL
|
|
if successRedirect != "" {
|
|
if successRedirectURL, err = url.Parse(successRedirect); err != nil {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "Invalid success_action_redirect")
|
|
return
|
|
}
|
|
}
|
|
|
|
contentType := formReq.MultipartFormValue("content-type")
|
|
contentDisposition := formReq.MultipartFormValue("content-disposition")
|
|
cacheControl := formReq.MultipartFormValue("cache-control")
|
|
if cacheControl != "" && !ValidateCacheControl(cacheControl) {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "Invalid cache-control")
|
|
return
|
|
}
|
|
|
|
expires := formReq.MultipartFormValue("expires")
|
|
if expires != "" && !ValidateCacheExpires(expires) {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "Invalid expires")
|
|
return
|
|
}
|
|
|
|
policy := formReq.MultipartFormValue("policy")
|
|
if policy == "" {
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%s)", errorCode.ErrorMessage, "Missing policy")
|
|
return
|
|
}
|
|
|
|
// read the file, the rest will be written to a temporary file if exceed
|
|
f, size, err := formReq.FormFile(10 << 20)
|
|
if err != nil {
|
|
log.LogErrorf("postObjectHandler: form file fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
errorCode = MalformedPOSTRequest.Copy()
|
|
errorCode.ErrorMessage = fmt.Sprintf("%s (%v)", errorCode.ErrorMessage, err)
|
|
return
|
|
}
|
|
defer f.Close()
|
|
if size > SinglePutLimit {
|
|
errorCode = EntityTooLarge
|
|
return
|
|
}
|
|
|
|
metadata := make(map[string]string)
|
|
// policy match forms
|
|
forms := make(map[string]string)
|
|
for name, values := range formReq.MultipartForm.Value {
|
|
name = strings.ToLower(name)
|
|
if strings.HasPrefix(name, XAmzMetaPrefix) {
|
|
forms[name] = strings.Join(values, ",")
|
|
metadata[name[len(XAmzMetaPrefix):]] = strings.Join(values, ",")
|
|
} else if len(values) > 0 {
|
|
forms[name] = values[0]
|
|
}
|
|
}
|
|
forms["bucket"] = param.Bucket()
|
|
forms["key"] = key
|
|
forms["content-length"] = strconv.FormatInt(size, 10)
|
|
|
|
// policy condition check
|
|
if err = PolicyConditionMatch(policy, forms); err != nil {
|
|
log.LogErrorf("postObjectHandler: policy match fail: requestID(%v) volume(%v) forms(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), forms, err)
|
|
return
|
|
}
|
|
|
|
// flow control
|
|
var reader io.Reader
|
|
if size > DefaultFlowLimitSize {
|
|
reader = rateLimit.GetReader(vol.owner, param.apiName, f)
|
|
} else {
|
|
reader = f
|
|
}
|
|
|
|
// put object
|
|
putOpt := &PutFileOption{
|
|
MIMEType: contentType,
|
|
Disposition: contentDisposition,
|
|
Tagging: tagging,
|
|
Metadata: metadata,
|
|
CacheControl: cacheControl,
|
|
Expires: expires,
|
|
ACL: aclInfo,
|
|
ObjectLock: objetLock,
|
|
}
|
|
start := time.Now()
|
|
fsFileInfo, err := vol.PutObject(key, reader, putOpt)
|
|
span.AppendTrackLog("file.w", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("postObjectHandler: put object fail: requestId(%v) volume(%v) path(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), key, err)
|
|
err = handlePutObjectErr(err)
|
|
return
|
|
}
|
|
|
|
// check content-md5 of actual data if specified in the request
|
|
if requestMD5 != "" && requestMD5 != fsFileInfo.ETag {
|
|
log.LogErrorf("postObjectHandler: MD5 validate fail: requestID(%v) volume(%v) path(%v) requestMD5(%v) serverMD5(%v)",
|
|
GetRequestID(r), vol.Name(), key, requestMD5, fsFileInfo.ETag)
|
|
errorCode = BadDigest
|
|
return
|
|
}
|
|
|
|
// set response header
|
|
etag := wrapUnescapedQuot(fsFileInfo.ETag)
|
|
w.Header()[ETag] = []string{etag}
|
|
|
|
// return response depending on success_action_xxx parameter
|
|
if successRedirectURL != nil {
|
|
query := successRedirectURL.Query()
|
|
query.Set("bucket", param.Bucket())
|
|
query.Set("key", key)
|
|
query.Set("etag", etag)
|
|
successRedirectURL.RawQuery = query.Encode()
|
|
w.Header().Set(Location, successRedirectURL.String())
|
|
w.WriteHeader(http.StatusSeeOther)
|
|
return
|
|
}
|
|
switch successStatus {
|
|
case "200":
|
|
w.WriteHeader(http.StatusOK)
|
|
case "201":
|
|
response := NewS3UploadObject()
|
|
response.Bucket = param.Bucket()
|
|
response.Key = key
|
|
response.ETag = etag
|
|
response.Location = "/" + response.Bucket + "/" + response.Key
|
|
writeResponse(w, http.StatusCreated, []byte(response.String()), ValueContentTypeXML)
|
|
default:
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
}
|
|
|
|
func handlePutObjectErr(err error) error {
|
|
if err == syscall.EINVAL {
|
|
return ObjectModeConflict
|
|
}
|
|
if err == syscall.EEXIST {
|
|
return ConflictUploadRequest
|
|
}
|
|
if err == io.ErrUnexpectedEOF {
|
|
return EntityTooSmall
|
|
}
|
|
return err
|
|
}
|
|
|
|
// Delete object
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObject.html .
|
|
func (o *ObjectNode) deleteObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("deleteObjectHandler: load volume fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// Audit deletion
|
|
log.LogInfof("Audit: delete object: requestID(%v) remote(%v) volume(%v) path(%v)",
|
|
GetRequestID(r), getRequestIP(r), vol.Name(), param.Object())
|
|
|
|
// Delete file
|
|
start := time.Now()
|
|
err = vol.DeletePath(param.Object())
|
|
span.AppendTrackLog("file.d", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("deleteObjectHandler: Volume delete file fail: "+
|
|
"requestID(%v) volume(%v) path(%v) err(%v)", GetRequestID(r), vol.Name(), param.Object(), err)
|
|
if strings.Contains(err.Error(), AccessDenied.ErrorMessage) {
|
|
err = AccessDenied
|
|
}
|
|
return
|
|
}
|
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// Get object tagging
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObjectTagging.html
|
|
func (o *ObjectNode) getObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("getObjectTaggingHandler: load volume fail: requestID(%v) volume(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
// get xattr
|
|
start := time.Now()
|
|
xattrInfo, err := vol.GetXAttr(param.object, XAttrKeyOSSTagging)
|
|
span.AppendTrackLog("xattr.r", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectTaggingHandler: get volume XAttr fail: requestID(%v) err(%v)", GetRequestID(r), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
|
|
ossTaggingData := xattrInfo.Get(XAttrKeyOSSTagging)
|
|
|
|
output, _ := ParseTagging(string(ossTaggingData))
|
|
response, err := MarshalXMLEntity(output)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectTaggingHandler: xml marshal result fail: requestID(%v) result(%v) err(%v)",
|
|
GetRequestID(r), output, err)
|
|
return
|
|
}
|
|
|
|
writeSuccessResponseXML(w, response)
|
|
}
|
|
|
|
// Put object tagging
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectTagging.html
|
|
func (o *ObjectNode) putObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("putObjectTaggingHandler: load volume fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
_, errorCode = VerifyContentLength(r, BodyLimit)
|
|
if errorCode != nil {
|
|
return
|
|
}
|
|
var requestBody []byte
|
|
if requestBody, err = io.ReadAll(r.Body); err != nil {
|
|
log.LogErrorf("putObjectTaggingHandler: read request body data fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
|
|
tagging := NewTagging()
|
|
if err = xml.Unmarshal(requestBody, tagging); err != nil {
|
|
log.LogWarnf("putObjectTaggingHandler: decode request body fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
if err = tagging.Validate(); err != nil {
|
|
log.LogErrorf("putObjectTaggingHandler: tagging validate fail: requestID(%v) tagging(%v) err(%v)",
|
|
GetRequestID(r), tagging, err)
|
|
return
|
|
}
|
|
|
|
start := time.Now()
|
|
err = vol.SetXAttr(param.object, XAttrKeyOSSTagging, []byte(tagging.Encode()), false)
|
|
span.AppendTrackLog("xattr.w", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("pubObjectTaggingHandler: set tagging xattr fail: requestID(%v) volume(%v) object(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
}
|
|
|
|
// Delete object tagging
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjectTagging.html
|
|
func (o *ObjectNode) deleteObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("deleteObjectTaggingHandler: load volume fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
start := time.Now()
|
|
err = vol.DeleteXAttr(param.object, XAttrKeyOSSTagging)
|
|
span.AppendTrackLog("xattr.d", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("deleteObjectTaggingHandler: volume delete tagging fail: requestID(%v) volume(%v) object(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// Put object extend attribute (xattr)
|
|
func (o *ObjectNode) putObjectXAttrHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if len(param.Bucket()) == 0 {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if len(param.Object()) == 0 {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.bucket); err != nil {
|
|
log.LogErrorf("pubObjectXAttrHandler: load volume fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
_, errorCode = VerifyContentLength(r, BodyLimit)
|
|
if errorCode != nil {
|
|
return
|
|
}
|
|
var requestBody []byte
|
|
if requestBody, err = io.ReadAll(r.Body); err != nil {
|
|
errorCode = &ErrorCode{
|
|
ErrorCode: "BadRequest",
|
|
ErrorMessage: err.Error(),
|
|
StatusCode: http.StatusBadRequest,
|
|
}
|
|
return
|
|
}
|
|
putXAttrRequest := PutXAttrRequest{}
|
|
if err = xml.Unmarshal(requestBody, &putXAttrRequest); err != nil {
|
|
errorCode = &ErrorCode{
|
|
ErrorCode: "BadRequest",
|
|
ErrorMessage: err.Error(),
|
|
StatusCode: http.StatusBadRequest,
|
|
}
|
|
return
|
|
}
|
|
key, value := putXAttrRequest.XAttr.Key, putXAttrRequest.XAttr.Value
|
|
if len(key) == 0 {
|
|
return
|
|
}
|
|
|
|
start := time.Now()
|
|
err = vol.SetXAttr(param.object, key, []byte(value), true)
|
|
span.AppendTrackLog("xattr.w", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("pubObjectXAttrHandler: volume set extend attribute fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
}
|
|
|
|
// Get object extend attribute (xattr)
|
|
func (o *ObjectNode) getObjectXAttrHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if len(param.Bucket()) == 0 {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if len(param.Object()) == 0 {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("getObjectXAttrHandler: load volume fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
var xattrKey string
|
|
if xattrKey = param.GetVar(ParamKey); len(xattrKey) == 0 {
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
|
|
start := time.Now()
|
|
info, err := vol.GetXAttr(param.object, xattrKey)
|
|
span.AppendTrackLog("xattr.r", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectXAttrHandler: get extend attribute fail: requestID(%v) volume(%v) object(%v) key(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), xattrKey, err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
|
|
output := GetXAttrOutput{
|
|
XAttr: &XAttr{
|
|
Key: xattrKey,
|
|
Value: string(info.Get(xattrKey)),
|
|
},
|
|
}
|
|
response, err := MarshalXMLEntity(&output)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectXAttrHandler: xml marshal result fail: requestID(%v) result(%v) err(%v)",
|
|
GetRequestID(r), output, err)
|
|
return
|
|
}
|
|
|
|
writeSuccessResponseXML(w, response)
|
|
}
|
|
|
|
// Delete object extend attribute (xattr)
|
|
func (o *ObjectNode) deleteObjectXAttrHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if len(param.Bucket()) == 0 {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if len(param.Object()) == 0 {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("deleteObjectXAttrHandler: load volume fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
var xattrKey string
|
|
if xattrKey = param.GetVar(ParamKey); len(xattrKey) == 0 {
|
|
errorCode = InvalidArgument
|
|
return
|
|
}
|
|
|
|
start := time.Now()
|
|
err = vol.DeleteXAttr(param.object, xattrKey)
|
|
span.AppendTrackLog("xattr.d", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("deleteObjectXAttrHandler: delete extend attribute fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
}
|
|
|
|
// List object xattrs
|
|
func (o *ObjectNode) listObjectXAttrs(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
param := ParseRequestParam(r)
|
|
if len(param.Bucket()) == 0 {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if len(param.Object()) == 0 {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.bucket); err != nil {
|
|
log.LogErrorf("listObjectXAttrs: load volume fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
return
|
|
}
|
|
|
|
// QPS and Concurrency Limit
|
|
rateLimit := o.AcquireRateLimiter()
|
|
if err = rateLimit.AcquireLimitResource(vol.owner, param.apiName); err != nil {
|
|
return
|
|
}
|
|
defer rateLimit.ReleaseLimitResource(vol.owner, param.apiName)
|
|
|
|
start := time.Now()
|
|
keys, err := vol.ListXAttrs(param.object)
|
|
span.AppendTrackLog("xattr.l", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("listObjectXAttrs: volume list extend attributes fail: requestID(%v) volume(%v) object(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
|
|
output := ListXAttrsOutput{
|
|
Keys: keys,
|
|
}
|
|
response, err := MarshalXMLEntity(&output)
|
|
if err != nil {
|
|
log.LogErrorf("listObjectXAttrs: marshal response body fail: requestID(%v) volume(%v) object(%v) err(%v)",
|
|
GetRequestID(r), param.Bucket(), param.Object(), err)
|
|
return
|
|
}
|
|
|
|
writeSuccessResponseXML(w, response)
|
|
}
|
|
|
|
// GetObjectRetention
|
|
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObjectRetention.html
|
|
func (o *ObjectNode) getObjectRetentionHandler(w http.ResponseWriter, r *http.Request) {
|
|
var (
|
|
err error
|
|
errorCode *ErrorCode
|
|
)
|
|
|
|
span := trace.SpanFromContextSafe(r.Context())
|
|
defer func() {
|
|
o.errorResponse(w, r, err, errorCode)
|
|
}()
|
|
|
|
// check args
|
|
param := ParseRequestParam(r)
|
|
if param.Bucket() == "" {
|
|
errorCode = InvalidBucketName
|
|
return
|
|
}
|
|
if param.Object() == "" {
|
|
errorCode = InvalidKey
|
|
return
|
|
}
|
|
var vol *Volume
|
|
if vol, err = o.getVol(param.Bucket()); err != nil {
|
|
log.LogErrorf("getObjectRetentionHandler: load volume fail: requestID(%v) err(%v)",
|
|
GetRequestID(r), err)
|
|
return
|
|
}
|
|
|
|
// get object meta
|
|
start := time.Now()
|
|
_, xattrs, err := vol.ObjectMeta(param.Object())
|
|
span.AppendTrackLog("meta.r", start, err)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectRetentionHandler: get file meta fail: requestId(%v) volume(%v) path(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), param.Object(), err)
|
|
if err == syscall.ENOENT {
|
|
errorCode = NoSuchKey
|
|
}
|
|
return
|
|
}
|
|
retainUntilDate := string(xattrs.Get(XAttrKeyOSSLock))
|
|
if retainUntilDate == "" {
|
|
errorCode = NoSuchObjectLockConfiguration
|
|
return
|
|
}
|
|
retainUntilDateInt64, err := strconv.ParseInt(retainUntilDate, 10, 64)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectRetentionHandler: parse retainUntilDate fail: requestId(%v) volume(%v) path(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), param.Object(), err)
|
|
return
|
|
}
|
|
var objectRetention ObjectRetention
|
|
objectRetention.Mode = ComplianceMode
|
|
objectRetention.RetainUntilDate = RetentionDate{Time: time.Unix(0, retainUntilDateInt64).UTC()}
|
|
b, err := xml.Marshal(objectRetention)
|
|
if err != nil {
|
|
log.LogErrorf("getObjectRetentionHandler: xml marshal fail: requestId(%v) volume(%v) result(%v) err(%v)",
|
|
GetRequestID(r), vol.Name(), objectRetention, err)
|
|
return
|
|
}
|
|
|
|
writeSuccessResponseXML(w, b)
|
|
}
|
|
|
|
func parsePartInfo(partNumber uint64, fileSize uint64) (uint64, uint64, uint64, uint64) {
|
|
var partSize uint64
|
|
var partCount uint64
|
|
var rangeLower uint64
|
|
var rangeUpper uint64
|
|
// partSize, partCount, rangeLower, rangeUpper
|
|
partSizeConst := ParallelDownloadPartSize
|
|
partCount = fileSize / uint64(partSizeConst)
|
|
lastSize := fileSize % uint64(partSizeConst)
|
|
if lastSize > 0 {
|
|
partCount += 1
|
|
}
|
|
|
|
rangeLower = uint64(partSizeConst) * (partNumber - 1)
|
|
if lastSize > 0 && partNumber == partCount {
|
|
partSize = lastSize
|
|
rangeUpper = fileSize - 1
|
|
} else {
|
|
partSize = uint64(partSizeConst)
|
|
rangeUpper = (partSize * partNumber) - 1
|
|
}
|
|
if partNumber > partCount {
|
|
return 0, 0, 0, 0
|
|
}
|
|
return partSize, partCount, rangeLower, rangeUpper
|
|
}
|
|
|
|
func GetContentLength(r *http.Request) int64 {
|
|
dcl := r.Header.Get(HeaderNameXAmzDecodedContentLength)
|
|
if dcl != "" {
|
|
length, err := strconv.ParseInt(dcl, 10, 64)
|
|
if err == nil {
|
|
return length
|
|
}
|
|
}
|
|
return r.ContentLength
|
|
}
|
|
|
|
func VerifyContentLength(r *http.Request, bodyLimit int64) (int64, *ErrorCode) {
|
|
dcl := r.Header.Get(HeaderNameXAmzDecodedContentLength)
|
|
length := r.ContentLength
|
|
if dcl != "" {
|
|
l, err := strconv.ParseInt(dcl, 10, 64)
|
|
if err == nil {
|
|
length = l
|
|
}
|
|
}
|
|
if length > bodyLimit {
|
|
return 0, EntityTooLarge
|
|
}
|
|
if length < 0 {
|
|
return 0, MissingContentLength
|
|
}
|
|
return length, nil
|
|
}
|
