Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Aggregrate resource info properties for virtual views #2215

Merged
merged 2 commits into from
Oct 28, 2021
Merged

Aggregrate resource info properties for virtual views #2215

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2f51dc9
Aggregrate resource info properties for virtual views
ishank011 Oct 28, 2021
9b1d711
Ignore errors when statting or listing across providers
ishank011 Oct 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions changelog/unreleased/eos-virtual-views.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
Enhancement: Aggregrate resource info properties for virtual views

https://github.com/cs3org/reva/pull/2215
226 changes: 93 additions & 133 deletions internal/grpc/services/gateway/storageprovider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ import (
"fmt"
"net/url"
"path"
"path/filepath"
"strings"
"sync"
"time"
Expand Down Expand Up @@ -1283,6 +1282,7 @@ func (s *svc) stat(ctx context.Context, req *provider.StatRequest) (*provider.St
Status: status.NewStatusFromErrType(ctx, "stat ref: "+req.Ref.String(), err),
}, nil
}
providers = getUniqueProviders(providers)

resPath := req.Ref.GetPath()
if len(providers) == 1 && (utils.IsRelativeReference(req.Ref) || resPath == "" || strings.HasPrefix(resPath, providers[0].ProviderPath)) {
Expand All @@ -1303,69 +1303,47 @@ func (s *svc) stat(ctx context.Context, req *provider.StatRequest) (*provider.St
}

func (s *svc) statAcrossProviders(ctx context.Context, req *provider.StatRequest, providers []*registry.ProviderInfo) (*provider.StatResponse, error) {
// TODO(ishank011): aggregrate properties such as etag, checksum, etc.
log := appctx.GetLogger(ctx)

infoFromProviders := make([]*provider.ResourceInfo, len(providers))
errors := make([]error, len(providers))
var wg sync.WaitGroup

for i, p := range providers {
wg.Add(1)
go s.statOnProvider(ctx, req, &infoFromProviders[i], p, &errors[i], &wg)
info := &provider.ResourceInfo{
Id: &provider.ResourceId{
StorageId: "/",
OpaqueId: uuid.New().String(),
},
Type: provider.ResourceType_RESOURCE_TYPE_CONTAINER,
Path: req.Ref.GetPath(),
MimeType: "httpd/unix-directory",
Size: 0,
Mtime: &types.Timestamp{},
}
wg.Wait()

var totalSize uint64
for i := range providers {
if errors[i] != nil {
log.Warn().Msgf("statting on provider %s returned err %+v", providers[i].ProviderPath, errors[i])
for _, p := range providers {
c, err := s.getStorageProviderClient(ctx, p)
if err != nil {
log.Err(err).Msg("error connecting to storage provider=" + p.Address)
continue
}
resp, err := c.Stat(ctx, req)
if err != nil {
log.Err(err).Msgf("gateway: error calling Stat %s: %+v", req.Ref.String(), p)
continue
}
if resp.Status.Code != rpc.Code_CODE_OK {
log.Err(status.NewErrorFromCode(rpc.Code_CODE_OK, "gateway"))
continue
}
if infoFromProviders[i] != nil {
totalSize += infoFromProviders[i].Size
if resp.Info != nil {
info.Size += resp.Info.Size
info.Mtime = utils.LaterTS(info.Mtime, resp.Info.Mtime)
}
}

// TODO(ishank011): aggregrate other properties for references spread across storage providers, eg. /eos
return &provider.StatResponse{
Status: status.NewOK(ctx),
Info: &provider.ResourceInfo{
Id: &provider.ResourceId{
StorageId: "/",
OpaqueId: uuid.New().String(),
},
Type: provider.ResourceType_RESOURCE_TYPE_CONTAINER,
Path: req.Ref.GetPath(),
Size: totalSize,
},
Info: info,
}, nil
}

func (s *svc) statOnProvider(ctx context.Context, req *provider.StatRequest, res **provider.ResourceInfo, p *registry.ProviderInfo, e *error, wg *sync.WaitGroup) {
defer wg.Done()
c, err := s.getStorageProviderClient(ctx, p)
if err != nil {
*e = errors.Wrap(err, "error connecting to storage provider="+p.Address)
return
}

if utils.IsAbsoluteReference(req.Ref) {
resPath := path.Clean(req.Ref.GetPath())
newPath := req.Ref.GetPath()
if resPath != "." && !strings.HasPrefix(resPath, p.ProviderPath) {
newPath = p.ProviderPath
}
req.Ref = &provider.Reference{Path: newPath}
}

r, err := c.Stat(ctx, req)
if err != nil {
*e = errors.Wrap(err, fmt.Sprintf("gateway: error calling Stat %s on %+v", req.Ref, p))
return
}
*res = r.Info
}

func (s *svc) Stat(ctx context.Context, req *provider.StatRequest) (*provider.StatResponse, error) {

if utils.IsRelativeReference(req.Ref) {
Expand Down Expand Up @@ -1651,60 +1629,74 @@ func (s *svc) listSharesFolder(ctx context.Context) (*provider.ListContainerResp
}

func (s *svc) listContainer(ctx context.Context, req *provider.ListContainerRequest) (*provider.ListContainerResponse, error) {
log := appctx.GetLogger(ctx)
providers, err := s.findProviders(ctx, req.Ref)
if err != nil {
return &provider.ListContainerResponse{
Status: status.NewStatusFromErrType(ctx, "listContainer ref: "+req.Ref.String(), err),
}, nil
}
providers = getUniqueProviders(providers)

infoFromProviders := make([][]*provider.ResourceInfo, len(providers))
errors := make([]error, len(providers))
indirects := make([]bool, len(providers))
var wg sync.WaitGroup

for i, p := range providers {
wg.Add(1)
go s.listContainerOnProvider(ctx, req, &infoFromProviders[i], p, &indirects[i], &errors[i], &wg)
resPath := req.Ref.GetPath()
if len(providers) == 1 && (utils.IsRelativeReference(req.Ref) || resPath == "" || strings.HasPrefix(resPath, providers[0].ProviderPath)) {
c, err := s.getStorageProviderClient(ctx, providers[0])
if err != nil {
return &provider.ListContainerResponse{
Status: status.NewInternal(ctx, err, "error connecting to storage provider="+providers[0].Address),
}, nil
}
rsp, err := c.ListContainer(ctx, req)
if err != nil || rsp.Status.Code != rpc.Code_CODE_OK {
return rsp, err
}
return rsp, nil
}
wg.Wait()

infos := []*provider.ResourceInfo{}
nestedInfos := make(map[string][]*provider.ResourceInfo)
for i := range providers {
if errors[i] != nil {
// return if there's only one mount, else skip this one
if len(providers) == 1 {
return &provider.ListContainerResponse{
Status: status.NewStatusFromErrType(ctx, "listContainer ref: "+req.Ref.String(), errors[i]),
}, nil
}
log.Warn().Msgf("listing container on provider %s returned err %+v", providers[i].ProviderPath, errors[i])
return s.listContainerAcrossProviders(ctx, req, providers)
}

func (s *svc) listContainerAcrossProviders(ctx context.Context, req *provider.ListContainerRequest, providers []*registry.ProviderInfo) (*provider.ListContainerResponse, error) {
nestedInfos := make(map[string]*provider.ResourceInfo)
log := appctx.GetLogger(ctx)

for _, p := range providers {
c, err := s.getStorageProviderClient(ctx, p)
if err != nil {
log.Err(err).Msg("error connecting to storage provider=" + p.Address)
continue
}
resp, err := c.ListContainer(ctx, req)
if err != nil {
log.Err(err).Msgf("gateway: error calling Stat %s: %+v", req.Ref.String(), p)
continue
}
if resp.Status.Code != rpc.Code_CODE_OK {
log.Err(status.NewErrorFromCode(rpc.Code_CODE_OK, "gateway"))
continue
}
for _, inf := range infoFromProviders[i] {
if indirects[i] {
p := inf.Path
// TODO do we need to trim prefix here for relative references?
nestedInfos[p] = append(nestedInfos[p], inf)

for _, info := range resp.Infos {
if p, ok := nestedInfos[info.Path]; ok {
// Since more than one providers contribute to this path,
// use a generic ID
p.Id = &provider.ResourceId{
StorageId: "/",
OpaqueId: uuid.New().String(),
}
// TODO(ishank011): aggregrate properties such as etag, checksum, etc.
p.Size += info.Size
p.Mtime = utils.LaterTS(p.Mtime, info.Mtime)
p.Type = provider.ResourceType_RESOURCE_TYPE_CONTAINER
p.MimeType = "httpd/unix-directory"
} else {
infos = append(infos, inf)
nestedInfos[info.Path] = info
}
}
}

for k := range nestedInfos {
inf := &provider.ResourceInfo{
Id: &provider.ResourceId{
StorageId: "/",
OpaqueId: uuid.New().String(),
},
Type: provider.ResourceType_RESOURCE_TYPE_CONTAINER,
Path: k,
Size: 0,
}
infos = append(infos, inf)
infos := make([]*provider.ResourceInfo, 0, len(nestedInfos))
for _, info := range nestedInfos {
infos = append(infos, info)
}

return &provider.ListContainerResponse{
Expand All @@ -1713,50 +1705,6 @@ func (s *svc) listContainer(ctx context.Context, req *provider.ListContainerRequ
}, nil
}

func (s *svc) listContainerOnProvider(ctx context.Context, req *provider.ListContainerRequest, res *[]*provider.ResourceInfo, p *registry.ProviderInfo, ind *bool, e *error, wg *sync.WaitGroup) {
defer wg.Done()
c, err := s.getStorageProviderClient(ctx, p)
if err != nil {
*e = errors.Wrap(err, "error connecting to storage provider="+p.Address)
return
}

if utils.IsAbsoluteReference(req.Ref) {
resPath := path.Clean(req.Ref.GetPath())
if resPath != "" && !strings.HasPrefix(resPath, p.ProviderPath) {
// The path which we're supposed to list encompasses this provider
// so just return the first child and mark it as indirect
rel, err := filepath.Rel(resPath, p.ProviderPath)
if err != nil {
*e = err
return
}
parts := strings.Split(rel, "/")
p := path.Join(resPath, parts[0])
*ind = true
*res = []*provider.ResourceInfo{
{
Id: &provider.ResourceId{
StorageId: "/",
OpaqueId: uuid.New().String(),
},
Type: provider.ResourceType_RESOURCE_TYPE_CONTAINER,
Path: p,
Size: 0,
},
}
return
}
}

r, err := c.ListContainer(ctx, req)
if err != nil {
*e = errors.Wrap(err, "gateway: error calling ListContainer")
return
}
*res = r.Infos
}

func (s *svc) ListContainer(ctx context.Context, req *provider.ListContainerRequest) (*provider.ListContainerResponse, error) {
log := appctx.GetLogger(ctx)

Expand Down Expand Up @@ -2197,6 +2145,18 @@ func (s *svc) findProviders(ctx context.Context, ref *provider.Reference) ([]*re
return res.Providers, nil
}

func getUniqueProviders(providers []*registry.ProviderInfo) []*registry.ProviderInfo {
unique := make(map[string]bool)
for _, p := range providers {
unique[p.Address] = true
}
p := make([]*registry.ProviderInfo, 0, len(unique))
for addr := range unique {
p = append(p, &registry.ProviderInfo{Address: addr})
}
return p
}

type etagWithTS struct {
Etag string
Timestamp time.Time
Expand Down
Loading