diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index 9dd7d290ed..fd4d50e0d6 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -1,11 +1,13 @@ package bleve import ( + "encoding/json" "errors" "fmt" "math" "path/filepath" "reflect" + "strings" "github.com/blevesearch/bleve/v2" "github.com/blevesearch/bleve/v2/analysis/analyzer/custom" @@ -16,27 +18,159 @@ 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" ) -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]any{"bolt_timeout": "5s"} + +// 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.Open(destination) + 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, nil + searchmapping.LogNewIndexCreated(logger, destination) + return index, searchmapping.Classification{Verdict: searchmapping.VerdictEqual}, nil + } + if err != nil { + return nil, searchmapping.Classification{}, err } - return index, err + r := &bleveReconciler{index: index, destination: destination} + classification, err := searchmapping.Reconcile(destination, r, logger) + if err != nil { + if r.index != nil { + _ = r.index.Close() + } + return nil, classification, err + } + + 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. 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() + r.index = nil + return false, fmt.Errorf("failed to store the updated index mapping: %w", err) + } + if err := r.index.Close(); err != nil { + r.index = nil + return true, err + } + index, err := bleve.OpenUsing(r.destination, openRuntimeConfig) + if err != nil { + r.index = nil + return true, err + } + r.index = index + return true, nil +} + +// 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 assumes stable +// bleve marshaling; the golden test guards against a marshaling-default drift. +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 + }) + + // 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) + classification.AddBreaking(reasons...) + + 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) { + for _, k := range searchmapping.SortedUnionKeys(stored, code) { + 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 index b09d2da12a..e6e4499347 100644 --- a/services/search/pkg/bleve/index_test.go +++ b/services/search/pkg/bleve/index_test.go @@ -1,13 +1,23 @@ package bleve_test import ( + "encoding/json" "fmt" + "os" "path/filepath" + "reflect" + 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/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" ) @@ -16,7 +26,7 @@ var _ = Describe("Index", func() { It("puts the index into a directory of its own generation", func() { root := GinkgoT().TempDir() - index, err := bleve.NewIndex(root) + index, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) DeferCleanup(index.Close) @@ -27,11 +37,11 @@ var _ = Describe("Index", func() { It("opens the index that is already there", func() { root := GinkgoT().TempDir() - index, err := bleve.NewIndex(root) + index, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) Expect(index.Close()).To(Succeed()) - reopened, err := bleve.NewIndex(root) + reopened, _, err := bleve.NewIndex(root, log.NopLogger()) Expect(err).ToNot(HaveOccurred()) DeferCleanup(reopened.Close) @@ -39,3 +49,214 @@ var _ = Describe("Index", func() { }) }) }) + +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, log.NopLogger()) + 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, log.NopLogger()) + 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, log.NopLogger()) + 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, log.NopLogger()) + 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, 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, log.NopLogger()) + 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, log.NopLogger()) + 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, log.NopLogger()) + 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 = "standard" + buildIndex(old, nil) + + _, _, err := bleve.NewIndex(root, log.NopLogger()) + 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, log.NopLogger()) + 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, log.NopLogger()) + Expect(err).To(MatchError(searchmapping.ErrManualActionRequired)) + }) + + It("refuses on a changed analyzer definition", func() { + old := codeMapping() + Expect(old.CustomAnalysis.Analyzers).To(HaveKey(searchmapping.WordsAnalyzer)) + old.CustomAnalysis.Analyzers[searchmapping.WordsAnalyzer] = map[string]any{ + "type": custom.Name, + "tokenizer": unicode.Name, + "token_filters": []string{lowercase.Name}, + } + buildIndex(old, nil) + + _, _, err := bleve.NewIndex(root, log.NopLogger()) + 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()) + }) + + // A diff here means existing indexes will classify as breaking (schema or + // bleve marshaling changed). Update the golden deliberately; on a marshaling + // change, bump search.SchemaVersion too or existing indexes refuse to start. + 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()) + + if os.Getenv("UPDATE_GOLDEN") != "" { + pretty, err := json.MarshalIndent(m, "", " ") + Expect(err).ToNot(HaveOccurred()) + Expect(os.WriteFile("testdata/mapping.golden.json", append(pretty, '\n'), 0o644)).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), goldenAdvice(golden, got)) + }) +}) + +// goldenAdvice classifies a golden diff so the failure says whether the change +// is additive (regenerate with UPDATE_GOLDEN=1) or breaking (bump +// search.SchemaVersion too). +func goldenAdvice(golden, got map[string]any) string { + dig := func(m map[string]any, path ...string) map[string]any { + for _, k := range path { + m, _ = m[k].(map[string]any) + } + return m + } + c := searchmapping.Classify(dig(golden, "default_mapping", "properties"), dig(got, "default_mapping", "properties"), nil) + if !reflect.DeepEqual(dig(golden, "analysis"), dig(got, "analysis")) { + c.AddBreaking("the analysis definitions changed") + } + switch c.Verdict { + case searchmapping.VerdictAdditive: + return fmt.Sprintf("additive schema change (new fields: %v): regenerate the golden with UPDATE_GOLDEN=1, no SchemaVersion bump needed", c.NewFields) + case searchmapping.VerdictBreaking: + return fmt.Sprintf("breaking schema change (%v): regenerate the golden with UPDATE_GOLDEN=1 and bump search.SchemaVersion", c.Reasons) + } + return "the mapping changed outside the classified tree (bleve marshaling drift?): regenerate the golden with UPDATE_GOLDEN=1 and see the SchemaVersion note above" +} 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..09f274b7f0 --- /dev/null +++ b/services/search/pkg/bleve/testdata/mapping.golden.json @@ -0,0 +1,969 @@ +{ + "default_mapping": { + "enabled": true, + "dynamic": true, + "properties": { + "Content": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "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": "keyword", + "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", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "MimeType": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "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": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Name_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "Name_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": true + } + ] + }, + "ParentID": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Path": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "docvalues": true + } + ] + }, + "RootID": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "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": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "docvalues": true + } + ] + }, + "Tags_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "Title": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "Title_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "Title_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": 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", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "albumArtist": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "albumArtist_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "albumArtist_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": true + } + ] + }, + "album_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "album_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": true + } + ] + }, + "artist": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "artist_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "artist_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": 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", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "composers_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "composers_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": true + } + ] + }, + "copyright": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "copyright_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "copyright_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": 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", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "genre_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "genre_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": 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", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "title_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "title_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": 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", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "cameraMake_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "cameraMake_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": true + } + ] + }, + "cameraModel": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "store": true, + "index": true, + "include_term_vectors": true, + "include_in_all": true, + "docvalues": true + } + ] + }, + "cameraModel_lowercase": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "keyword", + "index": true, + "include_term_vectors": true + } + ] + }, + "cameraModel_words": { + "enabled": true, + "dynamic": true, + "fields": [ + { + "type": "text", + "analyzer": "words", + "index": true, + "include_term_vectors": 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": { + "char_filters": { + "dot_to_space": { + "regexp": "\\.", + "replace": " ", + "type": "regexp" + } + }, + "analyzers": { + "words": { + "char_filters": [ + "dot_to_space" + ], + "token_filters": [ + "to_lower" + ], + "tokenizer": "unicode", + "type": "custom" + } + } + } +} diff --git a/services/search/pkg/command/server.go b/services/search/pkg/command/server.go index a93338f79c..dbfe28af9b 100644 --- a/services/search/pkg/command/server.go +++ b/services/search/pkg/command/server.go @@ -2,11 +2,9 @@ package command import ( "context" - "crypto/tls" "fmt" - "net/http" - "os" "os/signal" + "time" "github.com/opencloud-eu/opencloud/pkg/config/configlog" "github.com/opencloud-eu/opencloud/pkg/generators" @@ -30,8 +28,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" ) @@ -71,7 +67,7 @@ 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, _, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath, logger) if err != nil { return err } @@ -84,43 +80,15 @@ 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 } - indexName := opensearch.VersionedIndexName(cfg.Engine.OpenSearch.ResourceIndex.Name) - openSearchBackend, err := opensearch.NewBackend(indexName, client) + // a hung cluster must fail the start, not block it forever + startupCtx, cancelStartup := context.WithTimeout(ctx, time.Minute) + openSearchBackend, err := opensearch.NewBackend(startupCtx, cfg.Engine.OpenSearch.ResourceIndex.Name, client, logger) + 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 9ee2d9cb82..0a3a55c45b 100644 --- a/services/search/pkg/mapping/bleve.go +++ b/services/search/pkg/mapping/bleve.go @@ -12,8 +12,8 @@ import ( // struct via reflection. Field names come from json tags; overrides are // keyed by those names (or dotted paths for nested fields). // -// The returned mapping references the words analyzer for Fulltext fields; -// the caller registers it on the enclosing IndexMapping. +// The returned mapping references analyzer names that the caller must register +// 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..edc64563cf --- /dev/null +++ b/services/search/pkg/mapping/classify.go @@ -0,0 +1,174 @@ +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 reports a breaking schema change. Only reachable in +// development: a released instance versions the index by search.SchemaVersion, +// so a bump builds a fresh index instead. +func ManualActionRequiredError(index string, reasons []string) error { + return fmt.Errorf( + "%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 - "), + ) +} + +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 +} + +// AddBreaking records engine-specific breaking reasons (e.g. analyzer drift) +// found outside the properties tree and forces the verdict to breaking. It is a +// no-op when reasons is empty, so callers can pass their findings unconditionally. +func (c *Classification) AddBreaking(reasons ...string) { + if len(reasons) == 0 { + return + } + c.Verdict = VerdictBreaking + // clone so we never append into the caller's variadic slice + c.Reasons = append(slices.Clone(reasons), c.Reasons...) +} + +// Classify recursively compares a stored `properties` tree against the one from +// code (both generic JSON-decoded, not marshaled structs). dataFields reports +// whether a code-only field already holds data in the index (bleve dynamic +// fields make a new field breaking); 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) + 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 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)...) + } +} + +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 +} + +// SortedUnionKeys returns the sorted union of the keys of a and b. +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..e76cdb1cfd --- /dev/null +++ b/services/search/pkg/mapping/classify_test.go @@ -0,0 +1,137 @@ +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) } + } + + // setup produces the (stored, code, dataFields) arguments for one case. + type setup func() (stored, codeSchema map[string]any, dataFields func(string) bool) + + substrings := func(ss []string) []any { + out := make([]any, len(ss)) + for i, s := range ss { + out[i] = ContainSubstring(s) + } + return out + } + fields := func(ss []string) []any { + out := make([]any, len(ss)) + for i, s := range ss { + out[i] = s + } + return out + } + + // newFields/reasons are asserted only when non-nil; an empty slice asserts + // "none". + DescribeTable("verdict", + func(s setup, verdict Verdict, newFields, reasons []string) { + stored, codeSchema, dataFields := s() + + c := Classify(stored, codeSchema, dataFields) + Expect(c.Verdict).To(Equal(verdict)) + if newFields != nil { + Expect(c.NewFields).To(ConsistOf(fields(newFields)...)) + } + if reasons != nil { + Expect(c.Reasons).To(ConsistOf(substrings(reasons)...)) + } + }, + Entry("identical schemas are equal", + setup(func() (map[string]any, map[string]any, func(string) bool) { + return parse(code), parse(code), nil + }), VerdictEqual, []string{}, []string{}), + + Entry("a new top-level field is additive", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + delete(stored, "Size") + return stored, parse(code), nil + }), VerdictAdditive, []string{"Size"}, nil), + + Entry("a new nested field is additive", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + delete(stored["photo"].(map[string]any)["properties"].(map[string]any), "cameraMake") + return stored, parse(code), nil + }), VerdictAdditive, []string{"photo.cameraMake"}, nil), + + Entry("a new subtree lists every leaf", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + delete(stored, "photo") + return stored, parse(code), nil + }), VerdictAdditive, []string{"photo.cameraMake", "photo.cameraModel"}, nil), + + Entry("a new field that already holds data is breaking", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + delete(stored, "Size") + return stored, parse(code), hasData("Size") + }), VerdictBreaking, nil, []string{"Size"}), + + Entry("a new nested field that already holds data is breaking", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + delete(stored["photo"].(map[string]any)["properties"].(map[string]any), "cameraMake") + return stored, parse(code), hasData("photo.cameraMake") + }), VerdictBreaking, nil, []string{"photo.cameraMake"}), + + Entry("a new subtree with data below it is breaking", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + delete(stored, "photo") + return stored, parse(code), hasData("photo") + }), VerdictBreaking, nil, []string{"photo"}), + + Entry("a changed field definition is breaking", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + stored["Size"].(map[string]any)["type"] = "keyword" + return stored, parse(code), nil + }), VerdictBreaking, nil, []string{"Size"}), + + Entry("a field removed from the code schema is breaking", + setup(func() (map[string]any, map[string]any, func(string) bool) { + reduced := parse(code) + delete(reduced, "Size") + return parse(code), reduced, nil + }), VerdictBreaking, nil, []string{"removed or renamed"}), + + Entry("a changed object attribute is breaking", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + stored["photo"].(map[string]any)["dynamic"] = true + return stored, parse(code), nil + }), VerdictBreaking, nil, []string{"dynamic"}), + + Entry("breaking wins over additive", + setup(func() (map[string]any, map[string]any, func(string) bool) { + stored := parse(code) + delete(stored, "Size") + stored["Name"].(map[string]any)["type"] = "text" + return stored, parse(code), nil + }), VerdictBreaking, []string{"Size"}, []string{"Name"}), + ) +}) diff --git a/services/search/pkg/mapping/reconcile.go b/services/search/pkg/mapping/reconcile.go new file mode 100644 index 0000000000..82cc46feb6 --- /dev/null +++ b/services/search/pkg/mapping/reconcile.go @@ -0,0 +1,44 @@ +package mapping + +import ( + "github.com/opencloud-eu/opencloud/pkg/log" +) + +// SchemaReconciler is the engine-specific half of the startup schema check that +// Reconcile drives, keeping the verdict-to-action policy in one place. +type SchemaReconciler interface { + Classify() (Classification, error) + // ApplyAdditive applies an additive change. persisted is true once the + // schema is on disk, even if a later step (e.g. a bleve reopen) then fails. + ApplyAdditive() (persisted bool, err error) +} + +// Reconcile applies the shared verdict policy: equal is silent, breaking refuses +// with ManualActionRequiredError, additive is applied and warned about. +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: + 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 + } + } + + return classification, nil +} + +// LogNewIndexCreated logs that a fresh, empty index was created and how to +// 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/mapping/reconcile_test.go b/services/search/pkg/mapping/reconcile_test.go new file mode 100644 index 0000000000..2baa32c17f --- /dev/null +++ b/services/search/pkg/mapping/reconcile_test.go @@ -0,0 +1,78 @@ +package mapping + +import ( + "errors" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/opencloud-eu/opencloud/pkg/log" +) + +type fakeReconciler struct { + classification Classification + classifyErr error + persisted bool + applyErr error + applyCalls int +} + +func (f *fakeReconciler) Classify() (Classification, error) { + return f.classification, f.classifyErr +} + +func (f *fakeReconciler) ApplyAdditive() (bool, error) { + f.applyCalls++ + return f.persisted, f.applyErr +} + +var _ = Describe("Reconcile", func() { + logger := log.NopLogger() + + It("does nothing on an equal schema", func() { + r := &fakeReconciler{classification: Classification{Verdict: VerdictEqual}} + + c, err := Reconcile("idx", r, logger) + Expect(err).ToNot(HaveOccurred()) + Expect(c.Verdict).To(Equal(VerdictEqual)) + Expect(r.applyCalls).To(BeZero()) + }) + + It("refuses a breaking schema without applying it", func() { + r := &fakeReconciler{classification: Classification{Verdict: VerdictBreaking, Reasons: []string{"Name changed"}}} + + _, err := Reconcile("idx", r, logger) + Expect(err).To(MatchError(ErrManualActionRequired)) + Expect(err.Error()).To(ContainSubstring("Name changed")) + Expect(r.applyCalls).To(BeZero()) + }) + + It("applies an additive schema", func() { + r := &fakeReconciler{classification: Classification{Verdict: VerdictAdditive, NewFields: []string{"Size"}}, persisted: true} + + c, err := Reconcile("idx", r, logger) + Expect(err).ToNot(HaveOccurred()) + Expect(c.Verdict).To(Equal(VerdictAdditive)) + Expect(r.applyCalls).To(Equal(1)) + }) + + It("surfaces an apply error even when the schema was persisted", func() { + r := &fakeReconciler{ + classification: Classification{Verdict: VerdictAdditive, NewFields: []string{"Size"}}, + persisted: true, + applyErr: errors.New("reopen failed"), + } + + _, err := Reconcile("idx", r, logger) + Expect(err).To(MatchError(ContainSubstring("reopen failed"))) + Expect(r.applyCalls).To(Equal(1)) + }) + + It("propagates a classify error", func() { + r := &fakeReconciler{classifyErr: errors.New("read schema failed")} + + _, err := Reconcile("idx", r, logger) + Expect(err).To(MatchError(ContainSubstring("read schema failed"))) + Expect(r.applyCalls).To(BeZero()) + }) +}) diff --git a/services/search/pkg/opensearch/backend.go b/services/search/pkg/opensearch/backend.go index 49cf67fc54..330dcd577d 100644 --- a/services/search/pkg/opensearch/backend.go +++ b/services/search/pkg/opensearch/backend.go @@ -14,6 +14,7 @@ import ( "github.com/opencloud-eu/opencloud/pkg/conversions" "github.com/opencloud-eu/opencloud/pkg/kql" + "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" @@ -33,10 +34,10 @@ type Backend struct { } // NewBackend creates a backend on the versioned generation of the named index. -func NewBackend(name string, client *opensearchgoAPI.Client) (*Backend, error) { +func NewBackend(ctx context.Context, name string, client *opensearchgoAPI.Client, logger log.Logger) (*Backend, error) { index := VersionedIndexName(name) - pingResp, err := client.Ping(context.TODO(), &opensearchgoAPI.PingReq{}) + pingResp, err := client.Ping(ctx, &opensearchgoAPI.PingReq{}) switch { case err != nil: return nil, fmt.Errorf("%w, failed to ping opensearch: %w", ErrUnhealthyCluster, err) @@ -45,13 +46,13 @@ func NewBackend(name 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 dc83e057b8..94f9e31555 100644 --- a/services/search/pkg/opensearch/backend_test.go +++ b/services/search/pkg/opensearch/backend_test.go @@ -9,6 +9,7 @@ import ( opensearchgo "github.com/opensearch-project/opensearch-go/v4" opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi" + "github.com/opencloud-eu/opencloud/pkg/log" "github.com/opencloud-eu/opencloud/services/search/internal/opensearchtest" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" ) @@ -34,7 +35,7 @@ var _ = Describe("Backend", func() { }) Expect(err).ToNot(HaveOccurred(), "failed to create OpenSearch client") - backend, err := opensearch.NewBackend("test-engine-new-engine", client) + backend, err := opensearch.NewBackend(context.Background(), "test-engine-new-engine", client, log.NopLogger()) Expect(backend).To(BeNil()) Expect(err).To(MatchError(opensearch.ErrUnhealthyCluster)) }) @@ -57,7 +58,7 @@ var _ = Describe("Backend", func() { deleteIndexOnCleanup(tc, physical) var err error - backend, err = opensearch.NewBackend(indexName, tc.Client()) + backend, err = opensearch.NewBackend(context.Background(), indexName, tc.Client(), log.NopLogger()) Expect(err).ToNot(HaveOccurred()) }) diff --git a/services/search/pkg/opensearch/client.go b/services/search/pkg/opensearch/client.go new file mode 100644 index 0000000000..34972d6625 --- /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 +} diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index bbc38640f4..e79e053aa0 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. @@ -110,120 +114,150 @@ func buildResourceMapping() ([]byte, error) { return json.Marshal(index) } -func coveredAt(declared, index gjson.Result, declaredPath, indexPath string) (string, string, bool) { - declaredRaw := declared.Get(declaredPath).Raw - indexRaw := index.Get(indexPath).Raw - - var declaredValue, indexValue any - if err := json.Unmarshal([]byte(declaredRaw), &declaredValue); err != nil { - return declaredRaw, indexRaw, false - } - - if err := json.Unmarshal([]byte(indexRaw), &indexValue); err != nil { - return declaredRaw, indexRaw, false - } - - return declaredRaw, indexRaw, covered(declaredValue, indexValue) -} - -func covered(declared, index any) bool { - declaredMap, ok := declared.(map[string]any) - if !ok { - return reflect.DeepEqual(declared, index) - } - - indexMap, ok := index.(map[string]any) - if !ok { - return false - } - - for key, declaredValue := range declaredMap { - indexValue, ok := indexMap[key] - if !ok || !covered(declaredValue, indexValue) { - return false - } - } - - return true -} - -func (m IndexManager) Apply(ctx context.Context, name string, client *opensearchgoAPI.Client) error { +// 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 { return fmt.Errorf("failed to marshal index %s: %w", name, err) } + // Exists first: a pre-provisioned index must not require create privileges indicesExistsResp, err := client.Indices.Exists(ctx, opensearchgoAPI.IndicesExistsReq{ Indices: []string{name}, }) switch { case indicesExistsResp != nil && indicesExistsResp.StatusCode == 404: - break + createResp, createErr := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{ + Index: name, + Body: bytes.NewReader(localIndexB), + }) + var structErr *opensearchgo.StructError + switch { + case createErr == nil && createResp.Acknowledged: + searchmapping.LogNewIndexCreated(logger, name) + 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) } - 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.IndicesGetRespData)[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) - - var errs []error - - for k := range localIndexJson.Get("settings").Map() { - if lv, rv, ok := coveredAt(localIndexJson, remoteIndexJson, "settings."+k, "settings.index."+k); !ok { - errs = append(errs, fmt.Errorf("settings.%s local %s, remote %s", k, lv, rv)) - } - } - - for k := range localIndexJson.Get("mappings.properties").Map() { - if _, _, ok := coveredAt(localIndexJson, remoteIndexJson, "mappings.properties."+k, "mappings.properties."+k); !ok { - errs = append(errs, fmt.Errorf("mappings.properties.%s", k)) - } - } - - 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...), - ) - } - - return nil // Index is already up to date, no action needed - } - - createResp, err := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{ - Index: name, - Body: bytes.NewReader(localIndexB), + // the index exists: reconcile its schema through the shared verdict flow + resp, err := client.Indices.Get(ctx, opensearchgoAPI.IndicesGetReq{ + Indices: []string{name}, }) - switch { - 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) + if err != nil { + return fmt.Errorf("failed to get index %s: %w", name, err) } - return nil + remoteIndex, ok := (*resp.IndicesGetRespData)[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) + } + + 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 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 + 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(r.remote.Get("mappings.properties").Raw), + propertiesMap(r.local.Get("mappings.properties").Raw), + nil, + ) + classification.AddBreaking(reasons...) + return classification, nil +} + +// 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}, + 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 false, searchmapping.ManualActionRequiredError(r.name, []string{putErr.Err.Reason}) + case err != nil: + return false, fmt.Errorf("failed to update mapping of index %s: %w", r.name, err) + case !putResp.Acknowledged: + return false, fmt.Errorf("failed to update mapping of index %s: not acknowledged", r.name) + } + return true, nil +} + +// jsonEqual reports whether two raw JSON values are deeply equal. A missing +// gjson path is an empty string, so two unset values compare equal. +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 + } + if err := json.Unmarshal([]byte(b), &bv); err != nil { + return false + } + return reflect.DeepEqual(av, bv) +} + +// 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{} + if err := json.Unmarshal([]byte(raw), &props); err != nil || props == nil { + return map[string]any{} + } + 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 5837189e82..3dc8b02fec 100644 --- a/services/search/pkg/opensearch/index_test.go +++ b/services/search/pkg/opensearch/index_test.go @@ -1,14 +1,23 @@ package opensearch_test import ( + "bytes" + "encoding/json" "fmt" + "os" + "reflect" "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/internal/opensearchtest" + searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" "github.com/opencloud-eu/opencloud/services/search/pkg/search" ) @@ -26,6 +35,51 @@ func TestVersionedIndexName(t *testing.T) { ) } +// A diff here means the generated OpenSearch index definition changed: new +// indexes get the new shape, existing ones answer to the startup classifier. +// Update the golden deliberately (UPDATE_GOLDEN=1); a breaking change needs a +// search.SchemaVersion bump. +func TestGoldenMapping(t *testing.T) { + var pretty bytes.Buffer + require.NoError(t, json.Indent(&pretty, []byte(opensearch.IndexManagerLatest.String()), "", " ")) + pretty.WriteByte('\n') + + if os.Getenv("UPDATE_GOLDEN") != "" { + require.NoError(t, os.WriteFile("testdata/resource.golden.json", pretty.Bytes(), 0o644)) + } + + goldenB, err := os.ReadFile("testdata/resource.golden.json") + require.NoError(t, err) + + var got, golden map[string]any + require.NoError(t, json.Unmarshal(pretty.Bytes(), &got)) + require.NoError(t, json.Unmarshal(goldenB, &golden)) + require.Equal(t, golden, got, goldenAdvice(golden, got, "mappings.properties", "settings.analysis")) +} + +// goldenAdvice classifies a golden diff so the failure says whether the change +// is additive (regenerate with UPDATE_GOLDEN=1) or breaking (bump +// search.SchemaVersion too). +func goldenAdvice(golden, got map[string]any, propsPath, analysisPath string) string { + dig := func(m map[string]any, path string) map[string]any { + for _, k := range strings.Split(path, ".") { + m, _ = m[k].(map[string]any) + } + return m + } + c := searchmapping.Classify(dig(golden, propsPath), dig(got, propsPath), nil) + if !reflect.DeepEqual(dig(golden, analysisPath), dig(got, analysisPath)) { + c.AddBreaking("the analysis settings changed") + } + switch c.Verdict { + case searchmapping.VerdictAdditive: + return fmt.Sprintf("additive schema change (new fields: %v): regenerate the golden with UPDATE_GOLDEN=1, no SchemaVersion bump needed", c.NewFields) + case searchmapping.VerdictBreaking: + return fmt.Sprintf("breaking schema change (%v): regenerate the golden with UPDATE_GOLDEN=1 and bump search.SchemaVersion", c.Reasons) + } + return "the schema changed outside the classified tree: regenerate the golden with UPDATE_GOLDEN=1" +} + func TestIndexManager(t *testing.T) { t.Run("index plausibility", func(t *testing.T) { tests := []opensearchtest.TableTest[opensearch.IndexManager, struct{}]{ @@ -46,7 +100,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,38 +113,38 @@ 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("accepts an index that carries more than the definition declares", 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(), "mappings.properties.Path.fields.raw.type", "keyword") + body, err := sjson.Set(indexManager.String(), "settings.analysis.analyzer.lowercaseKeyword.tokenizer", "standard") require.NoError(t, err) tc.Require.IndicesCreate(indexName, strings.NewReader(body)) - require.NoError(t, indexManager.Apply(t.Context(), indexName, tc.Client())) + require.ErrorIs(t, indexManager.Apply(t.Context(), indexName, tc.Client(), log.NopLogger()), opensearch.ErrManualActionRequired) }) - t.Run("fails when the index misses something the definition declares", func(t *testing.T) { + 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.Delete(indexManager.String(), "mappings.properties.Path.analyzer") + body, err := sjson.Set(indexManager.String(), "settings.number_of_replicas", "2") require.NoError(t, err) tc.Require.IndicesCreate(indexName, strings.NewReader(body)) - require.ErrorIs(t, indexManager.Apply(t.Context(), indexName, tc.Client()), opensearch.ErrManualActionRequired) + 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("tolerates shard drift", func(t *testing.T) { indexManager := opensearch.IndexManagerLatest indexName := "opencloud-test-resource" @@ -101,6 +155,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.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" + + 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.GetIndices()[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.GetIndices()[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) }) } diff --git a/services/search/pkg/opensearch/testdata/resource.golden.json b/services/search/pkg/opensearch/testdata/resource.golden.json new file mode 100644 index 0000000000..7486322034 --- /dev/null +++ b/services/search/pkg/opensearch/testdata/resource.golden.json @@ -0,0 +1,291 @@ +{ + "mappings": { + "properties": { + "Content": { + "analyzer": "words", + "term_vector": "with_positions_offsets", + "type": "text" + }, + "Deleted": { + "type": "boolean" + }, + "Favorites": { + "type": "keyword" + }, + "Hidden": { + "type": "boolean" + }, + "ID": { + "type": "keyword" + }, + "MimeType": { + "doc_values": false, + "type": "wildcard" + }, + "Mtime": { + "type": "date" + }, + "Name": { + "type": "keyword" + }, + "Name_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "Name_words": { + "analyzer": "words", + "type": "text" + }, + "ParentID": { + "type": "keyword" + }, + "Path": { + "analyzer": "path_hierarchy", + "type": "text" + }, + "RootID": { + "type": "keyword" + }, + "Size": { + "type": "long" + }, + "Tags": { + "type": "keyword" + }, + "Tags_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "Title": { + "type": "keyword" + }, + "Title_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "Title_words": { + "analyzer": "words", + "type": "text" + }, + "Type": { + "type": "long" + }, + "audio": { + "properties": { + "album": { + "type": "keyword" + }, + "albumArtist": { + "type": "keyword" + }, + "albumArtist_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "albumArtist_words": { + "analyzer": "words", + "type": "text" + }, + "album_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "album_words": { + "analyzer": "words", + "type": "text" + }, + "artist": { + "type": "keyword" + }, + "artist_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "artist_words": { + "analyzer": "words", + "type": "text" + }, + "bitrate": { + "type": "long" + }, + "composers": { + "type": "keyword" + }, + "composers_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "composers_words": { + "analyzer": "words", + "type": "text" + }, + "copyright": { + "type": "keyword" + }, + "copyright_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "copyright_words": { + "analyzer": "words", + "type": "text" + }, + "disc": { + "type": "integer" + }, + "discCount": { + "type": "integer" + }, + "duration": { + "type": "long" + }, + "genre": { + "type": "keyword" + }, + "genre_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "genre_words": { + "analyzer": "words", + "type": "text" + }, + "hasDrm": { + "type": "boolean" + }, + "isVariableBitrate": { + "type": "boolean" + }, + "title": { + "type": "keyword" + }, + "title_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "title_words": { + "analyzer": "words", + "type": "text" + }, + "track": { + "type": "integer" + }, + "trackCount": { + "type": "integer" + }, + "year": { + "type": "integer" + } + } + }, + "image": { + "properties": { + "height": { + "type": "integer" + }, + "width": { + "type": "integer" + } + } + }, + "location": { + "properties": { + "altitude": { + "type": "double" + }, + "latitude": { + "type": "double" + }, + "longitude": { + "type": "double" + } + } + }, + "location_geopoint": { + "type": "geo_point" + }, + "photo": { + "properties": { + "cameraMake": { + "type": "keyword" + }, + "cameraMake_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "cameraMake_words": { + "analyzer": "words", + "type": "text" + }, + "cameraModel": { + "type": "keyword" + }, + "cameraModel_lowercase": { + "doc_values": false, + "type": "keyword" + }, + "cameraModel_words": { + "analyzer": "words", + "type": "text" + }, + "exposureDenominator": { + "type": "double" + }, + "exposureNumerator": { + "type": "double" + }, + "fNumber": { + "type": "double" + }, + "focalLength": { + "type": "double" + }, + "iso": { + "type": "integer" + }, + "orientation": { + "type": "integer" + }, + "takenDateTime": { + "type": "date" + } + } + } + } + }, + "settings": { + "analysis": { + "analyzer": { + "path_hierarchy": { + "tokenizer": "path_hierarchy", + "type": "custom" + }, + "words": { + "char_filter": [ + "dot_to_space" + ], + "filter": [ + "lowercase" + ], + "tokenizer": "standard", + "type": "custom" + } + }, + "char_filter": { + "dot_to_space": { + "mappings": [ + ". =\u003e \\u0020" + ], + "type": "mapping" + } + }, + "tokenizer": { + "path_hierarchy": { + "type": "path_hierarchy" + } + } + }, + "number_of_replicas": "1", + "number_of_shards": "1" + } +} diff --git a/services/search/pkg/parity/engines_test.go b/services/search/pkg/parity/engines_test.go index 7568dc7409..b49e0a928b 100644 --- a/services/search/pkg/parity/engines_test.go +++ b/services/search/pkg/parity/engines_test.go @@ -98,7 +98,7 @@ func newOpenSearch(name string, fixtures []search.Resource) testEngine { return testEngine{name: "opensearch", unavailable: err.Error()} } - backend, err := opensearch.NewBackend(name, tc.Client()) + backend, err := opensearch.NewBackend(context.Background(), name, tc.Client(), log.NopLogger()) if err != nil { return testEngine{name: "opensearch", unavailable: err.Error()} }