bulk: add configurable concurrency limit for ReadSecrets

changelog/fragments/1786037058-rate-limit-read-secrets.yaml

@@ -0,0 +1,14 @@
+kind: bug
+
+summary: Limit concurrent secret reads to reduce memory pressure under high agent check-in load
+
+description: |
+  Under high concurrent agent check-in load, fleet-server could open an
+  unbounded number of concurrent connections to Elasticsearch to resolve
+  policy secret references, contributing to out-of-memory conditions on
+  large deployments.
+
+  Secret reads are now capped at 32 concurrent in-flight requests.
+  Excess reads wait until a slot is available.
+
+component: fleet-server

internal/pkg/bulk/engine.go

@@ -17,12 +17,13 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/elastic/go-ucfg"
+
 	"github.com/elastic/fleet-server/v7/internal/pkg/apikey"
 	"github.com/elastic/fleet-server/v7/internal/pkg/build"
 	"github.com/elastic/fleet-server/v7/internal/pkg/config"
 	"github.com/elastic/fleet-server/v7/internal/pkg/es"
 	"github.com/elastic/fleet-server/v7/internal/pkg/logger/ecs"
-	"github.com/elastic/go-ucfg"
 
 	"github.com/rs/zerolog"
 	"go.elastic.co/apm/v2"
