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