bulk: add configurable concurrency limit for ReadSecrets

[ycombinator]

What is the problem this PR solves?

ReadSecrets makes direct HTTP calls to the ES Fleet secrets API (one per secret reference per agent checkin) without going through the bulk dispatch queue. This means it is not subject to the max_pending_bulk_dispatches cap introduced in #6751. Under high concurrent checkin load — as seen in large serverless projects — each checkin goroutine issues an independent ES connection, causing unbounded concurrent ES connections and additional memory pressure (goroutines, HTTP buffers, response allocations).

This was observed as a contributing factor in an OOM incident for a large production serverless security project. PR #7416 added the ReadSecrets per-checkin call ~24h before OOMs began.

How does this PR solve the problem?

Adds a semaphore.Weighted (readSecretsLimit) to Bulker, capping concurrent in-flight secret reads at 32 (matching apikeyLimit). ReadSecrets acquires a slot before each ES call and releases it immediately after — callers that arrive when all slots are taken block until one is free or their context is cancelled. Setting readSecretsLimit to nil (by passing WithMaxConcurrentSecretReads(0)) disables the cap entirely; the default is always 32.

The limit is intentionally not exposed as a user-facing config option, consistent with how apikeyMaxParallel is handled.

Files changed

• internal/pkg/bulk/engine.go: add readSecretsLimit *semaphore.Weighted to Bulker; initialize in NewBulker when limit > 0; nil-check acquire/release in ReadSecrets; add defaultMaxConcurrentSecretReads = 32
• internal/pkg/bulk/opt.go: add maxConcurrentSecretReads field, WithMaxConcurrentSecretReads BulkOpt, zerolog logging
• internal/pkg/bulk/secret_limit_test.go: unit tests covering the concurrency limit, context cancellation while waiting, zero-value (no limit), and default capacity

How to test this PR locally

go test -v -count=1 -run=TestReadSecrets ./internal/pkg/bulk

Design Checklist

• I have ensured my design is stateless and will work when multiple fleet-server instances are behind a load balancer.
• 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, etc.

Checklist

• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• 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

• Relates elastic/ingest-dev#8991
• Relates Store output API key secrets in .fleet-secrets instead of .fleet-agents #7416
• Relates feat: add configurable limit on concurrent bulk dispatch goroutines #6751

[mergify]

This pull request does not have a backport label. Could you fix it @ycombinator? 🙏
To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-./d./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Suppressed comments (5)

internal/pkg/bulk/engine.go:337

• ReadSecrets unconditionally acquires/releases readSecretsLimit. If the limit is disabled by configuration (e.g., max=0), this should be a no-op; additionally, the release should only happen when an acquire occurred.

		if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil {
			return nil, err
		}
		val, err := ReadSecret(ctx, esClient, id)
		b.readSecretsLimit.Release(1)

internal/pkg/bulk/opt.go:175

• PR description says the default for max_concurrent_secret_reads is 0 (no limit), but parseBulkOpts sets maxConcurrentSecretReads to defaultAPIKeyMaxParallel (32). This changes behavior by default and contradicts the option's "0 means no limit" semantics unless explicitly configured.

		apikeyMaxParallel:        defaultAPIKeyMaxParallel,
		blockQueueSz:             defaultBlockQueueSz,
		apikeyMaxReqSize:         defaultApikeyMaxReqSize,
		maxPendingBulkDispatches: defaultMaxPendingBulkDispatches,
		maxConcurrentSecretReads: defaultAPIKeyMaxParallel,

internal/pkg/bulk/secret_limit_test.go:120

• This test asserts an implicit default concurrency of 32, but the PR description states the default should be 0 (no limit). To avoid baking in a default and to keep the test valid regardless of the chosen default, configure the limit explicitly in the test.

// TestReadSecretsDefaultConcurrency verifies that a Bulker created without
// WithMaxConcurrentSecretReads initialises readSecretsLimit with the default
// capacity of defaultAPIKeyMaxParallel (32). It confirms the capacity
// indirectly: after filling all 32 slots via concurrent ReadSecrets calls
// that block in the transport, an additional Acquire with a cancelled context

internal/pkg/config/input.go:49

• PR description says the default for server.bulk.max_concurrent_secret_reads is 0 (no limit), but InitDefaults sets it to 32 (and introduces a constant for that default). This makes the new limit enabled by default and conflicts with the stated "preserving existing behaviour" default.

// defaultMaxConcurrentSecretReads matches defaultAPIKeyMaxParallel in the bulk package.
const defaultMaxConcurrentSecretReads = 32

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

