cephfs.go

package cephfs

import (
	"context"
	cephfs2 "github.com/ceph/go-ceph/cephfs"
	"github.com/ceph/go-ceph/cephfs/admin"
	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
	"github.com/cs3org/reva/pkg/errtypes"
	"github.com/cs3org/reva/pkg/storage"
	"github.com/cs3org/reva/pkg/storage/fs/registry"
	"github.com/cs3org/reva/pkg/storage/utils/chunking"
	"github.com/pkg/errors"
	"io"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
)

type cephfs struct{
	conf *Options
	chunkHandler *chunking.ChunkHandler
}

func init() {
	registry.Register("cephfs", New)
}

func New(m map[string]interface{}) (storage.FS, error) {
	return &cephfs{}, nil
}

func (fs *cephfs) GetHome(ctx context.Context) (string, error) {
	if fs.conf.DisableHome {
		return "", errtypes.NotSupported("cephfs: GetHome() home supported disabled")
	}

	u := fs.MakeUser(ctx)

	return FSadm.SubVolumePath("cephfs", "reva", u.Username)
}

func (fs *cephfs) CreateHome(ctx context.Context) (err error) {
	if fs.conf.DisableHome {
		return errtypes.NotSupported("cephfs: GetHome() home supported disabled")
	}

	u := fs.MakeUser(ctx)

	err = FSadm.CreateSubVolume("cephfs", "reva", u.Username, &admin.SubVolumeOptions{
		Size:              admin.ByteCount(fs.conf.UserQuotaBytes),
		Uid:               int(u.UidNumber),
		Gid:               int(u.GidNumber),
		Mode:              0755,
	})

	return
}

func (fs *cephfs) CreateDir(ctx context.Context, fn string) error {
	u := fs.MakeUser(ctx)

	return u.exec(func(mt Mount) error {
		return mt.MakeDir(fn, fs.conf.DirMode)
	})
}

func (fs *cephfs) Delete(ctx context.Context, ref *provider.Reference) (err error) {
	var path string
	u := fs.MakeUser(ctx)
	path, err = ResolveRef(ref); if err != nil { return }

	return u.exec(func(mt Mount) error {
		return mt.Unlink(path)
	})
}

func (fs *cephfs) Move(ctx context.Context, oldRef, newRef *provider.Reference) (err error) {
	var oldPath, newPath string
	u := fs.MakeUser(ctx)
	oldPath, err = ResolveRef(oldRef); if err != nil { return }
	newPath, err = ResolveRef(oldRef); if err != nil { return }

	return u.exec(func(mt Mount) error {
		return mt.Rename(oldPath, newPath)
	})
}

func (fs *cephfs) GetMD(ctx context.Context, ref *provider.Reference, mdKeys []string) (*provider.ResourceInfo, error) {
	u := fs.MakeUser(ctx)
	path, err := ResolveRef(ref); if err != nil { return nil, err }

	var iri interface{}
	iri, err = u.exec2(func(mt Mount) (i interface{}, e error) {
		var stat *cephfs2.CephStatx
		stat, e = mt.Statx(path, cephfs2.StatxAllStats, 0)

		return fileAsResourceInfo(ctx, mt, path, stat, mdKeys)
	})

	return iri.(*provider.ResourceInfo), err
}

func (fs *cephfs) ListFolder(ctx context.Context, ref *provider.Reference, mdKeys []string) (files []*provider.ResourceInfo, err error) {
	var path string
	u := fs.MakeUser(ctx)
	path, err = ResolveRef(ref); if err != nil { return }

	var dir *cephfs2.Directory
	err = u.exec(func(mt Mount) (e error) {
		dir, e = mt.OpenDir(path)
		if e != nil { return }
		defer dir.Close()

		var entry *cephfs2.DirEntryPlus
		var ri *provider.ResourceInfo
		for entry, e = dir.ReadDirPlus(cephfs2.StatxAllStats, 0); entry != nil && e == nil;
		entry, e = dir.ReadDirPlus(cephfs2.StatxAllStats, 0) {
			if entry.Name() == "." || entry.Name() == ".." { continue } //TODO: Maybe include?

			ri, err = fileAsResourceInfo(ctx, mt, filepath.Join(path, entry.Name()), entry.Statx(), mdKeys)
			if ri == nil || err != nil { continue }

			files = append(files, ri)
		}

		return
	})

	return
}

func (fs *cephfs) Download(ctx context.Context, ref *provider.Reference) (io.ReadCloser, error) {
	u := fs.MakeUser(ctx)
	path, err := ResolveRef(ref)
	if err != nil {
		return nil, errors.Wrap(err, "cephfs: error resolving ref")
	}

	if strings.HasPrefix(path, fs.conf.ShareFolder) {
		return nil, errtypes.PermissionDenied("cephfs: cannot download under the virtual share folder")
	}

	ir, err := u.exec2(func(mt Mount) (interface{}, error) {
		return mt.Open(path, os.O_RDONLY, defaultFilePerm)
	})

	return ir.(*cephfs2.File), err
}

