Skip to content

[9.5](backport #7662) feat: batch enrollment FindAgent searches with pre-refresh dedup - #7664

Merged
ycombinator merged 1 commit into
9.5from
mergify/bp/9.5/pr-7662
Aug 24, 2026
Merged

[9.5](backport #7662) feat: batch enrollment FindAgent searches with pre-refresh dedup#7664
ycombinator merged 1 commit into
9.5from
mergify/bp/9.5/pr-7662

Conversation

@mergify

@mergify mergify Bot commented Aug 20, 2026

Copy link
Copy Markdown
Contributor

What is the problem this PR solves?

At 30k+ agent scale, enrollment retries can create duplicate (ghost) agent documents in .fleet-agents. The root cause is ES near-real-time search latency: when an agent retries enrollment before the previous write is visible to search (default refresh interval: 1s on ES, 5s on Serverless), a concurrent retry also finds "no existing agent" and writes a second document with the same enrollment_id.

How does this PR solve the problem?

Adds a new kQueueEnrollSearch bulker queue that batches enrollment FindAgent searches and, before executing them, fires a single POST .fleet-agents/_refresh to make all recent writes visible. Within each batch:

  1. Pre-refresh: all unique refreshIndex values in the batch are refreshed so retries see the most recent writes.
  2. De-duplication: requests sharing the same enrollment_id (passed via WithDedupeKey) are grouped by FIFO order — the oldest (first) request is canonical and is included in the msearch; duplicates receive HTTP 429 (ErrEnrollDuplicate) and retry after backoff.
  3. Error propagation: if the canonical search itself returns an ES error, that error is forwarded to duplicates instead of ErrEnrollDuplicate, so operational failures are not silently hidden.

The queue flushes when N requests accumulate (default: 50) or after M seconds (default: 1s), whichever comes first — capping refresh calls at one per fleet-server instance per flush window. Both thresholds are configurable via inputs[].server.bulk.enroll.flush_interval and inputs[].server.bulk.enroll.flush_threshold_cnt.

How to test this PR locally

Run the integration test added in this PR:

ELASTICSEARCH_HOSTS=localhost:9200 ELASTICSEARCH_SERVICE_TOKEN=<token> \
  go test -tags integration ./internal/pkg/bulk/... -run TestEnrollSearchDedup -v

For manual testing, set a low flush_threshold_cnt (e.g. 2) in config and fire concurrent enrollment requests with the same enrollment_id. Only one agent document should be created.

Design Checklist

  • I have ensured my design is stateless and will work when multiple fleet-server instances are behind a load balancer. (Each instance maintains its own bulker queue; the pre-refresh + op_type: create write guarantees in ES ensure correctness across instances.)
  • I have or intend to scale test my changes, ensuring it will work reliably with 100K+ agents connected.
  • I have included fail safe mechanisms to limit the load on fleet-server: rate limiting, circuit breakers, caching, load shedding. (Flush interval and threshold count cap the refresh rate; duplicates shed load via 429.)

Checklist

  • I have added tests that prove my fix is effective or that my feature works
  • I have added an entry in ./changelog/fragments using the changelog tool

Related issues

@ycombinator @claude @mergify
feat: batch enrollment FindAgent searches with pre-refresh dedup (#7662)
352a61d 
* 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 <noreply@anthropic.com>

* chore: add changelog fragment for kQueueEnrollSearch enrollment dedup

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* refactor: rename EnrollBatcher → EnrollBulker in config and opts

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
(cherry picked from commit 3890bbb)
@mergify mergify Bot added the backport label Aug 20, 2026
@mergify
mergify Bot requested a review from a team as a code owner August 20, 2026 07:05
@mergify mergify Bot added the backport label Aug 20, 2026
@mergify
mergify Bot requested review from swiatekm and ycombinator August 20, 2026 07:05
@github-actions github-actions Bot added the Team:Elastic-Agent-Control-Plane Label for the Agent Control Plane team label Aug 20, 2026
@mergify mergify Bot mentioned this pull request Aug 20, 2026
Merged
5 tasks
@mergify

mergify Bot commented Aug 24, 2026

Copy link
Copy Markdown
Contributor Author

This pull request has not been merged yet. Could you please review and merge it @ycombinator? 🙏

@mergify

mergify Bot commented Aug 24, 2026

Copy link
Copy Markdown
Contributor Author

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

  • Queue this pull request

@ycombinator
ycombinator merged commit ee45f58 into 9.5 Aug 24, 2026
13 checks passed
@ycombinator
ycombinator deleted the mergify/bp/9.5/pr-7662 branch August 24, 2026 13:05
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@ycombinator ycombinator ycombinator approved these changes
@swiatekm swiatekm swiatekm approved these changes

Assignees

@ycombinator ycombinator

Labels

backport Team:Elastic-Agent-Control-Plane Label for the Agent Control Plane team

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@ycombinator @swiatekm