From 8c404440ef37ee33524c359f8b4b0ca2990ea6d3 Mon Sep 17 00:00:00 2001 From: Till Faelligen <2353100+S7evinK@users.noreply.github.com> Date: Wed, 15 Mar 2023 13:02:15 +0100 Subject: [PATCH] Remove fulltext from BaseDendrite --- internal/fulltext/bleve.go | 8 ++++++-- internal/fulltext/bleve_test.go | 29 +++++++++++++++-------------- setup/base/base.go | 17 ----------------- syncapi/syncapi.go | 16 +++++++++++++--- 4 files changed, 34 insertions(+), 36 deletions(-) diff --git a/internal/fulltext/bleve.go b/internal/fulltext/bleve.go index 7187861dd..65d320e79 100644 --- a/internal/fulltext/bleve.go +++ b/internal/fulltext/bleve.go @@ -18,10 +18,10 @@ package fulltext import ( + "context" "strings" "github.com/blevesearch/bleve/v2" - // side effect imports to allow all possible languages _ "github.com/blevesearch/bleve/v2/analysis/lang/ar" _ "github.com/blevesearch/bleve/v2/analysis/lang/cjk" @@ -77,12 +77,16 @@ func (i *IndexElement) SetContentType(v string) { } // New opens a new/existing fulltext index -func New(cfg config.Fulltext) (fts *Search, err error) { +func New(ctx context.Context, cfg config.Fulltext) (fts *Search, err error) { fts = &Search{} fts.FulltextIndex, err = openIndex(cfg) if err != nil { return nil, err } + go func() { + <-ctx.Done() + _ = fts.Close() + }() return fts, nil } diff --git a/internal/fulltext/bleve_test.go b/internal/fulltext/bleve_test.go index d16397a45..6ce691454 100644 --- a/internal/fulltext/bleve_test.go +++ b/internal/fulltext/bleve_test.go @@ -18,6 +18,7 @@ import ( "reflect" "testing" + "github.com/matrix-org/dendrite/setup/process" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" @@ -25,7 +26,7 @@ import ( "github.com/matrix-org/dendrite/setup/config" ) -func mustOpenIndex(t *testing.T, tempDir string) *fulltext.Search { +func mustOpenIndex(t *testing.T, tempDir string) (*fulltext.Search, *process.ProcessContext) { t.Helper() cfg := config.Fulltext{ Enabled: true, @@ -36,11 +37,12 @@ func mustOpenIndex(t *testing.T, tempDir string) *fulltext.Search { cfg.IndexPath = config.Path(tempDir) cfg.InMemory = false } - fts, err := fulltext.New(cfg) + ctx := process.NewProcessContext() + fts, err := fulltext.New(ctx.Context(), cfg) if err != nil { t.Fatal("failed to open fulltext index:", err) } - return fts + return fts, ctx } func mustAddTestData(t *testing.T, fts *fulltext.Search, firstStreamPos int64) (eventIDs, roomIDs []string) { @@ -93,19 +95,17 @@ func mustAddTestData(t *testing.T, fts *fulltext.Search, firstStreamPos int64) ( func TestOpen(t *testing.T) { dataDir := t.TempDir() - fts := mustOpenIndex(t, dataDir) - if err := fts.Close(); err != nil { - t.Fatal("unable to close fulltext index", err) - } + _, ctx := mustOpenIndex(t, dataDir) + ctx.ShutdownDendrite() // open existing index - fts = mustOpenIndex(t, dataDir) - defer fts.Close() + _, ctx = mustOpenIndex(t, dataDir) + ctx.ShutdownDendrite() } func TestIndex(t *testing.T) { - fts := mustOpenIndex(t, "") - defer fts.Close() + fts, ctx := mustOpenIndex(t, "") + defer ctx.ShutdownDendrite() // add some data var streamPos int64 = 1 @@ -128,8 +128,8 @@ func TestIndex(t *testing.T) { } func TestDelete(t *testing.T) { - fts := mustOpenIndex(t, "") - defer fts.Close() + fts, ctx := mustOpenIndex(t, "") + defer ctx.ShutdownDendrite() eventIDs, roomIDs := mustAddTestData(t, fts, 0) res1, err := fts.Search("lorem", roomIDs[:1], nil, 50, 0, false) if err != nil { @@ -224,7 +224,8 @@ func TestSearch(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - f := mustOpenIndex(t, "") + f, ctx := mustOpenIndex(t, "") + defer ctx.ShutdownDendrite() eventIDs, roomIDs := mustAddTestData(t, f, 0) var searchRooms []string for _, x := range tt.args.roomIndex { diff --git a/setup/base/base.go b/setup/base/base.go index dfe48ff3c..572bfd403 100644 --- a/setup/base/base.go +++ b/setup/base/base.go @@ -42,7 +42,6 @@ import ( "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/caching" - "github.com/matrix-org/dendrite/internal/fulltext" "github.com/matrix-org/dendrite/internal/httputil" "github.com/matrix-org/dendrite/internal/pushgateway" "github.com/matrix-org/dendrite/internal/sqlutil" @@ -83,7 +82,6 @@ type BaseDendrite struct { Database *sql.DB DatabaseWriter sqlutil.Writer EnableMetrics bool - Fulltext *fulltext.Search startupLock sync.Mutex } @@ -130,14 +128,6 @@ func NewBaseDendrite(cfg *config.Dendrite, options ...BaseDendriteOptions) *Base logrus.WithError(err).Panicf("failed to start opentracing") } - var fts *fulltext.Search - if cfg.SyncAPI.Fulltext.Enabled { - fts, err = fulltext.New(cfg.SyncAPI.Fulltext) - if err != nil { - logrus.WithError(err).Panicf("failed to create full text") - } - } - if cfg.Global.Sentry.Enabled { logrus.Info("Setting up Sentry for debugging...") err = sentry.Init(sentry.ClientOptions{ @@ -212,7 +202,6 @@ func NewBaseDendrite(cfg *config.Dendrite, options ...BaseDendriteOptions) *Base Database: db, // set if monolith with global connection pool only DatabaseWriter: writer, // set if monolith with global connection pool only EnableMetrics: enableMetrics, - Fulltext: fts, } } @@ -493,12 +482,6 @@ func (b *BaseDendrite) WaitForShutdown() { logrus.Warnf("failed to flush all Sentry events!") } } - if b.Fulltext != nil { - err := b.Fulltext.Close() - if err != nil { - logrus.Warnf("failed to close full text search!") - } - } logrus.Warnf("Dendrite is exiting now") } diff --git a/syncapi/syncapi.go b/syncapi/syncapi.go index 153f7af5d..543925016 100644 --- a/syncapi/syncapi.go +++ b/syncapi/syncapi.go @@ -17,6 +17,7 @@ package syncapi import ( "context" + "github.com/matrix-org/dendrite/internal/fulltext" "github.com/sirupsen/logrus" "github.com/matrix-org/dendrite/internal/caching" @@ -59,6 +60,15 @@ func AddPublicRoutes( logrus.WithError(err).Panicf("failed to load notifier ") } + var fts *fulltext.Search + if cfg.Fulltext.Enabled { + fts, err = fulltext.New(base.ProcessContext.Context(), cfg.Fulltext) + if err != nil { + logrus.WithError(err).Panicf("failed to create full text") + } + base.ProcessContext.ComponentStarted() + } + federationPresenceProducer := &producers.FederationAPIPresenceProducer{ Topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent), JetStream: js, @@ -86,7 +96,7 @@ func AddPublicRoutes( roomConsumer := consumers.NewOutputRoomEventConsumer( base.ProcessContext, cfg, js, syncDB, notifier, streams.PDUStreamProvider, - streams.InviteStreamProvider, rsAPI, base.Fulltext, + streams.InviteStreamProvider, rsAPI, fts, ) if err = roomConsumer.Start(); err != nil { logrus.WithError(err).Panicf("failed to start room server consumer") @@ -94,7 +104,7 @@ func AddPublicRoutes( clientConsumer := consumers.NewOutputClientDataConsumer( base.ProcessContext, cfg, js, natsClient, syncDB, notifier, - streams.AccountDataStreamProvider, base.Fulltext, + streams.AccountDataStreamProvider, fts, ) if err = clientConsumer.Start(); err != nil { logrus.WithError(err).Panicf("failed to start client data consumer") @@ -130,6 +140,6 @@ func AddPublicRoutes( routing.Setup( base.PublicClientAPIMux, requestPool, syncDB, userAPI, - rsAPI, cfg, base.Caches, base.Fulltext, + rsAPI, cfg, base.Caches, fts, ) }