bulk: add configurable concurrency limit for ReadSecrets - #7567
Open
ycombinator wants to merge 23 commits into
Open
bulk: add configurable concurrency limit for ReadSecrets#7567ycombinator wants to merge 23 commits into
ycombinator wants to merge 23 commits into
Conversation
Contributor
|
This pull request does not have a backport label. Could you fix it @ycombinator? 🙏
|
ycombinator
requested review from
belimawr,
blakerouse and
swiatekm
and removed request for
lorienhu and
samuelvl
August 6, 2026 16:41
ycombinator
added
the
backport-active-all
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
There was a problem hiding this comment.
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
ReadSecretsunconditionally acquires/releasesreadSecretsLimit. 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_readsis 0 (no limit), butparseBulkOptssetsmaxConcurrentSecretReadstodefaultAPIKeyMaxParallel(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_readsis 0 (no limit), butInitDefaultssets 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.
Contributor
TL;DRBuildkite Remediation
Investigation detailsRoot Cause
Evidence
Verification
Follow-upIf failure persists after committing What is this? | From workflow: PR Buildkite Detective Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not. |
ycombinator
force-pushed
the
rate-limit-read-secrets
branch
from
August 7, 2026 14:11
fe1f3f5 to
b415564
Compare
There was a problem hiding this comment.
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_readsis 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),ReadSecretswill 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) viaInitDefaultsand 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),
ReadSecretsneeds to guard Acquire/Release. As written, it will panic ifreadSecretsLimitis 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)
Comment on lines
+126
to
+133
| var wg sync.WaitGroup | ||
| for i := range defaultAPIKeyMaxParallel { | ||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| _, _ = b.ReadSecrets(context.Background(), []string{fmt.Sprintf("id%d", i)}) | ||
| }() | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Yes, it does. We are using Go v1.26.5.
Add bulkOptT field, WithMaxConcurrentSecretReads BulkOpt constructor, zerolog logging, and BulkOptsFromCfg wiring for the new max_concurrent_secret_reads config option. Related: elastic/ingest-dev#8991
ReadSecrets currently bypasses the bulk dispatch cap and makes unbounded concurrent direct HTTP calls to the ES Fleet secrets API. Under the high checkin concurrency seen in large serverless projects this causes both memory pressure and ES connection exhaustion. Add a readSecretsLimit *semaphore.Weighted to Bulker, initialized from the new max_concurrent_secret_reads config option (0 = no limit). ReadSecrets acquires one slot before performing secret reads and releases it on return, matching the pattern already used by apikeyLimit. Related: elastic/ingest-dev#8991
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
When maxConcurrentSecretReads is 0 (documented as "no limit"), NewBulker was creating a zero-capacity semaphore, causing all ReadSecrets calls to block until context cancellation instead of running without a limit. Fix by only initialising readSecretsLimit when the value is > 0, and nil-checking before Acquire/Release in ReadSecrets. Also introduce defaultMaxConcurrentSecretReads as its own constant (32) rather than reusing defaultAPIKeyMaxParallel, so the two limits can evolve independently. Drop the stale comment in config/input.go that referenced the apikey constant. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
apikeyMaxParallel is not user-configurable, so secret reads should follow the same pattern. Remove MaxConcurrentSecretReads from ServerBulk, drop the corresponding config constant, and remove the WithMaxConcurrentSecretReads wiring from parseBulkOptsFromConfig. The limit is now always defaultMaxConcurrentSecretReads (32). Update the changelog to drop the mention of a config option. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
go-elasticsearch performs a product check on the first request and requires the X-Elastic-Product: Elasticsearch response header. Without it, ReadSecrets calls that actually reach the transport fail with "the client noticed that the server is not Elasticsearch". Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
ycombinator
force-pushed
the
rate-limit-read-secrets
branch
from
August 7, 2026 15:16
616ffa4 to
c0d5581
Compare
There was a problem hiding this comment.
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:22
- The import block is not goimports-compliant:
github.com/elastic/go-ucfgis split into its own group and the third-party imports are fragmented around the localgithub.com/elastic/fleet-server/...imports. This will likely fail the repo’s import formatting checks; regroup third-party imports together and keep local imports in their own group.
"time"
"github.com/elastic/go-ucfg"
"github.com/elastic/fleet-server/v7/internal/pkg/apikey"
internal/pkg/bulk/opt.go:130
- The doc comment says this limits “concurrent ReadSecrets calls”, but the code actually limits concurrent in-flight secret read HTTP requests performed within ReadSecrets. Tweaking the wording would make the behavior clearer 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.
func WithMaxConcurrentSecretReads(max int) BulkOpt {
go fix rewrites the wg.Add(1)/go func()/defer wg.Done() pattern to wg.Go (available since Go 1.25). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.
Suppressed comments (3)
internal/pkg/bulk/secret_limit_test.go:153
- This is another unbounded busy-spin loop with no timeout. If
mt.inFlightnever reachesdefaultMaxConcurrentSecretReads, the test will hang indefinitely and consume CPU. Consider switching to a bounded wait (e.g.,require.Eventually) or using explicit synchronization (channels) instead of polling.
// Spin until all slots are occupied.
for mt.inFlight.Load() < int64(defaultMaxConcurrentSecretReads) {
// wait for all goroutines to enter the transport
}
internal/pkg/bulk/engine.go:34
- The import grouping/order here doesn’t match the convention used elsewhere in
internal/pkg/bulk(e.g.internal/pkg/bulk/opBulk.go:7-20, where third-party imports are grouped together and localgithub.com/elastic/fleet-server/...imports are in a separate group). This is likely to failgoimports/mage check:imports.
"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/rs/zerolog"
"go.elastic.co/apm/v2"
"golang.org/x/sync/semaphore"
"github.com/elastic/go-elasticsearch/v8"
"github.com/elastic/go-elasticsearch/v8/esapi"
)
internal/pkg/bulk/secret_limit_test.go:75
- This is an unbounded busy-spin loop with no timeout. If the request never enters the transport (e.g., unexpected error path), the test will hang indefinitely and burn CPU in CI. Prefer a bounded wait (e.g.,
require.Eventuallywith a short timeout) to fail fast and avoid tight spinning.
This issue also appears on line 150 of the same file.
// Spin until at least one request is in the transport.
for mt.inFlight.Load() < 1 {
// wait for the first goroutine to enter the transport
}
Comment on lines
+143
to
+148
| var wg sync.WaitGroup | ||
| for i := range defaultMaxConcurrentSecretReads { | ||
| wg.Go(func() { | ||
| _, _ = b.ReadSecrets(context.Background(), []string{fmt.Sprintf("id%d", i)}) | ||
| }) | ||
| } |
belimawr
reviewed
Aug 7, 2026
belimawr
left a comment
belimawr
left a comment
Member
There was a problem hiding this comment.
Only the busy wait are blocking.
One last nit pick (feel free to ignore it): you could use t.Context() instead of context.Background().
Comment on lines
+62
to
+79
| wg.Add(2) | ||
| go func() { | ||
| defer wg.Done() | ||
| _, errs[0] = b.ReadSecrets(context.Background(), []string{"id1"}) | ||
| }() | ||
| go func() { | ||
| defer wg.Done() | ||
| _, errs[1] = b.ReadSecrets(context.Background(), []string{"id2"}) | ||
| }() | ||
|
|
||
| // Spin until at least one request is in the transport. | ||
| for mt.inFlight.Load() < 1 { | ||
| // 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()) |
Member
There was a problem hiding this comment.
nit: the code can be modernised with wg.Go and use require.Eventually instead of a busy wait:
Suggested change
| wg.Add(2) | |
| go func() { | |
| defer wg.Done() | |
| _, errs[0] = b.ReadSecrets(context.Background(), []string{"id1"}) | |
| }() | |
| go func() { | |
| defer wg.Done() | |
| _, errs[1] = b.ReadSecrets(context.Background(), []string{"id2"}) | |
| }() | |
| // Spin until at least one request is in the transport. | |
| for mt.inFlight.Load() < 1 { | |
| // 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()) | |
| wg.Go(func() { _, errs[0] = b.ReadSecrets(t.Context(), []string{"id1"}) }) | |
| wg.Go(func() { _, errs[0] = b.ReadSecrets(t.Context(), []string{"id2"}) }) | |
| // With semaphore capacity 1, the second goroutine is blocked on Acquire | |
| // and cannot have entered the transport yet. | |
| require.Eventually( | |
| t, | |
| func() bool { return mt.inFlight.Load() == 1 }, | |
| time.Second, | |
| 10*time.Millisecond, | |
| "there are one or less requests in flight") |
Comment on lines
+150
to
+153
| // Spin until all slots are occupied. | ||
| for mt.inFlight.Load() < int64(defaultMaxConcurrentSecretReads) { | ||
| // wait for all goroutines to enter the transport | ||
| } |
Member
There was a problem hiding this comment.
That can be replaced by a require.Eventually
Suggested change
| // Spin until all slots are occupied. | |
| for mt.inFlight.Load() < int64(defaultMaxConcurrentSecretReads) { | |
| // wait for all goroutines to enter the transport | |
| } | |
| require.Eventually( | |
| t, | |
| func() bool { return mt.inFlight.Load() == defaultMaxConcurrentSecretReads }, | |
| time.Second, | |
| 10*time.Millisecond, | |
| "expecting %d in flight requests to block", | |
| defaultMaxConcurrentSecretReads, | |
| ) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the problem this PR solves?
ReadSecretsmakes 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 themax_pending_bulk_dispatchescap 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
ReadSecretsper-checkin call ~24h before OOMs began.How does this PR solve the problem?
Adds a
semaphore.Weighted(readSecretsLimit) toBulker, capping concurrent in-flight secret reads at 32 (matchingapikeyLimit).ReadSecretsacquires 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. SettingreadSecretsLimittonil(by passingWithMaxConcurrentSecretReads(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
apikeyMaxParallelis handled.Files changed
internal/pkg/bulk/engine.go: addreadSecretsLimit *semaphore.WeightedtoBulker; initialize inNewBulkerwhen limit > 0; nil-check acquire/release inReadSecrets; adddefaultMaxConcurrentSecretReads = 32internal/pkg/bulk/opt.go: addmaxConcurrentSecretReadsfield,WithMaxConcurrentSecretReadsBulkOpt, zerolog logginginternal/pkg/bulk/secret_limit_test.go: unit tests covering the concurrency limit, context cancellation while waiting, zero-value (no limit), and default capacityHow to test this PR locally
Design Checklist
Checklist
./changelog/fragmentsusing the changelog toolRelated issues