Add support to configure min GC intervals for soft and hard limits (#12450)

When servers are in a high memory usage mode, the amount of times GC is
called creates a high CPU usage which combined with a high ingest rate
limits capability to offload existing queued data.

Configuring the minimum interval even for hard limit, allows the system
to spend some CPU cycles between GCs to offload old data from
queues/batch processor. The most amount of data I've seen accumulated
are blocked in the batch processor queue (incoming channel) not in the
exporter queue.

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

.chloggen/add-support-for-gc.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: memorylimiter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support to configure min GC intervals for soft and hard limits.
+
+# One or more tracking issues or pull requests related to the change
+issues: [12450]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

extension/memorylimiterextension/factory.go

@@ -11,6 +11,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/extension/memorylimiterextension/internal/metadata"
+	"go.opentelemetry.io/collector/internal/memorylimiter"
 )
 
 // NewFactory returns a new factory for the Memory Limiter extension.
@@ -25,7 +26,7 @@ func NewFactory() extension.Factory {
 // CreateDefaultConfig creates the default configuration for extension. Notice
 // that the default configuration is expected to fail for this extension.
 func createDefaultConfig() component.Config {
-	return &Config{}
+	return memorylimiter.NewDefaultConfig()
 }
 
 func create(_ context.Context, set extension.Settings, cfg component.Config) (extension.Extension, error) {

extension/memorylimiterextension/memorylimiter_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 
-	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/internal/memorylimiter"
 	"go.opentelemetry.io/collector/internal/memorylimiter/iruntime"
 )
@@ -70,14 +70,20 @@ func TestMemoryPressureResponse(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			memorylimiter.GetMemoryFn = totalMemory
+			memorylimiter.GetMemoryFn = func() (uint64, error) {
+				return uint64(2048), nil
+			}
 			memorylimiter.ReadMemStatsFn = func(ms *runtime.MemStats) {
 				ms.Alloc = tt.memAlloc
 			}
+			t.Cleanup(func() {
+				memorylimiter.GetMemoryFn = iruntime.TotalMemory
+				memorylimiter.ReadMemStatsFn = runtime.ReadMemStats
+			})
 			ml, err := newMemoryLimiter(tt.mlCfg, zap.NewNop())
 			assert.NoError(t, err)
 
-			assert.NoError(t, ml.Start(ctx, &mockHost{}))
+			assert.NoError(t, ml.Start(ctx, componenttest.NewNopHost()))
 			ml.memLimiter.CheckMemLimits()
 			mustRefuse := ml.MustRefuse()
 			if tt.expectError {
@@ -88,20 +94,4 @@ func TestMemoryPressureResponse(t *testing.T) {
 			assert.NoError(t, ml.Shutdown(ctx))
 		})
 	}
-	t.Cleanup(func() {
-		memorylimiter.GetMemoryFn = iruntime.TotalMemory
-		memorylimiter.ReadMemStatsFn = runtime.ReadMemStats
-	})
-}
-
-type mockHost struct {
-	component.Host
-}
-
-func (h *mockHost) GetExtensions() map[component.ID]component.Component {
-	return make(map[component.ID]component.Component)
-}
-
-func totalMemory() (uint64, error) {
-	return uint64(2048), nil
 }

internal/memorylimiter/config.go

