─── 📂 경로 표준화 설정 ──────────────────────────────────────────────────

BASE_DATA_DIR = Path("./data") MERGED_DIR = BASE_DATA_DIR / "merged" OUTPUT_DIR = BASE_DATA_DIR / "output"

출력 디렉토리 자동 생성

OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

def parse_arguments(): parser = argparse.ArgumentParser(description="FinOps Governance & Anomaly Analyzer") # 💡 [교정] INTEGRATED/ALL 단계를 완전 폐기하고 타 스택과 완벽 대칭을 이루도록 대상 클러스터를 필수 요구 parser.add_argument("--cluster", type=str, required=True, choices=["COMPUTE", "STORAGE"], help="Target cluster type") return parser.parse_args()

def load_partitioned_data(cluster_target): """ 개편된 2단계 분할 정복 하위 폴더 위계를 스캔하여 일별 Parquet 파일들을 동적으로 결합합니다. """ # 💡 [교정] 타겟 클러스터 폴더 내부에 격리되어 저장된 일별 자산들만 정밀 타격 스캔 cl_dir = MERGED_DIR / cluster_target if not cl_dir.exists(): return pd.DataFrame(), pd.DataFrame(), pd.DataFrame()

pod_files    = glob.glob(str(cl_dir / "daily_enriched_*.parquet"))
ns_files     = glob.glob(str(cl_dir / "daily_ns_usage_*.parquet"))
pareto_files = glob.glob(str(cl_dir / "pareto_ns_*.parquet"))

print(f"🔍 [스캔 완료] {cluster_target} 파티션 하방 -> Enriched: {len(pod_files)}개 | NS요약: {len(ns_files)}개 | Pareto: {len(pareto_files)}개")

# 파일이 단 하나도 없을 경우의 빈 데이터프레임 예외 예방 가드레일
df_pod = pd.concat([pd.read_parquet(f) for f in pod_files], ignore_index=True) if pod_files else pd.DataFrame()
df_ns  = pd.concat([pd.read_parquet(f) for f in ns_files], ignore_index=True) if ns_files else pd.DataFrame()
df_pt  = pd.concat([pd.read_parquet(f) for f in pareto_files], ignore_index=True) if pareto_files else pd.DataFrame()

return df_pod, df_ns, df_pt

def main(): args = parse_arguments() cluster_target = args.cluster.upper() print(f"🚀 [Step4] Starting FinOps Governance Analyzer for Cluster: {cluster_target}...")

# 1. 쪼개져 있는 2단계 결과물 격리 로드
df_pod, df_ns, df_pt = load_partitioned_data(cluster_target)

if df_pod.empty:
    print(f"❌ [중단] '{cluster_target}' 클러스터에 정산된 2단계 가공 원부가 존재하지 않습니다. 앞단 배치를 확인하세요.")
    return

print(f"✅ 총 {len(df_pod):,}개의 컨테이너 타임라인 원부 통합 바인딩 완료.\n")

# 2. 🚨 [거버넌스 분석 레이어 1] 자원부족 및 OOM Killed 위험군 추출 (시트4 매핑용)
print("🚨 [Analysis 1] Extracting Critical Risk & OOM Killed Pods...")
df_risk = df_pod[
    (df_pod["status"] == "💥 OOM장애발생") | 
    (df_pod["status"] == "⚠️ Request부족") | 
    (df_pod["oom_strike_sum"] > 0)
].copy()

# 💡 [교정] 인덱스 미스매칭 크래시 방지를 위해 np.where 벡터화 구문으로 안전하게 조치 가이드 주입
if not df_risk.empty:
    df_risk["recommendation"] = np.where(
        df_risk["is_oom_killed"],
        "Upsize Memory Limit (OOM Detected)",
        "Upsize CPU Request/Limit (Throttling Detected)"
    )
else:
    df_risk["recommendation"] = None

# 3. 📉 [거버넌스 분석 레이어 2] 과다 할당 및 유휴 자산 추출 (시트2, 3 하향 후보용)
print("📉 [Analysis 2] Extracting Over-Allocated & Idle Infrastructure Assets...")
df_waste = df_pod[
    (df_pod["status"] == "📉 과다할당") & 
    (df_pod["cpu_waste_core_hours"] > 24)  # 하루 이상 코어 하나 통째로 낭비한 기준
].sort_values(by="cpu_waste_core_hours", ascending=False).copy()

# 4. 🛡️ [거버넌스 분석 레이어 3] 자원 미설정 배포 위반군 추출 (시트5 매핑용)
print("🛡️ [Analysis 3] Scanning Non-Compliant Missing Resource Specification Pods...")
df_violations = df_pod[
    (df_pod["has_no_request"] == True) | 
    (df_pod["has_no_limit"] == True)
].copy()

# 5. 💾 [결과 마감] 6단계 엑셀 빌더가 다이렉트로 인식할 수 있도록 산출물 저장
print(f"\n💾 [정산 마감] 분석 데이터 자산 output 레이어로 내보내기 진행 중...")

# 💡 [교정] 모놀리식 접미사를 전면 제거하고 명확한 클러스터 고유 타겟 식별자로 파일명 고정
master_gov_file = OUTPUT_DIR / f"governance_master_{cluster_target}.parquet"
risk_file       = OUTPUT_DIR / f"gov_risk_oom_{cluster_target}.parquet"
waste_file      = OUTPUT_DIR / f"gov_waste_candidates_{cluster_target}.parquet"
viol_file       = OUTPUT_DIR / f"gov_violations_{cluster_target}.parquet"

# Parquet 압축 저장
df_pod.to_parquet(master_gov_file, index=False)
df_risk.to_parquet(risk_file, index=False)
df_waste.to_parquet(waste_file, index=False)
df_violations.to_parquet(viol_file, index=False)

# 검증 서머리 출력
print(f"  -> 📦 [저장완료] 전사 {cluster_target} 마스터 원부 : {master_gov_file.name}")
print(f"  -> 💥 [저장완료] 고위험군/OOM 리스트    : {risk_file.name} (결과: {len(df_risk)}건)")
print(f"  -> 📉 [저장완료] 자원 하향조정 후보군  : {waste_file.name} (결과: {len(df_waste)}건)")
print(f"  -> 🛡️ [저장완료] 스펙 미설정 규격위반군: {viol_file.name} (결과: {len(df_violations)}건)")

print(f"\n🏁 === [Step4 완수] '{cluster_target}' 거버넌스 가공 원부가 무결하게 갱신되었습니다. ===")

if name == "main": main()

─── 🛡️ [Headless Environment Guard] Prevents crashes in K8s pods without display servers ───

import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import seaborn as sns

─── 📂 Directory Path Standardization (./data) ──────────────────────────

BASE_DATA_DIR = Path("./data") MERGED_DIR = BASE_DATA_DIR / "merged" BASE_PLOT_DIR = BASE_DATA_DIR / "output" / "plots"

Ensure base plot directory exists

BASE_PLOT_DIR.mkdir(parents=True, exist_ok=True)

─── 🔤 [No More Tofu] Use bulletproof built-in standard fonts ───

plt.rcParams['font.family'] = 'sans-serif' plt.rcParams['axes.unicode_minus'] = False # Prevents minus sign (-) corruption sns.set_theme(style="whitegrid")

─── 🛡️ [완치 가드레일] 빈 데이터셋 감지 시 플레이스홀더 PNG 생성 함수 ───

def check_and_handle_empty(df, output_path, chart_name): """ 데이터가 비어있으면 True를 반환하고, 6단계 엑셀 빌더가 터지지 않도록 'No Data Available' 문구가 적힌 플레이스홀더 이미지를 강제 생성합니다. """ if df is None or df.empty: print(f" ⚠️ [Empty Data] Generating placeholder for {chart_name} (Dataset is empty).") fig, ax = plt.subplots(figsize=(10, 5)) ax.text(0.5, 0.5, f"No Data Available\n({chart_name})", ha='center', va='center', fontsize=14, color='gray', weight='bold') ax.axis('off') plt.tight_layout() plt.savefig(output_path, dpi=100, bbox_inches='tight') plt.close() return True return False

def main(): parser = argparse.ArgumentParser(description="FinOps Analytics Visualization Engine") # 💡 [교정] INTEGRATED/ALL 단계를 배제하고 6단계와 완벽히 대칭되도록 명확한 대상 클러스터를 필수로 요구 parser.add_argument("--cluster", type=str, required=True, choices=["COMPUTE", "STORAGE"], help="Target cluster type") args = parser.parse_args()

cluster_target = args.cluster.upper()
print(f"🚀 [Step3] Starting FinOps Time-series English Visualization Engine for Cluster: {cluster_target}...")

# ─── 🗂️ 클러스터 격리 디렉토리 매핑 및 가변 결합 (분할 정복) ───
cl_dir = MERGED_DIR / cluster_target

if not cl_dir.exists():
    print(f"❌ Error: Target cluster directory '{cl_dir}' does not exist. Please run step2 first.")
    return

pod_files = list(cl_dir.glob("daily_enriched_*.parquet"))
ns_files  = list(cl_dir.glob("pareto_ns_*.parquet"))

if not pod_files:
    print(f"❌ Error: No processed daily Parquet files found for cluster '{cluster_target}' under {cl_dir}. Please run step2 first.")
    return

df_pod = pd.concat([pd.read_parquet(f) for f in pod_files], ignore_index=True)
df_ns  = pd.concat([pd.read_parquet(f) for f in ns_files], ignore_index=True) if ns_files else pd.DataFrame()

# 💡 [교정] 출력 경로 역시 INTEGRATED 우회 레이어 없이 해당 클러스터 폴더명으로 완전 격리 분리
PLOT_DIR = BASE_PLOT_DIR / cluster_target
PLOT_DIR.mkdir(parents=True, exist_ok=True)

print(f"✅ Data lake aggregated successfully -> Container rows: {len(df_pod):,}")

# 데이터 정렬 오염 차단용 형변환 Guard
df_pod["date"] = df_pod["date"].astype(str)

# Injecting efficiency metric layers
df_pod["cpu_util"] = np.where(df_pod["cpu_request_max"] > 0, (df_pod["cpu_usage_p95"] / df_pod["cpu_request_max"] * 100), 0)
df_pod["mem_util"] = np.where(df_pod["mem_request_max"] > 0, (df_pod["mem_usage_p95"] / df_pod["mem_request_max"] * 100), 0)
df_pod["lim_req_ratio"] = np.where(df_pod["cpu_request_max"] > 0, df_pod["cpu_limit_max"] / df_pod["cpu_request_max"], 0)

# ─── 📊 [Charts 1 & 2] Workload Type Allocated vs Actual Peak ───
print("⏳ [1/19] Rendering chart1_cpu_req_vs_usage_by_workload...")
out1 = PLOT_DIR / "chart1_cpu_req_vs_usage_by_workload.png"
if not check_and_handle_empty(df_pod, out1, "chart1_cpu_req_vs_usage_by_workload"):
    df_wl_cpu = df_pod.groupby("workload_type")[["cpu_request_max", "cpu_usage_p95"]].mean().reset_index()
    plt.figure(figsize=(10, 5))
    df_melt_cpu = df_wl_cpu.melt(id_vars="workload_type", value_vars=["cpu_request_max", "cpu_usage_p95"])
    sns.barplot(data=df_melt_cpu, x="workload_type", y="value", hue="variable", palette="Blues_r")
    plt.xticks(rotation=30, ha='right')
    plt.title("Average CPU Request vs P95 Peak Usage by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("CPU Cores")
    plt.tight_layout()
    plt.savefig(out1, dpi=100)
    plt.close()

print("⏳ [2/19] Rendering chart2_mem_req_vs_usage_by_workload...")
out2 = PLOT_DIR / "chart2_mem_req_vs_usage_by_workload.png"
if not check_and_handle_empty(df_pod, out2, "chart2_mem_req_vs_usage_by_workload"):
    df_wl_mem = df_pod.groupby("workload_type")[["mem_request_max", "mem_usage_p95"]].mean().reset_index()
    plt.figure(figsize=(10, 5))
    df_melt_mem = df_wl_mem.melt(id_vars="workload_type", value_vars=["mem_request_max", "mem_usage_p95"])
    sns.barplot(data=df_melt_mem, x="workload_type", y="value", hue="variable", palette="Purples_r")
    plt.xticks(rotation=30, ha='right')
    plt.title("Average Memory Request vs P95 Peak Usage (GB) by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("Memory Capacity (GB)")
    plt.tight_layout()
    plt.savefig(out2, dpi=100)
    plt.close()

# ─── 📊 [Chart 3] Daily CPU Waste Stacked Bar ───
print("⏳ [3/19] Rendering chart3_daily_waste_stack...")
out3 = PLOT_DIR / "chart3_daily_waste_stack.png"
df_daily_waste = df_pod.groupby(["date", "workload_type"])["cpu_waste_core_hours"].sum().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_daily_waste, out3, "chart3_daily_waste_stack"):
    df_daily_waste.plot(kind='bar', stacked=True, figsize=(11, 5), cmap="tab20")
    plt.title("Daily Total CPU Waste Core-Hours Stacked by Workload (KST)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Waste Volume (Core-Hours)")
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig(out3, dpi=100)
    plt.close()

# ─── 📊 [Charts 4 & 18] Efficiency Heatmaps ───
print("⏳ [4/19] Rendering chart4_cpu_efficiency_heatmap...")
out4 = PLOT_DIR / "chart4_cpu_efficiency_heatmap.png"
df_heat_cpu = df_pod.groupby(["workload_type", "date"])["cpu_util"].mean().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_heat_cpu, out4, "chart4_cpu_efficiency_heatmap"):
    plt.figure(figsize=(10, 5))
    sns.heatmap(df_heat_cpu, annot=True, fmt=".1f", cmap="RdYlGn", cbar=True)
    plt.title("Mean CPU Utilization Heatmap (%) (Workload Type x Date)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Workload Type")
    plt.tight_layout()
    plt.savefig(out4, dpi=100)
    plt.close()

print("⏳ [5/19] Rendering chart18_mem_waste_heatmap...")
out18 = PLOT_DIR / "chart18_mem_waste_heatmap.png"
df_heat_mem = df_pod.groupby(["workload_type", "date"])["mem_waste_gb_hours"].sum().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_heat_mem, out18, "chart18_mem_waste_heatmap"):
    plt.figure(figsize=(10, 5))
    sns.heatmap(df_heat_mem, annot=False, cmap="BuPu", cbar=True)
    plt.title("Total Memory Waste Volume Heatmap (GB-Hours)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Workload Type")
    plt.tight_layout()
    plt.savefig(out18, dpi=100)
    plt.close()

# ─── 📊 [Charts 5 & 11] Pareto Governance Charts ───
print("⏳ [6/19] Rendering chart5_pareto_ns_waste...")
out5 = PLOT_DIR / "chart5_pareto_ns_waste.png"
if not check_and_handle_empty(df_ns, out5, "chart5_pareto_ns_waste"):
    df_ns_top = df_ns.groupby("namespace")["total_waste_core_hours"].sum().reset_index().sort_values("total_waste_core_hours", ascending=False).head(15)
    df_ns_top["waste_share_pct"] = (df_ns_top["total_waste_core_hours"] / df_ns_top["total_waste_core_hours"].sum() * 100).round(2)
    df_ns_top["waste_cumsum_pct"] = df_ns_top["waste_share_pct"].cumsum().round(2)

    fig, ax1 = plt.subplots(figsize=(11, 5))
    sns.barplot(data=df_ns_top, x="namespace", y="total_waste_core_hours", ax=ax1, color="steelblue")
    ax1.set_xticklabels(ax1.get_xticklabels(), rotation=45, ha="right")
    ax2 = ax1.twinx()
    ax2.plot(df_ns_top["namespace"], df_ns_top["waste_cumsum_pct"], color="crimson", marker="o", linewidth=2)
    ax2.set_ylim(0, 110)
    plt.title("Top 15 Namespace Cost-Waste Pareto Chart (Cumulative Share %)")
    ax1.set_xlabel("Tenant Namespace")
    ax1.set_ylabel("Waste Volume (Core-Hours)")
    plt.tight_layout()
    plt.savefig(out5, dpi=100)
    plt.close()

print("⏳ [7/19] Rendering chart11_pareto_workload_waste...")
out11 = PLOT_DIR / "chart11_pareto_workload_waste.png"
if not check_and_handle_empty(df_pod, out11, "chart11_pareto_workload_waste"):
    df_wl_waste = df_pod.groupby("workload_type")["cpu_waste_core_hours"].sum().reset_index().sort_values("cpu_waste_core_hours", ascending=False)
    plt.figure(figsize=(10, 5))
    sns.barplot(data=df_wl_waste, x="workload_type", y="cpu_waste_core_hours", palette="Oranges_r")
    plt.xticks(rotation=30, ha="right")
    plt.title("Total CPU Waste Volume by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("Total Waste (Core-Hours)")
    plt.tight_layout()
    plt.savefig(out11, dpi=100)
    plt.close()

# ─── 📊 [Chart 6] Governance Status Donut Chart ───
print("⏳ [8/19] Rendering chart6_status_donut...")
out6 = PLOT_DIR / "chart6_status_donut.png"
if not check_and_handle_empty(df_pod, out6, "chart6_status_donut"):
    status_summary = df_pod["status"].value_counts()
    plt.figure(figsize=(6, 5))
    colors = ["#70AD47", "#1F4E79", "#FFC000", "#C00000"]
    plt.pie(status_summary, labels=status_summary.index, autopct='%1.1f%%', startangle=90, colors=colors[:len(status_summary)], wedgeprops=dict(width=0.4, edgecolor='w'))
    plt.title("Infrastructure Resource Governance Status Ratio")
    plt.tight_layout()
    plt.savefig(out6, dpi=100)
    plt.close()

# ─── 📊 [Charts 7 & 14] Bubble & Scatter Plots ───
print("⏳ [9/19] Rendering chart7_waste_footprint_bubble...")
out7 = PLOT_DIR / "chart7_waste_footprint_bubble.png"
if not check_and_handle_empty(df_pod, out7, "chart7_waste_footprint_bubble"):
    df_bubble = df_pod.groupby("workload_type").agg(
        x_alloc=("cpu_allocated_core_hours", "sum"),
        y_util=("cpu_util", "mean"),
        z_waste=("cpu_waste_core_hours", "sum")
    ).reset_index()
    plt.figure(figsize=(9, 6))
    sns.scatterplot(data=df_bubble, x="x_alloc", y="y_util", size="z_waste", hue="workload_type", sizes=(100, 2000), alpha=0.7, legend="brief")
    plt.title("Resource Footprint Bubble Chart (Allocated x Utilization x Waste Size)")
    plt.xlabel("Total Allocated Core-Hours")
    plt.ylabel("Average Utilization (%)")
    plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
    plt.tight_layout()
    plt.savefig(out7, dpi=100)
    plt.close()

print("⏳ [10/19] Rendering chart14_cpu_mem_waste_scatter...")
out14 = PLOT_DIR / "chart14_cpu_mem_waste_scatter.png"
if not check_and_handle_empty(df_pod, out14, "chart14_cpu_mem_waste_scatter"):
    plt.figure(figsize=(8, 6))
    sns.scatterplot(data=df_pod.head(5000), x="cpu_waste_core_hours", y="mem_waste_gb_hours", hue="workload_type", alpha=0.5)
    plt.title("Co-relation Scatter Plot: CPU Waste vs Memory Waste (Sampled)")
    plt.xlabel("CPU Waste (Core-Hours)")
    plt.ylabel("Memory Waste (GB-Hours)")
    plt.tight_layout()
    plt.savefig(out14, dpi=100)
    plt.close()

# ─── 📊 [Chart 8] Resource Shortage Footprint ───
print("⏳ [11/19] Rendering chart8_shortfall_footprint...")
out8 = PLOT_DIR / "chart8_shortfall_footprint.png"
df_short = df_pod.groupby(["workload_type", "date"])["cpu_shortage_cores"].sum().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_short, out8, "chart8_shortfall_footprint"):
    plt.figure(figsize=(10, 4))
    sns.heatmap(df_short, annot=False, cmap="YlOrRd", cbar=True)
    plt.title("Total CPU Shortfall (Deficit) Cores Footprint")
    plt.xlabel("Date (KST)")
    plt.ylabel("Workload Type")
    plt.tight_layout()
    plt.savefig(out8, dpi=100)
    plt.close()

