Merge pull request #3207 from aduffeck/concurrent-reindex

Reindex spaces concurrently

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"`
 

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,
 	}
 }
 

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