@@ -12,6 +12,7 @@ import (
 
 var (
 	errCheckIntervalOutOfRange        = errors.New("'check_interval' must be greater than zero")
+	errInconsistentGCMinInterval      = errors.New("'min_gc_interval_when_soft_limited' should be larger than 'min_gc_interval_when_hard_limited'")
 	errLimitOutOfRange                = errors.New("'limit_mib' or 'limit_percentage' must be greater than zero")
 	errSpikeLimitOutOfRange           = errors.New("'spike_limit_mib' must be smaller than 'limit_mib'")
 	errSpikeLimitPercentageOutOfRange = errors.New("'spike_limit_percentage' must be smaller than 'limit_percentage'")
@@ -26,6 +27,16 @@ type Config struct {
 	// checks will be performed.
 	CheckInterval time.Duration `mapstructure:"check_interval"`
 
+	// MinGCIntervalWhenSoftLimited minimum interval between forced GC when in soft (=limit_mib - spike_limit_mib) limited mode.
+	// Zero value means no minimum interval.
+	// GCs is a CPU-heavy operation and executing it too frequently may affect the recovery capabilities of the collector.
+	MinGCIntervalWhenSoftLimited time.Duration `mapstructure:"min_gc_interval_when_soft_limited"`
+
+	// MinGCIntervalWhenHardLimited minimum interval between forced GC when in hard (=limit_mib) limited mode.
+	// Zero value means no minimum interval.
+	// GCs is a CPU-heavy operation and executing it too frequently may affect the recovery capabilities of the collector.
+	MinGCIntervalWhenHardLimited time.Duration `mapstructure:"min_gc_interval_when_hard_limited"`
+
 	// MemoryLimitMiB is the maximum amount of memory, in MiB, targeted to be
 	// allocated by the process.
 	MemoryLimitMiB uint32 `mapstructure:"limit_mib"`
@@ -45,11 +56,20 @@ type Config struct {
 
 var _ component.Config = (*Config)(nil)
 
+func NewDefaultConfig() *Config {
+	return &Config{
+		MinGCIntervalWhenSoftLimited: 10 * time.Second,
+	}
+}
+
 // Validate checks if the processor configuration is valid
 func (cfg *Config) Validate() error {
 	if cfg.CheckInterval <= 0 {
 		return errCheckIntervalOutOfRange
 	}
+	if cfg.MinGCIntervalWhenSoftLimited < cfg.MinGCIntervalWhenHardLimited {
+		return errInconsistentGCMinInterval
+	}
 	if cfg.MemoryLimitMiB == 0 && cfg.MemoryLimitPercentage == 0 {
 		return errLimitOutOfRange
 	}

internal/memorylimiter/config_test.go

@@ -84,6 +84,17 @@ func TestConfigValidate(t *testing.T) {
 			},
 			err: errSpikeLimitPercentageOutOfRange,
 		},
+		{
+			name: "invalid gc intervals",
+			cfg: &Config{
+				CheckInterval:                100 * time.Millisecond,
+				MinGCIntervalWhenSoftLimited: 50 * time.Millisecond,
+				MinGCIntervalWhenHardLimited: 100 * time.Millisecond,
+				MemoryLimitMiB:               5722,
+				MemorySpikeLimitMiB:          1907,
+			},
+			err: errInconsistentGCMinInterval,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

internal/memorylimiter/go.mod

@@ -25,16 +25,18 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
+	github.com/rogpeppe/go-internal v1.13.1 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/collector/pdata v1.26.0 // indirect
 	go.opentelemetry.io/otel v1.34.0 // indirect
 	go.opentelemetry.io/otel/metric v1.34.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.34.0 // indirect
 	go.opentelemetry.io/otel/trace v1.34.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.33.0 // indirect
-	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/sys v0.29.0 // indirect
 	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a // indirect
 	google.golang.org/grpc v1.70.0 // indirect

internal/memorylimiter/go.sum

@@ -20,10 +20,6 @@ import (
 
 const (
 	mibBytes = 1024 * 1024
-
-	// Minimum interval between forced GC when in soft limited mode. We don't want to
-	// do GCs too frequently since it is a CPU-heavy operation.
-	minGCIntervalWhenSoftLimited = 10 * time.Second
 )
 
 var (
@@ -50,11 +46,14 @@ type MemoryLimiter struct {
 
 	ticker *time.Ticker
 
-	lastGCDone time.Time
+	minGCIntervalWhenSoftLimited time.Duration
+	minGCIntervalWhenHardLimited time.Duration
+	lastGCDone                   time.Time
 
-	// The function to read the mem values is set as a reference to help with
+	// The functions to read the mem values and run GC are set as a reference to help with
 	// testing different values.
 	readMemStatsFn func(m *runtime.MemStats)
+	runGCFn        func()
 
 	// Fields used for logging.
 	logger *zap.Logger
@@ -78,18 +77,20 @@ func NewMemoryLimiter(cfg *Config, logger *zap.Logger) (*MemoryLimiter, error) {
 		zap.Duration("check_interval", cfg.CheckInterval))
 
 	return &MemoryLimiter{
-		usageChecker:   *usageChecker,
-		memCheckWait:   cfg.CheckInterval,
-		ticker:         time.NewTicker(cfg.CheckInterval),
-		readMemStatsFn: ReadMemStatsFn,
-		logger:         logger,
-		mustRefuse:     &atomic.Bool{},
+		usageChecker:                 *usageChecker,
+		memCheckWait:                 cfg.CheckInterval,
+		ticker:                       time.NewTicker(cfg.CheckInterval),
+		minGCIntervalWhenSoftLimited: cfg.MinGCIntervalWhenSoftLimited,
+		minGCIntervalWhenHardLimited: cfg.MinGCIntervalWhenHardLimited,
+		lastGCDone:                   time.Now(),
+		readMemStatsFn:               ReadMemStatsFn,
+		runGCFn:                      runtime.GC,
+		logger:                       logger,
+		mustRefuse:                   &atomic.Bool{},
 	}, nil
 }
 
-// startMonitoring starts a single ticker'd goroutine per instance
-// that will check memory usage every checkInterval period.
-func (ml *MemoryLimiter) startMonitoring() {
+func (ml *MemoryLimiter) Start(_ context.Context, _ component.Host) error {
 	ml.refCounterLock.Lock()
 	defer ml.refCounterLock.Unlock()
 
@@ -110,10 +111,6 @@ func (ml *MemoryLimiter) startMonitoring() {
 			}
 		}()
 	}
-}
-
-func (ml *MemoryLimiter) Start(_ context.Context, _ component.Host) error {
-	ml.startMonitoring()
 	return nil
 }
 
@@ -167,7 +164,7 @@ func memstatToZapField(ms *runtime.MemStats) zap.Field {
 }
 
 func (ml *MemoryLimiter) doGCandReadMemStats() *runtime.MemStats {
-	runtime.GC()
+	ml.runGCFn()
 	ml.lastGCDone = time.Now()
 	ms := ml.readMemStats()
 	ml.logger.Info("Memory usage after GC.", memstatToZapField(ms))
@@ -180,38 +177,42 @@ func (ml *MemoryLimiter) CheckMemLimits() {
 
 	ml.logger.Debug("Currently used memory.", memstatToZapField(ms))
 
-	if ml.usageChecker.aboveHardLimit(ms) {
-		ml.logger.Warn("Memory usage is above hard limit. Forcing a GC.", memstatToZapField(ms))
-		ms = ml.doGCandReadMemStats()
-	}
-
-	// Remember current state.
-	wasRefusing := ml.mustRefuse.Load()
-
-	// Check if the memory usage is above the soft limit.
-	mustRefuse := ml.usageChecker.aboveSoftLimit(ms)
-
-	if wasRefusing && !mustRefuse {
-		// Was previously refusing but enough memory is available now, no need to limit.
-		ml.logger.Info("Memory usage back within limits. Resuming normal operation.", memstatToZapField(ms))
+	// Check if we are below the soft limit.
+	aboveSoftLimit := ml.usageChecker.aboveSoftLimit(ms)
+	if !aboveSoftLimit {
+		if ml.mustRefuse.Load() {
+			// Was previously refusing but enough memory is available now, no need to limit.
+			ml.logger.Info("Memory usage back within limits. Resuming normal operation.", memstatToZapField(ms))
+		}
+		ml.mustRefuse.Store(aboveSoftLimit)
+		return
 	}
 
-	if !wasRefusing && mustRefuse {
+	if ml.usageChecker.aboveHardLimit(ms) {
+		// We are above hard limit, do a GC if it wasn't done recently and see if
+		// it brings memory usage below the soft limit.
+		if time.Since(ml.lastGCDone) > ml.minGCIntervalWhenHardLimited {
+			ml.logger.Warn("Memory usage is above hard limit. Forcing a GC.", memstatToZapField(ms))
+			ms = ml.doGCandReadMemStats()
+			// Check the limit again to see if GC helped.
+			aboveSoftLimit = ml.usageChecker.aboveSoftLimit(ms)
+		}
+	} else {
 		// We are above soft limit, do a GC if it wasn't done recently and see if
 		// it brings memory usage below the soft limit.
-		if time.Since(ml.lastGCDone) > minGCIntervalWhenSoftLimited {
+		if time.Since(ml.lastGCDone) > ml.minGCIntervalWhenSoftLimited {
 			ml.logger.Info("Memory usage is above soft limit. Forcing a GC.", memstatToZapField(ms))
 			ms = ml.doGCandReadMemStats()
 			// Check the limit again to see if GC helped.
-			mustRefuse = ml.usageChecker.aboveSoftLimit(ms)
+			aboveSoftLimit = ml.usageChecker.aboveSoftLimit(ms)
 		}
+	}
 
-		if mustRefuse {
-			ml.logger.Warn("Memory usage is above soft limit. Refusing data.", memstatToZapField(ms))
-		}
+	if !ml.mustRefuse.Load() && aboveSoftLimit {
+		ml.logger.Warn("Memory usage is above soft limit. Refusing data.", memstatToZapField(ms))
 	}
 
-	ml.mustRefuse.Store(mustRefuse)
+	ml.mustRefuse.Store(aboveSoftLimit)
 }
 
 type memUsageChecker struct {