use tusd datastorages

changelog/unreleased/tusd-data-storage.md

@@ -0,0 +1,5 @@
+Enhancement: Use Tusd data storage implementations
+
+Decomposedfs now uses the data store implementation for uploads that comes with tusd instead of implementing the interface itself. This allows storing uploads directly in s3. When all bytes are transferred tusd will call `PreFinishResponseCallback` if the storage driver implements it.
+
+https://github.com/cs3org/reva/pull/4148

internal/grpc/services/storageprovider/storageprovider.go

@@ -338,6 +338,15 @@ func (s *service) InitiateFileUpload(ctx context.Context, req *provider.Initiate
 		}, nil
 	}
 
+	// FIXME: This is a hack to transport more metadata to the storage.FS InitiateUpload implementation
+	// we should use a request object that can carry
+	// * if-match
+	// * if-unmodified-since
+	// * uploadLength from the tus Upload-Length header
+	// * checksum from the tus Upload-Checksum header
+	// * mtime from the X-OC-Mtime header
+	// * expires from the s.conf.UploadExpiration ... should that not be part of the driver?
+	// * providerID
 	metadata := map[string]string{}
 	ifMatch := req.GetIfMatch()
 	if ifMatch != "" {

internal/http/services/dataprovider/dataprovider.go

@@ -143,7 +143,13 @@ func getDataTXs(c *config, fs storage.FS, publisher events.Publisher) (map[strin
 			if tx, err := f(c.DataTXs[t], publisher); err == nil {
 				if handler, err := tx.Handler(fs); err == nil {
 					txs[t] = handler
+					// FIXME we at least need to log this. the ocm received storage e.g. does not support tus
+					// } else {
+					// return nil, err
 				}
+				// FIXME we at least need to log this. the ocm received storage e.g. does not support tus
+				// } else {
+				// return nil, err
 			}
 		}
 	}

internal/http/services/owncloud/ocdav/put.go

@@ -302,11 +302,13 @@ func (s *svc) handlePut(ctx context.Context, w http.ResponseWriter, r *http.Requ
 
 	httpReq, err := rhttp.NewRequest(ctx, http.MethodPut, ep, r.Body)
 	if err != nil {
+		log.Error().Err(err).Msg("error creating new request to data service")
 		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
 	Propagator.Inject(ctx, propagation.HeaderCarrier(httpReq.Header))
 	httpReq.Header.Set(datagateway.TokenTransportHeader, token)
+	httpReq.ContentLength = r.ContentLength
 
 	httpRes, err := s.client.Do(httpReq)
 	if err != nil {

internal/http/services/owncloud/ocdav/tus.go

@@ -88,6 +88,10 @@ func (s *svc) handleSpacesTusPost(w http.ResponseWriter, r *http.Request, spaceI
 
 	sublog := appctx.GetLogger(ctx).With().Str("spaceid", spaceID).Str("path", r.URL.Path).Logger()
 
+	// use filename to build a storage space reference
+	// but what if upload happens directly to toh resourceid .. and filename is empty?
+	// currently there is always a validator thet requires the filename is not empty ...
+	// hm -> bug: clients currently cannot POST to an existing source with a resource id only
 	ref, err := spacelookup.MakeStorageSpaceReference(spaceID, path.Join(r.URL.Path, meta["filename"]))
 	if err != nil {
 		w.WriteHeader(http.StatusBadRequest)

internal/http/services/owncloud/ocdav/versions.go

@@ -180,6 +180,7 @@ func (h *VersionsHandler) doListVersions(w http.ResponseWriter, r *http.Request,
 			Type: provider.ResourceType_RESOURCE_TYPE_FILE,
 			Id: &provider.ResourceId{
 				StorageId: "versions",
+				SpaceId:   info.Id.SpaceId,
 				OpaqueId:  info.Id.OpaqueId + "@" + versions[i].GetKey(),
 			},
 			// Checksum

pkg/rhttp/datatx/manager/tus/tus.go

@@ -16,6 +16,9 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
+// Package tus implements a data tx manager that handles uploads using the TUS protocol.
+// reva storage drivers should implement the hasTusDatastore interface by using composition
+// of an upstream tusd.DataStore. If necessary they can also implement a tusd.DataStore directly.
 package tus
 
 import (
@@ -33,12 +36,12 @@ import (
 	"github.com/cs3org/reva/v2/pkg/appctx"
 	"github.com/cs3org/reva/v2/pkg/errtypes"
 	"github.com/cs3org/reva/v2/pkg/events"
+	"github.com/cs3org/reva/v2/pkg/logger"
 	"github.com/cs3org/reva/v2/pkg/rhttp/datatx"
 	"github.com/cs3org/reva/v2/pkg/rhttp/datatx/manager/registry"
 	"github.com/cs3org/reva/v2/pkg/rhttp/datatx/metrics"
 	"github.com/cs3org/reva/v2/pkg/storage"
 	"github.com/cs3org/reva/v2/pkg/storage/cache"
-	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
 	"github.com/mitchellh/mapstructure"
 )
@@ -76,46 +79,80 @@ func New(m map[string]interface{}, publisher events.Publisher) (datatx.DataTX, e
 }
 
 func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
-	composable, ok := fs.(composable)
-	if !ok {
-		return nil, errtypes.NotSupported("file system does not support the tus protocol")
+	zlog, err := logger.FromConfig(&logger.LogConf{
+		Output: "stderr",
+		Mode:   "json",
+		Level:  "error", // FIXME introduce shared config for logging
+	})
+	if err != nil {
+		return nil, errtypes.NotSupported("could not initialize log")
 	}
 
-	// A storage backend for tusd may consist of multiple different parts which
-	// handle upload creation, locking, termination and so on. The composer is a
-	// place where all those separated pieces are joined together. In this example
-	// we only use the file store but you may plug in multiple.
 	composer := tusd.NewStoreComposer()
 
-	// let the composable storage tell tus which extensions it supports
-	composable.UseIn(composer)
-
 	config := tusd.Config{
-		StoreComposer:         composer,
+		StoreComposer: composer,
+		PreUploadCreateCallback: func(hook tusd.HookEvent) error {
+			return errors.New("uploads must be created with a cs3 InitiateUpload call")
+		},
 		NotifyCompleteUploads: true,
-		Logger:                log.New(appctx.GetLogger(context.Background()), "", 0),
+		Logger:                log.New(zlog, "", 0),
+	}
+
+	var dataStore tusd.DataStore
+
+	cb, ok := fs.(hasTusDatastore)
+	if ok {
+		dataStore = cb.GetDataStore()
+		composable, ok := dataStore.(composable)
+		if !ok {
+			return nil, errtypes.NotSupported("tus datastore is not composable")
+		}
+		composable.UseIn(composer)
+		config.PreFinishResponseCallback = cb.PreFinishResponseCallback
+	} else {
+		composable, ok := fs.(composable)
+		if !ok {
+			return nil, errtypes.NotSupported("storage driver does not support the tus protocol")
+		}
+
+		// let the composable storage tell tus which extensions it supports
+		composable.UseIn(composer)
+		dataStore, ok = fs.(tusd.DataStore)
+		if !ok {
+			return nil, errtypes.NotSupported("storage driver does not support the tus datastore")
+		}
 	}
 
 	handler, err := tusd.NewUnroutedHandler(config)
 	if err != nil {
 		return nil, err
 	}
 
-	if _, ok := fs.(storage.UploadSessionLister); ok {
+	usl, ok := fs.(storage.UploadSessionLister)
+	if ok {
 		// We can currently only send updates if the fs is decomposedfs as we read very specific keys from the storage map of the tus info
 		go func() {
 			for {
 				ev := <-handler.CompleteUploads
 				// We should be able to get the upload progress with fs.GetUploadProgress, but currently tus will erase the info files
 				// so we create a Progress instance here that is used to read the correct properties
-				up := upload.Progress{
-					Info: ev.Upload,
+				sessions, err := usl.ListUploadSessions(context.Background(), storage.UploadSessionFilter{ID: &ev.Upload.ID})
+				if err != nil {
+					appctx.GetLogger(context.Background()).Error().Err(err).Str("id", ev.Upload.ID).Msg("failed to list upload session for upload")
+					continue
+				}
+				if len(sessions) != 1 {
+					appctx.GetLogger(context.Background()).Error().Err(err).Str("id", ev.Upload.ID).Msg("no upload session found")
+					continue
 				}
-				executant := up.Executant()
-				ref := up.Reference()
+				us := sessions[0]
+
+				executant := us.Executant()
+				ref := us.Reference()
 				datatx.InvalidateCache(&executant, &ref, m.statCache)
 				if m.publisher != nil {
-					if err := datatx.EmitFileUploadedEvent(up.SpaceOwner(), &executant, &ref, m.publisher); err != nil {
+					if err := datatx.EmitFileUploadedEvent(us.SpaceOwner(), &executant, &ref, m.publisher); err != nil {
 						appctx.GetLogger(context.Background()).Error().Err(err).Msg("failed to publish FileUploaded event")
 					}
 				}
@@ -137,7 +174,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 				metrics.UploadsActive.Sub(1)
 			}()
 			// set etag, mtime and file id
-			setHeaders(fs, w, r)
+			setHeaders(dataStore, usl, w, r)
 			handler.PostFile(w, r)
 		case "HEAD":
 			handler.HeadFile(w, r)
@@ -147,7 +184,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 				metrics.UploadsActive.Sub(1)
 			}()
 			// set etag, mtime and file id
-			setHeaders(fs, w, r)
+			setHeaders(dataStore, usl, w, r)
 			handler.PatchFile(w, r)
 		case "DELETE":
 			handler.DelFile(w, r)
@@ -174,14 +211,14 @@ type composable interface {
 	UseIn(composer *tusd.StoreComposer)
 }
 
-func setHeaders(fs storage.FS, w http.ResponseWriter, r *http.Request) {
+type hasTusDatastore interface {
+	PreFinishResponseCallback(hook tusd.HookEvent) error
+	GetDataStore() tusd.DataStore
+}
+
+func setHeaders(datastore tusd.DataStore, usl storage.UploadSessionLister, w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 	id := path.Base(r.URL.Path)
-	datastore, ok := fs.(tusd.DataStore)
-	if !ok {
-		appctx.GetLogger(ctx).Error().Interface("fs", fs).Msg("storage is not a tus datastore")
-		return
-	}
 	upload, err := datastore.GetUpload(ctx, id)
 	if err != nil {
 		appctx.GetLogger(ctx).Error().Err(err).Msg("could not get upload from storage")
@@ -192,14 +229,51 @@ func setHeaders(fs storage.FS, w http.ResponseWriter, r *http.Request) {
 		appctx.GetLogger(ctx).Error().Err(err).Msg("could not get upload info for upload")
 		return
 	}
-	expires := info.MetaData["expires"]
+	var expires string
+
+	var resourceid provider.ResourceId
+	var uploadSession storage.UploadSession
+	if usl != nil {
+		sessions, err := usl.ListUploadSessions(ctx, storage.UploadSessionFilter{ID: &id})
+		if err != nil {
+			appctx.GetLogger(context.Background()).Error().Err(err).Str("id", id).Msg("failed to list upload session for upload")
+			return
+		}
+		if len(sessions) != 1 {
+			appctx.GetLogger(context.Background()).Error().Err(err).Str("id", id).Msg("no upload session found")
+			return
+		}
+		uploadSession = sessions[0]
+
+		t := time.Time{}
+		if uploadSession.Expires() != t {
+			expires = uploadSession.Expires().Format(net.RFC1123)
+		}
+
+		reference := uploadSession.Reference()
+		resourceid = *reference.GetResourceId()
+	}
+
+	// FIXME expires should be part of the tus handler
+	// fallback for outdated storageproviders that implement a tus datastore
+	if expires == "" {
+		expires = info.MetaData["expires"]
+	}
+
 	if expires != "" {
 		w.Header().Set(net.HeaderTusUploadExpires, expires)
 	}
-	resourceid := provider.ResourceId{
-		StorageId: info.MetaData["providerID"],
-		SpaceId:   info.Storage["SpaceRoot"],
-		OpaqueId:  info.Storage["NodeId"],
+
+	// fallback for outdated storageproviders that implement a tus datastore
+	if resourceid.GetStorageId() == "" {
+		resourceid.StorageId = info.MetaData["providerID"]
 	}
+	if resourceid.GetSpaceId() == "" {
+		resourceid.SpaceId = info.MetaData["SpaceRoot"]
+	}
+	if resourceid.GetOpaqueId() == "" {
+		resourceid.OpaqueId = info.MetaData["NodeId"]
+	}
+
 	w.Header().Set(net.HeaderOCFileID, storagespace.FormatResourceID(resourceid))
 }

pkg/storage/fs/nextcloud/nextcloud_server_mock.go

@@ -113,7 +113,8 @@ var responses = map[string]Response{
 
 	`POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/GetMD {"ref":{"path":"/file"},"mdKeys":null}`:                                                                                                        {404, ``, serverStateEmpty},
 	`POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/InitiateUpload {"ref":{"path":"/file"},"uploadLength":0,"metadata":{"providerID":""}}`:                                                               {200, `{"simple": "yes","tus": "yes"}`, serverStateEmpty},
-	`POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/InitiateUpload {"ref":{"resource_id":{"storage_id":"f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c"},"path":"/versionedFile"},"uploadLength":0,"metadata":{}}`: {200, `{"simple": "yes","tus": "yes"}`, serverStateEmpty},
+	`POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/InitiateUpload {"ref":{"resource_id":{"storage_id":"f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c"},"path":"/versionedFile"},"uploadLength":1,"metadata":{}}`: {200, `{"simple": "yes","tus": "yes"}`, serverStateEmpty},
+	`POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/InitiateUpload {"ref":{"resource_id":{"storage_id":"f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c"},"path":"/versionedFile"},"uploadLength":2,"metadata":{}}`: {200, `{"simple": "yes","tus": "yes"}`, serverStateEmpty},
 
 	`POST /apps/sciencemesh/~f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c/api/storage/GetMD {"ref":{"path":"/yes"},"mdKeys":[]}`: {200, `{"opaque":{},"type":1,"id":{"opaque_id":"fileid-/yes"},"checksum":{},"etag":"deadbeef","mime_type":"text/plain","mtime":{"seconds":1234567890},"path":"/yes","permission_set":{},"size":1,"canonical_metadata":{},"arbitrary_metadata":{}}`, serverStateEmpty},
 

pkg/storage/fs/ocis/ocis.go

@@ -27,6 +27,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/fs/registry"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/options"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/tus"
 )
 
 func init() {
@@ -46,5 +47,7 @@ func New(m map[string]interface{}, stream events.Stream) (storage.FS, error) {
 		return nil, err
 	}
 
-	return decomposedfs.NewDefault(m, bs, stream)
+	tusDataStore := tus.NewFileStore(o.Root)
+
+	return decomposedfs.NewDefault(m, bs, tusDataStore, stream)
 }

pkg/storage/fs/s3ng/blobstore/blobstore.go

@@ -63,6 +63,10 @@ func New(endpoint, region, bucket, accessKey, secretKey string) (*Blobstore, err
 	}, nil
 }
 
+func (bs *Blobstore) MoveBlob(node *node.Node, source, bucket, key string) error {
+	return nil
+}
+
 // Upload stores some data in the blobstore under the given key
 func (bs *Blobstore) Upload(node *node.Node, source string) error {
 	reader, err := os.Open(source)

pkg/storage/fs/s3ng/option.go

@@ -43,6 +43,24 @@ type Options struct {
 
 	// Secret key for the s3 blobstore
 	S3SecretKey string `mapstructure:"s3.secret_key"`
+
+	// UploadObjectPrefix for the s3 blobstore
+	S3UploadObjectPrefix string `mapstructure:"s3.upload_object_prefix"`
+
+	// UploadMetadataPrefix for the s3 blobstore
+	S3UploadMetadataPrefix string `mapstructure:"s3.upload_metadata_prefix"`
+
+	// UploadTemporaryDirectory for the s3 blobstore
+	S3UploadTemporaryDirectory string `mapstructure:"s3.upload_temporary_directory"`
+
+	// DisableSSL for the s3 blobstore
+	S3DisableSSL bool `mapstructure:"s3.disable_ssl"`
+
+	// ForcePathStyle for the s3 blobstore
+	S3ForcePathStyle bool `mapstructure:"s3.force_path_style"`
+
+	// Root for the upload sessions
+	Root string `mapstructure:"root"`
 }
 
 // S3ConfigComplete return true if all required s3 fields are set

pkg/storage/fs/s3ng/s3ng.go

@@ -21,11 +21,16 @@ package s3ng
 import (
 	"fmt"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/cs3org/reva/v2/pkg/storage"
 	"github.com/cs3org/reva/v2/pkg/storage/fs/registry"
 	"github.com/cs3org/reva/v2/pkg/storage/fs/s3ng/blobstore"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/tus"
 )
 
 func init() {
@@ -49,5 +54,17 @@ func New(m map[string]interface{}, stream events.Stream) (storage.FS, error) {
 		return nil, err
 	}
 
-	return decomposedfs.NewDefault(m, bs, stream)
+	s3Config := aws.NewConfig()
+	s3Config.WithCredentials(credentials.NewStaticCredentials(o.S3AccessKey, o.S3SecretKey, "")).
+		WithEndpoint(o.S3Endpoint).
+		WithRegion(o.S3Region).
+		WithS3ForcePathStyle(o.S3ForcePathStyle).
+		WithDisableSSL(o.S3DisableSSL)
+
+	tusDataStore := tus.NewS3Store(o.Root, o.S3Bucket, s3.New(session.Must(session.NewSession()), s3Config))
+	tusDataStore.ObjectPrefix = o.S3UploadObjectPrefix
+	tusDataStore.MetadataObjectPrefix = o.S3UploadMetadataPrefix
+	tusDataStore.TemporaryDirectory = o.S3UploadTemporaryDirectory
+
+	return decomposedfs.NewDefault(m, bs, tusDataStore, stream)
 }