/*
* Minio Cloud Storage, (C) 2017, 2018 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package oss
import (
"context"
"encoding/xml"
"fmt"
"io"
"net/http"
"strconv"
"strings"
"github.com/aliyun/aliyun-oss-go-sdk/oss"
"github.com/dustin/go-humanize"
"github.com/minio/cli"
miniogopolicy "github.com/minio/minio-go/pkg/policy"
minio "github.com/minio/minio/cmd"
"github.com/minio/minio/cmd/logger"
"github.com/minio/minio/pkg/auth"
"github.com/minio/minio/pkg/hash"
"github.com/minio/minio/pkg/policy"
"github.com/minio/minio/pkg/policy/condition"
)
const (
ossS3MinPartSize = 5 * humanize.MiByte
ossMaxParts = 1000
ossMaxKeys = 1000
ossBackend = "oss"
)
func init() {
const ossGatewayTemplate = `NAME:
{{.HelpName}} - {{.Usage}}
USAGE:
{{.HelpName}} {{if .VisibleFlags}}[FLAGS]{{end}} [ENDPOINT]
{{if .VisibleFlags}}
FLAGS:
{{range .VisibleFlags}}{{.}}
{{end}}{{end}}
ENDPOINT:
OSS server endpoint. Default ENDPOINT is https://oss.aliyuncs.com
ENVIRONMENT VARIABLES:
ACCESS:
MINIO_ACCESS_KEY: Username or access key of OSS storage.
MINIO_SECRET_KEY: Password or secret key of OSS storage.
BROWSER:
MINIO_BROWSER: To disable web browser access, set this value to "off".
DOMAIN:
MINIO_DOMAIN: To enable virtual-host-style requests, set this value to Minio host domain name.
CACHE:
MINIO_CACHE_DRIVES: List of mounted drives or directories delimited by ";".
MINIO_CACHE_EXCLUDE: List of cache exclusion patterns delimited by ";".
MINIO_CACHE_EXPIRY: Cache expiry duration in days.
MINIO_CACHE_MAXUSE: Maximum permitted usage of the cache in percentage (0-100).
EXAMPLES:
1. Start minio gateway server for Aliyun OSS backend.
$ export MINIO_ACCESS_KEY=accesskey
$ export MINIO_SECRET_KEY=secretkey
$ {{.HelpName}}
2. Start minio gateway server for Aliyun OSS backend on custom endpoint.
$ export MINIO_ACCESS_KEY=Q3AM3UQ867SPQQA43P2F
$ export MINIO_SECRET_KEY=zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG
$ {{.HelpName}} https://oss.example.com
3. Start minio gateway server for Aliyun OSS backend with edge caching enabled.
$ export MINIO_ACCESS_KEY=accesskey
$ export MINIO_SECRET_KEY=secretkey
$ export MINIO_CACHE_DRIVES="/mnt/drive1;/mnt/drive2;/mnt/drive3;/mnt/drive4"
$ export MINIO_CACHE_EXCLUDE="bucket1/*;*.png"
$ export MINIO_CACHE_EXPIRY=40
$ export MINIO_CACHE_MAXUSE=80
$ {{.HelpName}}
`
minio.RegisterGatewayCommand(cli.Command{
Name: "oss",
Usage: "Alibaba Cloud (Aliyun) Object Storage Service (OSS).",
Action: ossGatewayMain,
CustomHelpTemplate: ossGatewayTemplate,
HideHelpCommand: true,
})
}
// Handler for 'minio gateway oss' command line.
func ossGatewayMain(ctx *cli.Context) {
if ctx.Args().First() == "help" {
cli.ShowCommandHelpAndExit(ctx, ossBackend, 1)
}
// Validate gateway arguments.
host := ctx.Args().First()
logger.FatalIf(minio.ValidateGatewayArguments(ctx.GlobalString("address"), host), "Invalid argument")
minio.StartGateway(ctx, &OSS{host})
}
// OSS implements Gateway.
type OSS struct {
host string
}
// Name implements Gateway interface.
func (g *OSS) Name() string {
return ossBackend
}
// NewGatewayLayer implements Gateway interface and returns OSS ObjectLayer.
func (g *OSS) NewGatewayLayer(creds auth.Credentials) (minio.ObjectLayer, error) {
var err error
// Regions and endpoints
// https://www.alibabacloud.com/help/doc-detail/31837.htm
if g.host == "" {
g.host = "https://oss.aliyuncs.com"
}
// Initialize oss client object.
client, err := oss.New(g.host, creds.AccessKey, creds.SecretKey)
if err != nil {
return nil, err
}
return &ossObjects{
Client: client,
}, nil
}
// Production - oss is production ready.
func (g *OSS) Production() bool {
return true
}
// appendS3MetaToOSSOptions converts metadata meant for S3 PUT/COPY
// object into oss.Option.
//
// S3 user-metadata is translated to OSS metadata by removing the
// `X-Amz-Meta-` prefix and converted into `X-Oss-Meta-`.
//
// Header names are canonicalized as in http.Header.
func appendS3MetaToOSSOptions(ctx context.Context, opts []oss.Option, s3Metadata map[string]string) ([]oss.Option, error) {
if opts == nil {
opts = make([]oss.Option, 0, len(s3Metadata))
}
for k, v := range s3Metadata {
k = http.CanonicalHeaderKey(k)
switch {
case strings.HasPrefix(k, "X-Amz-Meta-"):
metaKey := k[len("X-Amz-Meta-"):]
// NOTE(timonwong): OSS won't allow headers with underscore(_).
if strings.Contains(metaKey, "_") {
logger.LogIf(ctx, minio.UnsupportedMetadata{})
return nil, minio.UnsupportedMetadata{}
}
opts = append(opts, oss.Meta(metaKey, v))
case k == "X-Amz-Acl":
// Valid values: public-read, private, and public-read-write
opts = append(opts, oss.ObjectACL(oss.ACLType(v)))
case k == "X-Amz-Server-Side-Encryption":
opts = append(opts, oss.ServerSideEncryption(v))
case k == "X-Amz-Copy-Source-If-Match":
opts = append(opts, oss.CopySourceIfMatch(v))
case k == "X-Amz-Copy-Source-If-None-Match":
opts = append(opts, oss.CopySourceIfNoneMatch(v))
case k == "X-Amz-Copy-Source-If-Unmodified-Since":
if v, err := http.ParseTime(v); err == nil {
opts = append(opts, oss.CopySourceIfUnmodifiedSince(v))
}
case k == "X-Amz-Copy-Source-If-Modified-Since":
if v, err := http.ParseTime(v); err == nil {
opts = append(opts, oss.CopySourceIfModifiedSince(v))
}
case k == "Accept-Encoding":
opts = append(opts, oss.AcceptEncoding(v))
case k == "Cache-Control":
opts = append(opts, oss.CacheControl(v))
case k == "Content-Disposition":
opts = append(opts, oss.ContentDisposition(v))
case k == "Content-Encoding":
opts = append(opts, oss.ContentEncoding(v))
case k == "Content-Length":
if v, err := strconv.ParseInt(v, 10, 64); err == nil {
opts = append(opts, oss.ContentLength(v))
}
case k == "Content-MD5":
opts = append(opts, oss.ContentMD5(v))
case k == "Content-Type":
opts = append(opts, oss.ContentType(v))
case k == "Expires":
if v, err := http.ParseTime(v); err == nil {
opts = append(opts, oss.Expires(v))
}
}
}
return opts, nil
}
// ossMetaToS3Meta converts OSS metadata to S3 metadata.
// It is the reverse of appendS3MetaToOSSOptions.
func ossHeaderToS3Meta(header http.Header) map[string]string {
// Decoding technique for each key is used here is as follows
// Each '_' is converted to '-'
// Each '__' is converted to '_'
// With this basic assumption here are some of the expected
// translations for these keys.
// i: 'x_s3cmd__attrs' -> o: 'x-s3cmd_attrs' (mixed)
// i: 'x____test____value' -> o: 'x__test__value' (double '_')
decodeKey := func(key string) string {
tokens := strings.Split(key, "__")
for i := range tokens {
tokens[i] = strings.Replace(tokens[i], "_", "-", -1)
}
return strings.Join(tokens, "_")
}
s3Metadata := make(map[string]string)
for k := range header {
k = http.CanonicalHeaderKey(k)
switch {
case strings.HasPrefix(k, oss.HTTPHeaderOssMetaPrefix):
// Add amazon s3 meta prefix
metaKey := k[len(oss.HTTPHeaderOssMetaPrefix):]
metaKey = "X-Amz-Meta-" + decodeKey(metaKey)
metaKey = http.CanonicalHeaderKey(metaKey)
s3Metadata[metaKey] = header.Get(k)
case k == "Cache-Control":
fallthrough
case k == "Content-Encoding":
fallthrough
case k == "Content-Disposition":
fallthrough
case k == "Content-Length":
fallthrough
case k == "Content-MD5":
fallthrough
case k == "Content-Type":
s3Metadata[k] = header.Get(k)
}
}
return s3Metadata
}
// ossToObjectError converts OSS errors to minio object layer errors.
func ossToObjectError(err error, params ...string) error {
if err == nil {
return nil
}
bucket := ""
object := ""
uploadID := ""
switch len(params) {
case 3:
uploadID = params[2]
fallthrough
case 2:
object = params[1]
fallthrough
case 1:
bucket = params[0]
}
ossErr, ok := err.(oss.ServiceError)
if !ok {
// We don't interpret non OSS errors. As oss errors will
// have StatusCode to help to convert to object errors.
return err
}
switch ossErr.Code {
case "BucketAlreadyExists":
err = minio.BucketAlreadyOwnedByYou{Bucket: bucket}
case "BucketNotEmpty":
err = minio.BucketNotEmpty{Bucket: bucket}
case "InvalidBucketName":
err = minio.BucketNameInvalid{Bucket: bucket}
case "NoSuchBucket":
err = minio.BucketNotFound{Bucket: bucket}
case "NoSuchKey":
if object != "" {
err = minio.ObjectNotFound{Bucket: bucket, Object: object}
} else {
err = minio.BucketNotFound{Bucket: bucket}
}
case "InvalidObjectName":
err = minio.ObjectNameInvalid{Bucket: bucket, Object: object}
case "AccessDenied":
err = minio.PrefixAccessDenied{Bucket: bucket, Object: object}
case "NoSuchUpload":
err = minio.InvalidUploadID{UploadID: uploadID}
case "EntityTooSmall":
err = minio.PartTooSmall{}
case "SignatureDoesNotMatch":
err = minio.SignatureDoesNotMatch{}
case "InvalidPart":
err = minio.InvalidPart{}
}
return err
}
// ossObjects implements gateway for Aliyun Object Storage Service.
type ossObjects struct {
minio.GatewayUnsupported
Client *oss.Client
}
// Shutdown saves any gateway metadata to disk
// if necessary and reload upon next restart.
func (l *ossObjects) Shutdown(ctx context.Context) error {
return nil
}
// StorageInfo is not relevant to OSS backend.
func (l *ossObjects) StorageInfo(ctx context.Context) (si minio.StorageInfo) {
return
}
// ossIsValidBucketName verifies whether a bucket name is valid.
func ossIsValidBucketName(bucket string) bool {
// dot is not allowed in bucket name
if strings.Contains(bucket, ".") {
return false
}
if !minio.IsValidBucketName(bucket) {
return false
}
return true
}
// MakeBucketWithLocation creates a new container on OSS backend.
func (l *ossObjects) MakeBucketWithLocation(ctx context.Context, bucket, location string) error {
if !ossIsValidBucketName(bucket) {
logger.LogIf(ctx, minio.BucketNameInvalid{Bucket: bucket})
return minio.BucketNameInvalid{Bucket: bucket}
}
err := l.Client.CreateBucket(bucket)
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket)
}
// ossGeBucketInfo gets bucket metadata.
func ossGeBucketInfo(ctx context.Context, client *oss.Client, bucket string) (bi minio.BucketInfo, err error) {
bgir, err := client.GetBucketInfo(bucket)
if err != nil {
logger.LogIf(ctx, err)
return bi, ossToObjectError(err, bucket)
}
return minio.BucketInfo{
Name: bgir.BucketInfo.Name,
Created: bgir.BucketInfo.CreationDate,
}, nil
}
// GetBucketInfo gets bucket metadata.
func (l *ossObjects) GetBucketInfo(ctx context.Context, bucket string) (bi minio.BucketInfo, err error) {
return ossGeBucketInfo(ctx, l.Client, bucket)
}
// ListBuckets lists all OSS buckets.
func (l *ossObjects) ListBuckets(ctx context.Context) (buckets []minio.BucketInfo, err error) {
marker := oss.Marker("")
for {
lbr, err := l.Client.ListBuckets(marker)
if err != nil {
logger.LogIf(ctx, err)
return nil, ossToObjectError(err)
}
for _, bi := range lbr.Buckets {
buckets = append(buckets, minio.BucketInfo{
Name: bi.Name,
Created: bi.CreationDate,
})
}
marker = oss.Marker(lbr.NextMarker)
if !lbr.IsTruncated {
break
}
}
return buckets, nil
}
// DeleteBucket deletes a bucket on OSS.
func (l *ossObjects) DeleteBucket(ctx context.Context, bucket string) error {
err := l.Client.DeleteBucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket)
}
return nil
}
// fromOSSClientObjectProperties converts oss ObjectProperties to ObjectInfo.
func fromOSSClientObjectProperties(bucket string, o oss.ObjectProperties) minio.ObjectInfo {
// NOTE(timonwong): No Content-Type and user defined metadata.
// https://www.alibabacloud.com/help/doc-detail/31965.htm
return minio.ObjectInfo{
Bucket: bucket,
Name: o.Key,
ModTime: o.LastModified,
Size: o.Size,
ETag: minio.ToS3ETag(o.ETag),
}
}
// fromOSSClientListObjectsResult converts oss ListBucketResult to ListObjectsInfo.
func fromOSSClientListObjectsResult(bucket string, lor oss.ListObjectsResult) minio.ListObjectsInfo {
objects := make([]minio.ObjectInfo, len(lor.Objects))
for i, oi := range lor.Objects {
objects[i] = fromOSSClientObjectProperties(bucket, oi)
}
prefixes := make([]string, len(lor.CommonPrefixes))
copy(prefixes, lor.CommonPrefixes)
return minio.ListObjectsInfo{
IsTruncated: lor.IsTruncated,
NextMarker: lor.NextMarker,
Objects: objects,
Prefixes: prefixes,
}
}
// ossListObjects lists all blobs in OSS bucket filtered by prefix.
func ossListObjects(ctx context.Context, client *oss.Client, bucket, prefix, marker, delimiter string, maxKeys int) (loi minio.ListObjectsInfo, err error) {
buck, err := client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return loi, ossToObjectError(err, bucket)
}
// maxKeys should default to 1000 or less.
if maxKeys == 0 || maxKeys > ossMaxKeys {
maxKeys = ossMaxKeys
}
lor, err := buck.ListObjects(oss.Prefix(prefix), oss.Marker(marker), oss.Delimiter(delimiter), oss.MaxKeys(maxKeys))
if err != nil {
logger.LogIf(ctx, err)
return loi, ossToObjectError(err, bucket)
}
return fromOSSClientListObjectsResult(bucket, lor), nil
}
// ossListObjectsV2 lists all blobs in OSS bucket filtered by prefix.
func ossListObjectsV2(ctx context.Context, client *oss.Client, bucket, prefix, continuationToken, delimiter string, maxKeys int,
fetchOwner bool, startAfter string) (loi minio.ListObjectsV2Info, err error) {
// fetchOwner is not supported and unused.
marker := continuationToken
if marker == "" {
marker = startAfter
}
resultV1, err := ossListObjects(ctx, client, bucket, prefix, marker, delimiter, maxKeys)
if err != nil {
return loi, err
}
return minio.ListObjectsV2Info{
Objects: resultV1.Objects,
Prefixes: resultV1.Prefixes,
ContinuationToken: continuationToken,
NextContinuationToken: resultV1.NextMarker,
IsTruncated: resultV1.IsTruncated,
}, nil
}
// ListObjects lists all blobs in OSS bucket filtered by prefix.
func (l *ossObjects) ListObjects(ctx context.Context, bucket, prefix, marker, delimiter string, maxKeys int) (loi minio.ListObjectsInfo, err error) {
return ossListObjects(ctx, l.Client, bucket, prefix, marker, delimiter, maxKeys)
}
// ListObjectsV2 lists all blobs in OSS bucket filtered by prefix
func (l *ossObjects) ListObjectsV2(ctx context.Context, bucket, prefix, continuationToken, delimiter string, maxKeys int,
fetchOwner bool, startAfter string) (loi minio.ListObjectsV2Info, err error) {
return ossListObjectsV2(ctx, l.Client, bucket, prefix, continuationToken, delimiter, maxKeys, fetchOwner, startAfter)
}
// ossGetObject reads an object on OSS. Supports additional
// parameters like offset and length which are synonymous with
// HTTP Range requests.
//
// startOffset indicates the starting read location of the object.
// length indicates the total length of the object.
func ossGetObject(ctx context.Context, client *oss.Client, bucket, key string, startOffset, length int64, writer io.Writer, etag string) error {
if length < 0 && length != -1 {
logger.LogIf(ctx, fmt.Errorf("Invalid argument"))
return ossToObjectError(fmt.Errorf("Invalid argument"), bucket, key)
}
bkt, err := client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket, key)
}
var opts []oss.Option
if startOffset >= 0 && length >= 0 {
opts = append(opts, oss.Range(startOffset, startOffset+length-1))
}
object, err := bkt.GetObject(key, opts...)
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket, key)
}
defer object.Close()
if _, err := io.Copy(writer, object); err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket, key)
}
return nil
}
// GetObjectNInfo - returns object info and locked object ReadCloser
func (l *ossObjects) GetObjectNInfo(ctx context.Context, bucket, object string, rs *minio.HTTPRangeSpec, h http.Header, lockType minio.LockType) (gr *minio.GetObjectReader, err error) {
var objInfo minio.ObjectInfo
objInfo, err = l.GetObjectInfo(ctx, bucket, object, minio.ObjectOptions{})
if err != nil {
return nil, err
}
var startOffset, length int64
startOffset, length, err = rs.GetOffsetLength(objInfo.Size)
if err != nil {
return nil, err
}
pr, pw := io.Pipe()
go func() {
err := l.GetObject(ctx, bucket, object, startOffset, length, pw, objInfo.ETag, minio.ObjectOptions{})
pw.CloseWithError(err)
}()
// Setup cleanup function to cause the above go-routine to
// exit in case of partial read
pipeCloser := func() { pr.Close() }
return minio.NewGetObjectReaderFromReader(pr, objInfo, pipeCloser), nil
}
// GetObject reads an object on OSS. Supports additional
// parameters like offset and length which are synonymous with
// HTTP Range requests.
//
// startOffset indicates the starting read location of the object.
// length indicates the total length of the object.
func (l *ossObjects) GetObject(ctx context.Context, bucket, key string, startOffset, length int64, writer io.Writer, etag string, opts minio.ObjectOptions) error {
return ossGetObject(ctx, l.Client, bucket, key, startOffset, length, writer, etag)
}
func translatePlainError(err error) error {
errString := err.Error()
switch errString {
case "oss: service returned without a response body (404 Not Found)":
return oss.ServiceError{Code: "NoSuchKey"}
case "oss: service returned without a response body (400 Bad Request)":
return oss.ServiceError{Code: "AccessDenied"}
}
return err
}
// ossGetObjectInfo reads object info and replies back ObjectInfo.
func ossGetObjectInfo(ctx context.Context, client *oss.Client, bucket, object string) (objInfo minio.ObjectInfo, err error) {
bkt, err := client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return objInfo, ossToObjectError(err, bucket, object)
}
header, err := bkt.GetObjectDetailedMeta(object)
if err != nil {
logger.LogIf(ctx, translatePlainError(err))
return objInfo, ossToObjectError(translatePlainError(err), bucket, object)
}
// Build S3 metadata from OSS metadata
userDefined := ossHeaderToS3Meta(header)
modTime, _ := http.ParseTime(header.Get("Last-Modified"))
size, _ := strconv.ParseInt(header.Get("Content-Length"), 10, 64)
return minio.ObjectInfo{
Bucket: bucket,
Name: object,
ModTime: modTime,
Size: size,
ETag: minio.ToS3ETag(header.Get("ETag")),
UserDefined: userDefined,
ContentType: header.Get("Content-Type"),
ContentEncoding: header.Get("Content-Encoding"),
}, nil
}
// GetObjectInfo reads object info and replies back ObjectInfo.
func (l *ossObjects) GetObjectInfo(ctx context.Context, bucket, object string, opts minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) {
return ossGetObjectInfo(ctx, l.Client, bucket, object)
}
// ossPutObject creates a new object with the incoming data.
func ossPutObject(ctx context.Context, client *oss.Client, bucket, object string, data *hash.Reader, metadata map[string]string) (objInfo minio.ObjectInfo, err error) {
bkt, err := client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return objInfo, ossToObjectError(err, bucket, object)
}
// Build OSS metadata
opts, err := appendS3MetaToOSSOptions(ctx, nil, metadata)
if err != nil {
return objInfo, ossToObjectError(err, bucket, object)
}
err = bkt.PutObject(object, data, opts...)
if err != nil {
logger.LogIf(ctx, err)
return objInfo, ossToObjectError(err, bucket, object)
}
return ossGetObjectInfo(ctx, client, bucket, object)
}
// PutObject creates a new object with the incoming data.
func (l *ossObjects) PutObject(ctx context.Context, bucket, object string, data *hash.Reader, metadata map[string]string, opts minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) {
return ossPutObject(ctx, l.Client, bucket, object, data, metadata)
}
// CopyObject copies an object from source bucket to a destination bucket.
func (l *ossObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, srcInfo minio.ObjectInfo, srcOpts, dstOpts minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) {
bkt, err := l.Client.Bucket(srcBucket)
if err != nil {
logger.LogIf(ctx, err)
return objInfo, ossToObjectError(err, srcBucket, srcObject)
}
opts := make([]oss.Option, 0, len(srcInfo.UserDefined)+1)
// Set this header such that following CopyObject() always sets the right metadata on the destination.
// metadata input is already a trickled down value from interpreting x-oss-metadata-directive at
// handler layer. So what we have right now is supposed to be applied on the destination object anyways.
// So preserve it by adding "REPLACE" directive to save all the metadata set by CopyObject API.
opts = append(opts, oss.MetadataDirective(oss.MetaReplace))
// Build OSS metadata
opts, err = appendS3MetaToOSSOptions(ctx, opts, srcInfo.UserDefined)
if err != nil {
return objInfo, ossToObjectError(err, srcBucket, srcObject)
}
if _, err = bkt.CopyObjectTo(dstBucket, dstObject, srcObject, opts...); err != nil {
logger.LogIf(ctx, err)
return objInfo, ossToObjectError(err, srcBucket, srcObject)
}
return l.GetObjectInfo(ctx, dstBucket, dstObject, dstOpts)
}
// DeleteObject deletes a blob in bucket.
func (l *ossObjects) DeleteObject(ctx context.Context, bucket, object string) error {
bkt, err := l.Client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket, object)
}
err = bkt.DeleteObject(object)
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket, object)
}
return nil
}
// fromOSSClientListMultipartsInfo converts oss ListMultipartUploadResult to ListMultipartsInfo
func fromOSSClientListMultipartsInfo(lmur oss.ListMultipartUploadResult) minio.ListMultipartsInfo {
uploads := make([]minio.MultipartInfo, len(lmur.Uploads))
for i, um := range lmur.Uploads {
uploads[i] = minio.MultipartInfo{
Object: um.Key,
UploadID: um.UploadID,
Initiated: um.Initiated,
}
}
commonPrefixes := make([]string, len(lmur.CommonPrefixes))
copy(commonPrefixes, lmur.CommonPrefixes)
return minio.ListMultipartsInfo{
KeyMarker: lmur.KeyMarker,
UploadIDMarker: lmur.UploadIDMarker,
NextKeyMarker: lmur.NextKeyMarker,
NextUploadIDMarker: lmur.NextUploadIDMarker,
MaxUploads: lmur.MaxUploads,
IsTruncated: lmur.IsTruncated,
Uploads: uploads,
Prefix: lmur.Prefix,
Delimiter: lmur.Delimiter,
CommonPrefixes: commonPrefixes,
}
}
// ListMultipartUploads lists all multipart uploads.
func (l *ossObjects) ListMultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (lmi minio.ListMultipartsInfo, err error) {
bkt, err := l.Client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return lmi, ossToObjectError(err, bucket)
}
lmur, err := bkt.ListMultipartUploads(oss.Prefix(prefix), oss.KeyMarker(keyMarker), oss.UploadIDMarker(uploadIDMarker),
oss.Delimiter(delimiter), oss.MaxUploads(maxUploads))
if err != nil {
logger.LogIf(ctx, err)
return lmi, ossToObjectError(err, bucket)
}
return fromOSSClientListMultipartsInfo(lmur), nil
}
// NewMultipartUpload upload object in multiple parts.
func (l *ossObjects) NewMultipartUpload(ctx context.Context, bucket, object string, metadata map[string]string, o minio.ObjectOptions) (uploadID string, err error) {
bkt, err := l.Client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return uploadID, ossToObjectError(err, bucket, object)
}
// Build OSS metadata
opts, err := appendS3MetaToOSSOptions(ctx, nil, metadata)
if err != nil {
return uploadID, ossToObjectError(err, bucket, object)
}
lmur, err := bkt.InitiateMultipartUpload(object, opts...)
if err != nil {
logger.LogIf(ctx, err)
return uploadID, ossToObjectError(err, bucket, object)
}
return lmur.UploadID, nil
}
// PutObjectPart puts a part of object in bucket.
func (l *ossObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data *hash.Reader, opts minio.ObjectOptions) (pi minio.PartInfo, err error) {
bkt, err := l.Client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return pi, ossToObjectError(err, bucket, object)
}
imur := oss.InitiateMultipartUploadResult{
Bucket: bucket,
Key: object,
UploadID: uploadID,
}
size := data.Size()
up, err := bkt.UploadPart(imur, data, size, partID)
if err != nil {
logger.LogIf(ctx, err)
return pi, ossToObjectError(err, bucket, object)
}
return minio.PartInfo{
Size: size,
ETag: minio.ToS3ETag(up.ETag),
// NOTE(timonwong): LastModified is not supported
PartNumber: up.PartNumber,
}, nil
}
func ossBuildListObjectPartsParams(uploadID string, partNumberMarker, maxParts int) map[string]interface{} {
return map[string]interface{}{
"uploadId": uploadID,
"part-number-marker": strconv.Itoa(partNumberMarker),
"max-parts": strconv.Itoa(maxParts),
}
}
// fromOSSClientListPartsInfo converts OSS ListUploadedPartsResult to ListPartsInfo
func fromOSSClientListPartsInfo(lupr oss.ListUploadedPartsResult, partNumberMarker int) minio.ListPartsInfo {
parts := make([]minio.PartInfo, len(lupr.UploadedParts))
for i, up := range lupr.UploadedParts {
parts[i] = minio.PartInfo{
PartNumber: up.PartNumber,
LastModified: up.LastModified,
ETag: minio.ToS3ETag(up.ETag),
Size: int64(up.Size),
}
}
nextPartNumberMarker, _ := strconv.Atoi(lupr.NextPartNumberMarker)
return minio.ListPartsInfo{
Bucket: lupr.Bucket,
Object: lupr.Key,
UploadID: lupr.UploadID,
PartNumberMarker: partNumberMarker,
NextPartNumberMarker: nextPartNumberMarker,
MaxParts: lupr.MaxParts,
IsTruncated: lupr.IsTruncated,
Parts: parts,
}
}
func ossListObjectParts(client *oss.Client, bucket, object, uploadID string, partNumberMarker, maxParts int) (lupr oss.ListUploadedPartsResult, err error) {
params := ossBuildListObjectPartsParams(uploadID, partNumberMarker, maxParts)
resp, err := client.Conn.Do("GET", bucket, object, params, nil, nil, 0, nil)
if err != nil {
return lupr, err
}
// always drain output (response body)
defer minio.CloseResponse(resp.Body)
err = xml.NewDecoder(resp.Body).Decode(&lupr)
if err != nil {
return lupr, err
}
return lupr, nil
}
// CopyObjectPart creates a part in a multipart upload by copying
// existing object or a part of it.
func (l *ossObjects) CopyObjectPart(ctx context.Context, srcBucket, srcObject, destBucket, destObject, uploadID string,
partID int, startOffset, length int64, srcInfo minio.ObjectInfo, srcOpts, dstOpts minio.ObjectOptions) (p minio.PartInfo, err error) {
bkt, err := l.Client.Bucket(destBucket)
if err != nil {
logger.LogIf(ctx, err)
return p, ossToObjectError(err, destBucket)
}
// Build OSS metadata
opts, err := appendS3MetaToOSSOptions(ctx, nil, srcInfo.UserDefined)
if err != nil {
return p, ossToObjectError(err, srcBucket, srcObject)
}
completePart, err := bkt.UploadPartCopy(oss.InitiateMultipartUploadResult{
Key: destObject,
UploadID: uploadID,
}, srcBucket, srcObject, startOffset, length, partID, opts...)
if err != nil {
logger.LogIf(ctx, err)
return p, ossToObjectError(err, srcBucket, srcObject)
}
p.PartNumber = completePart.PartNumber
p.ETag = minio.ToS3ETag(completePart.ETag)
return p, nil
}
// ListObjectParts returns all object parts for specified object in specified bucket
func (l *ossObjects) ListObjectParts(ctx context.Context, bucket, object, uploadID string, partNumberMarker, maxParts int) (lpi minio.ListPartsInfo, err error) {
lupr, err := ossListObjectParts(l.Client, bucket, object, uploadID, partNumberMarker, maxParts)
if err != nil {
logger.LogIf(ctx, err)
return lpi, ossToObjectError(err, bucket, object, uploadID)
}
return fromOSSClientListPartsInfo(lupr, partNumberMarker), nil
}
// AbortMultipartUpload aborts a ongoing multipart upload.
func (l *ossObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string) error {
bkt, err := l.Client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket, object)
}
err = bkt.AbortMultipartUpload(oss.InitiateMultipartUploadResult{
Bucket: bucket,
Key: object,
UploadID: uploadID,
})
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket, object)
}
return nil
}
// CompleteMultipartUpload completes ongoing multipart upload and finalizes object.
func (l *ossObjects) CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []minio.CompletePart) (oi minio.ObjectInfo, err error) {
client := l.Client
bkt, err := client.Bucket(bucket)
if err != nil {
logger.LogIf(ctx, err)
return oi, ossToObjectError(err, bucket, object)
}
// Error out if uploadedParts except last part sizing < 5MiB.
// NOTE(timonwong): Actually, OSS wont't throw EntityTooSmall error, doing this check just for mint :(
var partNumberMarker int
lupr := oss.ListUploadedPartsResult{IsTruncated: true}
for lupr.IsTruncated {
lupr, err = ossListObjectParts(client, bucket, object, uploadID, partNumberMarker, ossMaxParts)
if err != nil {
logger.LogIf(ctx, err)
return oi, ossToObjectError(err, bucket, object, uploadID)
}
uploadedParts := lupr.UploadedParts
if !lupr.IsTruncated {
if len(uploadedParts) < 1 {
uploadedParts = nil
} else {
uploadedParts = uploadedParts[:len(uploadedParts)-1]
}
}
for _, part := range uploadedParts {
if part.Size < ossS3MinPartSize {
logger.LogIf(ctx, minio.PartTooSmall{
PartNumber: part.PartNumber,
PartSize: int64(part.Size),
PartETag: minio.ToS3ETag(part.ETag),
})
return oi, minio.PartTooSmall{
PartNumber: part.PartNumber,
PartSize: int64(part.Size),
PartETag: minio.ToS3ETag(part.ETag),
}
}
}
partNumberMarker, _ = strconv.Atoi(lupr.NextPartNumberMarker)
}
imur := oss.InitiateMultipartUploadResult{
Bucket: bucket,
Key: object,
UploadID: uploadID,
}
parts := make([]oss.UploadPart, len(uploadedParts))
for i, up := range uploadedParts {
parts[i] = oss.UploadPart{
PartNumber: up.PartNumber,
ETag: strings.TrimSuffix(up.ETag, "-1"), // Trim "-1" suffix in ETag as PutObjectPart().
}
}
_, err = bkt.CompleteMultipartUpload(imur, parts)
if err != nil {
logger.LogIf(ctx, err)
return oi, ossToObjectError(err, bucket, object)
}
return l.GetObjectInfo(ctx, bucket, object, minio.ObjectOptions{})
}
// SetBucketPolicy sets policy on bucket.
// OSS supports three types of bucket policies:
// oss.ACLPublicReadWrite: readwrite in minio terminology
// oss.ACLPublicRead: readonly in minio terminology
// oss.ACLPrivate: none in minio terminology
func (l *ossObjects) SetBucketPolicy(ctx context.Context, bucket string, bucketPolicy *policy.Policy) error {
policyInfo, err := minio.PolicyToBucketAccessPolicy(bucketPolicy)
if err != nil {
// This should not happen.
return ossToObjectError(err, bucket)
}
bucketPolicies := miniogopolicy.GetPolicies(policyInfo.Statements, bucket, "")
if len(bucketPolicies) != 1 {
logger.LogIf(ctx, minio.NotImplemented{})
return minio.NotImplemented{}
}
prefix := bucket + "/*" // For all objects inside the bucket.
for policyPrefix, bucketPolicy := range bucketPolicies {
if policyPrefix != prefix {
logger.LogIf(ctx, minio.NotImplemented{})
return minio.NotImplemented{}
}
var acl oss.ACLType
switch bucketPolicy {
case miniogopolicy.BucketPolicyNone:
acl = oss.ACLPrivate
case miniogopolicy.BucketPolicyReadOnly:
acl = oss.ACLPublicRead
case miniogopolicy.BucketPolicyReadWrite:
acl = oss.ACLPublicReadWrite
default:
logger.LogIf(ctx, minio.NotImplemented{})
return minio.NotImplemented{}
}
err := l.Client.SetBucketACL(bucket, acl)
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket)
}
}
return nil
}
// GetBucketPolicy will get policy on bucket.
func (l *ossObjects) GetBucketPolicy(ctx context.Context, bucket string) (*policy.Policy, error) {
result, err := l.Client.GetBucketACL(bucket)
if err != nil {
logger.LogIf(ctx, err)
return nil, ossToObjectError(err)
}
var readOnly, readWrite bool
switch result.ACL {
case string(oss.ACLPrivate):
// By default, all buckets starts with a "private" policy.
return nil, ossToObjectError(minio.BucketPolicyNotFound{}, bucket)
case string(oss.ACLPublicRead):
readOnly = true
case string(oss.ACLPublicReadWrite):
readWrite = true
default:
logger.LogIf(ctx, minio.NotImplemented{})
return nil, minio.NotImplemented{}
}
actionSet := policy.NewActionSet()
if readOnly {
actionSet.Add(policy.GetBucketLocationAction)
actionSet.Add(policy.ListBucketAction)
actionSet.Add(policy.GetObjectAction)
}
if readWrite {
actionSet.Add(policy.GetBucketLocationAction)
actionSet.Add(policy.ListBucketAction)
actionSet.Add(policy.GetObjectAction)
actionSet.Add(policy.ListBucketMultipartUploadsAction)
actionSet.Add(policy.AbortMultipartUploadAction)
actionSet.Add(policy.DeleteObjectAction)
actionSet.Add(policy.ListMultipartUploadPartsAction)
actionSet.Add(policy.PutObjectAction)
}
return &policy.Policy{
Version: policy.DefaultVersion,
Statements: []policy.Statement{
policy.NewStatement(
policy.Allow,
policy.NewPrincipal("*"),
actionSet,
policy.NewResourceSet(
policy.NewResource(bucket, ""),
policy.NewResource(bucket, "*"),
),
condition.NewFunctions(),
),
},
}, nil
}
// DeleteBucketPolicy deletes all policies on bucket.
func (l *ossObjects) DeleteBucketPolicy(ctx context.Context, bucket string) error {
err := l.Client.SetBucketACL(bucket, oss.ACLPrivate)
if err != nil {
logger.LogIf(ctx, err)
return ossToObjectError(err, bucket)
}
return nil
}