diff --git a/changelog/fragments/1787137647-enroll-search-bulker-queue.yaml b/changelog/fragments/1787137647-enroll-search-bulker-queue.yaml new file mode 100644 index 0000000000..0bf50aa211 --- /dev/null +++ b/changelog/fragments/1787137647-enroll-search-bulker-queue.yaml @@ -0,0 +1,20 @@ +kind: enhancement + +summary: Prevent ghost agent documents caused by concurrent enrollment retries + +description: | + At large scale, concurrent enrollment retries could create duplicate (ghost) agent + documents in Elasticsearch. This occurred when an agent retried enrollment before + its previous write was visible to search, causing multiple retries to each believe + no agent existed and each create a new document. + + Fleet Server now batches enrollment lookups and ensures index visibility before + searching, so retries consistently find an existing agent document rather than + creating a new one. Duplicate requests within a batch are handled efficiently + and prompted to retry, keeping the number of Elasticsearch operations low. + + The batching behaviour is configurable via + inputs[].server.bulk.enroll.flush_interval (default: 1s) and + inputs[].server.bulk.enroll.flush_threshold_cnt (default: 50). + +component: fleet-server diff --git a/internal/pkg/api/error.go b/internal/pkg/api/error.go index f40e512182..b16386c2c0 100644 --- a/internal/pkg/api/error.go +++ b/internal/pkg/api/error.go @@ -205,6 +205,15 @@ func NewHTTPErrResp(err error) HTTPErrResp { zerolog.WarnLevel, }, }, + { + bulk.ErrEnrollDuplicate, + HTTPErrResp{ + http.StatusTooManyRequests, + "EnrollDuplicate", + "concurrent enrollment for same id: retry", + zerolog.DebugLevel, + }, + }, { os.ErrDeadlineExceeded, HTTPErrResp{ diff --git a/internal/pkg/api/handleEnroll.go b/internal/pkg/api/handleEnroll.go index abdcbdd6da..aa3c2b3e24 100644 --- a/internal/pkg/api/handleEnroll.go +++ b/internal/pkg/api/handleEnroll.go @@ -215,8 +215,13 @@ func (et *EnrollerT) _enroll( vSpan, vCtx := apm.StartSpan(ctx, "checkEnrollmentID", "validate") enrollmentID = *req.EnrollmentId var err error - agent, err = dl.FindAgent(vCtx, et.bulker, dl.QueryAgentByEnrollmentID, dl.FieldEnrollmentID, enrollmentID) + agent, err = dl.FindAgent(vCtx, et.bulker, dl.QueryAgentByEnrollmentID, dl.FieldEnrollmentID, enrollmentID, + dl.WithBulkOpts(bulk.WithDedupeKey(enrollmentID, dl.FleetAgents))) if err != nil { + if errors.Is(err, bulk.ErrEnrollDuplicate) { + vSpan.End() + return nil, err + } zlog.Debug().Err(err). Str("EnrollmentId", enrollmentID). Msg("Agent with EnrollmentId not found") diff --git a/internal/pkg/bulk/block.go b/internal/pkg/bulk/block.go index eefe0bd0cf..1bfb7350f5 100644 --- a/internal/pkg/bulk/block.go +++ b/internal/pkg/bulk/block.go @@ -16,14 +16,16 @@ type Buf = danger.Buf // However, the multiOp API's will allocate directly in large blocks. type bulkT struct { - action actionT // requested actions - flags flagsT // execution flags - idx int32 // idx of originating request, used in mulitOp - ch chan respT // response channel, caller is waiting synchronously - buf Buf // json payload to be sent to elastic - next *bulkT // pointer to next bulkT, used for fast internal queueing - spanLink apm.SpanLink - hasSpanLink bool + action actionT // requested actions + flags flagsT // execution flags + idx int32 // idx of originating request, used in mulitOp + ch chan respT // response channel, caller is waiting synchronously + buf Buf // json payload to be sent to elastic + next *bulkT // pointer to next bulkT, used for fast internal queueing + spanLink apm.SpanLink + hasSpanLink bool + dedupeKey string // enrollment dedup key (enrollment_id); routes to kQueueEnrollSearch when set + refreshIndex string // index to refresh before msearch in kQueueEnrollSearch } type flagsT int8 @@ -79,6 +81,8 @@ func (blk *bulkT) reset() { blk.next = nil blk.spanLink = apm.SpanLink{} blk.hasSpanLink = false + blk.dedupeKey = "" + blk.refreshIndex = "" } type respT struct { diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 75ed0bcd23..28296cd63e 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -40,6 +40,7 @@ type APIKeyMetadata = apikey.APIKeyMetadata var ( ErrNoQuotes = errors.New("quoted literal not supported") ErrTooManyBulkDispatches = errors.New("too many pending bulk dispatches") + ErrEnrollDuplicate = errors.New("enrollment deduplicated: retry") ) type MultiOp struct { @@ -365,6 +366,10 @@ func stopTimer(t *time.Timer) { func blkToQueueType(blk *bulkT) queueType { queueIdx := kQueueBulk + if blk.action == ActionSearch && blk.dedupeKey != "" { + return kQueueEnrollSearch + } + forceRefresh := blk.flags.Has(flagRefresh) switch blk.action { @@ -399,6 +404,12 @@ func (b *Bulker) Run(ctx context.Context) error { stopTimer(timer) defer timer.Stop() + // Separate timer and counter for the enrollment search queue so it can flush + // independently at a shorter interval without affecting other queues. + enrollTimer := time.NewTimer(b.opts.enrollFlushInterval) + stopTimer(enrollTimer) + defer enrollTimer.Stop() + w := semaphore.NewWeighted(int64(b.opts.maxPending)) var queues [kNumQueues]queueT @@ -410,29 +421,42 @@ func (b *Bulker) Run(ctx context.Context) error { var itemCnt int var byteCnt int + var enrollItemCnt int doFlush := func() error { - for i := range queues { + if queueType(i) == kQueueEnrollSearch { + continue // flushed independently by doFlushEnroll + } q := &queues[i] if q.pending > 0 { - // Pass queue structure by value if err := b.flushQueue(ctx, w, *q); err != nil { return err } - // Reset local queue stored in array q.cnt = 0 q.head = nil q.pending = 0 } } - // Reset threshold counters itemCnt = 0 byteCnt = 0 + return nil + } + doFlushEnroll := func() error { + q := &queues[kQueueEnrollSearch] + if q.pending > 0 { + if err := b.flushQueue(ctx, w, *q); err != nil { + return err + } + q.cnt = 0 + q.head = nil + q.pending = 0 + } + enrollItemCnt = 0 return nil } @@ -453,28 +477,50 @@ func (b *Bulker) Run(ctx context.Context) error { q.cnt += 1 q.pending += blk.buf.Len() - // Update threshold counters - itemCnt += 1 - byteCnt += blk.buf.Len() - - // Start timer on first queued item - if itemCnt == 1 { - timer.Reset(b.opts.flushInterval) - } + if queueIdx == kQueueEnrollSearch { + enrollItemCnt++ + if enrollItemCnt == 1 { + enrollTimer.Reset(b.opts.enrollFlushInterval) + } + if enrollItemCnt >= b.opts.enrollFlushThresholdCnt { + zerolog.Ctx(ctx).Trace(). + Str("mod", kModBulk). + Int("enrollItemCnt", enrollItemCnt). + Msg("Flush enroll search on threshold") + err = doFlushEnroll() + stopTimer(enrollTimer) + } + } else { + // Update threshold counters + itemCnt += 1 + byteCnt += blk.buf.Len() + + // Start timer on first queued item + if itemCnt == 1 { + timer.Reset(b.opts.flushInterval) + } - // Threshold test, short circuit timer on pending count - if itemCnt >= b.opts.flushThresholdCnt || byteCnt >= b.opts.flushThresholdSz { - zerolog.Ctx(ctx).Trace(). - Str("mod", kModBulk). - Int("itemCnt", itemCnt). - Int("byteCnt", byteCnt). - Msg("Flush on threshold") + // Threshold test, short circuit timer on pending count + if itemCnt >= b.opts.flushThresholdCnt || byteCnt >= b.opts.flushThresholdSz { + zerolog.Ctx(ctx).Trace(). + Str("mod", kModBulk). + Int("itemCnt", itemCnt). + Int("byteCnt", byteCnt). + Msg("Flush on threshold") - err = doFlush() + err = doFlush() - stopTimer(timer) + stopTimer(timer) + } } + case <-enrollTimer.C: + zerolog.Ctx(ctx).Trace(). + Str("mod", kModBulk). + Int("enrollItemCnt", enrollItemCnt). + Msg("Flush enroll search on timer") + err = doFlushEnroll() + case <-timer.C: zerolog.Ctx(ctx).Trace(). Str("mod", kModBulk). @@ -550,6 +596,8 @@ func (b *Bulker) flushQueue(ctx context.Context, w *semaphore.Weighted, queue qu err = b.flushRead(flushCtx, queue) case kQueueSearch, kQueueFleetSearch: err = b.flushSearch(flushCtx, queue) + case kQueueEnrollSearch: + err = b.flushEnrollSearch(flushCtx, queue) case kQueueAPIKeyUpdate: err = b.flushUpdateAPIKey(flushCtx, queue) default: @@ -605,6 +653,8 @@ func (b *Bulker) newBlk(action actionT, opts optionsT) *bulkT { } blk.spanLink = opts.spanLink blk.hasSpanLink = opts.hasSpanLink + blk.dedupeKey = opts.DedupeKey + blk.refreshIndex = opts.RefreshIndex return blk } diff --git a/internal/pkg/bulk/enroll_search_integration_test.go b/internal/pkg/bulk/enroll_search_integration_test.go new file mode 100644 index 0000000000..2cd7616dbe --- /dev/null +++ b/internal/pkg/bulk/enroll_search_integration_test.go @@ -0,0 +1,74 @@ +// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one +// or more contributor license agreements. Licensed under the Elastic License 2.0; +// you may not use this file except in compliance with the Elastic License 2.0. + +//go:build integration + +package bulk + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + testlog "github.com/elastic/fleet-server/v7/internal/pkg/testing/log" +) + +// TestEnrollSearchDedup verifies that concurrent FindAgent searches sharing the +// same enrollment_id are de-duplicated within a single flush batch: exactly one +// request (the oldest, FIFO) is executed against ES and the rest receive +// ErrEnrollDuplicate so the agent handlers know to retry. +func TestEnrollSearchDedup(t *testing.T) { + const ( + numConcurrent = 10 + // Set threshold high enough that the timer drives the flush, not item count. + enrollThreshold = numConcurrent + 1 + enrollInterval = 300 * time.Millisecond + ) + + ctx := t.Context() + ctx = testlog.SetLogger(t).WithContext(ctx) + + // enrollThreshold > numConcurrent ensures all goroutines land in one batch. + index, bulker := SetupIndexWithBulk(ctx, t, testPolicy, + WithEnrollFlushThresholdCount(enrollThreshold), + WithEnrollFlushInterval(enrollInterval), + ) + + // Write a doc so the canonical search has something to find. + sample := NewRandomSample() + _, err := bulker.Create(ctx, index, "", sample.marshal(t), WithRefresh()) + require.NoError(t, err) + + dsl := []byte(`{"query":{"match_all":{}}}`) + dedupeKey := "test-enrollment-id-dedup" + + errs := make([]error, numConcurrent) + var wg sync.WaitGroup + wg.Add(numConcurrent) + for i := range numConcurrent { + go func(i int) { + defer wg.Done() + _, searchErr := bulker.Search(ctx, index, dsl, WithDedupeKey(dedupeKey, index)) + errs[i] = searchErr + }(i) + } + wg.Wait() + + var dupes, successes int + for _, e := range errs { + if errors.Is(e, ErrEnrollDuplicate) { + dupes++ + } else { + assert.NoError(t, e, "canonical request should not error") + successes++ + } + } + + assert.Equal(t, 1, successes, "exactly one canonical request should succeed") + assert.Equal(t, numConcurrent-1, dupes, "all other requests should receive ErrEnrollDuplicate") +} diff --git a/internal/pkg/bulk/opSearch.go b/internal/pkg/bulk/opSearch.go index 6ed83b6576..78ba3f97f6 100644 --- a/internal/pkg/bulk/opSearch.go +++ b/internal/pkg/bulk/opSearch.go @@ -243,3 +243,137 @@ func (b *Bulker) flushSearch(ctx context.Context, queue queueT) error { return nil } + +// flushEnrollSearch handles the kQueueEnrollSearch queue. It: +// 1. Groups items by dedupeKey — oldest (FIFO) occurrence is canonical, the rest are duplicates. +// 2. Refreshes all unique refreshIndex values in the batch. +// 3. Sends an msearch containing only the canonical items. +// 4. Dispatches results to canonical items; sends ErrEnrollDuplicate to duplicates (or the +// canonical's error if the search itself failed, to avoid hiding operational failures). +func (b *Bulker) flushEnrollSearch(ctx context.Context, queue queueT) error { + // Collect items in LIFO order from the queue (queue.head is newest), then reverse to FIFO + // so the oldest request becomes canonical for each dedupeKey. + var all []*bulkT + for n := queue.head; n != nil; n = n.next { + all = append(all, n) + } + for i, j := 0, len(all)-1; i < j; i, j = i+1, j-1 { + all[i], all[j] = all[j], all[i] + } + + // Group items: oldest occurrence of each dedupeKey is canonical, rest are dupes. + dupesByKey := make(map[string][]*bulkT) + var canonicals []*bulkT + for _, n := range all { + key := n.dedupeKey + if _, seen := dupesByKey[key]; !seen { + dupesByKey[key] = nil // mark key as seen; nil slice = no dupes yet + canonicals = append(canonicals, n) + } else { + dupesByKey[key] = append(dupesByKey[key], n) + } + } + + if len(canonicals) == 0 { + return nil + } + + // Refresh all unique indices in the batch before searching so retries see the most recent writes. + refreshIndices := make(map[string]struct{}) + for _, n := range canonicals { + if n.refreshIndex != "" { + refreshIndices[n.refreshIndex] = struct{}{} + } + } + if len(refreshIndices) > 0 { + idxSlice := make([]string, 0, len(refreshIndices)) + for idx := range refreshIndices { + idxSlice = append(idxSlice, idx) + } + refreshResp, err := esapi.IndicesRefreshRequest{Index: idxSlice}.Do(ctx, b.es) + if err != nil { + failQueue(queue, err) + return nil + } + if refreshResp.Body != nil { + refreshResp.Body.Close() + } + if refreshResp.IsError() { + err = fmt.Errorf("enroll search refresh failed: %s", refreshResp.String()) + failQueue(queue, err) + return nil + } + } + + // Build msearch body from canonical items only. + const kRoughEstimatePerItem = 256 + bufSz := max(len(canonicals)*kRoughEstimatePerItem, queue.pending) + buf := b.flushBufPool.Get().(*bytes.Buffer) //nolint:errcheck // we control what is placed in the pool + buf.Reset() + buf.Grow(bufSz) + defer b.flushBufPool.Put(buf) + + for _, n := range canonicals { + buf.Write(n.buf.Bytes()) + } + + span, ctx := apm.StartSpanOptions(ctx, flushSpanNames[queue.ty], queue.Type(), apm.SpanOptions{}) + defer span.End() + + msearchReq := esapi.MsearchRequest{Body: bytes.NewReader(buf.Bytes())} + res, err := msearchReq.Do(ctx, b.es) + if err != nil { + failQueue(queue, err) + return nil + } + if res.Body != nil { + defer res.Body.Close() + } + if res.IsError() { + err = parseError(res, zerolog.Ctx(ctx)) + failQueue(queue, err) + return nil + } + + buf.Reset() + if _, err = buf.ReadFrom(res.Body); err != nil { + return err + } + + var blk MsearchResponse + blk.Responses = make([]MsearchResponseItem, 0, len(canonicals)) + if err = easyjson.Unmarshal(buf.Bytes(), &blk); err != nil { + return err + } + if len(blk.Responses) != len(canonicals) { + return fmt.Errorf("enroll search queue length mismatch: got %d, want %d", len(blk.Responses), len(canonicals)) + } + + // WARNING: Once we start pushing items to the queue, the node pointers are invalid. + // Save any fields we need after the channel send before sending — the receiver + // calls freeBlk immediately on receipt, which races with any subsequent read of n. + for i, n := range canonicals { + response := &blk.Responses[i] + respErr := response.deriveError() + key := n.dedupeKey // must be captured before n.ch send + select { + case n.ch <- respT{err: respErr, idx: n.idx, data: response}: + default: + panic("Unexpected blocked response channel on flushEnrollSearch canonical") + } + // If the canonical's search itself failed, propagate that error to duplicates + // instead of ErrEnrollDuplicate so callers don't suppress real failures. + dupeErr := ErrEnrollDuplicate + if respErr != nil { + dupeErr = respErr + } + for _, dupe := range dupesByKey[key] { + select { + case dupe.ch <- respT{err: dupeErr}: + default: + panic("Unexpected blocked response channel on flushEnrollSearch dupe") + } + } + } + return nil +} diff --git a/internal/pkg/bulk/opt.go b/internal/pkg/bulk/opt.go index 751865dead..d0d16c0ac8 100644 --- a/internal/pkg/bulk/opt.go +++ b/internal/pkg/bulk/opt.go @@ -26,6 +26,8 @@ type optionsT struct { IgnoreUnavailable bool spanLink apm.SpanLink hasSpanLink bool + DedupeKey string + RefreshIndex string } type Opt func(*optionsT) @@ -63,6 +65,16 @@ func WithWaitForCheckpoints(checkpoints []int64) Opt { } } +// WithDedupeKey routes a search through kQueueEnrollSearch, which refreshes +// refreshIndex before executing the msearch and de-dupes concurrent requests +// with the same key, returning ErrEnrollDuplicate to all but the first. +func WithDedupeKey(key, refreshIndex string) Opt { + return func(opt *optionsT) { + opt.DedupeKey = key + opt.RefreshIndex = refreshIndex + } +} + //----- // Bulk API options @@ -78,6 +90,8 @@ type bulkOptT struct { maxConcurrentSecretReads int policyTokens []config.PolicyToken bi build.Info + enrollFlushInterval time.Duration + enrollFlushThresholdCnt int } type BulkOpt func(*bulkOptT) @@ -162,6 +176,17 @@ func WithBi(bi build.Info) BulkOpt { } } +// WithEnrollFlushInterval sets the flush interval for the kQueueEnrollSearch queue. +func WithEnrollFlushInterval(d time.Duration) BulkOpt { + return func(opt *bulkOptT) { opt.enrollFlushInterval = d } +} + +// WithEnrollFlushThresholdCount sets the item count that triggers an early flush +// of the kQueueEnrollSearch queue. +func WithEnrollFlushThresholdCount(cnt int) BulkOpt { + return func(opt *bulkOptT) { opt.enrollFlushThresholdCnt = cnt } +} + func parseBulkOpts(opts ...BulkOpt) bulkOptT { bopt := bulkOptT{ flushInterval: defaultFlushInterval, @@ -174,6 +199,8 @@ func parseBulkOpts(opts ...BulkOpt) bulkOptT { maxPendingBulkDispatches: defaultMaxPendingBulkDispatches, maxConcurrentSecretReads: defaultMaxConcurrentSecretReads, policyTokens: []config.PolicyToken{}, // default is empty + enrollFlushInterval: time.Second, + enrollFlushThresholdCnt: 50, } for _, f := range opts { @@ -193,6 +220,8 @@ func (o *bulkOptT) MarshalZerologObject(e *zerolog.Event) { e.Int("apikeyMaxReqSize", o.apikeyMaxReqSize) e.Int64("maxPendingBulkDispatches", o.maxPendingBulkDispatches) e.Int("maxConcurrentSecretReads", o.maxConcurrentSecretReads) + e.Dur("enrollFlushInterval", o.enrollFlushInterval) + e.Int("enrollFlushThresholdCnt", o.enrollFlushThresholdCnt) } // BulkOptsFromCfg transforms config to a slize of BulkOpt @@ -218,5 +247,7 @@ func BulkOptsFromCfg(cfg *config.Config) []BulkOpt { WithAPIKeyMaxRequestSize(cfg.Output.Elasticsearch.MaxContentLength), WithMaxPendingBulkDispatches(bulkCfg.MaxPendingBulkDispatches), WithPolicyTokens(policyTokens), + WithEnrollFlushInterval(bulkCfg.EnrollBulker.FlushInterval), + WithEnrollFlushThresholdCount(bulkCfg.EnrollBulker.FlushThresholdCount), } } diff --git a/internal/pkg/bulk/queue.go b/internal/pkg/bulk/queue.go index 8d6d350f1e..1c48344aa6 100644 --- a/internal/pkg/bulk/queue.go +++ b/internal/pkg/bulk/queue.go @@ -18,6 +18,7 @@ const ( kQueueRead kQueueSearch kQueueFleetSearch + kQueueEnrollSearch kQueueRefreshBulk kQueueRefreshRead kQueueAPIKeyUpdate @@ -34,6 +35,8 @@ func (q queueT) Type() string { return "search" case kQueueFleetSearch: return "fleetSearch" + case kQueueEnrollSearch: + return "enrollSearch" case kQueueRefreshBulk: return "refreshBulk" case kQueueRefreshRead: diff --git a/internal/pkg/config/input.go b/internal/pkg/config/input.go index d267744a97..1dd9b8d80a 100644 --- a/internal/pkg/config/input.go +++ b/internal/pkg/config/input.go @@ -49,12 +49,23 @@ type ServerTLS struct { Cert string `config:"cert"` } +type ServerBulkEnrollBulker struct { + FlushInterval time.Duration `config:"flush_interval"` + FlushThresholdCount int `config:"flush_threshold_cnt"` +} + +func (c *ServerBulkEnrollBulker) InitDefaults() { + c.FlushInterval = time.Second + c.FlushThresholdCount = 50 +} + type ServerBulk struct { - FlushInterval time.Duration `config:"flush_interval"` - FlushThresholdCount int `config:"flush_threshold_cnt"` - FlushThresholdSize int `config:"flush_threshold_size"` - FlushMaxPending int `config:"flush_max_pending"` - MaxPendingBulkDispatches int64 `config:"max_pending_bulk_dispatches"` + FlushInterval time.Duration `config:"flush_interval"` + FlushThresholdCount int `config:"flush_threshold_cnt"` + FlushThresholdSize int `config:"flush_threshold_size"` + FlushMaxPending int `config:"flush_max_pending"` + MaxPendingBulkDispatches int64 `config:"max_pending_bulk_dispatches"` + EnrollBulker ServerBulkEnrollBulker `config:"enroll"` } func (c *ServerBulk) InitDefaults() { @@ -62,6 +73,7 @@ func (c *ServerBulk) InitDefaults() { c.FlushThresholdCount = 2048 c.FlushThresholdSize = 1024 * 1024 c.FlushMaxPending = 8 + c.EnrollBulker.InitDefaults() } // Server is the configuration for the server diff --git a/internal/pkg/dl/agent.go b/internal/pkg/dl/agent.go index 43114db066..0b18a159ea 100644 --- a/internal/pkg/dl/agent.go +++ b/internal/pkg/dl/agent.go @@ -67,7 +67,7 @@ func GetAgent(ctx context.Context, bulker bulk.Bulk, agentID string, opt ...Opti func FindAgent(ctx context.Context, bulker bulk.Bulk, tmpl *dsl.Tmpl, name string, v any, opt ...Option) (model.Agent, error) { o := newOption(FleetAgents, opt...) - res, err := SearchWithOneParam(ctx, bulker, tmpl, o.indexName, name, v) + res, err := SearchWithOneParam(ctx, bulker, tmpl, o.indexName, name, v, o.bulkOpts...) if err != nil { return model.Agent{}, fmt.Errorf("failed searching for agent: %w", err) } diff --git a/internal/pkg/dl/common.go b/internal/pkg/dl/common.go index 39541c654c..faeade1d96 100644 --- a/internal/pkg/dl/common.go +++ b/internal/pkg/dl/common.go @@ -4,8 +4,11 @@ package dl +import "github.com/elastic/fleet-server/v7/internal/pkg/bulk" + type queryOption struct { indexName string + bulkOpts []bulk.Opt } // Option for the operation being made @@ -20,6 +23,13 @@ func WithIndexName(name string) Option { } } +// WithBulkOpts passes additional bulk.Opt values to the underlying bulker call. +func WithBulkOpts(opts ...bulk.Opt) Option { + return func(o *queryOption) { + o.bulkOpts = append(o.bulkOpts, opts...) + } +} + func newOption(defaultIndex string, opts ...Option) queryOption { o := queryOption{indexName: defaultIndex} for _, opt := range opts { diff --git a/internal/pkg/dl/search.go b/internal/pkg/dl/search.go index afbe7c7d1b..04353dc71b 100644 --- a/internal/pkg/dl/search.go +++ b/internal/pkg/dl/search.go @@ -26,12 +26,12 @@ func Search(ctx context.Context, bulker bulk.Bulk, tmpl *dsl.Tmpl, index string, return &res.HitsT, nil } -func SearchWithOneParam(ctx context.Context, bulker bulk.Bulk, tmpl *dsl.Tmpl, index string, name string, v any) (*es.HitsT, error) { +func SearchWithOneParam(ctx context.Context, bulker bulk.Bulk, tmpl *dsl.Tmpl, index string, name string, v any, opts ...bulk.Opt) (*es.HitsT, error) { query, err := tmpl.RenderOne(name, v) if err != nil { return nil, err } - res, err := bulker.Search(ctx, index, query) + res, err := bulker.Search(ctx, index, query, opts...) if err != nil { return nil, err }