mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-01 09:50:39 +01:00
shichanglin5
1166dead00
by adding processing of metadata and tagging in s3 api CopyObject (judging whether to copy or overwrite according to the directive header)
309 lines
8.8 KiB
Go
309 lines
8.8 KiB
Go
package s3api
|
|
|
|
import (
|
|
"fmt"
|
|
"github.com/chrislusf/seaweedfs/weed/glog"
|
|
headers "github.com/chrislusf/seaweedfs/weed/s3api/http"
|
|
xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
|
|
"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
|
|
"modernc.org/strutil"
|
|
"net/http"
|
|
"net/url"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/chrislusf/seaweedfs/weed/util"
|
|
)
|
|
|
|
func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|
|
|
dstBucket, dstObject := xhttp.GetBucketAndObject(r)
|
|
|
|
// Copy source path.
|
|
cpSrcPath, err := url.QueryUnescape(r.Header.Get("X-Amz-Copy-Source"))
|
|
if err != nil {
|
|
// Save unescaped string as is.
|
|
cpSrcPath = r.Header.Get("X-Amz-Copy-Source")
|
|
}
|
|
|
|
srcBucket, srcObject := pathToBucketAndObject(cpSrcPath)
|
|
|
|
glog.V(3).Infof("CopyObjectHandler %s %s => %s %s", srcBucket, srcObject, dstBucket, dstObject)
|
|
|
|
replaceMeta, replaceTagging := replaceDirective(r)
|
|
|
|
if (srcBucket == dstBucket && srcObject == dstObject || cpSrcPath == "") && (replaceMeta || replaceTagging) {
|
|
fullPath := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, dstBucket, dstObject))
|
|
dir, name := fullPath.DirAndName()
|
|
entry, err := s3a.getEntry(dir, name)
|
|
if err != nil || entry.IsDirectory {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
|
return
|
|
}
|
|
entry.Extended = processMetadataBytes(r.Header, entry.Extended, replaceMeta, replaceTagging)
|
|
err = s3a.touch(dir, name, entry)
|
|
if err != nil {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
|
return
|
|
}
|
|
writeSuccessResponseXML(w, r, CopyObjectResult{
|
|
ETag: fmt.Sprintf("%x", entry.Attributes.Md5),
|
|
LastModified: time.Now().UTC(),
|
|
})
|
|
return
|
|
}
|
|
|
|
// If source object is empty or bucket is empty, reply back invalid copy source.
|
|
if srcObject == "" || srcBucket == "" {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
|
return
|
|
}
|
|
srcPath := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, srcBucket, srcObject))
|
|
dir, name := srcPath.DirAndName()
|
|
if entry, err := s3a.getEntry(dir, name); err != nil || entry.IsDirectory {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
|
return
|
|
}
|
|
|
|
if srcBucket == dstBucket && srcObject == dstObject {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopyDest)
|
|
return
|
|
}
|
|
|
|
dstUrl := fmt.Sprintf("http://%s%s/%s%s?collection=%s",
|
|
s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, dstBucket, urlPathEscape(dstObject), dstBucket)
|
|
srcUrl := fmt.Sprintf("http://%s%s/%s%s",
|
|
s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, srcBucket, urlPathEscape(srcObject))
|
|
|
|
_, _, resp, err := util.DownloadFile(srcUrl, s3a.maybeGetFilerJwtAuthorizationToken(false))
|
|
if err != nil {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
|
return
|
|
}
|
|
defer util.CloseResponse(resp)
|
|
|
|
tagErr := processMetadata(r.Header, resp.Header, replaceMeta, replaceTagging, s3a.getTags, dir, name)
|
|
if tagErr != nil {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
|
return
|
|
}
|
|
glog.V(2).Infof("copy from %s to %s", srcUrl, dstUrl)
|
|
etag, errCode := s3a.putToFiler(r, dstUrl, resp.Body)
|
|
|
|
if errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
}
|
|
|
|
setEtag(w, etag)
|
|
|
|
response := CopyObjectResult{
|
|
ETag: etag,
|
|
LastModified: time.Now().UTC(),
|
|
}
|
|
|
|
writeSuccessResponseXML(w, r, response)
|
|
|
|
}
|
|
|
|
func pathToBucketAndObject(path string) (bucket, object string) {
|
|
path = strings.TrimPrefix(path, "/")
|
|
parts := strings.SplitN(path, "/", 2)
|
|
if len(parts) == 2 {
|
|
return parts[0], "/" + parts[1]
|
|
}
|
|
return parts[0], "/"
|
|
}
|
|
|
|
type CopyPartResult struct {
|
|
LastModified time.Time `xml:"LastModified"`
|
|
ETag string `xml:"ETag"`
|
|
}
|
|
|
|
func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Request) {
|
|
// https://docs.aws.amazon.com/AmazonS3/latest/dev/CopyingObjctsUsingRESTMPUapi.html
|
|
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_UploadPartCopy.html
|
|
dstBucket, _ := xhttp.GetBucketAndObject(r)
|
|
|
|
// Copy source path.
|
|
cpSrcPath, err := url.QueryUnescape(r.Header.Get("X-Amz-Copy-Source"))
|
|
if err != nil {
|
|
// Save unescaped string as is.
|
|
cpSrcPath = r.Header.Get("X-Amz-Copy-Source")
|
|
}
|
|
|
|
srcBucket, srcObject := pathToBucketAndObject(cpSrcPath)
|
|
// If source object is empty or bucket is empty, reply back invalid copy source.
|
|
if srcObject == "" || srcBucket == "" {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
|
return
|
|
}
|
|
|
|
uploadID := r.URL.Query().Get("uploadId")
|
|
partIDString := r.URL.Query().Get("partNumber")
|
|
|
|
partID, err := strconv.Atoi(partIDString)
|
|
if err != nil {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
|
|
return
|
|
}
|
|
|
|
glog.V(3).Infof("CopyObjectPartHandler %s %s => %s part %d", srcBucket, srcObject, dstBucket, partID)
|
|
|
|
// check partID with maximum part ID for multipart objects
|
|
if partID > globalMaxPartID {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxParts)
|
|
return
|
|
}
|
|
|
|
rangeHeader := r.Header.Get("x-amz-copy-source-range")
|
|
|
|
dstUrl := fmt.Sprintf("http://%s%s/%s/%04d.part?collection=%s",
|
|
s3a.option.Filer.ToHttpAddress(), s3a.genUploadsFolder(dstBucket), uploadID, partID, dstBucket)
|
|
srcUrl := fmt.Sprintf("http://%s%s/%s%s",
|
|
s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, srcBucket, urlPathEscape(srcObject))
|
|
|
|
dataReader, err := util.ReadUrlAsReaderCloser(srcUrl, s3a.maybeGetFilerJwtAuthorizationToken(false), rangeHeader)
|
|
if err != nil {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
|
return
|
|
}
|
|
defer dataReader.Close()
|
|
|
|
glog.V(2).Infof("copy from %s to %s", srcUrl, dstUrl)
|
|
etag, errCode := s3a.putToFiler(r, dstUrl, dataReader)
|
|
|
|
if errCode != s3err.ErrNone {
|
|
s3err.WriteErrorResponse(w, r, errCode)
|
|
return
|
|
}
|
|
|
|
setEtag(w, etag)
|
|
|
|
response := CopyPartResult{
|
|
ETag: etag,
|
|
LastModified: time.Now().UTC(),
|
|
}
|
|
|
|
writeSuccessResponseXML(w, r, response)
|
|
|
|
}
|
|
|
|
func replaceDirective(r *http.Request) (replaceMeta, replaceTagging bool) {
|
|
return r.Header.Get(headers.AmzUserMetaDirective) == "REPLACE", r.Header.Get(headers.AmzObjectTaggingDirective) == "REPLACE"
|
|
}
|
|
|
|
func processMetadata(reqHeader, existing http.Header, replaceMeta, replaceTagging bool, getTags func(parentDirectoryPath string, entryName string) (tags map[string]string, err error), dir, name string) (err error) {
|
|
if sc := reqHeader.Get(xhttp.AmzStorageClass); len(sc) == 0 {
|
|
if sc := existing[xhttp.AmzStorageClass]; len(sc) > 0 {
|
|
reqHeader[xhttp.AmzStorageClass] = sc
|
|
}
|
|
}
|
|
|
|
if !replaceMeta {
|
|
for header, _ := range reqHeader {
|
|
if strings.HasPrefix(header, xhttp.AmzUserMetaPrefix) {
|
|
delete(reqHeader, header)
|
|
}
|
|
}
|
|
for k, v := range existing {
|
|
if strings.HasPrefix(k, xhttp.AmzUserMetaPrefix) {
|
|
reqHeader[k] = v
|
|
}
|
|
}
|
|
}
|
|
|
|
if replaceTagging {
|
|
if tags := reqHeader.Get(xhttp.AmzObjectTagging); tags != "" {
|
|
for _, v := range strings.Split(tags, "&") {
|
|
tag := strings.Split(v, "=")
|
|
if len(tag) == 2 {
|
|
reqHeader[xhttp.AmzObjectTagging+"-"+tag[0]] = []string{tag[1]}
|
|
} else if len(tag) == 1 {
|
|
reqHeader[xhttp.AmzObjectTagging+"-"+tag[0]] = []string{}
|
|
}
|
|
}
|
|
}
|
|
delete(reqHeader, xhttp.AmzObjectTagging)
|
|
} else {
|
|
for header, _ := range reqHeader {
|
|
if strings.HasPrefix(header, xhttp.AmzObjectTagging) {
|
|
delete(reqHeader, header)
|
|
}
|
|
}
|
|
|
|
found := false
|
|
for k, _ := range existing {
|
|
if strings.HasPrefix(k, xhttp.AmzObjectTaggingPrefix) {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if found {
|
|
tags, err := getTags(dir, name)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var tagArr []string
|
|
for k, v := range tags {
|
|
tagArr = append(tagArr, fmt.Sprintf("%s=%s", k, v))
|
|
}
|
|
tagStr := strutil.JoinFields(tagArr, "&")
|
|
reqHeader.Set(xhttp.AmzObjectTagging, tagStr)
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
func processMetadataBytes(reqHeader http.Header, existing map[string][]byte, replaceMeta, replaceTagging bool) (metadata map[string][]byte) {
|
|
metadata = make(map[string][]byte)
|
|
|
|
if sc := existing[xhttp.AmzStorageClass]; len(sc) > 0 {
|
|
metadata[xhttp.AmzStorageClass] = sc
|
|
}
|
|
if sc := reqHeader.Get(xhttp.AmzStorageClass); len(sc) > 0 {
|
|
metadata[xhttp.AmzStorageClass] = []byte(sc)
|
|
}
|
|
|
|
if replaceMeta {
|
|
for k, v := range existing {
|
|
if strings.HasPrefix(k, xhttp.AmzUserMetaPrefix) {
|
|
metadata[k] = v
|
|
}
|
|
}
|
|
} else {
|
|
for header, values := range reqHeader {
|
|
if strings.HasPrefix(header, xhttp.AmzUserMetaPrefix) {
|
|
for _, value := range values {
|
|
metadata[header] = []byte(value)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if replaceTagging {
|
|
if tags := reqHeader.Get(xhttp.AmzObjectTagging); tags != "" {
|
|
for _, v := range strings.Split(tags, "&") {
|
|
tag := strings.Split(v, "=")
|
|
if len(tag) == 2 {
|
|
metadata[xhttp.AmzObjectTagging+"-"+tag[0]] = []byte(tag[1])
|
|
} else if len(tag) == 1 {
|
|
metadata[xhttp.AmzObjectTagging+"-"+tag[0]] = nil
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
for k, v := range existing {
|
|
if strings.HasPrefix(k, xhttp.AmzObjectTagging) {
|
|
metadata[k] = v
|
|
}
|
|
}
|
|
delete(metadata, xhttp.AmzTagCount)
|
|
}
|
|
|
|
return
|
|
}
|