reva bump 2.30.0
This commit is contained in:
+2
-1
@@ -50,6 +50,7 @@ type Option func(l *zerolog.Logger)
|
||||
// New creates a new logger.
|
||||
func New(opts ...Option) *zerolog.Logger {
|
||||
// create a default logger
|
||||
zerolog.SetGlobalLevel(zerolog.TraceLevel)
|
||||
zl := zerolog.New(os.Stderr).With().Timestamp().Caller().Logger()
|
||||
for _, opt := range opts {
|
||||
opt(&zl)
|
||||
@@ -127,7 +128,7 @@ type LogConf struct {
|
||||
|
||||
func fromConfig(conf *LogConf) (*zerolog.Logger, error) {
|
||||
if conf.Level == "" {
|
||||
conf.Level = zerolog.DebugLevel.String()
|
||||
conf.Level = zerolog.InfoLevel.String()
|
||||
}
|
||||
|
||||
var opts []Option
|
||||
|
||||
+24
-5
@@ -242,16 +242,35 @@ type tusdLogger struct {
|
||||
|
||||
// Handle handles the record
|
||||
func (l tusdLogger) Handle(_ context.Context, r slog.Record) error {
|
||||
var logev *zerolog.Event
|
||||
switch r.Level {
|
||||
case slog.LevelDebug:
|
||||
l.log.Debug().Msg(r.Message)
|
||||
logev = l.log.Debug()
|
||||
case slog.LevelInfo:
|
||||
l.log.Info().Msg(r.Message)
|
||||
logev = l.log.Info()
|
||||
case slog.LevelWarn:
|
||||
l.log.Warn().Msg(r.Message)
|
||||
logev = l.log.Warn()
|
||||
case slog.LevelError:
|
||||
l.log.Error().Msg(r.Message)
|
||||
logev = l.log.Error()
|
||||
}
|
||||
r.Attrs(func(a slog.Attr) bool {
|
||||
// Resolve the Attr's value before doing anything else.
|
||||
a.Value = a.Value.Resolve()
|
||||
// Ignore empty Attrs.
|
||||
if a.Equal(slog.Attr{}) {
|
||||
return true
|
||||
}
|
||||
switch a.Value.Kind() {
|
||||
case slog.KindBool:
|
||||
logev = logev.Bool(a.Key, a.Value.Bool())
|
||||
case slog.KindInt64:
|
||||
logev = logev.Int64(a.Key, a.Value.Int64())
|
||||
default:
|
||||
logev = logev.Str(a.Key, a.Value.String())
|
||||
}
|
||||
return true
|
||||
})
|
||||
logev.Msg(r.Message)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -262,7 +281,7 @@ func (l tusdLogger) Enabled(_ context.Context, _ slog.Level) bool { return true
|
||||
func (l tusdLogger) WithAttrs(attr []slog.Attr) slog.Handler {
|
||||
fields := make(map[string]interface{}, len(attr))
|
||||
for _, a := range attr {
|
||||
fields[a.Key] = a.Value
|
||||
fields[a.Key] = a.Value.String()
|
||||
}
|
||||
c := l.log.With().Fields(fields).Logger()
|
||||
sLog := tusdLogger{log: &c}
|
||||
|
||||
+10
-12
@@ -29,15 +29,13 @@ import (
|
||||
|
||||
// taken from https://golang.org/src/net/http/fs.go
|
||||
|
||||
// ErrSeeker is returned by ServeContent's sizeFunc when the content
|
||||
// doesn't seek properly. The underlying Seeker's error text isn't
|
||||
// included in the sizeFunc reply so it's not sent over HTTP to end
|
||||
// users.
|
||||
var ErrSeeker = errors.New("seeker can't seek")
|
||||
// ErrInvalidRange is returned by serveContent's parseRange if the Range is
|
||||
// malformed or invalid.
|
||||
var ErrInvalidRange = errors.New("invalid range")
|
||||
|
||||
// ErrNoOverlap is returned by serveContent's parseRange if first-byte-pos of
|
||||
// all of the byte-range-spec values is greater than the content size.
|
||||
var ErrNoOverlap = errors.New("invalid range: failed to overlap")
|
||||
var ErrNoOverlap = fmt.Errorf("%w: failed to overlap", ErrInvalidRange)
|
||||
|
||||
// HTTPRange specifies the byte range to be sent to the client.
|
||||
type HTTPRange struct {
|
||||
@@ -65,7 +63,7 @@ func ParseRange(s string, size int64) ([]HTTPRange, error) {
|
||||
}
|
||||
const b = "bytes="
|
||||
if !strings.HasPrefix(s, b) {
|
||||
return nil, errors.New("invalid range")
|
||||
return nil, ErrInvalidRange
|
||||
}
|
||||
ranges := []HTTPRange{}
|
||||
noOverlap := false
|
||||
@@ -76,7 +74,7 @@ func ParseRange(s string, size int64) ([]HTTPRange, error) {
|
||||
}
|
||||
i := strings.Index(ra, "-")
|
||||
if i < 0 {
|
||||
return nil, errors.New("invalid range")
|
||||
return nil, ErrInvalidRange
|
||||
}
|
||||
start, end := textproto.TrimString(ra[:i]), textproto.TrimString(ra[i+1:])
|
||||
var r HTTPRange
|
||||
@@ -85,7 +83,7 @@ func ParseRange(s string, size int64) ([]HTTPRange, error) {
|
||||
// range start relative to the end of the file.
|
||||
i, err := strconv.ParseInt(end, 10, 64)
|
||||
if err != nil {
|
||||
return nil, errors.New("invalid range")
|
||||
return nil, ErrInvalidRange
|
||||
}
|
||||
if i > size {
|
||||
i = size
|
||||
@@ -95,7 +93,7 @@ func ParseRange(s string, size int64) ([]HTTPRange, error) {
|
||||
} else {
|
||||
i, err := strconv.ParseInt(start, 10, 64)
|
||||
if err != nil || i < 0 {
|
||||
return nil, errors.New("invalid range")
|
||||
return nil, ErrInvalidRange
|
||||
}
|
||||
if i >= size {
|
||||
// If the range begins after the size of the content,
|
||||
@@ -110,7 +108,7 @@ func ParseRange(s string, size int64) ([]HTTPRange, error) {
|
||||
} else {
|
||||
i, err := strconv.ParseInt(end, 10, 64)
|
||||
if err != nil || r.Start > i {
|
||||
return nil, errors.New("invalid range")
|
||||
return nil, ErrInvalidRange
|
||||
}
|
||||
if i >= size {
|
||||
i = size - 1
|
||||
@@ -146,7 +144,7 @@ func RangesMIMESize(ranges []HTTPRange, contentType string, contentSize int64) (
|
||||
_, _ = mw.CreatePart(ra.MimeHeader(contentType, contentSize))
|
||||
encSize += ra.Length
|
||||
}
|
||||
mw.Close()
|
||||
_ = mw.Close()
|
||||
encSize += int64(w)
|
||||
return
|
||||
}
|
||||
|
||||
+8
-1
@@ -25,6 +25,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
|
||||
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
|
||||
@@ -72,6 +73,7 @@ type Lookup struct {
|
||||
Options *options.Options
|
||||
|
||||
IDCache IDCache
|
||||
IDHistoryCache IDCache
|
||||
metadataBackend metadata.Backend
|
||||
userMapper usermapper.Mapper
|
||||
tm node.TimeManager
|
||||
@@ -79,10 +81,15 @@ type Lookup struct {
|
||||
|
||||
// New returns a new Lookup instance
|
||||
func New(b metadata.Backend, um usermapper.Mapper, o *options.Options, tm node.TimeManager) *Lookup {
|
||||
idHistoryConf := o.Options.IDCache
|
||||
idHistoryConf.Database = o.Options.IDCache.Table + "_history"
|
||||
idHistoryConf.TTL = 1 * time.Minute
|
||||
|
||||
lu := &Lookup{
|
||||
Options: o,
|
||||
metadataBackend: b,
|
||||
IDCache: NewStoreIDCache(&o.Options),
|
||||
IDCache: NewStoreIDCache(o.Options.IDCache),
|
||||
IDHistoryCache: NewStoreIDCache(idHistoryConf),
|
||||
userMapper: um,
|
||||
tm: tm,
|
||||
}
|
||||
|
||||
Generated
Vendored
+16
-14
@@ -25,7 +25,7 @@ import (
|
||||
microstore "go-micro.dev/v4/store"
|
||||
|
||||
"github.com/opencloud-eu/reva/v2/pkg/appctx"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/options"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage/cache"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/store"
|
||||
)
|
||||
|
||||
@@ -34,31 +34,33 @@ type StoreIDCache struct {
|
||||
}
|
||||
|
||||
// NewMemoryIDCache returns a new MemoryIDCache
|
||||
func NewStoreIDCache(o *options.Options) *StoreIDCache {
|
||||
func NewStoreIDCache(c cache.Config) *StoreIDCache {
|
||||
return &StoreIDCache{
|
||||
cache: store.Create(
|
||||
store.Store(o.IDCache.Store),
|
||||
store.Size(o.IDCache.Size),
|
||||
microstore.Nodes(o.IDCache.Nodes...),
|
||||
microstore.Database(o.IDCache.Database),
|
||||
microstore.Table(o.IDCache.Table),
|
||||
store.DisablePersistence(o.IDCache.DisablePersistence),
|
||||
store.Authentication(o.IDCache.AuthUsername, o.IDCache.AuthPassword),
|
||||
store.Store(c.Store),
|
||||
store.Size(c.Size),
|
||||
microstore.Nodes(c.Nodes...),
|
||||
microstore.Database(c.Database),
|
||||
microstore.Table(c.Table),
|
||||
store.DisablePersistence(c.DisablePersistence),
|
||||
store.Authentication(c.AuthUsername, c.AuthPassword),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
// Delete removes an entry from the cache
|
||||
func (c *StoreIDCache) Delete(_ context.Context, spaceID, nodeID string) error {
|
||||
var rerr error
|
||||
v, err := c.cache.Read(cacheKey(spaceID, nodeID))
|
||||
if err == nil {
|
||||
err := c.cache.Delete(reverseCacheKey(string(v[0].Value)))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rerr = c.cache.Delete(reverseCacheKey(string(v[0].Value)))
|
||||
}
|
||||
|
||||
return c.cache.Delete(cacheKey(spaceID, nodeID))
|
||||
err = c.cache.Delete(cacheKey(spaceID, nodeID))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return rerr
|
||||
}
|
||||
|
||||
// DeleteByPath removes an entry from the cache
|
||||
|
||||
+13
-9
@@ -44,24 +44,28 @@ type Options struct {
|
||||
WatchType string `mapstructure:"watch_type"`
|
||||
WatchPath string `mapstructure:"watch_path"`
|
||||
WatchFolderKafkaBrokers string `mapstructure:"watch_folder_kafka_brokers"`
|
||||
|
||||
// InotifyWatcher specific options
|
||||
InotifyStatsFrequency time.Duration `mapstructure:"inotify_stats_frequency"`
|
||||
}
|
||||
|
||||
// New returns a new Options instance for the given configuration
|
||||
func New(m map[string]interface{}) (*Options, error) {
|
||||
o := &Options{}
|
||||
if err := mapstructure.Decode(m, o); err != nil {
|
||||
err = errors.Wrap(err, "error decoding conf")
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// default to hybrid metadatabackend for posixfs
|
||||
if _, ok := m["metadata_backend"]; !ok {
|
||||
m["metadata_backend"] = "hybrid"
|
||||
}
|
||||
if _, ok := m["scan_debounce_delay"]; !ok {
|
||||
m["scan_debounce_delay"] = 10 * time.Millisecond
|
||||
}
|
||||
if _, ok := m["inotify_stats_frequency"]; !ok {
|
||||
m["inotify_stats_frequency"] = 5 * time.Minute
|
||||
}
|
||||
|
||||
// debounced scan delay
|
||||
if o.ScanDebounceDelay == 0 {
|
||||
o.ScanDebounceDelay = 10 * time.Millisecond
|
||||
o := &Options{}
|
||||
if err := mapstructure.Decode(m, o); err != nil {
|
||||
err = errors.Wrap(err, "error decoding conf")
|
||||
return nil, err
|
||||
}
|
||||
|
||||
do, err := decomposedoptions.New(m)
|
||||
|
||||
+11
-6
@@ -164,17 +164,22 @@ func (tb *Trashbin) MoveToTrash(ctx context.Context, n *node.Node, path string)
|
||||
return err
|
||||
}
|
||||
|
||||
// purge metadata
|
||||
// 1. "Forget" the node
|
||||
if err = tb.lu.IDCache.DeleteByPath(ctx, path); err != nil {
|
||||
return err
|
||||
}
|
||||
err = tb.lu.MetadataBackend().Purge(ctx, n)
|
||||
|
||||
// 2. Move the node to the trash
|
||||
itemTrashPath := filepath.Join(trashPath, "files", key+".trashitem")
|
||||
err = os.Rename(path, itemTrashPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
itemTrashPath := filepath.Join(trashPath, "files", key+".trashitem")
|
||||
return os.Rename(path, itemTrashPath)
|
||||
// 3. Purge the node from the metadata backend. This will not delete the xattrs from the
|
||||
// node as it has already been moved but still remove it from the file metadata cache so
|
||||
// that the metadata is no longer available when reading the node.
|
||||
return tb.lu.MetadataBackend().Purge(ctx, n)
|
||||
}
|
||||
|
||||
// ListRecycle returns the list of available recycle items
|
||||
@@ -315,7 +320,7 @@ func (tb *Trashbin) RestoreRecycleItem(ctx context.Context, spaceID string, key,
|
||||
}
|
||||
// TODO the decomposed trash also checks the permissions on the restore node
|
||||
|
||||
_, id, _, err := tb.lu.MetadataBackend().IdentifyPath(ctx, trashPath)
|
||||
_, id, _, _, err := tb.lu.MetadataBackend().IdentifyPath(ctx, trashPath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -325,7 +330,7 @@ func (tb *Trashbin) RestoreRecycleItem(ctx context.Context, spaceID string, key,
|
||||
}
|
||||
|
||||
// update parent id in case it was restored to a different location
|
||||
_, parentID, _, err := tb.lu.MetadataBackend().IdentifyPath(ctx, filepath.Dir(restorePath))
|
||||
_, parentID, _, _, err := tb.lu.MetadataBackend().IdentifyPath(ctx, filepath.Dir(restorePath))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
+91
-65
@@ -243,17 +243,21 @@ func (t *Tree) Scan(path string, action EventAction, isDir bool) error {
|
||||
case ActionMoveFrom:
|
||||
t.log.Debug().Str("path", path).Bool("isDir", isDir).Msg("scanning path (ActionMoveFrom)")
|
||||
// 6. file/directory moved out of the watched directory
|
||||
// -> update directory
|
||||
err := t.HandleFileDelete(path)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Bool("isDir", isDir).Msg("failed to handle deleted item")
|
||||
}
|
||||
err = t.setDirty(filepath.Dir(path), true)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Bool("isDir", isDir).Msg("failed to mark directory as dirty")
|
||||
// -> remove from caches
|
||||
|
||||
// remember the id of the moved away item
|
||||
spaceID, nodeID, err := t.lookup.IDsForPath(context.Background(), path)
|
||||
if err == nil {
|
||||
err = t.lookup.IDHistoryCache.Set(context.Background(), spaceID, nodeID, path)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Msg("failed to cache the id of the moved item")
|
||||
}
|
||||
}
|
||||
|
||||
go func() { _ = t.WarmupIDCache(filepath.Dir(path), false, true) }()
|
||||
err = t.HandleFileDelete(path, false) // Do not send a item-trashed SSE in case of moves. They trigger a item-renamed event instead.
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Bool("isDir", isDir).Msg("failed to handle moved away item")
|
||||
}
|
||||
|
||||
case ActionDelete:
|
||||
t.log.Debug().Str("path", path).Bool("isDir", isDir).Msg("handling deleted item")
|
||||
@@ -261,7 +265,7 @@ func (t *Tree) Scan(path string, action EventAction, isDir bool) error {
|
||||
// 7. Deleted file or directory
|
||||
// -> update parent and all children
|
||||
|
||||
err := t.HandleFileDelete(path)
|
||||
err := t.HandleFileDelete(path, true)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Bool("isDir", isDir).Msg("failed to handle deleted item")
|
||||
}
|
||||
@@ -276,12 +280,20 @@ func (t *Tree) Scan(path string, action EventAction, isDir bool) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *Tree) HandleFileDelete(path string) error {
|
||||
func (t *Tree) HandleFileDelete(path string, sendSSE bool) error {
|
||||
spaceID, id, err := t.lookup.IDsForPath(context.Background(), path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
n := node.NewBaseNode(spaceID, id, t.lookup)
|
||||
if n.InternalPath() != path {
|
||||
return fmt.Errorf("internal path does not match path")
|
||||
}
|
||||
_, err = os.Stat(path)
|
||||
if err == nil || !os.IsNotExist(err) {
|
||||
t.log.Info().Str("path", path).Msg("file that was about to be cleared still exists/exists again. We'll leave it alone")
|
||||
return nil
|
||||
}
|
||||
|
||||
// purge metadata
|
||||
if err := t.lookup.IDCache.DeleteByPath(context.Background(), path); err != nil {
|
||||
@@ -291,6 +303,10 @@ func (t *Tree) HandleFileDelete(path string) error {
|
||||
t.log.Error().Err(err).Str("path", path).Msg("could not purge metadata")
|
||||
}
|
||||
|
||||
if !sendSSE {
|
||||
return nil
|
||||
}
|
||||
|
||||
parentNode, err := t.getNodeForPath(filepath.Dir(path))
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -355,6 +371,7 @@ func (t *Tree) findSpaceId(path string) (string, node.Attributes, error) {
|
||||
}
|
||||
|
||||
func (t *Tree) assimilate(item scanItem) error {
|
||||
t.log.Debug().Str("path", item.Path).Bool("rescan", item.ForceRescan).Bool("recurse", item.Recurse).Msg("assimilate")
|
||||
var err error
|
||||
|
||||
// First find the space id
|
||||
@@ -383,17 +400,20 @@ func (t *Tree) assimilate(item scanItem) error {
|
||||
}
|
||||
|
||||
// check for the id attribute again after grabbing the lock, maybe the file was assimilated/created by us in the meantime
|
||||
_, id, mtime, err := t.lookup.MetadataBackend().IdentifyPath(context.Background(), item.Path)
|
||||
_, id, parentID, mtime, err := t.lookup.MetadataBackend().IdentifyPath(context.Background(), item.Path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if id != "" {
|
||||
// the file has an id set, we already know it from the past
|
||||
n := node.NewBaseNode(spaceID, id, t.lookup)
|
||||
// n := node.NewBaseNode(spaceID, id, t.lookup)
|
||||
|
||||
previousPath, ok := t.lookup.GetCachedID(context.Background(), spaceID, id)
|
||||
previousParentID, _ := t.lookup.MetadataBackend().Get(context.Background(), n, prefixes.ParentidAttr)
|
||||
if previousPath == "" || !ok {
|
||||
previousPath, ok = t.lookup.IDHistoryCache.Get(context.Background(), spaceID, id)
|
||||
}
|
||||
// previousParentID, _ := t.lookup.MetadataBackend().Get(context.Background(), n, prefixes.ParentidAttr)
|
||||
|
||||
// compare metadata mtime with actual mtime. if it matches AND the path hasn't changed (move operation)
|
||||
// we can skip the assimilation because the file was handled by us
|
||||
@@ -405,7 +425,7 @@ func (t *Tree) assimilate(item scanItem) error {
|
||||
}
|
||||
|
||||
// was it moved or copied/restored with a clashing id?
|
||||
if ok && len(previousParentID) > 0 && previousPath != item.Path {
|
||||
if ok && len(parentID) > 0 && previousPath != item.Path {
|
||||
_, err := os.Stat(previousPath)
|
||||
if err == nil {
|
||||
// this id clashes with an existing item -> clear metadata and re-assimilate
|
||||
@@ -445,13 +465,13 @@ func (t *Tree) assimilate(item scanItem) error {
|
||||
}()
|
||||
}
|
||||
|
||||
parentID := attrs.String(prefixes.ParentidAttr)
|
||||
newParentID := attrs.String(prefixes.ParentidAttr)
|
||||
if len(parentID) > 0 {
|
||||
ref := &provider.Reference{
|
||||
ResourceId: &provider.ResourceId{
|
||||
StorageId: t.options.MountID,
|
||||
SpaceId: spaceID,
|
||||
OpaqueId: parentID,
|
||||
OpaqueId: newParentID,
|
||||
},
|
||||
Path: filepath.Base(item.Path),
|
||||
}
|
||||
@@ -459,7 +479,7 @@ func (t *Tree) assimilate(item scanItem) error {
|
||||
ResourceId: &provider.ResourceId{
|
||||
StorageId: t.options.MountID,
|
||||
SpaceId: spaceID,
|
||||
OpaqueId: string(previousParentID),
|
||||
OpaqueId: parentID,
|
||||
},
|
||||
Path: filepath.Base(previousPath),
|
||||
}
|
||||
@@ -615,54 +635,56 @@ assimilate:
|
||||
|
||||
n.SpaceRoot = &node.Node{BaseNode: node.BaseNode{SpaceID: spaceID, ID: spaceID}}
|
||||
|
||||
go func() {
|
||||
// Copy the previous current version to a revision
|
||||
currentNode := node.NewBaseNode(n.SpaceID, n.ID+node.CurrentIDDelimiter, t.lookup)
|
||||
currentPath := currentNode.InternalPath()
|
||||
stat, err := os.Stat(currentPath)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Str("currentPath", currentPath).Msg("could not stat current path")
|
||||
return
|
||||
}
|
||||
revisionPath := t.lookup.VersionPath(n.SpaceID, n.ID, stat.ModTime().UTC().Format(time.RFC3339Nano))
|
||||
if t.options.EnableFSRevisions {
|
||||
go func() {
|
||||
// Copy the previous current version to a revision
|
||||
currentNode := node.NewBaseNode(n.SpaceID, n.ID+node.CurrentIDDelimiter, t.lookup)
|
||||
currentPath := currentNode.InternalPath()
|
||||
stat, err := os.Stat(currentPath)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Str("currentPath", currentPath).Msg("could not stat current path")
|
||||
return
|
||||
}
|
||||
revisionPath := t.lookup.VersionPath(n.SpaceID, n.ID, stat.ModTime().UTC().Format(time.RFC3339Nano))
|
||||
|
||||
err = os.Rename(currentPath, revisionPath)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Str("revisionPath", revisionPath).Msg("could not create revision")
|
||||
return
|
||||
}
|
||||
err = os.Rename(currentPath, revisionPath)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Str("revisionPath", revisionPath).Msg("could not create revision")
|
||||
return
|
||||
}
|
||||
|
||||
// Copy the new version to the current version
|
||||
w, err := os.OpenFile(currentPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Str("currentPath", currentPath).Msg("could not open current path for writing")
|
||||
return
|
||||
}
|
||||
defer w.Close()
|
||||
r, err := os.OpenFile(n.InternalPath(), os.O_RDONLY, 0600)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Msg("could not open file for reading")
|
||||
return
|
||||
}
|
||||
defer r.Close()
|
||||
// Copy the new version to the current version
|
||||
w, err := os.OpenFile(currentPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Str("currentPath", currentPath).Msg("could not open current path for writing")
|
||||
return
|
||||
}
|
||||
defer w.Close()
|
||||
r, err := os.OpenFile(n.InternalPath(), os.O_RDONLY, 0600)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("path", path).Msg("could not open file for reading")
|
||||
return
|
||||
}
|
||||
defer r.Close()
|
||||
|
||||
_, err = io.Copy(w, r)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("currentPath", currentPath).Str("path", path).Msg("could not copy new version to current version")
|
||||
return
|
||||
}
|
||||
_, err = io.Copy(w, r)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("currentPath", currentPath).Str("path", path).Msg("could not copy new version to current version")
|
||||
return
|
||||
}
|
||||
|
||||
err = t.lookup.CopyMetadata(context.Background(), n, currentNode, func(attributeName string, value []byte) (newValue []byte, copy bool) {
|
||||
return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) ||
|
||||
attributeName == prefixes.TypeAttr ||
|
||||
attributeName == prefixes.BlobIDAttr ||
|
||||
attributeName == prefixes.BlobsizeAttr
|
||||
}, false)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("currentPath", currentPath).Str("path", path).Msg("failed to copy xattrs to 'current' file")
|
||||
return
|
||||
}
|
||||
}()
|
||||
err = t.lookup.CopyMetadata(context.Background(), n, currentNode, func(attributeName string, value []byte) (newValue []byte, copy bool) {
|
||||
return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) ||
|
||||
attributeName == prefixes.TypeAttr ||
|
||||
attributeName == prefixes.BlobIDAttr ||
|
||||
attributeName == prefixes.BlobsizeAttr
|
||||
}, false)
|
||||
if err != nil {
|
||||
t.log.Error().Err(err).Str("currentPath", currentPath).Str("path", path).Msg("failed to copy xattrs to 'current' file")
|
||||
return
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
err = t.Propagate(context.Background(), n, 0)
|
||||
if err != nil {
|
||||
@@ -735,7 +757,7 @@ func (t *Tree) WarmupIDCache(root string, assimilate, onlyDirty bool) error {
|
||||
sizes[path] += 0 // Make sure to set the size to 0 for empty directories
|
||||
}
|
||||
|
||||
nodeSpaceID, id, _, err := t.lookup.MetadataBackend().IdentifyPath(context.Background(), path)
|
||||
nodeSpaceID, id, _, _, err := t.lookup.MetadataBackend().IdentifyPath(context.Background(), path)
|
||||
if err == nil && len(id) > 0 {
|
||||
if len(nodeSpaceID) > 0 {
|
||||
spaceID = nodeSpaceID
|
||||
@@ -757,7 +779,7 @@ func (t *Tree) WarmupIDCache(root string, assimilate, onlyDirty bool) error {
|
||||
break
|
||||
}
|
||||
|
||||
spaceID, _, _, err = t.lookup.MetadataBackend().IdentifyPath(context.Background(), spaceCandidate)
|
||||
spaceID, _, _, _, err = t.lookup.MetadataBackend().IdentifyPath(context.Background(), spaceCandidate)
|
||||
if err == nil && len(spaceID) > 0 {
|
||||
err = scopeSpace(path)
|
||||
if err != nil {
|
||||
@@ -791,7 +813,11 @@ func (t *Tree) WarmupIDCache(root string, assimilate, onlyDirty bool) error {
|
||||
t.log.Error().Err(err).Str("path", path).Msg("could not assimilate item")
|
||||
}
|
||||
}
|
||||
return t.setDirty(path, false)
|
||||
|
||||
if info.IsDir() {
|
||||
return t.setDirty(path, false)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
for dir, size := range sizes {
|
||||
|
||||
Generated
Vendored
+137
-5
@@ -22,24 +22,40 @@ package tree
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/options"
|
||||
"github.com/pablodz/inotifywaitgo/inotifywaitgo"
|
||||
"github.com/rs/zerolog"
|
||||
)
|
||||
|
||||
type InotifyWatcher struct {
|
||||
tree *Tree
|
||||
log *zerolog.Logger
|
||||
tree *Tree
|
||||
options *options.Options
|
||||
log *zerolog.Logger
|
||||
}
|
||||
|
||||
func NewInotifyWatcher(tree *Tree, log *zerolog.Logger) (*InotifyWatcher, error) {
|
||||
func NewInotifyWatcher(tree *Tree, o *options.Options, log *zerolog.Logger) (*InotifyWatcher, error) {
|
||||
return &InotifyWatcher{
|
||||
tree: tree,
|
||||
log: log,
|
||||
tree: tree,
|
||||
options: o,
|
||||
log: log,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (iw *InotifyWatcher) Watch(path string) {
|
||||
if iw.options.InotifyStatsFrequency > 0 {
|
||||
go func() {
|
||||
for {
|
||||
iw.printStats()
|
||||
time.Sleep(iw.options.InotifyStatsFrequency)
|
||||
}
|
||||
}()
|
||||
}
|
||||
events := make(chan inotifywaitgo.FileEvent)
|
||||
errors := make(chan error)
|
||||
|
||||
@@ -104,3 +120,119 @@ func (iw *InotifyWatcher) Watch(path string) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// InotifyUsage holds the number of inotify watches and instances.
|
||||
type InotifyUsage struct {
|
||||
Watches int
|
||||
Instances int
|
||||
MaxWatches int
|
||||
MaxInstances int
|
||||
}
|
||||
|
||||
func countInotifyFDs(pid string) (int, int, error) {
|
||||
fds, err := os.ReadDir(filepath.Join("/proc", pid, "fd"))
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return 0, 0, nil // Process may have exited, treat as 0.
|
||||
}
|
||||
return 0, 0, fmt.Errorf("failed to read /proc/%s/fd: %w", pid, err)
|
||||
}
|
||||
|
||||
watches := 0
|
||||
instances := 0
|
||||
for _, fd := range fds {
|
||||
if !fd.IsDir() {
|
||||
if fd.Type()&os.ModeSymlink == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
link, err := os.Readlink(filepath.Join("/proc", pid, "fd", fd.Name()))
|
||||
if err != nil || (link != "inotify" && link != "anon_inode:inotify") {
|
||||
continue
|
||||
}
|
||||
|
||||
instances++
|
||||
fdinfoPath := filepath.Join("/proc", pid, "fdinfo", fd.Name())
|
||||
content, err := os.ReadFile(fdinfoPath)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("failed to read %s: %w", fdinfoPath, err)
|
||||
}
|
||||
|
||||
lines := strings.SplitSeq(string(content), "\n")
|
||||
for line := range lines {
|
||||
if strings.HasPrefix(line, "inotify") {
|
||||
watches++
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return watches, instances, nil
|
||||
}
|
||||
|
||||
func GetInotifyUsageFromProc() (InotifyUsage, error) {
|
||||
usage := InotifyUsage{}
|
||||
var err error
|
||||
|
||||
usage.MaxWatches, err = readProcFile("sys/fs/inotify/max_user_watches")
|
||||
if err != nil {
|
||||
return usage, fmt.Errorf("failed to read max_user_watches: %w", err)
|
||||
}
|
||||
usage.MaxInstances, err = readProcFile("sys/fs/inotify/max_user_instances")
|
||||
if err != nil {
|
||||
return usage, fmt.Errorf("failed to read max_user_instances: %w", err)
|
||||
}
|
||||
|
||||
dirs, err := os.ReadDir("/proc")
|
||||
if err != nil {
|
||||
return usage, fmt.Errorf("failed to read /proc: %w", err)
|
||||
}
|
||||
|
||||
totalWatches := 0
|
||||
totalInstances := 0
|
||||
for _, dir := range dirs {
|
||||
if dir.IsDir() {
|
||||
pid := dir.Name()
|
||||
if _, err := strconv.Atoi(pid); err == nil {
|
||||
watches, instances, err := countInotifyFDs(pid)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
totalWatches += watches
|
||||
totalInstances += instances
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
usage.Watches = totalWatches
|
||||
usage.Instances = totalInstances
|
||||
return usage, nil
|
||||
}
|
||||
|
||||
func readProcFile(filename string) (int, error) {
|
||||
filePath := filepath.Join("/proc", filename)
|
||||
content, err := os.ReadFile(filePath)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
i, err := strconv.Atoi(strings.TrimSpace(string(content)))
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to parse max_user_watches: %w", err)
|
||||
}
|
||||
return i, nil
|
||||
}
|
||||
|
||||
func (iw *InotifyWatcher) printStats() {
|
||||
t := time.Now()
|
||||
usage, err := GetInotifyUsageFromProc()
|
||||
if err != nil {
|
||||
iw.log.Error().Err(err).Msg("failed to get inotify usage")
|
||||
return
|
||||
}
|
||||
d := time.Since(t)
|
||||
|
||||
iw.log.Info().
|
||||
Str("watches", fmt.Sprintf("%d/%d (%.2f%%)", usage.Watches, usage.MaxWatches, float64(usage.Watches)/float64(usage.MaxWatches)*100)).
|
||||
Str("instances", fmt.Sprintf("%d/%d (%.2f%%)", usage.Instances, usage.MaxInstances, float64(usage.Instances)/float64(usage.MaxInstances)*100)).
|
||||
Str("duration", d.String()).
|
||||
Msg("Inotify usage stats")
|
||||
}
|
||||
|
||||
Generated
Vendored
+2
-1
@@ -7,6 +7,7 @@ package tree
|
||||
|
||||
import (
|
||||
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/options"
|
||||
"github.com/rs/zerolog"
|
||||
)
|
||||
|
||||
@@ -17,6 +18,6 @@ type NullWatcher struct{}
|
||||
func (*NullWatcher) Watch(path string) {}
|
||||
|
||||
// NewInotifyWatcher returns a new inotify watcher
|
||||
func NewInotifyWatcher(tree *Tree, log *zerolog.Logger) (*NullWatcher, error) {
|
||||
func NewInotifyWatcher(_ *Tree, _ *options.Options, _ *zerolog.Logger) (*NullWatcher, error) {
|
||||
return nil, errtypes.NotSupported("inotify watcher is not supported on this platform")
|
||||
}
|
||||
|
||||
+15
-17
@@ -129,7 +129,7 @@ func New(lu node.PathLookup, bs node.Blobstore, um usermapper.Mapper, trashbin *
|
||||
return nil, err
|
||||
}
|
||||
default:
|
||||
t.watcher, err = NewInotifyWatcher(t, log)
|
||||
t.watcher, err = NewInotifyWatcher(t, o, log)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -299,9 +299,11 @@ func (t *Tree) Move(ctx context.Context, oldNode *node.Node, newNode *node.Node)
|
||||
return errors.Wrap(err, "Decomposedfs: Move: error deleting target node "+newNode.ID)
|
||||
}
|
||||
}
|
||||
|
||||
// we are moving the node to a new parent, any target has been removed
|
||||
// bring old node to the new parent
|
||||
oldParent := oldNode.ParentPath()
|
||||
newParent := newNode.ParentPath()
|
||||
if newNode.ID == "" {
|
||||
newNode.ID = oldNode.ID
|
||||
}
|
||||
|
||||
// update target parentid and name
|
||||
attribs := node.Attributes{}
|
||||
@@ -311,29 +313,25 @@ func (t *Tree) Move(ctx context.Context, oldNode *node.Node, newNode *node.Node)
|
||||
return errors.Wrap(err, "Decomposedfs: could not update old node attributes")
|
||||
}
|
||||
|
||||
// rename node
|
||||
err = os.Rename(
|
||||
filepath.Join(oldNode.ParentPath(), oldNode.Name),
|
||||
filepath.Join(newNode.ParentPath(), newNode.Name),
|
||||
)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "Decomposedfs: could not move child")
|
||||
}
|
||||
|
||||
// update the id cache
|
||||
if newNode.ID == "" {
|
||||
newNode.ID = oldNode.ID
|
||||
}
|
||||
// invalidate old tree
|
||||
err = t.lookup.IDCache.DeleteByPath(ctx, filepath.Join(oldNode.ParentPath(), oldNode.Name))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := t.lookup.CacheID(ctx, newNode.SpaceID, newNode.ID, filepath.Join(newNode.ParentPath(), newNode.Name)); err != nil {
|
||||
t.log.Error().Err(err).Str("spaceID", newNode.SpaceID).Str("id", newNode.ID).Str("path", filepath.Join(newNode.ParentPath(), newNode.Name)).Msg("could not cache id")
|
||||
}
|
||||
|
||||
// rename node
|
||||
err = os.Rename(
|
||||
filepath.Join(oldParent, oldNode.Name),
|
||||
filepath.Join(newParent, newNode.Name),
|
||||
)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "Decomposedfs: could not move child")
|
||||
}
|
||||
|
||||
// rename the lock (if it exists)
|
||||
if _, err := os.Stat(lockFilePath); err == nil {
|
||||
err = os.Rename(lockFilePath, newNode.LockFilePath())
|
||||
|
||||
Generated
Vendored
+3
-2
@@ -45,13 +45,14 @@ func NewHybridBackend(offloadLimit int, metadataPathFunc MetadataPathFunc, o cac
|
||||
func (HybridBackend) Name() string { return "hybrid" }
|
||||
|
||||
// IdentifyPath returns the space id, node id and mtime of a file
|
||||
func (b HybridBackend) IdentifyPath(_ context.Context, path string) (string, string, time.Time, error) {
|
||||
func (b HybridBackend) IdentifyPath(_ context.Context, path string) (string, string, string, time.Time, error) {
|
||||
spaceID, _ := xattr.Get(path, prefixes.SpaceIDAttr)
|
||||
id, _ := xattr.Get(path, prefixes.IDAttr)
|
||||
parentID, _ := xattr.Get(path, prefixes.ParentidAttr)
|
||||
|
||||
mtimeAttr, _ := xattr.Get(path, prefixes.MTimeAttr)
|
||||
mtime, _ := time.Parse(time.RFC3339Nano, string(mtimeAttr))
|
||||
return string(spaceID), string(id), mtime, nil
|
||||
return string(spaceID), string(id), string(parentID), mtime, nil
|
||||
}
|
||||
|
||||
// Get an extended attribute value for the given key
|
||||
|
||||
Generated
Vendored
+7
-6
@@ -54,33 +54,34 @@ func NewMessagePackBackend(o cache.Config) MessagePackBackend {
|
||||
func (MessagePackBackend) Name() string { return "messagepack" }
|
||||
|
||||
// IdentifyPath returns the id and mtime of a file
|
||||
func (b MessagePackBackend) IdentifyPath(_ context.Context, path string) (string, string, time.Time, error) {
|
||||
func (b MessagePackBackend) IdentifyPath(_ context.Context, path string) (string, string, string, time.Time, error) {
|
||||
metaPath := filepath.Clean(path + ".mpk")
|
||||
source, err := os.Open(metaPath)
|
||||
// // No cached entry found. Read from storage and store in cache
|
||||
if err != nil {
|
||||
return "", "", time.Time{}, err
|
||||
return "", "", "", time.Time{}, err
|
||||
}
|
||||
msgBytes, err := io.ReadAll(source)
|
||||
if err != nil || len(msgBytes) == 0 {
|
||||
return "", "", time.Time{}, err
|
||||
return "", "", "", time.Time{}, err
|
||||
|
||||
}
|
||||
attribs := map[string][]byte{}
|
||||
err = msgpack.Unmarshal(msgBytes, &attribs)
|
||||
if err != nil {
|
||||
return "", "", time.Time{}, err
|
||||
return "", "", "", time.Time{}, err
|
||||
}
|
||||
|
||||
spaceID := attribs[prefixes.IDAttr]
|
||||
id := attribs[prefixes.IDAttr]
|
||||
parentID := attribs[prefixes.ParentidAttr]
|
||||
|
||||
mtimeAttr := attribs[prefixes.MTimeAttr]
|
||||
mtime, err := time.Parse(time.RFC3339Nano, string(mtimeAttr))
|
||||
if err != nil {
|
||||
return "", "", time.Time{}, err
|
||||
return "", "", "", time.Time{}, err
|
||||
}
|
||||
return string(spaceID), string(id), mtime, nil
|
||||
return string(spaceID), string(id), string(parentID), mtime, nil
|
||||
}
|
||||
|
||||
// All reads all extended attributes for a node
|
||||
|
||||
Generated
Vendored
+3
-3
@@ -47,7 +47,7 @@ type MetadataNode interface {
|
||||
// Backend defines the interface for file attribute backends
|
||||
type Backend interface {
|
||||
Name() string
|
||||
IdentifyPath(ctx context.Context, path string) (string, string, time.Time, error)
|
||||
IdentifyPath(ctx context.Context, path string) (string, string, string, time.Time, error)
|
||||
|
||||
All(ctx context.Context, n MetadataNode) (map[string][]byte, error)
|
||||
AllWithLockedSource(ctx context.Context, n MetadataNode, source io.Reader) (map[string][]byte, error)
|
||||
@@ -74,8 +74,8 @@ type NullBackend struct{}
|
||||
func (NullBackend) Name() string { return "null" }
|
||||
|
||||
// IdentifyPath returns the ids and mtime of a file
|
||||
func (NullBackend) IdentifyPath(ctx context.Context, path string) (string, string, time.Time, error) {
|
||||
return "", "", time.Time{}, errUnconfiguredError
|
||||
func (NullBackend) IdentifyPath(ctx context.Context, path string) (string, string, string, time.Time, error) {
|
||||
return "", "", "", time.Time{}, errUnconfiguredError
|
||||
}
|
||||
|
||||
// All reads all extended attributes for a node
|
||||
|
||||
Generated
Vendored
+3
-2
@@ -51,13 +51,14 @@ func NewXattrsBackend(o cache.Config) XattrsBackend {
|
||||
func (XattrsBackend) Name() string { return "xattrs" }
|
||||
|
||||
// IdentifyPath returns the space id, node id and mtime of a file
|
||||
func (b XattrsBackend) IdentifyPath(_ context.Context, path string) (string, string, time.Time, error) {
|
||||
func (b XattrsBackend) IdentifyPath(_ context.Context, path string) (string, string, string, time.Time, error) {
|
||||
spaceID, _ := xattr.Get(path, prefixes.SpaceIDAttr)
|
||||
id, _ := xattr.Get(path, prefixes.IDAttr)
|
||||
parentID, _ := xattr.Get(path, prefixes.ParentidAttr)
|
||||
|
||||
mtimeAttr, _ := xattr.Get(path, prefixes.MTimeAttr)
|
||||
mtime, _ := time.Parse(time.RFC3339Nano, string(mtimeAttr))
|
||||
return string(spaceID), string(id), mtime, nil
|
||||
return string(spaceID), string(id), string(parentID), mtime, nil
|
||||
}
|
||||
|
||||
// Get an extended attribute value for the given key
|
||||
|
||||
+24
-8
@@ -178,7 +178,9 @@ type BaseNode struct {
|
||||
SpaceID string
|
||||
ID string
|
||||
|
||||
lu PathLookup
|
||||
lu PathLookup
|
||||
internalPathID string
|
||||
internalPath string
|
||||
}
|
||||
|
||||
func NewBaseNode(spaceID, nodeID string, lu PathLookup) *BaseNode {
|
||||
@@ -194,7 +196,13 @@ func (n *BaseNode) GetID() string { return n.ID }
|
||||
|
||||
// InternalPath returns the internal path of the Node
|
||||
func (n *BaseNode) InternalPath() string {
|
||||
return n.lu.InternalPath(n.SpaceID, n.ID)
|
||||
if len(n.internalPath) > 0 && n.ID == n.internalPathID {
|
||||
return n.internalPath
|
||||
}
|
||||
|
||||
n.internalPath = n.lu.InternalPath(n.SpaceID, n.ID)
|
||||
n.internalPathID = n.ID
|
||||
return n.internalPath
|
||||
}
|
||||
|
||||
// Node represents a node in the tree and provides methods to get a Parent or Child instance
|
||||
@@ -209,6 +217,7 @@ type Node struct {
|
||||
SpaceRoot *Node
|
||||
|
||||
xattrsCache map[string][]byte
|
||||
disabled *bool
|
||||
nodeType *provider.ResourceType
|
||||
}
|
||||
|
||||
@@ -912,7 +921,7 @@ func (n *Node) AsResourceInfo(ctx context.Context, rp *provider.ResourcePermissi
|
||||
ri.Opaque = utils.AppendPlainToOpaque(ri.Opaque, "scantime", date.Format(time.RFC3339Nano))
|
||||
}
|
||||
|
||||
sublog.Debug().
|
||||
sublog.Trace().
|
||||
Interface("ri", ri).
|
||||
Msg("AsResourceInfo")
|
||||
|
||||
@@ -978,7 +987,7 @@ func (n *Node) readQuotaIntoOpaque(ctx context.Context, ri *provider.ResourceInf
|
||||
appctx.GetLogger(ctx).Error().Err(err).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("nodepath", n.InternalPath()).Str("quota", v).Msg("malformed quota")
|
||||
}
|
||||
case metadata.IsAttrUnset(err):
|
||||
appctx.GetLogger(ctx).Debug().Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("nodepath", n.InternalPath()).Msg("quota not set")
|
||||
appctx.GetLogger(ctx).Trace().Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("nodepath", n.InternalPath()).Msg("quota not set")
|
||||
default:
|
||||
appctx.GetLogger(ctx).Error().Err(err).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("nodepath", n.InternalPath()).Msg("could not read quota")
|
||||
}
|
||||
@@ -996,10 +1005,17 @@ func (n *Node) HasPropagation(ctx context.Context) (propagation bool) {
|
||||
// only used to check if a space is disabled
|
||||
// FIXME confusing with the trash logic
|
||||
func (n *Node) IsDisabled(ctx context.Context) bool {
|
||||
if _, err := n.GetDTime(ctx); err == nil {
|
||||
return true
|
||||
if n.disabled != nil {
|
||||
return *n.disabled
|
||||
}
|
||||
return false
|
||||
if _, err := n.GetDTime(ctx); err == nil {
|
||||
v := true
|
||||
n.disabled = &v
|
||||
} else {
|
||||
v := false
|
||||
n.disabled = &v
|
||||
}
|
||||
return *n.disabled
|
||||
}
|
||||
|
||||
// GetTreeSize reads the treesize from the extended attributes
|
||||
@@ -1116,7 +1132,7 @@ func (n *Node) ReadUserPermissions(ctx context.Context, u *userpb.User) (ap *pro
|
||||
}
|
||||
}
|
||||
|
||||
appctx.GetLogger(ctx).Debug().Interface("permissions", ap).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Interface("user", u).Msg("returning aggregated permissions")
|
||||
appctx.GetLogger(ctx).Trace().Interface("permissions", ap).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Interface("user", u).Msg("returning aggregated permissions")
|
||||
return ap, false, nil
|
||||
}
|
||||
|
||||
|
||||
+14
-3
@@ -73,8 +73,12 @@ func (fs *Decomposedfs) Upload(ctx context.Context, req storage.UploadRequest, u
|
||||
if err != nil {
|
||||
return &provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error opening assembled file")
|
||||
}
|
||||
defer fd.Close()
|
||||
defer os.RemoveAll(assembledFile)
|
||||
defer func() {
|
||||
_ = fd.Close()
|
||||
}()
|
||||
defer func() {
|
||||
_ = os.RemoveAll(assembledFile)
|
||||
}()
|
||||
req.Body = fd
|
||||
|
||||
size, err := session.WriteChunk(ctx, 0, req.Body)
|
||||
@@ -347,6 +351,7 @@ func (fs *Decomposedfs) UseIn(composer *tusd.StoreComposer) {
|
||||
composer.UseTerminater(fs)
|
||||
composer.UseConcater(fs)
|
||||
composer.UseLengthDeferrer(fs)
|
||||
composer.UseContentServer(fs)
|
||||
}
|
||||
|
||||
// To implement the core tus.io protocol as specified in https://tus.io/protocols/resumable-upload.html#core-protocol
|
||||
@@ -354,10 +359,16 @@ func (fs *Decomposedfs) UseIn(composer *tusd.StoreComposer) {
|
||||
// - the upload needs to implement the tusd.Upload interface: WriteChunk, GetInfo, GetReader and FinishUpload
|
||||
|
||||
// NewUpload returns a new tus Upload instance
|
||||
func (fs *Decomposedfs) NewUpload(ctx context.Context, info tusd.FileInfo) (tusd.Upload, error) {
|
||||
func (fs *Decomposedfs) NewUpload(_ context.Context, _ tusd.FileInfo) (tusd.Upload, error) {
|
||||
return nil, fmt.Errorf("not implemented, use InitiateUpload on the CS3 API to start a new upload")
|
||||
}
|
||||
|
||||
// AsServableUpload returns a ServableUpload
|
||||
// which implements the tusd.ServableUpload interface and
|
||||
func (fs *Decomposedfs) AsServableUpload(u tusd.Upload) tusd.ServableUpload {
|
||||
return u.(*upload.DecomposedFsSession)
|
||||
}
|
||||
|
||||
// GetUpload returns the Upload for the given upload id
|
||||
func (fs *Decomposedfs) GetUpload(ctx context.Context, id string) (tusd.Upload, error) {
|
||||
var ul tusd.Upload
|
||||
|
||||
Generated
Vendored
+98
-97
@@ -32,6 +32,7 @@ import (
|
||||
userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
|
||||
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
|
||||
typespb "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
|
||||
|
||||
"github.com/opencloud-eu/reva/v2/pkg/appctx"
|
||||
ctxpkg "github.com/opencloud-eu/reva/v2/pkg/ctx"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/node"
|
||||
@@ -46,49 +47,49 @@ type DecomposedFsSession struct {
|
||||
}
|
||||
|
||||
// Context returns a context with the user, logger and lockid used when initiating the upload session
|
||||
func (s *DecomposedFsSession) Context(ctx context.Context) context.Context { // restore logger from file info
|
||||
sub := s.store.log.With().Int("pid", os.Getpid()).Logger()
|
||||
func (session *DecomposedFsSession) Context(ctx context.Context) context.Context { // restore logger from file info
|
||||
sub := session.store.log.With().Int("pid", os.Getpid()).Logger()
|
||||
ctx = appctx.WithLogger(ctx, &sub)
|
||||
ctx = ctxpkg.ContextSetLockID(ctx, s.lockID())
|
||||
ctx = ctxpkg.ContextSetUser(ctx, s.executantUser())
|
||||
return ctxpkg.ContextSetInitiator(ctx, s.InitiatorID())
|
||||
ctx = ctxpkg.ContextSetLockID(ctx, session.lockID())
|
||||
ctx = ctxpkg.ContextSetUser(ctx, session.executantUser())
|
||||
return ctxpkg.ContextSetInitiator(ctx, session.InitiatorID())
|
||||
}
|
||||
|
||||
func (s *DecomposedFsSession) lockID() string {
|
||||
return s.info.MetaData["lockid"]
|
||||
func (session *DecomposedFsSession) lockID() string {
|
||||
return session.info.MetaData["lockid"]
|
||||
}
|
||||
func (s *DecomposedFsSession) executantUser() *userpb.User {
|
||||
func (session *DecomposedFsSession) executantUser() *userpb.User {
|
||||
var o *typespb.Opaque
|
||||
_ = json.Unmarshal([]byte(s.info.Storage["UserOpaque"]), &o)
|
||||
_ = json.Unmarshal([]byte(session.info.Storage["UserOpaque"]), &o)
|
||||
return &userpb.User{
|
||||
Id: &userpb.UserId{
|
||||
Type: userpb.UserType(userpb.UserType_value[s.info.Storage["UserType"]]),
|
||||
Idp: s.info.Storage["Idp"],
|
||||
OpaqueId: s.info.Storage["UserId"],
|
||||
Type: userpb.UserType(userpb.UserType_value[session.info.Storage["UserType"]]),
|
||||
Idp: session.info.Storage["Idp"],
|
||||
OpaqueId: session.info.Storage["UserId"],
|
||||
},
|
||||
Username: s.info.Storage["UserName"],
|
||||
DisplayName: s.info.Storage["UserDisplayName"],
|
||||
Username: session.info.Storage["UserName"],
|
||||
DisplayName: session.info.Storage["UserDisplayName"],
|
||||
Opaque: o,
|
||||
}
|
||||
}
|
||||
|
||||
// Purge deletes the upload session metadata and written binary data
|
||||
func (s *DecomposedFsSession) Purge(ctx context.Context) error {
|
||||
func (session *DecomposedFsSession) Purge(ctx context.Context) error {
|
||||
_, span := tracer.Start(ctx, "Purge")
|
||||
defer span.End()
|
||||
sessionPath := sessionPath(s.store.root, s.info.ID)
|
||||
sessionPath := sessionPath(session.store.root, session.info.ID)
|
||||
if err := os.Remove(sessionPath); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.Remove(s.binPath()); err != nil {
|
||||
if err := os.Remove(session.binPath()); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// TouchBin creates a file to contain the binary data. It's size will be used to keep track of the tus upload offset.
|
||||
func (s *DecomposedFsSession) TouchBin() error {
|
||||
file, err := os.OpenFile(s.binPath(), os.O_CREATE|os.O_WRONLY, defaultFilePerm)
|
||||
func (session *DecomposedFsSession) TouchBin() error {
|
||||
file, err := os.OpenFile(session.binPath(), os.O_CREATE|os.O_WRONLY, defaultFilePerm)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -98,17 +99,17 @@ func (s *DecomposedFsSession) TouchBin() error {
|
||||
// Persist writes the upload session metadata to disk
|
||||
// events can update the scan outcome and the finished event might read an empty file because of race conditions
|
||||
// so we need to lock the file while writing and use atomic writes
|
||||
func (s *DecomposedFsSession) Persist(ctx context.Context) error {
|
||||
func (session *DecomposedFsSession) Persist(ctx context.Context) error {
|
||||
_, span := tracer.Start(ctx, "Persist")
|
||||
defer span.End()
|
||||
sessionPath := sessionPath(s.store.root, s.info.ID)
|
||||
sessionPath := sessionPath(session.store.root, session.info.ID)
|
||||
// create folder structure (if needed)
|
||||
if err := os.MkdirAll(filepath.Dir(sessionPath), 0700); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var d []byte
|
||||
d, err := json.Marshal(s.info)
|
||||
d, err := json.Marshal(session.info)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -116,28 +117,28 @@ func (s *DecomposedFsSession) Persist(ctx context.Context) error {
|
||||
}
|
||||
|
||||
// ToFileInfo returns tus compatible FileInfo so the tus handler can access the upload offset
|
||||
func (s *DecomposedFsSession) ToFileInfo() tusd.FileInfo {
|
||||
return s.info
|
||||
func (session *DecomposedFsSession) ToFileInfo() tusd.FileInfo {
|
||||
return session.info
|
||||
}
|
||||
|
||||
// ProviderID returns the provider id
|
||||
func (s *DecomposedFsSession) ProviderID() string {
|
||||
return s.info.MetaData["providerID"]
|
||||
func (session *DecomposedFsSession) ProviderID() string {
|
||||
return session.info.MetaData["providerID"]
|
||||
}
|
||||
|
||||
// SpaceID returns the space id
|
||||
func (s *DecomposedFsSession) SpaceID() string {
|
||||
return s.info.Storage["SpaceRoot"]
|
||||
func (session *DecomposedFsSession) SpaceID() string {
|
||||
return session.info.Storage["SpaceRoot"]
|
||||
}
|
||||
|
||||
// NodeID returns the node id
|
||||
func (s *DecomposedFsSession) NodeID() string {
|
||||
return s.info.Storage["NodeId"]
|
||||
func (session *DecomposedFsSession) NodeID() string {
|
||||
return session.info.Storage["NodeId"]
|
||||
}
|
||||
|
||||
// NodeParentID returns the nodes parent id
|
||||
func (s *DecomposedFsSession) NodeParentID() string {
|
||||
return s.info.Storage["NodeParentId"]
|
||||
func (session *DecomposedFsSession) NodeParentID() string {
|
||||
return session.info.Storage["NodeParentId"]
|
||||
}
|
||||
|
||||
// NodeExists returns wether or not the node existed during InitiateUpload.
|
||||
@@ -148,63 +149,63 @@ func (s *DecomposedFsSession) NodeParentID() string {
|
||||
// A node should be created as part of InitiateUpload. When listing a directory
|
||||
// we can decide if we want to skip the entry, or expose uploed progress
|
||||
// information. But that is a bigger change and might involve client work.
|
||||
func (s *DecomposedFsSession) NodeExists() bool {
|
||||
return s.info.Storage["NodeExists"] == "true"
|
||||
func (session *DecomposedFsSession) NodeExists() bool {
|
||||
return session.info.Storage["NodeExists"] == "true"
|
||||
}
|
||||
|
||||
// HeaderIfMatch returns the if-match header for the upload session
|
||||
func (s *DecomposedFsSession) HeaderIfMatch() string {
|
||||
return s.info.MetaData["if-match"]
|
||||
func (session *DecomposedFsSession) HeaderIfMatch() string {
|
||||
return session.info.MetaData["if-match"]
|
||||
}
|
||||
|
||||
// HeaderIfNoneMatch returns the if-none-match header for the upload session
|
||||
func (s *DecomposedFsSession) HeaderIfNoneMatch() string {
|
||||
return s.info.MetaData["if-none-match"]
|
||||
func (session *DecomposedFsSession) HeaderIfNoneMatch() string {
|
||||
return session.info.MetaData["if-none-match"]
|
||||
}
|
||||
|
||||
// HeaderIfUnmodifiedSince returns the if-unmodified-since header for the upload session
|
||||
func (s *DecomposedFsSession) HeaderIfUnmodifiedSince() string {
|
||||
return s.info.MetaData["if-unmodified-since"]
|
||||
func (session *DecomposedFsSession) HeaderIfUnmodifiedSince() string {
|
||||
return session.info.MetaData["if-unmodified-since"]
|
||||
}
|
||||
|
||||
// Node returns the node for the session
|
||||
func (s *DecomposedFsSession) Node(ctx context.Context) (*node.Node, error) {
|
||||
return node.ReadNode(ctx, s.store.lu, s.SpaceID(), s.info.Storage["NodeId"], false, nil, true)
|
||||
func (session *DecomposedFsSession) Node(ctx context.Context) (*node.Node, error) {
|
||||
return node.ReadNode(ctx, session.store.lu, session.SpaceID(), session.info.Storage["NodeId"], false, nil, true)
|
||||
}
|
||||
|
||||
// ID returns the upload session id
|
||||
func (s *DecomposedFsSession) ID() string {
|
||||
return s.info.ID
|
||||
func (session *DecomposedFsSession) ID() string {
|
||||
return session.info.ID
|
||||
}
|
||||
|
||||
// Filename returns the name of the node which is not the same as the name af the file being uploaded for legacy chunked uploads
|
||||
func (s *DecomposedFsSession) Filename() string {
|
||||
return s.info.Storage["NodeName"]
|
||||
func (session *DecomposedFsSession) Filename() string {
|
||||
return session.info.Storage["NodeName"]
|
||||
}
|
||||
|
||||
// Chunk returns the chunk name when a legacy chunked upload was started
|
||||
func (s *DecomposedFsSession) Chunk() string {
|
||||
return s.info.Storage["Chunk"]
|
||||
func (session *DecomposedFsSession) Chunk() string {
|
||||
return session.info.Storage["Chunk"]
|
||||
}
|
||||
|
||||
// SetMetadata is used to fill the upload metadata that will be exposed to the end user
|
||||
func (s *DecomposedFsSession) SetMetadata(key, value string) {
|
||||
s.info.MetaData[key] = value
|
||||
func (session *DecomposedFsSession) SetMetadata(key, value string) {
|
||||
session.info.MetaData[key] = value
|
||||
}
|
||||
|
||||
// SetStorageValue is used to set metadata only relevant for the upload session implementation
|
||||
func (s *DecomposedFsSession) SetStorageValue(key, value string) {
|
||||
s.info.Storage[key] = value
|
||||
func (session *DecomposedFsSession) SetStorageValue(key, value string) {
|
||||
session.info.Storage[key] = value
|
||||
}
|
||||
|
||||
// SetSize will set the upload size of the underlying tus info.
|
||||
func (s *DecomposedFsSession) SetSize(size int64) {
|
||||
s.info.Size = size
|
||||
func (session *DecomposedFsSession) SetSize(size int64) {
|
||||
session.info.Size = size
|
||||
}
|
||||
|
||||
// SetSizeIsDeferred is uset to change the SizeIsDeferred property of the underlying tus info.
|
||||
func (s *DecomposedFsSession) SetSizeIsDeferred(value bool) {
|
||||
s.info.SizeIsDeferred = value
|
||||
func (session *DecomposedFsSession) SetSizeIsDeferred(value bool) {
|
||||
session.info.SizeIsDeferred = value
|
||||
}
|
||||
|
||||
// Dir returns the directory to which the upload is made
|
||||
@@ -227,115 +228,115 @@ func (s *DecomposedFsSession) SetSizeIsDeferred(value bool) {
|
||||
//
|
||||
// I think we can safely determine the path later, right before emitting the
|
||||
// event. And maybe make it configurable, because only audit needs it, anyway.
|
||||
func (s *DecomposedFsSession) Dir() string {
|
||||
return s.info.Storage["Dir"]
|
||||
func (session *DecomposedFsSession) Dir() string {
|
||||
return session.info.Storage["Dir"]
|
||||
}
|
||||
|
||||
// Size returns the upload size
|
||||
func (s *DecomposedFsSession) Size() int64 {
|
||||
return s.info.Size
|
||||
func (session *DecomposedFsSession) Size() int64 {
|
||||
return session.info.Size
|
||||
}
|
||||
|
||||
// SizeDiff returns the size diff that was calculated after postprocessing
|
||||
func (s *DecomposedFsSession) SizeDiff() int64 {
|
||||
sizeDiff, _ := strconv.ParseInt(s.info.MetaData["sizeDiff"], 10, 64)
|
||||
func (session *DecomposedFsSession) SizeDiff() int64 {
|
||||
sizeDiff, _ := strconv.ParseInt(session.info.MetaData["sizeDiff"], 10, 64)
|
||||
return sizeDiff
|
||||
}
|
||||
|
||||
// Reference returns a reference that can be used to access the uploaded resource
|
||||
func (s *DecomposedFsSession) Reference() provider.Reference {
|
||||
func (session *DecomposedFsSession) Reference() provider.Reference {
|
||||
return provider.Reference{
|
||||
ResourceId: &provider.ResourceId{
|
||||
StorageId: s.info.MetaData["providerID"],
|
||||
SpaceId: s.info.Storage["SpaceRoot"],
|
||||
OpaqueId: s.info.Storage["NodeId"],
|
||||
StorageId: session.info.MetaData["providerID"],
|
||||
SpaceId: session.info.Storage["SpaceRoot"],
|
||||
OpaqueId: session.info.Storage["NodeId"],
|
||||
},
|
||||
// Path is not used
|
||||
}
|
||||
}
|
||||
|
||||
// Executant returns the id of the user that initiated the upload session
|
||||
func (s *DecomposedFsSession) Executant() userpb.UserId {
|
||||
func (session *DecomposedFsSession) Executant() userpb.UserId {
|
||||
return userpb.UserId{
|
||||
Type: userpb.UserType(userpb.UserType_value[s.info.Storage["UserType"]]),
|
||||
Idp: s.info.Storage["Idp"],
|
||||
OpaqueId: s.info.Storage["UserId"],
|
||||
Type: userpb.UserType(userpb.UserType_value[session.info.Storage["UserType"]]),
|
||||
Idp: session.info.Storage["Idp"],
|
||||
OpaqueId: session.info.Storage["UserId"],
|
||||
}
|
||||
}
|
||||
|
||||
// SetExecutant is used to remember the user that initiated the upload session
|
||||
func (s *DecomposedFsSession) SetExecutant(u *userpb.User) {
|
||||
s.info.Storage["Idp"] = u.GetId().GetIdp()
|
||||
s.info.Storage["UserId"] = u.GetId().GetOpaqueId()
|
||||
s.info.Storage["UserType"] = utils.UserTypeToString(u.GetId().Type)
|
||||
s.info.Storage["UserName"] = u.GetUsername()
|
||||
s.info.Storage["UserDisplayName"] = u.GetDisplayName()
|
||||
func (session *DecomposedFsSession) SetExecutant(u *userpb.User) {
|
||||
session.info.Storage["Idp"] = u.GetId().GetIdp()
|
||||
session.info.Storage["UserId"] = u.GetId().GetOpaqueId()
|
||||
session.info.Storage["UserType"] = utils.UserTypeToString(u.GetId().Type)
|
||||
session.info.Storage["UserName"] = u.GetUsername()
|
||||
session.info.Storage["UserDisplayName"] = u.GetDisplayName()
|
||||
|
||||
b, _ := json.Marshal(u.GetOpaque())
|
||||
s.info.Storage["UserOpaque"] = string(b)
|
||||
session.info.Storage["UserOpaque"] = string(b)
|
||||
}
|
||||
|
||||
// Offset returns the current upload offset
|
||||
func (s *DecomposedFsSession) Offset() int64 {
|
||||
return s.info.Offset
|
||||
func (session *DecomposedFsSession) Offset() int64 {
|
||||
return session.info.Offset
|
||||
}
|
||||
|
||||
// SpaceOwner returns the id of the space owner
|
||||
func (s *DecomposedFsSession) SpaceOwner() *userpb.UserId {
|
||||
func (session *DecomposedFsSession) SpaceOwner() *userpb.UserId {
|
||||
return &userpb.UserId{
|
||||
// idp and type do not seem to be consumed and the node currently only stores the user id anyway
|
||||
OpaqueId: s.info.Storage["SpaceOwnerOrManager"],
|
||||
OpaqueId: session.info.Storage["SpaceOwnerOrManager"],
|
||||
}
|
||||
}
|
||||
|
||||
// Expires returns the time the upload session expires
|
||||
func (s *DecomposedFsSession) Expires() time.Time {
|
||||
func (session *DecomposedFsSession) Expires() time.Time {
|
||||
var t time.Time
|
||||
if value, ok := s.info.MetaData["expires"]; ok {
|
||||
if value, ok := session.info.MetaData["expires"]; ok {
|
||||
t, _ = utils.MTimeToTime(value)
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
// MTime returns the mtime to use for the uploaded file
|
||||
func (s *DecomposedFsSession) MTime() time.Time {
|
||||
func (session *DecomposedFsSession) MTime() time.Time {
|
||||
var t time.Time
|
||||
if value, ok := s.info.MetaData["mtime"]; ok {
|
||||
if value, ok := session.info.MetaData["mtime"]; ok {
|
||||
t, _ = utils.MTimeToTime(value)
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
// IsProcessing returns true if all bytes have been received. The session then has entered postprocessing state.
|
||||
func (s *DecomposedFsSession) IsProcessing() bool {
|
||||
func (session *DecomposedFsSession) IsProcessing() bool {
|
||||
// We might need a more sophisticated way to determine processing status soon
|
||||
return s.info.Size == s.info.Offset && s.info.MetaData["scanResult"] == ""
|
||||
return session.info.Size == session.info.Offset && session.info.MetaData["scanResult"] == ""
|
||||
}
|
||||
|
||||
// binPath returns the path to the file storing the binary data.
|
||||
func (s *DecomposedFsSession) binPath() string {
|
||||
return filepath.Join(s.store.root, "uploads", s.info.ID)
|
||||
func (session *DecomposedFsSession) binPath() string {
|
||||
return filepath.Join(session.store.root, "uploads", session.info.ID)
|
||||
}
|
||||
|
||||
// InitiatorID returns the id of the initiating client
|
||||
func (s *DecomposedFsSession) InitiatorID() string {
|
||||
return s.info.MetaData["initiatorid"]
|
||||
func (session *DecomposedFsSession) InitiatorID() string {
|
||||
return session.info.MetaData["initiatorid"]
|
||||
}
|
||||
|
||||
// SetScanData sets virus scan data to the upload session
|
||||
func (s *DecomposedFsSession) SetScanData(result string, date time.Time) {
|
||||
s.info.MetaData["scanResult"] = result
|
||||
s.info.MetaData["scanDate"] = date.Format(time.RFC3339)
|
||||
func (session *DecomposedFsSession) SetScanData(result string, date time.Time) {
|
||||
session.info.MetaData["scanResult"] = result
|
||||
session.info.MetaData["scanDate"] = date.Format(time.RFC3339)
|
||||
}
|
||||
|
||||
// ScanData returns the virus scan data
|
||||
func (s *DecomposedFsSession) ScanData() (string, time.Time) {
|
||||
date := s.info.MetaData["scanDate"]
|
||||
func (session *DecomposedFsSession) ScanData() (string, time.Time) {
|
||||
date := session.info.MetaData["scanDate"]
|
||||
if date == "" {
|
||||
return "", time.Time{}
|
||||
}
|
||||
d, _ := time.Parse(time.RFC3339, date)
|
||||
return s.info.MetaData["scanResult"], d
|
||||
return session.info.MetaData["scanResult"], d
|
||||
}
|
||||
|
||||
// sessionPath returns the path to the .info file storing the file's info.
|
||||
|
||||
Generated
Vendored
+67
-10
@@ -44,15 +44,15 @@ import (
|
||||
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/events"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/rhttp/datatx/metrics"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/rhttp/datatx/utils/download"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/metadata/prefixes"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/node"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/utils"
|
||||
)
|
||||
|
||||
var (
|
||||
tracer trace.Tracer
|
||||
ErrAlreadyExists = tusd.NewError("ERR_ALREADY_EXISTS", "file already exists", http.StatusConflict)
|
||||
defaultFilePerm = os.FileMode(0664)
|
||||
tracer trace.Tracer
|
||||
defaultFilePerm = os.FileMode(0664)
|
||||
)
|
||||
|
||||
func init() {
|
||||
@@ -60,7 +60,7 @@ func init() {
|
||||
}
|
||||
|
||||
// WriteChunk writes the stream from the reader to the given offset of the upload
|
||||
func (session *DecomposedFsSession) WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error) {
|
||||
func (session *DecomposedFsSession) WriteChunk(ctx context.Context, _ int64, src io.Reader) (int64, error) {
|
||||
ctx, span := tracer.Start(session.Context(ctx), "WriteChunk")
|
||||
defer span.End()
|
||||
_, subspan := tracer.Start(ctx, "os.OpenFile")
|
||||
@@ -69,7 +69,9 @@ func (session *DecomposedFsSession) WriteChunk(ctx context.Context, offset int64
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer file.Close()
|
||||
defer func() {
|
||||
_ = file.Close()
|
||||
}()
|
||||
|
||||
// calculate cheksum here? needed for the TUS checksum extension. https://tus.io/protocols/resumable-upload.html#checksum
|
||||
// TODO but how do we get the `Upload-Checksum`? WriteChunk() only has a context, offset and the reader ...
|
||||
@@ -259,7 +261,9 @@ func (session *DecomposedFsSession) ConcatUploads(_ context.Context, uploads []t
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer file.Close()
|
||||
defer func() {
|
||||
_ = file.Close()
|
||||
}()
|
||||
|
||||
for _, partialUpload := range uploads {
|
||||
fileUpload := partialUpload.(*DecomposedFsSession)
|
||||
@@ -268,7 +272,9 @@ func (session *DecomposedFsSession) ConcatUploads(_ context.Context, uploads []t
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer src.Close()
|
||||
defer func() {
|
||||
_ = src.Close()
|
||||
}()
|
||||
|
||||
if _, err := io.Copy(file, src); err != nil {
|
||||
return err
|
||||
@@ -298,9 +304,9 @@ func (session *DecomposedFsSession) Finalize(ctx context.Context) (err error) {
|
||||
}
|
||||
|
||||
func checkHash(expected string, h hash.Hash) error {
|
||||
hash := hex.EncodeToString(h.Sum(nil))
|
||||
if expected != hash {
|
||||
return errtypes.ChecksumMismatch(fmt.Sprintf("invalid checksum: expected %s got %x", expected, hash))
|
||||
shash := hex.EncodeToString(h.Sum(nil))
|
||||
if expected != shash {
|
||||
return errtypes.ChecksumMismatch(fmt.Sprintf("invalid checksum: expected %s got %x", expected, shash))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -399,6 +405,57 @@ func (session *DecomposedFsSession) URL(_ context.Context) (string, error) {
|
||||
return joinurl(session.store.tknopts.DataGatewayEndpoint, tkn), nil
|
||||
}
|
||||
|
||||
// ServeContent serves the content of the upload and implements the http.ServeContent interface needed by tusd,
|
||||
// it is used by the tusd handler to serve the content of the upload and supports range requests
|
||||
func (session *DecomposedFsSession) ServeContent(ctx context.Context, w http.ResponseWriter, req *http.Request) error {
|
||||
_, span := tracer.Start(session.Context(ctx), "ServeContent")
|
||||
defer span.End()
|
||||
|
||||
f, err := os.Open(session.binPath())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
_ = f.Close()
|
||||
}()
|
||||
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var r io.Reader = f
|
||||
if err := func() error {
|
||||
if req.Header.Get("Range") == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
ranges, err := download.ParseRange(req.Header.Get("Range"), info.Size())
|
||||
switch {
|
||||
case len(ranges) == 0:
|
||||
fallthrough
|
||||
case errors.Is(err, download.ErrInvalidRange):
|
||||
// ignore invalid range and return the whole file
|
||||
return nil
|
||||
case err != nil:
|
||||
return err
|
||||
}
|
||||
|
||||
r = io.NewSectionReader(f, ranges[0].Start, ranges[0].Length)
|
||||
w.WriteHeader(http.StatusPartialContent)
|
||||
w.Header().Set("Content-Range", ranges[0].ContentRange(info.Size()))
|
||||
return nil
|
||||
}(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err := io.Copy(w, r); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// replace with url.JoinPath after switching to go1.19
|
||||
func joinurl(paths ...string) string {
|
||||
var s strings.Builder
|
||||
|
||||
Reference in New Issue
Block a user