fix(resource): refresh process metadata on PID reuse

[1fanwang]

Summary

When Linux reuses a PID, Kepler can keep the previous process's command and container assignment for the new process. If the new process starts with less cumulative CPU time, its metadata can also be skipped while the old cache entry is retained.

Kepler now replaces the process-cache entry when cumulative CPU time drops. The replacement reads current procfs data and starts its CPU delta from the new process's total.

Testing Done

I have no production sighting. I reproduced PID reuse with Kepler's procfs reader and real child processes in a Linux PID namespace. The new process reused the old PID with zero CPU time. Before cache invalidation, Kepler kept the old command name; after the change, it read the new command.

From the repository root, save the source below as internal/resource/pid_reuse_live_test.go and run:

docker run --rm --privileged -v "$PWD:/src" -v kepler-go-modcache:/root/.cache/go-mod -v kepler-go-buildcache:/root/.cache/go-build -w /src -e GOMODCACHE=/root/.cache/go-mod -e KEPLER_PID_REUSE_E2E=1 golang:1.25-alpine sh -ec 'go test -v ./internal/resource -run "^TestLivePIDReuseCacheInvalidation$" -count=1'

Before cache invalidation:

=== RUN   TestLivePIDReuseCacheInvalidation
    pid_reuse_live_test.go:19: pid_namespace=pid:[4026534732]
    pid_reuse_live_test.go:88: pid=7671 reused=true cpu_before=0.2700 cpu_after=0.0000 comm_before=sh comm_after=sh
    pid_reuse_live_test.go:89: 
        	Error Trace:	/src/internal/resource/pid_reuse_live_test.go:89
        	Error:      	Expected and actual point to the same object: 0x4000262780 &resource.Process{PID:7671, Comm:"sh", Exe:"/bin/busybox", Type:"regular", Container:(*resource.Container)(nil), VirtualMachine:(*resource.VirtualMachine)(nil), CPUTotalTime:0, CPUTimeDelta:0}
        	Test:       	TestLivePIDReuseCacheInvalidation
    pid_reuse_live_test.go:90: 
        	Error Trace:	/src/internal/resource/pid_reuse_live_test.go:90
        	Error:      	Not equal: 
        	            	expected: "sleep"
        	            	actual  : "sh"
        	            	
        	            	Diff:
        	            	--- Expected
        	            	+++ Actual
        	            	@@ -1 +1 @@
        	            	-sleep
        	            	+sh
        	Test:       	TestLivePIDReuseCacheInvalidation
--- FAIL: TestLivePIDReuseCacheInvalidation (0.29s)
FAIL
FAIL	github.com/sustainable-computing-io/kepler/internal/resource	0.298s
FAIL
exit=1

After cache invalidation:

=== RUN   TestLivePIDReuseCacheInvalidation
    pid_reuse_live_test.go:19: pid_namespace=pid:[4026534732]
    pid_reuse_live_test.go:88: pid=85 reused=true cpu_before=0.2700 cpu_after=0.0000 comm_before=sh comm_after=sleep
--- PASS: TestLivePIDReuseCacheInvalidation (0.28s)
PASS
ok  	github.com/sustainable-computing-io/kepler/internal/resource	0.293s
exit=0

Reproducer source: pid_reuse_live_test.go

package resource

