From 44485a99bf12b9bfa934dd151804122efd23fe32 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:16:37 -0700 Subject: [PATCH 01/27] bulk: add max_concurrent_secret_reads config option ReadSecrets makes direct HTTP calls to ES (bypassing the bulk dispatch cap) for each secret reference in an agent checkin. Under high concurrent checkin load this can produce unbounded concurrent ES connections and memory pressure. Add a MaxConcurrentSecretReads config field to ServerBulk, backed by a semaphore.Weighted in the Bulker struct. When set, ReadSecrets acquires a slot before proceeding and releases it on return, bounding the number of simultaneous secret reads fleet-server can perform. Default is 0 (no limit) to preserve existing behaviour. Fleet-controller can set this via server.bulk.max_concurrent_secret_reads in the project config secret. Related: https://github.com/elastic/ingest-dev/issues/8991 --- internal/pkg/config/input.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/pkg/config/input.go b/internal/pkg/config/input.go index ed67d3779f..627904fc54 100644 --- a/internal/pkg/config/input.go +++ b/internal/pkg/config/input.go @@ -50,6 +50,7 @@ type ServerBulk struct { FlushThresholdSize int `config:"flush_threshold_size"` FlushMaxPending int `config:"flush_max_pending"` MaxPendingBulkDispatches int64 `config:"max_pending_bulk_dispatches"` + MaxConcurrentSecretReads int64 `config:"max_concurrent_secret_reads"` } func (c *ServerBulk) InitDefaults() { From 0d275808cb019f9ff3daa496072bf3edef8de49b Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:16:48 -0700 Subject: [PATCH 02/27] bulk: wire max_concurrent_secret_reads through BulkOpt Add bulkOptT field, WithMaxConcurrentSecretReads BulkOpt constructor, zerolog logging, and BulkOptsFromCfg wiring for the new max_concurrent_secret_reads config option. Related: https://github.com/elastic/ingest-dev/issues/8991 --- internal/pkg/bulk/opt.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/internal/pkg/bulk/opt.go b/internal/pkg/bulk/opt.go index e9b8141fc2..dfc86c5fd4 100644 --- a/internal/pkg/bulk/opt.go +++ b/internal/pkg/bulk/opt.go @@ -75,6 +75,7 @@ type bulkOptT struct { apikeyMaxParallel int apikeyMaxReqSize int maxPendingBulkDispatches int64 + maxConcurrentSecretReads int64 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 int64) 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) { @@ -182,6 +191,7 @@ func (o *bulkOptT) MarshalZerologObject(e *zerolog.Event) { e.Int("apikeyMaxParallel", o.apikeyMaxParallel) e.Int("apikeyMaxReqSize", o.apikeyMaxReqSize) e.Int64("maxPendingBulkDispatches", o.maxPendingBulkDispatches) + e.Int64("maxConcurrentSecretReads", o.maxConcurrentSecretReads) } // BulkOptsFromCfg transforms config to a slize of BulkOpt @@ -206,6 +216,7 @@ func BulkOptsFromCfg(cfg *config.Config) []BulkOpt { WithAPIKeyMaxParallel(maxKeyParallel), WithAPIKeyMaxRequestSize(cfg.Output.Elasticsearch.MaxContentLength), WithMaxPendingBulkDispatches(bulkCfg.MaxPendingBulkDispatches), + WithMaxConcurrentSecretReads(bulkCfg.MaxConcurrentSecretReads), WithPolicyTokens(policyTokens), } } From 0b9ad699b93bc1090263a9226e312d91cf5b41c3 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:18:03 -0700 Subject: [PATCH 03/27] bulk: enforce concurrent ReadSecrets limit via semaphore 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: https://github.com/elastic/ingest-dev/issues/8991 --- internal/pkg/bulk/engine.go | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index b5078f6985..6bb2239735 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -121,6 +121,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 @@ -139,7 +140,8 @@ const ( defaultAPIKeyMaxParallel = 32 defaultApikeyMaxReqSize = 100 * 1024 * 1024 defaultFlushContextTimeout = time.Minute * 1 - defaultMaxPendingBulkDispatches int64 = 0 // 0 means no limit + defaultMaxPendingBulkDispatches int64 = 0 // 0 means no limit + defaultMaxConcurrentSecretReads int64 = 0 // 0 means no limit // dispatchAbortDrainTimeout bounds how long the drain helper waits for // a late response from the Run loop on an abort from the second @@ -164,6 +166,12 @@ func NewBulker(es esapi.Transport, tracer *apm.Tracer, opts ...BulkOpt) *Bulker blkPool: sync.Pool{New: poolFunc}, flushBufPool: sync.Pool{New: func() any { return new(bytes.Buffer) }}, apikeyLimit: semaphore.NewWeighted(int64(bopts.apikeyMaxParallel)), + readSecretsLimit: func() *semaphore.Weighted { + if bopts.maxConcurrentSecretReads > 0 { + return semaphore.NewWeighted(bopts.maxConcurrentSecretReads) + } + return nil + }(), tracer: tracer, remoteOutputConfigMap: make(map[string]map[string]any), // remote ES bulkers @@ -325,6 +333,12 @@ func (b *Bulker) hasChangedAndUpdateRemoteOutputConfig(zlog zerolog.Logger, name // read secrets one by one as there is no bulk API yet to read them in one request func (b *Bulker) ReadSecrets(ctx context.Context, secretIds []string) (map[string]string, error) { + if b.readSecretsLimit != nil { + if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { + return nil, err + } + defer b.readSecretsLimit.Release(1) + } result := make(map[string]string) esClient := b.Client() for _, id := range secretIds { From ff0b7a4b5fdd76931c9e5bdfd27bb93de7234db2 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:51:00 -0700 Subject: [PATCH 04/27] bulk: fix goimports: restore constant alignment, set default to 32 --- internal/pkg/bulk/engine.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 6bb2239735..069dc164b0 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -140,8 +140,8 @@ const ( defaultAPIKeyMaxParallel = 32 defaultApikeyMaxReqSize = 100 * 1024 * 1024 defaultFlushContextTimeout = time.Minute * 1 - defaultMaxPendingBulkDispatches int64 = 0 // 0 means no limit - defaultMaxConcurrentSecretReads int64 = 0 // 0 means no limit + defaultMaxPendingBulkDispatches int64 = 0 // 0 means no limit + defaultMaxConcurrentSecretReads int64 = 32 // dispatchAbortDrainTimeout bounds how long the drain helper waits for // a late response from the Run loop on an abort from the second From 4b16bc7a9b549458a9ae551c1daa1160e099d170 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:51:49 -0700 Subject: [PATCH 05/27] bulk: set maxConcurrentSecretReads default in parseBulkOpts --- internal/pkg/bulk/opt.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/pkg/bulk/opt.go b/internal/pkg/bulk/opt.go index dfc86c5fd4..a93aeced1a 100644 --- a/internal/pkg/bulk/opt.go +++ b/internal/pkg/bulk/opt.go @@ -172,6 +172,7 @@ func parseBulkOpts(opts ...BulkOpt) bulkOptT { blockQueueSz: defaultBlockQueueSz, apikeyMaxReqSize: defaultApikeyMaxReqSize, maxPendingBulkDispatches: defaultMaxPendingBulkDispatches, + maxConcurrentSecretReads: defaultMaxConcurrentSecretReads, policyTokens: []config.PolicyToken{}, // default is empty } From 8dd3e0677f4035e11d469960473d30c9b4366987 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:51:50 -0700 Subject: [PATCH 06/27] config: default max_concurrent_secret_reads to 32 --- internal/pkg/config/input.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/pkg/config/input.go b/internal/pkg/config/input.go index 627904fc54..2735e03706 100644 --- a/internal/pkg/config/input.go +++ b/internal/pkg/config/input.go @@ -58,6 +58,7 @@ func (c *ServerBulk) InitDefaults() { c.FlushThresholdCount = 2048 c.FlushThresholdSize = 1024 * 1024 c.FlushMaxPending = 8 + c.MaxConcurrentSecretReads = 32 } // Server is the configuration for the server From 24351f468ec53760759c8f5dcd0519ac2eb3f794 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:53:36 -0700 Subject: [PATCH 07/27] bulk: remove spurious extra space in constant comment --- internal/pkg/bulk/engine.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 069dc164b0..0708049abd 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -140,7 +140,7 @@ const ( defaultAPIKeyMaxParallel = 32 defaultApikeyMaxReqSize = 100 * 1024 * 1024 defaultFlushContextTimeout = time.Minute * 1 - defaultMaxPendingBulkDispatches int64 = 0 // 0 means no limit + defaultMaxPendingBulkDispatches int64 = 0 // 0 means no limit defaultMaxConcurrentSecretReads int64 = 32 // dispatchAbortDrainTimeout bounds how long the drain helper waits for From b933c15483a4b624b02788bf1182b028b7e843e6 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:55:23 -0700 Subject: [PATCH 08/27] bulk: reuse defaultAPIKeyMaxParallel for readSecretsLimit default --- internal/pkg/bulk/engine.go | 1 - 1 file changed, 1 deletion(-) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 0708049abd..480ca11346 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -141,7 +141,6 @@ const ( defaultApikeyMaxReqSize = 100 * 1024 * 1024 defaultFlushContextTimeout = time.Minute * 1 defaultMaxPendingBulkDispatches int64 = 0 // 0 means no limit - defaultMaxConcurrentSecretReads int64 = 32 // dispatchAbortDrainTimeout bounds how long the drain helper waits for // a late response from the Run loop on an abort from the second From 3ecf3116082f5d40ba8242dc94ffd557e4bb7dfb Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:55:24 -0700 Subject: [PATCH 09/27] bulk: reuse defaultAPIKeyMaxParallel for readSecretsLimit default --- internal/pkg/bulk/opt.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/pkg/bulk/opt.go b/internal/pkg/bulk/opt.go index a93aeced1a..d340bc22e5 100644 --- a/internal/pkg/bulk/opt.go +++ b/internal/pkg/bulk/opt.go @@ -172,7 +172,7 @@ func parseBulkOpts(opts ...BulkOpt) bulkOptT { blockQueueSz: defaultBlockQueueSz, apikeyMaxReqSize: defaultApikeyMaxReqSize, maxPendingBulkDispatches: defaultMaxPendingBulkDispatches, - maxConcurrentSecretReads: defaultMaxConcurrentSecretReads, + maxConcurrentSecretReads: int64(defaultAPIKeyMaxParallel), policyTokens: []config.PolicyToken{}, // default is empty } From d6f5f1647d2c171a49b026bb84e0bb592d8eb112 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 09:59:15 -0700 Subject: [PATCH 10/27] bulk: always initialize readSecretsLimit, matching apikeyLimit pattern --- internal/pkg/bulk/engine.go | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 480ca11346..97160b91fd 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -165,12 +165,7 @@ func NewBulker(es esapi.Transport, tracer *apm.Tracer, opts ...BulkOpt) *Bulker blkPool: sync.Pool{New: poolFunc}, flushBufPool: sync.Pool{New: func() any { return new(bytes.Buffer) }}, apikeyLimit: semaphore.NewWeighted(int64(bopts.apikeyMaxParallel)), - readSecretsLimit: func() *semaphore.Weighted { - if bopts.maxConcurrentSecretReads > 0 { - return semaphore.NewWeighted(bopts.maxConcurrentSecretReads) - } - return nil - }(), + readSecretsLimit: semaphore.NewWeighted(int64(bopts.maxConcurrentSecretReads)), tracer: tracer, remoteOutputConfigMap: make(map[string]map[string]any), // remote ES bulkers @@ -332,12 +327,10 @@ func (b *Bulker) hasChangedAndUpdateRemoteOutputConfig(zlog zerolog.Logger, name // read secrets one by one as there is no bulk API yet to read them in one request func (b *Bulker) ReadSecrets(ctx context.Context, secretIds []string) (map[string]string, error) { - if b.readSecretsLimit != nil { - if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { - return nil, err - } - defer b.readSecretsLimit.Release(1) + if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { + return nil, err } + defer b.readSecretsLimit.Release(1) result := make(map[string]string) esClient := b.Client() for _, id := range secretIds { From cca2152bab4f5be5c234a53aea8ef1a7f7304885 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 10:04:44 -0700 Subject: [PATCH 11/27] bulk: acquire readSecretsLimit per secret call, matching apikeyLimit granularity --- internal/pkg/bulk/engine.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 97160b91fd..1639cd0b84 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -327,14 +327,14 @@ func (b *Bulker) hasChangedAndUpdateRemoteOutputConfig(zlog zerolog.Logger, name // read secrets one by one as there is no bulk API yet to read them in one request func (b *Bulker) ReadSecrets(ctx context.Context, secretIds []string) (map[string]string, error) { - if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { - return nil, err - } - defer b.readSecretsLimit.Release(1) result := make(map[string]string) esClient := b.Client() for _, id := range secretIds { + if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { + return nil, err + } val, err := ReadSecret(ctx, esClient, id) + 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") From afaf5ed87fdb4e3100579811903294bb984bab61 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 10:07:57 -0700 Subject: [PATCH 12/27] bulk: change maxConcurrentSecretReads to int, matching apikeyMaxParallel --- internal/pkg/bulk/opt.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/pkg/bulk/opt.go b/internal/pkg/bulk/opt.go index d340bc22e5..2d78b80439 100644 --- a/internal/pkg/bulk/opt.go +++ b/internal/pkg/bulk/opt.go @@ -75,7 +75,7 @@ type bulkOptT struct { apikeyMaxParallel int apikeyMaxReqSize int maxPendingBulkDispatches int64 - maxConcurrentSecretReads int64 + maxConcurrentSecretReads int policyTokens []config.PolicyToken bi build.Info } @@ -127,7 +127,7 @@ 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 int64) BulkOpt { +func WithMaxConcurrentSecretReads(max int) BulkOpt { return func(opt *bulkOptT) { opt.maxConcurrentSecretReads = max } @@ -172,7 +172,7 @@ func parseBulkOpts(opts ...BulkOpt) bulkOptT { blockQueueSz: defaultBlockQueueSz, apikeyMaxReqSize: defaultApikeyMaxReqSize, maxPendingBulkDispatches: defaultMaxPendingBulkDispatches, - maxConcurrentSecretReads: int64(defaultAPIKeyMaxParallel), + maxConcurrentSecretReads: defaultAPIKeyMaxParallel, policyTokens: []config.PolicyToken{}, // default is empty } @@ -192,7 +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.Int64("maxConcurrentSecretReads", o.maxConcurrentSecretReads) + e.Int("maxConcurrentSecretReads", o.maxConcurrentSecretReads) } // BulkOptsFromCfg transforms config to a slize of BulkOpt From 2fc3907742a89ac81f062a81821f22998770e5a0 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 10:07:58 -0700 Subject: [PATCH 13/27] config: use named constant and int type for MaxConcurrentSecretReads --- internal/pkg/config/input.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/internal/pkg/config/input.go b/internal/pkg/config/input.go index 2735e03706..a27f2b717d 100644 --- a/internal/pkg/config/input.go +++ b/internal/pkg/config/input.go @@ -44,13 +44,16 @@ type ServerTLS struct { Cert string `config:"cert"` } +// 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 int64 `config:"max_concurrent_secret_reads"` + MaxConcurrentSecretReads int `config:"max_concurrent_secret_reads"` } func (c *ServerBulk) InitDefaults() { @@ -58,7 +61,7 @@ func (c *ServerBulk) InitDefaults() { c.FlushThresholdCount = 2048 c.FlushThresholdSize = 1024 * 1024 c.FlushMaxPending = 8 - c.MaxConcurrentSecretReads = 32 + c.MaxConcurrentSecretReads = defaultMaxConcurrentSecretReads } // Server is the configuration for the server From 9158e59f6ffc8130a62314dab1b60dd49187959d Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 10:08:13 -0700 Subject: [PATCH 14/27] bulk: add comment on readSecretsLimit acquire --- internal/pkg/bulk/engine.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 1639cd0b84..b882bddfb7 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -330,6 +330,7 @@ func (b *Bulker) ReadSecrets(ctx context.Context, secretIds []string) (map[strin result := make(map[string]string) esClient := b.Client() for _, id := range secretIds { + // limit concurrent direct ES secret reads, matching apikeyLimit granularity if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { return nil, err } From 5d857b520a50d321e8244b368233e6409ec0ec18 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 10:09:16 -0700 Subject: [PATCH 15/27] bulk: trim acquire comment --- internal/pkg/bulk/engine.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index b882bddfb7..2ef49608f3 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -330,7 +330,7 @@ func (b *Bulker) ReadSecrets(ctx context.Context, secretIds []string) (map[strin result := make(map[string]string) esClient := b.Client() for _, id := range secretIds { - // limit concurrent direct ES secret reads, matching apikeyLimit granularity + // limit concurrent direct ES secret reads if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { return nil, err } From 3b9df172432032e1ad583fd3b4e4b78d5a75f55c Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 10:24:18 -0700 Subject: [PATCH 16/27] changelog: add fragment for ReadSecrets concurrency limit --- .../1786037058-rate-limit-read-secrets.yaml | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 changelog/fragments/1786037058-rate-limit-read-secrets.yaml diff --git a/changelog/fragments/1786037058-rate-limit-read-secrets.yaml b/changelog/fragments/1786037058-rate-limit-read-secrets.yaml new file mode 100644 index 0000000000..4303e2da8a --- /dev/null +++ b/changelog/fragments/1786037058-rate-limit-read-secrets.yaml @@ -0,0 +1,19 @@ +kind: bug + +summary: Add configurable concurrency limit for ReadSecrets to reduce memory pressure under high checkin load + +description: | + ReadSecrets makes direct HTTP calls to the ES Fleet secrets API (one per + secret reference per agent checkin) outside the bulk dispatch queue, so it + is not subject to the max_pending_bulk_dispatches cap. Under high concurrent + checkin load this can cause unbounded concurrent ES connections and additional + goroutine and heap pressure, contributing to OOM conditions on large + serverless projects. + + A new max_concurrent_secret_reads option (server.bulk.max_concurrent_secret_reads, + default 32) is added to ServerBulk, backed by a semaphore.Weighted in the + Bulker struct. ReadSecrets acquires one slot per secret read and releases it + immediately after, matching the existing apikeyLimit pattern. Setting the + value to 0 is not supported. + +component: fleet-server From 2fea54e13f97311491123856723b15a3e622c482 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 10:29:12 -0700 Subject: [PATCH 17/27] bulk: add unit tests for ReadSecrets concurrency limit --- internal/pkg/bulk/secret_limit_test.go | 149 +++++++++++++++++++++++++ 1 file changed, 149 insertions(+) create mode 100644 internal/pkg/bulk/secret_limit_test.go diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go new file mode 100644 index 0000000000..522951ddf7 --- /dev/null +++ b/internal/pkg/bulk/secret_limit_test.go @@ -0,0 +1,149 @@ +// 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" + + "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{} + inFlight atomic.Int64 +} + +func (m *blockingTransport) RoundTrip(_ *http.Request) (*http.Response, error) { + m.inFlight.Add(1) + defer m.inFlight.Add(-1) + <-m.gate + body := `{"value":"test"}` + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + 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{})} + b := newTestBulkerWithTransport(t, mt, WithMaxConcurrentSecretReads(1)) + + var wg sync.WaitGroup + errs := make([]error, 2) + + 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()) + + // Unblock both goroutines and wait for them to finish. + close(mt.gate) + 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(context.Background(), 1) + require.NoError(t, err) + defer b.readSecretsLimit.Release(1) + + // Call ReadSecrets with an already-cancelled context. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err = b.ReadSecrets(ctx, []string{"id1"}) + require.ErrorIs(t, err, context.Canceled) +} + +// 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 +// returns context.Canceled immediately. +func TestReadSecretsDefaultConcurrency(t *testing.T) { + mt := &blockingTransport{gate: make(chan struct{})} + + // No WithMaxConcurrentSecretReads option → uses defaultAPIKeyMaxParallel. + b := newTestBulkerWithTransport(t, mt) + + require.NotNil(t, b.readSecretsLimit) + + // Launch defaultAPIKeyMaxParallel 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 defaultAPIKeyMaxParallel { + wg.Add(1) + go func() { + defer wg.Done() + _, _ = b.ReadSecrets(context.Background(), []string{fmt.Sprintf("id%d", i)}) + }() + } + + // Spin until all slots are occupied. + for mt.inFlight.Load() < int64(defaultAPIKeyMaxParallel) { + // wait for all goroutines to enter the transport + } + + // The semaphore is now full; a cancelled-context acquire must return immediately. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := b.readSecretsLimit.Acquire(ctx, 1) + require.ErrorIs(t, err, context.Canceled) + + // Unblock all goroutines. + close(mt.gate) + wg.Wait() +} From 7b6ab54e9e74fde1e7966771b8f06c6f1bd6fe69 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Thu, 6 Aug 2026 16:35:45 -0700 Subject: [PATCH 18/27] fix: reorder imports in engine.go and simplify changelog fragment Co-Authored-By: Claude Sonnet 4.6 --- .../1786037058-rate-limit-read-secrets.yaml | 22 ++++++++----------- internal/pkg/bulk/engine.go | 5 +++-- 2 files changed, 12 insertions(+), 15 deletions(-) diff --git a/changelog/fragments/1786037058-rate-limit-read-secrets.yaml b/changelog/fragments/1786037058-rate-limit-read-secrets.yaml index 4303e2da8a..24e1091c59 100644 --- a/changelog/fragments/1786037058-rate-limit-read-secrets.yaml +++ b/changelog/fragments/1786037058-rate-limit-read-secrets.yaml @@ -1,19 +1,15 @@ kind: bug -summary: Add configurable concurrency limit for ReadSecrets to reduce memory pressure under high checkin load +summary: Add configurable limit on concurrent secret reads to reduce memory pressure under high agent check-in load description: | - ReadSecrets makes direct HTTP calls to the ES Fleet secrets API (one per - secret reference per agent checkin) outside the bulk dispatch queue, so it - is not subject to the max_pending_bulk_dispatches cap. Under high concurrent - checkin load this can cause unbounded concurrent ES connections and additional - goroutine and heap pressure, contributing to OOM conditions on large - serverless projects. - - A new max_concurrent_secret_reads option (server.bulk.max_concurrent_secret_reads, - default 32) is added to ServerBulk, backed by a semaphore.Weighted in the - Bulker struct. ReadSecrets acquires one slot per secret read and releases it - immediately after, matching the existing apikeyLimit pattern. Setting the - value to 0 is not supported. + 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. + + 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. component: fleet-server diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 2ef49608f3..69830afc0a 100644 --- a/internal/pkg/bulk/engine.go +++ b/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" @@ -165,7 +166,7 @@ func NewBulker(es esapi.Transport, tracer *apm.Tracer, opts ...BulkOpt) *Bulker 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)), + readSecretsLimit: semaphore.NewWeighted(int64(bopts.maxConcurrentSecretReads)), tracer: tracer, remoteOutputConfigMap: make(map[string]map[string]any), // remote ES bulkers From 315c5cebdbd18822bc879f6e386d26ce27f784d1 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Fri, 7 Aug 2026 07:34:20 -0700 Subject: [PATCH 19/27] bulk: fix zero-value readSecretsLimit and add own default constant 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 --- internal/pkg/bulk/engine.go | 19 ++++++++++----- internal/pkg/bulk/opt.go | 2 +- internal/pkg/bulk/secret_limit_test.go | 33 +++++++++++++++++++------- internal/pkg/config/input.go | 1 - 4 files changed, 38 insertions(+), 17 deletions(-) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 69830afc0a..68ab97a444 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -139,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 @@ -159,19 +160,22 @@ 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), 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, remoteOutputConfigMap: make(map[string]map[string]any), // remote ES bulkers bulkerMap: make(map[string]Bulk), } + if bopts.maxConcurrentSecretReads > 0 { + b.readSecretsLimit = semaphore.NewWeighted(int64(bopts.maxConcurrentSecretReads)) + } + return b } func (b *Bulker) GetBulker(outputName string) Bulk { @@ -331,12 +335,15 @@ func (b *Bulker) ReadSecrets(ctx context.Context, secretIds []string) (map[strin result := make(map[string]string) esClient := b.Client() for _, id := range secretIds { - // limit concurrent direct ES secret reads - if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { - return nil, err + if b.readSecretsLimit != nil { + if err := b.readSecretsLimit.Acquire(ctx, 1); err != nil { + return nil, err + } } val, err := ReadSecret(ctx, esClient, id) - b.readSecretsLimit.Release(1) + 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") diff --git a/internal/pkg/bulk/opt.go b/internal/pkg/bulk/opt.go index 2d78b80439..d643a69818 100644 --- a/internal/pkg/bulk/opt.go +++ b/internal/pkg/bulk/opt.go @@ -172,7 +172,7 @@ func parseBulkOpts(opts ...BulkOpt) bulkOptT { blockQueueSz: defaultBlockQueueSz, apikeyMaxReqSize: defaultApikeyMaxReqSize, maxPendingBulkDispatches: defaultMaxPendingBulkDispatches, - maxConcurrentSecretReads: defaultAPIKeyMaxParallel, + maxConcurrentSecretReads: defaultMaxConcurrentSecretReads, policyTokens: []config.PolicyToken{}, // default is empty } diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go index 522951ddf7..58b7e14b94 100644 --- a/internal/pkg/bulk/secret_limit_test.go +++ b/internal/pkg/bulk/secret_limit_test.go @@ -106,25 +106,40 @@ func TestReadSecretsContextCancelledWhileWaiting(t *testing.T) { 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(context.Background(), []string{"id1"}) + require.NoError(t, err) +} + // 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 +// 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{})} - // No WithMaxConcurrentSecretReads option → uses defaultAPIKeyMaxParallel. + // No WithMaxConcurrentSecretReads option → uses defaultMaxConcurrentSecretReads. b := newTestBulkerWithTransport(t, mt) require.NotNil(t, b.readSecretsLimit) - // Launch defaultAPIKeyMaxParallel 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. + // 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 defaultAPIKeyMaxParallel { + for i := range defaultMaxConcurrentSecretReads { wg.Add(1) go func() { defer wg.Done() @@ -133,7 +148,7 @@ func TestReadSecretsDefaultConcurrency(t *testing.T) { } // Spin until all slots are occupied. - for mt.inFlight.Load() < int64(defaultAPIKeyMaxParallel) { + for mt.inFlight.Load() < int64(defaultMaxConcurrentSecretReads) { // wait for all goroutines to enter the transport } diff --git a/internal/pkg/config/input.go b/internal/pkg/config/input.go index a27f2b717d..d13f373a9d 100644 --- a/internal/pkg/config/input.go +++ b/internal/pkg/config/input.go @@ -44,7 +44,6 @@ type ServerTLS struct { Cert string `config:"cert"` } -// defaultMaxConcurrentSecretReads matches defaultAPIKeyMaxParallel in the bulk package. const defaultMaxConcurrentSecretReads = 32 type ServerBulk struct { From 4a29c59bd13cb1707c5d6bdb9efcaf9793a31321 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Fri, 7 Aug 2026 07:36:10 -0700 Subject: [PATCH 20/27] bulk: remove max_concurrent_secret_reads from user-facing config 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 --- changelog/fragments/1786037058-rate-limit-read-secrets.yaml | 5 ++--- internal/pkg/bulk/opt.go | 1 - internal/pkg/config/input.go | 4 ---- 3 files changed, 2 insertions(+), 8 deletions(-) diff --git a/changelog/fragments/1786037058-rate-limit-read-secrets.yaml b/changelog/fragments/1786037058-rate-limit-read-secrets.yaml index 24e1091c59..6fc726e7c8 100644 --- a/changelog/fragments/1786037058-rate-limit-read-secrets.yaml +++ b/changelog/fragments/1786037058-rate-limit-read-secrets.yaml @@ -1,6 +1,6 @@ kind: bug -summary: Add configurable limit on concurrent secret reads to reduce memory pressure under high agent check-in load +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 @@ -8,8 +8,7 @@ description: | policy secret references, contributing to out-of-memory conditions on large deployments. - A new configuration option, server.bulk.max_concurrent_secret_reads - (default 32), limits how many secret reads can be in-flight at once. + Secret reads are now capped at 32 concurrent in-flight requests. Excess reads wait until a slot is available. component: fleet-server diff --git a/internal/pkg/bulk/opt.go b/internal/pkg/bulk/opt.go index d643a69818..751865dead 100644 --- a/internal/pkg/bulk/opt.go +++ b/internal/pkg/bulk/opt.go @@ -217,7 +217,6 @@ func BulkOptsFromCfg(cfg *config.Config) []BulkOpt { WithAPIKeyMaxParallel(maxKeyParallel), WithAPIKeyMaxRequestSize(cfg.Output.Elasticsearch.MaxContentLength), WithMaxPendingBulkDispatches(bulkCfg.MaxPendingBulkDispatches), - WithMaxConcurrentSecretReads(bulkCfg.MaxConcurrentSecretReads), WithPolicyTokens(policyTokens), } } diff --git a/internal/pkg/config/input.go b/internal/pkg/config/input.go index d13f373a9d..ed67d3779f 100644 --- a/internal/pkg/config/input.go +++ b/internal/pkg/config/input.go @@ -44,15 +44,12 @@ type ServerTLS struct { Cert string `config:"cert"` } -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() { @@ -60,7 +57,6 @@ func (c *ServerBulk) InitDefaults() { c.FlushThresholdCount = 2048 c.FlushThresholdSize = 1024 * 1024 c.FlushMaxPending = 8 - c.MaxConcurrentSecretReads = defaultMaxConcurrentSecretReads } // Server is the configuration for the server From 8ff68188f182088db5d615de91af3d8f58f9f067 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Fri, 7 Aug 2026 07:40:33 -0700 Subject: [PATCH 21/27] bulk: add comment explaining nil readSecretsLimit for zero value Co-Authored-By: Claude Sonnet 4.6 --- internal/pkg/bulk/engine.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index 68ab97a444..d5723255d6 100644 --- a/internal/pkg/bulk/engine.go +++ b/internal/pkg/bulk/engine.go @@ -172,6 +172,7 @@ 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)) } From c0d5581cb59390d040f2c44745fa0e3b414374ef Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Fri, 7 Aug 2026 07:43:57 -0700 Subject: [PATCH 22/27] test: add X-Elastic-Product header to blockingTransport responses 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 --- internal/pkg/bulk/secret_limit_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go index 58b7e14b94..ed52e794bc 100644 --- a/internal/pkg/bulk/secret_limit_test.go +++ b/internal/pkg/bulk/secret_limit_test.go @@ -29,10 +29,12 @@ 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: make(http.Header), + Header: h, Body: io.NopCloser(strings.NewReader(body)), }, nil } From 8a19124e535f7a78c7d806ecc57b26cc075e11d2 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Fri, 7 Aug 2026 08:59:08 -0700 Subject: [PATCH 23/27] test: use wg.Go in TestReadSecretsDefaultConcurrency 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 --- internal/pkg/bulk/secret_limit_test.go | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go index ed52e794bc..ad4930c5be 100644 --- a/internal/pkg/bulk/secret_limit_test.go +++ b/internal/pkg/bulk/secret_limit_test.go @@ -142,11 +142,9 @@ func TestReadSecretsDefaultConcurrency(t *testing.T) { // slot and block in the transport, filling all capacity. var wg sync.WaitGroup for i := range defaultMaxConcurrentSecretReads { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, _ = b.ReadSecrets(context.Background(), []string{fmt.Sprintf("id%d", i)}) - }() + }) } // Spin until all slots are occupied. From aaa3d9de8eacbe1e5b2167b3b4d39769fa7f5835 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 10 Aug 2026 08:40:06 -0700 Subject: [PATCH 24/27] test: bound secret read concurrency waits --- internal/pkg/bulk/secret_limit_test.go | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go index ad4930c5be..ef2150aaf5 100644 --- a/internal/pkg/bulk/secret_limit_test.go +++ b/internal/pkg/bulk/secret_limit_test.go @@ -13,6 +13,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/elastic/go-elasticsearch/v8" "github.com/stretchr/testify/require" @@ -69,10 +70,9 @@ func TestReadSecretsLimitsConcurrency(t *testing.T) { _, 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 - } + 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. @@ -147,10 +147,9 @@ func TestReadSecretsDefaultConcurrency(t *testing.T) { }) } - // 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() >= 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(context.Background()) From 1d0a31d7fb3597e0fb67cea92f1ef180554abe7b Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 10 Aug 2026 08:40:19 -0700 Subject: [PATCH 25/27] test: use test context for secret read limits --- internal/pkg/bulk/secret_limit_test.go | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go index ef2150aaf5..a7064feb00 100644 --- a/internal/pkg/bulk/secret_limit_test.go +++ b/internal/pkg/bulk/secret_limit_test.go @@ -63,11 +63,11 @@ func TestReadSecretsLimitsConcurrency(t *testing.T) { wg.Add(2) go func() { defer wg.Done() - _, errs[0] = b.ReadSecrets(context.Background(), []string{"id1"}) + _, errs[0] = b.ReadSecrets(t.Context(), []string{"id1"}) }() go func() { defer wg.Done() - _, errs[1] = b.ReadSecrets(context.Background(), []string{"id2"}) + _, errs[1] = b.ReadSecrets(t.Context(), []string{"id2"}) }() require.Eventually(t, func() bool { @@ -96,12 +96,12 @@ func TestReadSecretsContextCancelledWhileWaiting(t *testing.T) { b := newTestBulkerWithTransport(t, mt, WithMaxConcurrentSecretReads(1)) // Manually hold the only semaphore slot so ReadSecrets must wait. - err := b.readSecretsLimit.Acquire(context.Background(), 1) + 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(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) cancel() _, err = b.ReadSecrets(ctx, []string{"id1"}) @@ -119,7 +119,7 @@ func TestReadSecretsNoLimitWhenZero(t *testing.T) { require.Nil(t, b.readSecretsLimit) - _, err := b.ReadSecrets(context.Background(), []string{"id1"}) + _, err := b.ReadSecrets(t.Context(), []string{"id1"}) require.NoError(t, err) } @@ -143,7 +143,7 @@ func TestReadSecretsDefaultConcurrency(t *testing.T) { var wg sync.WaitGroup for i := range defaultMaxConcurrentSecretReads { wg.Go(func() { - _, _ = b.ReadSecrets(context.Background(), []string{fmt.Sprintf("id%d", i)}) + _, _ = b.ReadSecrets(t.Context(), []string{fmt.Sprintf("id%d", i)}) }) } @@ -152,7 +152,7 @@ func TestReadSecretsDefaultConcurrency(t *testing.T) { }, 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(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) cancel() err := b.readSecretsLimit.Acquire(ctx, 1) require.ErrorIs(t, err, context.Canceled) From 229dfcefb7b24fd756dadd820e02e1d58b3567a4 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 10 Aug 2026 08:46:06 -0700 Subject: [PATCH 26/27] test: use wait group go for secret reads --- internal/pkg/bulk/secret_limit_test.go | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go index a7064feb00..5e470c18ae 100644 --- a/internal/pkg/bulk/secret_limit_test.go +++ b/internal/pkg/bulk/secret_limit_test.go @@ -60,15 +60,12 @@ func TestReadSecretsLimitsConcurrency(t *testing.T) { var wg sync.WaitGroup errs := make([]error, 2) - wg.Add(2) - go func() { - defer wg.Done() + wg.Go(func() { _, errs[0] = b.ReadSecrets(t.Context(), []string{"id1"}) - }() - go func() { - defer wg.Done() + }) + wg.Go(func() { _, errs[1] = b.ReadSecrets(t.Context(), []string{"id2"}) - }() + }) require.Eventually(t, func() bool { return mt.inFlight.Load() >= 1 From bd742bad89f7b4c1c87f499bc312efe545987f52 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 10 Aug 2026 11:29:48 -0700 Subject: [PATCH 27/27] test: unblock secret read workers on failure --- internal/pkg/bulk/secret_limit_test.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go index 5e470c18ae..24a62d7dae 100644 --- a/internal/pkg/bulk/secret_limit_test.go +++ b/internal/pkg/bulk/secret_limit_test.go @@ -23,9 +23,14 @@ import ( // 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) @@ -55,6 +60,7 @@ func newTestBulkerWithTransport(t *testing.T, transport http.RoundTripper, opts // 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 @@ -76,7 +82,7 @@ func TestReadSecretsLimitsConcurrency(t *testing.T) { require.Equal(t, int64(1), mt.inFlight.Load()) // Unblock both goroutines and wait for them to finish. - close(mt.gate) + mt.unblock() wg.Wait() require.NoError(t, errs[0]) @@ -128,6 +134,7 @@ func TestReadSecretsNoLimitWhenZero(t *testing.T) { // 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) @@ -155,6 +162,6 @@ func TestReadSecretsDefaultConcurrency(t *testing.T) { require.ErrorIs(t, err, context.Canceled) // Unblock all goroutines. - close(mt.gate) + mt.unblock() wg.Wait() }