26J30b2

"""
[신규 2단계 - KST 정렬 및 공통 설정 이식판] 
실행: python step2_pipeline.py
"""

import re
import pandas as pd
import numpy as np
from pathlib import Path

# ── 📦 config.py 로부터 마스터 설정 인프라 수입 ──
from config import RAW_DIR, MERGED_DIR, NODE_PREFIX_PATTERN, get_workload_type

def filter_target_nodes(df):
    """config.py에 등재된 정규식을 활용해 워커 노드 입구 컷 필터 집행"""
    node_pattern = NODE_PREFIX_PATTERN
    print(f"🌐 [노드 정규식 필터] config 원부 패턴 조회 적용: '{node_pattern}'")
    
    try:
        re.compile(node_pattern)
    except re.error:
        print(f"🚨 [경고] config.py의 정규식 문법 오류 포착 -> 전체 허용 백업 가동")
        node_pattern = ".*"

    initial_rows = len(df)
    df_filtered = df[df["node"].str.contains(node_pattern, regex=True, na=False, case=False)].reset_index(drop=True)
    print(f"  -> ✂️  필터링 결과: {initial_rows:,}행 중 {initial_rows - len(df_filtered):,}행 노이즈 제거 -> {len(df_filtered):,}행 생존")
    return df_filtered

def derive_cluster_from_node(node_name):
    """노드 명명 규칙을 기반으로 테넌트 클러스터 역추적"""
    n = str(node_name).lower()
    if "icdlh" in n or "prod" in n: return "prod-data-lakehouse"
    elif "stage" in n or "stg" in n: return "stage-lakehouse"
    return "dev-sandbox"

