26J30d2

"""
[신규 1단계 - KST 정합 및 데이터 레이크 실시간 아카이빙 통합판]
설명: Thanos에서 1m 해상도로 긁은 정각 청크를 파일 생성 즉시 사내 MinIO AIStor에 백업하여 
      타노스 장기 시계열 조회 타임아웃 문제를 차단합니다.
실행: python step1_prom_fetch.py --days 1 --step 1m
"""

import os
import argparse
import requests
import boto3
import pandas as pd
from datetime import datetime, timedelta, timezone
from pathlib import Path

# config 원부로부터 스토리지 및 인프라 명세 징수
from config import RAW_DIR, DEFAULT_THANOS_URL, MINIO_RAW_BUCKET, get_finops_promql_queries

KST = timezone(timedelta(hours=9))

def parse_arguments():
    parser = argparse.ArgumentParser(description="FinOps Thanos KST Native Data Collector & Archiver")
    parser.add_argument("--url", type=str, default=DEFAULT_THANOS_URL, help="Thanos Endpoint URL")
    parser.add_argument("--days", type=int, default=None, help="최근 며칠간의 데이터를 KST 정각 기준으로 수집할지 정의")
    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)")
    parser.add_argument("--step", type=str, default="1m", help="수집 해상도 주기 (1m)")
    parser.add_argument("--cluster", type=str, default=None, help="클러스터 필터 키워드")
    parser.add_argument("--namespace", type=str, default=None, help="네임스페이스 필터")
    parser.add_argument("--force", action="store_true", help="기존 파일 무시하고 전수 강제 수집")
    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]):
        return None
        
    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 build_promql_selectors(cluster, namespace):
    selectors = ['container!=""']
    if cluster: selectors.append(f'cluster="{cluster}"')
    if namespace:
        if any(char in namespace for char in ["|", ".*", "^", "$"]):
            selectors.append(f'namespace=~"{namespace}"')
        else:
            selectors.append(f'namespace="{namespace}"')
    return "{" + ", ".join(selectors) + "}"

def fetch_range_chunk(url, query, start_ts, end_ts, step):
    endpoint = f"{url}/api/v1/query_range"
    params = {"query": query, "start": start_ts, "end": end_ts, "step": step}
    try:
        response = requests.get(endpoint, params=params, timeout=120)
        if response.status_code != 200: return None
        res_json = response.json()
        if res_json.get("status") != "success": return None
        return res_json["data"]["result"]
    except Exception:
        return None

def process_matrix_to_df(matrix_result, metric_name):
    if not matrix_result: return pd.DataFrame()
    records = []
    for item in matrix_result:
        metric_labels = item.get("metric", {})
        cluster = metric_labels.get("cluster", "default")
        namespace = metric_labels.get("namespace", "unknown")
        pod = metric_labels.get("pod", "unknown")
        container = metric_labels.get("container", "unknown")
        node = metric_labels.get("node", metric_labels.get("instance", "unknown")).split(":")[0]
        
        for val_pair in item.get("values", []):
            utc_dt = pd.to_datetime(int(val_pair[0]), unit='s').tz_localize('UTC')
            kst_dt = utc_dt.tz_convert('Asia/Seoul').tz_localize(None)
            
            records.append({
                "timestamp": kst_dt,
                "cluster": cluster, "namespace": namespace, "pod": pod, "container": container, "node": node,
                metric_name: float(val_pair[1])
            })
    return pd.DataFrame(records)

def main():
    args = parse_arguments()
    s3_client = get_minio_client()
    now_kst = datetime.now(KST)
    
    if args.days:
        start_dt = (now_kst - timedelta(days=args.days)).replace(hour=0, minute=0, second=0, microsecond=0)
        end_dt = now_kst
    elif 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:
        print("❌ 오류: 수집 기간 인자가 지정되지 않았습니다.")
        return

    selector = build_promql_selectors(args.cluster, args.namespace)
    queries = get_finops_promql_queries(selector)

    print(f"🚀 FinOps KST 네이티브 수집 & 원천 아카이빙 기동 -> URL: {args.url}")
    print(f"📅 범위 (KST): {start_dt.strftime('%Y-%m-%d %H:%M:%S')} ~ {end_dt.strftime('%Y-%m-%d %H:%M:%S')}\n")

    chunk_delta = timedelta(hours=6)
    current_start = start_dt

    while current_start < end_dt:
        current_end = min(current_start + chunk_delta, end_dt)
        chunk_str = current_start.strftime("%Y%m%d_%H")
        
        is_completed_chunk = (current_start + chunk_delta) <= now_kst
        
        if is_completed_chunk:
            file_name = f"prom_raw_{chunk_str}.parquet"
        else:
            file_name = f"prom_raw_{chunk_str}_active.parquet"
            
        out_file = RAW_DIR / file_name
        
        # ── 🛡️ [마감 청크 MinIO 이중 체크 및 Skip] ──
        if not args.force and is_completed_chunk:
            # 로컬에 이미 파일이 있거나, MinIO에 이미 올라가 있다면 타노스 호출 생략
            if out_file.exists() and out_file.stat().st_size > 0:
                print(f"⏭️  [SKIP] 로컬 검증 통과: {file_name}")
                current_start = current_end
                continue
            if s3_client:
                try:
                    s3_client.head_object(Bucket=MINIO_RAW_BUCKET, Key=f"raw/{file_name}")
                    print(f"⏭️  [SKIP] MinIO 원천 데이터 레이크 내 존재 확인 완료: {file_name}")
                    current_start = current_end
                    continue
                except Exception:
                    pass

        start_ts = int(current_start.timestamp())
        end_ts = int(current_end.timestamp())
        
        print(f"⏳ [Thanos 징수] {current_start.strftime('%m-%d %H:%M')} ~ {current_end.strftime('%m-%d %H:%M')} KST")
        
        chunk_df_list = []
        for metric_name, query_expr in queries.items():
            raw_matrix = fetch_range_chunk(args.url, query_expr, start_ts, end_ts, args.step)
            df_metric = process_matrix_to_df(raw_matrix, metric_name)
            if not df_metric.empty:
                chunk_df_list.append(df_metric)
        
        if chunk_df_list:
            merged_chunk = chunk_df_list[0]
            for next_df in chunk_df_list[1:]:
                join_keys = ["timestamp", "cluster", "namespace", "pod", "container", "node"]
                merged_chunk = pd.merge(merged_chunk, next_df, on=join_keys, how="outer")
            
            # 1. 로컬 emptyDir 임시 저장
            merged_chunk.to_parquet(out_file, index=False)
            print(f"  -> 💾 로컬 캐싱 완료: {file_name} ({len(merged_chunk):,}행)")
            
            # 2. 🌟 [핵심 반영] 마감 완료된 원천 Parquet는 즉시 MinIO AIStor 데이터 레이크로 영구 아카이빙 영사!!
            if is_completed_chunk and s3_client:
                print(f"  -> 🪣 [MinIO AIStor 아카이빙] {file_name} -> 버킷 [{MINIO_RAW_BUCKET}] 동기화 중...")
                s3_client.upload_file(str(out_file), MINIO_RAW_BUCKET, f"raw/{file_name}")
                
                # 마감되었으므로 구형 임시 active 파일 청소
                active_file = RAW_DIR / f"prom_raw_{chunk_str}_active.parquet"
                if active_file.exists(): active_file.unlink()
        else:
            print("  -> ⚠️ 해당 구간 메트릭 없음.")

        current_start = current_end

    print("\n✅ KST 정각 정렬 및 MinIO 원천 데이터 레이크 아카이빙 프로세스 완료.")

if __name__ == "__main__":
    main()