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

Add a IndexSpace method to the search provider

Browse Source 
IndexSpace recursively traverses a given space and (re-)index all files
it finds.
This commit is contained in:
André Duffeck
2022-04-27 09:55:32 +02:00
parent 827885abad
commit 656ef9d6ae
13 changed files with 545 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/search/pkg/search/index/index.go
-3
View File
@@ -39,7 +39,6 @@ type indexDocument struct {
ID string
Name string
Etag string
Size uint64
Mtime string
MimeType string
@@ -123,7 +122,6 @@ func toEntity(ref *sprovider.Reference, ri *sprovider.ResourceInfo) *indexDocume
Path: ref.Path,
ID: idToBleveId(ri.Id),
Name: ri.Path,
Etag: ri.Etag,
Size: ri.Size,
MimeType: ri.MimeType,
}
@@ -154,7 +152,6 @@ func fromFields(fields map[string]interface{}) (*searchmsg.Match, error) {
},
Name: fields["Name"].(string),
Size: uint64(fields["Size"].(float64)),
Etag: fields["Etag"].(string),
MimeType: fields["MimeType"].(string),
},
}
extensions/search/pkg/search/index/index_test.go
-2
View File
@@ -46,7 +46,6 @@ var _ = Describe("Index", func() {
},
Path: "foo.pdf",
Size: 12345,
Etag: "abcde",
MimeType: "application/pdf",
Mtime: &typesv1beta1.Timestamp{Seconds: 4000},
}
@@ -124,7 +123,6 @@ var _ = Describe("Index", func() {
Expect(match.Entity.Id.OpaqueId).To(Equal(ri.Id.OpaqueId))
Expect(match.Entity.Name).To(Equal(ri.Path))
Expect(match.Entity.Size).To(Equal(ri.Size))
Expect(match.Entity.Etag).To(Equal(ri.Etag))
Expect(match.Entity.MimeType).To(Equal(ri.MimeType))
Expect(uint64(match.Entity.LastModifiedTime.AsTime().Unix())).To(Equal(ri.Mtime.Seconds))
})
extensions/search/pkg/search/mocks/ProviderClient.go
+23
View File
@@ -15,6 +15,29 @@ type ProviderClient struct {
mock.Mock
}
// IndexSpace provides a mock function with given fields: ctx, req
func (_m *ProviderClient) IndexSpace(ctx context.Context, req *v0.IndexSpaceRequest) (*v0.IndexSpaceResponse, error) {
ret := _m.Called(ctx, req)
var r0 *v0.IndexSpaceResponse
if rf, ok := ret.Get(0).(func(context.Context, *v0.IndexSpaceRequest) *v0.IndexSpaceResponse); ok {
r0 = rf(ctx, req)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*v0.IndexSpaceResponse)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, *v0.IndexSpaceRequest) error); ok {
r1 = rf(ctx, req)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// Search provides a mock function with given fields: ctx, req
func (_m *ProviderClient) Search(ctx context.Context, req *v0.SearchRequest) (*v0.SearchResponse, error) {
ret := _m.Called(ctx, req)
extensions/search/pkg/search/provider/searchprovider.go
+82 -10
View File
@@ -2,19 +2,23 @@ package provider
import (
"context"
"fmt"
"path/filepath"
"strings"
gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
providerv1beta1 "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
typesv1beta1 "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
ctxpkg "github.com/cs3org/reva/v2/pkg/ctx"
"github.com/cs3org/reva/v2/pkg/errtypes"
"github.com/cs3org/reva/v2/pkg/events"
"github.com/cs3org/reva/v2/pkg/storage/utils/walker"
"github.com/cs3org/reva/v2/pkg/utils"
"github.com/owncloud/ocis/extensions/search/pkg/search"
"github.com/owncloud/ocis/ocis-pkg/log"
"google.golang.org/grpc/metadata"
searchmsg "github.com/owncloud/ocis/protogen/gen/ocis/messages/search/v0"
@@ -22,16 +26,24 @@ import (
)
type Provider struct {
logger log.Logger
gwClient gateway.GatewayAPIClient
indexClient search.IndexClient
machineAuthAPIKey string
}
func New(gwClient gateway.GatewayAPIClient, indexClient search.IndexClient, machineAuthAPIKey string, eventsChan <-chan interface{}) *Provider {
func New(gwClient gateway.GatewayAPIClient, indexClient search.IndexClient, machineAuthAPIKey string, eventsChan <-chan interface{}, logger log.Logger) *Provider {
p := &Provider{
gwClient: gwClient,
indexClient: indexClient,
machineAuthAPIKey: machineAuthAPIKey,
logger: logger,
}
go func() {
for {
ev := <-eventsChan
var ref *providerv1beta1.Reference
var ref *provider.Reference
var owner *user.User
switch e := ev.(type) {
case events.FileUploaded:
@@ -46,23 +58,29 @@ func New(gwClient gateway.GatewayAPIClient, indexClient search.IndexClient, mach
// Get auth
ownerCtx := ctxpkg.ContextSetUser(context.Background(), owner)
authRes, err := gwClient.Authenticate(ownerCtx, &gateway.AuthenticateRequest{
authRes, err := p.gwClient.Authenticate(ownerCtx, &gateway.AuthenticateRequest{
Type: "machine",
ClientId: "userid:" + owner.Id.OpaqueId,
ClientSecret: machineAuthAPIKey,
})
if err != nil || authRes.GetStatus().GetCode() != rpc.Code_CODE_OK {
// TODO: log error
p.logger.Error().Err(err).Interface("authRes", authRes).Msg("error using machine auth")
}
ownerCtx = metadata.AppendToOutgoingContext(ownerCtx, ctxpkg.TokenHeader, authRes.Token)
// Stat changed resource resource
statRes, err := gwClient.Stat(ownerCtx, &providerv1beta1.StatRequest{Ref: ref})
statRes, err := gwClient.Stat(ownerCtx, &provider.StatRequest{Ref: ref})
if err != nil || statRes.Status.Code != rpc.Code_CODE_OK {
// TODO: log error
p.logger.Error().Err(err).Interface("statRes", statRes).Msg("failed to stat the changed resource")
}
err = p.indexClient.Add(ref, statRes.Info)
if err != nil {
p.logger.Error().Err(err).Msg("error adding resource to the index")
} else {
p.logDocCount()
}
indexClient.Add(ref, statRes.Info)
}
}()
@@ -82,7 +100,7 @@ func (p *Provider) Search(ctx context.Context, req *searchsvc.SearchRequest) (*s
return nil, errtypes.PreconditionFailed("empty query provided")
}
listSpacesRes, err := p.gwClient.ListStorageSpaces(ctx, &providerv1beta1.ListStorageSpacesRequest{
listSpacesRes, err := p.gwClient.ListStorageSpaces(ctx, &provider.ListStorageSpacesRequest{
Opaque: &typesv1beta1.Opaque{Map: map[string]*typesv1beta1.OpaqueEntry{
"path": {
Decoder: "plain",
@@ -98,7 +116,7 @@ func (p *Provider) Search(ctx context.Context, req *searchsvc.SearchRequest) (*s
for _, space := range listSpacesRes.StorageSpaces {
pathPrefix := ""
if space.SpaceType == "grant" {
gpRes, err := p.gwClient.GetPath(ctx, &providerv1beta1.GetPathRequest{
gpRes, err := p.gwClient.GetPath(ctx, &provider.GetPathRequest{
ResourceId: space.Root,
})
if err != nil {
@@ -136,3 +154,57 @@ func (p *Provider) Search(ctx context.Context, req *searchsvc.SearchRequest) (*s
Matches: matches,
}, nil
}
func (p *Provider) IndexSpace(ctx context.Context, req *searchsvc.IndexSpaceRequest) (*searchsvc.IndexSpaceResponse, error) {
// get user
res, err := p.gwClient.GetUserByClaim(context.Background(), &user.GetUserByClaimRequest{
Claim: "username",
Value: req.UserId,
})
if err != nil || res.Status.Code != rpc.Code_CODE_OK {
fmt.Println("error: Could not get user by userid")
return nil, err
}
// Get auth context
ownerCtx := ctxpkg.ContextSetUser(context.Background(), res.User)
authRes, err := p.gwClient.Authenticate(ownerCtx, &gateway.AuthenticateRequest{
Type: "machine",
ClientId: "userid:" + res.User.Id.OpaqueId,
ClientSecret: p.machineAuthAPIKey,
})
if err != nil || authRes.GetStatus().GetCode() != rpc.Code_CODE_OK {
return nil, err
}
if authRes.GetStatus().GetCode() != rpc.Code_CODE_OK {
return nil, fmt.Errorf("could not get authenticated context for user")
}
ownerCtx = metadata.AppendToOutgoingContext(ownerCtx, ctxpkg.TokenHeader, authRes.Token)
// Walk the space and index all files
walker := walker.NewWalker(p.gwClient)
rootId := &provider.ResourceId{StorageId: req.SpaceId, OpaqueId: req.SpaceId}
err = walker.Walk(ownerCtx, rootId, func(wd string, info *provider.ResourceInfo, err error) error {
if err != nil {
p.logger.Error().Err(err).Msg("error walking the tree")
}
ref := &provider.Reference{
Path: utils.MakeRelativePath(filepath.Join(wd, info.Path)),
ResourceId: rootId,
}
err = p.indexClient.Add(ref, info)
if err != nil {
p.logger.Error().Err(err).Msg("error adding resource to the index")
} else {
p.logger.Debug().Interface("ref", ref).Msg("added resource to index")
}
return nil
})
if err != nil {
return nil, err
}
p.logDocCount()
return &searchsvc.IndexSpaceResponse{}, nil
}
extensions/search/pkg/search/provider/searchprovider_test.go
+40 -13
View File
@@ -16,6 +16,7 @@ import (
cs3mocks "github.com/cs3org/reva/v2/tests/cs3mocks/mocks"
"github.com/owncloud/ocis/extensions/search/pkg/search/mocks"
provider "github.com/owncloud/ocis/extensions/search/pkg/search/provider"
"github.com/owncloud/ocis/ocis-pkg/log"
searchmsg "github.com/owncloud/ocis/protogen/gen/ocis/messages/search/v0"
searchsvc "github.com/owncloud/ocis/protogen/gen/ocis/services/search/v0"
)
@@ -29,6 +30,12 @@ var _ = Describe("Searchprovider", func() {
ctx context.Context
eventsChan chan interface{}
logger = log.NewLogger()
user = &userv1beta1.User{
Id: &userv1beta1.UserId{
OpaqueId: "user",
},
}
otherUser = &userv1beta1.User{
Id: &userv1beta1.UserId{
OpaqueId: "otheruser",
@@ -71,27 +78,27 @@ var _ = Describe("Searchprovider", func() {
gwClient = &cs3mocks.GatewayAPIClient{}
indexClient = &mocks.IndexClient{}
p = provider.New(gwClient, indexClient, "", eventsChan)
p = provider.New(gwClient, indexClient, "", eventsChan, logger)
gwClient.On("Authenticate", mock.Anything, mock.Anything).Return(&gateway.AuthenticateResponse{
Status: status.NewOK(ctx),
Token: "authtoken",
}, nil)
gwClient.On("Stat", mock.Anything, mock.Anything).Return(&sprovider.StatResponse{
Status: status.NewOK(context.Background()),
Info: ri,
}, nil)
indexClient.On("DocCount").Return(uint64(1), nil)
})
Describe("New", func() {
It("returns a new instance", func() {
p := provider.New(gwClient, indexClient, "", eventsChan)
p := provider.New(gwClient, indexClient, "", eventsChan, logger)
Expect(p).ToNot(BeNil())
})
})
Describe("events", func() {
BeforeEach(func() {
gwClient.On("Authenticate", mock.Anything, mock.Anything).Return(&gateway.AuthenticateResponse{
Token: "authtoken",
}, nil)
gwClient.On("Stat", mock.Anything, mock.Anything).Return(&sprovider.StatResponse{
Status: status.NewOK(context.Background()),
Info: ri,
}, nil)
})
It("trigger an index update when a file has been uploaded", func() {
called := false
indexClient.On("Add", mock.Anything, mock.MatchedBy(func(riToIndex *sprovider.ResourceInfo) bool {
@@ -100,7 +107,8 @@ var _ = Describe("Searchprovider", func() {
called = true
})
eventsChan <- events.FileUploaded{
FileID: ref,
FileID: ref,
Executant: user.Id,
}
Eventually(func() bool {
@@ -109,6 +117,25 @@ var _ = Describe("Searchprovider", func() {
})
})
Describe("IndexSpace", func() {
It("walks the space and indexes all files", func() {
gwClient.On("GetUserByClaim", mock.Anything, mock.Anything).Return(&userv1beta1.GetUserByClaimResponse{
Status: status.NewOK(context.Background()),
User: user,
}, nil)
indexClient.On("Add", mock.Anything, mock.MatchedBy(func(riToIndex *sprovider.ResourceInfo) bool {
return riToIndex.Id.OpaqueId == ri.Id.OpaqueId
})).Return(nil)
res, err := p.IndexSpace(ctx, &searchsvc.IndexSpaceRequest{
SpaceId: "storageid",
UserId: "user",
})
Expect(err).ToNot(HaveOccurred())
Expect(res).ToNot(BeNil())
})
})
Describe("Search", func() {
It("fails when an empty query is given", func() {
res, err := p.Search(ctx, &searchsvc.SearchRequest{
extensions/search/pkg/search/search.go
+3
View File
@@ -31,10 +31,13 @@ import (
// ProviderClient is the interface to the search provider service
type ProviderClient interface {
Search(ctx context.Context, req *searchsvc.SearchRequest) (*searchsvc.SearchResponse, error)
IndexSpace(ctx context.Context, req *searchsvc.IndexSpaceRequest) (*searchsvc.IndexSpaceResponse, error)
}
// IndexClient is the interface to the search index
type IndexClient interface {
Search(ctx context.Context, req *searchsvc.SearchIndexRequest) (*searchsvc.SearchIndexResponse, error)
Add(ref *providerv1beta1.Reference, ri *providerv1beta1.ResourceInfo) error
Remove(ri *providerv1beta1.ResourceInfo) error
DocCount() (uint64, error)
}
extensions/search/pkg/service/v0/service.go
+6 -1
View File
@@ -61,7 +61,7 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, error) {
logger.Fatal().Err(err).Str("addr", cfg.Reva.Address).Msg("could not get reva client")
}
provider := searchprovider.New(gwclient, index, cfg.MachineAuthAPIKey, evts)
provider := searchprovider.New(gwclient, index, cfg.MachineAuthAPIKey, evts, logger)
return &Service{
id: cfg.GRPC.Namespace + "." + cfg.Service.Name,
@@ -99,3 +99,8 @@ func (s Service) Search(ctx context.Context, in *searchsvc.SearchRequest, out *s
out.NextPageToken = res.NextPageToken
return nil
}
func (s Service) IndexSpace(ctx context.Context, in *searchsvc.IndexSpaceRequest, out *searchsvc.IndexSpaceResponse) error {
_, err := s.provider.IndexSpace(ctx, in)
return err
}