feat: add NewCoordinatorFromConfig

Author: LarsJurgensen

internal/grpc/services/storageprovider/storageprovider.go

@@ -212,9 +212,10 @@ func New(m map[string]interface{}, ss *grpc.Server, log *zerolog.Logger) (rgrpc.
 		return nil, err
 	}
 
-	coord, err := getCoordinator(c, fs, evstream, log)
+	// storageprovider only initiates uploads; the data path assembles chunks, so no chunking here.
+	coord, err := upload.NewCoordinatorFromConfig(c.UploadDirectory, c.Drivers[c.Driver], fs, evstream, log, false)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("storageprovider: %w", err)
 	}
 
 	service := &Service{
@@ -228,22 +229,6 @@ func New(m map[string]interface{}, ss *grpc.Server, log *zerolog.Logger) (rgrpc.
 	return service, nil
 }
 
-// getCoordinator builds the coordinator that initiates uploads for the driver
-// this service mounts. It stages sessions in the same directory the dataprovider
-// appends bytes to, so an upload initiated here can be continued there.
-func getCoordinator(c *config, fs storage.FS, publisher events.Publisher, log *zerolog.Logger) (upload.Coordinator, error) {
-	store := upload.NewFileStoreFromConfig(c.UploadDirectory, c.Drivers[c.Driver], log)
-	if store == nil {
-		return nil, fmt.Errorf("storageprovider: cannot determine the upload directory, set upload_directory")
-	}
-	if err := store.Setup(); err != nil {
-		return nil, fmt.Errorf("storageprovider: upload directory setup failed: %w", err)
-	}
-
-	// No chunk folder: only the data path assembles chunks.
-	return upload.NewCoordinator(fs, store, "", publisher), nil
-}
-
 func (s *Service) SetArbitraryMetadata(ctx context.Context, req *provider.SetArbitraryMetadataRequest) (*provider.SetArbitraryMetadataResponse, error) {
 	ctx = ctxpkg.ContextSetLockID(ctx, req.LockId)
 

internal/http/services/dataprovider/dataprovider.go

@@ -106,9 +106,10 @@ func New(m map[string]interface{}, log *zerolog.Logger) (global.Service, error)
 		return nil, err
 	}
 
-	coord, err := getCoordinator(conf, fs, evstream, log)
+	// the data path assembles chunks, so enable chunking
+	coord, err := upload.NewCoordinatorFromConfig(conf.UploadDirectory, conf.Drivers[conf.Driver], fs, evstream, log, true)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("dataprovider: %w", err)
 	}
 
 	// only the data path consumes postprocessing results: one consumer group gets
@@ -141,19 +142,6 @@ func getFS(c *config, stream events.Stream, log *zerolog.Logger) (storage.FS, er
 	return nil, fmt.Errorf("driver not found: %s", c.Driver)
 }
 
-// getCoordinator builds the coordinator that owns the upload lifecycle for the
-// driver this service mounts.
-func getCoordinator(c *config, fs storage.FS, publisher events.Publisher, log *zerolog.Logger) (upload.Coordinator, error) {
-	store := upload.NewFileStoreFromConfig(c.UploadDirectory, c.Drivers[c.Driver], log)
-	if store == nil {
-		return nil, fmt.Errorf("dataprovider: cannot determine the upload directory, set upload_directory")
-	}
-	if err := store.Setup(); err != nil {
-		return nil, fmt.Errorf("dataprovider: upload directory setup failed: %w", err)
-	}
-	return upload.NewCoordinator(fs, store, store.UploadDir(), publisher), nil
-}
-
 func getDataTXs(c *config, coord upload.Coordinator, fs 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{})

pkg/upload/coordinator.go

@@ -11,6 +11,7 @@ import (
 	user "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/rs/zerolog"
 	tusd "github.com/tus/tusd/v2/pkg/handler"
 
 	"github.com/owncloud/reva/v2/pkg/appctx"
@@ -62,6 +63,24 @@ func NewCoordinator(fs storage.FS, store SessionStore, chunkFolder string, pub e
 	return c
 }
 
+// NewCoordinatorFromConfig sets up a coordinator and its file store. Pass
+// withChunking=false when only initiating uploads: chunk assembly happens on the
+// data path, so the storageprovider doesn't need it.
+func NewCoordinatorFromConfig(uploadDir string, driverConf map[string]interface{}, fs storage.FS, pub events.Publisher, log *zerolog.Logger, withChunking bool) (Coordinator, error) {
+	store := NewFileStoreFromConfig(uploadDir, driverConf, log)
+	if store == nil {
+		return nil, fmt.Errorf("cannot determine the upload directory, set upload_directory")
+	}
+	if err := store.Setup(); err != nil {
+		return nil, fmt.Errorf("upload directory setup failed: %w", err)
+	}
+	chunkFolder := ""
+	if withChunking {
+		chunkFolder = store.UploadDir()
+	}
+	return NewCoordinator(fs, store, chunkFolder, pub), nil
+}
+
 // InitiateUpload resolves the target, then creates and persists the session that
 // bytes are appended to.
 func (c *coordinator) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) {