Merge pull request #2202 from sthaha/feat-max-terminated-process

feat: implement energy-based terminated workload tracking

Author: Sunil Thaha

config/config.go

@@ -49,6 +49,10 @@ type (
 		Interval  time.Duration `yaml:"interval"`  // Interval for monitoring resources
 		Staleness time.Duration `yaml:"staleness"` // Time after which calculated values are considered stale
 
+		// MaxTerminated controls terminated workload tracking behavior:
+		// <0: Any negative value indicates to track unlimited terminated workloads (no capacity limit)
+		// =0: Disable terminated workload tracking completely
+		// >0: Track top N terminated workloads by energy consumption
 		MaxTerminated int `yaml:"maxTerminated"`
 	}
 
@@ -297,8 +301,8 @@ func RegisterFlags(app *kingpin.Application) ConfigUpdaterFn {
 	// monitor
 	monitorInterval := app.Flag(MonitorIntervalFlag,
 		"Interval for monitoring resources (processes, container, vm, etc...); 0 to disable").Default("5s").Duration()
-	maxTerminated := app.Flag(MonitorMaxTerminatedFlag,
-		"Maximum number of terminated workloads to keep in memory until exported; 0 for unlimited").Default("500").Int()
+	monitorMaxTerminated := app.Flag(MonitorMaxTerminatedFlag,
+		"Maximum number of terminated workloads to track; 0 to disable, -1 for unlimited").Default("500").Int()
 
 	enablePprof := app.Flag(pprofEnabledFlag, "Enable pprof debug endpoints").Default("false").Bool()
 	webConfig := app.Flag(WebConfigFlag, "Web config file path").Default("").String()
@@ -338,9 +342,8 @@ func RegisterFlags(app *kingpin.Application) ConfigUpdaterFn {
 		if flagsSet[MonitorIntervalFlag] {
 			cfg.Monitor.Interval = *monitorInterval
 		}
-
 		if flagsSet[MonitorMaxTerminatedFlag] {
-			cfg.Monitor.MaxTerminated = *maxTerminated
+			cfg.Monitor.MaxTerminated = *monitorMaxTerminated
 		}
 
 		if flagsSet[pprofEnabledFlag] {

internal/device/cpu_power_meter.go

@@ -32,4 +32,9 @@ type CPUPowerMeter interface {
 
 	// Zones() returns a slice of the energy measurement zones
 	Zones() ([]EnergyZone, error)
+
+	// PrimaryEnergyZone() returns the zone with the highest energy coverage/priority
+	// This zone represents the most comprehensive energy measurement available
+	// E.g. Psys > Package > Core > DRAM > Uncore
+	PrimaryEnergyZone() (EnergyZone, error)
 }

internal/device/fake_cpu_power_meter.go

@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"math/rand"
 	"path/filepath"
+	"strings"
 	"sync"
 )
 
@@ -151,3 +152,25 @@ func (m *fakeRaplMeter) Name() string {
 func (m *fakeRaplMeter) Zones() ([]EnergyZone, error) {
 	return m.zones, nil
 }
+
+// PrimaryEnergyZone returns the zone with the highest energy coverage/priority
+func (m *fakeRaplMeter) PrimaryEnergyZone() (EnergyZone, error) {
+	zones, err := m.Zones()
+	if err != nil {
+		return nil, err
+	}
+
+	if len(zones) == 0 {
+		return nil, fmt.Errorf("no zones available in fake meter")
+	}
+
+	// For fake meter, prefer package if available, otherwise first zone
+	for _, zone := range zones {
+		if strings.Contains(strings.ToLower(zone.Name()), "package") {
+			return zone, nil
+		}
+	}
+
+	// Fallback to first zone
+	return zones[0], nil
+}

internal/device/rapl_sysfs_power_meter.go

@@ -17,6 +17,7 @@ type raplPowerMeter struct {
 	cachedZones []EnergyZone
 	logger      *slog.Logger
 	zoneFilter  []string
+	topZone     EnergyZone
 }
 
 type OptionFn func(*raplPowerMeter)
@@ -192,6 +193,43 @@ func (r *raplPowerMeter) zoneNames(zones []EnergyZone) []string {
 	return names
 }
 
+// PrimaryEnergyZone returns the zone with the highest energy coverage/priority
+func (r *raplPowerMeter) PrimaryEnergyZone() (EnergyZone, error) {
+	// Return cached zone if already initialized
+	if r.topZone != nil {
+		return r.topZone, nil
+	}
+
+	zones, err := r.Zones()
+	if err != nil {
+		return nil, err
+	}
+
+	if len(zones) == 0 {
+		return nil, fmt.Errorf("no energy zones available")
+	}
+
+	zoneMap := map[string]EnergyZone{}
+	for _, zone := range zones {
+		zoneMap[strings.ToLower(zone.Name())] = zone
+	}
+
+	// Priority hierarchy for RAPL zones (highest to lowest priority)
+	priorityOrder := []string{"psys", "package", "core", "dram", "uncore"}
+
+	// Find highest priority zone available
+	for _, p := range priorityOrder {
+		if zone, exists := zoneMap[p]; exists {
+			r.topZone = zone
+			return zone, nil
+		}
+	}
+
+	// Fallback to first zone if none match our preferences
+	r.topZone = zones[0]
+	return zones[0], nil
+}
+
 // isStandardRaplPath checks if a RAPL zone path is in the standard format
 func isStandardRaplPath(path string) bool {
 	return strings.Contains(path, "/intel-rapl:")

internal/device/rapl_sysfs_power_meter_test.go

@@ -445,3 +445,143 @@ func TestCPUPowerMeter_InitNoZones(t *testing.T) {
 	assert.Equal(t, "no RAPL zones found", err.Error(), "Start() should return a specific error message")
 	mockReader.AssertExpectations(t)
 }
+
+// TestPrimaryEnergyZone tests the PrimaryEnergyZone method
+func TestPrimaryEnergyZone(t *testing.T) {
+	t.Run("Priority hierarchy", func(t *testing.T) {
+		tests := []struct {
+			name     string
+			zones    []EnergyZone
+			expected string
+		}{{
+			name: "psys has highest priority",
+			zones: []EnergyZone{
+				mockZone{name: "package", index: 0},
+				mockZone{name: "psys", index: 0},
+				mockZone{name: "core", index: 0},
+			},
+			expected: "psys",
+		}, {
+			name: "package has priority over core",
+			zones: []EnergyZone{
+				mockZone{name: "core", index: 0},
+				mockZone{name: "package", index: 0},
+				mockZone{name: "dram", index: 0},
+			},
+			expected: "package",
+		}, {
+			name: "core has priority over dram",
+			zones: []EnergyZone{
+				mockZone{name: "dram", index: 0},
+				mockZone{name: "core", index: 0},
+				mockZone{name: "uncore", index: 0},
+			},
+			expected: "core",
+		}, {
+			name: "dram has priority over uncore",
+			zones: []EnergyZone{
+				mockZone{name: "uncore", index: 0},
+				mockZone{name: "dram", index: 0},
+			},
+			expected: "dram",
+		}}
+
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				mockReader := &mockRaplReader{}
+				mockReader.On("Zones").Return(tt.zones, nil)
+
+				meter := &raplPowerMeter{reader: mockReader, logger: slog.Default()}
+				zone, err := meter.PrimaryEnergyZone()
+
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expected, zone.Name())
+				mockReader.AssertExpectations(t)
+			})
+		}
+	})
+
+	t.Run("Case insensitive matching", func(t *testing.T) {
+		mockReader := &mockRaplReader{}
+		mockReader.On("Zones").Return([]EnergyZone{
+			mockZone{name: "PACKAGE", index: 0},
+			mockZone{name: "Core", index: 0},
+		}, nil)
+
+		meter := &raplPowerMeter{reader: mockReader, logger: slog.Default()}
+		zone, err := meter.PrimaryEnergyZone()
+
+		assert.NoError(t, err)
+		assert.Equal(t, "PACKAGE", zone.Name())
+		mockReader.AssertExpectations(t)
+	})
+
+	t.Run("Fallback to first zone", func(t *testing.T) {
+		zones := []EnergyZone{
+			mockZone{name: "unknown1", index: 0},
+			mockZone{name: "unknown2", index: 1},
+		}
+		mockReader := &mockRaplReader{}
+		mockReader.On("Zones").Return(zones, nil)
+
+		meter := &raplPowerMeter{reader: mockReader, logger: slog.Default()}
+		zone, err := meter.PrimaryEnergyZone()
+
+		assert.NoError(t, err)
+		// NOTE: since reader.Zones() does not guarantee the order after filtering,
+		// we cannot assert zone.Name() == "unknown1", thus assert the zone returned
+		// any of the zones passed as input
+		zoneName := zone.Name()
+		assert.Contains(t, []string{"unknown1", "unknown2"}, zoneName)
+		mockReader.AssertExpectations(t)
+	})
+
+	t.Run("Caching behavior", func(t *testing.T) {
+		mockReader := &mockRaplReader{}
+		mockReader.On("Zones").Return([]EnergyZone{
+			mockZone{name: "package", index: 0},
+		}, nil).Once()
+
+		meter := &raplPowerMeter{reader: mockReader, logger: slog.Default()}
+
+		// First call should read from zones and cache topZone
+		zone1, err := meter.PrimaryEnergyZone()
+		assert.NoError(t, err)
+		assert.Equal(t, "package", zone1.Name())
+
+		// Second call should use cached topZone directly
+		zone2, err := meter.PrimaryEnergyZone()
+		assert.NoError(t, err)
+		assert.Equal(t, "package", zone2.Name())
+
+		mockReader.AssertExpectations(t)
+	})
+
+	t.Run("Error handling", func(t *testing.T) {
+		t.Run("Zones() returns error", func(t *testing.T) {
+			mockReader := &mockRaplReader{}
+			mockReader.On("Zones").Return([]EnergyZone{}, errors.New("zones error"))
+
+			meter := &raplPowerMeter{reader: mockReader, logger: slog.Default()}
+			zone, err := meter.PrimaryEnergyZone()
+
+			assert.Error(t, err)
+			assert.Nil(t, zone)
+			assert.Contains(t, err.Error(), "zones error")
+			mockReader.AssertExpectations(t)
+		})
+
+		t.Run("Empty zones list", func(t *testing.T) {
+			mockReader := &mockRaplReader{}
+			mockReader.On("Zones").Return([]EnergyZone{}, nil)
+
+			meter := &raplPowerMeter{reader: mockReader, logger: slog.Default()}
+			zone, err := meter.PrimaryEnergyZone()
+
+			assert.Error(t, err)
+			assert.Nil(t, zone)
+			assert.Contains(t, err.Error(), "no RAPL zones found")
+			mockReader.AssertExpectations(t)
+		})
+	})
+}

