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 // The pod's own lifecycle facts, which explain errors the log lines alone // cannot: a 503 burst while a replica is being replaced, a crash loop the // kubelet is already retrying, or a pod that runs but never becomes ready. Ready bool // container passing its readiness probe Phase string // Pending | Running | Succeeded | Failed | Unknown Restarts int32 // restarts of the selected pod's containers PodRestarts int32 // restarts summed over every pod of the Deployment Reason string // CrashLoopBackOff, OOMKilled, ImagePullBackOff, ... ("" while healthy) ExitCode *int32 // exit code of the last container termination Message string // message of the last container termination LastTerminatedAt time.Time // when the last container terminated } // rollingWindow is how recently a container must have died for the resulting // connection errors to be attributed to its replacement still coming up. const rollingWindow = 5 * time.Minute // CrashLooping reports whether the pod is stuck restarting instead of serving. // Restarting the Deployment then only repeats what the kubelet is already // doing, so the answer is a new release, not a restart. func (p PodLog) CrashLooping() bool { if p.Reason == "CrashLoopBackOff" { return true } return p.Restarts > 0 && !p.LastTerminatedAt.IsZero() } // Restarting reports whether a pod of this Deployment is right now being // replaced, which is what makes a short 503 burst expected rather than new. func (p PodLog) Restarting() bool { return !p.LastTerminatedAt.IsZero() && time.Since(p.LastTerminatedAt) < rollingWindow } // 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] state := podState(pod, live) 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 withState(state, 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 withState(state, PodLog{Pod: pod.Name, Lines: lines, Found: true}) } // withState keeps the query result but carries the lifecycle facts along. func withState(state, result PodLog) PodLog { result.Ready, result.Phase, result.Restarts, result.PodRestarts = state.Ready, state.Phase, state.Restarts, state.PodRestarts result.Reason, result.ExitCode, result.Message, result.LastTerminatedAt = state.Reason, state.ExitCode, state.Message, state.LastTerminatedAt return result } // podState reads the lifecycle facts off the selected pod and totals the // restarts of its siblings, so one unhealthy replica is not mistaken for a // healthy deployment. func podState(selected corev1.Pod, all []corev1.Pod) PodLog { state := PodLog{Found: true, Ready: len(selected.Status.ContainerStatuses) > 0, Phase: string(selected.Status.Phase)} // The most recent termination across the pod's containers is the // interesting one: a pod being replaced is explained by its last death. keepTermination := func(term *corev1.ContainerStateTerminated) { if term == nil || !term.FinishedAt.Time.After(state.LastTerminatedAt) { return } state.LastTerminatedAt = term.FinishedAt.Time state.Message = term.Message code := term.ExitCode state.ExitCode = &code switch { case term.Reason != "": state.Reason = term.Reason case term.Message != "": state.Reason = "Error" } } for i := range selected.Status.ContainerStatuses { cs := &selected.Status.ContainerStatuses[i] state.Restarts += cs.RestartCount if !cs.Ready { state.Ready = false } keepTermination(cs.LastTerminationState.Terminated) } if selected.Status.Reason != "" { state.Reason = selected.Status.Reason } for _, p := range all { for _, cs := range p.Status.ContainerStatuses { state.PodRestarts += cs.RestartCount } } return state }