# ─── 📊 [Charts 9 & 10] Boxplot Distribution Analysis ───
print("⏳ [12/19] Rendering chart9_boxplot_cpu_util_by_workload...")
out9 = PLOT_DIR / "chart9_boxplot_cpu_util_by_workload.png"
if not check_and_handle_empty(df_pod, out9, "chart9_boxplot_cpu_util_by_workload"):
    plt.figure(figsize=(11, 5))
    sns.boxplot(data=df_pod, x="workload_type", y="cpu_util", palette="Set3")
    plt.xticks(rotation=30, ha="right")
    plt.ylim(-5, 105)
    plt.title("CPU Utilization P95 Boxplot Distribution by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("P95 Actual Utilization (%)")
    plt.tight_layout()
    plt.savefig(out9, dpi=100)
    plt.close()

print("⏳ [13/19] Rendering chart10_boxplot_mem_util_by_workload...")
out10 = PLOT_DIR / "chart10_boxplot_mem_util_by_workload.png"
if not check_and_handle_empty(df_pod, out10, "chart10_boxplot_mem_util_by_workload"):
    plt.figure(figsize=(11, 5))
    sns.boxplot(data=df_pod, x="workload_type", y="mem_util", palette="Pastel1")
    plt.xticks(rotation=30, ha="right")
    plt.ylim(-5, 105)
    plt.title("Memory Utilization P95 Boxplot Distribution by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("P95 Actual Utilization (%)")
    plt.tight_layout()
    plt.savefig(out10, dpi=100)
    plt.close()

# ─── 📊 [Charts 12 & 15] Trend and Counts ───
print("⏳ [14/19] Rendering chart12_daily_waste_trend_by_workload...")
out12 = PLOT_DIR / "chart12_daily_waste_trend_by_workload.png"
df_trend_wl = df_pod.groupby(["date", "workload_type"])["cpu_waste_core_hours"].sum().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_trend_wl, out12, "chart12_daily_waste_trend_by_workload"):
    plt.figure(figsize=(11, 5))
    sns.lineplot(data=df_trend_wl, markers=True, dashes=False, linewidth=2)
    plt.title("Daily CPU Waste Timeline Trend Line by Workload Type")
    plt.xlabel("Date (KST)")
    plt.ylabel("Waste Volume (Core-Hours)")
    plt.xticks(rotation=30)
    plt.tight_layout()
    plt.savefig(out12, dpi=100)
    plt.close()

print("⏳ [15/19] Rendering chart15_oom_status_by_workload...")
out15 = PLOT_DIR / "chart15_oom_status_by_workload.png"
if not check_and_handle_empty(df_pod, out15, "chart15_oom_status_by_workload"):
    plt.figure(figsize=(11, 5))
    sns.countplot(data=df_pod, x="workload_type", hue="status", palette="muted")
    plt.xticks(rotation=30, ha="right")
    plt.title("Governance Status Distribution Count per Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("Pod Count")
    plt.legend(loc="upper right")
    plt.tight_layout()
    plt.savefig(out15, dpi=100)
    plt.close()

# ─── 📊 [Charts 13 & 17] Violin and Overcommit Boxplot ───
print("⏳ [16/19] Rendering chart13_violin_cpu_util...")
out13 = PLOT_DIR / "chart13_violin_cpu_util.png"
if not check_and_handle_empty(df_pod, out13, "chart13_violin_cpu_util"):
    plt.figure(figsize=(10, 5))
    sns.violinplot(data=df_pod, x="workload_type", y="cpu_util", inner="quartile", palette="pastel")
    plt.xticks(rotation=30, ha="right")
    plt.title("CPU Utilization Kernel Density Violin Plot")
    plt.xlabel("Workload Type")
    plt.ylabel("CPU Utilization (%)")
    plt.tight_layout()
    plt.savefig(out13, dpi=100)
    plt.close()

print("⏳ [17/19] Rendering chart17_cpu_limit_request_ratio...")
out17 = PLOT_DIR / "chart17_cpu_limit_request_ratio.png"
if not check_and_handle_empty(df_pod, out17, "chart17_cpu_limit_request_ratio"):
    plt.figure(figsize=(10, 5))
    sns.boxplot(data=df_pod, x="workload_type", y="lim_req_ratio", palette="vlag")
    plt.xticks(rotation=30, ha="right")
    plt.title("Kubernetes Pod CPU Limit / Request Overcommit Ratio")
    plt.xlabel("Workload Type")
    plt.ylabel("Limit / Request Ratio")
    plt.tight_layout()
    plt.savefig(out17, dpi=100)
    plt.close()

# ─── 📊 [Charts 19 & 20] Capacity vs Actual Trends ───
print("⏳ [18/19] Rendering chart19_daily_cpu_per_workload...")
out19 = PLOT_DIR / "chart19_daily_cpu_per_workload.png"
if not check_and_handle_empty(df_pod, out19, "chart19_daily_cpu_per_workload"):
    df_daily_cpu_req = df_pod.groupby("date")["cpu_request_max"].sum()
    df_daily_cpu_use = df_pod.groupby("date")["cpu_usage_p95"].sum()
    plt.figure(figsize=(11, 5))
    plt.fill_between(df_daily_cpu_req.index, df_daily_cpu_req.values, label="Total CPU Request Cores", color="skyblue", alpha=0.4)
    plt.plot(df_daily_cpu_use.index, df_daily_cpu_use.values, label="Total CPU P95 Actual Cores", color="navy", linewidth=2.5, marker="o")
    plt.title("Daily Total CPU Capacity Allocation vs Actual Peak Consumption (KST)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Total CPU Cores")
    plt.xticks(rotation=30)
    plt.legend(loc="upper left")
    plt.tight_layout()
    plt.savefig(out19, dpi=100)
    plt.close()

print("⏳ [19/19] Rendering chart20_daily_mem_per_workload...")
out20 = PLOT_DIR / "chart20_daily_mem_per_workload.png"
if not check_and_handle_empty(df_pod, out20, "chart20_daily_mem_per_workload"):
    df_daily_mem_req = df_pod.groupby("date")["mem_request_max"].sum()
    df_daily_mem_use = df_pod.groupby("date")["mem_usage_p95"].sum()
    plt.figure(figsize=(11, 5))
    plt.fill_between(df_daily_mem_req.index, df_daily_mem_req.values, label="Total Memory Request (GB)", color="plum", alpha=0.4)
    plt.plot(df_daily_mem_use.index, df_daily_mem_use.values, label="Total Memory P95 Actual (GB)", color="purple", linewidth=2.5, marker="o")
    plt.title("Daily Total Memory Capacity Allocation vs Actual Peak Consumption (KST)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Total Memory (GB)")
    plt.xticks(rotation=30)
    plt.legend(loc="upper left")
    plt.tight_layout()
    plt.savefig(out20, dpi=100)
    plt.close()

print(f"\n🏁 === [Step3 Success] All charts generated cleanly under ./data/output/plots/{cluster_target} ===")

if name == "main": main()

import config from config import ( RAW_DIR, MERGED_DIR, CLUSTER_NODE_PATTERNS, get_workload_type, classify_node_infrastructure )

KST = timezone(timedelta(hours=9)) W_CPU, W_MEM, W_PV = 1.0, 0.11, 0.02

def parse_arguments(): parser = argparse.ArgumentParser(description="FinOps Polars-Powered Reprocessing Engine") parser.add_argument("--days", type=int, default=None) parser.add_argument("--start-date", type=str, default=None) parser.add_argument("--end-date", type=str, default=None) parser.add_argument("--cluster", type=str, default="ALL", choices=["COMPUTE", "STORAGE", "ALL"]) parser.add_argument("--workload-domain", type=str, default=None) return parser.parse_args()

def get_minio_client(): return boto3.client( "s3", endpoint_url=os.getenv("MINIO_ENDPOINT"), aws_access_key_id=os.getenv("MINIO_ACCESS_KEY"), aws_secret_access_key=os.getenv("MINIO_SECRET_KEY"), region_name="us-east-1", config=boto3.session.Config(signature_version="s3v4") )

def sync_down_raw_from_minio(s3_client, bucket_name, start_dt, end_dt): chunk_delta = timedelta(hours=1) current_start = start_dt print(f"\n📡 [AIStor 스마트 캐시 엔진] 원천 데이터 가용성 체크 시작...")

while current_start < end_dt:
    chunk_str = current_start.strftime("%Y%m%d_%H")
    f_name = f"prom_raw_{chunk_str}.parquet"
    object_key = f"raw/{f_name}"
    local_path = RAW_DIR / f_name

    if local_path.exists() and local_path.stat().st_size > 0:
        print(f"   ⚡ [Cache Hit] 로컬 캐시 자산 활용 (다운로드 스킵): {f_name}")
    else:
        try:
            s3_client.head_object(Bucket=bucket_name, Key=object_key)
            print(f"   📥 [Cache Miss] AIStor 백업으로부터 다운로드: {object_key}")
            s3_client.download_file(bucket_name, object_key, str(local_path))
        except ClientError as e:
            if e.response['Error']['Code'] != "404":
                print(f"   ❌ AIStor 통신 실패 에러 ({f_name}): {str(e)}")
        except Exception as e:
            print(f"   ⚠️ 일반 다운로드 예외 발생 ({f_name}): {str(e)}")

    current_start += chunk_delta

def main(): args = parse_arguments() s3_client = get_minio_client() bucket_name = os.getenv("MINIO_RAW_BUCKET", "devops-test") now_kst = datetime.now(KST)

if args.start_date and args.end_date:
    start_dt = datetime.strptime(args.start_date, "%Y-%m-%d").replace(tzinfo=KST)
    end_dt = (datetime.strptime(args.end_date, "%Y-%m-%d") + timedelta(days=1)).replace(tzinfo=KST)
elif args.days:
    start_dt = (now_kst - timedelta(days=args.days)).replace(hour=0, minute=0, second=0, microsecond=0)
    end_dt = now_kst
else:
    yesterday = (now_kst - timedelta(days=1)).date()
    start_dt = datetime.combine(yesterday, datetime.min.time()).replace(tzinfo=KST)
    end_dt = datetime.combine(now_kst.date(), datetime.min.time()).replace(tzinfo=KST)

sync_down_raw_from_minio(s3_client, bucket_name, start_dt, end_dt)

raw_files = list(RAW_DIR.glob("prom_raw_*.parquet"))
if not raw_files:
    print("\n🚨 [중단] 정산 가동할 물리 원천 파일이 존재하지 않습니다.")
    return

print(f"\n📊 [Polars Lazy Engine 기동] {len(raw_files)}개 소스 파티션 스캔 링킹...")
lf_raw = pl.scan_parquet(str(RAW_DIR / "prom_raw_*.parquet"))

start_naive = start_dt.replace(tzinfo=None)
end_naive = end_dt.replace(tzinfo=None)

lf_filtered = lf_raw.filter(
    (pl.col("timestamp") >= start_naive) & (pl.col("timestamp") < end_naive)
).with_columns(
    pl.col("timestamp").dt.strftime("%Y%m%d").alias("date")
)

print("⏳ 대용량 인프라 정산 메트릭 다차원 분산 집계 중 (Rust Core 스레드 가동)...")
df_raw = lf_filtered.collect()

if df_raw.is_empty():
    print("⚠️  해당 기간 내 정산할 데이터 로우가 없습니다.")
    return

df_raw = df_raw.with_columns([
    pl.col("node").map_elements(
        lambda x: {"cluster_type": classify_node_infrastructure(x)[0], "workload_domain": classify_node_infrastructure(x)[1]},
        return_dtype=pl.Struct({"cluster_type": pl.String, "workload_domain": pl.String})
    ).alias("infra_struct"),
    pl.col("pod").map_elements(get_workload_type, return_dtype=pl.String).alias("workload_type")
]).with_columns([
    pl.col("infra_struct").struct.field("cluster_type").alias("cluster_type"),
    pl.col("infra_struct").struct.field("workload_domain").alias("workload_domain")
]).drop("infra_struct")

target_dates = df_raw["date"].unique().to_list()
target_clusters = ["COMPUTE", "STORAGE"] if args.cluster.upper() == "ALL" else [args.cluster.upper()]

print(f"🎯 분할 연산 타겟 스케줄러 가동 -> 일자 풀: {target_dates} | 클러스터 풀: {target_clusters}")

GB_DIV = 1024 ** 3

for date_chunk in target_dates:
    for cluster_chunk in target_clusters:

        df_slice = df_raw.filter((pl.col("date") == date_chunk) & (pl.col("cluster_type") == cluster_chunk))
        if df_slice.is_empty():
            continue

        if args.workload_domain:
            df_slice = df_slice.filter(pl.col("workload_domain") == args.workload_domain)
            if df_slice.is_empty(): continue

        print(f"   ⏳ [Polars 집계 연산] {date_chunk} ➡️ {cluster_chunk} 고성능 그룹바이 컴파일...")

        target_fields = ["cpu_request", "cpu_limit", "cpu_usage", "cpu_throttled", "mem_request", "mem_limit", "mem_usage", "mem_rss", "oom_event", "pv_capacity", "pv_used"]
        for col in target_fields:
            if col not in df_slice.columns:
                df_slice = df_slice.with_columns(pl.lit(0.0).alias(col))

        # [테이블 1] 상세 메트릭 원부 집계
        df_pod = df_slice.group_by([
            "date", "cluster_type", "workload_domain", "namespace", "workload_type", "node", "pod", "container"
        ]).agg([
            pl.len().alias("minutes_running"),
            pl.col("cpu_request").max().alias("cpu_request_max"),
            pl.col("cpu_limit").max().alias("cpu_limit_max"),
            pl.col("cpu_usage").quantile(0.95).alias("cpu_usage_p95"),
            pl.col("cpu_throttled").max().alias("cpu_throttled_max"),
            (pl.col("mem_request").max() / GB_DIV).alias("mem_request_max"),
            (pl.col("mem_limit").max() / GB_DIV).alias("mem_limit_max"),
            (pl.col("mem_usage").quantile(0.95) / GB_DIV).alias("mem_usage_p95"),
            (pl.col("mem_rss").quantile(0.95) / GB_DIV).alias("mem_rss_p95"),
            pl.col("oom_event").sum().alias("oom_strike_sum"),
            (pl.col("pv_capacity").max() / GB_DIV).alias("pv_capacity_max"),

            # 💡 [교정] pv_used_p95 대신 표준 네이밍 규약인 pv_usage_p95 로 엘리어싱하여 하방 ColumnNotFoundError 완벽 차단
            (pl.col("pv_used").quantile(0.95) / GB_DIV).alias("pv_usage_p95")
        ]).fill_null(0.0)

        df_pod = df_pod.with_columns([
            (pl.col("cpu_request_max") * (pl.col("minutes_running") / 60.0)).alias("cpu_allocated_core_hours"),
            (pl.col("cpu_usage_p95") * (pl.col("minutes_running") / 60.0)).alias("cpu_usage_core_hours"),
            (pl.col("mem_request_max") * (pl.col("minutes_running") / 60.0)).alias("mem_allocated_gb_hours"),
            (pl.col("mem_usage_p95") * (pl.col("minutes_running") / 60.0)).alias("mem_usage_gb_hours"),
            (pl.col("pv_capacity_max") * (pl.col("minutes_running") / 60.0)).alias("pv_allocated_gb_hours"),
            (pl.col("pv_usage_p95") * (pl.col("minutes_running") / 60.0)).alias("pv_usage_gb_hours"),
            (pl.col("oom_strike_sum") > 0).alias("is_oom_killed"),
            (pl.col("cpu_request_max") == 0).alias("has_no_request"),
            (pl.col("cpu_limit_max") == 0).alias("has_no_limit"),
            ((pl.col("cpu_usage_p95") - pl.col("cpu_request_max")).clip(lower_bound=0)).alias("cpu_shortage_cores")
        ]).with_columns([
            ((pl.col("cpu_allocated_core_hours") - pl.col("cpu_usage_core_hours")).clip(lower_bound=0)).alias("cpu_waste_core_hours"),
            ((pl.col("mem_allocated_gb_hours") - pl.col("mem_usage_gb_hours")).clip(lower_bound=0)).alias("mem_waste_gb_hours"),
            ((pl.col("pv_allocated_gb_hours") - pl.col("pv_usage_gb_hours")).clip(lower_bound=0)).alias("pv_waste_gb_hours"),
        ])

        df_pod = df_pod.with_columns(
            pl.when(pl.col("is_oom_killed")).then(pl.lit("💥 OOM장애발생"))
            .when((pl.col("cpu_shortage_cores") > 0.5) | (pl.col("cpu_throttled_max") > 0.2)).then(pl.lit("⚠️ Request부족"))
            .when((pl.col("cpu_waste_core_hours") > 10) | (pl.col("pv_waste_gb_hours") > 50)).then(pl.lit("📉 과다할당"))
            .otherwise(pl.lit("✅ 최적화완료")).alias("status")
        )

        # 📅 [테이블 2] Namespace 사용량 요약 빌드
        df_daily_ns = df_pod.group_by(["date", "namespace"]).agg([
            pl.col("cpu_usage_core_hours").sum().alias("cpu_used_ch"),
            pl.col("cpu_allocated_core_hours").sum().alias("cpu_alloc_ch"),
            pl.col("cpu_waste_core_hours").sum().alias("cpu_waste_ch"),
            pl.col("mem_usage_gb_hours").sum().alias("mem_used_gh"),
            pl.col("mem_allocated_gb_hours").sum().alias("mem_alloc_gh"),
            pl.col("pv_usage_gb_hours").sum().alias("pv_used_gh"),
            pl.col("pv_allocated_gb_hours").sum().alias("pv_alloc_gh")
        ]).with_columns(
            ((pl.col("cpu_alloc_ch") * W_CPU) + (pl.col("mem_alloc_gh") * W_MEM) + (pl.col("pv_alloc_gh") * W_PV)).round(1).alias("final_usage_score")
        )

        # 📉 [테이블 3] 파레토 원부 롤업 빌드 
        df_ns = df_pod.group_by("namespace").agg([
            pl.col("cpu_waste_core_hours").sum().alias("total_waste_core_hours"),
            pl.col("minutes_running").sum().alias("minutes_running_sum"),
            pl.col("container").count().alias("container_cnt"),
            pl.col("cpu_allocated_core_hours").sum().alias("total_allocated_core_hours")
        ]).sort("total_waste_core_hours", descending=True)

        global_total_waste = df_ns["total_waste_core_hours"].sum() or 0.1
        df_ns = df_ns.with_columns([
            pl.lit(date_chunk).alias("date"),
            ((pl.col("total_waste_core_hours") / global_total_waste) * 100).round(2).alias("waste_share_pct")
        ]).with_columns(
            pl.col("waste_share_pct").cum_sum().round(2).alias("waste_cumsum_pct")
        )

        # ─── 💾 [로컬 저장 명세 고정 바인딩] ───
        # 💡 모호한 INTEGRATED 구조가 배제되고 COMPUTE, STORAGE 등 고유 클러스터 서브디렉토리 위계로 정확히 격리 적재됩니다.
        cluster_partition_dir = MERGED_DIR / cluster_chunk
        cluster_partition_dir.mkdir(parents=True, exist_ok=True)

        target_output_file = cluster_partition_dir / f"daily_enriched_{cluster_chunk}_{date_chunk}.parquet"
        ns_output_file     = cluster_partition_dir / f"daily_ns_usage_{cluster_chunk}_{date_chunk}.parquet"
        pareto_output_file = cluster_partition_dir / f"pareto_ns_{cluster_chunk}_{date_chunk}.parquet"

        df_pod.write_parquet(target_output_file)
        df_daily_ns.write_parquet(ns_output_file)
        df_ns.write_parquet(pareto_output_file)

        print(f"     💾 [로컬 저장] 파티션 3대 원부 컴파일 성료 (클러스터: {cluster_chunk} / 날짜: {date_chunk})")

        # ─── 📡 [AIStor 영구 적재 가드레일 주입] ───
        print(f"     Submitting artifacts to AIStor Tables storage system...")
        try:
            # 💡 원격 AIStor(MinIO) 오브젝트 키 역시 클러스터 이름별 경로 구획을 완벽하게 동기화하여 뒤섞임을 원천 차단합니다.
            s3_client.upload_file(str(target_output_file), bucket_name, f"merged/{cluster_chunk}/{target_output_file.name}")
            s3_client.upload_file(str(ns_output_file), bucket_name, f"merged/{cluster_chunk}/{ns_output_file.name}")
            s3_client.upload_file(str(pareto_output_file), bucket_name, f"merged/{cluster_chunk}/{pareto_output_file.name}")
            print(f"     ✅ [AIStor 업로드 완수] 테이블 자산 적재 완료 -> merged/{cluster_chunk}/{pareto_output_file.name}")
        except Exception as e:
            print(f"     ❌ [AIStor 통신장애] 오브젝트 전송 실패: {str(e)}")

print("\n🏁 === [Step2 Polars 배치 완료] 로컬 및 AIStor 테이블 연동이 완벽히 마감되었습니다. ===")

if name == "main": main()

─── 📂 모든 디렉토리 경로 ./data 상대 경로 표준화 ──────────────────────────

BASE_DATA_DIR = Path("./data") MERGED_DIR = BASE_DATA_DIR / "merged" PLOT_DIR = BASE_DATA_DIR / "output" / "plots" OUT_DIR = BASE_DATA_DIR / "output"

OUT_DIR.mkdir(parents=True, exist_ok=True)

C = dict( hdr_dark="1F4E79", hdr_mid="2E75B6", hdr_light="BDD7EE", accent="ED7D31", red="C00000", green="70AD47", yellow="FFC000", gray_row="F2F2F2", white="FFFFFF", border="9DC3E6", summary_bg="EBF3FB", purple="7030A0", teal="008080", )

def ft(bold=False, size=10, color="000000", name="Arial"): return Font(name=name, bold=bold, size=size, color=color)

def fill(hex_color): return PatternFill("solid", fgColor=hex_color)

def thin_border(): t = Side(style="thin", color=C["border"]) return Border(left=t, right=t, top=t, bottom=t)

def center(wrap=False): return Alignment(horizontal="center", vertical="center", wrap_text=wrap)

def left(wrap=False): return Alignment(horizontal="left", vertical="center", wrap_text=wrap)

def set_col_widths(ws, widths): for col, w in widths.items(): ws.column_dimensions[col].width = w

def apply_header_row(ws, row_idx, headers, bg=None, fg=C["white"], size=10): bg = bg or C["hdr_dark"] for c, h in enumerate(headers, 1): cell = ws.cell(row=row_idx, column=c, value=h) cell.font = ft(bold=True, size=size, color=fg) cell.fill = fill(bg) cell.alignment = center(wrap=True) cell.border = thin_border()

def apply_data_rows(ws, df, start_row, num_formats=None, status_col_idx=None, zebra=True): nf = num_formats or {} for r_offset, row in enumerate(df.itertuples(index=False)): row_num = start_row + r_offset bg_hex = C["gray_row"] if (r_offset % 2 == 1 and zebra) else C["white"] for c_idx, val in enumerate(row, 1): cell = ws.cell(row=row_num, column=c_idx, value=val) cell.font = ft(size=9) cell.fill = fill(bg_hex) cell.alignment = left() cell.border = thin_border() if c_idx in nf: cell.number_format = nf[c_idx] if status_col_idx and c_idx == status_col_idx: v = str(val) if "OOM" in v or "Killed" in v: cell.fill = fill("FFCCCC"); cell.font = ft(bold=True, size=9, color=C["red"]) elif "Shortage" in v or "부족" in v: cell.fill = fill("FFF2CC"); cell.font = ft(bold=True, size=9, color="7F6000") elif "Over" in v or "과다" in v: cell.fill = fill("DDEEFF"); cell.font = ft(bold=True, size=9, color=C["hdr_dark"]) elif "Optim" in v or "최적" in v: cell.fill = fill("E2EFDA"); cell.font = ft(bold=True, size=9, color="375623") return start_row + len(df)

def freeze_and_filter(ws, row=2): ws.freeze_panes = ws.cell(row=row+1, column=1) ws.auto_filter.ref = ws.dimensions

💡 [교정] 차트 픽업 시 클러스터별 하위 폴더(infra_tag)를 조준하도록 경로 인자 주입

def add_chart_image(ws, path_key, anchor_cell, infra_tag, w=860, h=400, label=None): path = PLOT_DIR / infra_tag / path_key if not path.exists(): print(f" ⚠️ [차트 유실] {path_key} 파일이 {PLOT_DIR / infra_tag}에 없어 삽입을 건너뜁니다.") return row_num = int(''.join(filter(str.isdigit, anchor_cell))) if label: col_letter = "".join(filter(str.isalpha, anchor_cell)) col_idx = 1 if col_letter == "A" else (9 if col_letter == "I" else 1) ws.cell(row=row_num-1, column=col_idx, value=label).font = ft(bold=True, size=11, color=C["hdr_dark"]) img = XLImage(str(path)) img.width = w; img.height = h ws.add_image(img, anchor_cell) print(f" -> 🎨 차트 맵핑 완료: {path_key} ➡️ {anchor_cell} 셀 ({infra_tag} 파티션)")

─── Sheet 0: 전사 종합 요약 ───────────────────────────────

def build_sheet_summary(wb, df_pod, df_ns, infra_tag): print(f"⏳ [0/9] '0. 전사종합요약' 대시보드 탭 구축 개시... ({infra_tag})") ws = wb.active ws.title = "0. 전사종합요약" ws.sheet_view.showGridLines = False ws.row_dimensions[1].height = 42

ws.merge_cells("A1:H1")
t = ws["A1"]
t.value = f"Resource Governance Master Report [{infra_tag}]  (KST 기준)"
t.font = ft(bold=True, size=15, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()

oom_cnt    = int(df_pod["is_oom_killed"].sum())
no_req_cnt = int((df_pod["has_no_request"] | df_pod["has_no_limit"]).sum())
alloc_ch   = df_pod["cpu_allocated_core_hours"].sum()
usage_ch   = df_pod["cpu_usage_core_hours"].sum()
waste_ch   = df_pod["cpu_waste_core_hours"].sum()
eff_pct    = usage_ch / max(alloc_ch, 0.001) * 100

mem_waste  = df_pod["mem_waste_gb_hours"].sum()
pv_waste   = df_pod["pv_waste_gb_hours"].sum()
throttle_risks = int((df_pod["cpu_throttled_max"] > 0.2).sum())
kst_dates  = sorted(df_pod["date"].unique())
date_range = f"{kst_dates[0]} ~ {kst_dates[-1]} (KST)"

kpis = [
    ("정산 대상 인프라 도메인",       infra_tag,                              "002060",      C["white"]),
    ("분석 기간 (KST)",              date_range,                             C["hdr_dark"], C["white"]),
    ("총 관측 컨테이너 볼륨",         f"{len(df_pod):,} 개",                C["hdr_dark"], C["white"]),
    ("OOMKilled 장애 컨테이너",       f"{oom_cnt:,} 개",                    C["red"],      C["white"]),
    ("연산 스로틀링 위험군 컨테이너",   f"{throttle_risks:,} 개",                 "C00000",      C["white"]),
    ("리소스 설정 규격 위반군",       f"{no_req_cnt:,} 개",                  "E26B0A",      C["white"]),
    ("전사 CPU 무효 선점 낭비량",     f"{waste_ch:,.1f} Core-H",            C["hdr_mid"],  C["white"]),
    ("전사 Memory 무효 선점 낭비량",   f"{mem_waste:,.1f} GB-H",             "6B4F9B",      C["white"]),
    ("전사 PV 스토리지 알박기 낭비량", f"{pv_waste:,.1f} GB-H",              "008080",      C["white"]),
    ("전사 CPU 평균 실효 활용률",     f"{eff_pct:.1f} %",                    "375623",      C["white"]),
]

for i, (label, value, bg, fg) in enumerate(kpis):
    row = 3 + i
    ws.row_dimensions[row].height = 24
    lc = ws.cell(row=row, column=1, value=label)
    lc.font = ft(bold=True, size=10, color=C["hdr_dark"])
    lc.fill = fill(C["summary_bg"]); lc.alignment = left(); lc.border = thin_border()
    ws.merge_cells(f"A{row}:C{row}")
    vc = ws.cell(row=row, column=4, value=value)
    vc.font = ft(bold=True, size=11, color=fg)
    vc.fill = fill(bg); vc.alignment = center(); vc.border = thin_border()
    ws.merge_cells(f"D{row}:F{row}")

set_col_widths(ws, {"A":28,"B":14,"C":14,"D":22,"E":14,"F":14,"G":20,"H":20})

chart_row = 16
add_chart_image(ws, "chart6_status_donut.png",   f"A{chart_row}", infra_tag, w=420, h=300)
add_chart_image(ws, "chart5_pareto_ns_waste.png", f"E{chart_row}", infra_tag, w=560, h=300)

─── Sheet 1: 파레토 분석 NS ──────────────────────────────

def build_sheet_pareto(wb, df_ns, infra_tag): print(f"⏳ [1/9] '1. 파레토분석_NS' 시트 렌더링 중... ({infra_tag})") ws = wb.create_sheet("1. 파레토분석_NS") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:I1")
t = ws["A1"]
t.value = f"Namespace별 CPU Waste 파레토 분석 [{infra_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["Namespace","실행시간 합계(분)","컨테이너 수","할당 Core-H","낭비 Core-H","낭비 비중(%)","누적 비중(%)","등급"]
apply_header_row(ws, 2, headers, bg=C["hdr_mid"])

df_disp = df_ns.copy()
df_disp["등급"] = df_disp["waste_cumsum_pct"].apply(
    lambda x: "Critical (Top 20%)" if x<=20 else ("High (Top 50%)" if x<=50 else ("Medium" if x<=80 else "Low")))
col_order = ["namespace","minutes_running_sum","container_cnt","total_allocated_core_hours","total_waste_core_hours","waste_share_pct","waste_cumsum_pct","등급"]
df_out = df_disp[col_order]
nf = {4:"#,##0.0", 5:"#,##0.0", 6:"0.00", 7:"0.00"}
end_row = apply_data_rows(ws, df_out, start_row=3, num_formats=nf)

ws.conditional_formatting.add(f"E3:E{end_row}", DataBarRule(start_type="min", end_type="max", color="2E75B6", showValue=True))
set_col_widths(ws, {"A":22,"B":18,"C":14,"D":16,"E":16,"F":12,"G":12,"H":22})
freeze_and_filter(ws)

─── 🎉 Sheet 1-2: 일단위 Namespace별 자원 사용량 탭 ───────────────────

def build_sheet_ns_daily(wb, df_daily_ns, infra_tag): print(f"⏳ [2/9] '1-2. 일단위_NS별_사용량' 종합 청산 장표 마크 중... (행 수: {len(df_daily_ns):,}건)") ws = wb.create_sheet("1-2. 일단위_NS별_사용량") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:J1")
t = ws["A1"]
t.value = f"Namespace 일일 자원 소모량 및 통합 사용량 점수 지표 [{infra_tag}]"
t.font = ft(bold=True, size=12, color=C["white"])
t.fill = fill("1F4E79"); t.alignment = center()
ws.row_dimensions[1].height = 35

headers = [
    "관측 일자(KST)", "네임스페이스(Namespace)", 
    "CPU 사용(Core-H)", "CPU 할당(Core-H)", "CPU 낭비(Core-H)",
    "Mem 사용(GB-H)", "Mem 할당(GB-H)",
    "PV 사용(GB-H)", "PV 할당(GB-H)", "통합 사용량 점수 (Score)"
]
apply_header_row(ws, 2, headers, bg=C["hdr_mid"])
ws.row_dimensions[2].height = 26

col_order = ["date", "namespace", "cpu_used_ch", "cpu_alloc_ch", "cpu_waste_ch", "mem_used_gh", "mem_alloc_gh", "pv_used_gh", "pv_alloc_gh", "final_usage_score"]
df_out = df_daily_ns[col_order].sort_values(by=["date", "final_usage_score"], ascending=[True, False])

nf = {3:"#,##0.0", 4:"#,##0.0", 5:"#,##0.0", 6:"#,##0.0", 7:"#,##0.0", 8:"#,##0.0", 9:"#,##0.0", 10:"#,##0.0"}
end_row = apply_data_rows(ws, df_out, start_row=3, num_formats=nf)

ws.conditional_formatting.add(f"J3:J{end_row}", DataBarRule(start_type="min", end_type="max", color="375623", showValue=True))

widths = {"A":14, "B":22, "C":18, "D":18, "E":18, "F":18, "G":18, "H":18, "I":18, "J":24}
set_col_widths(ws, widths)
freeze_and_filter(ws, row=2)

─── Sheet 2: CPU 분석 ────────────────────────────────────

def build_sheet_cpu(wb, df_pod, infra_tag): top30 = max(1, int(len(df_pod)*0.30)) print(f"⏳ [3/9] '2. CPU Request_Usage 분석' 시트 빌드 중... (상위 30% 격리: {top30}행)") ws = wb.create_sheet("2. CPU Request_Usage 분석") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:M1")
t = ws["A1"]
t.value = f"CPU Resource Efficiency Analysis [{infra_tag}] — Request / Limit / Usage / Throttling"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["날짜(KST)","클러스터","네임스페이스","워크로드","Pod","컨테이너",
           "CPU Request","CPU Limit","CPU P95","Throttle Peak","활용률(%)","낭비 Core-H","상태"]
apply_header_row(ws, 2, headers, bg=C["hdr_mid"])

df_out = df_pod.sort_values("cpu_waste_core_hours", ascending=False).head(top30).copy()
df_out["util"] = np.where(df_out["cpu_request_max"]>0, (df_out["cpu_usage_p95"]/df_out["cpu_request_max"]*100).round(1), 0)
df_out["status_en"] = df_out["status"].map({
    "💥 OOM장애발생":"OOM Killed","⚠️ Request부족":"Request Shortage",
    "📉 과다할당":"Over-allocated","✅ 최적화완료":"Optimized"}).fillna("Unknown")

cols = ["date","cluster","namespace","workload_type","pod","container",
        "cpu_request_max","cpu_limit_max","cpu_usage_p95","cpu_throttled_max","util","cpu_waste_core_hours","status_en"]
df_disp = df_out[cols].reset_index(drop=True)

nf = {7:"0.000",8:"0.000",9:"0.000",10:"0.000",11:"0.0",12:"#,##0.0"}
end_row = apply_data_rows(ws, df_disp, start_row=3, num_formats=nf, status_col_idx=13)

ws.conditional_formatting.add(f"K3:K{end_row}", ColorScaleRule(start_type="num", start_value=0, start_color="FF0000", mid_type="num", mid_value=50, mid_color="FFFF00", end_type="num", end_value=100, end_color="00B050"))
set_col_widths(ws, {"A":12,"B":16,"C":18,"D":16,"E":26,"F":14,"G":12,"H":12,"I":12,"J":14,"K":12,"L":14,"M":16})
freeze_and_filter(ws)

add_chart_image(ws, "chart1_cpu_req_vs_usage_by_workload.png", f"A{end_row+3}", infra_tag, w=860, h=400, label="[ CPU Request / Limit / P95 by Workload ]")
add_chart_image(ws, "chart9_boxplot_cpu_util_by_workload.png",  f"A{end_row+26}", infra_tag, w=860, h=400, label="[ CPU Utilization Boxplot by Workload ]")

─── Sheet 3: Memory 및 PV 스토리지 분석 ─────────────────────────────────

def build_sheet_memory(wb, df_pod, infra_tag): top30 = max(1, int(len(df_pod)*0.30)) print(f"⏳ [4/9] '3. Memory_PV 입체분석' 고도화 시트 로딩 중... ({infra_tag})") ws = wb.create_sheet("3. Memory_PV 입체분석") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:O1")
t = ws["A1"]
t.value = f"Memory RSS & Persistent Volume Storage Cross-Governance Analysis [{infra_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["날짜(KST)","클러스터","네임스페이스","워크로드","Pod","컨테이너",
           "Mem Request","Mem P95","Mem RSS","PV Cap(GB)","PV Used(GB)","PV Waste(GB-H)","활용률(%)","낭비 GB-H","상태"]
apply_header_row(ws, 2, headers, bg=C["hdr_mid"])

df_out = df_pod.sort_values(by=["mem_waste_gb_hours", "pv_waste_gb_hours"], ascending=[False, False]).head(top30).copy()
df_out["util"] = np.where(df_out["mem_request_max"]>0, (df_out["mem_usage_p95"]/df_out["mem_request_max"]*100).round(1), 0)
df_out["status_en"] = df_out["status"].map({
    "💥 OOM장애발생":"OOM Killed","⚠️ Request부족":"Request Shortage",
    "📉 과다할당":"Over-allocated","✅ 최적화완료":"Optimized"}).fillna("Unknown")

cols = ["date","cluster","namespace","workload_type","pod","container",
        "mem_request_max","mem_usage_p95","mem_rss_p95","pv_capacity_max","pv_used_p95","pv_waste_gb_hours","util","mem_waste_gb_hours","status_en"]
df_disp = df_out[cols].reset_index(drop=True)

nf = {7:"0.00", 8:"0.00", 9:"0.00", 10:"0.00", 11:"0.00", 12:"#,##0.0", 13:"0.0", 14:"#,##0.0"}
end_row = apply_data_rows(ws, df_disp, start_row=3, num_formats=nf, status_col_idx=15)

ws.conditional_formatting.add(f"M3:M{end_row}", ColorScaleRule(start_type="num", start_value=0, start_color="FF0000", mid_type="num", mid_value=50, mid_color="FFFF00", end_type="num", end_value=100, end_color="00B050"))
set_col_widths(ws, {"A":12,"B":16,"C":18,"D":16,"E":26,"F":14,"G":12,"H":12,"I":12,"J":12,"K":12,"L":14,"M":12,"N":12,"O":16})
freeze_and_filter(ws)

add_chart_image(ws, "chart2_mem_req_vs_usage_by_workload.png", f"A{end_row+3}", infra_tag, w=860, h=400, label="[ Memory Specs vs WorkingSet vs RSS Peak ]")
add_chart_image(ws, "chart20_daily_mem_per_workload.png", f"A{end_row+26}", infra_tag, w=960, h=540, label="[ Daily Memory & PV Provisioning Trends ]")

─── Sheet 4: OOM 및 자원 부족 ───────────────────────────

def build_sheet_oom(wb, df_pod, infra_tag): print(f"⏳ [5/9] '4. 자원부족및OOM장애군' 리스크 인덱스 추출 중... ({infra_tag})") ws = wb.create_sheet("4. 자원부족및OOM장애군") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:L1")
t = ws["A1"]
t.value = f"OOMKilled / CPU CFS Throttled 병목 위험 워크로드 명세 [{infra_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["red"]); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["날짜(KST)","클러스터","네임스페이스","워크로드","Pod","컨테이너",
           "상태","CPU Request","CPU P95","Throttle Peak","Mem Limit(GB)","Mem P95(GB)"]
apply_header_row(ws, 2, headers, bg=C["red"])

df_out = df_pod[
    (df_pod["cpu_shortage_cores"]>0) | (df_pod["is_oom_killed"]) | (df_pod["cpu_throttled_max"] > 0.2)
].sort_values(["is_oom_killed","cpu_throttled_max","cpu_shortage_cores"], ascending=[False,False,False]).copy()

df_out["status_en"] = df_out["status"].map({
    "💥 OOM장애발생":"OOM Killed","⚠️ Request부족":"Request Shortage",
    "📉 과다할당":"Over-allocated","✅ 최적화완료":"Optimized"}).fillna("Unknown")

cols = ["date","cluster","namespace","workload_type","pod","container",
        "status_en","cpu_request_max","cpu_usage_p95","cpu_throttled_max","mem_limit_max","mem_usage_p95"]
df_disp = df_out[cols].reset_index(drop=True)

nf = {8:"0.000",9:"0.000",10:"0.000",11:"0.00",12:"0.00"}
end_row = apply_data_rows(ws, df_disp, start_row=3, num_formats=nf, status_col_idx=7)
set_col_widths(ws, {"A":12,"B":18,"C":20,"D":18,"E":28,"F":16,"G":16,"H":12,"I":12,"J":14,"K":14,"L":14})
freeze_and_filter(ws)

add_chart_image(ws, "chart8_shortfall_footprint.png", f"A{end_row+3}", infra_tag, w=860, h=400, label="[ CPU CFS Throttling Risk Heatmap Footprint ]")

─── Sheet 5: 리소스 미설정 위반 ─────────────────────────

def build_sheet_violations(wb, df_pod, infra_tag): print(f"⏳ [6/9] '5. 리소스미설정위반군' 규격 미설정 목록 마크 중... ({infra_tag})") ws = wb.create_sheet("5. 리소스미설정위반군") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:L1")
t = ws["A1"]
t.value = "Resource Request / Limit 미설정 위반 컨테이너 (KST)"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill("E26B0A"); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["날짜(KST)","클러스터","네임스페이스","워크로드","Pod","컨테이너",
           "Request 미설정","Limit 미설정","CPU Request","CPU Limit","Mem Request(GB)","Mem Limit(GB)"]
apply_header_row(ws, 2, headers, bg="E26B0A")

df_out = df_pod[df_pod["has_no_request"]|df_pod["has_no_limit"]].sort_values("minutes_running", ascending=False).copy()
df_out["req_flag"] = df_out["has_no_request"].map({True:"MISSING",False:"OK"})
df_out["lim_flag"] = df_out["has_no_limit"].map({True:"MISSING",False:"OK"})
cols = ["date","cluster","namespace","workload_type","pod","container",
        "req_flag","lim_flag","cpu_request_max","cpu_limit_max","mem_request_max","mem_limit_max"]
df_disp = df_out[cols].reset_index(drop=True)

nf = {9:"0.000",10:"0.000",11:"0.00",12:"0.00"}
end_row = apply_data_rows(ws, df_disp, start_row=3, num_formats=nf)
set_col_widths(ws, {"A":12,"B":18,"C":20,"D":18,"E":28,"F":16,"G":14,"H":14,"I":14,"J":14,"K":14,"L":14})
freeze_and_filter(ws)

─── Sheet 6: 일별 트렌드 (KST) ──────────────────────────

def build_sheet_trends(wb, df_pod, infra_tag): print(f"⏳ [7/9] '6. 일별트렌드_KST' 누적 통계 분석 중... ({infra_tag})") ws = wb.create_sheet("6. 일별트렌드_KST") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:M1")
t = ws["A1"]
t.value = f"Daily Resource Governance & Loss Trend [{infra_tag}] — KST"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 32

df_daily = df_pod.groupby("date").agg(
    containers=("container","count"),
    cpu_alloc=("cpu_allocated_core_hours","sum"),
    cpu_used=("cpu_usage_core_hours","sum"),
    cpu_waste=("cpu_waste_core_hours","sum"),
    mem_alloc=("mem_allocated_gb_hours","sum"),
    mem_waste=("mem_waste_gb_hours","sum"),
    pv_alloc=("pv_allocated_gb_hours","sum"),
    pv_waste_sum=("pv_waste_gb_hours","sum"),
    oom_cnt=("is_oom_killed","sum"),
    shortage_cnt=("cpu_shortage_cores", lambda x: (x>0).sum()),
).reset_index()

df_daily["cpu_util_pct"] = (df_daily["cpu_used"]/df_daily["cpu_alloc"].clip(lower=0.001)*100).round(1)
df_daily["mem_util_pct"] = ((df_daily["mem_alloc"]-df_daily["mem_waste"])/df_daily["mem_alloc"].clip(lower=0.001)*100).round(1)

headers = ["날짜(KST)","컨테이너 볼륨","CPU 할당 Core-H","CPU 사용 Core-H","CPU 낭비 Core-H",
           "Mem 할당 GB-H","Mem 낭비 GB-H","PV 할당 GB-H","PV 낭비 GB-H","OOM 사살","자원부족군","CPU활용률(%)","Mem활용률(%)"]
apply_header_row(ws, 3, headers, bg=C["hdr_mid"])

nf = {3:"#,##0.0",4:"#,##0.0",5:"#,##0.0",6:"#,##0.0",7:"#,##0.0",8:"#,##0.0",9:"#,##0.0",10:"#,##0",11:"#,##0",12:"0.1",13:"0.1"}
df_disp = df_daily[["date","containers","cpu_alloc","cpu_used","cpu_waste",
                     "mem_alloc","mem_waste","pv_alloc","pv_waste_sum","oom_cnt","shortage_cnt","cpu_util_pct","mem_util_pct"]]
end_row = apply_data_rows(ws, df_disp, start_row=4, num_formats=nf)

set_col_widths(ws, {"A":14,"B":12,"C":16,"D":16,"E":16,"F":16,"G":14,"H":16,"I":14,"J":12,"K":12,"L":14,"M":14})
freeze_and_filter(ws, row=4)

add_chart_image(ws, "chart3_daily_waste_stack.png", f"A{end_row+3}", infra_tag, 860, 400, "[ Daily CPU Waste Stacked Timeline ]")
add_chart_image(ws, "chart4_cpu_efficiency_heatmap.png", f"A{end_row+26}", infra_tag, 860, 400, "[ CPU Utilization Heatmap Grid ]")

─── Sheet 7: 워크로드별 심층 분석 ─────────────────

def build_sheet_workload(wb, df_pod, infra_tag): print(f"⏳ [8/9] '7. 워크로드별_심층분석' 오픈소스 스택 분석 표 작성 중... ({infra_tag})") ws = wb.create_sheet("7. 워크로드별_심층분석") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:N1")
t = ws["A1"]
t.value = f"Workload Type별 오픈소스 기술 스택 심층 자원 효율화 분석 [{infra_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["purple"]); t.alignment = center()
ws.row_dimensions[1].height = 32

df_wl = df_pod.groupby("workload_type").agg(
    containers=("container","count"),
    cpu_req_avg=("cpu_request_max","mean"),
    cpu_p95_avg=("cpu_usage_p95","mean"),
    cpu_waste_sum=("cpu_waste_core_hours","sum"),
    cpu_throttle_pk=("cpu_throttled_max","max"),
    mem_req_avg=("mem_request_max","mean"),
    mem_p95_avg=("mem_usage_p95","mean"),
    mem_rss_avg=("mem_rss_p95","mean"),
    mem_waste_sum=("mem_waste_gb_hours","sum"),
    pv_capacity_avg=("pv_capacity_max","mean"),
    pv_waste_sum=("pv_waste_gb_hours","sum"),
    oom_cnt=("is_oom_killed","sum")
).reset_index()

df_wl["cpu_util_pct"] = (df_wl["cpu_p95_avg"]/df_wl["cpu_req_avg"].replace(0,np.nan)*100).round(1).fillna(0)
df_wl["mem_util_pct"] = (df_wl["mem_p95_avg"]/df_wl["mem_req_avg"].replace(0,np.nan)*100).round(1).fillna(0)
df_wl = df_wl.sort_values("cpu_waste_sum", ascending=False).reset_index(drop=True)

headers2 = ["워크로드 타입","컨테이너 수","CPU Req(avg)","CPU P95(avg)","CPU 낭비 총합","Throttle Peak",
            "Mem Req(avg)","Mem P95(avg)","Mem RSS(avg)","Mem 낭비 총합","PV Cap(avg)","PV 낭비 총합","CPU활용(%)","Mem활용(%)"]
apply_header_row(ws, 3, headers2, bg=C["purple"])

cols_out = ["workload_type","containers","cpu_req_avg","cpu_p95_avg","cpu_waste_sum","cpu_throttle_pk",
            "mem_req_avg","mem_p95_avg","mem_rss_avg","mem_waste_sum","pv_capacity_avg","pv_waste_sum","cpu_util_pct","mem_util_pct"]
df_disp = df_wl[cols_out]

nf = {3:"0.000", 4:"0.000", 5:"#,##0.0", 6:"0.000", 7:"0.00", 8:"0.00", 9:"0.00", 10:"#,##0.0", 11:"0.00", 12:"#,##0.0", 13:"0.1", 14:"0.1"}
end_row = apply_data_rows(ws, df_disp, start_row=4, num_formats=nf)

set_col_widths(ws, {"A":18,"B":12,"C":14,"D":14,"E":14,"F":14,"G":14,"H":14,"I":14,"J":14,"K":14,"L":14,"M":12,"N":12})
freeze_and_filter(ws, row=4)

add_chart_image(ws, "chart15_oom_status_by_workload.png", f"A{end_row+3}", infra_tag, 860, 400, "[ Status Distribution count per Open-Source Stack ]")
add_chart_image(ws, "chart14_cpu_mem_waste_scatter.png", f"A{end_row+26}", infra_tag, 760, 480, "[ Multi-Resource Loss Correlation Bubble Map ]")

─── 🎉 [요청 반영] Sheet 8: 미사용 추가 차트 모음 전용 시트 ─────────────────

def build_sheet_extra_charts(wb, infra_tag, date_tag): print(f"⏳ [9/9] '8. 추가차트모음' 시트 격리 배치 중... ({infra_tag})") ws = wb.create_sheet("8. 추가차트모음") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:Q1")
t = ws["A1"]
t.value = f"Step 3 Additional Unused Dashboards & Reports [{infra_tag} - {date_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 35

used_ids = [1, 2, 3, 5, 6, 9, 10, 15]
remaining_ids = [i for i in range(1, 20) if i not in used_ids]

current_row = 3
col_positions = ["A", "I"]  # 2열 바둑판 교차 격자 구조
col_idx = 0

for cid in remaining_ids:
    found_file = None
    patterns = [f"chart{cid}_*.png", f"chart_{cid}_*.png", f"chart{cid}.png", f"chart_{cid}.png"]
    for pat in patterns:
        # 💡 [교정] 추가 차트 탐색 시 클러스터별 하위 폴더 격리 스캔
        matches = list((PLOT_DIR / infra_tag).glob(pat))
        if matches:
            found_file = matches[0].name
            break

    if found_file:
        anchor = f"{col_positions[col_idx]}{current_row}"
        add_chart_image(ws, found_file, anchor, infra_tag, w=480, h=280, label=f"Chart {cid}: {found_file}")
        col_idx += 1
        if col_idx >= len(col_positions):
            col_idx = 0
            current_row += 16  # 차트 간격 오프셋
    else:
        print(f"  ℹ️  [추가 차트] Chart ID {cid} 번 파일이 {infra_tag} 폴더에 없어 생략합니다.")

─── Main ─────────────────────────────────────────────────

def main(): print("🚀 [Step6 개시] res_usage_ 계열 고도화 엑셀 리포터 엔진 가동 (인프라/날짜 패턴 자동 검색)...")

# ─── 🔄 [교정] 2단계 개편안에 맞춰 하위 격리 디렉토리 위계(*/*)로 검색 와일드카드 변경 ───
search_pattern = str(MERGED_DIR / "*" / "pareto_ns_*.parquet")
pareto_files = glob.glob(search_pattern)

if not pareto_files:
    print(f"❌ 오류: 탐색된 파레토 가공원부 파일이 없습니다. 경로를 확인하세요: {search_pattern}")
    return

processed_count = 0
for file_path in pareto_files:
    filename = os.path.basename(file_path)
    match = re.search(r"pareto_ns_(.*)_(.*)\.parquet", filename)
    if not match:
        continue

    infra_tag = match.group(1)   # 예: COMPUTE 또는 STORAGE
    date_tag  = match.group(2)   # 예: 20260702

    # 💡 [교정] 모놀리식 주소를 폐기하고 클러스터 격리 폴더명(/infra_tag/)을 경로 중간에 강제 주입
    p1 = MERGED_DIR / infra_tag / f"daily_enriched_{infra_tag}_{date_tag}.parquet"
    p2 = Path(file_path)
    p3 = MERGED_DIR / infra_tag / f"daily_ns_usage_{infra_tag}_{date_tag}.parquet"

    if not (p1.exists() and p3.exists()):
        print(f"⚠️  [세트 불완전] {infra_tag} | {date_tag} 에 쌍이 되는 가공 데이터 폴더 스캔 실패로 건너뜜.")
        continue

    print(f"\n" + "="*60)
    print(f"🔄 [파티션 매칭 감지] Cluster/Infra: {infra_tag} | Date: {date_tag} 컴파일 가동")
    print("="*60)

    df_pod      = pd.read_parquet(p1)
    df_ns       = pd.read_parquet(p2)
    df_daily_ns = pd.read_parquet(p3)

    excel_name = f"res_usage_report_{infra_tag.lower()}_{date_tag}.xlsx"

    wb = Workbook()
    build_sheet_summary(wb, df_pod, df_ns, infra_tag)
    build_sheet_pareto(wb, df_ns, infra_tag)
    build_sheet_ns_daily(wb, df_daily_ns, infra_tag)
    build_sheet_cpu(wb, df_pod, infra_tag)
    build_sheet_memory(wb, df_pod, infra_tag)
    build_sheet_oom(wb, df_pod, infra_tag)
    build_sheet_violations(wb, df_pod, infra_tag)
    build_sheet_trends(wb, df_pod, infra_tag)
    build_sheet_workload(wb, df_pod, infra_tag)
    build_sheet_extra_charts(wb, infra_tag, date_tag)

    out_path = OUT_DIR / excel_name
    print(f"💾 openpyxl 스트림 디스크 저장 가동 중... ➡️ {out_path}")
    wb.save(out_path)
    print(f"📦 [로컬 컴파일 마감 완료]: {excel_name} ({out_path.stat().st_size/1024:.0f} KB)")

    # ─── 🪣 사내 MinIO AIStor 'devops-test' 버킷 자동 배포 레이어 ───
    minio_endpoint   = os.getenv("MINIO_ENDPOINT")
    minio_access_key = os.getenv("MINIO_ACCESS_KEY")
    minio_secret_key = os.getenv("MINIO_SECRET_KEY")
    bucket_name      = os.getenv("MINIO_REPORT_BUCKET", "devops-test")

    if all([minio_endpoint, minio_access_key, minio_secret_key]):
        try:
            endpoint_clean = minio_endpoint.replace("http://", "").replace("https://", "")
            secure_flag = minio_endpoint.startswith("https://")

            minio_client = Minio(
                endpoint_clean,
                access_key=minio_access_key,
                secret_key=minio_secret_key,
                secure=secure_flag
            )

            if not minio_client.bucket_exists(bucket_name):
                print(f"🪣  [MinIO] 버킷이 존재하지 않아 신규 생성합니다: {bucket_name}")
                minio_client.make_bucket(bucket_name)

            object_key = f"reports/{infra_tag.lower()}/{excel_name}"
            print(f"🪣  [오브젝트 스토리지 싱크] devops-test 버킷 배포 ➡️ MinIO://{object_key} 파일 스트림 업로드 중...")

            minio_client.fput_file(bucket_name, object_key, str(out_path))
            print(f"🏁 === [전사 배포 마감 성공] 고도화 {infra_tag} 마스터 엑셀 리포트 배포 자산화 완료. ===")
        except Exception as e:
            print(f"❌ [배포 에러] 사내 MinIO AIStor 원격 업로드 중 예외 발생: {str(e)}")
    else:
        print("⚠️ [안내] 접속 환경변수(MINIO_ENDPOINT 등) 생략으로 로컬 아카이빙 처리 후 해당 파티션을 마감합니다.")

    processed_count += 1

print(f"\n🏁 === [전체 자동 매칭 빌드 완료] 총 {processed_count}개 세트 인프라 마스터 리포트 빌드 프로세스가 종결되었습니다. ===")

if name == "main": main()

from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter from openpyxl.drawing.image import Image from openpyxl.formatting.rule import DataBarRule

─── 📂 경로 및 타임존 표준화 ─────────────────────────────────────────────

BASE_DATA_DIR = Path("./data") OUTPUT_DIR = BASE_DATA_DIR / "output" BASE_PLOT_DIR = OUTPUT_DIR / "plots"

KST = timezone(timedelta(hours=9))

─── 🎨 엔터프라이즈 Muted Blue 테마 디자인 시트 스타일 정의 ──────────────────

C = dict( hdr_dark="1F4E79", hdr_mid="2E75B6", hdr_light="BDD7EE", accent="ED7D31", red="C00000", green="70AD47", yellow="FFC000", gray_row="F2F2F2", white="FFFFFF", border="9DC3E6", summary_bg="EBF3FB" )

def ft(bold=False, size=10, color="000000", name="Arial"): return Font(name=name, bold=bold, size=size, color=color) def fill(hex_color): return PatternFill("solid", fgColor=hex_color) def thin_border(): t = Side(style="thin", color=C["border"]) return Border(left=t, right=t, top=t, bottom=t) def center(wrap=False): return Alignment(horizontal="center", vertical="center", wrap_text=wrap) def left(wrap=False): return Alignment(horizontal="left", vertical="center", wrap_text=wrap)

def parse_arguments(): parser = argparse.ArgumentParser(description="FinOps Master Excel Sheet Builder") # 💡 INTEGRATED를 배제하고 무조건 명확한 대상 클러스터를 요구하도록 고정 parser.add_argument("--cluster", type=str, required=True, choices=["COMPUTE", "STORAGE"], help="Target cluster type") return parser.parse_args()

def get_minio_client(): return boto3.client( "s3", endpoint_url=os.getenv("MINIO_ENDPOINT"), aws_access_key_id=os.getenv("MINIO_ACCESS_KEY"), aws_secret_access_key=os.getenv("MINIO_SECRET_KEY"), region_name="us-east-1", config=boto3.session.Config(signature_version="s3v4") )

def main(): args = parse_arguments() cluster_target = args.cluster.upper()

print(f"🚀 [Step6] Executing Excel Master Builder for Target Cluster: {cluster_target}...")

# ─── 🗂️ [적응형 매핑 주입 1] 4단계 거버넌스 산출 파일 바인딩 ───
risk_file  = OUTPUT_DIR / f"gov_risk_oom_{cluster_target}.parquet"
waste_file = OUTPUT_DIR / f"gov_waste_candidates_{cluster_target}.parquet"
viol_file  = OUTPUT_DIR / f"gov_violations_{cluster_target}.parquet"

# ─── 🎨 [적응형 매핑 주입 2] 3단계 차트 보관 디렉토리 격리 바인딩 ───
PLOT_DIR = BASE_PLOT_DIR / cluster_target

if not risk_file.exists():
    print(f"❌ Error: Required governance data for '{cluster_target}' not found. Please run step4 first.")
    return

# 데이터 로드
df_risk  = pd.read_parquet(risk_file)
df_waste = pd.read_parquet(waste_file)
df_viol  = pd.read_parquet(viol_file)

wb = Workbook()

# -------------------------------------------------------------------------
# 시트 1: 0. 전사종합요약 & 격리 대시보드 차트 탑재
# -------------------------------------------------------------------------
ws_dash = wb.active
ws_dash.title = "0. 종합 대시보드"
ws_dash.sheet_view.showGridLines = False

ws_dash.merge_cells("A1:N2")
title_cell = ws_dash["A1"]
title_cell.value = f"FinOps Infrastructure Governance Executive Summary [{cluster_target} CLUSTER]"
title_cell.font = ft(bold=True, size=14, color=C["white"])
title_cell.fill = fill(C["hdr_dark"])
title_cell.alignment = center()

# 💡 격리된 폴더 경로에서 이미지 차트 서칭 및 임베딩
chart_map = {
    "B4": "chart6_status_donut.png",
    "H4": "chart1_cpu_req_vs_usage_by_workload.png" if cluster_target == "COMPUTE" else "chart2_mem_req_vs_usage_by_workload.png",
    "B20": "chart5_pareto_ns_waste.png",
    "H20": "chart4_cpu_efficiency_heatmap.png" if cluster_target == "COMPUTE" else "chart18_mem_waste_heatmap.png"
}

print("🎨 Embedding cluster-isolated analytical charts into Dashboard sheet...")
for cell_loc, chart_name in chart_map.items():
    img_path = PLOT_DIR / chart_name
    if img_path.exists():
        img = Image(str(img_path))
        # 대시보드 레이아웃 균형을 위해 차트 해상도 스케일링 조율
        img.width, img.height = 420, 260
        ws_dash.add_image(img, cell_loc)
    else:
        print(f"  ⚠️ Missing chart asset: {img_path.name} (Placeholder will be used in Excel view)")

# -------------------------------------------------------------------------
# 시트 2: 4. 자원부족및OOM장애군 (Step4 산출물 매핑)
# -------------------------------------------------------------------------
ws_risk = wb.create_sheet("1. 자원부족 및 OOM 장애군")
ws_risk.sheet_view.showGridLines = False

headers_risk = ["Date", "Namespace", "Workload Type", "Pod Name", "Container", "Minutes Running", "Max Request", "Max Limit", "P95 Usage", "OOM Count", "Status", "Recommendation"]

# 헤더 렌더링
for col_idx, h in enumerate(headers_risk, 1):
    cell = ws_risk.cell(row=1, column=col_idx, value=h)
    cell.font = ft(bold=True, color=C["white"])
    cell.fill = fill(C["hdr_mid"])
    cell.alignment = center()
    cell.border = thin_border()

# 데이터 쓰기
for r_idx, row in enumerate(df_risk.itertuples(index=False), 2):
    bg_color = C["gray_row"] if r_idx % 2 == 1 else C["white"]
    for c_idx, val in enumerate(row, 1):
        cell = ws_risk.cell(row=r_idx, column=c_idx, value=val)
        cell.font = ft(size=9)
        cell.fill = fill(bg_color)
        cell.border = thin_border()
        if c_idx in [1, 2, 3, 11]: cell.alignment = center()
        if c_idx in [7, 8, 9]: cell.number_format = '#,##0.2f'

# -------------------------------------------------------------------------
# 시트 3: 2. CPU_Memory_과다할당 권고군 (Step4 산출물 매핑)
# -------------------------------------------------------------------------
ws_waste = wb.create_sheet("2. 과다할당 권고군")
ws_waste.sheet_view.showGridLines = False

headers_waste = ["Date", "Namespace", "Workload Type", "Pod Name", "Container", "Minutes Running", "Max Request", "Max Limit", "P95 Usage", "Throttled Max", "Allocated Hours", "Usage Hours", "Waste Hours", "Status"]

for col_idx, h in enumerate(headers_waste, 1):
    cell = ws_waste.cell(row=1, column=col_idx, value=h)
    cell.font = ft(bold=True, color=C["white"])
    cell.fill = fill(C["hdr_mid"])
    cell.alignment = center()
    cell.border = thin_border()

for r_idx, row in enumerate(df_waste.itertuples(index=False), 2):
    bg_color = C["gray_row"] if r_idx % 2 == 1 else C["white"]
    for c_idx, val in enumerate(row, 1):
        cell = ws_waste.cell(row=r_idx, column=c_idx, value=val)
        cell.font = ft(size=9)
        cell.fill = fill(bg_color)
        cell.border = thin_border()
        if c_idx in [1, 2, 3, 14]: cell.alignment = center()
        if c_idx in [7, 8, 9, 11, 12, 13]: cell.number_format = '#,##0.1f'

# 📊 낭비 자산 가시성을 높이기 위한 데이터바 조건부 서식 부여 (Waste Hours 컬럼 저격)
if len(df_waste) > 0:
    ws_waste.conditional_formatting.add(
        f"M2:M{len(df_waste)+1}", 
        DataBarRule(start_type="num", start_value=0, end_type="max", color="F2DCDB", showValue=True)
    )

# -------------------------------------------------------------------------
# 시트 4: 5. 리소스 미설정 위반군 (Step4 산출물 매핑)
# -------------------------------------------------------------------------
ws_viol = wb.create_sheet("3. 스펙 미설정 위반군")
ws_viol.sheet_view.showGridLines = False

for col_idx, h in enumerate(headers_risk, 1): # 위험군과 데이터 구조가 동일하므로 헤더 재사용
    cell = ws_viol.cell(row=1, column=col_idx, value=h)
    cell.font = ft(bold=True, color=C["white"])
    cell.fill = fill(C["hdr_dark"])
    cell.alignment = center()
    cell.border = thin_border()

for r_idx, row in enumerate(df_viol.itertuples(index=False), 2):
    bg_color = C["gray_row"] if r_idx % 2 == 1 else C["white"]
    for c_idx, val in enumerate(row, 1):
        cell = ws_viol.cell(row=r_idx, column=c_idx, value=val)
        cell.font = ft(size=9)
        cell.fill = fill(bg_color)
        cell.border = thin_border()
        if c_idx in [1, 2, 3, 11]: cell.alignment = center()

# 📐 전 시트 스캔 및 글로벌 열 너비 최적화 자동화 구문
print("📐 Executing column auto-fit layout optimizer...")
for ws in wb.worksheets:
    if ws.title == "0. 종합 대시보드": continue
    ws.freeze_panes = "A2"
    ws.auto_filter.ref = ws.dimensions
    for col in ws.columns:
        max_len = max(len(str(cell.value or '')) for cell in col)
        col_letter = get_column_letter(col[0].column)
        ws.column_dimensions[col_letter].width = max(max_len + 3, 12)

# 💾 [최종 마감 및 S3 슈팅]
target_date_str = datetime.now(KST).strftime("%Y%m%d")
excel_filename = f"res_usage_report_{cluster_target.lower()}_{target_date_str}.xlsx"
local_excel_path = OUTPUT_DIR / excel_filename

wb.save(local_excel_path)
print(f"✅ Local report compilation completed: {local_excel_path.name}")

# AIStor 원격 업로드 집행
s3_client = get_minio_client()
bucket_name = os.getenv("MINIO_REPORT_BUCKET", "devops-test")
object_key = f"reports/{cluster_target.lower()}/{excel_filename}"

print(f"📤 Uploading final executive ledger to AIStor Tables Ecosystem...")
try:
    s3_client.upload_file(str(local_excel_path), bucket_name, object_key)
    print(f"🏁 [FinOps 완수] Report securely stored ➡️ S3://{bucket_name}/{object_key}")
except Exception as e:
    print(f"❌ S3 Upload failed due to network anomaly: {str(e)}")

if name == "main": main()

따라서 이 API는 1) 이전에 설계한 CNPG 메타 DB를 조회하는 방식(가장 추천) 또는 2) MinIO AIStor API를 극한으로 최적화하여 래핑하는 방식 중 하나를 선택하여 구현해야 함.

