feaute: object storage support set/get http headers

This commit is contained in:
Qiu Jian
2019-12-14 17:22:27 +08:00
parent f220d0c89a
commit dd732c2bb1
18 changed files with 689 additions and 126 deletions
+56 -16
View File
@@ -27,6 +27,7 @@ import (
"yunion.io/x/log"
"yunion.io/x/pkg/errors"
"yunion.io/x/s3cli"
"net/http"
)
type TBucketACLType string
@@ -42,6 +43,13 @@ const (
ACLPublicRead = TBucketACLType(s3cli.CANNED_ACL_PUBLIC_READ)
ACLPublicReadWrite = TBucketACLType(s3cli.CANNED_ACL_PUBLIC_READ_WRITE)
ACLUnknown = TBucketACLType("")
META_HEADER_CONTENT_TYPE = "Content-Type"
META_HEADER_CACHE_CONTROL = "Cache-Control"
META_HEADER_CONTENT_DISPOSITION = "Content-Disposition"
META_HEADER_CONTENT_ENCODING = "Content-Encoding"
META_HEADER_CONTENT_LANGUAGE = "Content-Language"
META_HEADER_CONTENT_MD5 = "Content-MD5"
)
type SBucketStats struct {
@@ -69,7 +77,7 @@ type SBaseCloudObject struct {
StorageClass string
ETag string
LastModified time.Time
ContentType string
Meta http.Header
}
type SListObjectResult struct {
@@ -140,14 +148,15 @@ type ICloudBucket interface {
ListObjects(prefix string, marker string, delimiter string, maxCount int) (SListObjectResult, error)
GetIObjects(prefix string, isRecursive bool) ([]ICloudObject, error)
CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, contType string, cannedAcl TBucketACLType, storageClassStr string) error
CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, cannedAcl TBucketACLType, storageClassStr string, meta http.Header) error
GetObject(ctx context.Context, key string, rangeOpt *SGetObjectRange) (io.ReadCloser, error)
DeleteObject(ctx context.Context, keys string) error
GetTempUrl(method string, key string, expire time.Duration) (string, error)
PutObject(ctx context.Context, key string, input io.Reader, sizeBytes int64, contType string, cannedAcl TBucketACLType, storageClassStr string) error
NewMultipartUpload(ctx context.Context, key string, contType string, cannedAcl TBucketACLType, storageClassStr string) (string, error)
PutObject(ctx context.Context, key string, input io.Reader, sizeBytes int64, cannedAcl TBucketACLType, storageClassStr string, meta http.Header) error
NewMultipartUpload(ctx context.Context, key string, cannedAcl TBucketACLType, storageClassStr string, meta http.Header) (string, error)
UploadPart(ctx context.Context, key string, uploadId string, partIndex int, input io.Reader, partSize int64) (string, error)
CopyPart(ctx context.Context, key string, uploadId string, partIndex int, srcBucketName string, srcKey string, srcOffset int64, srcLength int64) (string, error)
CompleteMultipartUpload(ctx context.Context, key string, uploadId string, partEtags []string) error
@@ -162,7 +171,7 @@ type ICloudObject interface {
GetLastModified() time.Time
GetStorageClass() string
GetETag() string
GetContentType() string
GetMeta() http.Header
GetAcl() TBucketACLType
SetAcl(acl TBucketACLType) error
@@ -175,7 +184,7 @@ func ICloudObject2JSONObject(obj ICloudObject) jsonutils.JSONObject {
StorageClass string
ETag string
LastModified time.Time
ContentType string
Meta http.Header
Acl string
}{
Key: obj.GetKey(),
@@ -183,7 +192,7 @@ func ICloudObject2JSONObject(obj ICloudObject) jsonutils.JSONObject {
StorageClass: obj.GetStorageClass(),
ETag: obj.GetETag(),
LastModified: obj.GetLastModified(),
ContentType: obj.GetContentType(),
Meta: obj.GetMeta(),
Acl: string(obj.GetAcl()),
}
return jsonutils.Marshal(obj2)
@@ -209,8 +218,8 @@ func (o *SBaseCloudObject) GetETag() string {
return o.ETag
}
func (o *SBaseCloudObject) GetContentType() string {
return o.ContentType
func (o *SBaseCloudObject) GetMeta() http.Header {
return o.Meta
}
func GetIBucketById(region ICloudRegion, name string) (ICloudBucket, error) {
@@ -323,14 +332,14 @@ func Makedir(ctx context.Context, bucket ICloudBucket, key string) error {
}
}
path := strings.Join(segs, "/") + "/"
err := bucket.PutObject(ctx, path, strings.NewReader(""), 0, "", ACLPrivate, "")
err := bucket.PutObject(ctx, path, strings.NewReader(""), 0, bucket.GetAcl(), "", nil)
if err != nil {
return errors.Wrap(err, "PutObject")
}
return nil
}
func UploadObject(ctx context.Context, bucket ICloudBucket, key string, blocksz int64, input io.Reader, sizeBytes int64, contType string, cannedAcl TBucketACLType, storageClass string, debug bool) error {
func UploadObject(ctx context.Context, bucket ICloudBucket, key string, blocksz int64, input io.Reader, sizeBytes int64, cannedAcl TBucketACLType, storageClass string, meta http.Header, debug bool) error {
if blocksz <= 0 {
blocksz = MAX_PUT_OBJECT_SIZEBYTES
}
@@ -338,7 +347,7 @@ func UploadObject(ctx context.Context, bucket ICloudBucket, key string, blocksz
if debug {
log.Debugf("too small, put object in one shot")
}
return bucket.PutObject(ctx, key, input, sizeBytes, contType, cannedAcl, storageClass)
return bucket.PutObject(ctx, key, input, sizeBytes, cannedAcl, storageClass, meta)
}
partSize := blocksz
partCount := sizeBytes / partSize
@@ -358,7 +367,7 @@ func UploadObject(ctx context.Context, bucket ICloudBucket, key string, blocksz
if debug {
log.Debugf("multipart upload part count %d part size %d", partCount, partSize)
}
uploadId, err := bucket.NewMultipartUpload(ctx, key, contType, cannedAcl, storageClass)
uploadId, err := bucket.NewMultipartUpload(ctx, key, cannedAcl, storageClass, meta)
if err != nil {
return errors.Wrap(err, "bucket.NewMultipartUpload")
}
@@ -407,7 +416,31 @@ func DeletePrefix(ctx context.Context, bucket ICloudBucket, prefix string) error
return nil
}
func CopyObject(ctx context.Context, blocksz int64, dstBucket ICloudBucket, dstKey string, srcBucket ICloudBucket, srcKey string, debug bool) error {
func MergeMeta(src http.Header, dst http.Header) http.Header {
if src != nil && dst != nil {
ret := http.Header{}
for k, vs := range src {
for _, v := range vs {
ret.Add(k, v)
}
}
for k, vs := range dst {
for _, v := range vs {
ret.Add(k, v)
}
}
return ret
} else if src != nil && dst == nil {
return src
} else if src == nil && dst != nil {
return dst
} else {
return nil
}
}
func CopyObject(ctx context.Context, blocksz int64, dstBucket ICloudBucket, dstKey string, srcBucket ICloudBucket, srcKey string, dstMeta http.Header, debug bool) error {
srcObj, err := GetIObject(srcBucket, srcKey)
if err != nil {
return errors.Wrap(err, "GetIObject")
@@ -425,7 +458,7 @@ func CopyObject(ctx context.Context, blocksz int64, dstBucket ICloudBucket, dstK
return errors.Wrap(err, "srcBucket.GetObject")
}
defer srcStream.Close()
err = dstBucket.PutObject(ctx, dstKey, srcStream, sizeBytes, srcObj.GetContentType(), srcObj.GetAcl(), srcObj.GetStorageClass())
err = dstBucket.PutObject(ctx, dstKey, srcStream, sizeBytes, srcObj.GetAcl(), srcObj.GetStorageClass(), MergeMeta(srcObj.GetMeta(), dstMeta))
if err != nil {
return errors.Wrap(err, "dstBucket.PutObject")
}
@@ -449,7 +482,7 @@ func CopyObject(ctx context.Context, blocksz int64, dstBucket ICloudBucket, dstK
if debug {
log.Debugf("multipart upload part count %d part size %d", partCount, partSize)
}
uploadId, err := dstBucket.NewMultipartUpload(ctx, dstKey, srcObj.GetContentType(), srcObj.GetAcl(), srcObj.GetStorageClass())
uploadId, err := dstBucket.NewMultipartUpload(ctx, dstKey, srcObj.GetAcl(), srcObj.GetStorageClass(), MergeMeta(srcObj.GetMeta(), dstMeta))
if err != nil {
return errors.Wrap(err, "bucket.NewMultipartUpload")
}
@@ -513,3 +546,10 @@ func CopyPart(ctx context.Context,
}
return etag, nil
}
func ObjectSetMeta(ctx context.Context,
bucket ICloudBucket, obj ICloudObject,
meta http.Header,
) error {
return bucket.CopyObject(ctx, obj.GetKey(), bucket.GetName(), obj.GetKey(), obj.GetAcl(), obj.GetStorageClass(), meta)
}
+5 -1
View File
@@ -935,7 +935,11 @@ func (bucket *SBucket) PerformUpload(
}
}
err = cloudprovider.UploadObject(ctx, iBucket, key, 0, appParams.Request.Body, sizeBytes, contType, cloudprovider.TBucketACLType(aclStr), storageClass, false)
meta := http.Header{}
if len(contType) > 0 {
meta.Add(cloudprovider.META_HEADER_CONTENT_TYPE, contType)
}
err = cloudprovider.UploadObject(ctx, iBucket, key, 0, appParams.Request.Body, sizeBytes, cloudprovider.TBucketACLType(aclStr), storageClass, meta, false)
if err != nil {
return nil, httperrors.NewInternalServerError("put object error %s", err)
}
+56 -28
View File
@@ -27,6 +27,7 @@ import (
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/multicloud"
"net/http"
)
type SBucket struct {
@@ -160,7 +161,6 @@ func (b *SBucket) ListObjects(prefix string, marker string, delimiter string, ma
SizeBytes: object.Size,
ETag: object.ETag,
LastModified: object.LastModified,
ContentType: object.Type,
},
}
result.Objects = append(result.Objects, obj)
@@ -183,7 +183,32 @@ func (b *SBucket) GetIObjects(prefix string, isRecursive bool) ([]cloudprovider.
return cloudprovider.GetIObjects(b, prefix, isRecursive)
}
func (b *SBucket) PutObject(ctx context.Context, key string, input io.Reader, sizeBytes int64, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func metaOpts(opts []oss.Option, meta http.Header) []oss.Option {
for k, v := range meta {
if len(v) == 0 {
continue
}
switch http.CanonicalHeaderKey(k) {
case cloudprovider.META_HEADER_CONTENT_TYPE:
opts = append(opts, oss.ContentType(v[0]))
case cloudprovider.META_HEADER_CONTENT_MD5:
opts = append(opts, oss.ContentMD5(v[0]))
case cloudprovider.META_HEADER_CONTENT_LANGUAGE:
opts = append(opts, oss.ContentLanguage(v[0]))
case cloudprovider.META_HEADER_CONTENT_ENCODING:
opts = append(opts, oss.ContentEncoding(v[0]))
case cloudprovider.META_HEADER_CONTENT_DISPOSITION:
opts = append(opts, oss.ContentDisposition(v[0]))
case cloudprovider.META_HEADER_CACHE_CONTROL:
opts = append(opts, oss.CacheControl(v[0]))
default:
opts = append(opts, oss.Meta(http.CanonicalHeaderKey(k), v[0]))
}
}
return opts
}
func (b *SBucket) PutObject(ctx context.Context, key string, input io.Reader, sizeBytes int64, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
osscli, err := b.region.GetOssClient()
if err != nil {
return errors.Wrap(err, "GetOssClient")
@@ -196,16 +221,17 @@ func (b *SBucket) PutObject(ctx context.Context, key string, input io.Reader, si
if sizeBytes > 0 {
opts = append(opts, oss.ContentLength(sizeBytes))
}
if len(contType) > 0 {
opts = append(opts, oss.ContentType(contType))
if meta != nil {
opts = metaOpts(opts, meta)
}
if len(cannedAcl) > 0 {
acl, err := str2Acl(string(cannedAcl))
if err != nil {
return errors.Wrap(err, "")
}
opts = append(opts, oss.ObjectACL(acl))
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
acl, err := str2Acl(string(cannedAcl))
if err != nil {
return errors.Wrap(err, "")
}
opts = append(opts, oss.ObjectACL(acl))
if len(storageClassStr) > 0 {
storageClass, err := str2StorageClass(storageClassStr)
if err != nil {
@@ -216,7 +242,7 @@ func (b *SBucket) PutObject(ctx context.Context, key string, input io.Reader, si
return bucket.PutObject(key, input, opts...)
}
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) (string, error) {
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) (string, error) {
osscli, err := b.region.GetOssClient()
if err != nil {
return "", errors.Wrap(err, "GetOssClient")
@@ -226,16 +252,17 @@ func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType s
return "", errors.Wrap(err, "Bucket")
}
opts := make([]oss.Option, 0)
if len(contType) > 0 {
opts = append(opts, oss.ContentType(contType))
if meta != nil {
opts = metaOpts(opts, meta)
}
if len(cannedAcl) > 0 {
acl, err := str2Acl(string(cannedAcl))
if err != nil {
return "", errors.Wrap(err, "")
}
opts = append(opts, oss.ObjectACL(acl))
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
acl, err := str2Acl(string(cannedAcl))
if err != nil {
return "", errors.Wrap(err, "str2Acl")
}
opts = append(opts, oss.ObjectACL(acl))
if len(storageClassStr) > 0 {
storageClass, err := str2StorageClass(storageClassStr)
if err != nil {
@@ -361,7 +388,7 @@ func (b *SBucket) GetTempUrl(method string, key string, expire time.Duration) (s
return urlStr, nil
}
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
osscli, err := b.region.GetOssClient()
if err != nil {
return errors.Wrap(err, "GetOssClient")
@@ -371,16 +398,17 @@ func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, src
return errors.Wrap(err, "Bucket")
}
opts := make([]oss.Option, 0)
if len(contType) > 0 {
opts = append(opts, oss.ContentType(contType))
if meta != nil {
opts = metaOpts(opts, meta)
}
if len(cannedAcl) > 0 {
acl, err := str2Acl(string(cannedAcl))
if err != nil {
return errors.Wrap(err, "")
}
opts = append(opts, oss.ObjectACL(acl))
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
acl, err := str2Acl(string(cannedAcl))
if err != nil {
return errors.Wrap(err, "str2Acl")
}
opts = append(opts, oss.ObjectACL(acl))
if len(storageClassStr) > 0 {
storageClass, err := str2StorageClass(storageClassStr)
if err != nil {
+24
View File
@@ -21,6 +21,7 @@ import (
"yunion.io/x/log"
"yunion.io/x/onecloud/pkg/cloudprovider"
"net/http"
)
type SObject struct {
@@ -76,3 +77,26 @@ func (o *SObject) SetAcl(aclStr cloudprovider.TBucketACLType) error {
}
return nil
}
func (o *SObject) GetMeta() http.Header {
if o.Meta != nil {
return o.Meta
}
osscli, err := o.bucket.region.GetOssClient()
if err != nil {
log.Errorf("o.bucket.region.GetOssClient error %s", err)
return nil
}
bucket, err := osscli.Bucket(o.bucket.Name)
if err != nil {
log.Errorf("osscli.Bucket error %s", err)
return nil
}
result, err := bucket.GetObjectDetailedMeta(o.Key)
if err != nil {
log.Errorf("bucket.GetObjectACL error %s", err)
return nil
}
o.Meta = result
return result
}
+1 -1
View File
@@ -178,7 +178,7 @@ func (self *SStoragecache) uploadImage(ctx context.Context, userCred mcclient.To
return "", err
}
log.Debugf("To upload image to bucket %s ...", bucketName)
err = cloudprovider.UploadObject(context.Background(), bucket, image.ImageId, 0, reader, sizeByte, "", "", "", false)
err = cloudprovider.UploadObject(context.Background(), bucket, image.ImageId, 0, reader, sizeByte, "", "", nil, false)
// err = bucket.PutObject(image.ImageId, reader)
if err != nil {
log.Errorf("PutObject error %s %s", image.ImageId, err)
+88 -13
View File
@@ -31,6 +31,7 @@ import (
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/multicloud"
"yunion.io/x/onecloud/pkg/util/fileutils2"
"net/http"
)
type SBucket struct {
@@ -183,7 +184,6 @@ func (b *SBucket) ListObjects(prefix string, marker string, delimiter string, ma
SizeBytes: *object.Size,
ETag: *object.ETag,
LastModified: *object.LastModified,
ContentType: "",
},
}
result.Objects = append(result.Objects, obj)
@@ -210,7 +210,7 @@ func (b *SBucket) GetIObjects(prefix string, isRecursive bool) ([]cloudprovider.
return cloudprovider.GetIObjects(b, prefix, isRecursive)
}
func (b *SBucket) PutObject(ctx context.Context, key string, body io.Reader, sizeBytes int64, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) PutObject(ctx context.Context, key string, body io.Reader, sizeBytes int64, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
if sizeBytes < 0 {
return errors.Error("content length expected")
}
@@ -228,12 +228,37 @@ func (b *SBucket) PutObject(ctx context.Context, key string, body io.Reader, siz
defer seeker.Close()
input.SetBody(seeker)
input.SetContentLength(sizeBytes)
if len(contType) > 0 {
input.SetContentType(contType)
if meta != nil {
metaHdr := make(map[string]*string)
for k, v := range meta {
if len(v) == 0 || len(v[0]) == 0 {
continue
}
switch http.CanonicalHeaderKey(k) {
case cloudprovider.META_HEADER_CACHE_CONTROL:
input.SetCacheControl(v[0])
case cloudprovider.META_HEADER_CONTENT_TYPE:
input.SetContentType(v[0])
case cloudprovider.META_HEADER_CONTENT_MD5:
input.SetContentMD5(v[0])
case cloudprovider.META_HEADER_CONTENT_LANGUAGE:
input.SetContentLanguage(v[0])
case cloudprovider.META_HEADER_CONTENT_ENCODING:
input.SetContentEncoding(v[0])
case cloudprovider.META_HEADER_CONTENT_DISPOSITION:
input.SetContentDisposition(v[0])
default:
metaHdr[k] = &v[0]
}
}
if len(metaHdr) > 0 {
input.SetMetadata(metaHdr)
}
}
if len(cannedAcl) > 0 {
input.SetACL(string(cannedAcl))
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
input.SetACL(string(cannedAcl))
if len(storageClassStr) > 0 {
input.SetStorageClass(storageClassStr)
}
@@ -244,7 +269,7 @@ func (b *SBucket) PutObject(ctx context.Context, key string, body io.Reader, siz
return nil
}
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) (string, error) {
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) (string, error) {
s3cli, err := b.region.GetS3Client()
if err != nil {
return "", errors.Wrap(err, "GetS3Client")
@@ -252,12 +277,35 @@ func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType s
input := &s3.CreateMultipartUploadInput{}
input.SetBucket(b.Name)
input.SetKey(key)
if len(contType) > 0 {
input.SetContentType(contType)
if meta != nil {
metaHdr := make(map[string]*string)
for k, v := range meta {
if len(v) == 0 || len(v[0]) == 0 {
continue
}
switch http.CanonicalHeaderKey(k) {
case cloudprovider.META_HEADER_CACHE_CONTROL:
input.SetCacheControl(v[0])
case cloudprovider.META_HEADER_CONTENT_TYPE:
input.SetContentType(v[0])
case cloudprovider.META_HEADER_CONTENT_LANGUAGE:
input.SetContentLanguage(v[0])
case cloudprovider.META_HEADER_CONTENT_ENCODING:
input.SetContentEncoding(v[0])
case cloudprovider.META_HEADER_CONTENT_DISPOSITION:
input.SetContentDisposition(v[0])
default:
metaHdr[k] = &v[0]
}
}
if len(metaHdr) > 0 {
input.SetMetadata(metaHdr)
}
}
if len(cannedAcl) > 0 {
input.SetACL(string(cannedAcl))
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
input.SetACL(string(cannedAcl))
if len(storageClassStr) > 0 {
input.SetStorageClass(storageClassStr)
}
@@ -380,7 +428,7 @@ func (b *SBucket) GetTempUrl(method string, key string, expire time.Duration) (s
return url, nil
}
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
s3cli, err := b.region.GetS3Client()
if err != nil {
return errors.Wrap(err, "GetS3Client")
@@ -391,8 +439,35 @@ func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, src
input.SetKey(destKey)
input.SetCopySource(fmt.Sprintf("%s/%s", srcBucket, url.PathEscape(srcKey)))
input.SetStorageClass(storageClassStr)
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
input.SetACL(string(cannedAcl))
input.SetContentType(contType)
if meta != nil {
metaHdr := make(map[string]*string)
for k, v := range meta {
if len(v) == 0 || len(v[0]) == 0 {
continue
}
switch http.CanonicalHeaderKey(k) {
case cloudprovider.META_HEADER_CACHE_CONTROL:
input.SetCacheControl(v[0])
case cloudprovider.META_HEADER_CONTENT_TYPE:
input.SetContentType(v[0])
case cloudprovider.META_HEADER_CONTENT_LANGUAGE:
input.SetContentLanguage(v[0])
case cloudprovider.META_HEADER_CONTENT_ENCODING:
input.SetContentEncoding(v[0])
case cloudprovider.META_HEADER_CONTENT_DISPOSITION:
input.SetContentDisposition(v[0])
default:
metaHdr[k] = &v[0]
}
}
if len(metaHdr) > 0 {
input.SetMetadata(metaHdr)
}
}
_, err = s3cli.CopyObject(input)
if err != nil {
return errors.Wrap(err, "CopyObject")
+43
View File
@@ -21,6 +21,7 @@ import (
"yunion.io/x/pkg/errors"
"yunion.io/x/onecloud/pkg/cloudprovider"
"net/http"
)
type SObject struct {
@@ -66,3 +67,45 @@ func (o *SObject) SetAcl(aclStr cloudprovider.TBucketACLType) error {
}
return nil
}
func (o *SObject) GetMeta() http.Header {
log.Infof("GetMeta for %s/%s", o.bucket.Name, o.Key)
if o.Meta != nil {
return o.Meta
}
s3cli, err := o.bucket.region.GetS3Client()
if err != nil {
log.Errorf("o.bucket.region.GetS3Client fail %s", err)
return nil
}
input := &s3.HeadObjectInput{}
input.SetBucket(o.bucket.Name)
input.SetKey(o.Key)
output, err := s3cli.HeadObject(input)
if err != nil {
log.Errorf("s3cli.HeadObject fail %s", err)
return nil
}
ret := http.Header{}
for k, v := range output.Metadata {
if v != nil && len(*v) > 0 {
ret.Add(k, *v)
}
}
if output.CacheControl != nil && len(*output.CacheControl) > 0 {
ret.Set(cloudprovider.META_HEADER_CACHE_CONTROL, *output.CacheControl)
}
if output.ContentType != nil && len(*output.ContentType) > 0 {
ret.Set(cloudprovider.META_HEADER_CONTENT_TYPE, *output.ContentType)
}
if output.ContentDisposition != nil && len(*output.ContentDisposition) > 0 {
ret.Set(cloudprovider.META_HEADER_CONTENT_DISPOSITION, *output.ContentDisposition)
}
if output.ContentEncoding != nil && len(*output.ContentEncoding) > 0 {
ret.Set(cloudprovider.META_HEADER_CONTENT_ENCODING, *output.ContentEncoding)
}
if output.ContentLanguage != nil && len(*output.ContentLanguage) > 0 {
ret.Set(cloudprovider.META_HEADER_CONTENT_LANGUAGE, *output.ContentLanguage)
}
return nil
}
+1 -1
View File
@@ -187,7 +187,7 @@ func (self *SStoragecache) uploadImage(ctx context.Context, userCred mcclient.To
if err != nil {
return "", errors.Wrap(err, "GetIBucketByName")
}
err = cloudprovider.UploadObject(ctx, bucket, image.ImageId, 0, reader, sizeBytes, "", "", "", false)
err = cloudprovider.UploadObject(ctx, bucket, image.ImageId, 0, reader, sizeBytes, "", "", nil, false)
if err != nil {
return "", errors.Wrap(err, "cloudprovider.UploadObject")
}
+57 -10
View File
@@ -37,6 +37,7 @@ import (
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/httperrors"
"yunion.io/x/onecloud/pkg/multicloud"
"net/http"
)
type SContainer struct {
@@ -524,7 +525,7 @@ func (self *SContainer) CopySnapshot(snapshotId, fileName string) (*storage.Blob
return blobRef, blobRef.GetProperties(&storage.GetBlobPropertiesOptions{})
}
func (self *SContainer) UploadStream(key string, reader io.Reader, contType string) error {
func (self *SContainer) UploadStream(key string, reader io.Reader, meta http.Header) error {
blobService, err := self.storageaccount.getBlobServiceClient()
if err != nil {
return errors.Wrap(err, "getBlobServiceClient")
@@ -532,7 +533,32 @@ func (self *SContainer) UploadStream(key string, reader io.Reader, contType stri
containerRef := blobService.GetContainerReference(self.Name)
blobRef := containerRef.GetBlobReference(key)
blobRef.Properties.BlobType = storage.BlobTypeBlock
blobRef.Properties.ContentType = contType
if meta != nil {
for k, v := range meta {
if len(v) == 0 || len(v[0]) == 0{
continue
}
switch http.CanonicalHeaderKey(k) {
case cloudprovider.META_HEADER_CACHE_CONTROL:
blobRef.Properties.CacheControl = v[0]
case cloudprovider.META_HEADER_CONTENT_TYPE:
blobRef.Properties.ContentType = v[0]
case cloudprovider.META_HEADER_CONTENT_MD5:
blobRef.Properties.ContentMD5 = v[0]
case cloudprovider.META_HEADER_CONTENT_ENCODING:
blobRef.Properties.ContentEncoding = v[0]
case cloudprovider.META_HEADER_CONTENT_LANGUAGE:
blobRef.Properties.ContentLanguage = v[0]
case cloudprovider.META_HEADER_CONTENT_DISPOSITION:
blobRef.Properties.ContentDisposition = v[0]
default:
if blobRef.Metadata == nil {
blobRef.Metadata = storage.BlobMetadata{}
}
blobRef.Metadata[k] = v[0]
}
}
}
return blobRef.CreateBlockBlobFromReader(reader, &storage.PutBlobOptions{})
}
@@ -692,12 +718,12 @@ func (self *SStorageAccount) getOrCreateContainer(containerName string, create b
return container, nil
}
func (self *SStorageAccount) UploadStream(containerName string, key string, reader io.Reader, contType string) error {
func (self *SStorageAccount) UploadStream(containerName string, key string, reader io.Reader, meta http.Header) error {
container, err := self.getOrCreateContainer(containerName, true)
if err != nil {
return errors.Wrap(err, "getOrCreateContainer")
}
return container.UploadStream(key, reader, contType)
return container.UploadStream(key, reader, meta)
}
func (b *SStorageAccount) MaxPartSizeBytes() int64 {
@@ -896,6 +922,28 @@ func (b *SStorageAccount) ListObjects(prefix string, marker string, delimiter st
}
for i := range oResult.Blobs {
blob := oResult.Blobs[i]
meta := http.Header{}
for k, v := range blob.Metadata {
meta.Add(k, v)
}
if len(blob.Properties.CacheControl) > 0 {
meta.Set(cloudprovider.META_HEADER_CACHE_CONTROL, blob.Properties.CacheControl)
}
if len(blob.Properties.ContentType) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_TYPE, blob.Properties.ContentType)
}
if len(blob.Properties.ContentDisposition) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_DISPOSITION, blob.Properties.ContentDisposition)
}
if len(blob.Properties.ContentLanguage) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_LANGUAGE, blob.Properties.ContentLanguage)
}
if len(blob.Properties.ContentEncoding) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_ENCODING, blob.Properties.ContentEncoding)
}
if len(blob.Properties.ContentMD5) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_MD5, blob.Properties.ContentMD5)
}
o := &SObject{
container: &container,
SBaseCloudObject: cloudprovider.SBaseCloudObject{
@@ -904,7 +952,7 @@ func (b *SStorageAccount) ListObjects(prefix string, marker string, delimiter st
StorageClass: "",
ETag: blob.Properties.Etag,
LastModified: time.Time(blob.Properties.LastModified),
ContentType: blob.Properties.ContentType,
Meta: meta,
},
}
result.Objects = append(result.Objects, o)
@@ -956,14 +1004,14 @@ func splitKeyAndBlob(path string) (string, string, error) {
return containerName, key, nil
}
func (b *SStorageAccount) PutObject(ctx context.Context, key string, reader io.Reader, sizeBytes int64, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SStorageAccount) PutObject(ctx context.Context, key string, reader io.Reader, sizeBytes int64, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
containerName, blob, err := splitKey(key)
if err != nil {
return errors.Wrap(err, "splitKey")
}
if len(blob) > 0 {
// put blob
err = b.UploadStream(containerName, blob, reader, contType)
err = b.UploadStream(containerName, blob, reader, meta)
if err != nil {
return errors.Wrap(err, "UploadStream")
}
@@ -980,7 +1028,7 @@ func (b *SStorageAccount) PutObject(ctx context.Context, key string, reader io.R
return nil
}
func (b *SStorageAccount) NewMultipartUpload(ctx context.Context, key string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) (string, error) {
func (b *SStorageAccount) NewMultipartUpload(ctx context.Context, key string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) (string, error) {
containerName, blob, err := splitKeyAndBlob(key)
if err != nil {
return "", errors.Wrap(err, "splitKey")
@@ -999,7 +1047,6 @@ func (b *SStorageAccount) NewMultipartUpload(ctx context.Context, key string, co
if err != nil {
return "", errors.Wrap(err, "CreateBlockBlob")
}
uploadId, err := blobRef.AcquireLease(-1, "", nil)
if err != nil {
return "", errors.Wrap(err, "blobRef.AcquireLease")
@@ -1149,7 +1196,7 @@ func (b *SStorageAccount) GetTempUrl(method string, key string, expire time.Dura
return container.SignUrl(method, blob, expire)
}
func (b *SStorageAccount) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SStorageAccount) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
srcIBucket, err := b.region.GetIBucketByName(srcBucket)
if err != nil {
return errors.Wrap(err, "GetIBucketByName")
+70 -16
View File
@@ -27,6 +27,8 @@ import (
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/multicloud"
"yunion.io/x/onecloud/pkg/multicloud/huawei/obs"
"net/http"
"yunion.io/x/pkg/utils"
)
type SBucket struct {
@@ -205,7 +207,6 @@ func (b *SBucket) ListObjects(prefix string, marker string, delimiter string, ma
SizeBytes: object.Size,
ETag: object.ETag,
LastModified: object.LastModified,
ContentType: "",
},
}
result.Objects = append(result.Objects, obj)
@@ -227,7 +228,7 @@ func (b *SBucket) ListObjects(prefix string, marker string, delimiter string, ma
return result, nil
}
func (b *SBucket) PutObject(ctx context.Context, key string, reader io.Reader, sizeBytes int64, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) PutObject(ctx context.Context, key string, reader io.Reader, sizeBytes int64, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
obscli, err := b.region.getOBSClient()
if err != nil {
return errors.Wrap(err, "GetOBSClient")
@@ -246,11 +247,32 @@ func (b *SBucket) PutObject(ctx context.Context, key string, reader io.Reader, s
return err
}
}
if len(cannedAcl) > 0 {
input.ACL = obs.AclType(string(cannedAcl))
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
if len(contType) > 0 {
input.ContentType = contType
input.ACL = obs.AclType(string(cannedAcl))
if meta != nil {
val := meta.Get(cloudprovider.META_HEADER_CONTENT_TYPE)
if len(val) > 0 {
input.ContentType = val
}
val = meta.Get(cloudprovider.META_HEADER_CONTENT_MD5)
if len(val) > 0 {
input.ContentMD5 = val
}
extraMeta := make(map[string]string)
for k, v := range meta {
if utils.IsInStringArray(k, []string {
cloudprovider.META_HEADER_CONTENT_TYPE,
cloudprovider.META_HEADER_CONTENT_MD5,
}) {
continue
}
if len(v[0]) > 0 {
extraMeta[k] = v[0]
}
}
input.Metadata = extraMeta
}
_, err = obscli.PutObject(input)
if err != nil {
@@ -259,7 +281,7 @@ func (b *SBucket) PutObject(ctx context.Context, key string, reader io.Reader, s
return nil
}
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) (string, error) {
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) (string, error) {
obscli, err := b.region.getOBSClient()
if err != nil {
return "", errors.Wrap(err, "GetOBSClient")
@@ -268,12 +290,28 @@ func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType s
input := &obs.InitiateMultipartUploadInput{}
input.Bucket = b.Name
input.Key = key
if len(contType) > 0 {
input.ContentType = contType
if meta != nil {
val := meta.Get(cloudprovider.META_HEADER_CONTENT_TYPE)
if len(val) > 0 {
input.ContentType = val
}
extraMeta := make(map[string]string)
for k, v := range meta {
if utils.IsInStringArray(k, []string {
cloudprovider.META_HEADER_CONTENT_TYPE,
}) {
continue
}
if len(v[0]) > 0 {
extraMeta[k] = v[0]
}
}
input.Metadata = extraMeta
}
if len(cannedAcl) > 0 {
input.ACL = obs.AclType(string(cannedAcl))
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
input.ACL = obs.AclType(string(cannedAcl))
if len(storageClassStr) > 0 {
input.StorageClass, err = str2StorageClass(storageClassStr)
if err != nil {
@@ -428,7 +466,7 @@ func (b *SBucket) SetLimit(limit cloudprovider.SBucketStats) error {
return nil
}
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
obscli, err := b.region.getOBSClient()
if err != nil {
return errors.Wrap(err, "GetOBSClient")
@@ -442,11 +480,27 @@ func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, src
return err
}
}
if len(cannedAcl) > 0 {
input.ACL = obs.AclType(string(cannedAcl))
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
if len(contType) > 0 {
input.ContentType = contType
input.ACL = obs.AclType(string(cannedAcl))
if meta != nil {
val := meta.Get(cloudprovider.META_HEADER_CONTENT_TYPE)
if len(val) > 0 {
input.ContentType = val
}
extraMeta := make(map[string]string)
for k, v := range meta {
if utils.IsInStringArray(k, []string {
cloudprovider.META_HEADER_CONTENT_TYPE,
}) {
continue
}
if len(v[0]) > 0 {
extraMeta[k] = v[0]
}
}
input.Metadata = extraMeta
}
_, err = obscli.CopyObject(input)
if err != nil {
+29
View File
@@ -20,6 +20,7 @@ import (
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/multicloud/huawei/obs"
"net/http"
)
type SObject struct {
@@ -62,3 +63,31 @@ func (o *SObject) SetAcl(aclStr cloudprovider.TBucketACLType) error {
}
return nil
}
func (o *SObject) GetMeta() http.Header {
if o.Meta != nil {
return o.Meta
}
obscli, err := o.bucket.region.getOBSClient()
if err != nil {
log.Errorf("getOBSClient fail %s", err)
return nil
}
input := &obs.GetObjectMetadataInput{}
input.Bucket = o.bucket.Name
input.Key = o.Key
output, err := obscli.GetObjectMetadata(input)
if err != nil {
log.Errorf("obscli.GetObjectMetadata fail %s", err)
return nil
}
meta := http.Header{}
for k, v := range output.Metadata {
meta.Add(k, v)
}
if len(output.ContentType) > 0 {
meta.Add(cloudprovider.META_HEADER_CONTENT_TYPE, output.ContentType)
}
o.Meta = meta
return meta
}
+1 -1
View File
@@ -194,7 +194,7 @@ func (self *SStoragecache) uploadImage(ctx context.Context, userCred mcclient.To
return "", errors.Wrap(err, "GetIBucketByName")
}
err = cloudprovider.UploadObject(context.Background(), bucket, image.ImageId, 0, reader, sizeByte, "", "", "", false)
err = cloudprovider.UploadObject(context.Background(), bucket, image.ImageId, 0, reader, sizeByte, "", "", nil, false)
if err != nil {
return "", errors.Wrap(err, "cloudprovider.UploadObject")
}
+65 -13
View File
@@ -28,6 +28,7 @@ import (
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/multicloud"
"yunion.io/x/pkg/utils"
)
type SBucket struct {
@@ -141,6 +142,9 @@ func (bucket *SBucket) ListObjects(prefix string, marker string, delimiter strin
ret.Objects = make([]cloudprovider.ICloudObject, len(result.Contents))
for i := range result.Contents {
object := result.Contents[i]
if len(object.ContentType) > 0 {
object.Metadata.Set(cloudprovider.META_HEADER_CONTENT_TYPE, object.ContentType)
}
ret.Objects[i] = &SObject{
bucket: bucket,
SBaseCloudObject: cloudprovider.SBaseCloudObject{
@@ -149,7 +153,7 @@ func (bucket *SBucket) ListObjects(prefix string, marker string, delimiter strin
SizeBytes: object.Size,
ETag: object.ETag,
LastModified: object.LastModified,
ContentType: object.ContentType,
Meta: object.Metadata,
},
}
}
@@ -169,6 +173,9 @@ func (bucket *SBucket) GetIObjects(prefix string, isRecursive bool) ([]cloudprov
if !isRecursive && prefix == object.Key {
continue
}
if len(object.ContentType) > 0 {
object.Metadata.Set(cloudprovider.META_HEADER_CONTENT_TYPE, object.ContentType)
}
obj := &SObject{
bucket: bucket,
SBaseCloudObject: cloudprovider.SBaseCloudObject{
@@ -177,7 +184,7 @@ func (bucket *SBucket) GetIObjects(prefix string, isRecursive bool) ([]cloudprov
SizeBytes: object.Size,
ETag: object.ETag,
LastModified: object.LastModified,
ContentType: object.ContentType,
Meta: object.Metadata,
},
}
ret = append(ret, obj)
@@ -185,11 +192,51 @@ func (bucket *SBucket) GetIObjects(prefix string, isRecursive bool) ([]cloudprov
return ret, nil
}
func (bucket *SBucket) PutObject(ctx context.Context, key string, input io.Reader, sizeBytes int64, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func metaInitOptions(meta http.Header) s3cli.PutObjectOptions {
opts := s3cli.PutObjectOptions{}
if len(contType) > 0 {
opts.ContentType = contType
if meta != nil {
val := meta.Get(cloudprovider.META_HEADER_CONTENT_TYPE)
if len(val) > 0 {
opts.ContentType = val
}
val = meta.Get(cloudprovider.META_HEADER_CACHE_CONTROL)
if len(val) > 0 {
opts.CacheControl = val
}
val = meta.Get(cloudprovider.META_HEADER_CONTENT_DISPOSITION)
if len(val) > 0 {
opts.ContentDisposition = val
}
val = meta.Get(cloudprovider.META_HEADER_CONTENT_ENCODING)
if len(val) > 0 {
opts.ContentEncoding = val
}
val = meta.Get(cloudprovider.META_HEADER_CONTENT_LANGUAGE)
if len(val) > 0 {
opts.ContentLanguage = val
}
userMeta := make(map[string]string)
for k, v := range meta {
if utils.IsInStringArray(k, []string{
cloudprovider.META_HEADER_CONTENT_TYPE,
cloudprovider.META_HEADER_CACHE_CONTROL,
cloudprovider.META_HEADER_CONTENT_DISPOSITION,
cloudprovider.META_HEADER_CONTENT_ENCODING,
cloudprovider.META_HEADER_CONTENT_LANGUAGE,
}) {
continue
}
if len(v) > 0 {
userMeta[http.CanonicalHeaderKey(k)] = v[0]
}
}
opts.UserMetadata = userMeta
}
return opts
}
func (bucket *SBucket) PutObject(ctx context.Context, key string, input io.Reader, sizeBytes int64, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
opts := metaInitOptions(meta)
if len(storageClassStr) > 0 {
opts.StorageClass = storageClassStr
}
@@ -202,6 +249,9 @@ func (bucket *SBucket) PutObject(ctx context.Context, key string, input io.Reade
if err != nil {
return errors.Wrap(err, "cloudprovider.GetIObject")
}
if len(cannedAcl) == 0 {
cannedAcl = bucket.GetAcl()
}
err = obj.SetAcl(cannedAcl)
if err != nil && errors.Cause(err) != cloudprovider.ErrNotImplemented {
return errors.Wrap(err, "obj.SetAcl")
@@ -209,11 +259,8 @@ func (bucket *SBucket) PutObject(ctx context.Context, key string, input io.Reade
return nil
}
func (bucket *SBucket) NewMultipartUpload(ctx context.Context, key string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) (string, error) {
opts := s3cli.PutObjectOptions{}
if len(contType) > 0 {
opts.ContentType = contType
}
func (bucket *SBucket) NewMultipartUpload(ctx context.Context, key string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) (string, error) {
opts := metaInitOptions(meta)
if len(storageClassStr) > 0 {
opts.StorageClass = storageClassStr
}
@@ -275,10 +322,12 @@ func (bucket *SBucket) GetTempUrl(method string, key string, expire time.Duratio
return url.String(), nil
}
func (bucket *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (bucket *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, dstMeta http.Header) error {
meta := make(map[string]string)
if len(contType) > 0 {
meta[http.CanonicalHeaderKey("Content-Type")] = contType
if dstMeta != nil {
for k, v := range dstMeta {
meta[http.CanonicalHeaderKey(k)] = v[0]
}
}
if len(storageClassStr) > 0 {
meta[http.CanonicalHeaderKey("x-amz-storage-class")] = storageClassStr
@@ -296,6 +345,9 @@ func (bucket *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket
if err != nil {
return errors.Wrap(err, "GetIObject")
}
if len(cannedAcl) == 0 {
cannedAcl = bucket.GetAcl()
}
err = obj.SetAcl(cannedAcl)
if err != nil {
return errors.Wrap(err, "obj.SetAcl")
+97 -6
View File
@@ -20,8 +20,10 @@ import (
"io"
"os"
"time"
"net/http"
"strings"
"github.com/pkg/errors"
"yunion.io/x/pkg/errors"
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/util/printutils"
@@ -29,6 +31,46 @@ import (
"yunion.io/x/onecloud/pkg/util/streamutils"
)
type ObjectHeaderOptions struct {
CacheControl string `help:"Cache-Control"`
ContentType string `help:"Content-Type"`
ContentEncoding string `help:"Content-Encoding"`
ContentLanguage string `help:"Content-Language"`
ContentDisposition string `help:"Content-Disposition"`
ContentMD5 string `help:"Content-MD5"`
Meta []string `help:"header, common seperatored key and value, e.g. max-age:100"`
}
func objectHeaderOptions2Meta(args ObjectHeaderOptions) http.Header {
meta := http.Header{}
for _, kv := range args.Meta {
parts := strings.Split(kv, ":")
if len(parts) == 2 && len(parts[0]) > 0 && len(parts[1]) > 0 {
meta.Add(parts[0], parts[1])
}
}
if len(args.CacheControl) > 0 {
meta.Set(cloudprovider.META_HEADER_CACHE_CONTROL, args.CacheControl)
}
if len(args.ContentType) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_TYPE, args.ContentType)
}
if len(args.ContentEncoding) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_ENCODING, args.ContentEncoding)
}
if len(args.ContentMD5) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_MD5, args.ContentMD5)
}
if len(args.ContentLanguage) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_LANGUAGE, args.ContentLanguage)
}
if len(args.ContentDisposition) > 0 {
meta.Set(cloudprovider.META_HEADER_CONTENT_DISPOSITION, args.ContentDisposition)
}
return meta
}
func S3Shell() {
type BucketListOptions struct {
}
@@ -196,9 +238,11 @@ func S3Shell() {
BlockSize int64 `help:"blocksz in MB" default:"100"`
Acl string `help:"acl" choices:"private|public-read|public-read-write"`
ContentType string `help:"content-type"`
Acl string `help:"acl" choices:"private|public-read|public-read-write"`
StorageClass string `help:"storage class"`
ObjectHeaderOptions
}
shellutils.R(&BucketPutObjectOptions{}, "put-object", "Put object into a bucket", func(cli cloudprovider.ICloudRegion, args *BucketPutObjectOptions) error {
bucket, err := cli.GetIBucketById(args.BUCKET)
@@ -223,7 +267,8 @@ func S3Shell() {
} else {
input = os.Stdout
}
err = cloudprovider.UploadObject(context.Background(), bucket, args.KEY, args.BlockSize*1000*1000, input, fSize, args.ContentType, cloudprovider.TBucketACLType(args.Acl), args.StorageClass, true)
meta := objectHeaderOptions2Meta(args.ObjectHeaderOptions)
err = cloudprovider.UploadObject(context.Background(), bucket, args.KEY, args.BlockSize*1000*1000, input, fSize, cloudprovider.TBucketACLType(args.Acl), args.StorageClass, meta, true)
if err != nil {
return err
}
@@ -364,6 +409,8 @@ func S3Shell() {
Debug bool `help:"show debug info"`
BlockSize int64 `help:"block size in MB"`
Native bool `help:"Use native copy"`
ObjectHeaderOptions
}
shellutils.R(&BucketObjectCopyOptions{}, "object-copy", "Copy object", func(cli cloudprovider.ICloudRegion, args *BucketObjectCopyOptions) error {
ctx := context.Background()
@@ -379,13 +426,14 @@ func S3Shell() {
if err != nil {
return err
}
meta := objectHeaderOptions2Meta(args.ObjectHeaderOptions)
if args.Native {
err = dstBucket.CopyObject(ctx, args.DSTKEY, args.SRC, args.SRCKEY, srcObj.GetContentType(), srcObj.GetAcl(), srcObj.GetStorageClass())
err = dstBucket.CopyObject(ctx, args.DSTKEY, args.SRC, args.SRCKEY, srcObj.GetAcl(), srcObj.GetStorageClass(), meta)
if err != nil {
return err
}
} else {
err = cloudprovider.CopyObject(ctx, args.BlockSize*1000*1000, dstBucket, args.DSTKEY, srcBucket, args.SRCKEY, args.Debug)
err = cloudprovider.CopyObject(ctx, args.BlockSize*1000*1000, dstBucket, args.DSTKEY, srcBucket, args.SRCKEY, meta, args.Debug)
if err != nil {
return err
}
@@ -393,4 +441,47 @@ func S3Shell() {
fmt.Println("Success!")
return nil
})
type ObjectMetaOptions struct {
BUCKET string `help:"bucket name"`
KEY string `help:"object key"`
}
shellutils.R(&ObjectMetaOptions{}, "object-meta", "Show object meta header", func(cli cloudprovider.ICloudRegion, args *ObjectMetaOptions) error {
bucket, err := cli.GetIBucketByName(args.BUCKET)
if err != nil {
return err
}
obj, err := cloudprovider.GetIObject(bucket, args.KEY)
if err != nil {
return err
}
meta := obj.GetMeta()
for k, v := range meta {
fmt.Println(k, ": ", v[0])
}
return nil
})
type ObjectSetMetaOptions struct {
BUCKET string `help:"bucket name"`
KEY string `help:"object key"`
ObjectHeaderOptions
}
shellutils.R(&ObjectSetMetaOptions{}, "object-set-meta", "Set object meta header", func(cli cloudprovider.ICloudRegion, args *ObjectSetMetaOptions) error {
bucket, err := cli.GetIBucketByName(args.BUCKET)
if err != nil {
return err
}
obj, err := cloudprovider.GetIObject(bucket, args.KEY)
if err != nil {
return err
}
meta := objectHeaderOptions2Meta(args.ObjectHeaderOptions)
err = cloudprovider.ObjectSetMeta(context.Background(), bucket, obj, meta)
if err != nil {
return err
}
return nil
})
}
+83 -16
View File
@@ -29,6 +29,7 @@ import (
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/multicloud"
"net/http"
)
type SBucket struct {
@@ -198,7 +199,6 @@ func (b *SBucket) ListObjects(prefix string, marker string, delimiter string, ma
SizeBytes: int64(object.Size),
ETag: object.ETag,
LastModified: lastModified,
ContentType: "",
},
}
result.Objects = append(result.Objects, obj)
@@ -220,7 +220,7 @@ func (b *SBucket) ListObjects(prefix string, marker string, delimiter string, ma
return result, nil
}
func (b *SBucket) PutObject(ctx context.Context, key string, reader io.Reader, sizeBytes int64, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) PutObject(ctx context.Context, key string, reader io.Reader, sizeBytes int64, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
coscli, err := b.region.GetCosClient(b)
if err != nil {
return errors.Wrap(err, "GetCosClient")
@@ -232,12 +232,35 @@ func (b *SBucket) PutObject(ctx context.Context, key string, reader io.Reader, s
if sizeBytes > 0 {
opts.ContentLength = int(sizeBytes)
}
if len(contType) > 0 {
opts.ContentType = contType
if meta != nil {
extraHdr := http.Header{}
for k, v := range meta {
if len(v) == 0 || len(v[0]) == 0 {
continue
}
switch http.CanonicalHeaderKey(k) {
case cloudprovider.META_HEADER_CACHE_CONTROL:
opts.CacheControl = v[0]
case cloudprovider.META_HEADER_CONTENT_TYPE:
opts.ContentType = v[0]
case cloudprovider.META_HEADER_CONTENT_MD5:
opts.ContentMD5 = v[0]
case cloudprovider.META_HEADER_CONTENT_ENCODING:
opts.ContentEncoding = v[0]
case cloudprovider.META_HEADER_CONTENT_DISPOSITION:
opts.ContentDisposition = v[0]
default:
extraHdr.Add(k, v[0])
}
}
if len(extraHdr) > 0 {
opts.XCosMetaXXX = &extraHdr
}
}
if len(cannedAcl) > 0 {
opts.XCosACL = string(cannedAcl)
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
opts.XCosACL = string(cannedAcl)
if len(storageClassStr) > 0 {
opts.XCosStorageClass = storageClassStr
}
@@ -248,7 +271,7 @@ func (b *SBucket) PutObject(ctx context.Context, key string, reader io.Reader, s
return nil
}
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) (string, error) {
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) (string, error) {
coscli, err := b.region.GetCosClient(b)
if err != nil {
return "", errors.Wrap(err, "GetCosClient")
@@ -257,12 +280,35 @@ func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType s
ACLHeaderOptions: &cos.ACLHeaderOptions{},
ObjectPutHeaderOptions: &cos.ObjectPutHeaderOptions{},
}
if len(contType) > 0 {
opts.ContentType = contType
if meta != nil {
extraHdr := http.Header{}
for k, v := range meta {
if len(v) == 0 || len(v[0]) == 0 {
continue
}
switch http.CanonicalHeaderKey(k) {
case cloudprovider.META_HEADER_CACHE_CONTROL:
opts.CacheControl = v[0]
case cloudprovider.META_HEADER_CONTENT_TYPE:
opts.ContentType = v[0]
case cloudprovider.META_HEADER_CONTENT_MD5:
opts.ContentMD5 = v[0]
case cloudprovider.META_HEADER_CONTENT_ENCODING:
opts.ContentEncoding = v[0]
case cloudprovider.META_HEADER_CONTENT_DISPOSITION:
opts.ContentDisposition = v[0]
default:
extraHdr.Add(k, v[0])
}
}
if len(extraHdr) > 0 {
opts.XCosMetaXXX = &extraHdr
}
}
if len(cannedAcl) > 0 {
opts.XCosACL = string(cannedAcl)
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
opts.XCosACL = string(cannedAcl)
if len(storageClassStr) > 0 {
opts.XCosStorageClass = storageClassStr
}
@@ -356,7 +402,7 @@ func (b *SBucket) GetTempUrl(method string, key string, expire time.Duration) (s
return url.String(), nil
}
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucketName, srcKey string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucketName, srcKey string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
coscli, err := b.region.GetCosClient(b)
if err != nil {
return errors.Wrap(err, "GetCosClient")
@@ -365,14 +411,35 @@ func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucketName,
ObjectCopyHeaderOptions: &cos.ObjectCopyHeaderOptions{},
ACLHeaderOptions: &cos.ACLHeaderOptions{},
}
if len(cannedAcl) > 0 {
opts.XCosACL = string(cannedAcl)
if len(cannedAcl) == 0 {
cannedAcl = b.GetAcl()
}
opts.XCosACL = string(cannedAcl)
if len(storageClassStr) > 0 {
opts.XCosStorageClass = storageClassStr
}
if len(contType) > 0 {
opts.ContentType = contType
if meta != nil {
extraHdr := http.Header{}
for k, v := range meta {
if len(v) == 0 || len(v[0]) == 0 {
continue
}
switch http.CanonicalHeaderKey(k) {
case cloudprovider.META_HEADER_CACHE_CONTROL:
opts.CacheControl = v[0]
case cloudprovider.META_HEADER_CONTENT_TYPE:
opts.ContentType = v[0]
case cloudprovider.META_HEADER_CONTENT_ENCODING:
opts.ContentEncoding = v[0]
case cloudprovider.META_HEADER_CONTENT_DISPOSITION:
opts.ContentDisposition = v[0]
default:
extraHdr.Add(k, v[0])
}
}
if len(extraHdr) > 0 {
opts.XCosMetaXXX = &extraHdr
}
}
srcBucket := SBucket{
region: b.region,
+5
View File
@@ -23,6 +23,7 @@ import (
"yunion.io/x/pkg/errors"
"yunion.io/x/onecloud/pkg/cloudprovider"
"net/http"
)
type SObject struct {
@@ -65,3 +66,7 @@ func (o *SObject) SetAcl(aclStr cloudprovider.TBucketACLType) error {
}
return nil
}
func (o *SObject) GetMeta() http.Header {
return nil
}
+1 -1
View File
@@ -187,7 +187,7 @@ func (self *SStoragecache) uploadImage(ctx context.Context, userCred mcclient.To
if err != nil {
return "", errors.Wrap(err, "GetIBucketByName")
}
err = cloudprovider.UploadObject(context.Background(), bucket, image.ImageId, 0, reader, sizeBytes, "", "", "", false)
err = cloudprovider.UploadObject(context.Background(), bucket, image.ImageId, 0, reader, sizeBytes, "", "", nil, false)
// err = bucket.PutObject(context.Background(), image.ImageId, reader, sizeBytes, "", "", "")
if err != nil {
log.Errorf("UploadObject error %s %s", image.ImageId, err)
+7 -3
View File
@@ -182,6 +182,10 @@ func (self *SFile) SetAcl(cloudprovider.TBucketACLType) error {
return nil
}
func (self *SFile) GetMeta() http.Header {
return nil
}
func doRequest(req *http.Request) (jsonutils.JSONObject, error) {
res, err := httputils.GetDefaultClient().Do(req)
if err != nil {
@@ -343,11 +347,11 @@ func (b *SBucket) ListObjects(prefix string, marker string, delimiter string, ma
return result, nil
}
func (b *SBucket) PutObject(ctx context.Context, key string, input io.Reader, sizeBytes int64, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) PutObject(ctx context.Context, key string, input io.Reader, sizeBytes int64, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
return cloudprovider.ErrNotSupported
}
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) (string, error) {
func (b *SBucket) NewMultipartUpload(ctx context.Context, key string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) (string, error) {
return "", cloudprovider.ErrNotSupported
}
@@ -375,7 +379,7 @@ func (b *SBucket) GetTempUrl(method string, key string, expire time.Duration) (s
return "", cloudprovider.ErrNotSupported
}
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, contType string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string) error {
func (b *SBucket) CopyObject(ctx context.Context, destKey string, srcBucket, srcKey string, cannedAcl cloudprovider.TBucketACLType, storageClassStr string, meta http.Header) error {
return cloudprovider.ErrNotSupported
}