feat: detect and report container OOM kills

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,

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
+}

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)
+	}
+}

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)
 			}
 		},
 	})