Cephfs: support keyrings with IDs (cs3org#2488)

changelog/1.18.0_2022-02-11/cephfs-driver.md

@@ -1,3 +1,3 @@
 Enhancement: Reva CephFS module v0.2.1
 
-https://github.com/cs3org/reva/pull/1209
+https://github.com/cs3org/reva/pull/1209

pkg/storage/fs/cephfs/cephfs.go

@@ -77,7 +77,7 @@ func New(m map[string]interface{}) (fs storage.FS, err error) {
 		return nil, errors.New("cephfs: can't create caches")
 	}
 
-	adminConn := newAdminConn(c.IndexPool)
+	adminConn := newAdminConn(c)
 	if adminConn == nil {
 		return nil, errors.Wrap(err, "cephfs: Couldn't create admin connections")
 	}
@@ -119,7 +119,7 @@ func (fs *cephfs) CreateHome(ctx context.Context) (err error) {
 	}
 
 	err = walkPath(user.home, func(path string) error {
-		return fs.adminConn.adminMount.MakeDir(path, dirPermDefault)
+		return fs.adminConn.adminMount.MakeDir(path, fs.conf.DirPerms)
 	}, false)
 	if err != nil {
 		return getRevaError(err)
@@ -136,7 +136,7 @@ func (fs *cephfs) CreateHome(ctx context.Context) (err error) {
 	}
 
 	user.op(func(cv *cacheVal) {
-		err = cv.mount.MakeDir(removeLeadingSlash(fs.conf.ShareFolder), dirPermDefault)
+		err = cv.mount.MakeDir(removeLeadingSlash(fs.conf.ShareFolder), fs.conf.DirPerms)
 		if err != nil && err.Error() == errFileExists {
 			err = nil
 		}
@@ -153,7 +153,7 @@ func (fs *cephfs) CreateDir(ctx context.Context, ref *provider.Reference) error
 	}
 
 	user.op(func(cv *cacheVal) {
-		if err = cv.mount.MakeDir(path, dirPermDefault); err != nil {
+		if err = cv.mount.MakeDir(path, fs.conf.DirPerms); err != nil {
 			return
 		}
 
@@ -494,7 +494,7 @@ func (fs *cephfs) CreateReference(ctx context.Context, path string, targetURI *u
 		if !strings.HasPrefix(strings.TrimPrefix(path, user.home), fs.conf.ShareFolder) {
 			err = errors.New("cephfs: can't create reference outside a share folder")
 		} else {
-			err = cv.mount.MakeDir(path, dirPermDefault)
+			err = cv.mount.MakeDir(path, fs.conf.DirPerms)
 		}
 	})
 	if err != nil {
@@ -562,46 +562,62 @@ func (fs *cephfs) UnsetArbitraryMetadata(ctx context.Context, ref *provider.Refe
 	return getRevaError(err)
 }
 
+func (fs *cephfs) TouchFile(ctx context.Context, ref *provider.Reference) error {
+	user := fs.makeUser(ctx)
+	path, err := user.resolveRef(ref)
+	if err != nil {
+		return getRevaError(err)
+	}
+
+	user.op(func(cv *cacheVal) {
+		var file *cephfs2.File
+		defer closeFile(file)
+		if file, err = cv.mount.Open(path, os.O_CREATE|os.O_WRONLY, fs.conf.FilePerms); err != nil {
+			return
+		}
+
+		//TODO(tmourati): Add entry id logic
+	})
+
+	return getRevaError(err)
+}
+
 func (fs *cephfs) EmptyRecycle(ctx context.Context) error {
-	return errtypes.NotSupported("cephfs: empty recycle not supported")
+	return errtypes.NotSupported("unimplemented")
 }
 
 func (fs *cephfs) CreateStorageSpace(ctx context.Context, req *provider.CreateStorageSpaceRequest) (r *provider.CreateStorageSpaceResponse, err error) {
-	return nil, errors.New("cephfs: createStorageSpace not supported")
+	return nil, errtypes.NotSupported("unimplemented")
 }
 
 func (fs *cephfs) ListRecycle(ctx context.Context, basePath, key, relativePath string) ([]*provider.RecycleItem, error) {
-	panic("implement me")
+	return nil, errtypes.NotSupported("unimplemented")
 }
 
 func (fs *cephfs) RestoreRecycleItem(ctx context.Context, basePath, key, relativePath string, restoreRef *provider.Reference) error {
-	return errors.New("cephfs: restoreRecycleItem not supported")
+	return errtypes.NotSupported("unimplemented")
 }
 
 func (fs *cephfs) PurgeRecycleItem(ctx context.Context, basePath, key, relativePath string) error {
-	return errors.New("cephfs: purgeRecycleItem not supported")
+	return errtypes.NotSupported("unimplemented")
 }
 
-func (fs *cephfs) ListStorageSpaces(ctx context.Context, filter []*provider.ListStorageSpacesRequest_Filter, permissions map[string]struct{}) ([]*provider.StorageSpace, error) {
-	return nil, errors.New("cephfs: listStorageSpaces not supported")
+func (fs *cephfs) ListStorageSpaces(ctx context.Context, filter []*provider.ListStorageSpacesRequest_Filter) ([]*provider.StorageSpace, error) {
+	return nil, errtypes.NotSupported("unimplemented")
 }
 
 func (fs *cephfs) UpdateStorageSpace(ctx context.Context, req *provider.UpdateStorageSpaceRequest) (*provider.UpdateStorageSpaceResponse, error) {
-	return nil, errors.New("cephfs: updateStorageSpace not supported")
+	return nil, errtypes.NotSupported("unimplemented")
 }
 
-func (fs *cephfs) TouchFile(ctx context.Context, ref *provider.Reference) error {
+func (fs *cephfs) SetLock(ctx context.Context, ref *provider.Reference, lock *provider.Lock) error {
 	return errtypes.NotSupported("unimplemented")
 }
 
 func (fs *cephfs) GetLock(ctx context.Context, ref *provider.Reference) (*provider.Lock, error) {
 	return nil, errtypes.NotSupported("unimplemented")
 }
 
-func (fs *cephfs) SetLock(ctx context.Context, ref *provider.Reference, lock *provider.Lock) error {
-	return errtypes.NotSupported("unimplemented")
-}
-
 func (fs *cephfs) RefreshLock(ctx context.Context, ref *provider.Reference, lock *provider.Lock) error {
 	return errtypes.NotSupported("unimplemented")
 }

pkg/storage/fs/cephfs/chunking.go

@@ -120,7 +120,7 @@ func (c *ChunkHandler) saveChunk(path string, r io.ReadCloser) (finish bool, chu
 	c.user.op(func(cv *cacheVal) {
 		var tmpFile *cephfs2.File
 		target := filepath.Join(c.chunkFolder, chunkTempFilename)
-		tmpFile, err = cv.mount.Open(target, os.O_CREATE|os.O_WRONLY, filePermDefault)
+		tmpFile, err = cv.mount.Open(target, os.O_CREATE|os.O_WRONLY, c.user.fs.conf.FilePerms)
 		defer closeFile(tmpFile)
 		if err != nil {
 			return
@@ -170,7 +170,7 @@ func (c *ChunkHandler) saveChunk(path string, r io.ReadCloser) (finish bool, chu
 		}
 
 		chunk = filepath.Join(c.chunkFolder, c.getChunkTempFileName())
-		assembledFile, err = cv.mount.Open(chunk, os.O_CREATE|os.O_WRONLY, filePermDefault)
+		assembledFile, err = cv.mount.Open(chunk, os.O_CREATE|os.O_WRONLY, c.user.fs.conf.FilePerms)
 		defer closeFile(assembledFile)
 		defer deleteFile(cv.mount, chunk)
 		if err != nil {

pkg/storage/fs/cephfs/connections.go

@@ -108,33 +108,40 @@ func (c *connections) clearCache() {
 }
 
 type adminConn struct {
-	indexPoolName string
+	// indexPoolName string
 	subvolAdmin   *admin.FSAdmin
 	adminMount    Mount
 	radosConn     *rados2.Conn
-	radosIO       *rados2.IOContext
+	// radosIO       *rados2.IOContext
 }
 
-func newAdminConn(poolName string) *adminConn {
-	rados, err := rados2.NewConn()
+func newAdminConn(conf *Options) *adminConn {
+	rados, err := rados2.NewConnWithUser(conf.ClientID)
 	if err != nil {
 		return nil
 	}
-	if err = rados.ReadDefaultConfigFile(); err != nil {
+	if err = rados.ReadConfigFile(conf.Config); err != nil {
+		return nil
+	}
+
+	if err = rados.SetConfigOption("keyring", conf.Keyring); err != nil {
 		return nil
 	}
 
 	if err = rados.Connect(); err != nil {
 		return nil
 	}
 
+	// TODO: May use later for file ids
+	/*
 	pools, err := rados.ListPools()
 	if err != nil {
 		rados.Shutdown()
 		return nil
 	}
 
 	var radosIO *rados2.IOContext
+	poolName := conf.IndexPool
 	if in(poolName, pools) {
 		radosIO, err = rados.OpenIOContext(poolName)
 		if err != nil {
@@ -153,37 +160,43 @@ func newAdminConn(poolName string) *adminConn {
 			return nil
 		}
 	}
+	*/
 
 	mount, err := cephfs2.CreateFromRados(rados)
 	if err != nil {
 		rados.Shutdown()
 		return nil
 	}
 
-	if err = mount.Mount(); err != nil {
+	if err = mount.MountWithRoot(conf.Root); err != nil {
 		rados.Shutdown()
 		destroyCephConn(mount, nil)
 		return nil
 	}
 
-	return &adminConn{
-		poolName,
+	return &adminConn {
+		// poolName,
 		admin.NewFromConn(rados),
 		mount,
 		rados,
-		radosIO,
+		// radosIO,
 	}
 }
 
 func newConn(user *User) *cacheVal {
 	var perm *cephfs2.UserPerm
-	mount, err := cephfs2.CreateMount()
+	mount, err := cephfs2.CreateMountWithId(user.fs.conf.ClientID)
 	if err != nil {
 		return destroyCephConn(mount, perm)
 	}
-	if err = mount.ReadDefaultConfigFile(); err != nil {
+	if err = mount.ReadConfigFile(user.fs.conf.Config); err != nil {
 		return destroyCephConn(mount, perm)
 	}
+
+	if err = mount.SetConfigOption("keyring", user.fs.conf.Keyring); err != nil {
+		return destroyCephConn(mount, perm)
+	}
+
 	if err = mount.Init(); err != nil {
 		return destroyCephConn(mount, perm)
 	}
@@ -195,7 +208,7 @@ func newConn(user *User) *cacheVal {
 		}
 	}
 
-	if err = mount.MountWithRoot("/"); err != nil {
+	if err = mount.MountWithRoot(user.fs.conf.Root); err != nil {
 		return destroyCephConn(mount, perm)
 	}
 

pkg/storage/fs/cephfs/options.go

@@ -29,15 +29,20 @@ import (
 
 // Options for the cephfs module
 type Options struct {
+	ClientID string `mapstructure:"client_id"`
+	Config   string `mapstructure:"config"`
 	GatewaySvc   string `mapstructure:"gatewaysvc"`
 	IndexPool    string `mapstructure:"index_pool"`
+	Keyring      string `mapstructure:"keyring"`
 	Root         string `mapstructure:"root"`
 	ShadowFolder string `mapstructure:"shadow_folder"`
 	ShareFolder  string `mapstructure:"share_folder"`
 	UploadFolder string `mapstructure:"uploads"`
 	UserLayout   string `mapstructure:"user_layout"`
 
 	DisableHome    bool   `mapstructure:"disable_home"`
+	DirPerms       uint32 `mapstructure:"dir_perms"`
+	FilePerms      uint32 `mapstructure:"file_perms"`
 	UserQuotaBytes uint64 `mapstructure:"user_quota_bytes"`
 	HiddenDirs     map[string]bool
 }
@@ -49,10 +54,26 @@ func (c *Options) fillDefaults() {
 		c.IndexPool = "path_index"
 	}
 
+	if c.Config == "" {
+		c.Config = "/etc/ceph/ceph.conf"
+	} else {
+		c.Config = addLeadingSlash(c.Config) //force absolute path in case leading "/" is omitted
+	}
+
+	if c.ClientID == "" {
+		c.ClientID = "admin"
+	}
+
+	if c.Keyring == "" {
+		c.Keyring = "/etc/ceph/keyring"
+	} else {
+		c.Keyring = addLeadingSlash(c.Keyring)
+	}
+
 	if c.Root == "" {
 		c.Root = "/home"
 	} else {
-		c.Root = addLeadingSlash(c.Root) //force absolute path in case leading "/" is omitted
+		c.Root = addLeadingSlash(c.Root)
 	}
 
 	if c.ShadowFolder == "" {
@@ -82,7 +103,13 @@ func (c *Options) fillDefaults() {
 		removeLeadingSlash(c.ShadowFolder): true,
 	}
 
-	c.DisableHome = false // it is currently only home based
+	if c.DirPerms == 0 {
+		c.DirPerms = dirPermDefault
+	}
+
+	if c.FilePerms == 0 {
+		c.FilePerms = filePermDefault
+	}
 
 	if c.UserQuotaBytes == 0 {
 		c.UserQuotaBytes = 50000000000

pkg/storage/fs/cephfs/upload.go

@@ -179,7 +179,7 @@ func (fs *cephfs) NewUpload(ctx context.Context, info tusd.FileInfo) (upload tus
 	user.op(func(cv *cacheVal) {
 		var f *cephfs2.File
 		defer closeFile(f)
-		f, err = cv.mount.Open(binPath, os.O_CREATE|os.O_WRONLY, filePermDefault)
+		f, err = cv.mount.Open(binPath, os.O_CREATE|os.O_WRONLY, fs.conf.FilePerms)
 		if err != nil {
 			return
 		}
@@ -332,7 +332,7 @@ func (upload *fileUpload) writeInfo() error {
 	user := upload.fs.makeUser(upload.ctx)
 	user.op(func(cv *cacheVal) {
 		var file io.WriteCloser
-		if file, err = cv.mount.Open(upload.infoPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, filePermDefault); err != nil {
+		if file, err = cv.mount.Open(upload.infoPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, upload.fs.conf.FilePerms); err != nil {
 			return
 		}
 		defer file.Close()

pkg/storage/fs/cephfs/user.go

@@ -53,7 +53,11 @@ type User struct {
 
 func (fs *cephfs) makeUser(ctx context.Context) *User {
 	u := ctx2.ContextMustGetUser(ctx)
-	home := filepath.Join(fs.conf.Root, templates.WithUser(u, fs.conf.UserLayout))
+	home := fs.conf.Root
+	if !fs.conf.DisableHome {
+		home = filepath.Join(fs.conf.Root, templates.WithUser(u, fs.conf.UserLayout))
+	}
+
 	return &User{u, fs, ctx, home}
 }
 

pkg/storage/fs/cephfs/utils.go

@@ -29,7 +29,6 @@ import (
 	"os"
 	"path/filepath"
 	"strconv"
-	"strings"
 
 	cephfs2 "github.com/ceph/go-ceph/cephfs"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
@@ -42,8 +41,8 @@ type Mount = *cephfs2.MountInfo
 type Statx = *cephfs2.CephStatx
 
 var dirPermFull = uint32(0777)
-var dirPermDefault = uint32(0775)
-var filePermDefault = uint32(0660)
+var dirPermDefault = uint32(0700)
+var filePermDefault = uint32(0640)
 
 func closeDir(directory *cephfs2.Directory) {
 	if directory != nil {
@@ -75,8 +74,10 @@ func isDir(t provider.ResourceType) bool {
 	return t == provider.ResourceType_RESOURCE_TYPE_CONTAINER
 }
 
+// TODO: Use when fileids are available
+/*
 func (fs *cephfs) makeFIDPath(fid string) string {
-	return "" //filepath.Join(fs.conf.EIDFolder, fid)
+	return "" // filepath.Join(fs.conf.EIDFolder, fid) EIDFolder does not exist
 }
 
 func (fs *cephfs) makeFID(absolutePath string, inode string) (rid *provider.ResourceId, err error) {
@@ -99,6 +100,7 @@ func (fs *cephfs) getFIDPath(cv *cacheVal, path string) (fid string, err error)
 
 	return fs.makeFIDPath(string(buffer)), err
 }
+ */
 
 func calcChecksum(filepath string, mt Mount, stat Statx) (checksum string, err error) {
 	file, err := mt.Open(filepath, os.O_RDONLY, 0)
@@ -195,6 +197,8 @@ func walkPath(path string, f func(string) error, reverse bool) (err error) {
 	return
 }
 
+// TODO: Use when fileids are available
+/*
 func (fs *cephfs) writeIndex(oid string, value string) (err error) {
 	return fs.adminConn.radosIO.WriteFull(oid, []byte(value))
 }
@@ -243,3 +247,4 @@ func (fs *cephfs) resolveIndex(oid string) (fullPath string, err error) {
 		currPath.Reset()
 	}
 }
+ */