방안 1. 이전에 기획한 CNPG 메타 DB를 활용하는 방식 (강력 추천)

Kafka ➡️ Redis ➡️ CNPG 파이프라인을 통해 메타데이터를 이미 적재하고 있다면, S3를 찌를 필요가 전혀 없음. API 서버는 단순한 SQL 쿼리 실행기 역할만 수행하므로 리스크가 제로에 가까움.

① 특정 Prefix 아래의 총 사용량 및 파일 수 조회 (/usage)

• 동작 방식: Prefix 테이블에서 ID를 찾고, 파티셔닝된 Object 테이블에서 COUNT와 SUM을 집계하여 리턴함.
• API 구현 로직 (SQL 기반): ```sql
• • API 서버 내부에서 실행할 쿼리 SELECT COUNT(*) as total_objects, COALESCE(SUM(size), 0) as total_bytes FROM storage_objects WHERE prefix_id = (SELECT id FROM storage_prefixes WHERE full_path = :request_prefix);

* **이점:** 수억 건의 데이터가 있어도 인덱스 타겟 파티션 1개만 스캔하므로 **수 밀리초(ms) 안에 응답**이 끝남.

#### ② 특정 Prefix 아래의 객체 리스트 조회 (`/objects`)

* **위험 요소:** 만약 특정 prefix 아래에 수백만 개의 파일이 있다면, 이를 한 번에 JSON 배열로 리턴하는 순간 API 서버 메모리가 고갈(OOM)됨.
* **해결책:** Cursor 기반 페이지네이션(Pagination)을 API 스펙에 반드시 강제해야 함.
* **API 구현 스펙 예시:**
* **요청:** `GET /api/v1/storage/objects?prefix=dir1/&limit=1000&cursor_file=last_seen_file.txt`
* **쿼리:**
```sql
SELECT name, size, created_at
FROM storage_objects
WHERE prefix_id = :prefix_id AND name > :cursor_file
ORDER BY name ASC
LIMIT :limit;

