sevenhill/QSfera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(search): implement support for versioned os index templates

Browse Source
This commit is contained in:
fschade
2025-08-28 09:30:59 +02:00
parent a7d4ff4872
commit 1c92f3db00
11 changed files with 379 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/search/pkg/opensearch/engine.go
+11 -3
View File
@@ -24,7 +24,9 @@ type Engine struct {
}
func NewEngine(index string, client *opensearchgoAPI.Client) (*Engine, error) {
_, healthy, err := clusterHealth(context.Background(), client, []string{index})
// first check if the cluster is healthy, we cannot expect that the index exists at this point,
// so we pass nil for the indices parameter and only check the cluster health
_, healthy, err := clusterHealth(context.Background(), client, nil)
switch {
case err != nil:
return nil, fmt.Errorf("failed to get cluster health: %w", err)
@@ -32,6 +34,12 @@ func NewEngine(index string, client *opensearchgoAPI.Client) (*Engine, error) {
return nil, fmt.Errorf("cluster health is not healthy")
}
// apply the index template, this will create the index if it does not exist,
// or update it if it does exist
if err := IndexTemplateResourceV1.Apply(context.Background(), client); err != nil {
return nil, fmt.Errorf("failed to apply index template: %w", err)
}
return &Engine{index: index, client: client}, nil
}
@@ -57,7 +65,7 @@ func (e *Engine) Search(ctx context.Context, sir *searchService.SearchIndexReque
if sir.Ref != nil {
boolQuery.Filter(
NewMatchPhraseQuery("RootID").Query(
NewTermQuery[string]("RootID").Value(
storagespace.FormatResourceID(
&storageProvider.ResourceId{
StorageId: sir.Ref.GetResourceId().GetStorageId(),
@@ -74,7 +82,7 @@ func (e *Engine) Search(ctx context.Context, sir *searchService.SearchIndexReque
return nil, fmt.Errorf("failed to marshal query: %w", err)
}
resp, err := e.client.Search(context.Background(), &opensearchgoAPI.SearchReq{
resp, err := e.client.Search(ctx, &opensearchgoAPI.SearchReq{
Indices: []string{e.index},
Body: bytes.NewReader(body),
})
services/search/pkg/opensearch/engine_convert.go
+90
View File
@@ -0,0 +1,90 @@
package opensearch
import (
"fmt"
"strings"
"time"
"github.com/opencloud-eu/reva/v2/pkg/storagespace"
opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
"google.golang.org/protobuf/types/known/timestamppb"
searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
)
func searchHitToSearchMessageMatch(hit opensearchgoAPI.SearchHit) (*searchMessage.Match, error) {
resource, err := convert[engine.Resource](hit.Source)
if err != nil {
return nil, fmt.Errorf("failed to convert hit source: %w", err)
}
resourceRootID, err := storagespace.ParseID(resource.RootID)
if err != nil {
return nil, err
}
resourceID, err := storagespace.ParseID(resource.ID)
if err != nil {
return nil, err
}
resourceParentID, _ := storagespace.ParseID(resource.ParentID)
match := &searchMessage.Match{
Score: hit.Score,
Entity: &searchMessage.Entity{
Ref: &searchMessage.Reference{
ResourceId: &searchMessage.ResourceID{
StorageId: resourceRootID.GetStorageId(),
SpaceId: resourceRootID.GetSpaceId(),
OpaqueId: resourceRootID.GetOpaqueId(),
},
Path: resource.Path,
},
Id: &searchMessage.ResourceID{
StorageId: resourceID.GetStorageId(),
SpaceId: resourceID.GetSpaceId(),
OpaqueId: resourceID.GetOpaqueId(),
},
Name: resource.Name,
ParentId: &searchMessage.ResourceID{
StorageId: resourceParentID.GetStorageId(),
SpaceId: resourceParentID.GetSpaceId(),
OpaqueId: resourceParentID.GetOpaqueId(),
},
Size: resource.Size,
Type: resource.Type,
MimeType: resource.MimeType,
Deleted: resource.Deleted,
Tags: resource.Tags,
//Highlights: getFragmentValue(hit.Fragments, "Content", 0),
Audio: func() *searchMessage.Audio {
if !strings.HasPrefix(resource.MimeType, "audio/") {
return nil
}
audio, _ := convert[*searchMessage.Audio](resource.Audio)
return audio
}(),
Image: func() *searchMessage.Image {
image, _ := convert[*searchMessage.Image](resource.Image)
return image
}(),
Location: func() *searchMessage.GeoCoordinates {
geoCoordinates, _ := convert[*searchMessage.GeoCoordinates](resource.Location)
return geoCoordinates
}(),
Photo: func() *searchMessage.Photo {
photo, _ := convert[*searchMessage.Photo](resource.Photo)
return photo
}(),
},
}
if mtime, err := time.Parse(time.RFC3339, resource.Mtime); err == nil {
match.Entity.LastModifiedTime = &timestamppb.Timestamp{Seconds: mtime.Unix(), Nanos: int32(mtime.Nanosecond())}
}
return match, nil
}
services/search/pkg/opensearch/engine_convert_test.go
+37
View File
@@ -0,0 +1,37 @@
package opensearch_test
import (
"encoding/json"
"testing"
opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
"github.com/stretchr/testify/assert"
searchMessage "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/search/v0"
"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test"
)
func TestSearchHitToSearchMessageMatch(t *testing.T) {
resource := opensearchtest.Testdata.Resources.Full
resource.MimeType = "audio/anything"
hit := opensearchgoAPI.SearchHit{
Score: 1.1,
Source: json.RawMessage(opensearchtest.ToJSON(t, resource)),
}
match, err := opensearch.SearchHitToSearchMessageMatch(hit)
assert.NoError(t, err)
assert.Equal(t, hit.Score, match.Score)
assert.Equal(t, resource.Name, match.Entity.Name)
t.Run("converts the audio field to the expected type", func(t *testing.T) {
// searchMessage.Audio contains int64, int32 ... values that are converted to strings by the JSON marshaler,
// so we need to convert the resource.Audio to align the expectations for the JSON comparison.
audio, err := opensearch.Convert[*searchMessage.Audio](resource.Audio)
assert.NoError(t, err)
assert.Equal(t, resource.Audio.Bitrate, match.Entity.Audio.Bitrate)
assert.JSONEq(t, opensearchtest.ToJSON(t, audio), opensearchtest.ToJSON(t, match.Entity.Audio))
})
}
services/search/pkg/opensearch/engine_test.go
+28 -35
View File
@@ -6,7 +6,6 @@ import (
opensearchgo "github.com/opensearch-project/opensearch-go/v4"
opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
searchService "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
@@ -24,17 +23,16 @@ func TestNewEngine(t *testing.T) {
require.NoError(t, err, "failed to create OpenSearch client")
engine, err := opensearch.NewEngine("test-engine-new-engine", client)
assert.Nil(t, engine)
assert.ErrorIs(t, err, opensearch.ErrUnhealthyCluster)
require.Nil(t, engine)
require.ErrorIs(t, err, opensearch.ErrUnhealthyCluster)
})
}
func TestEngine_Search(t *testing.T) {
index := "test-engine-search"
index := "opencloud-default-resource"
tc := opensearchtest.NewDefaultTestClient(t)
tc.Require.IndicesReset([]string{index})
tc.Require.IndicesCount([]string{index}, "", 0)
tc.Require.IndicesCreate(index, "")
defer tc.Require.IndicesDelete([]string{index})
@@ -43,16 +41,16 @@ func TestEngine_Search(t *testing.T) {
tc.Require.IndicesCount([]string{index}, "", 1)
engine, err := opensearch.NewEngine(index, tc.Client())
assert.NoError(t, err)
require.NoError(t, err)
t.Run("most simple search", func(t *testing.T) {
resp, err := engine.Search(t.Context(), &searchService.SearchIndexRequest{
Query: fmt.Sprintf(`"%s"`, document.Name),
})
assert.NoError(t, err)
require.NoError(t, err)
require.Len(t, resp.Matches, 1)
assert.Equal(t, int32(1), resp.TotalMatches)
assert.Equal(t, document.ID, fmt.Sprintf("%s$%s!%s", resp.Matches[0].Entity.Id.StorageId, resp.Matches[0].Entity.Id.SpaceId, resp.Matches[0].Entity.Id.OpaqueId))
require.Equal(t, int32(1), resp.TotalMatches)
require.Equal(t, document.ID, fmt.Sprintf("%s$%s!%s", resp.Matches[0].Entity.Id.StorageId, resp.Matches[0].Entity.Id.SpaceId, resp.Matches[0].Entity.Id.OpaqueId))
})
t.Run("ignores files that are marked as deleted", func(t *testing.T) {
@@ -66,28 +64,27 @@ func TestEngine_Search(t *testing.T) {
resp, err := engine.Search(t.Context(), &searchService.SearchIndexRequest{
Query: fmt.Sprintf(`"%s"`, document.Name),
})
assert.NoError(t, err)
require.NoError(t, err)
require.Len(t, resp.Matches, 1)
assert.Equal(t, int32(1), resp.TotalMatches)
assert.Equal(t, document.ID, fmt.Sprintf("%s$%s!%s", resp.Matches[0].Entity.Id.StorageId, resp.Matches[0].Entity.Id.SpaceId, resp.Matches[0].Entity.Id.OpaqueId))
require.Equal(t, int32(1), resp.TotalMatches)
require.Equal(t, document.ID, fmt.Sprintf("%s$%s!%s", resp.Matches[0].Entity.Id.StorageId, resp.Matches[0].Entity.Id.SpaceId, resp.Matches[0].Entity.Id.OpaqueId))
})
}
func TestEngine_Upsert(t *testing.T) {
index := "test-engine-upsert"
index := "opencloud-default-resource"
tc := opensearchtest.NewDefaultTestClient(t)
tc.Require.IndicesReset([]string{index})
tc.Require.IndicesCount([]string{index}, "", 0)
tc.Require.IndicesCreate(index, "")
defer tc.Require.IndicesDelete([]string{index})
engine, err := opensearch.NewEngine(index, tc.Client())
assert.NoError(t, err)
require.NoError(t, err)
t.Run("upsert with full document", func(t *testing.T) {
document := opensearchtest.Testdata.Resources.Full
assert.NoError(t, engine.Upsert(document.ID, document))
require.NoError(t, engine.Upsert(document.ID, document))
tc.Require.IndicesCount([]string{index}, "", 1)
})
@@ -96,16 +93,15 @@ func TestEngine_Upsert(t *testing.T) {
func TestEngine_Move(t *testing.T) {}
func TestEngine_Delete(t *testing.T) {
index := "test-engine-delete"
index := "opencloud-default-resource"
tc := opensearchtest.NewDefaultTestClient(t)
tc.Require.IndicesReset([]string{index})
tc.Require.IndicesCount([]string{index}, "", 0)
tc.Require.IndicesCreate(index, "")
defer tc.Require.IndicesDelete([]string{index})
engine, err := opensearch.NewEngine(index, tc.Client())
assert.NoError(t, err)
require.NoError(t, err)
t.Run("mark document as deleted", func(t *testing.T) {
document := opensearchtest.Testdata.Resources.Full
@@ -116,7 +112,7 @@ func TestEngine_Delete(t *testing.T) {
opensearch.NewTermQuery[bool]("Deleted").Value(true),
).String(), 0)
assert.NoError(t, engine.Delete(document.ID))
require.NoError(t, engine.Delete(document.ID))
tc.Require.IndicesCount([]string{index}, opensearch.NewRootQuery(
opensearch.NewTermQuery[bool]("Deleted").Value(true),
).String(), 1)
@@ -124,16 +120,15 @@ func TestEngine_Delete(t *testing.T) {
}
func TestEngine_Restore(t *testing.T) {
index := "test-engine-restore"
index := "opencloud-default-resource"
tc := opensearchtest.NewDefaultTestClient(t)
tc.Require.IndicesReset([]string{index})
tc.Require.IndicesCount([]string{index}, "", 0)
tc.Require.IndicesCreate(index, "")
defer tc.Require.IndicesDelete([]string{index})
engine, err := opensearch.NewEngine(index, tc.Client())
assert.NoError(t, err)
require.NoError(t, err)
t.Run("mark document as not deleted", func(t *testing.T) {
document := opensearchtest.Testdata.Resources.Full
@@ -145,7 +140,7 @@ func TestEngine_Restore(t *testing.T) {
opensearch.NewTermQuery[bool]("Deleted").Value(true),
).String(), 1)
assert.NoError(t, engine.Restore(document.ID))
require.NoError(t, engine.Restore(document.ID))
tc.Require.IndicesCount([]string{index}, opensearch.NewRootQuery(
opensearch.NewTermQuery[bool]("Deleted").Value(true),
).String(), 0)
@@ -153,39 +148,37 @@ func TestEngine_Restore(t *testing.T) {
}
func TestEngine_Purge(t *testing.T) {
index := "test-engine-purge"
index := "opencloud-default-resource"
tc := opensearchtest.NewDefaultTestClient(t)
tc.Require.IndicesReset([]string{index})
tc.Require.IndicesCount([]string{index}, "", 0)
tc.Require.IndicesCreate(index, "")
defer tc.Require.IndicesDelete([]string{index})
engine, err := opensearch.NewEngine(index, tc.Client())
assert.NoError(t, err)
require.NoError(t, err)
t.Run("purge with full document", func(t *testing.T) {
document := opensearchtest.Testdata.Resources.Full
tc.Require.DocumentCreate(index, document.ID, opensearchtest.ToJSON(t, document))
tc.Require.IndicesCount([]string{index}, "", 1)
assert.NoError(t, engine.Purge(document.ID))
require.NoError(t, engine.Purge(document.ID))
tc.Require.IndicesCount([]string{index}, "", 0)
})
}
func TestEngine_DocCount(t *testing.T) {
index := "test-engine-doc-count"
index := "opencloud-default-resource"
tc := opensearchtest.NewDefaultTestClient(t)
tc.Require.IndicesReset([]string{index})
tc.Require.IndicesCount([]string{index}, "", 0)
tc.Require.IndicesCreate(index, "")
defer tc.Require.IndicesDelete([]string{index})
engine, err := opensearch.NewEngine(index, tc.Client())
assert.NoError(t, err)
require.NoError(t, err)
t.Run("ignore deleted documents", func(t *testing.T) {
document := opensearchtest.Testdata.Resources.Full
@@ -193,8 +186,8 @@ func TestEngine_DocCount(t *testing.T) {
tc.Require.IndicesCount([]string{index}, "", 1)
count, err := engine.DocCount()
assert.NoError(t, err)
assert.Equal(t, uint64(1), count)
require.NoError(t, err)
require.Equal(t, uint64(1), count)
tc.Require.Update(index, document.ID, opensearchtest.ToJSON(t, map[string]any{
"doc": map[string]any{
@@ -205,7 +198,7 @@ func TestEngine_DocCount(t *testing.T) {
tc.Require.IndicesCount([]string{index}, "", 1)
count, err = engine.DocCount()
assert.NoError(t, err)
assert.Equal(t, uint64(0), count)
require.NoError(t, err)
require.Equal(t, uint64(0), count)
})
}
services/search/pkg/opensearch/export_test.go
+10
View File
@@ -0,0 +1,10 @@
package opensearch
var (
SearchHitToSearchMessageMatch = searchHitToSearchMessageMatch
BuilderToBoolQuery = builderToBoolQuery
)
func Convert[T any](v any) (T, error) {
return convert[T](v)
}
services/search/pkg/opensearch/internal/indices/resource_v1.json
+34
View File
@@ -0,0 +1,34 @@
{
"index_patterns": [
"opencloud-default-resource"
],
"template": {
"settings": {
"number_of_shards": 1,
"number_of_replicas": 1
},
"mappings": {
"properties": {
"ID": {
"type": "keyword"
},
"ParentID": {
"type": "keyword"
},
"RootID": {
"type": "keyword"
},
"Deleted": {
"type": "boolean"
},
"Hidden": {
"type": "boolean"
}
}
}
},
"version": 1,
"_meta": {
"description": "using component templates"
}
}
services/search/pkg/opensearch/internal/test/os.go
+4 -1
View File
@@ -89,7 +89,10 @@ func (tc *TestClient) IndicesRefresh(ctx context.Context, indices []string, allo
Indices: indices,
})
isAllowed := resp != nil && slices.Contains(allow, resp.Inspect().Response.StatusCode)
isAllowed := resp != nil
isAllowed = isAllowed && resp.Inspect().Response != nil
isAllowed = isAllowed && slices.Contains(allow, resp.Inspect().Response.StatusCode)
if err != nil && !isAllowed {
return fmt.Errorf("failed to refresh indices %v: %w", indices, err)
}
services/search/pkg/opensearch/os.go
+30
View File
@@ -0,0 +1,30 @@
package opensearch
import (
"context"
"fmt"
"time"
opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
)
func clusterHealth(ctx context.Context, client *opensearchgoAPI.Client, indices []string) (*opensearchgoAPI.ClusterHealthResp, bool, error) {
resp, err := client.Cluster.Health(ctx, &opensearchgoAPI.ClusterHealthReq{
Indices: indices,
Params: opensearchgoAPI.ClusterHealthParams{
Local: opensearchgoAPI.ToPointer(true),
Timeout: 5 * time.Second,
},
})
switch {
case err != nil:
return nil, false, fmt.Errorf("%w, failed to get cluster health: %w", ErrUnhealthyCluster, err)
case resp.TimedOut:
return resp, false, fmt.Errorf("%w, cluster health request timed out", ErrUnhealthyCluster)
case resp.Status != "green" && resp.Status != "yellow":
return resp, false, fmt.Errorf("%w, cluster health is not green or yellow: %s", ErrUnhealthyCluster, resp.Status)
default:
return resp, true, nil
}
}
services/search/pkg/opensearch/os_dsl_test.go
+36
View File
@@ -0,0 +1,36 @@
package opensearch_test
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test"
)
func TestBuilderToBoolQuery(t *testing.T) {
tests := []opensearchtest.TableTest[opensearch.Builder, *opensearch.BoolQuery]{
{
Name: "term-query",
Got: opensearch.NewTermQuery[string]("Name").Value("openCloud"),
Want: opensearch.NewBoolQuery().Must(opensearch.NewTermQuery[string]("Name").Value("openCloud")),
},
{
Name: "root-query",
Got: opensearch.NewRootQuery(opensearch.NewTermQuery[string]("Name").Value("openCloud")),
Want: opensearch.NewBoolQuery().Must(opensearch.NewTermQuery[string]("Name").Value("openCloud")),
},
{
Name: "bool-query",
Got: opensearch.NewBoolQuery().Must(opensearch.NewTermQuery[string]("Name").Value("openCloud")),
Want: opensearch.NewBoolQuery().Must(opensearch.NewTermQuery[string]("Name").Value("openCloud")),
},
}
for _, test := range tests {
t.Run(test.Name, func(t *testing.T) {
assert.JSONEq(t, opensearchtest.ToJSON(t, test.Want), opensearchtest.ToJSON(t, opensearch.BuilderToBoolQuery(test.Got)))
})
}
}
services/search/pkg/opensearch/os_index_template.go
+65
View File
@@ -0,0 +1,65 @@
package opensearch
import (
"bytes"
"context"
"embed"
"fmt"
"path"
opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
)
var (
IndexTemplateResourceV1 IndexTemplate = [2]string{"opencloud-default-resource", "resource_v1.json"}
)
//go:embed internal/indices/*.json
var indexTemplates embed.FS
type IndexTemplate [2]string
func (t IndexTemplate) Name() string {
return t[0]
}
func (t IndexTemplate) String() string {
b, err := t.MarshalJSON()
if err != nil {
return ""
}
return string(b)
}
func (t IndexTemplate) MarshalJSON() ([]byte, error) {
file := t[1]
body, err := indexTemplates.ReadFile(path.Join("./internal/indices", file))
switch {
case err != nil:
return nil, fmt.Errorf("failed to read index template file %s: %w", file, err)
case len(body) <= 0:
return nil, fmt.Errorf("index template file %s is empty", file)
}
return body, nil
}
func (t IndexTemplate) Apply(ctx context.Context, client *opensearchgoAPI.Client) error {
body, err := t.MarshalJSON()
if err != nil {
return fmt.Errorf("failed to inspect index template %s: %w", t[1], err)
}
resp, err := client.IndexTemplate.Create(ctx, opensearchgoAPI.IndexTemplateCreateReq{
IndexTemplate: t.Name(),
Body: bytes.NewBuffer(body),
})
switch {
case err != nil:
return fmt.Errorf("failed to create index template %s: %w", t.Name(), err)
case !resp.Acknowledged:
return fmt.Errorf("failed to create index template %s: not acknowledged", t.Name())
default:
return nil
}
}
services/search/pkg/opensearch/os_index_template_test.go
+34
View File
@@ -0,0 +1,34 @@
package opensearch_test
import (
"testing"
"github.com/stretchr/testify/require"
"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
opensearchtest "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test"
)
func TestIndexTemplates(t *testing.T) {
tc := opensearchtest.NewDefaultTestClient(t)
t.Run("index templates plausibility", func(t *testing.T) {
tests := []opensearchtest.TableTest[opensearch.IndexTemplate, struct{}]{
{
Name: "empty",
Got: opensearch.IndexTemplateResourceV1,
},
}
for _, test := range tests {
t.Run(test.Name, func(t *testing.T) {
body, err := test.Got.MarshalJSON()
require.NoError(t, err)
require.NotEmpty(t, body)
require.NotEmpty(t, test.Got.String())
require.JSONEq(t, test.Got.String(), string(body))
require.NotEmpty(t, test.Got.Name())
require.NoError(t, test.Got.Apply(t.Context(), tc.Client()))
})
}
})
}