From 352a61d7b5daf46c2a19739e5cd2969e619ce207 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 20 Aug 2026 00:05:19 -0700 Subject: [PATCH] feat: batch enrollment FindAgent searches with pre-refresh dedup (#7662) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: batch enrollment FindAgent searches with pre-refresh dedup via kQueueEnrollSearch Adds a dedicated kQueueEnrollSearch bulker queue that prevents duplicate agent documents caused by concurrent enrollment retries with the same enrollment_id. The queue: - Flushes independently from other queues (default: every 1s or 50 items) - Calls POST .fleet-agents/_refresh before executing the msearch batch, ensuring retries see agent documents written by earlier requests - De-dupes concurrent requests with the same enrollment_id: the first (canonical) request gets the search result; duplicates receive ErrEnrollDuplicate (HTTP 429) so the agent retries after backoff Co-Authored-By: Claude Sonnet 4.6 * chore: add changelog fragment for kQueueEnrollSearch enrollment dedup Co-Authored-By: Claude Sonnet 4.6 * refactor: rename EnrollBatcher → EnrollBulker in config and opts Co-Authored-By: Claude Sonnet 4.6 * fix: correct LIFO→FIFO order and refresh all unique indices in flushEnrollSearch Address two Copilot review findings in flushEnrollSearch: 1. The bulker queue is LIFO (head = newest item), so iterating queue.head directly made the *newest* request canonical, not the oldest. Collect all items into a slice and reverse it so the oldest request is canonical for each dedupeKey (FIFO semantics). 2. Only canonicals[0].refreshIndex was refreshed, silently skipping any additional indices present in the batch. Collect all unique refreshIndex values from every canonical and refresh them all. Co-Authored-By: Claude Sonnet 4.6 * fix: propagate canonical search error to duplicates in flushEnrollSearch When a canonical request's msearch returns an ES error, duplicates were receiving ErrEnrollDuplicate, which callers treat as a benign retry signal. This silently hid the real failure. Now: if deriveError() on a canonical response is non-nil, that error is forwarded to all duplicates sharing the same dedupeKey, so callers surface the actual operational failure rather than retrying indefinitely under the impression it was just a duplicate. ErrEnrollDuplicate is still sent when the canonical search succeeded. Co-Authored-By: Claude Sonnet 4.6 * test: add integration test for enrollment search dedup (kQueueEnrollSearch) Fires numConcurrent goroutines all calling bulker.Search with the same dedupeKey against a test index. Sets enrollFlushThresholdCount above the concurrent count so all requests land in one batch and are flushed by the timer. Asserts exactly one canonical result and numConcurrent-1 ErrEnrollDuplicate responses. Co-Authored-By: Claude Sonnet 4.6 * fix: address golangci-lint failures (nolintlint, goimports) - Add explanation to nolint:errcheck directive in flushEnrollSearch (nolintlint requires a reason comment) - Realign ServerBulk struct fields after adding the wider ServerBulkEnrollBulker type (goimports formatting) Co-Authored-By: Claude Sonnet 4.6 * fix: call EnrollBulker.InitDefaults() from ServerBulk.InitDefaults() ucfg calls each nested struct's InitDefaults() automatically when loading config from a file, so the EnrollBulker defaults (1s, 50) were correctly applied at runtime. However, tests that construct expected configs by calling InitDefaults() manually through the chain were getting zero values for EnrollBulker, causing TestConfig to fail. Co-Authored-By: Claude Sonnet 4.6 * fix: eliminate data race in flushEnrollSearch dispatch loop After sending on n.ch, the receiving goroutine (Search) immediately calls freeBlk(n) -> reset(), which zeroes all bulkT fields including dedupeKey. Reading n.dedupeKey after the channel send to look up duplicate waiters was therefore a data race detected by -race. Fix: capture n.dedupeKey into a local variable before the send, so the map lookup uses a stack-allocated copy that is not subject to concurrent modification. This matches the pattern already used in flushSearch, which saves n.next before its channel send with the comment "n is invalid immediately on channel send". Co-Authored-By: Claude Sonnet 4.6 * fix: rename enroll_bulker config key to enroll, simplify changelog - inputs[].server.bulk.enroll_bulker -> inputs[].server.bulk.enroll (per blakerouse feedback: _bulker suffix is redundant in this context) - Rewrite changelog to focus on the user-visible problem (ghost agents) rather than implementation details Co-Authored-By: Claude Sonnet 4.6 --------- Co-authored-by: Claude Sonnet 4.6 (cherry picked from commit 3890bbb1b2acfa72071dd7ac26e6e06e0c1efdb6) --- ...1787137647-enroll-search-bulker-queue.yaml | 20 +++ internal/pkg/api/error.go | 9 ++ internal/pkg/api/handleEnroll.go | 7 +- internal/pkg/bulk/block.go | 20 +-- internal/pkg/bulk/engine.go | 92 +++++++++--- .../bulk/enroll_search_integration_test.go | 74 ++++++++++ internal/pkg/bulk/opSearch.go | 134 ++++++++++++++++++ internal/pkg/bulk/opt.go | 31 ++++ internal/pkg/bulk/queue.go | 3 + internal/pkg/config/input.go | 22 ++- internal/pkg/dl/agent.go | 2 +- internal/pkg/dl/common.go | 10 ++ internal/pkg/dl/search.go | 4 +- 13 files changed, 390 insertions(+), 38 deletions(-) create mode 100644 changelog/fragments/1787137647-enroll-search-bulker-queue.yaml create mode 100644 internal/pkg/bulk/enroll_search_integration_test.go 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 }