방안 2. DB를 쓰지 않고 MinIO AIStor API를 직접 래핑하는 방식

만약 실시간 정합성을 위해 메타 DB를 거치지 않고 MinIO 스토리지를 직접 찔러야 한다면, API 서버 내부에서 MinIO Go/Python SDK를 사용할 때 S3 호환 파라미터를 세밀하게 통제해야 함.

① 특정 Prefix 아래의 사용량 조회 ➡️ MinIO Admin API 활용

• S3 API에는 특정 폴더의 크기를 바로 리턴해주는 표준 API가 없음. 원래는 전체 리스팅을 해서 합산을 해야 하지만, 상용 MinIO AIStor는 Admin API를 통해 서버 측에서 계산된 프리픽스별 디스크 사용량(DataUsageInfo)을 제공함.
• API 서버 구현 방식 (Python MinIO Admin SDK 예시):

  # API 서버 내부 가상 코드
  from minio import Minio
  # Admin 클라이언트를 통해 S3 리스팅 없이 서버 측 집계 데이터 반환 요청
  # (주의: 이 또한 수억 건인 경우 백엔드 가비지 컬렉션 주기에 따라 약간의 레이턴시가 있을 수 있음)
  usage_info = minio_admin_client.get_bucket_usage(bucket_name, prefix=target_prefix)
  return {
    "size_bytes": usage_info.size,
    "object_count": usage_info.object_count
  }
  

#### ② 특정 Prefix 아래의 객체 리스트 조회 ➡️ S3 Delimiter & ContinuationToken 활용

* **계층형 탐색 강제 (`Delimiter='/'`):** 수억 개의 하위 파일을 다 가져오지 않고, 파일 시스템 탐색기처럼 딱 현재 레벨의 하위 폴더(Prefix)와 파일 목록만 끊어서 가져오도록 래핑해야 함.
* **ContinuationToken 토스:** MinIO가 리턴해주는 페이징용 포인터(`NextContinuationToken`)를 API 서버가 보관하지 말고, 최종 클라이언트(포털 또는 Workbench)에게 그대로 넘겨주어 무상태(Stateless)를 유지해야 함.
* **API 서버 구현 스펙 예시:**
* **요청:** `GET /api/v1/storage/raw-list?prefix=dir1/&token=xyz...`
* **API 서버 내부 로직:**
```python
# API 서버가 MinIO SDK를 호출할 때 설정을 극단적으로 타이트하게 제한
objects = minio_client.list_objects(
    bucket_name, 
    prefix=requested_prefix, 
    recursive=False, # ◀ 수억 건 폭주를 막기 위해 하위 폴더 재귀 탐색 절대 금지
    delimiter='/',   # ◀ 딱 한 단계 아래의 구조만 단건 분리
    max_keys=100,    # ◀ 한 번에 최대 100개씩만 페이징
    start_after=requested_token
)

💡 DevOps 관점의 최종 권장 아키텍처 결론

• 사용량/용량 조회(COUNT/SUM): MinIO Admin API를 래핑하거나, Prometheus 메트릭 엔드포인트를 API 서버가 파싱해서 던져주는 방식이 안전함.
• 객체 리스트 조회(ListObjects): 6,000개 테넌트 규모에서는 결국 유저들이 각자 자기가 구축한 메타 DB(방안 1)를 조회하게 유도하고, 진짜 S3 스토리지 다이렉트 리스팅 API는 아주 특수한 관리자 UI나 Workbench 초기 로딩 시점에만 Strict 페이징(Recursive=False, MaxKeys=100)을 걸어 제한적으로 제공하는 것이 리스크를 관리하는 정석임.

이 경우 API 서버의 역할은 무거운 데이터를 직접 적재하는 것이 아니라, MinIO AIStor가 내부적으로 이미 잘 만들어서 제공하고 있는 기본 기능들(S3 API, Admin API, Prometheus 메트릭, Audit Log Webhook)을 조합하여 상위에 보안과 거버넌스 레이어를 씌워주는 "경량 API 게이트웨이(BFF, Backend for Frontend)" 형태로 가져가는 것이 가장 이상적임.

기존 API 서버에 추가 개발하여 DevOps 거버넌스를 고도화할 수 있는 핵심 API 세트를 아래와 같이 제안함.

1. 보안 및 인증 위임 레이어 (Access Control API)

6,000명이 넘는 개인 테넌트 유저가 MinIO의 Raw AccessKey/SecretKey를 직접 소유하게 만드는 것은 보안상 매우 위험함. 이를 API 서버가 숨겨주고 제어해야 함.

• POST /api/v1/storage/presigned (임시 접근 URL 발급 API)
• 기능: Workbench 커스텀 앱이나 유저가 파일 업로드/다운로드가 필요할 때 API 서버 요청함. API 서버는 내부 마스터 권한으로 MinIO SDK를 호출하여 특정 Object에 대해 5분~30분간만 유효한 Presigned URL을 생성해 반환함.
• 이점: 유저에게 영구 자격증명을 주지 않고도 S3 API 보안을 완벽하게 통제할 수 있음.

• GET /api/v1/storage/sts-token (임시 자격증명 발급 API)
• 기능: AWS STS(AssumeRole) 호환 기능을 활용하여, 해당 유저의 개인 Namespace/Prefix 구역만 딱 1시간 동안 접근할 수 있는 임시 토큰을 발급함. Workbench 내 데이터 사이언스 코드(Python 등)가 대량의 S3 I/O를 직접 쳐야 할 때 유용함.

2. 무거운 DB 없는 실시간 거버넌스 레이어 (Metrics API Wrapper)

MinIO AIStor는 버킷별, 프리픽스별 용량 및 객체 수 통계를 내부적으로 이미 계산하여 Prometheus 엔드포인트(/minio/v2/metrics/cluster)로 실시간 제공하고 있음. 메타 DB를 안 만들어도 이 메트릭을 가공하면 원하는 조회가 가능함.

• GET /api/v1/governance/tenant/{user_id}/summary (개인별 용량/객체 수 즉시 조회 API)
• 기능: API 서버가 호출되면, MinIO의 Prometheus 메트릭 파이프라인에서 해당 유저의 프리픽스/버킷 관련 메트릭(minio_bucket_usage_total_bytes, minio_bucket_usage_object_total_count)만 긁어서 정제 후 JSON으로 리턴함.
• 이점: 수억 건의 객체가 있어도 S3 ListObjects를 치지 않고 O(1) 성능으로 유저별 실시간 사용량과 쿼터 임박 여부를 대시보드(포탈)에 뿌려줄 수 있음.

3. 프로비저닝 및 자원 통제 레이어 (Lifecycle & Quota API)

Workbench와 연동하여 사용자가 추가되거나 휴면 상태가 될 때 자원을 자동 제어하는 API임.

• POST /api/v1/governance/tenant/setup (테넌트 스토리지 자동 개설 API)
• 기능: 신규 유저가 Workbench를 할당받을 때 연동 호출됨. MinIO Admin API SDK를 이용해 1) 개인 버킷/프리픽스 생성, 2) 전용 정책(Policy) 바인딩, 3) 스토리지 용량 한도(Hard Quota) 설정을 코드 한 줄로 원스톱 처리함.

• PUT /api/v1/governance/tenant/lifecycle (수명주기 동적 제어 API)
• 기능: 주간 분석(Step 6) 결과 낭비군으로 분류되거나 일정 기간 미접속한 유저의 프리픽스에 대해 미설정 상태였던 객체 만료 정책(Object Lifecycle Management, 예: 30일 지나면 자동 삭제)을 강제로 주입하거나 쿼터를 축소함.

4. 실시간 감사 및 킬러 쿼리 차단 레이어 (Audit Webhook Receiver API)

MinIO AIStor는 모든 S3 API 요청 이력을 JSON 형태로 외부 웹훅에 실시간 전송하는 기능(Audit Logs Webhook)을 내장하고 있음. 기존 API 서버에 이를 받아먹는 리시버 엔드포인트를 하나 뚫어두는 것을 강력히 권장함.

• POST /api/v1/logs/audit-receiver (감사 로그 웹훅 수신 및 필터링 API)
• 기능: MinIO가 쏴주는 실시간 실황 로그를 수신함. 데이터베이스에 다 저장할 필요 없이, API 서버 메모리 상에서 특정 위험 패턴만 필터링하여 이벤트를 발생시킴.
• 추적 대상 패턴 리스트:

1. 403 Access Denied 폭주 유저: Workbench 내부에서 잘못된 스크립트나 해킹 시도로 타인의 영역을 찌르는 유저 탐지.
2. Bulk Delete 패턴: 특정 프리픽스 하위의 자산을 무단으로 대량 삭제하는 행위 실시간 인지.
3. Slow S3 Operations 탐지: 하나의 버킷에 몇억 건이 있는 상태에서 유저가 실수로 Delimiter 없이 전체 ListObjects를 날려 스토리지 클러스터 전체에 고부하를 주는 행위를 실시간으로 가로채어 해당 유저 정보를 DevOps 알람망(Slack 등)에 공유함.

요약

무거운 아키텍처를 걷어낸 지금 상태에서는 "MinIO가 이미 알고 있는 정보(Prometheus 메트릭, Webhook 로그)"를 API 서버가 중간에서 가볍게 통역(가공)만 해주는 API들을 추가하는 것이 DevOps 관점에서 리스크가 제로에 가까우며 가장 가성비 높은 고도화 전략임.

1. 제어 평면(Control Plane) 안정성 및 확장성 심화

대규모 환경에서는 워커 노드보다 마스터 노드의 제어부 병목으로 인해 클러스터 전체가 먹통이 되는 경우가 더 빈번하게 발생함.

• API Priority and Fairness (APF) 세부 튜닝
• 6,000여 명의 유저 및 각종 자동화 에이전트(ArgoCD, Prometheus 등)가 동시에 API 서버를 찌를 때 발생하는 병목을 제어해야 함.
• 크리티컬한 인프라 에이전트(Cilium, 코어 컨트롤러)의 요청이 개인 유저의 무분별한 kubectl 요청에 밀려 드롭되지 않도록 FlowSchema와 PriorityLevelConfiguration을 커스텀하게 분리 및 가중치 조정해야 함.

• etcd 쿼터 및 조각화(Fragmentation) 모니터링 자동화
• etcd 용량 상한을 8GB로 늘렸더라도 무분별한 객체 생성/삭제가 반복되면 DB 내부 단편화가 진행되어 성능이 폭락함.
• etcdctl defrag 명령을 상시 안전하게 실행할 수 있는 유지보수 런타임 워커를 구현하고, etcd의 wal_fsync_duration_seconds 메트릭을 통해 디스크 I/O 레이턴시를 밀리초(ms) 단위로 감시해야 함.

2. 노드 OS 커널(Kernel) 및 컨테이너 런타임 최적화

쿠버네티스는 결국 리눅스 커널 위에서 돌기 때문에, 노드 OS 레벨의 자원 고갈(Resource Exhaustion) 방지책이 필수적임.

• 대규모 네트워크 커넥션 수용을 위한 호스트 sysctl 튜닝
• 수만 개의 파드가 동시다발적으로 통신할 때 호스트 커널의 네트워크 버퍼가 가득 차 패킷 드롭이 발생함.
• 각 워커 노드의 /etc/sysctl.conf에 아래 파라미터를 강제 적용하는 데몬셋(DaemonSet) 또는 OS 이미지 튜닝이 요구됨.
• net.core.somaxconn: 최소 32768 이상 상향 (TCP 백로그 큐 확장)
• net.ipv4.tcp_max_syn_backlog: 대규모 인그레스 노드의 SYN Flood 방어 및 큐 확장
• net.core.netdev_max_backlog: 커널이 처리하기 전 네트워크 카드가 받아두는 패킷 큐 상향

• 커널 스레드 및 메모리 맵 한계 확장
• 개인 테넌트 파드가 스레드를 무분별하게 생성하거나 메모리를 과도하게 매핑할 때 노드 커널 락(Lock)이 걸리는 현상을 방지해야 함.
• kernel.pid_max(최소 4194304) 및 vm.max_map_count(최소 262144 이상)를 조정하여 프로세스/스레드 ID 고갈을 방지함.

• Kubelet oomScoreAdj 및 Eviction 정책 정교화
• 노드에 메모리 고갈 발생 시 커널 OOM 킬러가 무작위로 인프라 파드를 죽이지 않도록 통제해야 함.
• Kubelet 시스템 컴포넌트와 Cilium 에이전트, 스토리지 CSI의 oomScoreAdj 값을 가장 낮게 고정하여(-997~-998) 최후까지 생존하도록 방어벽을 쳐야 함.

3. Cilium eBPF 대규모 환경 특화 고급 튜닝

Cilium은 대규모 환경에서 최상의 성능을 내지만, 설정 값을 대규모 카디널리티에 맞게 늘려주지 않으면 eBPF 맵(Map) 내부 공간 부족으로 네트워크가 통째로 단절됨.

• Cilium eBPF Map Size 한계치 상향 고정
• 6,000개 네임스페이스의 수많은 파드가 커넥션을 맺으면 Cilium 내부의 Connection Tracking(ct) 맵과 NAT 맵이 가득 참.
• CiliumConfig에서 아래 파라미터를 기본값보다 대폭 상향하여 BPF map is full 에러를 선제 차단해야 함.
• bpf-ct-global-any-max: 커넥션 추적 테이블 용량 확장 (최소 1,000,000 이상)
• bpf-nat-global-max: NAT 매핑 테이블 용량 확장

• Cilium Host Routing 및 eBPF 기반 마스커레이딩 활성화
• 패킷이 노드의 iptables 서브시스템을 타게 되면 노드 CPU 사용량이 폭증하고 레이턴시가 늘어남.
• bpf.masquerade=true 및 tunnel=disabled (Direct Routing / BGP 환경) 설정을 통해 리눅스 네트워크 스택을 완전히 우회하고 eBPF가 네트워크 카드 장치에서 파드 가상 인터페이스(veth)로 패킷을 다이렉트 바이패스하도록 아키텍처를 고정해야 함.

• 네트워크 정책(CiliumNetworkPolicy) 결합 오버헤드 통제
• 수천 개의 테넌트가 개별적으로 네트워크 정책을 남발하면 eBPF 프로그램 컴파일 및 전파 오버헤드로 인해 Kubelet 파드 생성 속도가 극도로 저하됨.
• 개별 정책보다는 네임스페이스 레이블 패턴 기반의 글로벌 정책(CiliumClusterwideNetworkPolicy) 위주로 추상화하여 노드 전파 카디널리티를 최소화해야 함.

4. 대규모 GitOps 아키텍처 및 SRE 모니터링 부하 통제

6,000개가 넘는 애플리케이션 정의를 ArgoCD 등으로 동기화할 때 발생하는 성능 감쇄를 튜닝해야 함.

• ArgoCD 컨트롤러 셔딩(Sharding) 및 API 쿼트 분산
• 단일 ArgoCD Application Controller가 6,000개 테넌트의 상태 매칭(Reconcile)을 수행하면 메모리 고갈(OOM) 및 쿠버네티스 API 속도 제한(Rate Limit)에 걸려 동기화가 무한 지연됨.
• ARGOCD_CONTROLLER_REPLICAS를 증설하고 클러스터 셔딩 알고리즘을 도입하여 동기화 부하를 분산해야 하며, ArgoCD 설정에서 K8s API 자원 조회 시 주기적 풀링(Polling) 대신 웹훅(Webhook) 트리거 및 캐시 기반 매칭 구조로 전환해야 함.

• SRE 관점의 Cilium 핵심 텔레메트릭 메트릭 감시
• 프로메테우스에서 단순히 파드 CPU/Mem만 볼 것이 아니라, 아래 Cilium 전용 저수준 메트릭을 SRE 골든 시그널 대시보드에 무조건 포함해야 함.
• cilium_drop_count_total: 커널 단에서 드롭된 패킷 수 및 원인 코드(Reason) 추적
• cilium_bpf_map_ops_total: eBPF 맵 연산 실패율 감시
• cilium_errors_total: 에이전트 내부 오류 발생 카운트

5. 스토리지(MinIO AIStor) 네트워크 I/O 병목 및 백업 거버넌스

데이터 레이크하우스 아키텍처의 핵심인 데이터 무결성 및 고속 전송을 위한 SRE 관점의 인프라 제어 지점임.

• 네트워크 인터페이스 카드(NIC) 본딩 및 인터럽트(IRQ) 분산
• MinIO 대규모 분산 환경에서 특정 노드의 CPU 0번 코어만 네트워크 인터페이스의 소프트웨어 인터럽트(softirq) 처리에 매몰되어 시스템이 멈추는 현상이 발생할 수 있음.
• OS 레벨에서 irqbalance 설정을 최적화하고 다중 큐(Multi-queue NIC) 설정을 켜서 데이터 레이크하우스 트래픽 처리가 전체 CPU 코어에 균등 분산되도록 통제해야 함.

• 테넌트별 자원 회수(Garbage Collection) 및 벨레로(Velero) 백업 세분화
• 개인이 쓰다 버린 휴면 네임스페이스의 스토리지 알박기 자원을 정기적으로 회수하기 위해, 주기적으로 주간 분석 리포트의 낭비 지표와 연동하여 자동 kubectl delete를 수행하는 가비지 컬렉터 워커 가동이 필요함.
• 6,000개 전체를 한 번에 백업하면 스냅샷 레이어에서 타임아웃이 나므로, 자원 중요도(Tiering) 레이블에 따라 백업 주기와 대상을 세분화하여 오브젝트 스토리지에 격리 저장해야 함.

구상 중인 클러스터 전체 자원 관리 및 거버넌스 고도화를 위한 아키텍처 수립 전략을 4가지 핵심 필라(Pillar)로 나누어 상세히 정리해 드립니다.

1. 6,000+ 개인 네임스페이스(Namespace) 대규모 관리 전략

개인형 테넌트 환경의 핵심은 "중앙 통제된 자동화"입니다. 유저가 직접 네임스페이스를 제어하게 두지 않고, 포털이나 GitOps 파이프라인이 생성 시점부터 규격을 강제해야 합니다.

① 메타데이터 표준화 (Labels & Annotations)

고성능 모니터링 및 자동 청소(Garbage Collection)를 위해 네임스페이스에 반드시 다음 메타데이터를 주입해야 합니다.

• Labels (스케줄링 및 비용 정산용):
• governance.io/tenant-owner: "user_id" (개인 식별)
• governance.io/tier: "developer" (자원 할당 등급)
• governance.io/sync-status: "managed" (ArgoCD 동기화 대상 여부)

• Annotations (생명주기 및 거버넌스 제어용):
• governance.io/ttl-expire-date: "2026-12-31" (휴면 계정 자동 삭제 워커용 배정 만료일)
• governance.io/purpose: "sandbox-testing"

② ResourceQuota 설계 (etcd 폭발 방지책)

6,000개 네임스페이스에서 무분별하게 객체(Object)가 생성되면 etcd 메모리가 고갈됩니다. 자원 용량뿐만 아니라 객체 개수 제한(Object Count Quota)을 무조건 걸어야 합니다.

• 개인용 기본 ResourceQuota 프로필 예시:

  apiVersion: v1
  kind: ResourceQuota
  metadata:
  name: personal-user-quota
  spec:
  hard:
    # 1. 컴퓨팅 자원 상한 (Tight하게 설정)
    requests.cpu: "2"
    requests.memory: "4Gi"
    limits.cpu: "4"
    limits.memory: "8Gi"
    # 2. etcd 보호를 위한 객체 개수 제약 (가장 중요)
    pods: "5"
    services: "3"
    secrets: "10"
    configmaps: "10"
    persistentvolumeclaims: "2"
    replicationcontrollers: "0" # 레거시 차단
  

#### ③ LimitRange 설계 (기본값 강제화)

개발자가 `request/limit` 설정을 누락하고 파드를 배포하면 노드가 동반 고갈(Noisy Neighbor)됩니다. 네임스페이스 단위로 기본값을 주입합니다.

* **권장 비율:** CPU는 오버커밋을 허용하되(`Request:Limit = 1:2`), Memory는 시스템 안정성을 위해 최대한 동일하게(`Request:Limit = 1:1`) 가져가는 것이 대규모 환경의 OOM 방지에 유리합니다.

---

### 2. 주요 쿠버네티스 오브젝트 자원 거버넌스

각 핵심 리소스들이 무분별하게 슬롯을 차지하거나 모자라지 않도록 격리하는 세부 제어 가이드라인입니다.

| 오브젝트 종류 | 대규모 환경에서의 자원 관리 및 거버넌스 방법론 |
| --- | --- |
| **Service / Ingress** | • 개인 NS 내 `Type: LoadBalancer` 생성 권한을 RBAC으로 완전히 차단합니다.<br>

<br>• 오직 `ClusterIP`만 허용하고, 외부 노출은 중앙 플랫폼 팀이 관리하는 공유형 Ingress(또는 Gateway API) 프록시를 통해 라우팅 규칙만 연동합니다. |
| **Secret / ConfigMap** | • etcd 단일 객체 한계(1MB) 및 전체 용량 압박을 막기 위해 개인 NS당 총 용량(합산 크기)을 모니터링해야 합니다.<br>

<br>• ArgoCD 플러그인이나 외부 Secret Store CSI Driver(Vault 연동)를 사용하여 파드 메모리에 직접 주입하고, k8s 내부 Secret 객체 남발을 억제합니다. |
| **PV / PVC / StorageClass** | • `ResourceQuota`에 `requests.storage: "50Gi"` 형태로 네임스페이스당 스토리지 총량을 강력하게 제한합니다.<br>

<br>• 개인용 StorageClass는 프로덕션용 고성능 NVMe와 분리하여 용량 제한 및 Thin Provisioning이 적용된 저가형 스토리지 풀로 바인딩합니다.<br>

<br>• `ReclaimPolicy`는 `Delete`로 설정하여 개인 NS 삭제 시 디스크 잔재가 남지 않도록 자동화합니다. |
| **ServiceAccount / Role / RoleBinding** | • 개인 유저에게는 오직 해당 네임스페이스 내부만 제어할 수 있는 `Role`(`edit` 또는 `view` 수준)만 부여합니다.<br>

<br>• `ClusterRoleBinding` 권한을 철저히 차단하여, 개인 NS의 파드가 다른 사용자나 시스템 인프라 영역(kube-system 등)을 공격하거나 조회할 수 없도록 논리적 멀티테넌시(Soft Isolation)를 완성합니다. |

---

### 3. 성능 최적화 및 오토스케일링 (HPA, VPA, KEDA) 튜닝

대규모 환경에서 오토스케일러들이 서로 충돌(Flapping)하지 않도록 역할을 명확히 인프라 정책으로 정의해야 합니다.

#### ① VPA (Vertical Pod Autoscaler) ➡️ 개인 워크로드 전담

* **전략:** 6,000명의 개발자는 본인 파드의 적정 자원량을 모릅니다. 개인 NS에 배포되는 파드들은 VPA를 `updateMode: "Auto"` 또는 안심할 수 없다면 `Initial`(생성 시점에만 적용)로 지정합니다.
* **주의사항:** 무한 증설을 막기 위해 VPA `ResourcePolicy` 내부에 `maxAllowed` 값을 설정하여 개인이 선점할 수 있는 단일 파드의 최대 크기 상한을 걸어두어야 합니다.

#### ② HPA (Horizontal Pod Autoscaler) ➡️ 공유 플랫폼 서비스 전담

* **전략:** 데이터 레이크하우스 쿼리 엔진이나 공용 미들웨어는 HPA를 씁니다.
* **트래픽 급변(Spike) 대응 튜닝:** 급격한 스케일 업/다운으로 클러스터가 요동치는 것을 막기 위해 `behavior` 설정을 튜닝해야 합니다.
```yaml
behavior:
  scaleDown:
    stabilizationWindowSeconds: 600 # 다운스케일은 10분간 관측 후 보수적으로 진행
    policies:
    - type: Percent
      value: 10
      periodSeconds: 60

③ KEDA (Event-driven Autoscaler) ➡️ 비동기 파이프라인 전담

• 앞서 구상하신 AIStor ➡️ Kafka 이벤트를 받아 처리하는 컨수머(Consumer) 앱 영역에 KEDA를 전면 배치합니다. CPU 기준이 아닌 Kafka Lag(밀린 메시지 수)를 기준으로 파드 개수를 0개에서 수십 개까지 고속 스케일링하게 세팅합니다. (activationScaleObject 활용으로 0개 유지가 핵심 - 자원 절약)

⚠️ 절대 금지 규칙: 동일한 파드의 동일한 메트릭(예: CPU 사용량)에 HPA와 VPA를 동시에 인게이지하면 안 됩니다. 자원 크기를 늘릴지, 파드 수를 늘릴지 두 엔진이 경합하다가 시스템이 다운됩니다.

4. 고부하 대규모 클러스터 및 Cilium 고급 튜닝 요소 리스트

6,000개 네임스페이스 구조에서 노드 수백 대가 안정적으로 구동되기 위한 플랫폼 인프라 최적화 체크리스트입니다.