@@ -121,6 +122,7 @@ type Bulker struct {
 	blkPool               sync.Pool
 	flushBufPool          sync.Pool
 	apikeyLimit           *semaphore.Weighted
+	readSecretsLimit      *semaphore.Weighted
 	tracer                *apm.Tracer
 	cancelFn              context.CancelFunc
 	pendingBulkDispatches atomic.Int64
@@ -137,6 +139,7 @@ const (
 	defaultMaxPending                     = 32
 	defaultBlockQueueSz                   = 32 // Small capacity to allow multiOp to spin fast
 	defaultAPIKeyMaxParallel              = 32
+	defaultMaxConcurrentSecretReads       = 32
 	defaultApikeyMaxReqSize               = 100 * 1024 * 1024
 	defaultFlushContextTimeout            = time.Minute * 1
 	defaultMaxPendingBulkDispatches int64 = 0 // 0 means no limit
@@ -157,7 +160,7 @@ func NewBulker(es esapi.Transport, tracer *apm.Tracer, opts ...BulkOpt) *Bulker
 		return &bulkT{ch: make(chan respT, 1)}
 	}
 
-	return &Bulker{
+	b := &Bulker{
 		opts:                  bopts,
 		es:                    es,
 		ch:                    make(chan *bulkT, bopts.blockQueueSz),
@@ -169,6 +172,11 @@ func NewBulker(es esapi.Transport, tracer *apm.Tracer, opts ...BulkOpt) *Bulker
 		// remote ES bulkers
 		bulkerMap: make(map[string]Bulk),
 	}
+	// 0 means no limit; leave readSecretsLimit nil so ReadSecrets skips the semaphore.
+	if bopts.maxConcurrentSecretReads > 0 {
+		b.readSecretsLimit = semaphore.NewWeighted(int64(bopts.maxConcurrentSecretReads))
+	}
+	return b
 }
 
 func (b *Bulker) GetBulker(outputName string) Bulk {
@@ -328,7 +336,15 @@ func (b *Bulker) ReadSecrets(ctx context.Context, secretIds []string) (map[strin
 	result := make(map[string]string)
 	esClient := b.Client()
 	for _, id := range secretIds {
+		if b.readSecretsLimit != nil {
+			if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil {
+				return nil, err
+			}
+		}
 		val, err := ReadSecret(ctx, esClient, id)
+		if b.readSecretsLimit != nil {
+			b.readSecretsLimit.Release(1)
+		}
 		if err != nil {
 			if errors.Is(err, ErrSecretNotFound) {
 				zerolog.Ctx(ctx).Warn().Str("secret_id", id).Msg("secret not found; policy will load without it")

internal/pkg/bulk/opt.go

@@ -75,6 +75,7 @@ type bulkOptT struct {
 	apikeyMaxParallel        int
 	apikeyMaxReqSize         int
 	maxPendingBulkDispatches int64
+	maxConcurrentSecretReads int
 	policyTokens             []config.PolicyToken
 	bi                       build.Info
 }
@@ -124,6 +125,14 @@ func WithMaxPendingBulkDispatches(max int64) BulkOpt {
 	}
 }
 
+// WithMaxConcurrentSecretReads sets the upper bound on concurrent ReadSecrets calls.
+// When the limit is reached, ReadSecrets blocks until a slot is available. 0 means no limit.
+func WithMaxConcurrentSecretReads(max int) BulkOpt {
+	return func(opt *bulkOptT) {
+		opt.maxConcurrentSecretReads = max
+	}
+}
+
 // WithAPIKeyMaxParallel sets the number of api key operations outstanding
 func WithAPIKeyMaxParallel(max int) BulkOpt {
 	return func(opt *bulkOptT) {
@@ -163,6 +172,7 @@ func parseBulkOpts(opts ...BulkOpt) bulkOptT {
 		blockQueueSz:             defaultBlockQueueSz,
 		apikeyMaxReqSize:         defaultApikeyMaxReqSize,
 		maxPendingBulkDispatches: defaultMaxPendingBulkDispatches,
+		maxConcurrentSecretReads: defaultMaxConcurrentSecretReads,
 		policyTokens:             []config.PolicyToken{}, // default is empty
 	}
 
@@ -182,6 +192,7 @@ func (o *bulkOptT) MarshalZerologObject(e *zerolog.Event) {
 	e.Int("apikeyMaxParallel", o.apikeyMaxParallel)
 	e.Int("apikeyMaxReqSize", o.apikeyMaxReqSize)
 	e.Int64("maxPendingBulkDispatches", o.maxPendingBulkDispatches)
+	e.Int("maxConcurrentSecretReads", o.maxConcurrentSecretReads)
 }
 
 // BulkOptsFromCfg transforms config to a slize of BulkOpt

internal/pkg/bulk/secret_limit_test.go

@@ -0,0 +1,167 @@
+// 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.
+
+package bulk
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/elastic/go-elasticsearch/v8"
+	"github.com/stretchr/testify/require"
+)
+
+// blockingTransport is an http.RoundTripper that blocks until gate is closed
+// and tracks how many requests are currently in-flight.
+type blockingTransport struct {
+	gate     chan struct{}
+	gateOnce sync.Once
+	inFlight atomic.Int64
+}
+
+func (m *blockingTransport) unblock() {
+	m.gateOnce.Do(func() { close(m.gate) })
+}
+
+func (m *blockingTransport) RoundTrip(_ *http.Request) (*http.Response, error) {
+	m.inFlight.Add(1)
+	defer m.inFlight.Add(-1)
+	<-m.gate
+	h := http.Header{}
+	h.Set("X-Elastic-Product", "Elasticsearch")
+	body := `{"value":"test"}`
+	return &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     h,
+		Body:       io.NopCloser(strings.NewReader(body)),
+	}, nil
+}
+
+func newTestBulkerWithTransport(t *testing.T, transport http.RoundTripper, opts ...BulkOpt) *Bulker {
+	t.Helper()
+	esClient, err := elasticsearch.NewClient(elasticsearch.Config{
+		Transport: transport,
+		Addresses: []string{"http://localhost:9200"},
+	})
+	require.NoError(t, err)
+	return NewBulker(esClient, nil, opts...)
+}
+
+// TestReadSecretsLimitsConcurrency verifies that WithMaxConcurrentSecretReads(1)
+// allows at most one ReadSecret HTTP call in-flight at a time across concurrent
+// ReadSecrets callers.
+func TestReadSecretsLimitsConcurrency(t *testing.T) {
+	mt := &blockingTransport{gate: make(chan struct{})}
+	defer mt.unblock()
+	b := newTestBulkerWithTransport(t, mt, WithMaxConcurrentSecretReads(1))
+
+	var wg sync.WaitGroup
+	errs := make([]error, 2)
+
+	wg.Go(func() {
+		_, errs[0] = b.ReadSecrets(t.Context(), []string{"id1"})
+	})
+	wg.Go(func() {
+		_, errs[1] = b.ReadSecrets(t.Context(), []string{"id2"})
+	})
+
+	require.Eventually(t, func() bool {
+		return mt.inFlight.Load() >= 1
+	}, time.Second, time.Millisecond, "wait for the first goroutine to enter the transport")
+
+	// With semaphore capacity 1, the second goroutine is blocked on Acquire
+	// and cannot have entered the transport yet.
+	require.Equal(t, int64(1), mt.inFlight.Load())
+
+	// Unblock both goroutines and wait for them to finish.
+	mt.unblock()
+	wg.Wait()
+
+	require.NoError(t, errs[0])
+	require.NoError(t, errs[1])
+}
+
+// TestReadSecretsContextCancelledWhileWaiting verifies that ReadSecrets returns
+// context.Canceled immediately when the semaphore is full and the caller's context
+// is already cancelled.
+func TestReadSecretsContextCancelledWhileWaiting(t *testing.T) {
+	mt := &blockingTransport{gate: make(chan struct{})}
+	defer close(mt.gate)
+
+	b := newTestBulkerWithTransport(t, mt, WithMaxConcurrentSecretReads(1))
+
+	// Manually hold the only semaphore slot so ReadSecrets must wait.
+	err := b.readSecretsLimit.Acquire(t.Context(), 1)
+	require.NoError(t, err)
+	defer b.readSecretsLimit.Release(1)
+
+	// Call ReadSecrets with an already-cancelled context.
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	_, err = b.ReadSecrets(ctx, []string{"id1"})
+	require.ErrorIs(t, err, context.Canceled)
+}
+
+// TestReadSecretsNoLimitWhenZero verifies that WithMaxConcurrentSecretReads(0)
+// disables the concurrency limit: readSecretsLimit is nil and ReadSecrets
+// completes without blocking.
+func TestReadSecretsNoLimitWhenZero(t *testing.T) {
+	mt := &blockingTransport{gate: make(chan struct{})}
+	close(mt.gate) // unblocked so ReadSecrets returns immediately
+
+	b := newTestBulkerWithTransport(t, mt, WithMaxConcurrentSecretReads(0))
+
+	require.Nil(t, b.readSecretsLimit)
+
+	_, err := b.ReadSecrets(t.Context(), []string{"id1"})
+	require.NoError(t, err)
+}
+
+// TestReadSecretsDefaultConcurrency verifies that a Bulker created without
+// WithMaxConcurrentSecretReads initialises readSecretsLimit with the default
+// capacity of defaultMaxConcurrentSecretReads (32). It confirms the capacity
+// indirectly: after filling all slots via concurrent ReadSecrets calls that
+// block in the transport, an additional Acquire with a cancelled context
+// returns context.Canceled immediately.
+func TestReadSecretsDefaultConcurrency(t *testing.T) {
+	mt := &blockingTransport{gate: make(chan struct{})}
+	defer mt.unblock()
+
+	// No WithMaxConcurrentSecretReads option → uses defaultMaxConcurrentSecretReads.
+	b := newTestBulkerWithTransport(t, mt)
+
+	require.NotNil(t, b.readSecretsLimit)
+
+	// Launch defaultMaxConcurrentSecretReads goroutines, each calling ReadSecrets
+	// with a single unique secret ID. Each goroutine will acquire one semaphore
+	// slot and block in the transport, filling all capacity.
+	var wg sync.WaitGroup
+	for i := range defaultMaxConcurrentSecretReads {
+		wg.Go(func() {
+			_, _ = b.ReadSecrets(t.Context(), []string{fmt.Sprintf("id%d", i)})
+		})
+	}
+
+	require.Eventually(t, func() bool {
+		return mt.inFlight.Load() >= int64(defaultMaxConcurrentSecretReads)
+	}, time.Second, time.Millisecond, "wait for all goroutines to enter the transport")
+
+	// The semaphore is now full; a cancelled-context acquire must return immediately.
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	err := b.readSecretsLimit.Acquire(ctx, 1)
+	require.ErrorIs(t, err, context.Canceled)
+
+	// Unblock all goroutines.
+	mt.unblock()
+	wg.Wait()
+}