如何使用 Python 检测 Kubernetes 中的 CrashLoopBackOff（分步指南）

Source: Dev.to

请提供您希望翻译的完整文本内容，我将为您翻译成简体中文并保留原有的格式。

介绍

如果你在使用 Kubernetes，可能已经遇到过 CrashLoopBackOff 错误——这是 Kubernetes 环境中最常见且令人沮丧的问题之一。

传统的调试方式包括：

• 运行 kubectl 命令
• 手动检查日志
• 猜测根本原因

这个过程既慢又低效。在本指南中，你将学习如何通过结合 Pod 状态和日志分析，使用 Python 自动检测 CrashLoopBackOff。

什么是 CrashLoopBackOff？

CrashLoopBackOff 发生在：

• 容器启动
• 立即崩溃
• Kubernetes 重启它
• 循环重复

示例：

kubectl get pods

输出：

sample-app   0/1   CrashLoopBackOff   3 (15s ago)

目标

• 自动检测 CrashLoopBackOff
• 获取日志
• 生成结构化洞察
• 减少手动调试

第一步：使用 Python 获取 Kubernetes Pods

我们将使用 subprocess 来调用 kubectl：

import subprocess
import json

def list_pods(namespace):
    result = subprocess.run(
        ["kubectl", "get", "pods", "-n", namespace, "-o", "json"],
        capture_output=True,
        text=True
    )
    pods = json.loads(result.stdout)
    pod_list = []
    for item in pods["items"]:
        name = item["metadata"]["name"]
        state = item["status"]["containerStatuses"][0]["state"]
        if "waiting" in state:
            reason = state["waiting"]["reason"]
        else:
            reason = "Running"
        pod_list.append({
            "name": name,
            "state": reason
        })
    return pod_list

第2步：检测 CrashLoopBackOff

一旦我们获取了 pod 状态，检测就很直接：

def detect_failures(pods):
    failures = []
    for pod in pods:
        if pod["state"] in ["CrashLoopBackOff", "ImagePullBackOff", "ErrImagePull"]:
            failures.append({
                "pod_name": pod["name"],
                "issue": pod["state"],
                "severity": "CRITICAL"
            })
    return failures

第 3 步：获取 Pod 日志

检索日志以进行更深入的分析：

def get_pod_logs(namespace, pod_name):
    result = subprocess.run(
        ["kubectl", "logs", "-n", namespace, pod_name],
        capture_output=True,
        text=True
    )
    return result.stdout

第4步：解析日志中的错误

从日志中提取重要信号：

def parse_logs(logs):
    issues = []
    for line in logs.split("\n"):
        if "ERROR" in line:
            issues.append({
                "level": "WARNING",
                "message": line
            })
    return issues

第5步：合并状态 + 日志

def analyze_pod(namespace, pod):
    pod_name = pod["name"]
    pod_state = pod["state"]

    if pod_state == "CrashLoopBackOff":
        return {
            "pod_name": pod_name,
            "status": "unhealthy",
            "issues_found": [{
                "level": "CRITICAL",
                "message": f"Pod in {pod_state}"
            }]
        }

    logs = get_pod_logs(namespace, pod_name)
    log_issues = parse_logs(logs)

    if log_issues:
        return {
            "pod_name": pod_name,
            "status": "unhealthy",
            "issues_found": log_issues
        }

    return {
        "pod_name": pod_name,
        "status": "healthy",
        "issues_found": []
    }

{
  "pod_name": "sample-app",
  "status": "unhealthy",
  "issues_found": [
    {
      "level": "CRITICAL",
      "message": "Pod in CrashLoopBackOff"
    }
  ]
}

为什么这种方法有效

• 自动化故障检测
• 减少手动调试工作量
• 提供结构化洞察
• 适用于实时系统

关键要点

有效的 Kubernetes 调试结合了：

• Pod 状态
• 日志
• 上下文分析

更大的系统的一部分

此逻辑是更大 AI 驱动的 Kubernetes 调试器的一部分，该调试器能够：

• 自动检测故障
• 分析日志
• 提出修复建议

项目链接

GitHub: https://github.com/sumitpurandare/kube-ai