① K8s 코어 및 OS 커널 레벨 튜닝

• etcd 성능 극대화:
• etcd 데이터 디렉토리를 반드시 가장 빠른 로컬 NVMe SSD 단독 드라이브로 격리합니다.
• --quota-backend-bytes: 기본 2GB에서 상한선인 8GB로 확장하여 대규모 네임스페이스 객체 수용 능력을 확보합니다.
• --auto-compaction-retention: etcd 단편화 방지를 위해 주기적 컴팩션을 1 (1시간 단위) 혹은 분 단위로 타이트하게 설정합니다.

• API Server 포화 방지:
• --max-requests-inflight 및 --max-mutating-requests-inflight 값을 인프라 사양에 맞춰 기본값보다 2~3배 상향하여 대규모 수집기(Prometheus 등)와 GitOps 동기화 요청 폭주 시 발생하는 429 에러를 방어합니다.

• Kubelet 이베이전(Eviction) 기준 고도화:
• 노드 다운을 막기 위해 imageGCHighThresholdPercent: 80, imageGCLowThresholdPercent: 70으로 설정하여 디스크 부족 현상을 선제 관리합니다.

② Cilium CNI 기반 네트워크 및 보안 튜닝

• XDP (eXpress Data Path) 활성화:
• bpf.masquerade=true 및 enable-xdp=true 설정을 켭니다. 패킷 처리가 리눅스 커널 네트워킹 서브시스템(iptables/conntrack) 진입 전 네트워크 카드(NIC) 드라이버 레이어에서 eBPF로 처리되므로, 고부하 환경에서 네트워크 처리로 인한 커널 CPU 오버헤드가 극적으로 감소합니다.

• BGP 및 ECMP (Equal-Cost Multi-Path) 최적화:
• Cilium 내장 BGP Control Plane을 탑재하여 탑오브랙(ToR) 스위치와 다중 경로 라우팅(ECMP)을 맺습니다. 특정 네트워크 노드 장비 장애 시 서브밀리초 단위의 Failover를 보장합니다.

• Hubble 텔레메트릭 오버헤드 통제:
• Hubble 오픈소스 가시성은 매우 강력하지만, 6,000개 테넌트의 모든 L7 트래픽(HTTP/gRPC)을 파싱하면 모니터링 에이전트 자체가 노드 자원을 고갈시킵니다.
• 가이드: 개인 NS 구역은 L3/L4 레벨로 흐름 감지를 제한하고, 오직 공유 플랫폼 구역에만 L7 파싱 및 Hubble 메트릭 수집을 선택적으로 활성화(clean-up 정책 연계)해야 합니다.

• ClusterMesh 최적화:
• 멀티 클러스터 간 전역 서비스 동기화 시, 상태 전파 카디널리티를 줄이기 위해 meshAuth 기능을 커널단에서 최적화하고 에이전트 간 하트비트 주기를 조율합니다.

이 거버넌스 하이레벨 가이드라인을 플랫폼 구축 명세의 초석으로 삼으시면, 대규모 멀티테넌트 환경에서도 인프라 통제력을 잃지 않고 안정적인 데이터 플랫폼 레이크하우스를 완성하실 수 있습니다. 이 중 특정 컴포넌트(예: etcd 세부 파라미터나 Cilium BGP 연동 YAML 구성)의 구체적인 구현 단계로 진입하실 때 다시 말씀해 주세요!

구축하신 시스템의 목적 부합성을 최종 검증해 드리고, 매주 개발 부서 담당자들과 실무 협의(Communication) 시 무기로 활용하실 수 있는 ‘데이터 기반 자원 조정 가이드라인 핵심 자료’를 정리해 드립니다.

Part 1. 파이프라인 아키텍처 및 컨셉 점검 (Validation)

구상하신 운영 방식은 목적에 100% 부합하며, 다음과 같은 아키텍처적 강점을 가집니다.

1. 글로벌 Fetch와 클러스터별 루프 분리 (step1 ➡️ step2~6)

• 대규모 환경에서 프로메테우스 스크랩(Scrape) 부하를 최소화하기 위해 글로벌 메트릭을 새벽에 단 한 번 가져오고(step1), 다운스트림 가공 과정만 도메인별(COMPUTE, STORAGE) 서브폴더 파티션으로 격리해 순회하는 구조는 데이터 레이크하우스 구축 시 성능 경합을 피할 수 있는 가장 훌륭한 선택입니다.

2. AIStor 내 데이터 생명주기 및 적재 규칙

• raw/, enrich/, excel/로 일단위 보관 체계를 잡은 것은 원시 로그 추적 및 장기 트렌드 분석을 위해 완벽한 구조입니다.

3. 배치 범위의 유연성 (Daily vs. Weekly/Monthly)

• start-date와 end-date를 파라미터로 받아 하방 파이프라인을 재구동할 수 있게 설계한 덕분에, 주간 단위 배치 워크로드의 변동성(Spike)이나 월간 정산 주기에 맞춰 유연한 통계 윈도우(Time-window)를 가져갈 수 있습니다.

Part 2. 개발 담당자 협의용 자원 조정 가이드라인 (기본 자료)

개발자들은 대개 서비스 안정성을 이유로 자원 축소(Downscaling)에 보수적(반발)입니다. 따라서 협의 시에는 감정적 설득이 아닌, "우리가 수집한 P95 데이터와 커널 가시성(Throttling/OOM)"을 기반으로 객체화된 가이드를 제시해야 합니다.

1. 자원 조정의 대원칙 (Platform & Dev Shared Goal)

• 목적: 애플리케이션의 안정성(Reliability)을 최우선으로 확보하되, 무의미하게 버려지는 선점 자원(Locked Resource)을 회수하여 전체 인프라의 가용 슬롯을 확보함.
• 약속: 가이드라인에 따라 자원을 조정한 후 모니터링 중 장애 징후가 보이면, 즉시 이전 상태로 Rollback 하거나 즉각 재조정(Upscaling)을 보장함.

2. 유형별 자원 조정 판단 기준 및 권장 액션 (Thresholds)

주간 데이터에서 최소 7일 이상 지속(지속성 7회 이상) 감지된 워크로드를 대상으로 아래 기준을 적용합니다.

🔴 유형 A. 과다 할당 위반군 (Over-allocated / Waste)

개발팀이 안정성을 이유로 필요 이상의 자원을 선점해 둔 상태입니다.

• 판단 기준: 7일간 CPU/Memory 평균 실효 활용률(Util) < 15%이면서 동시에 P95 Usage가 Request의 30% 미만인 경우.
• 가이드 가치 (Talk): *"현재 할당된 스펙의 80% 이상이 단 한 번도 사용되지 않고 낭비되고 있어 타 모듈의 파드 배치를 방해하고 있습니다."*
• 조정 가이드: * Request 조정: 관측된 P95 최대 사용량 * 1.25 (25% 마진) 수준으로 Request 하향 조정을 권장합니다.
• Limit 설정: Limit은 현재 값을 유지하거나 Request의 2배 수준으로 확보하여 급작스러운 스파이크는 흡수할 수 있도록 안전장치를 제공합니다.

🟡 유형 B. 자원 부족 및 병목 위험군 (Shortage / Performance Risk)

실제 서비스 장애(OOM)가 났거나 커널 단에서 연산 속도 제한이 걸려 성능이 저하되고 있는 위험 상태입니다.

• 판단 기준: 1. Memory: 관측 기간 내 OOMKilled(is_oom_killed) 1회 이상 발생 또는 Memory RSS 사용량이 Limit의 90% 상회.

2. CPU: CFS Throttling Peak(cpu_throttled_max) > 20% 상태가 빈번하여 애플리케이션 지연(Latency) 유발 처리가 확인된 경우.

• 가이드 가치 (Talk): *"해당 모듈은 리소스 부족으로 인해 컨테이너가 강제 종료되거나 커널 단에서 연산 스로틀링(지연)이 걸려 서비스 품질에 악영향을 주고 있습니다."*
• 조정 가이드:
• Memory: 즉시 Limit 규격을 기존 Limit * 1.3 (30% 상향) 조정하거나, 가공 원부의 Mem RSS 최고점 대비 20% 마진을 두어 메모리 누수(Leak) 여부 확인을 요청합니다.
• CPU: Request를 CPU P95 수준으로 현실화하고, Limit을 대폭 늘려주거나 자바/Go의 런타임 스레드 풀 수치를 k8s 쿼터와 매칭하도록 가이드합니다.

🟠 유형 C. 규격 미설정 위반군 (Violation / Missing Settings)

Request나 Limit 중 하나라도 누락되어 노드 전반의 동반 자원 고갈(Noisy Neighbor)을 유발하는 상태입니다.

• 판단 기준: has_no_request == True 또는 has_no_limit == True
• 가이드 가치 (Talk): *"자원 설정이 누락되어 해당 파드가 유실되거나 노드 내 다른 정상 서비스의 자원을 침범할 위험이 있습니다. 클러스터 거버넌스 규격 준수가 강제됩니다."*
• 조정 가이드: 지난 7일간 수집된 실제 daily_enriched 데이터의 평균 사용량을 기반으로 최소한의 Request/Limit 세팅을 의무화합니다.

3. 개발 담당자 커뮤니케이션용 핵심 지표 용어 해석 사전

개발자들이 인프라 메트릭을 쉽게 이해할 수 있도록 리포트 전달 시 아래 용어 설명을 동봉합니다.

• P95 사용량 (95th Percentile): 전체 실행 시간 중 상위 5%에 해당하는 피크 수치입니다. 평균치만 보고 자원을 줄이면 피크 시 서비스가 터지므로, 플랫폼 팀은 개발팀의 안정성을 보장하기 위해 이 P95 수치를 기준선으로 대화합니다.
• CPU Throttling (스로틀링): 컨테이너가 배정된 CPU 한도(Limit)를 초과하여 커널이 강제로 프로세스를 멈추고 대기시킨 비율입니다. 이 수치가 높으면 소스 코드가 아무리 좋아도 인프라 쿼터 제약 때문에 API 응답 속도가 느려집니다.
• Memory RSS vs WorkingSet: WorkingSet은 커널 캐시를 포함한 느슨한 지표이지만, RSS는 프로세스가 물리 메모리에 직접 상주 시킨 진짜 자원 소모량입니다. OOMKilled는 이 RSS가 Limit을 칠 때 발생하므로, 3번 시트의 RSS 데이터를 보고 메모리 증설 여부를 정확히 판단할 수 있습니다.

Part 3. 주간 커뮤니케이션 가동 워크플로우 예시 (일주일에 한번)

매주 월요일 아침, 주간 lookback 파이프라인을 돌려 나온 excel 결과물에서 가장 낭비가 심한 상위 5개 네임스페이스(Step6 파레토 Critical 등급)와 장애 위험 상위 5개 네임스페이스를 추출하여 아래 템플릿 형태로 던져 소통하시면 됩니다.

[주간 자원 거버넌스 공람 템플릿 예시]

• 수신: 메인 모듈 개발 리드
• 발신: 플랫폼 DevOps 파트
• 내용: 지난 주간 COMPUTE 클러스터 정산 결과 기반 최적화 대상 권고

1. 대상 워크로드: order-api-pod (Namespace: production-order)
2. 인프라 진단 데이터:

• 할당 자원: CPU Request 8 Cores / Memory Limit 16 GB
• 실제 사용량: CPU P95 0.8 Cores (활용률 10%) / Mem RSS Peak 4 GB (활용률 25%)
• 특이사항: 7일 연속 피크 타임에도 자원 소모량 변동 없음 (과다 할당 등급: Critical)

3. 플랫폼 팀 제안 권고사항:

• CPU Request를 8 Cores ➡️ 2 Cores로 하향 조정 제안 (Limit은 8 Cores 유지하여 스파이크 보장)
• 회수된 자원은 다음 주 배포 예정인 신규 인프라 데이터 레이크 파드 배치 비용으로 재활용됩니다.

이 정도의 데이터 무기와 기준선을 가지고 개발팀과 협의 체계를 정립하시면, 플랫폼 엔지니어로서 클러스터 전체 리소스 통제권을 강력하게 쥐고 가실 수 있습니다.

개발팀 소통 프로세스를 정립하시면서, 다음 단계인 "K8s 전체 자원 관리 및 거버넌스 고도화 전략(VPA/HPA 연계, 오버커밋 제어 등)"으로 넘어가고 싶으실 때 언제든 질문해 주세요!

1. 기존 데이터 Parquet 로드

    df_pod      = pd.read_parquet(p1)
    df_ns       = pd.read_parquet(p2)
    df_daily_ns = pd.read_parquet(p3)

    # ─── 🎯 [최종 통합 완결 패치] 모든 하방 시트의 KeyError를 여기서 종결 ───

    # [공통] cluster 컬럼 누락 방어 주입
    if 'cluster' not in df_pod.columns:
        df_pod['cluster'] = infra_tag
    if 'cluster' not in df_daily_ns.columns:
        df_daily_ns['cluster'] = infra_tag

    # 💡 [3번 탭 패치 흡수] pv_used_p95 명명 규칙 보정 및 기본값 방어
    if "pv_used_p95" not in df_pod.columns:
        if "pv_usage_p95" in df_pod.columns:
            df_pod["pv_used_p95"] = df_pod["pv_usage_p95"]
        else:
            df_pod["pv_used_p95"] = 0.0  # PV 메트릭이 없는 노드 환경 대비
    if "pv_capacity_max" not in df_pod.columns:
        df_pod["pv_capacity_max"] = 0.0

    # [5번 탭 대응] minutes_running 실행 시간 컬럼명 변형 방어
    if "minutes_running" not in df_pod.columns:
        for alt_col in ["duration_minutes", "running_minutes", "duration"]:
            if alt_col in df_pod.columns:
                df_pod["minutes_running"] = df_pod[alt_col]
                break
        else:
            df_pod["minutes_running"] = 0

    # [6번 탭 대응] mem_allocated_gb_hours 변형 방어
    if "mem_allocated_gb_hours" not in df_pod.columns:
        for alt_col in ["mem_alloc_gb_hours", "mem_alloc_gh", "memory_allocated_gb_hours"]:
            if alt_col in df_pod.columns:
                df_pod["mem_allocated_gb_hours"] = df_pod[alt_col]
                break

    # [6번 탭 대응] pv_allocated_gb_hours 변형 방어
    if "pv_allocated_gb_hours" not in df_pod.columns:
        for alt_col in ["pv_alloc_gb_hours", "pv_alloc_gh", "pv_capacity_gb_hours", "pv_capacity_max"]:
            if alt_col in df_pod.columns:
                df_pod["pv_allocated_gb_hours"] = df_pod[alt_col]
                break

    # ─── 🏁 방어 레이어 마감 후 엑셀 컴파일 및 시트 빌드 로직 정상 진행 ───
    excel_name = f"res_usage_report_{infra_tag.lower()}_{date_tag}.xlsx"
    # ... (이하 동일) ...

... (중략) ...

    print(f"\n" + "="*60)
    print(f"🔄 [서브폴더 파티션 검증 완료] Cluster: {infra_tag} | Date: {date_tag} 컴파일 가동")
    print("="*60)

    # 1. 기존 데이터 로드
    df_pod      = pd.read_parquet(p1)
    df_ns       = pd.read_parquet(p2)
    df_daily_ns = pd.read_parquet(p3)

    # ─── 🎯 [KeyError 해결 패치] 누락된 cluster 컬럼을 동적으로 주입 ───
    if 'cluster' not in df_pod.columns:
        df_pod['cluster'] = infra_tag  # 명세서 작성을 위해 대문자 값(COMPUTE/STORAGE) 강제 주입

    if 'cluster' not in df_daily_ns.columns:
        df_daily_ns['cluster'] = infra_tag

    # 기존 excel_name 정의 및 하위 build_sheet 호출부로 이어짐...
    excel_name = f"res_usage_report_{infra_tag.lower()}_{date_tag}.xlsx"
    # ... (이하 동일) ...

... 기존 openpyxl 및 minio 임포트 생략 ...

─── Main ─────────────────────────────────────────────────

def main(): print("🚀 [Step6 개시] res_usage_ 계열 고도화 엑셀 리포터 엔진 가동...")

# ─── 🎯 [요청 반영] --cluster Argument 파싱 레이어 구성 ───
parser = argparse.ArgumentParser(description="Step6 리소스 정산 엑셀 마스터 빌더")
parser.add_argument(
    "--cluster", 
    type=str, 
    choices=["COMPUTE", "STORAGE", "compute", "storage"],
    help="대상 클러스터 컨텍스트 명시 (COMPUTE 또는 STORAGE)"
)
args = parser.parse_args()

# 1순위: --cluster 인자값, 2순위: CLUSTER_MODE 환경변수
cluster_mode = args.cluster or os.getenv("CLUSTER_MODE", "")
cluster_mode = cluster_mode.upper()  # 대문자 정형화

if cluster_mode:
    print(f"🎯 클러스터 Argument/컨텍스트 고정 완료: [merged/{cluster_mode}/]")
    search_pattern = str(MERGED_DIR / cluster_mode / f"pareto_ns_{cluster_mode}_*.parquet")
else:
    print("🔍 클러스터 미지정: merged/ 하위 모든 클러스터 서브폴더(*)를 자동 순회 탐색합니다.")
    search_pattern = str(MERGED_DIR / "*" / "pareto_ns_*.parquet")

pareto_files = glob.glob(search_pattern)

if not pareto_files:
    print(f"❌ 오류: 탐색된 파레토 가공원부(*.parquet) 파일이 없습니다. 타겟 경로를 확인하세요: {search_pattern}")
    return

processed_count = 0
for file_path in pareto_files:
    filename = os.path.basename(file_path)
    match = re.search(r"pareto_ns_(.*)_(.*)\.parquet", filename)
    if not match:
        continue

    infra_tag = match.group(1)   # 예: COMPUTE 또는 STORAGE
    date_tag  = match.group(2)   # 예: 20260702

    # 하위 클러스터 서브디렉토리 경로 명시적 조인
    p1 = MERGED_DIR / infra_tag / f"daily_enriched_{infra_tag}_{date_tag}.parquet"
    p2 = Path(file_path)
    p3 = MERGED_DIR / infra_tag / f"daily_ns_usage_{infra_tag}_{date_tag}.parquet"

    if not (p1.exists() and p3.exists()):
        print(f"⚠️  [세트 불완전] merged/{infra_tag}/ 하위에 쌍이 되는 가공 데이터가 유실되어 작업을 건너뜁니다.")
        continue

    print(f"\n" + "="*60)
    print(f"🔄 [서브폴더 파티션 검증 완료] Cluster: {infra_tag} | Date: {date_tag} 컴파일 가동")
    print("="*60)

    df_pod = pd.read_parquet(p1)
    df_ns = pd.read_parquet(p2)
    df_daily_ns = pd.read_parquet(p3)

    excel_name = f"res_usage_report_{infra_tag.lower()}_{date_tag}.xlsx"

    wb = Workbook()
    build_sheet_summary(wb, df_pod, df_ns, infra_tag)
    build_sheet_pareto(wb, df_ns, infra_tag)
    build_sheet_ns_daily(wb, df_daily_ns, infra_tag)
    build_sheet_cpu(wb, df_pod, infra_tag)
    build_sheet_memory(wb, df_pod, infra_tag)
    build_sheet_oom(wb, df_pod, infra_tag)
    build_sheet_violations(wb, df_pod, infra_tag)
    build_sheet_trends(wb, df_pod, infra_tag)
    build_sheet_workload(wb, df_pod, infra_tag)
    build_sheet_extra_charts(wb, infra_tag, date_tag)

    out_path = OUT_DIR / excel_name
    print(f"💾 openpyxl 스트림 디스크 저장 가동 중... ➡️ {out_path}")
    wb.save(out_path)
    print(f"📦 [로컬 컴파일 마감 완료]: {excel_name} ({out_path.stat().st_size/1024:.0f} KB)")

    # ─── 🪣 사내 MinIO AIStor 자동 배포 레이어 ───
    minio_endpoint   = os.getenv("MINIO_ENDPOINT")
    minio_access_key = os.getenv("MINIO_ACCESS_KEY")
    minio_secret_key = os.getenv("MINIO_SECRET_KEY")
    bucket_name      = os.getenv("MINIO_REPORT_BUCKET", "devops-test")

    if all([minio_endpoint, minio_access_key, minio_secret_key]):
        try:
            endpoint_clean = minio_endpoint.replace("http://", "").replace("https://", "")
            secure_flag = minio_endpoint.startswith("https://")

            minio_client = Minio(
                endpoint_clean,
                access_key=minio_access_key,
                secret_key=minio_secret_key,
                secure=secure_flag
            )

            if not minio_client.bucket_exists(bucket_name):
                minio_client.make_bucket(bucket_name)

            object_key = f"reports/{infra_tag.lower()}/{excel_name}"
            print(f"🪣  [오브젝트 스토리지 싱크] devops-test 버킷 배포 ➡️ MinIO://{object_key} 업로드 중...")
            minio_client.fput_file(bucket_name, object_key, str(out_path))
            print("🏁 === [전사 배포 마감 성공] 고도화 res_usage_ 마스터 엑셀 리포트 배포 자산화 완료 ===")
        except Exception as e:
            print(f"❌ [배포 에러] 사내 MinIO 원격 업로드 중 예외 발생: {str(e)}")
    else:
        print("⚠️ [안내] 접속 환경변수 생략으로 로컬 아카이빙 처리 후 해당 파티션을 마감합니다.")

    processed_count += 1

print(f"\n🏁 === [전체 자동 매칭 빌드 완료] 총 {processed_count}개 세트 인프라 마스터 리포트 빌드 프로세스가 종결되었습니다. ===")

─── 📂 모든 디렉토리 경로 ./data 상대 경로 표준화 ──────────────────────────

BASE_DATA_DIR = Path("./data") MERGED_DIR = BASE_DATA_DIR / "merged" PLOT_DIR = BASE_DATA_DIR / "output" / "plots" OUT_DIR = BASE_DATA_DIR / "output"

OUT_DIR.mkdir(parents=True, exist_ok=True)

C = dict( hdr_dark="1F4E79", hdr_mid="2E75B6", hdr_light="BDD7EE", accent="ED7D31", red="C00000", green="70AD47", yellow="FFC000", gray_row="F2F2F2", white="FFFFFF", border="9DC3E6", summary_bg="EBF3FB", purple="7030A0", teal="008080", )

def ft(bold=False, size=10, color="000000", name="Arial"): return Font(name=name, bold=bold, size=size, color=color)

def fill(hex_color): return PatternFill("solid", fgColor=hex_color)

def thin_border(): t = Side(style="thin", color=C["border"]) return Border(left=t, right=t, top=t, bottom=t)

def center(wrap=False): return Alignment(horizontal="center", vertical="center", wrap_text=wrap)

def left(wrap=False): return Alignment(horizontal="left", vertical="center", wrap_text=wrap)

def set_col_widths(ws, widths): for col, w in widths.items(): ws.column_dimensions[col].width = w

def apply_header_row(ws, row_idx, headers, bg=None, fg=C["white"], size=10): bg = bg or C["hdr_dark"] for c, h in enumerate(headers, 1): cell = ws.cell(row=row_idx, column=c, value=h) cell.font = ft(bold=True, size=size, color=fg) cell.fill = fill(bg) cell.alignment = center(wrap=True) cell.border = thin_border()

def apply_data_rows(ws, df, start_row, num_formats=None, status_col_idx=None, zebra=True): nf = num_formats or {} for r_offset, row in enumerate(df.itertuples(index=False)): row_num = start_row + r_offset bg_hex = C["gray_row"] if (r_offset % 2 == 1 and zebra) else C["white"] for c_idx, val in enumerate(row, 1): cell = ws.cell(row=row_num, column=c_idx, value=val) cell.font = ft(size=9) cell.fill = fill(bg_hex) cell.alignment = left() cell.border = thin_border() if c_idx in nf: cell.number_format = nf[c_idx] if status_col_idx and c_idx == status_col_idx: v = str(val) if "OOM" in v or "Killed" in v: cell.fill = fill("FFCCCC"); cell.font = ft(bold=True, size=9, color=C["red"]) elif "Shortage" in v or "부족" in v: cell.fill = fill("FFF2CC"); cell.font = ft(bold=True, size=9, color="7F6000") elif "Over" in v or "과다" in v: cell.fill = fill("DDEEFF"); cell.font = ft(bold=True, size=9, color=C["hdr_dark"]) elif "Optim" in v or "최적" in v: cell.fill = fill("E2EFDA"); cell.font = ft(bold=True, size=9, color="375623") return start_row + len(df)

def freeze_and_filter(ws, row=2): ws.freeze_panes = ws.cell(row=row+1, column=1) ws.auto_filter.ref = ws.dimensions

def add_chart_image(ws, path_key, anchor_cell, w=860, h=400, label=None): path = PLOT_DIR / path_key if not path.exists(): print(f" ⚠️ [차트 유실] {path_key} 파일이 {PLOT_DIR}에 없어 삽입을 건너뜁니다.") return row_num = int(''.join(filter(str.isdigit, anchor_cell))) if label: col_letter = "".join(filter(str.isalpha, anchor_cell)) col_idx = 1 if col_letter == "A" else (9 if col_letter == "I" else 1) ws.cell(row=row_num-1, column=col_idx, value=label).font = ft(bold=True, size=11, color=C["hdr_dark"]) img = XLImage(str(path)) img.width = w; img.height = h ws.add_image(img, anchor_cell) print(f" -> 🎨 차트 맵핑 완료: {path_key} ➡️ {anchor_cell} 셀")

─── Sheet 0: 전사 종합 요약 ───────────────────────────────

def build_sheet_summary(wb, df_pod, df_ns, infra_tag): print("⏳ [0/9] '0. 전사종합요약' 대시보드 탭 구축 개시...") ws = wb.active ws.title = "0. 전사종합요약" ws.sheet_view.showGridLines = False ws.row_dimensions[1].height = 42

