Merge pull request #2209 from sthaha/fix-terminated-min-power

feat: allow setting energy threshold for terminated workload tracking

Author: vprashar2929

cmd/kepler/main.go

@@ -154,6 +154,7 @@ func createServices(logger *slog.Logger, cfg *config.Config) ([]service.Service,
 		monitor.WithInterval(cfg.Monitor.Interval),
 		monitor.WithMaxStaleness(cfg.Monitor.Staleness),
 		monitor.WithMaxTerminated(cfg.Monitor.MaxTerminated),
+		monitor.WithMinTerminatedEnergyThreshold(monitor.Energy(cfg.Monitor.MinTerminatedEnergyThreshold)*monitor.Joule),
 	)
 
 	apiServer := server.NewAPIServer(

compose/dev/kepler-dev/etc/kepler/config.yaml

@@ -24,6 +24,10 @@ monitor:
   # to be kept in memory until the data is exported; 0 disables the limit
   maxTerminated: 500
 
+  # minimum energy threshold (in joules) for terminated workloads
+  # terminated workloads with energy consumption below this threshold will be filtered out
+  minTerminatedEnergyThreshold: 10
+
 host:
   sysfs: /host/sys # Path to sysfs filesystem (default: /sys)
   procfs: /host/proc # Path to procfs filesystem (default: /proc)

config/config.go

@@ -54,6 +54,12 @@ type (
 		// =0: Disable terminated workload tracking completely
 		// >0: Track top N terminated workloads by energy consumption
 		MaxTerminated int `yaml:"maxTerminated"`
+
+		// MinTerminatedEnergyThreshold sets the minimum energy consumption threshold for terminated workloads
+		// Only terminated workloads with energy consumption above this threshold will be tracked
+		// Value is in joules (e.g., 10 = 10 joules)
+		// TODO: Add support for parsing energy units like "10J", "500mJ", "2kJ"
+		MinTerminatedEnergyThreshold int64 `yaml:"minTerminatedEnergyThreshold"`
 	}
 
 	// Exporter configuration
@@ -201,7 +207,8 @@ func DefaultConfig() *Config {
 			Interval:  5 * time.Second,
 			Staleness: 500 * time.Millisecond,
 
-			MaxTerminated: 500,
+			MaxTerminated:                500,
+			MinTerminatedEnergyThreshold: 10, // 10 Joules
 		},
 		Exporter: Exporter{
 			Stdout: StdoutExporter{
@@ -476,8 +483,9 @@ func (c *Config) Validate(skips ...SkipValidation) error {
 		if c.Monitor.Staleness < 0 {
 			errs = append(errs, fmt.Sprintf("invalid monitor staleness: %s can't be negative", c.Monitor.Staleness))
 		}
-		if c.Monitor.MaxTerminated < 0 {
-			errs = append(errs, fmt.Sprintf("invalid monitor max terminated: %d can't be negative", c.Monitor.MaxTerminated))
+
+		if c.Monitor.MinTerminatedEnergyThreshold < 0 {
+			errs = append(errs, fmt.Sprintf("invalid monitor min terminated energy threshold: %d can't be negative", c.Monitor.MinTerminatedEnergyThreshold))
 		}
 	}
 	{ // Kubernetes

config/config_test.go

@@ -633,14 +633,29 @@ func TestMonitorConfig(t *testing.T) {
 		assert.NoError(t, cfg.Validate())
 
 		cfg.Monitor.MaxTerminated = -10
-		assert.ErrorContains(t, cfg.Validate(), "invalid configuration: invalid monitor max terminated")
+		assert.NoError(t, cfg.Validate(), "invalid configuration: invalid monitor max terminated")
 
 		cfg.Monitor.MaxTerminated = 0
 		assert.NoError(t, cfg.Validate(), "maxTerminated=0 should be valid (unlimited)")
 
 		cfg.Monitor.MaxTerminated = 1000
 		assert.NoError(t, cfg.Validate())
 	})
+
+	t.Run("minTerminatedEnergyThreshold", func(t *testing.T) {
+		cfg := DefaultConfig()
+		assert.Equal(t, int64(10), cfg.Monitor.MinTerminatedEnergyThreshold, "default minTerminatedEnergyThreshold should be 10")
+		assert.NoError(t, cfg.Validate())
+
+		cfg.Monitor.MinTerminatedEnergyThreshold = -10
+		assert.ErrorContains(t, cfg.Validate(), "invalid configuration: invalid monitor min terminated energy threshold")
+
+		cfg.Monitor.MinTerminatedEnergyThreshold = 0
+		assert.NoError(t, cfg.Validate(), "minTerminatedEnergyThreshold=0 should be valid (no filtering)")
+
+		cfg.Monitor.MinTerminatedEnergyThreshold = 1000
+		assert.NoError(t, cfg.Validate())
+	})
 }
 
 func TestMonitorConfigFlags(t *testing.T) {
@@ -676,9 +691,9 @@ func TestMonitorConfigFlags(t *testing.T) {
 		args:     []string{"--monitor.max-terminated=0"},
 		expected: expect{interval: 5 * time.Second, staleness: 500 * time.Millisecond, maxTerminated: 0, parseError: nil},
 	}, {
-		name:     "invalid-max-terminated",
+		name:     "negative-max-terminated",
 		args:     []string{"--monitor.max-terminated=-10"},
-		expected: expect{cfgErr: fmt.Errorf("invalid configuration: invalid monitor max terminated")},
+		expected: expect{interval: 5 * time.Second, staleness: 500 * time.Millisecond, maxTerminated: -10, parseError: nil},
 	}}
 
 	for _, tc := range tt {
@@ -731,15 +746,50 @@ monitor:
 		assert.Equal(t, 0, cfg.Monitor.MaxTerminated)
 	})
 
-	t.Run("yaml-config-maxTerminated-invalid", func(t *testing.T) {
+	t.Run("yaml-config-maxTerminated-negative", func(t *testing.T) {
 		yamlData := `
 monitor:
   maxTerminated: -100
+`
+		reader := strings.NewReader(yamlData)
+		cfg, err := Load(reader)
+		assert.NoError(t, err)
+		assert.Equal(t, -100, cfg.Monitor.MaxTerminated)
+	})
+}
+
+func TestMonitorMinTerminatedEnergyThresholdYAML(t *testing.T) {
+	t.Run("yaml-config-minTerminatedEnergyThreshold", func(t *testing.T) {
+		yamlData := `
+monitor:
+  minTerminatedEnergyThreshold: 50
+`
+		reader := strings.NewReader(yamlData)
+		cfg, err := Load(reader)
+		assert.NoError(t, err)
+		assert.Equal(t, int64(50), cfg.Monitor.MinTerminatedEnergyThreshold)
+	})
+
+	t.Run("yaml-config-minTerminatedEnergyThreshold-zero", func(t *testing.T) {
+		yamlData := `
+monitor:
+  minTerminatedEnergyThreshold: 0
+`
+		reader := strings.NewReader(yamlData)
+		cfg, err := Load(reader)
+		assert.NoError(t, err)
+		assert.Equal(t, int64(0), cfg.Monitor.MinTerminatedEnergyThreshold)
+	})
+
+	t.Run("yaml-config-minTerminatedEnergyThreshold-invalid", func(t *testing.T) {
+		yamlData := `
+monitor:
+  minTerminatedEnergyThreshold: -100
 `
 		reader := strings.NewReader(yamlData)
 		_, err := Load(reader)
 		assert.Error(t, err)
-		assert.Contains(t, err.Error(), "invalid monitor max terminated")
+		assert.Contains(t, err.Error(), "invalid monitor min terminated energy threshold")
 	})
 }
 

docs/configuration/configuration.md

@@ -83,6 +83,7 @@ monitor:
   interval: 5s        # Monitor refresh interval (default: 5s)
   staleness: 1000ms   # Duration after which data is considered stale (default: 1000ms)
   maxTerminated: 500  # Maximum number of terminated workloads to keep in memory (default: 500)
+  minTerminatedEnergyThreshold: 10  # Minimum energy threshold for terminated workloads (default: 10)
 
 host:
   sysfs: /sys   # Path to sysfs filesystem (default: /sys)
@@ -154,6 +155,7 @@ monitor:
   interval: 5s
   staleness: 1000ms
   maxTerminated: 500
+  minTerminatedEnergyThreshold: 10
 ```
 
 - **interval**: The monitor's refresh interval. All processes with a lifetime less than this interval will be ignored. Setting to 0s disables monitor refreshes.
@@ -162,6 +164,8 @@ monitor:
 
 - **maxTerminated**: Maximum number of terminated workloads (processes, containers, VMs, pods) to keep in memory until the data is exported. This prevents unbounded memory growth in high-churn environments. Set to 0 for unlimited (no limit). When the limit is reached, the least power consuming terminated workloads are removed first.
 
+- **minTerminatedEnergyThreshold**: Minimum energy consumption threshold (in joules) for terminated workloads to be tracked. Only terminated workloads with energy consumption above this threshold will be included in the tracking. This helps filter out short-lived processes that consume minimal energy. Default is 10 joules.
+
 ### 🗄️ Host Configuration
 
 ```yaml

hack/config.yaml

@@ -24,6 +24,10 @@ monitor:
   # to be kept in memory until the data is exported; 0 disables the limit
   maxTerminated: 500
 
+  # minimum energy threshold (in joules) for terminated workloads
+  # terminated workloads with energy consumption below this threshold will be filtered out
+  minTerminatedEnergyThreshold: 10
+
 host:
   sysfs: /sys # Path to sysfs filesystem (default: /sys)
   procfs: /proc # Path to procfs filesystem (default: /proc)

internal/monitor/container.go

@@ -85,17 +85,9 @@ func (pm *PowerMonitor) calculateContainerPower(prev, newSnapshot *Snapshot) err
 			continue
 		}
 
-		// Only include terminated containers that have consumed energy
-		if prevContainer.Zones.HasZeroEnergy() {
-			pm.logger.Debug("Filtering out terminated container with zero energy", "id", id)
-			continue
-		}
-		pm.logger.Debug("Including terminated container with non-zero energy", "id", id)
-
 		// Add to internal tracker (which will handle priority-based retention)
 		// NOTE: Each terminated container is only added once since a container cannot be terminated twice
-		terminatedContainer := prevContainer.Clone()
-		pm.terminatedContainersTracker.Add(terminatedContainer)
+		pm.terminatedContainersTracker.Add(prevContainer.Clone())
 	}
 
 	// process running containers

internal/monitor/container_power_test.go

@@ -655,11 +655,12 @@ func TestTerminatedContainerTracking(t *testing.T) {
 		resInformer := &MockResourceInformer{}
 
 		monitor := &PowerMonitor{
-			logger:        logger,
-			cpu:           mockMeter,
-			clock:         fakeClock,
-			resources:     resInformer,
-			maxTerminated: 500,
+			logger:                       logger,
+			cpu:                          mockMeter,
+			clock:                        fakeClock,
+			resources:                    resInformer,
+			maxTerminated:                500,
+			minTerminatedEnergyThreshold: 1 * Joule, // Set threshold to filter zero-energy containers
 		}
 
 		err := monitor.Init()

internal/monitor/monitor.go

@@ -40,11 +40,17 @@ type PowerMonitor struct {
 	logger *slog.Logger
 	cpu    device.CPUPowerMeter
 
-	interval      time.Duration
-	clock         clock.WithTicker
-	maxStaleness  time.Duration
-	maxTerminated int
-	resources     resource.Informer
+	interval time.Duration
+	clock    clock.WithTicker
+
+	// related to snapshots
+	maxStaleness time.Duration
+
+	// related to terminated resource tracking
+	maxTerminated                int
+	minTerminatedEnergyThreshold Energy
+
+	resources resource.Informer
 
 	// signals when a snapshot has been updated
 	dataCh chan struct{}
@@ -86,14 +92,18 @@ func NewPowerMonitor(meter device.CPUPowerMeter, applyOpts ...OptionFn) *PowerMo
 	ctx, cancel := context.WithCancel(context.Background())
 
 	monitor := &PowerMonitor{
-		logger:           opts.logger.With("service", "monitor"),
-		cpu:              meter,
-		clock:            opts.clock,
-		interval:         opts.interval,
-		resources:        opts.resources,
-		dataCh:           make(chan struct{}, 1),
-		maxStaleness:     opts.maxStaleness,
-		maxTerminated:    opts.maxTerminated,
+		logger:    opts.logger.With("service", "monitor"),
+		cpu:       meter,
+		clock:     opts.clock,
+		interval:  opts.interval,
+		resources: opts.resources,
+		dataCh:    make(chan struct{}, 1),
+
+		maxStaleness: opts.maxStaleness,
+
+		maxTerminated:                opts.maxTerminated,
+		minTerminatedEnergyThreshold: opts.minTerminatedEnergyThreshold,
+
 		collectionCtx:    ctx,
 		collectionCancel: cancel,
 	}
@@ -119,11 +129,19 @@ func (pm *PowerMonitor) Init() error {
 	pm.logger.Info("Using primary energy zone for terminated workload tracking",
 		"zone", primaryEnergyZone.Name())
 
-	// Initialize terminated workload trackers with the primary energy zone
-	pm.terminatedProcessesTracker = NewTerminatedResourceTracker[*Process](primaryEnergyZone, pm.maxTerminated, pm.logger)
-	pm.terminatedContainersTracker = NewTerminatedResourceTracker[*Container](primaryEnergyZone, pm.maxTerminated, pm.logger)
-	pm.terminatedVMsTracker = NewTerminatedResourceTracker[*VirtualMachine](primaryEnergyZone, pm.maxTerminated, pm.logger)
-	pm.terminatedPodsTracker = NewTerminatedResourceTracker[*Pod](primaryEnergyZone, pm.maxTerminated, pm.logger)
+	// Initialize terminated workload trackers with the primary energy zone and minimum energy threshold
+	pm.terminatedProcessesTracker = NewTerminatedResourceTracker[*Process](
+		primaryEnergyZone, pm.maxTerminated,
+		pm.minTerminatedEnergyThreshold, pm.logger)
+	pm.terminatedContainersTracker = NewTerminatedResourceTracker[*Container](
+		primaryEnergyZone, pm.maxTerminated,
+		pm.minTerminatedEnergyThreshold, pm.logger)
+	pm.terminatedVMsTracker = NewTerminatedResourceTracker[*VirtualMachine](
+		primaryEnergyZone, pm.maxTerminated,
+		pm.minTerminatedEnergyThreshold, pm.logger)
+	pm.terminatedPodsTracker = NewTerminatedResourceTracker[*Pod](
+		primaryEnergyZone, pm.maxTerminated,
+		pm.minTerminatedEnergyThreshold, pm.logger)
 
 	// signal now so that exporters can construct descriptors
 	pm.signalNewData()

internal/monitor/options.go

@@ -12,23 +12,25 @@ import (
 )
 
 type Opts struct {
-	logger        *slog.Logger
-	interval      time.Duration
-	clock         clock.WithTicker
-	resources     resource.Informer
-	maxStaleness  time.Duration
-	maxTerminated int
+	logger                       *slog.Logger
+	interval                     time.Duration
+	clock                        clock.WithTicker
+	resources                    resource.Informer
+	maxStaleness                 time.Duration
+	maxTerminated                int
+	minTerminatedEnergyThreshold Energy
 }
 
 // NewConfig returns a new Config with defaults set
 func DefaultOpts() Opts {
 	return Opts{
-		logger:        slog.Default(),
-		interval:      5 * time.Second,
-		clock:         clock.RealClock{},
-		maxStaleness:  500 * time.Millisecond,
-		resources:     nil,
-		maxTerminated: 500,
+		logger:                       slog.Default(),
+		interval:                     5 * time.Second,
+		clock:                        clock.RealClock{},
+		maxStaleness:                 500 * time.Millisecond,
+		resources:                    nil,
+		maxTerminated:                500,
+		minTerminatedEnergyThreshold: 10 * Joule,
 	}
 }
 
@@ -76,3 +78,10 @@ func WithMaxTerminated(max int) OptionFn {
 		o.maxTerminated = max
 	}
 }
+
+// WithMinTerminatedEnergyThreshold sets the minimum energy threshold for terminated workloads
+func WithMinTerminatedEnergyThreshold(threshold Energy) OptionFn {
+	return func(o *Opts) {
+		o.minTerminatedEnergyThreshold = threshold
+	}
+}