Merge pull request #2145 from vimalk78/pod-power-tests

chore(pod): added tests for pod power

Author: Sunil Thaha

config/config.go

@@ -386,15 +386,14 @@ func (c *Config) Validate(skips ...SkipValidation) error {
 		}
 	}
 	{ // Kubernetes
-		if c.Kube.Config != "" && !ptr.Deref(c.Kube.Enabled, false) {
-			errs = append(errs, fmt.Sprintf("%s supplied but %s set to false", KubeConfigFlag, KubernetesFlag))
-		}
-		if c.Kube.Node != "" && !ptr.Deref(c.Kube.Enabled, false) {
-			errs = append(errs, fmt.Sprintf("%s supplied but %s set to false", KubeNodeNameFlag, KubernetesFlag))
-		}
-		if ptr.Deref(c.Kube.Enabled, false) && c.Kube.Config != "" {
-			if err := canReadFile(c.Kube.Config); err != nil {
-				errs = append(errs, fmt.Sprintf("unreadable kubeconfig: %s", c.Kube.Config))
+		if ptr.Deref(c.Kube.Enabled, false) {
+			if c.Kube.Config != "" {
+				if err := canReadFile(c.Kube.Config); err != nil {
+					errs = append(errs, fmt.Sprintf("unreadable kubeconfig: %s", c.Kube.Config))
+				}
+			}
+			if c.Kube.Node == "" {
+				errs = append(errs, fmt.Sprintf("%s not supplied but %s set to true", KubeNodeNameFlag, KubernetesFlag))
 			}
 		}
 	}

config/config_test.go

@@ -301,21 +301,13 @@ func TestInvalidConfigurationValues(t *testing.T) {
 		},
 		error: "unreadable kubeconfig",
 	}, {
-		name: "kube not enabled, kubeconfig supplied",
+		name: "kube enabled, nodeName not supplied",
 		config: &Config{
 			Kube: Kube{
-				Config: "/some/existing/file",
-			},
-		},
-		error: "kube.config supplied but kube.enable set to false",
-	}, {
-		name: "kube not enabled, nodeName supplied",
-		config: &Config{
-			Kube: Kube{
-				Node: "dummyNode",
+				Enabled: ptr.To(true),
 			},
 		},
-		error: "kube.node-name supplied but kube.enable set to false",
+		error: "kube.node-name not supplied but kube.enable set to true",
 	}}
 
 	// test yaml marshall

internal/exporter/prometheus/collector/power_collector.go

@@ -95,7 +95,7 @@ func timeDesc(level, device string, labels []string) *prometheus.Desc {
 // by fetching all data in a single snapshot during collection
 func NewPowerCollector(monitor PowerDataProvider, logger *slog.Logger) *PowerCollector {
 	const (
-		// these labels should rename the same across all descriptors to ease querying
+		// these labels should remain the same across all descriptors to ease querying
 		zone   = "zone"
 		cntrID = "container_id"
 		vmID   = "vm_id"
@@ -124,8 +124,8 @@ func NewPowerCollector(monitor PowerDataProvider, logger *slog.Logger) *PowerCol
 		processCPUWattsDescriptor:  wattsDesc("process", "cpu", []string{"pid", "comm", "exe", "type", cntrID, vmID, zone}),
 		processCPUTimeDescriptor:   timeDesc("process", "cpu", []string{"pid", "comm", "exe", "type", cntrID, vmID}),
 
-		containerCPUJoulesDescriptor: joulesDesc("container", "cpu", []string{cntrID, "container_name", "runtime", zone}),
-		containerCPUWattsDescriptor:  wattsDesc("container", "cpu", []string{cntrID, "container_name", "runtime", zone}),
+		containerCPUJoulesDescriptor: joulesDesc("container", "cpu", []string{cntrID, "container_name", "runtime", zone, podID}),
+		containerCPUWattsDescriptor:  wattsDesc("container", "cpu", []string{cntrID, "container_name", "runtime", zone, podID}),
 
 		vmCPUJoulesDescriptor: joulesDesc("vm", "cpu", []string{vmID, "vm_name", "hypervisor", zone}),
 		vmCPUWattsDescriptor:  wattsDesc("vm", "cpu", []string{vmID, "vm_name", "hypervisor", zone}),
@@ -329,13 +329,16 @@ func (c *PowerCollector) collectContainerMetrics(ch chan<- prometheus.Metric, co
 				usage.EnergyTotal.Joules(),
 				id, container.Name, string(container.Runtime),
 				zoneName,
+				container.PodID,
 			)
 
 			ch <- prometheus.MustNewConstMetric(
 				c.containerCPUWattsDescriptor,
 				prometheus.GaugeValue,
 				usage.Power.Watts(),
-				id, container.Name, string(container.Runtime), zoneName,
+				id, container.Name, string(container.Runtime),
+				zoneName,
+				container.PodID,
 			)
 		}
 	}

internal/k8s/pod/pod.go

@@ -10,6 +10,7 @@ import (
 	"log/slog"
 	"strings"
 
+	"github.com/sustainable-computing-io/kepler/internal/logger"
 	"github.com/sustainable-computing-io/kepler/internal/service"
 	"go.uber.org/zap/zapcore"
 	corev1 "k8s.io/api/core/v1"
@@ -131,11 +132,7 @@ func (pi *podInformer) Init() error {
 	}
 	pi.manager = mgr
 
-	opts := zap.Options{
-		Development: true, // enables DebugLevel by default
-		Level:       zapcore.DebugLevel,
-	}
-	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)).WithCallDepth(0))
+	pi.setControllerRuntimeLogLevel()
 
 	pi.logger.Info("pod informer initialized")
 
