From c19706029b797a2a69de08e6443bd8d197d347fd Mon Sep 17 00:00:00 2001 From: George Hahn Date: Thu, 23 Jul 2026 17:28:55 -0500 Subject: [PATCH 1/2] Add adaptive (memory-feedback) staged-startup pacing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The V1 pacing was a fixed timer between coarse stages. That's crude on two axes: it's time-based (wastes time when steps are cheap, too fast when heavy) and order-sensitive (two heavy subsystems in the same stage still stack). Add a shared pacer (pkg/util/stagedstart) with two modes, selected by staged_start.mode: - interval (default, = V1): fixed staged_start.stage_interval between steps. - adaptive: run steps one at a time and advance as soon as the process's own memory settles (growth over settle_window < settle_threshold_bytes after step_min), then reclaim. Because no two heavy inits overlap, startup order stops affecting the memory peak — no manual ordering needed. Bounded worst case: each adaptive step is capped at step_max; on timeout the next step proceeds and a warning is logged, so a process under memory pressure makes progress (and fails loudly) instead of hanging. The settle signal is this process's Go-runtime memory retained from the OS via the stdlib runtime/metrics package — portable across every OS, non-STW, and no new dependency. The sampler is injectable so a truer RSS source can be swapped in. All three staged binaries (core agent sequencer, system-probe module loader, security-agent) now share this pacer. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../subcommands/start/command.go | 30 +-- .../startupsequencer/impl/startupsequencer.go | 67 +++--- .../impl/startupsequencer_test.go | 34 ++- pkg/config/setup/common_settings.go | 15 ++ pkg/system-probe/api/module/loader.go | 34 +-- pkg/util/stagedstart/pacer.go | 220 ++++++++++++++++++ pkg/util/stagedstart/pacer_test.go | 143 ++++++++++++ ...ged-startup-adaptive-4b1e9c7a6d2f8e05.yaml | 11 + 8 files changed, 467 insertions(+), 87 deletions(-) create mode 100644 pkg/util/stagedstart/pacer.go create mode 100644 pkg/util/stagedstart/pacer_test.go create mode 100644 releasenotes/notes/staged-startup-adaptive-4b1e9c7a6d2f8e05.yaml diff --git a/cmd/security-agent/subcommands/start/command.go b/cmd/security-agent/subcommands/start/command.go index a3a68c10e6cb..1580b9175373 100644 --- a/cmd/security-agent/subcommands/start/command.go +++ b/cmd/security-agent/subcommands/start/command.go @@ -15,8 +15,6 @@ import ( _ "net/http/pprof" // Blank import used because this isn't directly used in this file "os" "os/signal" - "runtime" - "runtime/debug" "strings" "syscall" "time" @@ -77,6 +75,7 @@ import ( "github.com/DataDog/datadog-agent/pkg/util/defaultpaths" "github.com/DataDog/datadog-agent/pkg/util/fxutil" "github.com/DataDog/datadog-agent/pkg/util/profiling" + "github.com/DataDog/datadog-agent/pkg/util/stagedstart" "github.com/DataDog/datadog-agent/pkg/util/startstop" "github.com/DataDog/datadog-agent/pkg/version" ) @@ -211,23 +210,18 @@ func Commands(globalParams *command.GlobalParams) []*cobra.Command { } // stagedStartPace spreads security-agent subsystem startup when staged startup -// is enabled: it reclaims the transient memory allocated by prior initialization -// and pauses briefly before starting the next subsystem, so their startup memory -// peaks do not stack into a single spike (both within this process and relative -// to the other agent processes starting at the same time). When staged startup -// is disabled this is a no-op. +// is enabled: before starting the named subsystem it reclaims transient memory +// from prior initialization and waits (a fixed interval, or until this process's +// memory settles in adaptive mode) so their startup memory peaks do not stack +// into a single spike, both within this process and relative to the other agent +// processes starting at the same time. When staged startup is disabled this is a +// no-op. func stagedStartPace(cfg config.Component, logger log.Component, name string) { - if !cfg.GetBool("staged_start.enabled") { - return - } - if cfg.GetBool("staged_start.free_os_memory") { - runtime.GC() - debug.FreeOSMemory() - } - if interval := cfg.GetDuration("staged_start.stage_interval"); interval > 0 { - logger.Infof("staged startup: pacing %s startup by %s", name, interval) - time.Sleep(interval) - } + pacer := stagedstart.NewPacer(stagedstart.ConfigFromReader(cfg), + func(f string, a ...interface{}) { logger.Infof(f, a...) }, + func(f string, a ...interface{}) { logger.Warnf(f, a...) }, + ) + pacer.Pace(context.Background(), name) } // start will start the security-agent. diff --git a/comp/core/startupsequencer/impl/startupsequencer.go b/comp/core/startupsequencer/impl/startupsequencer.go index 7250afb8b57e..bf48e9e3019d 100644 --- a/comp/core/startupsequencer/impl/startupsequencer.go +++ b/comp/core/startupsequencer/impl/startupsequencer.go @@ -8,14 +8,13 @@ package startupsequencerimpl import ( "context" - "runtime" - "runtime/debug" "sync" "time" config "github.com/DataDog/datadog-agent/comp/core/config" log "github.com/DataDog/datadog-agent/comp/core/log/def" startupsequencer "github.com/DataDog/datadog-agent/comp/core/startupsequencer/def" + "github.com/DataDog/datadog-agent/pkg/util/stagedstart" ) // Requires defines the dependencies of the startupsequencer component. @@ -35,10 +34,9 @@ type deferredStart struct { } type sequencer struct { - log log.Component - enabled bool - interval time.Duration - freeOSMemory bool + log log.Component + enabled bool + pacer *stagedstart.Pacer mu sync.Mutex begun bool @@ -47,12 +45,15 @@ type sequencer struct { // NewComponent returns the staged startup sequencer. func NewComponent(reqs Requires) Provides { + cfg := stagedstart.ConfigFromReader(reqs.Config) return Provides{ Comp: &sequencer{ - log: reqs.Log, - enabled: reqs.Config.GetBool("staged_start.enabled"), - interval: reqs.Config.GetDuration("staged_start.stage_interval"), - freeOSMemory: reqs.Config.GetBool("staged_start.free_os_memory"), + log: reqs.Log, + enabled: cfg.Enabled, + pacer: stagedstart.NewPacer(cfg, + func(f string, a ...interface{}) { reqs.Log.Infof(f, a...) }, + func(f string, a ...interface{}) { reqs.Log.Warnf(f, a...) }, + ), }, } } @@ -94,34 +95,32 @@ func (s *sequencer) Begin(ctx context.Context) { } func (s *sequencer) run(ctx context.Context, deferred [startupsequencer.NumStages][]deferredStart) { - s.log.Infof("staged startup: beginning (stage interval %s)", s.interval) + // Flatten stages into a single ordered list. Stages are only a coarse + // priority (critical-path subsystems first); the pacer keeps consecutive + // items from allocating on top of one another, so the exact order within + // that priority does not affect the memory peak. + var items []deferredStart for stage := startupsequencer.Stage(0); stage < startupsequencer.NumStages; stage++ { - for _, d := range deferred[stage] { - if ctx.Err() != nil { - s.log.Infof("staged startup: aborted before %q (context cancelled)", d.name) - return - } - start := time.Now() - if err := d.fn(ctx); err != nil { - s.log.Errorf("staged startup: %q (stage %d) failed: %v", d.name, stage, err) - } else { - s.log.Debugf("staged startup: started %q (stage %d) in %s", d.name, stage, time.Since(start)) - } - } + items = append(items, deferred[stage]...) + } - // Return transient memory allocated during this stage to the OS before - // the next stage allocates, keeping the peak RSS close to steady state. - if s.freeOSMemory { - runtime.GC() - debug.FreeOSMemory() + s.log.Info("staged startup: beginning") + for i, d := range items { + if ctx.Err() != nil { + s.log.Infof("staged startup: aborted before %q (context cancelled)", d.name) + return + } + start := time.Now() + if err := d.fn(ctx); err != nil { + s.log.Errorf("staged startup: %q failed: %v", d.name, err) + } else { + s.log.Debugf("staged startup: started %q in %s", d.name, time.Since(start)) } - if stage < startupsequencer.NumStages-1 { - select { - case <-ctx.Done(): - return - case <-time.After(s.interval): - } + // Pace before releasing the next item (reclaims transient memory and, + // in adaptive mode, waits until this item's allocation settles). + if i < len(items)-1 { + s.pacer.Pace(ctx, d.name) } } s.log.Info("staged startup: complete") diff --git a/comp/core/startupsequencer/impl/startupsequencer_test.go b/comp/core/startupsequencer/impl/startupsequencer_test.go index 29910e511020..3e85ebe2da74 100644 --- a/comp/core/startupsequencer/impl/startupsequencer_test.go +++ b/comp/core/startupsequencer/impl/startupsequencer_test.go @@ -17,19 +17,21 @@ import ( logmock "github.com/DataDog/datadog-agent/comp/core/log/mock" startupsequencer "github.com/DataDog/datadog-agent/comp/core/startupsequencer/def" + "github.com/DataDog/datadog-agent/pkg/util/stagedstart" ) +// newTestSequencer builds a sequencer whose pacer is a no-op (no delay, no +// reclaim) so ordering/inline tests run instantly. func newTestSequencer(t *testing.T, enabled bool) *sequencer { + cfg := stagedstart.Config{Enabled: enabled} return &sequencer{ - log: logmock.New(t), - enabled: enabled, - interval: time.Millisecond, - freeOSMemory: false, + log: logmock.New(t), + enabled: enabled, + pacer: stagedstart.NewPacer(cfg, nil, nil), } } -// When disabled, Defer must run the work synchronously and propagate its error, -// so a Defer call from an OnStart hook is identical to running the work inline. +// When disabled, Defer must run the work synchronously and propagate its error. func TestDisabledRunsInline(t *testing.T) { s := newTestSequencer(t, false) @@ -45,7 +47,6 @@ func TestDisabledRunsInline(t *testing.T) { err = s.Defer(startupsequencer.StageChecks, "y", func(context.Context) error { return sentinel }) assert.ErrorIs(t, err, sentinel, "error must propagate when staging is disabled") - // Begin is a no-op when disabled. s.Begin(context.Background()) } @@ -66,7 +67,6 @@ func TestEnabledRunsInStageOrder(t *testing.T) { } done := make(chan struct{}) - // Register out of stage order to prove the sequencer orders by stage. require.NoError(t, s.Defer(startupsequencer.StageBackground, "background", func(ctx context.Context) error { _ = record("background")(ctx) close(done) @@ -75,7 +75,6 @@ func TestEnabledRunsInStageOrder(t *testing.T) { require.NoError(t, s.Defer(startupsequencer.StageCritical, "critical", record("critical"))) require.NoError(t, s.Defer(startupsequencer.StageIngest, "ingest", record("ingest"))) - // Nothing should have run before Begin. mu.Lock() assert.Empty(t, order, "no deferred work should run before Begin") mu.Unlock() @@ -93,8 +92,7 @@ func TestEnabledRunsInStageOrder(t *testing.T) { assert.Equal(t, []string{"critical", "ingest", "background"}, order) } -// Work registered after the sequence has begun must still run (inline), not be -// silently dropped. +// Work registered after the sequence has begun must still run (inline). func TestLateRegistrationRunsInline(t *testing.T) { s := newTestSequencer(t, true) s.Begin(context.Background()) @@ -107,10 +105,12 @@ func TestLateRegistrationRunsInline(t *testing.T) { assert.True(t, ran, "work registered after Begin should run inline") } -// A cancelled context must stop the sequence rather than running later stages. +// A cancelled context must stop the sequence rather than running later items. func TestContextCancellationStops(t *testing.T) { s := newTestSequencer(t, true) - s.interval = time.Hour // ensure we block between stages + // A long pacer interval makes the sequencer block between items so we can + // cancel mid-sequence. + s.pacer = stagedstart.NewPacer(stagedstart.Config{Enabled: true, Interval: time.Hour}, nil, nil) first := make(chan struct{}) require.NoError(t, s.Defer(startupsequencer.StageCritical, "first", func(context.Context) error { @@ -126,11 +126,9 @@ func TestContextCancellationStops(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) s.Begin(ctx) - <-first // first stage ran; sequencer now sleeping before stage 2 - cancel() // cancel during the inter-stage wait + <-first // first item ran; sequencer now paused before the second + cancel() // cancel during the inter-item wait - // The second stage is gated behind a one-hour inter-stage wait, so the only - // way it could run is if cancellation were ignored. time.Sleep(50 * time.Millisecond) - assert.False(t, secondRan, "later stage must not run after cancellation") + assert.False(t, secondRan, "later item must not run after cancellation") } diff --git a/pkg/config/setup/common_settings.go b/pkg/config/setup/common_settings.go index 531b69463695..d9227f6fdd1d 100644 --- a/pkg/config/setup/common_settings.go +++ b/pkg/config/setup/common_settings.go @@ -83,6 +83,21 @@ func initCoreAgentFull(config pkgconfigmodel.Setup) { config.BindEnvAndSetDefault("staged_start.stage_interval", 30*time.Second) // If true, transient memory is returned to the OS between staged startup stages. config.BindEnvAndSetDefault("staged_start.free_os_memory", true) + // Pacing mode between staged startup steps: "interval" (fixed stage_interval + // delay) or "adaptive" (advance as this process's memory settles, bounded by + // step_max so a process under memory pressure proceeds instead of hanging). + config.BindEnvAndSetDefault("staged_start.mode", "interval") + // Adaptive mode: minimum wait per step (lets a step's async work begin). + config.BindEnvAndSetDefault("staged_start.step_min", 1*time.Second) + // Adaptive mode: hard cap per step. On timeout the next step proceeds anyway + // (and a warning is logged) so startup never hangs on memory that won't settle. + config.BindEnvAndSetDefault("staged_start.step_max", 15*time.Second) + // Adaptive mode: a step is "settled" once retained memory grows less than + // settle_threshold_bytes over this window. + config.BindEnvAndSetDefault("staged_start.settle_window", 2*time.Second) + config.BindEnvAndSetDefault("staged_start.settle_threshold_bytes", int64(2*1024*1024)) + // Adaptive mode: how often to sample memory while waiting for a step to settle. + config.BindEnvAndSetDefault("staged_start.poll_interval", 250*time.Millisecond) // If true, then new version of disk v2 check will be used. // Otherwise, the python version of disk check will be used. diff --git a/pkg/system-probe/api/module/loader.go b/pkg/system-probe/api/module/loader.go index c405fd043e98..06c6b6204b45 100644 --- a/pkg/system-probe/api/module/loader.go +++ b/pkg/system-probe/api/module/loader.go @@ -11,8 +11,6 @@ import ( "fmt" "maps" "net/http" - "runtime" - "runtime/debug" "runtime/pprof" "sync" "time" @@ -21,6 +19,7 @@ import ( rcclient "github.com/DataDog/datadog-agent/comp/remote-config/rcclient/def" sysconfigtypes "github.com/DataDog/datadog-agent/pkg/system-probe/config/types" "github.com/DataDog/datadog-agent/pkg/util/log" + "github.com/DataDog/datadog-agent/pkg/util/stagedstart" ) var l *loader @@ -90,24 +89,25 @@ func Register(cfg *sysconfigtypes.Config, httpMux *http.ServeMux, factories []*F // Staged startup: creating a module loads its eBPF assets, which allocates a // large but transient amount of scratch (BTF parsing, bytecode, CO-RE // relocations). Loading every module back-to-back stacks those transients - // into a single startup memory peak. When enabled, reclaim each module's - // scratch and pause briefly between modules so the peak stays close to the - // steady-state footprint. The total added delay is bounded by one stage - // interval regardless of how many modules are enabled. - stagedStart := deps.CoreConfig.GetBool("staged_start.enabled") - freeOSMemory := deps.CoreConfig.GetBool("staged_start.free_os_memory") - var moduleDelay time.Duration - if stagedStart && len(enabledModulesFactories) > 1 { - moduleDelay = deps.CoreConfig.GetDuration("staged_start.stage_interval") / time.Duration(len(enabledModulesFactories)) + // into a single startup memory peak. When enabled, the pacer reclaims each + // module's scratch and waits between modules (a fixed slice of the stage + // interval, or until memory settles in adaptive mode) so the peak stays + // close to the steady-state footprint. + pacerCfg := stagedstart.ConfigFromReader(deps.CoreConfig) + // In fixed-interval mode, spread the whole stage interval across the modules + // (rather than waiting a full interval per module). Adaptive mode self-bounds + // per step via step_max and ignores this. + if !pacerCfg.Adaptive && len(enabledModulesFactories) > 1 { + pacerCfg.Interval /= time.Duration(len(enabledModulesFactories)) } + pacer := stagedstart.NewPacer(pacerCfg, + func(f string, a ...interface{}) { log.Infof(f, a...) }, + func(f string, a ...interface{}) { log.Warnf(f, a...) }, + ) for i, factory := range enabledModulesFactories { - if stagedStart && i > 0 { - if freeOSMemory { - runtime.GC() - debug.FreeOSMemory() - } - time.Sleep(moduleDelay) + if i > 0 { + pacer.Pace(context.Background(), string(enabledModulesFactories[i-1].Name)) } var err error diff --git a/pkg/util/stagedstart/pacer.go b/pkg/util/stagedstart/pacer.go new file mode 100644 index 000000000000..c17c50628caa --- /dev/null +++ b/pkg/util/stagedstart/pacer.go @@ -0,0 +1,220 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2016-present Datadog, Inc. + +// Package stagedstart provides a shared pacer used to spread expensive +// subsystem startup across time so the Agent's startup memory high-water-mark +// stays close to its steady-state footprint. +// +// A Pacer is called between startup steps. In "interval" mode it inserts a +// fixed delay. In "adaptive" mode it waits until this process's memory settles +// (the previous step has finished allocating and any transient scratch has been +// reclaimed) before releasing the next step, bounded by a hard maximum so a +// process under memory pressure fails loudly rather than hanging. Either way it +// then returns transient memory to the OS. +// +// The signal is this process's own Go-runtime memory retained from the OS, +// read via the stdlib runtime/metrics package — portable across every OS and +// non-stop-the-world. It is a proxy for the process's RSS contribution; the +// sampler is injectable so a truer RSS source (or a test double) can be used. +package stagedstart + +import ( + "context" + "runtime" + "runtime/debug" + "runtime/metrics" + "time" +) + +// configReader is the subset of the Agent config interface the pacer needs. +// Any config.Component / model.Reader satisfies it structurally. +type configReader interface { + GetBool(string) bool + GetString(string) string + GetDuration(string) time.Duration + GetInt64(string) int64 +} + +// Config controls the pacer. Zero value is a disabled no-op pacer. +type Config struct { + Enabled bool + FreeOSMemory bool + Adaptive bool // true => memory-feedback pacing; false => fixed interval + + // Interval mode. + Interval time.Duration + + // Adaptive mode. + StepMin time.Duration // minimum wait per step (let async work start) + StepMax time.Duration // hard cap per step; on hit, proceed and warn + SettleWindow time.Duration // memory growth is measured over this window + SettleThreshold uint64 // growth below this (bytes) over the window => settled + PollInterval time.Duration // sampling cadence +} + +// ConfigFromReader builds a Config from staged_start.* Agent config keys. +func ConfigFromReader(c configReader) Config { + threshold := c.GetInt64("staged_start.settle_threshold_bytes") + if threshold < 0 { + threshold = 0 + } + return Config{ + Enabled: c.GetBool("staged_start.enabled"), + FreeOSMemory: c.GetBool("staged_start.free_os_memory"), + Adaptive: c.GetString("staged_start.mode") == "adaptive", + Interval: c.GetDuration("staged_start.stage_interval"), + StepMin: c.GetDuration("staged_start.step_min"), + StepMax: c.GetDuration("staged_start.step_max"), + SettleWindow: c.GetDuration("staged_start.settle_window"), + SettleThreshold: uint64(threshold), + PollInterval: c.GetDuration("staged_start.poll_interval"), + } +} + +// Pacer paces startup steps. Construct with NewPacer; the zero value is not usable. +type Pacer struct { + cfg Config + + // Injectable for testing; defaulted by NewPacer. + sample func() uint64 + reclaim func() + now func() time.Time + // sleep waits for d or until ctx is done; returns false if ctx was cancelled. + sleep func(ctx context.Context, d time.Duration) bool + + infof func(string, ...interface{}) + warnf func(string, ...interface{}) +} + +// NewPacer returns a Pacer for the given config. infof/warnf may be nil. +func NewPacer(cfg Config, infof, warnf func(string, ...interface{})) *Pacer { + if infof == nil { + infof = func(string, ...interface{}) {} + } + if warnf == nil { + warnf = func(string, ...interface{}) {} + } + return &Pacer{ + cfg: cfg, + sample: retainedFromOS, + reclaim: func() { runtime.GC(); debug.FreeOSMemory() }, + now: time.Now, + sleep: sleepCtx, + infof: infof, + warnf: warnf, + } +} + +// Enabled reports whether the pacer will do anything. +func (p *Pacer) Enabled() bool { return p != nil && p.cfg.Enabled } + +// Pace is called between startup steps. It waits (fixed interval, or until +// memory settles in adaptive mode), then returns transient memory to the OS. +// name identifies the step just completed, for logging. It returns early if ctx +// is cancelled. It is a no-op when the pacer is disabled. +func (p *Pacer) Pace(ctx context.Context, name string) { + if !p.Enabled() { + return + } + if p.cfg.Adaptive { + p.waitUntilSettled(ctx, name) + } else if p.cfg.Interval > 0 { + p.sleep(ctx, p.cfg.Interval) + } + if ctx.Err() != nil { + return + } + if p.cfg.FreeOSMemory { + p.reclaim() + } +} + +// waitUntilSettled blocks until this process's retained memory stops growing +// (growth over SettleWindow < SettleThreshold) after at least StepMin, or until +// StepMax elapses — at which point it proceeds anyway and warns loudly, so a +// process under sustained memory pressure makes progress (and fails loudly) +// instead of hanging. +func (p *Pacer) waitUntilSettled(ctx context.Context, name string) { + start := p.now() + deadline := start.Add(p.cfg.StepMax) + + type point struct { + t time.Time + v uint64 + } + var window []point + + for { + if ctx.Err() != nil { + return + } + now := p.now() + window = append(window, point{now, p.sample()}) + // Drop samples older than the settle window. + cutoff := now.Add(-p.cfg.SettleWindow) + i := 0 + for i < len(window) && window[i].t.Before(cutoff) { + i++ + } + window = window[i:] + + elapsed := now.Sub(start) + if elapsed >= p.cfg.StepMin && now.Sub(window[0].t) >= p.cfg.SettleWindow { + var lo, hi uint64 = window[0].v, window[0].v + for _, pt := range window { + if pt.v < lo { + lo = pt.v + } + if pt.v > hi { + hi = pt.v + } + } + if hi-lo < p.cfg.SettleThreshold { + p.infof("staged startup: %q settled after %s", name, elapsed.Round(time.Millisecond)) + return + } + } + + if !now.Before(deadline) { + p.warnf("staged startup: %q did not settle within %s; proceeding (memory may still be growing)", name, p.cfg.StepMax) + return + } + + if !p.sleep(ctx, p.cfg.PollInterval) { + return + } + } +} + +// retainedFromOS returns the bytes this process's Go runtime currently holds +// from the OS (total mapped minus released). Non-stop-the-world and portable. +func retainedFromOS() uint64 { + samples := []metrics.Sample{ + {Name: "/memory/classes/total:bytes"}, + {Name: "/memory/classes/heap/released:bytes"}, + } + metrics.Read(samples) + total := samples[0].Value.Uint64() + released := samples[1].Value.Uint64() + if released > total { + return 0 + } + return total - released +} + +// sleepCtx waits for d or until ctx is done; returns false if ctx was cancelled. +func sleepCtx(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return ctx.Err() == nil + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} diff --git a/pkg/util/stagedstart/pacer_test.go b/pkg/util/stagedstart/pacer_test.go new file mode 100644 index 000000000000..eb685ff90024 --- /dev/null +++ b/pkg/util/stagedstart/pacer_test.go @@ -0,0 +1,143 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2016-present Datadog, Inc. + +package stagedstart + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeClock drives now() and sleep() deterministically: sleeping just advances +// virtual time, so tests run instantly and reproducibly on any OS. +type fakeClock struct { + t time.Time +} + +func (c *fakeClock) now() time.Time { return c.t } + +func (c *fakeClock) sleep(ctx context.Context, d time.Duration) bool { + if ctx.Err() != nil { + return false + } + c.t = c.t.Add(d) + return true +} + +func newTestPacer(cfg Config, clock *fakeClock, sample func() uint64) (*Pacer, *int32, *int32) { + var reclaims, warns int32 + p := NewPacer(cfg, nil, func(string, ...interface{}) { atomic.AddInt32(&warns, 1) }) + p.now = clock.now + p.sleep = clock.sleep + p.sample = sample + p.reclaim = func() { atomic.AddInt32(&reclaims, 1) } + return p, &reclaims, &warns +} + +func TestPaceDisabledIsNoop(t *testing.T) { + clock := &fakeClock{t: time.Unix(0, 0)} + p, reclaims, _ := newTestPacer(Config{Enabled: false}, clock, func() uint64 { return 0 }) + p.Pace(context.Background(), "x") + assert.Equal(t, int32(0), atomic.LoadInt32(reclaims), "disabled pacer must not reclaim") + assert.Equal(t, time.Unix(0, 0), clock.t, "disabled pacer must not sleep") +} + +func TestPaceIntervalMode(t *testing.T) { + clock := &fakeClock{t: time.Unix(0, 0)} + cfg := Config{Enabled: true, FreeOSMemory: true, Adaptive: false, Interval: 5 * time.Second} + p, reclaims, _ := newTestPacer(cfg, clock, func() uint64 { return 0 }) + + p.Pace(context.Background(), "x") + + assert.Equal(t, 5*time.Second, clock.t.Sub(time.Unix(0, 0)), "interval mode should wait exactly the interval") + assert.Equal(t, int32(1), atomic.LoadInt32(reclaims), "should reclaim once after the interval") +} + +func TestAdaptiveSettlesWhenMemoryFlattens(t *testing.T) { + clock := &fakeClock{t: time.Unix(0, 0)} + start := clock.t + cfg := Config{ + Enabled: true, FreeOSMemory: true, Adaptive: true, + StepMin: 1 * time.Second, StepMax: 30 * time.Second, + SettleWindow: 2 * time.Second, SettleThreshold: 2 << 20, PollInterval: 250 * time.Millisecond, + } + // Memory grows fast for the first 3s, then holds flat. + sample := func() uint64 { + elapsed := clock.t.Sub(start) + if elapsed < 3*time.Second { + return 100<<20 + uint64(elapsed/(250*time.Millisecond))*(10<<20) + } + return 220 << 20 + } + p, reclaims, warns := newTestPacer(cfg, clock, sample) + + p.Pace(context.Background(), "heavy") + + elapsed := clock.t.Sub(start) + assert.GreaterOrEqual(t, elapsed, 3*time.Second, "must not settle while still growing") + assert.Less(t, elapsed, cfg.StepMax, "should settle well before the hard cap") + assert.Equal(t, int32(1), atomic.LoadInt32(reclaims), "should reclaim once after settling") + assert.Equal(t, int32(0), atomic.LoadInt32(warns), "clean settle should not warn") +} + +// The bounded worst case: memory grows forever (memory pressure). The pacer must +// NOT hang — it proceeds at StepMax and warns loudly. +func TestAdaptiveBoundedWorstCase(t *testing.T) { + clock := &fakeClock{t: time.Unix(0, 0)} + start := clock.t + cfg := Config{ + Enabled: true, FreeOSMemory: true, Adaptive: true, + StepMin: 1 * time.Second, StepMax: 10 * time.Second, + SettleWindow: 2 * time.Second, SettleThreshold: 2 << 20, PollInterval: 250 * time.Millisecond, + } + sample := func() uint64 { + return 100<<20 + uint64(clock.t.Sub(start)/(250*time.Millisecond))*(10<<20) // never flattens + } + p, reclaims, warns := newTestPacer(cfg, clock, sample) + + p.Pace(context.Background(), "runaway") + + elapsed := clock.t.Sub(start) + assert.GreaterOrEqual(t, elapsed, cfg.StepMax, "must wait at least the hard cap before giving up") + assert.Less(t, elapsed, cfg.StepMax+time.Second, "must not exceed the hard cap by more than one poll") + assert.Equal(t, int32(1), atomic.LoadInt32(warns), "hitting the cap must warn loudly") + assert.Equal(t, int32(1), atomic.LoadInt32(reclaims), "should still reclaim after giving up") +} + +func TestAdaptiveRespectsStepMin(t *testing.T) { + clock := &fakeClock{t: time.Unix(0, 0)} + start := clock.t + cfg := Config{ + Enabled: true, Adaptive: true, + StepMin: 5 * time.Second, StepMax: 30 * time.Second, + SettleWindow: 1 * time.Second, SettleThreshold: 1 << 20, PollInterval: 250 * time.Millisecond, + } + // Perfectly flat from the start — but StepMin must still be honored. + p, _, _ := newTestPacer(cfg, clock, func() uint64 { return 42 << 20 }) + + p.Pace(context.Background(), "flat") + + assert.GreaterOrEqual(t, clock.t.Sub(start), cfg.StepMin, "must wait at least StepMin even if already flat") +} + +func TestAdaptiveStopsOnContextCancel(t *testing.T) { + clock := &fakeClock{t: time.Unix(0, 0)} + cfg := Config{ + Enabled: true, Adaptive: true, + StepMin: 1 * time.Second, StepMax: time.Hour, + SettleWindow: 2 * time.Second, SettleThreshold: 1 << 20, PollInterval: 250 * time.Millisecond, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + p, reclaims, _ := newTestPacer(cfg, clock, func() uint64 { return 0 }) + + require.NotPanics(t, func() { p.Pace(ctx, "cancelled") }) + assert.Equal(t, int32(0), atomic.LoadInt32(reclaims), "cancelled pace must not reclaim") +} diff --git a/releasenotes/notes/staged-startup-adaptive-4b1e9c7a6d2f8e05.yaml b/releasenotes/notes/staged-startup-adaptive-4b1e9c7a6d2f8e05.yaml new file mode 100644 index 000000000000..69bb44870884 --- /dev/null +++ b/releasenotes/notes/staged-startup-adaptive-4b1e9c7a6d2f8e05.yaml @@ -0,0 +1,11 @@ +--- +enhancements: + - | + Staged startup gained an adaptive pacing mode. Instead of waiting a fixed + delay between startup steps, ``staged_start.mode: adaptive`` advances to the + next step as soon as the process's memory settles (the previous step has + finished allocating and its transient scratch has been reclaimed), bounded + by ``staged_start.step_max`` so a process under memory pressure proceeds and + logs a warning rather than hanging. Because steps run one at a time and each + waits for the previous to settle, startup order no longer affects the memory + peak. The default remains ``interval`` (fixed ``staged_start.stage_interval``). From dfced88f6390442be4226e40f05e8e06da7f1b70 Mon Sep 17 00:00:00 2001 From: George Hahn Date: Thu, 23 Jul 2026 17:31:15 -0500 Subject: [PATCH 2/2] Default staged_start.mode to adaptive on-branch for SMP measurement Measurement scaffolding so the SMP comparison exercises adaptive pacing; revisit the shipping default (interval vs adaptive) before merge. Co-Authored-By: Claude Opus 4.8 (1M context) --- pkg/config/setup/common_settings.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pkg/config/setup/common_settings.go b/pkg/config/setup/common_settings.go index d9227f6fdd1d..0683683e135d 100644 --- a/pkg/config/setup/common_settings.go +++ b/pkg/config/setup/common_settings.go @@ -86,7 +86,9 @@ func initCoreAgentFull(config pkgconfigmodel.Setup) { // Pacing mode between staged startup steps: "interval" (fixed stage_interval // delay) or "adaptive" (advance as this process's memory settles, bounded by // step_max so a process under memory pressure proceeds instead of hanging). - config.BindEnvAndSetDefault("staged_start.mode", "interval") + // NOTE: temporarily defaulted to "adaptive" on this branch so SMP exercises + // it; revisit the shipping default before merge. + config.BindEnvAndSetDefault("staged_start.mode", "adaptive") // Adaptive mode: minimum wait per step (lets a step's async work begin). config.BindEnvAndSetDefault("staged_start.step_min", 1*time.Second) // Adaptive mode: hard cap per step. On timeout the next step proceeds anyway