ws.merge_cells("A1:H1")
t = ws["A1"]
t.value = f"Resource Governance Master Report [{infra_tag}]  (KST 기준)"
t.font = ft(bold=True, size=15, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()

oom_cnt    = int(df_pod["is_oom_killed"].sum())
no_req_cnt = int((df_pod["has_no_request"] | df_pod["has_no_limit"]).sum())
alloc_ch   = df_pod["cpu_allocated_core_hours"].sum()
usage_ch   = df_pod["cpu_usage_core_hours"].sum()
waste_ch   = df_pod["cpu_waste_core_hours"].sum()
eff_pct    = usage_ch / max(alloc_ch, 0.001) * 100

mem_waste  = df_pod["mem_waste_gb_hours"].sum()
pv_waste   = df_pod["pv_waste_gb_hours"].sum()
throttle_risks = int((df_pod["cpu_throttled_max"] > 0.2).sum())
kst_dates  = sorted(df_pod["date"].unique())
date_range = f"{kst_dates[0]} ~ {kst_dates[-1]} (KST)"

kpis = [
    ("정산 대상 인프라 도메인",       infra_tag,                              "002060",      C["white"]),
    ("분석 기간 (KST)",              date_range,                             C["hdr_dark"], C["white"]),
    ("총 관측 컨테이너 볼륨",         f"{len(df_pod):,} 개",                C["hdr_dark"], C["white"]),
    ("OOMKilled 장애 컨테이너",       f"{oom_cnt:,} 개",                    C["red"],      C["white"]),
    ("연산 스로틀링 위험군 컨테이너",   f"{throttle_risks:,} 개",                 "C00000",      C["white"]),
    ("리소스 설정 규격 위반군",       f"{no_req_cnt:,} 개",                  "E26B0A",      C["white"]),
    ("전사 CPU 무효 선점 낭비량",     f"{waste_ch:,.1f} Core-H",            C["hdr_mid"],  C["white"]),
    ("전사 Memory 무효 선점 낭비량",   f"{mem_waste:,.1f} GB-H",             "6B4F9B",      C["white"]),
    ("전사 PV 스토리지 알박기 낭비량", f"{pv_waste:,.1f} GB-H",              "008080",      C["white"]),
    ("전사 CPU 평균 실효 활용률",     f"{eff_pct:.1f} %",                    "375623",      C["white"]),
]

for i, (label, value, bg, fg) in enumerate(kpis):
    row = 3 + i
    ws.row_dimensions[row].height = 24
    lc = ws.cell(row=row, column=1, value=label)
    lc.font = ft(bold=True, size=10, color=C["hdr_dark"])
    lc.fill = fill(C["summary_bg"]); lc.alignment = left(); lc.border = thin_border()
    ws.merge_cells(f"A{row}:C{row}")
    vc = ws.cell(row=row, column=4, value=value)
    vc.font = ft(bold=True, size=11, color=fg)
    vc.fill = fill(bg); vc.alignment = center(); vc.border = thin_border()
    ws.merge_cells(f"D{row}:F{row}")

set_col_widths(ws, {"A":28,"B":14,"C":14,"D":22,"E":14,"F":14,"G":20,"H":20})

chart_row = 16
add_chart_image(ws, "chart6_status_donut.png",   f"A{chart_row}", w=420, h=300)
add_chart_image(ws, "chart5_pareto_ns_waste.png", f"E{chart_row}", w=560, h=300)

─── Sheet 1: 파레토 분석 NS ──────────────────────────────

def build_sheet_pareto(wb, df_ns, infra_tag): print(f"⏳ [1/9] '1. 파레토분석_NS' 시트 렌더링 중...") ws = wb.create_sheet("1. 파레토분석_NS") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:I1")
t = ws["A1"]
t.value = f"Namespace별 CPU Waste 파레토 분석 [{infra_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["Namespace","실행시간 합계(분)","컨테이너 수","할당 Core-H","낭비 Core-H","낭비 비중(%)","누적 비중(%)","등급"]
apply_header_row(ws, 2, headers, bg=C["hdr_mid"])

df_disp = df_ns.copy()
df_disp["등급"] = df_disp["waste_cumsum_pct"].apply(
    lambda x: "Critical (Top 20%)" if x<=20 else ("High (Top 50%)" if x<=50 else ("Medium" if x<=80 else "Low")))
col_order = ["namespace","minutes_running_sum","container_cnt","total_allocated_core_hours","total_waste_core_hours","waste_share_pct","waste_cumsum_pct","등급"]
df_out = df_disp[col_order]
nf = {4:"#,##0.0", 5:"#,##0.0", 6:"0.00", 7:"0.00"}
end_row = apply_data_rows(ws, df_out, start_row=3, num_formats=nf)

ws.conditional_formatting.add(f"E3:E{end_row}", DataBarRule(start_type="min", end_type="max", color="2E75B6", showValue=True))
set_col_widths(ws, {"A":22,"B":18,"C":14,"D":16,"E":16,"F":12,"G":12,"H":22})
freeze_and_filter(ws)

─── 🎉 Sheet 1-2: 일단위 Namespace별 자원 사용량 탭 ───────────────────

def build_sheet_ns_daily(wb, df_daily_ns, infra_tag): print(f"⏳ [2/9] '1-2. 일단위_NS별_사용량' 종합 청산 피벗 장표 마크 중... (행 수: {len(df_daily_ns):,}건)") ws = wb.create_sheet("1-2. 일단위_NS별_사용량") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:J1")
t = ws["A1"]
t.value = f"Namespace 일일 자원 소모량 및 통합 사용량 점수 지표 [{infra_tag}]"
t.font = ft(bold=True, size=12, color=C["white"])
t.fill = fill("1F4E79"); t.alignment = center()
ws.row_dimensions[1].height = 35

headers = [
    "관측 일자(KST)", "네임스페이스(Namespace)", 
    "CPU 사용(Core-H)", "CPU 할당(Core-H)", "CPU 낭비(Core-H)",
    "Mem 사용(GB-H)", "Mem 할당(GB-H)",
    "PV 사용(GB-H)", "PV 할당(GB-H)", "통합 사용량 점수 (Score)"
]
apply_header_row(ws, 2, headers, bg=C["hdr_mid"])
ws.row_dimensions[2].height = 26

col_order = ["date", "namespace", "cpu_used_ch", "cpu_alloc_ch", "cpu_waste_ch", "mem_used_gh", "mem_alloc_gh", "pv_used_gh", "pv_alloc_gh", "final_usage_score"]
df_out = df_daily_ns[col_order].sort_values(by=["date", "final_usage_score"], ascending=[True, False])

nf = {3:"#,##0.0", 4:"#,##0.0", 5:"#,##0.0", 6:"#,##0.0", 7:"#,##0.0", 8:"#,##0.0", 9:"#,##0.0", 10:"#,##0.0"}
end_row = apply_data_rows(ws, df_out, start_row=3, num_formats=nf)

ws.conditional_formatting.add(f"J3:J{end_row}", DataBarRule(start_type="min", end_type="max", color="375623", showValue=True))

widths = {"A":14, "B":22, "C":18, "D":18, "E":18, "F":18, "G":18, "H":18, "I":18, "J":24}
set_col_widths(ws, widths)
freeze_and_filter(ws, row=2)

─── Sheet 2: CPU 분석 ────────────────────────────────────

def build_sheet_cpu(wb, df_pod, infra_tag): top30 = max(1, int(len(df_pod)*0.30)) print(f"⏳ [3/9] '2. CPU Request_Usage 분석' 시트 빌드 중... (상위 30% 격리: {top30}행)") ws = wb.create_sheet("2. CPU Request_Usage 분석") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:M1")
t = ws["A1"]
t.value = f"CPU Resource Efficiency Analysis [{infra_tag}] — Request / Limit / Usage / Throttling"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["날짜(KST)","클러스터","네임스페이스","워크로드","Pod","컨테이너",
           "CPU Request","CPU Limit","CPU P95","Throttle Peak","활용률(%)","낭비 Core-H","상태"]
apply_header_row(ws, 2, headers, bg=C["hdr_mid"])

df_out = df_pod.sort_values("cpu_waste_core_hours", ascending=False).head(top30).copy()
df_out["util"] = np.where(df_out["cpu_request_max"]>0, (df_out["cpu_usage_p95"]/df_out["cpu_request_max"]*100).round(1), 0)
df_out["status_en"] = df_out["status"].map({
    "💥 OOM장애발생":"OOM Killed","⚠️ Request부족":"Request Shortage",
    "📉 과다할당":"Over-allocated","✅ 최적화완료":"Optimized"}).fillna("Unknown")

cols = ["date","cluster","namespace","workload_type","pod","container",
        "cpu_request_max","cpu_limit_max","cpu_usage_p95","cpu_throttled_max","util","cpu_waste_core_hours","status_en"]
df_disp = df_out[cols].reset_index(drop=True)

nf = {7:"0.000",8:"0.000",9:"0.000",10:"0.000",11:"0.0",12:"#,##0.0"}
end_row = apply_data_rows(ws, df_disp, start_row=3, num_formats=nf, status_col_idx=13)

ws.conditional_formatting.add(f"K3:K{end_row}", ColorScaleRule(start_type="num", start_value=0, start_color="FF0000", mid_type="num", mid_value=50, mid_color="FFFF00", end_type="num", end_value=100, end_color="00B050"))
set_col_widths(ws, {"A":12,"B":16,"C":18,"D":16,"E":26,"F":14,"G":12,"H":12,"I":12,"J":14,"K":12,"L":14,"M":16})
freeze_and_filter(ws)

add_chart_image(ws, "chart1_cpu_req_vs_usage_by_workload.png", f"A{end_row+3}", w=860, h=400, label="[ CPU Request / Limit / P95 by Workload ]")
add_chart_image(ws, "chart9_boxplot_cpu_util_by_workload.png",  f"A{end_row+26}", w=860, h=400, label="[ CPU Utilization Boxplot by Workload ]")

─── Sheet 3: Memory 및 PV 스토리지 분석 ─────────────────────────────────

def build_sheet_memory(wb, df_pod, infra_tag): top30 = max(1, int(len(df_pod)*0.30)) print(f"⏳ [4/9] '3. Memory_PV 입체분석' 고도화 시트 로딩 중...") ws = wb.create_sheet("3. Memory_PV 입체분석") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:O1")
t = ws["A1"]
t.value = f"Memory RSS & Persistent Volume Storage Cross-Governance Analysis [{infra_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["날짜(KST)","클러스터","네임스페이스","워크로드","Pod","컨테이너",
           "Mem Request","Mem P95","Mem RSS","PV Cap(GB)","PV Used(GB)","PV Waste(GB-H)","활용률(%)","낭비 GB-H","상태"]
apply_header_row(ws, 2, headers, bg=C["hdr_mid"])

df_out = df_pod.sort_values(by=["mem_waste_gb_hours", "pv_waste_gb_hours"], ascending=[False, False]).head(top30).copy()
df_out["util"] = np.where(df_out["mem_request_max"]>0, (df_out["mem_usage_p95"]/df_out["mem_request_max"]*100).round(1), 0)
df_out["status_en"] = df_out["status"].map({
    "💥 OOM장애발생":"OOM Killed","⚠️ Request부족":"Request Shortage",
    "📉 과다할당":"Over-allocated","✅ 최적화완료":"Optimized"}).fillna("Unknown")

cols = ["date","cluster","namespace","workload_type","pod","container",
        "mem_request_max","mem_usage_p95","mem_rss_p95","pv_capacity_max","pv_used_p95","pv_waste_gb_hours","util","mem_waste_gb_hours","status_en"]
df_disp = df_out[cols].reset_index(drop=True)

nf = {7:"0.00", 8:"0.00", 9:"0.00", 10:"0.00", 11:"0.00", 12:"#,##0.0", 13:"0.0", 14:"#,##0.0"}
end_row = apply_data_rows(ws, df_disp, start_row=3, num_formats=nf, status_col_idx=15)

ws.conditional_formatting.add(f"M3:M{end_row}", ColorScaleRule(start_type="num", start_value=0, start_color="FF0000", mid_type="num", mid_value=50, mid_color="FFFF00", end_type="num", end_value=100, end_color="00B050"))
set_col_widths(ws, {"A":12,"B":16,"C":18,"D":16,"E":26,"F":14,"G":12,"H":12,"I":12,"J":12,"K":12,"L":14,"M":12,"N":12,"O":16})
freeze_and_filter(ws)

add_chart_image(ws, "chart2_mem_req_vs_usage_by_workload.png", f"A{end_row+3}", w=860, h=400, label="[ Memory Specs vs WorkingSet vs RSS Peak ]")
add_chart_image(ws, "chart20_daily_mem_per_workload.png", f"A{end_row+26}", w=960, h=540, label="[ Daily Memory & PV Provisioning Trends ]")

─── Sheet 4: OOM 및 자원 부족 ───────────────────────────

def build_sheet_oom(wb, df_pod, infra_tag): print("⏳ [5/9] '4. 자원부족및OOM장애군' 리스크 인덱스 추출 중...") ws = wb.create_sheet("4. 자원부족및OOM장애군") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:L1")
t = ws["A1"]
t.value = f"OOMKilled / CPU CFS Throttled 병목 위험 워크로드 명세 [{infra_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["red"]); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["날짜(KST)","클러스터","네임스페이스","워크로드","Pod","컨테이너",
           "상태","CPU Request","CPU P95","Throttle Peak","Mem Limit(GB)","Mem P95(GB)"]
apply_header_row(ws, 2, headers, bg=C["red"])

df_out = df_pod[
    (df_pod["cpu_shortage_cores"]>0) | (df_pod["is_oom_killed"]) | (df_pod["cpu_throttled_max"] > 0.2)
].sort_values(["is_oom_killed","cpu_throttled_max","cpu_shortage_cores"], ascending=[False,False,False]).copy()

df_out["status_en"] = df_out["status"].map({
    "💥 OOM장애발생":"OOM Killed","⚠️ Request부족":"Request Shortage",
    "📉 과다할당":"Over-allocated","✅ 최적화완료":"Optimized"}).fillna("Unknown")

cols = ["date","cluster","namespace","workload_type","pod","container",
        "status_en","cpu_request_max","cpu_usage_p95","cpu_throttled_max","mem_limit_max","mem_usage_p95"]
df_disp = df_out[cols].reset_index(drop=True)

nf = {8:"0.000",9:"0.000",10:"0.000",11:"0.00",12:"0.00"}
end_row = apply_data_rows(ws, df_disp, start_row=3, num_formats=nf, status_col_idx=7)
set_col_widths(ws, {"A":12,"B":18,"C":20,"D":18,"E":28,"F":16,"G":16,"H":12,"I":12,"J":14,"K":14,"L":14})
freeze_and_filter(ws)

add_chart_image(ws, "chart8_shortfall_footprint.png", f"A{end_row+3}", w=860, h=400, label="[ CPU CFS Throttling Risk Heatmap Footprint ]")

─── Sheet 5: 리소스 미설정 위반 ─────────────────────────

def build_sheet_violations(wb, df_pod, infra_tag): print("⏳ [6/9] '5. 리소스미설정위반군' 규격 미설정 목록 마크 중...") ws = wb.create_sheet("5. 리소스미설정위반군") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:L1")
t = ws["A1"]
t.value = "Resource Request / Limit 미설정 위반 컨테이너 (KST)"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill("E26B0A"); t.alignment = center()
ws.row_dimensions[1].height = 32

headers = ["날짜(KST)","클러스터","네임스페이스","워크로드","Pod","컨테이너",
           "Request 미설정","Limit 미설정","CPU Request","CPU Limit","Mem Request(GB)","Mem Limit(GB)"]
apply_header_row(ws, 2, headers, bg="E26B0A")

df_out = df_pod[df_pod["has_no_request"]|df_pod["has_no_limit"]].sort_values("minutes_running", ascending=False).copy()
df_out["req_flag"] = df_out["has_no_request"].map({True:"MISSING",False:"OK"})
df_out["lim_flag"] = df_out["has_no_limit"].map({True:"MISSING",False:"OK"})
cols = ["date","cluster","namespace","workload_type","pod","container",
        "req_flag","lim_flag","cpu_request_max","cpu_limit_max","mem_request_max","mem_limit_max"]
df_disp = df_out[cols].reset_index(drop=True)

nf = {9:"0.000",10:"0.000",11:"0.00",12:"0.00"}
end_row = apply_data_rows(ws, df_disp, start_row=3, num_formats=nf)
set_col_widths(ws, {"A":12,"B":18,"C":20,"D":18,"E":28,"F":16,"G":14,"H":14,"I":14,"J":14,"K":14,"L":14})
freeze_and_filter(ws)

─── Sheet 6: 일별 트렌드 (KST) ──────────────────────────

def build_sheet_trends(wb, df_pod, infra_tag): print("⏳ [7/9] '6. 일별트렌드_KST' 누적 손실액 분석 중...") ws = wb.create_sheet("6. 일별트렌드_KST") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:M1")
t = ws["A1"]
t.value = f"Daily Resource Governance & Loss Trend [{infra_tag}] — KST"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 32

df_daily = df_pod.groupby("date").agg(
    containers=("container","count"),
    cpu_alloc=("cpu_allocated_core_hours","sum"),
    cpu_used=("cpu_usage_core_hours","sum"),
    cpu_waste=("cpu_waste_core_hours","sum"),
    mem_alloc=("mem_allocated_gb_hours","sum"),
    mem_waste=("mem_waste_gb_hours","sum"),
    pv_alloc=("pv_allocated_gb_hours","sum"),
    pv_waste_sum=("pv_waste_gb_hours","sum"),
    oom_cnt=("is_oom_killed","sum"),
    shortage_cnt=("cpu_shortage_cores", lambda x: (x>0).sum()),
).reset_index()

df_daily["cpu_util_pct"] = (df_daily["cpu_used"]/df_daily["cpu_alloc"].clip(lower=0.001)*100).round(1)
df_daily["mem_util_pct"] = ((df_daily["mem_alloc"]-df_daily["mem_waste"])/df_daily["mem_alloc"].clip(lower=0.001)*100).round(1)

headers = ["날짜(KST)","컨테이너 볼륨","CPU 할당 Core-H","CPU 사용 Core-H","CPU 낭비 Core-H",
           "Mem 할당 GB-H","Mem 낭비 GB-H","PV 할당 GB-H","PV 낭비 GB-H","OOM 사살","자원부족군","CPU활용률(%)","Mem활용률(%)"]
apply_header_row(ws, 3, headers, bg=C["hdr_mid"])

nf = {3:"#,##0.0",4:"#,##0.0",5:"#,##0.0",6:"#,##0.0",7:"#,##0.0",8:"#,##0.0",9:"#,##0.0",10:"#,##0",11:"#,##0",12:"0.1",13:"0.1"}
df_disp = df_daily[["date","containers","cpu_alloc","cpu_used","cpu_waste",
                     "mem_alloc","mem_waste","pv_alloc","pv_waste_sum","oom_cnt","shortage_cnt","cpu_util_pct","mem_util_pct"]]
end_row = apply_data_rows(ws, df_disp, start_row=4, num_formats=nf)

set_col_widths(ws, {"A":14,"B":12,"C":16,"D":16,"E":16,"F":16,"G":14,"H":16,"I":14,"J":12,"K":12,"L":14,"M":14})
freeze_and_filter(ws, row=4)

add_chart_image(ws, "chart3_daily_waste_stack.png", f"A{end_row+3}", 860, 400, "[ Daily CPU Waste Stacked Timeline ]")
add_chart_image(ws, "chart4_cpu_efficiency_heatmap.png", f"A{end_row+26}", 860, 400, "[ CPU Utilization Heatmap Grid ]")

─── Sheet 7: 워크로드별 심층 분석 ─────────────────

def build_sheet_workload(wb, df_pod, infra_tag): print("⏳ [8/9] '7. 워크로드별_심층분석' 오픈소스 기술 스택 평점표 작성 중...") ws = wb.create_sheet("7. 워크로드별_심층분석") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:N1")
t = ws["A1"]
t.value = f"Workload Type별 오픈소스 기술 스택 심층 자원 효율화 분석 [{infra_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["purple"]); t.alignment = center()
ws.row_dimensions[1].height = 32

df_wl = df_pod.groupby("workload_type").agg(
    containers=("container","count"),
    cpu_req_avg=("cpu_request_max","mean"),
    cpu_p95_avg=("cpu_usage_p95","mean"),
    cpu_waste_sum=("cpu_waste_core_hours","sum"),
    cpu_throttle_pk=("cpu_throttled_max","max"),
    mem_req_avg=("mem_request_max","mean"),
    mem_p95_avg=("mem_usage_p95","mean"),
    mem_rss_avg=("mem_rss_p95","mean"),
    mem_waste_sum=("mem_waste_gb_hours","sum"),
    pv_capacity_avg=("pv_capacity_max","mean"),
    pv_waste_sum=("pv_waste_gb_hours","sum"),
    oom_cnt=("is_oom_killed","sum")
).reset_index()

df_wl["cpu_util_pct"] = (df_wl["cpu_p95_avg"]/df_wl["cpu_req_avg"].replace(0,np.nan)*100).round(1).fillna(0)
df_wl["mem_util_pct"] = (df_wl["mem_p95_avg"]/df_wl["mem_req_avg"].replace(0,np.nan)*100).round(1).fillna(0)
df_wl = df_wl.sort_values("cpu_waste_sum", ascending=False).reset_index(drop=True)

headers2 = ["워크로드 타입","컨테이너 수","CPU Req(avg)","CPU P95(avg)","CPU 낭비 총합","Throttle Peak",
            "Mem Req(avg)","Mem P95(avg)","Mem RSS(avg)","Mem 낭비 총합","PV Cap(avg)","PV 낭비 총합","CPU활용(%)","Mem활용(%)"]
apply_header_row(ws, 3, headers2, bg=C["purple"])

cols_out = ["workload_type","containers","cpu_req_avg","cpu_p95_avg","cpu_waste_sum","cpu_throttle_pk",
            "mem_req_avg","mem_p95_avg","mem_rss_avg","mem_waste_sum","pv_capacity_avg","pv_waste_sum","cpu_util_pct","mem_util_pct"]
df_disp = df_wl[cols_out]

nf = {3:"0.000", 4:"0.000", 5:"#,##0.0", 6:"0.000", 7:"0.00", 8:"0.00", 9:"0.00", 10:"#,##0.0", 11:"0.00", 12:"#,##0.0", 13:"0.1", 14:"0.1"}
end_row = apply_data_rows(ws, df_disp, start_row=4, num_formats=nf)

set_col_widths(ws, {"A":18,"B":12,"C":14,"D":14,"E":14,"F":14,"G":14,"H":14,"I":14,"J":14,"K":14,"L":14,"M":12,"N":12})
freeze_and_filter(ws, row=4)

add_chart_image(ws, "chart15_oom_status_by_workload.png", f"A{end_row+3}", 860, 400, "[ Status Distribution count per Open-Source Stack ]")
add_chart_image(ws, "chart14_cpu_mem_waste_scatter.png", f"A{end_row+26}", 760, 480, "[ Multi-Resource Loss Correlation Bubble Map ]")

─── 🎉 [요청 반영] Sheet 8: 미사용 추가 차트 모음 전용 시트 ─────────────────

def build_sheet_extra_charts(wb, infra_tag, date_tag): print("⏳ [9/9] '8. 추가차트모음' 시트 배치 작업 진행 중...") ws = wb.create_sheet("8. 추가차트모음") ws.sheet_view.showGridLines = False

ws.merge_cells("A1:Q1")
t = ws["A1"]
t.value = f"Step 3 Additional Unused Dashboards & Reports [{infra_tag} - {date_tag}]"
t.font = ft(bold=True, size=13, color=C["white"])
t.fill = fill(C["hdr_dark"]); t.alignment = center()
ws.row_dimensions[1].height = 35

# 유저 지정 주력 사용 8개 차트 외의 나머지 번호 자동 색출 및 다중 포맷 탐색
used_ids = [1, 2, 3, 5, 6, 9, 10, 15]
remaining_ids = [i for i in range(1, 20) if i not in used_ids]

current_row = 3
col_positions = ["A", "I"]  # 2열 바둑판 교차 격자 배치 구조
col_idx = 0

for cid in remaining_ids:
    found_file = None
    # 동적/고정 프리픽스 호환성 검증 패치
    patterns = [f"chart{cid}_*.png", f"chart_{cid}_*.png", f"chart{cid}.png", f"chart_{cid}.png"]
    for pat in patterns:
        matches = list(PLOT_DIR.glob(pat))
        if matches:
            found_file = matches[0].name
            break

    if found_file:
        anchor = f"{col_positions[col_idx]}{current_row}"
        # 차트 크기 및 타이틀 라벨 할당
        add_chart_image(ws, found_file, anchor, w=480, h=280, label=f"Chart {cid}: {found_file}")
        col_idx += 1
        if col_idx >= len(col_positions):
            col_idx = 0
            current_row += 16  # 차트 간 행 간격 오프셋 설정
    else:
        print(f"  ℹ️  [추가 차트 보관함] Chart ID {cid} 번 이미지 파일이 없어 스킵합니다.")

─── Main ─────────────────────────────────────────────────

def main(): print("🚀 [Step6 개시] res_usage_ 계열 고도화 엑셀 리포터 엔진 가동 (인프라/날짜 패턴 자동 검색)...")

# ─── 🔄 [방법 B] 파레토 Parquet 파일 패턴을 기반으로 동적 파티션 검색 ───
search_pattern = str(MERGED_DIR / "pareto_ns_*.parquet")
pareto_files = glob.glob(search_pattern)

if not pareto_files:
    print(f"❌ 오류: 탐색된 파레토 가공원부(*.parquet) 파일이 없습니다. 경로를 확인하세요: {search_pattern}")
    return

processed_count = 0
for file_path in pareto_files:
    filename = os.path.basename(file_path)
    match = re.search(r"pareto_ns_(.*)_(.*)\.parquet", filename)
    if not match:
        continue

    infra_tag = match.group(1)   # 예: COMPUTE 또는 STORAGE
    date_tag  = match.group(2)   # 예: 20260702

    # 고도화 패치된 분할정복 연동 파일명 매핑 규칙 바인딩
    p1 = MERGED_DIR / f"daily_enriched_{infra_tag}_{date_tag}.parquet"
    p2 = Path(file_path)
    p3 = MERGED_DIR / f"daily_ns_usage_{infra_tag}_{date_tag}.parquet"

    if not (p1.exists() and p3.exists()):
        print(f"⚠️  [세트 불완전] {infra_tag} | {date_tag} 에 쌍이 되는 가공 데이터가 유실되어 작업을 건너뜁니다.")
        continue

    print(f"\n" + "="*60)
    print(f"🔄 [파티션 매칭 감지] Cluster/Infra: {infra_tag} | Date: {date_tag} 컴파일 가동")
    print("="*60)

    df_pod      = pd.read_parquet(p1)
    df_ns       = pd.read_parquet(p2)
    df_daily_ns = pd.read_parquet(p3)

    excel_name = f"res_usage_report_{infra_tag.lower()}_{date_tag}.xlsx"

    wb = Workbook()
    build_sheet_summary(wb, df_pod, df_ns, infra_tag)
    build_sheet_pareto(wb, df_ns, infra_tag)
    build_sheet_ns_daily(wb, df_daily_ns, infra_tag)
    build_sheet_cpu(wb, df_pod, infra_tag)
    build_sheet_memory(wb, df_pod, infra_tag)
    build_sheet_oom(wb, df_pod, infra_tag)
    build_sheet_violations(wb, df_pod, infra_tag)
    build_sheet_trends(wb, df_pod, infra_tag)
    build_sheet_workload(wb, df_pod, infra_tag)
    build_sheet_extra_charts(wb, infra_tag, date_tag)  # ◀ [신설 추가 차트 시트 연결]

    out_path = OUT_DIR / excel_name
    print(f"💾 openpyxl 스트림 디스크 저장 가동 중... ➡️ {out_path}")
    wb.save(out_path)
    print(f"📦 [로컬 컴파일 마감 완료]: {excel_name} ({out_path.stat().st_size/1024:.0f} KB)")

    # ─── 🪣 사내 MinIO AIStor 'devops-test' 버킷 자동 배포 레이어 (boto3 -> minio 상용 완전 교체) ───
    minio_endpoint   = os.getenv("MINIO_ENDPOINT")
    minio_access_key = os.getenv("MINIO_ACCESS_KEY")
    minio_secret_key = os.getenv("MINIO_SECRET_KEY")
    bucket_name      = os.getenv("MINIO_REPORT_BUCKET", "devops-test")

    if all([minio_endpoint, minio_access_key, minio_secret_key]):
        try:
            # 엔드포인트 내 http/https 스키마 제거 프로토콜 보정 작업
            endpoint_clean = minio_endpoint.replace("http://", "").replace("https://", "")
            secure_flag = minio_endpoint.startswith("https://")

            minio_client = Minio(
                endpoint_clean,
                access_key=minio_access_key,
                secret_key=minio_secret_key,
                secure=secure_flag
            )

            # 원격 타겟 버킷 검증 및 유동적 생성 통제
            if not minio_client.bucket_exists(bucket_name):
                print(f"🪣  [MinIO] 버킷이 존재하지 않아 신규 생성합니다: {bucket_name}")
                minio_client.make_bucket(bucket_name)

            object_key = f"reports/{infra_tag.lower()}/{excel_name}"
            print(f"🪣  [오브젝트 스토리지 싱크] devops-test 버킷 배포 ➡️ MinIO://{object_key} 파일 스트림 업로드 중...")

            # fput_file을 이용한 안정적인 파일 스트림 배포 자산화
            minio_client.fput_file(bucket_name, object_key, str(out_path))
            print("🏁 === [전사 배포 마감 성공] 고도화 res_usage_ 마스터 엑셀 리포트 배포 자산화가 완료되었습니다. ===")
        except Exception as e:
            print(f"❌ [배포 에러] 사내 MinIO AIStor 원격 업로드 중 예외 발생: {str(e)}")
    else:
        print("⚠️ [안내] 접속 환경변수(MINIO_ENDPOINT 등) 생략으로 로컬 아카이빙 처리 후 해당 파티션을 마감합니다.")

    processed_count += 1

print(f"\n🏁 === [전체 자동 매칭 빌드 완료] 총 {processed_count}개 세트 인프라 마스터 리포트 빌드 프로세스가 종결되었습니다. ===")

if name == "main": main()

로깅 설정

logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[logging.StreamHandler(sys.stdout)] ) logger = logging.getLogger(name)

환경 설정 (기본값 설정 및 환경변수 오버라이드 가능)

