code logic for reader.go

[SamYuan1990]

kepler/pkg/collector/reader.go

Lines 255 to 366 in b941ce5

	ticker := time.NewTicker(samplePeriod)
	go func() {
	_ = gpu.GetGpuEnergy() // reset power usage counter
	c.resetBPFTables()
	c.readEnergy()
	nodeEnergy.SetValues(sensorEnergy, pkgEnergy, 0, map[string]float64{}) // set initial energy
	acpiPowerMeter.Run()
	for {
	// wait ticker
	<-ticker.C
	lock.Lock()
	c.resetCurrValue()
	var coreNDelta, dramNDelta, uncoreNDelta, pkgNDelta, gpuNDelta []float64
	// read node-level settings (frequency)
	cpuFrequency = acpiPowerMeter.GetCPUCoreFrequency()
	// read pod metrics
	pidPodName, containerIDPodName := c.readBPFEvent()
	c.readCgroup(containerIDPodName)
	c.readKubelet()
	// convert to pod metrics to array
	var podMetricValues [][]float64
	var podNameList []string
	for podName, v := range podEnergy {
	values := v.ToEstimatorValues()
	podMetricValues = append(podMetricValues, values)
	podNameList = append(podNameList, podName)
	}
	// TO-DO: handle metrics read by GPU device in the same way as the other usage metrics
	// read gpu power
	gpuPerPid, _ := gpu.GetCurrGpuEnergyPerPid() // power not energy
	podGPUDelta := make(map[string]float64)
	for pid, podName := range pidPodName {
	gpuPower := gpuPerPid[pid]
	if _, found := podGPUDelta[podName]; !found {
	podGPUDelta[podName] = 0
	} else {
	podGPUDelta[podName] += gpuPower
	}
	}
	for _, podName := range podNameList {
	gpuNDelta = append(gpuNDelta, podGPUDelta[podName])
	}
	// read and compute power (energy delta)
	var totalCorePower, totalDRAMPower, totalUncorePower, totalPkgPower, totalGPUPower uint64
	for _, val := range gpuNDelta {
	totalGPUPower += uint64(val)
	}
	c.readEnergy()
	sumUsage := model.GetSumUsageMap(metricNames, podMetricValues)
	nodeEnergy.SetValues(sensorEnergy, pkgEnergy, totalGPUPower, sumUsage)
	for pkgIDKey, pkgStat := range nodeEnergy.EnergyInPkg.Stat {
	coreDelta, dramDelta, uncoreDelta := nodeEnergy.GetCurrEnergyPerpkgID(pkgIDKey)
	pkgDelta := pkgStat.Curr
	coreNDelta = append(coreNDelta, float64(coreDelta))
	dramNDelta = append(dramNDelta, float64(dramDelta))
	uncoreNDelta = append(uncoreNDelta, float64(uncoreDelta))
	pkgNDelta = append(pkgNDelta, float64(pkgDelta))
	totalCorePower += coreDelta
	totalDRAMPower += dramDelta
	totalUncorePower += uncoreDelta
	totalPkgPower += pkgDelta
	}
	// get power from usage ratio
	podCore, podDRAM, podUncore, podPkg := model.GetPowerFromUsageRatio(podMetricValues, totalCorePower, totalDRAMPower, totalUncorePower, totalPkgPower, sumUsage)
	// get dynamic power from usage metrics
	podDynamicPower := model.GetDynamicPower(metricNames, podMetricValues, coreNDelta, dramNDelta, uncoreNDelta, pkgNDelta, gpuNDelta)
	// get other energy - divide equally
	podOther := uint64(0)
	podCount := len(podNameList)
	if podCount > 0 {
	podOther = nodeEnergy.EnergyInOther / uint64(podCount)
	}
	// set pod energy
	for i, podName := range podNameList {
	if err := podEnergy[podName].EnergyInCore.AddNewCurr(podCore[i]); err != nil {
	klog.V(5).Infoln(err)
	}
	if err := podEnergy[podName].EnergyInDRAM.AddNewCurr(podDRAM[i]); err != nil {
	klog.V(5).Infoln(err)
	}
	if err := podEnergy[podName].EnergyInUncore.AddNewCurr(podUncore[i]); err != nil {
	klog.V(5).Infoln(err)
	}
	if err := podEnergy[podName].EnergyInPkg.AddNewCurr(podPkg[i]); err != nil {
	klog.V(5).Infoln(err)
	}
	podGPU := uint64(math.Ceil(podGPUDelta[podName]))
	if err := podEnergy[podName].EnergyInGPU.AddNewCurr(podGPU); err != nil {
	klog.V(5).Infoln(err)
	}
	if err := podEnergy[podName].EnergyInOther.AddNewCurr(podOther); err != nil {
	klog.V(5).Infoln(err)
	}
	}
	if len(podDynamicPower) != 0 {
	for i, podName := range podNameList {
	power := uint64(podDynamicPower[i])
	if err := podEnergy[podName].DynEnergy.AddNewCurr(power); err != nil {
	klog.V(5).Infoln(err)
	}
	}
	klog.V(3).Infof("Get pod powers: %v \n %v from %v (%d x %d)\n", podMetricValues[0:2], podDynamicPower, metricNames, len(podMetricValues), len(podMetricValues[0]))
	}
	for _, v := range podEnergy {
	klog.V(3).Infoln(v)
	}
	klog.V(3).Infoln(nodeEnergy)
	lock.Unlock()
	}

1. lock.lock() not follow with defer lock.unlock()
2. this read() function seems been invoked only once? hence may I know the reason for reset() functions?
3. for loop logic as round robin, should we consider a go routine base on pod id? (each pod id as a key and a go routine for update?)
4. gpu device decouple from current logic? in case on no gpu?

[SamYuan1990]

@rootfs , what do you think?

[rootfs]

the code block is called in a go routine triggered by timer. The reset() function is to clean up the ebpf table before first entry and after each read.

A per pod id based go routine will not help here since it is not known which pod will be collected by ebpf in the kernel. Once the table is filled by ebpf, the userspace read is quick enough (i tested it on my laptop). Having a dedicated go routine for each pod id doesn't improve the performance but it will add resources to the processing.

gpu device is associated with each pod, how do you want to separate from the current logic?

[SamYuan1990]

so far if there no gpu device, the code still go through with gpu logic for example: podGPUDelta
as we have *enableGPU, if GPU is disabled, can we skip gpu related logic?

[rootfs]

sounds a good PR candidate

[marceloamaral]

lock.lock() not follow with defer lock.unlock()

We unlock in the end and there is no branch in between, so we don't need the defer

this read() function seems been invoked only once? hence may I know the reason for reset() functions?

Yes. We have an infinity loop reading the data, as expected from this function
As @rootfs said before, we must reset the counters every time that we will read the data from eBPF

for loop logic as round robin, should we consider a go routine base on pod id? (each pod id as a key and a go routine for update?)

We don't need to create a go routine for each pod.

gpu device decouple from current logic? in case on no gpu?

I think it is better to keep everything together