diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index f26e3ff809..0e0e28c85f 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" @@ -17,27 +21,170 @@ 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]interface{}{"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) - 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, nil + searchmapping.LogNewIndexCreated(logger, destination) + return index, searchmapping.Classification{Verdict: searchmapping.VerdictEqual}, nil } + if err != nil { + return nil, searchmapping.Classification{}, 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 +} - return index, err +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 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 { + 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) + if len(reasons) > 0 { + classification.Verdict = searchmapping.VerdictBreaking + classification.Reasons = append(reasons, classification.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) { + 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..ad3b41293e --- /dev/null +++ b/services/search/pkg/bleve/index_test.go @@ -0,0 +1,202 @@ +package bleve_test + +import ( + "encoding/json" + "fmt" + "os" + "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/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" +) + +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 = "fulltext" + 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("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, 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 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 62a1a6795c..83bb5d3799 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,16 @@ 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 + 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..2e0e7e100a --- /dev/null +++ b/services/search/pkg/mapping/classify.go @@ -0,0 +1,164 @@ +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. 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: 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 +} + +// Classify recursively compares a stored `properties` tree against the one +// 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) + 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 +} + +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..c4b4d35817 --- /dev/null +++ b/services/search/pkg/mapping/classify_test.go @@ -0,0 +1,128 @@ +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 + 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/mapping/reconcile.go b/services/search/pkg/mapping/reconcile.go new file mode 100644 index 0000000000..6f8c056127 --- /dev/null +++ b/services/search/pkg/mapping/reconcile.go @@ -0,0 +1,49 @@ +package mapping + +import ( + "github.com/opencloud-eu/opencloud/pkg/log" +) + +// 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 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) +} + +// 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: + 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/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/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 +} diff --git a/services/search/pkg/opensearch/index.go b/services/search/pkg/opensearch/index.go index 18bc0eb99b..5fcf5bd467 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,155 @@ 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 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.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 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) + } + remoteIndexB, err := json.Marshal(remoteIndex) + if err != nil { + return fmt.Errorf("failed to marshal 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)) - } - } + r := &osReconciler{ + ctx: ctx, + name: name, + client: client, + local: gjson.ParseBytes(localIndexB), + remote: gjson.ParseBytes(remoteIndexB), + } + _, err = searchmapping.Reconcile(name, r, logger) + return 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)) - } - } +// 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 +} - 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...), - ) - } +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))) + } - return nil // Index is already up to date, no action needed + classification := searchmapping.Classify( + propertiesMap(r.remote.Get("mappings.properties").Raw), + propertiesMap(r.local.Get("mappings.properties").Raw), + nil, + ) + reasons = append(reasons, classification.Reasons...) + if len(reasons) > 0 { + classification.Verdict = searchmapping.VerdictBreaking + classification.Reasons = reasons } + return classification, nil +} - createResp, err := client.Indices.Create(ctx, opensearchgoAPI.IndicesCreateReq{ - Index: name, - Body: bytes.NewReader(localIndexB), +// 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 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 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. 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 + } + if err := json.Unmarshal([]byte(b), &bv); err != nil { + return false + } + return reflect.DeepEqual(av, bv) +} - return nil +// 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 cdbc01ac85..c7ce1740b7 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,10 +63,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("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.analysis.analyzer.lowercaseKeyword.tokenizer", "standard") + 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("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("tolerates shard drift", func(t *testing.T) { indexManager := opensearch.IndexManagerLatest indexName := "opencloud-test-resource" @@ -73,6 +105,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.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) }) }