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

bump reva

Browse Source 
Signed-off-by: jkoberg <jkoberg@owncloud.com>
This commit is contained in:
jkoberg
2023-06-21 11:58:24 +02:00
parent 9cace5738f
commit 7bfe6bdf7d
11 changed files with 336 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/jsoncs3.go
Generated Vendored
+118 -86
View File
@@ -29,6 +29,12 @@ import (
"sync"
"time"
gatewayv1beta1 "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
"github.com/go-micro/plugins/v4/events/natsjs"
"github.com/google/uuid"
"github.com/mitchellh/mapstructure"
"github.com/pkg/errors"
@@ -37,11 +43,6 @@ import (
"golang.org/x/sync/errgroup"
"google.golang.org/genproto/protobuf/field_mask"
gatewayv1beta1 "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
"github.com/cs3org/reva/v2/pkg/appctx"
ctxpkg "github.com/cs3org/reva/v2/pkg/ctx"
"github.com/cs3org/reva/v2/pkg/errtypes"
@@ -57,7 +58,6 @@ import (
"github.com/cs3org/reva/v2/pkg/storage/utils/metadata" // nolint:staticcheck // we need the legacy package to convert V1 to V2 messages
"github.com/cs3org/reva/v2/pkg/storagespace"
"github.com/cs3org/reva/v2/pkg/utils"
"github.com/go-micro/plugins/v4/events/natsjs"
)
/*
@@ -333,83 +333,112 @@ func (m *Manager) Share(ctx context.Context, md *provider.ResourceInfo, g *colla
Mtime: ts,
}
err = m.Cache.Add(ctx, md.Id.StorageId, md.Id.SpaceId, shareID, s)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.Cache.Sync(ctx, md.Id.StorageId, md.Id.SpaceId); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return nil, err
}
err = m.Cache.Add(ctx, md.Id.StorageId, md.Id.SpaceId, shareID, s)
// TODO try more often?
}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return nil, err
}
eg, ctx := errgroup.WithContext(ctx)
err = m.CreatedCache.Add(ctx, s.GetCreator().GetOpaqueId(), shareID)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.CreatedCache.Sync(ctx, s.GetCreator().GetOpaqueId()); err != nil {
eg.Go(func() error {
err := m.Cache.Add(ctx, md.Id.StorageId, md.Id.SpaceId, shareID, s)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.Cache.Sync(ctx, md.Id.StorageId, md.Id.SpaceId); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
err = m.Cache.Add(ctx, md.Id.StorageId, md.Id.SpaceId, shareID, s)
// TODO try more often?
}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return nil, err
}
err = m.CreatedCache.Add(ctx, s.GetCreator().GetOpaqueId(), shareID)
// TODO try more often?
}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return nil, err
}
return err
})
eg.Go(func() error {
err := m.CreatedCache.Add(ctx, s.GetCreator().GetOpaqueId(), shareID)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.CreatedCache.Sync(ctx, s.GetCreator().GetOpaqueId()); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
err = m.CreatedCache.Add(ctx, s.GetCreator().GetOpaqueId(), shareID)
// TODO try more often?
}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
}
return err
})
spaceID := md.Id.StorageId + shareid.IDDelimiter + md.Id.SpaceId
// set flag for grantee to have access to share
switch g.Grantee.Type {
case provider.GranteeType_GRANTEE_TYPE_USER:
userid := g.Grantee.GetUserId().GetOpaqueId()
eg.Go(func() error {
userid := g.Grantee.GetUserId().GetOpaqueId()
rs := &collaboration.ReceivedShare{
Share: s,
State: collaboration.ShareState_SHARE_STATE_PENDING,
}
err = m.UserReceivedStates.Add(ctx, userid, spaceID, rs)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.UserReceivedStates.Sync(ctx, s.GetCreator().GetOpaqueId()); err != nil {
rs := &collaboration.ReceivedShare{
Share: s,
State: collaboration.ShareState_SHARE_STATE_PENDING,
}
err := m.UserReceivedStates.Add(ctx, userid, spaceID, rs)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.UserReceivedStates.Sync(ctx, s.GetCreator().GetOpaqueId()); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
err = m.UserReceivedStates.Add(ctx, userid, spaceID, rs)
// TODO try more often?
}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return nil, err
}
err = m.UserReceivedStates.Add(ctx, userid, spaceID, rs)
// TODO try more often?
}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return nil, err
}
return err
})
case provider.GranteeType_GRANTEE_TYPE_GROUP:
groupid := g.Grantee.GetGroupId().GetOpaqueId()
err := m.GroupReceivedCache.Add(ctx, groupid, shareID)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.GroupReceivedCache.Sync(ctx, groupid); err != nil {
eg.Go(func() error {
groupid := g.Grantee.GetGroupId().GetOpaqueId()
err := m.GroupReceivedCache.Add(ctx, groupid, shareID)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.GroupReceivedCache.Sync(ctx, groupid); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
err = m.GroupReceivedCache.Add(ctx, groupid, shareID)
// TODO try more often?
}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return nil, err
}
err = m.GroupReceivedCache.Add(ctx, groupid, shareID)
// TODO try more often?
}
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return nil, err
}
return err
})
}
if err = eg.Wait(); err != nil {
return nil, err
}
span.SetStatus(codes.Ok, "")
return s, nil
}
@@ -1087,33 +1116,36 @@ func (m *Manager) removeShare(ctx context.Context, s *collaboration.Share) error
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "removeShare")
defer span.End()
storageID, spaceID, _ := shareid.Decode(s.Id.OpaqueId)
err := m.Cache.Remove(ctx, storageID, spaceID, s.Id.OpaqueId)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.Cache.Sync(ctx, storageID, spaceID); err != nil {
return err
eg, ctx := errgroup.WithContext(ctx)
eg.Go(func() error {
storageID, spaceID, _ := shareid.Decode(s.Id.OpaqueId)
err := m.Cache.Remove(ctx, storageID, spaceID, s.Id.OpaqueId)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.Cache.Sync(ctx, storageID, spaceID); err != nil {
return err
}
err = m.Cache.Remove(ctx, storageID, spaceID, s.Id.OpaqueId)
// TODO try more often?
}
err = m.Cache.Remove(ctx, storageID, spaceID, s.Id.OpaqueId)
// TODO try more often?
}
if err != nil {
return err
}
// remove from created cache
err = m.CreatedCache.Remove(ctx, s.GetCreator().GetOpaqueId(), s.Id.OpaqueId)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.CreatedCache.Sync(ctx, s.GetCreator().GetOpaqueId()); err != nil {
return err
}
err = m.CreatedCache.Remove(ctx, s.GetCreator().GetOpaqueId(), s.Id.OpaqueId)
// TODO try more often?
}
if err != nil {
return err
}
})
eg.Go(func() error {
// remove from created cache
err := m.CreatedCache.Remove(ctx, s.GetCreator().GetOpaqueId(), s.Id.OpaqueId)
if _, ok := err.(errtypes.IsPreconditionFailed); ok {
if err := m.CreatedCache.Sync(ctx, s.GetCreator().GetOpaqueId()); err != nil {
return err
}
err = m.CreatedCache.Remove(ctx, s.GetCreator().GetOpaqueId(), s.Id.OpaqueId)
// TODO try more often?
}
return err
})
// TODO remove from grantee cache
return nil
return eg.Wait()
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/spaces.go
Generated Vendored
+96 -54
View File
@@ -27,6 +27,7 @@ import (
"path/filepath"
"strconv"
"strings"
"sync/atomic"
"time"
userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
@@ -49,6 +50,7 @@ import (
"github.com/cs3org/reva/v2/pkg/utils"
"github.com/google/uuid"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
const (
@@ -429,66 +431,106 @@ func (fs *Decomposedfs) ListStorageSpaces(ctx context.Context, filter []*provide
// But what about sharding nodes by space?
// an efficient lookup would be possible if we received a spaceid&opaqueid in the request
// the personal spaces must also use the nodeid and not the name
numShares := atomic.Int64{}
errg, ctx := errgroup.WithContext(ctx)
work := make(chan string, len(matches))
results := make(chan *provider.StorageSpace, len(matches))
numShares := 0
for match := range matches {
var err error
// TODO introduce metadata.IsLockFile(path)
// do not investigate flock files any further. They indicate file locks but are not relevant here.
if strings.HasSuffix(match, filelocks.LockFileSuffix) {
continue
}
// skip metadata files
if fs.lu.MetadataBackend().IsMetaFile(match) {
continue
}
// always read link in case storage space id != node id
spaceID, nodeID, err = ReadSpaceAndNodeFromIndexLink(match)
if err != nil {
appctx.GetLogger(ctx).Error().Err(err).Str("match", match).Msg("could not read link, skipping")
continue
}
n, err := node.ReadNode(ctx, fs.lu, spaceID, nodeID, true, nil, true)
if err != nil {
appctx.GetLogger(ctx).Error().Err(err).Str("id", nodeID).Msg("could not read node, skipping")
continue
}
if !n.Exists {
continue
}
space, err := fs.storageSpaceFromNode(ctx, n, checkNodePermissions)
if err != nil {
switch err.(type) {
case errtypes.IsPermissionDenied:
// ok
case errtypes.NotFound:
// ok
default:
appctx.GetLogger(ctx).Error().Err(err).Str("id", nodeID).Msg("could not convert to storage space")
// Distribute work
errg.Go(func() error {
defer close(work)
for match := range matches {
select {
case work <- match:
case <-ctx.Done():
return ctx.Err()
}
continue
}
return nil
})
// FIXME type share evolved to grant on the edge branch ... make it configurable if the driver should support them or not for now ... ignore type share
if space.SpaceType == spaceTypeShare {
numShares++
// do not list shares as spaces for the owner
continue
}
// TODO apply more filters
_, ok1 := spaceTypes[spaceTypeAny]
_, ok2 := spaceTypes[space.SpaceType]
if ok1 || ok2 {
spaces = append(spaces, space)
}
// Spawn workers that'll concurrently work the queue
numWorkers := 20
if len(matches) < numWorkers {
numWorkers = len(matches)
}
for i := 0; i < numWorkers; i++ {
errg.Go(func() error {
for match := range work {
var err error
// TODO introduce metadata.IsLockFile(path)
// do not investigate flock files any further. They indicate file locks but are not relevant here.
if strings.HasSuffix(match, filelocks.LockFileSuffix) {
continue
}
// skip metadata files
if fs.lu.MetadataBackend().IsMetaFile(match) {
continue
}
// always read link in case storage space id != node id
spaceID, nodeID, err = ReadSpaceAndNodeFromIndexLink(match)
if err != nil {
appctx.GetLogger(ctx).Error().Err(err).Str("match", match).Msg("could not read link, skipping")
continue
}
n, err := node.ReadNode(ctx, fs.lu, spaceID, nodeID, true, nil, true)
if err != nil {
appctx.GetLogger(ctx).Error().Err(err).Str("id", nodeID).Msg("could not read node, skipping")
continue
}
if !n.Exists {
continue
}
space, err := fs.storageSpaceFromNode(ctx, n, checkNodePermissions)
if err != nil {
switch err.(type) {
case errtypes.IsPermissionDenied:
// ok
case errtypes.NotFound:
// ok
default:
appctx.GetLogger(ctx).Error().Err(err).Str("id", nodeID).Msg("could not convert to storage space")
}
continue
}
// FIXME type share evolved to grant on the edge branch ... make it configurable if the driver should support them or not for now ... ignore type share
if space.SpaceType == spaceTypeShare {
numShares.Add(1)
// do not list shares as spaces for the owner
continue
}
// TODO apply more filters
_, ok1 := spaceTypes[spaceTypeAny]
_, ok2 := spaceTypes[space.SpaceType]
if ok1 || ok2 {
select {
case results <- space:
case <-ctx.Done():
return ctx.Err()
}
}
}
return nil
})
}
// Wait for things to settle down, then close results chan
go func() {
_ = errg.Wait() // error is checked later
close(results)
}()
for r := range results {
spaces = append(spaces, r)
}
// if there are no matches (or they happened to be spaces for the owner) and the node is a child return a space
if len(matches) <= numShares && nodeID != spaceID {
if int64(len(matches)) <= numShares.Load() && nodeID != spaceID {
// try node id
n, err := node.ReadNode(ctx, fs.lu, spaceID, nodeID, true, nil, false) // permission to read disabled space is checked in storageSpaceFromNode
if err != nil {
vendor/github.com/cs3org/reva/v2/pkg/utils/ldap/reconnect.go
Generated Vendored
+5 -5
View File
@@ -163,10 +163,10 @@ func (c *ConnWithReconnect) getConnection() (*ldap.Conn, error) {
}
func (c *ConnWithReconnect) ldapAutoConnect(config Config) {
l, err := c.ldapConnect(config)
if err != nil {
c.logger.Debug().Err(err).Msg("autoconnect could not get ldap Connection")
}
var (
l *ldap.Conn
err error
)
for {
select {
@@ -203,7 +203,7 @@ func (c *ConnWithReconnect) ldapConnect(config Config) (*ldap.Conn, error) {
}
if err != nil {
c.logger.Debug().Err(err).Msg("could not get ldap Connection")
c.logger.Error().Err(err).Msg("could not get ldap Connection")
return nil, err
}
c.logger.Debug().Msg("LDAP Connected")