From 4f71ff85526ef0143bc281b8d11ed17ac4f239bc Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Thu, 16 Jul 2026 21:52:59 +0200 Subject: [PATCH 1/3] feat(search): schema-revision index migration for bleve and OpenSearch Migrate the search index in place instead of a full Tika re-crawl when the shared search.SchemaRevision is bumped. bleve (single instance): rebuild from the existing documents into a temp index and atomically swap, with crash recovery; auto-migrate on startup (opt-out via SEARCH_ENGINE_BLEVE_AUTO_MIGRATE). OpenSearch (scaled): versioned indices -v (no alias), filled from the newest older index via a create-only scroll reindex - idempotent, never overwrites, safe to run live. 'opencloud search index migrate' fills the current-revision index; 'opencloud search index prune' drops older ones. Per-document fixups are gated on the source schema version. Trash preserved, geopoint siblings synthesized, all fields round-tripped. --- services/search/pkg/bleve/index.go | 6 + services/search/pkg/bleve/migrate.go | 262 ++++++++++++++ .../search/pkg/bleve/migrate_fields_test.go | 89 +++++ services/search/pkg/bleve/migrate_test.go | 164 +++++++++ services/search/pkg/command/index.go | 2 + services/search/pkg/command/migrate.go | 103 ++++++ services/search/pkg/command/migrate_test.go | 49 +++ services/search/pkg/command/server.go | 53 +-- .../pkg/config/defaults/defaultconfig.go | 3 +- services/search/pkg/config/engine.go | 3 +- services/search/pkg/opensearch/client.go | 53 +++ services/search/pkg/opensearch/index_test.go | 1 + services/search/pkg/opensearch/migrate.go | 319 ++++++++++++++++++ .../pkg/opensearch/migrate_fields_test.go | 79 +++++ .../search/pkg/opensearch/migrate_test.go | 249 ++++++++++++++ services/search/pkg/search/search.go | 6 + 16 files changed, 1400 insertions(+), 41 deletions(-) create mode 100644 services/search/pkg/bleve/migrate.go create mode 100644 services/search/pkg/bleve/migrate_fields_test.go create mode 100644 services/search/pkg/bleve/migrate_test.go create mode 100644 services/search/pkg/command/migrate.go create mode 100644 services/search/pkg/command/migrate_test.go create mode 100644 services/search/pkg/opensearch/client.go create mode 100644 services/search/pkg/opensearch/migrate.go create mode 100644 services/search/pkg/opensearch/migrate_fields_test.go create mode 100644 services/search/pkg/opensearch/migrate_test.go diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index 097fea9560..ed093e8078 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -46,6 +46,12 @@ func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { if err != nil { return nil, searchmapping.Classification{}, err } + // stamp the current revision so a fresh index is not seen as outdated + // and needlessly migrated on the next start + if err := writeRevision(index); err != nil { + _ = index.Close() + return nil, searchmapping.Classification{}, err + } return index, searchmapping.Classification{Verdict: searchmapping.VerdictEqual}, nil } diff --git a/services/search/pkg/bleve/migrate.go b/services/search/pkg/bleve/migrate.go new file mode 100644 index 0000000000..760844f8dd --- /dev/null +++ b/services/search/pkg/bleve/migrate.go @@ -0,0 +1,262 @@ +package bleve + +import ( + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "strconv" + + "github.com/blevesearch/bleve/v2" + + "github.com/opencloud-eu/opencloud/pkg/log" + searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +const ( + revisionKey = "_schema_revision" + migratingSuffix = ".migrating" + backupSuffix = ".bak" + migrateBatch = 100 + migrateLogInterval = 50000 // how often buildReplacement logs progress, in documents +) + +// OpenOrMigrate opens the index and, when autoMigrate is set and the stored +// schema is incompatible, rebuilds it in place first. The migration triggers on +// a schema revision bump, not on a mapping diff, so a breaking change without a +// matching bump still refuses to start. Returns the number of documents migrated +// (0 for none). Opening first means a fresh install and the common up-to-date +// case open the index only once; the migration path is taken only on a mismatch. +func OpenOrMigrate(root string, autoMigrate bool, logger log.Logger) (bleve.Index, searchmapping.Classification, int, error) { + idx, classification, openErr := NewIndex(root) + if openErr == nil || !autoMigrate || !errors.Is(openErr, searchmapping.ErrManualActionRequired) { + // opened cleanly, or auto-migrate is off, or the failure is not a + // schema mismatch we could migrate away + return idx, classification, 0, openErr + } + + // the stored schema is incompatible; rebuild only if the revision was bumped, + // otherwise surface the original error (a schema change without a bump) + migrated, err := MigrateIndex(root, logger) + if err != nil { + return nil, classification, 0, err + } + if migrated == 0 { + return nil, classification, 0, openErr + } + idx, classification, err = NewIndex(root) + return idx, classification, migrated, err +} + +// MigrateIndex rebuilds the index from the documents it already holds when the +// stored schema revision is older than search.SchemaRevision, else a no-op. It +// takes the exclusive bolt lock, so the search service must be stopped. +func MigrateIndex(root string, logger log.Logger) (int, error) { + dest := filepath.Join(root, "bleve") + tmp := dest + migratingSuffix + bak := dest + backupSuffix + + if err := recoverMigration(root); err != nil { + return 0, err + } + + old, err := bleve.OpenUsing(dest, openRuntimeConfig) + if errors.Is(err, bleve.ErrorIndexPathDoesNotExist) { + return 0, nil // nothing to migrate yet; NewIndex creates a fresh index + } + if err != nil { + return 0, fmt.Errorf("open index (is the search service still running?): %w", err) + } + + stored, err := readRevision(old) + if err != nil { + _ = old.Close() + return 0, err + } + if stored >= search.SchemaRevision { + _ = old.Close() + return 0, nil + } + + total, err := old.DocCount() + if err != nil { + _ = old.Close() + return 0, err + } + logger.Info().Uint64("documents", total).Msg("starting search index migration") + + copied, err := buildReplacement(old, tmp, total, logger) + if err != nil { + _ = old.Close() + _ = os.RemoveAll(tmp) + return 0, err + } + + // verify before the swap, so a failure leaves the original untouched + if err := verifyMigrated(tmp, total); err != nil { + _ = old.Close() + _ = os.RemoveAll(tmp) + return 0, err + } + + if err := swap(old, dest, tmp, bak); err != nil { + return 0, err + } + logger.Info().Int("documents", copied).Msg("search index migration complete") + return copied, nil +} + +// buildReplacement copies every document from src into a fresh index at tmp via +// the same Deserialize -> PrepareForIndex round-trip the Move/Delete/Restore +// paths use, and stamps the current revision. +func buildReplacement(src bleve.Index, tmp string, total uint64, logger log.Logger) (int, error) { + if err := os.RemoveAll(tmp); err != nil { + return 0, err + } + m, err := NewMapping() + if err != nil { + return 0, err + } + dst, err := bleve.New(tmp, m) + if err != nil { + return 0, err + } + + batch, err := NewBatch(dst, migrateBatch) + if err != nil { + _ = dst.Close() + return 0, err + } + + var copied int + nextLog := migrateLogInterval + var after []string + for { + req := bleve.NewSearchRequest(bleve.NewMatchAllQuery()) + req.Size = migrateBatch + req.Fields = []string{"*"} + req.SortBy([]string{"_id"}) // total order, no skips or dupes across pages + req.SearchAfter = after + res, err := src.Search(req) + if err != nil { + _ = dst.Close() + return 0, err + } + if len(res.Hits) == 0 { + break + } + for _, hit := range res.Hits { + r := legacyHitToResource(hit.Fields) + if err := batch.Upsert(hit.ID, r); err != nil { // hit.ID: the bleve key is authoritative + _ = dst.Close() + return 0, err + } + copied++ + } + if copied >= nextLog { + logger.Info().Msgf("migrating search index: %d of %d documents", copied, total) + nextLog += migrateLogInterval + } + after = []string{res.Hits[len(res.Hits)-1].ID} + } + if err := batch.Push(); err != nil { + _ = dst.Close() + return 0, err + } + if err := writeRevision(dst); err != nil { + _ = dst.Close() + return 0, err + } + return copied, dst.Close() +} + +// legacyHitToResource is the fixup hook before re-indexing, symmetric to the +// opensearch side. bleve Deserialize is fail-soft, so nothing is stripped here. +func legacyHitToResource(fields map[string]any) search.Resource { + return *searchmapping.Deserialize[search.Resource](fields) +} + +// verifyMigrated checks the rebuilt index holds want documents before the swap +// replaces the original. The count catches a pagination drop or batch-push loss; +// the mapping needs no check, it is built deterministically from NewMapping(). +func verifyMigrated(path string, want uint64) error { + idx, err := bleve.OpenUsing(path, openRuntimeConfig) + if err != nil { + return err + } + defer func() { _ = idx.Close() }() + + got, err := idx.DocCount() + if err != nil { + return err + } + if got != want { + return fmt.Errorf("migrated index holds %d documents, expected %d", got, want) + } + return nil +} + +// swap replaces the live index with the rebuilt one. old still holds the lock +// and is closed here right before the rename. +func swap(old bleve.Index, dest, tmp, bak string) error { + if err := old.Close(); err != nil { + return err + } + if err := os.Rename(dest, bak); err != nil { + return err + } + if err := os.Rename(tmp, dest); err != nil { + _ = os.Rename(bak, dest) // best effort roll back + return err + } + return os.RemoveAll(bak) +} + +// recoverMigration cleans up a crashed MigrateIndex and must run before the +// index is opened, else NewIndex would recreate an empty index mid-swap. Rolls +// back, never forward: a truncated tmp is a valid index, completeness is not +// observable. +func recoverMigration(root string) error { + dest := filepath.Join(root, "bleve") + tmp := dest + migratingSuffix + bak := dest + backupSuffix + + if _, err := os.Stat(dest); err == nil { + return errors.Join(os.RemoveAll(tmp), os.RemoveAll(bak)) // index present: tmp/bak are leftovers + } else if !errors.Is(err, fs.ErrNotExist) { + return err + } + + if _, err := os.Stat(bak); err == nil { + // crashed mid-swap: roll back + if err := os.RemoveAll(tmp); err != nil { + return err + } + return os.Rename(bak, dest) + } else if !errors.Is(err, fs.ErrNotExist) { + return err + } + + return os.RemoveAll(tmp) // fresh install, or index removed by hand +} + +// writeRevision stamps the current revision so a fresh or migrated index is not +// seen as outdated. +func writeRevision(index bleve.Index) error { + return index.SetInternal([]byte(revisionKey), []byte(strconv.Itoa(search.SchemaRevision))) +} + +// readRevision returns the stored schema revision, or 0 for an index that +// predates the marker. +func readRevision(index bleve.Index) (int, error) { + raw, err := index.GetInternal([]byte(revisionKey)) + if err != nil { + return 0, err + } + if len(raw) == 0 { + return 0, nil + } + return strconv.Atoi(string(raw)) +} diff --git a/services/search/pkg/bleve/migrate_fields_test.go b/services/search/pkg/bleve/migrate_fields_test.go new file mode 100644 index 0000000000..b570fa0424 --- /dev/null +++ b/services/search/pkg/bleve/migrate_fields_test.go @@ -0,0 +1,89 @@ +package bleve_test + +import ( + "path/filepath" + "sort" + "testing" + "time" + + bleveSearch "github.com/blevesearch/bleve/v2" + libregraph "github.com/opencloud-eu/libre-graph-api-go" + "github.com/stretchr/testify/require" + + "github.com/opencloud-eu/opencloud/pkg/conversions" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" + "github.com/opencloud-eu/opencloud/services/search/pkg/content" + searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +// fullyPopulated exercises every field, including the facets and slices the +// count-only test skipped. +func fullyPopulated() search.Resource { + return search.Resource{ + ID: "1$2!3", + RootID: "1$2!2", + ParentID: "1$2!2", + Path: "./a/b/song.mp3", + Type: 2, + Deleted: true, + Hidden: true, + Document: content.Document{ + Title: "the title", + Name: "song.mp3", + Content: "some extracted content", + Size: 123456, + Mtime: conversions.ToPointer(time.Date(2026, 3, 15, 12, 0, 0, 0, time.UTC)), + MimeType: "audio/mpeg", + Tags: []string{"alpha", "beta", "gamma"}, + Favorites: []string{"user-a", "user-b"}, + Audio: &libregraph.Audio{ + Album: libregraph.PtrString("the album"), + Artist: libregraph.PtrString("the artist"), + Track: libregraph.PtrInt32(7), + Year: libregraph.PtrInt32(1998), + HasDrm: libregraph.PtrBool(false), + }, + Image: &libregraph.Image{Width: libregraph.PtrInt32(1920), Height: libregraph.PtrInt32(1080)}, + Photo: &libregraph.Photo{CameraMake: libregraph.PtrString("Canon"), Iso: libregraph.PtrInt32(400), FNumber: libregraph.PtrFloat64(2.8)}, + Location: &libregraph.GeoCoordinates{ + Longitude: libregraph.PtrFloat64(11.103870357204285), + Latitude: libregraph.PtrFloat64(49.48675890884328), + Altitude: libregraph.PtrFloat64(300.0), + }, + }, + } +} + +func TestMigratePreservesAllFields(t *testing.T) { + root := t.TempDir() + want := fullyPopulated() + + old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t)) + require.NoError(t, err) + require.NoError(t, old.Index(want.ID, want)) + require.NoError(t, old.Close()) + + _, err = bleve.MigrateIndex(root, log.NopLogger()) + require.NoError(t, err) + + idx, _, err := bleve.NewIndex(root) + require.NoError(t, err) + defer func() { _ = idx.Close() }() + + // read the migrated document back through the production reconstruction path + req := bleveSearch.NewSearchRequest(bleveSearch.NewMatchAllQuery()) + req.Fields = []string{"*"} + res, err := idx.Search(req) + require.NoError(t, err) + require.Len(t, res.Hits, 1) + got := searchmapping.Deserialize[search.Resource](res.Hits[0].Fields) + + require.NotNil(t, got.Mtime, "Mtime") + require.True(t, want.Mtime.Equal(*got.Mtime), "Mtime: want %v got %v", want.Mtime, got.Mtime) + got.Mtime = want.Mtime // normalize time.Time location/monotonic + sort.Strings(got.Tags) // bleve may return multi-value fields reordered + sort.Strings(got.Favorites) // + require.Equal(t, want, *got) +} diff --git a/services/search/pkg/bleve/migrate_test.go b/services/search/pkg/bleve/migrate_test.go new file mode 100644 index 0000000000..0c1ab3489a --- /dev/null +++ b/services/search/pkg/bleve/migrate_test.go @@ -0,0 +1,164 @@ +package bleve_test + +import ( + "path/filepath" + "testing" + "time" + + bleveSearch "github.com/blevesearch/bleve/v2" + bleveMapping "github.com/blevesearch/bleve/v2/mapping" + "github.com/blevesearch/bleve/v2/search/query" + libregraph "github.com/opencloud-eu/libre-graph-api-go" + "github.com/stretchr/testify/require" + + "github.com/opencloud-eu/opencloud/pkg/conversions" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" + "github.com/opencloud-eu/opencloud/services/search/pkg/content" + searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +// oldMainMapping reconstructs the pre-refactor bleve mapping: only Name, Tags, +// Favorites and Content explicit, everything else dynamic. Reuses NewMapping's +// registered analyzers so indexing works. +func oldMainMapping(t *testing.T) bleveMapping.IndexMapping { + m, err := bleve.NewMapping() + require.NoError(t, err) + impl := m.(*bleveMapping.IndexMappingImpl) + + doc := bleveSearch.NewDocumentMapping() + name := bleveSearch.NewTextFieldMapping() + name.Analyzer = "lowercaseKeyword" + lc := bleveSearch.NewTextFieldMapping() + lc.Analyzer = "lowercaseKeyword" + lc.IncludeInAll = false + ft := bleveSearch.NewTextFieldMapping() + ft.Analyzer = "fulltext" + ft.IncludeInAll = false + doc.AddFieldMappingsAt("Name", name) + doc.AddFieldMappingsAt("Tags", lc) + doc.AddFieldMappingsAt("Favorites", lc) + doc.AddFieldMappingsAt("Content", ft) + impl.DefaultMapping = doc + return impl +} + +func fullDoc(id, name string, deleted bool) search.Resource { + lon, lat, alt := 11.103870357204285, 49.48675890884328, 300.0 + return search.Resource{ + ID: id, + RootID: "1$1!1", + ParentID: "1$1!1", + Path: "./" + name, + Type: 2, + Deleted: deleted, + Document: content.Document{ + Name: name, + Title: "title", + Content: "hello world", + Size: 42, + Mtime: conversions.ToPointer(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)), + MimeType: "image/jpeg", + Tags: []string{"tag1"}, + Location: &libregraph.GeoCoordinates{Longitude: &lon, Latitude: &lat, Altitude: &alt}, + }, + } +} + +func TestOpenOrMigrate(t *testing.T) { + buildOld := func(t *testing.T) string { + root := t.TempDir() + old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t)) + require.NoError(t, err) + require.NoError(t, old.Index("1$1!1", fullDoc("1$1!1", "a.jpg", false))) + require.NoError(t, old.Close()) + return root + } + + t.Run("migrates an outdated revision when auto-migrate is on", func(t *testing.T) { + // the old index carries no revision (0), the code revision is higher + idx, classification, migrated, err := bleve.OpenOrMigrate(buildOld(t), true, log.NopLogger()) + require.NoError(t, err) + require.NotNil(t, idx) + defer func() { _ = idx.Close() }() + require.Equal(t, searchmapping.VerdictEqual, classification.Verdict) + require.Equal(t, 1, migrated) + }) + + t.Run("refuses to start when auto-migrate is off", func(t *testing.T) { + idx, _, migrated, err := bleve.OpenOrMigrate(buildOld(t), false, log.NopLogger()) + require.ErrorIs(t, err, searchmapping.ErrManualActionRequired) + require.Nil(t, idx) + require.Equal(t, 0, migrated) + }) + + t.Run("does not migrate a fresh index", func(t *testing.T) { + root := t.TempDir() + // NewIndex creates a fresh index and stamps the current revision + idx, _, err := bleve.NewIndex(root) + require.NoError(t, err) + require.NoError(t, idx.Close()) + + // a fresh index is already at the current revision, so no migration runs + idx2, classification, migrated, err := bleve.OpenOrMigrate(root, true, log.NopLogger()) + require.NoError(t, err) + defer func() { _ = idx2.Close() }() + require.Equal(t, searchmapping.VerdictEqual, classification.Verdict) + require.Equal(t, 0, migrated) + }) +} + +func TestMigrateIndex(t *testing.T) { + root := t.TempDir() + + // 1) an index left behind by the old release: main-style mapping, no revision + old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t)) + require.NoError(t, err) + docs := []search.Resource{ + fullDoc("1$1!1", "photo.jpg", false), + fullDoc("1$1!2", "song.mp3", false), + fullDoc("1$1!3", "trashed.txt", true), // trash must survive the migration + } + for _, r := range docs { + require.NoError(t, old.Index(r.ID, r)) + } + require.NoError(t, old.Close()) + + // 2) sanity: the old index is incompatible, the service would refuse to start + _, _, err = bleve.NewIndex(root) + require.ErrorIs(t, err, searchmapping.ErrManualActionRequired) + + // 3) migrate + n, err := bleve.MigrateIndex(root, log.NopLogger()) + require.NoError(t, err) + require.Equal(t, 3, n) + + // 4) the migrated index classifies as equal and holds every document + migrated, classification, err := bleve.NewIndex(root) + require.NoError(t, err) + require.Equal(t, searchmapping.VerdictEqual, classification.Verdict) + + count, err := migrated.DocCount() + require.NoError(t, err) + require.Equal(t, uint64(3), count) + + // 4a) the trashed document survived (a rescan would have dropped it) + trashed, err := migrated.Document("1$1!3") + require.NoError(t, err) + require.NotNil(t, trashed) + + // 5) location_geopoint was synthesized during migration (absent in the old index) + near := query.NewGeoDistanceQuery(11.103870357204285, 49.48675890884328, "1km") + near.SetField("location" + searchmapping.GeopointSuffix) + res, err := migrated.Search(bleveSearch.NewSearchRequest(near)) + require.NoError(t, err) + require.Equal(t, uint64(3), res.Total, "all three docs match a geo-distance query on the new sibling field") + + require.NoError(t, migrated.Close()) // release the lock before the second run + + // 6) idempotent: a second run is a no-op because the revision is now current + n, err = bleve.MigrateIndex(root, log.NopLogger()) + require.NoError(t, err) + require.Equal(t, 0, n) +} diff --git a/services/search/pkg/command/index.go b/services/search/pkg/command/index.go index 3951ee6fd3..e459e6e0f7 100644 --- a/services/search/pkg/command/index.go +++ b/services/search/pkg/command/index.go @@ -95,5 +95,7 @@ func Index(cfg *config.Config) *cobra.Command { "disable TLS for the gRPC connection.", ) + indexCmd.AddCommand(Migrate(cfg), Prune(cfg)) + return indexCmd } diff --git a/services/search/pkg/command/migrate.go b/services/search/pkg/command/migrate.go new file mode 100644 index 0000000000..c389e56257 --- /dev/null +++ b/services/search/pkg/command/migrate.go @@ -0,0 +1,103 @@ +package command + +import ( + "fmt" + + "github.com/spf13/cobra" + + "github.com/opencloud-eu/opencloud/pkg/config/configlog" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" + "github.com/opencloud-eu/opencloud/services/search/pkg/config" + "github.com/opencloud-eu/opencloud/services/search/pkg/config/parser" + "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" +) + +// Migrate rebuilds the index for the current schema revision from the documents +// it holds, no re-crawl. Stop the service first. Idempotent. +func Migrate(cfg *config.Config) *cobra.Command { + return &cobra.Command{ + Use: "migrate", + Short: "rebuild the search index for the current schema revision", + Long: "Rebuild the search index from the documents it already holds when the schema revision changed, instead of re-crawling storage. Stop the search service before running it. Idempotent: a no-op when the index is already current.", + PreRunE: func(cmd *cobra.Command, args []string) error { + return configlog.ReturnFatal(parser.ParseConfig(cfg)) + }, + RunE: func(cmd *cobra.Command, args []string) error { + logger := log.Configure(cfg.Service.Name, cfg.Commons, cfg.LogLevel) + + switch cfg.Engine.Type { + case "bleve": + n, err := bleve.MigrateIndex(cfg.Engine.Bleve.Datapath, logger) + if err != nil { + return err + } + reportMigration(logger, n) + case "open-search": + client, err := opensearch.NewClient(cfg.Engine.OpenSearch.Client) + if err != nil { + return err + } + n, err := opensearch.MigrateIndex(cmd.Context(), cfg.Engine.OpenSearch.ResourceIndex.Name, client, logger) + if err != nil { + return err + } + reportMigration(logger, n) + default: + return fmt.Errorf("unknown search engine: %s", cfg.Engine.Type) + } + return nil + }, + } +} + +// rescanRecommendation is appended after a rebuild/migration: the index holds +// the migrated documents, but a full crawl is the only guaranteed-clean state. +const rescanRecommendation = "re-indexing with 'opencloud search index --all-spaces --force-rescan' is still recommended for a guaranteed-clean index" + +func reportMigration(logger log.Logger, migrated int) { + if migrated == 0 { + logger.Info().Msg("the search index is already at the current schema revision, nothing to migrate") + return + } + logger.Info().Int("documents", migrated).Msg("the search index was migrated; " + rescanRecommendation) +} + +// Prune deletes the OpenSearch indices left by older schema revisions. Run it +// after a rollout has fully drained the old instances. bleve keeps one index in +// place, so there is nothing to prune there. +func Prune(cfg *config.Config) *cobra.Command { + return &cobra.Command{ + Use: "prune", + Short: "delete search indices older than the current schema revision", + Long: "Delete search indices left by older schema revisions, after a rollout has fully drained the old instances. OpenSearch only; refuses if the current-revision index does not exist yet.", + PreRunE: func(cmd *cobra.Command, args []string) error { + return configlog.ReturnFatal(parser.ParseConfig(cfg)) + }, + RunE: func(cmd *cobra.Command, args []string) error { + logger := log.Configure(cfg.Service.Name, cfg.Commons, cfg.LogLevel) + + switch cfg.Engine.Type { + case "bleve": + logger.Info().Msg("bleve keeps a single index in place, nothing to prune") + case "open-search": + client, err := opensearch.NewClient(cfg.Engine.OpenSearch.Client) + if err != nil { + return err + } + n, err := opensearch.PruneOldIndices(cmd.Context(), client, cfg.Engine.OpenSearch.ResourceIndex.Name, logger) + if err != nil { + return err + } + if n == 0 { + logger.Info().Msg("no old search indices to prune") + } else { + logger.Info().Int("indices", n).Msg("pruned old search indices") + } + default: + return fmt.Errorf("unknown search engine: %s", cfg.Engine.Type) + } + return nil + }, + } +} diff --git a/services/search/pkg/command/migrate_test.go b/services/search/pkg/command/migrate_test.go new file mode 100644 index 0000000000..d42c3b9d46 --- /dev/null +++ b/services/search/pkg/command/migrate_test.go @@ -0,0 +1,49 @@ +package command_test + +import ( + "context" + "path/filepath" + "testing" + + bleveSearch "github.com/blevesearch/bleve/v2" + "github.com/stretchr/testify/require" + + "github.com/opencloud-eu/opencloud/pkg/shared" + "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" + "github.com/opencloud-eu/opencloud/services/search/pkg/command" + "github.com/opencloud-eu/opencloud/services/search/pkg/config/defaults" + searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" +) + +func TestMigrateCommandBleve(t *testing.T) { + root := t.TempDir() + + // an index left behind by an older release: a plain dynamic mapping, which + // classifies as breaking against the code schema + old, err := bleveSearch.New(filepath.Join(root, "bleve"), bleveSearch.NewIndexMapping()) + require.NoError(t, err) + require.NoError(t, old.Index("doc1", map[string]any{"Name": "file.txt", "Mtime": "2026-01-01T00:00:00Z"})) + require.NoError(t, old.Close()) + + cfg := defaults.DefaultConfig() + defaults.EnsureDefaults(cfg) + cfg.Commons = &shared.Commons{Log: &shared.Log{}} // normally filled by the config pipeline + cfg.Engine.Type = "bleve" + cfg.Engine.Bleve.Datapath = root + + run := func() error { + cmd := command.Migrate(cfg) + cmd.SetContext(context.Background()) + return cmd.RunE(cmd, nil) // bypass PreRunE (parser requires service credentials) + } + + // first run migrates, the index then classifies equal + require.NoError(t, run()) + idx, classification, err := bleve.NewIndex(root) + require.NoError(t, err) + require.NoError(t, idx.Close()) + require.Equal(t, searchmapping.VerdictEqual, classification.Verdict) + + // second run is a no-op (idempotent), still succeeds + require.NoError(t, run()) +} diff --git a/services/search/pkg/command/server.go b/services/search/pkg/command/server.go index 5e9d8fee05..9723ae371b 100644 --- a/services/search/pkg/command/server.go +++ b/services/search/pkg/command/server.go @@ -2,10 +2,7 @@ package command import ( "context" - "crypto/tls" "fmt" - "net/http" - "os" "os/signal" "time" @@ -32,8 +29,6 @@ import ( "github.com/opencloud-eu/reva/v2/pkg/events/raw" "github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool" - opensearchgo "github.com/opensearch-project/opensearch-go/v4" - opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi" "github.com/spf13/cobra" ) @@ -73,7 +68,13 @@ func Server(cfg *config.Config) *cobra.Command { var eng search.Engine switch cfg.Engine.Type { case "bleve": - idx, classification, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath) + // with AutoMigrate a schema revision bump rebuilds the index in + // place instead of refusing to start. Safe for bleve because a + // single instance holds the datapath lock. + idx, classification, migrated, err := bleve.OpenOrMigrate(cfg.Engine.Bleve.Datapath, cfg.Engine.Bleve.AutoMigrate, logger) + if migrated > 0 { + logger.Info().Int("documents", migrated).Msgf("the bleve index at %s was rebuilt for the new schema revision; %s", cfg.Engine.Bleve.Datapath, rescanRecommendation) + } // warn before the error check: the new mapping may already be // persisted, then later startups classify equal and stay silent if classification.Verdict == searchmapping.VerdictAdditive { @@ -93,44 +94,18 @@ func Server(cfg *config.Config) *cobra.Command { eng = bleve.NewBackend(idx, bleveQuery.DefaultCreator, logger) case "open-search": - clientConfig := opensearchgo.Config{ - Addresses: cfg.Engine.OpenSearch.Client.Addresses, - Username: cfg.Engine.OpenSearch.Client.Username, - Password: cfg.Engine.OpenSearch.Client.Password, - Header: cfg.Engine.OpenSearch.Client.Header, - RetryOnStatus: cfg.Engine.OpenSearch.Client.RetryOnStatus, - DisableRetry: cfg.Engine.OpenSearch.Client.DisableRetry, - EnableRetryOnTimeout: cfg.Engine.OpenSearch.Client.EnableRetryOnTimeout, - MaxRetries: cfg.Engine.OpenSearch.Client.MaxRetries, - CompressRequestBody: cfg.Engine.OpenSearch.Client.CompressRequestBody, - DiscoverNodesOnStart: cfg.Engine.OpenSearch.Client.DiscoverNodesOnStart, - DiscoverNodesInterval: cfg.Engine.OpenSearch.Client.DiscoverNodesInterval, - EnableMetrics: cfg.Engine.OpenSearch.Client.EnableMetrics, - EnableDebugLogger: cfg.Engine.OpenSearch.Client.EnableDebugLogger, - Transport: &http.Transport{ - TLSClientConfig: &tls.Config{ - MinVersion: tls.VersionTLS12, - InsecureSkipVerify: cfg.Engine.OpenSearch.Client.Insecure, - }, - }, - } - - if cfg.Engine.OpenSearch.Client.CACert != "" { - certBytes, err := os.ReadFile(cfg.Engine.OpenSearch.Client.CACert) - if err != nil { - return fmt.Errorf("failed to read CA cert: %w", err) - } - clientConfig.CACert = certBytes - } - - client, err := opensearchgoAPI.NewClient(opensearchgoAPI.Config{Client: clientConfig}) + client, err := opensearch.NewClient(cfg.Engine.OpenSearch.Client) if err != nil { - return fmt.Errorf("failed to create OpenSearch client: %w", err) + return err } + // the revision is in the index name, so each instance uses its own + // index and rollouts never share one; a pre-rollout migrate builds it + indexName := opensearch.TargetIndex(cfg.Engine.OpenSearch.ResourceIndex.Name) + // a hung cluster must fail the start, not block it forever startupCtx, cancelStartup := context.WithTimeout(ctx, time.Minute) - openSearchBackend, err := opensearch.NewBackend(startupCtx, cfg.Engine.OpenSearch.ResourceIndex.Name, client, logger) + openSearchBackend, err := opensearch.NewBackend(startupCtx, indexName, client, logger) cancelStartup() if err != nil { return fmt.Errorf("failed to create OpenSearch backend: %w", err) diff --git a/services/search/pkg/config/defaults/defaultconfig.go b/services/search/pkg/config/defaults/defaultconfig.go index 3e504350fa..137d91c824 100644 --- a/services/search/pkg/config/defaults/defaultconfig.go +++ b/services/search/pkg/config/defaults/defaultconfig.go @@ -37,7 +37,8 @@ func DefaultConfig() *config.Config { Engine: config.Engine{ Type: "bleve", Bleve: config.EngineBleve{ - Datapath: filepath.Join(defaults.BaseDataPath(), "search"), + Datapath: filepath.Join(defaults.BaseDataPath(), "search"), + AutoMigrate: true, }, OpenSearch: config.EngineOpenSearch{ ResourceIndex: config.EngineOpenSearchResourceIndex{ diff --git a/services/search/pkg/config/engine.go b/services/search/pkg/config/engine.go index a22d2a92d1..e7e5109069 100644 --- a/services/search/pkg/config/engine.go +++ b/services/search/pkg/config/engine.go @@ -14,7 +14,8 @@ type Engine struct { // EngineBleve configures the bleve engine type EngineBleve struct { - Datapath string `yaml:"data_path" env:"SEARCH_ENGINE_BLEVE_DATA_PATH" desc:"The directory where the filesystem will store search data. If not defined, the root directory derives from $OC_BASE_DATA_PATH/search." introductionVersion:"1.0.0"` + Datapath string `yaml:"data_path" env:"SEARCH_ENGINE_BLEVE_DATA_PATH" desc:"The directory where the filesystem will store search data. If not defined, the root directory derives from $OC_BASE_DATA_PATH/search." introductionVersion:"1.0.0"` + AutoMigrate bool `yaml:"auto_migrate" env:"SEARCH_ENGINE_BLEVE_AUTO_MIGRATE" desc:"Rebuild the index on startup when the schema revision was bumped, instead of refusing to start. Safe for bleve because a single instance holds the data path lock. Defaults to 'true'; set to 'false' to migrate manually with 'opencloud search migrate'." introductionVersion:"%%NEXT%%"` } // EngineOpenSearch configures the OpenSearch engine diff --git a/services/search/pkg/opensearch/client.go b/services/search/pkg/opensearch/client.go new file mode 100644 index 0000000000..ec2e6d9b96 --- /dev/null +++ b/services/search/pkg/opensearch/client.go @@ -0,0 +1,53 @@ +package opensearch + +import ( + "crypto/tls" + "fmt" + "net/http" + "os" + + opensearchgo "github.com/opensearch-project/opensearch-go/v4" + opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi" + + "github.com/opencloud-eu/opencloud/services/search/pkg/config" +) + +// NewClient builds an OpenSearch API client from the engine client config. It is +// shared by the server startup and the migrate CLI so both connect identically. +func NewClient(cfg config.EngineOpenSearchClient) (*opensearchgoAPI.Client, error) { + clientConfig := opensearchgo.Config{ + Addresses: cfg.Addresses, + Username: cfg.Username, + Password: cfg.Password, + Header: cfg.Header, + RetryOnStatus: cfg.RetryOnStatus, + DisableRetry: cfg.DisableRetry, + EnableRetryOnTimeout: cfg.EnableRetryOnTimeout, + MaxRetries: cfg.MaxRetries, + CompressRequestBody: cfg.CompressRequestBody, + DiscoverNodesOnStart: cfg.DiscoverNodesOnStart, + DiscoverNodesInterval: cfg.DiscoverNodesInterval, + EnableMetrics: cfg.EnableMetrics, + EnableDebugLogger: cfg.EnableDebugLogger, + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{ + MinVersion: tls.VersionTLS12, + InsecureSkipVerify: cfg.Insecure, + }, + }, + } + + if cfg.CACert != "" { + certBytes, err := os.ReadFile(cfg.CACert) + if err != nil { + return nil, fmt.Errorf("failed to read CA cert: %w", err) + } + clientConfig.CACert = certBytes + } + + client, err := opensearchgoAPI.NewClient(opensearchgoAPI.Config{Client: clientConfig}) + if err != nil { + return nil, fmt.Errorf("failed to create OpenSearch client: %w", err) + } + return client, nil +} diff --git a/services/search/pkg/opensearch/index_test.go b/services/search/pkg/opensearch/index_test.go index d6b7204880..11b866d277 100644 --- a/services/search/pkg/opensearch/index_test.go +++ b/services/search/pkg/opensearch/index_test.go @@ -51,6 +51,7 @@ func TestIndexManager(t *testing.T) { require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) }) + t.Run("fails to create index if it already exists but is not up to date", func(t *testing.T) { indexManager := opensearch.IndexManagerLatest indexName := "opencloud-test-resource" diff --git a/services/search/pkg/opensearch/migrate.go b/services/search/pkg/opensearch/migrate.go new file mode 100644 index 0000000000..cd047f5167 --- /dev/null +++ b/services/search/pkg/opensearch/migrate.go @@ -0,0 +1,319 @@ +package opensearch + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "time" + + opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi" + + "github.com/opencloud-eu/opencloud/pkg/conversions" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +const ( + // migratePageSize is the scroll page size for reading the source index. + migratePageSize = 1000 + // migrateLogInterval is how often reindex logs progress, in documents. + migrateLogInterval = 50000 + // migrateScrollKeepAlive keeps the source snapshot alive across scroll pages. + migrateScrollKeepAlive = 5 * time.Minute +) + +// TargetIndex is the concrete index for the current schema revision. The +// revision is the index name suffix, so instances of different revisions use +// different indices and never share one. +func TargetIndex(base string) string { + return fmt.Sprintf("%s-v%d", base, search.SchemaRevision) +} + +// MigrateIndex fills the current-revision index (-v) from the newest +// older-revision index, in-process. Because the revision is in the index name, +// instances of different revisions never share an index and no alias flip is +// needed: run it as a pre-rollout step so new instances find their index ready. +// Reindexing uses create-only bulk ops, so a document already present is skipped, +// never overwritten: the run is idempotent and safe against a target an instance +// is already writing to. A no-op when there is no older index to migrate from +// (fresh install, or pruned). Returns the number of documents newly created. +func MigrateIndex(ctx context.Context, base string, client *opensearchgoAPI.Client, logger log.Logger) (int, error) { + target := TargetIndex(base) + + source, oldVersion, err := previousIndex(ctx, client, base) + if err != nil { + return 0, err + } + if source == "" { + return 0, nil // nothing older to migrate from + } + + exists, err := indexExists(ctx, client, target) + if err != nil { + return 0, err + } + if !exists { + mappingB, err := IndexManagerLatest.MarshalJSON() + if err != nil { + return 0, err + } + resp, err := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{ + Index: target, + Body: strings.NewReader(string(mappingB)), + }) + switch { + case err != nil: + return 0, fmt.Errorf("create %s: %w", target, err) + case !resp.Acknowledged: + return 0, fmt.Errorf("create %s not acknowledged", target) + } + } + logger.Info().Str("source", source).Str("target", target).Msg("starting search index migration") + + created, skipped, err := reindex(ctx, client, source, target, oldVersion, logger) + if err != nil { + return 0, err + } + logger.Info().Str("target", target).Int("created", created).Int("skipped", skipped).Msg("search index migration complete") + return created, nil +} + +// PruneOldIndices deletes every index older than the current schema revision: +// the versioned -v with k < SchemaRevision plus the legacy unversioned +// . Run it after a rollout has fully drained the old instances. It refuses +// when the current-revision index does not exist yet, so a premature prune can +// not wipe the only data. +func PruneOldIndices(ctx context.Context, client *opensearchgoAPI.Client, base string, logger log.Logger) (int, error) { + current, err := indexExists(ctx, client, TargetIndex(base)) + if err != nil { + return 0, err + } + if !current { + return 0, fmt.Errorf("current index %s does not exist; migrate before pruning", TargetIndex(base)) + } + + older, err := olderVersionedIndices(ctx, client, base) + if err != nil { + return 0, err + } + old := make([]string, 0, len(older)) + for _, name := range older { + old = append(old, name) + } + if legacy, err := indexExists(ctx, client, base); err != nil { + return 0, err + } else if legacy { + old = append(old, base) + } + if len(old) == 0 { + return 0, nil + } + + delResp, err := client.Indices.Delete(ctx, opensearchgoAPI.IndicesDeleteReq{Indices: old}) + if err != nil { + return 0, fmt.Errorf("delete old indices %v: %w", old, err) + } + if !delResp.Acknowledged { + return 0, fmt.Errorf("prune not acknowledged") + } + logger.Info().Strs("indices", old).Msg("pruned old search indices") + return len(old), nil +} + +// previousIndex returns the newest index to migrate from and its schema version: +// the highest-revision -v below the current SchemaRevision, or, if none +// exists yet, the legacy unversioned index (version 0) left by a +// pre-versioning release. name is "" when there is nothing older to migrate from. +func previousIndex(ctx context.Context, client *opensearchgoAPI.Client, base string) (name string, version int, err error) { + older, err := olderVersionedIndices(ctx, client, base) + if err != nil { + return "", 0, err + } + best := -1 + for k := range older { + if k > best { + best = k + } + } + if best >= 0 { + return older[best], best, nil + } + + // no versioned index yet: fall back to the legacy unversioned index (version 0) + legacy, err := indexExists(ctx, client, base) + if err != nil { + return "", 0, err + } + if legacy { + return base, 0, nil + } + return "", 0, nil +} + +// olderVersionedIndices returns the existing -v indices with k below the +// current SchemaRevision, keyed by version. +func olderVersionedIndices(ctx context.Context, client *opensearchgoAPI.Client, base string) (map[int]string, error) { + resp, err := client.Indices.Get(ctx, opensearchgoAPI.IndicesGetReq{Indices: []string{base + "-v*"}}) + if err != nil { + return nil, fmt.Errorf("list %s-v* indices: %w", base, err) + } + prefix := base + "-v" + out := map[int]string{} + for name := range resp.Indices { + suffix, ok := strings.CutPrefix(name, prefix) + if !ok { + continue + } + if k, err := strconv.Atoi(suffix); err == nil && k < search.SchemaRevision { + out[k] = name + } + } + return out, nil +} + +// indexExists reports whether the concrete index exists. A transient error is +// returned so callers do not treat "unknown" as "absent". +func indexExists(ctx context.Context, client *opensearchgoAPI.Client, index string) (bool, error) { + resp, err := client.Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{Indices: []string{index}}) + switch { + case resp != nil && resp.StatusCode == 404: + return false, nil + case err != nil: + return false, fmt.Errorf("check index %s: %w", index, err) + } + return true, nil +} + +// reindex copies source into target with create-only bulk ops (same +// PrepareForIndex transform as the live write path). It uses a scroll: a +// consistent snapshot of the source even while old instances keep writing to it. +// Returns the number of documents created and skipped (already present). +func reindex(ctx context.Context, client *opensearchgoAPI.Client, source, target string, oldVersion int, logger log.Logger) (int, int, error) { + first, err := client.Search(ctx, &opensearchgoAPI.SearchReq{ + Indices: []string{source}, + Params: opensearchgoAPI.SearchParams{Scroll: migrateScrollKeepAlive}, + // track_total_hits so Total.Value is exact, not capped at 10000: the + // completeness check below relies on it + Body: strings.NewReader(fmt.Sprintf(`{"size":%d,"track_total_hits":true,"query":{"match_all":{}}}`, migratePageSize)), + }) + if err != nil { + return 0, 0, err + } + scrollID := "" + if first.ScrollID != nil { + scrollID = *first.ScrollID + } + defer func() { + if scrollID != "" { + _, _ = client.Scroll.Delete(ctx, opensearchgoAPI.ScrollDeleteReq{ScrollIDs: []string{scrollID}}) + } + }() + + total := first.Hits.Total.Value + var created, skipped int + nextLog := migrateLogInterval + hits := first.Hits.Hits + for len(hits) > 0 { + c, s, err := bulkCreate(ctx, client, target, hits, oldVersion) + if err != nil { + return created, skipped, err + } + created += c + skipped += s + if created+skipped >= nextLog { + logger.Info().Msgf("migrating search index: %d of %d documents", created+skipped, total) + nextLog += migrateLogInterval + } + next, err := client.Scroll.Get(ctx, opensearchgoAPI.ScrollGetReq{ + ScrollID: scrollID, + Params: opensearchgoAPI.ScrollGetParams{Scroll: migrateScrollKeepAlive}, + }) + if err != nil { + return created, skipped, err + } + if next.ScrollID != nil { + scrollID = *next.ScrollID + } + hits = next.Hits.Hits + } + if created+skipped != total { + return created, skipped, fmt.Errorf("processed %d of %d source documents", created+skipped, total) + } + return created, skipped, nil +} + +// bulkCreate inserts hits into target with create semantics: a document already +// present is a 409 conflict, counted as skipped and never overwritten. +func bulkCreate(ctx context.Context, client *opensearchgoAPI.Client, target string, hits []opensearchgoAPI.SearchHit, oldVersion int) (int, int, error) { + var body strings.Builder + for _, hit := range hits { + r, err := legacyHitToResource(hit.Source, oldVersion) + if err != nil { + return 0, 0, fmt.Errorf("convert document %s: %w", hit.ID, err) + } + doc, err := mapping.PrepareForIndex(r, r.SearchFieldOverrides()) + if err != nil { + return 0, 0, fmt.Errorf("prepare document %s: %w", hit.ID, err) + } + action, err := json.Marshal(map[string]any{"create": map[string]any{"_index": target, "_id": hit.ID}}) + if err != nil { + return 0, 0, err + } + docJSON, err := json.Marshal(doc) + if err != nil { + return 0, 0, err + } + body.Write(action) + body.WriteByte('\n') + body.Write(docJSON) + body.WriteByte('\n') + } + + resp, err := client.Bulk(ctx, opensearchgoAPI.BulkReq{Body: strings.NewReader(body.String())}) + if err != nil { + return 0, 0, fmt.Errorf("bulk create: %w", err) + } + var created, skipped int + for _, item := range resp.Items { + res, ok := item["create"] + if !ok { + continue + } + switch { + case res.Status == 409: + skipped++ // already present, keep the existing (newer) document + case res.Error != nil || res.Status >= 300: + reason := "" + if res.Error != nil { + reason = res.Error.Type + ": " + res.Error.Reason + } + return created, skipped, fmt.Errorf("create document %s failed: %s", res.ID, reason) + default: + created++ + } + } + return created, skipped, nil +} + +// legacyHitToResource is the per-document fixup applied while reindexing. +// oldVersion is the schema version of the SOURCE index (0 for the legacy +// unversioned index), so fixups are gated on where the document comes from: the +// empty-Mtime strip only runs for documents from the pre-date schema. The switch +// is the hook for future revision-dependent migrations. +func legacyHitToResource(source json.RawMessage, oldVersion int) (search.Resource, error) { + var m map[string]any + if err := json.Unmarshal(source, &m); err != nil { + return search.Resource{}, err + } + switch { + case oldVersion <= 1: + // pre-date schema: an empty Mtime is not a valid date, strip it + if v, ok := m["Mtime"]; ok && v == "" { + delete(m, "Mtime") + } + } + return conversions.To[search.Resource](m) +} diff --git a/services/search/pkg/opensearch/migrate_fields_test.go b/services/search/pkg/opensearch/migrate_fields_test.go new file mode 100644 index 0000000000..d3ab8e9a06 --- /dev/null +++ b/services/search/pkg/opensearch/migrate_fields_test.go @@ -0,0 +1,79 @@ +package opensearch_test + +import ( + "context" + "strings" + "testing" + "time" + + libregraph "github.com/opencloud-eu/libre-graph-api-go" + "github.com/stretchr/testify/require" + + "github.com/opencloud-eu/opencloud/pkg/conversions" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/search/pkg/content" + "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" + opensearchtest "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +func fullyPopulated() search.Resource { + return search.Resource{ + ID: "1$2!3", + RootID: "1$2!2", + ParentID: "1$2!2", + Path: "./a/b/song.mp3", + Type: 2, + Deleted: true, + Hidden: true, + Document: content.Document{ + Title: "the title", + Name: "song.mp3", + Content: "some extracted content", + Size: 123456, + Mtime: conversions.ToPointer(time.Date(2026, 3, 15, 12, 0, 0, 0, time.UTC)), + MimeType: "audio/mpeg", + Tags: []string{"alpha", "beta", "gamma"}, + Favorites: []string{"user-a", "user-b"}, + Audio: &libregraph.Audio{ + Album: libregraph.PtrString("the album"), + Artist: libregraph.PtrString("the artist"), + Track: libregraph.PtrInt32(7), + Year: libregraph.PtrInt32(1998), + HasDrm: libregraph.PtrBool(false), + }, + Image: &libregraph.Image{Width: libregraph.PtrInt32(1920), Height: libregraph.PtrInt32(1080)}, + Photo: &libregraph.Photo{CameraMake: libregraph.PtrString("Canon"), Iso: libregraph.PtrInt32(400), FNumber: libregraph.PtrFloat64(2.8)}, + Location: &libregraph.GeoCoordinates{ + Longitude: libregraph.PtrFloat64(11.103870357204285), + Latitude: libregraph.PtrFloat64(49.48675890884328), + Altitude: libregraph.PtrFloat64(300.0), + }, + }, + } +} + +func TestMigratePreservesAllFields(t *testing.T) { + ctx := context.Background() + base := "opencloud-migrate-fields" + tc, target := newMigrateTest(t, base) + + want := fullyPopulated() + tc.Require.IndicesCreate(base, strings.NewReader(`{}`)) + tc.Require.DocumentCreate(base, want.ID, strings.NewReader(opensearchtest.JSONMustMarshal(t, want))) + tc.Require.IndicesCount([]string{base}, nil, 1) + + _, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger()) + require.NoError(t, err) + + // read the migrated document back from the versioned target; every field must round-trip + hits := tc.Require.Search(target, strings.NewReader(`{"size":10,"query":{"match_all":{}}}`)) + resources := opensearchtest.SearchHitsMustBeConverted[search.Resource](t, hits.Hits) + require.Len(t, resources, 1) + got := resources[0] + + require.NotNil(t, got.Mtime, "Mtime") + require.True(t, want.Mtime.Equal(*got.Mtime), "Mtime: want %v got %v", want.Mtime, got.Mtime) + got.Mtime = want.Mtime // normalize time.Time location/monotonic before the deep-equal + require.Equal(t, want, got) +} diff --git a/services/search/pkg/opensearch/migrate_test.go b/services/search/pkg/opensearch/migrate_test.go new file mode 100644 index 0000000000..a05c68b6be --- /dev/null +++ b/services/search/pkg/opensearch/migrate_test.go @@ -0,0 +1,249 @@ +package opensearch_test + +import ( + "context" + "fmt" + "strings" + "testing" + "time" + + libregraph "github.com/opencloud-eu/libre-graph-api-go" + opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi" + "github.com/stretchr/testify/require" + + "github.com/opencloud-eu/opencloud/pkg/conversions" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/search/pkg/content" + "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" + opensearchtest "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +// newMigrateTest returns a client and the current-revision target for a fresh +// migration test, deleting base, its versioned siblings and the target both +// before the test and via t.Cleanup. +func newMigrateTest(t *testing.T, base string) (*opensearchtest.TestClient, string) { + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + indices := []string{base, base + "-v0", opensearch.TargetIndex(base)} + reset := func() { + for _, i := range indices { + _, _ = tc.Client().Indices.Delete(context.Background(), opensearchgoAPI.IndicesDeleteReq{Indices: []string{i}}) + } + } + reset() + t.Cleanup(reset) + return tc, opensearch.TargetIndex(base) +} + +func migrateDoc(id, name string, deleted bool) search.Resource { + lon, lat, alt := 11.103870357204285, 49.48675890884328, 300.0 + return search.Resource{ + ID: id, + RootID: "1$1!1", + ParentID: "1$1!1", + Path: "./" + name, + Type: 2, + Deleted: deleted, + Document: content.Document{ + Name: name, + Content: "hello world", + Size: 42, + Mtime: conversions.ToPointer(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)), + MimeType: "image/jpeg", + Tags: []string{"tag1"}, + Location: &libregraph.GeoCoordinates{Longitude: &lon, Latitude: &lat, Altitude: &alt}, + }, + } +} + +func contentDoc(id, text string) search.Resource { + return search.Resource{ + ID: id, + RootID: "1$1!1", + ParentID: "1$1!1", + Path: "./" + id, + Type: 2, + Document: content.Document{ + Name: id, + Content: text, + Mtime: conversions.ToPointer(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)), + }, + } +} + +// bulkIndex writes total migrateDoc documents into base, in chunks. +func bulkIndex(t *testing.T, tc *opensearchtest.TestClient, base string, total int) { + const chunk = 5000 + for i := 0; i < total; i += chunk { + var bulk strings.Builder + for j := i; j < i+chunk && j < total; j++ { + id := fmt.Sprintf("doc%05d", j) + fmt.Fprintf(&bulk, "{\"create\":{\"_index\":%q,\"_id\":%q}}\n", base, id) + bulk.WriteString(opensearchtest.JSONMustMarshal(t, migrateDoc(id, id+".txt", false))) + bulk.WriteString("\n") + } + _, err := tc.Client().Bulk(context.Background(), opensearchgoAPI.BulkReq{Body: strings.NewReader(bulk.String())}) + require.NoError(t, err) + } +} + +func TestMigrateIndex(t *testing.T) { + ctx := context.Background() + base := "opencloud-migrate-basic" + tc, target := newMigrateTest(t, base) + + // 1) a legacy unversioned index from a pre-versioning release: dynamic mapping, + // documents including a trashed one, location as a plain object (no geopoint) + tc.Require.IndicesCreate(base, strings.NewReader(`{}`)) + for _, d := range []search.Resource{ + migrateDoc("doc1", "photo.jpg", false), + migrateDoc("doc2", "song.mp3", false), + migrateDoc("doc3", "trashed.txt", true), + } { + tc.Require.DocumentCreate(base, d.ID, strings.NewReader(opensearchtest.JSONMustMarshal(t, d))) + } + tc.Require.IndicesCount([]string{base}, nil, 3) + + // 2) migrate builds the versioned target from the legacy index + n, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger()) + require.NoError(t, err) + require.Equal(t, 3, n) + + // 3) the versioned target exists and holds every document + tc.Require.IndicesCount([]string{target}, nil, 3) + + // 3a) the trashed document survived (a rescan would have dropped it) + trash := `{"query":{"term":{"Deleted":true}}}` + tc.Require.IndicesCount([]string{target}, strings.NewReader(trash), 1) + + // 4) location_geopoint was synthesized during migration (absent in the legacy index) + geo := fmt.Sprintf(`{"query":{"geo_distance":{"distance":"1km","location%s":{"lat":%f,"lon":%f}}}}`, + "_geopoint", 49.48675890884328, 11.103870357204285) + tc.Require.IndicesCount([]string{target}, strings.NewReader(geo), 3) + + // 5) the service starts against the target and Apply classifies it equal + require.NoError(t, opensearch.IndexManagerLatest.Apply(ctx, target, tc.Client(), log.NopLogger())) + + // 6) idempotent: the target now exists, so a second run creates nothing + n, err = opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger()) + require.NoError(t, err) + require.Equal(t, 0, n) +} + +// TestMigrateLargeIndex covers reindexing past a single scroll page and past +// OpenSearch's default track_total_hits cap (10000) — the completeness check +// must use the exact total. +func TestMigrateLargeIndex(t *testing.T) { + for _, row := range []struct { + name string + total int + }{ + {"pages beyond one scroll page", 2500}, + {"exceeds the 10k track_total_hits cap", 10500}, + } { + t.Run(row.name, func(t *testing.T) { + ctx := context.Background() + base := fmt.Sprintf("opencloud-migrate-large-%d", row.total) + tc, target := newMigrateTest(t, base) + + tc.Require.IndicesCreate(base, strings.NewReader(`{}`)) + bulkIndex(t, tc, base, row.total) + tc.Require.IndicesRefresh([]string{base}, nil) + tc.Require.IndicesCount([]string{base}, nil, row.total) + + n, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger()) + require.NoError(t, err) + require.Equal(t, row.total, n, "every document across all pages must be migrated") + tc.Require.IndicesCount([]string{target}, nil, row.total) + }) + } +} + +// TestMigrateBackfillsWithoutOverwriting simulates the post-rollout / late run: +// the target already exists (an instance created it and wrote newer documents), +// and migrate must backfill the missing ones with create-only ops, never +// overwriting the newer ones. +func TestMigrateBackfillsWithoutOverwriting(t *testing.T) { + ctx := context.Background() + base := "opencloud-migrate-backfill" + tc, target := newMigrateTest(t, base) + + // legacy index with three documents (old content) + tc.Require.IndicesCreate(base, strings.NewReader(`{}`)) + for _, id := range []string{"a", "b", "c"} { + tc.Require.DocumentCreate(base, id, strings.NewReader(opensearchtest.JSONMustMarshal(t, contentDoc(id, "OLD")))) + } + tc.Require.IndicesCount([]string{base}, nil, 3) + + // a new instance already created the target and wrote a newer "a" plus a new "d" + tc.Require.IndicesCreate(target, strings.NewReader(opensearch.IndexManagerLatest.String())) + tc.Require.DocumentCreate(target, "a", strings.NewReader(opensearchtest.JSONMustMarshal(t, contentDoc("a", "NEW")))) + tc.Require.DocumentCreate(target, "d", strings.NewReader(opensearchtest.JSONMustMarshal(t, contentDoc("d", "NEW")))) + tc.Require.IndicesRefresh([]string{target}, nil) + + // migrate backfills b and c, must not overwrite the newer a + n, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger()) + require.NoError(t, err) + require.Equal(t, 2, n, "only b and c are newly created (a is skipped, d untouched)") + + tc.Require.IndicesRefresh([]string{target}, nil) + tc.Require.IndicesCount([]string{target}, nil, 4) // a, b, c, d + + hits := tc.Require.Search(target, strings.NewReader(`{"query":{"ids":{"values":["a"]}}}`)) + got := opensearchtest.SearchHitsMustBeConverted[search.Resource](t, hits.Hits) + require.Len(t, got, 1) + require.Equal(t, "NEW", got[0].Content, "the newer document must not be overwritten by the migration") +} + +func TestPruneOldIndices(t *testing.T) { + ctx := context.Background() + base := "opencloud-prune" + tc, target := newMigrateTest(t, base) + old := base + "-v0" + + // refuses while the current-revision index does not exist yet + _, err := opensearch.PruneOldIndices(ctx, tc.Client(), base, log.NopLogger()) + require.Error(t, err, "prune must refuse when the current index is missing") + + // legacy, an older versioned index, and the current index all exist + tc.Require.IndicesCreate(base, strings.NewReader(`{}`)) + tc.Require.IndicesCreate(old, strings.NewReader(`{}`)) + tc.Require.IndicesCreate(target, strings.NewReader(opensearch.IndexManagerLatest.String())) + + n, err := opensearch.PruneOldIndices(ctx, tc.Client(), base, log.NopLogger()) + require.NoError(t, err) + require.Equal(t, 2, n, "legacy base and base-v0 are pruned, base-v1 stays") + + for _, gone := range []string{base, old} { + resp, _ := tc.Client().Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{Indices: []string{gone}}) + require.Equal(t, 404, resp.StatusCode, "%s must be pruned", gone) + } + resp, err := tc.Client().Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{Indices: []string{target}}) + require.NoError(t, err) + require.Equal(t, 200, resp.StatusCode, "the current index must survive prune") +} + +// TestMigrateStripsEmptyMtime covers the one legacy value that would otherwise +// break the reindex: an empty Mtime string, which is not a valid date. It must +// be stripped so the document survives instead of being rejected by the new +// date mapping. +func TestMigrateStripsEmptyMtime(t *testing.T) { + ctx := context.Background() + base := "opencloud-migrate-empty-mtime" + tc, target := newMigrateTest(t, base) + + // an old document with an empty Mtime, exactly as the pre-fix code wrote it + tc.Require.IndicesCreate(base, strings.NewReader(`{}`)) + tc.Require.DocumentCreate(base, "x", strings.NewReader(`{"ID":"x","Name":"f.txt","Path":"./f.txt","Mtime":""}`)) + tc.Require.IndicesCount([]string{base}, nil, 1) + + n, err := opensearch.MigrateIndex(ctx, base, tc.Client(), log.NopLogger()) + require.NoError(t, err) + require.Equal(t, 1, n, "the empty-mtime document must survive, not be dropped") + + tc.Require.IndicesCount([]string{target}, nil, 1) + hits := tc.Require.Search(target, strings.NewReader(`{"query":{"match_all":{}}}`)) + resources := opensearchtest.SearchHitsMustBeConverted[search.Resource](t, hits.Hits) + require.Len(t, resources, 1) + require.Nil(t, resources[0].Mtime, "the empty mtime should be dropped, not carried over") +} diff --git a/services/search/pkg/search/search.go b/services/search/pkg/search/search.go index c0dac8fe99..047d7fc45b 100644 --- a/services/search/pkg/search/search.go +++ b/services/search/pkg/search/search.go @@ -49,6 +49,12 @@ type BatchOperator interface { Push() error } +// SchemaRevision is the version of the index schema defined by Resource and its +// SearchFieldOverrides. Bump it deliberately when a change needs a rebuild: the +// bump triggers the migration, not a classifier diff. Both backends share it, +// bleve stores it in the index, opensearch carries it as the index name suffix. +const SchemaRevision = 1 + // Resource is the entity that is stored in the index. type Resource struct { content.Document From 4788a98cdc9b4493417940dd6692b2e2d9cafb03 Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Fri, 17 Jul 2026 09:00:49 +0200 Subject: [PATCH 2/3] explore(search): bleve reset+rescan instead of in-place migration Alternative that drops the in-place rebuild machinery (buildReplacement, swap, recoverMigration, verifyMigrated, the fixup hook) and just wipes the outdated bleve index so a full 'search index --force-rescan' repopulates it. ~272 lines lighter, but trades the fast no-rescan migration for a Tika re-crawl on every schema bump. Kept in history for reference; reverted in the next commit - the in-place migration is worth the ~300 lines for bleve too. --- services/search/pkg/bleve/migrate.go | 234 +++--------------- .../search/pkg/bleve/migrate_fields_test.go | 89 ------- services/search/pkg/bleve/migrate_test.go | 21 +- 3 files changed, 36 insertions(+), 308 deletions(-) delete mode 100644 services/search/pkg/bleve/migrate_fields_test.go diff --git a/services/search/pkg/bleve/migrate.go b/services/search/pkg/bleve/migrate.go index 760844f8dd..27aba850f2 100644 --- a/services/search/pkg/bleve/migrate.go +++ b/services/search/pkg/bleve/migrate.go @@ -3,7 +3,6 @@ package bleve import ( "errors" "fmt" - "io/fs" "os" "path/filepath" "strconv" @@ -15,235 +14,66 @@ import ( "github.com/opencloud-eu/opencloud/services/search/pkg/search" ) -const ( - revisionKey = "_schema_revision" - migratingSuffix = ".migrating" - backupSuffix = ".bak" - migrateBatch = 100 - migrateLogInterval = 50000 // how often buildReplacement logs progress, in documents -) +const revisionKey = "_schema_revision" // OpenOrMigrate opens the index and, when autoMigrate is set and the stored -// schema is incompatible, rebuilds it in place first. The migration triggers on -// a schema revision bump, not on a mapping diff, so a breaking change without a -// matching bump still refuses to start. Returns the number of documents migrated -// (0 for none). Opening first means a fresh install and the common up-to-date -// case open the index only once; the migration path is taken only on a mismatch. +// schema is incompatible, resets it first so a re-crawl can repopulate it at the +// current schema. The reset triggers on a schema revision bump, not on a mapping +// diff, so a breaking change without a matching bump still refuses to start. func OpenOrMigrate(root string, autoMigrate bool, logger log.Logger) (bleve.Index, searchmapping.Classification, int, error) { idx, classification, openErr := NewIndex(root) if openErr == nil || !autoMigrate || !errors.Is(openErr, searchmapping.ErrManualActionRequired) { - // opened cleanly, or auto-migrate is off, or the failure is not a - // schema mismatch we could migrate away return idx, classification, 0, openErr } - // the stored schema is incompatible; rebuild only if the revision was bumped, - // otherwise surface the original error (a schema change without a bump) - migrated, err := MigrateIndex(root, logger) + // incompatible schema: MigrateIndex drops a bumped-revision index so the + // reopen below rebuilds it empty; a breaking change without a bump is a no-op + // there and is surfaced by that reopen instead + dropped, err := MigrateIndex(root, logger) if err != nil { return nil, classification, 0, err } - if migrated == 0 { - return nil, classification, 0, openErr - } idx, classification, err = NewIndex(root) - return idx, classification, migrated, err + return idx, classification, dropped, err } -// MigrateIndex rebuilds the index from the documents it already holds when the -// stored schema revision is older than search.SchemaRevision, else a no-op. It -// takes the exclusive bolt lock, so the search service must be stopped. +// MigrateIndex resets the index when its stored schema revision is older than +// search.SchemaRevision: it drops the outdated index so a full re-crawl +// ('opencloud search index --all-spaces --force-rescan') repopulates it at the +// current schema. No-op when the index is missing or already current. It takes +// the exclusive bolt lock, so the search service must be stopped. Returns the +// number of documents that were dropped. func MigrateIndex(root string, logger log.Logger) (int, error) { dest := filepath.Join(root, "bleve") - tmp := dest + migratingSuffix - bak := dest + backupSuffix - - if err := recoverMigration(root); err != nil { - return 0, err - } - old, err := bleve.OpenUsing(dest, openRuntimeConfig) + idx, err := bleve.OpenUsing(dest, openRuntimeConfig) if errors.Is(err, bleve.ErrorIndexPathDoesNotExist) { - return 0, nil // nothing to migrate yet; NewIndex creates a fresh index + return 0, nil // fresh install, nothing to migrate } if err != nil { return 0, fmt.Errorf("open index (is the search service still running?): %w", err) } - - stored, err := readRevision(old) - if err != nil { - _ = old.Close() - return 0, err - } - if stored >= search.SchemaRevision { - _ = old.Close() - return 0, nil - } - - total, err := old.DocCount() - if err != nil { - _ = old.Close() - return 0, err - } - logger.Info().Uint64("documents", total).Msg("starting search index migration") - - copied, err := buildReplacement(old, tmp, total, logger) - if err != nil { - _ = old.Close() - _ = os.RemoveAll(tmp) - return 0, err - } - - // verify before the swap, so a failure leaves the original untouched - if err := verifyMigrated(tmp, total); err != nil { - _ = old.Close() - _ = os.RemoveAll(tmp) - return 0, err - } - - if err := swap(old, dest, tmp, bak); err != nil { - return 0, err - } - logger.Info().Int("documents", copied).Msg("search index migration complete") - return copied, nil -} - -// buildReplacement copies every document from src into a fresh index at tmp via -// the same Deserialize -> PrepareForIndex round-trip the Move/Delete/Restore -// paths use, and stamps the current revision. -func buildReplacement(src bleve.Index, tmp string, total uint64, logger log.Logger) (int, error) { - if err := os.RemoveAll(tmp); err != nil { - return 0, err - } - m, err := NewMapping() - if err != nil { - return 0, err - } - dst, err := bleve.New(tmp, m) - if err != nil { - return 0, err - } - - batch, err := NewBatch(dst, migrateBatch) - if err != nil { - _ = dst.Close() - return 0, err + stored, rerr := readRevision(idx) + count, cerr := idx.DocCount() + _ = idx.Close() + switch { + case rerr != nil: + return 0, rerr + case cerr != nil: + return 0, cerr + case stored >= search.SchemaRevision: + return 0, nil // already current } - var copied int - nextLog := migrateLogInterval - var after []string - for { - req := bleve.NewSearchRequest(bleve.NewMatchAllQuery()) - req.Size = migrateBatch - req.Fields = []string{"*"} - req.SortBy([]string{"_id"}) // total order, no skips or dupes across pages - req.SearchAfter = after - res, err := src.Search(req) - if err != nil { - _ = dst.Close() - return 0, err - } - if len(res.Hits) == 0 { - break - } - for _, hit := range res.Hits { - r := legacyHitToResource(hit.Fields) - if err := batch.Upsert(hit.ID, r); err != nil { // hit.ID: the bleve key is authoritative - _ = dst.Close() - return 0, err - } - copied++ - } - if copied >= nextLog { - logger.Info().Msgf("migrating search index: %d of %d documents", copied, total) - nextLog += migrateLogInterval - } - after = []string{res.Hits[len(res.Hits)-1].ID} - } - if err := batch.Push(); err != nil { - _ = dst.Close() - return 0, err - } - if err := writeRevision(dst); err != nil { - _ = dst.Close() + if err := os.RemoveAll(dest); err != nil { return 0, err } - return copied, dst.Close() -} - -// legacyHitToResource is the fixup hook before re-indexing, symmetric to the -// opensearch side. bleve Deserialize is fail-soft, so nothing is stripped here. -func legacyHitToResource(fields map[string]any) search.Resource { - return *searchmapping.Deserialize[search.Resource](fields) -} - -// verifyMigrated checks the rebuilt index holds want documents before the swap -// replaces the original. The count catches a pagination drop or batch-push loss; -// the mapping needs no check, it is built deterministically from NewMapping(). -func verifyMigrated(path string, want uint64) error { - idx, err := bleve.OpenUsing(path, openRuntimeConfig) - if err != nil { - return err - } - defer func() { _ = idx.Close() }() - - got, err := idx.DocCount() - if err != nil { - return err - } - if got != want { - return fmt.Errorf("migrated index holds %d documents, expected %d", got, want) - } - return nil -} - -// swap replaces the live index with the rebuilt one. old still holds the lock -// and is closed here right before the rename. -func swap(old bleve.Index, dest, tmp, bak string) error { - if err := old.Close(); err != nil { - return err - } - if err := os.Rename(dest, bak); err != nil { - return err - } - if err := os.Rename(tmp, dest); err != nil { - _ = os.Rename(bak, dest) // best effort roll back - return err - } - return os.RemoveAll(bak) -} - -// recoverMigration cleans up a crashed MigrateIndex and must run before the -// index is opened, else NewIndex would recreate an empty index mid-swap. Rolls -// back, never forward: a truncated tmp is a valid index, completeness is not -// observable. -func recoverMigration(root string) error { - dest := filepath.Join(root, "bleve") - tmp := dest + migratingSuffix - bak := dest + backupSuffix - - if _, err := os.Stat(dest); err == nil { - return errors.Join(os.RemoveAll(tmp), os.RemoveAll(bak)) // index present: tmp/bak are leftovers - } else if !errors.Is(err, fs.ErrNotExist) { - return err - } - - if _, err := os.Stat(bak); err == nil { - // crashed mid-swap: roll back - if err := os.RemoveAll(tmp); err != nil { - return err - } - return os.Rename(bak, dest) - } else if !errors.Is(err, fs.ErrNotExist) { - return err - } - - return os.RemoveAll(tmp) // fresh install, or index removed by hand + logger.Warn().Msgf("the bleve index at %s was reset for schema revision %d (%d documents dropped); run 'opencloud search index --all-spaces --force-rescan' to repopulate", root, search.SchemaRevision, count) + return int(count), nil } -// writeRevision stamps the current revision so a fresh or migrated index is not -// seen as outdated. +// writeRevision stamps the current revision so a fresh index is not seen as +// outdated on the next start. func writeRevision(index bleve.Index) error { return index.SetInternal([]byte(revisionKey), []byte(strconv.Itoa(search.SchemaRevision))) } diff --git a/services/search/pkg/bleve/migrate_fields_test.go b/services/search/pkg/bleve/migrate_fields_test.go deleted file mode 100644 index b570fa0424..0000000000 --- a/services/search/pkg/bleve/migrate_fields_test.go +++ /dev/null @@ -1,89 +0,0 @@ -package bleve_test - -import ( - "path/filepath" - "sort" - "testing" - "time" - - bleveSearch "github.com/blevesearch/bleve/v2" - libregraph "github.com/opencloud-eu/libre-graph-api-go" - "github.com/stretchr/testify/require" - - "github.com/opencloud-eu/opencloud/pkg/conversions" - "github.com/opencloud-eu/opencloud/pkg/log" - "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" - "github.com/opencloud-eu/opencloud/services/search/pkg/content" - searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" - "github.com/opencloud-eu/opencloud/services/search/pkg/search" -) - -// fullyPopulated exercises every field, including the facets and slices the -// count-only test skipped. -func fullyPopulated() search.Resource { - return search.Resource{ - ID: "1$2!3", - RootID: "1$2!2", - ParentID: "1$2!2", - Path: "./a/b/song.mp3", - Type: 2, - Deleted: true, - Hidden: true, - Document: content.Document{ - Title: "the title", - Name: "song.mp3", - Content: "some extracted content", - Size: 123456, - Mtime: conversions.ToPointer(time.Date(2026, 3, 15, 12, 0, 0, 0, time.UTC)), - MimeType: "audio/mpeg", - Tags: []string{"alpha", "beta", "gamma"}, - Favorites: []string{"user-a", "user-b"}, - Audio: &libregraph.Audio{ - Album: libregraph.PtrString("the album"), - Artist: libregraph.PtrString("the artist"), - Track: libregraph.PtrInt32(7), - Year: libregraph.PtrInt32(1998), - HasDrm: libregraph.PtrBool(false), - }, - Image: &libregraph.Image{Width: libregraph.PtrInt32(1920), Height: libregraph.PtrInt32(1080)}, - Photo: &libregraph.Photo{CameraMake: libregraph.PtrString("Canon"), Iso: libregraph.PtrInt32(400), FNumber: libregraph.PtrFloat64(2.8)}, - Location: &libregraph.GeoCoordinates{ - Longitude: libregraph.PtrFloat64(11.103870357204285), - Latitude: libregraph.PtrFloat64(49.48675890884328), - Altitude: libregraph.PtrFloat64(300.0), - }, - }, - } -} - -func TestMigratePreservesAllFields(t *testing.T) { - root := t.TempDir() - want := fullyPopulated() - - old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t)) - require.NoError(t, err) - require.NoError(t, old.Index(want.ID, want)) - require.NoError(t, old.Close()) - - _, err = bleve.MigrateIndex(root, log.NopLogger()) - require.NoError(t, err) - - idx, _, err := bleve.NewIndex(root) - require.NoError(t, err) - defer func() { _ = idx.Close() }() - - // read the migrated document back through the production reconstruction path - req := bleveSearch.NewSearchRequest(bleveSearch.NewMatchAllQuery()) - req.Fields = []string{"*"} - res, err := idx.Search(req) - require.NoError(t, err) - require.Len(t, res.Hits, 1) - got := searchmapping.Deserialize[search.Resource](res.Hits[0].Fields) - - require.NotNil(t, got.Mtime, "Mtime") - require.True(t, want.Mtime.Equal(*got.Mtime), "Mtime: want %v got %v", want.Mtime, got.Mtime) - got.Mtime = want.Mtime // normalize time.Time location/monotonic - sort.Strings(got.Tags) // bleve may return multi-value fields reordered - sort.Strings(got.Favorites) // - require.Equal(t, want, *got) -} diff --git a/services/search/pkg/bleve/migrate_test.go b/services/search/pkg/bleve/migrate_test.go index 0c1ab3489a..dbaf9c3697 100644 --- a/services/search/pkg/bleve/migrate_test.go +++ b/services/search/pkg/bleve/migrate_test.go @@ -7,7 +7,6 @@ import ( bleveSearch "github.com/blevesearch/bleve/v2" bleveMapping "github.com/blevesearch/bleve/v2/mapping" - "github.com/blevesearch/bleve/v2/search/query" libregraph "github.com/opencloud-eu/libre-graph-api-go" "github.com/stretchr/testify/require" @@ -129,35 +128,23 @@ func TestMigrateIndex(t *testing.T) { _, _, err = bleve.NewIndex(root) require.ErrorIs(t, err, searchmapping.ErrManualActionRequired) - // 3) migrate + // 3) migrate resets the outdated index, dropping its documents n, err := bleve.MigrateIndex(root, log.NopLogger()) require.NoError(t, err) require.Equal(t, 3, n) - // 4) the migrated index classifies as equal and holds every document + // 4) the reset index classifies as equal and is empty, awaiting a re-crawl migrated, classification, err := bleve.NewIndex(root) require.NoError(t, err) require.Equal(t, searchmapping.VerdictEqual, classification.Verdict) count, err := migrated.DocCount() require.NoError(t, err) - require.Equal(t, uint64(3), count) - - // 4a) the trashed document survived (a rescan would have dropped it) - trashed, err := migrated.Document("1$1!3") - require.NoError(t, err) - require.NotNil(t, trashed) - - // 5) location_geopoint was synthesized during migration (absent in the old index) - near := query.NewGeoDistanceQuery(11.103870357204285, 49.48675890884328, "1km") - near.SetField("location" + searchmapping.GeopointSuffix) - res, err := migrated.Search(bleveSearch.NewSearchRequest(near)) - require.NoError(t, err) - require.Equal(t, uint64(3), res.Total, "all three docs match a geo-distance query on the new sibling field") + require.Equal(t, uint64(0), count, "reset drops the documents; a re-crawl repopulates") require.NoError(t, migrated.Close()) // release the lock before the second run - // 6) idempotent: a second run is a no-op because the revision is now current + // 5) idempotent: a second run is a no-op because the revision is now current n, err = bleve.MigrateIndex(root, log.NopLogger()) require.NoError(t, err) require.Equal(t, 0, n) From 3c98a31b9db2cca74d0d64e90500bdebec03460b Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Fri, 17 Jul 2026 09:01:04 +0200 Subject: [PATCH 3/3] Revert "explore(search): bleve reset+rescan instead of in-place migration" Keep the in-place bleve migration: the fast no-rescan rebuild is worth its ~300 lines even for bleve (avoids a Tika re-crawl on schema bumps). The reset variant stays in the previous commit for reference. This reverts commit 4788a98cdc. --- services/search/pkg/bleve/migrate.go | 234 +++++++++++++++--- .../search/pkg/bleve/migrate_fields_test.go | 89 +++++++ services/search/pkg/bleve/migrate_test.go | 21 +- 3 files changed, 308 insertions(+), 36 deletions(-) create mode 100644 services/search/pkg/bleve/migrate_fields_test.go diff --git a/services/search/pkg/bleve/migrate.go b/services/search/pkg/bleve/migrate.go index 27aba850f2..760844f8dd 100644 --- a/services/search/pkg/bleve/migrate.go +++ b/services/search/pkg/bleve/migrate.go @@ -3,6 +3,7 @@ package bleve import ( "errors" "fmt" + "io/fs" "os" "path/filepath" "strconv" @@ -14,66 +15,235 @@ import ( "github.com/opencloud-eu/opencloud/services/search/pkg/search" ) -const revisionKey = "_schema_revision" +const ( + revisionKey = "_schema_revision" + migratingSuffix = ".migrating" + backupSuffix = ".bak" + migrateBatch = 100 + migrateLogInterval = 50000 // how often buildReplacement logs progress, in documents +) // OpenOrMigrate opens the index and, when autoMigrate is set and the stored -// schema is incompatible, resets it first so a re-crawl can repopulate it at the -// current schema. The reset triggers on a schema revision bump, not on a mapping -// diff, so a breaking change without a matching bump still refuses to start. +// schema is incompatible, rebuilds it in place first. The migration triggers on +// a schema revision bump, not on a mapping diff, so a breaking change without a +// matching bump still refuses to start. Returns the number of documents migrated +// (0 for none). Opening first means a fresh install and the common up-to-date +// case open the index only once; the migration path is taken only on a mismatch. func OpenOrMigrate(root string, autoMigrate bool, logger log.Logger) (bleve.Index, searchmapping.Classification, int, error) { idx, classification, openErr := NewIndex(root) if openErr == nil || !autoMigrate || !errors.Is(openErr, searchmapping.ErrManualActionRequired) { + // opened cleanly, or auto-migrate is off, or the failure is not a + // schema mismatch we could migrate away return idx, classification, 0, openErr } - // incompatible schema: MigrateIndex drops a bumped-revision index so the - // reopen below rebuilds it empty; a breaking change without a bump is a no-op - // there and is surfaced by that reopen instead - dropped, err := MigrateIndex(root, logger) + // the stored schema is incompatible; rebuild only if the revision was bumped, + // otherwise surface the original error (a schema change without a bump) + migrated, err := MigrateIndex(root, logger) if err != nil { return nil, classification, 0, err } + if migrated == 0 { + return nil, classification, 0, openErr + } idx, classification, err = NewIndex(root) - return idx, classification, dropped, err + return idx, classification, migrated, err } -// MigrateIndex resets the index when its stored schema revision is older than -// search.SchemaRevision: it drops the outdated index so a full re-crawl -// ('opencloud search index --all-spaces --force-rescan') repopulates it at the -// current schema. No-op when the index is missing or already current. It takes -// the exclusive bolt lock, so the search service must be stopped. Returns the -// number of documents that were dropped. +// MigrateIndex rebuilds the index from the documents it already holds when the +// stored schema revision is older than search.SchemaRevision, else a no-op. It +// takes the exclusive bolt lock, so the search service must be stopped. func MigrateIndex(root string, logger log.Logger) (int, error) { dest := filepath.Join(root, "bleve") + tmp := dest + migratingSuffix + bak := dest + backupSuffix + + if err := recoverMigration(root); err != nil { + return 0, err + } - idx, err := bleve.OpenUsing(dest, openRuntimeConfig) + old, err := bleve.OpenUsing(dest, openRuntimeConfig) if errors.Is(err, bleve.ErrorIndexPathDoesNotExist) { - return 0, nil // fresh install, nothing to migrate + return 0, nil // nothing to migrate yet; NewIndex creates a fresh index } if err != nil { return 0, fmt.Errorf("open index (is the search service still running?): %w", err) } - stored, rerr := readRevision(idx) - count, cerr := idx.DocCount() - _ = idx.Close() - switch { - case rerr != nil: - return 0, rerr - case cerr != nil: - return 0, cerr - case stored >= search.SchemaRevision: - return 0, nil // already current + + stored, err := readRevision(old) + if err != nil { + _ = old.Close() + return 0, err + } + if stored >= search.SchemaRevision { + _ = old.Close() + return 0, nil + } + + total, err := old.DocCount() + if err != nil { + _ = old.Close() + return 0, err + } + logger.Info().Uint64("documents", total).Msg("starting search index migration") + + copied, err := buildReplacement(old, tmp, total, logger) + if err != nil { + _ = old.Close() + _ = os.RemoveAll(tmp) + return 0, err + } + + // verify before the swap, so a failure leaves the original untouched + if err := verifyMigrated(tmp, total); err != nil { + _ = old.Close() + _ = os.RemoveAll(tmp) + return 0, err + } + + if err := swap(old, dest, tmp, bak); err != nil { + return 0, err + } + logger.Info().Int("documents", copied).Msg("search index migration complete") + return copied, nil +} + +// buildReplacement copies every document from src into a fresh index at tmp via +// the same Deserialize -> PrepareForIndex round-trip the Move/Delete/Restore +// paths use, and stamps the current revision. +func buildReplacement(src bleve.Index, tmp string, total uint64, logger log.Logger) (int, error) { + if err := os.RemoveAll(tmp); err != nil { + return 0, err + } + m, err := NewMapping() + if err != nil { + return 0, err + } + dst, err := bleve.New(tmp, m) + if err != nil { + return 0, err + } + + batch, err := NewBatch(dst, migrateBatch) + if err != nil { + _ = dst.Close() + return 0, err } - if err := os.RemoveAll(dest); err != nil { + var copied int + nextLog := migrateLogInterval + var after []string + for { + req := bleve.NewSearchRequest(bleve.NewMatchAllQuery()) + req.Size = migrateBatch + req.Fields = []string{"*"} + req.SortBy([]string{"_id"}) // total order, no skips or dupes across pages + req.SearchAfter = after + res, err := src.Search(req) + if err != nil { + _ = dst.Close() + return 0, err + } + if len(res.Hits) == 0 { + break + } + for _, hit := range res.Hits { + r := legacyHitToResource(hit.Fields) + if err := batch.Upsert(hit.ID, r); err != nil { // hit.ID: the bleve key is authoritative + _ = dst.Close() + return 0, err + } + copied++ + } + if copied >= nextLog { + logger.Info().Msgf("migrating search index: %d of %d documents", copied, total) + nextLog += migrateLogInterval + } + after = []string{res.Hits[len(res.Hits)-1].ID} + } + if err := batch.Push(); err != nil { + _ = dst.Close() + return 0, err + } + if err := writeRevision(dst); err != nil { + _ = dst.Close() return 0, err } - logger.Warn().Msgf("the bleve index at %s was reset for schema revision %d (%d documents dropped); run 'opencloud search index --all-spaces --force-rescan' to repopulate", root, search.SchemaRevision, count) - return int(count), nil + return copied, dst.Close() +} + +// legacyHitToResource is the fixup hook before re-indexing, symmetric to the +// opensearch side. bleve Deserialize is fail-soft, so nothing is stripped here. +func legacyHitToResource(fields map[string]any) search.Resource { + return *searchmapping.Deserialize[search.Resource](fields) +} + +// verifyMigrated checks the rebuilt index holds want documents before the swap +// replaces the original. The count catches a pagination drop or batch-push loss; +// the mapping needs no check, it is built deterministically from NewMapping(). +func verifyMigrated(path string, want uint64) error { + idx, err := bleve.OpenUsing(path, openRuntimeConfig) + if err != nil { + return err + } + defer func() { _ = idx.Close() }() + + got, err := idx.DocCount() + if err != nil { + return err + } + if got != want { + return fmt.Errorf("migrated index holds %d documents, expected %d", got, want) + } + return nil +} + +// swap replaces the live index with the rebuilt one. old still holds the lock +// and is closed here right before the rename. +func swap(old bleve.Index, dest, tmp, bak string) error { + if err := old.Close(); err != nil { + return err + } + if err := os.Rename(dest, bak); err != nil { + return err + } + if err := os.Rename(tmp, dest); err != nil { + _ = os.Rename(bak, dest) // best effort roll back + return err + } + return os.RemoveAll(bak) +} + +// recoverMigration cleans up a crashed MigrateIndex and must run before the +// index is opened, else NewIndex would recreate an empty index mid-swap. Rolls +// back, never forward: a truncated tmp is a valid index, completeness is not +// observable. +func recoverMigration(root string) error { + dest := filepath.Join(root, "bleve") + tmp := dest + migratingSuffix + bak := dest + backupSuffix + + if _, err := os.Stat(dest); err == nil { + return errors.Join(os.RemoveAll(tmp), os.RemoveAll(bak)) // index present: tmp/bak are leftovers + } else if !errors.Is(err, fs.ErrNotExist) { + return err + } + + if _, err := os.Stat(bak); err == nil { + // crashed mid-swap: roll back + if err := os.RemoveAll(tmp); err != nil { + return err + } + return os.Rename(bak, dest) + } else if !errors.Is(err, fs.ErrNotExist) { + return err + } + + return os.RemoveAll(tmp) // fresh install, or index removed by hand } -// writeRevision stamps the current revision so a fresh index is not seen as -// outdated on the next start. +// writeRevision stamps the current revision so a fresh or migrated index is not +// seen as outdated. func writeRevision(index bleve.Index) error { return index.SetInternal([]byte(revisionKey), []byte(strconv.Itoa(search.SchemaRevision))) } diff --git a/services/search/pkg/bleve/migrate_fields_test.go b/services/search/pkg/bleve/migrate_fields_test.go new file mode 100644 index 0000000000..b570fa0424 --- /dev/null +++ b/services/search/pkg/bleve/migrate_fields_test.go @@ -0,0 +1,89 @@ +package bleve_test + +import ( + "path/filepath" + "sort" + "testing" + "time" + + bleveSearch "github.com/blevesearch/bleve/v2" + libregraph "github.com/opencloud-eu/libre-graph-api-go" + "github.com/stretchr/testify/require" + + "github.com/opencloud-eu/opencloud/pkg/conversions" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" + "github.com/opencloud-eu/opencloud/services/search/pkg/content" + searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +// fullyPopulated exercises every field, including the facets and slices the +// count-only test skipped. +func fullyPopulated() search.Resource { + return search.Resource{ + ID: "1$2!3", + RootID: "1$2!2", + ParentID: "1$2!2", + Path: "./a/b/song.mp3", + Type: 2, + Deleted: true, + Hidden: true, + Document: content.Document{ + Title: "the title", + Name: "song.mp3", + Content: "some extracted content", + Size: 123456, + Mtime: conversions.ToPointer(time.Date(2026, 3, 15, 12, 0, 0, 0, time.UTC)), + MimeType: "audio/mpeg", + Tags: []string{"alpha", "beta", "gamma"}, + Favorites: []string{"user-a", "user-b"}, + Audio: &libregraph.Audio{ + Album: libregraph.PtrString("the album"), + Artist: libregraph.PtrString("the artist"), + Track: libregraph.PtrInt32(7), + Year: libregraph.PtrInt32(1998), + HasDrm: libregraph.PtrBool(false), + }, + Image: &libregraph.Image{Width: libregraph.PtrInt32(1920), Height: libregraph.PtrInt32(1080)}, + Photo: &libregraph.Photo{CameraMake: libregraph.PtrString("Canon"), Iso: libregraph.PtrInt32(400), FNumber: libregraph.PtrFloat64(2.8)}, + Location: &libregraph.GeoCoordinates{ + Longitude: libregraph.PtrFloat64(11.103870357204285), + Latitude: libregraph.PtrFloat64(49.48675890884328), + Altitude: libregraph.PtrFloat64(300.0), + }, + }, + } +} + +func TestMigratePreservesAllFields(t *testing.T) { + root := t.TempDir() + want := fullyPopulated() + + old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t)) + require.NoError(t, err) + require.NoError(t, old.Index(want.ID, want)) + require.NoError(t, old.Close()) + + _, err = bleve.MigrateIndex(root, log.NopLogger()) + require.NoError(t, err) + + idx, _, err := bleve.NewIndex(root) + require.NoError(t, err) + defer func() { _ = idx.Close() }() + + // read the migrated document back through the production reconstruction path + req := bleveSearch.NewSearchRequest(bleveSearch.NewMatchAllQuery()) + req.Fields = []string{"*"} + res, err := idx.Search(req) + require.NoError(t, err) + require.Len(t, res.Hits, 1) + got := searchmapping.Deserialize[search.Resource](res.Hits[0].Fields) + + require.NotNil(t, got.Mtime, "Mtime") + require.True(t, want.Mtime.Equal(*got.Mtime), "Mtime: want %v got %v", want.Mtime, got.Mtime) + got.Mtime = want.Mtime // normalize time.Time location/monotonic + sort.Strings(got.Tags) // bleve may return multi-value fields reordered + sort.Strings(got.Favorites) // + require.Equal(t, want, *got) +} diff --git a/services/search/pkg/bleve/migrate_test.go b/services/search/pkg/bleve/migrate_test.go index dbaf9c3697..0c1ab3489a 100644 --- a/services/search/pkg/bleve/migrate_test.go +++ b/services/search/pkg/bleve/migrate_test.go @@ -7,6 +7,7 @@ import ( bleveSearch "github.com/blevesearch/bleve/v2" bleveMapping "github.com/blevesearch/bleve/v2/mapping" + "github.com/blevesearch/bleve/v2/search/query" libregraph "github.com/opencloud-eu/libre-graph-api-go" "github.com/stretchr/testify/require" @@ -128,23 +129,35 @@ func TestMigrateIndex(t *testing.T) { _, _, err = bleve.NewIndex(root) require.ErrorIs(t, err, searchmapping.ErrManualActionRequired) - // 3) migrate resets the outdated index, dropping its documents + // 3) migrate n, err := bleve.MigrateIndex(root, log.NopLogger()) require.NoError(t, err) require.Equal(t, 3, n) - // 4) the reset index classifies as equal and is empty, awaiting a re-crawl + // 4) the migrated index classifies as equal and holds every document migrated, classification, err := bleve.NewIndex(root) require.NoError(t, err) require.Equal(t, searchmapping.VerdictEqual, classification.Verdict) count, err := migrated.DocCount() require.NoError(t, err) - require.Equal(t, uint64(0), count, "reset drops the documents; a re-crawl repopulates") + require.Equal(t, uint64(3), count) + + // 4a) the trashed document survived (a rescan would have dropped it) + trashed, err := migrated.Document("1$1!3") + require.NoError(t, err) + require.NotNil(t, trashed) + + // 5) location_geopoint was synthesized during migration (absent in the old index) + near := query.NewGeoDistanceQuery(11.103870357204285, 49.48675890884328, "1km") + near.SetField("location" + searchmapping.GeopointSuffix) + res, err := migrated.Search(bleveSearch.NewSearchRequest(near)) + require.NoError(t, err) + require.Equal(t, uint64(3), res.Total, "all three docs match a geo-distance query on the new sibling field") require.NoError(t, migrated.Close()) // release the lock before the second run - // 5) idempotent: a second run is a no-op because the revision is now current + // 6) idempotent: a second run is a no-op because the revision is now current n, err = bleve.MigrateIndex(root, log.NopLogger()) require.NoError(t, err) require.Equal(t, 0, n)