Disable mem-ballast-size-mib from command line option

Author: mxiamxia

CHANGELOG.md

@@ -3,9 +3,11 @@
 ## Unreleased
 
 ## 🛑 Breaking changes 🛑
-
 - Remove Resize() from pdata slice APIs (#3675)
+- Remove `mem-ballast-size-mib` option in command line (#3626)
+  - Use [`ballast extension`](./extension/ballastextension/README.md) to set memory ballast instead. 
 
+  
 ## v0.30.0 Beta
 
 ## 🛑 Breaking changes 🛑
@@ -39,7 +41,6 @@
 
 - `scraperhelper`: Include the scraper name in log messages (#3487)
 - `scraperhelper`: fix case when returned pdata is empty (#3520) 
-- `otlpexporter`: Allow endpoint to be configured with a scheme of `http` or `https` (#3575)
 - Record the correct number of points not metrics in Kafka receiver (#3553) 
 - Validate the Prometheus configuration (#3589) 
 

service/collector.go

@@ -35,6 +35,7 @@ import (
 	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/config/experimental/configsource"
 	"go.opentelemetry.io/collector/consumer/consumererror"
+	"go.opentelemetry.io/collector/extension/ballastextension"
 	"go.opentelemetry.io/collector/internal/collector/telemetry"
 	"go.opentelemetry.io/collector/service/internal/builder"
 	"go.opentelemetry.io/collector/service/parserprovider"
@@ -286,30 +287,37 @@ func (col *Collector) execute(ctx context.Context) error {
 	)
 	col.stateChannel <- Starting
 
-	// Set memory ballast
-	ballast, ballastSizeBytes := col.createMemoryBallast()
+	//  Add `mem-ballast-size-mib` warning message if it is still enabled
+	//  TODO: will remove all `mem-ballast-size-mib` footprints after some baking time.
+	if builder.MemBallastSize() > 0 {
+		col.logger.Warn("`mem-ballast-size-mib` command line option has been deprecated. Please use `ballast extension` instead!")
+	}
 
 	col.asyncErrorChannel = make(chan error)
 
-	// Setup everything.
-	err := col.setupTelemetry(ballastSizeBytes)
+	err := col.setupConfigurationComponents(ctx)
 	if err != nil {
 		return err
 	}
 
-	err = col.setupConfigurationComponents(ctx)
+	// Get ballastSizeBytes if ballast extension is enabled
+	ballastSizeBytes := col.getBallastSize()
+
+	// Setup Telemetry.
+	err = col.setupTelemetry(ballastSizeBytes)
 	if err != nil {
 		return err
 	}
 
+	col.service.GetExtensions()
+
 	// Everything is ready, now run until an event requiring shutdown happens.
 	col.runAndWaitForShutdownEvent()
 
 	// Accumulate errors and proceed with shutting down remaining components.
 	var errs []error
 
 	// Begin shutdown sequence.
-	runtime.KeepAlive(ballast)
 	col.logger.Info("Starting shutdown...")
 
 	if closable, ok := col.parserProvider.(parserprovider.Closeable); ok {
@@ -335,17 +343,6 @@ func (col *Collector) execute(ctx context.Context) error {
 	return consumererror.Combine(errs)
 }
 
-func (col *Collector) createMemoryBallast() ([]byte, uint64) {
-	ballastSizeMiB := builder.MemBallastSize()
-	if ballastSizeMiB > 0 {
-		ballastSizeBytes := uint64(ballastSizeMiB) * 1024 * 1024
-		ballast := make([]byte, ballastSizeBytes)
-		col.logger.Info("Using memory ballast", zap.Int("MiBs", ballastSizeMiB))
-		return ballast, ballastSizeBytes
-	}
-	return nil, 0
-}
-
 // reloadService shutdowns the current col.service and setups a new one according
 // to the latest configuration. It requires that col.parserProvider and col.factories
 // are properly populated to finish successfully.
@@ -370,3 +367,15 @@ func (col *Collector) reloadService(ctx context.Context) error {
 
 	return nil
 }
+
+func (col *Collector) getBallastSize() uint64 {
+	var ballastSize uint64
+	extensions := col.service.GetExtensions()
+	for _, extension := range extensions {
+		if ext, ok := extension.(*ballastextension.MemoryBallast); ok {
+			ballastSize = ext.GetBallastSize()
+			break
+		}
+	}
+	return ballastSize
+}

service/internal/builder/builder.go

@@ -36,6 +36,7 @@ var (
 )
 
 // Flags adds flags related to basic building of the collector server to the given flagset.
+// Deprecated: keep this flag for preventing the breaking change. Use `ballast extension` instead.
 func Flags(flags *flag.FlagSet) {
 	memBallastSize = flags.Uint(memBallastFlag, 0,
 		fmt.Sprintf("Flag to specify size of memory (MiB) ballast to set. Ballast is not used when this is not specified. "+

service/internal/telemetry/process_telemetry.go

@@ -111,8 +111,8 @@ var viewRSSMemory = &view.View{
 func NewProcessMetricsViews(ballastSizeBytes uint64) (*ProcessMetricsViews, error) {
 	pmv := &ProcessMetricsViews{
 		prevTimeUnixNano: time.Now().UnixNano(),
-		ballastSizeBytes: ballastSizeBytes,
 		views:            []*view.View{viewProcessUptime, viewAllocMem, viewTotalAllocMem, viewSysMem, viewCPUSeconds, viewRSSMemory},
+		ballastSizeBytes: ballastSizeBytes,
 		done:             make(chan struct{}),
 	}
 
@@ -176,8 +176,10 @@ func (pmv *ProcessMetricsViews) updateViews() {
 
 func (pmv *ProcessMetricsViews) readMemStats(ms *runtime.MemStats) {
 	runtime.ReadMemStats(ms)
-	ms.Alloc -= pmv.ballastSizeBytes
-	ms.HeapAlloc -= pmv.ballastSizeBytes
-	ms.HeapSys -= pmv.ballastSizeBytes
-	ms.HeapInuse -= pmv.ballastSizeBytes
+	if pmv.ballastSizeBytes > 0 {
+		ms.Alloc -= pmv.ballastSizeBytes
+		ms.HeapAlloc -= pmv.ballastSizeBytes
+		ms.HeapSys -= pmv.ballastSizeBytes
+		ms.HeapInuse -= pmv.ballastSizeBytes
+	}
 }

service/internal/telemetry/process_telemetry_test.go

@@ -25,7 +25,6 @@ import (
 
 func TestProcessTelemetry(t *testing.T) {
 	const ballastSizeBytes uint64 = 0
-
 	pmv, err := NewProcessMetricsViews(ballastSizeBytes)
 	require.NoError(t, err)
 	assert.NotNil(t, pmv)