diff --git a/services/search/pkg/config/config.go b/services/search/pkg/config/config.go index 128061b76f..a5ce9abdc4 100644 --- a/services/search/pkg/config/config.go +++ b/services/search/pkg/config/config.go @@ -28,6 +28,7 @@ type Config struct { Extractor Extractor `yaml:"extractor"` ContentExtractionSizeLimit uint64 `yaml:"content_extraction_size_limit" env:"SEARCH_CONTENT_EXTRACTION_SIZE_LIMIT" desc:"Maximum file size in bytes that is allowed for content extraction." introductionVersion:"1.0.0"` BatchSize int `yaml:"batch_size" env:"SEARCH_BATCH_SIZE" desc:"The number of documents to process in a single batch. Defaults to 500." introductionVersion:"1.0.0"` + ReindexMaxConcurrency int `yaml:"reindex_concurrency" env:"SEARCH_REINDEX_MAX_CONCURRENCY" desc:"The maximum number of spaces that are reindexed concurrently when reindexing all spaces." introductionVersion:"%%NEXT%%"` ServiceAccount ServiceAccount `yaml:"service_account"` diff --git a/services/search/pkg/config/defaults/defaultconfig.go b/services/search/pkg/config/defaults/defaultconfig.go index 3e504350fa..7145dfdc70 100644 --- a/services/search/pkg/config/defaults/defaultconfig.go +++ b/services/search/pkg/config/defaults/defaultconfig.go @@ -65,6 +65,7 @@ func DefaultConfig() *config.Config { }, ContentExtractionSizeLimit: 20 * 1024 * 1024, // Limit content extraction to <20MB files by default BatchSize: 50, + ReindexMaxConcurrency: 3, } } diff --git a/services/search/pkg/service/grpc/v0/service.go b/services/search/pkg/service/grpc/v0/service.go index dddb64d4b6..f382159c31 100644 --- a/services/search/pkg/service/grpc/v0/service.go +++ b/services/search/pkg/service/grpc/v0/service.go @@ -19,6 +19,7 @@ import ( "github.com/opencloud-eu/reva/v2/pkg/utils" merrors "go-micro.dev/v4/errors" "go-micro.dev/v4/metadata" + "golang.org/x/sync/errgroup" grpcmetadata "google.golang.org/grpc/metadata" "github.com/opencloud-eu/opencloud/pkg/log" @@ -144,13 +145,28 @@ func (s Service) IndexSpace(_ context.Context, in *searchsvc.IndexSpaceRequest, return errors.New(resp.GetStatus().GetMessage()) } + // Index all spaces concurrently, limited to a configurable number of spaces + // being reindexed at the same time. + concurrency := max(s.cfg.ReindexMaxConcurrency, 1) + var g errgroup.Group + g.SetLimit(concurrency) + for _, space := range resp.GetStorageSpaces() { - if err := s.searcher.IndexSpace(space.GetId(), in.GetForceReindex()); err != nil { - return err - } + g.Go(func() error { + s.log.Info().Str("space_id", space.GetId().GetOpaqueId()).Msg("indexing space") + + err := s.searcher.IndexSpace(space.GetId(), in.GetForceReindex()) + if err != nil { + s.log.Error().Err(err).Str("space_id", space.GetId().GetOpaqueId()).Msg("failed to index space") + } else { + s.log.Info().Str("space_id", space.GetId().GetOpaqueId()).Msg("finished indexing space") + } + + return nil + }) } - return nil + return g.Wait() } // FromCache pulls a search result from cache