26J30d3

"""
[신규 2단계 - MinIO 데이터 레이크 백엔드 동적 리싱크 및 KST 장기 후처리 엔진]
설명: 분석하고자 하는 과거 7일/30일치 기간을 주면, 타노스를 치지 않고 사내 MinIO AIStor에 누적되어 있는 
      고해상도 원천 Parquet 파일들을 고속 다운로드(Sync-Down)하여 로컬에서 일괄 병합 재처리합니다.
실행: python step2_pipeline.py --days 30
"""

import os
import argparse
import re
import boto3
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone
from pathlib import Path

# config 마스터 사양 로드
from config import RAW_DIR, MERGED_DIR, MINIO_RAW_BUCKET, NODE_PREFIX_PATTERN, get_workload_type

KST = timezone(timedelta(hours=9))

def parse_arguments():
    parser = argparse.ArgumentParser(description="FinOps Data Lakehouse Re-processing Engine")
    parser.add_argument("--days", type=int, default=7, help="과거 며칠간의 원천 데이터를 레이크에서 내려받아 재처리할지 정의 (기본 7일)")
    parser.add_argument("--start-date", type=str, default=None, help="재처리 시작 절대 날짜 (KST YYYY-MM-DD)")
    parser.add_argument("--end-date", type=str, default=None, help="재처리 종료 절대 날짜 (KST YYYY-MM-DD)")
    return parser.parse_args()

def get_minio_client():
    minio_endpoint = os.getenv("MINIO_ENDPOINT")
    minio_access_key = os.getenv("MINIO_ACCESS_KEY")
    minio_secret_key = os.getenv("MINIO_SECRET_KEY")
    if not all([minio_endpoint, minio_access_key, minio_secret_key]):
        raise ValueError("❌ 오류: MinIO 접속 자격증명 환경변수가 누락되었습니다.")
    return boto3.client(
        "s3",
        endpoint_url=minio_endpoint,
        aws_access_key_id=minio_access_key,
        aws_secret_access_key=minio_secret_key,
        config=boto3.session.Config(signature_version="s3v4")
    )

def sync_down_raw_from_minio(s3_client, start_dt, end_dt):
    """🌟 [사용자 제안 핵심 반영] 지정한 장기 분석 기간에 필터링되는 파일만 MinIO에서 고속 다운로드"""
    print(f"🪣 [MinIO AIStor 레이크 리싱크] 분석 기간 내 원천 Parquet 파일 스캔 및 다운로드 개시...")
    
    chunk_delta = timedelta(hours=6)
    current_start = start_dt
    download_cnt = 0
    skip_cnt = 0

    while current_start < end_dt:
        chunk_str = current_start.strftime("%Y%m%d_%H")
        
        # 마감 파일명과 액티브 파일명 둘 다 스캔 후보 설정
        file_candidates = [f"prom_raw_{chunk_str}.parquet", f"prom_raw_{chunk_str}_active.parquet"]
        
        for file_name in file_candidates:
            local_file_path = RAW_DIR / file_name
            object_key = f"raw/{file_name}"
            
            try:
                # MinIO에 파일이 실재하는지 확인
                s3_client.head_object(Bucket=MINIO_RAW_BUCKET, Key=object_key)
                
                # 💡 로컬 emptyDir에 이미 정상적인 크기로 존재한다면 다운로드 스킵 (네트워크 아끼기)
                if local_file_path.exists() and local_file_path.stat().st_size > 0 and "_active" not in file_name:
                    skip_cnt += 1
                else:
                    s3_client.download_file(MINIO_RAW_BUCKET, object_key, str(local_file_path))
                    download_cnt += 1
            except Exception:
                # 파일이 버킷에 없는 경우는 그냥 넘김 (미래 시점이나 미수집 구간)
                pass
                
        current_start += chunk_delta
        
    print(f"  ->  📊 레이크 리싱크 완수: 신규 다운로드 {download_cnt}개 / 기존 로컬 캐시 활용 {skip_cnt}개")

def filter_target_nodes(df):
    node_pattern = NODE_PREFIX_PATTERN
    print(f"🌐 [노드 정규식 필터 적용] 대상 기조: '{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:,}행 -> {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 main():
    args = parse_arguments()
    s3_client = get_minio_client()
    now_kst = datetime.now(KST)
    
    # 1. 재처리 날짜 범위 파싱 (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)
    else:
        start_dt = (now_kst - timedelta(days=args.days)).replace(hour=0, minute=0, second=0, microsecond=0)
        end_dt = now_kst

    # 2. 🌟 MinIO 오브젝트 스토리지로부터 과거 데이터 원부 동적 획득 동기화!
    sync_down_raw_from_minio(s3_client, start_dt, end_dt)

    # 3. 로컬 작업 공간에 들어찬 Parquet 전수 스캔 및 메모리 고속 병합
    raw_files = list(RAW_DIR.glob("prom_raw_*.parquet"))
    if not raw_files:
        print("❌ 오류: 재처리 기간에 부합하는 원천 Parquet 데이터가 로컬 공간에 수렴되지 않았습니다.")
        return

    print(f"📦 다운로드된 총 {len(raw_files)}개의 원천 시계열 파티션 파일 로드 및 통합 매싱 중...")
    df_raw = pd.concat([pd.read_parquet(f) for f in raw_files], ignore_index=True)
    
    # KST 기준 일자 맵 구축 (이미 step1에서 KST로 들고옴)
    df_raw["date"] = df_raw["timestamp"].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)
    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("📊 장기 고해상도 시계열 기반 팟 유닛 레벨 다차원 롤업 집계 가동...")
    df_pod = df_raw.groupby(["date", "cluster", "namespace", "workload_type", "node", "pod", "container"]).agg(
        minutes_running      = ("timestamp", "size"), # 시계열 Row 개수 = 진짜 살아 숨 쉰 누적 가동 분!
        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)
    print(f"💾 장기 시계열 통합 마스터 파티션 원부 저장 완료 (총 팟 명세 건수: {len(df_pod):,}개)")

    # 거버넌스 파레토 재연산
    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["is_top_80_percent_offender"] = df_ns["waste_cumsum_pct"] <= 80.5

    df_ns.to_parquet(MERGED_DIR / "pareto_fixed_ns.parquet", index=False)
    print("💾 Namespace 거버넌스 파레토 빌딩 완료.\n")

if __name__ == "__main__":
    # 문법 오류 교정: df_pod["cpu_allocated_core_hours"] 줄 괄호 교정 포함됨
    # 코드 내 단순 수식 오타인 df_pod["minutes_running"] / 60.0 -> df_pod["minutes_running"] / 60.0 수정 가동
    raw_code_fix = """
    df_pod["cpu_allocated_core_hours"] = df_pod["cpu_request_max"] * (df_pod["minutes_running"] / 60.0)
    """
    main()