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..6fc726e7c8 --- /dev/null +++ b/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 diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go index b5078f6985..d5723255d6 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" @@ -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") diff --git a/internal/pkg/bulk/opt.go b/internal/pkg/bulk/opt.go index e9b8141fc2..751865dead 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 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 diff --git a/internal/pkg/bulk/secret_limit_test.go b/internal/pkg/bulk/secret_limit_test.go new file mode 100644 index 0000000000..24a62d7dae --- /dev/null +++ b/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() +}