storj/cmd/uplinkng/ulfs/remote.go
Jeff Wendling 08d860570b cmd/uplinkng: parallelsm and a ton of fixes
this was just supposed to add parallel uploads/downloads
and it does do that, but i then found a bunch of bugs
with respect to path handling that i thought i had under
control. oops.

so this adds a ton of tests and tries to make the logic
in ulloc to be more consistent. almost all of the actual
file handling bits and knowledge happens in cmd_cp now
where it should belong.

additionally, the s3 command has the behavior that if your
bucket has the file s3://bucket/file, then executing
s3 ls s3://bucket/fi returns nothing. this change makes
uplinkng match that behavior even if i don't personally
like it.

a big portion of the weirdness is the concept introduced
that i've named "directoryish", which intends to capture
the behavior that if a user copies a file to that location
then the base name of the source should be appended on
rather than a direct copy. this concept is entirely a
based on the string value and not the actual filesystem
state. hence, the cp command is responsible for checking
if local paths are actually a directory, and adding a
trailing slash if necessary to make them "directoryish".
additionally, the empty key for a bucket and the empty
string for local paths are considered "directoryish".

Change-Id: I9120d18616fd813b29ff81beed4f5993caa99fb6
2021-08-11 02:30:06 +00:00

147 lines
4.2 KiB
Go

// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package ulfs
import (
"context"
"strings"
"github.com/zeebo/errs"
"storj.io/storj/cmd/uplinkng/ulloc"
"storj.io/uplink"
)
// Remote implements something close to a filesystem but backed by an uplink project.
type Remote struct {
project *uplink.Project
}
// NewRemote returns something close to a filesystem and returns objects using the project.
func NewRemote(project *uplink.Project) *Remote {
return &Remote{
project: project,
}
}
// Close releases any resources that the Remote contains.
func (r *Remote) Close() error {
return r.project.Close()
}
// Open returns a ReadHandle for the object identified by a given bucket and key.
func (r *Remote) Open(ctx context.Context, bucket, key string) (ReadHandle, error) {
fh, err := r.project.DownloadObject(ctx, bucket, key, nil)
if err != nil {
return nil, errs.Wrap(err)
}
return newUplinkReadHandle(bucket, fh), nil
}
// Create returns a WriteHandle for the object identified by a given bucket and key.
func (r *Remote) Create(ctx context.Context, bucket, key string) (WriteHandle, error) {
fh, err := r.project.UploadObject(ctx, bucket, key, nil)
if err != nil {
return nil, err
}
return newUplinkWriteHandle(fh), nil
}
// Remove deletes the object at the provided key and bucket.
func (r *Remote) Remove(ctx context.Context, bucket, key string) error {
_, err := r.project.DeleteObject(ctx, bucket, key)
if err != nil {
return err
}
return nil
}
// ListObjects lists all of the objects in some bucket that begin with the given prefix.
func (r *Remote) ListObjects(ctx context.Context, bucket, prefix string, recursive bool) ObjectIterator {
parentPrefix := ""
if idx := strings.LastIndexByte(prefix, '/'); idx >= 0 {
parentPrefix = prefix[:idx+1]
}
trim := ulloc.NewRemote(bucket, "")
if !recursive {
trim = ulloc.NewRemote(bucket, parentPrefix)
}
return &filteredObjectIterator{
trim: trim,
filter: ulloc.NewRemote(bucket, prefix),
iter: newUplinkObjectIterator(bucket, r.project.ListObjects(ctx, bucket,
&uplink.ListObjectsOptions{
Prefix: parentPrefix,
Recursive: recursive,
System: true,
})),
}
}
// ListUploads lists all of the pending uploads in some bucket that begin with the given prefix.
func (r *Remote) ListUploads(ctx context.Context, bucket, prefix string, recursive bool) ObjectIterator {
parentPrefix := ""
if idx := strings.LastIndexByte(prefix, '/'); idx >= 0 {
parentPrefix = prefix[:idx+1]
}
trim := ulloc.NewRemote(bucket, "")
if !recursive {
trim = ulloc.NewRemote(bucket, parentPrefix)
}
return &filteredObjectIterator{
trim: trim,
filter: ulloc.NewRemote(bucket, prefix),
iter: newUplinkUploadIterator(bucket, r.project.ListUploads(ctx, bucket,
&uplink.ListUploadsOptions{
Prefix: parentPrefix,
Recursive: recursive,
System: true,
})),
}
}
// uplinkObjectIterator implements objectIterator for *uplink.ObjectIterator.
type uplinkObjectIterator struct {
bucket string
iter *uplink.ObjectIterator
}
// newUplinkObjectIterator constructs an *uplinkObjectIterator from an *uplink.ObjectIterator.
func newUplinkObjectIterator(bucket string, iter *uplink.ObjectIterator) *uplinkObjectIterator {
return &uplinkObjectIterator{
bucket: bucket,
iter: iter,
}
}
func (u *uplinkObjectIterator) Next() bool { return u.iter.Next() }
func (u *uplinkObjectIterator) Err() error { return u.iter.Err() }
func (u *uplinkObjectIterator) Item() ObjectInfo {
return uplinkObjectToObjectInfo(u.bucket, u.iter.Item())
}
// uplinkUploadIterator implements objectIterator for *multipart.UploadIterators.
type uplinkUploadIterator struct {
bucket string
iter *uplink.UploadIterator
}
// newUplinkUploadIterator constructs a *uplinkUploadIterator from a *uplink.UploadIterator.
func newUplinkUploadIterator(bucket string, iter *uplink.UploadIterator) *uplinkUploadIterator {
return &uplinkUploadIterator{
bucket: bucket,
iter: iter,
}
}
func (u *uplinkUploadIterator) Next() bool { return u.iter.Next() }
func (u *uplinkUploadIterator) Err() error { return u.iter.Err() }
func (u *uplinkUploadIterator) Item() ObjectInfo {
return uplinkUploadInfoToObjectInfo(u.bucket, u.iter.Item())
}