ocis: allow setting owner of root (#1225)

cs3org · Oct 7, 2020 · bccddc4 · bccddc4
1 parent a761b31
commit bccddc4
Show file tree

Hide file tree

Showing 5 changed files with 92 additions and 31 deletions.
diff --git a/changelog/unreleased/ocis-set-owner.md b/changelog/unreleased/ocis-set-owner.md
@@ -0,0 +1,5 @@
+Enhancement: Allow setting the owner when using the ocis driver
+
+To support the metadata storage we allow setting the owner of the root node so that subsequent requests with that owner can be used to manage the storage.
+
+https://github.com/cs3org/reva/pull/1225
diff --git a/pkg/storage/fs/ocis/ocis.go b/pkg/storage/fs/ocis/ocis.go
@@ -26,6 +26,7 @@ import (
 	"path/filepath"
 	"strings"
 
+	userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/pkg/appctx"
 	"github.com/cs3org/reva/pkg/errtypes"
@@ -122,6 +123,7 @@ func New(m map[string]interface{}) (storage.FS, error) {
 	}
 	o.init(m)
 
+	// create data paths for internal layout
 	dataPaths := []string{
 		filepath.Join(o.Root, "nodes"),
 		// notes contain symlinks from nodes/<u-u-i-d>/uploads/<uploadid> to ../../uploads/<uploadid>
@@ -141,9 +143,14 @@ func New(m map[string]interface{}) (storage.FS, error) {
 		Options: o,
 	}
 
-	// the root node has an empty name, or use `.` ?
-	// the root node has no parent, or use `root` ?
-	if err = createNode(&Node{lu: lu, ID: "root"}, nil); err != nil {
+	// the root node has an empty name
+	// the root node has no parent
+	if err = createNode(
+		&Node{lu: lu, ID: "root"},
+		&userv1beta1.UserId{
+			OpaqueId: o.Owner,
+		},
+	); err != nil {
 		return nil, err
 	}
 

diff --git a/pkg/storage/fs/ocis/option.go b/pkg/storage/fs/ocis/option.go
@@ -40,6 +40,9 @@ type Options struct {
 
 	// propagate size changes as treesize
 	TreeSizeAccounting bool `mapstructure:"treesize_accounting"`
+
+	// set an owner for the root node
+	Owner string `mapstructure:"owner"`
 }
 
 // newOptions initializes the available default options.

diff --git a/pkg/storage/fs/ocis/tree.go b/pkg/storage/fs/ocis/tree.go
@@ -94,17 +94,27 @@ func (t *Tree) CreateDir(ctx context.Context, node *Node) (err error) {
 	// create a directory node
 	node.ID = uuid.New().String()
 
-	if t.lu.Options.EnableHome {
-		if u, ok := user.ContextGetUser(ctx); ok {
-			err = createNode(node, u.Id)
-		} else {
-			log := appctx.GetLogger(ctx)
-			log.Error().Msg("home support enabled but no user in context")
-			err = errors.Wrap(errtypes.UserRequired("userrequired"), "error getting user from ctx")
-		}
-	} else {
+	// who will become the owner?
+	u, ok := user.ContextGetUser(ctx)
+	switch {
+	case ok:
+		// we have a user in context
+		err = createNode(node, u.Id)
+	case t.lu.Options.EnableHome:
+		// enable home requires a user
+		log := appctx.GetLogger(ctx)
+		log.Error().Msg("home support enabled but no user in context")
+		err = errors.Wrap(errtypes.UserRequired("userrequired"), "error getting user from ctx")
+	case t.lu.Options.Owner != "":
+		// fallback to owner?
+		err = createNode(node, &userpb.UserId{
+			OpaqueId: t.lu.Options.Owner,
+		})
+	default:
+		// fallback to parent owner?
 		err = createNode(node, nil)
 	}
+
 	if err != nil {
 		return nil
 	}

diff --git a/pkg/storage/fs/ocis/upload.go b/pkg/storage/fs/ocis/upload.go
@@ -83,12 +83,14 @@ func (fs *ocisfs) Upload(ctx context.Context, ref *provider.Reference, r io.Read
 	nodePath := fs.lu.toInternalPath(n.ID)
 
 	var tmp *os.File
-	tmp, err = ioutil.TempFile(nodePath, "._reva_atomic_upload")
+	tmp, err = ioutil.TempFile(filepath.Dir(nodePath), "._reva_atomic_upload")
 	if err != nil {
 		return errors.Wrap(err, "ocisfs: error creating tmp fn at "+nodePath)
 	}
 
 	_, err = io.Copy(tmp, r)
+	r.Close()
+	tmp.Close()
 	if err != nil {
 		return errors.Wrap(err, "ocisfs: error writing to tmp file "+tmp.Name())
 	}
@@ -97,25 +99,53 @@ func (fs *ocisfs) Upload(ctx context.Context, ref *provider.Reference, r io.Read
 	//_ = os.RemoveAll(path.Join(nodePath, "content"))
 	appctx.GetLogger(ctx).Warn().Msg("TODO move old content to version")
 
+	appctx.GetLogger(ctx).Debug().Str("tmp", tmp.Name()).Str("ipath", nodePath).Msg("moving tmp to content")
 	if err = os.Rename(tmp.Name(), nodePath); err != nil {
 		return
 	}
 
-	if fs.o.EnableHome {
-		if u, ok := user.ContextGetUser(ctx); ok {
-			err = n.writeMetadata(u.Id)
-		} else {
-			log := appctx.GetLogger(ctx)
-			log.Error().Msg("home support enabled but no user in context")
-			err = errors.Wrap(errtypes.UserRequired("userrequired"), "error getting user from ctx")
-		}
-	} else {
+	// who will become the owner?
+	u, ok := user.ContextGetUser(ctx)
+	switch {
+	case ok:
+		err = n.writeMetadata(u.Id)
+	case fs.o.EnableHome:
+		log := appctx.GetLogger(ctx)
+		log.Error().Msg("home support enabled but no user in context")
+		err = errors.Wrap(errtypes.UserRequired("userrequired"), "error getting user from ctx")
+	case fs.o.Owner != "":
+		err = n.writeMetadata(&userpb.UserId{
+			OpaqueId: fs.o.Owner,
+		})
+	default:
+		// fallback to parent owner?
 		err = n.writeMetadata(nil)
 	}
 	if err != nil {
 		return
 	}
 
+	// link child name to parent if it is new
+	childNameLink := filepath.Join(fs.lu.toInternalPath(n.ParentID), n.Name)
+	var link string
+	link, err = os.Readlink(childNameLink)
+	if err == nil && link != "../"+n.ID {
+		log.Err(err).
+			Interface("node", n).
+			Str("childNameLink", childNameLink).
+			Str("link", link).
+			Msg("ocisfs: child name link has wrong target id, repairing")
+
+		if err = os.Remove(childNameLink); err != nil {
+			return errors.Wrap(err, "ocisfs: could not remove symlink child entry")
+		}
+	}
+	if os.IsNotExist(err) || link != "../"+n.ID {
+		if err = os.Symlink("../"+n.ID, childNameLink); err != nil {
+			return errors.Wrap(err, "ocisfs: could not symlink child entry")
+		}
+	}
+
 	if fs.o.TreeTimeAccounting {
 		// mark the home node as the end of propagation q
 		if err = xattr.Set(nodePath, propagationAttr, []byte("1")); err != nil {
@@ -442,15 +472,21 @@ func (upload *fileUpload) FinishUpload(ctx context.Context) (err error) {
 			Msg("ocisfs: could not rename")
 		return
 	}
-	if upload.fs.o.EnableHome {
-		if u, ok := user.ContextGetUser(upload.ctx); ok {
-			err = n.writeMetadata(u.Id)
-		} else {
-			log := appctx.GetLogger(upload.ctx)
-			log.Error().Msg("home support enabled but no user in context")
-			err = errors.Wrap(errtypes.UserRequired("userrequired"), "error getting user from ctx")
-		}
-	} else {
+	// who will become the owner?
+	u, ok := user.ContextGetUser(upload.ctx)
+	switch {
+	case ok:
+		err = n.writeMetadata(u.Id)
+	case upload.fs.o.EnableHome:
+		log := appctx.GetLogger(upload.ctx)
+		log.Error().Msg("home support enabled but no user in context")
+		err = errors.Wrap(errtypes.UserRequired("userrequired"), "error getting user from upload ctx")
+	case upload.fs.o.Owner != "":
+		err = n.writeMetadata(&userpb.UserId{
+			OpaqueId: upload.fs.o.Owner,
+		})
+	default:
+		// fallback to parent owner?
 		err = n.writeMetadata(nil)
 	}
 	if err != nil {