Enhance storage registry with virtual views and regular expressions. (cs3org#1570)

Author: ishank011

Diff for: changelog/unreleased/storage-registry-refactor.md

@@ -0,0 +1,9 @@
+Enhancement: Enhance storage registry with virtual views and regular expressions
+
+Add the functionality to the storage registry service to handle user requests
+for references which can span across multiple storage providers, particularly
+useful for cases where directories are sharded across providers or virtual views
+are expected.
+
+https://github.com/cs3org/cs3apis/pull/116
+https://github.com/cs3org/reva/pull/1570

Diff for: go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/cheggaaa/pb v1.0.29
 	github.com/coreos/go-oidc v2.2.1+incompatible
 	github.com/cs3org/cato v0.0.0-20200828125504-e418fc54dd5e
-	github.com/cs3org/go-cs3apis v0.0.0-20210316113645-e4a74cb8761c
+	github.com/cs3org/go-cs3apis v0.0.0-20210322124405-872bbbf14d0b
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/eventials/go-tus v0.0.0-20200718001131-45c7ec8f5d59
 	github.com/ffurano/grpc-proto v0.0.0-20210312134900-65801a1ca184

Diff for: go.sum

@@ -142,8 +142,8 @@ github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7Do
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/cs3org/cato v0.0.0-20200828125504-e418fc54dd5e h1:tqSPWQeueWTKnJVMJffz4pz0o1WuQxJ28+5x5JgaHD8=
 github.com/cs3org/cato v0.0.0-20200828125504-e418fc54dd5e/go.mod h1:XJEZ3/EQuI3BXTp/6DUzFr850vlxq11I6satRtz0YQ4=
-github.com/cs3org/go-cs3apis v0.0.0-20210316113645-e4a74cb8761c h1:2vcWjiaFkJMhMZHeTbkkXWwhhAOTAIKpul8yjAo95UU=
-github.com/cs3org/go-cs3apis v0.0.0-20210316113645-e4a74cb8761c/go.mod h1:UXha4TguuB52H14EMoSsCqDj7k8a/t7g4gVP+bgY5LY=
+github.com/cs3org/go-cs3apis v0.0.0-20210322124405-872bbbf14d0b h1:80DK9Yufaj1YJ0fPb6x1WZfijHWA+CMstq3MEZs/8To=
+github.com/cs3org/go-cs3apis v0.0.0-20210322124405-872bbbf14d0b/go.mod h1:UXha4TguuB52H14EMoSsCqDj7k8a/t7g4gVP+bgY5LY=
 github.com/cucumber/godog v0.8.1/go.mod h1:vSh3r/lM+psC1BPXvdkSEuNjmXfpVqrMGYAElF6hxnA=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

Diff for: internal/grpc/services/gateway/storageprovider.go

@@ -24,6 +24,7 @@ import (
 	"net/url"
 	"path"
 	"strings"
+	"sync"
 	"time"
 
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
@@ -35,10 +36,9 @@ import (
 	"github.com/cs3org/reva/pkg/rgrpc/status"
 	"github.com/cs3org/reva/pkg/rgrpc/todo/pool"
 	"github.com/cs3org/reva/pkg/storage/utils/etag"
-	"github.com/cs3org/reva/pkg/storage/utils/templates"
-	"github.com/cs3org/reva/pkg/user"
 	"github.com/cs3org/reva/pkg/utils"
 	"github.com/dgrijalva/jwt-go"
+	"github.com/google/uuid"
 	"github.com/pkg/errors"
 )
 
@@ -193,6 +193,7 @@ func (s *svc) getHome(_ context.Context) string {
 	// TODO(labkode): issue #601, /home will be hardcoded.
 	return "/home"
 }
+
 func (s *svc) InitiateFileDownload(ctx context.Context, req *provider.InitiateFileDownloadRequest) (*gateway.InitiateFileDownloadResponse, error) {
 	log := appctx.GetLogger(ctx)
 	p, st := s.getPath(ctx, req.Ref)
@@ -366,6 +367,7 @@ func (s *svc) InitiateFileDownload(ctx context.Context, req *provider.InitiateFi
 }
 
 func (s *svc) initiateFileDownload(ctx context.Context, req *provider.InitiateFileDownloadRequest) (*gateway.InitiateFileDownloadResponse, error) {
+	// TODO(ishank011): enable downloading references spread across storage providers, eg. /eos
 	c, err := s.find(ctx, req.Ref)
 	if err != nil {
 		return &gateway.InitiateFileDownloadResponse{
@@ -857,6 +859,7 @@ func (s *svc) Delete(ctx context.Context, req *provider.DeleteRequest) (*provide
 }
 
 func (s *svc) delete(ctx context.Context, req *provider.DeleteRequest) (*provider.DeleteResponse, error) {
+	// TODO(ishank011): enable deleting references spread across storage providers, eg. /eos
 	c, err := s.find(ctx, req.Ref)
 	if err != nil {
 		return &provider.DeleteResponse{
@@ -974,19 +977,20 @@ func (s *svc) Move(ctx context.Context, req *provider.MoveRequest) (*provider.Mo
 }
 
 func (s *svc) move(ctx context.Context, req *provider.MoveRequest) (*provider.MoveResponse, error) {
-	srcP, err := s.findProvider(ctx, req.Source)
+	srcList, err := s.findProviders(ctx, req.Source)
 	if err != nil {
 		return &provider.MoveResponse{
 			Status: status.NewStatusFromErrType(ctx, "move src="+req.Source.String(), err),
 		}, nil
 	}
 
-	dstP, err := s.findProvider(ctx, req.Destination)
+	dstList, err := s.findProviders(ctx, req.Destination)
 	if err != nil {
 		return &provider.MoveResponse{
 			Status: status.NewStatusFromErrType(ctx, "move dst="+req.Destination.String(), err),
 		}, nil
 	}
+	srcP, dstP := srcList[0], dstList[0]
 
 	// if providers are not the same we do not implement cross storage copy yet.
 	if srcP.Address != dstP.Address {
@@ -1007,6 +1011,7 @@ func (s *svc) move(ctx context.Context, req *provider.MoveRequest) (*provider.Mo
 }
 
 func (s *svc) SetArbitraryMetadata(ctx context.Context, req *provider.SetArbitraryMetadataRequest) (*provider.SetArbitraryMetadataResponse, error) {
+	// TODO(ishank011): enable for references spread across storage providers, eg. /eos
 	c, err := s.find(ctx, req.Ref)
 	if err != nil {
 		return &provider.SetArbitraryMetadataResponse{
@@ -1023,6 +1028,7 @@ func (s *svc) SetArbitraryMetadata(ctx context.Context, req *provider.SetArbitra
 }
 
 func (s *svc) UnsetArbitraryMetadata(ctx context.Context, req *provider.UnsetArbitraryMetadataRequest) (*provider.UnsetArbitraryMetadataResponse, error) {
+	// TODO(ishank011): enable for references spread across storage providers, eg. /eos
 	c, err := s.find(ctx, req.Ref)
 	if err != nil {
 		return &provider.UnsetArbitraryMetadataResponse{
@@ -1142,14 +1148,89 @@ func (s *svc) statSharesFolder(ctx context.Context) (*provider.StatResponse, err
 }
 
 func (s *svc) stat(ctx context.Context, req *provider.StatRequest) (*provider.StatResponse, error) {
-	c, err := s.find(ctx, req.Ref)
+	providers, err := s.findProviders(ctx, req.Ref)
 	if err != nil {
 		return &provider.StatResponse{
-			Status: status.NewStatusFromErrType(ctx, "stat ref="+req.Ref.String(), err),
+			Status: status.NewStatusFromErrType(ctx, "stat ref: "+req.Ref.String(), err),
 		}, nil
 	}
 
-	return c.Stat(ctx, req)
+	resPath := req.Ref.GetPath()
+	if len(providers) == 1 && (resPath == "" || strings.HasPrefix(resPath, providers[0].ProviderPath)) {
+		c, err := s.getStorageProviderClient(ctx, providers[0])
+		if err != nil {
+			return &provider.StatResponse{
+				Status: status.NewInternal(ctx, err, "error connecting to storage provider="+providers[0].Address),
+			}, nil
+		}
+		return c.Stat(ctx, req)
+	}
+
+	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)
+	}
+	wg.Wait()
+
+	var totalSize uint64
+	for i := range providers {
+		if errors[i] != nil {
+			return &provider.StatResponse{
+				Status: status.NewStatusFromErrType(ctx, "stat ref: "+req.Ref.String(), errors[i]),
+			}, nil
+		}
+		if infoFromProviders[i] != nil {
+			totalSize += infoFromProviders[i].Size
+		}
+	}
+
+	// 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: resPath,
+			Size: totalSize,
+		},
+	}, 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
+	}
+
+	resPath := path.Clean(req.Ref.GetPath())
+	newPath := req.Ref.GetPath()
+	if resPath != "" && !strings.HasPrefix(resPath, p.ProviderPath) {
+		newPath = p.ProviderPath
+	}
+	r, err := c.Stat(ctx, &provider.StatRequest{
+		Ref: &provider.Reference{
+			Spec: &provider.Reference_Path{
+				Path: newPath,
+			},
+		},
+	})
+	if err != nil {
+		*e = errors.Wrap(err, "gateway: error calling ListContainer")
+		return
+	}
+	if res == nil {
+		res = &provider.ResourceInfo{}
+	}
+	*res = *r.Info
 }
 
 func (s *svc) Stat(ctx context.Context, req *provider.StatRequest) (*provider.StatResponse, error) {
@@ -1454,19 +1535,88 @@ func (s *svc) listSharesFolder(ctx context.Context) (*provider.ListContainerResp
 }
 
 func (s *svc) listContainer(ctx context.Context, req *provider.ListContainerRequest) (*provider.ListContainerResponse, error) {
-	c, err := s.find(ctx, req.Ref)
+	providers, err := s.findProviders(ctx, req.Ref)
 	if err != nil {
 		return &provider.ListContainerResponse{
-			Status: status.NewStatusFromErrType(ctx, "listContainer ref="+req.Ref.String(), err),
+			Status: status.NewStatusFromErrType(ctx, "listContainer ref: "+req.Ref.String(), err),
 		}, nil
 	}
 
-	res, err := c.ListContainer(ctx, req)
+	resPath := path.Clean(req.Ref.GetPath())
+	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.listContainerOnProvider(ctx, req, &infoFromProviders[i], p, &errors[i], &wg)
+	}
+	wg.Wait()
+
+	infos := []*provider.ResourceInfo{}
+	indirects := make(map[string][]*provider.ResourceInfo)
+	for i := range providers {
+		if errors[i] != nil {
+			return &provider.ListContainerResponse{
+				Status: status.NewStatusFromErrType(ctx, "listContainer ref: "+req.Ref.String(), errors[i]),
+			}, nil
+		}
+		for _, inf := range infoFromProviders[i] {
+			if parent := path.Dir(inf.Path); resPath != "" && resPath != parent {
+				parts := strings.Split(strings.TrimPrefix(inf.Path, resPath), "/")
+				p := path.Join(resPath, parts[1])
+				indirects[p] = append(indirects[p], inf)
+			} else {
+				infos = append(infos, inf)
+			}
+		}
+	}
+
+	for k, v := range indirects {
+		inf := &provider.ResourceInfo{
+			Id: &provider.ResourceId{
+				StorageId: "/",
+				OpaqueId:  uuid.New().String(),
+			},
+			Type: provider.ResourceType_RESOURCE_TYPE_CONTAINER,
+			Etag: etag.GenerateEtagFromResources(nil, v),
+			Path: k,
+			Size: 0,
+		}
+		infos = append(infos, inf)
+	}
+
+	return &provider.ListContainerResponse{
+		Status: status.NewOK(ctx),
+		Infos:  infos,
+	}, nil
+}
+
+func (s *svc) listContainerOnProvider(ctx context.Context, req *provider.ListContainerRequest, res *[]*provider.ResourceInfo, p *registry.ProviderInfo, e *error, wg *sync.WaitGroup) {
+	defer wg.Done()
+	c, err := s.getStorageProviderClient(ctx, p)
 	if err != nil {
-		return nil, errors.Wrap(err, "gateway: error calling ListContainer")
+		*e = errors.Wrap(err, "error connecting to storage provider="+p.Address)
+		return
 	}
 
-	return res, nil
+	resPath := path.Clean(req.Ref.GetPath())
+	newPath := req.Ref.GetPath()
+	if resPath != "" && !strings.HasPrefix(resPath, p.ProviderPath) {
+		newPath = p.ProviderPath
+	}
+	r, err := c.ListContainer(ctx, &provider.ListContainerRequest{
+		Ref: &provider.Reference{
+			Spec: &provider.Reference_Path{
+				Path: newPath,
+			},
+		},
+	})
+	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) {
@@ -1888,11 +2038,11 @@ func (s *svc) findByPath(ctx context.Context, path string) (provider.ProviderAPI
 }
 
 func (s *svc) find(ctx context.Context, ref *provider.Reference) (provider.ProviderAPIClient, error) {
-	p, err := s.findProvider(ctx, ref)
+	p, err := s.findProviders(ctx, ref)
 	if err != nil {
 		return nil, err
 	}
-	return s.getStorageProviderClient(ctx, p)
+	return s.getStorageProviderClient(ctx, p[0])
 }
 
 func (s *svc) getStorageProviderClient(_ context.Context, p *registry.ProviderInfo) (provider.ProviderAPIClient, error) {
@@ -1905,37 +2055,13 @@ func (s *svc) getStorageProviderClient(_ context.Context, p *registry.ProviderIn
 	return c, nil
 }
 
-func (s *svc) findProvider(ctx context.Context, ref *provider.Reference) (*registry.ProviderInfo, error) {
-	home := s.getHome(ctx)
-	if strings.HasPrefix(ref.GetPath(), home) && s.c.HomeMapping != "" {
-		if u, ok := user.ContextGetUser(ctx); ok {
-			layout := templates.WithUser(u, s.c.HomeMapping)
-			newRef := &provider.Reference{
-				Spec: &provider.Reference_Path{
-					Path: path.Join(layout, strings.TrimPrefix(ref.GetPath(), home)),
-				},
-			}
-			res, err := s.getStorageProvider(ctx, newRef)
-			if err != nil {
-				// if we get a NotFound error, default to the original reference
-				if _, ok := err.(errtypes.IsNotFound); !ok {
-					return nil, err
-				}
-			} else {
-				return res, nil
-			}
-		}
-	}
-	return s.getStorageProvider(ctx, ref)
-}
-
-func (s *svc) getStorageProvider(ctx context.Context, ref *provider.Reference) (*registry.ProviderInfo, error) {
+func (s *svc) findProviders(ctx context.Context, ref *provider.Reference) ([]*registry.ProviderInfo, error) {
 	c, err := pool.GetStorageRegistryClient(s.c.StorageRegistryEndpoint)
 	if err != nil {
 		return nil, errors.Wrap(err, "gateway: error getting storage registry client")
 	}
 
-	res, err := c.GetStorageProvider(ctx, &registry.GetStorageProviderRequest{
+	res, err := c.GetStorageProviders(ctx, &registry.GetStorageProvidersRequest{
 		Ref: ref,
 	})
 
@@ -1958,11 +2084,11 @@ func (s *svc) getStorageProvider(ctx context.Context, ref *provider.Reference) (
 		}
 	}
 
-	if res.Provider == nil {
+	if res.Providers == nil {
 		return nil, errors.New("gateway: provider is nil")
 	}
 
-	return res.Provider, nil
+	return res.Providers, nil
 }
 
 type etagWithTS struct {