[codex] Add worker pod termination status logs

controlplane/k8s_pool_helpers.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/posthog/duckgres/controlplane/configstore"
 	"github.com/posthog/duckgres/server"
+	corev1 "k8s.io/api/core/v1"
 )
 
 // --- Helpers ---
@@ -212,3 +213,97 @@ func (p *K8sWorkerPool) logw(id int) *slog.Logger {
 	}
 	return slog.With("worker", id)
 }
+
+func workerPodStatusLogAttrs(pod *corev1.Pod) []any {
+	if pod == nil {
+		return nil
+	}
+	attrs := []any{
+		"pod_phase", string(pod.Status.Phase),
+	}
+	if pod.Status.Reason != "" {
+		attrs = append(attrs, "pod_reason", pod.Status.Reason)
+	}
+	if pod.Status.Message != "" {
+		attrs = append(attrs, "pod_message", pod.Status.Message)
+	}
+	if pod.Spec.NodeName != "" {
+		attrs = append(attrs, "pod_node", pod.Spec.NodeName)
+	}
+	if pod.Status.PodIP != "" {
+		attrs = append(attrs, "pod_ip", pod.Status.PodIP)
+	}
+	if pod.DeletionTimestamp != nil {
+		attrs = append(attrs, "pod_deletion_timestamp", pod.DeletionTimestamp.Time.UTC().Format(time.RFC3339Nano))
+	}
+	if status, ok := workerContainerStatus(pod.Status.ContainerStatuses); ok {
+		attrs = append(attrs, containerStatusLogAttrs(status)...)
+	}
+	return attrs
+}
+
+func workerContainerStatus(statuses []corev1.ContainerStatus) (corev1.ContainerStatus, bool) {
+	if len(statuses) == 0 {
+		return corev1.ContainerStatus{}, false
+	}
+	for _, status := range statuses {
+		if status.Name == "duckdb-worker" {
+			return status, true
+		}
+	}
+	return statuses[0], true
+}
+
+func containerStatusLogAttrs(status corev1.ContainerStatus) []any {
+	attrs := []any{
+		"container_name", status.Name,
+		"container_restart_count", status.RestartCount,
+	}
+	switch {
+	case status.State.Terminated != nil:
+		term := status.State.Terminated
+		attrs = append(attrs,
+			"container_state", "terminated",
+			"container_reason", term.Reason,
+			"container_exit_code", term.ExitCode,
+			"container_signal", term.Signal,
+		)
+		if term.Message != "" {
+			attrs = append(attrs, "container_message", term.Message)
+		}
+		if !term.StartedAt.IsZero() {
+			attrs = append(attrs, "container_started_at", term.StartedAt.Time.UTC().Format(time.RFC3339Nano))
+		}
+		if !term.FinishedAt.IsZero() {
+			attrs = append(attrs, "container_finished_at", term.FinishedAt.Time.UTC().Format(time.RFC3339Nano))
+		}
+	case status.State.Waiting != nil:
+		waiting := status.State.Waiting
+		attrs = append(attrs,
+			"container_state", "waiting",
+			"container_reason", waiting.Reason,
+		)
+		if waiting.Message != "" {
+			attrs = append(attrs, "container_message", waiting.Message)
+		}
+	case status.State.Running != nil:
+		running := status.State.Running
+		attrs = append(attrs, "container_state", "running")
+		if !running.StartedAt.IsZero() {
+			attrs = append(attrs, "container_started_at", running.StartedAt.Time.UTC().Format(time.RFC3339Nano))
+		}
+	default:
+		attrs = append(attrs, "container_state", "unknown")
+	}
+	if term := status.LastTerminationState.Terminated; term != nil {
+		attrs = append(attrs,
+			"last_container_reason", term.Reason,
+			"last_container_exit_code", term.ExitCode,
+			"last_container_signal", term.Signal,
+		)
+		if !term.FinishedAt.IsZero() {
+			attrs = append(attrs, "last_container_finished_at", term.FinishedAt.Time.UTC().Format(time.RFC3339Nano))
+		}
+	}
+	return attrs
+}

controlplane/k8s_pool_lifecycle.go

@@ -438,15 +438,44 @@ func (p *K8sWorkerPool) HealthCheckLoop(ctx context.Context, interval time.Durat
 								}
 								observeControlPlaneWorkers(workerCount)
 
-								p.logw(lease.workerID).Error("K8s worker unresponsive, deleting pod.", "consecutive_failures", count)
 								if onCrash != nil {
 									onCrash(lease.workerID)
 								}
-								// Delete the pod to force cleanup
+
+								// Snapshot pod/container state before the delete below removes
+								// the easiest source of OOMKilled/Evicted/exit-code evidence.
 								podName := p.workerPodName(removedWorker)
-								_ = p.clientset.CoreV1().Pods(p.namespace).Delete(ctx, podName, metav1.DeleteOptions{
-									GracePeriodSeconds: int64Ptr(10),
-								})
+								deleteAttrs := []any{"consecutive_failures", count}
+								if podName != "" && p.clientset != nil {
+									statusCtx, statusCancel := context.WithTimeout(ctx, 2*time.Second)
+									pod, err := p.clientset.CoreV1().Pods(p.namespace).Get(statusCtx, podName, metav1.GetOptions{})
+									statusCancel()
+									if err != nil {
+										deleteAttrs = append(deleteAttrs, "pod_status_error", err)
+									} else {
+										deleteAttrs = append(deleteAttrs, workerPodStatusLogAttrs(pod)...)
+									}
+								}
+								logger := slog.With(workerLogAttrs(removedWorker)...)
+								logger.Error("K8s worker unresponsive, deleting pod.", deleteAttrs...)
+								deleteResultAttrs := append(append([]any{}, deleteAttrs...), "grace_period_seconds", 10)
+								if podName == "" || p.clientset == nil {
+									logger.Warn("K8s worker pod delete skipped.", append(deleteResultAttrs, "reason", "missing_pod_or_client")...)
+								} else {
+									deleteCtx, deleteCancel := context.WithTimeout(context.Background(), 10*time.Second)
+									err := p.clientset.CoreV1().Pods(p.namespace).Delete(deleteCtx, podName, metav1.DeleteOptions{
+										GracePeriodSeconds: int64Ptr(10),
+									})
+									deleteCancel()
+									switch {
+									case err == nil:
+										logger.Info("K8s worker pod delete requested.", deleteResultAttrs...)
+									case errors.IsNotFound(err):
+										logger.Info("K8s worker pod already gone before delete request completed.", append(deleteResultAttrs, "error", err)...)
+									default:
+										logger.Warn("K8s worker pod delete request failed.", append(deleteResultAttrs, "error", err)...)
+									}
+								}
 								if removedWorker.client != nil {
 									_ = removedWorker.client.Close()
 								}

controlplane/k8s_pool_reconcile.go

@@ -400,7 +400,14 @@ func (p *K8sWorkerPool) onPodTerminated(pod *corev1.Pod) {
 	case <-w.done:
 		// Already closed
 	default:
-		slog.Warn("Worker pod terminated.", "worker", id, "worker_pod", pod.Name, "org", pod.Labels["duckgres/active-org"], "phase", pod.Status.Phase)
+		attrs := []any{
+			"worker", id,
+			"worker_pod", pod.Name,
+			"org", pod.Labels["duckgres/active-org"],
+			"phase", pod.Status.Phase,
+		}
+		attrs = append(attrs, workerPodStatusLogAttrs(pod)...)
+		slog.Warn("Worker pod terminated.", attrs...)
 		close(w.done)
 	}
 }