From d7b8feaa187ed4a6fbe9d62040948279dbbd4dca Mon Sep 17 00:00:00 2001 From: Dominik Schmidt Date: Thu, 16 Jul 2026 01:20:28 +0200 Subject: [PATCH] fix(search): report rejected documents when indexing to OpenSearch --- services/search/pkg/opensearch/batch.go | 13 ++++- services/search/pkg/opensearch/batch_test.go | 53 ++++++++++++++++++++ 2 files changed, 64 insertions(+), 2 deletions(-) create mode 100644 services/search/pkg/opensearch/batch_test.go diff --git a/services/search/pkg/opensearch/batch.go b/services/search/pkg/opensearch/batch.go index ff76b06b16..20c890826e 100644 --- a/services/search/pkg/opensearch/batch.go +++ b/services/search/pkg/opensearch/batch.go @@ -210,11 +210,20 @@ func (b *Batch) Push() error { body.WriteString("\n") } - if _, err := b.client.Bulk(context.Background(), opensearchgoAPI.BulkReq{ + resp, err := b.client.Bulk(context.Background(), opensearchgoAPI.BulkReq{ Body: strings.NewReader(body.String()), Params: opensearchgoAPI.BulkParams{Refresh: "wait_for"}, - }); err != nil { + }) + switch { + case err != nil: return fmt.Errorf("failed to execute bulk operations: %w", err) + case resp.Errors: + items, err := json.Marshal(resp.Items) + if err != nil { + return fmt.Errorf("failed to marshal bulk response: %w", err) + } + + return fmt.Errorf("failed to execute bulk operations, response: %s", items) } bulkOperations = nil diff --git a/services/search/pkg/opensearch/batch_test.go b/services/search/pkg/opensearch/batch_test.go new file mode 100644 index 0000000000..efd29e5597 --- /dev/null +++ b/services/search/pkg/opensearch/batch_test.go @@ -0,0 +1,53 @@ +package opensearch_test + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch" + "github.com/opencloud-eu/opencloud/services/search/pkg/opensearch/internal/test" +) + +func TestBatch_Push(t *testing.T) { + tc := opensearchtest.NewDefaultTestClient(t, defaultConfig.Engine.OpenSearch.Client) + + t.Run("reports the documents the bulk API rejected", func(t *testing.T) { + indexName := "opencloud-test-batch-push-rejected" + tc.Require.IndicesReset([]string{indexName}) + defer tc.Require.IndicesDelete([]string{indexName}) + + // Name is a string, mapping it as a long makes every document fail to parse. + tc.Require.IndicesCreate(indexName, strings.NewReader(`{"mappings":{"properties":{"Name":{"type":"long"}}}}`)) + + batch, err := opensearch.NewBatch(tc.Client(), indexName, 10) + require.NoError(t, err) + + document := opensearchtest.Testdata.Resources.File + require.NoError(t, batch.Upsert(document.ID, document)) + + err = batch.Push() + require.Error(t, err) + require.ErrorContains(t, err, document.ID) + require.ErrorContains(t, err, "mapper_parsing_exception") + tc.Require.IndicesCount([]string{indexName}, nil, 0) + }) + + t.Run("pushes the documents the bulk API accepted", func(t *testing.T) { + indexName := "opencloud-test-batch-push-accepted" + tc.Require.IndicesReset([]string{indexName}) + defer tc.Require.IndicesDelete([]string{indexName}) + + tc.Require.IndicesCreate(indexName, strings.NewReader(opensearch.IndexManagerLatest.String())) + + batch, err := opensearch.NewBatch(tc.Client(), indexName, 10) + require.NoError(t, err) + + document := opensearchtest.Testdata.Resources.File + require.NoError(t, batch.Upsert(document.ID, document)) + require.NoError(t, batch.Push()) + + tc.Require.IndicesCount([]string{indexName}, nil, 1) + }) +}