From de64adb7bd8c80f9f6892fa50d201bf083870d7f Mon Sep 17 00:00:00 2001 From: Konstantins Feofantovs Date: Wed, 29 Jul 2026 18:43:00 +0300 Subject: [PATCH] feat: detect and report container OOM kills MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pod watcher now scans container statuses (init and regular) for terminations with reason OOMKilled โ€” both lastState (restarted containers) and current state (restartPolicy Never) โ€” and reports each kill exactly once via a new OnOOMKill callback. The event pairs the kill with the container's declared memory limit, plus finishedAt, exit code, restart count, and the resolved workload. OOM kills arrive as pod status updates, so the watcher gains an UpdateFunc handler; deduplication is an in-memory map keyed by container and finishedAt, purged on pod deletion. Losing it on agent restart is harmless: the same kill is re-reported, never missed. Uses only the pods/status access already promised in docs/security.md ยง4, which names this exact detection. Co-Authored-By: Claude Fable 5 --- cmd/agent/main.go | 13 +++ internal/collector/oom.go | 113 ++++++++++++++++++++++ internal/collector/oom_test.go | 165 +++++++++++++++++++++++++++++++++ internal/collector/pods.go | 27 +++++- test/e2e/oom_test.go | 113 ++++++++++++++++++++++ 5 files changed, 430 insertions(+), 1 deletion(-) create mode 100644 internal/collector/oom.go create mode 100644 internal/collector/oom_test.go create mode 100644 test/e2e/oom_test.go diff --git a/cmd/agent/main.go b/cmd/agent/main.go index ffc55ea..d12f4ea 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -79,6 +79,19 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac "containers", p.Containers, ) }) + podWatcher.OnOOMKill(func(o collector.OOMKill) { + logger.Warn("oom kill observed", + "namespace", o.Namespace, + "pod", o.Pod, + "container", o.Container, + "workload_kind", o.Workload.Kind, + "workload_name", o.Workload.Name, + "finished_at", o.FinishedAt, + "exit_code", o.ExitCode, + "restart_count", o.RestartCount, + "memory_limit_bytes", o.MemoryLimitBytes, + ) + }) nodeWatcher := collector.NewNodeWatcher(clientset, func(n collector.NodeInfo) { logger.Info("node observed", "node", n.Name, diff --git a/internal/collector/oom.go b/internal/collector/oom.go new file mode 100644 index 0000000..934563f --- /dev/null +++ b/internal/collector/oom.go @@ -0,0 +1,113 @@ +package collector + +import ( + "strings" + "time" + + corev1 "k8s.io/api/core/v1" +) + +// oomKilledReason is the reason the container runtime sets on a terminated +// container state when the kernel OOM killer ended it. +const oomKilledReason = "OOMKilled" + +// OOMKill is one observed out-of-memory kill of a container, paired with the +// memory limit the container declared โ€” the two facts that together make the +// "limit too low" story. MemoryLimitBytes is nil when no limit was set. +type OOMKill struct { + Namespace string `json:"namespace"` + Pod string `json:"pod"` + Container string `json:"container"` + Workload WorkloadRef `json:"workload"` + FinishedAt time.Time `json:"finished_at"` + ExitCode int32 `json:"exit_code"` + RestartCount int32 `json:"restart_count"` + MemoryLimitBytes *int64 `json:"memory_limit_bytes,omitempty"` +} + +// OnOOMKill registers fn to be called once per observed OOM kill. Must be +// called before Run. fn is called from the informer goroutine and must not +// block. +func (w *PodWatcher) OnOOMKill(fn func(OOMKill)) { + w.onOOM = fn +} + +// reportOOMKills scans a pod's container statuses for OOM-killed terminations +// and reports each one exactly once. A kill shows up in state.terminated +// while the container lies dead (restartPolicy Never) and moves to +// lastState.terminated after a restart; both carry the same finishedAt, so +// the dedup key covers the move. +func (w *PodWatcher) reportOOMKills(pod *corev1.Pod) { + if w.onOOM == nil { + return + } + statuses := make([]corev1.ContainerStatus, 0, len(pod.Status.InitContainerStatuses)+len(pod.Status.ContainerStatuses)) + statuses = append(statuses, pod.Status.InitContainerStatuses...) + statuses = append(statuses, pod.Status.ContainerStatuses...) + for _, status := range statuses { + for _, term := range []*corev1.ContainerStateTerminated{ + status.LastTerminationState.Terminated, + status.State.Terminated, + } { + if term == nil || term.Reason != oomKilledReason { + continue + } + key := oomKey(pod, status.Name, term) + w.mu.Lock() + _, dup := w.reportedOOMs[key] + if !dup { + w.reportedOOMs[key] = struct{}{} + } + w.mu.Unlock() + if dup { + continue + } + w.onOOM(OOMKill{ + Namespace: pod.Namespace, + Pod: pod.Name, + Container: status.Name, + Workload: w.resolveWorkload(pod), + FinishedAt: term.FinishedAt.Time, + ExitCode: term.ExitCode, + RestartCount: status.RestartCount, + MemoryLimitBytes: memoryLimitOf(pod, status.Name), + }) + } + } +} + +// forgetOOMKills drops the dedup entries of a deleted pod so the map does not +// grow for the lifetime of the agent. Losing the whole map on restart is +// harmless: the same OOM gets reported again, nothing is missed. +func (w *PodWatcher) forgetOOMKills(pod *corev1.Pod) { + prefix := pod.Namespace + "/" + pod.Name + "/" + w.mu.Lock() + defer w.mu.Unlock() + for key := range w.reportedOOMs { + if strings.HasPrefix(key, prefix) { + delete(w.reportedOOMs, key) + } + } +} + +// oomKey identifies one termination: container names cannot contain "/", and +// finishedAt distinguishes repeated kills of the same container. +func oomKey(pod *corev1.Pod, container string, term *corev1.ContainerStateTerminated) string { + return pod.Namespace + "/" + pod.Name + "/" + container + "@" + term.FinishedAt.UTC().Format(time.RFC3339Nano) +} + +// memoryLimitOf returns the declared memory limit of the named container, +// nil when the container has none. +func memoryLimitOf(pod *corev1.Pod, container string) *int64 { + for _, c := range pod.Spec.InitContainers { + if c.Name == container { + return resourcesOf(&c).MemoryLimitBytes + } + } + for _, c := range pod.Spec.Containers { + if c.Name == container { + return resourcesOf(&c).MemoryLimitBytes + } + } + return nil +} diff --git a/internal/collector/oom_test.go b/internal/collector/oom_test.go new file mode 100644 index 0000000..8466f0c --- /dev/null +++ b/internal/collector/oom_test.go @@ -0,0 +1,165 @@ +package collector + +import ( + "context" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/utils/ptr" +) + +func terminated(reason string, exitCode int32, finishedAt time.Time) corev1.ContainerState { + return corev1.ContainerState{ + Terminated: &corev1.ContainerStateTerminated{ + Reason: reason, + ExitCode: exitCode, + FinishedAt: metav1.NewTime(finishedAt), + }, + } +} + +// oomWatcher starts a PodWatcher over the given pod and returns the OOM +// events channel plus the started clientset for follow-up updates. +func oomWatcher(t *testing.T, initial *corev1.Pod) (*fake.Clientset, chan OOMKill) { + t.Helper() + clientset := fake.NewClientset(initial) + + events := make(chan OOMKill, 16) + watcher := NewPodWatcher(clientset, func(PodInfo) {}) + watcher.OnOOMKill(func(o OOMKill) { events <- o }) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- watcher.Run(ctx) }() + t.Cleanup(func() { + cancel() + if err := <-done; err != nil { + t.Errorf("watcher returned error: %v", err) + } + }) + return clientset, events +} + +func waitOOM(t *testing.T, events chan OOMKill) OOMKill { + t.Helper() + select { + case o := <-events: + return o + case <-time.After(5 * time.Second): + t.Fatal("no OOM kill reported before timeout") + return OOMKill{} + } +} + +func TestReportsOOMKillOnceAcrossStatusUpdates(t *testing.T) { + firstKill := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) + p := pod("checkout-1", ptr.To(controllerRef("StatefulSet", "checkout"))) + clientset, events := oomWatcher(t, p) + + // The app container gets OOM-killed and restarted. + update := p.DeepCopy() + update.Status.ContainerStatuses = []corev1.ContainerStatus{{ + Name: "app", + RestartCount: 1, + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}, + LastTerminationState: terminated(oomKilledReason, 137, firstKill), + }} + if _, err := clientset.CoreV1().Pods("shop").Update(t.Context(), update, metav1.UpdateOptions{}); err != nil { + t.Fatalf("updating pod: %v", err) + } + + got := waitOOM(t, events) + if got.Namespace != "shop" || got.Pod != "checkout-1" || got.Container != "app" { + t.Errorf("event = %s/%s container %s, want shop/checkout-1 container app", got.Namespace, got.Pod, got.Container) + } + if got.ExitCode != 137 || got.RestartCount != 1 || !got.FinishedAt.Equal(firstKill) { + t.Errorf("event = %+v, want exit 137, restarts 1, finished at %s", got, firstKill) + } + if got.Workload != (WorkloadRef{Kind: "StatefulSet", Name: "checkout"}) { + t.Errorf("workload = %+v, want StatefulSet/checkout", got.Workload) + } + // The app container declares a 1Gi limit (see pod() in pods_test.go). + if got.MemoryLimitBytes == nil || *got.MemoryLimitBytes != 1<<30 { + t.Errorf("memory limit = %v, want %d", got.MemoryLimitBytes, int64(1<<30)) + } + + // An unrelated status update carrying the same lastState must not + // re-report; a genuinely new kill must. Receiving the second kill as + // the *next* event proves the middle update produced none. + unrelated := update.DeepCopy() + unrelated.Status.Phase = corev1.PodRunning + unrelated.Labels = map[string]string{"touched": "yes"} + if _, err := clientset.CoreV1().Pods("shop").Update(t.Context(), unrelated, metav1.UpdateOptions{}); err != nil { + t.Fatalf("updating pod: %v", err) + } + secondKill := firstKill.Add(time.Minute) + again := unrelated.DeepCopy() + again.Status.ContainerStatuses[0].RestartCount = 2 + again.Status.ContainerStatuses[0].LastTerminationState = terminated(oomKilledReason, 137, secondKill) + if _, err := clientset.CoreV1().Pods("shop").Update(t.Context(), again, metav1.UpdateOptions{}); err != nil { + t.Fatalf("updating pod: %v", err) + } + + next := waitOOM(t, events) + if !next.FinishedAt.Equal(secondKill) { + t.Errorf("next event finished at %s, want %s (the first kill was reported twice)", next.FinishedAt, secondKill) + } + if next.RestartCount != 2 { + t.Errorf("next event restart count = %d, want 2", next.RestartCount) + } +} + +func TestReportsOOMKillPresentAtStart(t *testing.T) { + // restartPolicy Never: the kill sits in state.terminated, and the pod + // already looks like this when the watcher starts (cache replay path). + killedAt := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) + p := pod("one-shot", nil) + p.Status.Phase = corev1.PodFailed + p.Status.ContainerStatuses = []corev1.ContainerStatus{{ + Name: "sidecar", + State: terminated(oomKilledReason, 137, killedAt), + }} + _, events := oomWatcher(t, p) + + got := waitOOM(t, events) + if got.Pod != "one-shot" || got.Container != "sidecar" || got.ExitCode != 137 { + t.Errorf("event = %+v, want one-shot/sidecar exit 137", got) + } + // The sidecar container declares no limits. + if got.MemoryLimitBytes != nil { + t.Errorf("memory limit = %d, want unset", *got.MemoryLimitBytes) + } +} + +func TestIgnoresNonOOMTermination(t *testing.T) { + crashedAt := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) + p := pod("crasher", nil) + clientset, events := oomWatcher(t, p) + + // A plain crash (reason Error) must not be reported as an OOM kill. + update := p.DeepCopy() + update.Status.ContainerStatuses = []corev1.ContainerStatus{{ + Name: "app", + RestartCount: 1, + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}, + LastTerminationState: terminated("Error", 1, crashedAt), + }} + if _, err := clientset.CoreV1().Pods("shop").Update(t.Context(), update, metav1.UpdateOptions{}); err != nil { + t.Fatalf("updating pod: %v", err) + } + // Follow with a real OOM: it must be the first event received. + oom := update.DeepCopy() + oom.Status.ContainerStatuses[0].RestartCount = 2 + oom.Status.ContainerStatuses[0].LastTerminationState = terminated(oomKilledReason, 137, crashedAt.Add(time.Minute)) + if _, err := clientset.CoreV1().Pods("shop").Update(t.Context(), oom, metav1.UpdateOptions{}); err != nil { + t.Fatalf("updating pod: %v", err) + } + + got := waitOOM(t, events) + if got.RestartCount != 2 || got.ExitCode != 137 { + t.Errorf("event = %+v, want the OOM kill (restarts 2), not the plain crash", got) + } +} diff --git a/internal/collector/pods.go b/internal/collector/pods.go index 6568ea1..e146591 100644 --- a/internal/collector/pods.go +++ b/internal/collector/pods.go @@ -5,6 +5,7 @@ package collector import ( "context" "fmt" + "sync" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -63,16 +64,24 @@ type PodInfo struct { type PodWatcher struct { clientset kubernetes.Interface onPod func(PodInfo) + onOOM func(OOMKill) rsLister appslisters.ReplicaSetLister jobLister batchlisters.JobLister + + mu sync.Mutex + reportedOOMs map[string]struct{} } // NewPodWatcher returns a watcher that calls onPod for every pod present at // start and for every pod created afterwards. onPod is called from the // informer goroutine and must not block. func NewPodWatcher(clientset kubernetes.Interface, onPod func(PodInfo)) *PodWatcher { - return &PodWatcher{clientset: clientset, onPod: onPod} + return &PodWatcher{ + clientset: clientset, + onPod: onPod, + reportedOOMs: make(map[string]struct{}), + } } // Run blocks until ctx is canceled. ReplicaSet and Job caches are synced @@ -107,6 +116,22 @@ func (w *PodWatcher) Run(ctx context.Context) error { AddFunc: func(obj any) { if pod, ok := obj.(*corev1.Pod); ok { w.onPod(w.describe(pod)) + w.reportOOMKills(pod) + } + }, + // OOM kills happen after a pod appears and arrive as status + // updates; the pod itself is not re-reported. + UpdateFunc: func(_, obj any) { + if pod, ok := obj.(*corev1.Pod); ok { + w.reportOOMKills(pod) + } + }, + DeleteFunc: func(obj any) { + if unknown, ok := obj.(cache.DeletedFinalStateUnknown); ok { + obj = unknown.Obj + } + if pod, ok := obj.(*corev1.Pod); ok { + w.forgetOOMKills(pod) } }, }) diff --git a/test/e2e/oom_test.go b/test/e2e/oom_test.go new file mode 100644 index 0000000..891c8a9 --- /dev/null +++ b/test/e2e/oom_test.go @@ -0,0 +1,113 @@ +//go:build e2e + +package e2e + +import ( + "context" + "encoding/json" + "fmt" + "os" + "sync" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/RebuildStackCo/runtime-agent/internal/collector" +) + +func TestOOMKillDetectionAgainstRealCluster(t *testing.T) { + clientset := clusterClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + ns := fmt.Sprintf("runtime-agent-e2e-oom-%d", os.Getpid()) + if _, err := clientset.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: ns}, + }, metav1.CreateOptions{}); err != nil { + t.Fatalf("creating namespace: %v", err) + } + t.Cleanup(func() { + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), time.Minute) + defer cleanupCancel() + _ = clientset.CoreV1().Namespaces().Delete(cleanupCtx, ns, metav1.DeleteOptions{}) + }) + + // The shell doubles a string until the 32Mi memory limit kills it. + // The shell is PID 1, so the OOM kill terminates the whole container + // and the runtime marks it OOMKilled. restartPolicy Never keeps the + // kill in state.terminated, deterministic to assert on. + memoryHog := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "memory-hog"}, + Spec: corev1.PodSpec{ + RestartPolicy: corev1.RestartPolicyNever, + Containers: []corev1.Container{{ + Name: "hog", + Image: "docker.io/library/busybox:1.36", + Command: []string{"sh", "-c", `s=x; while true; do s="$s$s"; done`}, + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + corev1.ResourceMemory: resource.MustParse("32Mi"), + }, + }, + }}, + }, + } + if _, err := clientset.CoreV1().Pods(ns).Create(ctx, memoryHog, metav1.CreateOptions{}); err != nil { + t.Fatalf("creating pod: %v", err) + } + + var mu sync.Mutex + var kills []collector.OOMKill + watcher := collector.NewPodWatcher(clientset, func(collector.PodInfo) {}) + watcher.OnOOMKill(func(o collector.OOMKill) { + observed, _ := json.Marshal(o) + t.Logf("oom kill observed: %s", observed) + if o.Namespace != ns { + return + } + mu.Lock() + kills = append(kills, o) + mu.Unlock() + }) + watcherDone := make(chan error, 1) + go func() { watcherDone <- watcher.Run(ctx) }() + + // Image pull plus a few seconds of allocation; 3 minutes is generous. + deadline := time.Now().Add(3 * time.Minute) + var got collector.OOMKill + for { + mu.Lock() + if len(kills) > 0 { + got = kills[0] + } + mu.Unlock() + if got.Pod != "" { + break + } + if time.Now().After(deadline) { + t.Fatal("no OOM kill reported before timeout") + } + time.Sleep(time.Second) + } + + if got.Pod != "memory-hog" || got.Container != "hog" { + t.Errorf("kill = %s/%s, want memory-hog/hog", got.Pod, got.Container) + } + if got.ExitCode != 137 { + t.Errorf("exit code = %d, want 137", got.ExitCode) + } + if got.MemoryLimitBytes == nil || *got.MemoryLimitBytes != 32<<20 { + t.Errorf("memory limit = %v, want %d", got.MemoryLimitBytes, int64(32<<20)) + } + if got.Workload.Kind != "none" { + t.Errorf("workload kind = %q, want none (bare pod)", got.Workload.Kind) + } + + cancel() + if err := <-watcherDone; err != nil { + t.Fatalf("watcher returned error: %v", err) + } +}