import (
	"fmt"
	"os"
	"os/exec"
	"strconv"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestLivePIDReuseCacheInvalidation(t *testing.T) {
	require.Equal(t, "1", os.Getenv("KEPLER_PID_REUSE_E2E"))
	pidNamespace, err := os.Readlink("/proc/self/ns/pid")
	require.NoError(t, err)
	t.Logf("pid_namespace=%s", pidNamespace)

	reader, err := NewProcFSReader("/proc")
	require.NoError(t, err)
	informer, err := NewInformer(WithProcReader(reader))
	require.NoError(t, err)

	oldCommand := exec.Command("/bin/sh", "-c", "while :; do :; done")
	require.NoError(t, oldCommand.Start())
	t.Cleanup(func() {
		if oldCommand.ProcessState == nil {
			_ = oldCommand.Process.Kill()
			_ = oldCommand.Wait()
		}
	})
	oldPID := oldCommand.Process.Pid
	var oldCached *Process
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		oldProc, err := liveProcByPID(reader, oldPID)
		require.NoError(t, err)
		oldCached, err = informer.updateProcessCache(oldProc)
		require.NoError(t, err)
		if oldCached.CPUTotalTime > 0 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	time.Sleep(250 * time.Millisecond)
	oldProc, err := liveProcByPID(reader, oldPID)
	require.NoError(t, err)
	oldCached, err = informer.updateProcessCache(oldProc)
	require.NoError(t, err)
	require.Greater(t, oldCached.CPUTotalTime, float64(0))
	oldCachedCPUTime := oldCached.CPUTotalTime
	oldComm := oldCached.Comm
	_ = oldCommand.Process.Kill()
	_ = oldCommand.Wait()

	require.NoError(t, os.WriteFile("/proc/sys/kernel/ns_last_pid", []byte(strconv.Itoa(oldPID-1)), 0o600))
	newCommand := exec.Command("/bin/sh", "-c", "exec sleep 30")
	require.NoError(t, newCommand.Start())
	t.Cleanup(func() {
		if newCommand.ProcessState == nil {
			_ = newCommand.Process.Kill()
			_ = newCommand.Wait()
		}
	})
	require.Equal(t, oldPID, newCommand.Process.Pid)
	var newProc procInfo
	deadline = time.Now().Add(time.Second)
	for time.Now().Before(deadline) {
		newProc, err = liveProcByPID(reader, oldPID)
		require.NoError(t, err)
		newComm, err := newProc.Comm()
		require.NoError(t, err)
		if newComm == "sleep" {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	newComm, err := newProc.Comm()
	require.NoError(t, err)
	require.Equal(t, "sleep", newComm)
	newCPUTime, err := newProc.CPUTime()
	require.NoError(t, err)
	require.Less(t, newCPUTime, oldCached.CPUTotalTime)
	newCached, err := informer.updateProcessCache(newProc)
	require.NoError(t, err)
	t.Logf("pid=%d reused=true cpu_before=%.4f cpu_after=%.4f comm_before=%s comm_after=%s", oldPID, oldCachedCPUTime, newCached.CPUTotalTime, oldComm, newCached.Comm)
	assert.NotSame(t, oldCached, newCached)
	assert.Equal(t, "sleep", newCached.Comm)
	assert.GreaterOrEqual(t, newCached.CPUTimeDelta, float64(0))
}

func liveProcByPID(reader *procFSReader, pid int) (procInfo, error) {
	procs, err := reader.AllProcs()
	if err != nil {
		return nil, err
	}
	for _, proc := range procs {
		if proc.PID() == pid {
			return proc, nil
		}
	}
	return nil, fmt.Errorf("pid %d not found in procfs", pid)
}

[sthaha]

I think a comment would be useful here for future maintainers

[1fanwang]

Have you seen this in the wild? I also think this should be used to invalidate the cache so the the proc info is re-read.

@sthaha, I have not seen a production occurrence. I reproduced PID reuse in a local Linux PID namespace. The fix rebuilds the process cache when cumulative CPU time drops, so Kepler reads the new process metadata.

Done in e8d6018.

[sthaha]

LGTM! Have you seen this in the wild?

I also think this should be used to invalidate the cache so the the proc info is re-read.

[Copilot]

Copilot review overview

🟡 Changes recommended

Equal CPU totals can still retain stale metadata after PID reuse.

Review effort: Balanced
Findings: 1

Open (1)

• Detect PID reuse when cumulative CPU time is equal or zero · New

What changed in this PR

Adds PID-reuse handling to refresh stale process metadata and CPU accounting.

Changes:

• Rebuilds cached processes when cumulative CPU time decreases.
• Adds unit and refresh-level regression tests.

File	Description
internal/​resource/​informer.go	Detects PID reuse and rebuilds process metadata.
internal/​resource/​procfs_reader_test.go	Tests metadata and CPU-delta reset behavior.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.