-1
@@ -35,7 +35,6 @@ import (
|
||||
_ "github.com/opencloud-eu/reva/v2/pkg/storage/fs/ocis"
|
||||
_ "github.com/opencloud-eu/reva/v2/pkg/storage/fs/owncloudsql"
|
||||
_ "github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix"
|
||||
_ "github.com/opencloud-eu/reva/v2/pkg/storage/fs/s3"
|
||||
_ "github.com/opencloud-eu/reva/v2/pkg/storage/fs/s3ng"
|
||||
// Add your own here
|
||||
)
|
||||
|
||||
+10
-1
@@ -659,7 +659,16 @@ assimilate:
|
||||
|
||||
var n *node.Node
|
||||
if fi.IsDir() {
|
||||
attributes.SetInt64(prefixes.TypeAttr, int64(provider.ResourceType_RESOURCE_TYPE_CONTAINER))
|
||||
// The Space's name attribute might not match the directory name. Use the name as
|
||||
// it was set before. Also the space root doesn't have a 'type' attribute
|
||||
// currently so only set it for normal directories.
|
||||
if t.isSpaceRoot(path) {
|
||||
if previousAttribs != nil && previousAttribs[prefixes.NameAttr] != nil {
|
||||
attributes[prefixes.NameAttr] = previousAttribs[prefixes.NameAttr]
|
||||
}
|
||||
} else {
|
||||
attributes.SetInt64(prefixes.TypeAttr, int64(provider.ResourceType_RESOURCE_TYPE_CONTAINER))
|
||||
}
|
||||
attributes.SetInt64(prefixes.TreesizeAttr, 0)
|
||||
if previousAttribs != nil && previousAttribs[prefixes.TreesizeAttr] != nil {
|
||||
attributes[prefixes.TreesizeAttr] = previousAttribs[prefixes.TreesizeAttr]
|
||||
|
||||
+6
-1
@@ -720,7 +720,7 @@ func (t *Tree) createDirNode(ctx context.Context, n *node.Node) (err error) {
|
||||
}
|
||||
|
||||
func (t *Tree) isIgnored(path string) bool {
|
||||
return isLockFile(path) || isTrash(path) || t.isUpload(path) || t.isInternal(path) || t.isRootPath(path)
|
||||
return isLockFile(path) || isTrash(path) || t.isUpload(path) || t.isInternal(path) || t.isRootPath(path) || t.isSpaceRoot(path)
|
||||
}
|
||||
|
||||
func (t *Tree) isUpload(path string) bool {
|
||||
@@ -737,6 +737,11 @@ func (t *Tree) isRootPath(path string) bool {
|
||||
path == t.projectSpacesRoot
|
||||
}
|
||||
|
||||
func (t *Tree) isSpaceRoot(path string) bool {
|
||||
parent := filepath.Dir(path)
|
||||
return parent == t.personalSpacesRoot || parent == t.projectSpacesRoot
|
||||
}
|
||||
|
||||
func (t *Tree) isInternal(path string) bool {
|
||||
return t.isIndex(path) || strings.Contains(path, lookup.MetadataDir)
|
||||
}
|
||||
|
||||
-698
@@ -1,698 +0,0 @@
|
||||
// Copyright 2018-2021 CERN
|
||||
//
|
||||
// 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.
|
||||
//
|
||||
// In applying this license, CERN does not waive the privileges and immunities
|
||||
// granted to it by virtue of its status as an Intergovernmental Organization
|
||||
// or submit itself to any jurisdiction.
|
||||
|
||||
package s3
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/awserr"
|
||||
"github.com/aws/aws-sdk-go/aws/credentials"
|
||||
"github.com/aws/aws-sdk-go/aws/session"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
"github.com/aws/aws-sdk-go/service/s3/s3manager"
|
||||
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
|
||||
types "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
|
||||
"github.com/mitchellh/mapstructure"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/appctx"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/events"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/mime"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage/fs/registry"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/rs/zerolog"
|
||||
)
|
||||
|
||||
func init() {
|
||||
registry.Register("s3", New)
|
||||
}
|
||||
|
||||
type config struct {
|
||||
Region string `mapstructure:"region"`
|
||||
AccessKey string `mapstructure:"access_key"`
|
||||
SecretKey string `mapstructure:"secret_key"`
|
||||
Endpoint string `mapstructure:"endpoint"`
|
||||
Bucket string `mapstructure:"bucket"`
|
||||
Prefix string `mapstructure:"prefix"`
|
||||
}
|
||||
|
||||
func parseConfig(m map[string]interface{}) (*config, error) {
|
||||
c := &config{}
|
||||
if err := mapstructure.Decode(m, c); err != nil {
|
||||
err = errors.Wrap(err, "error decoding conf")
|
||||
return nil, err
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// New returns an implementation to of the storage.FS interface that talk to
|
||||
// a s3 api.
|
||||
func New(m map[string]interface{}, _ events.Stream, _ *zerolog.Logger) (storage.FS, error) {
|
||||
c, err := parseConfig(m)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
awsConfig := aws.NewConfig().
|
||||
WithHTTPClient(http.DefaultClient).
|
||||
WithMaxRetries(aws.UseServiceDefaultRetries).
|
||||
WithLogger(aws.NewDefaultLogger()).
|
||||
WithLogLevel(aws.LogOff).
|
||||
WithSleepDelay(time.Sleep).
|
||||
WithCredentials(credentials.NewStaticCredentials(c.AccessKey, c.SecretKey, "")).
|
||||
WithEndpoint(c.Endpoint).
|
||||
WithS3ForcePathStyle(true).
|
||||
WithDisableSSL(true)
|
||||
|
||||
if c.Region != "" {
|
||||
awsConfig.WithRegion(c.Region)
|
||||
} else {
|
||||
awsConfig.WithRegion("us-east-1")
|
||||
}
|
||||
|
||||
sess, err := session.NewSession(awsConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if sess == nil {
|
||||
return nil, errors.New("creating the S3 session")
|
||||
}
|
||||
|
||||
s3Client := s3.New(sess)
|
||||
|
||||
return &s3FS{client: s3Client, config: c}, nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) Shutdown(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) addRoot(p string) string {
|
||||
np := path.Join(fs.config.Prefix, p)
|
||||
return np
|
||||
}
|
||||
|
||||
func (fs *s3FS) resolve(ctx context.Context, ref *provider.Reference) (string, error) {
|
||||
if strings.HasPrefix(ref.Path, "/") {
|
||||
return fs.addRoot(ref.GetPath()), nil
|
||||
}
|
||||
|
||||
if ref.ResourceId != nil && ref.ResourceId.OpaqueId != "" {
|
||||
fn := path.Join("/", strings.TrimPrefix(ref.ResourceId.OpaqueId, "fileid-"))
|
||||
fn = fs.addRoot(fn)
|
||||
return fn, nil
|
||||
}
|
||||
|
||||
// reference is invalid
|
||||
return "", fmt.Errorf("invalid reference %+v", ref)
|
||||
}
|
||||
|
||||
func (fs *s3FS) removeRoot(np string) string {
|
||||
p := strings.TrimPrefix(np, fs.config.Prefix)
|
||||
if p == "" {
|
||||
p = "/"
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
type s3FS struct {
|
||||
client *s3.S3
|
||||
config *config
|
||||
}
|
||||
|
||||
// permissionSet returns the permission set for the current user
|
||||
func (fs *s3FS) permissionSet(ctx context.Context) *provider.ResourcePermissions {
|
||||
// TODO fix permissions for share recipients by traversing reading acls up to the root? cache acls for the parent node and reuse it
|
||||
return &provider.ResourcePermissions{
|
||||
// owner has all permissions
|
||||
AddGrant: true,
|
||||
CreateContainer: true,
|
||||
Delete: true,
|
||||
GetPath: true,
|
||||
GetQuota: true,
|
||||
InitiateFileDownload: true,
|
||||
InitiateFileUpload: true,
|
||||
ListContainer: true,
|
||||
ListFileVersions: true,
|
||||
ListGrants: true,
|
||||
ListRecycle: true,
|
||||
Move: true,
|
||||
PurgeRecycle: true,
|
||||
RemoveGrant: true,
|
||||
RestoreFileVersion: true,
|
||||
RestoreRecycleItem: true,
|
||||
Stat: true,
|
||||
UpdateGrant: true,
|
||||
}
|
||||
}
|
||||
|
||||
func (fs *s3FS) normalizeObject(ctx context.Context, o *s3.Object, fn string) *provider.ResourceInfo {
|
||||
fn = fs.removeRoot(path.Join("/", fn))
|
||||
isDir := strings.HasSuffix(*o.Key, "/")
|
||||
md := &provider.ResourceInfo{
|
||||
Id: &provider.ResourceId{
|
||||
OpaqueId: "fileid-" + strings.TrimPrefix(fn, "/"),
|
||||
},
|
||||
Path: fn,
|
||||
Type: getResourceType(isDir),
|
||||
Etag: *o.ETag,
|
||||
MimeType: mime.Detect(isDir, fn),
|
||||
PermissionSet: fs.permissionSet(ctx),
|
||||
Size: uint64(*o.Size),
|
||||
Mtime: &types.Timestamp{
|
||||
Seconds: uint64(o.LastModified.Unix()),
|
||||
},
|
||||
}
|
||||
appctx.GetLogger(ctx).Debug().
|
||||
Interface("object", o).
|
||||
Interface("metadata", md).
|
||||
Msg("normalized Object")
|
||||
return md
|
||||
}
|
||||
|
||||
func getResourceType(isDir bool) provider.ResourceType {
|
||||
if isDir {
|
||||
return provider.ResourceType_RESOURCE_TYPE_CONTAINER
|
||||
}
|
||||
return provider.ResourceType_RESOURCE_TYPE_CONTAINER
|
||||
}
|
||||
|
||||
func (fs *s3FS) normalizeHead(ctx context.Context, o *s3.HeadObjectOutput, fn string) *provider.ResourceInfo {
|
||||
fn = fs.removeRoot(path.Join("/", fn))
|
||||
isDir := strings.HasSuffix(fn, "/")
|
||||
md := &provider.ResourceInfo{
|
||||
Id: &provider.ResourceId{OpaqueId: "fileid-" + strings.TrimPrefix(fn, "/")},
|
||||
Path: fn,
|
||||
Type: getResourceType(isDir),
|
||||
Etag: *o.ETag,
|
||||
MimeType: mime.Detect(isDir, fn),
|
||||
PermissionSet: fs.permissionSet(ctx),
|
||||
Size: uint64(*o.ContentLength),
|
||||
Mtime: &types.Timestamp{
|
||||
Seconds: uint64(o.LastModified.Unix()),
|
||||
},
|
||||
}
|
||||
appctx.GetLogger(ctx).Debug().
|
||||
Interface("head", o).
|
||||
Interface("metadata", md).
|
||||
Msg("normalized Head")
|
||||
return md
|
||||
}
|
||||
func (fs *s3FS) normalizeCommonPrefix(ctx context.Context, p *s3.CommonPrefix) *provider.ResourceInfo {
|
||||
fn := fs.removeRoot(path.Join("/", *p.Prefix))
|
||||
md := &provider.ResourceInfo{
|
||||
Id: &provider.ResourceId{OpaqueId: "fileid-" + strings.TrimPrefix(fn, "/")},
|
||||
Path: fn,
|
||||
Type: getResourceType(true),
|
||||
Etag: "TODO(labkode)",
|
||||
MimeType: mime.Detect(true, fn),
|
||||
PermissionSet: fs.permissionSet(ctx),
|
||||
Size: 0,
|
||||
Mtime: &types.Timestamp{
|
||||
Seconds: 0,
|
||||
},
|
||||
}
|
||||
appctx.GetLogger(ctx).Debug().
|
||||
Interface("prefix", p).
|
||||
Interface("metadata", md).
|
||||
Msg("normalized CommonPrefix")
|
||||
return md
|
||||
}
|
||||
|
||||
// GetPathByID returns the path pointed by the file id
|
||||
// In this implementation the file id is that path of the file without the first slash
|
||||
// thus the file id always points to the filename
|
||||
func (fs *s3FS) GetPathByID(ctx context.Context, id *provider.ResourceId) (string, error) {
|
||||
return path.Join("/", strings.TrimPrefix(id.OpaqueId, "fileid-")), nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) AddGrant(ctx context.Context, ref *provider.Reference, g *provider.Grant) error {
|
||||
return errtypes.NotSupported("s3: operation not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) DenyGrant(ctx context.Context, ref *provider.Reference, g *provider.Grantee) error {
|
||||
return errtypes.NotSupported("s3: operation not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) ListGrants(ctx context.Context, ref *provider.Reference) ([]*provider.Grant, error) {
|
||||
return nil, errtypes.NotSupported("s3: operation not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) RemoveGrant(ctx context.Context, ref *provider.Reference, g *provider.Grant) error {
|
||||
return errtypes.NotSupported("s3: operation not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) UpdateGrant(ctx context.Context, ref *provider.Reference, g *provider.Grant) error {
|
||||
return errtypes.NotSupported("s3: operation not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) GetQuota(ctx context.Context, ref *provider.Reference) (uint64, uint64, uint64, error) {
|
||||
return 0, 0, 0, nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) SetArbitraryMetadata(ctx context.Context, ref *provider.Reference, md *provider.ArbitraryMetadata) error {
|
||||
return errtypes.NotSupported("s3: operation not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) UnsetArbitraryMetadata(ctx context.Context, ref *provider.Reference, keys []string) error {
|
||||
return errtypes.NotSupported("s3: operation not supported")
|
||||
}
|
||||
|
||||
// GetLock returns an existing lock on the given reference
|
||||
func (fs *s3FS) GetLock(ctx context.Context, ref *provider.Reference) (*provider.Lock, error) {
|
||||
return nil, errtypes.NotSupported("unimplemented")
|
||||
}
|
||||
|
||||
// SetLock puts a lock on the given reference
|
||||
func (fs *s3FS) SetLock(ctx context.Context, ref *provider.Reference, lock *provider.Lock) error {
|
||||
return errtypes.NotSupported("unimplemented")
|
||||
}
|
||||
|
||||
// RefreshLock refreshes an existing lock on the given reference
|
||||
func (fs *s3FS) RefreshLock(ctx context.Context, ref *provider.Reference, lock *provider.Lock, existingLockID string) error {
|
||||
return errtypes.NotSupported("unimplemented")
|
||||
}
|
||||
|
||||
// Unlock removes an existing lock from the given reference
|
||||
func (fs *s3FS) Unlock(ctx context.Context, ref *provider.Reference, lock *provider.Lock) error {
|
||||
return errtypes.NotSupported("unimplemented")
|
||||
}
|
||||
|
||||
func (fs *s3FS) CreateReference(ctx context.Context, path string, targetURI *url.URL) error {
|
||||
// TODO(jfd):implement
|
||||
return errtypes.NotSupported("s3: operation not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) GetHome(ctx context.Context) (string, error) {
|
||||
return "", errtypes.NotSupported("eos: not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) CreateHome(ctx context.Context) error {
|
||||
return errtypes.NotSupported("s3fs: not supported")
|
||||
}
|
||||
|
||||
func (fs *s3FS) CreateDir(ctx context.Context, ref *provider.Reference) error {
|
||||
log := appctx.GetLogger(ctx)
|
||||
|
||||
fn, err := fs.resolve(ctx, ref)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
fn = fs.addRoot(fn) + "/" // append / to indicate folder // TODO only if fn does not end in /
|
||||
|
||||
input := &s3.PutObjectInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Key: aws.String(fn),
|
||||
ContentType: aws.String("application/octet-stream"),
|
||||
ContentLength: aws.Int64(0),
|
||||
}
|
||||
|
||||
result, err := fs.client.PutObject(input)
|
||||
if err != nil {
|
||||
log.Error().Err(err)
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
if aerr.Code() == s3.ErrCodeNoSuchBucket {
|
||||
return errtypes.NotFound(ref.Path)
|
||||
}
|
||||
}
|
||||
// FIXME we also need already exists error, webdav expects 405 MethodNotAllowed
|
||||
return errors.Wrap(err, "s3fs: error creating dir "+ref.Path)
|
||||
}
|
||||
|
||||
log.Debug().Interface("result", result) // todo cache etag?
|
||||
return nil
|
||||
}
|
||||
|
||||
// TouchFile as defined in the storage.FS interface
|
||||
func (fs *s3FS) TouchFile(ctx context.Context, ref *provider.Reference, markprocessing bool, mtime string) error {
|
||||
return fmt.Errorf("unimplemented: TouchFile")
|
||||
}
|
||||
|
||||
func (fs *s3FS) Delete(ctx context.Context, ref *provider.Reference) error {
|
||||
log := appctx.GetLogger(ctx)
|
||||
|
||||
fn, err := fs.resolve(ctx, ref)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error resolving ref")
|
||||
}
|
||||
|
||||
// first we need to find out if fn is a dir or a file
|
||||
|
||||
_, err = fs.client.HeadObject(&s3.HeadObjectInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Key: aws.String(fn),
|
||||
})
|
||||
if err != nil {
|
||||
log.Error().Err(err)
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
switch aerr.Code() {
|
||||
case s3.ErrCodeNoSuchBucket:
|
||||
case s3.ErrCodeNoSuchKey:
|
||||
return errtypes.NotFound(fn)
|
||||
}
|
||||
}
|
||||
// it might be a directory, so we can batch delete the prefix + /
|
||||
iter := s3manager.NewDeleteListIterator(fs.client, &s3.ListObjectsInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Prefix: aws.String(fn + "/"),
|
||||
})
|
||||
batcher := s3manager.NewBatchDeleteWithClient(fs.client)
|
||||
if err := batcher.Delete(aws.BackgroundContext(), iter); err != nil {
|
||||
return err
|
||||
}
|
||||
// ok, we are done
|
||||
return nil
|
||||
}
|
||||
|
||||
// we found an object, let's get rid of it
|
||||
result, err := fs.client.DeleteObject(&s3.DeleteObjectInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Key: aws.String(fn),
|
||||
})
|
||||
if err != nil {
|
||||
log.Error().Err(err)
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
switch aerr.Code() {
|
||||
case s3.ErrCodeNoSuchBucket:
|
||||
case s3.ErrCodeNoSuchKey:
|
||||
return errtypes.NotFound(fn)
|
||||
}
|
||||
}
|
||||
return errors.Wrap(err, "s3fs: error deleting "+fn)
|
||||
}
|
||||
|
||||
log.Debug().Interface("result", result)
|
||||
return nil
|
||||
}
|
||||
|
||||
// CreateStorageSpace creates a storage space
|
||||
func (fs *s3FS) CreateStorageSpace(ctx context.Context, req *provider.CreateStorageSpaceRequest) (*provider.CreateStorageSpaceResponse, error) {
|
||||
return nil, errtypes.NotSupported("unimplemented: CreateStorageSpace")
|
||||
}
|
||||
|
||||
func (fs *s3FS) moveObject(ctx context.Context, oldKey string, newKey string) error {
|
||||
|
||||
// Copy
|
||||
// TODO double check CopyObject can deal with >5GB files.
|
||||
// Docs say we need to use multipart upload: https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectCOPY.html
|
||||
_, err := fs.client.CopyObject(&s3.CopyObjectInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
CopySource: aws.String("/" + fs.config.Bucket + oldKey),
|
||||
Key: aws.String(newKey),
|
||||
})
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
if aerr.Code() == s3.ErrCodeNoSuchBucket {
|
||||
return errtypes.NotFound(oldKey)
|
||||
}
|
||||
return err
|
||||
}
|
||||
// TODO cache etag and mtime?
|
||||
|
||||
// Delete
|
||||
_, err = fs.client.DeleteObject(&s3.DeleteObjectInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Key: aws.String(oldKey),
|
||||
})
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
switch aerr.Code() {
|
||||
case s3.ErrCodeNoSuchBucket:
|
||||
case s3.ErrCodeNoSuchKey:
|
||||
return errtypes.NotFound(oldKey)
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) Move(ctx context.Context, oldRef, newRef *provider.Reference) error {
|
||||
log := appctx.GetLogger(ctx)
|
||||
|
||||
fn, err := fs.resolve(ctx, oldRef)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error resolving ref")
|
||||
}
|
||||
|
||||
newName, err := fs.resolve(ctx, newRef)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error resolving ref")
|
||||
}
|
||||
|
||||
// first we need to find out if fn is a dir or a file
|
||||
|
||||
_, err = fs.client.HeadObject(&s3.HeadObjectInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Key: aws.String(fn),
|
||||
})
|
||||
if err != nil {
|
||||
log.Error().Err(err)
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
switch aerr.Code() {
|
||||
case s3.ErrCodeNoSuchBucket:
|
||||
case s3.ErrCodeNoSuchKey:
|
||||
return errtypes.NotFound(fn)
|
||||
}
|
||||
}
|
||||
|
||||
// move directory
|
||||
input := &s3.ListObjectsV2Input{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Prefix: aws.String(fn + "/"),
|
||||
}
|
||||
isTruncated := true
|
||||
|
||||
for isTruncated {
|
||||
output, err := fs.client.ListObjectsV2(input)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "s3FS: error listing "+fn)
|
||||
}
|
||||
|
||||
for _, o := range output.Contents {
|
||||
log.Debug().
|
||||
Interface("object", *o).
|
||||
Str("fn", fn).
|
||||
Msg("found Object")
|
||||
|
||||
err := fs.moveObject(ctx, *o.Key, strings.Replace(*o.Key, fn+"/", newName+"/", 1))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
input.ContinuationToken = output.NextContinuationToken
|
||||
isTruncated = *output.IsTruncated
|
||||
}
|
||||
// ok, we are done
|
||||
return nil
|
||||
}
|
||||
|
||||
// move single object
|
||||
err = fs.moveObject(ctx, fn, newName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) GetMD(ctx context.Context, ref *provider.Reference, mdKeys []string, fieldMask []string) (*provider.ResourceInfo, error) {
|
||||
log := appctx.GetLogger(ctx)
|
||||
|
||||
fn, err := fs.resolve(ctx, ref)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "error resolving ref")
|
||||
}
|
||||
|
||||
// first try a head, works for files
|
||||
log.Debug().
|
||||
Str("fn", fn).
|
||||
Msg("trying HEAD")
|
||||
|
||||
input := &s3.HeadObjectInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Key: aws.String(fn),
|
||||
}
|
||||
output, err := fs.client.HeadObject(input)
|
||||
if err != nil {
|
||||
log.Error().Err(err)
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
switch aerr.Code() {
|
||||
case s3.ErrCodeNoSuchBucket:
|
||||
case s3.ErrCodeNoSuchKey:
|
||||
return nil, errtypes.NotFound(fn)
|
||||
}
|
||||
}
|
||||
log.Debug().
|
||||
Str("fn", fn).
|
||||
Msg("trying to list prefix")
|
||||
// try by listing parent to find directory
|
||||
input := &s3.ListObjectsV2Input{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Prefix: aws.String(fn),
|
||||
Delimiter: aws.String("/"), // limit to a single directory
|
||||
}
|
||||
isTruncated := true
|
||||
|
||||
for isTruncated {
|
||||
output, err := fs.client.ListObjectsV2(input)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "s3FS: error listing "+fn)
|
||||
}
|
||||
|
||||
for i := range output.CommonPrefixes {
|
||||
log.Debug().
|
||||
Interface("object", output.CommonPrefixes[i]).
|
||||
Str("fn", fn).
|
||||
Msg("found CommonPrefix")
|
||||
if *output.CommonPrefixes[i].Prefix == fn+"/" {
|
||||
return fs.normalizeCommonPrefix(ctx, output.CommonPrefixes[i]), nil
|
||||
}
|
||||
}
|
||||
|
||||
input.ContinuationToken = output.NextContinuationToken
|
||||
isTruncated = *output.IsTruncated
|
||||
}
|
||||
return nil, errtypes.NotFound(fn)
|
||||
}
|
||||
|
||||
return fs.normalizeHead(ctx, output, fn), nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) ListFolder(ctx context.Context, ref *provider.Reference, mdKeys, fieldMask []string) ([]*provider.ResourceInfo, error) {
|
||||
fn, err := fs.resolve(ctx, ref)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "error resolving ref")
|
||||
}
|
||||
|
||||
input := &s3.ListObjectsV2Input{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Prefix: aws.String(fn + "/"),
|
||||
Delimiter: aws.String("/"), // limit to a single directory
|
||||
}
|
||||
isTruncated := true
|
||||
|
||||
finfos := []*provider.ResourceInfo{}
|
||||
|
||||
for isTruncated {
|
||||
output, err := fs.client.ListObjectsV2(input)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "s3FS: error listing "+fn)
|
||||
}
|
||||
|
||||
for i := range output.CommonPrefixes {
|
||||
finfos = append(finfos, fs.normalizeCommonPrefix(ctx, output.CommonPrefixes[i]))
|
||||
}
|
||||
|
||||
for i := range output.Contents {
|
||||
finfos = append(finfos, fs.normalizeObject(ctx, output.Contents[i], *output.Contents[i].Key))
|
||||
}
|
||||
|
||||
input.ContinuationToken = output.NextContinuationToken
|
||||
isTruncated = *output.IsTruncated
|
||||
}
|
||||
// TODO sort fileinfos?
|
||||
return finfos, nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) Download(ctx context.Context, ref *provider.Reference, openReaderfunc func(*provider.ResourceInfo) bool) (*provider.ResourceInfo, io.ReadCloser, error) {
|
||||
log := appctx.GetLogger(ctx)
|
||||
|
||||
fn, err := fs.resolve(ctx, ref)
|
||||
if err != nil {
|
||||
return nil, nil, errors.Wrap(err, "error resolving ref")
|
||||
}
|
||||
|
||||
ri, err := fs.GetMD(ctx, ref, nil, []string{"size", "mimetype", "etag"})
|
||||
if err != nil {
|
||||
return nil, nil, errors.Wrap(err, "error getting metadata")
|
||||
}
|
||||
|
||||
if !openReaderfunc(ri) {
|
||||
return ri, nil, nil
|
||||
}
|
||||
|
||||
// use GetObject instead of s3manager.Downloader:
|
||||
// the result.Body is a ReadCloser, which allows streaming
|
||||
// TODO double check we are not caching bytes in memory
|
||||
r, err := fs.client.GetObject(&s3.GetObjectInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Key: aws.String(fn),
|
||||
})
|
||||
if err != nil {
|
||||
log.Error().Err(err)
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
switch aerr.Code() {
|
||||
case s3.ErrCodeNoSuchBucket:
|
||||
case s3.ErrCodeNoSuchKey:
|
||||
return nil, nil, errtypes.NotFound(fn)
|
||||
}
|
||||
}
|
||||
return nil, nil, errors.Wrap(err, "s3fs: error deleting "+fn)
|
||||
}
|
||||
return ri, r.Body, nil
|
||||
}
|
||||
|
||||
func (fs *s3FS) ListRevisions(ctx context.Context, ref *provider.Reference) ([]*provider.FileVersion, error) {
|
||||
return nil, errtypes.NotSupported("list revisions")
|
||||
}
|
||||
|
||||
func (fs *s3FS) DownloadRevision(ctx context.Context, ref *provider.Reference, revisionKey string, openReaderfunc func(*provider.ResourceInfo) bool) (*provider.ResourceInfo, io.ReadCloser, error) {
|
||||
return nil, nil, errtypes.NotSupported("download revision")
|
||||
}
|
||||
|
||||
func (fs *s3FS) RestoreRevision(ctx context.Context, ref *provider.Reference, revisionKey string) error {
|
||||
return errtypes.NotSupported("restore revision")
|
||||
}
|
||||
|
||||
func (fs *s3FS) PurgeRecycleItem(ctx context.Context, ref *provider.Reference, key, relativePath string) error {
|
||||
return errtypes.NotSupported("purge recycle item")
|
||||
}
|
||||
|
||||
func (fs *s3FS) EmptyRecycle(ctx context.Context, ref *provider.Reference) error {
|
||||
return errtypes.NotSupported("empty recycle")
|
||||
}
|
||||
|
||||
func (fs *s3FS) ListRecycle(ctx context.Context, ref *provider.Reference, key, relativePath string) ([]*provider.RecycleItem, error) {
|
||||
return nil, errtypes.NotSupported("list recycle")
|
||||
}
|
||||
|
||||
func (fs *s3FS) RestoreRecycleItem(ctx context.Context, ref *provider.Reference, key, relativePath string, restoreRef *provider.Reference) error {
|
||||
return errtypes.NotSupported("restore recycle")
|
||||
}
|
||||
|
||||
func (fs *s3FS) ListStorageSpaces(ctx context.Context, filter []*provider.ListStorageSpacesRequest_Filter, unrestricted bool) ([]*provider.StorageSpace, error) {
|
||||
return nil, errtypes.NotSupported("list storage spaces")
|
||||
}
|
||||
|
||||
// UpdateStorageSpace updates a storage space
|
||||
func (fs *s3FS) UpdateStorageSpace(ctx context.Context, req *provider.UpdateStorageSpaceRequest) (*provider.UpdateStorageSpaceResponse, error) {
|
||||
return nil, errtypes.NotSupported("update storage space")
|
||||
}
|
||||
|
||||
// DeleteStorageSpace deletes a storage space
|
||||
func (fs *s3FS) DeleteStorageSpace(ctx context.Context, req *provider.DeleteStorageSpaceRequest) error {
|
||||
return errtypes.NotSupported("delete storage space")
|
||||
}
|
||||
-75
@@ -1,75 +0,0 @@
|
||||
// Copyright 2018-2021 CERN
|
||||
//
|
||||
// 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.
|
||||
//
|
||||
// In applying this license, CERN does not waive the privileges and immunities
|
||||
// granted to it by virtue of its status as an Intergovernmental Organization
|
||||
// or submit itself to any jurisdiction.
|
||||
|
||||
package s3
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/awserr"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
"github.com/aws/aws-sdk-go/service/s3/s3manager"
|
||||
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/appctx"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
func (fs *s3FS) Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) {
|
||||
log := appctx.GetLogger(ctx)
|
||||
|
||||
fn, err := fs.resolve(ctx, req.Ref)
|
||||
if err != nil {
|
||||
return &provider.ResourceInfo{}, errors.Wrap(err, "error resolving ref")
|
||||
}
|
||||
|
||||
upParams := &s3manager.UploadInput{
|
||||
Bucket: aws.String(fs.config.Bucket),
|
||||
Key: aws.String(fn),
|
||||
Body: req.Body,
|
||||
}
|
||||
uploader := s3manager.NewUploaderWithClient(fs.client)
|
||||
result, err := uploader.Upload(upParams)
|
||||
|
||||
if err != nil {
|
||||
log.Error().Err(err)
|
||||
if aerr, ok := err.(awserr.Error); ok {
|
||||
if aerr.Code() == s3.ErrCodeNoSuchBucket {
|
||||
return &provider.ResourceInfo{}, errtypes.NotFound(fn)
|
||||
}
|
||||
}
|
||||
return &provider.ResourceInfo{}, errors.Wrap(err, "s3fs: error creating object "+fn)
|
||||
}
|
||||
|
||||
log.Debug().Interface("result", result) // todo cache etag?
|
||||
|
||||
// return id, etag and mtime
|
||||
ri, err := fs.GetMD(ctx, req.Ref, []string{}, []string{"id", "etag", "mtime"})
|
||||
if err != nil {
|
||||
return &provider.ResourceInfo{}, err
|
||||
}
|
||||
|
||||
return ri, nil
|
||||
}
|
||||
|
||||
// InitiateUpload returns upload ids corresponding to different protocols it supports
|
||||
func (fs *s3FS) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) {
|
||||
return nil, errtypes.NotSupported("op not supported")
|
||||
}
|
||||
Reference in New Issue
Block a user