DATA_DIR = os.getenv("DATA_DIR", "./data/merged") CHARTS_DIR = os.getenv("CHARTS_DIR", "./data/charts") # 차트 이미지 폴더 기본값 OUTPUT_DIR = os.getenv("OUTPUT_DIR", "./data") MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT", "minio.ai-stor.local:9000") MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY", "admin") MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY", "password") MINIO_BUCKET = os.getenv("MINIO_BUCKET", "devops-test") USE_SSL = os.getenv("MINIO_USE_SSL", "False").lower() in ("true", "1", "yes")

def get_minio_client(): """MinIO AIStor 클라이언트 초기화""" try: return Minio( MINIO_ENDPOINT, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=USE_SSL ) except Exception as e: logger.error(f"MinIO 클라이언트 초기화 실패: {e}") return None

def upload_to_aistor(local_file_path, object_name): """생성된 엑셀 리포트를 MinIO AIStor devops-test 버킷에 업로드""" client = get_minio_client() if not client: logger.warning("MinIO 클라이언트를 사용할 수 없어 업로드를 건너뜁니다.") return False

try:
    if not client.bucket_exists(MINIO_BUCKET):
        logger.info(f"버킷이 존재하지 않아 생성합니다: {MINIO_BUCKET}")
        client.make_bucket(MINIO_BUCKET)

    logger.info(f"AIStor 업로드 시작: {local_file_path} -> bucket: {MINIO_BUCKET}/{object_name}")
    client.fput_file(MINIO_BUCKET, object_name, local_file_path)
    logger.info("AIStor 업로드 완료 성공.")
    return True
except S3Error as e:
    logger.error(f"AIStor 업로드 중 S3 에러 발생: {e}")
except Exception as e:
    logger.error(f"AIStor 업로드 중 알 수 없는 오류 발생: {e}")
return False

def style_sheet(ws, title_text): """시트 기본 스타일링 (헤더, 그리드라인, 폰트 등)""" ws.views.sheetView[0].showGridLines = True

font_family = "Arial"
header_fill = PatternFill(start_color="2F4F4F", end_color="2F4F4F", fill_type="solid") # Muted Dark Slate
header_font = Font(name=font_family, size=11, bold=True, color="FFFFFF")
data_font = Font(name=font_family, size=10, color="000000")
zebra_fill = PatternFill(start_color="F9FBFB", end_color="F9FBFB", fill_type="solid")

thin_border = Border(
    left=Side(style='thin', color='E0E0E0'),
    right=Side(style='thin', color='E0E0E0'),
    top=Side(style='thin', color='E0E0E0'),
    bottom=Side(style='thin', color='E0E0E0')
)

for row_idx, row in enumerate(ws.iter_rows(min_row=1, max_row=ws.max_row, min_col=1, max_col=ws.max_column), start=1):
    is_header = (row_idx == 1)
    for cell in row:
        if is_header:
            cell.fill = header_fill
            cell.font = header_font
            cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
        else:
            cell.font = data_font
            cell.border = thin_border
            if row_idx % 2 == 0:
                cell.fill = zebra_fill

            if isinstance(cell.value, (int, float)):
                if cell.value == 0:
                    cell.number_format = '#,##0'
                elif 'pct' in str(cell.column) or '비율' in str(ws.cell(1, cell.column).value or ''):
                    cell.number_format = '0.0%'
                elif isinstance(cell.value, float):
                    cell.number_format = '#,##0.00'
                else:
                    cell.number_format = '#,##0'

for col in ws.columns:
    max_len = 0
    col_letter = openpyxl.utils.get_column_letter(col[0].column)
    for cell in col:
        if cell.value:
            max_len = max(max_len, len(str(cell.value)))
    ws.column_dimensions[col_letter].width = max(max_len + 3, 12)

ws.freeze_panes = 'A2'

def add_pareto_chart(ws, data_len): """네임스페이스별 CPU 낭비 현황 오픈pyxl 자체 파레토 차트 삽입""" if data_len <= 0: logger.warning("차트를 생성할 데이터가 없어 차트 삽입을 건너뜁니다.") return

try:
    chart = BarChart()
    chart.type = "col"
    chart.style = 10
    chart.title = "Namespace CPU Waste Pareto Analysis (Programmatic)"
    chart.y_axis.title = "CPU Waste (Core-Hours)"
    chart.x_axis.title = "Namespace"

    data = Reference(ws, min_col=2, min_row=1, max_row=data_len + 1)
    cats = Reference(ws, min_col=1, min_row=2, max_row=data_len + 1)

    chart.add_data(data, titles_from_data=True)
    chart.set_categories(cats)

    line_chart = LineChart()
    line_data = Reference(ws, min_col=3, min_row=1, max_row=data_len + 1)
    line_chart.add_data(line_data, titles_from_data=True)

    line_chart.y_axis.axId = 200
    line_chart.y_axis.title = "Cumulative Percentage"
    line_chart.y_axis.crosses = "max"

    chart += line_chart
    ws.add_chart(chart, f"E2")
    logger.info("자체 스크립트 기반 파레토 차트가 시트(E2 위치)에 삽입되었습니다.")
except Exception as e:
    logger.error(f"파레토 차트 생성 중 오류 발생: {e}")

def inject_step3_external_charts(wb, cluster, date): """ Step 3에서 생성된 19개 차트 중 미사용 중인 나머지 11개 차트를 별도의 'Charts' 전용 시트를 생성하여 바둑판 배열로 자동 삽입합니다. - 기존 사용 8개: 1, 2, 3, 5, 6, 9, 10, 15 - 신규 추가 11개: 나머지 번호들 """ used_charts = [1, 2, 3, 5, 6, 9, 10, 15] all_charts = list(range(1, 20)) remaining_charts = [c for c in all_charts if c not in used_charts]

logger.info(f"[{cluster} | {date}] 미사용 차트({len(remaining_charts)}개) 일괄 배치를 위한 'Charts' 시트 생성 중...")

ws_charts = wb.create_sheet(title="Charts")
ws_charts.views.sheetView[0].showGridLines = True

# 폰트 세팅
title_font = Font(name="Arial", size=14, bold=True, color="2F4F4F")
ws_charts["A1"] = f"Step 3 Additional Dashboards & Charts ({cluster} - {date})"
ws_charts["A1"].font = title_font

# 배치를 위한 변수 (2열 구조 배열)
col_positions = ['B', 'N'] # B열, N열에 차트를 나란히 배치하여 겹침 방지
current_row = 3
col_idx = 0

for chart_id in remaining_charts:
    # 파일명 매핑 규칙 수용 (동적 파티션 이름 및 고정 이름 매칭 버퍼 확보)
    possible_filenames = [
        f"chart_{chart_id}_{cluster}_{date}.png",
        f"chart_{cluster}_{date}_{chart_id}.png",
        f"chart_{chart_id}.png",
        f"chart_fixed_{chart_id}.png"
    ]

    chart_file_path = None
    # 설정된 차트 경로 및 데이터 통합 경로 검전
    for fname in possible_filenames:
        for d in [CHARTS_DIR, DATA_DIR, "./data", "."]:
            p = os.path.join(d, fname)
            if os.path.exists(p):
                chart_file_path = p
                break
        if chart_file_path:
            break

    if chart_file_path:
        try:
            img = OpenpyxlImage(chart_file_path)
            cell_address = f"{col_positions[col_idx]}{current_row}"
            ws_charts.add_image(img, cell_address)
            logger.info(f" -> 차트 {chart_id} 삽입 완료: {chart_file_path} -> cell: {cell_address}")

            # 다음 이미지 위치 계산 (2열 배치)
            col_idx += 1
            if col_idx >= len(col_positions):
                col_idx = 0
                current_row += 20 # 차트 높이를 고려하여 20행 아래로 이동
        except Exception as e:
            logger.error(f" -> 차트 {chart_id}({chart_file_path}) 삽입 중 openpyxl 예외 발생: {e}")
    else:
        # 실시간 로깅 상태창 제공
        logger.warning(f" -> 차트 {chart_id} 원부 이미지를 찾을 수 없어 건너뜁니다. (검색 패턴 샘플: chart_{chart_id}_*.png)")

def generate_report(cluster, date): """동적 매핑된 파일들을 결합하여 최종 엑셀 리포트 빌드""" usage_file = os.path.join(DATA_DIR, f"daily_ns_usage_{cluster}{date}.parquet") pareto_file = os.path.join(DATA_DIR, f"pareto_ns{cluster}{date}.parquet") enriched_file = os.path.join(DATA_DIR, f"daily_enriched{cluster}_{date}.parquet")

# 파일 유실 여부 검증
missing_files = [f for f in [usage_file, pareto_file, enriched_file] if not os.path.exists(f)]
if missing_files:
    logger.error(f"[{cluster} | {date}] 필수 가공원부가 유실되어 빌드를 건너뜁니다: {missing_files}")
    return

logger.info(f"[{cluster} | {date}] 엑셀 리포트 가공 시작...")

# 데이터 로드
df_usage = pd.read_parquet(usage_file)
df_pareto = pd.read_parquet(pareto_file)
df_enriched = pd.read_parquet(enriched_file)

# 워크북 및 시트 구성
wb = openpyxl.Workbook()

# 기본 시트를 Summary로 변경
ws_summary = wb.active
ws_summary.title = "Summary Metrics"

# 요약 정보 작성
ws_summary.append(["Metric Key", "Value"])
ws_summary.append(["Cluster Name", cluster])
ws_summary.append(["Analysis Date", date])
ws_summary.append(["Total Namespaces", len(df_usage['namespace'].unique()) if 'namespace' in df_usage.columns else 0])
if 'cpu_waste' in df_pareto.columns:
    ws_summary.append(["Total CPU Waste (Core-Hours)", float(df_pareto['cpu_waste'].sum())])
style_sheet(ws_summary, "Summary Metrics")

# 1. Daily Namespace Usage 시트
ws_usage = wb.create_sheet(title="Daily Namespace Usage")
for r in dataframe_to_rows(df_usage, index=False, header=True):
    ws_usage.append(r)
style_sheet(ws_usage, "Daily Namespace Usage")

# 2. Pareto Analysis 시트
ws_pareto = wb.create_sheet(title="Pareto Analysis")
for r in dataframe_to_rows(df_pareto, index=False, header=True):
    ws_pareto.append(r)
style_sheet(ws_pareto, "Pareto Analysis")
# 파레토 차트 내부 렌더링 추가
add_pareto_chart(ws_pareto, len(df_pareto))

# 3. Enriched Metrics 시트
ws_enriched = wb.create_sheet(title="Enriched Metrics")
for r in dataframe_to_rows(df_enriched, index=False, header=True):
    ws_enriched.append(r)
style_sheet(ws_enriched, "Enriched Metrics")

# [요구사항 반영] Step 3 외부 차트 19개 중 미사용 11개 차트 전용 시트 주입 로직
inject_step3_external_charts(wb, cluster, date)

if 'cpu_waste' in df_pareto.columns:
    red_fill = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
    ws_pareto.conditional_formatting.add(
        f"B2:B{len(df_pareto)+1}",
        CellIsRule(operator='greaterThan', formula=['100'], stopIfTrue=True, fill=red_fill)
    )

# 파일 저장 (요청된 프리픽스 매핑 반영)
output_filename = f"res_usage_report_{cluster}_{date}.xlsx"
local_output_path = os.path.join(OUTPUT_DIR, output_filename)

os.makedirs(OUTPUT_DIR, exist_ok=True)
wb.save(local_output_path)
logger.info(f"로컬 엑셀 리포트 저장 성공: {local_output_path}")

# AIStor 자동 업로드 연계
upload_to_aistor(local_output_path, output_filename)

def main(): logger.info(f"가공원부 자동 검색 스캔 시작 (경로: {DATA_DIR})")

# 방법 B: pareto_ns_*.parquet 패턴을 탐색하여 동적으로 cluster와 date 패러미터 추출
search_pattern = os.path.join(DATA_DIR, "pareto_ns_*.parquet")
pareto_files = glob.glob(search_pattern)

if not pareto_files:
    logger.warning(f"탐색된 파레토 가공원부(*.parquet) 파일이 없습니다. 경로를 확인하세요: {search_pattern}")
    return

processed_count = 0
for file_path in pareto_files:
    filename = os.path.basename(file_path)
    match = re.search(r"pareto_ns_(.*)_(.*)\.parquet", filename)
    if match:
        cluster = match.group(1)
        date = match.group(2)

        logger.info(f"파티션 감지 성공 -> Cluster: {cluster}, Date: {date}")
        generate_report(cluster, date)
        processed_count += 1

logger.info(f"전체 자동 매칭 빌드 프로세스 완료. (처리 완료 세트 수: {processed_count})")

if name == "main": main()

import config from config import ( RAW_DIR, MERGED_DIR, CLUSTER_NODE_PATTERNS, get_workload_type, classify_node_infrastructure )

KST = timezone(timedelta(hours=9)) W_CPU, W_MEM, W_PV = 1.0, 0.11, 0.02

def parse_arguments(): parser = argparse.ArgumentParser(description="FinOps Polars-Powered Reprocessing Engine") parser.add_argument("--days", type=int, default=None) parser.add_argument("--start-date", type=str, default=None) parser.add_argument("--end-date", type=str, default=None) parser.add_argument("--cluster", type=str, default="ALL", choices=["COMPUTE", "STORAGE", "ALL"]) parser.add_argument("--workload-domain", type=str, default=None) return parser.parse_args()

def get_minio_client(): return boto3.client( "s3", endpoint_url=os.getenv("MINIO_ENDPOINT"), aws_access_key_id=os.getenv("MINIO_ACCESS_KEY"), aws_secret_access_key=os.getenv("MINIO_SECRET_KEY"), region_name="us-east-1", config=boto3.session.Config(signature_version="s3v4") )

def sync_down_raw_from_minio(s3_client, bucket_name, start_dt, end_dt): chunk_delta = timedelta(hours=1) current_start = start_dt print(f"\n📡 [AIStor 스마트 캐시 엔진] 원천 데이터 가용성 체크 시작...")

while current_start < end_dt:
    chunk_str = current_start.strftime("%Y%m%d_%H")
    f_name = f"prom_raw_{chunk_str}.parquet"
    object_key = f"raw/{f_name}"
    local_path = RAW_DIR / f_name

    if local_path.exists() and local_path.stat().st_size > 0:
        print(f"   ⚡ [Cache Hit] 로컬 캐시 자산 활용 (다운로드 스킵): {f_name}")
    else:
        try:
            s3_client.head_object(Bucket=bucket_name, Key=object_key)
            print(f"   📥 [Cache Miss] AIStor 백업으로부터 다운로드: {object_key}")
            s3_client.download_file(bucket_name, object_key, str(local_path))
        except ClientError as e:
            if e.response['Error']['Code'] != "404":
                print(f"   ❌ AIStor 통신 실패 에러 ({f_name}): {str(e)}")
        except Exception as e:
            print(f"   ⚠️ 일반 다운로드 예외 발생 ({f_name}): {str(e)}")

    current_start += chunk_delta

def main(): args = parse_arguments() s3_client = get_minio_client() bucket_name = os.getenv("MINIO_RAW_BUCKET", "devops-test") now_kst = datetime.now(KST)

if args.start_date and args.end_date:
    start_dt = datetime.strptime(args.start_date, "%Y-%m-%d").replace(tzinfo=KST)
    end_dt = (datetime.strptime(args.end_date, "%Y-%m-%d") + timedelta(days=1)).replace(tzinfo=KST)
elif args.days:
    start_dt = (now_kst - timedelta(days=args.days)).replace(hour=0, minute=0, second=0, microsecond=0)
    end_dt = now_kst
else:
    yesterday = (now_kst - timedelta(days=1)).date()
    start_dt = datetime.combine(yesterday, datetime.min.time()).replace(tzinfo=KST)
    end_dt = datetime.combine(now_kst.date(), datetime.min.time()).replace(tzinfo=KST)

sync_down_raw_from_minio(s3_client, bucket_name, start_dt, end_dt)

raw_files = list(RAW_DIR.glob("prom_raw_*.parquet"))
if not raw_files:
    print("\n🚨 [중단] 정산 가동할 물리 원천 파일이 존재하지 않습니다.")
    return

print(f"\n📊 [Polars Lazy Engine 기동] {len(raw_files)}개 소스 파티션 스캔 링킹...")
lf_raw = pl.scan_parquet(str(RAW_DIR / "prom_raw_*.parquet"))

start_naive = start_dt.replace(tzinfo=None)
end_naive = end_dt.replace(tzinfo=None)

lf_filtered = lf_raw.filter(
    (pl.col("timestamp") >= start_naive) & (pl.col("timestamp") < end_naive)
).with_columns(
    pl.col("timestamp").dt.strftime("%Y%m%d").alias("date")
)

print("⏳ 대용량 인프라 정산 메트릭 다차원 분산 집계 중 (Rust Core 스레드 가동)...")
df_raw = lf_filtered.collect()

if df_raw.is_empty():
    print("⚠️  해당 기간 내 정산할 데이터 로우가 없습니다.")
    return

df_raw = df_raw.with_columns([
    pl.col("node").map_elements(
        lambda x: {"cluster_type": classify_node_infrastructure(x)[0], "workload_domain": classify_node_infrastructure(x)[1]},
        return_dtype=pl.Struct({"cluster_type": pl.String, "workload_domain": pl.String})
    ).alias("infra_struct"),
    pl.col("pod").map_elements(get_workload_type, return_dtype=pl.String).alias("workload_type")
]).with_columns([
    pl.col("infra_struct").struct.field("cluster_type").alias("cluster_type"),
    pl.col("infra_struct").struct.field("workload_domain").alias("workload_domain")
]).drop("infra_struct")

target_dates = df_raw["date"].unique().to_list()
target_clusters = ["COMPUTE", "STORAGE"] if args.cluster.upper() == "ALL" else [args.cluster.upper()]

print(f"🎯 분할 연산 타겟 스케줄러 가동 -> 일자 풀: {target_dates} | 클러스터 풀: {target_clusters}")

GB_DIV = 1024 ** 3

for date_chunk in target_dates:
    for cluster_chunk in target_clusters:

        df_slice = df_raw.filter((pl.col("date") == date_chunk) & (pl.col("cluster_type") == cluster_chunk))
        if df_slice.is_empty():
            continue

        if args.workload_domain:
            df_slice = df_slice.filter(pl.col("workload_domain") == args.workload_domain)
            if df_slice.is_empty(): continue

        print(f"   ⏳ [Polars 집계 연산] {date_chunk} ➡️ {cluster_chunk} 고성능 그룹바이 컴파일...")

        target_fields = ["cpu_request", "cpu_limit", "cpu_usage", "cpu_throttled", "mem_request", "mem_limit", "mem_usage", "mem_rss", "oom_event", "pv_capacity", "pv_used"]
        for col in target_fields:
            if col not in df_slice.columns:
                df_slice = df_slice.with_columns(pl.lit(0.0).alias(col))

        # [테이블 1] 상세 메트릭 원부 집계
        df_pod = df_slice.group_by([
            "date", "cluster_type", "workload_domain", "namespace", "workload_type", "node", "pod", "container"
        ]).agg([
            pl.len().alias("minutes_running"),
            pl.col("cpu_request").max().alias("cpu_request_max"),
            pl.col("cpu_limit").max().alias("cpu_limit_max"),
            pl.col("cpu_usage").quantile(0.95).alias("cpu_usage_p95"),
            pl.col("cpu_throttled").max().alias("cpu_throttled_max"),
            (pl.col("mem_request").max() / GB_DIV).alias("mem_request_max"),
            (pl.col("mem_limit").max() / GB_DIV).alias("mem_limit_max"),
            (pl.col("mem_usage").quantile(0.95) / GB_DIV).alias("mem_usage_p95"),
            (pl.col("mem_rss").quantile(0.95) / GB_DIV).alias("mem_rss_p95"),
            pl.col("oom_event").sum().alias("oom_strike_sum"),
            (pl.col("pv_capacity").max() / GB_DIV).alias("pv_capacity_max"),
            (pl.col("pv_used").quantile(0.95) / GB_DIV).alias("pv_used_p95")
        ]).fill_null(0.0)

        df_pod = df_pod.with_columns([
            (pl.col("cpu_request_max") * (pl.col("minutes_running") / 60.0)).alias("cpu_allocated_core_hours"),
            (pl.col("cpu_usage_p95") * (pl.col("minutes_running") / 60.0)).alias("cpu_usage_core_hours"),
            (pl.col("mem_request_max") * (pl.col("minutes_running") / 60.0)).alias("mem_allocated_gb_hours"),
            (pl.col("mem_usage_p95") * (pl.col("minutes_running") / 60.0)).alias("mem_usage_gb_hours"),
            (pl.col("pv_capacity_max") * (pl.col("minutes_running") / 60.0)).alias("pv_allocated_gb_hours"),
            (pl.col("pv_usage_p95") * (pl.col("minutes_running") / 60.0)).alias("pv_usage_gb_hours"),
            (pl.col("oom_strike_sum") > 0).alias("is_oom_killed"),
            (pl.col("cpu_request_max") == 0).alias("has_no_request"),
            (pl.col("cpu_limit_max") == 0).alias("has_no_limit"),
            ((pl.col("cpu_usage_p95") - pl.col("cpu_request_max")).clip(lower_bound=0)).alias("cpu_shortage_cores")
        ]).with_columns([
            ((pl.col("cpu_allocated_core_hours") - pl.col("cpu_usage_core_hours")).clip(lower_bound=0)).alias("cpu_waste_core_hours"),
            ((pl.col("mem_allocated_gb_hours") - pl.col("mem_usage_gb_hours")).clip(lower_bound=0)).alias("mem_waste_gb_hours"),
            ((pl.col("pv_allocated_gb_hours") - pl.col("pv_usage_gb_hours")).clip(lower_bound=0)).alias("pv_waste_gb_hours"),
        ])

        df_pod = df_pod.with_columns(
            pl.when(pl.col("is_oom_killed")).then(pl.lit("💥 OOM장애발생"))
            .when((pl.col("cpu_shortage_cores") > 0.5) | (pl.col("cpu_throttled_max") > 0.2)).then(pl.lit("⚠️ Request부족"))
            .when((pl.col("cpu_waste_core_hours") > 10) | (pl.col("pv_waste_gb_hours") > 50)).then(pl.lit("📉 과다할당"))
            .otherwise(pl.lit("✅ 최적화완료")).alias("status")
        )

        # 📅 [테이블 2] Namespace 사용량 요약 빌드
        df_daily_ns = df_pod.group_by(["date", "namespace"]).agg([
            pl.col("cpu_usage_core_hours").sum().alias("cpu_used_ch"),
            pl.col("cpu_allocated_core_hours").sum().alias("cpu_alloc_ch"),
            pl.col("cpu_waste_core_hours").sum().alias("cpu_waste_ch"),
            pl.col("mem_usage_gb_hours").sum().alias("mem_used_gh"),
            pl.col("mem_allocated_gb_hours").sum().alias("mem_alloc_gh"),
            pl.col("pv_usage_gb_hours").sum().alias("pv_used_gh"),
            pl.col("pv_allocated_gb_hours").sum().alias("pv_alloc_gh")
        ]).with_columns(
            ((pl.col("cpu_alloc_ch") * W_CPU) + (pl.col("mem_alloc_gh") * W_MEM) + (pl.col("pv_alloc_gh") * W_PV)).round(1).alias("final_usage_score")
        )

        # 📉 [테이블 3] 파레토 원부 롤업 빌드 
        df_ns = df_pod.group_by("namespace").agg([
            pl.col("cpu_waste_core_hours").sum().alias("total_waste_core_hours"),
            pl.col("minutes_running").sum().alias("minutes_running_sum"),
            pl.col("container").count().alias("container_cnt"),
            pl.col("cpu_allocated_core_hours").sum().alias("total_allocated_core_hours")
        ]).sort("total_waste_core_hours", descending=True)

        global_total_waste = df_ns["total_waste_core_hours"].sum() or 0.1
        df_ns = df_ns.with_columns([
            pl.lit(date_chunk).alias("date"),
            ((pl.col("total_waste_core_hours") / global_total_waste) * 100).round(2).alias("waste_share_pct")
        ]).with_columns(
            pl.col("waste_share_pct").cum_sum().round(2).alias("waste_cumsum_pct")
        )

        # ─── 💾 [로컬 저장 명세 고정 바인딩] ───
        cluster_partition_dir = MERGED_DIR / cluster_chunk
        cluster_partition_dir.mkdir(parents=True, exist_ok=True)

        target_output_file = cluster_partition_dir / f"daily_enriched_{cluster_chunk}_{date_chunk}.parquet"
        ns_output_file     = cluster_partition_dir / f"daily_ns_usage_{cluster_chunk}_{date_chunk}.parquet"
        pareto_output_file = cluster_partition_dir / f"pareto_ns_{cluster_chunk}_{date_chunk}.parquet"

        df_pod.write_parquet(target_output_file)
        df_daily_ns.write_parquet(ns_output_file)
        df_ns.write_parquet(pareto_output_file) # ◀ 파레토 로컬 디스크 정착

        print(f"     💾 [로컬 저장] 파티션 3대 원부 컴파일 성료 (클러스터: {cluster_chunk})")

        # ─── 📡 [AIStor 영구 적재 가드레일 주입] ───
        print(f"     Submitting artifacts to AIStor Tables storage system...")
        try:
            s3_client.upload_file(str(target_output_file), bucket_name, f"merged/{cluster_chunk}/{target_output_file.name}")
            s3_client.upload_file(str(ns_output_file), bucket_name, f"merged/{cluster_chunk}/{ns_output_file.name}")
            s3_client.upload_file(str(pareto_output_file), bucket_name, f"merged/{cluster_chunk}/{pareto_output_file.name}")
            print(f"     ✅ [AIStor 업로드 완수] 테이블 자산 적재 완료 -> {pareto_output_file.name}")
        except Exception as e:
            print(f"     ❌ [AIStor 통신장애] 오브젝트 전송 실패: {str(e)}")

print("\n🏁 === [Step2 Polars 배치 완료] 로컬 및 AIStor 테이블 연동이 완벽히 마감되었습니다. ===")

if name == "main": main()

... [기존 로컬 저장 구역] ...

        cluster_partition_dir = MERGED_DIR / cluster_chunk
        cluster_partition_dir.mkdir(parents=True, exist_ok=True)

        target_output_file = cluster_partition_dir / f"daily_enriched_{cluster_chunk}_{date_chunk}.parquet"
        ns_output_file     = cluster_partition_dir / f"daily_ns_usage_{cluster_chunk}_{date_chunk}.parquet"
        pareto_output_file = cluster_partition_dir / f"pareto_ns_{cluster_chunk}_{date_chunk}.parquet"

        # (로컬 쓰기 집행 완료 상태)
        # df_pod.to_parquet(target_output_file, index=False)
        # ...

        # ─── 📡 [신규 추가] AIStor(MinIO) 테이블 레이어로 즉시 슈팅 (Sync-Up) ───
        print(f"     📤 [AIStor 적재] 정산 완료본 오브젝트 스토리지로 업로드 중...")
        try:
            # 1. 상세 원부 업로드 -> s3://devops-test/merged/STORAGE/daily_enriched_STORAGE_20260630.parquet
            s3_client.upload_file(
                str(target_output_file), bucket_name, 
                f"merged/{cluster_chunk}/{target_output_file.name}"
            )
            # 2. 일별 사용량 요약 업로드
            s3_client.upload_file(
                str(ns_output_file), bucket_name, 
                f"merged/{cluster_chunk}/{ns_output_file.name}"
            )
            # 3. 파레토 지분 원부 업로드
            s3_client.upload_file(
                str(pareto_output_file), bucket_name, 
                f"merged/{cluster_chunk}/{pareto_output_file.name}"
            )
            print(f"     ✅ [업로드 완수] AIStor Table 동기화 완료.")

        except Exception as e:
            print(f"     ❌ [업로드 실패] AIStor 네트워크 통신 장애: {str(e)}")
        # ──────────────────────────────────────────────────────────────────────

─── 📂 경로 표준화 설정 ──────────────────────────────────────────────────

BASE_DATA_DIR = Path("./data") MERGED_DIR = BASE_DATA_DIR / "merged" OUTPUT_DIR = BASE_DATA_DIR / "output"

출력 디렉토리 자동 생성

OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

def parse_arguments(): parser = argparse.ArgumentParser(description="FinOps Governance & Anomaly Analyzer") parser.add_argument("--cluster", type=str, default="ALL", choices=["COMPUTE", "STORAGE", "ALL"]) return parser.parse_args()

