Ocisdev 900

internal/grpc/services/storageprovider/storageprovider.go

@@ -25,6 +25,7 @@ import (
 	"net/url"
 	"os"
 	"path"
+	"path/filepath"
 	"sort"
 	"strconv"
 	"strings"
@@ -47,6 +48,7 @@ import (
 	"github.com/owncloud/reva/v2/pkg/storage"
 	"github.com/owncloud/reva/v2/pkg/storage/fs/registry"
 	"github.com/owncloud/reva/v2/pkg/storagespace"
+	pkgupload "github.com/owncloud/reva/v2/pkg/upload"
 	"github.com/owncloud/reva/v2/pkg/utils"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
@@ -70,6 +72,7 @@ type config struct {
 	CustomMimeTypesJSON string                            `mapstructure:"custom_mimetypes_json" docs:"nil;An optional mapping file with the list of supported custom file extensions and corresponding mime types."`
 	MountID             string                            `mapstructure:"mount_id"`
 	UploadExpiration    int64                             `mapstructure:"upload_expiration" docs:"0;Duration for how long uploads will be valid."`
+	UploadDirectory     string                            `mapstructure:"upload_directory" docs:";Local directory for staging upload sessions. Overrides the driver's root. Required for drivers that have no local filesystem root."`
 	Events              eventconfig                       `mapstructure:"events" docs:"0;Event stream configuration"`
 }
 
@@ -81,6 +84,8 @@ type eventconfig struct {
 	EnableTLS            bool   `mapstructure:"nats_enable_tls" docs:"events tls switch"`
 	AuthUsername         string `mapstructure:"nats_username" docs:"event stream username"`
 	AuthPassword         string `mapstructure:"nats_password" docs:"event stream password"`
+	ConsumerGroup        string `mapstructure:"consumer_group" docs:"dcfs;Consumer group for the upload coordinator"`
+	NumConsumers         int    `mapstructure:"numconsumers" docs:"1;Number of concurrent postprocessing event consumers"`
 }
 
 func (c *config) init() {
@@ -101,11 +106,20 @@ func (c *config) init() {
 	if len(c.AvailableXS) == 0 {
 		c.AvailableXS = map[string]uint32{"md5": 100, "unset": 1000}
 	}
+
+	if c.Events.ConsumerGroup == "" {
+		c.Events.ConsumerGroup = "dcfs"
+	}
+	if c.Events.NumConsumers <= 0 {
+		c.Events.NumConsumers = 1
+	}
 }
 
+
 type Service struct {
 	conf          *config
 	Storage       storage.FS
+	Coordinator   pkgupload.Coordinator
 	dataServerURL *url.URL
 	availableXS   []*provider.ResourceChecksumPriority
 }
@@ -202,9 +216,33 @@ func New(m map[string]interface{}, ss *grpc.Server, log *zerolog.Logger) (rgrpc.
 		return nil, err
 	}
 
+	store := pkgupload.NewFileStoreFromConfig(c.UploadDirectory, c.Drivers[c.Driver], log)
+	if store == nil {
+		return nil, fmt.Errorf("storageprovider: cannot determine upload directory, set upload_directory in config or driver root")
+	}
+	if err := store.Setup(); err != nil {
+		return nil, fmt.Errorf("storageprovider: upload directory setup failed: %w", err)
+	}
+	evstream, err := estreamFromConfig(c.Events)
+	if err != nil {
+		return nil, err
+	}
+	async := evstream != nil
+	coord, err := pkgupload.NewCoordinator(fs, store, evstream, async,
+		c.MountID, c.Events.ConsumerGroup, c.Events.NumConsumers, log,
+		filepath.Join(store.Root(), "uploads"))
+	if err != nil {
+		return nil, err
+	}
+	if async {
+		if err := coord.Start(evstream); err != nil {
+			return nil, err
+		}
+	}
 	service := &Service{
 		conf:          c,
 		Storage:       fs,
+		Coordinator:   coord,
 		dataServerURL: u,
 		availableXS:   xsTypes,
 	}
@@ -427,7 +465,7 @@ func (s *Service) InitiateFileUpload(ctx context.Context, req *provider.Initiate
 		metadata["expires"] = strconv.Itoa(int(expirationTimestamp.Seconds))
 	}
 
-	uploadIDs, err := s.Storage.InitiateUpload(ctx, req.Ref, uploadLength, metadata)
+	uploadIDs, err := s.Coordinator.InitiateUpload(ctx, req.Ref, uploadLength, metadata)
 	if err != nil {
 		var st *rpc.Status
 		switch err.(type) {
@@ -1299,7 +1337,15 @@ func estreamFromConfig(c eventconfig) (events.Stream, error) {
 		return nil, nil
 	}
 
-	return stream.NatsFromConfig("storageprovider", false, stream.NatsConfig(c))
+	return stream.NatsFromConfig("storageprovider", false, stream.NatsConfig{
+		Endpoint:             c.Endpoint,
+		Cluster:              c.Cluster,
+		TLSInsecure:          c.TLSInsecure,
+		TLSRootCACertificate: c.TLSRootCACertificate,
+		EnableTLS:            c.EnableTLS,
+		AuthUsername:         c.AuthUsername,
+		AuthPassword:         c.AuthPassword,
+	})
 }
 
 func canLockPublicShare(ctx context.Context) bool {

internal/http/services/dataprovider/dataprovider.go

@@ -21,6 +21,7 @@ package dataprovider
 import (
 	"fmt"
 	"net/http"
+	"path/filepath"
 
 	"github.com/mitchellh/mapstructure"
 	"github.com/rs/zerolog"
@@ -33,6 +34,7 @@ import (
 	"github.com/owncloud/reva/v2/pkg/rhttp/router"
 	"github.com/owncloud/reva/v2/pkg/storage"
 	"github.com/owncloud/reva/v2/pkg/storage/fs/registry"
+	pkgupload "github.com/owncloud/reva/v2/pkg/upload"
 )
 
 func init() {
@@ -44,13 +46,17 @@ type config struct {
 	Driver             string                            `mapstructure:"driver" docs:"localhome;The storage driver to be used."`
 	Drivers            map[string]map[string]interface{} `mapstructure:"drivers" docs:"url:pkg/storage/fs/localhome/localhome.go;The configuration for the storage driver"`
 	DataTXs            map[string]map[string]interface{} `mapstructure:"data_txs" docs:"url:pkg/rhttp/datatx/manager/simple/simple.go;The configuration for the data tx protocols"`
+	MountID            string                            `mapstructure:"mount_id"`
 	NatsAddress        string                            `mapstructure:"nats_address"`
 	NatsClusterID      string                            `mapstructure:"nats_clusterID"`
 	NatsTLSInsecure    bool                              `mapstructure:"nats_tls_insecure"`
 	NatsRootCACertPath string                            `mapstructure:"nats_root_ca_cert_path"`
 	NatsEnableTLS      bool                              `mapstructure:"nats_enable_tls"`
 	NatsUsername       string                            `mapstructure:"nats_username"`
 	NatsPassword       string                            `mapstructure:"nats_password"`
+	ConsumerGroup      string                            `mapstructure:"consumer_group"`
+	NumConsumers       int                               `mapstructure:"numconsumers"`
+	UploadDirectory    string                            `mapstructure:"upload_directory" docs:";Local directory for staging upload sessions. Overrides the driver's root. Required for drivers that have no local filesystem root."`
 }
 
 func (c *config) init() {
@@ -60,6 +66,12 @@ func (c *config) init() {
 	if c.Driver == "" {
 		c.Driver = "localhome"
 	}
+	if c.ConsumerGroup == "" {
+		c.ConsumerGroup = "dcfs"
+	}
+	if c.NumConsumers <= 0 {
+		c.NumConsumers = 1
+	}
 }
 
 type svc struct {
@@ -104,7 +116,27 @@ func New(m map[string]interface{}, log *zerolog.Logger) (global.Service, error)
 		return nil, err
 	}
 
-	dataTXs, err := getDataTXs(conf, fs, evstream, log)
+	store := pkgupload.NewFileStoreFromConfig(conf.UploadDirectory, conf.Drivers[conf.Driver], log)
+	if store == nil {
+		return nil, fmt.Errorf("dataprovider: cannot determine upload directory, set upload_directory in config or driver root")
+	}
+	if err := store.Setup(); err != nil {
+		return nil, fmt.Errorf("dataprovider: upload directory setup failed: %w", err)
+	}
+	async := evstream != nil
+	coord, err := pkgupload.NewCoordinator(fs, store, evstream, async,
+		conf.MountID, conf.ConsumerGroup, conf.NumConsumers, log,
+		filepath.Join(store.Root(), "uploads"))
+	if err != nil {
+		return nil, err
+	}
+	if async {
+		if err := coord.Start(evstream); err != nil {
+			return nil, err
+		}
+	}
+
+	dataTXs, err := getDataTXs(conf, coord, fs, evstream, log)
 	if err != nil {
 		return nil, err
 	}
@@ -126,7 +158,7 @@ func getFS(c *config, stream events.Stream, log *zerolog.Logger) (storage.FS, er
 	return nil, fmt.Errorf("driver not found: %s", c.Driver)
 }
 
-func getDataTXs(c *config, fs storage.FS, publisher events.Publisher, log *zerolog.Logger) (map[string]http.Handler, error) {
+func getDataTXs(c *config, coord pkgupload.Coordinator, driver storage.FS, publisher events.Publisher, log *zerolog.Logger) (map[string]http.Handler, error) {
 	if c.DataTXs == nil {
 		c.DataTXs = make(map[string]map[string]interface{})
 	}
@@ -146,7 +178,7 @@ func getDataTXs(c *config, fs storage.FS, publisher events.Publisher, log *zerol
 	for t := range c.DataTXs {
 		if f, ok := datatxregistry.NewFuncs[t]; ok {
 			if tx, err := f(c.DataTXs[t], publisher, log); err == nil {
-				if handler, err := tx.Handler(fs); err == nil {
+				if handler, err := tx.Handler(coord, driver); err == nil {
 					txs[t] = handler
 				}
 			}

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

@@ -666,6 +666,13 @@ func (s *svc) prepareCopy(ctx context.Context, w http.ResponseWriter, r *http.Re
 		return nil
 	}
 
+	if !srcStatRes.GetInfo().GetPermissionSet().GetInitiateFileDownload() {
+		w.WriteHeader(http.StatusForbidden)
+		b, err := errors.Marshal(http.StatusForbidden, "no permission to copy from this resource", "", "")
+		errors.HandleWebdavError(log, w, b, err)
+		return nil
+	}
+
 	dstStatReq := &provider.StatRequest{Ref: dstRef}
 	dstStatRes, err := client.Stat(ctx, dstStatReq)
 	switch {

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

@@ -587,6 +587,10 @@ func (p *Handler) getResourceInfos(ctx context.Context, w http.ResponseWriter, r
 			var status *rpc.Status
 			info, status, err = p.statSpace(ctx, spaceRef, metadataKeys, fieldMaskPaths)
 			if err != nil || status.GetCode() != rpc.Code_CODE_OK {
+				if status.GetCode() == rpc.Code_CODE_TOO_EARLY {
+					w.WriteHeader(http.StatusTooEarly)
+					return nil, false, false
+				}
 				continue
 			}
 		}

pkg/ocm/storage/received/ocm.go

@@ -539,6 +539,9 @@ func (d *driver) GetLock(ctx context.Context, ref *provider.Reference) (*provide
 		return nil, err
 	}
 
+	if token == "" {
+		return nil, errtypes.NotFound("no lock found")
+	}
 	return &provider.Lock{LockId: token, Type: provider.LockType_LOCK_TYPE_EXCL}, nil
 }
 

pkg/ocm/storage/received/upload.go

@@ -76,8 +76,8 @@ func (d *driver) InitiateUpload(ctx context.Context, ref *provider.Reference, up
 	}, nil
 }
 
-func (d *driver) MarkProcessing(ctx context.Context, ref *provider.Reference, processing bool, sessionID string) error {
-	return errtypes.NotSupported("op not supported")
+func (d *driver) MarkProcessing(_ context.Context, _ *provider.Reference, _ bool, _ string) error {
+	return nil
 }
 
 func (d *driver) CommitUpload(ctx context.Context, ref *provider.Reference, source storage.UploadSource) (*provider.ResourceInfo, error) {

pkg/rhttp/datatx/datatx.go

@@ -28,12 +28,13 @@ import (
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/owncloud/reva/v2/pkg/events"
 	"github.com/owncloud/reva/v2/pkg/storage"
+	pkgupload "github.com/owncloud/reva/v2/pkg/upload"
 	"github.com/owncloud/reva/v2/pkg/utils"
 )
 
 // DataTX provides an abstraction around various data transfer protocols.
 type DataTX interface {
-	Handler(fs storage.FS) (http.Handler, error)
+	Handler(coord pkgupload.Coordinator, driver storage.FS) (http.Handler, error)
 }
 
 // EmitFileUploadedEvent is a helper function which publishes a FileUploaded event

pkg/rhttp/datatx/manager/simple/simple.go

@@ -40,6 +40,7 @@ import (
 	"github.com/owncloud/reva/v2/pkg/storage"
 	"github.com/owncloud/reva/v2/pkg/storage/cache"
 	"github.com/owncloud/reva/v2/pkg/storagespace"
+	pkgupload "github.com/owncloud/reva/v2/pkg/upload"
 	"github.com/owncloud/reva/v2/pkg/utils"
 )
 
@@ -78,7 +79,7 @@ func New(m map[string]interface{}, publisher events.Publisher, log *zerolog.Logg
 	}, nil
 }
 
-func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
+func (m *manager) Handler(coord pkgupload.Coordinator, driver storage.FS) (http.Handler, error) {
 	h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		sublog := m.log.With().Str("path", r.URL.Path).Logger()
 		r = r.WithContext(appctx.WithLogger(r.Context(), &sublog))
@@ -92,7 +93,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 					metrics.DownloadsActive.Sub(1)
 				}()
 			}
-			download.GetOrHeadFile(w, r, fs, "")
+			download.GetOrHeadFile(w, r, driver, "")
 		case "PUT":
 			metrics.UploadsActive.Add(1)
 			defer func() {
@@ -114,7 +115,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 				ctx = ctxpkg.ContextSetLockID(ctx, lockID)
 			}
 
-			info, err := fs.Upload(ctx, storage.UploadRequest{
+			info, err := coord.Upload(ctx, storage.UploadRequest{
 				Ref:    ref,
 				Body:   r.Body,
 				Length: r.ContentLength,

pkg/rhttp/datatx/manager/spaces/spaces.go

@@ -42,6 +42,7 @@ import (
 	"github.com/owncloud/reva/v2/pkg/storage"
 	"github.com/owncloud/reva/v2/pkg/storage/cache"
 	"github.com/owncloud/reva/v2/pkg/storagespace"
+	pkgupload "github.com/owncloud/reva/v2/pkg/upload"
 	"github.com/owncloud/reva/v2/pkg/utils"
 )
 
@@ -80,7 +81,7 @@ func New(m map[string]interface{}, publisher events.Publisher, log *zerolog.Logg
 	}, nil
 }
 
-func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
+func (m *manager) Handler(coord pkgupload.Coordinator, driver storage.FS) (http.Handler, error) {
 	h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		var spaceID string
 		spaceID, r.URL.Path = router.ShiftPath(r.URL.Path)
@@ -97,7 +98,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 					metrics.DownloadsActive.Sub(1)
 				}()
 			}
-			download.GetOrHeadFile(w, r, fs, spaceID)
+			download.GetOrHeadFile(w, r, driver, spaceID)
 		case "PUT":
 			metrics.UploadsActive.Add(1)
 			defer func() {
@@ -117,7 +118,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
 				Path:       fn,
 			}
 			var info *provider.ResourceInfo
-			info, err = fs.Upload(ctx, storage.UploadRequest{
+			info, err = coord.Upload(ctx, storage.UploadRequest{
 				Ref:    ref,
 				Body:   r.Body,
 				Length: r.ContentLength,