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

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

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{

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")

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 {

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
 }

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")
+}