def load_partitioned_data(cluster_target): """ 개편된 2단계 분할 정복 하위 폴더 위계를 스캔하여 일별 Parquet 파일들을 동적으로 결합합니다. """ target_clusters = ["COMPUTE", "STORAGE"] if cluster_target.upper() == "ALL" else [cluster_target.upper()]

pod_files = []
ns_files = []
pareto_files = []

# 💡 [적응형 픽업 핵심 구역] 지정된 클러스터 폴더 하방의 모든 일별 파일들을 수집
for cl in target_clusters:
    cl_dir = MERGED_DIR / cl
    if cl_dir.exists():
        pod_files.extend(glob.glob(str(cl_dir / "daily_enriched_*.parquet")))
        ns_files.extend(glob.glob(str(cl_dir / "daily_ns_usage_*.parquet")))
        pareto_files.extend(glob.glob(str(cl_dir / "pareto_ns_*.parquet")))

print(f"🔍 [스캔 완료] Enriched 파일: {len(pod_files)}개 | NS요약 파일: {len(ns_files)}개 | Pareto 파일: {len(pareto_files)}개")

# 파일이 단 하나도 없을 경우의 빈 데이터프레임 예외 예방 가드레일
df_pod = pd.concat([pd.read_parquet(f) for f in pod_files], ignore_index=True) if pod_files else pd.DataFrame()
df_ns  = pd.concat([pd.read_parquet(f) for f in ns_files], ignore_index=True) if ns_files else pd.DataFrame()
df_pt  = pd.concat([pd.read_parquet(f) for f in pareto_files], ignore_index=True) if pareto_files else pd.DataFrame()

return df_pod, df_ns, df_pt

def main(): args = parse_arguments() print(f"🚀 [Step4] Starting FinOps Governance Analyzer for Cluster: {args.cluster}...")

# 1. 쪼개져 있는 2단계 결과물 전체 로드 (적응형 복구)
df_pod, df_ns, df_pt = load_partitioned_data(args.cluster)

if df_pod.empty:
    print(f"❌ [중단] '{args.cluster}' 클러스터에 정산된 2단계 가공 원부가 존재하지 않습니다. 앞단 배치를 확인하세요.")
    return

cluster_suffix = args.cluster.upper() if args.cluster.upper() != "ALL" else "INTEGRATED"
print(f"✅ 총 {len(df_pod):,}개의 컨테이너 타임라인 원부 통합 바인딩 완료.\n")

# 2. 🚨 [거버넌스 분석 레이어 1] 자원부족 및 OOM Killed 위험군 추출 (시트4 매핑용)
print("🚨 [Analysis 1] Extracting Critical Risk & OOM Killed Pods...")
df_risk = df_pod[
    (df_pod["status"] == "💥 OOM장애발생") | 
    (df_pod["status"] == "⚠️ Request부족") | 
    (df_pod["oom_strike_sum"] > 0)
].copy()

# 조치 가이드 컬럼 인젝션
if not df_risk.empty:
    df_risk["recommendation"] = pd.DataFrame(
        # OOM일 경우 Limit 상향, 쓰로틀링/자원부족일 경우 Request 상향 가이드 명시
        [ "Upsize Memory Limit (OOM Detected)" if x else "Upsize CPU Request/Limit (Throttling Detected)" for x in df_risk["is_oom_killed"] ]
    )
else:
    df_risk["recommendation"] = None

# 3. 📉 [거버넌스 분석 레이어 2] 과다 할당 및 유휴 알박기 자산 추출 (시트2, 3 하향 후보용)
print("📉 [Analysis 2] Extracting Over-Allocated & Idle Infrastructure Assets...")
df_waste = df_pod[
    (df_pod["status"] == "📉 과다할당") & 
    (df_pod["cpu_waste_core_hours"] > 24)  # 하루 이상 코어 하나 통째로 낭비한 기준
].sort_values(by="cpu_waste_core_hours", ascending=False).copy()

# 4. 🛡️ [거버넌스 분석 레이어 3] 자원 미설정 배포 위반군 추출 (시트5 매핑용)
print("🛡️ [Analysis 3] Scanning Non-Compliant Missing Resource Specification Pods...")
df_violations = df_pod[
    (df_pod["has_no_request"] == True) | 
    (df_pod["has_no_limit"] == True)
].copy()

# 5. 💾 [결과 마감] 6단계 엑셀 빌더가 다이렉트로 인식할 수 있도록 산출물 저장
print(f"\n💾 [정산 마감] 분석 데이터 자산 output 레이어로 내보내기 진행 중...")

# 6단계 엑셀 빌더 경로 명명 규칙 통합 바인딩
master_gov_file = OUTPUT_DIR / f"governance_master_{cluster_suffix}.parquet"
risk_file       = OUTPUT_DIR / f"gov_risk_oom_{cluster_suffix}.parquet"
waste_file      = OUTPUT_DIR / f"gov_waste_candidates_{cluster_suffix}.parquet"
viol_file       = OUTPUT_DIR / f"gov_violations_{cluster_suffix}.parquet"

# Parquet 압축 저장
df_pod.to_parquet(master_gov_file, index=False)
df_risk.to_parquet(risk_file, index=False)
df_waste.to_parquet(waste_file, index=False)
df_violations.to_parquet(viol_file, index=False)

# 검증 서머리 출력
print(f"  -> 📦 [저장완료] 전사 통합 마스터 원부 : {master_gov_file.name}")
print(f"  -> 💥 [저장완료] 고위험군/OOM 리스트    : {risk_file.name} (결과: {len(df_risk)}건)")
print(f"  -> 📉 [저장완료] 자원 하향조정 후보군  : {waste_file.name} (결과: {len(df_waste)}건)")
print(f"  -> 🛡️ [저장완료] 스펙 미설정 규격위반군: {viol_file.name} (결과: {len(df_violations)}건)")

print(f"\n🏁 === [Step4 완수] '{cluster_suffix}' 거버넌스 가공 원부가 무결하게 갱신되었습니다. ===")

if name == "main": main()

─── 🛡️ [Headless Environment Guard] Prevents crashes in K8s pods without display servers ───

import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import seaborn as sns

─── 📂 Directory Path Standardization (./data) ──────────────────────────

BASE_DATA_DIR = Path("./data") MERGED_DIR = BASE_DATA_DIR / "merged" BASE_PLOT_DIR = BASE_DATA_DIR / "output" / "plots"

Ensure base plot directory exists

BASE_PLOT_DIR.mkdir(parents=True, exist_ok=True)

─── 🔤 [No More Tofu] Use bulletproof built-in standard fonts ───

plt.rcParams['font.family'] = 'sans-serif' plt.rcParams['axes.unicode_minus'] = False # Prevents minus sign (-) corruption sns.set_theme(style="whitegrid")

─── 🛡️ [완치 가드레일] 빈 데이터셋 감지 시 플레이스홀더 PNG 생성 함수 ───

def check_and_handle_empty(df, output_path, chart_name): """ 데이터가 비어있으면 True를 반환하고, 6단계 엑셀 빌더가 터지지 않도록 'No Data Available' 문구가 적힌 플레이스홀더 이미지를 강제 생성합니다. """ if df is None or df.empty: print(f" ⚠️ [Empty Data] Generating placeholder for {chart_name} (Dataset is empty).") fig, ax = plt.subplots(figsize=(10, 5)) ax.text(0.5, 0.5, f"No Data Available\n({chart_name})", ha='center', va='center', fontsize=14, color='gray', weight='bold') ax.axis('off') plt.tight_layout() plt.savefig(output_path, dpi=100, bbox_inches='tight') plt.close() return True return False

def main(): parser = argparse.ArgumentParser(description="FinOps Analytics Visualization Engine") parser.add_argument("--cluster", type=str, default="ALL", choices=["COMPUTE", "STORAGE", "ALL"]) args = parser.parse_args()

print(f"🚀 [Step3] Starting FinOps Time-series English Visualization Engine for Cluster: {args.cluster}...")

# ─── 🗂️ 클러스터별 파티션 파일 서치 및 가변 결합 (분할 정복) ───
target_clusters = ["COMPUTE", "STORAGE"] if args.cluster.upper() == "ALL" else [args.cluster.upper()]

all_pod_dfs = []
all_ns_dfs = []

for cl in target_clusters:
    cl_dir = MERGED_DIR / cl
    if cl_dir.exists():
        all_pod_dfs.extend([pd.read_parquet(f) for f in cl_dir.glob("daily_enriched_*.parquet")])
        all_ns_dfs.extend([pd.read_parquet(f) for f in cl_dir.glob("pareto_ns_*.parquet")])

if not all_pod_dfs:
    print(f"❌ Error: No processed daily Parquet files found for cluster '{args.cluster}'. Please run step2 first.")
    return

df_pod = pd.concat(all_pod_dfs, ignore_index=True)
df_ns  = pd.concat(all_ns_dfs, ignore_index=True) if all_ns_dfs else pd.DataFrame()

# 출력 경로 역시 클러스터별 하위 디렉토리로 격리 파티셔닝
cluster_type_str = args.cluster.upper() if args.cluster.upper() != "ALL" else "INTEGRATED"
PLOT_DIR = BASE_PLOT_DIR / cluster_type_str
PLOT_DIR.mkdir(parents=True, exist_ok=True)

print(f"✅ Data lake aggregated successfully -> Container rows: {len(df_pod):,}")

# 데이터 정렬 오염 차단용 형변환 Guard
df_pod["date"] = df_pod["date"].astype(str)

# Injecting efficiency metric layers
df_pod["cpu_util"] = np.where(df_pod["cpu_request_max"] > 0, (df_pod["cpu_usage_p95"] / df_pod["cpu_request_max"] * 100), 0)
df_pod["mem_util"] = np.where(df_pod["mem_request_max"] > 0, (df_pod["mem_usage_p95"] / df_pod["mem_request_max"] * 100), 0)
df_pod["lim_req_ratio"] = np.where(df_pod["cpu_request_max"] > 0, df_pod["cpu_limit_max"] / df_pod["cpu_request_max"], 0)

# ─── 📊 [Charts 1 & 2] Workload Type Allocated vs Actual Peak ───
print("⏳ [1/19] Rendering chart1_cpu_req_vs_usage_by_workload...")
out1 = PLOT_DIR / "chart1_cpu_req_vs_usage_by_workload.png"
if not check_and_handle_empty(df_pod, out1, "chart1_cpu_req_vs_usage_by_workload"):
    df_wl_cpu = df_pod.groupby("workload_type")[["cpu_request_max", "cpu_usage_p95"]].mean().reset_index()
    plt.figure(figsize=(10, 5))
    df_melt_cpu = df_wl_cpu.melt(id_vars="workload_type", value_vars=["cpu_request_max", "cpu_usage_p95"])
    sns.barplot(data=df_melt_cpu, x="workload_type", y="value", hue="variable", palette="Blues_r")
    plt.xticks(rotation=30, ha='right')
    plt.title("Average CPU Request vs P95 Peak Usage by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("CPU Cores")
    plt.tight_layout()
    plt.savefig(out1, dpi=100)
    plt.close()

print("⏳ [2/19] Rendering chart2_mem_req_vs_usage_by_workload...")
out2 = PLOT_DIR / "chart2_mem_req_vs_usage_by_workload.png"
if not check_and_handle_empty(df_pod, out2, "chart2_mem_req_vs_usage_by_workload"):
    df_wl_mem = df_pod.groupby("workload_type")[["mem_request_max", "mem_usage_p95"]].mean().reset_index()
    plt.figure(figsize=(10, 5))
    df_melt_mem = df_wl_mem.melt(id_vars="workload_type", value_vars=["mem_request_max", "mem_usage_p95"])
    sns.barplot(data=df_melt_mem, x="workload_type", y="value", hue="variable", palette="Purples_r")
    plt.xticks(rotation=30, ha='right')
    plt.title("Average Memory Request vs P95 Peak Usage (GB) by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("Memory Capacity (GB)")
    plt.tight_layout()
    plt.savefig(out2, dpi=100)
    plt.close()

# ─── 📊 [Chart 3] Daily CPU Waste Stacked Bar ───
print("⏳ [3/19] Rendering chart3_daily_waste_stack...")
out3 = PLOT_DIR / "chart3_daily_waste_stack.png"
df_daily_waste = df_pod.groupby(["date", "workload_type"])["cpu_waste_core_hours"].sum().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_daily_waste, out3, "chart3_daily_waste_stack"):
    df_daily_waste.plot(kind='bar', stacked=True, figsize=(11, 5), cmap="tab20")
    plt.title("Daily Total CPU Waste Core-Hours Stacked by Workload (KST)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Waste Volume (Core-Hours)")
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig(out3, dpi=100)
    plt.close()

# ─── 📊 [Charts 4 & 18] Efficiency Heatmaps ───
print("⏳ [4/19] Rendering chart4_cpu_efficiency_heatmap...")
out4 = PLOT_DIR / "chart4_cpu_efficiency_heatmap.png"
df_heat_cpu = df_pod.groupby(["workload_type", "date"])["cpu_util"].mean().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_heat_cpu, out4, "chart4_cpu_efficiency_heatmap"):
    plt.figure(figsize=(10, 5))
    sns.heatmap(df_heat_cpu, annot=True, fmt=".1f", cmap="RdYlGn", cbar=True)
    plt.title("Mean CPU Utilization Heatmap (%) (Workload Type x Date)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Workload Type")
    plt.tight_layout()
    plt.savefig(out4, dpi=100)
    plt.close()

print("⏳ [5/19] Rendering chart18_mem_waste_heatmap...")
out18 = PLOT_DIR / "chart18_mem_waste_heatmap.png"
df_heat_mem = df_pod.groupby(["workload_type", "date"])["mem_waste_gb_hours"].sum().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_heat_mem, out18, "chart18_mem_waste_heatmap"):
    plt.figure(figsize=(10, 5))
    sns.heatmap(df_heat_mem, annot=False, cmap="BuPu", cbar=True)
    plt.title("Total Memory Waste Volume Heatmap (GB-Hours)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Workload Type")
    plt.tight_layout()
    plt.savefig(out18, dpi=100)
    plt.close()

# ─── 📊 [Charts 5 & 11] Pareto Governance Charts ───
print("⏳ [6/19] Rendering chart5_pareto_ns_waste...")
out5 = PLOT_DIR / "chart5_pareto_ns_waste.png"
if not check_and_handle_empty(df_ns, out5, "chart5_pareto_ns_waste"):
    # 단일 테넌트 중복 방지용 정렬 및 상위 스캔
    df_ns_top = df_ns.groupby("namespace")["total_waste_core_hours"].sum().reset_index().sort_values("total_waste_core_hours", ascending=False).head(15)
    df_ns_top["waste_share_pct"] = (df_ns_top["total_waste_core_hours"] / df_ns_top["total_waste_core_hours"].sum() * 100).round(2)
    df_ns_top["waste_cumsum_pct"] = df_ns_top["waste_share_pct"].cumsum().round(2)

    fig, ax1 = plt.subplots(figsize=(11, 5))
    sns.barplot(data=df_ns_top, x="namespace", y="total_waste_core_hours", ax=ax1, color="steelblue")
    ax1.set_xticklabels(ax1.get_xticklabels(), rotation=45, ha="right")
    ax2 = ax1.twinx()
    ax2.plot(df_ns_top["namespace"], df_ns_top["waste_cumsum_pct"], color="crimson", marker="o", linewidth=2)
    ax2.set_ylim(0, 110)
    plt.title("Top 15 Namespace Cost-Waste Pareto Chart (Cumulative Share %)")
    ax1.set_xlabel("Tenant Namespace")
    ax1.set_ylabel("Waste Volume (Core-Hours)")
    plt.tight_layout()
    plt.savefig(out5, dpi=100)
    plt.close()

print("⏳ [7/19] Rendering chart11_pareto_workload_waste...")
out11 = PLOT_DIR / "chart11_pareto_workload_waste.png"
if not check_and_handle_empty(df_pod, out11, "chart11_pareto_workload_waste"):
    df_wl_waste = df_pod.groupby("workload_type")["cpu_waste_core_hours"].sum().reset_index().sort_values("cpu_waste_core_hours", ascending=False)
    plt.figure(figsize=(10, 5))
    sns.barplot(data=df_wl_waste, x="workload_type", y="cpu_waste_core_hours", palette="Oranges_r")
    plt.xticks(rotation=30, ha="right")
    plt.title("Total CPU Waste Volume by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("Total Waste (Core-Hours)")
    plt.tight_layout()
    plt.savefig(out11, dpi=100)
    plt.close()

# ─── 📊 [Chart 6] Governance Status Donut Chart ───
print("⏳ [8/19] Rendering chart6_status_donut...")
out6 = PLOT_DIR / "chart6_status_donut.png"
if not check_and_handle_empty(df_pod, out6, "chart6_status_donut"):
    status_summary = df_pod["status"].value_counts()
    plt.figure(figsize=(6, 5))
    colors = ["#70AD47", "#1F4E79", "#FFC000", "#C00000"]
    plt.pie(status_summary, labels=status_summary.index, autopct='%1.1f%%', startangle=90, colors=colors[:len(status_summary)], wedgeprops=dict(width=0.4, edgecolor='w'))
    plt.title("Infrastructure Resource Governance Status Ratio")
    plt.tight_layout()
    plt.savefig(out6, dpi=100)
    plt.close()

# ─── 📊 [Charts 7 & 14] Bubble & Scatter Plots ───
print("⏳ [9/19] Rendering chart7_waste_footprint_bubble...")
out7 = PLOT_DIR / "chart7_waste_footprint_bubble.png"
if not check_and_handle_empty(df_pod, out7, "chart7_waste_footprint_bubble"):
    df_bubble = df_pod.groupby("workload_type").agg(
        x_alloc=("cpu_allocated_core_hours", "sum"),
        y_util=("cpu_util", "mean"),
        z_waste=("cpu_waste_core_hours", "sum")
    ).reset_index()
    plt.figure(figsize=(9, 6))
    sns.scatterplot(data=df_bubble, x="x_alloc", y="y_util", size="z_waste", hue="workload_type", sizes=(100, 2000), alpha=0.7, legend="brief")
    plt.title("Resource Footprint Bubble Chart (Allocated x Utilization x Waste Size)")
    plt.xlabel("Total Allocated Core-Hours")
    plt.ylabel("Average Utilization (%)")
    plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
    plt.tight_layout()
    plt.savefig(out7, dpi=100)
    plt.close()

print("⏳ [10/19] Rendering chart14_cpu_mem_waste_scatter...")
out14 = PLOT_DIR / "chart14_cpu_mem_waste_scatter.png"
if not check_and_handle_empty(df_pod, out14, "chart14_cpu_mem_waste_scatter"):
    plt.figure(figsize=(8, 6))
    sns.scatterplot(data=df_pod.head(5000), x="cpu_waste_core_hours", y="mem_waste_gb_hours", hue="workload_type", alpha=0.5)
    plt.title("Co-relation Scatter Plot: CPU Waste vs Memory Waste (Sampled)")
    plt.xlabel("CPU Waste (Core-Hours)")
    plt.ylabel("Memory Waste (GB-Hours)")
    plt.tight_layout()
    plt.savefig(out14, dpi=100)
    plt.close()

# ─── 📊 [Chart 8] Resource Shortage Footprint ───
print("⏳ [11/19] Rendering chart8_shortfall_footprint...")
out8 = PLOT_DIR / "chart8_shortfall_footprint.png"
df_short = df_pod.groupby(["workload_type", "date"])["cpu_shortage_cores"].sum().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_short, out8, "chart8_shortfall_footprint"):
    plt.figure(figsize=(10, 4))
    sns.heatmap(df_short, annot=False, cmap="YlOrRd", cbar=True)
    plt.title("Total CPU Shortfall (Deficit) Cores Footprint")
    plt.xlabel("Date (KST)")
    plt.ylabel("Workload Type")
    plt.tight_layout()
    plt.savefig(out8, dpi=100)
    plt.close()

# ─── 📊 [Charts 9 & 10] Boxplot Distribution Analysis ───
print("⏳ [12/19] Rendering chart9_boxplot_cpu_util_by_workload...")
out9 = PLOT_DIR / "chart9_boxplot_cpu_util_by_workload.png"
if not check_and_handle_empty(df_pod, out9, "chart9_boxplot_cpu_util_by_workload"):
    plt.figure(figsize=(11, 5))
    sns.boxplot(data=df_pod, x="workload_type", y="cpu_util", palette="Set3")
    plt.xticks(rotation=30, ha="right")
    plt.ylim(-5, 105)
    plt.title("CPU Utilization P95 Boxplot Distribution by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("P95 Actual Utilization (%)")
    plt.tight_layout()
    plt.savefig(out9, dpi=100)
    plt.close()

print("⏳ [13/19] Rendering chart10_boxplot_mem_util_by_workload...")
out10 = PLOT_DIR / "chart10_boxplot_mem_util_by_workload.png"
if not check_and_handle_empty(df_pod, out10, "chart10_boxplot_mem_util_by_workload"):
    plt.figure(figsize=(11, 5))
    sns.boxplot(data=df_pod, x="workload_type", y="mem_util", palette="Pastel1")
    plt.xticks(rotation=30, ha="right")
    plt.ylim(-5, 105)
    plt.title("Memory Utilization P95 Boxplot Distribution by Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("P95 Actual Utilization (%)")
    plt.tight_layout()
    plt.savefig(out10, dpi=100)
    plt.close()

# ─── 📊 [Charts 12 & 15] Trend and Counts ───
print("⏳ [14/19] Rendering chart12_daily_waste_trend_by_workload...")
out12 = PLOT_DIR / "chart12_daily_waste_trend_by_workload.png"
df_trend_wl = df_pod.groupby(["date", "workload_type"])["cpu_waste_core_hours"].sum().unstack().fillna(0) if not df_pod.empty else pd.DataFrame()
if not check_and_handle_empty(df_trend_wl, out12, "chart12_daily_waste_trend_by_workload"):
    plt.figure(figsize=(11, 5))
    sns.lineplot(data=df_trend_wl, markers=True, dashes=False, linewidth=2)
    plt.title("Daily CPU Waste Timeline Trend Line by Workload Type")
    plt.xlabel("Date (KST)")
    plt.ylabel("Waste Volume (Core-Hours)")
    plt.xticks(rotation=30)
    plt.tight_layout()
    plt.savefig(out12, dpi=100)
    plt.close()

print("⏳ [15/19] Rendering chart15_oom_status_by_workload...")
out15 = PLOT_DIR / "chart15_oom_status_by_workload.png"
if not check_and_handle_empty(df_pod, out15, "chart15_oom_status_by_workload"):
    plt.figure(figsize=(11, 5))
    sns.countplot(data=df_pod, x="workload_type", hue="status", palette="muted")
    plt.xticks(rotation=30, ha="right")
    plt.title("Governance Status Distribution Count per Workload Type")
    plt.xlabel("Workload Type")
    plt.ylabel("Pod Count")
    plt.legend(loc="upper right")
    plt.tight_layout()
    plt.savefig(out15, dpi=100)
    plt.close()

# ─── 📊 [Charts 13 & 17] Violin and Overcommit Boxplot ───
print("⏳ [16/19] Rendering chart13_violin_cpu_util...")
out13 = PLOT_DIR / "chart13_violin_cpu_util.png"
if not check_and_handle_empty(df_pod, out13, "chart13_violin_cpu_util"):
    plt.figure(figsize=(10, 5))
    sns.violinplot(data=df_pod, x="workload_type", y="cpu_util", inner="quartile", palette="pastel")
    plt.xticks(rotation=30, ha="right")
    plt.title("CPU Utilization Kernel Density Violin Plot")
    plt.xlabel("Workload Type")
    plt.ylabel("CPU Utilization (%)")
    plt.tight_layout()
    plt.savefig(out13, dpi=100)
    plt.close()

print("⏳ [17/19] Rendering chart17_cpu_limit_request_ratio...")
out17 = PLOT_DIR / "chart17_cpu_limit_request_ratio.png"
if not check_and_handle_empty(df_pod, out17, "chart17_cpu_limit_request_ratio"):
    plt.figure(figsize=(10, 5))
    sns.boxplot(data=df_pod, x="workload_type", y="lim_req_ratio", palette="vlag")
    plt.xticks(rotation=30, ha="right")
    plt.title("Kubernetes Pod CPU Limit / Request Overcommit Ratio")
    plt.xlabel("Workload Type")
    plt.ylabel("Limit / Request Ratio")
    plt.tight_layout()
    plt.savefig(out17, dpi=100)
    plt.close()

# ─── 📊 [Charts 19 & 20] Capacity vs Actual Trends ───
print("⏳ [18/19] Rendering chart19_daily_cpu_per_workload...")
out19 = PLOT_DIR / "chart19_daily_cpu_per_workload.png"
if not check_and_handle_empty(df_pod, out19, "chart19_daily_cpu_per_workload"):
    df_daily_cpu_req = df_pod.groupby("date")["cpu_request_max"].sum()
    df_daily_cpu_use = df_pod.groupby("date")["cpu_usage_p95"].sum()
    plt.figure(figsize=(11, 5))
    plt.fill_between(df_daily_cpu_req.index, df_daily_cpu_req.values, label="Total CPU Request Cores", color="skyblue", alpha=0.4)
    plt.plot(df_daily_cpu_use.index, df_daily_cpu_use.values, label="Total CPU P95 Actual Cores", color="navy", linewidth=2.5, marker="o")
    plt.title("Daily Total CPU Capacity Allocation vs Actual Peak Consumption (KST)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Total CPU Cores")
    plt.xticks(rotation=30)
    plt.legend(loc="upper left")
    plt.tight_layout()
    plt.savefig(out19, dpi=100)
    plt.close()

print("⏳ [19/19] Rendering chart20_daily_mem_per_workload...")
out20 = PLOT_DIR / "chart20_daily_mem_per_workload.png"
if not check_and_handle_empty(df_pod, out20, "chart20_daily_mem_per_workload"):
    df_daily_mem_req = df_pod.groupby("date")["mem_request_max"].sum()
    df_daily_mem_use = df_pod.groupby("date")["mem_usage_p95"].sum()
    plt.figure(figsize=(11, 5))
    plt.fill_between(df_daily_mem_req.index, df_daily_mem_req.values, label="Total Memory Request (GB)", color="plum", alpha=0.4)
    plt.plot(df_daily_mem_use.index, df_daily_mem_use.values, label="Total Memory P95 Actual (GB)", color="purple", linewidth=2.5, marker="o")
    plt.title("Daily Total Memory Capacity Allocation vs Actual Peak Consumption (KST)")
    plt.xlabel("Date (KST)")
    plt.ylabel("Total Memory (GB)")
    plt.xticks(rotation=30)
    plt.legend(loc="upper left")
    plt.tight_layout()
    plt.savefig(out20, dpi=100)
    plt.close()

print(f"\n🏁 === [Step3 Success] All charts generated cleanly under ./data/output/plots/{cluster_type_str} ===")

if name == "main": main()

1. 크론잡 마스터 등록

kubectl apply -f finops-daily-cronjob.yaml

2. 등록 상태 및 차기 스케줄 타임라인 확인

kubectl get cronjob finops-daily-pipeline -n devops-finops

3. 🔥 [수동 즉시 트리거] 크론잡 설정을 기반으로 일회성 Job 강제 생성

kubectl create job --from=cronjob/finops-daily-pipeline finops-adhoc-test -n devops-finops

4. 생성된 팟의 실시간 스트리밍 로그 추적 (step1의 레코드 출력과 step2의 StarRocks 연동 상태 확인)

kubectl get pods -n devops-finops | grep finops-adhoc-test kubectl logs -f <조회된-파드명> -n devops-finops

```yaml spec: jobTemplate: spec: # ─── 🛡️ Job 전체의 최대 실패 허용 횟수를 0으로 제한 ─── backoffLimit: 0

  template:
    spec:
      # ─── 🛡️ 팟이 죽었을 때 컨테이너를 다시 살리지 않고 무조건 종료 ───
      restartPolicy: Never 

      containers:
      - name: pipeline-runner
        image: harbor.internal.zone/devops/finops-pipeline:v2

spec: jobTemplate: spec: # ─── 🛡️ 최초 실패 후 딱 1번만 더 새 팟을 띄워 시도 ─── backoffLimit: 1

  template:
    spec:
      # 💡 새 팟을 다른 노드에 스케줄링하기 위해 Never로 지정하는 것이 유리합니다.
      restartPolicy: Never 

      containers:
      - name: pipeline-runner