• The changelog fragment states the new option defaults to 32, but the PR description says the default is 0 (no limit). This should match the actual default behavior to avoid misleading operators.

  A new configuration option, server.bulk.max_concurrent_secret_reads
  (default 32), limits how many secret reads can be in-flight at once.
  Excess reads wait until a slot is available.

[github-actions]

TL;DR

Buildkite Run check-ci failed because CI auto-fix checks produced unstaged changes, so the NoChanges guard failed at git update-index --refresh. This is a CI hygiene failure (format/fix drift), not a runtime logic failure.

Remediation

• Run mage check:ci locally on rate-limit-read-secrets, then commit all resulting changes (including internal/pkg/bulk/engine.go and internal/pkg/bulk/secret_limit_test.go).
• Push and re-run Buildkite; this step should pass once the tree remains clean after Check.Imports/Check.Fix.

Investigation details

Root Cause

Run check-ci executes .buildkite/scripts/check_ci.sh, which calls mage check:ci (.buildkite/scripts/check_ci.sh:13). check:ci runs Generate, Check.Imports, Check.Fix, Check.Headers, Check.Notice, then Check.NoChanges (magefile.go:625-628).

Check.NoChanges enforces a clean tree via git update-index --refresh and git diff-index --exit-code HEAD -- (magefile.go:596-616). In this build, those checks detected file rewrites still needed:

• internal/pkg/bulk/engine.go (alignment/formatting around NewBulker, e.g. readSecretsLimit initializer at ~L167 in PR head)
• internal/pkg/bulk/secret_limit_test.go (TestReadSecretsDefaultConcurrency, loop around ~L125 where CI wants wg.Go(...) style rewrite)

Evidence

• Build: https://buildkite.com/elastic/fleet-server/builds/16139
• Job/step: Run check-ci
• Key log excerpt:

diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go
@@ -125,11 +125,9 @@ func TestReadSecretsDefaultConcurrency(t *testing.T) {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		wg.Go(func() {
			_, _ = b.ReadSecrets(context.Background(), []string{fmt.Sprintf("id%d", i)})
-		}()
+		})
internal/pkg/bulk/secret_limit_test.go: needs update
Error: git update-index failure: running "git update-index --refresh" failed with exit code 1

Verification

• Not run locally in this detective workflow; analysis is based on the provided Buildkite failure artifacts and PR metadata.
• Checked for matching flaky-test issue signal (label:flaky-test + git update-index) and found none.

Follow-up

If failure persists after committing mage check:ci output, capture the next Run check-ci log; it will likely be a new gate/failure signature rather than this one.

---

What is this? | From workflow: PR Buildkite Detective

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Suppressed comments (3)

internal/pkg/bulk/engine.go:170

• max_concurrent_secret_reads is documented as "0 means no limit", but NewBulker always creates a weighted semaphore with that value. If the config/opt sets 0 (or a negative value), ReadSecrets will block forever on Acquire (or potentially panic), effectively breaking secret resolution.

Consider treating values <= 0 as "unlimited" by leaving readSecretsLimit nil (or otherwise bypassing Acquire/Release), matching the existing apikeyLimit pattern of only limiting when configured.

		blkPool:               sync.Pool{New: poolFunc},
		flushBufPool:          sync.Pool{New: func() any { return new(bytes.Buffer) }},
		apikeyLimit:           semaphore.NewWeighted(int64(bopts.apikeyMaxParallel)),
		readSecretsLimit:      semaphore.NewWeighted(int64(bopts.maxConcurrentSecretReads)),
		tracer:                tracer,

internal/pkg/config/input.go:65

• PR description says the default should be 0 (no limit) to preserve existing behavior, but the code and changelog set a non-zero default (32) via InitDefaults and the bulk option defaults.

Please align the default behavior and docs (PR description, changelog, and bulk option defaults) so operators don’t accidentally throttle or, if they set 0 expecting "unlimited", hit a deadlock/panic depending on implementation.

// defaultMaxConcurrentSecretReads matches defaultAPIKeyMaxParallel in the bulk package.
const defaultMaxConcurrentSecretReads = 32

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"`
	MaxConcurrentSecretReads int           `config:"max_concurrent_secret_reads"`
}

func (c *ServerBulk) InitDefaults() {
	c.FlushInterval = 250 * time.Millisecond
	c.FlushThresholdCount = 2048
	c.FlushThresholdSize = 1024 * 1024
	c.FlushMaxPending = 8
	c.MaxConcurrentSecretReads = defaultMaxConcurrentSecretReads
}

internal/pkg/bulk/engine.go:339

