Store extended attributes in files in stead of in extended attributes

changelog/unreleased/materialized-xattrs.md

@@ -0,0 +1,5 @@
+Enhancement: Introduce ini file based metadata backend
+
+We added a new metadata backend for the decomposed storage driver that uses an additional `.ini` file to store file metadata. This allows scaling beyond some filesystem specific xattr limitations.
+
+https://github.com/cs3org/reva/pull/3649

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -24,6 +24,7 @@ package decomposedfs
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"net/url"
 	"os"
@@ -51,6 +52,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs/prefixes"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/filelocks"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/templates"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
@@ -101,6 +103,15 @@ func NewDefault(m map[string]interface{}, bs tree.Blobstore, es events.Stream) (
 		return nil, err
 	}
 
+	switch o.MetadataBackend {
+	case "xattrs":
+		xattrs.UseXattrsBackend()
+	case "ini":
+		xattrs.UseIniBackend()
+	default:
+		return nil, fmt.Errorf("unknown metadata backend %s, only 'ini' or 'xattrs' (default) supported", o.MetadataBackend)
+	}
+
 	lu := &lookup.Lookup{}
 	lu.Options = o
 
@@ -566,7 +577,7 @@ func (fs *Decomposedfs) CreateDir(ctx context.Context, ref *provider.Reference)
 
 	if fs.o.TreeTimeAccounting || fs.o.TreeSizeAccounting {
 		// mark the home node as the end of propagation
-		if err = n.SetXattr(xattrs.PropagationAttr, "1"); err != nil {
+		if err = n.SetXattr(prefixes.PropagationAttr, "1"); err != nil {
 			appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Msg("could not mark node to propagate")
 
 			// FIXME: This does not return an error at all, but results in a severe situation that the

pkg/storage/utils/decomposedfs/grants.go

@@ -31,6 +31,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/ace"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs/prefixes"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 )
@@ -202,9 +203,9 @@ func (fs *Decomposedfs) RemoveGrant(ctx context.Context, ref *provider.Reference
 
 	var attr string
 	if g.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_GROUP {
-		attr = xattrs.GrantGroupAcePrefix + g.Grantee.GetGroupId().OpaqueId
+		attr = prefixes.GrantGroupAcePrefix + g.Grantee.GetGroupId().OpaqueId
 	} else {
-		attr = xattrs.GrantUserAcePrefix + g.Grantee.GetUserId().OpaqueId
+		attr = prefixes.GrantUserAcePrefix + g.Grantee.GetUserId().OpaqueId
 	}
 
 	if err = xattrs.Remove(grantNode.InternalPath(), attr); err != nil {
@@ -309,7 +310,7 @@ func (fs *Decomposedfs) storeGrant(ctx context.Context, n *node.Node, g *provide
 	// set the grant
 	e := ace.FromGrant(g)
 	principal, value := e.Marshal()
-	if err := n.SetXattr(xattrs.GrantPrefix+principal, string(value)); err != nil {
+	if err := n.SetXattr(prefixes.GrantPrefix+principal, string(value)); err != nil {
 		appctx.GetLogger(ctx).Error().Err(err).
 			Str("principal", principal).Msg("Could not set grant for principal")
 		return err
@@ -329,15 +330,15 @@ func extractACEsFromAttrs(ctx context.Context, fsfn string, attrs []string) (ent
 	log := appctx.GetLogger(ctx)
 	entries = []*ace.ACE{}
 	for i := range attrs {
-		if strings.HasPrefix(attrs[i], xattrs.GrantPrefix) {
+		if strings.HasPrefix(attrs[i], prefixes.GrantPrefix) {
 			var value string
 			var err error
 			if value, err = xattrs.Get(fsfn, attrs[i]); err != nil {
 				log.Error().Err(err).Str("attr", attrs[i]).Msg("could not read attribute")
 				continue
 			}
 			var e *ace.ACE
-			principal := attrs[i][len(xattrs.GrantPrefix):]
+			principal := attrs[i][len(prefixes.GrantPrefix):]
 			if e, err = ace.Unmarshal(principal, []byte(value)); err != nil {
 				log.Error().Err(err).Str("principal", principal).Str("attr", attrs[i]).Msg("could not unmarshal ace")
 				continue

pkg/storage/utils/decomposedfs/grants_test.go

@@ -28,9 +28,9 @@ import (
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	helpers "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/testhelpers"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs/prefixes"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"github.com/pkg/xattr"
 	"github.com/stretchr/testify/mock"
 )
 
@@ -140,9 +140,9 @@ var _ = Describe("Grants", func() {
 
 				o := env.Owner.GetId()
 				localPath := n.InternalPath()
-				attr, err := xattr.Get(localPath, xattrs.GrantUserAcePrefix+grant.Grantee.GetUserId().OpaqueId)
+				attr, err := xattrs.Get(localPath, prefixes.GrantUserAcePrefix+grant.Grantee.GetUserId().OpaqueId)
 				Expect(err).ToNot(HaveOccurred())
-				Expect(string(attr)).To(Equal(fmt.Sprintf("\x00t=A:f=:p=rw:c=%s:e=0\n", o.GetOpaqueId()))) // NOTE: this tests ace package
+				Expect(attr).To(Equal(fmt.Sprintf("\x00t=A:f=:p=rw:c=%s:e=0\n", o.GetOpaqueId()))) // NOTE: this tests ace package
 			})
 
 			It("creates a storage space per created grant", func() {

pkg/storage/utils/decomposedfs/lookup/lookup.go

@@ -30,6 +30,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/options"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs/prefixes"
 )
 
 // Lookup implements transformations from filepath to node and back
@@ -138,7 +139,7 @@ func (lu *Lookup) WalkPath(ctx context.Context, r *node.Node, p string, followRe
 		}
 
 		if followReferences {
-			if attrBytes, err := r.Xattr(xattrs.ReferenceAttr); err == nil {
+			if attrBytes, err := r.Xattr(prefixes.ReferenceAttr); err == nil {
 				realNodeID := attrBytes
 				ref, err := xattrs.ReferenceFromAttr([]byte(realNodeID))
 				if err != nil {

pkg/storage/utils/decomposedfs/metadata.go

@@ -30,6 +30,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/errtypes"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/xattrs/prefixes"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/pkg/errors"
@@ -109,7 +110,7 @@ func (fs *Decomposedfs) SetArbitraryMetadata(ctx context.Context, ref *provider.
 		}
 	}
 	for k, v := range md.Metadata {
-		attrName := xattrs.MetadataPrefix + k
+		attrName := prefixes.MetadataPrefix + k
 		if err = n.SetXattr(attrName, v); err != nil {
 			errs = append(errs, errors.Wrap(err, "Decomposedfs: could not set metadata attribute "+attrName+" to "+k))
 		}
@@ -182,7 +183,7 @@ func (fs *Decomposedfs) UnsetArbitraryMetadata(ctx context.Context, ref *provide
 				errs = append(errs, errors.Wrap(errtypes.UserRequired("userrequired"), "user has no id"))
 				continue
 			}
-			fa := fmt.Sprintf("%s:%s:%s@%s", xattrs.FavPrefix, utils.UserTypeToString(uid.GetType()), uid.GetOpaqueId(), uid.GetIdp())
+			fa := fmt.Sprintf("%s:%s:%s@%s", prefixes.FavPrefix, utils.UserTypeToString(uid.GetType()), uid.GetOpaqueId(), uid.GetIdp())
 			if err := n.RemoveXattr(fa); err != nil {
 				if xattrs.IsAttrUnset(err) {
 					continue // already gone, ignore
@@ -194,7 +195,7 @@ func (fs *Decomposedfs) UnsetArbitraryMetadata(ctx context.Context, ref *provide
 				errs = append(errs, errors.Wrap(err, "could not unset favorite flag"))
 			}
 		default:
-			if err = n.RemoveXattr(xattrs.MetadataPrefix + k); err != nil {
+			if err = n.RemoveXattr(prefixes.MetadataPrefix + k); err != nil {
 				if xattrs.IsAttrUnset(err) {
 					continue // already gone, ignore
 				}