def run_enrich_and_pareto_pipeline():
    raw_files = list(RAW_DIR.glob("prom_raw_*.parquet"))
    if not raw_files:
        print("❌ 에러: data/raw/ 폴더에 수집된 원천 Parquet 시계열 파일이 없습니다.")
        return

    print(f"📦 총 {len(raw_files)}개의 청크 파일 메모리 맵 병합 가동...")
    df_raw = pd.concat([pd.read_parquet(f) for f in raw_files], ignore_index=True)
    
    # ── ⏱️ [요청 반영] GMT -> KST 타임존 9시간 시프트 및 일자 변환 레이어 ──
    print("⏰ GMT(UTC) 타임스탬프 기반 KST(+9H) 시간 축 보정 정렬 중...")
    df_raw["timestamp_kst"] = df_raw["timestamp"] + pd.Timedelta(hours=9)
    df_raw["date"] = df_raw["timestamp_kst"].dt.strftime("%Y-%m-%d")
    
    # 공통 노드 필터 집행
    df_raw = filter_target_nodes(df_raw)
    if df_raw.empty:
        print("⚠️ 필터 조건에 부합하는 노드가 없어 후처리를 종료합니다.")
        return

    # 클러스터 식별
    df_raw["cluster"] = df_raw["node"].apply(derive_cluster_from_node)
    
    # ── 🧩 [요청 반영] config.py 중앙 제어형 분류 함수 이식 ──
    print("⚙️ config.py 내재화 규칙 기반 워크로드 기술 도메인 맵 주입 중...")
    df_raw["workload_type"] = df_raw["pod"].apply(get_workload_type)

    # 지표 누락 대비 보정용 폴백 레이어
    for col in ["cpu_limit", "mem_limit", "oom_event"]:
        if col not in df_raw.columns: df_raw[col] = 0.0

    print("📊 1분 타임시리즈 롤업 집계 가동 (KST 일자 파티션 기준 수렴)...")
    df_pod = df_raw.groupby(["date", "cluster", "namespace", "workload_type", "node", "pod", "container"]).agg(
        minutes_running      = ("timestamp_kst", "size"), # KST 기반 연속 가동 분 분할 역산
        cpu_request_max      = ("cpu_request", "max"),
        cpu_limit_max        = ("cpu_limit", "max"),
        cpu_usage_p95        = ("cpu_usage", lambda x: x.quantile(0.95)),
        mem_request_max      = ("mem_request", "max"),
        mem_limit_max        = ("mem_limit", "max"),
        mem_usage_p95        = ("mem_usage", lambda x: x.quantile(0.95)),
        oom_strike_sum       = ("oom_event", "sum")
    ).reset_index().fillna(0)

    # 바이트 -> GB 정규화
    for col in ["mem_request_max", "mem_limit_max", "mem_usage_p95"]:
        df_pod[col] = df_pod[col] / (1024**3)

    # Core-Hours 및 GB-Hours 산출 연산
    df_pod["cpu_allocated_core_hours"] = df_pod["cpu_request_max"] * (df_pod["minutes_running"] / 60.0)
    df_pod["cpu_usage_core_hours"]     = df_pod["cpu_usage_p95"] * (df_pod["minutes_running"] / 60.0)
    df_pod["cpu_waste_core_hours"]     = (df_pod["cpu_allocated_core_hours"] - df_pod["cpu_usage_core_hours"]).clip(lower=0)

    df_pod["mem_allocated_gb_hours"]   = df_pod["mem_request_max"] * (df_pod["minutes_running"] / 60.0)
    df_pod["mem_usage_gb_hours"]       = df_pod["mem_usage_p95"] * (df_pod["minutes_running"] / 60.0)
    df_pod["mem_waste_gb_hours"]       = (df_pod["mem_allocated_gb_hours"] - df_pod["mem_usage_gb_hours"]).clip(lower=0)

    # 거버넌스 단속용 플래그 주입
    df_pod["is_oom_killed"] = df_pod["oom_strike_sum"] > 0
    df_pod["has_no_request"] = (df_pod["cpu_request_max"] == 0) | (df_pod["mem_request_max"] == 0)
    df_pod["has_no_limit"] = (df_pod["cpu_limit_max"] == 0) | (df_pod["mem_limit_max"] == 0)

    # 공급 부족 크래시 마진 확인
    df_pod["cpu_shortage_cores"] = (df_pod["cpu_usage_p95"] - df_pod["cpu_request_max"]).clip(lower=0)
    df_pod["status"] = np.where(df_pod["is_oom_killed"], "💥 OOM장애발생",
                        np.where(df_pod["cpu_shortage_cores"] > 0.5, "⚠️ Request부족", 
                        np.where(df_pod["cpu_waste_core_hours"] > 10, "📉 과다할당", "✅ 최적화완료")))

    # 최종 병합 데이터 마스터 디스크 박음
    df_pod.to_parquet(MERGED_DIR / "enriched_fixed_7d.parquet", index=False)
    
    # Namespace 파레토 연산
    df_ns = df_pod.groupby("namespace").agg(
        minutes_running_sum = ("minutes_running", "sum"),
        container_cnt       = ("container", "count"),
        total_allocated_core_hours = ("cpu_allocated_core_hours", "sum"),
        total_waste_core_hours     = ("cpu_waste_core_hours", "sum")
    ).reset_index().sort_values(by="total_waste_core_hours", ascending=False).reset_index(drop=True)
    global_total_waste = df_ns["total_waste_core_hours"].sum() if df_ns["total_waste_core_hours"].sum() > 0 else 0.1
    df_ns["waste_share_pct"] = (df_ns["total_waste_core_hours"] / global_total_waste * 100).round(2)
    df_ns["waste_cumsum_pct"] = df_ns["waste_share_pct"].cumsum().round(2)
    
    df_ns.to_parquet(MERGED_DIR / "pareto_fixed_ns.parquet", index=False)
    print("💾 KST 보정 및 중앙식 워크로드 매핑 완료본 Parquet 2종 세이브 성료.\n")

if __name__ == "__main__":
    run_enrich_and_pareto_pipeline()