• After making the semaphore optional (nil when unlimited), ReadSecrets needs to guard Acquire/Release. As written, it will panic if readSecretsLimit is nil, and it also makes it harder to reason about the intended "0 means no limit" behavior.

		if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil {
			return nil, err
		}
		val, err := ReadSecret(ctx, esClient, id)
		b.readSecretsLimit.Release(1)

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Suppressed comments (2)

internal/pkg/bulk/engine.go:178

• readSecretsLimit is initialized directly from maxConcurrentSecretReads (default 32) without considering the ES client's MaxConnPerHost budgeting that BulkOptsFromCfg applies to apikeyMaxParallel. If an operator lowers max_conn_per_host (or raises flush_max_pending), apikeyMaxParallel will shrink to leave room for bulk flushes, but secret reads will still allow 32 concurrent requests and can consume the remaining connections, potentially starving bulk flushes and other direct ES calls.

Consider capping the secret-read semaphore at apikeyMaxParallel so it respects the same connection budget.

	// 0 means no limit; leave readSecretsLimit nil so ReadSecrets skips the semaphore.
	if bopts.maxConcurrentSecretReads > 0 {
		b.readSecretsLimit = semaphore.NewWeighted(int64(bopts.maxConcurrentSecretReads))
	}

internal/pkg/bulk/opt.go:130

• The doc comment says this limits "concurrent ReadSecrets calls", but the implementation acquires/releases around each ReadSecret HTTP request inside the loop. The limit therefore bounds concurrent secret-read requests (potentially allowing multiple ReadSecrets invocations to overlap), not the number of active ReadSecrets calls. Tightening the wording will avoid confusion for callers of this option.

// 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 {

[swiatekm]

The logic looks correct to me, but why is this a separate ES connection for each goroutine? Shouldn't it multiplex or go through a pool?

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

internal/pkg/bulk/secret_limit_test.go:133

• If this test fails before reaching the final close(mt.gate), the goroutines blocked in blockingTransport.RoundTrip will never be released, which can hang the test suite. Add a deferred “safe close” for mt.gate so goroutines always unblock even on FailNow paths.

	mt := &blockingTransport{gate: make(chan struct{})}

	// No WithMaxConcurrentSecretReads option → uses defaultMaxConcurrentSecretReads.
	b := newTestBulkerWithTransport(t, mt)

[ycombinator]

The logic looks correct to me, but why is this a separate ES connection for each goroutine? Shouldn't it multiplex or go through a pool?

The requests do go through a connection pool. The ES client used by the individual requests is shared. That client's HTTP transport pools and reuses connections.

The requests remain individual, one per secret, because the Fleet secrets ES API has no bulk-read endpoint. This change limits concurrent in-flight reads to 32.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Suppressed comments (3)

internal/pkg/bulk/opt.go:199

• Typo in comment: "slize" → "slice".

	e.Int("maxConcurrentSecretReads", o.maxConcurrentSecretReads)
}

// BulkOptsFromCfg transforms config to a slize of BulkOpt
// used to bridge to configuration subsystem

internal/pkg/bulk/engine.go:347

• The semaphore slot is released only on the normal return path. If ReadSecret panics (or is otherwise interrupted before the explicit Release), the slot will be leaked and future ReadSecrets calls can block indefinitely. Wrap the ReadSecret call so Release is deferred after a successful Acquire.

		if b.readSecretsLimit != nil {
			if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil {
				return nil, err
			}
		}

internal/pkg/bulk/opt.go:129

• The docstring says this limits concurrent "ReadSecrets" calls, but the semaphore actually limits concurrent in-flight secret reads (each ReadSecret HTTP call). Rewording avoids confusion for callers.

// 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.

[blakerouse]

Looks good.

[ycombinator]

@Mergifyio backport 9.5 9.4 8.19

[mergify]

backport 9.5 9.4 8.19

✅ Backports have been created

Details

• #7627 [9.5](backport #7567) bulk: add configurable concurrency limit for ReadSecrets has been created for branch 9.5
• #7628 [9.4](backport #7567) bulk: add configurable concurrency limit for ReadSecrets has been created for branch 9.4
• #7629 [8.19](backport #7567) bulk: add configurable concurrency limit for ReadSecrets has been created for branch 8.19 but encountered conflicts

Cherry-pick of 5bef901 has failed:

On branch mergify/bp/8.19/pr-7567
Your branch is up to date with 'origin/8.19'.

You are currently cherry-picking commit 5bef901.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Changes to be committed:
	new file:   changelog/fragments/1786037058-rate-limit-read-secrets.yaml
	modified:   internal/pkg/bulk/opt.go
	new file:   internal/pkg/bulk/secret_limit_test.go

Unmerged paths:
  (use "git add <file>..." to mark resolution)
	both modified:   internal/pkg/bulk/engine.go

To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally