Python을 사용하여 Kubernetes에서 CrashLoopBackOff 감지하는 방법 (단계별 가이드)

Source: Dev.to

(위 링크에 있는 전체 글을 제공해 주시면, 해당 내용을 한국어로 번역해 드리겠습니다.)

소개

Kubernetes를 사용하고 있다면, 가장 흔하고 답답한 문제 중 하나인 CrashLoopBackOff 오류를 겪어봤을 가능성이 높습니다.

전통적으로 디버깅은 다음과 같은 과정을 포함합니다:

• kubectl 명령 실행
• 로그를 수동으로 확인
• 근본 원인을 추측

이 과정은 느리고 비효율적입니다. 이 가이드에서는 파드 상태와 로그 분석을 결합하여 Python으로 CrashLoopBackOff를 자동으로 감지하는 방법을 배웁니다.

CrashLoopBackOff란 무엇인가?

CrashLoopBackOff는 다음과 같은 경우에 발생합니다:

• 컨테이너가 시작됨
• 즉시 충돌함
• Kubernetes가 이를 재시작함
• 이 사이클이 반복됨

예시:

kubectl get pods

Output:

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

Goal

• CrashLoopBackOff를 자동으로 감지
• 로그를 가져옴
• 구조화된 인사이트 생성
• 수동 디버깅 감소

단계 1: 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 감지

파드 상태를 확보하면, 감지는 간단합니다:

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단계: 파드 로그 가져오기

깊은 분석을 위해 로그를 가져옵니다:

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"
    }
  ]
}

왜 이 접근 방식이 효과적인가

• 실패 감지를 자동화합니다
• 수동 디버깅 작업을 줄입니다
• 구조화된 인사이트를 제공합니다
• 실시간 시스템에서 작동합니다

핵심 요점

• Pod 상태
• 로그
• 맥락 분석

더 큰 시스템의 일부

• 실패를 자동으로 감지합니다
• 로그를 분석합니다
• 수정안을 제시합니다

프로젝트 링크

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