From 7de7ad23bfc45092e90e27f9428af7c5190c24e0 Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 8 Jul 2026 21:17:31 +0200 Subject: [PATCH 01/17] feat(search): check the index schema on startup and refuse breaking changes Both engines now diff the stored/live index schema against the schema generated from code when the service starts. A shared recursive classifier in the mapping package is the single oracle: - equal: start normally. - additive (new fields without any indexed data): applied in place. OpenSearch gets a PUT _mapping with the full code properties, bleve persists the code mapping into the index (SetInternal + reopen) so the new fields are properly typed immediately and later startups classify equal. A startup warning lists the new fields because documents indexed before the upgrade lack them until re-indexed. - breaking (changed definitions or analyzers, removed or renamed fields, or new fields that already contain data of unknown form): refuse to start with an error describing the rebuild procedure (delete the index, start, run "opencloud search index --all-spaces") and the OC_EXCLUDE_RUN_SERVICES=search escape hatch. PUT _mapping is deliberately only the apply mechanism, never the judge: its merge semantics cannot see removals or renames and it accepts in-place updatable param changes with an ack. bleve additionally checks idx.Fields() so previously dynamically indexed data (which leaves no schema trace in bleve) is caught, matching by exact name and by path prefix. While at it: the OpenSearch startup check runs with a real, minute-bounded context instead of context.TODO(), bleve indexes are opened with a 5s bolt_timeout so a second process fails fast instead of hanging on the file lock, and the reversed errors.Is arguments in bleve.NewIndex were fixed. https://github.com/opencloud-eu/opencloud/issues/3092 --- .../change-search-index-schema-handling.md | 22 +++ services/search/pkg/bleve/index.go | 151 ++++++++++++++- services/search/pkg/bleve/index_test.go | 182 ++++++++++++++++++ services/search/pkg/command/server.go | 16 +- services/search/pkg/mapping/bleve.go | 4 +- services/search/pkg/mapping/classify.go | 169 ++++++++++++++++ services/search/pkg/mapping/classify_test.go | 129 +++++++++++++ services/search/pkg/opensearch/backend.go | 9 +- .../search/pkg/opensearch/backend_test.go | 17 +- services/search/pkg/opensearch/index.go | 173 +++++++++-------- services/search/pkg/opensearch/index_test.go | 98 +++++++++- 11 files changed, 869 insertions(+), 101 deletions(-) create mode 100644 changelog/unreleased/change-search-index-schema-handling.md create mode 100644 services/search/pkg/bleve/index_test.go create mode 100644 services/search/pkg/mapping/classify.go create mode 100644 services/search/pkg/mapping/classify_test.go diff --git a/changelog/unreleased/change-search-index-schema-handling.md b/changelog/unreleased/change-search-index-schema-handling.md new file mode 100644 index 0000000000..1177da09ca --- /dev/null +++ b/changelog/unreleased/change-search-index-schema-handling.md @@ -0,0 +1,22 @@ +Change: Check the search index schema on startup (existing indexes need a rebuild) + +The search service now compares the schema of an existing search index with +the schema expected by the code at startup, for both the bleve and the +OpenSearch engine. A purely additive change (new fields that have never been +indexed) is applied in place and the service starts; documents indexed before +the upgrade do not contain the new fields until they are re-indexed. Any other +difference (changed field definitions, changed analyzers, removed or renamed +fields, or new fields that already contain data of unknown form) makes the +service refuse to start instead of silently returning wrong or incomplete +search results. + +Upgrading to this version is a breaking change for BOTH engines: every search +index built by a previous version differs from the new schema and the service +will refuse to start. To rebuild: stop the service, delete the search index +(the bleve directory or the OpenSearch index), start the service (an empty +index with the new schema is created) and run +"opencloud search index --all-spaces" to re-index all files. To bring an +instance up without search until a maintenance window, set +OC_EXCLUDE_RUN_SERVICES=search. + +https://github.com/opencloud-eu/opencloud/issues/3092 diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index f26e3ff809..392bd8bccb 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -1,11 +1,15 @@ package bleve import ( + "encoding/json" "errors" "fmt" + "maps" "math" "path/filepath" "reflect" + "slices" + "strings" "github.com/blevesearch/bleve/v2" "github.com/blevesearch/bleve/v2/analysis/analyzer/custom" @@ -21,23 +25,156 @@ import ( "github.com/opencloud-eu/opencloud/services/search/pkg/search" ) -func NewIndex(root string) (bleve.Index, error) { +// bolt_timeout makes a second process on the same datapath fail after 5s +// instead of blocking forever on the file lock. +var openRuntimeConfig = map[string]interface{}{"bolt_timeout": "5s"} + +// NewIndex opens (or creates) the bleve index at root and classifies the +// stored schema against the one generated from code. On a breaking change it +// refuses with ErrManualActionRequired. On an additive one the code schema is +// persisted into the index (the bleve analogue of an OpenSearch PUT _mapping), +// so the new fields are properly typed from now on and later startups classify +// equal; the caller must still warn that documents indexed before the upgrade +// lack the Classification.NewFields until they are re-indexed. +func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { destination := filepath.Join(root, fmt.Sprintf("bleve-v%d", search.SchemaVersion)) - index, err := bleve.Open(destination) - if errors.Is(bleve.ErrorIndexPathDoesNotExist, err) { + index, err := bleve.OpenUsing(destination, openRuntimeConfig) + if errors.Is(err, bleve.ErrorIndexPathDoesNotExist) { indexMapping, err := NewMapping() if err != nil { - return nil, err + return nil, searchmapping.Classification{}, err } index, err = bleve.New(destination, indexMapping) if err != nil { - return nil, err + return nil, searchmapping.Classification{}, err + } + + return index, searchmapping.Classification{Verdict: searchmapping.VerdictEqual}, nil + } + if err != nil { + return nil, searchmapping.Classification{}, err + } + + classification, codeB, err := classifyStoredMapping(index) + if err != nil { + _ = index.Close() + return nil, searchmapping.Classification{}, err + } + + switch classification.Verdict { + case searchmapping.VerdictBreaking: + _ = index.Close() + return nil, classification, searchmapping.ManualActionRequiredError(destination, classification.Reasons) + case searchmapping.VerdictAdditive: + // The classifier guarantees everything else is identical and the new + // fields hold no data yet, so storing the code mapping only adds + // fields. Reopen so the live mapping picks it up: without this the + // new fields would be indexed dynamically and the data-aware rule + // would turn them breaking on the next startup. + if err := index.SetInternal([]byte("_mapping"), codeB); err != nil { + _ = index.Close() + return nil, searchmapping.Classification{}, fmt.Errorf("failed to store the updated index mapping: %w", err) + } + if err := index.Close(); err != nil { + return nil, searchmapping.Classification{}, err + } + index, err = bleve.OpenUsing(destination, openRuntimeConfig) + if err != nil { + return nil, searchmapping.Classification{}, err + } + } + + return index, classification, nil +} + +// classifyStoredMapping diffs the mapping stored in the index against +// NewMapping() and also returns the marshaled code mapping. Fields that are +// new in the code schema but already have data in the index (previously +// indexed dynamically) are breaking, their de-facto form is unknown. The JSON +// compare is stable within one bleve version; if a bleve upgrade changes +// marshaling defaults it fails towards breaking, normalize the affected key +// here if that ever fires. +func classifyStoredMapping(index bleve.Index) (searchmapping.Classification, []byte, error) { + storedB, err := index.GetInternal([]byte("_mapping")) + if err != nil { + return searchmapping.Classification{}, nil, fmt.Errorf("failed to read the stored index mapping: %w", err) + } + codeMapping, err := NewMapping() + if err != nil { + return searchmapping.Classification{}, nil, err + } + codeB, err := json.Marshal(codeMapping) + if err != nil { + return searchmapping.Classification{}, nil, err + } + + var stored, code map[string]any + if err := json.Unmarshal(storedB, &stored); err != nil { + return searchmapping.Classification{}, nil, fmt.Errorf("failed to parse the stored index mapping: %w", err) + } + if err := json.Unmarshal(codeB, &code); err != nil { + return searchmapping.Classification{}, nil, err + } + + fields, err := index.Fields() + if err != nil { + return searchmapping.Classification{}, nil, fmt.Errorf("failed to list the indexed fields: %w", err) + } + indexedFields := make(map[string]struct{}, len(fields)) + for _, f := range fields { + if !strings.HasPrefix(f, "_") { // skip bleve-internal fields like _all + indexedFields[f] = struct{}{} + } + } + + storedDM, _ := stored["default_mapping"].(map[string]any) + codeDM, _ := code["default_mapping"].(map[string]any) + storedProps, _ := storedDM["properties"].(map[string]any) + codeProps, _ := codeDM["properties"].(map[string]any) + + classification := searchmapping.Classify(storedProps, codeProps, func(path string) bool { + if _, ok := indexedFields[path]; ok { + return true } + nested := path + "." + for f := range indexedFields { + if strings.HasPrefix(f, nested) { + return true + } + } + return false + }) - return index, nil + // everything outside default_mapping.properties (analyzer definitions, + // default analyzer, dynamic flags, ...) must match exactly + var reasons []string + compareKeysExcept(stored, code, "default_mapping", "", &reasons) + compareKeysExcept(storedDM, codeDM, "properties", "default_mapping.", &reasons) + if len(reasons) > 0 { + classification.Verdict = searchmapping.VerdictBreaking + classification.Reasons = append(reasons, classification.Reasons...) } - return index, err + return classification, codeB, nil +} + +// compareKeysExcept deep-compares all keys present on either side except skip. +func compareKeysExcept(stored, code map[string]any, skip, prefix string, reasons *[]string) { + keys := slices.Collect(maps.Keys(stored)) + for k := range code { + if _, ok := stored[k]; !ok { + keys = append(keys, k) + } + } + slices.Sort(keys) + for _, k := range keys { + if k == skip { + continue + } + if !reflect.DeepEqual(stored[k], code[k]) { + *reasons = append(*reasons, fmt.Sprintf("%s%s changed", prefix, k)) + } + } } func NewMapping() (mapping.IndexMapping, error) { diff --git a/services/search/pkg/bleve/index_test.go b/services/search/pkg/bleve/index_test.go new file mode 100644 index 0000000000..32d285d9f0 --- /dev/null +++ b/services/search/pkg/bleve/index_test.go @@ -0,0 +1,182 @@ +package bleve_test + +import ( + "fmt" + "path/filepath" + + bleveSearch "github.com/blevesearch/bleve/v2" + "github.com/blevesearch/bleve/v2/analysis/analyzer/custom" + "github.com/blevesearch/bleve/v2/analysis/token/lowercase" + "github.com/blevesearch/bleve/v2/analysis/tokenizer/unicode" + bleveMapping "github.com/blevesearch/bleve/v2/mapping" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" + searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +var _ = Describe("NewIndex", func() { + var root string + + BeforeEach(func() { + root = GinkgoT().TempDir() + }) + + codeMapping := func() *bleveMapping.IndexMappingImpl { + m, err := bleve.NewMapping() + Expect(err).ToNot(HaveOccurred()) + impl, ok := m.(*bleveMapping.IndexMappingImpl) + Expect(ok).To(BeTrue()) + return impl + } + + // buildIndex simulates an index left behind by an older release + buildIndex := func(m bleveMapping.IndexMapping, docs map[string]map[string]any) { + idx, err := bleveSearch.New(filepath.Join(root, fmt.Sprintf("bleve-v%d", search.SchemaVersion)), m) + Expect(err).ToNot(HaveOccurred()) + for id, doc := range docs { + Expect(idx.Index(id, doc)).To(Succeed()) + } + Expect(idx.Close()).To(Succeed()) + } + + It("creates a fresh index", func() { + idx, classification, err := bleve.NewIndex(root) + Expect(err).ToNot(HaveOccurred()) + Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual)) + Expect(idx.Close()).To(Succeed()) + }) + + It("opens an index with an identical schema", func() { + buildIndex(codeMapping(), nil) + + idx, classification, err := bleve.NewIndex(root) + Expect(err).ToNot(HaveOccurred()) + Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual)) + Expect(classification.NewFields).To(BeEmpty()) + Expect(idx.Close()).To(Succeed()) + }) + + It("treats a genuinely new field as additive", func() { + old := codeMapping() + Expect(old.DefaultMapping.Properties).To(HaveKey("Title")) + delete(old.DefaultMapping.Properties, "Title") + buildIndex(old, nil) + + idx, classification, err := bleve.NewIndex(root) + Expect(err).ToNot(HaveOccurred()) + Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive)) + Expect(classification.NewFields).To(ConsistOf("Title")) + Expect(idx.Index("1", map[string]any{"Title": "hello"})).To(Succeed()) + Expect(idx.Close()).To(Succeed()) + }) + + It("treats a new nested field as additive", func() { + old := codeMapping() + photo := old.DefaultMapping.Properties["photo"] + Expect(photo).ToNot(BeNil()) + Expect(photo.Properties).To(HaveKey("cameraMake")) + delete(photo.Properties, "cameraMake") + buildIndex(old, nil) + + idx, classification, err := bleve.NewIndex(root) + Expect(err).ToNot(HaveOccurred()) + Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive)) + Expect(classification.NewFields).To(ConsistOf("photo.cameraMake")) + Expect(idx.Close()).To(Succeed()) + }) + + It("persists an additive schema change so later startups classify it as equal", func() { + old := codeMapping() + delete(old.DefaultMapping.Properties, "Title") + buildIndex(old, nil) + + idx, classification, err := bleve.NewIndex(root) + Expect(err).ToNot(HaveOccurred()) + Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive)) + Expect(idx.Index("1", map[string]any{"Title": "hello"})).To(Succeed()) + Expect(idx.Close()).To(Succeed()) + + idx, classification, err = bleve.NewIndex(root) + Expect(err).ToNot(HaveOccurred()) + Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual)) + Expect(idx.Close()).To(Succeed()) + }) + + It("refuses when a new field already has data in the index", func() { + old := codeMapping() + Expect(old.DefaultMapping.Properties).To(HaveKey("Mtime")) + delete(old.DefaultMapping.Properties, "Mtime") + buildIndex(old, map[string]map[string]any{"1": {"Mtime": "2026-01-02T03:04:05Z"}}) + + idx, _, err := bleve.NewIndex(root) + Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) + Expect(idx).To(BeNil()) + }) + + It("refuses when a new object field already has nested data in the index", func() { + old := codeMapping() + Expect(old.DefaultMapping.Properties).To(HaveKey("photo")) + delete(old.DefaultMapping.Properties, "photo") + buildIndex(old, map[string]map[string]any{"1": {"photo": map[string]any{"cameraMake": "ACME"}}}) + + _, _, err := bleve.NewIndex(root) + Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) + }) + + It("refuses on a changed field definition", func() { + old := codeMapping() + name := old.DefaultMapping.Properties["Name"] + Expect(name).ToNot(BeNil()) + Expect(name.Fields).ToNot(BeEmpty()) + name.Fields[0].Analyzer = "fulltext" + buildIndex(old, nil) + + _, _, err := bleve.NewIndex(root) + Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) + }) + + It("refuses when a stored field was removed from the code schema", func() { + old := codeMapping() + old.DefaultMapping.AddFieldMappingsAt("Legacy", bleveSearch.NewTextFieldMapping()) + buildIndex(old, nil) + + _, _, err := bleve.NewIndex(root) + Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) + }) + + It("refuses when a default_mapping attribute changed", func() { + old := codeMapping() + old.DefaultMapping.Dynamic = false + buildIndex(old, nil) + + _, _, err := bleve.NewIndex(root) + Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) + }) + + It("refuses on a changed analyzer definition", func() { + old := codeMapping() + Expect(old.CustomAnalysis.Analyzers).To(HaveKey("fulltext")) + old.CustomAnalysis.Analyzers["fulltext"] = map[string]any{ + "type": custom.Name, + "tokenizer": unicode.Name, + "token_filters": []string{lowercase.Name}, + } + buildIndex(old, nil) + + _, _, err := bleve.NewIndex(root) + Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) + }) +}) + +var _ = Describe("NewMapping", func() { + It("only references registered analyzers", func() { + m, err := bleve.NewMapping() + Expect(err).ToNot(HaveOccurred()) + impl, ok := m.(*bleveMapping.IndexMappingImpl) + Expect(ok).To(BeTrue()) + Expect(impl.Validate()).To(Succeed()) + }) +}) diff --git a/services/search/pkg/command/server.go b/services/search/pkg/command/server.go index 62a1a6795c..2459bc4105 100644 --- a/services/search/pkg/command/server.go +++ b/services/search/pkg/command/server.go @@ -7,6 +7,7 @@ import ( "net/http" "os" "os/signal" + "time" "github.com/opencloud-eu/opencloud/pkg/config/configlog" "github.com/opencloud-eu/opencloud/pkg/generators" @@ -20,6 +21,7 @@ import ( "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/content" + searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" "github.com/opencloud-eu/opencloud/services/search/pkg/metrics" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" bleveQuery "github.com/opencloud-eu/opencloud/services/search/pkg/query/bleve" @@ -71,11 +73,17 @@ func Server(cfg *config.Config) *cobra.Command { var eng search.Engine switch cfg.Engine.Type { case "bleve": - idx, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath) + idx, classification, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath) if err != nil { return err } + if classification.Verdict == searchmapping.VerdictAdditive { + logger.Warn(). + Strs("fields", classification.NewFields). + Msgf("the bleve index at %s was built with an older schema; the new fields were added to the index schema, but documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces", cfg.Engine.Bleve.Datapath) + } + defer func() { if err = idx.Close(); err != nil { logger.Error().Err(err).Msg("could not close bleve index") @@ -119,8 +127,12 @@ func Server(cfg *config.Config) *cobra.Command { return fmt.Errorf("failed to create OpenSearch client: %w", err) } + // bound the startup schema check so a hung cluster fails the + // start instead of blocking forever + startupCtx, cancelStartup := context.WithTimeout(ctx, time.Minute) indexName := opensearch.VersionedIndexName(cfg.Engine.OpenSearch.ResourceIndex.Name) - openSearchBackend, err := opensearch.NewBackend(indexName, client) + 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/mapping/bleve.go b/services/search/pkg/mapping/bleve.go index f357e4f305..ba4f2ec540 100644 --- a/services/search/pkg/mapping/bleve.go +++ b/services/search/pkg/mapping/bleve.go @@ -14,8 +14,8 @@ import ( // // The returned mapping references analyzer names (Analyzer field on the // FieldOpts, plus "fulltext" / "path_hierarchy" for the corresponding Types); -// the caller is responsible for registering those analyzers on the enclosing -// IndexMapping. +// the caller must register every referenced analyzer on the enclosing +// IndexMapping (IndexMapping.Validate catches missing ones). func BleveBuildMapping(t reflect.Type, overrides map[string]FieldOpts) (*bleveMapping.DocumentMapping, error) { return buildBleveDocMapping(t, overrides, "") } diff --git a/services/search/pkg/mapping/classify.go b/services/search/pkg/mapping/classify.go new file mode 100644 index 0000000000..9fa357ef7c --- /dev/null +++ b/services/search/pkg/mapping/classify.go @@ -0,0 +1,169 @@ +package mapping + +import ( + "encoding/json" + "errors" + "fmt" + "maps" + "reflect" + "slices" + "strings" +) + +// ErrManualActionRequired marks schema changes that cannot be applied in place. +var ErrManualActionRequired = errors.New("manual action required") + +// ManualActionRequiredError builds the operator-facing error for a breaking +// schema change. index is the index name (OpenSearch) or path (bleve). +func ManualActionRequiredError(index string, reasons []string) error { + return fmt.Errorf( + "%w: search index %s was built with a different schema (%s). "+ + "There is no in-place migration: stop the service, delete %s, "+ + "start the service (an empty index with the new schema is created), "+ + "then rebuild the content by running: opencloud search index --all-spaces. "+ + "To bring the instance up without search until a maintenance window, "+ + "set OC_EXCLUDE_RUN_SERVICES=search", + ErrManualActionRequired, index, strings.Join(reasons, "; "), index, + ) +} + +type Verdict string + +const ( + VerdictEqual Verdict = "equal" + VerdictAdditive Verdict = "additive" + VerdictBreaking Verdict = "breaking" +) + +// Classification is the outcome of diffing a stored index schema against the +// schema generated from code. +type Classification struct { + Verdict Verdict + // NewFields are dotted paths of fields that only exist in the code schema. + NewFields []string + // Reasons are human-readable breaking differences. + Reasons []string +} + +// Classify recursively compares a stored `properties` tree against the one +// generated from code. Both sides must be generic JSON-decoded values +// (map[string]any, []any, float64), not marshaled Go structs, so values +// compare structurally. +// +// dataFields reports whether the index holds data at or below a dotted field +// path even though it is absent from the stored schema (bleve indexes dynamic +// fields without a schema trace). Engines that record dynamic fields in the +// live schema (OpenSearch) pass nil. +func Classify(stored, code map[string]any, dataFields func(path string) bool) Classification { + c := Classification{Verdict: VerdictEqual} + classifyProperties(stored, code, dataFields, "", &c) + if c.Verdict == VerdictEqual && len(c.NewFields) > 0 { + c.Verdict = VerdictAdditive + } + return c +} + +func classifyProperties(stored, code map[string]any, dataFields func(string) bool, prefix string, c *Classification) { + for _, k := range slices.Sorted(maps.Keys(stored)) { + path := joinPath(prefix, k) + codeNode, ok := code[k] + if !ok { + c.breaking(fmt.Sprintf("field %s exists in the index but not in the code schema (removed or renamed)", path)) + continue + } + classifyNode(stored[k], codeNode, dataFields, path, c) + } + + for _, k := range slices.Sorted(maps.Keys(code)) { + if _, ok := stored[k]; ok { + continue + } + path := joinPath(prefix, k) + if dataFields != nil && dataFields(path) { + c.breaking(fmt.Sprintf("field %s is new in the code schema but the index already contains data for it (previously indexed dynamically)", path)) + continue + } + c.NewFields = append(c.NewFields, leafPaths(code[k], path)...) + } +} + +func classifyNode(stored, code any, dataFields func(string) bool, path string, c *Classification) { + storedMap, sOK := stored.(map[string]any) + codeMap, cOK := code.(map[string]any) + if !sOK || !cOK { + if !reflect.DeepEqual(stored, code) { + c.breaking(fmt.Sprintf("field %s changed: index %s, code %s", path, compactJSON(stored), compactJSON(code))) + } + return + } + + for _, k := range sortedUnionKeys(storedMap, codeMap) { + if k == "properties" { + continue + } + sv, sHas := storedMap[k] + cv, cHas := codeMap[k] + if sHas && cHas && reflect.DeepEqual(sv, cv) { + continue + } + c.breaking(fmt.Sprintf("field %s: %s changed: index %s, code %s", path, k, optJSON(sv, sHas), optJSON(cv, cHas))) + } + + storedProps, _ := storedMap["properties"].(map[string]any) + codeProps, _ := codeMap["properties"].(map[string]any) + if len(storedProps) > 0 || len(codeProps) > 0 { + classifyProperties(storedProps, codeProps, dataFields, path, c) + } +} + +// leafPaths lists the dotted paths of all leaf fields at or below node. +func leafPaths(node any, path string) []string { + if nodeMap, ok := node.(map[string]any); ok { + if props, ok := nodeMap["properties"].(map[string]any); ok && len(props) > 0 { + var leaves []string + for _, k := range slices.Sorted(maps.Keys(props)) { + leaves = append(leaves, leafPaths(props[k], path+"."+k)...) + } + return leaves + } + } + return []string{path} +} + +func (c *Classification) breaking(reason string) { + c.Verdict = VerdictBreaking + c.Reasons = append(c.Reasons, reason) +} + +func joinPath(prefix, k string) string { + if prefix == "" { + return k + } + return prefix + "." + k +} + +func sortedUnionKeys(a, b map[string]any) []string { + keys := slices.Collect(maps.Keys(a)) + for k := range b { + if _, ok := a[k]; !ok { + keys = append(keys, k) + } + } + slices.Sort(keys) + return keys +} + +func optJSON(v any, present bool) string { + if !present { + return "(unset)" + } + return compactJSON(v) +} + +func compactJSON(v any) string { + b, err := json.Marshal(v) + if err != nil { + return fmt.Sprintf("%v", v) + } + return string(b) +} diff --git a/services/search/pkg/mapping/classify_test.go b/services/search/pkg/mapping/classify_test.go new file mode 100644 index 0000000000..37ea5fbd1b --- /dev/null +++ b/services/search/pkg/mapping/classify_test.go @@ -0,0 +1,129 @@ +package mapping + +import ( + "encoding/json" + "slices" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Classify", func() { + code := `{ + "Name": {"type": "keyword"}, + "Size": {"type": "long"}, + "photo": {"properties": {"cameraMake": {"type": "keyword"}, "cameraModel": {"type": "keyword"}}} + }` + + parse := func(s string) map[string]any { + var m map[string]any + Expect(json.Unmarshal([]byte(s), &m)).To(Succeed()) + return m + } + + hasData := func(fields ...string) func(string) bool { + return func(path string) bool { return slices.Contains(fields, path) } + } + + It("classifies identical schemas as equal", func() { + c := Classify(parse(code), parse(code), nil) + Expect(c.Verdict).To(Equal(VerdictEqual)) + Expect(c.NewFields).To(BeEmpty()) + Expect(c.Reasons).To(BeEmpty()) + }) + + It("classifies a new top-level field as additive", func() { + stored := parse(code) + delete(stored, "Size") + + c := Classify(stored, parse(code), nil) + Expect(c.Verdict).To(Equal(VerdictAdditive)) + Expect(c.NewFields).To(ConsistOf("Size")) + Expect(c.Reasons).To(BeEmpty()) + }) + + It("classifies a new nested field as additive", func() { + stored := parse(code) + delete(stored["photo"].(map[string]any)["properties"].(map[string]any), "cameraMake") + + c := Classify(stored, parse(code), nil) + Expect(c.Verdict).To(Equal(VerdictAdditive)) + Expect(c.NewFields).To(ConsistOf("photo.cameraMake")) + }) + + It("lists every leaf of a new subtree", func() { + stored := parse(code) + delete(stored, "photo") + + c := Classify(stored, parse(code), nil) + Expect(c.Verdict).To(Equal(VerdictAdditive)) + Expect(c.NewFields).To(ConsistOf("photo.cameraMake", "photo.cameraModel")) + }) + + It("breaks when a new field already has data in the index", func() { + stored := parse(code) + delete(stored, "Size") + + c := Classify(stored, parse(code), hasData("Size")) + Expect(c.Verdict).To(Equal(VerdictBreaking)) + Expect(c.Reasons).To(ConsistOf(ContainSubstring("Size"))) + }) + + It("breaks when a new nested field already has data in the index", func() { + stored := parse(code) + delete(stored["photo"].(map[string]any)["properties"].(map[string]any), "cameraMake") + + c := Classify(stored, parse(code), hasData("photo.cameraMake")) + Expect(c.Verdict).To(Equal(VerdictBreaking)) + Expect(c.Reasons).To(ConsistOf(ContainSubstring("photo.cameraMake"))) + }) + + It("breaks when a new subtree already has data below it", func() { + stored := parse(code) + delete(stored, "photo") + + // the callback is consulted with the subtree root; reporting data at + // or below it is the caller's job + c := Classify(stored, parse(code), hasData("photo")) + Expect(c.Verdict).To(Equal(VerdictBreaking)) + Expect(c.Reasons).To(ConsistOf(ContainSubstring("photo"))) + }) + + It("breaks on a changed field definition", func() { + stored := parse(code) + stored["Size"].(map[string]any)["type"] = "keyword" + + c := Classify(stored, parse(code), nil) + Expect(c.Verdict).To(Equal(VerdictBreaking)) + Expect(c.Reasons).To(ConsistOf(ContainSubstring("Size"))) + }) + + It("breaks on a field that was removed from the code schema", func() { + reduced := parse(code) + delete(reduced, "Size") + + c := Classify(parse(code), reduced, nil) + Expect(c.Verdict).To(Equal(VerdictBreaking)) + Expect(c.Reasons).To(ConsistOf(ContainSubstring("removed or renamed"))) + }) + + It("breaks on a changed object attribute", func() { + stored := parse(code) + stored["photo"].(map[string]any)["dynamic"] = true + + c := Classify(stored, parse(code), nil) + Expect(c.Verdict).To(Equal(VerdictBreaking)) + Expect(c.Reasons).To(ConsistOf(ContainSubstring("dynamic"))) + }) + + It("lets breaking win over additive", func() { + stored := parse(code) + delete(stored, "Size") + stored["Name"].(map[string]any)["type"] = "text" + + c := Classify(stored, parse(code), nil) + Expect(c.Verdict).To(Equal(VerdictBreaking)) + Expect(c.NewFields).To(ConsistOf("Size")) + Expect(c.Reasons).To(ConsistOf(ContainSubstring("Name"))) + }) +}) diff --git a/services/search/pkg/opensearch/backend.go b/services/search/pkg/opensearch/backend.go index 1f27c136bf..a7f28bbdfa 100644 --- a/services/search/pkg/opensearch/backend.go +++ b/services/search/pkg/opensearch/backend.go @@ -13,6 +13,7 @@ import ( "github.com/opencloud-eu/reva/v2/pkg/utils" "github.com/opencloud-eu/opencloud/pkg/conversions" + "github.com/opencloud-eu/opencloud/pkg/log" searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0" searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/convert" @@ -31,8 +32,8 @@ type Backend struct { client *opensearchgoAPI.Client } -func NewBackend(index string, client *opensearchgoAPI.Client) (*Backend, error) { - pingResp, err := client.Ping(context.TODO(), &opensearchgoAPI.PingReq{}) +func NewBackend(ctx context.Context, index string, client *opensearchgoAPI.Client, logger log.Logger) (*Backend, error) { + pingResp, err := client.Ping(ctx, &opensearchgoAPI.PingReq{}) switch { case err != nil: return nil, fmt.Errorf("%w, failed to ping opensearch: %w", ErrUnhealthyCluster, err) @@ -41,13 +42,13 @@ func NewBackend(index string, client *opensearchgoAPI.Client) (*Backend, error) } // apply the index template - if err := IndexManagerLatest.Apply(context.TODO(), index, client); err != nil { + if err := IndexManagerLatest.Apply(ctx, index, client, logger); err != nil { return nil, fmt.Errorf("failed to apply index template: %w", err) } // first check if the cluster is healthy - resp, err := client.Cluster.Health(context.TODO(), &opensearchgoAPI.ClusterHealthReq{ + resp, err := client.Cluster.Health(ctx, &opensearchgoAPI.ClusterHealthReq{ Indices: []string{index}, Params: opensearchgoAPI.ClusterHealthParams{ Local: opensearchgoAPI.ToPointer(true), diff --git a/services/search/pkg/opensearch/backend_test.go b/services/search/pkg/opensearch/backend_test.go index 8422dd10ad..2fb55f03b6 100644 --- a/services/search/pkg/opensearch/backend_test.go +++ b/services/search/pkg/opensearch/backend_test.go @@ -9,6 +9,7 @@ import ( opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi" "github.com/stretchr/testify/require" + "github.com/opencloud-eu/opencloud/pkg/log" searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test" @@ -24,7 +25,7 @@ func TestNewBackend(t *testing.T) { }) require.NoError(t, err, "failed to create OpenSearch client") - backend, err := opensearch.NewBackend("test-engine-new-engine", client) + backend, err := opensearch.NewBackend(t.Context(), "test-engine-new-engine", client, log.NopLogger()) require.Nil(t, backend) require.ErrorIs(t, err, opensearch.ErrUnhealthyCluster) }) @@ -38,7 +39,7 @@ func TestEngine_Search(t *testing.T) { defer tc.Require.IndicesDelete([]string{indexName}) - backend, err := opensearch.NewBackend(indexName, tc.Client()) + backend, err := opensearch.NewBackend(t.Context(), indexName, tc.Client(), log.NopLogger()) require.NoError(t, err) document := opensearchtest.Testdata.Resources.File @@ -81,7 +82,7 @@ func TestEngine_Upsert(t *testing.T) { defer tc.Require.IndicesDelete([]string{indexName}) - backend, err := opensearch.NewBackend(indexName, tc.Client()) + backend, err := opensearch.NewBackend(t.Context(), indexName, tc.Client(), log.NopLogger()) require.NoError(t, err) t.Run("upsert with full document", func(t *testing.T) { @@ -110,7 +111,7 @@ func TestEngine_Move(t *testing.T) { defer tc.Require.IndicesDelete([]string{indexName}) - backend, err := opensearch.NewBackend(indexName, tc.Client()) + backend, err := opensearch.NewBackend(t.Context(), indexName, tc.Client(), log.NopLogger()) require.NoError(t, err) t.Run("moves the document to a new path", func(t *testing.T) { @@ -147,7 +148,7 @@ func TestEngine_Delete(t *testing.T) { defer tc.Require.IndicesDelete([]string{indexName}) - backend, err := opensearch.NewBackend(indexName, tc.Client()) + backend, err := opensearch.NewBackend(t.Context(), indexName, tc.Client(), log.NopLogger()) require.NoError(t, err) t.Run("mark document as deleted", func(t *testing.T) { @@ -180,7 +181,7 @@ func TestEngine_Restore(t *testing.T) { defer tc.Require.IndicesDelete([]string{indexName}) - backend, err := opensearch.NewBackend(indexName, tc.Client()) + backend, err := opensearch.NewBackend(t.Context(), indexName, tc.Client(), log.NopLogger()) require.NoError(t, err) t.Run("mark document as not deleted", func(t *testing.T) { @@ -214,7 +215,7 @@ func TestEngine_Purge(t *testing.T) { defer tc.Require.IndicesDelete([]string{indexName}) - backend, err := opensearch.NewBackend(indexName, tc.Client()) + backend, err := opensearch.NewBackend(t.Context(), indexName, tc.Client(), log.NopLogger()) require.NoError(t, err) t.Run("purge with full document", func(t *testing.T) { @@ -266,7 +267,7 @@ func TestEngine_DocCount(t *testing.T) { defer tc.Require.IndicesDelete([]string{indexName}) - backend, err := opensearch.NewBackend(indexName, tc.Client()) + backend, err := opensearch.NewBackend(t.Context(), indexName, tc.Client(), log.NopLogger()) require.NoError(t, err) t.Run("ignore deleted documents", func(t *testing.T) { diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index 18bc0eb99b..d4896599a8 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -3,21 +3,25 @@ package opensearch import ( "bytes" "context" + "encoding/json" "errors" "fmt" "maps" "reflect" + "strings" - "github.com/go-jose/go-jose/v3/json" + opensearchgo "github.com/opensearch-project/opensearch-go/v4" opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi" "github.com/tidwall/gjson" + "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" ) var ( - ErrManualActionRequired = errors.New("manual action required") + // ErrManualActionRequired is the shared sentinel, see the mapping package. + ErrManualActionRequired = searchmapping.ErrManualActionRequired // IndexManagerLatest identifies the current resource mapping; its version is // derived from search.SchemaVersion so it never drifts from the index name. @@ -99,99 +103,118 @@ func buildResourceMapping() ([]byte, error) { return json.Marshal(index) } -func (m IndexManager) Apply(ctx context.Context, name string, client *opensearchgoAPI.Client) error { +// Apply ensures the index exists and matches the schema generated from code. +// A missing index is created, an additive schema change is applied in place +// via PUT _mapping, a breaking one returns ErrManualActionRequired. The +// classifier decides what is additive; PUT _mapping is only the mechanism to +// apply it (its merge semantics cannot detect removals or renames). +func (m IndexManager) Apply(ctx context.Context, name string, client *opensearchgoAPI.Client, logger log.Logger) error { localIndexB, err := m.MarshalJSON() if err != nil { return fmt.Errorf("failed to marshal index %s: %w", name, err) } - indicesExistsResp, err := client.Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{ - Indices: []string{name}, + createResp, err := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{ + Index: name, + Body: bytes.NewReader(localIndexB), }) + var createErr *opensearchgo.StructError switch { - case indicesExistsResp != nil && indicesExistsResp.StatusCode == 404: - break - case err != nil: - return fmt.Errorf("failed to check if index %s exists: %w", name, err) - case indicesExistsResp == nil: - return fmt.Errorf("indicesExistsResp is nil for index %s", name) + case err == nil && createResp.Acknowledged: + return nil + case err == nil: + return fmt.Errorf("failed to create index %s: not acknowledged", name) + case !errors.As(err, &createErr) || createErr.Err.Type != "resource_already_exists_exception": + // transport errors, disk-full etc. stay plain fatal, the restart policy retries + return fmt.Errorf("failed to create index %s: %w", name, err) } - if indicesExistsResp.StatusCode == 200 { - resp, err := client.Indices.Get(ctx, opensearchgoAPI.IndicesGetReq{ - Indices: []string{name}, - }) - if err != nil { - return fmt.Errorf("failed to get index %s: %w", name, err) - } - - remoteIndex, ok := resp.Indices[name] - if !ok { - return fmt.Errorf("index %s not found in response", name) - } - remoteIndexB, err := json.Marshal(remoteIndex) - if err != nil { - return fmt.Errorf("failed to marshal index %s: %w", name, err) - } - - localIndexJson := gjson.ParseBytes(localIndexB) - remoteIndexJson := gjson.ParseBytes(remoteIndexB) - - compare := func(lvPath, rvPath string) (any, any, bool) { - lv := localIndexJson.Get(lvPath).Raw - rv := remoteIndexJson.Get(rvPath).Raw - - var lvv, rvv any - if err := json.Unmarshal([]byte(lv), &lvv); err != nil { - return nil, nil, false - } - - if err := json.Unmarshal([]byte(rv), &rvv); err != nil { - return nil, nil, false - } - - return lv, rv, reflect.DeepEqual(lvv, rvv) - } - - var errs []error + // the index already exists: compare settings and classify the mapping diff + resp, err := client.Indices.Get(ctx, opensearchgoAPI.IndicesGetReq{ + Indices: []string{name}, + }) + if err != nil { + return fmt.Errorf("failed to get index %s: %w", name, err) + } - for k := range localIndexJson.Get("settings").Map() { - if lv, rv, ok := compare("settings."+k, "settings.index."+k); !ok { - errs = append(errs, fmt.Errorf("settings.%s local %s, remote %s", k, lv, rv)) - } - } + remoteIndex, ok := resp.Indices[name] + if !ok { + return fmt.Errorf("index %s not found in response", name) + } + remoteIndexB, err := json.Marshal(remoteIndex) + if err != nil { + return fmt.Errorf("failed to marshal index %s: %w", name, err) + } - for k := range localIndexJson.Get("mappings.properties").Map() { - if _, _, ok := compare("mappings.properties."+k, "mappings.properties."+k); !ok { - errs = append(errs, fmt.Errorf("mappings.properties.%s", k)) - } - } + localIndexJson := gjson.ParseBytes(localIndexB) + remoteIndexJson := gjson.ParseBytes(remoteIndexB) - if errs != nil { - return fmt.Errorf( - "index %s already exists with a different mapping than the requested version. "+ - "There is no in-place migration today: drop the index in OpenSearch (DELETE /%s) "+ - "and restart the search service. The index will be recreated with the new mapping. "+ - "%w: %w", - name, name, - ErrManualActionRequired, - errors.Join(errs...), - ) + var reasons []string + for k := range localIndexJson.Get("settings").Map() { + lv := localIndexJson.Get("settings." + k).Raw + rv := remoteIndexJson.Get("settings.index." + k).Raw + if !jsonEqual(lv, rv) { + reasons = append(reasons, fmt.Sprintf("settings.%s changed: index %s, code %s", k, rawOrUnset(rv), rawOrUnset(lv))) } + } - return nil // Index is already up to date, no action needed + classification := searchmapping.Classify( + propertiesMap(remoteIndexJson.Get("mappings.properties").Raw), + propertiesMap(localIndexJson.Get("mappings.properties").Raw), + nil, + ) + reasons = append(reasons, classification.Reasons...) + if len(reasons) > 0 { + return searchmapping.ManualActionRequiredError(name, reasons) + } + if len(classification.NewFields) == 0 { + return nil // schema is up to date } - createResp, err := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{ - Index: name, - Body: bytes.NewReader(localIndexB), + // additive: the classifier guarantees every existing field matches the + // remote state, so putting the full code properties can only add fields + putResp, err := client.Indices.Mapping.Put(ctx, opensearchgoAPI.MappingPutReq{ + Indices: []string{name}, + Body: strings.NewReader(localIndexJson.Get("mappings").Raw), }) + var putErr *opensearchgo.StructError switch { + case err != nil && errors.As(err, &putErr) && putErr.Err.Type == "illegal_argument_exception" && + (strings.Contains(putErr.Err.Reason, "cannot be changed") || strings.Contains(putErr.Err.Reason, "Cannot update parameter")): + // backstop, should be unreachable after the classification above + return searchmapping.ManualActionRequiredError(name, []string{putErr.Err.Reason}) case err != nil: - return fmt.Errorf("failed to create index %s: %w", name, err) - case !createResp.Acknowledged: - return fmt.Errorf("failed to create index %s: not acknowledged", name) + return fmt.Errorf("failed to update mapping of index %s: %w", name, err) + case !putResp.Acknowledged: + return fmt.Errorf("failed to update mapping of index %s: not acknowledged", name) } + logger.Info().Strs("fields", classification.NewFields).Str("index", name).Msg("extended the search index mapping with new fields") return nil } + +func jsonEqual(a, b string) bool { + var av, bv any + if err := json.Unmarshal([]byte(a), &av); err != nil { + return false + } + if err := json.Unmarshal([]byte(b), &bv); err != nil { + return false + } + return reflect.DeepEqual(av, bv) +} + +// propertiesMap parses a raw mappings.properties object; missing or empty +// input yields an empty map, which classifies as purely additive. +func propertiesMap(raw string) map[string]any { + props := map[string]any{} + _ = json.Unmarshal([]byte(raw), &props) + return props +} + +func rawOrUnset(raw string) string { + if raw == "" { + return "(unset)" + } + return raw +} diff --git a/services/search/pkg/opensearch/index_test.go b/services/search/pkg/opensearch/index_test.go index cdbc01ac85..34d267e288 100644 --- a/services/search/pkg/opensearch/index_test.go +++ b/services/search/pkg/opensearch/index_test.go @@ -5,9 +5,13 @@ import ( "strings" "testing" + opensearchgo "github.com/opensearch-project/opensearch-go/v4" + opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi" "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" "github.com/tidwall/sjson" + "github.com/opencloud-eu/opencloud/pkg/log" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test" "github.com/opencloud-eu/opencloud/services/search/pkg/search" @@ -46,7 +50,7 @@ func TestIndexManager(t *testing.T) { require.NotEmpty(t, body) require.NotEmpty(t, test.Got.String()) require.JSONEq(t, test.Got.String(), string(body)) - require.NoError(t, test.Got.Apply(t.Context(), indexName, tc.Client())) + require.NoError(t, test.Got.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) }) } }) @@ -59,7 +63,7 @@ func TestIndexManager(t *testing.T) { tc.Require.IndicesReset([]string{indexName}) tc.Require.IndicesCreate(indexName, strings.NewReader(indexManager.String())) - require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client())) + 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) { @@ -73,6 +77,94 @@ func TestIndexManager(t *testing.T) { require.NoError(t, err) tc.Require.IndicesCreate(indexName, strings.NewReader(body)) - require.ErrorIs(t, indexManager.Apply(t.Context(), indexName, tc.Client()), opensearch.ErrManualActionRequired) + require.ErrorIs(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger()), opensearch.ErrManualActionRequired) + }) + + t.Run("is idempotent", func(t *testing.T) { + indexManager := opensearch.IndexManagerLatest + indexName := "opencloud-test-resource" + + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + tc.Require.IndicesReset([]string{indexName}) + + require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) + require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) + }) + + t.Run("adds a new field to an existing index in place", func(t *testing.T) { + indexManager := opensearch.IndexManagerLatest + indexName := "opencloud-test-resource" + + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + tc.Require.IndicesReset([]string{indexName}) + + body, err := sjson.Delete(indexManager.String(), "mappings.properties.Title") + require.NoError(t, err) + tc.Require.IndicesCreate(indexName, strings.NewReader(body)) + + require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) + + resp, err := tc.Client().Indices.Mapping.Get(t.Context(), &opensearchgoAPI.MappingGetReq{Indices: []string{indexName}}) + require.NoError(t, err) + require.True(t, gjson.GetBytes(resp.Indices[indexName].Mappings, "properties.Title").Exists()) + }) + + t.Run("adds a new nested field to an existing index in place", func(t *testing.T) { + indexManager := opensearch.IndexManagerLatest + indexName := "opencloud-test-resource" + + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + tc.Require.IndicesReset([]string{indexName}) + + body, err := sjson.Delete(indexManager.String(), "mappings.properties.photo.properties.cameraMake") + require.NoError(t, err) + tc.Require.IndicesCreate(indexName, strings.NewReader(body)) + + require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) + + resp, err := tc.Client().Indices.Mapping.Get(t.Context(), &opensearchgoAPI.MappingGetReq{Indices: []string{indexName}}) + require.NoError(t, err) + require.True(t, gjson.GetBytes(resp.Indices[indexName].Mappings, "properties.photo.properties.cameraMake").Exists()) + }) + + t.Run("fails when an existing field changed its definition", func(t *testing.T) { + indexManager := opensearch.IndexManagerLatest + indexName := "opencloud-test-resource" + + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + tc.Require.IndicesReset([]string{indexName}) + + body, err := sjson.Set(indexManager.String(), "mappings.properties.Deleted.type", "keyword") + require.NoError(t, err) + tc.Require.IndicesCreate(indexName, strings.NewReader(body)) + + require.ErrorIs(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger()), opensearch.ErrManualActionRequired) + }) + + t.Run("fails when the index contains a field the code schema does not know", func(t *testing.T) { + indexManager := opensearch.IndexManagerLatest + indexName := "opencloud-test-resource" + + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + tc.Require.IndicesReset([]string{indexName}) + + body, err := sjson.Set(indexManager.String(), "mappings.properties.legacyField.type", "keyword") + require.NoError(t, err) + tc.Require.IndicesCreate(indexName, strings.NewReader(body)) + + require.ErrorIs(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger()), opensearch.ErrManualActionRequired) + }) + + t.Run("transport errors do not demand manual action", func(t *testing.T) { + client, err := opensearchgoAPI.NewClient(opensearchgoAPI.Config{ + Client: opensearchgo.Config{ + Addresses: []string{"http://localhost:1025"}, + }, + }) + require.NoError(t, err) + + err = opensearch.IndexManagerLatest.Apply(t.Context(), "opencloud-test-resource", client, log.NopLogger()) + require.Error(t, err) + require.NotErrorIs(t, err, opensearch.ErrManualActionRequired) }) } From b1373878b2119d1907e279508ea2a404e0d7f8bc Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 8 Jul 2026 21:31:11 +0200 Subject: [PATCH 02/17] chore(search): drop the changelog entry --- .../change-search-index-schema-handling.md | 22 ------------------- 1 file changed, 22 deletions(-) delete mode 100644 changelog/unreleased/change-search-index-schema-handling.md diff --git a/changelog/unreleased/change-search-index-schema-handling.md b/changelog/unreleased/change-search-index-schema-handling.md deleted file mode 100644 index 1177da09ca..0000000000 --- a/changelog/unreleased/change-search-index-schema-handling.md +++ /dev/null @@ -1,22 +0,0 @@ -Change: Check the search index schema on startup (existing indexes need a rebuild) - -The search service now compares the schema of an existing search index with -the schema expected by the code at startup, for both the bleve and the -OpenSearch engine. A purely additive change (new fields that have never been -indexed) is applied in place and the service starts; documents indexed before -the upgrade do not contain the new fields until they are re-indexed. Any other -difference (changed field definitions, changed analyzers, removed or renamed -fields, or new fields that already contain data of unknown form) makes the -service refuse to start instead of silently returning wrong or incomplete -search results. - -Upgrading to this version is a breaking change for BOTH engines: every search -index built by a previous version differs from the new schema and the service -will refuse to start. To rebuild: stop the service, delete the search index -(the bleve directory or the OpenSearch index), start the service (an empty -index with the new schema is created) and run -"opencloud search index --all-spaces" to re-index all files. To bring an -instance up without search until a maintenance window, set -OC_EXCLUDE_RUN_SERVICES=search. - -https://github.com/opencloud-eu/opencloud/issues/3092 From 3b52bec87d78a4db2a661958be0b2e61e7cd9bac Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 8 Jul 2026 21:34:59 +0200 Subject: [PATCH 03/17] chore(search): mention the impact of disabling search in the refuse message --- services/search/pkg/mapping/classify.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/services/search/pkg/mapping/classify.go b/services/search/pkg/mapping/classify.go index 9fa357ef7c..f87d30282e 100644 --- a/services/search/pkg/mapping/classify.go +++ b/services/search/pkg/mapping/classify.go @@ -14,7 +14,8 @@ import ( var ErrManualActionRequired = errors.New("manual action required") // ManualActionRequiredError builds the operator-facing error for a breaking -// schema change. index is the index name (OpenSearch) or path (bleve). +// schema change, shared by both engines. index is the index name (OpenSearch) +// or path (bleve). func ManualActionRequiredError(index string, reasons []string) error { return fmt.Errorf( "%w: search index %s was built with a different schema (%s). "+ @@ -22,7 +23,9 @@ func ManualActionRequiredError(index string, reasons []string) error { "start the service (an empty index with the new schema is created), "+ "then rebuild the content by running: opencloud search index --all-spaces. "+ "To bring the instance up without search until a maintenance window, "+ - "set OC_EXCLUDE_RUN_SERVICES=search", + "set OC_EXCLUDE_RUN_SERVICES=search; until the service is back, search "+ + "and features built on it (e.g. the search bar and the tag list) are "+ + "unavailable", ErrManualActionRequired, index, strings.Join(reasons, "; "), index, ) } From bb215a12d44e827499e0f450d886a924bb3b891c Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 8 Jul 2026 21:44:09 +0200 Subject: [PATCH 04/17] chore(search): warn on additive opensearch changes and name the exact delete step Addresses the two Copilot review comments on the PR: the additive opensearch log now matches the bleve warning (level and re-index hint), and the refuse message spells out how to delete the index per engine (DELETE / vs removing the bleve directory). --- services/search/pkg/bleve/index.go | 2 +- services/search/pkg/mapping/classify.go | 10 ++++++---- services/search/pkg/opensearch/index.go | 6 +++--- 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index 392bd8bccb..521e00b929 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -64,7 +64,7 @@ func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { switch classification.Verdict { case searchmapping.VerdictBreaking: _ = index.Close() - return nil, classification, searchmapping.ManualActionRequiredError(destination, classification.Reasons) + return nil, classification, searchmapping.ManualActionRequiredError(destination, "delete the index directory "+destination, classification.Reasons) case searchmapping.VerdictAdditive: // The classifier guarantees everything else is identical and the new // fields hold no data yet, so storing the code mapping only adds diff --git a/services/search/pkg/mapping/classify.go b/services/search/pkg/mapping/classify.go index f87d30282e..dfaac5ebdc 100644 --- a/services/search/pkg/mapping/classify.go +++ b/services/search/pkg/mapping/classify.go @@ -15,18 +15,20 @@ var ErrManualActionRequired = errors.New("manual action required") // ManualActionRequiredError builds the operator-facing error for a breaking // schema change, shared by both engines. index is the index name (OpenSearch) -// or path (bleve). -func ManualActionRequiredError(index string, reasons []string) error { +// or path (bleve); deleteStep is the engine-specific instruction to remove the +// index, e.g. "delete the index (DELETE /name)" or "delete the index +// directory /path". +func ManualActionRequiredError(index, deleteStep string, reasons []string) error { return fmt.Errorf( "%w: search index %s was built with a different schema (%s). "+ - "There is no in-place migration: stop the service, delete %s, "+ + "There is no in-place migration: stop the service, %s, "+ "start the service (an empty index with the new schema is created), "+ "then rebuild the content by running: opencloud search index --all-spaces. "+ "To bring the instance up without search until a maintenance window, "+ "set OC_EXCLUDE_RUN_SERVICES=search; until the service is back, search "+ "and features built on it (e.g. the search bar and the tag list) are "+ "unavailable", - ErrManualActionRequired, index, strings.Join(reasons, "; "), index, + ErrManualActionRequired, index, strings.Join(reasons, "; "), deleteStep, ) } diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index d4896599a8..209832b8b1 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -165,7 +165,7 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch ) reasons = append(reasons, classification.Reasons...) if len(reasons) > 0 { - return searchmapping.ManualActionRequiredError(name, reasons) + return searchmapping.ManualActionRequiredError(name, fmt.Sprintf("delete the index (DELETE /%s)", name), reasons) } if len(classification.NewFields) == 0 { return nil // schema is up to date @@ -182,14 +182,14 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch case err != nil && errors.As(err, &putErr) && putErr.Err.Type == "illegal_argument_exception" && (strings.Contains(putErr.Err.Reason, "cannot be changed") || strings.Contains(putErr.Err.Reason, "Cannot update parameter")): // backstop, should be unreachable after the classification above - return searchmapping.ManualActionRequiredError(name, []string{putErr.Err.Reason}) + return searchmapping.ManualActionRequiredError(name, fmt.Sprintf("delete the index (DELETE /%s)", name), []string{putErr.Err.Reason}) case err != nil: return fmt.Errorf("failed to update mapping of index %s: %w", name, err) case !putResp.Acknowledged: return fmt.Errorf("failed to update mapping of index %s: not acknowledged", name) } - logger.Info().Strs("fields", classification.NewFields).Str("index", name).Msg("extended the search index mapping with new fields") + logger.Warn().Strs("fields", classification.NewFields).Str("index", name).Msg("extended the search index mapping with new fields; documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces") return nil } From 63e38adb9a3cf756d0993ec79a4b8c769b0d5c1e Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 8 Jul 2026 21:49:43 +0200 Subject: [PATCH 05/17] chore(search): tighten doc comments --- services/search/pkg/bleve/index.go | 30 ++++++++------------ services/search/pkg/command/server.go | 3 +- services/search/pkg/mapping/classify.go | 18 ++++-------- services/search/pkg/mapping/classify_test.go | 3 +- services/search/pkg/opensearch/index.go | 9 +++--- 5 files changed, 24 insertions(+), 39 deletions(-) diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index 521e00b929..1e4e908c0f 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -30,12 +30,10 @@ import ( var openRuntimeConfig = map[string]interface{}{"bolt_timeout": "5s"} // NewIndex opens (or creates) the bleve index at root and classifies the -// stored schema against the one generated from code. On a breaking change it -// refuses with ErrManualActionRequired. On an additive one the code schema is -// persisted into the index (the bleve analogue of an OpenSearch PUT _mapping), -// so the new fields are properly typed from now on and later startups classify -// equal; the caller must still warn that documents indexed before the upgrade -// lack the Classification.NewFields until they are re-indexed. +// stored schema against NewMapping(). Breaking changes refuse with +// ErrManualActionRequired, additive ones are persisted into the index (the +// bleve analogue of PUT _mapping); the caller must warn that pre-upgrade +// documents lack the Classification.NewFields until re-indexed. func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { destination := filepath.Join(root, fmt.Sprintf("bleve-v%d", search.SchemaVersion)) index, err := bleve.OpenUsing(destination, openRuntimeConfig) @@ -66,11 +64,9 @@ func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { _ = index.Close() return nil, classification, searchmapping.ManualActionRequiredError(destination, "delete the index directory "+destination, classification.Reasons) case searchmapping.VerdictAdditive: - // The classifier guarantees everything else is identical and the new - // fields hold no data yet, so storing the code mapping only adds - // fields. Reopen so the live mapping picks it up: without this the - // new fields would be indexed dynamically and the data-aware rule - // would turn them breaking on the next startup. + // Safe: everything else is identical and the new fields hold no data. + // Reopen so the live mapping picks the change up; otherwise the fields + // get indexed dynamically and flip to breaking on the next start. if err := index.SetInternal([]byte("_mapping"), codeB); err != nil { _ = index.Close() return nil, searchmapping.Classification{}, fmt.Errorf("failed to store the updated index mapping: %w", err) @@ -87,13 +83,11 @@ func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { return index, classification, nil } -// classifyStoredMapping diffs the mapping stored in the index against -// NewMapping() and also returns the marshaled code mapping. Fields that are -// new in the code schema but already have data in the index (previously -// indexed dynamically) are breaking, their de-facto form is unknown. The JSON -// compare is stable within one bleve version; if a bleve upgrade changes -// marshaling defaults it fails towards breaking, normalize the affected key -// here if that ever fires. +// classifyStoredMapping diffs the stored mapping against NewMapping() and +// returns the marshaled code mapping. New-in-code fields that already hold +// data (previously indexed dynamically) are breaking. The compare is only +// stable within one bleve version: a changed marshaling default fails towards +// breaking, normalize the affected key here if that ever fires. func classifyStoredMapping(index bleve.Index) (searchmapping.Classification, []byte, error) { storedB, err := index.GetInternal([]byte("_mapping")) if err != nil { diff --git a/services/search/pkg/command/server.go b/services/search/pkg/command/server.go index 2459bc4105..4086e64cd8 100644 --- a/services/search/pkg/command/server.go +++ b/services/search/pkg/command/server.go @@ -127,8 +127,7 @@ func Server(cfg *config.Config) *cobra.Command { return fmt.Errorf("failed to create OpenSearch client: %w", err) } - // bound the startup schema check so a hung cluster fails the - // start instead of blocking forever + // a hung cluster must fail the start, not block it forever startupCtx, cancelStartup := context.WithTimeout(ctx, time.Minute) indexName := opensearch.VersionedIndexName(cfg.Engine.OpenSearch.ResourceIndex.Name) openSearchBackend, err := opensearch.NewBackend(startupCtx, indexName, client, logger) diff --git a/services/search/pkg/mapping/classify.go b/services/search/pkg/mapping/classify.go index dfaac5ebdc..7e824a06e9 100644 --- a/services/search/pkg/mapping/classify.go +++ b/services/search/pkg/mapping/classify.go @@ -14,10 +14,8 @@ import ( var ErrManualActionRequired = errors.New("manual action required") // ManualActionRequiredError builds the operator-facing error for a breaking -// schema change, shared by both engines. index is the index name (OpenSearch) -// or path (bleve); deleteStep is the engine-specific instruction to remove the -// index, e.g. "delete the index (DELETE /name)" or "delete the index -// directory /path". +// schema change. index names the index, deleteStep is the engine-specific +// instruction to remove it. func ManualActionRequiredError(index, deleteStep string, reasons []string) error { return fmt.Errorf( "%w: search index %s was built with a different schema (%s). "+ @@ -51,14 +49,10 @@ type Classification struct { } // Classify recursively compares a stored `properties` tree against the one -// generated from code. Both sides must be generic JSON-decoded values -// (map[string]any, []any, float64), not marshaled Go structs, so values -// compare structurally. -// -// dataFields reports whether the index holds data at or below a dotted field -// path even though it is absent from the stored schema (bleve indexes dynamic -// fields without a schema trace). Engines that record dynamic fields in the -// live schema (OpenSearch) pass nil. +// generated from code. Both sides must be generic JSON-decoded values, not +// marshaled Go structs. dataFields reports whether the index holds data at or +// below a dotted field path absent from the stored schema (bleve dynamic +// fields); engines without that blind spot pass nil. func Classify(stored, code map[string]any, dataFields func(path string) bool) Classification { c := Classification{Verdict: VerdictEqual} classifyProperties(stored, code, dataFields, "", &c) diff --git a/services/search/pkg/mapping/classify_test.go b/services/search/pkg/mapping/classify_test.go index 37ea5fbd1b..c4b4d35817 100644 --- a/services/search/pkg/mapping/classify_test.go +++ b/services/search/pkg/mapping/classify_test.go @@ -82,8 +82,7 @@ var _ = Describe("Classify", func() { stored := parse(code) delete(stored, "photo") - // the callback is consulted with the subtree root; reporting data at - // or below it is the caller's job + // the callback is consulted with the subtree root c := Classify(stored, parse(code), hasData("photo")) Expect(c.Verdict).To(Equal(VerdictBreaking)) Expect(c.Reasons).To(ConsistOf(ContainSubstring("photo"))) diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index 209832b8b1..bc7fe53f0a 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -103,11 +103,10 @@ func buildResourceMapping() ([]byte, error) { return json.Marshal(index) } -// Apply ensures the index exists and matches the schema generated from code. -// A missing index is created, an additive schema change is applied in place -// via PUT _mapping, a breaking one returns ErrManualActionRequired. The -// classifier decides what is additive; PUT _mapping is only the mechanism to -// apply it (its merge semantics cannot detect removals or renames). +// Apply ensures the index exists and matches the schema generated from code: +// created if missing, additive changes applied via PUT _mapping, breaking ones +// refused with ErrManualActionRequired. The classifier judges, PUT _mapping +// only applies (its merge semantics hide removals and renames). func (m IndexManager) Apply(ctx context.Context, name string, client *opensearchgoAPI.Client, logger log.Logger) error { localIndexB, err := m.MarshalJSON() if err != nil { From f14455550af9c76231cb4e96849ab533677f127a Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Thu, 9 Jul 2026 01:27:26 +0200 Subject: [PATCH 06/17] fix(search): address max-review findings - the additive warnings advertise --all-spaces --force-rescan; a plain walk skips unchanged documents and never backfills the new fields - Apply checks index existence first again, so a pre-provisioned index needs no create privilege and odd create-error shapes (string error bodies, cluster blocks) cannot fail a healthy startup; Create on 404 keeps the typed already-exists swallow as the creation-race backstop - number_of_replicas drift is not breaking, it is runtime-tunable and needs no rebuild - bleve returns the classification alongside post-persist errors and the server warns before the error check, so the one-time additive warning is not lost when close or reopen fails - a golden fixture pins the marshaled bleve mapping so a dependency bump that changes marshaling fails in CI instead of refusing every installation in the field --- services/search/pkg/bleve/index.go | 9 +- services/search/pkg/bleve/index_test.go | 19 + .../pkg/bleve/testdata/mapping.golden.json | 677 ++++++++++++++++++ services/search/pkg/command/server.go | 11 +- services/search/pkg/opensearch/index.go | 41 +- services/search/pkg/opensearch/index_test.go | 14 + 6 files changed, 750 insertions(+), 21 deletions(-) create mode 100644 services/search/pkg/bleve/testdata/mapping.golden.json diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index 1e4e908c0f..f541b1941a 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -67,16 +67,19 @@ func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { // Safe: everything else is identical and the new fields hold no data. // Reopen so the live mapping picks the change up; otherwise the fields // get indexed dynamically and flip to breaking on the next start. + // return the classification even on errors: the mapping may already be + // persisted, so the next start classifies equal and the caller's + // warning is the only chance to surface the new fields if err := index.SetInternal([]byte("_mapping"), codeB); err != nil { _ = index.Close() - return nil, searchmapping.Classification{}, fmt.Errorf("failed to store the updated index mapping: %w", err) + return nil, classification, fmt.Errorf("failed to store the updated index mapping: %w", err) } if err := index.Close(); err != nil { - return nil, searchmapping.Classification{}, err + return nil, classification, err } index, err = bleve.OpenUsing(destination, openRuntimeConfig) if err != nil { - return nil, searchmapping.Classification{}, err + return nil, classification, err } } diff --git a/services/search/pkg/bleve/index_test.go b/services/search/pkg/bleve/index_test.go index 32d285d9f0..ae7ddf0ded 100644 --- a/services/search/pkg/bleve/index_test.go +++ b/services/search/pkg/bleve/index_test.go @@ -1,7 +1,9 @@ package bleve_test import ( + "encoding/json" "fmt" + "os" "path/filepath" bleveSearch "github.com/blevesearch/bleve/v2" @@ -179,4 +181,21 @@ var _ = Describe("NewMapping", func() { Expect(ok).To(BeTrue()) Expect(impl.Validate()).To(Succeed()) }) + + // A diff here means existing indexes will classify as breaking (schema or + // bleve marshaling changed); update the golden file only deliberately. + It("matches the committed golden mapping", func() { + m, err := bleve.NewMapping() + Expect(err).ToNot(HaveOccurred()) + b, err := json.Marshal(m) + Expect(err).ToNot(HaveOccurred()) + var got, golden map[string]any + Expect(json.Unmarshal(b, &got)).To(Succeed()) + + goldenB, err := os.ReadFile("testdata/mapping.golden.json") + Expect(err).ToNot(HaveOccurred()) + Expect(json.Unmarshal(goldenB, &golden)).To(Succeed()) + + Expect(got).To(Equal(golden)) + }) }) diff --git a/services/search/pkg/bleve/testdata/mapping.golden.json b/services/search/pkg/bleve/testdata/mapping.golden.json new file mode 100644 index 0000000000..f9c598a05b --- /dev/null +++ b/services/search/pkg/bleve/testdata/mapping.golden.json @@ -0,0 +1,677 @@ +{ + "default_mapping": { + "enabled": true, + "dynamic": true, + "properties": { + "Content": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "fulltext", + "store": true, + "index": true, + "include_term_vectors": true, + "docvalues": true + } + ] + }, + "Deleted": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "boolean", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Favorites": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "lowercaseKeyword", + "store": true, + "index": true, + "include_term_vectors": true, + "docvalues": true + } + ] + }, + "Hidden": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "boolean", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "ID": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "MimeType": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Mtime": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "datetime", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Name": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "lowercaseKeyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "ParentID": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Path": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "RootID": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Size": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Tags": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "lowercaseKeyword", + "store": true, + "index": true, + "include_term_vectors": true, + "docvalues": true + } + ] + }, + "Title": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Type": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "audio": { + "enabled": true, + "dynamic": true, + "properties": { + "album": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "albumArtist": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "artist": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "bitrate": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "composers": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "copyright": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "disc": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "discCount": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "duration": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "genre": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "hasDrm": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "boolean", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "isVariableBitrate": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "boolean", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "title": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "track": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "trackCount": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "year": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + } + } + }, + "image": { + "enabled": true, + "dynamic": true, + "properties": { + "height": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "width": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + } + } + }, + "location": { + "enabled": true, + "dynamic": true, + "properties": { + "altitude": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "latitude": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "longitude": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + } + } + }, + "location_geopoint": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "geopoint", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "photo": { + "enabled": true, + "dynamic": true, + "properties": { + "cameraMake": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "cameraModel": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "exposureDenominator": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "exposureNumerator": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "fNumber": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "focalLength": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "iso": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "orientation": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "number", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "takenDateTime": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "datetime", + "store": true, + "index": true, + "include_in_all": true, + "docvalues": true + } + ] + } + } + } + } + }, + "type_field": "_type", + "default_type": "_default", + "default_analyzer": "keyword", + "default_datetime_parser": "dateTimeOptional", + "default_field": "_all", + "store_dynamic": true, + "index_dynamic": true, + "docvalues_dynamic": true, + "analysis": { + "analyzers": { + "fulltext": { + "token_filters": [ + "to_lower", + "stemmer_porter" + ], + "tokenizer": "unicode", + "type": "custom" + }, + "lowercaseKeyword": { + "token_filters": [ + "to_lower" + ], + "tokenizer": "single", + "type": "custom" + } + } + } +} diff --git a/services/search/pkg/command/server.go b/services/search/pkg/command/server.go index 4086e64cd8..59fb16befa 100644 --- a/services/search/pkg/command/server.go +++ b/services/search/pkg/command/server.go @@ -74,14 +74,15 @@ func Server(cfg *config.Config) *cobra.Command { switch cfg.Engine.Type { case "bleve": idx, classification, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath) - if err != nil { - return err - } - + // 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 { logger.Warn(). Strs("fields", classification.NewFields). - Msgf("the bleve index at %s was built with an older schema; the new fields were added to the index schema, but documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces", cfg.Engine.Bleve.Datapath) + Msgf("the bleve index at %s was built with an older schema; the new fields were added to the index schema, but documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces --force-rescan", cfg.Engine.Bleve.Datapath) + } + if err != nil { + return err } defer func() { diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index bc7fe53f0a..a457bc446a 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -113,22 +113,34 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch return fmt.Errorf("failed to marshal index %s: %w", name, err) } - createResp, err := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{ - Index: name, - Body: bytes.NewReader(localIndexB), + // Exists first: a pre-provisioned index must not require create privileges + indicesExistsResp, err := client.Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{ + Indices: []string{name}, }) - var createErr *opensearchgo.StructError switch { - case err == nil && createResp.Acknowledged: - return nil - case err == nil: - return fmt.Errorf("failed to create index %s: not acknowledged", name) - case !errors.As(err, &createErr) || createErr.Err.Type != "resource_already_exists_exception": - // transport errors, disk-full etc. stay plain fatal, the restart policy retries - return fmt.Errorf("failed to create index %s: %w", name, err) + case indicesExistsResp != nil && indicesExistsResp.StatusCode == 404: + createResp, createErr := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{ + Index: name, + Body: bytes.NewReader(localIndexB), + }) + var structErr *opensearchgo.StructError + switch { + case createErr == nil && createResp.Acknowledged: + return nil + case createErr == nil: + return fmt.Errorf("failed to create index %s: not acknowledged", name) + case !errors.As(createErr, &structErr) || structErr.Err.Type != "resource_already_exists_exception": + // transport errors, disk-full etc. stay plain fatal, the restart policy retries + return fmt.Errorf("failed to create index %s: %w", name, createErr) + } + // lost the creation race to another instance, compare against its index + case err != nil: + return fmt.Errorf("failed to check if index %s exists: %w", name, err) + case indicesExistsResp == nil: + return fmt.Errorf("indicesExistsResp is nil for index %s", name) } - // the index already exists: compare settings and classify the mapping diff + // the index exists: compare settings and classify the mapping diff resp, err := client.Indices.Get(ctx, opensearchgoAPI.IndicesGetReq{ Indices: []string{name}, }) @@ -150,6 +162,9 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch var reasons []string for k := range localIndexJson.Get("settings").Map() { + if k == "number_of_replicas" { + continue // runtime-tunable via PUT _settings, drift needs no rebuild + } lv := localIndexJson.Get("settings." + k).Raw rv := remoteIndexJson.Get("settings.index." + k).Raw if !jsonEqual(lv, rv) { @@ -188,7 +203,7 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch return fmt.Errorf("failed to update mapping of index %s: not acknowledged", name) } - logger.Warn().Strs("fields", classification.NewFields).Str("index", name).Msg("extended the search index mapping with new fields; documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces") + logger.Warn().Strs("fields", classification.NewFields).Str("index", name).Msg("extended the search index mapping with new fields; documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces --force-rescan") return nil } diff --git a/services/search/pkg/opensearch/index_test.go b/services/search/pkg/opensearch/index_test.go index 34d267e288..f9ae3e3199 100644 --- a/services/search/pkg/opensearch/index_test.go +++ b/services/search/pkg/opensearch/index_test.go @@ -80,6 +80,20 @@ func TestIndexManager(t *testing.T) { require.ErrorIs(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger()), opensearch.ErrManualActionRequired) }) + t.Run("tolerates replica drift", func(t *testing.T) { + indexManager := opensearch.IndexManagerLatest + indexName := "opencloud-test-resource" + + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + tc.Require.IndicesReset([]string{indexName}) + + body, err := sjson.Set(indexManager.String(), "settings.number_of_replicas", "2") + require.NoError(t, err) + tc.Require.IndicesCreate(indexName, strings.NewReader(body)) + + require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) + }) + t.Run("is idempotent", func(t *testing.T) { indexManager := opensearch.IndexManagerLatest indexName := "opencloud-test-resource" From d08b4f8a8752f542f019534b189fa0f4c2538e1b Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 15 Jul 2026 17:37:08 +0200 Subject: [PATCH 07/17] fix(search): name the service to stop in the schema mismatch error --- services/search/pkg/mapping/classify.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/services/search/pkg/mapping/classify.go b/services/search/pkg/mapping/classify.go index 7e824a06e9..594e6d7650 100644 --- a/services/search/pkg/mapping/classify.go +++ b/services/search/pkg/mapping/classify.go @@ -19,8 +19,8 @@ var ErrManualActionRequired = errors.New("manual action required") func ManualActionRequiredError(index, deleteStep string, reasons []string) error { return fmt.Errorf( "%w: search index %s was built with a different schema (%s). "+ - "There is no in-place migration: stop the service, %s, "+ - "start the service (an empty index with the new schema is created), "+ + "There is no in-place migration: with the OpenCloud search service stopped, %s, "+ + "then start it again (an empty index with the new schema is created), "+ "then rebuild the content by running: opencloud search index --all-spaces. "+ "To bring the instance up without search until a maintenance window, "+ "set OC_EXCLUDE_RUN_SERVICES=search; until the service is back, search "+ From 2bac094fe554158c0234a01ac238aad9940e24b3 Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 15 Jul 2026 18:22:36 +0200 Subject: [PATCH 08/17] fix(search): list the schema mismatch reasons on separate lines --- services/search/pkg/mapping/classify.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/services/search/pkg/mapping/classify.go b/services/search/pkg/mapping/classify.go index 594e6d7650..696980e0fc 100644 --- a/services/search/pkg/mapping/classify.go +++ b/services/search/pkg/mapping/classify.go @@ -18,7 +18,7 @@ var ErrManualActionRequired = errors.New("manual action required") // instruction to remove it. func ManualActionRequiredError(index, deleteStep string, reasons []string) error { return fmt.Errorf( - "%w: search index %s was built with a different schema (%s). "+ + "%w: search index %s was built with a different schema:\n - %s\n"+ "There is no in-place migration: with the OpenCloud search service stopped, %s, "+ "then start it again (an empty index with the new schema is created), "+ "then rebuild the content by running: opencloud search index --all-spaces. "+ @@ -26,7 +26,7 @@ func ManualActionRequiredError(index, deleteStep string, reasons []string) error "set OC_EXCLUDE_RUN_SERVICES=search; until the service is back, search "+ "and features built on it (e.g. the search bar and the tag list) are "+ "unavailable", - ErrManualActionRequired, index, strings.Join(reasons, "; "), deleteStep, + ErrManualActionRequired, index, strings.Join(reasons, "\n - "), deleteStep, ) } @@ -79,7 +79,7 @@ func classifyProperties(stored, code map[string]any, dataFields func(string) boo } path := joinPath(prefix, k) if dataFields != nil && dataFields(path) { - c.breaking(fmt.Sprintf("field %s is new in the code schema but the index already contains data for it (previously indexed dynamically)", path)) + c.breaking(fmt.Sprintf("field %s is explicitly mapped now, but the index already holds data that was indexed dynamically for it, of an unknown type", path)) continue } c.NewFields = append(c.NewFields, leafPaths(code[k], path)...) From 6450af2e6df0f155c5266272df6ec9693dc07d40 Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 19:55:52 +0200 Subject: [PATCH 09/17] refactor(search): make the breaking-schema error developer-facing --- services/search/pkg/bleve/index.go | 2 +- services/search/pkg/mapping/classify.go | 22 +++++++++------------- services/search/pkg/opensearch/index.go | 4 ++-- 3 files changed, 12 insertions(+), 16 deletions(-) diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index f541b1941a..50726b97be 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -62,7 +62,7 @@ func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { switch classification.Verdict { case searchmapping.VerdictBreaking: _ = index.Close() - return nil, classification, searchmapping.ManualActionRequiredError(destination, "delete the index directory "+destination, classification.Reasons) + return nil, classification, searchmapping.ManualActionRequiredError(destination, classification.Reasons) case searchmapping.VerdictAdditive: // Safe: everything else is identical and the new fields hold no data. // Reopen so the live mapping picks the change up; otherwise the fields diff --git a/services/search/pkg/mapping/classify.go b/services/search/pkg/mapping/classify.go index 696980e0fc..2e0e7e100a 100644 --- a/services/search/pkg/mapping/classify.go +++ b/services/search/pkg/mapping/classify.go @@ -13,20 +13,16 @@ import ( // ErrManualActionRequired marks schema changes that cannot be applied in place. var ErrManualActionRequired = errors.New("manual action required") -// ManualActionRequiredError builds the operator-facing error for a breaking -// schema change. index names the index, deleteStep is the engine-specific -// instruction to remove it. -func ManualActionRequiredError(index, deleteStep string, reasons []string) error { +// ManualActionRequiredError reports a breaking schema change. Because the index +// is versioned by search.SchemaVersion, a released instance never hits this: a +// version bump builds a fresh index. It fires in development when the mapping is +// changed in a breaking way without bumping search.SchemaVersion, so the fix is +// to bump it (or revert the change). +func ManualActionRequiredError(index string, reasons []string) error { return fmt.Errorf( - "%w: search index %s was built with a different schema:\n - %s\n"+ - "There is no in-place migration: with the OpenCloud search service stopped, %s, "+ - "then start it again (an empty index with the new schema is created), "+ - "then rebuild the content by running: opencloud search index --all-spaces. "+ - "To bring the instance up without search until a maintenance window, "+ - "set OC_EXCLUDE_RUN_SERVICES=search; until the service is back, search "+ - "and features built on it (e.g. the search bar and the tag list) are "+ - "unavailable", - ErrManualActionRequired, index, strings.Join(reasons, "\n - "), deleteStep, + "%w: the search mapping in code differs from index %s in a breaking way:\n - %s\n"+ + "bump search.SchemaVersion to build a fresh index, or revert the mapping change", + ErrManualActionRequired, index, strings.Join(reasons, "\n - "), ) } diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index a457bc446a..3ead1bed48 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -179,7 +179,7 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch ) reasons = append(reasons, classification.Reasons...) if len(reasons) > 0 { - return searchmapping.ManualActionRequiredError(name, fmt.Sprintf("delete the index (DELETE /%s)", name), reasons) + return searchmapping.ManualActionRequiredError(name, reasons) } if len(classification.NewFields) == 0 { return nil // schema is up to date @@ -196,7 +196,7 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch case err != nil && errors.As(err, &putErr) && putErr.Err.Type == "illegal_argument_exception" && (strings.Contains(putErr.Err.Reason, "cannot be changed") || strings.Contains(putErr.Err.Reason, "Cannot update parameter")): // backstop, should be unreachable after the classification above - return searchmapping.ManualActionRequiredError(name, fmt.Sprintf("delete the index (DELETE /%s)", name), []string{putErr.Err.Reason}) + return searchmapping.ManualActionRequiredError(name, []string{putErr.Err.Reason}) case err != nil: return fmt.Errorf("failed to update mapping of index %s: %w", name, err) case !putResp.Acknowledged: From 494627b0f8bb566504566769ed80c7b78ef8cb3f Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 20:11:25 +0200 Subject: [PATCH 10/17] refactor(search): only enforce analysis settings, tolerate operational drift --- services/search/pkg/opensearch/index.go | 17 ++++++++--------- services/search/pkg/opensearch/index_test.go | 18 ++++++++++++++++-- 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index 3ead1bed48..a8b275d25c 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -160,16 +160,15 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch localIndexJson := gjson.ParseBytes(localIndexB) remoteIndexJson := gjson.ParseBytes(remoteIndexB) + // Only the analysis settings (analyzers/tokenizers/filters) affect how data + // is indexed and queried; a drift there yields wrong results. Shard/replica + // counts and other operational knobs are the operator's to tune (and a + // pre-provisioned index's to own), so they are not compared. var reasons []string - for k := range localIndexJson.Get("settings").Map() { - if k == "number_of_replicas" { - continue // runtime-tunable via PUT _settings, drift needs no rebuild - } - lv := localIndexJson.Get("settings." + k).Raw - rv := remoteIndexJson.Get("settings.index." + k).Raw - if !jsonEqual(lv, rv) { - reasons = append(reasons, fmt.Sprintf("settings.%s changed: index %s, code %s", k, rawOrUnset(rv), rawOrUnset(lv))) - } + lv := localIndexJson.Get("settings.analysis").Raw + rv := remoteIndexJson.Get("settings.index.analysis").Raw + if !jsonEqual(lv, rv) { + reasons = append(reasons, fmt.Sprintf("settings.analysis changed: index %s, code %s", rawOrUnset(rv), rawOrUnset(lv))) } classification := searchmapping.Classify( diff --git a/services/search/pkg/opensearch/index_test.go b/services/search/pkg/opensearch/index_test.go index f9ae3e3199..c7ce1740b7 100644 --- a/services/search/pkg/opensearch/index_test.go +++ b/services/search/pkg/opensearch/index_test.go @@ -66,14 +66,14 @@ 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) { + t.Run("fails when the analysis settings drift", func(t *testing.T) { indexManager := opensearch.IndexManagerLatest indexName := "opencloud-test-resource" tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) tc.Require.IndicesReset([]string{indexName}) - body, err := sjson.Set(indexManager.String(), "settings.number_of_shards", "2") + body, err := sjson.Set(indexManager.String(), "settings.analysis.analyzer.lowercaseKeyword.tokenizer", "standard") require.NoError(t, err) tc.Require.IndicesCreate(indexName, strings.NewReader(body)) @@ -94,6 +94,20 @@ func TestIndexManager(t *testing.T) { require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) }) + t.Run("tolerates shard drift", func(t *testing.T) { + indexManager := opensearch.IndexManagerLatest + indexName := "opencloud-test-resource" + + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + tc.Require.IndicesReset([]string{indexName}) + + body, err := sjson.Set(indexManager.String(), "settings.number_of_shards", "2") + require.NoError(t, err) + tc.Require.IndicesCreate(indexName, strings.NewReader(body)) + + require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger())) + }) + t.Run("is idempotent", func(t *testing.T) { indexManager := opensearch.IndexManagerLatest indexName := "opencloud-test-resource" From 3e7941dc0c7750a7d68a9279953e0cf7805f31bb Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 20:23:30 +0200 Subject: [PATCH 11/17] refactor(search): harden the index-diff helpers against unset input --- services/search/pkg/opensearch/index.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index a8b275d25c..d96fcdbf7b 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -206,7 +206,14 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch return nil } +// jsonEqual reports whether two raw JSON values are deeply equal. gjson yields +// an empty string for a path that does not exist; two such unset values are +// equal, an unset value never equals a present one, and a value that fails to +// parse counts as unequal. func jsonEqual(a, b string) bool { + if a == "" || b == "" { + return a == b + } var av, bv any if err := json.Unmarshal([]byte(a), &av); err != nil { return false @@ -217,11 +224,14 @@ func jsonEqual(a, b string) bool { return reflect.DeepEqual(av, bv) } -// propertiesMap parses a raw mappings.properties object; missing or empty -// input yields an empty map, which classifies as purely additive. +// propertiesMap parses a raw mappings.properties object into a map. Missing, +// empty, null or malformed input yields an empty (non-nil) map, which +// classifies as purely additive. func propertiesMap(raw string) map[string]any { props := map[string]any{} - _ = json.Unmarshal([]byte(raw), &props) + if err := json.Unmarshal([]byte(raw), &props); err != nil || props == nil { + return map[string]any{} + } return props } From ee824f52708924a420a93125bac5b92688c4c58e Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 20:36:58 +0200 Subject: [PATCH 12/17] refactor(search): extract opensearch.NewClient out of server startup --- services/search/pkg/command/server.go | 39 +----------------- services/search/pkg/opensearch/client.go | 52 ++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 37 deletions(-) create mode 100644 services/search/pkg/opensearch/client.go diff --git a/services/search/pkg/command/server.go b/services/search/pkg/command/server.go index 59fb16befa..69ec297d7a 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" ) @@ -93,39 +88,9 @@ 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 } // a hung cluster must fail the start, not block it forever diff --git a/services/search/pkg/opensearch/client.go b/services/search/pkg/opensearch/client.go new file mode 100644 index 0000000000..0b85b03a3b --- /dev/null +++ b/services/search/pkg/opensearch/client.go @@ -0,0 +1,52 @@ +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. +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 +} From a0af07d383c64db0ad5fcb1285b2c493a1e7de5c Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 21:27:51 +0200 Subject: [PATCH 13/17] refactor(search): route schema verdict handling through a shared mapping.Reconcile --- services/search/pkg/bleve/index.go | 79 ++++++++++++++---------- services/search/pkg/bleve/index_test.go | 25 ++++---- services/search/pkg/command/server.go | 10 +-- services/search/pkg/mapping/reconcile.go | 37 +++++++++++ services/search/pkg/opensearch/index.go | 69 +++++++++++++-------- 5 files changed, 140 insertions(+), 80 deletions(-) create mode 100644 services/search/pkg/mapping/reconcile.go diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index 50726b97be..d01bc17d9d 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -21,6 +21,7 @@ import ( "github.com/blevesearch/bleve/v2/mapping" storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" + "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" ) @@ -29,12 +30,10 @@ import ( // instead of blocking forever on the file lock. var openRuntimeConfig = map[string]interface{}{"bolt_timeout": "5s"} -// NewIndex opens (or creates) the bleve index at root and classifies the -// stored schema against NewMapping(). Breaking changes refuse with -// ErrManualActionRequired, additive ones are persisted into the index (the -// bleve analogue of PUT _mapping); the caller must warn that pre-upgrade -// documents lack the Classification.NewFields until re-indexed. -func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { +// NewIndex opens (or creates) the bleve index at root and reconciles the stored +// schema against NewMapping() via searchmapping.Reconcile: a breaking change +// refuses to start, an additive one is persisted into the index and warned about. +func NewIndex(root string, logger log.Logger) (bleve.Index, searchmapping.Classification, error) { destination := filepath.Join(root, fmt.Sprintf("bleve-v%d", search.SchemaVersion)) index, err := bleve.OpenUsing(destination, openRuntimeConfig) if errors.Is(err, bleve.ErrorIndexPathDoesNotExist) { @@ -53,37 +52,51 @@ func NewIndex(root string) (bleve.Index, searchmapping.Classification, error) { return nil, searchmapping.Classification{}, err } - classification, codeB, err := classifyStoredMapping(index) + r := &bleveReconciler{index: index, destination: destination} + classification, err := searchmapping.Reconcile(destination, r, logger) if err != nil { - _ = index.Close() - return nil, searchmapping.Classification{}, err - } - - switch classification.Verdict { - case searchmapping.VerdictBreaking: - _ = index.Close() - return nil, classification, searchmapping.ManualActionRequiredError(destination, classification.Reasons) - case searchmapping.VerdictAdditive: - // Safe: everything else is identical and the new fields hold no data. - // Reopen so the live mapping picks the change up; otherwise the fields - // get indexed dynamically and flip to breaking on the next start. - // return the classification even on errors: the mapping may already be - // persisted, so the next start classifies equal and the caller's - // warning is the only chance to surface the new fields - if err := index.SetInternal([]byte("_mapping"), codeB); err != nil { - _ = index.Close() - return nil, classification, fmt.Errorf("failed to store the updated index mapping: %w", err) - } - if err := index.Close(); err != nil { - return nil, classification, err - } - index, err = bleve.OpenUsing(destination, openRuntimeConfig) - if err != nil { - return nil, classification, err + if r.index != nil { + _ = r.index.Close() } + return nil, classification, err } - return index, classification, nil + return r.index, classification, nil +} + +// bleveReconciler adapts a bleve index to searchmapping.SchemaReconciler. +type bleveReconciler struct { + index bleve.Index + destination string + codeB []byte // marshaled code mapping, produced by Classify, used by ApplyAdditive +} + +func (r *bleveReconciler) Classify() (searchmapping.Classification, error) { + classification, codeB, err := classifyStoredMapping(r.index) + r.codeB = codeB + return classification, err +} + +// ApplyAdditive persists the code mapping and reopens so the live mapping picks +// it up; otherwise the new fields get indexed dynamically and flip to breaking +// on the next start. On failure it closes the index and clears the handle. +func (r *bleveReconciler) ApplyAdditive() error { + if err := r.index.SetInternal([]byte("_mapping"), r.codeB); err != nil { + _ = r.index.Close() + r.index = nil + return fmt.Errorf("failed to store the updated index mapping: %w", err) + } + if err := r.index.Close(); err != nil { + r.index = nil + return err + } + index, err := bleve.OpenUsing(r.destination, openRuntimeConfig) + if err != nil { + r.index = nil + return err + } + r.index = index + return nil } // classifyStoredMapping diffs the stored mapping against NewMapping() and diff --git a/services/search/pkg/bleve/index_test.go b/services/search/pkg/bleve/index_test.go index ae7ddf0ded..ad3b41293e 100644 --- a/services/search/pkg/bleve/index_test.go +++ b/services/search/pkg/bleve/index_test.go @@ -14,6 +14,7 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + "github.com/opencloud-eu/opencloud/pkg/log" "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" "github.com/opencloud-eu/opencloud/services/search/pkg/search" @@ -45,7 +46,7 @@ var _ = Describe("NewIndex", func() { } It("creates a fresh index", func() { - idx, classification, err := bleve.NewIndex(root) + idx, classification, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual)) Expect(idx.Close()).To(Succeed()) @@ -54,7 +55,7 @@ var _ = Describe("NewIndex", func() { It("opens an index with an identical schema", func() { buildIndex(codeMapping(), nil) - idx, classification, err := bleve.NewIndex(root) + idx, classification, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual)) Expect(classification.NewFields).To(BeEmpty()) @@ -67,7 +68,7 @@ var _ = Describe("NewIndex", func() { delete(old.DefaultMapping.Properties, "Title") buildIndex(old, nil) - idx, classification, err := bleve.NewIndex(root) + idx, classification, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive)) Expect(classification.NewFields).To(ConsistOf("Title")) @@ -83,7 +84,7 @@ var _ = Describe("NewIndex", func() { delete(photo.Properties, "cameraMake") buildIndex(old, nil) - idx, classification, err := bleve.NewIndex(root) + idx, classification, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive)) Expect(classification.NewFields).To(ConsistOf("photo.cameraMake")) @@ -95,13 +96,13 @@ var _ = Describe("NewIndex", func() { delete(old.DefaultMapping.Properties, "Title") buildIndex(old, nil) - idx, classification, err := bleve.NewIndex(root) + idx, classification, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive)) Expect(idx.Index("1", map[string]any{"Title": "hello"})).To(Succeed()) Expect(idx.Close()).To(Succeed()) - idx, classification, err = bleve.NewIndex(root) + idx, classification, err = bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual)) Expect(idx.Close()).To(Succeed()) @@ -113,7 +114,7 @@ var _ = Describe("NewIndex", func() { delete(old.DefaultMapping.Properties, "Mtime") buildIndex(old, map[string]map[string]any{"1": {"Mtime": "2026-01-02T03:04:05Z"}}) - idx, _, err := bleve.NewIndex(root) + idx, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) Expect(idx).To(BeNil()) }) @@ -124,7 +125,7 @@ var _ = Describe("NewIndex", func() { delete(old.DefaultMapping.Properties, "photo") buildIndex(old, map[string]map[string]any{"1": {"photo": map[string]any{"cameraMake": "ACME"}}}) - _, _, err := bleve.NewIndex(root) + _, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) }) @@ -136,7 +137,7 @@ var _ = Describe("NewIndex", func() { name.Fields[0].Analyzer = "fulltext" buildIndex(old, nil) - _, _, err := bleve.NewIndex(root) + _, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) }) @@ -145,7 +146,7 @@ var _ = Describe("NewIndex", func() { old.DefaultMapping.AddFieldMappingsAt("Legacy", bleveSearch.NewTextFieldMapping()) buildIndex(old, nil) - _, _, err := bleve.NewIndex(root) + _, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) }) @@ -154,7 +155,7 @@ var _ = Describe("NewIndex", func() { old.DefaultMapping.Dynamic = false buildIndex(old, nil) - _, _, err := bleve.NewIndex(root) + _, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) }) @@ -168,7 +169,7 @@ var _ = Describe("NewIndex", func() { } buildIndex(old, nil) - _, _, err := bleve.NewIndex(root) + _, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) }) }) diff --git a/services/search/pkg/command/server.go b/services/search/pkg/command/server.go index 69ec297d7a..83bb5d3799 100644 --- a/services/search/pkg/command/server.go +++ b/services/search/pkg/command/server.go @@ -18,7 +18,6 @@ import ( "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/content" - searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" "github.com/opencloud-eu/opencloud/services/search/pkg/metrics" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" bleveQuery "github.com/opencloud-eu/opencloud/services/search/pkg/query/bleve" @@ -68,14 +67,7 @@ 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) - // 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 { - logger.Warn(). - Strs("fields", classification.NewFields). - Msgf("the bleve index at %s was built with an older schema; the new fields were added to the index schema, but documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces --force-rescan", cfg.Engine.Bleve.Datapath) - } + idx, _, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath, logger) if err != nil { return err } diff --git a/services/search/pkg/mapping/reconcile.go b/services/search/pkg/mapping/reconcile.go new file mode 100644 index 0000000000..8f02604d8c --- /dev/null +++ b/services/search/pkg/mapping/reconcile.go @@ -0,0 +1,37 @@ +package mapping + +import ( + "github.com/opencloud-eu/opencloud/pkg/log" +) + +// SchemaReconciler is the engine-specific half of the startup schema check. +// Classify reads the stored and code schema and returns the verdict (with any +// engine-specific extras, e.g. analyzer/settings drift, already folded in); +// ApplyAdditive applies an additive change to the live index. Reconcile drives +// them so the verdict-to-action mapping lives in one place for every backend. +type SchemaReconciler interface { + Classify() (Classification, error) + ApplyAdditive() error +} + +// Reconcile runs the shared schema-verdict flow: an equal schema starts +// silently, a breaking one refuses with ManualActionRequiredError, an additive +// one is applied and warned about. index names the index in the messages. +func Reconcile(index string, r SchemaReconciler, logger log.Logger) (Classification, error) { + classification, err := r.Classify() + if err != nil { + return classification, err + } + + switch classification.Verdict { + case VerdictBreaking: + return classification, ManualActionRequiredError(index, classification.Reasons) + case VerdictAdditive: + if err := r.ApplyAdditive(); err != nil { + return classification, err + } + logger.Warn().Strs("fields", classification.NewFields).Str("index", index).Msg("extended the search index mapping with new fields; documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces --force-rescan") + } + + return classification, nil +} diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index d96fcdbf7b..4acc840ff5 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -103,10 +103,11 @@ func buildResourceMapping() ([]byte, error) { return json.Marshal(index) } -// Apply ensures the index exists and matches the schema generated from code: -// created if missing, additive changes applied via PUT _mapping, breaking ones -// refused with ErrManualActionRequired. The classifier judges, PUT _mapping -// only applies (its merge semantics hide removals and renames). +// Apply ensures the index exists and matches the schema generated from code: it +// is created if missing, otherwise its schema is reconciled via +// searchmapping.Reconcile (see osReconciler). PUT _mapping only applies the +// change; the classifier judges, because its merge semantics hide removals and +// renames. func (m IndexManager) Apply(ctx context.Context, name string, client *opensearchgoAPI.Client, logger log.Logger) error { localIndexB, err := m.MarshalJSON() if err != nil { @@ -140,14 +141,13 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch return fmt.Errorf("indicesExistsResp is nil for index %s", name) } - // the index exists: compare settings and classify the mapping diff + // the index exists: reconcile its schema through the shared verdict flow resp, err := client.Indices.Get(ctx, opensearchgoAPI.IndicesGetReq{ Indices: []string{name}, }) if err != nil { return fmt.Errorf("failed to get index %s: %w", name, err) } - remoteIndex, ok := resp.Indices[name] if !ok { return fmt.Errorf("index %s not found in response", name) @@ -157,52 +157,69 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch return fmt.Errorf("failed to marshal index %s: %w", name, err) } - localIndexJson := gjson.ParseBytes(localIndexB) - remoteIndexJson := gjson.ParseBytes(remoteIndexB) + r := &osReconciler{ + ctx: ctx, + name: name, + client: client, + local: gjson.ParseBytes(localIndexB), + remote: gjson.ParseBytes(remoteIndexB), + } + _, err = searchmapping.Reconcile(name, r, logger) + return err +} + +// osReconciler adapts an existing OpenSearch index to searchmapping.SchemaReconciler. +type osReconciler struct { + ctx context.Context + name string + client *opensearchgoAPI.Client + local gjson.Result + remote gjson.Result +} +func (r *osReconciler) Classify() (searchmapping.Classification, error) { // Only the analysis settings (analyzers/tokenizers/filters) affect how data // is indexed and queried; a drift there yields wrong results. Shard/replica // counts and other operational knobs are the operator's to tune (and a // pre-provisioned index's to own), so they are not compared. var reasons []string - lv := localIndexJson.Get("settings.analysis").Raw - rv := remoteIndexJson.Get("settings.index.analysis").Raw + lv := r.local.Get("settings.analysis").Raw + rv := r.remote.Get("settings.index.analysis").Raw if !jsonEqual(lv, rv) { reasons = append(reasons, fmt.Sprintf("settings.analysis changed: index %s, code %s", rawOrUnset(rv), rawOrUnset(lv))) } classification := searchmapping.Classify( - propertiesMap(remoteIndexJson.Get("mappings.properties").Raw), - propertiesMap(localIndexJson.Get("mappings.properties").Raw), + propertiesMap(r.remote.Get("mappings.properties").Raw), + propertiesMap(r.local.Get("mappings.properties").Raw), nil, ) reasons = append(reasons, classification.Reasons...) if len(reasons) > 0 { - return searchmapping.ManualActionRequiredError(name, reasons) - } - if len(classification.NewFields) == 0 { - return nil // schema is up to date + classification.Verdict = searchmapping.VerdictBreaking + classification.Reasons = reasons } + return classification, nil +} - // additive: the classifier guarantees every existing field matches the - // remote state, so putting the full code properties can only add fields - putResp, err := client.Indices.Mapping.Put(ctx, opensearchgoAPI.MappingPutReq{ - Indices: []string{name}, - Body: strings.NewReader(localIndexJson.Get("mappings").Raw), +// ApplyAdditive puts the full code properties; the classifier guarantees every +// existing field already matches the remote state, so this can only add fields. +func (r *osReconciler) ApplyAdditive() error { + putResp, err := r.client.Indices.Mapping.Put(r.ctx, opensearchgoAPI.MappingPutReq{ + Indices: []string{r.name}, + Body: strings.NewReader(r.local.Get("mappings").Raw), }) var putErr *opensearchgo.StructError switch { case err != nil && errors.As(err, &putErr) && putErr.Err.Type == "illegal_argument_exception" && (strings.Contains(putErr.Err.Reason, "cannot be changed") || strings.Contains(putErr.Err.Reason, "Cannot update parameter")): // backstop, should be unreachable after the classification above - return searchmapping.ManualActionRequiredError(name, []string{putErr.Err.Reason}) + return searchmapping.ManualActionRequiredError(r.name, []string{putErr.Err.Reason}) case err != nil: - return fmt.Errorf("failed to update mapping of index %s: %w", name, err) + return fmt.Errorf("failed to update mapping of index %s: %w", r.name, err) case !putResp.Acknowledged: - return fmt.Errorf("failed to update mapping of index %s: not acknowledged", name) + return fmt.Errorf("failed to update mapping of index %s: not acknowledged", r.name) } - - logger.Warn().Strs("fields", classification.NewFields).Str("index", name).Msg("extended the search index mapping with new fields; documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces --force-rescan") return nil } From 3e06bedb10aedf8f376ac74c64ab9fdabc6ed097 Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 21:32:38 +0200 Subject: [PATCH 14/17] refactor(search): warn on a persisted additive change even when the reopen fails --- services/search/pkg/bleve/index.go | 13 +++++++------ services/search/pkg/mapping/reconcile.go | 14 +++++++++++--- services/search/pkg/opensearch/index.go | 11 ++++++----- 3 files changed, 24 insertions(+), 14 deletions(-) diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index d01bc17d9d..1e95641349 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -79,24 +79,25 @@ func (r *bleveReconciler) Classify() (searchmapping.Classification, error) { // ApplyAdditive persists the code mapping and reopens so the live mapping picks // it up; otherwise the new fields get indexed dynamically and flip to breaking -// on the next start. On failure it closes the index and clears the handle. -func (r *bleveReconciler) ApplyAdditive() error { +// on the next start. Once SetInternal succeeds it reports persisted=true even if +// the reopen then fails. On failure it closes the index and clears the handle. +func (r *bleveReconciler) ApplyAdditive() (bool, error) { if err := r.index.SetInternal([]byte("_mapping"), r.codeB); err != nil { _ = r.index.Close() r.index = nil - return fmt.Errorf("failed to store the updated index mapping: %w", err) + return false, fmt.Errorf("failed to store the updated index mapping: %w", err) } if err := r.index.Close(); err != nil { r.index = nil - return err + return true, err } index, err := bleve.OpenUsing(r.destination, openRuntimeConfig) if err != nil { r.index = nil - return err + return true, err } r.index = index - return nil + return true, nil } // classifyStoredMapping diffs the stored mapping against NewMapping() and diff --git a/services/search/pkg/mapping/reconcile.go b/services/search/pkg/mapping/reconcile.go index 8f02604d8c..a812128c2b 100644 --- a/services/search/pkg/mapping/reconcile.go +++ b/services/search/pkg/mapping/reconcile.go @@ -11,7 +11,12 @@ import ( // them so the verdict-to-action mapping lives in one place for every backend. type SchemaReconciler interface { Classify() (Classification, error) - ApplyAdditive() error + // ApplyAdditive applies an additive change to the live index and reports + // whether the schema was persisted. It returns persisted=true even when a + // later step then fails (e.g. a bleve reopen), so Reconcile can still warn: + // the change is on disk, a subsequent start classifies equal and stays + // silent, so this is the only chance to surface the new fields. + ApplyAdditive() (persisted bool, err error) } // Reconcile runs the shared schema-verdict flow: an equal schema starts @@ -27,10 +32,13 @@ func Reconcile(index string, r SchemaReconciler, logger log.Logger) (Classificat case VerdictBreaking: return classification, ManualActionRequiredError(index, classification.Reasons) case VerdictAdditive: - if err := r.ApplyAdditive(); err != nil { + persisted, err := r.ApplyAdditive() + if persisted { + logger.Warn().Strs("fields", classification.NewFields).Str("index", index).Msg("extended the search index mapping with new fields; documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces --force-rescan") + } + if err != nil { return classification, err } - logger.Warn().Strs("fields", classification.NewFields).Str("index", index).Msg("extended the search index mapping with new fields; documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces --force-rescan") } return classification, nil diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index 4acc840ff5..00df4b1ef3 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -204,7 +204,8 @@ func (r *osReconciler) Classify() (searchmapping.Classification, error) { // ApplyAdditive puts the full code properties; the classifier guarantees every // existing field already matches the remote state, so this can only add fields. -func (r *osReconciler) ApplyAdditive() error { +// The PUT is atomic, so persisted is true only on success. +func (r *osReconciler) ApplyAdditive() (bool, error) { putResp, err := r.client.Indices.Mapping.Put(r.ctx, opensearchgoAPI.MappingPutReq{ Indices: []string{r.name}, Body: strings.NewReader(r.local.Get("mappings").Raw), @@ -214,13 +215,13 @@ func (r *osReconciler) ApplyAdditive() error { case err != nil && errors.As(err, &putErr) && putErr.Err.Type == "illegal_argument_exception" && (strings.Contains(putErr.Err.Reason, "cannot be changed") || strings.Contains(putErr.Err.Reason, "Cannot update parameter")): // backstop, should be unreachable after the classification above - return searchmapping.ManualActionRequiredError(r.name, []string{putErr.Err.Reason}) + return false, searchmapping.ManualActionRequiredError(r.name, []string{putErr.Err.Reason}) case err != nil: - return fmt.Errorf("failed to update mapping of index %s: %w", r.name, err) + return false, fmt.Errorf("failed to update mapping of index %s: %w", r.name, err) case !putResp.Acknowledged: - return fmt.Errorf("failed to update mapping of index %s: not acknowledged", r.name) + return false, fmt.Errorf("failed to update mapping of index %s: not acknowledged", r.name) } - return nil + return true, nil } // jsonEqual reports whether two raw JSON values are deeply equal. gjson yields From 740bd05116f3ab8acb5c555074bc43118b743a88 Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 21:33:59 +0200 Subject: [PATCH 15/17] feat(search): log a reindex hint when a fresh search index is created --- services/search/pkg/bleve/index.go | 1 + services/search/pkg/opensearch/index.go | 1 + 2 files changed, 2 insertions(+) diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index 1e95641349..4d229bf98f 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -46,6 +46,7 @@ func NewIndex(root string, logger log.Logger) (bleve.Index, searchmapping.Classi return nil, searchmapping.Classification{}, err } + logger.Info().Str("index", destination).Msg("created a new empty search index; if this OpenCloud instance already held files, they are not in it yet, index them by running: opencloud search index --all-spaces --force-rescan") return index, searchmapping.Classification{Verdict: searchmapping.VerdictEqual}, nil } if err != nil { diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index 00df4b1ef3..db0125da42 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -127,6 +127,7 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch var structErr *opensearchgo.StructError switch { case createErr == nil && createResp.Acknowledged: + logger.Info().Str("index", name).Msg("created a new empty search index; if this OpenCloud instance already held files, they are not in it yet, index them by running: opencloud search index --all-spaces --force-rescan") return nil case createErr == nil: return fmt.Errorf("failed to create index %s: not acknowledged", name) From 589b69b2ef8edee0b2bdaaf81881d91b096ee340 Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 22:07:10 +0200 Subject: [PATCH 16/17] refactor(search): single-source the new-index log message --- services/search/pkg/bleve/index.go | 2 +- services/search/pkg/mapping/reconcile.go | 7 +++++++ services/search/pkg/opensearch/index.go | 2 +- 3 files changed, 9 insertions(+), 2 deletions(-) diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index 4d229bf98f..d4f7787ba9 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -46,7 +46,7 @@ func NewIndex(root string, logger log.Logger) (bleve.Index, searchmapping.Classi return nil, searchmapping.Classification{}, err } - logger.Info().Str("index", destination).Msg("created a new empty search index; if this OpenCloud instance already held files, they are not in it yet, index them by running: opencloud search index --all-spaces --force-rescan") + searchmapping.LogNewIndexCreated(logger, destination) return index, searchmapping.Classification{Verdict: searchmapping.VerdictEqual}, nil } if err != nil { diff --git a/services/search/pkg/mapping/reconcile.go b/services/search/pkg/mapping/reconcile.go index a812128c2b..460218f8f0 100644 --- a/services/search/pkg/mapping/reconcile.go +++ b/services/search/pkg/mapping/reconcile.go @@ -43,3 +43,10 @@ func Reconcile(index string, r SchemaReconciler, logger log.Logger) (Classificat return classification, nil } + +// LogNewIndexCreated logs that a fresh, empty index was created and how to +// backfill it. Both backends call it after creating their index (that path does +// not run through Reconcile); an existing, up-to-date index stays silent. +func LogNewIndexCreated(logger log.Logger, index string) { + logger.Info().Str("index", index).Msg("created a new empty search index; if this OpenCloud instance already held files, they are not in it yet, index them by running: opencloud search index --all-spaces --force-rescan") +} diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index db0125da42..f2e42286ef 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -127,7 +127,7 @@ func (m IndexManager) Apply(ctx context.Context, name string, client *opensearch var structErr *opensearchgo.StructError switch { case createErr == nil && createResp.Acknowledged: - logger.Info().Str("index", name).Msg("created a new empty search index; if this OpenCloud instance already held files, they are not in it yet, index them by running: opencloud search index --all-spaces --force-rescan") + searchmapping.LogNewIndexCreated(logger, name) return nil case createErr == nil: return fmt.Errorf("failed to create index %s: not acknowledged", name) From a1b05351e81545a50b21e35f19709b7415e9f20e Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Wed, 29 Jul 2026 22:19:49 +0200 Subject: [PATCH 17/17] chore(search): tighten the schema-reconcile comments --- services/search/pkg/bleve/index.go | 10 ++++------ services/search/pkg/mapping/reconcile.go | 21 +++++++++------------ services/search/pkg/opensearch/index.go | 15 +++++---------- 3 files changed, 18 insertions(+), 28 deletions(-) diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index d4f7787ba9..0e0e28c85f 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -30,9 +30,8 @@ import ( // instead of blocking forever on the file lock. var openRuntimeConfig = map[string]interface{}{"bolt_timeout": "5s"} -// NewIndex opens (or creates) the bleve index at root and reconciles the stored -// schema against NewMapping() via searchmapping.Reconcile: a breaking change -// refuses to start, an additive one is persisted into the index and warned about. +// NewIndex opens (or creates) the bleve index at root and reconciles its schema +// against NewMapping() via searchmapping.Reconcile. func NewIndex(root string, logger log.Logger) (bleve.Index, searchmapping.Classification, error) { destination := filepath.Join(root, fmt.Sprintf("bleve-v%d", search.SchemaVersion)) index, err := bleve.OpenUsing(destination, openRuntimeConfig) @@ -79,9 +78,8 @@ func (r *bleveReconciler) Classify() (searchmapping.Classification, error) { } // ApplyAdditive persists the code mapping and reopens so the live mapping picks -// it up; otherwise the new fields get indexed dynamically and flip to breaking -// on the next start. Once SetInternal succeeds it reports persisted=true even if -// the reopen then fails. On failure it closes the index and clears the handle. +// it up. persisted=true once SetInternal succeeds, even if the reopen then +// fails; on error it closes the index and clears the handle. func (r *bleveReconciler) ApplyAdditive() (bool, error) { if err := r.index.SetInternal([]byte("_mapping"), r.codeB); err != nil { _ = r.index.Close() diff --git a/services/search/pkg/mapping/reconcile.go b/services/search/pkg/mapping/reconcile.go index 460218f8f0..6f8c056127 100644 --- a/services/search/pkg/mapping/reconcile.go +++ b/services/search/pkg/mapping/reconcile.go @@ -4,18 +4,16 @@ import ( "github.com/opencloud-eu/opencloud/pkg/log" ) -// SchemaReconciler is the engine-specific half of the startup schema check. -// Classify reads the stored and code schema and returns the verdict (with any -// engine-specific extras, e.g. analyzer/settings drift, already folded in); -// ApplyAdditive applies an additive change to the live index. Reconcile drives -// them so the verdict-to-action mapping lives in one place for every backend. +// SchemaReconciler is the engine-specific half of the startup schema check: +// Classify reads stored vs code schema (extras like analyzer drift folded in), +// ApplyAdditive applies an additive change. Reconcile drives them so the +// verdict-to-action policy lives in one place for every backend. type SchemaReconciler interface { Classify() (Classification, error) - // ApplyAdditive applies an additive change to the live index and reports - // whether the schema was persisted. It returns persisted=true even when a - // later step then fails (e.g. a bleve reopen), so Reconcile can still warn: - // the change is on disk, a subsequent start classifies equal and stays - // silent, so this is the only chance to surface the new fields. + // ApplyAdditive applies an additive change and reports whether the schema + // was persisted. persisted=true even if a later step fails (e.g. a bleve + // reopen), so Reconcile still warns: it is on disk, the next start + // classifies equal and stays silent. ApplyAdditive() (persisted bool, err error) } @@ -45,8 +43,7 @@ func Reconcile(index string, r SchemaReconciler, logger log.Logger) (Classificat } // LogNewIndexCreated logs that a fresh, empty index was created and how to -// backfill it. Both backends call it after creating their index (that path does -// not run through Reconcile); an existing, up-to-date index stays silent. +// backfill it. The create path does not run through Reconcile. func LogNewIndexCreated(logger log.Logger, index string) { logger.Info().Str("index", index).Msg("created a new empty search index; if this OpenCloud instance already held files, they are not in it yet, index them by running: opencloud search index --all-spaces --force-rescan") } diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index f2e42286ef..5fcf5bd467 100644 --- a/services/search/pkg/opensearch/index.go +++ b/services/search/pkg/opensearch/index.go @@ -103,11 +103,8 @@ func buildResourceMapping() ([]byte, error) { return json.Marshal(index) } -// Apply ensures the index exists and matches the schema generated from code: it -// is created if missing, otherwise its schema is reconciled via -// searchmapping.Reconcile (see osReconciler). PUT _mapping only applies the -// change; the classifier judges, because its merge semantics hide removals and -// renames. +// Apply ensures the index exists and matches the code schema: created if +// missing, otherwise reconciled via searchmapping.Reconcile (see osReconciler). func (m IndexManager) Apply(ctx context.Context, name string, client *opensearchgoAPI.Client, logger log.Logger) error { localIndexB, err := m.MarshalJSON() if err != nil { @@ -179,8 +176,7 @@ type osReconciler struct { } func (r *osReconciler) Classify() (searchmapping.Classification, error) { - // Only the analysis settings (analyzers/tokenizers/filters) affect how data - // is indexed and queried; a drift there yields wrong results. Shard/replica + // Only the analysis settings affect indexing correctness; shard/replica // counts and other operational knobs are the operator's to tune (and a // pre-provisioned index's to own), so they are not compared. var reasons []string @@ -203,9 +199,8 @@ func (r *osReconciler) Classify() (searchmapping.Classification, error) { return classification, nil } -// ApplyAdditive puts the full code properties; the classifier guarantees every -// existing field already matches the remote state, so this can only add fields. -// The PUT is atomic, so persisted is true only on success. +// ApplyAdditive puts the full code properties (only additions, per the +// classifier). The PUT is atomic, so persisted is true only on success. func (r *osReconciler) ApplyAdditive() (bool, error) { putResp, err := r.client.Indices.Mapping.Put(r.ctx, opensearchgoAPI.MappingPutReq{ Indices: []string{r.name},