@@ -252,3 +249,27 @@ func getConfig(kubeConfigPath string) (*rest.Config, error) {
 func (i *podInformer) Name() string {
 	return "podInformer"
 }
+
+func (pi *podInformer) setControllerRuntimeLogLevel() {
+	level := logger.LogLevel()
+	opts := zap.Options{
+		Level: slogLevelToZapLevel(level),
+	}
+	if level == slog.LevelDebug {
+		opts.Development = true
+	}
+	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)).WithCallDepth(0))
+}
+
+func slogLevelToZapLevel(level slog.Level) zapcore.Level {
+	switch {
+	case level <= slog.LevelDebug:
+		return zapcore.DebugLevel
+	case level <= slog.LevelInfo:
+		return zapcore.InfoLevel
+	case level <= slog.LevelWarn:
+		return zapcore.WarnLevel
+	default:
+		return zapcore.ErrorLevel
+	}
+}

internal/k8s/pod/pod_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"go.uber.org/zap/zapcore"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -297,3 +298,22 @@ func TestPodInformer_RunIntegration(t *testing.T) {
 		}
 	})
 }
+
+func TestSlogLevelToZapLevel(t *testing.T) {
+	tests := []struct {
+		input    slog.Level
+		expected zapcore.Level
+	}{
+		{slog.LevelDebug, zapcore.DebugLevel},
+		{slog.LevelInfo, zapcore.InfoLevel},
+		{slog.LevelWarn, zapcore.WarnLevel},
+		{slog.LevelError, zapcore.ErrorLevel},
+		{slog.Level(-10), zapcore.DebugLevel},
+		{slog.Level(10), zapcore.ErrorLevel},
+	}
+
+	for _, tc := range tests {
+		result := slogLevelToZapLevel(tc.input)
+		assert.Equal(t, tc.expected, result, "Conversion failed for slog level: %v", tc.input)
+	}
+}

internal/logger/logger.go

@@ -11,11 +11,17 @@ import (
 	"strings"
 )
 
+var logLevel slog.Level
+
 func New(level, format string, w io.Writer) *slog.Logger {
-	logLevel := parseLogLevel(level)
+	logLevel = parseLogLevel(level)
 	return slog.New(handlerForFormat(format, logLevel, w))
 }
 
+func LogLevel() slog.Level {
+	return logLevel
+}
+
 func handlerForFormat(format string, logLevel slog.Level, w io.Writer) slog.Handler {
 	switch format {
 	case "json":

internal/monitor/container.go

@@ -25,6 +25,9 @@ func (pm *PowerMonitor) firstContainerRead(snapshot *Snapshot) error {
 			CPUTotalTime: ctnr.CPUTotalTime,
 			Zones:        make(ZoneUsageMap, len(zones)),
 		}
+		if ctnr.Pod != nil {
+			container.PodID = ctnr.Pod.ID
+		}
 
 		// Initialize each zone with zero values
 		for _, zone := range zones {
@@ -78,6 +81,9 @@ func (pm *PowerMonitor) calculateContainerPower(prev, newSnapshot *Snapshot) err
 			CPUTotalTime: c.CPUTotalTime,
 			Zones:        make(ZoneUsageMap),
 		}
+		if c.Pod != nil {
+			container.PodID = c.Pod.ID
+		}
 
 		// Calculate CPU time ratio for this container
 

internal/monitor/monitor.go

@@ -320,7 +320,7 @@ func (pm *PowerMonitor) firstReading(newSnapshot *Snapshot) error {
 
 	// First read for pods
 	if err := pm.firstPodRead(newSnapshot); err != nil {
-		return fmt.Errorf(containerPowerError, err)
+		return fmt.Errorf(podPowerError, err)
 	}
 
 	return nil

internal/monitor/monitor_collection_test.go

@@ -30,7 +30,6 @@ func TestCollectionLoop(t *testing.T) {
 	resourceInformer := &MockResourceInformer{}
 	resourceInformer.SetExpectations(t, tr)
 	resourceInformer.On("Refresh").Return(nil)
-	// resourceInformer.On("Pods").Return(nil)
 
 	monitor := NewPowerMonitor(
 		mockMeter,

internal/resource/informer.go

@@ -219,7 +219,7 @@ func (ri *resourceInformer) Refresh() error {
 					containersNoPod = append(containersNoPod, container.ID)
 				} else {
 					ri.logger.Debug("Failed to get pod for container", "container", container.ID, "error", err)
-					refreshErrs = errors.Join(refreshErrs, err)
+					refreshErrs = errors.Join(refreshErrs, fmt.Errorf("failed to get pod for container: %w", err))
 				}
 				continue
 			}
@@ -290,11 +290,9 @@ func (ri *resourceInformer) Refresh() error {
 	ri.vms.Terminated = vmsTerminated
 
 	// Find terminated pods
-	totalPodsDelta := float64(0)
 	podsTerminated := make(map[string]*Pod)
 	for id, pod := range ri.podCache {
 		if _, isRunning := podsRunning[id]; isRunning {
-			totalPodsDelta += pod.CPUTimeDelta
 			continue
 		}
 		podsTerminated[id] = pod