Skip to content

Commit 5cd2fbc

Browse files
committed
add empty upload coordinator skeleton
1 parent 5fe82c1 commit 5cd2fbc

17 files changed

Lines changed: 2693 additions & 72 deletions

File tree

internal/grpc/services/storageprovider/storageprovider.go

Lines changed: 26 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import (
2525
"net/url"
2626
"os"
2727
"path"
28+
"path/filepath"
2829
"sort"
2930
"strconv"
3031
"strings"
@@ -47,6 +48,7 @@ import (
4748
"github.com/owncloud/reva/v2/pkg/storage"
4849
"github.com/owncloud/reva/v2/pkg/storage/fs/registry"
4950
"github.com/owncloud/reva/v2/pkg/storagespace"
51+
"github.com/owncloud/reva/v2/pkg/upload"
5052
"github.com/owncloud/reva/v2/pkg/utils"
5153
"github.com/pkg/errors"
5254
"github.com/rs/zerolog"
@@ -69,6 +71,7 @@ type config struct {
6971
AvailableXS map[string]uint32 `mapstructure:"available_checksums" docs:"nil;List of available checksums."`
7072
CustomMimeTypesJSON string `mapstructure:"custom_mimetypes_json" docs:"nil;An optional mapping file with the list of supported custom file extensions and corresponding mime types."`
7173
MountID string `mapstructure:"mount_id"`
74+
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."`
7275
UploadExpiration int64 `mapstructure:"upload_expiration" docs:"0;Duration for how long uploads will be valid."`
7376
Events eventconfig `mapstructure:"events" docs:"0;Event stream configuration"`
7477
}
@@ -106,6 +109,7 @@ func (c *config) init() {
106109
type Service struct {
107110
conf *config
108111
Storage storage.FS
112+
Coordinator upload.Coordinator
109113
dataServerURL *url.URL
110114
availableXS []*provider.ResourceChecksumPriority
111115
}
@@ -175,11 +179,29 @@ func New(m map[string]interface{}, ss *grpc.Server, log *zerolog.Logger) (rgrpc.
175179

176180
c.init()
177181

178-
fs, err := getFS(c, log)
182+
evstream, err := estreamFromConfig(c.Events)
179183
if err != nil {
180184
return nil, err
181185
}
182186

187+
fs, err := getFS(c, evstream, log)
188+
if err != nil {
189+
return nil, err
190+
}
191+
192+
// Build the coordinator-owned session store. UploadDirectory (service level)
193+
// takes precedence over the driver root, so rootless drivers can still get a
194+
// coordinator. The store points at the same root as the driver's data path so
195+
// it can read the sessions the coordinator writes (decomposedfs on-disk format).
196+
store := upload.NewFileStoreFromConfig(c.UploadDirectory, c.Drivers[c.Driver], log)
197+
if store == nil {
198+
return nil, fmt.Errorf("storageprovider: cannot determine upload directory, set upload_directory in config or driver root")
199+
}
200+
if err := store.Setup(); err != nil {
201+
return nil, fmt.Errorf("storageprovider: upload directory setup failed: %w", err)
202+
}
203+
coordinator := upload.NewCoordinator(fs, store, filepath.Join(store.Root(), "uploads"), evstream)
204+
183205
// parse data server url
184206
u, err := url.Parse(c.DataServerURL)
185207
if err != nil {
@@ -205,6 +227,7 @@ func New(m map[string]interface{}, ss *grpc.Server, log *zerolog.Logger) (rgrpc.
205227
service := &Service{
206228
conf: c,
207229
Storage: fs,
230+
Coordinator: coordinator,
208231
dataServerURL: u,
209232
availableXS: xsTypes,
210233
}
@@ -427,7 +450,7 @@ func (s *Service) InitiateFileUpload(ctx context.Context, req *provider.Initiate
427450
metadata["expires"] = strconv.Itoa(int(expirationTimestamp.Seconds))
428451
}
429452

430-
uploadIDs, err := s.Storage.InitiateUpload(ctx, req.Ref, uploadLength, metadata)
453+
uploadIDs, err := s.Coordinator.InitiateUpload(ctx, req.Ref, uploadLength, metadata)
431454
if err != nil {
432455
var st *rpc.Status
433456
switch err.(type) {
@@ -1266,12 +1289,7 @@ func (s *Service) addMissingStorageProviderID(resourceID *provider.ResourceId, s
12661289
}
12671290
}
12681291

1269-
func getFS(c *config, log *zerolog.Logger) (storage.FS, error) {
1270-
evstream, err := estreamFromConfig(c.Events)
1271-
if err != nil {
1272-
return nil, err
1273-
}
1274-
1292+
func getFS(c *config, evstream events.Stream, log *zerolog.Logger) (storage.FS, error) {
12751293
if f, ok := registry.NewFuncs[c.Driver]; ok {
12761294
driverConf := c.Drivers[c.Driver]
12771295
driverConf["mount_id"] = c.MountID // pass the mount id to the driver

internal/http/services/dataprovider/dataprovider.go

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ package dataprovider
2121
import (
2222
"fmt"
2323
"net/http"
24+
"path/filepath"
2425

2526
"github.com/mitchellh/mapstructure"
2627
"github.com/rs/zerolog"
@@ -33,6 +34,7 @@ import (
3334
"github.com/owncloud/reva/v2/pkg/rhttp/router"
3435
"github.com/owncloud/reva/v2/pkg/storage"
3536
"github.com/owncloud/reva/v2/pkg/storage/fs/registry"
37+
pkgupload "github.com/owncloud/reva/v2/pkg/upload"
3638
)
3739

3840
func init() {
@@ -51,6 +53,7 @@ type config struct {
5153
NatsEnableTLS bool `mapstructure:"nats_enable_tls"`
5254
NatsUsername string `mapstructure:"nats_username"`
5355
NatsPassword string `mapstructure:"nats_password"`
56+
UploadDirectory string `mapstructure:"upload_directory" docs:";Local directory for staging upload sessions. Overrides the driver root. Required for drivers without a local root."`
5457
}
5558

5659
func (c *config) init() {
@@ -104,7 +107,18 @@ func New(m map[string]interface{}, log *zerolog.Logger) (global.Service, error)
104107
return nil, err
105108
}
106109

107-
dataTXs, err := getDataTXs(conf, fs, evstream, log)
110+
// The data provider finishes uploads that the storage provider initiated, so
111+
// its store must resolve to the same upload directory.
112+
store := pkgupload.NewFileStoreFromConfig(conf.UploadDirectory, conf.Drivers[conf.Driver], log)
113+
if store == nil {
114+
return nil, fmt.Errorf("dataprovider: cannot determine upload directory, set upload_directory in config or driver root")
115+
}
116+
if err := store.Setup(); err != nil {
117+
return nil, fmt.Errorf("dataprovider: upload directory setup failed: %w", err)
118+
}
119+
coord := pkgupload.NewCoordinator(fs, store, filepath.Join(store.Root(), "uploads"), evstream)
120+
121+
dataTXs, err := getDataTXs(conf, coord, fs, evstream, log)
108122
if err != nil {
109123
return nil, err
110124
}
@@ -126,7 +140,7 @@ func getFS(c *config, stream events.Stream, log *zerolog.Logger) (storage.FS, er
126140
return nil, fmt.Errorf("driver not found: %s", c.Driver)
127141
}
128142

129-
func getDataTXs(c *config, fs storage.FS, publisher events.Publisher, log *zerolog.Logger) (map[string]http.Handler, error) {
143+
func getDataTXs(c *config, coord pkgupload.Coordinator, driver storage.FS, publisher events.Publisher, log *zerolog.Logger) (map[string]http.Handler, error) {
130144
if c.DataTXs == nil {
131145
c.DataTXs = make(map[string]map[string]interface{})
132146
}
@@ -146,7 +160,7 @@ func getDataTXs(c *config, fs storage.FS, publisher events.Publisher, log *zerol
146160
for t := range c.DataTXs {
147161
if f, ok := datatxregistry.NewFuncs[t]; ok {
148162
if tx, err := f(c.DataTXs[t], publisher, log); err == nil {
149-
if handler, err := tx.Handler(fs); err == nil {
163+
if handler, err := tx.Handler(coord, driver); err == nil {
150164
txs[t] = handler
151165
}
152166
}

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -319,7 +319,11 @@ func (s *svc) handleTusPost(ctx context.Context, w http.ResponseWriter, r *http.
319319
sReq.Ref.Path = uReq.Ref.GetPath()
320320
sReq.Ref.ResourceId = nil
321321
} else {
322-
if resid, err := storagespace.ParseID(httpRes.Header.Get(net.HeaderOCFileID)); err == nil {
322+
// A new file has no node id until the upload finishes, which is after the
323+
// data server wrote this header, so it may be absent. sReq.Ref then keeps
324+
// the path-based reference it was built with, which already names the
325+
// file and stats just as well.
326+
if resid, err := storagespace.ParseID(httpRes.Header.Get(net.HeaderOCFileID)); err == nil && resid.GetOpaqueId() != "" {
323327
sReq.Ref = &provider.Reference{
324328
ResourceId: &resid,
325329
}

pkg/ocm/storage/received/ocm.go

Lines changed: 77 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -253,7 +253,10 @@ func (d *driver) CreateDir(ctx context.Context, ref *provider.Reference) (*stora
253253
}
254254

255255
func (d *driver) Delete(ctx context.Context, ref *provider.Reference) (*storage.DeleteResult, error) {
256-
client, _, rel, err := d.webdavClient(ctx, nil, ref)
256+
// The coordinator rolls back a failed upload from the tusd finish path, which
257+
// carries no request auth token, so the share must be resolved as the service
258+
// account on the executant's behalf.
259+
client, _, rel, err := d.serviceWebdavClient(ctx, ref)
257260
if err != nil {
258261
return nil, err
259262
}
@@ -264,14 +267,28 @@ func (d *driver) Delete(ctx context.Context, ref *provider.Reference) (*storage.
264267
}
265268

266269
func (d *driver) TouchFile(ctx context.Context, ref *provider.Reference, markprocessing bool, mtime string) (*storage.TouchFileResult, error) {
267-
client, _, rel, err := d.webdavClient(ctx, nil, ref)
270+
// The coordinator calls this from the upload finish path, which runs under
271+
// tusd with no request auth token, so the share must be resolved as the
272+
// service account on the executant's behalf.
273+
client, _, rel, err := d.serviceWebdavClient(ctx, ref)
268274
if err != nil {
269275
return nil, err
270276
}
271277
if err := client.Write(rel, []byte{}, 0); err != nil {
272278
return nil, err
273279
}
274-
return &storage.TouchFileResult{}, nil
280+
// The coordinator threads ResourceID into the session so the following
281+
// CommitUpload can address the file, so return the real ids rather than an
282+
// empty result. rel is the remote path, encoded the way GetMD encodes it.
283+
shareID, _ := shareInfoFromReference(ref)
284+
return &storage.TouchFileResult{
285+
SpaceID: shareID.GetOpaqueId(),
286+
ResourceID: &provider.ResourceId{
287+
StorageId: utils.OCMStorageProviderID,
288+
SpaceId: shareID.GetOpaqueId(),
289+
OpaqueId: base64.StdEncoding.EncodeToString([]byte(filepath.Join("/", rel))),
290+
},
291+
}, nil
275292
}
276293

277294
func (d *driver) Move(ctx context.Context, oldRef, newRef *provider.Reference) (*storage.MoveResult, error) {
@@ -347,6 +364,44 @@ func convertStatToResourceInfo(ref *provider.Reference, f fs.FileInfo, share *oc
347364
return &ri, nil
348365
}
349366

367+
// extractLock reads the DAV:lockdiscovery property and returns the active lock,
368+
// or nil when the resource carries no usable lock.
369+
//
370+
// gowebdav's Props.GetString formats a missing key as the string "<nil>", so an
371+
// absent lockdiscovery cannot be told apart by emptiness. Instead we require the
372+
// value to parse as <d:activelock> and to carry a lock token: a lock we cannot
373+
// name is one the caller could never match against its own, so treating it as
374+
// absent is the only useful reading.
375+
func extractLock(props gowebdav.Props) *provider.Lock {
376+
raw := props.GetString(xml.Name{Space: "DAV:", Local: "lockdiscovery"})
377+
378+
// Element names are matched on their local part only: the value is the raw
379+
// innerxml of lockdiscovery, so the xmlns:d declaration that bound the "d"
380+
// prefix stayed behind on the enclosing element and the prefix is unbound
381+
// here.
382+
var al struct {
383+
LockScope struct {
384+
Exclusive *struct{} `xml:"exclusive"`
385+
Shared *struct{} `xml:"shared"`
386+
} `xml:"lockscope"`
387+
LockToken struct {
388+
Href string `xml:"href"`
389+
} `xml:"locktoken"`
390+
}
391+
if err := xml.Unmarshal([]byte(raw), &al); err != nil {
392+
return nil
393+
}
394+
if al.LockToken.Href == "" {
395+
return nil
396+
}
397+
398+
lockType := provider.LockType_LOCK_TYPE_EXCL
399+
if al.LockScope.Shared != nil && al.LockScope.Exclusive == nil {
400+
lockType = provider.LockType_LOCK_TYPE_SHARED
401+
}
402+
return &provider.Lock{LockId: al.LockToken.Href, Type: lockType}
403+
}
404+
350405
func extractChecksum(props gowebdav.Props) *provider.ResourceChecksum {
351406
checksums := props.GetString(xml.Name{Space: "http://owncloud.org/ns", Local: "checksums"})
352407
if checksums == "" {
@@ -534,12 +589,29 @@ func (d *driver) GetLock(ctx context.Context, ref *provider.Reference) (*provide
534589
return nil, err
535590
}
536591

537-
token, err := client.GetLock(rel)
592+
// gowebdav's GetLock cannot be used: it stats with a fixed property set that
593+
// omits lockdiscovery, so it reports the same value whatever the lock state.
594+
// getetag is requested alongside because an unlocked resource returns
595+
// lockdiscovery in a 404 propstat, and gowebdav only reads the 200 one — with
596+
// no other property to carry it, that propstat would be empty and the stat
597+
// would fail as if the file were missing.
598+
info, err := client.StatWithProps(rel, []string{"lockdiscovery", "getetag"})
538599
if err != nil {
539600
return nil, err
540601
}
541602

542-
return &provider.Lock{LockId: token, Type: provider.LockType_LOCK_TYPE_EXCL}, nil
603+
props, ok := info.Sys().(gowebdav.Props)
604+
if !ok {
605+
return nil, errtypes.InternalError("ocm: unexpected stat result for " + ref.GetPath())
606+
}
607+
608+
lock := extractLock(props)
609+
if lock == nil {
610+
// Callers distinguish "locked" from "not locked" by testing the returned
611+
// lock against nil, so a lock must never be invented here.
612+
return nil, errtypes.NotFound("no lock found")
613+
}
614+
return lock, nil
543615
}
544616

545617
func (d *driver) RefreshLock(ctx context.Context, ref *provider.Reference, lock *provider.Lock, existingLockID string) error {

0 commit comments

Comments
 (0)