func (fs *cephfs) ListRevisions(ctx context.Context, ref *provider.Reference) ([]*provider.FileVersion, error) {
	panic("implement me")
}

func (fs *cephfs) DownloadRevision(ctx context.Context, ref *provider.Reference, key string) (io.ReadCloser, error) {
	panic("implement me")
}

func (fs *cephfs) RestoreRevision(ctx context.Context, ref *provider.Reference, key string) error {
	panic("implement me")
}

func (fs *cephfs) ListRecycle(ctx context.Context) ([]*provider.RecycleItem, error) {
	panic("implement me")
}

func (fs *cephfs) RestoreRecycleItem(ctx context.Context, key string) error {
	panic("implement me")
}

func (fs *cephfs) PurgeRecycleItem(ctx context.Context, key string) error {
	return errors.New("cephfs: Recycled items can't be purged, they are handled by snapshots, which are read-only")
}

func (fs *cephfs) EmptyRecycle(ctx context.Context) error {
	return errors.New("cephfs: recycle is based on snapshots and can't be edited")
}

func (fs *cephfs) GetPathByID(ctx context.Context, id *provider.ResourceId) (string, error) {
	return "", errors.New("cephfs: file ids not implemented")
}

func (fs *cephfs) AddGrant(ctx context.Context, ref *provider.Reference, g *provider.Grant) (err error) {
	var path string
	u := fs.MakeUser(ctx)
	path, err = ResolveRef(ref); if err != nil { return }

	err = u.exec(func(mt Mount) (e error) {
		return changePerms(mt, g, path, UpdateGrant)
	})

	return
}

func (fs *cephfs) RemoveGrant(ctx context.Context, ref *provider.Reference, g *provider.Grant) (err error) {
	var path string
	u := fs.MakeUser(ctx)
	path, err = ResolveRef(ref); if err != nil { return }

	err = u.exec(func(mt Mount) (e error) {
		return changePerms(mt, g, path, RemoveGrant)
	})

	return
}

func (fs *cephfs) UpdateGrant(ctx context.Context, ref *provider.Reference, g *provider.Grant) (err error) {
	var path string
	u := fs.MakeUser(ctx)
	path, err = ResolveRef(ref); if err != nil { return }

	err = u.exec(func(mt Mount) (e error) {
		return changePerms(mt, g, path, UpdateGrant)
	})

	return
}

func (fs *cephfs) ListGrants(ctx context.Context, ref *provider.Reference) (glist []*provider.Grant, err error) {
	var path string
	u := fs.MakeUser(ctx)
	if path, err = ResolveRef(ref); err != nil { return }

	err = u.exec(func(mt Mount) error {
		glist = GetFullPermissionSet(mt, path)

		if glist == nil {
			return errors.New("cephfs: error listing grants on " + path)
		}

		return nil
	})

	return
}

func (fs *cephfs) GetQuota(ctx context.Context) (total uint64, used uint64, err error) {
	u := fs.MakeUser(ctx)

	err = u.exec(func(mt Mount) error {
		buf, e := mt.GetXattr("/", "ceph.quota.max_bytes")
		if e != nil {
			total = 0
		} else {
			total, e = strconv.ParseUint(string(buf), 10, 64)
		}

		buf, e = mt.GetXattr("/", "ceph.dir.rbytes")
		if e == nil {
			used, e = strconv.ParseUint(string(buf), 10, 64)
		}

		return e
	})

	return
}

func (fs *cephfs) CreateReference(ctx context.Context, path string, targetURI *url.URL) (err error) {
	if !strings.HasPrefix(path, fs.conf.ShareFolder) {
		return errors.New("cephfs: can't create reference outside a share folder")
	}
	u := fs.MakeUser(ctx)

	err = u.exec(func(mt Mount) error {
		return mt.MakeDir(path, fs.conf.DirMode)
	})
	if err != nil { return }

	err = u.exec(func(mt Mount) error {
		return mt.SetXattr(path, "reva.stat.ref", []byte(targetURI.String()), 0)
	})

	return
}

func (fs *cephfs) Shutdown(ctx context.Context) (err error) {
	if Admin != nil {
		_ = Admin.Release()
	}
	ClearCache()

	return
}

func (fs *cephfs) SetArbitraryMetadata(ctx context.Context, ref *provider.Reference, md *provider.ArbitraryMetadata) (err error) {
	var path string
	u := fs.MakeUser(ctx)
	path, err = ResolveRef(ref); if err != nil { return }

	for k, v := range md.Metadata {
		e := u.exec(func(mt Mount) error {
			return mt.SetXattr(path, k, []byte(v), 0)
		})
		if e != nil { err = errors.Wrap(err, e.Error()) }
	}

	return
}

func (fs *cephfs) UnsetArbitraryMetadata(ctx context.Context, ref *provider.Reference, keys []string) (err error) {
	var path string
	u := fs.MakeUser(ctx)
	path, err = ResolveRef(ref); if err != nil { return }

	for _, key := range keys {
		e := u.exec(func(mt Mount) error {
			return mt.RemoveXattr(path, key)
		})
		if e != nil { err = errors.Wrap(err, e.Error()) }
	}

	return
}