internal/exporter/prometheus/collector/power_collector.go

@@ -311,13 +311,12 @@ func (c *PowerCollector) collectProcessMetrics(ch chan<- prometheus.Metric, stat
 
 	// No need to lock, already done by the calling function
 	for pid, proc := range processes {
-		pidStr := fmt.Sprintf("%d", pid)
 
 		ch <- prometheus.MustNewConstMetric(
 			c.processCPUTimeDescriptor,
 			prometheus.CounterValue,
 			proc.CPUTotalTime,
-			pidStr, proc.Comm, proc.Exe, string(proc.Type),
+			pid, proc.Comm, proc.Exe, string(proc.Type),
 			proc.ContainerID, proc.VirtualMachineID,
 		)
 
@@ -327,7 +326,7 @@ func (c *PowerCollector) collectProcessMetrics(ch chan<- prometheus.Metric, stat
 				c.processCPUJoulesDescriptor,
 				prometheus.CounterValue,
 				usage.EnergyTotal.Joules(),
-				pidStr, proc.Comm, proc.Exe, string(proc.Type), state,
+				pid, proc.Comm, proc.Exe, string(proc.Type), state,
 				proc.ContainerID, proc.VirtualMachineID,
 				zoneName,
 			)
@@ -336,7 +335,7 @@ func (c *PowerCollector) collectProcessMetrics(ch chan<- prometheus.Metric, stat
 				c.processCPUWattsDescriptor,
 				prometheus.GaugeValue,
 				usage.Power.Watts(),
-				pidStr, proc.Comm, proc.Exe, string(proc.Type), state,
+				pid, proc.Comm, proc.Exe, string(proc.Type), state,
 				proc.ContainerID, proc.VirtualMachineID,
 				zoneName,
 			)

internal/exporter/prometheus/collector/power_collector_test.go

@@ -215,7 +215,7 @@ func TestPowerCollector(t *testing.T) {
 	}
 
 	testProcesses := monitor.Processes{
-		123: {
+		"123": {
 			PID:          123,
 			Comm:         "test-process",
 			Exe:          "/usr/bin/123",
@@ -527,7 +527,7 @@ func TestTerminatedProcessExport(t *testing.T) {
 			},
 		},
 		Processes: monitor.Processes{
-			123: &monitor.Process{
+			"123": &monitor.Process{
 				PID:          123,
 				Comm:         "running-proc",
 				Exe:          "/usr/bin/running-proc",
@@ -544,7 +544,7 @@ func TestTerminatedProcessExport(t *testing.T) {
 			},
 		},
 		TerminatedProcesses: monitor.Processes{
-			456: &monitor.Process{
+			"456": &monitor.Process{
 				PID:          456,
 				Comm:         "terminated-proc",
 				Exe:          "/usr/bin/terminated-proc",
@@ -633,7 +633,7 @@ func TestEnhancedErrorReporting(t *testing.T) {
 			},
 		},
 		Processes: monitor.Processes{
-			123: &monitor.Process{
+			"123": &monitor.Process{
 				PID:          123,
 				Comm:         "actual-proc",
 				Exe:          "/usr/bin/actual-proc",
@@ -650,7 +650,7 @@ func TestEnhancedErrorReporting(t *testing.T) {
 			},
 		},
 		TerminatedProcesses: monitor.Processes{
-			456: &monitor.Process{
+			"456": &monitor.Process{
 				PID:          456,
 				Comm:         "terminated-proc",
 				Exe:          "/usr/bin/terminated-proc",
@@ -771,7 +771,7 @@ func TestPowerCollector_MetricsLevelFiltering(t *testing.T) {
 					UsageRatio: 0.5,
 				},
 				Processes: monitor.Processes{
-					123: &monitor.Process{
+					"123": &monitor.Process{
 						PID:          123,
 						Comm:         "test-process",
 						Exe:          "/usr/bin/test-process",