k8s_label_exporter.py

#!/usr/bin/env python3
"""
Kubernetes Node Label CSV Exporter
노드 레이블을 CSV 형태로 출력하는 프로그램
"""

import subprocess
import json
import csv
import sys
import argparse
from typing import Dict, List, Set
from io import StringIO

def run_kubectl_command(cmd: List[str]) -> str:
"""kubectl 명령어 실행"""
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
return result.stdout
except subprocess.CalledProcessError as e:
print(f"Error executing kubectl: {e}", file=sys.stderr)
print(f"Error output: {e.stderr}", file=sys.stderr)
sys.exit(1)
except FileNotFoundError:
print("Error: kubectl command not found. Please make sure kubectl is installed and in PATH.", file=sys.stderr)
sys.exit(1)

def get_node_labels() -> Dict[str, Dict[str, str]]:
"""모든 노드의 레이블 정보를 가져옴"""
cmd = ["kubectl", "get", "nodes", "-o", "json"]
output = run_kubectl_command(cmd)

try:
    nodes_data = json.loads(output)
except json.JSONDecodeError as e:
    print(f"Error parsing kubectl output: {e}", file=sys.stderr)
    sys.exit(1)

node_labels = {}
for node in nodes_data.get('items', []):
    node_name = node['metadata']['name']
    labels = node['metadata'].get('labels', {})
    node_labels[node_name] = labels

return node_labels

def export_pattern1(node_labels: Dict[str, Dict[str, str]], output_file: str = None, delimiter: str = '\t'):
"""패턴 1: Label 기준으로 출력 (Label -> host1, host2, ... 형태)"""
if not node_labels:
print("No nodes found.")
return

# 모든 레이블 키 수집
all_label_keys = set()
for labels in node_labels.values():
    all_label_keys.update(labels.keys())

all_label_keys = sorted(all_label_keys)
node_names = sorted(node_labels.keys())

# CSV 데이터 생성
output = StringIO()
writer = csv.writer(output, delimiter=delimiter)

# 헤더 작성
header = ['Label'] + node_names
writer.writerow(header)

# 각 레이블 키에 대해 행 작성
for label_key in all_label_keys:
    row = [label_key]
    for node_name in node_names:
        value = node_labels[node_name].get(label_key, '')
        row.append(value)
    writer.writerow(row)

# 출력
content = output.getvalue()
if output_file:
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write(content)
    print(f"Pattern 1 output saved to: {output_file}")
else:
    print("=== Pattern 1: Label 기준 출력 ===")
    print(content)

def export_pattern2(node_labels: Dict[str, Dict[str, str]], output_file: str = None, delimiter: str = '\t'):
"""패턴 2: Hostname 기준으로 레이블 존재 여부 표시"""
if not node_labels:
print("No nodes found.")
return

# 모든 레이블 키=값 조합 수집
all_label_pairs = set()
for labels in node_labels.values():
    for key, value in labels.items():
        all_label_pairs.add(f"{key}={value}")

all_label_pairs = sorted(all_label_pairs)
node_names = sorted(node_labels.keys())

# CSV 데이터 생성
output = StringIO()
writer = csv.writer(output, delimiter=delimiter)

# 헤더 작성
header = ['Labels'] + all_label_pairs
writer.writerow(header)

# 각 노드에 대해 행 작성
for node_name in node_names:
    row = [node_name]
    node_label_pairs = set()
    for key, value in node_labels[node_name].items():
        node_label_pairs.add(f"{key}={value}")
    
    for label_pair in all_label_pairs:
        if label_pair in node_label_pairs:
            row.append('O')  # 있음
        else:
            row.append('X')  # 없음
    writer.writerow(row)

# 출력
content = output.getvalue()
if output_file:
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write(content)
    print(f"Pattern 2 output saved to: {output_file}")
else:
    print("=== Pattern 2: Hostname 기준 레이블 존재 여부 ===")
    print(content)

def filter_labels(node_labels: Dict[str, Dict[str, str]], include_keys: List[str] = None, exclude_keys: List[str] = None) -> Dict[str, Dict[str, str]]:
"""레이블 필터링"""
if not include_keys and not exclude_keys:
return node_labels

filtered_labels = {}
for node_name, labels in node_labels.items():
    filtered_node_labels = {}
    for key, value in labels.items():
        # include_keys가 지정된 경우, 해당 키만 포함
        if include_keys and key not in include_keys:
            continue
        # exclude_keys가 지정된 경우, 해당 키 제외
        if exclude_keys and key in exclude_keys:
            continue
        filtered_node_labels[key] = value
    filtered_labels[node_name] = filtered_node_labels

return filtered_labels

def main():
parser = argparse.ArgumentParser(description='Export Kubernetes node labels to CSV format')
parser.add_argument('--pattern', type=int, choices=[1, 2], default=1,
help='Output pattern: 1=Label기준, 2=Hostname기준 (default: 1)')
parser.add_argument('--output', '-o', type=str,
help='Output file path (default: stdout)')
parser.add_argument('--delimiter', '-d', type=str, default='\t',
help='CSV delimiter (default: tab)')
parser.add_argument('--include-keys', type=str, nargs='+',
help='Include only specified label keys')
parser.add_argument('--exclude-keys', type=str, nargs='+',
help='Exclude specified label keys')
parser.add_argument('--both', action='store_true',
help='Export both patterns')

args = parser.parse_args()

print("Fetching node labels from Kubernetes cluster...")
node_labels = get_node_labels()

if not node_labels:
    print("No nodes found in the cluster.")
    return

print(f"Found {len(node_labels)} nodes.")

# 레이블 필터링
if args.include_keys or args.exclude_keys:
    node_labels = filter_labels(node_labels, args.include_keys, args.exclude_keys)

# 출력 실행
if args.both:
    # 두 패턴 모두 출력
    output_file1 = f"{args.output}_pattern1.csv" if args.output else None
    output_file2 = f"{args.output}_pattern2.csv" if args.output else None
    
    export_pattern1(node_labels, output_file1, args.delimiter)
    print()
    export_pattern2(node_labels, output_file2, args.delimiter)
elif args.pattern == 1:
    export_pattern1(node_labels, args.output, args.delimiter)
elif args.pattern == 2:
    export_pattern2(node_labels, args.output, args.delimiter)

if name == "main":
main()