From 5185eebabc2b4b758a3dc722398eb71fe6363a02 Mon Sep 17 00:00:00 2001 From: Firas Frikha Date: Mon, 31 Aug 2026 09:39:37 +0100 Subject: [PATCH] feat: remove InitiateUpload/Upload from storage.FS --- pkg/ocm/storage/received/upload.go | 379 ------------- pkg/rhttp/datatx/manager/simple/simple.go | 2 +- pkg/rhttp/datatx/manager/spaces/spaces.go | 2 +- pkg/storage/fs/cephfs/upload.go | 398 ------------- pkg/storage/fs/hello/unimplemented.go | 10 - pkg/storage/fs/kiteworks/kiteworks.go | 8 - pkg/storage/fs/kiteworks/kiteworks_test.go | 4 - pkg/storage/fs/nextcloud/nextcloud.go | 44 -- .../fs/nextcloud/nextcloud_server_mock.go | 6 + pkg/storage/fs/nextcloud/nextcloud_test.go | 55 +- pkg/storage/fs/owncloudsql/upload.go | 523 ------------------ pkg/storage/fs/posix/posix.go | 38 -- pkg/storage/fs/s3/upload.go | 48 -- pkg/storage/storage.go | 11 - pkg/storage/uploads.go | 19 - pkg/storage/utils/decomposedfs/upload.go | 353 ------------ .../utils/decomposedfs/upload/session.go | 4 +- pkg/storage/utils/decomposedfs/upload_test.go | 98 +--- pkg/storage/utils/eosfs/upload.go | 73 --- pkg/storage/utils/localfs/upload.go | 369 ------------ pkg/storage/utils/middleware/middleware.go | 92 --- pkg/upload/coordinator.go | 16 +- pkg/upload/coordinator_test.go | 6 +- pkg/upload/put_test.go | 12 +- tests/helpers/helpers.go | 24 +- 25 files changed, 83 insertions(+), 2511 deletions(-) diff --git a/pkg/ocm/storage/received/upload.go b/pkg/ocm/storage/received/upload.go index 3e72c1f26bf..cef04d347e4 100644 --- a/pkg/ocm/storage/received/upload.go +++ b/pkg/ocm/storage/received/upload.go @@ -20,63 +20,21 @@ package ocm import ( "context" - "crypto/md5" - "crypto/sha1" - "encoding/hex" - "encoding/json" - "errors" - "fmt" - "hash" - "hash/adler32" - "io" "net/http" - "os" - "path/filepath" - "strings" - "github.com/google/uuid" "github.com/studio-b12/gowebdav" - tusd "github.com/tus/tusd/v2/pkg/handler" - userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" ocmpb "github.com/cs3org/go-cs3apis/cs3/sharing/ocm/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" - "github.com/owncloud/reva/v2/pkg/appctx" ctxpkg "github.com/owncloud/reva/v2/pkg/ctx" "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/storage" "github.com/owncloud/reva/v2/pkg/utils" ) -var defaultFilePerm = os.FileMode(0664) - func (d *driver) ListUploadSessions(ctx context.Context, filter storage.UploadSessionFilter) ([]storage.UploadSession, error) { return []storage.UploadSession{}, nil } -func (d *driver) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - shareID, rel := shareInfoFromReference(ref) - p := getPathFromShareIDAndRelPath(shareID, rel) - - info := tusd.FileInfo{ - MetaData: tusd.MetaData{ - "filename": filepath.Base(p), - "dir": filepath.Dir(p), - }, - Size: uploadLength, - } - - upload, err := d.NewUpload(ctx, info) - if err != nil { - return nil, err - } - - info, _ = upload.GetInfo(ctx) - - return map[string]string{ - "simple": info.ID, - "tus": info.ID, - }, nil -} // MarkProcessing is a no-op: the file lives on the remote instance, so there is no // local node to flag while postprocessing runs. @@ -124,340 +82,3 @@ func (d *driver) serviceWebdavClient(ctx context.Context, ref *provider.Referenc } return d.webdavClient(serviceUserCtx, executant.GetId(), ref) } - -func (d *driver) Upload(ctx context.Context, req storage.UploadRequest, _ storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - shareID, _ := shareInfoFromReference(req.Ref) - u, err := d.GetUpload(ctx, shareID.OpaqueId) - if err != nil { - return &provider.ResourceInfo{}, err - } - - info, err := u.GetInfo(ctx) - if err != nil { - return &provider.ResourceInfo{}, err - } - - defer cleanup(&upload{Info: info}) - - client, _, rel, err := d.webdavClient(ctx, nil, &provider.Reference{ - Path: filepath.Join(info.MetaData["dir"], info.MetaData["filename"]), - }) - if err != nil { - return &provider.ResourceInfo{}, err - } - client.SetInterceptor(func(method string, rq *http.Request) { - // Set the content length on the request struct directly instead of the header. - // The content-length header gets reset by the golang http library before - // sendind out the request, resulting in chunked encoding to be used which - // breaks the quota checks in ocdav. - if method == "PUT" { - rq.ContentLength = req.Length - } - }) - - locktoken, _ := ctxpkg.ContextGetLockID(ctx) - return &provider.ResourceInfo{}, client.WriteStream(rel, req.Body, 0, locktoken) -} - -// UseIn tells the tus upload middleware which extensions it supports. -func (d *driver) UseIn(composer *tusd.StoreComposer) { - composer.UseCore(d) - composer.UseTerminater(d) - composer.UseConcater(d) - composer.UseLengthDeferrer(d) -} - -// AsTerminatableUpload returns a TerminatableUpload -// To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination -// the storage needs to implement AsTerminatableUpload -func (d *driver) AsTerminatableUpload(up tusd.Upload) tusd.TerminatableUpload { - return up.(*upload) -} - -// AsLengthDeclarableUpload returns a LengthDeclarableUpload -// To implement the creation-defer-length extension as specified in https://tus.io/protocols/resumable-upload.html#creation -// the storage needs to implement AsLengthDeclarableUpload -func (d *driver) AsLengthDeclarableUpload(up tusd.Upload) tusd.LengthDeclarableUpload { - return up.(*upload) -} - -// AsConcatableUpload returns a ConcatableUpload -// To implement the concatenation extension as specified in https://tus.io/protocols/resumable-upload.html#concatenation -// the storage needs to implement AsConcatableUpload -func (d *driver) AsConcatableUpload(up tusd.Upload) tusd.ConcatableUpload { - return up.(*upload) -} - -// To implement the core tus.io protocol as specified in https://tus.io/protocols/resumable-upload.html#core-protocol -// - the storage needs to implement NewUpload and GetUpload -// - the upload needs to implement the tusd.Upload interface: WriteChunk, GetInfo, GetReader and FinishUpload - -// NewUpload returns a new tus Upload instance -func (d *driver) NewUpload(ctx context.Context, info tusd.FileInfo) (tusd.Upload, error) { - return NewUpload(ctx, d, d.c.StorageRoot, info) -} - -// GetUpload returns the Upload for the given upload id -func (d *driver) GetUpload(ctx context.Context, id string) (tusd.Upload, error) { - return GetUpload(ctx, d, d.c.StorageRoot, id) -} -func NewUpload(ctx context.Context, d *driver, storageRoot string, info tusd.FileInfo) (tusd.Upload, error) { - if info.MetaData["filename"] == "" { - return nil, errors.New("Decomposedfs: missing filename in metadata") - } - if info.MetaData["dir"] == "" { - return nil, errors.New("Decomposedfs: missing dir in metadata") - } - - uploadRoot := filepath.Join(storageRoot, "uploads") - info.ID = uuid.New().String() - - user, ok := ctxpkg.ContextGetUser(ctx) - if !ok { - return nil, errors.New("no user in context") - } - info.MetaData["user"] = user.GetId().GetOpaqueId() - info.MetaData["idp"] = user.GetId().GetIdp() - - info.Storage = map[string]string{ - "Type": "OCM", - "Path": uploadRoot, - } - - u := &upload{ - Info: info, - Ctx: ctx, - d: d, - } - - err := os.MkdirAll(uploadRoot, 0755) - if err != nil { - return nil, err - } - - file, err := os.OpenFile(u.BinPath(), os.O_CREATE|os.O_WRONLY, defaultFilePerm) - if err != nil { - return nil, err - } - defer file.Close() - - err = u.Persist() - if err != nil { - return nil, err - } - return u, nil -} - -func GetUpload(ctx context.Context, d *driver, storageRoot string, id string) (tusd.Upload, error) { - info := tusd.FileInfo{} - data, err := os.ReadFile(filepath.Join(storageRoot, "uploads", id+".info")) - if err != nil { - return nil, err - } - err = json.Unmarshal(data, &info) - if err != nil { - return nil, err - } - upload := &upload{ - Info: info, - Ctx: ctx, - d: d, - } - return upload, nil -} - -type upload struct { - Info tusd.FileInfo - Ctx context.Context - - d *driver -} - -func (u *upload) InfoPath() string { - return filepath.Join(u.Info.Storage["Path"], u.Info.ID+".info") -} - -func (u *upload) BinPath() string { - return filepath.Join(u.Info.Storage["Path"], u.Info.ID) -} - -func (u *upload) Persist() error { - data, err := json.Marshal(u.Info) - if err != nil { - return err - } - return os.WriteFile(u.InfoPath(), data, defaultFilePerm) -} - -func (u *upload) WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error) { - file, err := os.OpenFile(u.BinPath(), os.O_WRONLY|os.O_APPEND, defaultFilePerm) - if err != nil { - return 0, err - } - defer file.Close() - - // calculate cheksum here? needed for the TUS checksum extension. https://tus.io/protocols/resumable-upload.html#checksum - // TODO but how do we get the `Upload-Checksum`? WriteChunk() only has a context, offset and the reader ... - // It is sent with the PATCH request, well or in the POST when the creation-with-upload extension is used - // but the tus handler uses a context.Background() so we cannot really check the header and put it in the context ... - n, err := io.Copy(file, src) - - // If the HTTP PATCH request gets interrupted in the middle (e.g. because - // the user wants to pause the upload), Go's net/http returns an io.ErrUnexpectedEOF. - // However, for the ocis driver it's not important whether the stream has ended - // on purpose or accidentally. - if err != nil && err != io.ErrUnexpectedEOF { - return n, err - } - - u.Info.Offset += n - return n, u.Persist() -} - -func (u *upload) GetInfo(ctx context.Context) (tusd.FileInfo, error) { - return u.Info, nil -} - -func (u *upload) GetReader(ctx context.Context) (io.ReadCloser, error) { - return os.Open(u.BinPath()) -} - -func (u *upload) FinishUpload(ctx context.Context) error { - log := appctx.GetLogger(u.Ctx) - - // calculate the checksum of the written bytes - // they will all be written to the metadata later, so we cannot omit any of them - // TODO only calculate the checksum in sync that was requested to match, the rest could be async ... but the tests currently expect all to be present - // TODO the hashes all implement BinaryMarshaler so we could try to persist the state for resumable upload. we would neet do keep track of the copied bytes ... - sha1h := sha1.New() - md5h := md5.New() - adler32h := adler32.New() - { - f, err := os.Open(u.BinPath()) - if err != nil { - // we can continue if no oc checksum header is set - log.Info().Err(err).Str("binPath", u.BinPath()).Msg("error opening binPath") - } - defer f.Close() - - r1 := io.TeeReader(f, sha1h) - r2 := io.TeeReader(r1, md5h) - - _, err = io.Copy(adler32h, r2) - if err != nil { - log.Info().Err(err).Msg("error copying checksums") - } - } - - defer cleanup(u) - // compare if they match the sent checksum - // TODO the tus checksum extension would do this on every chunk, but I currently don't see an easy way to pass in the requested checksum. for now we do it in FinishUpload which is also called for chunked uploads - if u.Info.MetaData["checksum"] != "" { - var err error - parts := strings.SplitN(u.Info.MetaData["checksum"], " ", 2) - if len(parts) != 2 { - return errtypes.BadRequest("invalid checksum format. must be '[algorithm] [checksum]'") - } - switch parts[0] { - case "sha1": - err = u.checkHash(parts[1], sha1h) - case "md5": - err = u.checkHash(parts[1], md5h) - case "adler32": - err = u.checkHash(parts[1], adler32h) - default: - err = errtypes.BadRequest("unsupported checksum algorithm: " + parts[0]) - } - if err != nil { - return err - } - } - - // send to the remote storage via webdav - // shareID, rel := shareInfoFromReference(u.Info.MetaData["ref"]) - // p := getPathFromShareIDAndRelPath(shareID, rel) - - gwc, err := u.d.gateway.Next() - if err != nil { - return err - } - serviceUserCtx, err := utils.GetServiceUserContext(u.d.c.ServiceAccountID, gwc, u.d.c.ServiceAccountSecret) - if err != nil { - return err - } - client, _, rel, err := u.d.webdavClient(serviceUserCtx, &userpb.UserId{ - OpaqueId: u.Info.MetaData["user"], - Idp: u.Info.MetaData["idp"], - }, &provider.Reference{ - Path: filepath.Join(u.Info.MetaData["dir"], u.Info.MetaData["filename"]), - }) - if err != nil { - return err - } - - client.SetInterceptor(func(method string, rq *http.Request) { - // Set the content length on the request struct directly instead of the header. - // The content-length header gets reset by the golang http library before - // sendind out the request, resulting in chunked encoding to be used which - // breaks the quota checks in ocdav. - if method == "PUT" { - rq.ContentLength = u.Info.Size - } - }) - - f, err := os.Open(u.BinPath()) - if err != nil { - return err - } - defer f.Close() - return client.WriteStream(rel, f, 0, "") -} - -func (u *upload) Terminate(ctx context.Context) error { - cleanup(u) - return nil -} - -func (u *upload) ConcatUploads(_ context.Context, uploads []tusd.Upload) error { - file, err := os.OpenFile(u.BinPath(), os.O_WRONLY|os.O_APPEND, defaultFilePerm) - if err != nil { - return err - } - defer file.Close() - - for _, partialUpload := range uploads { - fileUpload := partialUpload.(*upload) - - src, err := os.Open(fileUpload.BinPath()) - if err != nil { - return err - } - defer src.Close() - - if _, err := io.Copy(file, src); err != nil { - return err - } - } - return nil -} - -func (u *upload) DeclareLength(ctx context.Context, length int64) error { - u.Info.Size = length - u.Info.SizeIsDeferred = false - return nil -} - -func (u *upload) checkHash(expected string, h hash.Hash) error { - if expected != hex.EncodeToString(h.Sum(nil)) { - return errtypes.ChecksumMismatch(fmt.Sprintf("invalid checksum: expected %s got %x", u.Info.MetaData["checksum"], h.Sum(nil))) - } - return nil -} - -func cleanup(u *upload) { - if u == nil { - return - } - _ = os.Remove(u.BinPath()) - _ = os.Remove(u.InfoPath()) -} diff --git a/pkg/rhttp/datatx/manager/simple/simple.go b/pkg/rhttp/datatx/manager/simple/simple.go index 057c0d90bd5..fb2ca687f1d 100644 --- a/pkg/rhttp/datatx/manager/simple/simple.go +++ b/pkg/rhttp/datatx/manager/simple/simple.go @@ -115,7 +115,7 @@ func (m *manager) Handler(coord upload.Coordinator, driver storage.FS) (http.Han ctx = ctxpkg.ContextSetLockID(ctx, lockID) } - info, err := coord.Upload(ctx, storage.UploadRequest{ + info, err := coord.Upload(ctx, upload.Request{ Ref: ref, Body: r.Body, Length: r.ContentLength, diff --git a/pkg/rhttp/datatx/manager/spaces/spaces.go b/pkg/rhttp/datatx/manager/spaces/spaces.go index 514bccf1df4..5a018c31859 100644 --- a/pkg/rhttp/datatx/manager/spaces/spaces.go +++ b/pkg/rhttp/datatx/manager/spaces/spaces.go @@ -118,7 +118,7 @@ func (m *manager) Handler(coord upload.Coordinator, driver storage.FS) (http.Han Path: fn, } var info *provider.ResourceInfo - info, err = coord.Upload(ctx, storage.UploadRequest{ + info, err = coord.Upload(ctx, upload.Request{ Ref: ref, Body: r.Body, Length: r.ContentLength, diff --git a/pkg/storage/fs/cephfs/upload.go b/pkg/storage/fs/cephfs/upload.go index bcd42033a8a..ce8d862b8de 100644 --- a/pkg/storage/fs/cephfs/upload.go +++ b/pkg/storage/fs/cephfs/upload.go @@ -22,131 +22,13 @@ package cephfs import ( - "bytes" "context" - "encoding/json" - "io" - "os" - "path/filepath" - cephfs2 "github.com/ceph/go-ceph/cephfs" - userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" - "github.com/google/uuid" - "github.com/owncloud/reva/v2/pkg/appctx" - ctx2 "github.com/owncloud/reva/v2/pkg/ctx" "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/storage" - "github.com/owncloud/reva/v2/pkg/utils" - "github.com/pkg/errors" - tusd "github.com/tus/tusd/v2/pkg/handler" ) -func (fs *cephfs) Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - user := fs.makeUser(ctx) - upload, err := fs.GetUpload(ctx, req.Ref.GetPath()) - if err != nil { - metadata := map[string]string{"sizedeferred": "true"} - uploadIDs, err := fs.InitiateUpload(ctx, req.Ref, 0, metadata) - if err != nil { - return &provider.ResourceInfo{}, err - } - if upload, err = fs.GetUpload(ctx, uploadIDs["simple"]); err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "cephfs: error retrieving upload") - } - } - - uploadInfo := upload.(*fileUpload) - - p := uploadInfo.info.Storage["InternalDestination"] - ok, err := IsChunked(p) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "cephfs: error checking path") - } - if ok { - var assembledFile string - p, assembledFile, err = NewChunkHandler(ctx, fs).WriteChunk(p, req.Body) - if err != nil { - return &provider.ResourceInfo{}, err - } - if p == "" { - if err = uploadInfo.Terminate(ctx); err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "cephfs: error removing auxiliary files") - } - return &provider.ResourceInfo{}, errtypes.PartialContent(req.Ref.String()) - } - uploadInfo.info.Storage["InternalDestination"] = p - - user.op(func(cv *cacheVal) { - req.Body, err = cv.mount.Open(assembledFile, os.O_RDONLY, 0) - }) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "cephfs: error opening assembled file") - } - defer req.Body.Close() - defer user.op(func(cv *cacheVal) { - _ = cv.mount.Unlink(assembledFile) - }) - } - ri := &provider.ResourceInfo{ - // fill with at least fileid, mtime and etag - Id: &provider.ResourceId{ - StorageId: uploadInfo.info.MetaData["providerID"], - SpaceId: uploadInfo.info.Storage["SpaceRoot"], - OpaqueId: uploadInfo.info.Storage["NodeId"], - }, - Etag: uploadInfo.info.MetaData["etag"], - } - - if mtime, err := utils.MTimeToTS(uploadInfo.info.MetaData["mtime"]); err == nil { - ri.Mtime = &mtime - } - - if _, err := uploadInfo.WriteChunk(ctx, 0, req.Body); err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "cephfs: error writing to binary file") - } - - return ri, uploadInfo.FinishUpload(ctx) -} - -func (fs *cephfs) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - user := fs.makeUser(ctx) - np, err := user.resolveRef(ref) - if err != nil { - return nil, errors.Wrap(err, "cephfs: error resolving reference") - } - - info := tusd.FileInfo{ - MetaData: tusd.MetaData{ - "filename": filepath.Base(np), - "dir": filepath.Dir(np), - }, - Size: uploadLength, - } - - if metadata != nil { - info.MetaData["providerID"] = metadata["providerID"] - if metadata["mtime"] != "" { - info.MetaData["mtime"] = metadata["mtime"] - } - if _, ok := metadata["sizedeferred"]; ok { - info.SizeIsDeferred = true - } - } - - upload, err := fs.NewUpload(ctx, info) - if err != nil { - return nil, err - } - - info, _ = upload.GetInfo(ctx) - - return map[string]string{ - "simple": info.ID, - "tus": info.ID, - }, nil -} - func (fs *cephfs) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error { return errtypes.NotSupported("op not supported") } @@ -162,283 +44,3 @@ func (fs *cephfs) PrepareUpload(_ context.Context, _ *provider.Reference, _ stri func (fs *cephfs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error { return nil } - -// UseIn tells the tus upload middleware which extensions it supports. -func (fs *cephfs) UseIn(composer *tusd.StoreComposer) { - composer.UseCore(fs) - composer.UseTerminater(fs) -} - -func (fs *cephfs) NewUpload(ctx context.Context, info tusd.FileInfo) (upload tusd.Upload, err error) { - log := appctx.GetLogger(ctx) - log.Debug().Interface("info", info).Msg("cephfs: NewUpload") - - user := fs.makeUser(ctx) - - fn := info.MetaData["filename"] - if fn == "" { - return nil, errors.New("cephfs: missing filename in metadata") - } - info.MetaData["filename"] = filepath.Clean(info.MetaData["filename"]) - - dir := info.MetaData["dir"] - if dir == "" { - return nil, errors.New("cephfs: missing dir in metadata") - } - info.MetaData["dir"] = filepath.Clean(info.MetaData["dir"]) - - np := filepath.Join(info.MetaData["dir"], info.MetaData["filename"]) - - info.ID = uuid.New().String() - - binPath := fs.getUploadPath(info.ID) - - info.Storage = map[string]string{ - "Type": "Cephfs", - "BinPath": binPath, - "InternalDestination": np, - - "Idp": user.Id.Idp, - "UserId": user.Id.OpaqueId, - "UserName": user.Username, - "UserType": utils.UserTypeToString(user.Id.Type), - - "LogLevel": log.GetLevel().String(), - } - - // Create binary file with no content - user.op(func(cv *cacheVal) { - var f *cephfs2.File - defer closeFile(f) - f, err = cv.mount.Open(binPath, os.O_CREATE|os.O_WRONLY, filePermDefault) - if err != nil { - return - } - }) - //TODO: if we get two same upload ids, the second one can't upload at all - if err != nil { - return - } - - upload = &fileUpload{ - info: info, - binPath: binPath, - infoPath: binPath + ".info", - fs: fs, - ctx: ctx, - } - - if !info.SizeIsDeferred && info.Size == 0 { - log.Debug().Interface("info", info).Msg("cephfs: finishing upload for empty file") - // no need to create info file and finish directly - err = upload.FinishUpload(ctx) - - return - } - - // writeInfo creates the file by itself if necessary - err = upload.(*fileUpload).writeInfo() - - return -} - -func (fs *cephfs) getUploadPath(uploadID string) string { - return filepath.Join(fs.conf.UploadFolder, uploadID) -} - -// GetUpload returns the Upload for the given upload id -func (fs *cephfs) GetUpload(ctx context.Context, id string) (fup tusd.Upload, err error) { - binPath := fs.getUploadPath(id) - info := tusd.FileInfo{} - if err != nil { - return nil, errtypes.NotFound("bin path for upload " + id + " not found") - } - infoPath := binPath + ".info" - - var data bytes.Buffer - f, err := fs.adminConn.adminMount.Open(infoPath, os.O_RDONLY, 0) - if err != nil { - return - } - _, err = io.Copy(&data, f) - if err != nil { - return - } - if err = json.Unmarshal(data.Bytes(), &info); err != nil { - return - } - - u := &userpb.User{ - Id: &userpb.UserId{ - Idp: info.Storage["Idp"], - OpaqueId: info.Storage["UserId"], - }, - Username: info.Storage["UserName"], - } - ctx = ctx2.ContextSetUser(ctx, u) - user := fs.makeUser(ctx) - - var stat Statx - user.op(func(cv *cacheVal) { - stat, err = cv.mount.Statx(binPath, cephfs2.StatxSize, 0) - }) - if err != nil { - return - } - info.Offset = int64(stat.Size) - - return &fileUpload{ - info: info, - binPath: binPath, - infoPath: infoPath, - fs: fs, - ctx: ctx, - }, nil -} - -type fileUpload struct { - // info stores the current information about the upload - info tusd.FileInfo - // infoPath is the path to the .info file - infoPath string - // binPath is the path to the binary file (which has no extension) - binPath string - // only fs knows how to handle metadata and versions - fs *cephfs - // a context with a user - ctx context.Context -} - -// GetInfo returns the FileInfo -func (upload *fileUpload) GetInfo(ctx context.Context) (tusd.FileInfo, error) { - return upload.info, nil -} - -// GetReader returns an io.Reader for the upload -func (upload *fileUpload) GetReader(ctx context.Context) (file io.ReadCloser, err error) { - user := upload.fs.makeUser(upload.ctx) - user.op(func(cv *cacheVal) { - file, err = cv.mount.Open(upload.binPath, os.O_RDONLY, 0) - }) - return -} - -// WriteChunk writes the stream from the reader to the given offset of the upload -func (upload *fileUpload) WriteChunk(ctx context.Context, offset int64, src io.Reader) (n int64, err error) { - var file io.WriteCloser - user := upload.fs.makeUser(upload.ctx) - user.op(func(cv *cacheVal) { - file, err = cv.mount.Open(upload.binPath, os.O_WRONLY|os.O_APPEND, 0) - }) - if err != nil { - return 0, err - } - defer file.Close() - - n, err = io.Copy(file, src) - - // If the HTTP PATCH request gets interrupted in the middle (e.g. because - // the user wants to pause the upload), Go's net/http returns an io.ErrUnexpectedEOF. - // However, for OwnCloudStore it's not important whether the stream has ended - // on purpose or accidentally. - if err != nil { - if err != io.ErrUnexpectedEOF { - return n, err - } - } - - upload.info.Offset += n - err = upload.writeInfo() - - return n, err -} - -// writeInfo updates the entire information. Everything will be overwritten. -func (upload *fileUpload) writeInfo() error { - data, err := json.Marshal(upload.info) - - if err != nil { - return err - } - user := upload.fs.makeUser(upload.ctx) - user.op(func(cv *cacheVal) { - var file io.WriteCloser - if file, err = cv.mount.Open(upload.infoPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, filePermDefault); err != nil { - return - } - defer file.Close() - - _, err = io.Copy(file, bytes.NewReader(data)) - }) - - return err -} - -// FinishUpload finishes an upload and moves the file to the internal destination -func (upload *fileUpload) FinishUpload(ctx context.Context) (err error) { - - np := upload.info.Storage["InternalDestination"] - - // TODO check etag with If-Match header - // if destination exists - // if _, err := os.Stat(np); err == nil { - // the local storage does not store metadata - // the fileid is based on the path, so no we do not need to copy it to the new file - // the local storage does not track revisions - // } - - // if destination exists - // if _, err := os.Stat(np); err == nil { - // create revision - // if err := upload.fs.archiveRevision(upload.ctx, np); err != nil { - // return err - // } - // } - - user := upload.fs.makeUser(upload.ctx) - log := appctx.GetLogger(ctx) - - user.op(func(cv *cacheVal) { - err = cv.mount.Rename(upload.binPath, np) - }) - if err != nil { - return errors.Wrap(err, upload.binPath) - } - - // only delete the upload if it was successfully written to the fs - user.op(func(cv *cacheVal) { - err = cv.mount.Unlink(upload.infoPath) - }) - if err != nil { - if err.Error() != errNotFound { - log.Err(err).Interface("info", upload.info).Msg("cephfs: could not delete upload metadata") - } - } - - // TODO: set mtime if specified in metadata - - return -} - -// To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination -// - the storage needs to implement AsTerminatableUpload -// - the upload needs to implement Terminate - -// AsTerminatableUpload returns a a TerminatableUpload -func (fs *cephfs) AsTerminatableUpload(upload tusd.Upload) tusd.TerminatableUpload { - return upload.(*fileUpload) -} - -// Terminate terminates the upload -func (upload *fileUpload) Terminate(ctx context.Context) (err error) { - user := upload.fs.makeUser(upload.ctx) - - user.op(func(cv *cacheVal) { - if err = cv.mount.Unlink(upload.infoPath); err != nil { - return - } - err = cv.mount.Unlink(upload.binPath) - }) - - return -} diff --git a/pkg/storage/fs/hello/unimplemented.go b/pkg/storage/fs/hello/unimplemented.go index a018c96c8ac..daf25bdc56c 100644 --- a/pkg/storage/fs/hello/unimplemented.go +++ b/pkg/storage/fs/hello/unimplemented.go @@ -74,16 +74,6 @@ func (fs *hellofs) Move(ctx context.Context, oldRef, newRef *provider.Reference) return nil, errtypes.NotSupported("unimplemented") } -// Upload creates or updates a resource of type file with a new revision -func (fs *hellofs) Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - return nil, errtypes.NotSupported("hellofs: upload not supported") -} - -// InitiateUpload returns a list of protocols with urls that can be used to append bytes to a new upload session -func (fs *hellofs) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - return nil, errtypes.NotSupported("hellofs: initiate upload not supported") -} - // MarkProcessing toggles a processing flag on the resource. func (fs *hellofs) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error { return errtypes.NotSupported("hellofs: mark processing not supported") diff --git a/pkg/storage/fs/kiteworks/kiteworks.go b/pkg/storage/fs/kiteworks/kiteworks.go index 6f7f00edc93..2e0f4a98d75 100644 --- a/pkg/storage/fs/kiteworks/kiteworks.go +++ b/pkg/storage/fs/kiteworks/kiteworks.go @@ -259,14 +259,6 @@ func (d *Driver) Move(_ context.Context, _, _ *provider.Reference) (*storage.Mov return nil, errtypes.NotSupported("kiteworks: read-only driver") } -func (d *Driver) InitiateUpload(_ context.Context, _ *provider.Reference, _ int64, _ map[string]string) (map[string]string, error) { - return nil, errtypes.NotSupported("kiteworks: read-only driver") -} - -func (d *Driver) Upload(_ context.Context, _ storage.UploadRequest, _ storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - return nil, errtypes.NotSupported("kiteworks: read-only driver") -} - func (d *Driver) MarkProcessing(_ context.Context, _ *provider.Reference, _ bool, _ string) error { return errtypes.NotSupported("kiteworks: read-only driver") } diff --git a/pkg/storage/fs/kiteworks/kiteworks_test.go b/pkg/storage/fs/kiteworks/kiteworks_test.go index a451ac31ce0..123f734c2c9 100644 --- a/pkg/storage/fs/kiteworks/kiteworks_test.go +++ b/pkg/storage/fs/kiteworks/kiteworks_test.go @@ -234,9 +234,5 @@ var _ = Describe("kiteworks driver", func() { err := d.AddGrant(fix.ctx, &provider.Reference{ResourceId: &provider.ResourceId{SpaceId: fix.spaceID}}, &provider.Grant{}) Expect(err).To(Satisfy(notSupported)) }) - It("rejects InitiateUpload", func() { - _, err := d.InitiateUpload(fix.ctx, &provider.Reference{ResourceId: &provider.ResourceId{SpaceId: fix.spaceID}}, 0, nil) - Expect(err).To(Satisfy(notSupported)) - }) }) }) diff --git a/pkg/storage/fs/nextcloud/nextcloud.go b/pkg/storage/fs/nextcloud/nextcloud.go index b410cfda1bf..0d620191083 100644 --- a/pkg/storage/fs/nextcloud/nextcloud.go +++ b/pkg/storage/fs/nextcloud/nextcloud.go @@ -405,34 +405,6 @@ func (nc *StorageDriver) ListFolder(ctx context.Context, ref *provider.Reference return pointers, err } -// InitiateUpload as defined in the storage.FS interface -func (nc *StorageDriver) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - type paramsObj struct { - Ref *provider.Reference `json:"ref"` - UploadLength int64 `json:"uploadLength"` - Metadata map[string]string `json:"metadata"` - } - bodyObj := ¶msObj{ - Ref: ref, - UploadLength: uploadLength, - Metadata: metadata, - } - bodyStr, _ := json.Marshal(bodyObj) - log := appctx.GetLogger(ctx) - log.Info().Msgf("InitiateUpload %s", bodyStr) - - _, respBody, err := nc.do(ctx, Action{"InitiateUpload", string(bodyStr)}) - if err != nil { - return nil, err - } - respMap := make(map[string]string) - err = json.Unmarshal(respBody, &respMap) - if err != nil { - return nil, err - } - return respMap, err -} - // MarkProcessing as defined in the storage.FS interface. // No sciencemesh endpoint toggles the flag alone. A no-op, not NotSupported: // the coordinator treats a failed mark as fatal. @@ -457,22 +429,6 @@ func (nc *StorageDriver) RollbackUpload(_ context.Context, _ *provider.Reference return nil } -// Upload as defined in the storage.FS interface -func (nc *StorageDriver) Upload(ctx context.Context, req storage.UploadRequest, _ storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - err := nc.doUpload(ctx, req.Ref.Path, req.Body) - if err != nil { - return &provider.ResourceInfo{}, err - } - - // return id, etag and mtime - ri, err := nc.GetMD(ctx, req.Ref, []string{}, []string{"id", "etag", "mtime"}) - if err != nil { - return &provider.ResourceInfo{}, err - } - - return ri, nil -} - // Download as defined in the storage.FS interface func (nc *StorageDriver) Download(ctx context.Context, ref *provider.Reference, openReaderfunc func(*provider.ResourceInfo) bool) (*provider.ResourceInfo, io.ReadCloser, error) { md, err := nc.GetMD(ctx, ref, []string{}, nil) diff --git a/pkg/storage/fs/nextcloud/nextcloud_server_mock.go b/pkg/storage/fs/nextcloud/nextcloud_server_mock.go index 1221575a855..7d58ab688fc 100644 --- a/pkg/storage/fs/nextcloud/nextcloud_server_mock.go +++ b/pkg/storage/fs/nextcloud/nextcloud_server_mock.go @@ -114,6 +114,12 @@ var responses = map[string]Response{ `POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/GetMD {"ref":{"path":"/versionedFile"},"mdKeys":null} EMPTY`: {200, `{"opaque":{},"type":1,"id":{"opaque_id":"fileid-/some/path"},"checksum":{},"etag":"deadbeef","mime_type":"text/plain","mtime":{"seconds":1234567890},"path":"/versionedFile","permission_set":{},"size":2,"canonical_metadata":{},"arbitrary_metadata":{"metadata":{"da":"ta","some":"arbi","trary":"meta"}}}`, serverStateEmpty}, `POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/GetMD {"ref":{"path":"/versionedFile"},"mdKeys":null} FILE-RESTORED`: {200, `{"opaque":{},"type":1,"id":{"opaque_id":"fileid-/some/path"},"checksum":{},"etag":"deadbeef","mime_type":"text/plain","mtime":{"seconds":1234567890},"path":"/versionedFile","permission_set":{},"size":1,"canonical_metadata":{},"arbitrary_metadata":{"metadata":{"da":"ta","some":"arbi","trary":"meta"}}}`, serverStateFileRestored}, + // The coordinator resolves the upload target itself, so it stats the file with the + // full reference before every upload. Reporting the file as existing keeps it on the + // overwrite path, which needs neither a parent stat nor a TouchFile (whose mtime is + // wall-clock derived and could never match this table's exact-body keys). + `POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/GetMD {"ref":{"resource_id":{"storage_id":"f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c"},"path":"/versionedFile"},"mdKeys":[]}`: {200, `{"opaque":{},"type":1,"id":{"storage_id":"f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c","opaque_id":"fileid-/versionedFile"},"parent_id":{"opaque_id":"fileid-/"},"name":"versionedFile","checksum":{},"etag":"deadbeef","mime_type":"text/plain","mtime":{"seconds":1234567890},"path":"/versionedFile","permission_set":{"initiate_file_upload":true,"stat":true,"list_container":true,"list_file_versions":true,"restore_file_version":true},"size":1,"canonical_metadata":{},"arbitrary_metadata":{"metadata":{}}}`, serverStateEmpty}, + `POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/GetPathByID {"storage_id":"00000000-0000-0000-0000-000000000000","opaque_id":"fileid-/some/path"} EMPTY`: {200, "/subdir", serverStateEmpty}, `POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/GetMD {"ref":{"path":"/file"},"mdKeys":null}`: {404, ``, serverStateEmpty}, diff --git a/pkg/storage/fs/nextcloud/nextcloud_test.go b/pkg/storage/fs/nextcloud/nextcloud_test.go index a18a96c4226..68420a13fef 100644 --- a/pkg/storage/fs/nextcloud/nextcloud_test.go +++ b/pkg/storage/fs/nextcloud/nextcloud_test.go @@ -375,58 +375,9 @@ var _ = Describe("Nextcloud", func() { }) }) - // InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) - Describe("InitiateUpload", func() { - It("calls the InitiateUpload endpoint", func() { - nc, called, teardown := setUpNextcloudServer() - defer teardown() - // https://github.com/cs3org/go-cs3apis/blob/970eec3/cs3/storage/provider/v1beta1/resources.pb.go#L550-L561 - ref := &provider.Reference{ - ResourceId: &provider.ResourceId{ - StorageId: "storage-id", - OpaqueId: "opaque-id", - }, - Path: "/some/path", - } - uploadLength := int64(12345) - metadata := map[string]string{ - "key1": "val1", - "key2": "val2", - "key3": "val3", - } - results, err := nc.InitiateUpload(ctx, ref, uploadLength, metadata) - Expect(err).ToNot(HaveOccurred()) - Expect(results).To(Equal(map[string]string{ - "not": "sure", - "what": "should be", - "returned": "here", - })) - checkCalled(called, `POST /apps/sciencemesh/~tester/api/storage/InitiateUpload {"ref":{"resource_id":{"storage_id":"storage-id","opaque_id":"opaque-id"},"path":"/some/path"},"uploadLength":12345,"metadata":{"key1":"val1","key2":"val2","key3":"val3"}}`) - }) - }) - - // Upload(ctx context.Context, ref *provider.Reference, r io.ReadCloser) error - Describe("Upload", func() { - It("calls the Upload endpoint", func() { - nc, called, teardown := setUpNextcloudServer() - defer teardown() - // https://github.com/cs3org/go-cs3apis/blob/970eec3/cs3/storage/provider/v1beta1/resources.pb.go#L550-L561 - ref := &provider.Reference{ - ResourceId: &provider.ResourceId{ - StorageId: "storage-id", - OpaqueId: "opaque-id", - }, - Path: "some/file/path.txt", - } - stringReader := strings.NewReader("shiny!") - stringReadCloser := io.NopCloser(stringReader) - _, err := nc.Upload(ctx, storage.UploadRequest{Ref: ref, Body: stringReadCloser, Length: stringReader.Size()}, nil) - Expect(err).ToNot(HaveOccurred()) - Expect(len(*called)).To(Equal(2)) - Expect((*called)[0]).To(Equal(`PUT /apps/sciencemesh/~tester/api/storage/Upload/some/file/path.txt shiny!`)) - Expect((*called)[1]).To(Equal(`POST /apps/sciencemesh/~tester/api/storage/GetMD {"ref":{"resource_id":{"storage_id":"storage-id","opaque_id":"opaque-id"},"path":"some/file/path.txt"},"mdKeys":[]}`)) - }) - }) + // The legacy InitiateUpload and Upload specs are gone with the methods: the + // coordinator initiates uploads itself and CommitUpload issues the same PUT the + // Upload spec asserted, so its coverage lives in the CommitUpload Describe below. // MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error Describe("MarkProcessing", func() { diff --git a/pkg/storage/fs/owncloudsql/upload.go b/pkg/storage/fs/owncloudsql/upload.go index 8b44f5094a3..358d35e45c0 100644 --- a/pkg/storage/fs/owncloudsql/upload.go +++ b/pkg/storage/fs/owncloudsql/upload.go @@ -20,155 +20,12 @@ package owncloudsql import ( "context" - "encoding/json" - "fmt" - "io" - iofs "io/fs" - "os" - "path/filepath" - "strconv" - "time" - userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" - "github.com/google/uuid" - "github.com/owncloud/reva/v2/pkg/appctx" - "github.com/owncloud/reva/v2/pkg/conversions" - ctxpkg "github.com/owncloud/reva/v2/pkg/ctx" "github.com/owncloud/reva/v2/pkg/errtypes" - "github.com/owncloud/reva/v2/pkg/logger" - "github.com/owncloud/reva/v2/pkg/mime" "github.com/owncloud/reva/v2/pkg/storage" - "github.com/owncloud/reva/v2/pkg/storage/utils/chunking" - "github.com/owncloud/reva/v2/pkg/storage/utils/templates" - "github.com/owncloud/reva/v2/pkg/utils" - "github.com/pkg/errors" - "github.com/rs/zerolog/log" - tusd "github.com/tus/tusd/v2/pkg/handler" ) -var defaultFilePerm = os.FileMode(0664) - -func (fs *owncloudsqlfs) Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - upload, err := fs.GetUpload(ctx, req.Ref.GetPath()) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "owncloudsql: error retrieving upload") - } - - uploadInfo := upload.(*fileUpload) - - p := uploadInfo.info.Storage["InternalDestination"] - if chunking.IsChunked(p) { - var assembledFile string - p, assembledFile, err = fs.chunkHandler.WriteChunk(p, req.Body) - if err != nil { - return &provider.ResourceInfo{}, err - } - if p == "" { - if err = uploadInfo.Terminate(ctx); err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "owncloudsql: error removing auxiliary files") - } - return &provider.ResourceInfo{}, errtypes.PartialContent(req.Ref.String()) - } - uploadInfo.info.Storage["InternalDestination"] = p - fd, err := os.Open(assembledFile) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "owncloudsql: error opening assembled file") - } - defer fd.Close() - defer os.RemoveAll(assembledFile) - req.Body = fd - } - - if _, err := uploadInfo.WriteChunk(ctx, 0, req.Body); err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "owncloudsql: error writing to binary file") - } - - if err := uploadInfo.FinishUpload(ctx); err != nil { - return &provider.ResourceInfo{}, err - } - - if uff != nil { - info := uploadInfo.info - uploadRef := &provider.Reference{ - ResourceId: &provider.ResourceId{ - StorageId: info.MetaData["providerID"], - SpaceId: info.Storage["SpaceRoot"], - OpaqueId: info.Storage["SpaceRoot"], - }, - Path: utils.MakeRelativePath(filepath.Join(info.MetaData["dir"], info.MetaData["filename"])), - } - owner, ok := ctxpkg.ContextGetUser(uploadInfo.ctx) - if !ok { - return &provider.ResourceInfo{}, errtypes.PreconditionFailed("error getting user from uploadinfo context") - } - // spaces support in localfs needs to be revisited: - // * info.Storage["SpaceRoot"] is never set - // * there is no space owner or manager that could be passed to the UploadFinishedFunc - uff(owner.Id, owner.Id, uploadRef) - } - - ri := &provider.ResourceInfo{ - // fill with at least fileid, mtime and etag - Id: &provider.ResourceId{ - StorageId: uploadInfo.info.MetaData["providerID"], - SpaceId: uploadInfo.info.Storage["StorageId"], - OpaqueId: uploadInfo.info.Storage["fileid"], - }, - Etag: uploadInfo.info.MetaData["etag"], - } - - if mtime, err := utils.MTimeToTS(uploadInfo.info.MetaData["mtime"]); err == nil { - ri.Mtime = &mtime - } - - return ri, nil -} - -// InitiateUpload returns upload ids corresponding to different protocols it supports -// TODO read optional content for small files in this request -func (fs *owncloudsqlfs) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - ip, err := fs.resolve(ctx, ref) - if err != nil { - return nil, errors.Wrap(err, "owncloudsql: error resolving reference") - } - - // permissions are checked in NewUpload below - - p := fs.toStoragePath(ctx, ip) - - info := tusd.FileInfo{ - MetaData: tusd.MetaData{ - "filename": filepath.Base(p), - "dir": filepath.Dir(p), - "mtime": strconv.FormatInt(time.Now().Unix(), 10), - }, - Size: uploadLength, - } - - if metadata != nil { - info.MetaData["providerID"] = metadata["providerID"] - if metadata["mtime"] != "" { - info.MetaData["mtime"] = metadata["mtime"] - } - if _, ok := metadata["sizedeferred"]; ok { - info.SizeIsDeferred = true - } - } - - upload, err := fs.NewUpload(ctx, info) - if err != nil { - return nil, err - } - - info, _ = upload.GetInfo(ctx) - - return map[string]string{ - "simple": info.ID, - "tus": info.ID, - }, nil -} - func (fs *owncloudsqlfs) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error { return errtypes.NotSupported("op not supported") } @@ -184,383 +41,3 @@ func (fs *owncloudsqlfs) PrepareUpload(_ context.Context, _ *provider.Reference, func (fs *owncloudsqlfs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error { return nil } - -// UseIn tells the tus upload middleware which extensions it supports. -func (fs *owncloudsqlfs) UseIn(composer *tusd.StoreComposer) { - composer.UseCore(fs) - composer.UseTerminater(fs) - composer.UseConcater(fs) - composer.UseLengthDeferrer(fs) -} - -// To implement the core tus.io protocol as specified in https://tus.io/protocols/resumable-upload.html#core-protocol -// - the storage needs to implement NewUpload and GetUpload -// - the upload needs to implement the tusd.Upload interface: WriteChunk, GetInfo, GetReader and FinishUpload - -func (fs *owncloudsqlfs) NewUpload(ctx context.Context, info tusd.FileInfo) (upload tusd.Upload, err error) { - - log := appctx.GetLogger(ctx) - log.Debug().Interface("info", info).Msg("owncloudsql: NewUpload") - - if info.MetaData["filename"] == "" { - return nil, errors.New("owncloudsql: missing filename in metadata") - } - info.MetaData["filename"] = filepath.Clean(info.MetaData["filename"]) - - dir := info.MetaData["dir"] - if dir == "" { - return nil, errors.New("owncloudsql: missing dir in metadata") - } - info.MetaData["dir"] = filepath.Clean(info.MetaData["dir"]) - - ip := fs.toInternalPath(ctx, filepath.Join(info.MetaData["dir"], info.MetaData["filename"])) - - // check permissions - var perm *provider.ResourcePermissions - var perr error - var fsInfo iofs.FileInfo - // if destination exists - if fsInfo, err = os.Stat(ip); err == nil { - // check permissions of file to be overwritten - perm, perr = fs.readPermissions(ctx, ip) - } else { - // check permissions of parent folder - perm, perr = fs.readPermissions(ctx, filepath.Dir(ip)) - } - if perr == nil { - if !perm.InitiateFileUpload { - return nil, errtypes.PermissionDenied("") - } - } else { - if os.IsNotExist(err) { - return nil, errtypes.NotFound(fs.toStoragePath(ctx, filepath.Dir(ip))) - } - return nil, errors.Wrap(err, "owncloudsql: error reading permissions") - } - - // if we are trying to overwriting a folder with a file - if fsInfo != nil && fsInfo.IsDir() { - return nil, errtypes.PreconditionFailed("resource is not a file") - } - - log.Debug().Interface("info", info).Msg("owncloudsql: resolved filename") - - info.ID = uuid.New().String() - - binPath, err := fs.getUploadPath(ctx, info.ID) - if err != nil { - return nil, errors.Wrap(err, "owncloudsql: error resolving upload path") - } - usr := ctxpkg.ContextMustGetUser(ctx) - storageID, err := fs.getStorage(ctx, ip) - if err != nil { - return nil, err - } - info.Storage = map[string]string{ - "Type": "OwnCloudStore", - "BinPath": binPath, - "InternalDestination": ip, - "Permissions": strconv.Itoa((int)(conversions.RoleFromResourcePermissions(perm, false).OCSPermissions())), - - "Idp": usr.Id.Idp, - "UserId": usr.Id.OpaqueId, - "UserName": usr.Username, - - "LogLevel": log.GetLevel().String(), - - "StorageId": strconv.Itoa(storageID), - } - // Create binary file in the upload folder with no content - log.Debug().Interface("info", info).Msg("owncloudsql: built storage info") - file, err := os.OpenFile(binPath, os.O_CREATE|os.O_WRONLY, defaultFilePerm) - if err != nil { - return nil, err - } - defer file.Close() - - u := &fileUpload{ - info: info, - binPath: binPath, - infoPath: filepath.Join(fs.c.UploadInfoDir, info.ID+".info"), - fs: fs, - ctx: ctx, - } - - // writeInfo creates the file by itself if necessary - err = u.writeInfo() - if err != nil { - return nil, err - } - - return u, nil -} - -func (fs *owncloudsqlfs) getUploadPath(ctx context.Context, uploadID string) (string, error) { - u, ok := ctxpkg.ContextGetUser(ctx) - if !ok { - err := errors.Wrap(errtypes.UserRequired("userrequired"), "error getting user from ctx") - return "", err - } - layout := templates.WithUser(u, fs.c.UserLayout) - return filepath.Join(fs.c.DataDirectory, layout, "uploads", uploadID), nil -} - -// GetUpload returns the Upload for the given upload id -func (fs *owncloudsqlfs) GetUpload(ctx context.Context, id string) (tusd.Upload, error) { - infoPath := filepath.Join(fs.c.UploadInfoDir, id+".info") - - info := tusd.FileInfo{} - data, err := os.ReadFile(infoPath) - if err != nil { - if os.IsNotExist(err) { - // Interpret os.ErrNotExist as 404 Not Found - err = tusd.ErrNotFound - } - return nil, err - } - if err := json.Unmarshal(data, &info); err != nil { - return nil, err - } - - stat, err := os.Stat(info.Storage["BinPath"]) - if err != nil { - return nil, err - } - - info.Offset = stat.Size() - - u := &userpb.User{ - Id: &userpb.UserId{ - Idp: info.Storage["Idp"], - OpaqueId: info.Storage["UserId"], - }, - Username: info.Storage["UserName"], - } - - ctx = ctxpkg.ContextSetUser(ctx, u) - // TODO configure the logger the same way ... store and add traceid in file info - - var opts []logger.Option - opts = append(opts, logger.WithLevel(info.Storage["LogLevel"])) - opts = append(opts, logger.WithWriter(os.Stderr, logger.ConsoleMode)) - l := logger.New(opts...) - - sub := l.With().Int("pid", os.Getpid()).Logger() - - ctx = appctx.WithLogger(ctx, &sub) - - return &fileUpload{ - info: info, - binPath: info.Storage["BinPath"], - infoPath: infoPath, - fs: fs, - ctx: ctx, - }, nil -} - -type fileUpload struct { - // info stores the current information about the upload - info tusd.FileInfo - // infoPath is the path to the .info file - infoPath string - // binPath is the path to the binary file (which has no extension) - binPath string - // only fs knows how to handle metadata and versions - fs *owncloudsqlfs - // a context with a user - // TODO add logger as well? - ctx context.Context -} - -// GetInfo returns the FileInfo -func (upload *fileUpload) GetInfo(ctx context.Context) (tusd.FileInfo, error) { - return upload.info, nil -} - -// WriteChunk writes the stream from the reader to the given offset of the upload -func (upload *fileUpload) WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error) { - file, err := os.OpenFile(upload.binPath, os.O_WRONLY|os.O_APPEND, defaultFilePerm) - if err != nil { - return 0, err - } - defer file.Close() - - n, err := io.Copy(file, src) - - // If the HTTP PATCH request gets interrupted in the middle (e.g. because - // the user wants to pause the upload), Go's net/http returns an io.ErrUnexpectedEOF. - // However, for OwnCloudStore it's not important whether the stream has ended - // on purpose or accidentally. - if err != nil { - if err != io.ErrUnexpectedEOF { - return n, err - } - } - - upload.info.Offset += n - err = upload.writeInfo() // TODO info is written here ... we need to truncate in DiscardChunk - - return n, err -} - -// GetReader returns an io.Reader for the upload -func (upload *fileUpload) GetReader(ctx context.Context) (io.ReadCloser, error) { - return os.Open(upload.binPath) -} - -// writeInfo updates the entire information. Everything will be overwritten. -func (upload *fileUpload) writeInfo() error { - log.Debug().Str("path", upload.infoPath).Msg("Writing info file") - data, err := json.Marshal(upload.info) - if err != nil { - return err - } - return os.WriteFile(upload.infoPath, data, defaultFilePerm) -} - -// FinishUpload finishes an upload and moves the file to the internal destination -func (upload *fileUpload) FinishUpload(ctx context.Context) error { - - ip := upload.info.Storage["InternalDestination"] - - // if destination exists - // TODO check etag with If-Match header - if _, err := os.Stat(ip); err == nil { - // create revision - if err := upload.fs.archiveRevision(upload.ctx, upload.fs.getVersionsPath(upload.ctx, ip), ip); err != nil { - return err - } - } - - sha1h, md5h, adler32h, err := upload.fs.HashFile(upload.binPath) - if err != nil { - log.Err(err).Msg("owncloudsql: could not open file for checksumming") - } - - err = os.Rename(upload.binPath, ip) - if err != nil { - log.Err(err).Interface("info", upload.info). - Str("binPath", upload.binPath). - Str("ipath", ip). - Msg("owncloudsql: could not rename") - return err - } - - var fi os.FileInfo - fi, err = os.Stat(ip) - if err != nil { - return err - } - - perms, err := strconv.Atoi(upload.info.Storage["Permissions"]) - if err != nil { - return err - } - - if upload.info.MetaData["mtime"] == "" { - upload.info.MetaData["mtime"] = fmt.Sprintf("%d", fi.ModTime().Unix()) - } - if upload.info.MetaData["etag"] == "" { - upload.info.MetaData["etag"] = calcEtag(upload.ctx, fi) - } - - data := map[string]interface{}{ - "path": upload.fs.toDatabasePath(ip), - "checksum": fmt.Sprintf("SHA1:%032x MD5:%032x ADLER32:%032x", sha1h, md5h, adler32h), - "etag": upload.info.MetaData["etag"], - "size": upload.info.Size, - "mimetype": mime.Detect(false, ip), - "permissions": perms, - "mtime": upload.info.MetaData["mtime"], - "storage_mtime": upload.info.MetaData["mtime"], - } - var fileid int - fileid, err = upload.fs.filecache.InsertOrUpdate(ctx, upload.info.Storage["StorageId"], data, false) - if err != nil { - return err - } - upload.info.Storage["fileid"] = fmt.Sprintf("%d", fileid) - - // only delete the upload if it was successfully written to the storage - if err := os.Remove(upload.infoPath); err != nil { - if !os.IsNotExist(err) { - log.Err(err).Interface("info", upload.info).Msg("owncloudsql: could not delete upload info") - return err - } - } - - return upload.fs.propagate(upload.ctx, ip) -} - -// To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination -// - the storage needs to implement AsTerminatableUpload -// - the upload needs to implement Terminate - -// AsTerminatableUpload returns a TerminatableUpload -func (fs *owncloudsqlfs) AsTerminatableUpload(upload tusd.Upload) tusd.TerminatableUpload { - return upload.(*fileUpload) -} - -// Terminate terminates the upload -func (upload *fileUpload) Terminate(ctx context.Context) error { - if err := os.Remove(upload.infoPath); err != nil { - if !os.IsNotExist(err) { - return err - } - } - if err := os.Remove(upload.binPath); err != nil { - if !os.IsNotExist(err) { - return err - } - } - return nil -} - -// To implement the creation-defer-length extension as specified in https://tus.io/protocols/resumable-upload.html#creation -// - the storage needs to implement AsLengthDeclarableUpload -// - the upload needs to implement DeclareLength - -// AsLengthDeclarableUpload returns a LengthDeclarableUpload -func (fs *owncloudsqlfs) AsLengthDeclarableUpload(upload tusd.Upload) tusd.LengthDeclarableUpload { - return upload.(*fileUpload) -} - -// DeclareLength updates the upload length information -func (upload *fileUpload) DeclareLength(ctx context.Context, length int64) error { - upload.info.Size = length - upload.info.SizeIsDeferred = false - return upload.writeInfo() -} - -// To implement the concatenation extension as specified in https://tus.io/protocols/resumable-upload.html#concatenation -// - the storage needs to implement AsConcatableUpload -// - the upload needs to implement ConcatUploads - -// AsConcatableUpload returns a ConcatableUpload -func (fs *owncloudsqlfs) AsConcatableUpload(upload tusd.Upload) tusd.ConcatableUpload { - return upload.(*fileUpload) -} - -// ConcatUploads concatenates multiple uploads -func (upload *fileUpload) ConcatUploads(ctx context.Context, uploads []tusd.Upload) (err error) { - file, err := os.OpenFile(upload.binPath, os.O_WRONLY|os.O_APPEND, defaultFilePerm) - if err != nil { - return err - } - defer file.Close() - - for _, partialUpload := range uploads { - fileUpload := partialUpload.(*fileUpload) - - src, err := os.Open(fileUpload.binPath) - if err != nil { - return err - } - - if _, err := io.Copy(file, src); err != nil { - return err - } - } - - return -} diff --git a/pkg/storage/fs/posix/posix.go b/pkg/storage/fs/posix/posix.go index ecb68fe2cd2..406f17145d8 100644 --- a/pkg/storage/fs/posix/posix.go +++ b/pkg/storage/fs/posix/posix.go @@ -28,7 +28,6 @@ import ( "syscall" "github.com/rs/zerolog" - tusd "github.com/tus/tusd/v2/pkg/handler" microstore "go-micro.dev/v4/store" "github.com/owncloud/reva/v2/pkg/events" @@ -46,7 +45,6 @@ import ( "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/metadata" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/node" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/permissions" - "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/upload" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/usermapper" "github.com/owncloud/reva/v2/pkg/storage/utils/middleware" "github.com/owncloud/reva/v2/pkg/store" @@ -184,39 +182,3 @@ func New(m map[string]interface{}, stream events.Stream, log *zerolog.Logger) (s func (fs *posixFS) ListUploadSessions(ctx context.Context, filter storage.UploadSessionFilter) ([]storage.UploadSession, error) { return fs.FS.(storage.UploadSessionLister).ListUploadSessions(ctx, filter) } - -// UseIn tells the tus upload middleware which extensions it supports. -func (fs *posixFS) UseIn(composer *tusd.StoreComposer) { - fs.FS.(storage.ComposableFS).UseIn(composer) -} - -// NewUpload returns a new tus Upload instance -func (fs *posixFS) NewUpload(ctx context.Context, info tusd.FileInfo) (upload tusd.Upload, err error) { - return fs.FS.(tusd.DataStore).NewUpload(ctx, info) -} - -// NewUpload returns a new tus Upload instance -func (fs *posixFS) GetUpload(ctx context.Context, id string) (upload tusd.Upload, err error) { - return fs.FS.(tusd.DataStore).GetUpload(ctx, id) -} - -// AsTerminatableUpload returns a TerminatableUpload -// To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination -// the storage needs to implement AsTerminatableUpload -func (fs *posixFS) AsTerminatableUpload(up tusd.Upload) tusd.TerminatableUpload { - return up.(*upload.OcisSession) -} - -// AsLengthDeclarableUpload returns a LengthDeclarableUpload -// To implement the creation-defer-length extension as specified in https://tus.io/protocols/resumable-upload.html#creation -// the storage needs to implement AsLengthDeclarableUpload -func (fs *posixFS) AsLengthDeclarableUpload(up tusd.Upload) tusd.LengthDeclarableUpload { - return up.(*upload.OcisSession) -} - -// AsConcatableUpload returns a ConcatableUpload -// To implement the concatenation extension as specified in https://tus.io/protocols/resumable-upload.html#concatenation -// the storage needs to implement AsConcatableUpload -func (fs *posixFS) AsConcatableUpload(up tusd.Upload) tusd.ConcatableUpload { - return up.(*upload.OcisSession) -} diff --git a/pkg/storage/fs/s3/upload.go b/pkg/storage/fs/s3/upload.go index 30ed4ddb987..c5747775181 100644 --- a/pkg/storage/fs/s3/upload.go +++ b/pkg/storage/fs/s3/upload.go @@ -21,59 +21,11 @@ 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/owncloud/reva/v2/pkg/appctx" "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/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") -} - func (fs *s3FS) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error { return errtypes.NotSupported("op not supported") } diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 0716672f79f..2293c667f1f 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -28,7 +28,6 @@ import ( userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" registry "github.com/cs3org/go-cs3apis/cs3/storage/registry/v1beta1" - tusd "github.com/tus/tusd/v2/pkg/handler" ) type MoveResult struct { @@ -144,10 +143,6 @@ type FS interface { Delete(ctx context.Context, ref *provider.Reference) (*DeleteResult, error) // Move changes the path of a resource Move(ctx context.Context, oldRef, newRef *provider.Reference) (*MoveResult, error) - // InitiateUpload returns a list of protocols with urls that can be used to append bytes to a new upload session - InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) - // Upload creates or updates a resource of type file with a new revision - Upload(ctx context.Context, req UploadRequest, uploadFunc UploadFinishedFunc) (*provider.ResourceInfo, error) // MarkProcessing toggles a processing flag on the resource. MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error // CommitUpload writes the staged bytes from source to the resource at ref. @@ -271,12 +266,6 @@ type UploadSource struct { // UnscopeFunc is a function that unscopes a user type UnscopeFunc func() -// Composable is the interface that a struct needs to implement -// to be composable, so that it can support the TUS methods -type ComposableFS interface { - UseIn(composer *tusd.StoreComposer) -} - // Registry is the interface that storage registries implement // for discovering storage providers type Registry interface { diff --git a/pkg/storage/uploads.go b/pkg/storage/uploads.go index 44b70274534..6c37ced988b 100644 --- a/pkg/storage/uploads.go +++ b/pkg/storage/uploads.go @@ -20,31 +20,12 @@ package storage import ( "context" - "io" "time" userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" - tusd "github.com/tus/tusd/v2/pkg/handler" ) -// UploadFinishedFunc is a callback function used in storage drivers to indicate that an upload has finished -type UploadFinishedFunc func(spaceOwner, executant *userpb.UserId, ref *provider.Reference) - -// UploadRequest us used in FS.Upload() to carry required upload metadata -type UploadRequest struct { - Ref *provider.Reference - Body io.ReadCloser - Length int64 -} - -// UploadsManager defines the interface for storage drivers that allow for managing uploads -// Deprecated: No longer used. Storage drivers should implement the UploadSessionLister. -type UploadsManager interface { - ListUploads() ([]tusd.FileInfo, error) - PurgeExpiredUploads(chan<- tusd.FileInfo) error -} - // UploadSessionLister defines the interface for FS implementations that allow listing and purging upload sessions type UploadSessionLister interface { // ListUploadSessions returns the upload sessions matching the given filter diff --git a/pkg/storage/utils/decomposedfs/upload.go b/pkg/storage/utils/decomposedfs/upload.go index 7ae2034e908..0354f0df27c 100644 --- a/pkg/storage/utils/decomposedfs/upload.go +++ b/pkg/storage/utils/decomposedfs/upload.go @@ -28,322 +28,18 @@ import ( "time" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" - "github.com/google/uuid" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes" "github.com/pkg/errors" "github.com/rogpeppe/go-internal/lockedfile" - tusd "github.com/tus/tusd/v2/pkg/handler" "github.com/owncloud/reva/v2/pkg/appctx" - ctxpkg "github.com/owncloud/reva/v2/pkg/ctx" "github.com/owncloud/reva/v2/pkg/errtypes" - "github.com/owncloud/reva/v2/pkg/rhttp/datatx/metrics" "github.com/owncloud/reva/v2/pkg/storage" - "github.com/owncloud/reva/v2/pkg/storage/utils/chunking" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/node" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/upload" - "github.com/owncloud/reva/v2/pkg/storagespace" "github.com/owncloud/reva/v2/pkg/utils" ) -// Upload uploads data to the given resource -// TODO(OCISDEV-901): remove Upload once all drivers are migrated to CommitUpload and the coordinator (OCISDEV-900) is in place. -func (fs *Decomposedfs) Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - _, span := tracer.Start(ctx, "Upload") - defer span.End() - up, err := fs.GetUpload(ctx, req.Ref.GetPath()) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error retrieving upload") - } - - session := up.(*upload.OcisSession) - - ctx = session.Context(ctx) - - if session.Chunk() != "" { // check chunking v1 - p, assembledFile, err := fs.chunkHandler.WriteChunk(session.Chunk(), req.Body) - if err != nil { - return &provider.ResourceInfo{}, err - } - if p == "" { - if err = session.Terminate(ctx); err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error removing auxiliary files") - } - return &provider.ResourceInfo{}, errtypes.PartialContent(req.Ref.String()) - } - fd, err := os.Open(assembledFile) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error opening assembled file") - } - defer fd.Close() - defer os.RemoveAll(assembledFile) - req.Body = fd - - size, err := session.WriteChunk(ctx, 0, req.Body) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error writing to binary file") - } - session.SetSize(size) - } else { - size, err := session.WriteChunk(ctx, 0, req.Body) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error writing to binary file") - } - if size != req.Length { - return &provider.ResourceInfo{}, errtypes.PartialContent("Decomposedfs: unexpected end of stream") - } - } - - if err := session.FinishUploadDecomposed(ctx); err != nil { - return &provider.ResourceInfo{}, err - } - - if uff != nil { - uploadRef := &provider.Reference{ - ResourceId: &provider.ResourceId{ - StorageId: session.ProviderID(), - SpaceId: session.SpaceID(), - OpaqueId: session.SpaceID(), - }, - Path: utils.MakeRelativePath(filepath.Join(session.Dir(), session.Filename())), - } - executant := session.Executant() - uff(session.SpaceOwner(), &executant, uploadRef) - } - - ri := &provider.ResourceInfo{ - // fill with at least fileid, mtime and etag - Id: &provider.ResourceId{ - StorageId: session.ProviderID(), - SpaceId: session.SpaceID(), - OpaqueId: session.NodeID(), - }, - } - - // add etag to metadata - ri.Etag, _ = node.CalculateEtag(session.NodeID(), session.MTime()) - - if !session.MTime().IsZero() { - ri.Mtime = utils.TimeToTS(session.MTime()) - } - - return ri, nil -} - -// InitiateUpload returns upload ids corresponding to different protocols it supports -// TODO(OCISDEV-901): remove InitiateUpload once all drivers are migrated to CommitUpload and the coordinator (OCISDEV-900) is in place. -func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - _, span := tracer.Start(ctx, "InitiateUpload") - defer span.End() - log := appctx.GetLogger(ctx) - log.Debug().Interface("ref", ref).Msg("decomposedfs:InitiateUpload:start") - - // remember the path from the reference - refpath := ref.GetPath() - var chunk *chunking.ChunkBLOBInfo - var err error - if chunking.IsChunked(refpath) { // check chunking v1 - chunk, err = chunking.GetChunkBLOBInfo(refpath) - if err != nil { - return nil, errtypes.BadRequest(err.Error()) - } - ref.Path = chunk.Path - } - n, err := fs.lu.NodeFromResource(ctx, ref) - switch err.(type) { - case nil: - // ok - case errtypes.IsNotFound: - return nil, errtypes.PreconditionFailed(err.Error()) - default: - return nil, err - } - - // permissions are checked in NewUpload below - - relative, err := fs.lu.Path(ctx, n, node.NoCheck) - // TODO why do we need the path here? - // jfd: it is used later when emitting the UploadReady event ... - // AAAND refPath might be . when accessing with an id / relative reference ... which causes NodeName to become . But then dir will also always be . - // That is why we still have to read the path here: so that the event we emit contains a relative reference with a path relative to the space root. WTF - if err != nil { - return nil, err - } - - lockID, _ := ctxpkg.ContextGetLockID(ctx) - - session := fs.sessionStore.New(ctx) - session.SetMetadata("filename", n.Name) - session.SetStorageValue("NodeName", n.Name) - if chunk != nil { - session.SetStorageValue("Chunk", filepath.Base(refpath)) - } - session.SetMetadata("dir", filepath.Dir(relative)) - session.SetStorageValue("Dir", filepath.Dir(relative)) - session.SetMetadata("lockid", lockID) - - session.SetSize(uploadLength) - session.SetStorageValue("SpaceRoot", n.SpaceRoot.ID) // TODO SpaceRoot -> SpaceID - session.SetStorageValue("SpaceOwnerOrManager", n.SpaceOwnerOrManager(ctx).GetOpaqueId()) // TODO needed for what? - - spaceGID, ok := ctx.Value(CtxKeySpaceGID).(uint32) - if ok { - session.SetStorageValue("SpaceGid", fmt.Sprintf("%d", spaceGID)) - } - - iid, _ := ctxpkg.ContextGetInitiator(ctx) - session.SetMetadata("initiatorid", iid) - - if metadata != nil { - session.SetMetadata("providerID", metadata["providerID"]) - if mtime, ok := metadata["mtime"]; ok { - if mtime != "null" { - session.SetMetadata("mtime", metadata["mtime"]) - } - } - if expiration, ok := metadata["expires"]; ok { - if expiration != "null" { - session.SetMetadata("expires", metadata["expires"]) - } - } - if _, ok := metadata["sizedeferred"]; ok { - session.SetSizeIsDeferred(true) - } - if checksum, ok := metadata["checksum"]; ok { - parts := strings.SplitN(checksum, " ", 2) - if len(parts) != 2 { - return nil, errtypes.BadRequest("invalid checksum format. must be '[algorithm] [checksum]'") - } - switch parts[0] { - case "sha1", "md5", "adler32": - session.SetMetadata("checksum", checksum) - default: - return nil, errtypes.BadRequest("unsupported checksum algorithm: " + parts[0]) - } - } - - // only check preconditions if they are not empty // TODO or is this a bad request? - if metadata["if-match"] != "" { - session.SetMetadata("if-match", metadata["if-match"]) - } - if metadata["if-none-match"] != "" { - session.SetMetadata("if-none-match", metadata["if-none-match"]) - } - if metadata["if-unmodified-since"] != "" { - session.SetMetadata("if-unmodified-since", metadata["if-unmodified-since"]) - } - } - - if session.MTime().IsZero() { - session.SetMetadata("mtime", utils.TimeToOCMtime(time.Now())) - } - - log.Debug().Str("uploadid", session.ID()).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Interface("metadata", metadata).Msg("Decomposedfs: resolved filename") - - _, err = node.CheckQuota(ctx, n.SpaceRoot, n.Exists, uint64(n.Blobsize), uint64(session.Size())) - if err != nil { - return nil, err - } - - if session.Filename() == "" { - return nil, errors.New("Decomposedfs: missing filename in metadata") - } - if session.Dir() == "" { - return nil, errors.New("Decomposedfs: missing dir in metadata") - } - - // the parent owner will become the new owner - parent, perr := n.Parent(ctx) - if perr != nil { - return nil, errors.Wrap(perr, "Decomposedfs: error getting parent "+n.ParentID) - } - - // check permissions - var ( - checkNode *node.Node - path string - ) - if n.Exists { - // check permissions of file to be overwritten - checkNode = n - path, _ = storagespace.FormatReference(&provider.Reference{ResourceId: &provider.ResourceId{ - SpaceId: checkNode.SpaceID, - OpaqueId: checkNode.ID, - }}) - } else { - // check permissions of parent - checkNode = parent - path, _ = storagespace.FormatReference(&provider.Reference{ResourceId: &provider.ResourceId{ - SpaceId: checkNode.SpaceID, - OpaqueId: checkNode.ID, - }, Path: n.Name}) - } - rp, err := fs.p.AssemblePermissions(ctx, checkNode) - switch { - case err != nil: - return nil, err - case !rp.InitiateFileUpload: - return nil, errtypes.PermissionDenied(path) - } - - // are we trying to overwriting a folder with a file? - if n.Exists && n.IsDir(ctx) { - return nil, errtypes.PreconditionFailed("resource is not a file") - } - - // check lock - if err := n.CheckLock(ctx); err != nil { - return nil, err - } - - usr := ctxpkg.ContextMustGetUser(ctx) - - // fill future node info - if n.Exists { - if session.HeaderIfNoneMatch() == "*" { - return nil, errtypes.Aborted(fmt.Sprintf("parent %s already has a child %s, id %s", n.ParentID, n.Name, n.ID)) - } - session.SetStorageValue("NodeId", n.ID) - session.SetStorageValue("NodeExists", "true") - } else { - session.SetStorageValue("NodeId", uuid.New().String()) - } - session.SetStorageValue("NodeParentId", n.ParentID) - session.SetExecutant(usr) - session.SetStorageValue("LogLevel", log.GetLevel().String()) - - log.Debug().Interface("session", session).Msg("Decomposedfs: built session info") - - err = fs.um.RunInBaseScope(func() error { - // Create binary file in the upload folder with no content - // It will be used when determining the current offset of an upload - err := session.TouchBin() - if err != nil { - return err - } - - return session.Persist(ctx) - }) - if err != nil { - return nil, err - } - metrics.UploadSessionsInitiated.Inc() - - if uploadLength == 0 { - // Directly finish this upload - err = session.FinishUploadDecomposed(ctx) - if err != nil { - return nil, err - } - } - - log.Debug().Str("uploadid", session.ID()).Msg("decomposedfs:InitiateUpload:complete") - return map[string]string{ - "simple": session.ID(), - "tus": session.ID(), - }, nil -} - // MarkProcessing toggles a processing flag on the resource. func (fs *Decomposedfs) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error { n, err := fs.lu.NodeFromResource(ctx, ref) @@ -752,34 +448,6 @@ func validateChecksums(ctx context.Context, lu node.PathLookup, n *node.Node, ve return nil } -// UseIn tells the tus upload middleware which extensions it supports. -func (fs *Decomposedfs) UseIn(composer *tusd.StoreComposer) { - composer.UseCore(fs) - composer.UseTerminater(fs) - composer.UseConcater(fs) - composer.UseLengthDeferrer(fs) -} - -// To implement the core tus.io protocol as specified in https://tus.io/protocols/resumable-upload.html#core-protocol -// - the storage needs to implement NewUpload and GetUpload -// - the upload needs to implement the tusd.Upload interface: WriteChunk, GetInfo, GetReader and FinishUpload - -// NewUpload returns a new tus Upload instance -func (fs *Decomposedfs) NewUpload(ctx context.Context, info tusd.FileInfo) (tusd.Upload, error) { - return nil, fmt.Errorf("not implemented, use InitiateUpload on the CS3 API to start a new upload") -} - -// GetUpload returns the Upload for the given upload id -func (fs *Decomposedfs) GetUpload(ctx context.Context, id string) (tusd.Upload, error) { - var ul tusd.Upload - var err error - _ = fs.um.RunInBaseScope(func() error { - ul, err = fs.sessionStore.Get(ctx, id) - return nil - }) - return ul, err -} - // ListUploadSessions returns the upload sessions for the given filter func (fs *Decomposedfs) ListUploadSessions(ctx context.Context, filter storage.UploadSessionFilter) ([]storage.UploadSession, error) { var sessions []*upload.OcisSession @@ -829,24 +497,3 @@ func (fs *Decomposedfs) ListUploadSessions(ctx context.Context, filter storage.U } return filteredSessions, nil } - -// AsTerminatableUpload returns a TerminatableUpload -// To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination -// the storage needs to implement AsTerminatableUpload -func (fs *Decomposedfs) AsTerminatableUpload(up tusd.Upload) tusd.TerminatableUpload { - return up.(*upload.OcisSession) -} - -// AsLengthDeclarableUpload returns a LengthDeclarableUpload -// To implement the creation-defer-length extension as specified in https://tus.io/protocols/resumable-upload.html#creation -// the storage needs to implement AsLengthDeclarableUpload -func (fs *Decomposedfs) AsLengthDeclarableUpload(up tusd.Upload) tusd.LengthDeclarableUpload { - return up.(*upload.OcisSession) -} - -// AsConcatableUpload returns a ConcatableUpload -// To implement the concatenation extension as specified in https://tus.io/protocols/resumable-upload.html#concatenation -// the storage needs to implement AsConcatableUpload -func (fs *Decomposedfs) AsConcatableUpload(up tusd.Upload) tusd.ConcatableUpload { - return up.(*upload.OcisSession) -} diff --git a/pkg/storage/utils/decomposedfs/upload/session.go b/pkg/storage/utils/decomposedfs/upload/session.go index 73c0e280347..f7f3b5dac97 100644 --- a/pkg/storage/utils/decomposedfs/upload/session.go +++ b/pkg/storage/utils/decomposedfs/upload/session.go @@ -247,8 +247,8 @@ func (s *OcisSession) SetSizeIsDeferred(value bool) { // postprocessing finished. I wonder why the UploadReady contains a finished // flag ... maybe multiple distinct events would make more sense. // - build the reference that is passed to the FileUploaded event in the -// UploadFinishedFunc callback passed to the Upload call used for simple -// datatx put requests +// upload.FinishedFunc callback passed to the coordinator's Upload call used +// for simple datatx put requests // // AFAICT only search and audit services consume the path. // - search needs to index from the root anyway. And it only needs the most diff --git a/pkg/storage/utils/decomposedfs/upload_test.go b/pkg/storage/utils/decomposedfs/upload_test.go index bd0a3e858ad..9023b6bacd8 100644 --- a/pkg/storage/utils/decomposedfs/upload_test.go +++ b/pkg/storage/utils/decomposedfs/upload_test.go @@ -29,7 +29,6 @@ import ( v1beta11 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" ruser "github.com/owncloud/reva/v2/pkg/ctx" - "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool" "github.com/owncloud/reva/v2/pkg/storage" "github.com/owncloud/reva/v2/pkg/storage/cache" @@ -37,8 +36,6 @@ import ( "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/aspects" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/lookup" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/metadata" - "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes" - "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/node" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/options" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/permissions" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/permissions/mocks" @@ -47,6 +44,7 @@ import ( treemocks "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/tree/mocks" "github.com/owncloud/reva/v2/pkg/storagespace" "github.com/owncloud/reva/v2/pkg/store" + "github.com/owncloud/reva/v2/pkg/upload" "github.com/owncloud/reva/v2/tests/helpers" "github.com/rs/zerolog" "github.com/stretchr/testify/mock" @@ -61,6 +59,7 @@ var _ = Describe("File uploads", func() { ref *provider.Reference rootRef *provider.Reference fs storage.FS + coord upload.Coordinator user *userpb.User ctx context.Context @@ -148,6 +147,13 @@ var _ = Describe("File uploads", func() { fs, err = decomposedfs.New(o, aspects, &zerolog.Logger{}) Expect(err).ToNot(HaveOccurred()) + // The coordinator owns the upload flow; the driver only sees the slim contract. + // No publisher and no chunking: without a postprocessing consumer an upload + // finishes synchronously, so the specs below observe the driver's final state. + log := zerolog.Nop() + coord, err = upload.NewCoordinatorFromConfig(GinkgoT().TempDir(), nil, fs, nil, &log, false) + Expect(err).ToNot(HaveOccurred()) + resp, err := fs.CreateStorageSpace(ctx, &provider.CreateStorageSpaceRequest{Owner: user, Type: "personal"}) Expect(err).ToNot(HaveOccurred()) Expect(resp.Status.Code).To(Equal(v1beta11.Code_CODE_OK)) @@ -156,63 +162,9 @@ var _ = Describe("File uploads", func() { ref.ResourceId = &resID }) - Context("the user's quota is exceeded", func() { - BeforeEach(func() { - pmock.On("AssemblePermissions", mock.Anything, mock.Anything, mock.Anything).Return(&provider.ResourcePermissions{ - Stat: true, - GetQuota: true, - }, nil) - }) - When("the user wants to initiate a file upload", func() { - It("fails", func() { - var originalFunc = node.CheckQuota - node.CheckQuota = func(ctx context.Context, spaceRoot *node.Node, overwrite bool, oldSize, newSize uint64) (quotaSufficient bool, err error) { - return false, errtypes.InsufficientStorage("quota exceeded") - } - _, err := fs.InitiateUpload(ctx, ref, 10, map[string]string{}) - Expect(err).To(MatchError(errtypes.InsufficientStorage("quota exceeded"))) - node.CheckQuota = originalFunc - }) - }) - }) - - Context("the user has insufficient permissions", func() { - BeforeEach(func() { - pmock.On("AssemblePermissions", mock.Anything, mock.Anything, mock.Anything).Return(&provider.ResourcePermissions{ - Stat: true, - }, nil) - }) - - When("the user wants to initiate a file upload", func() { - It("fails", func() { - msg := "error: permission denied: u-s-e-r-id!u-s-e-r-id/foo" - _, err := fs.InitiateUpload(ctx, ref, 10, map[string]string{}) - Expect(err).To(MatchError(msg)) - }) - }) - }) - - Context("with insufficient permissions, home node", func() { - JustBeforeEach(func() { - var err error - // the space name attribute is the stop condition in the lookup - h, err := lu.NodeFromResource(ctx, rootRef) - Expect(err).ToNot(HaveOccurred()) - err = h.SetXattrString(ctx, prefixes.SpaceNameAttr, "username") - Expect(err).ToNot(HaveOccurred()) - pmock.On("AssemblePermissions", mock.Anything, mock.Anything, mock.Anything).Return(&provider.ResourcePermissions{ - Stat: true, - }, nil) - }) - - When("the user wants to initiate a file upload", func() { - It("fails", func() { - msg := "error: permission denied: u-s-e-r-id!u-s-e-r-id/foo" - _, err := fs.InitiateUpload(ctx, ref, 10, map[string]string{}) - Expect(err).To(MatchError(msg)) - }) - }) - }) + // The quota and permission checks now live in the coordinator, which resolves them + // through GetQuota/GetMD rather than node.CheckQuota. They are asserted against the + // coordinator directly in pkg/upload/initiate_test.go. Context("with sufficient permissions", func() { BeforeEach(func() { @@ -227,7 +179,7 @@ var _ = Describe("File uploads", func() { When("the user initiates a non zero byte file upload", func() { It("succeeds", func() { - uploadIds, err := fs.InitiateUpload(ctx, ref, 10, map[string]string{}) + uploadIds, err := coord.InitiateUpload(ctx, ref, 10, map[string]string{}) Expect(err).ToNot(HaveOccurred()) Expect(len(uploadIds)).To(Equal(2)) @@ -242,9 +194,9 @@ var _ = Describe("File uploads", func() { When("the user initiates a zero byte file upload", func() { It("succeeds", func() { - bs.On("Upload", mock.AnythingOfType("*node.Node"), mock.AnythingOfType("string"), mock.Anything). + bs.On("UploadFromReader", mock.AnythingOfType("*node.Node"), mock.Anything, mock.AnythingOfType("int64")). Return(nil) - uploadIds, err := fs.InitiateUpload(ctx, ref, 0, map[string]string{}) + uploadIds, err := coord.InitiateUpload(ctx, ref, 0, map[string]string{}) Expect(err).ToNot(HaveOccurred()) Expect(len(uploadIds)).To(Equal(2)) @@ -257,9 +209,9 @@ var _ = Describe("File uploads", func() { }) It("fails when trying to upload empty data. 0-byte uploads are finished during initialization already", func() { - bs.On("Upload", mock.AnythingOfType("*node.Node"), mock.AnythingOfType("string"), mock.Anything). + bs.On("UploadFromReader", mock.AnythingOfType("*node.Node"), mock.Anything, mock.AnythingOfType("int64")). Return(nil) - uploadIds, err := fs.InitiateUpload(ctx, ref, 0, map[string]string{}) + uploadIds, err := coord.InitiateUpload(ctx, ref, 0, map[string]string{}) Expect(err).ToNot(HaveOccurred()) Expect(len(uploadIds)).To(Equal(2)) @@ -267,7 +219,7 @@ var _ = Describe("File uploads", func() { uploadRef := &provider.Reference{Path: "/" + uploadIds["simple"]} - _, err = fs.Upload(ctx, storage.UploadRequest{ + _, err = coord.Upload(ctx, upload.Request{ Ref: uploadRef, Body: io.NopCloser(bytes.NewReader([]byte(""))), Length: 0, @@ -283,7 +235,7 @@ var _ = Describe("File uploads", func() { fileContent = []byte("0123456789") ) - uploadIds, err := fs.InitiateUpload(ctx, ref, 10, map[string]string{}) + uploadIds, err := coord.InitiateUpload(ctx, ref, 10, map[string]string{}) Expect(err).ToNot(HaveOccurred()) Expect(len(uploadIds)).To(Equal(2)) @@ -292,23 +244,25 @@ var _ = Describe("File uploads", func() { uploadRef := &provider.Reference{Path: "/" + uploadIds["simple"]} - bs.On("Upload", mock.AnythingOfType("*node.Node"), mock.AnythingOfType("string"), mock.Anything). + bs.On("UploadFromReader", mock.AnythingOfType("*node.Node"), mock.Anything, mock.AnythingOfType("int64")). Return(nil). Run(func(args mock.Arguments) { - data, err := os.ReadFile(args.Get(1).(string)) + // CommitUpload streams the staged bytes to the blobstore, + // so assert on the reader's content rather than a path. + data, err := io.ReadAll(args.Get(1).(io.Reader)) Expect(err).ToNot(HaveOccurred()) Expect(data).To(Equal([]byte("0123456789"))) }) - _, err = fs.Upload(ctx, storage.UploadRequest{ + _, err = coord.Upload(ctx, upload.Request{ Ref: uploadRef, Body: io.NopCloser(bytes.NewReader(fileContent)), Length: int64(len(fileContent)), }, nil) Expect(err).ToNot(HaveOccurred()) - bs.AssertCalled(GinkgoT(), "Upload", mock.Anything, mock.Anything, mock.Anything) + bs.AssertCalled(GinkgoT(), "UploadFromReader", mock.Anything, mock.Anything, mock.Anything) resources, err := fs.ListFolder(ctx, rootRef, []string{}, []string{}) @@ -325,7 +279,7 @@ var _ = Describe("File uploads", func() { ) uploadRef := &provider.Reference{Path: "/some-non-existent-upload-reference"} - _, err := fs.Upload(ctx, storage.UploadRequest{ + _, err := coord.Upload(ctx, upload.Request{ Ref: uploadRef, Body: io.NopCloser(bytes.NewReader(fileContent)), Length: int64(len(fileContent)), diff --git a/pkg/storage/utils/eosfs/upload.go b/pkg/storage/utils/eosfs/upload.go index 0b72b6f8e58..e45a600cd28 100644 --- a/pkg/storage/utils/eosfs/upload.go +++ b/pkg/storage/utils/eosfs/upload.go @@ -20,85 +20,12 @@ package eosfs import ( "context" - "os" - "path" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/storage" - "github.com/owncloud/reva/v2/pkg/storage/utils/chunking" - "github.com/pkg/errors" ) -func (fs *eosfs) Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - p, err := fs.resolve(ctx, req.Ref) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "eos: error resolving reference") - } - - if fs.isShareFolder(ctx, p) { - return &provider.ResourceInfo{}, errtypes.PermissionDenied("eos: cannot upload under the virtual share folder") - } - - if chunking.IsChunked(p) { - var assembledFile string - p, assembledFile, err = fs.chunkHandler.WriteChunk(p, req.Body) - if err != nil { - return &provider.ResourceInfo{}, err - } - if p == "" { - return &provider.ResourceInfo{}, errtypes.PartialContent(req.Ref.String()) - } - fd, err := os.Open(assembledFile) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "eos: error opening assembled file") - } - defer fd.Close() - defer os.RemoveAll(assembledFile) - req.Body = fd - } - - fn := fs.wrap(ctx, p) - - u, err := getUser(ctx) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "eos: no user in ctx") - } - - // We need the auth corresponding to the parent directory - // as the file might not exist at the moment - auth, err := fs.getUserAuth(ctx, u, path.Dir(fn)) - if err != nil { - return &provider.ResourceInfo{}, err - } - - if err := fs.c.Write(ctx, auth, fn, req.Body); err != nil { - return &provider.ResourceInfo{}, err - } - - eosFileInfo, err := fs.c.GetFileInfoByPath(ctx, auth, fn) - if err != nil { - return &provider.ResourceInfo{}, err - } - - ri, err := fs.convertToResourceInfo(ctx, eosFileInfo) - if err != nil { - return &provider.ResourceInfo{}, err - } - - return ri, nil -} - -func (fs *eosfs) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - p, err := fs.resolve(ctx, ref) - if err != nil { - return nil, err - } - return map[string]string{ - "simple": p, - }, nil -} - func (fs *eosfs) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error { return errtypes.NotSupported("op not supported") } diff --git a/pkg/storage/utils/localfs/upload.go b/pkg/storage/utils/localfs/upload.go index 7161996c714..e8619c05a44 100644 --- a/pkg/storage/utils/localfs/upload.go +++ b/pkg/storage/utils/localfs/upload.go @@ -20,137 +20,12 @@ package localfs import ( "context" - "encoding/json" - "io" - "os" - "path/filepath" - userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" - "github.com/google/uuid" - "github.com/owncloud/reva/v2/pkg/appctx" - ctxpkg "github.com/owncloud/reva/v2/pkg/ctx" "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/storage" - "github.com/owncloud/reva/v2/pkg/storage/utils/chunking" - "github.com/owncloud/reva/v2/pkg/utils" - "github.com/pkg/errors" - tusd "github.com/tus/tusd/v2/pkg/handler" ) -var defaultFilePerm = os.FileMode(0664) - -func (fs *localfs) Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - upload, err := fs.GetUpload(ctx, req.Ref.GetPath()) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "localfs: error retrieving upload") - } - - uploadInfo := upload.(*fileUpload) - - p := uploadInfo.info.Storage["InternalDestination"] - if chunking.IsChunked(p) { - var assembledFile string - p, assembledFile, err = fs.chunkHandler.WriteChunk(p, req.Body) - if err != nil { - return &provider.ResourceInfo{}, err - } - if p == "" { - if err = uploadInfo.Terminate(ctx); err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "localfs: error removing auxiliary files") - } - return &provider.ResourceInfo{}, errtypes.PartialContent(req.Ref.String()) - } - uploadInfo.info.Storage["InternalDestination"] = p - fd, err := os.Open(assembledFile) - if err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "localfs: error opening assembled file") - } - defer fd.Close() - defer os.RemoveAll(assembledFile) - req.Body = fd - } - - if _, err := uploadInfo.WriteChunk(ctx, 0, req.Body); err != nil { - return &provider.ResourceInfo{}, errors.Wrap(err, "localfs: error writing to binary file") - } - - if err := uploadInfo.FinishUpload(ctx); err != nil { - return &provider.ResourceInfo{}, err - } - - if uff != nil { - info := uploadInfo.info - uploadRef := &provider.Reference{ - ResourceId: &provider.ResourceId{ - StorageId: info.MetaData["providerID"], - SpaceId: info.Storage["SpaceRoot"], - OpaqueId: info.Storage["SpaceRoot"], - }, - Path: utils.MakeRelativePath(filepath.Join(info.MetaData["dir"], info.MetaData["filename"])), - } - owner, ok := ctxpkg.ContextGetUser(uploadInfo.ctx) - if !ok { - return &provider.ResourceInfo{}, errtypes.PreconditionFailed("error getting user from uploadinfo context") - } - // spaces support in localfs needs to be revisited: - // * info.Storage["SpaceRoot"] is never set - // * there is no space owner or manager that could be passed to the UploadFinishedFunc - uff(owner.Id, owner.Id, uploadRef) - } - - // 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 -// It resolves the resource and then reuses the NewUpload function -// Currently requires the uploadLength to be set -// TODO to implement LengthDeferrerDataStore make size optional -// TODO read optional content for small files in this request -func (fs *localfs) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - - np, err := fs.resolve(ctx, ref) - if err != nil { - return nil, errors.Wrap(err, "localfs: error resolving reference") - } - - info := tusd.FileInfo{ - MetaData: tusd.MetaData{ - "filename": filepath.Base(np), - "dir": filepath.Dir(np), - }, - Size: uploadLength, - } - - if metadata != nil { - info.MetaData["providerID"] = metadata["providerID"] - if metadata["mtime"] != "" { - info.MetaData["mtime"] = metadata["mtime"] - } - if _, ok := metadata["sizedeferred"]; ok { - info.SizeIsDeferred = true - } - } - - upload, err := fs.NewUpload(ctx, info) - if err != nil { - return nil, err - } - - info, _ = upload.GetInfo(ctx) - - return map[string]string{ - "simple": info.ID, - "tus": info.ID, - }, nil -} - func (fs *localfs) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error { return errtypes.NotSupported("op not supported") } @@ -166,247 +41,3 @@ func (fs *localfs) PrepareUpload(_ context.Context, _ *provider.Reference, _ str func (fs *localfs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error { return nil } - -// UseIn tells the tus upload middleware which extensions it supports. -func (fs *localfs) UseIn(composer *tusd.StoreComposer) { - composer.UseCore(fs) - composer.UseTerminater(fs) - // TODO composer.UseConcater(fs) - // TODO composer.UseLengthDeferrer(fs) -} - -// NewUpload creates a new upload using the size as the file's length. To determine where to write the binary data -// the Fileinfo metadata must contain a dir and a filename. -// returns a unique id which is used to identify the upload. The properties Size and MetaData will be filled. -func (fs *localfs) NewUpload(ctx context.Context, info tusd.FileInfo) (upload tusd.Upload, err error) { - - log := appctx.GetLogger(ctx) - log.Debug().Interface("info", info).Msg("localfs: NewUpload") - - fn := info.MetaData["filename"] - if fn == "" { - return nil, errors.New("localfs: missing filename in metadata") - } - info.MetaData["filename"] = filepath.Clean(info.MetaData["filename"]) - - dir := info.MetaData["dir"] - if dir == "" { - return nil, errors.New("localfs: missing dir in metadata") - } - info.MetaData["dir"] = filepath.Clean(info.MetaData["dir"]) - - np := fs.wrap(ctx, filepath.Join(info.MetaData["dir"], info.MetaData["filename"])) - - log.Debug().Interface("info", info).Msg("localfs: resolved filename") - - info.ID = uuid.New().String() - - binPath, err := fs.getUploadPath(ctx, info.ID) - if err != nil { - return nil, errors.Wrap(err, "localfs: error resolving upload path") - } - usr := ctxpkg.ContextMustGetUser(ctx) - info.Storage = map[string]string{ - "Type": "LocalStore", - "BinPath": binPath, - "InternalDestination": np, - - "Idp": usr.Id.Idp, - "UserId": usr.Id.OpaqueId, - "UserName": usr.Username, - "UserType": utils.UserTypeToString(usr.Id.Type), - - "LogLevel": log.GetLevel().String(), - } - // Create binary file with no content - file, err := os.OpenFile(binPath, os.O_CREATE|os.O_WRONLY, defaultFilePerm) - if err != nil { - return nil, err - } - defer file.Close() - - u := &fileUpload{ - info: info, - binPath: binPath, - infoPath: binPath + ".info", - fs: fs, - ctx: ctx, - } - - // writeInfo creates the file by itself if necessary - err = u.writeInfo() - if err != nil { - return nil, err - } - - return u, nil -} - -func (fs *localfs) getUploadPath(ctx context.Context, uploadID string) (string, error) { - return filepath.Join(fs.conf.Uploads, uploadID), nil -} - -// GetUpload returns the Upload for the given upload id -func (fs *localfs) GetUpload(ctx context.Context, id string) (tusd.Upload, error) { - binPath, err := fs.getUploadPath(ctx, id) - if err != nil { - return nil, err - } - infoPath := binPath + ".info" - info := tusd.FileInfo{} - data, err := os.ReadFile(infoPath) - if err != nil { - if os.IsNotExist(err) { - // Interpret os.ErrNotExist as 404 Not Found - err = tusd.ErrNotFound - } - return nil, err - } - if err := json.Unmarshal(data, &info); err != nil { - return nil, err - } - - stat, err := os.Stat(binPath) - if err != nil { - return nil, err - } - - info.Offset = stat.Size() - - u := &userpb.User{ - Id: &userpb.UserId{ - Idp: info.Storage["Idp"], - OpaqueId: info.Storage["UserId"], - Type: utils.UserTypeMap(info.Storage["UserType"]), - }, - Username: info.Storage["UserName"], - } - - ctx = ctxpkg.ContextSetUser(ctx, u) - - return &fileUpload{ - info: info, - binPath: binPath, - infoPath: infoPath, - fs: fs, - ctx: ctx, - }, nil -} - -type fileUpload struct { - // info stores the current information about the upload - info tusd.FileInfo - // infoPath is the path to the .info file - infoPath string - // binPath is the path to the binary file (which has no extension) - binPath string - // only fs knows how to handle metadata and versions - fs *localfs - // a context with a user - ctx context.Context -} - -// GetInfo returns the FileInfo -func (upload *fileUpload) GetInfo(ctx context.Context) (tusd.FileInfo, error) { - return upload.info, nil -} - -// GetReader returns an io.Reader for the upload -func (upload *fileUpload) GetReader(ctx context.Context) (io.ReadCloser, error) { - return os.Open(upload.binPath) -} - -// WriteChunk writes the stream from the reader to the given offset of the upload -func (upload *fileUpload) WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error) { - file, err := os.OpenFile(upload.binPath, os.O_WRONLY|os.O_APPEND, defaultFilePerm) - if err != nil { - return 0, err - } - defer file.Close() - - n, err := io.Copy(file, src) - - // If the HTTP PATCH request gets interrupted in the middle (e.g. because - // the user wants to pause the upload), Go's net/http returns an io.ErrUnexpectedEOF. - // However, for OwnCloudStore it's not important whether the stream has ended - // on purpose or accidentally. - if err != nil { - if err != io.ErrUnexpectedEOF { - return n, err - } - } - - upload.info.Offset += n - err = upload.writeInfo() - - return n, err -} - -// writeInfo updates the entire information. Everything will be overwritten. -func (upload *fileUpload) writeInfo() error { - data, err := json.Marshal(upload.info) - if err != nil { - return err - } - return os.WriteFile(upload.infoPath, data, defaultFilePerm) -} - -// FinishUpload finishes an upload and moves the file to the internal destination -func (upload *fileUpload) FinishUpload(ctx context.Context) error { - - np := upload.info.Storage["InternalDestination"] - - // TODO check etag with If-Match header - // if destination exists - // if _, err := os.Stat(np); err == nil { - // the local storage does not store metadata - // the fileid is based on the path, so no we do not need to copy it to the new file - // the local storage does not track revisions - //} - - // if destination exists - if _, err := os.Stat(np); err == nil { - // create revision - if err := upload.fs.archiveRevision(upload.ctx, np); err != nil { - return err - } - } - - err := os.Rename(upload.binPath, np) - if err != nil { - return err - } - - // only delete the upload if it was successfully written to the fs - if err := os.Remove(upload.infoPath); err != nil { - if !os.IsNotExist(err) { - log := appctx.GetLogger(ctx) - log.Err(err).Interface("info", upload.info).Msg("localfs: could not delete upload info") - } - } - - // TODO: set mtime if specified in metadata - - // metadata propagation is left to the storage implementation - return err -} - -// To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination -// - the storage needs to implement AsTerminatableUpload -// - the upload needs to implement Terminate - -// AsTerminatableUpload returns a a TerminatableUpload -func (fs *localfs) AsTerminatableUpload(upload tusd.Upload) tusd.TerminatableUpload { - return upload.(*fileUpload) -} - -// Terminate terminates the upload -func (upload *fileUpload) Terminate(ctx context.Context) error { - if err := os.Remove(upload.infoPath); err != nil { - return err - } - if err := os.Remove(upload.binPath); err != nil { - return err - } - return nil -} diff --git a/pkg/storage/utils/middleware/middleware.go b/pkg/storage/utils/middleware/middleware.go index dd57f73cae4..f37584000cc 100644 --- a/pkg/storage/utils/middleware/middleware.go +++ b/pkg/storage/utils/middleware/middleware.go @@ -24,10 +24,8 @@ import ( "net/url" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" - tusd "github.com/tus/tusd/v2/pkg/handler" "github.com/owncloud/reva/v2/pkg/storage" - "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/upload" "github.com/owncloud/reva/v2/pkg/storagespace" ) @@ -55,42 +53,6 @@ func (f *FS) ListUploadSessions(ctx context.Context, filter storage.UploadSessio return f.next.(storage.UploadSessionLister).ListUploadSessions(ctx, filter) } -// UseIn tells the tus upload middleware which extensions it supports. -func (f *FS) UseIn(composer *tusd.StoreComposer) { - f.next.(storage.ComposableFS).UseIn(composer) -} - -// NewUpload returns a new tus Upload instance -func (f *FS) NewUpload(ctx context.Context, info tusd.FileInfo) (upload tusd.Upload, err error) { - return f.next.(tusd.DataStore).NewUpload(ctx, info) -} - -// NewUpload returns a new tus Upload instance -func (f *FS) GetUpload(ctx context.Context, id string) (upload tusd.Upload, err error) { - return f.next.(tusd.DataStore).GetUpload(ctx, id) -} - -// AsTerminatableUpload returns a TerminatableUpload -// To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination -// the storage needs to implement AsTerminatableUpload -func (f *FS) AsTerminatableUpload(up tusd.Upload) tusd.TerminatableUpload { - return up.(*upload.OcisSession) -} - -// AsLengthDeclarableUpload returns a LengthDeclarableUpload -// To implement the creation-defer-length extension as specified in https://tus.io/protocols/resumable-upload.html#creation -// the storage needs to implement AsLengthDeclarableUpload -func (f *FS) AsLengthDeclarableUpload(up tusd.Upload) tusd.LengthDeclarableUpload { - return up.(*upload.OcisSession) -} - -// AsConcatableUpload returns a ConcatableUpload -// To implement the concatenation extension as specified in https://tus.io/protocols/resumable-upload.html#concatenation -// the storage needs to implement AsConcatableUpload -func (f *FS) AsConcatableUpload(up tusd.Upload) tusd.ConcatableUpload { - return up.(*upload.OcisSession) -} - func (f *FS) GetHome(ctx context.Context) (string, error) { var ( err error @@ -307,60 +269,6 @@ func (f *FS) ListFolder(ctx context.Context, ref *provider.Reference, mdKeys, fi return res0, res1 } -func (f *FS) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { - var ( - err error - unhook UnHook - unhooks []UnHook - ) - for _, hook := range f.hooks { - ctx, unhook, err = hook("InitiateUpload", ctx, ref.GetResourceId().GetSpaceId()) - if err != nil { - return nil, err - } - if unhook != nil { - unhooks = append(unhooks, unhook) - } - } - - res0, res1 := f.next.InitiateUpload(ctx, ref, uploadLength, metadata) - - for _, unhook := range unhooks { - if err := unhook(); err != nil { - return nil, err - } - } - - return res0, res1 -} - -func (f *FS) Upload(ctx context.Context, req storage.UploadRequest, uploadFunc storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { - var ( - err error - unhook UnHook - unhooks []UnHook - ) - for _, hook := range f.hooks { - ctx, unhook, err = hook("Upload", ctx, req.Ref.GetResourceId().GetSpaceId()) - if err != nil { - return &provider.ResourceInfo{}, err - } - if unhook != nil { - unhooks = append(unhooks, unhook) - } - } - - res0, res1 := f.next.Upload(ctx, req, uploadFunc) - - for _, unhook := range unhooks { - if err := unhook(); err != nil { - return &provider.ResourceInfo{}, err - } - } - - return res0, res1 -} - func (f *FS) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error { var ( err error diff --git a/pkg/upload/coordinator.go b/pkg/upload/coordinator.go index a0b004e954e..d853b5139b2 100644 --- a/pkg/upload/coordinator.go +++ b/pkg/upload/coordinator.go @@ -3,6 +3,7 @@ package upload import ( "context" "fmt" + "io" "os" "path/filepath" "strings" @@ -24,6 +25,17 @@ import ( "github.com/owncloud/reva/v2/pkg/utils" ) +// Request carries the metadata of a non-resumable (PUT) upload. +type Request struct { + Ref *provider.Reference + Body io.ReadCloser + Length int64 +} + +// FinishedFunc is called once an upload has finished, so that the caller can +// publish a FileUploaded event through its own publisher. +type FinishedFunc func(spaceOwner, executant *user.UserId, ref *provider.Reference) + // Coordinator owns the upload lifecycle: initiation, data transfer and listing. type Coordinator interface { // InitiateUpload returns the protocols and ids that bytes can be appended to. @@ -35,7 +47,7 @@ type Coordinator interface { // ListUploadSessions returns the upload sessions matching the given filter. ListUploadSessions(ctx context.Context, filter storage.UploadSessionFilter) ([]storage.UploadSession, error) // Upload writes the whole body of a non-resumable (PUT) upload and finishes it. - Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) + Upload(ctx context.Context, req Request, uff FinishedFunc) (*provider.ResourceInfo, error) // StartPostprocessing subscribes to postprocessing results and enables async // uploads. Call once, before serving requests. StartPostprocessing(stream events.Consumer, group, mountID string, numConsumers int) error @@ -198,7 +210,7 @@ func (c *coordinator) applyRequestMetadata(session Session, metadata map[string] // Upload writes the whole body of a non-resumable (PUT) upload and finishes it. // req.Ref.Path carries the session id minted by InitiateUpload. -func (c *coordinator) Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error) { +func (c *coordinator) Upload(ctx context.Context, req Request, uff FinishedFunc) (*provider.ResourceInfo, error) { // The request path arrives rooted, while session ids are stored unrooted. session, err := c.store.Get(ctx, strings.TrimPrefix(req.Ref.GetPath(), "/")) if err != nil { diff --git a/pkg/upload/coordinator_test.go b/pkg/upload/coordinator_test.go index a8161f6f497..9fe8015a3db 100644 --- a/pkg/upload/coordinator_test.go +++ b/pkg/upload/coordinator_test.go @@ -299,7 +299,7 @@ var _ = Describe("coordinator", func() { Expect(session.TouchBin()).To(Succeed()) Expect(session.Persist(ctx)).To(Succeed()) - ri, err := c.Upload(ctx, storage.UploadRequest{ + ri, err := c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/" + session.ID()}, Body: io.NopCloser(strings.NewReader(body)), Length: bodyLen, @@ -319,7 +319,7 @@ var _ = Describe("coordinator", func() { Expect(session.TouchBin()).To(Succeed()) Expect(session.Persist(ctx)).To(Succeed()) - _, err := c.Upload(ctx, storage.UploadRequest{ + _, err := c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/" + session.ID()}, Body: io.NopCloser(strings.NewReader("short")), Length: bodyLen, @@ -340,7 +340,7 @@ var _ = Describe("coordinator", func() { Expect(session.TouchBin()).To(Succeed()) Expect(session.Persist(ctx)).To(Succeed()) - _, err := c.Upload(ctx, storage.UploadRequest{ + _, err := c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/" + session.ID()}, Body: io.NopCloser(strings.NewReader(body)), Length: bodyLen, diff --git a/pkg/upload/put_test.go b/pkg/upload/put_test.go index fecade1e327..23ed0f9b2a0 100644 --- a/pkg/upload/put_test.go +++ b/pkg/upload/put_test.go @@ -59,7 +59,7 @@ var _ = Describe("Upload", func() { // put is the call the dataprovider makes: the session id rides in the ref path. put := func(session Session, content string) (*provider.ResourceInfo, error) { - return c.Upload(ctx, storage.UploadRequest{ + return c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/" + session.ID()}, Body: io.NopCloser(strings.NewReader(content)), Length: int64(len(content)), @@ -97,7 +97,7 @@ var _ = Describe("Upload", func() { }) It("reports an unknown session id as not found", func() { - _, err := c.Upload(ctx, storage.UploadRequest{ + _, err := c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/no-such-session"}, Body: io.NopCloser(strings.NewReader(body)), Length: bodyLen, @@ -111,7 +111,7 @@ var _ = Describe("Upload", func() { It("rejects a body shorter than the declared length", func() { session := initiated("") - _, err := c.Upload(ctx, storage.UploadRequest{ + _, err := c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/" + session.ID()}, Body: io.NopCloser(strings.NewReader("short")), Length: bodyLen, @@ -137,7 +137,7 @@ var _ = Describe("Upload", func() { It("propagates a failure to read the body", func() { session := initiated("") - _, err := c.Upload(ctx, storage.UploadRequest{ + _, err := c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/" + session.ID()}, Body: io.NopCloser(failingReader{err: errors.New("connection reset by peer")}), Length: bodyLen, @@ -163,7 +163,7 @@ var _ = Describe("Upload", func() { var gotRef *provider.Reference var gotExecutant *userpb.UserId - _, err := c.Upload(ctx, storage.UploadRequest{ + _, err := c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/" + session.ID()}, Body: io.NopCloser(strings.NewReader(body)), Length: bodyLen, @@ -182,7 +182,7 @@ var _ = Describe("Upload", func() { fs.commitErr = errors.New("blobstore unavailable") called := false - _, err := c.Upload(ctx, storage.UploadRequest{ + _, err := c.Upload(ctx, Request{ Ref: &provider.Reference{Path: "/" + session.ID()}, Body: io.NopCloser(strings.NewReader(body)), Length: bodyLen, diff --git a/tests/helpers/helpers.go b/tests/helpers/helpers.go index 2942f399c59..ad56de47e0b 100644 --- a/tests/helpers/helpers.go +++ b/tests/helpers/helpers.go @@ -32,6 +32,7 @@ import ( "github.com/owncloud/ocis/v2/services/webdav/pkg/net" "github.com/pkg/errors" + "github.com/rs/zerolog" "github.com/studio-b12/gowebdav" gatewayv1beta1 "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1" @@ -42,6 +43,7 @@ import ( "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/rhttp" "github.com/owncloud/reva/v2/pkg/storage" + "github.com/owncloud/reva/v2/pkg/upload" "github.com/owncloud/reva/v2/pkg/utils" ) @@ -96,10 +98,26 @@ func TempJSONFile(c any) (string, error) { return TempFile(bytes.NewBuffer(data)) } -// Upload can be used to initiate an upload and do the upload to a storage.FS in one step +// Upload can be used to initiate an upload and do the upload to a storage.FS in one step. +// The upload coordinator drives the flow; the driver only sees the slim contract. func Upload(ctx context.Context, fs storage.FS, ref *provider.Reference, content []byte) error { + // The session files only live until CommitUpload has read the bytes back, so a + // throwaway directory is enough. No publisher and no chunk folder: without a + // postprocessing consumer the upload finishes synchronously and never chunks. + uploadDir, err := os.MkdirTemp("", "reva-test-uploads-*") + if err != nil { + return err + } + defer os.RemoveAll(uploadDir) + + log := zerolog.Nop() + coord, err := upload.NewCoordinatorFromConfig(uploadDir, nil, fs, nil, &log, false) + if err != nil { + return err + } + length := int64(len(content)) - uploadIds, err := fs.InitiateUpload(ctx, ref, length, map[string]string{}) + uploadIds, err := coord.InitiateUpload(ctx, ref, length, map[string]string{}) if err != nil { return err } @@ -109,7 +127,7 @@ func Upload(ctx context.Context, fs storage.FS, ref *provider.Reference, content return errors.New("simple upload method not available") } uploadRef := &provider.Reference{Path: "/" + uploadID} - _, err = fs.Upload(ctx, storage.UploadRequest{ + _, err = coord.Upload(ctx, upload.Request{ Ref: uploadRef, Body: io.NopCloser(bytes.NewReader(content)), Length: int64(len(content)),