#!/usr/bin/env python3
"""
Kubernetes 노드 리소스 분석기 v2.1 - Python Edition with CSV Export
주요 기능:
• 노드별 CPU/메모리 할당 vs 요청량 vs 실제사용량 분석
• 점유율(Request 기준)과 사용률(실제 사용량) 구분 표시
• 1G 이상 CPU 요청 Pod의 효율성 분석
• 색상을 통한 직관적 상태 표시
• CSV 파일로 분석 결과 출력 (NEW!)
• 안정적인 Python 기반 계산
사용법:
python3 k8s_analyzer.py [--csv-output]
python3 k8s_analyzer.py --csv-output ./reports/
작성일: 2025-07-28
버전: v2.1 - CSV 출력 기능 추가
"""
import subprocess
import json
import re
import sys
import csv
import os
import argparse
from datetime import datetime
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, asdict
색상 정의
class Colors:
RED = '\033[1;31m'
GREEN = '\033[1;32m'
YELLOW = '\033[1;33m'
BLUE = '\033[1;34m'
PURPLE = '\033[1;35m'
CYAN = '\033[1;36m'
GRAY = '\033[0;37m'
BOLD = '\033[1m'
NC = '\033[0m' # No Color
@dataclass
class ResourceInfo:
"""리소스 정보를 담는 데이터 클래스"""
allocatable_cpu: float
allocatable_memory: float
requested_cpu: float
requested_memory: float
cpu_occupy_percent: float
memory_occupy_percent: float
actual_cpu_percent: Optional[float] = None
actual_memory_percent: Optional[float] = None
@dataclass
class PodResourceInfo:
"""Pod 리소스 정보를 담는 데이터 클래스"""
namespace: str
name: str
node: str
cpu_request: float
memory_request: float
actual_cpu: Optional[float] = None
efficiency: Optional[float] = None
@dataclass
class ClusterSummary:
"""클러스터 전체 요약 정보"""
total_nodes: int
total_allocatable_cpu: float
total_allocatable_memory: float
total_requested_cpu: float
total_requested_memory: float
avg_cpu_occupy_percent: float
avg_memory_occupy_percent: float
high_usage_nodes: int
pending_pods: int
analysis_timestamp: str
class K8sResourceAnalyzer:
"""Kubernetes 리소스 분석기 메인 클래스"""
def __init__(self, csv_output_dir: Optional[str] = None):
self.metrics_available = self._check_metrics_server()
self.nodes_data: Dict[str, ResourceInfo] = {}
self.high_cpu_pods: List[PodResourceInfo] = []
self.pending_pods: List[str] = []
self.csv_output_dir = csv_output_dir
self.analysis_timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
def _run_kubectl(self, command: str) -> str:
"""kubectl 명령어 실행"""
try:
result = subprocess.run(
f"kubectl {command}",
shell=True,
capture_output=True,
text=True,
check=False
)
return result.stdout
except Exception as e:
print(f"{Colors.RED}Error running kubectl: {e}{Colors.NC}")
return ""
def _check_metrics_server(self) -> bool:
"""Metrics Server 사용 가능 여부 확인"""
result = subprocess.run(
"kubectl top nodes",
shell=True,
capture_output=True,
text=True
)
return result.returncode == 0
def _convert_cpu_to_giga(self, cpu_str: str) -> float:
"""CPU 값을 기가코어로 변환"""
if not cpu_str or cpu_str == "null":
return 0.0
# 밀리코어 (예: 1500m)
if cpu_str.endswith('m'):
return float(cpu_str[:-1]) / 1000
# 정수 (예: 2)
elif cpu_str.isdigit():
return float(cpu_str)
# 소수점 (예: 1.5)
else:
try:
return float(cpu_str)
except ValueError:
return 0.0
def _convert_memory_to_giga(self, mem_str: str) -> float:
"""메모리 값을 기가바이트로 변환"""
if not mem_str or mem_str == "null":
return 0.0
# Ki 단위
if mem_str.endswith('Ki'):
return float(mem_str[:-2]) / (1024 * 1024)
# Mi 단위
elif mem_str.endswith('Mi'):
return float(mem_str[:-2]) / 1024
# Gi 단위
elif mem_str.endswith('Gi'):
return float(mem_str[:-2])
# Ti 단위
elif mem_str.endswith('Ti'):
return float(mem_str[:-2]) * 1024
else:
try:
# 바이트로 가정
return float(mem_str) / (1024 ** 3)
except ValueError:
return 0.0
def _calculate_percentage(self, numerator: float, denominator: float) -> float:
"""퍼센트 계산"""
if denominator == 0 or numerator == 0:
return 0.0
return (numerator / denominator) * 100
def _get_color_by_percentage(self, percentage: float) -> str:
"""퍼센트에 따른 색상 반환"""
if percentage >= 80:
return Colors.RED
elif percentage >= 60:
return Colors.YELLOW
else:
return Colors.GREEN
def _get_node_actual_usage(self, node: str) -> Tuple[Optional[float], Optional[float]]:
"""노드의 실제 사용률 가져오기 (kubectl top 기반)"""
if not self.metrics_available:
return None, None
top_output = self._run_kubectl("top nodes --no-headers")
for line in top_output.strip().split('\n'):
if line.startswith(node + ' '):
parts = line.split()
if len(parts) >= 5:
cpu_percent = float(parts[2].rstrip('%'))
memory_percent = float(parts[4].rstrip('%'))
return cpu_percent, memory_percent
return None, None
def print_header(self, title: str):
"""헤더 출력"""
border = "=" * (len(title) + 10)
print(f"\n{Colors.CYAN}{border}{Colors.NC}")
print(f"{Colors.CYAN} {title}{Colors.NC}")
print(f"{Colors.CYAN}{border}{Colors.NC}\n")
def analyze_nodes(self):
"""노드별 리소스 분석"""
print(f"{Colors.BLUE}🔍 노드 정보 수집 중...{Colors.NC}")
# 노드 목록 가져오기
nodes_output = self._run_kubectl("get nodes --no-headers -o custom-columns=':metadata.name'")
nodes = [node.strip() for node in nodes_output.strip().split('\n') if node.strip()]
for node in nodes:
print(f"{Colors.GRAY} 분석 중: {node}{Colors.NC}")
# 노드 상세 정보 가져오기
node_info = self._run_kubectl(f"describe node {node}")
# Allocatable 리소스 추출
allocatable_section = False
allocatable_cpu = "0"
allocatable_memory = "0"
for line in node_info.split('\n'):
if 'Allocatable:' in line:
allocatable_section = True
continue
if allocatable_section and 'cpu:' in line:
allocatable_cpu = line.split()[1]
elif allocatable_section and 'memory:' in line:
allocatable_memory = line.split()[1]
elif allocatable_section and not line.startswith(' '):
allocatable_section = False
# Allocated resources 추출
allocated_section = False
requested_cpu = "0"
requested_memory = "0"
for line in node_info.split('\n'):
if 'Allocated resources:' in line:
allocated_section = True
continue
if allocated_section and 'cpu' in line and 'requests' not in line.lower():
# cpu 줄에서 요청량 추출 (예: cpu 1500m (75%) 3000m (150%))
parts = line.split()
if len(parts) >= 2:
requested_cpu = parts[1].strip('()')
elif allocated_section and 'memory' in line and 'requests' not in line.lower():
parts = line.split()
if len(parts) >= 2:
requested_memory = parts[1].strip('()')
elif allocated_section and line.strip() and not line.startswith(' '):
allocated_section = False
# 단위 변환
alloc_cpu_g = self._convert_cpu_to_giga(allocatable_cpu)
alloc_mem_gi = self._convert_memory_to_giga(allocatable_memory)
req_cpu_g = self._convert_cpu_to_giga(requested_cpu)
req_mem_gi = self._convert_memory_to_giga(requested_memory)
# 점유율 계산
cpu_occupy = self._calculate_percentage(req_cpu_g, alloc_cpu_g)
memory_occupy = self._calculate_percentage(req_mem_gi, alloc_mem_gi)
# 실제 사용률 가져오기
actual_cpu, actual_memory = self._get_node_actual_usage(node)
# 데이터 저장
self.nodes_data[node] = ResourceInfo(
allocatable_cpu=alloc_cpu_g,
allocatable_memory=alloc_mem_gi,
requested_cpu=req_cpu_g,
requested_memory=req_mem_gi,
cpu_occupy_percent=cpu_occupy,
memory_occupy_percent=memory_occupy,
actual_cpu_percent=actual_cpu,
actual_memory_percent=actual_memory
)
def display_node_analysis(self):
"""노드 분석 결과 표시"""
self.print_header("📊 노드별 리소스 분석 결과")
# 메트릭 서버 상태 표시
if self.metrics_available:
print(f"{Colors.GREEN}✅ Metrics Server 사용 가능 - 실제 사용률 표시{Colors.NC}")
else:
print(f"{Colors.YELLOW}⚠️ Metrics Server 없음 - 점유율만 표시{Colors.NC}")
print(f"\n{Colors.BOLD}💡 점유율: Pod Request 기준 / 사용률: 실제 소비량 기준{Colors.NC}\n")
# 테이블 헤더
if self.metrics_available:
print(f"{Colors.BOLD}┌─────────────────────┬─────────────┬─────────────┬─────────────┬─────────────┬─────────────┬─────────────┬─────────────┬─────────────┐{Colors.NC}")
print(f"{Colors.BOLD}│ 노드명 │ CPU 할당(G) │ CPU 요청(G) │ CPU 점유율% │ CPU 사용률% │ 메모리할당Gi│ 메모리요청Gi│메모리점유율%│메모리사용률%│{Colors.NC}")
print(f"{Colors.BOLD}├─────────────────────┼─────────────┼─────────────┼─────────────┼─────────────┼─────────────┼─────────────┼─────────────┼─────────────┤{Colors.NC}")
else:
print(f"{Colors.BOLD}┌─────────────────────┬─────────────┬─────────────┬─────────────┬─────────────┬─────────────┬─────────────┐{Colors.NC}")
print(f"{Colors.BOLD}│ 노드명 │ CPU 할당(G) │ CPU 요청(G) │ CPU 점유율% │ 메모리할당Gi│ 메모리요청Gi│메모리점유율%│{Colors.NC}")
print(f"{Colors.BOLD}├─────────────────────┼─────────────┼─────────────┼─────────────┼─────────────┼─────────────┼─────────────┤{Colors.NC}")
# 각 노드 데이터 출력
for node_name, data in self.nodes_data.items():
cpu_occupy_color = self._get_color_by_percentage(data.cpu_occupy_percent)
memory_occupy_color = self._get_color_by_percentage(data.memory_occupy_percent)
node_display = node_name[:19] if len(node_name) > 19 else node_name
if self.metrics_available and data.actual_cpu_percent is not None:
cpu_usage_color = self._get_color_by_percentage(data.actual_cpu_percent)
memory_usage_color = self._get_color_by_percentage(data.actual_memory_percent)
print(f"│ {node_display:<19} │ {data.allocatable_cpu:>10.3f}G │ {data.requested_cpu:>10.3f}G │ "
f"{cpu_occupy_color}{data.cpu_occupy_percent:>10.1f}%{Colors.NC} │ "
f"{cpu_usage_color}{data.actual_cpu_percent:>10.1f}%{Colors.NC} │ "
f"{data.allocatable_memory:>10.2f}Gi │ {data.requested_memory:>10.2f}Gi │ "
f"{memory_occupy_color}{data.memory_occupy_percent:>11.1f}%{Colors.NC} │ "
f"{memory_usage_color}{data.actual_memory_percent:>12.1f}%{Colors.NC} │")
else:
print(f"│ {node_display:<19} │ {data.allocatable_cpu:>10.3f}G │ {data.requested_cpu:>10.3f}G │ "
f"{cpu_occupy_color}{data.cpu_occupy_percent:>10.1f}%{Colors.NC} │ "
f"{data.allocatable_memory:>10.2f}Gi │ {data.requested_memory:>10.2f}Gi │ "
f"{memory_occupy_color}{data.memory_occupy_percent:>11.1f}%{Colors.NC} │")
# 테이블 푸터
if self.metrics_available:
print(f"{Colors.BOLD}└─────────────────────┴─────────────┴─────────────┴─────────────┴─────────────┴─────────────┴─────────────┴─────────────┴─────────────┘{Colors.NC}")
else:
print(f"{Colors.BOLD}└─────────────────────┴─────────────┴─────────────┴─────────────┴─────────────┴─────────────┴─────────────┘{Colors.NC}")
def analyze_high_cpu_pods(self):
"""1G 이상 CPU 요청 Pod 분석"""
self.print_header("🔥 대용량 CPU 요청 Pod 분석 (1G 이상)")
# Pod 정보 가져오기 (JSON 형태)
pods_json = self._run_kubectl("get pods --all-namespaces -o json")
try:
pods_data = json.loads(pods_json)
self.high_cpu_pods = [] # 클래스 변수에 저장
for pod in pods_data.get('items', []):
namespace = pod['metadata']['namespace']
name = pod['metadata']['name']
node = pod['spec'].get('nodeName', 'Unscheduled')
# 컨테이너별 CPU 요청량 합계 계산
total_cpu_request = 0
total_memory_request = 0
for container in pod['spec'].get('containers', []):
resources = container.get('resources', {})
requests = resources.get('requests', {})
cpu_request = requests.get('cpu', '0')
memory_request = requests.get('memory', '0')
total_cpu_request += self._convert_cpu_to_giga(cpu_request)
total_memory_request += self._convert_memory_to_giga(memory_request)
# 1G 이상인 Pod만 수집
if total_cpu_request >= 1.0:
# 실제 사용량 가져오기 (kubectl top pod 사용)
actual_cpu = None
efficiency = None
if self.metrics_available and node != 'Unscheduled':
top_pod_output = self._run_kubectl(f"top pod {name} -n {namespace} --no-headers")
if top_pod_output.strip():
try:
parts = top_pod_output.strip().split()
if len(parts) >= 2:
cpu_usage = parts[1]
actual_cpu = self._convert_cpu_to_giga(cpu_usage)
if total_cpu_request > 0:
efficiency = (actual_cpu / total_cpu_request) * 100
except:
pass
self.high_cpu_pods.append(PodResourceInfo(
namespace=namespace,
name=name,
node=node,
cpu_request=total_cpu_request,
memory_request=total_memory_request,
actual_cpu=actual_cpu,
efficiency=efficiency
))
# 결과 출력
if self.high_cpu_pods:
print(f"{Colors.BOLD}┌─────────────────────┬─────────────────────────────────────┬─────────────┬─────────────┬─────────────┬──────────────────────────┐{Colors.NC}")
print(f"{Colors.BOLD}│ Node │ Pod Name │ CPU 요청(G) │ CPU 사용(G) │ 효율성 (%) │ 상태 │{Colors.NC}")
print(f"{Colors.BOLD}├─────────────────────┼─────────────────────────────────────┼─────────────┼─────────────┼─────────────┼──────────────────────────┤{Colors.NC}")
for pod in sorted(self.high_cpu_pods, key=lambda x: x.cpu_request, reverse=True):
node_display = pod.node[:19] if len(pod.node) > 19 else pod.node
pod_display = f"{pod.namespace}/{pod.name}"
if len(pod_display) > 35:
pod_display = pod_display[:32] + "..."
actual_cpu_str = f"{pod.actual_cpu:.3f}" if pod.actual_cpu is not None else "N/A"
if pod.efficiency is not None:
if pod.efficiency < 30:
status = f"{Colors.RED}과다 할당{Colors.NC}"
efficiency_color = Colors.RED
elif pod.efficiency < 60:
status = f"{Colors.YELLOW}비효율적{Colors.NC}"
efficiency_color = Colors.YELLOW
elif pod.efficiency > 90:
status = f"{Colors.PURPLE}최적 사용{Colors.NC}"
efficiency_color = Colors.PURPLE
else:
status = f"{Colors.GREEN}적정 수준{Colors.NC}"
efficiency_color = Colors.GREEN
efficiency_str = f"{efficiency_color}{pod.efficiency:.1f}{Colors.NC}"
else:
if pod.node == 'Unscheduled':
status = f"{Colors.RED}스케줄 대기{Colors.NC}"
else:
status = f"{Colors.GRAY}데이터 없음{Colors.NC}"
efficiency_str = f"{Colors.GRAY}N/A{Colors.NC}"
print(f"│ {node_display:<19} │ {pod_display:<35} │ {pod.cpu_request:>10.3f}G │ {actual_cpu_str:>10s}G │ {efficiency_str:>10s}% │ {status:<24} │")
print(f"{Colors.BOLD}└─────────────────────┴─────────────────────────────────────┴─────────────┴─────────────┴─────────────┴──────────────────────────┘{Colors.NC}")
else:
print(f"{Colors.GREEN}✅ 1G 이상 CPU를 요청하는 Pod가 없습니다{Colors.NC}")
except json.JSONDecodeError:
print(f"{Colors.RED}❌ Pod 정보를 가져오는데 실패했습니다{Colors.NC}")
def check_scheduling_issues(self):
"""스케줄링 문제 확인"""
self.print_header("🚫 스케줄링 이슈 분석")
# Pending Pod 확인
pending_output = self._run_kubectl("get pods --all-namespaces --field-selector status.phase=Pending --no-headers")
self.pending_pods = [] # 클래스 변수에 저장
if pending_output.strip():
print(f"{Colors.RED}스케줄링 실패한 Pod들:{Colors.NC}")
for line in pending_output.strip().split('\n'):
if line.strip():
parts = line.split()
if len(parts) >= 2:
pod_info = f"{parts[0]}/{parts[1]}"
self.pending_pods.append(pod_info)
print(f" {Colors.RED}• {pod_info}{Colors.NC} - {parts[3] if len(parts) > 3 else 'Pending'}")
else:
print(f"{Colors.GREEN}✅ 모든 Pod가 정상적으로 스케줄되었습니다{Colors.NC}")
# 최근 스케줄링 실패 이벤트
print(f"\n{Colors.BLUE}최근 스케줄링 실패 이벤트:{Colors.NC}")
events_output = self._run_kubectl("get events --all-namespaces --field-selector reason=FailedScheduling --sort-by='.lastTimestamp'")
if events_output.strip() and len(events_output.strip().split('\n')) > 1:
lines = events_output.strip().split('\n')
# 헤더 제외하고 최근 5개만 표시
recent_events = lines[-5:] if len(lines) > 5 else lines[1:]
for event in recent_events:
print(f" {Colors.YELLOW}{event}{Colors.NC}")
else:
print(f"{Colors.GREEN}✅ 최근 스케줄링 실패 이벤트가 없습니다{Colors.NC}")
def _generate_cluster_summary(self) -> ClusterSummary:
"""클러스터 전체 요약 정보 생성"""
total_allocatable_cpu = sum(data.allocatable_cpu for data in self.nodes_data.values())
total_allocatable_memory = sum(data.allocatable_memory for data in self.nodes_data.values())
total_requested_cpu = sum(data.requested_cpu for data in self.nodes_data.values())
total_requested_memory = sum(data.requested_memory for data in self.nodes_data.values())
avg_cpu_occupy = sum(data.cpu_occupy_percent for data in self.nodes_data.values()) / len(self.nodes_data) if self.nodes_data else 0
avg_memory_occupy = sum(data.memory_occupy_percent for data in self.nodes_data.values()) / len(self.nodes_data) if self.nodes_data else 0
high_usage_nodes = sum(1 for data in self.nodes_data.values()
if data.cpu_occupy_percent >= 80 or data.memory_occupy_percent >= 80)
return ClusterSummary(
total_nodes=len(self.nodes_data),
total_allocatable_cpu=total_allocatable_cpu,
total_allocatable_memory=total_allocatable_memory,
total_requested_cpu=total_requested_cpu,
total_requested_memory=total_requested_memory,
avg_cpu_occupy_percent=avg_cpu_occupy,
avg_memory_occupy_percent=avg_memory_occupy,
high_usage_nodes=high_usage_nodes,
pending_pods=len(self.pending_pods),
analysis_timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
)
def export_to_csv(self):
"""분석 결과를 CSV 파일로 출력"""
if not self.csv_output_dir:
return
# 출력 디렉토리 생성
if not os.path.exists(self.csv_output_dir):
os.makedirs(self.csv_output_dir)
print(f"\n{Colors.BLUE}📄 CSV 파일 생성 중...{Colors.NC}")
# 1. 노드 리소스 정보 CSV
nodes_csv_path = os.path.join(self.csv_output_dir, f"k8s_nodes_analysis_{self.analysis_timestamp}.csv")
with open(nodes_csv_path, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = [
'node_name', 'allocatable_cpu_g', 'allocatable_memory_gi',
'requested_cpu_g', 'requested_memory_gi',
'cpu_occupy_percent', 'memory_occupy_percent',
'actual_cpu_percent', 'actual_memory_percent',
'cpu_status', 'memory_status', 'overall_status'
]
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for node_name, data in self.nodes_data.items():
# 상태 결정
cpu_status = 'HIGH' if data.cpu_occupy_percent >= 80 else 'MEDIUM' if data.cpu_occupy_percent >= 60 else 'NORMAL'
memory_status = 'HIGH' if data.memory_occupy_percent >= 80 else 'MEDIUM' if data.memory_occupy_percent >= 60 else 'NORMAL'
overall_status = 'WARNING' if cpu_status == 'HIGH' or memory_status == 'HIGH' else 'CAUTION' if cpu_status == 'MEDIUM' or memory_status == 'MEDIUM' else 'HEALTHY'
writer.writerow({
'node_name': node_name,
'allocatable_cpu_g': f"{data.allocatable_cpu:.3f}",
'allocatable_memory_gi': f"{data.allocatable_memory:.2f}",
'requested_cpu_g': f"{data.requested_cpu:.3f}",
'requested_memory_gi': f"{data.requested_memory:.2f}",
'cpu_occupy_percent': f"{data.cpu_occupy_percent:.1f}",
'memory_occupy_percent': f"{data.memory_occupy_percent:.1f}",
'actual_cpu_percent': f"{data.actual_cpu_percent:.1f}" if data.actual_cpu_percent is not None else "N/A",
'actual_memory_percent': f"{data.actual_memory_percent:.1f}" if data.actual_memory_percent is not None else "N/A",
'cpu_status': cpu_status,
'memory_status': memory_status,
'overall_status': overall_status
})
print(f" {Colors.GREEN}✅ 노드 분석 결과: {nodes_csv_path}{Colors.NC}")
# 2. 대용량 CPU Pod 정보 CSV
if self.high_cpu_pods:
pods_csv_path = os.path.join(self.csv_output_dir, f"k8s_high_cpu_pods_{self.analysis_timestamp}.csv")
with open(pods_csv_path, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = [
'namespace', 'pod_name', 'node', 'cpu_request_g', 'memory_request_gi',
'actual_cpu_g', 'efficiency_percent', 'efficiency_status', 'scheduling_status'
]
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for pod in self.high_cpu_pods:
# 효율성 상태 결정
if pod.efficiency is not None:
if pod.efficiency < 30:
efficiency_status = 'OVER_ALLOCATED'
elif pod.efficiency < 60:
efficiency_status = 'INEFFICIENT'
elif pod.efficiency > 90:
efficiency_status = 'OPTIMAL'
else:
efficiency_status = 'GOOD'
else:
efficiency_status = 'NO_DATA'
scheduling_status = 'PENDING' if pod.node == 'Unscheduled' else 'RUNNING'
writer.writerow({
'namespace': pod.namespace,
'pod_name': pod.name,
'node': pod.node,
'cpu_request_g': f"{pod.cpu_request:.3f}",
'memory_request_gi': f"{pod.memory_request:.2f}",
'actual_cpu_g': f"{pod.actual_cpu:.3f}" if pod.actual_cpu is not None else "N/A",
'efficiency_percent': f"{pod.efficiency:.1f}" if pod.efficiency is not None else "N/A",
'efficiency_status': efficiency_status,
'scheduling_status': scheduling_status
})
print(f" {Colors.GREEN}✅ 대용량 CPU Pod 분석 결과: {pods_csv_path}{Colors.NC}")
# 3. 클러스터 요약 정보 CSV
summary = self._generate_cluster_summary()
summary_csv_path = os.path.join(self.csv_output_dir, f"k8s_cluster_summary_{self.analysis_timestamp}.csv")
with open(summary_csv_path, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = list(asdict(summary).keys())
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
writer.writerow(asdict(summary))
print(f" {Colors.GREEN}✅ 클러스터 요약 정보: {summary_csv_path}{Colors.NC}")
# 4. Pending Pod 목록 CSV (있는 경우에만)
if self.pending_pods:
pending_csv_path = os.path.join(self.csv_output_dir, f"k8s_pending_pods_{self.analysis_timestamp}.csv")
with open(pending_csv_path, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = ['namespace_pod', 'analysis_timestamp']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for pod_info in self.pending_pods:
writer.writerow({
'namespace_pod': pod_info,
'analysis_timestamp': summary.analysis_timestamp
})
print(f" {Colors.GREEN}✅ Pending Pod 목록: {pending_csv_path}{Colors.NC}")
print(f"\n{Colors.CYAN}📊 CSV 출력 완료! 총 {len([f for f in os.listdir(self.csv_output_dir) if f.endswith('.csv') and self.analysis_timestamp in f])}개 파일 생성{Colors.NC}")else 'Pending'}")
else:
print(f"{Colors.GREEN}✅ 모든 Pod가 정상적으로 스케줄되었습니다{Colors.NC}")
# 최근 스케줄링 실패 이벤트
print(f"\n{Colors.BLUE}최근 스케줄링 실패 이벤트:{Colors.NC}")
events_output = self._run_kubectl("get events --all-namespaces --field-selector reason=FailedScheduling --sort-by='.lastTimestamp'")
if events_output.strip() and len(events_output.strip().split('\n')) > 1:
lines = events_output.strip().split('\n')
# 헤더 제외하고 최근 5개만 표시
recent_events = lines[-5:] if len(lines) > 5 else lines[1:]
for event in recent_events:
print(f" {Colors.YELLOW}{event}{Colors.NC}")
else:
print(f"{Colors.GREEN}✅ 최근 스케줄링 실패 이벤트가 없습니다{Colors.NC}")
def display_alerts(self):
"""경고 알림 표시"""
print(f"\n{Colors.BOLD}🚨 상태 알림:{Colors.NC}")
warnings = []
for node_name, data in self.nodes_data.items():
if data.cpu_occupy_percent >= 80 or data.memory_occupy_percent >= 80:
warnings.append(f" {Colors.RED}⚠️ {node_name}: 높은 점유율 "
f"(CPU: {data.cpu_occupy_percent:.1f}%, Memory: {data.memory_occupy_percent:.1f}%) - 새 Pod 스케줄링 어려움{Colors.NC}")
if (self.metrics_available and data.actual_cpu_percent and data.actual_memory_percent and
(data.actual_cpu_percent >= 80 or data.actual_memory_percent >= 80)):
warnings.append(f" {Colors.YELLOW}🔥 {node_name}: 높은 실제 사용률 "
f"(CPU: {data.actual_cpu_percent:.1f}%, Memory: {data.actual_memory_percent:.1f}%) - 성능 저하 우려{Colors.NC}")
if warnings:
for warning in warnings:
print(warning)
else:
print(f" {Colors.GREEN}✅ 모든 노드의 리소스 상태가 정상 범위입니다{Colors.NC}")
def display_guide(self):
"""분석 가이드 표시"""
self.print_header("💡 분석 가이드 및 권장사항")
print(f"{Colors.BOLD}🎯 지표 설명:{Colors.NC}")
print(f" {Colors.CYAN}📊 점유율{Colors.NC}: Pod Request 기준 리소스 예약 비율 (스케줄링에 영향)")
print(f" {Colors.CYAN}🔥 사용률{Colors.NC}: 실제로 소비되고 있는 리소스 비율 (성능에 영향)")
print(f"\n{Colors.BOLD}🎯 색상 가이드:{Colors.NC}")
print(f" {Colors.GREEN}●{Colors.NC} 초록색: 정상 범위 (0-60%)")
print(f" {Colors.YELLOW}●{Colors.NC} 노란색: 주의 범위 (60-80%)")
print(f" {Colors.RED}●{Colors.NC} 빨간색: 위험 범위 (80%+)")
print(f"\n{Colors.BOLD}🛠️ 권장 조치사항:{Colors.NC}")
print(f" {Colors.PURPLE}•{Colors.NC} 점유율 80% 이상: 새로운 노드 추가 또는 Pod 재분배")
print(f" {Colors.PURPLE}•{Colors.NC} 사용률 80% 이상: CPU/메모리 최적화 또는 스케일업")
print(f" {Colors.PURPLE}•{Colors.NC} 효율성 30% 미만: Pod Request 값 다운사이징")
print(f" {Colors.PURPLE}•{Colors.NC} 효율성 90% 초과: Pod Request 값 업사이징 (안정성 확보)")
print(f"\n{Colors.BOLD}🔧 유용한 명령어:{Colors.NC}")
print(f" {Colors.GRAY}# 실시간 노드 사용률: kubectl top nodes{Colors.NC}")
print(f" {Colors.GRAY}# 실시간 Pod 사용률: kubectl top pods --all-namespaces{Colors.NC}")
print(f" {Colors.GRAY}# 노드 상세 정보: kubectl describe node <node-name>{Colors.NC}")
def run_analysis(self):
"""전체 분석 실행"""
self.print_header("🚀 KUBERNETES 리소스 분석기 v2.1 - Python Edition with CSV Export")
try:
# 1. 노드 분석
self.analyze_nodes()
# 2. 결과 표시
self.display_node_analysis()
# 3. 경고 알림
self.display_alerts()
# 4. 대용량 CPU Pod 분석
self.analyze_high_cpu_pods()
# 5. 스케줄링 이슈 확인
self.check_scheduling_issues()
# 6. CSV 출력 (옵션이 설정된 경우)
if self.csv_output_dir:
self.export_to_csv()
# 7. 가이드 표시
self.display_guide()
# 8. 완료 메시지
self.print_header("✅ 분석 완료 - Python v2.1")
print(f"{Colors.GREEN}Python 기반 안정적인 리소스 분석이 완료되었습니다! 🎯{Colors.NC}")
if self.csv_output_dir:
print(f"{Colors.CYAN}📊 CSV 파일들이 {self.csv_output_dir} 디렉토리에 저장되었습니다.{Colors.NC}")
print(f"{Colors.BLUE}📋 스케줄링 문제는 점유율을, 성능 문제는 사용률을, 최적화는 효율성을 확인하세요.{Colors.NC}")
except KeyboardInterrupt:
print(f"\n{Colors.YELLOW}분석이 중단되었습니다.{Colors.NC}")
sys.exit(1)
except Exception as e:
print(f"{Colors.RED}분석 중 오류가 발생했습니다: {e}{Colors.NC}")
sys.exit(1)def parse_arguments():
"""명령행 인수 파싱"""
parser = argparse.ArgumentParser(
description='Kubernetes 리소스 분석기 v2.1',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
사용 예시:
python3 k8s_analyzer.py # 콘솔 출력만
python3 k8s_analyzer.py --csv-output # 현재 디렉토리에 CSV 저장
python3 k8s_analyzer.py --csv-output ./reports # ./reports 디렉토리에 CSV 저장
python3 k8s_analyzer.py -c /tmp/k8s_reports # /tmp/k8s_reports 디렉토리에 CSV 저장
CSV 파일 종류:
• k8s_nodes_analysis_YYYYMMDD_HHMMSS.csv - 노드별 리소스 분석
• k8s_high_cpu_pods_YYYYMMDD_HHMMSS.csv - 1G 이상 CPU Pod 분석
• k8s_cluster_summary_YYYYMMDD_HHMMSS.csv - 클러스터 전체 요약
• k8s_pending_pods_YYYYMMDD_HHMMSS.csv - Pending Pod 목록 (있는 경우만)
"""
)
parser.add_argument(
'--csv-output', '-c',
nargs='?',
const='.',
help='CSV 파일을 저장할 디렉토리 (기본값: 현재 디렉토리)'
)
parser.add_argument(
'--version', '-v',
action='version',
version='Kubernetes 리소스 분석기 v2.1'
)
return parser.parse_args()def main():
"""메인 함수"""
# 명령행 인수 파싱
args = parse_arguments()
# Python 버전 체크
if sys.version_info < (3, 6):
print(f"{Colors.RED}Python 3.6 이상이 필요합니다. 현재 버전: {sys.version}{Colors.NC}")
sys.exit(1)
# kubectl 명령어 존재 여부 확인
result = subprocess.run("which kubectl", shell=True, capture_output=True)
if result.returncode != 0:
print(f"{Colors.RED}kubectl 명령어를 찾을 수 없습니다. Kubernetes CLI를 설치해주세요.{Colors.NC}")
sys.exit(1)
# 클러스터 연결 확인
result = subprocess.run("kubectl cluster-info", shell=True, capture_output=True)
if result.returncode != 0:
print(f"{Colors.RED}Kubernetes 클러스터에 연결할 수 없습니다. kubectl 설정을 확인해주세요.{Colors.NC}")
sys.exit(1)
# CSV 출력 디렉토리 검증
if args.csv_output:
try:
# 상대 경로를 절대 경로로 변환
csv_dir = os.path.abspath(args.csv_output)
print(f"{Colors.BLUE}📁 CSV 출력 디렉토리: {csv_dir}{Colors.NC}")
except Exception as e:
print(f"{Colors.RED}CSV 출력 디렉토리 설정 오류: {e}{Colors.NC}")
sys.exit(1)
else:
csv_dir = None
# 분석기 실행
print(f"{Colors.CYAN}🚀 Kubernetes 리소스 분석을 시작합니다...{Colors.NC}")
analyzer = K8sResourceAnalyzer(csv_output_dir=csv_dir)
analyzer.run_analysis()if name == "main":
main()
bytebliss