sevenhill/QSfera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bump reva to get latest changes for LDAP client

Browse Source
This commit is contained in:
root
2023-07-11 16:05:32 +02:00
committed by Ralf Haferkamp
parent 99f27e569d
commit 6989b17a13
34 changed files with 1293 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vendor/github.com/cs3org/reva/v2/pkg/eosclient/eosgrpc/eos_grpc/Rpc.pb.go
Generated Vendored
+3 -2
View File
@@ -47,10 +47,11 @@
package eos_grpc
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
)
const (
vendor/github.com/cs3org/reva/v2/pkg/events/postprocessing.go
Generated Vendored
+13
View File
@@ -184,3 +184,16 @@ func (ResumePostprocessing) Unmarshal(v []byte) (interface{}, error) {
err := json.Unmarshal(v, &e)
return e, err
}
// RestartPostprocessing will be emitted by postprocessing service if it doesn't know about an upload
type RestartPostprocessing struct {
UploadID string
Timestamp *types.Timestamp
}
// Unmarshal to fulfill umarshaller interface
func (RestartPostprocessing) Unmarshal(v []byte) (interface{}, error) {
e := RestartPostprocessing{}
err := json.Unmarshal(v, &e)
return e, err
}
vendor/github.com/cs3org/reva/v2/pkg/events/stream/nats.go
Generated Vendored
+1
View File
@@ -52,6 +52,7 @@ func NatsFromConfig(cfg NatsConfig) (events.Stream, error) {
natsjs.TLSConfig(tlsConf),
natsjs.Address(cfg.Endpoint),
natsjs.ClusterID(cfg.Cluster),
natsjs.SynchronousPublish(true),
)
}
vendor/github.com/cs3org/reva/v2/pkg/micro/ocdav/option.go
Generated Vendored
+10
View File
@@ -28,6 +28,7 @@ import (
"github.com/cs3org/reva/v2/pkg/storage/favorite"
"github.com/rs/zerolog"
"go-micro.dev/v4/broker"
"go.opentelemetry.io/otel/trace"
)
// Option defines a single option function.
@@ -54,6 +55,8 @@ type Options struct {
TracingCollector string
TracingEndpoint string
TraceProvider trace.TracerProvider
MetricsEnabled bool
MetricsNamespace string
MetricsSubsystem string
@@ -234,6 +237,13 @@ func WithTracingExporter(exporter string) Option {
}
}
// WithTraceProvider option
func WithTraceProvider(provider trace.TracerProvider) Option {
return func(o *Options) {
o.TraceProvider = provider
}
}
// Version provides a function to set the Version config option.
func Version(val string) Option {
return func(o *Options) {
vendor/github.com/cs3org/reva/v2/pkg/micro/ocdav/service.go
Generated Vendored
+16 -14
View File
@@ -51,7 +51,6 @@ const (
// Service initializes the ocdav service and underlying http server.
func Service(opts ...Option) (micro.Service, error) {
sopts := newOptions(opts...)
// set defaults
@@ -86,19 +85,23 @@ func Service(opts ...Option) (micro.Service, error) {
// chi.RegisterMethod(ocdav.MethodMkcol)
// chi.RegisterMethod(ocdav.MethodReport)
r := chi.NewRouter()
topts := []rtrace.Option{
rtrace.WithExporter(sopts.TracingExporter),
rtrace.WithEndpoint(sopts.TracingEndpoint),
rtrace.WithCollector(sopts.TracingCollector),
rtrace.WithServiceName(sopts.Name),
tp := sopts.TraceProvider
if tp == nil {
topts := []rtrace.Option{
rtrace.WithExporter(sopts.TracingExporter),
rtrace.WithEndpoint(sopts.TracingEndpoint),
rtrace.WithCollector(sopts.TracingCollector),
rtrace.WithServiceName(sopts.Name),
}
if sopts.TracingEnabled {
topts = append(topts, rtrace.WithEnabled())
}
if sopts.TracingInsecure {
topts = append(topts, rtrace.WithInsecure())
}
tp = rtrace.NewTracerProvider(topts...)
}
if sopts.TracingEnabled {
topts = append(topts, rtrace.WithEnabled())
}
if sopts.TracingInsecure {
topts = append(topts, rtrace.WithInsecure())
}
tp := rtrace.NewTracerProvider(topts...)
if err := useMiddlewares(r, &sopts, revaService, tp); err != nil {
return nil, err
}
@@ -132,7 +135,6 @@ func Service(opts ...Option) (micro.Service, error) {
}
func setDefaults(sopts *Options) error {
// set defaults
if sopts.Name == "" {
sopts.Name = ServerName
vendor/github.com/cs3org/reva/v2/pkg/storage/cache/stat.go
Generated Vendored
+1 -1
View File
@@ -36,7 +36,7 @@ import (
var tracer trace.Tracer
func init() {
tracer = otel.Tracer("github.com/cs3org/reva/pkg/storage/utils/decomposedfs/lookup")
tracer = otel.Tracer("github.com/cs3org/reva/pkg/storage/cache")
}
// NewStatCache creates a new StatCache
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go
Generated Vendored
+70 -26
View File
@@ -32,9 +32,9 @@ import (
"path/filepath"
"strconv"
"strings"
"syscall"
"time"
user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
ctxpkg "github.com/cs3org/reva/v2/pkg/ctx"
@@ -48,9 +48,9 @@ import (
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/migrator"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/mtimesyncedcache"
"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/spaceidindex"
"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/filelocks"
@@ -66,7 +66,15 @@ import (
"golang.org/x/sync/errgroup"
)
var tracer trace.Tracer
var (
tracer trace.Tracer
_registeredEvents = []events.Unmarshaller{
events.PostprocessingFinished{},
events.PostprocessingStepFinished{},
events.RestartPostprocessing{},
}
)
func init() {
tracer = otel.Tracer("github.com/cs3org/reva/pkg/storage/utils/decomposedfs")
@@ -104,8 +112,10 @@ type Decomposedfs struct {
stream events.Stream
cache cache.StatCache
UserCache *ttlcache.Cache
spaceIDCache mtimesyncedcache.Cache[string, map[string]string]
UserCache *ttlcache.Cache
userSpaceIndex *spaceidindex.Index
groupSpaceIndex *spaceidindex.Index
spaceTypeIndex *spaceidindex.Index
}
// NewDefault returns an instance with default components
@@ -169,16 +179,34 @@ func New(o *options.Options, lu *lookup.Lookup, p Permissions, tp Tree, es event
if o.LockCycleDurationFactor != 0 {
filelocks.SetLockCycleDurationFactor(o.LockCycleDurationFactor)
}
userSpaceIndex := spaceidindex.New(filepath.Join(o.Root, "indexes"), "by-user-id")
err = userSpaceIndex.Init()
if err != nil {
return nil, err
}
groupSpaceIndex := spaceidindex.New(filepath.Join(o.Root, "indexes"), "by-group-id")
err = groupSpaceIndex.Init()
if err != nil {
return nil, err
}
spaceTypeIndex := spaceidindex.New(filepath.Join(o.Root, "indexes"), "by-type")
err = spaceTypeIndex.Init()
if err != nil {
return nil, err
}
fs := &Decomposedfs{
tp: tp,
lu: lu,
o: o,
p: p,
chunkHandler: chunking.NewChunkHandler(filepath.Join(o.Root, "uploads")),
stream: es,
cache: cache.GetStatCache(o.StatCache.Store, o.StatCache.Nodes, o.StatCache.Database, "stat", time.Duration(o.StatCache.TTL)*time.Second, o.StatCache.Size),
UserCache: ttlcache.NewCache(),
tp: tp,
lu: lu,
o: o,
p: p,
chunkHandler: chunking.NewChunkHandler(filepath.Join(o.Root, "uploads")),
stream: es,
cache: cache.GetStatCache(o.StatCache.Store, o.StatCache.Nodes, o.StatCache.Database, "stat", time.Duration(o.StatCache.TTL)*time.Second, o.StatCache.Size),
UserCache: ttlcache.NewCache(),
userSpaceIndex: userSpaceIndex,
groupSpaceIndex: groupSpaceIndex,
spaceTypeIndex: spaceTypeIndex,
}
if o.AsyncFileUploads {
@@ -187,7 +215,7 @@ func New(o *options.Options, lu *lookup.Lookup, p Permissions, tp Tree, es event
return nil, errors.New("need nats for async file processing")
}
ch, err := events.Consume(fs.stream, "dcfs", events.PostprocessingFinished{}, events.PostprocessingStepFinished{})
ch, err := events.Consume(fs.stream, "dcfs", _registeredEvents...)
if err != nil {
return nil, err
}
@@ -285,7 +313,34 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
); err != nil {
log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("Failed to publish UploadReady event")
}
case events.RestartPostprocessing:
up, err := upload.Get(ctx, ev.UploadID, fs.lu, fs.tp, fs.o.Root, fs.stream, fs.o.AsyncFileUploads, fs.o.Tokens)
if err != nil {
log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("Failed to get upload")
continue
}
n, err := node.ReadNode(ctx, fs.lu, up.Info.Storage["SpaceRoot"], up.Info.Storage["NodeId"], false, nil, true)
if err != nil {
log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("could not read node")
continue
}
s, err := up.URL(up.Ctx)
if err != nil {
log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("could not create url")
continue
}
// restart postprocessing
if err := events.Publish(fs.stream, events.BytesReceived{
UploadID: up.Info.ID,
URL: s,
SpaceOwner: n.SpaceOwnerOrManager(up.Ctx),
ExecutingUser: &user.User{Id: &user.UserId{OpaqueId: "postprocessing-restart"}}, // send nil instead?
ResourceID: &provider.ResourceId{SpaceId: n.SpaceID, OpaqueId: n.ID},
Filename: up.Info.Storage["NodeName"],
Filesize: uint64(up.Info.Size),
}); err != nil {
log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("Failed to publish BytesReceived event")
}
case events.PostprocessingStepFinished:
if ev.FinishedStep != events.PPStepAntivirus {
// atm we are only interested in antivirus results
@@ -515,17 +570,6 @@ func (fs *Decomposedfs) CreateHome(ctx context.Context) (err error) {
return nil
}
// The os not exists error is buried inside the xattr error,
// so we cannot just use os.IsNotExists().
func isAlreadyExists(err error) bool {
if xerr, ok := err.(*os.LinkError); ok {
if serr, ok2 := xerr.Err.(syscall.Errno); ok2 {
return serr == syscall.EEXIST
}
}
return false
}
// GetHome is called to look up the home path for a user
// It is NOT supposed to return the internal path but the external path
func (fs *Decomposedfs) GetHome(ctx context.Context) (string, error) {
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/grants.go
Generated Vendored
+2 -5
View File
@@ -20,7 +20,6 @@ package decomposedfs
import (
"context"
"os"
"path/filepath"
"strings"
@@ -220,14 +219,12 @@ func (fs *Decomposedfs) RemoveGrant(ctx context.Context, ref *provider.Reference
switch {
case g.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER:
// remove from user index
userIDPath := filepath.Join(fs.o.Root, "indexes", "by-user-id", g.Grantee.GetUserId().GetOpaqueId(), grantNode.SpaceID)
if err := os.Remove(userIDPath); err != nil {
if err := fs.userSpaceIndex.Remove(g.Grantee.GetUserId().GetOpaqueId(), grantNode.SpaceID); err != nil {
return err
}
case g.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_GROUP:
// remove from group index
userIDPath := filepath.Join(fs.o.Root, "indexes", "by-group-id", g.Grantee.GetGroupId().GetOpaqueId(), grantNode.SpaceID)
if err := os.Remove(userIDPath); err != nil {
if err := fs.groupSpaceIndex.Remove(g.Grantee.GetGroupId().GetOpaqueId(), grantNode.SpaceID); err != nil {
return err
}
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup/lookup.go
Generated Vendored
+5 -5
View File
@@ -284,7 +284,7 @@ func refFromCS3(b []byte) (*provider.Reference, error) {
func (lu *Lookup) CopyMetadata(ctx context.Context, src, target string, filter func(attributeName string) bool) (err error) {
// Acquire a read log on the source node
// write lock existing node before reading treesize or tree time
f, err := lockedfile.Open(lu.MetadataBackend().MetadataPath(src))
lock, err := lockedfile.OpenFile(lu.MetadataBackend().LockfilePath(src), os.O_RDONLY|os.O_CREATE, 0600)
if err != nil {
return err
}
@@ -293,7 +293,7 @@ func (lu *Lookup) CopyMetadata(ctx context.Context, src, target string, filter f
return errors.Wrap(err, "xattrs: Unable to lock source to read")
}
defer func() {
rerr := f.Close()
rerr := lock.Close()
// if err is non nil we do not overwrite that
if err == nil {
@@ -301,7 +301,7 @@ func (lu *Lookup) CopyMetadata(ctx context.Context, src, target string, filter f
}
}()
return lu.CopyMetadataWithSourceLock(ctx, src, target, filter, f)
return lu.CopyMetadataWithSourceLock(ctx, src, target, filter, lock)
}
// CopyMetadataWithSourceLock copies all extended attributes from source to target.
@@ -312,11 +312,11 @@ func (lu *Lookup) CopyMetadataWithSourceLock(ctx context.Context, sourcePath, ta
switch {
case lockedSource == nil:
return errors.New("no lock provided")
case lockedSource.File.Name() != lu.MetadataBackend().MetadataPath(sourcePath):
case lockedSource.File.Name() != lu.MetadataBackend().LockfilePath(sourcePath):
return errors.New("lockpath does not match filepath")
}
attrs, err := lu.metadataBackend.AllWithLockedSource(ctx, sourcePath, lockedSource)
attrs, err := lu.metadataBackend.All(ctx, sourcePath)
if err != nil {
return err
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/messagepack_backend.go
Generated Vendored
+45 -43
View File
@@ -20,7 +20,9 @@ package metadata
import (
"context"
"errors"
"io"
"io/fs"
"os"
"path/filepath"
"strconv"
@@ -28,6 +30,7 @@ import (
"time"
"github.com/cs3org/reva/v2/pkg/storage/cache"
"github.com/google/renameio/v2"
"github.com/pkg/xattr"
"github.com/rogpeppe/go-internal/lockedfile"
"github.com/shamaton/msgpack/v2"
@@ -142,74 +145,60 @@ func (b MessagePackBackend) saveAttributes(ctx context.Context, path string, set
span.End()
}()
lockPath := b.LockfilePath(path)
metaPath := b.MetadataPath(path)
if acquireLock {
_, subspan := tracer.Start(ctx, "lockedfile.OpenFile")
f, err = lockedfile.OpenFile(metaPath, os.O_RDWR|os.O_CREATE, 0600)
subspan.End()
} else {
_, subspan := tracer.Start(ctx, "os.OpenFile")
f, err = os.OpenFile(metaPath, os.O_RDWR|os.O_CREATE, 0600)
f, err = lockedfile.OpenFile(lockPath, os.O_RDWR|os.O_CREATE, 0600)
subspan.End()
defer f.Close()
}
if err != nil {
return err
}
defer f.Close()
// Invalidate cache early
_, subspan := tracer.Start(ctx, "metaCache.RemoveMetadata")
_ = b.metaCache.RemoveMetadata(b.cacheKey(path))
subspan.End()
// Read current state
_, subspan = tracer.Start(ctx, "io.ReadAll")
_, subspan := tracer.Start(ctx, "os.ReadFile")
var msgBytes []byte
msgBytes, err = io.ReadAll(f)
msgBytes, err = os.ReadFile(metaPath)
subspan.End()
if err != nil {
return err
}
attribs := map[string][]byte{}
if len(msgBytes) > 0 {
switch {
case err != nil:
if !errors.Is(err, fs.ErrNotExist) {
return err
}
case len(msgBytes) == 0:
// ugh. an empty file? bail out
return errors.New("encountered empty metadata file")
default:
// only unmarshal if we read data
err = msgpack.Unmarshal(msgBytes, &attribs)
if err != nil {
return err
}
}
// set new metadata
// prepare metadata
for key, val := range setAttribs {
attribs[key] = val
}
for _, key := range deleteAttribs {
delete(attribs, key)
}
// Truncate file
_, err = f.Seek(0, io.SeekStart)
if err != nil {
return err
}
_, subspan = tracer.Start(ctx, "f.Truncate")
err = f.Truncate(0)
subspan.End()
if err != nil {
return err
}
// Write new metadata to file
var d []byte
d, err = msgpack.Marshal(attribs)
if err != nil {
return err
}
_, subspan = tracer.Start(ctx, "f.Write")
_, err = f.Write(d)
subspan.End()
// overwrite file atomically
_, subspan = tracer.Start(ctx, "renameio.Writefile")
err = renameio.WriteFile(metaPath, d, 0600)
if err != nil {
return err
}
subspan.End()
_, subspan = tracer.Start(ctx, "metaCache.PushToCache")
err = b.metaCache.PushToCache(b.cacheKey(path), attribs)
@@ -227,9 +216,13 @@ func (b MessagePackBackend) loadAttributes(ctx context.Context, path string, sou
}
metaPath := b.MetadataPath(path)
var msgBytes []byte
if source == nil {
_, subspan := tracer.Start(ctx, "lockedfile.Open")
source, err = lockedfile.Open(metaPath)
// // No cached entry found. Read from storage and store in cache
_, subspan := tracer.Start(ctx, "os.OpenFile")
// source, err = lockedfile.Open(metaPath)
source, err = os.Open(metaPath)
subspan.End()
// // No cached entry found. Read from storage and store in cache
if err != nil {
@@ -246,12 +239,16 @@ func (b MessagePackBackend) loadAttributes(ctx context.Context, path string, sou
return attribs, nil // no attributes set yet
}
}
defer source.(*lockedfile.File).Close()
_, subspan = tracer.Start(ctx, "io.ReadAll")
msgBytes, err = io.ReadAll(source)
source.(*os.File).Close()
subspan.End()
} else {
_, subspan := tracer.Start(ctx, "io.ReadAll")
msgBytes, err = io.ReadAll(source)
subspan.End()
}
_, subspan := tracer.Start(ctx, "io.ReadAll")
msgBytes, err := io.ReadAll(source)
subspan.End()
if err != nil {
return nil, err
}
@@ -262,7 +259,7 @@ func (b MessagePackBackend) loadAttributes(ctx context.Context, path string, sou
}
}
_, subspan = tracer.Start(ctx, "metaCache.PushToCache")
_, subspan := tracer.Start(ctx, "metaCache.PushToCache")
err = b.metaCache.PushToCache(b.cacheKey(path), attribs)
subspan.End()
if err != nil {
@@ -273,7 +270,9 @@ func (b MessagePackBackend) loadAttributes(ctx context.Context, path string, sou
}
// IsMetaFile returns whether the given path represents a meta file
func (MessagePackBackend) IsMetaFile(path string) bool { return strings.HasSuffix(path, ".mpk") }
func (MessagePackBackend) IsMetaFile(path string) bool {
return strings.HasSuffix(path, ".mpk") || strings.HasSuffix(path, ".mlock")
}
// Purge purges the data of a given path
func (b MessagePackBackend) Purge(path string) error {
@@ -304,6 +303,9 @@ func (b MessagePackBackend) Rename(oldPath, newPath string) error {
// MetadataPath returns the path of the file holding the metadata for the given path
func (MessagePackBackend) MetadataPath(path string) string { return path + ".mpk" }
// LockfilePath returns the path of the lock file
func (MessagePackBackend) LockfilePath(path string) string { return path + ".mlock" }
func (b MessagePackBackend) cacheKey(path string) string {
// rootPath is guaranteed to have no trailing slash
// the cache key shouldn't begin with a slash as some stores drop it which can cause
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/metadata.go
Generated Vendored
+4
View File
@@ -52,6 +52,7 @@ type Backend interface {
Rename(oldPath, newPath string) error
IsMetaFile(path string) bool
MetadataPath(path string) string
LockfilePath(path string) string
AllWithLockedSource(ctx context.Context, path string, source io.Reader) (map[string][]byte, error)
}
@@ -110,6 +111,9 @@ func (NullBackend) Rename(oldPath, newPath string) error { return errUnconfigure
// MetadataPath returns the path of the file holding the metadata for the given path
func (NullBackend) MetadataPath(path string) string { return "" }
// LockfilePath returns the path of the lock file
func (NullBackend) LockfilePath(path string) string { return "" }
// AllWithLockedSource reads all extended attributes from the given reader
// The path argument is used for storing the data in the cache
func (NullBackend) AllWithLockedSource(ctx context.Context, path string, source io.Reader) (map[string][]byte, error) {
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/xattrs_backend.go
Generated Vendored
+5 -1
View File
@@ -24,6 +24,7 @@ import (
"os"
"path/filepath"
"strconv"
"strings"
"github.com/cs3org/reva/v2/pkg/storage/utils/filelocks"
"github.com/pkg/errors"
@@ -156,7 +157,7 @@ func (XattrsBackend) Remove(ctx context.Context, filePath string, key string) (e
}
// IsMetaFile returns whether the given path represents a meta file
func (XattrsBackend) IsMetaFile(path string) bool { return false }
func (XattrsBackend) IsMetaFile(path string) bool { return strings.HasSuffix(path, ".meta.lock") }
// Purge purges the data of a given path
func (XattrsBackend) Purge(path string) error { return nil }
@@ -167,6 +168,9 @@ func (XattrsBackend) Rename(oldPath, newPath string) error { return nil }
// MetadataPath returns the path of the file holding the metadata for the given path
func (XattrsBackend) MetadataPath(path string) string { return path }
// LockfilePath returns the path of the lock file
func (XattrsBackend) LockfilePath(path string) string { return path + ".mlock" }
func cleanupLockfile(f *lockedfile.File) {
_ = f.Close()
_ = os.Remove(f.Name())
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/migrator/0004_switch_to_messagepack_space_index.go
Generated Vendored
+120
View File
@@ -0,0 +1,120 @@
// Copyright 2018-2023 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package migrator
import (
"os"
"path/filepath"
"github.com/shamaton/msgpack/v2"
)
// Migration0004 migrates the directory tree based space indexes to messagepack
func (m *Migrator) Migration0004() (Result, error) {
root := m.lu.InternalRoot()
// migrate user indexes
users, err := os.ReadDir(filepath.Join(root, "indexes", "by-user-id"))
if err != nil {
m.log.Warn().Err(err).Msg("error listing user indexes")
}
for _, user := range users {
if !user.IsDir() {
continue
}
id := user.Name()
indexPath := filepath.Join(root, "indexes", "by-user-id", id+".mpk")
dirIndexPath := filepath.Join(root, "indexes", "by-user-id", id)
cacheKey := "by-user-id:" + id
m.log.Info().Str("root", m.lu.InternalRoot()).Msg("Migrating " + indexPath + " to messagepack index format...")
err := migrateSpaceIndex(indexPath, dirIndexPath, cacheKey)
if err != nil {
m.log.Error().Err(err).Str("path", dirIndexPath).Msg("error migrating index")
}
}
// migrate group indexes
groups, err := os.ReadDir(filepath.Join(root, "indexes", "by-group-id"))
if err != nil {
m.log.Warn().Err(err).Msg("error listing group indexes")
}
for _, group := range groups {
if !group.IsDir() {
continue
}
id := group.Name()
indexPath := filepath.Join(root, "indexes", "by-group-id", id+".mpk")
dirIndexPath := filepath.Join(root, "indexes", "by-group-id", id)
cacheKey := "by-group-id:" + id
m.log.Info().Str("root", m.lu.InternalRoot()).Msg("Migrating " + indexPath + " to messagepack index format...")
err := migrateSpaceIndex(indexPath, dirIndexPath, cacheKey)
if err != nil {
m.log.Error().Err(err).Str("path", dirIndexPath).Msg("error migrating index")
}
}
// migrate project indexes
for _, spaceType := range []string{"personal", "project", "share"} {
indexPath := filepath.Join(root, "indexes", "by-type", spaceType+".mpk")
dirIndexPath := filepath.Join(root, "indexes", "by-type", spaceType)
cacheKey := "by-type:" + spaceType
_, err := os.Stat(dirIndexPath)
if err != nil {
continue
}
m.log.Info().Str("root", m.lu.InternalRoot()).Msg("Migrating " + indexPath + " to messagepack index format...")
err = migrateSpaceIndex(indexPath, dirIndexPath, cacheKey)
if err != nil {
m.log.Error().Err(err).Str("path", dirIndexPath).Msg("error migrating index")
}
}
m.log.Info().Msg("done.")
return resultSucceeded, nil
}
func migrateSpaceIndex(indexPath, dirIndexPath, cacheKey string) error {
links := map[string][]byte{}
m, err := filepath.Glob(dirIndexPath + "/*")
if err != nil {
return err
}
for _, match := range m {
link, err := os.Readlink(match)
if err != nil {
continue
}
links[filepath.Base(match)] = []byte(link)
}
// rewrite index as file
d, err := msgpack.Marshal(links)
if err != nil {
return err
}
err = os.WriteFile(indexPath, d, 0600)
if err != nil {
return err
}
return os.RemoveAll(dirIndexPath)
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/migrator/migrator.go
Generated Vendored
+1 -1
View File
@@ -29,7 +29,7 @@ import (
"github.com/rs/zerolog"
)
var allMigrations = []string{"0001", "0002", "0003"}
var allMigrations = []string{"0001", "0002", "0003", "0004"}
const (
resultFailed = "failed"
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/revisions.go
Generated Vendored
+3 -3
View File
@@ -70,7 +70,7 @@ func (fs *Decomposedfs) ListRevisions(ctx context.Context, ref *provider.Referen
np := n.InternalPath()
if items, err := filepath.Glob(np + node.RevisionIDDelimiter + "*"); err == nil {
for i := range items {
if fs.lu.MetadataBackend().IsMetaFile(items[i]) {
if fs.lu.MetadataBackend().IsMetaFile(items[i]) || strings.HasSuffix(items[i], ".mlock") {
continue
}
@@ -237,7 +237,7 @@ func (fs *Decomposedfs) RestoreRevision(ctx context.Context, ref *provider.Refer
attributeName == prefixes.BlobsizeAttr
})
if err != nil {
return errtypes.InternalError("failed to copy blob xattrs to version node")
return errtypes.InternalError("failed to copy blob xattrs to version node: " + err.Error())
}
// remember mtime from node as new revision mtime
@@ -256,7 +256,7 @@ func (fs *Decomposedfs) RestoreRevision(ctx context.Context, ref *provider.Refer
attributeName == prefixes.BlobsizeAttr
})
if err != nil {
return errtypes.InternalError("failed to copy blob xattrs to old revision to node")
return errtypes.InternalError("failed to copy blob xattrs to old revision to node: " + err.Error())
}
revisionSize, err := fs.lu.MetadataBackend().GetInt64(ctx, restoredRevisionPath, prefixes.BlobsizeAttr)
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/spaceidindex/spaceidindex.go
Generated Vendored
+153
View File
@@ -0,0 +1,153 @@
package spaceidindex
import (
"io"
"os"
"path/filepath"
"time"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/mtimesyncedcache"
"github.com/pkg/errors"
"github.com/rogpeppe/go-internal/lockedfile"
"github.com/shamaton/msgpack/v2"
)
// Index holds space id indexes
type Index struct {
root string
name string
cache mtimesyncedcache.Cache[string, map[string]string]
}
type readWriteCloseSeekTruncater interface {
io.ReadWriteCloser
io.Seeker
Truncate(int64) error
}
// New returns a new index instance
func New(root, name string) *Index {
return &Index{
root: root,
name: name,
}
}
// Init initializes the index and makes sure it can be used
func (i *Index) Init() error {
// Make sure to work on an existing tree
return os.MkdirAll(filepath.Join(i.root, i.name), 0700)
}
// Load returns the content of an index
func (i *Index) Load(index string) (map[string]string, error) {
indexPath := filepath.Join(i.root, i.name, index+".mpk")
fi, err := os.Stat(indexPath)
if err != nil {
return nil, err
}
return i.readSpaceIndex(indexPath, i.name+":"+index, fi.ModTime())
}
// Add adds an entry to an index
func (i *Index) Add(index, key string, value string) error {
return i.updateIndex(index, map[string]string{key: value}, []string{})
}
// Remove removes an entry from the index
func (i *Index) Remove(index, key string) error {
return i.updateIndex(index, map[string]string{}, []string{key})
}
func (i *Index) updateIndex(index string, addLinks map[string]string, removeLinks []string) error {
indexPath := filepath.Join(i.root, i.name, index+".mpk")
var err error
// acquire writelock
var f readWriteCloseSeekTruncater
f, err = lockedfile.OpenFile(indexPath, os.O_RDWR|os.O_CREATE, 0600)
if err != nil {
return errors.Wrap(err, "unable to lock index to write")
}
defer func() {
rerr := f.Close()
// if err is non nil we do not overwrite that
if err == nil {
err = rerr
}
}()
// Read current state
msgBytes, err := io.ReadAll(f)
if err != nil {
return err
}
links := map[string]string{}
if len(msgBytes) > 0 {
err = msgpack.Unmarshal(msgBytes, &links)
if err != nil {
return err
}
}
// set new metadata
for key, val := range addLinks {
links[key] = val
}
for _, key := range removeLinks {
delete(links, key)
}
// Truncate file
_, err = f.Seek(0, io.SeekStart)
if err != nil {
return err
}
err = f.Truncate(0)
if err != nil {
return err
}
// Write new metadata to file
d, err := msgpack.Marshal(links)
if err != nil {
return errors.Wrap(err, "unable to marshal index")
}
_, err = f.Write(d)
if err != nil {
return errors.Wrap(err, "unable to write index")
}
return nil
}
func (i *Index) readSpaceIndex(indexPath, cacheKey string, mtime time.Time) (map[string]string, error) {
return i.cache.LoadOrStore(cacheKey, mtime, func() (map[string]string, error) {
// Acquire a read log on the index file
f, err := lockedfile.Open(indexPath)
if err != nil {
return nil, errors.Wrap(err, "unable to lock index to read")
}
defer func() {
rerr := f.Close()
// if err is non nil we do not overwrite that
if err == nil {
err = rerr
}
}()
// Read current state
msgBytes, err := io.ReadAll(f)
if err != nil {
return nil, errors.Wrap(err, "unable to read index")
}
links := map[string]string{}
if len(msgBytes) > 0 {
err = msgpack.Unmarshal(msgBytes, &links)
if err != nil {
return nil, errors.Wrap(err, "unable to parse index")
}
}
return links, nil
})
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/spaces.go
Generated Vendored
+29 -138
View File
@@ -298,31 +298,13 @@ func (fs *Decomposedfs) ListStorageSpaces(ctx context.Context, filter []*provide
}
matches := map[string]struct{}{}
var allMatches map[string]string
var err error
if requestedUserID != nil {
allMatches := map[string]string{}
indexPath := filepath.Join(fs.o.Root, "indexes", "by-user-id", requestedUserID.GetOpaqueId())
fi, err := os.Stat(indexPath)
if err == nil {
allMatches, err = fs.spaceIDCache.LoadOrStore("by-user-id:"+requestedUserID.GetOpaqueId(), fi.ModTime(), func() (map[string]string, error) {
path := filepath.Join(fs.o.Root, "indexes", "by-user-id", requestedUserID.GetOpaqueId(), "*")
m, err := filepath.Glob(path)
if err != nil {
return nil, err
}
matches := map[string]string{}
for _, match := range m {
link, err := os.Readlink(match)
if err != nil {
continue
}
matches[match] = link
}
return matches, nil
})
}
allMatches, err = fs.userSpaceIndex.Load(requestedUserID.GetOpaqueId())
if err != nil {
return nil, err
return nil, errors.Wrap(err, "error reading user index")
}
if nodeID == spaceIDAny {
@@ -344,29 +326,12 @@ func (fs *Decomposedfs) ListStorageSpaces(ctx context.Context, filter []*provide
}
for _, group := range user.Groups {
indexPath := filepath.Join(fs.o.Root, "indexes", "by-group-id", group)
fi, err := os.Stat(indexPath)
allMatches, err = fs.groupSpaceIndex.Load(group)
if err != nil {
continue
}
allMatches, err := fs.spaceIDCache.LoadOrStore("by-group-id:"+group, fi.ModTime(), func() (map[string]string, error) {
path := filepath.Join(fs.o.Root, "indexes", "by-group-id", group, "*")
m, err := filepath.Glob(path)
if err != nil {
return nil, err
if os.IsNotExist(err) {
continue // no spaces for this group
}
matches := map[string]string{}
for _, match := range m {
link, err := os.Readlink(match)
if err != nil {
continue
}
matches[match] = link
}
return matches, nil
})
if err != nil {
return nil, err
return nil, errors.Wrap(err, "error reading group index")
}
if nodeID == spaceIDAny {
@@ -381,33 +346,22 @@ func (fs *Decomposedfs) ListStorageSpaces(ctx context.Context, filter []*provide
}
if requestedUserID == nil {
if _, ok := spaceTypes[spaceTypeAny]; ok {
// TODO do not hardcode dirs
spaceTypes = map[string]struct{}{
"personal": {},
"project": {},
"share": {},
}
}
for spaceType := range spaceTypes {
indexPath := filepath.Join(fs.o.Root, "indexes", "by-type")
if spaceType != spaceTypeAny {
indexPath = filepath.Join(indexPath, spaceType)
}
fi, err := os.Stat(indexPath)
allMatches, err = fs.spaceTypeIndex.Load(spaceType)
if err != nil {
continue
}
allMatches, err := fs.spaceIDCache.LoadOrStore("by-type:"+spaceType, fi.ModTime(), func() (map[string]string, error) {
path := filepath.Join(fs.o.Root, "indexes", "by-type", spaceType, "*")
m, err := filepath.Glob(path)
if err != nil {
return nil, err
if os.IsNotExist(err) {
continue // no spaces for this space type
}
matches := map[string]string{}
for _, match := range m {
link, err := os.Readlink(match)
if err != nil {
continue
}
matches[match] = link
}
return matches, nil
})
if err != nil {
return nil, err
return nil, errors.Wrap(err, "error reading type index")
}
if nodeID == spaceIDAny {
@@ -764,13 +718,12 @@ func (fs *Decomposedfs) DeleteStorageSpace(ctx context.Context, req *provider.De
return errtypes.NewErrtypeFromStatus(status.NewInvalid(ctx, "can't purge enabled space"))
}
// TODO invalidate ALL indexes in msgpack, not only by type
spaceType, err := n.XattrString(ctx, prefixes.SpaceTypeAttr)
if err != nil {
return err
}
// remove type index
spaceTypePath := filepath.Join(fs.o.Root, "indexes", "by-type", spaceType, spaceID)
if err := os.Remove(spaceTypePath); err != nil {
if err := fs.spaceTypeIndex.Remove(spaceType, spaceID); err != nil {
return err
}
@@ -817,80 +770,18 @@ func (fs *Decomposedfs) updateIndexes(ctx context.Context, grantee *provider.Gra
}
func (fs *Decomposedfs) linkSpaceByUser(ctx context.Context, userID, spaceID string) error {
if userID == "" {
return nil
}
// create user index dir
// TODO: pathify userID
if err := os.MkdirAll(filepath.Join(fs.o.Root, "indexes", "by-user-id", userID), 0700); err != nil {
return err
}
err := os.Symlink("../../../spaces/"+lookup.Pathify(spaceID, 1, 2)+"/nodes/"+lookup.Pathify(spaceID, 4, 2), filepath.Join(fs.o.Root, "indexes/by-user-id", userID, spaceID))
if err != nil {
if isAlreadyExists(err) {
appctx.GetLogger(ctx).Debug().Err(err).Str("space", spaceID).Str("user-id", userID).Msg("symlink already exists")
// FIXME: is it ok to wipe this err if the symlink already exists?
err = nil //nolint
} else {
// TODO how should we handle error cases here?
appctx.GetLogger(ctx).Error().Err(err).Str("space", spaceID).Str("user-id", userID).Msg("could not create symlink")
}
}
return nil
target := "../../../spaces/" + lookup.Pathify(spaceID, 1, 2) + "/nodes/" + lookup.Pathify(spaceID, 4, 2)
return fs.userSpaceIndex.Add(userID, spaceID, target)
}
func (fs *Decomposedfs) linkSpaceByGroup(ctx context.Context, groupID, spaceID string) error {
if groupID == "" {
return nil
}
// create group index dir
// TODO: pathify groupid
if err := os.MkdirAll(filepath.Join(fs.o.Root, "indexes", "by-group-id", groupID), 0700); err != nil {
return err
}
err := os.Symlink("../../../spaces/"+lookup.Pathify(spaceID, 1, 2)+"/nodes/"+lookup.Pathify(spaceID, 4, 2), filepath.Join(fs.o.Root, "indexes/by-group-id", groupID, spaceID))
if err != nil {
if isAlreadyExists(err) {
appctx.GetLogger(ctx).Debug().Err(err).Str("space", spaceID).Str("group-id", groupID).Msg("symlink already exists")
// FIXME: is it ok to wipe this err if the symlink already exists?
err = nil //nolint
} else {
// TODO how should we handle error cases here?
appctx.GetLogger(ctx).Error().Err(err).Str("space", spaceID).Str("group-id", groupID).Msg("could not create symlink")
}
}
return nil
target := "../../../spaces/" + lookup.Pathify(spaceID, 1, 2) + "/nodes/" + lookup.Pathify(spaceID, 4, 2)
return fs.groupSpaceIndex.Add(groupID, spaceID, target)
}
// TODO: implement linkSpaceByGroup
func (fs *Decomposedfs) linkStorageSpaceType(ctx context.Context, spaceType string, spaceID string) error {
if spaceType == "" {
return nil
}
// create space type dir
if err := os.MkdirAll(filepath.Join(fs.o.Root, "indexes", "by-type", spaceType), 0700); err != nil {
return err
}
// link space in spacetypes
err := os.Symlink("../../../spaces/"+lookup.Pathify(spaceID, 1, 2)+"/nodes/"+lookup.Pathify(spaceID, 4, 2), filepath.Join(fs.o.Root, "indexes", "by-type", spaceType, spaceID))
if err != nil {
if isAlreadyExists(err) {
appctx.GetLogger(ctx).Debug().Err(err).Str("space", spaceID).Str("spacetype", spaceType).Msg("symlink already exists")
// FIXME: is it ok to wipe this err if the symlink already exists?
} else {
// TODO how should we handle error cases here?
appctx.GetLogger(ctx).Error().Err(err).Str("space", spaceID).Str("spacetype", spaceType).Msg("could not create symlink")
return err
}
}
// touch index root to invalidate caches
now := time.Now()
return os.Chtimes(filepath.Join(fs.o.Root, "indexes", "by-type"), now, now)
target := "../../../spaces/" + lookup.Pathify(spaceID, 1, 2) + "/nodes/" + lookup.Pathify(spaceID, 4, 2)
return fs.spaceTypeIndex.Add(spaceType, spaceID, target)
}
func (fs *Decomposedfs) storageSpaceFromNode(ctx context.Context, n *node.Node, checkPermissions bool) (*provider.StorageSpace, error) {
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree/tree.go
Generated Vendored
+3 -13
View File
@@ -40,7 +40,6 @@ import (
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes"
"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/filelocks"
"github.com/cs3org/reva/v2/pkg/utils"
"github.com/google/uuid"
"github.com/pkg/errors"
@@ -750,17 +749,8 @@ func (t *Tree) Propagate(ctx context.Context, n *node.Node, sizeDiff int64) (err
// lock parent before reading treesize or tree time
_, subspan := tracer.Start(ctx, "lockedfile.OpenFile")
var parentFilename string
switch t.lookup.MetadataBackend().(type) {
case metadata.MessagePackBackend:
parentFilename = t.lookup.MetadataBackend().MetadataPath(n.ParentPath())
f, err = lockedfile.OpenFile(parentFilename, os.O_RDWR|os.O_CREATE, 0600)
case metadata.XattrsBackend:
// we have to use dedicated lockfiles to lock directories
// this only works because the xattr backend also locks folders with separate lock files
parentFilename = n.ParentPath() + filelocks.LockFileSuffix
f, err = lockedfile.OpenFile(parentFilename, os.O_RDWR|os.O_CREATE, 0600)
}
parentFilename := t.lookup.MetadataBackend().LockfilePath(n.ParentPath())
f, err = lockedfile.OpenFile(parentFilename, os.O_RDWR|os.O_CREATE, 0600)
subspan.End()
if err != nil {
sublog.Error().Err(err).
@@ -777,7 +767,7 @@ func (t *Tree) Propagate(ctx context.Context, n *node.Node, sizeDiff int64) (err
}
}()
if n, err = n.ParentWithReader(ctx, f); err != nil {
if n, err = n.Parent(ctx); err != nil {
sublog.Error().Err(err).
Msg("Propagation failed. Could not read parent node.")
return err
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/processing.go
Generated Vendored
+7 -14
View File
@@ -38,7 +38,6 @@ import (
"github.com/cs3org/reva/v2/pkg/logger"
"github.com/cs3org/reva/v2/pkg/storage/utils/chunking"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/options"
@@ -329,23 +328,17 @@ func initNewNode(upload *Upload, n *node.Node, fsize uint64) (*lockedfile.File,
}
// create and write lock new node metadata
f, err := lockedfile.OpenFile(upload.lu.MetadataBackend().MetadataPath(n.InternalPath()), os.O_RDWR|os.O_CREATE, 0600)
f, err := lockedfile.OpenFile(upload.lu.MetadataBackend().LockfilePath(n.InternalPath()), os.O_RDWR|os.O_CREATE, 0600)
if err != nil {
return nil, err
}
switch upload.lu.MetadataBackend().(type) {
case metadata.MessagePackBackend:
// for the ini and metadata backend we also need to touch the actual node file here.
// it stores the mtime of the resource, which must not change when we update the ini file
h, err := os.OpenFile(n.InternalPath(), os.O_CREATE, 0600)
if err != nil {
return f, err
}
h.Close()
case metadata.XattrsBackend:
// nothing to do
// we also need to touch the actual node file here it stores the mtime of the resource
h, err := os.OpenFile(n.InternalPath(), os.O_CREATE, 0600)
if err != nil {
return f, err
}
h.Close()
if _, err := node.CheckQuota(upload.Ctx, n.SpaceRoot, false, 0, fsize); err != nil {
return f, err
@@ -403,7 +396,7 @@ func updateExistingNode(upload *Upload, n *node.Node, spaceID string, fsize uint
targetPath := n.InternalPath()
// write lock existing node before reading treesize or tree time
f, err := lockedfile.OpenFile(upload.lu.MetadataBackend().MetadataPath(targetPath), os.O_RDWR, 0600)
f, err := lockedfile.OpenFile(upload.lu.MetadataBackend().LockfilePath(targetPath), os.O_RDWR|os.O_CREATE, 0600)
if err != nil {
return nil, err
}
vendor/github.com/cs3org/reva/v2/pkg/utils/ldap/reconnect.go
Generated Vendored
+22 -1
View File
@@ -240,7 +240,23 @@ func (c *ConnWithReconnect) StartTLS(*tls.Config) error {
}
// Close implements the ldap.Client interface
func (c *ConnWithReconnect) Close() {}
func (c *ConnWithReconnect) Close() (err error) {
conn, err := c.getConnection()
if err != nil {
return err
}
return conn.Close()
}
func (c *ConnWithReconnect) GetLastError() error {
conn, err := c.getConnection()
if err != nil {
return err
}
return conn.GetLastError()
}
// IsClosing implements the ldap.Client interface
func (c *ConnWithReconnect) IsClosing() bool {
@@ -304,3 +320,8 @@ func (c *ConnWithReconnect) TLSConnectionState() (tls.ConnectionState, bool) {
func (c *ConnWithReconnect) Unbind() error {
return ldap.NewError(ldap.LDAPResultNotSupported, fmt.Errorf("not implemented"))
}
// DirSync implements the ldap.Client interface
func (c *ConnWithReconnect) DirSync(searchRequest *ldap.SearchRequest, flags, maxAttrCount int64, cookie []byte) (*ldap.SearchResult, error) {
return nil, ldap.NewError(ldap.LDAPResultNotSupported, fmt.Errorf("not implemented"))
}