Report quota per storage space (#2152)

changelog/unreleased/get-quota-storage-space.md

@@ -0,0 +1,7 @@
+Enhancement: Add a reference parameter to the getQuota request
+
+Implementation of [cs3org/cs3apis#147](https://github.com/cs3org/cs3apis/pull/147)
+
+Make the cs3apis accept a Reference in the getQuota Request to limit the call to a specific storage space.
+
+https://github.com/cs3org/reva/issues/2152

internal/grpc/services/gateway/storageprovider.go

@@ -2135,7 +2135,7 @@ func (s *svc) GetQuota(ctx context.Context, req *gateway.GetQuotaRequest) (*prov
 
 	res, err := c.GetQuota(ctx, &provider.GetQuotaRequest{
 		Opaque: req.GetOpaque(),
-		// Ref:    req.GetRef(), // TODO send which storage space ... or root
+		Ref:    req.GetRef(),
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "gateway: error calling GetQuota")

internal/grpc/services/storageprovider/storageprovider.go

@@ -1181,7 +1181,13 @@ func (s *service) CreateSymlink(ctx context.Context, req *provider.CreateSymlink
 }
 
 func (s *service) GetQuota(ctx context.Context, req *provider.GetQuotaRequest) (*provider.GetQuotaResponse, error) {
-	total, used, err := s.storage.GetQuota(ctx)
+	newRef, err := s.unwrap(ctx, req.Ref)
+	if err != nil {
+		return &provider.GetQuotaResponse{
+			Status: status.NewInternal(ctx, err, "error unwrapping path"),
+		}, nil
+	}
+	total, used, err := s.storage.GetQuota(ctx, newRef)
 	if err != nil {
 		var st *rpc.Status
 		switch err.(type) {

pkg/storage/fs/nextcloud/nextcloud.go

@@ -679,7 +679,7 @@ func (nc *StorageDriver) ListGrants(ctx context.Context, ref *provider.Reference
 }
 
 // GetQuota as defined in the storage.FS interface
-func (nc *StorageDriver) GetQuota(ctx context.Context) (uint64, uint64, error) {
+func (nc *StorageDriver) GetQuota(ctx context.Context, ref *provider.Reference) (uint64, uint64, error) {
 	log := appctx.GetLogger(ctx)
 	log.Info().Msg("GetQuota")
 

pkg/storage/fs/nextcloud/nextcloud_test.go

@@ -881,7 +881,7 @@ var _ = Describe("Nextcloud", func() {
 		It("calls the GetQuota endpoint", func() {
 			nc, called, teardown := setUpNextcloudServer()
 			defer teardown()
-			maxBytes, maxFiles, err := nc.GetQuota(ctx)
+			maxBytes, maxFiles, err := nc.GetQuota(ctx, nil)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(maxBytes).To(Equal(uint64(456)))
 			Expect(maxFiles).To(Equal(uint64(123)))

pkg/storage/fs/owncloud/owncloud_unix.go

@@ -30,6 +30,7 @@ import (
 	"strings"
 	"syscall"
 
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/pkg/appctx"
 )
 
@@ -68,7 +69,7 @@ func calcEtag(ctx context.Context, fi os.FileInfo) string {
 	return fmt.Sprintf("\"%s\"", strings.Trim(etag, "\""))
 }
 
-func (fs *ocfs) GetQuota(ctx context.Context) (uint64, uint64, error) {
+func (fs *ocfs) GetQuota(ctx context.Context, ref *provider.Reference) (uint64, uint64, error) {
 	// TODO quota of which storage space?
 	// we could use the logged in user, but when a user has access to multiple storages this falls short
 	// for now return quota of root

pkg/storage/fs/owncloud/owncloud_windows.go

@@ -29,6 +29,7 @@ import (
 	"os"
 	"strings"
 
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/pkg/appctx"
 	"golang.org/x/sys/windows"
 )
@@ -55,7 +56,7 @@ func calcEtag(ctx context.Context, fi os.FileInfo) string {
 	return fmt.Sprintf("\"%s\"", strings.Trim(etag, "\""))
 }
 
-func (fs *ocfs) GetQuota(ctx context.Context) (uint64, uint64, error) {
+func (fs *ocfs) GetQuota(ctx context.Context, ref *provider.Reference) (uint64, uint64, error) {
 	// TODO quota of which storage space?
 	// we could use the logged in user, but when a user has access to multiple storages this falls short
 	// for now return quota of root

pkg/storage/fs/owncloudsql/owncloudsql_unix.go

@@ -30,6 +30,7 @@ import (
 	"strings"
 	"syscall"
 
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/pkg/appctx"
 )
 
@@ -68,7 +69,7 @@ func calcEtag(ctx context.Context, fi os.FileInfo) string {
 	return strings.Trim(etag, "\"")
 }
 
-func (fs *owncloudsqlfs) GetQuota(ctx context.Context) (uint64, uint64, error) {
+func (fs *owncloudsqlfs) GetQuota(ctx context.Context, ref *provider.Reference) (uint64, uint64, error) {
 	// TODO quota of which storage space?
 	// we could use the logged in user, but when a user has access to multiple storages this falls short
 	// for now return quota of root

pkg/storage/fs/owncloudsql/owncloudsql_windows.go

@@ -29,6 +29,7 @@ import (
 	"os"
 	"strings"
 
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/pkg/appctx"
 	"golang.org/x/sys/windows"
 )
@@ -55,7 +56,7 @@ func calcEtag(ctx context.Context, fi os.FileInfo) string {
 	return fmt.Sprintf("\"%s\"", strings.Trim(etag, "\""))
 }
 
-func (fs *owncloudsqlfs) GetQuota(ctx context.Context) (uint64, uint64, error) {
+func (fs *owncloudsqlfs) GetQuota(ctx context.Context, ref *provider.Reference) (uint64, uint64, error) {
 	// TODO quota of which storage space?
 	// we could use the logged in user, but when a user has access to multiple storages this falls short
 	// for now return quota of root

pkg/storage/fs/s3/s3.go

@@ -268,7 +268,7 @@ func (fs *s3FS) UpdateGrant(ctx context.Context, ref *provider.Reference, g *pro
 	return errtypes.NotSupported("s3: operation not supported")
 }
 
-func (fs *s3FS) GetQuota(ctx context.Context) (uint64, uint64, error) {
+func (fs *s3FS) GetQuota(ctx context.Context, ref *provider.Reference) (uint64, uint64, error) {
 	return 0, 0, nil
 }
 

pkg/storage/storage.go

@@ -52,7 +52,7 @@ type FS interface {
 	RemoveGrant(ctx context.Context, ref *provider.Reference, g *provider.Grant) error
 	UpdateGrant(ctx context.Context, ref *provider.Reference, g *provider.Grant) error
 	ListGrants(ctx context.Context, ref *provider.Reference) ([]*provider.Grant, error)
-	GetQuota(ctx context.Context) ( /*TotalBytes*/ uint64 /*UsedBytes*/, uint64, error)
+	GetQuota(ctx context.Context, ref *provider.Reference) ( /*TotalBytes*/ uint64 /*UsedBytes*/, uint64, error)
 	CreateReference(ctx context.Context, path string, targetURI *url.URL) error
 	Shutdown(ctx context.Context) error
 	SetArbitraryMetadata(ctx context.Context, ref *provider.Reference, md *provider.ArbitraryMetadata) error

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -128,10 +128,16 @@ func (fs *Decomposedfs) Shutdown(ctx context.Context) error {
 
 // GetQuota returns the quota available
 // TODO Document in the cs3 should we return quota or free space?
-func (fs *Decomposedfs) GetQuota(ctx context.Context) (total uint64, inUse uint64, err error) {
+func (fs *Decomposedfs) GetQuota(ctx context.Context, ref *provider.Reference) (total uint64, inUse uint64, err error) {
 	var n *node.Node
-	if n, err = fs.lu.HomeOrRootNode(ctx); err != nil {
-		return 0, 0, err
+	if ref != nil {
+		if n, err = fs.lu.NodeFromResource(ctx, ref); err != nil {
+			return 0, 0, err
+		}
+	} else {
+		if n, err = fs.lu.HomeOrRootNode(ctx); err != nil {
+			return 0, 0, err
+		}
 	}
 
 	if !n.Exists {
@@ -206,7 +212,7 @@ func (fs *Decomposedfs) CreateHome(ctx context.Context) (err error) {
 		return
 	}
 
-	if fs.o.TreeTimeAccounting {
+	if fs.o.TreeTimeAccounting || fs.o.TreeSizeAccounting {
 		homePath := h.InternalPath()
 		// mark the home node as the end of propagation
 		if err = xattr.Set(homePath, xattrs.PropagationAttr, []byte("1")); err != nil {
@@ -215,6 +221,10 @@ func (fs *Decomposedfs) CreateHome(ctx context.Context) (err error) {
 		}
 	}
 
+	if err := n.SetMetadata(xattrs.SpaceNameAttr, u.DisplayName); err != nil {
+		return err
+	}
+
 	// add storage space
 	if err := fs.createStorageSpace(ctx, "personal", h.ID); err != nil {
 		return err
@@ -289,7 +299,7 @@ func (fs *Decomposedfs) CreateDir(ctx context.Context, ref *provider.Reference)
 
 	err = fs.tp.CreateDir(ctx, n)
 
-	if fs.o.TreeTimeAccounting {
+	if fs.o.TreeTimeAccounting || fs.o.TreeSizeAccounting {
 		nodePath := n.InternalPath()
 		// mark the home node as the end of propagation
 		if err = xattr.Set(nodePath, xattrs.PropagationAttr, []byte("1")); err != nil {

pkg/storage/utils/decomposedfs/lookup.go

@@ -45,23 +45,26 @@ func (lu *Lookup) NodeFromResource(ctx context.Context, ref *provider.Reference)
 	if ref.ResourceId != nil {
 		// check if a storage space reference is used
 		// currently, the decomposed fs uses the root node id as the space id
-		n, err := lu.NodeFromID(ctx, ref.ResourceId)
+		spaceRoot, err := lu.NodeFromID(ctx, ref.ResourceId)
 		if err != nil {
 			return nil, err
 		}
-
-		p := filepath.Clean(ref.Path)
-		if p != "." {
-			// walk the relative path
-			n, err = lu.WalkPath(ctx, n, p, false, func(ctx context.Context, n *node.Node) error {
-				return nil
-			})
-			if err != nil {
-				return nil, err
+		n := spaceRoot
+		// is this a relative reference?
+		if ref.Path != "" {
+			p := filepath.Clean(ref.Path)
+			if p != "." {
+				// walk the relative path
+				n, err = lu.WalkPath(ctx, n, p, false, func(ctx context.Context, n *node.Node) error {
+					return nil
+				})
+				if err != nil {
+					return nil, err
+				}
 			}
-			return n, nil
+			// use reference id as space root for relative references
+			n.SpaceRoot = spaceRoot
 		}
-
 		return n, nil
 	}
 
@@ -78,23 +81,27 @@ func (lu *Lookup) NodeFromPath(ctx context.Context, fn string, followReferences
 	log := appctx.GetLogger(ctx)
 	log.Debug().Interface("fn", fn).Msg("NodeFromPath()")
 
-	n, err := lu.HomeOrRootNode(ctx)
+	root, err := lu.HomeOrRootNode(ctx)
 	if err != nil {
 		return nil, err
 	}
 
+	n := root
 	// TODO collect permissions of the current user on every segment
 	fn = filepath.Clean(fn)
 	if fn != "/" && fn != "." {
 		n, err = lu.WalkPath(ctx, n, fn, followReferences, func(ctx context.Context, n *node.Node) error {
 			log.Debug().Interface("node", n).Msg("NodeFromPath() walk")
+			if n.SpaceRoot != nil && n.SpaceRoot != root {
+				root = n.SpaceRoot
+			}
 			return nil
 		})
 		if err != nil {
 			return nil, err
 		}
 	}
-
+	n.SpaceRoot = root
 	return n, nil
 }
 
@@ -129,7 +136,9 @@ func (lu *Lookup) Path(ctx context.Context, n *node.Node) (p string, err error)
 
 // RootNode returns the root node of the storage
 func (lu *Lookup) RootNode(ctx context.Context) (*node.Node, error) {
-	return node.New("root", "", "", 0, "", nil, lu), nil
+	n := node.New("root", "", "", 0, "", nil, lu)
+	n.Exists = true
+	return n, nil
 }
 
 // HomeNode returns the home node of a user

pkg/storage/utils/decomposedfs/node/node.go

@@ -65,13 +65,14 @@ const (
 
 // Node represents a node in the tree and provides methods to get a Parent or Child instance
 type Node struct {
-	ParentID string
-	ID       string
-	Name     string
-	Blobsize int64
-	BlobID   string
-	owner    *userpb.UserId
-	Exists   bool
+	ParentID  string
+	ID        string
+	Name      string
+	Blobsize  int64
+	BlobID    string
+	owner     *userpb.UserId
+	Exists    bool
+	SpaceRoot *Node
 
 	lu PathLookup
 }
@@ -191,6 +192,10 @@ func ReadNode(ctx context.Context, lu PathLookup, id string) (n *Node, err error
 	default:
 		return nil, errtypes.InternalError(err.Error())
 	}
+	// check if this is a space root
+	if _, err = xattr.Get(nodePath, xattrs.SpaceNameAttr); err == nil {
+		n.SpaceRoot = n
+	}
 	// lookup name in extended attributes
 	if attrBytes, err = xattr.Get(nodePath, xattrs.NameAttr); err == nil {
 		n.Name = string(attrBytes)
@@ -239,9 +244,10 @@ func (n *Node) Child(ctx context.Context, name string) (*Node, error) {
 	if err != nil {
 		if os.IsNotExist(err) || isNotDir(err) {
 			c := &Node{
-				lu:       n.lu,
-				ParentID: n.ID,
-				Name:     name,
+				lu:        n.lu,
+				ParentID:  n.ID,
+				Name:      name,
+				SpaceRoot: n.SpaceRoot,
 			}
 			return c, nil // if the file does not exist we return a node that has Exists = false
 		}
@@ -268,8 +274,9 @@ func (n *Node) Parent() (p *Node, err error) {
 		return nil, fmt.Errorf("Decomposedfs: root has no parent")
 	}
 	p = &Node{
-		lu: n.lu,
-		ID: n.ParentID,
+		lu:        n.lu,
+		ID:        n.ParentID,
+		SpaceRoot: n.SpaceRoot,
 	}
 
 	parentPath := n.lu.InternalPath(n.ParentID)
@@ -608,11 +615,18 @@ func (n *Node) AsResourceInfo(ctx context.Context, rp *provider.ResourcePermissi
 	// quota
 	if _, ok := mdKeysMap[QuotaKey]; (nodeType == provider.ResourceType_RESOURCE_TYPE_CONTAINER) && returnAllKeys || ok {
 		var quotaPath string
-		if r, err := n.lu.HomeOrRootNode(ctx); err == nil {
-			quotaPath = r.InternalPath()
-			readQuotaIntoOpaque(ctx, quotaPath, ri)
+		if n.SpaceRoot == nil {
+			root, err := n.lu.HomeOrRootNode(ctx)
+			if err == nil {
+				quotaPath = root.InternalPath()
+			} else {
+				sublog.Debug().Err(err).Msg("error determining the space root node for quota")
+			}
 		} else {
-			sublog.Error().Err(err).Msg("error determining home or root node for quota")
+			quotaPath = n.SpaceRoot.InternalPath()
+		}
+		if quotaPath != "" {
+			readQuotaIntoOpaque(ctx, quotaPath, ri)
 		}
 	}
 

pkg/storage/utils/decomposedfs/spaces.go

@@ -66,6 +66,14 @@ func (fs *Decomposedfs) CreateStorageSpace(ctx context.Context, req *provider.Cr
 		return nil, err
 	}
 
+	// always enable propagation on the storage space root
+	nodePath := n.InternalPath()
+	// mark the space root node as the end of propagation
+	if err = xattr.Set(nodePath, xattrs.PropagationAttr, []byte("1")); err != nil {
+		appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Msg("could not mark node to propagate")
+		return nil, err
+	}
+
 	if err := fs.createHiddenSpaceFolder(ctx, n); err != nil {
 		return nil, err
 	}

pkg/storage/utils/decomposedfs/tree/tree.go

@@ -578,8 +578,12 @@ func (t *Tree) Propagate(ctx context.Context, n *node.Node) (err error) {
 	// is propagation enabled for the parent node?
 
 	var root *node.Node
-	if root, err = t.lookup.HomeOrRootNode(ctx); err != nil {
-		return
+	if n.SpaceRoot == nil {
+		if root, err = t.lookup.HomeOrRootNode(ctx); err != nil {
+			return
+		}
+	} else {
+		root = n.SpaceRoot
 	}
 
 	// use a sync time and don't rely on the mtime of the current node, as the stat might not change when a rename happened too quickly