fix notifications

internal/topology/podlogs.go

@@ -0,0 +1,87 @@
+package topology
+
+import (
+	"bufio"
+	"context"
+	"sort"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// PodLog is what looking up a running pod behind a Deployment found.
+type PodLog struct {
+	Pod   string   // empty unless Found
+	Lines []string // most recent log lines, oldest first
+	Found bool     // a pod exists and was queried (Lines may still be empty: a quiet pod)
+	Err   string   // set when the lookup itself failed; Found is false
+}
+
+// podLogTimeout bounds one lookup so a slow or unreachable API server never
+// stalls alert delivery.
+const podLogTimeout = 5 * time.Second
+
+// PodLogs finds a pod behind namespace/deployment and returns its last `tail`
+// log lines. A Deployment with no live pods right now - scaled to 0, or mid
+// rollout with none ready yet - comes back with Found=false and no Err: that
+// absence is itself the answer, not a failure.
+func (r *Registry) PodLogs(ctx context.Context, namespace, deployment string, tail int) PodLog {
+	if r == nil || r.client == nil {
+		return PodLog{}
+	}
+	ctx, cancel := context.WithTimeout(ctx, podLogTimeout)
+	defer cancel()
+
+	dep, err := r.client.AppsV1().Deployments(namespace).Get(ctx, deployment, metav1.GetOptions{})
+	if err != nil {
+		return PodLog{Err: err.Error()}
+	}
+	sel, err := metav1.LabelSelectorAsSelector(dep.Spec.Selector)
+	if err != nil {
+		return PodLog{Err: err.Error()}
+	}
+	pods, err := r.client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: sel.String()})
+	if err != nil {
+		return PodLog{Err: err.Error()}
+	}
+	live := make([]corev1.Pod, 0, len(pods.Items))
+	for _, p := range pods.Items {
+		if p.DeletionTimestamp == nil {
+			live = append(live, p)
+		}
+	}
+	if len(live) == 0 {
+		return PodLog{} // scaled to 0, or nothing ready yet
+	}
+	// Prefer a Running pod; among ties prefer the newest, most likely to hold
+	// the log lines behind whatever just triggered the alert.
+	sort.Slice(live, func(i, j int) bool {
+		ri, rj := live[i].Status.Phase == corev1.PodRunning, live[j].Status.Phase == corev1.PodRunning
+		if ri != rj {
+			return ri
+		}
+		return live[i].CreationTimestamp.After(live[j].CreationTimestamp.Time)
+	})
+	pod := live[0]
+	container := ""
+	if len(pod.Spec.Containers) > 0 {
+		container = pod.Spec.Containers[0].Name
+	}
+	tailLines := int64(tail)
+	stream, err := r.client.CoreV1().Pods(namespace).GetLogs(pod.Name, &corev1.PodLogOptions{Container: container, TailLines: &tailLines}).Stream(ctx)
+	if err != nil {
+		return PodLog{Pod: pod.Name, Found: true, Err: err.Error()}
+	}
+	defer stream.Close()
+	var lines []string
+	sc := bufio.NewScanner(stream)
+	sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+	for sc.Scan() {
+		lines = append(lines, sc.Text())
+	}
+	if len(lines) > tail {
+		lines = lines[len(lines)-tail:]
+	}
+	return PodLog{Pod: pod.Name, Lines: lines, Found: true}
+}

internal/topology/watcher.go

@@ -24,9 +24,10 @@ import (
 // Registry serves the latest Snapshot. A nil *Registry resolves nothing, so
 // callers need no cluster to run.
 type Registry struct {
-	snap  atomic.Pointer[Snapshot]
-	dirty atomic.Bool
-	built atomic.Int64 // unix seconds of the last rebuild
+	snap   atomic.Pointer[Snapshot]
+	dirty  atomic.Bool
+	built  atomic.Int64 // unix seconds of the last rebuild
+	client kubernetes.Interface // for on-demand pod lookups; nil outside a cluster
 }
 
 // Resolve returns the backends serving host+path.
@@ -79,7 +80,7 @@ func NewFromEnv(ctx context.Context) (*Registry, error) {
 	svc := factory.Core().V1().Services().Informer()
 	dep := factory.Apps().V1().Deployments().Informer()
 
-	r := &Registry{}
+	r := &Registry{client: client}
 	mark := cache.ResourceEventHandlerFuncs{
 		AddFunc:    func(any) { r.dirty.Store(true) },
 		UpdateFunc: func(_, _ any) { r.dirty.Store(true) },