26S14x

K8s 클러스터 장애 발생 시 알림 수신 $\rightarrow$ 관측 데이터(Prometheus/OpenSearch/K8s API) 수집 $\rightarrow$ 내부 런북(RAG) 매칭 $\rightarrow$ 원인 분석 및 초동 조치 가이드 리포팅으로 이어지는 준실시간 AI SRE 에이전트 시스템의 전체 아키텍처와 워크플로우 설계안입니다.

---

전체 시스템 아키텍처

[1. Event Trigger]
Alertmanager / Webhook / Slack Bot
         │
         ▼
[2. Agent Orchestrator (LangGraph / State Machine)]
         │
         ├───► [3. Context Collector (MCP / Tool Calling)]
         │        ├── Prometheus API (PromQL 시계열 메트릭, Throttling, Node Load)
         │        ├── OpenSearch API (로그 패턴, 에러 스택트레이스, 빈도 집계)
         │        └── Kubernetes API (Pod Events, OOMKilled, Describe, RestartCount)
         │
         ├───► [4. Runbook RAG Engine]
         │        ├── Vector DB (운영 런북, 장애 히스토리, 포스트모템 문서)
         │        └── Hybrid Search (키워드 + 시맨틱 임베딩)
         │
         ▼
[5. LLM Reasoning Core (vLLM / Enterprise LLM)]
         │  - 타임라인 재구성 (Metrics + Logs + Events 교차 검증)
         │  - 런북 조치 방안 매핑
         │  - 환각 방지: 메트릭 수치 및 로그 원문 인용 검증
         │
         ▼
[6. Notification & Action]
Slack / Opsgenie (요약 리포트, 원인 추정, 권장 명령어 제시)
  └─ [Human-in-the-loop] 엔지니어 승인 시 조치 스크립트 실행

---

단계별 핵심 구성 요소

1. 트리거 및 이벤트 수신 (Event Intake)

• 입력 소스: Prometheus Alertmanager Webhook, K8s Event Exporter, 혹은 엔지니어의 On-demand 호출(Slack 명령 /diagnose pod-name).
• 이벤트 파싱: 발생 시각($T_0$), 영향받는 네임스페이스, Pod/노드 레이블, 알림명(e.g., KubePodCrashLooping, NodeMemoryPressure)을 구조화된 컨텍스트 객체로 변환합니다.

2. 에이전트 오케스트레이션 (Workflow Controller)

• 단순 1회성 LLM 호출이 아닌, 상태 기반 워크플로우(LangGraph 또는 경량 Go/Python 워커)로 구성합니다.
• 무분별하게 모든 데이터를 한 번에 긁어오면 LLM 컨텍스트 윈도우가 터지거나 토큰 비용이 급증하므로, "초기 진단 $\rightarrow$ 추가 데이터 필요 여부 판단 $\rightarrow$ 정밀 타겟 수집" 2단계로 진행합니다.

3. 데이터 수집 툴킷 (Observability Collectors)
에이전트가 호출할 Tool 인터페이스를 표준화(Function Calling / MCP)합니다.

• Prometheus Tool:
• 타깃 워크로드의 $T_0 \pm 15$분 전후 CPU/메모리/네트워크 I/O 사용량 및 Throttling 메트릭 조회.
• 골든 시그널(Latency, Traffic, Errors, Saturation) 자동 쿼리 템플릿화.

• OpenSearch Tool:
• 해당 시간대 타깃 Pod의 log.level: (ERROR OR WARN OR FATAL) 필터링.
• 동일 에러 메시지 발생 빈도(Top Aggregations) 및 최초 발생 시점 추출.

• K8s API Tool:
• kubectl get events -n <ns> --field-selector involvedObject.name=<pod> 동등 데이터 수집.
• Pod의 lastState.terminated 이유(e.g., Exit Code 137 OOM 등) 확인.

4. 런북 및 지식 베이스 검색 (RAG Engine)

• 데이터셋: 사내 Confluence/Wiki의 서비스별 트러블슈팅 가이드, 인프라 운영 매뉴얼, 과거 장애 리포트(Post-mortem).
• 검색 파이프라인:
• 에러 키워드(e.g., OutOfMemory, Heartbeat Lost, Connection refused)와 서비스명으로 1차 하이브리드 검색(Sparse + Dense).
• 가장 연관성 높은 2~3개 런북 섹션만 프롬프트에 주입(Context Injection).

5. 추론 및 리포팅 (Analysis & Synthesis)

• 교차 검증: "Prometheus에서 14:00에 메모리가 100%에 도달했고, OpenSearch에 14:01 OOM 로그가 기록되었으며, K8s Event에 14:02 Killing pod가 확인됨"과 같이 시간순 인과관계를 조립합니다.
• 출력 포맷: 엔지니어가 3초 안에 파악할 수 있는 스캔 가능한 리포트 생성.

---

출력 결과물 예시 (Slack/메신저 리포트)

🚨 [장애 진단 리포트] prod-starrocks-be-2 비정상 재시작
1. 인과관계 요약

• 14:02:10 OpenSearch: 대용량 쿼리 실행으로 인한 Direct Memory 할당 스파이크 확인.
• 14:02:35 Prometheus: 노드 메모리 사용률 98% 도달 및 PSI(Pressure Stall Information) 급증.
• 14:02:40 K8s Event: cgroup OOMKilled 발생 (ExitCode: 137).

2. 사내 런북 매칭 (DOC-OLAP-042: BE OOM 대응 가이드)

• 해당 파드는 쿼리 멤리밋(query_mem_limit)이 기본값(80%)으로 설정되어 있어 스파이크에 취약함.

3. 권장 초동 조치

• 파드 메모리 cgroup 확장 또는 임시 쿼리 캔슬:

# 비정상 쿼리 세션 확인
mysql -h <fe-ip> -P 9030 -e "SHOW PROCESSLIST;"

---

구축 로드맵 (단계별 추천)

• Phase 1 (정적 데이터 취합 파이프라인 구축):
• LLM 없이 Prometheus, OpenSearch, K8s Event를 Alert 발생 시점에 자동으로 모아서 하나의 JSON/Markdown으로 묶어주는 수집기(Collector) 먼저 개발.

• Phase 2 (RAG 지식 베이스 연동):
• 사내 Markdown/Wiki 문서를 파싱하여 Vector DB(OpenSearch의 k-NN 인덱스 재활용 가능)에 적재하고 검색 정확도 검증.

• Phase 3 (Agentic 워크플로우 결합):
• LangGraph 등을 활용하여 수집된 데이터와 런북을 기반으로 원인 분석 및 보고서를 작성하는 LLM 오케스트레이터 배포.

• Phase 4 (안전한 자동화 조치 - Human-in-the-loop):

• 슬랙 인터랙티브 버튼(Button: "파드 재기동 승인", "Quota 임시 증설")을 통해 엔지니어의 승인 하에 실행되는 2단계 자동화 구현.

  ==

  MVP 단계에서 기존 모니터링/인프라망과 에이전트 시스템을 직접 API 연동하지 않고(망 분리/격리 환경), 엔지니어가 수동으로 추출한 덤프 데이터를 주입해 분석하는 구조로 전환할 때 변경·준비해야 할 사항들입니다.

---

1. 데이터 추출 표준화 (단일 덤프 스크립트 작성)

엔지니어가 장애 발생 시 여러 UI(Grafana, OpenSearch Dashboards, Lens/k9s)를 일일이 캡처하거나 복사하는 수고를 없애기 위해, 로컬/배스천(Bastion)에서 한 번에 텍스트 파일로 뽑아내는 1회성 덤프 스크립트가 필요합니다.

• K8s 리소스 추출 (dump_k8s.sh):
• 대상 Pod의 kubectl describe pod <name> -n <ns> (Events, Last State, OOM 여부 포함).
• 동일 네임스페이스의 최근 Event 목록 (kubectl get events --sort-by=.metadata.creationTimestamp).

• OpenSearch 로그 추출 (curl 스크립트):
• 발생 시각($T_0$) 기준 $\pm 10$분 동안의 에러 로그만 뽑아 NDJSON 또는 텍스트로 저장:

curl -X POST "$OPENSEARCH_URL/$INDEX/_search" -H 'Content-Type: application/json' -d '{
  "query": {
    "bool": {
      "must": [
        { "range": { "@timestamp": { "gte": "now-15m", "lte": "now" } } },
        { "match": { "log.level": "ERROR" } },
        { "match": { "k8s.pod.name": "'"$TARGET_POD"'" } }
      ]
    }
  },
  "_source": ["@timestamp", "message", "log.level"],
  "size": 100
}' > logs.json

• Prometheus 메트릭 요약 추출:
• 시계열 그래프 전체를 전달하면 토큰이 낭비되므로, 핵심 수치(CPU Throttling rate, Memory 사용량, PSI 수치)의 쿼리 결과(Vector/Matrix) JSON을 스크립트로 다운로드.

---

2. 에이전트 시스템 입력 인터페이스 변경

실시간 Webhook 수신부 대신, 수동 파일 업로드나 디렉토리 감시 기반으로 입력 구조를 단순화합니다.

• 인터페이스 구현 방식:
• Option A (가장 단순): 특정 인풋 디렉토리(/workspace/incident_inputs/)에 파일(k8s.txt, logs.json, metrics.json)을 넣고 CLI 명령 실행 (python analyze.py --incident-dir ./incident_inputs).
• Option B (GUI): Streamlit, Gradio 또는 Open-WebUI를 활용해 웹 화면에서 텍스트 붙여넣기 및 파일 드래그 앤 드롭 지원.

• 입력 번들링 (Bundle Schema):
• 에이전트가 어떤 맥락인지 바로 인지할 수 있도록 메타데이터(장애 발생 추정 시각, 증상 요약, 시스템명)를 담는 간단한 meta.yaml 정의.

---

3. 컨텍스트 정제 및 토큰 축약기(Preprocessor) 추가

API를 직접 호출할 때는 필요한 항목만 필터링해서 가져올 수 있었지만, 수동 덤프 데이터는 불필요한 노이즈와 거대한 로그 덤프를 포함하기 쉽습니다. 이를 LLM에 그대로 넣으면 컨텍스트 윈도우 초과 및 환각이 발생합니다.

• 로그 중복 제거 및 클러스터링:
• 반복되는 Java/C++ 스택트레이스나 동일 에러 로그는 해시 기반으로 묶어 [동일 에러 45회 반복: Connection timeout ...] 형태로 요약.

• K8s Spec 불필요 필드 제거:
• kubectl describe나 YAML에서 metadata.managedFields, status.conditions 중 비정상 상태가 아닌 정상 조건들은 정규식/스크립트로 파싱하여 날리고, Events와 Last State Terminated Reason만 추출.

---

4. 런북(RAG) 지식 베이스의 로컬화

외부 Confluence/Wiki API를 실시간 조회할 수 없으므로, 운영 문서를 로컬 파일 기반으로 사전 빌드해야 합니다.

• 사전 인덱싱 (Offline Indexing):
• 대상 서비스(e.g., StarRocks, PostgreSQL, 노드 운영)와 관련된 기존 문제 해결 가이드, 포스트모템 Markdown 파일들을 사전에 로컬 디렉토리에 수집.

• 경량 로컬 Vector DB 구성:
• 네트워크 통신이 필요 없는 ChromaDB, FAISS, 또는 Qdrant Local Mode를 사용하여 운영 문서 텍스트 청크(Chunk)와 임베딩을 로컬 디스크에 영구 저장(Persistence).

---

5. 에이전트 오케스트레이션 로직 간소화

툴 호출(Tool Calling/MCP)을 통해 외부 시스템을 스스로 찌르는 복잡한 다단계 에이전트 루프가 필요 없어집니다.

• 단순화된 파이프라인:

1. Load: 업로드된 k8s.txt, logs.json, metrics.json 로드 및 구조화.
2. Retrieve: 로그/이벤트에서 추출한 핵심 키워드(OOMKilled, Heartbeat Lost 등)로 로컬 Vector DB에서 관련 런북 1~2개 검색.
3. Synthesize: 프롬프트에 [수집 데이터 요약] + [관련 런북 내용]을 함께 주입하고, 장애 타임라인 재구성 및 조치 가이드 생성 프롬프트 실행.

• 프레임워크 부담 완화:

• LangGraph의 복잡한 동적 브랜칭 없이도, 단일 체인(LangChain LCEL 또는 순수 Python 스크립트)만으로 MVP 구현이 가능해집니다.

  ==

  대량의 모니터링 원천 데이터를 경량 LLM(SLM)이 탑재된 GPU Node에서 1차 압축·요약(Distillation/Preprocessing)한 뒤, 정제된 최종 결과 파일만 Agent 시스템으로 넘기는 구조입니다.

---

1. GPU Node 사전 준비 사항

GPU 노드에서는 긴 컨텍스트의 로그와 JSON 메트릭을 고속으로 읽고 구조화된 텍스트로 요약하는 전처리 전용 로컬 서빙 파이프라인을 갖춥니다.

• LLM 서빙 엔진: vLLM (높은 처리량, 배치 처리, PagedAttention 지원)
• 추천 모델: 긴 컨텍스트 윈도우(32k~128k)와 JSON 파싱 능력이 우수한 가벼운 모델
• Qwen2.5-7B-Instruct (또는 14B), Llama-3.1-8B-Instruct

• vLLM 기동 예시:

python3 -m vllm.entrypoints.openai.api_server \
  --model Qwen/Qwen2.5-7B-Instruct \
  --port 8000 \
  --max-model-len 32768 \
  --gpu-memory-utilization 0.85

• 필수 패키지: jq, curl, python3-pip, pip install openai pandas

---

2. 자동화 파이프라인 구성

작업 흐름은 [원본 데이터 수집] $\rightarrow$ [룰 기반 1차 컷팅] $\rightarrow$ [GPU Node 로컬 LLM 2차 압축] $\rightarrow$ [정제된 파일 번들 생성] 순으로 진행됩니다.

[운영/배스천 환경]
  dump_raw.sh (Prometheus, OpenSearch, K8s CLI)
       │
       ▼ (대용량 JSON/로그 전송)
[GPU Node]
  1. rule_based_trim.py (중복 로그 집계, spec 필드 제거)
  2. llm_summarizer.py  (vLLM 로컬 호출: 타임라인 & 이상치 요약)
       │
       ▼ (수 KB 크기의 요약본 생성)
  output: incident_summary_<timestamp>.json

---

3. 자동화 스크립트

① 데이터 수집 스크립트 (collect_raw.sh)

배스천 또는 운영 노드에서 실행하여 타깃 Pod의 데이터를 신속히 추출합니다.

#!/usr/bin/env bash
set -euo pipefail

TARGET_POD=${1:-"prod-starrocks-be-0"}
NAMESPACE=${2:-"default"}
OUTPUT_DIR="./raw_dump_${TARGET_POD}_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "[1/3] K8s Metadata & Events 수집..."
kubectl describe pod "$TARGET_POD" -n "$NAMESPACE" > "$OUTPUT_DIR/k8s_describe.txt"
kubectl get events -n "$NAMESPACE" --sort-by='.lastTimestamp' | tail -n 100 > "$OUTPUT_DIR/k8s_events.txt"

echo "[2/3] OpenSearch 최근 15분 에러 로그 덤프..."
# 환경에 맞게 OPENSEARCH_HOST 수정
OPENSEARCH_HOST=${OPENSEARCH_HOST:-"http://localhost:9200"}
curl -s -X POST "$OPENSEARCH_HOST/*/_search" \
  -H 'Content-Type: application/json' \
  -d '{
    "size": 500,
    "query": {
      "bool": {
        "must": [
          { "range": { "@timestamp": { "gte": "now-15m" } } },
          { "match": { "kubernetes.pod_name": "'"$TARGET_POD"'" } }
        ]
      }
    },
    "_source": ["@timestamp", "log.level", "message"]
  }' > "$OUTPUT_DIR/raw_logs.json"

echo "[3/3] Prometheus 최근 메트릭 요약 덤프..."
PROM_HOST=${PROM_HOST:-"http://localhost:9090"}
# CPU Throttling 및 메모리 사용량 쿼리
curl -s -G --data-urlencode "query=rate(container_cpu_cfs_throttled_periods_total{pod=\"$TARGET_POD\"}[5m])" \
  "$PROM_HOST/api/v1/query" > "$OUTPUT_DIR/metric_throttling.json"
curl -s -G --data-urlencode "query=container_memory_working_set_bytes{pod=\"$TARGET_POD\"}" \
  "$PROM_HOST/api/v1/query" > "$OUTPUT_DIR/metric_memory.json"

echo "수집 완료: $OUTPUT_DIR"

② GPU 노드 요약 및 압축 엔진 (gpu_compressor.py)

GPU 노드에서 실행되며, 룰 기반 전처리와 vLLM을 순차 실행해 최종 경량 리포트만 산출합니다.

#!/usr/bin/env python3
import json
import os
import sys
import glob
from openai import OpenAI

# 1. vLLM 로컬 엔드포인트 설정
client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="EMPTY"
)

def rule_based_clean(dump_dir):
    """불필요한 노이즈 1차 제거 및 중복 로그 그룹화"""
    cleaned_context = []

    # K8s Events 정제
    event_path = os.path.join(dump_dir, "k8s_events.txt")
    if os.path.exists(event_path):
        with open(event_path, "r") as f:
            lines = [line.strip() for line in f.readlines() if any(k in line for k in ["Warning", "Failed", "Killing", "OOM"])]
            cleaned_context.append("### K8s Warning/Error Events:\n" + "\n".join(lines[-20:]))

    # K8s Describe 내 핵심 상태 추출
    desc_path = os.path.join(dump_dir, "k8s_describe.txt")
    if os.path.exists(desc_path):
        with open(desc_path, "r") as f:
            content = f.read()
            # Last State 및 Exit Code 주변 발췌
            target_keywords = ["Last State:", "Exit Code:", "Reason:", "Restart Count:"]
            hits = [line for line in content.splitlines() if any(k in line for k in target_keywords)]
            cleaned_context.append("### Pod State History:\n" + "\n".join(hits))

    # OpenSearch 로그 중복 카운팅
    log_path = os.path.join(dump_dir, "raw_logs.json")
    if os.path.exists(log_path):
        try:
            with open(log_path, "r") as f:
                data = json.load(f)
                hits = data.get("hits", {}).get("hits", [])
                log_map = {}
                for h in hits:
                    msg = h.get("_source", {}).get("message", "")[:200] # 앞부분 200자 기준 그룹핑
                    lvl = h.get("_source", {}).get("log.level", "INFO")
                    ts = h.get("_source", {}).get("@timestamp", "")
                    key = f"[{lvl}] {msg}"
                    if key not in log_map:
                        log_map[key] = {"count": 1, "last_seen": ts}
                    else:
                        log_map[key]["count"] += 1
                
                # 상위 중복 로그 정리
                log_summary = [f"{v['count']}회 반복 ({v['last_seen']}): {k}" for k, v in sorted(log_map.items(), key=lambda x: x[1]['count'], reverse=True)[:15]]
                cleaned_context.append("### Error Log Patterns:\n" + "\n".join(log_summary))
        except Exception as e:
            cleaned_context.append(f"Log parsing error: {e}")

    return "\n\n".join(cleaned_context)

def summarize_with_gpu(text_context):
    """GPU 로컬 vLLM을 활용해 2차 핵심 압축"""
    prompt = f"""당신은 SRE 데이터 전처리 엔진입니다. 
아래 대규모 인프라 덤프 데이터(K8s Event, State, Error Logs)를 분석하여 메인 AI Agent가 읽기 편하도록 핵심만 요약하세요.

[요구사항]
1. 발생한 장애 증상(OOM, Panic, Throttling 등) 단일 명시
2. 타임라인 순서로 주요 사건 3~5줄 나열
3. 에러를 유발한 원인 메시지/스택트레이스 원문 최대 2줄 발췌
4. 불필요한 서술어를 빼고 건조한 사실(Facts) 위주로 정리

---
{text_context}
---
"""
    response = client.chat.completions.create(
        model="Qwen/Qwen2.5-7B-Instruct",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
        max_tokens=1000
    )
    return response.choices[0].message.content

def main():
    if len(sys.argv) < 2:
        print("Usage: python3 gpu_compressor.py <raw_dump_dir>")
        sys.exit(1)

    dump_dir = sys.argv[1]
    print("[1] 룰 기반 1차 컷팅 진행 중...")
    clean_data = rule_based_clean(dump_dir)

    print("[2] GPU(vLLM) 기반 2차 증류 요약 중...")
    summary = summarize_with_gpu(clean_data)

    out_file = os.path.join(dump_dir, "distilled_incident_context.md")
    with open(out_file, "w") as f:
        f.write(summary)

    print(f"\n최종 압축 완료! Agent 전달용 파일 생성됨: {out_file}")
    print("--------------------------------------------------")
    print(summary)

if __name__ == "__main__":
    main()

---

4. 생성되는 최종 전달 파일 예시 (distilled_incident_context.md)

수십 MB 크기의 로그와 복잡한 JSON 대신, Agent 시스템에는 아래와 같이 약 1~2KB 크기의 정제된 마크다운 파일만 전달됩니다.

### 1. 장애 증상
- Pod 상태: CrashLoopBackOff (Exit Code: 137, OOMKilled)
- 영향 대상: prod-starrocks-be-0 (Node: k8s-worker-gpu-04)

### 2. 주요 타임라인
- 14:05:12 | OpenSearch: `QueryEngine: Memory allocation failed for HashJoinNode` (총 42회 반복)
- 14:05:40 | Prometheus: cgroup memory working set이 limit(64Gi)의 99.4% 도달
- 14:05:45 | K8s: System OOM Killer 발동, `Killing container starrocks-be` 이벤트 감지
- 14:06:01 | K8s: Pod 재기동 시도 실패 (BackOff)

### 3. 핵심 에러 발췌
- `[ERROR] Memory limit exceeded: query memory limit=53687091200, current usage=53687093120, process memory limit=68719476736`

이 파일을 Agent 시스템으로 넘기면, 토큰 소모를 95% 이상 절감하면서 RAG(런북) 매칭과 최종 조치 분석을 즉각 수행할 수 있습니다.

==

GPU 노드에서 생성된 정제 파일(distilled_incident_context.md)을 입력받아, 사내 런북(RAG)을 검색하고 LLM을 통해 최종 장애 분석 리포트를 도출하는 2단계 에이전트 스크립트입니다.

외부 네트워크나 별도 인프라 연동 없이, 로컬에 저장된 런북 마크다운 문서들을 로컬 Vector DB(ChromaDB)에 사전 임베딩해 두고 추론을 수행합니다.

---

1. 사전 패키지 설치

pip install chromadb sentence-transformers openai

---

2. 런북 로컬 인덱싱 스크립트 (build_runbook_db.py)

사내 트러블슈팅 가이드나 런북 Markdown 파일들(docs/*.md)을 읽어 로컬 ChromaDB에 벡터 인덱스를 구축합니다. (1회 실행)

#!/usr/bin/env python3
import os
import glob
import chromadb
from chromadb.utils import embedding_functions

# 1. 로컬 Chroma 클라이언트 설정 (파일 영구 저장)
CHROMA_DATA_PATH = "./chroma_runbooks_db"
client = chromadb.PersistentClient(path=CHROMA_DATA_PATH)

# 가볍고 다국어/한국어 성능이 준수한 경량 로컬 임베딩 모델
emb_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

collection = client.get_or_create_collection(
    name="sre_runbooks",
    embedding_function=emb_fn
)

def chunk_markdown(content, chunk_size=800, overlap=100):
    chunks = []
    start = 0
    while start < len(content):
        end = start + chunk_size
        chunks.append(content[start:end])
        start += chunk_size - overlap
    return chunks

def index_runbooks(docs_dir="./runbooks"):
    md_files = glob.glob(os.path.join(docs_dir, "**/*.md"), recursive=True)
    if not md_files:
        print(f"문서 디렉토리({docs_dir})에 .md 파일이 없습니다.")
        return

    doc_ids, doc_texts, doc_metas = [], [], []
    idx = 0

    for file_path in md_files:
        with open(file_path, "r", encoding="utf-8") as f:
            text = f.read()
            chunks = chunk_markdown(text)
            for chunk_idx, chunk in enumerate(chunks):
                doc_ids.append(f"doc_{idx}_{chunk_idx}")
                doc_texts.append(chunk)
                doc_metas.append({"source": os.path.basename(file_path), "chunk": chunk_idx})
            idx += 1

    # 기존 데이터 삭제 후 갱신 적재
    collection.add(
        documents=doc_texts,
        metadatas=doc_metas,
        ids=doc_ids
    )
    print(f"총 {len(doc_texts)}개 런북 청크 인덱싱 완료 (저장 경로: {CHROMA_DATA_PATH})")

if __name__ == "__main__":
    index_runbooks()

---

3. 최종 진단 에이전트 스크립트 (analyze_incident.py)

GPU 노드가 넘겨준 distilled_incident_context.md를 로드하고, 관련 런북을 검색(RAG)한 뒤 최종 조치 가이드를 생성합니다.

#!/usr/bin/env python3
import sys
import os
import chromadb
from chromadb.utils import embedding_functions
from openai import OpenAI

# 1. 로컬 vLLM 엔드포인트 설정 (또는 엔터프라이즈 API)
llm_client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="EMPTY"
)

# 2. 로컬 ChromaDB 연결
CHROMA_DATA_PATH = "./chroma_runbooks_db"
chroma_client = chromadb.PersistentClient(path=CHROMA_DATA_PATH)
emb_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)
collection = chroma_client.get_collection(
    name="sre_runbooks",
    embedding_function=emb_fn
)

def search_relevant_runbooks(query_text, top_k=2):
    """장애 컨텍스트 기반 관련 런북 검색"""
    results = collection.query(
        query_texts=[query_text],
        n_results=top_k
    )
    
    retrieved_contexts = []
    if results and "documents" in results and results["documents"]:
        for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
            retrieved_contexts.append(f"--- [참고 문서: {meta['source']}] ---\n{doc}")
    
    return "\n\n".join(retrieved_contexts)

def run_rca_analysis(incident_context, runbook_context):
    """LLM 인과관계 분석 및 초동 조치 가이드 생성"""
    system_prompt = """당신은 대규모 Kubernetes 클러스터 및 데이터 플랫폼 인프라를 운영하는 수석 SRE 진단 에이전트입니다.
주어진 [장애 요약 데이터]와 사내 [운영 런북 문서]를 교차 검증하여 인과관계를 밝히고, 엔지니어가 즉시 수행할 수 있는 구체적인 조치 절차를 제시하세요.

답변은 간결하고 실질적인 명령어나 설정값 위주로 작성하며, 사실에 기반하지 않은 추측은 배제하세요."""

    user_prompt = f"""[장애 요약 데이터]
{incident_context}

[관련 사내 런북 가이드]
{runbook_context}

[요청 사항]
아래 포맷에 맞춰 진단 리포트를 작성해줘:
1. **장애 근본 원인(RCA) 분석**: 메트릭, 로그, 상태 이력을 결합한 인과관계 설명
2. **사내 런북 매칭 결과**: 매칭된 가이드 명칭 및 적용 타당성
3. **즉시 실행 권장 조치**: Pod/Node 트러블슈팅 또는 쿼리/워크로드 튜닝 명령어 블록
4. **재발 방지 대책**: cgroup/리소스 설정, 쿼리 제한 등 영구 조치 사항
"""

    response = llm_client.chat.completions.create(
        model="Qwen/Qwen2.5-7B-Instruct",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.2,
        max_tokens=1500
    )
    return response.choices[0].message.content

def main():
    if len(sys.argv) < 2:
        print("Usage: python3 analyze_incident.py <distilled_incident_file.md>")
        sys.exit(1)

    incident_file = sys.argv[1]
    if not os.path.exists(incident_file):
        print(f"파일을 찾을 수 없습니다: {incident_file}")
        sys.exit(1)

    with open(incident_file, "r", encoding="utf-8") as f:
        incident_context = f.read()

    print("[1] 관련 사내 런북 검색 중 (RAG)...")
    runbook_context = search_relevant_runbooks(incident_context, top_k=2)

    print("[2] 원인 분석 및 조치 가이드 생성 중...")
    final_report = run_rca_analysis(incident_context, runbook_context)

    # 결과 저장
    output_report_path = incident_file.replace(".md", "_final_report.md")
    with open(output_report_path, "w", encoding="utf-8") as f:
        f.write(final_report)

    print("\n================ [최종 장애 진단 리포트] ================\n")
    print(final_report)
    print(f"\n리포트 파일 저장 완료: {output_report_path}")

if __name__ == "__main__":
    main()

---

4. 실행 방법

1. 런북 문서 배치 및 색인:
   ./runbooks/ 디렉토리에 기존 장애 대응 매뉴얼(예: starrocks_oom_guide.md, k8s_node_pressure.md 등)을 넣고 빌드합니다.

python3 build_runbook_db.py

2. 진단 수행:
   GPU 노드에서 전달받은 요약 파일을 지정하여 실행합니다.

python3 analyze_incident.py ./raw_dump_xxx/distilled_incident_context.md

==

3단계: Agentic 워크플로우 결합(LangGraph/상태 기반 오케스트레이터)은 앞서 작성한 단순 단방향 스크립트(수집 $\rightarrow$ 검색 $\rightarrow$ 생성)의 한계를 넘어, 에이전트가 데이터의 부족 여부를 스스로 판단하여 추가 질의를 하거나 다각도로 가설을 검증(Reflection/Re-Act)하는 지능형 진단 루프를 구축하는 단계입니다.

단순 체인과 달리 장애 원인이 불명확할 때(예: "OOM인데 쿼리 문제인지 OS 캐시 경합인지 불확실") 추가 문서를 능동적으로 재검색하거나 분석 깊이를 조절할 수 있습니다.

---

1. 상태 기반 에이전트(State Graph) 아키텍처

LangGraph를 사용해 순환 제어(Looping & Branching)가 가능한 상태 머신을 구성합니다.

       [Start]
          │
          ▼
   [Load Incident] ──── 장애 컨텍스트 수신
          │
          ▼
  [Hypothesis Node] ─── 초기 가설 수립 (예: Query Memory Leak? Cgroup Limit Exceeded?)
          │
          ▼
   [Retrieve Node] ──── 가설 기반 관련 런북/이력 타겟 검색
          │
          ▼
   [Evaluate Node] ─── 런북과 메트릭/로그 간 근거 충분성 검증 (Self-Reflection)
          │
    ┌─────┴─────────────────────┐
 (불충분 / 다른 가설 필요)   (충분함 / 인과관계 성립)
    │                           │
    ▼                           ▼
[Refine Query]            [Generate RCA] ─── 최종 리포트 및 액션 플랜 작성
    │                           │
    └───────────────────────────► [End]

---

2. 환경 구성 및 패키지 준비

pip install langgraph langchain langchain-community chromadb sentence-transformers openai

---

3. LangGraph 기반 장애 진단 에이전트 구현 (agentic_rca.py)

① 상태(State) 정의 및 의존성 주입

#!/usr/bin/env python3
import os
import sys
from typing import TypedDict, List, Annotated
import chromadb
from chromadb.utils import embedding_functions
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

# 상태 정의: 그래프 노드 간 공유되는 컨텍스트
class IncidentState(TypedDict):
    incident_text: str          # 1단계에서 압축된 장애 컨텍스트
    hypotheses: List[str]       # 생성된 장애 원인 가설 목록
    search_queries: List[str]   # 런북 검색용 타겟 쿼리
    runbook_matches: List[str]  # 검색된 런북 스니펫
    evaluation: str             # 가설 적합성 평가 ("SUFFICIENT" or "RETRY")
    iteration_count: int        # 무한 루프 방지용 카운터
    final_report: str           # 최종 도출 보고서

# 로컬 vLLM 엔드포인트 연결 (OpenAI 호환)
llm = ChatOpenAI(
    base_url="http://localhost:8000/v1",
    api_key="EMPTY",
    model="Qwen/Qwen2.5-7B-Instruct",
    temperature=0.1
)

# ChromaDB 연결
chroma_client = chromadb.PersistentClient(path="./chroma_runbooks_db")
emb_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)
collection = chroma_client.get_collection(name="sre_runbooks", embedding_function=emb_fn)

② 핵심 노드 함수 구현

def formulate_hypothesis(state: IncidentState) -> IncidentState:
    """[노드 1] 증상을 보고 1~2개의 잠재적 원인 가설 및 검색 키워드 추출"""
    prompt = f"""다음 K8s 장애 컨텍스트를 분석하여 가장 가능성 높은 장애 원인 가설 1~2개와, 
사내 런북에서 찾아야 할 핵심 검색 키워드를 JSON이 아닌 쉼표 구분으로 제시하세요.

[장애 내용]
{state['incident_text']}

[출력 형식]
가설: <가설 내용>
검색어: <키워드1, 키워드2>
"""
    res = llm.invoke(prompt).content
    
    # 간단 파싱
    lines = res.strip().split("\n")
    queries = []
    for line in lines:
        if line.startswith("검색어:"):
            queries = [q.strip() for q in line.replace("검색어:", "").split(",") if q.strip()]

    return {
        "hypotheses": [res],
        "search_queries": queries if queries else ["OOMKilled", "CrashLoopBackOff"],
        "iteration_count": state.get("iteration_count", 0) + 1
    }

def retrieve_runbooks(state: IncidentState) -> IncidentState:
    """[노드 2] 생성된 타깃 쿼리로 런북 RAG 검색"""
    retrieved = list(state.get("runbook_matches", []))
    for q in state["search_queries"]:
        results = collection.query(query_texts=[q], n_results=1)
        if results["documents"] and results["documents"][0]:
            doc_text = results["documents"][0][0]
            meta = results["metadatas"][0][0]["source"]
            retrieved.append(f"[{meta}]\n{doc_text}")
            
    return {"runbook_matches": list(set(retrieved))}

def evaluate_evidence(state: IncidentState) -> IncidentState:
    """[노드 3] 검색된 런북이 장애 증상을 충분히 설명하는지 자체 검증 (Reflection)"""
    # 2회 이상 시도했으면 무조건 통과 (Safe Exit)
    if state["iteration_count"] >= 2:
        return {"evaluation": "SUFFICIENT"}

    prompt = f"""수석 SRE 관점에서 평가하세요.
현재 검색된 사내 런북이 주어진 장애를 명확히 진단하고 조치하기에 충분한 정보인지 판별하세요.

[장애 데이터]
{state['incident_text']}

[검색된 런북]
{chr(10).join(state['runbook_matches'])}

답변은 반드시 다음 중 하나로만 시작하세요:
- 판정: SUFFICIENT (충분함)
- 판정: RETRY (정보 부족, 다른 각도의 키워드 필요)
"""
    res = llm.invoke(prompt).content
    verdict = "SUFFICIENT" if "SUFFICIENT" in res else "RETRY"
    return {"evaluation": verdict}

def generate_rca_report(state: IncidentState) -> IncidentState:
    """[노드 4] 최종 RCA 및 조치 가이드 합성"""
    prompt = f"""당신은 대규모 K8s 플랫폼 수석 SRE입니다.
수집된 장애 데이터와 검증된 런북을 교차 분석하여 정밀 진단 보고서를 작성하세요.

[장애 데이터]
{state['incident_text']}

[참고 런북]
{chr(10).join(state['runbook_matches'])}

[작성 가이드]
1. 원인 분석: 메트릭 수치와 로그 이벤트를 결합한 명확한 타임라인 설명
2. 런북 기반 즉각 조치 명령어 (Bash/SQL 블록)
3. 영구 재발 방지 권고안 (cgroup/operator 설정 등)
"""
    report = llm.invoke(prompt).content
    return {"final_report": report}

③ 그래프 조립 및 실행

def route_evaluation(state: IncidentState):
    """조건부 분기: 근거가 충분하면 보고서 작성, 부족하면 키워드 재정의"""
    if state["evaluation"] == "SUFFICIENT":
        return "generate_rca"
    return "formulate_hypothesis"

# 그래프 빌드
builder = StateGraph(IncidentState)

builder.add_node("formulate_hypothesis", formulate_hypothesis)
builder.add_node("retrieve_runbooks", retrieve_runbooks)
builder.add_node("evaluate_evidence", evaluate_evidence)
builder.add_node("generate_rca", generate_rca_report)

builder.set_entry_point("formulate_hypothesis")
builder.add_edge("formulate_hypothesis", "retrieve_runbooks")
builder.add_edge("retrieve_runbooks", "evaluate_evidence")
builder.add_conditional_edges(
    "evaluate_evidence",
    route_evaluation,
    {
        "generate_rca": "generate_rca",
        "formulate_hypothesis": "formulate_hypothesis"
    }
)
builder.add_edge("generate_rca", END)

app = builder.compile()

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python3 agentic_rca.py <distilled_incident.md>")
        sys.exit(1)

    with open(sys.argv[1], "r", encoding="utf-8") as f:
        content = f.read()

    initial_state: IncidentState = {
        "incident_text": content,
        "hypotheses": [],
        "search_queries": [],
        "runbook_matches": [],
        "evaluation": "",
        "iteration_count": 0,
        "final_report": ""
    }

    print(">>> Agentic RCA 진단 루프 시작...")
    final_output = app.invoke(initial_state)

    print("\n================ [최종 Agentic 진단 리포트] ================\n")
    print(final_output["final_report"])

---

4. 2단계 대비 3단계 도입 시 얻는 핵심 차이점

• 자가 교정(Self-Correction):
  초기 키워드 검색(ExitCode 137)으로 엉뚱한 노드 리소스 고갈 문서가 걸렸을 때, 평가 노드(evaluate_evidence)가 이를 감지하고 StarRocks HashJoin query memory 같은 세부 키워드로 재검색하여 적중률을 끌어올립니다.
• 환각 방지(Hallucination Guard):
  근거 런북이 불충분할 경우 억지로 조치 명령어를 지어내지 않고, 재검색 한도 초과 시 "일치하는 런북 없음: 기본 Pod 진단 절차 적용"으로 안전하게 Fallback 됩니다.

==

4단계: Human-in-the-loop 기반 안전한 자동화 조치는 AI 에이전트가 도출한 진단 결과를 바탕으로, 엔지니어가 메신저(Slack/Teams 등)에서 버튼 클릭 한 번으로 검증된 복구 작업을 안전하게 승인·실행하는 단계입니다.

대규모 프로덕션 클러스터에서 LLM에게 임의 쓰기 권한(kubectl delete/patch 등)을 직접 부여하면 환각(Hallucination)으로 인한 대형 사고가 발생할 수 있습니다. 따라서 "AI는 사전 정의된 파라미터화된 액션만 제안"하고, "실행 권한과 최종 결정은 온콜 엔지니어"가 갖는 아키텍처가 필수적입니다.

---

1. 전체 아키텍처 및 통제 흐름

[3단계 Agentic RCA 완료]
         │
         ▼
[Action Dispatcher] ── Action 매핑 (사전문서화된 안전 스크립트만 선택)
         │
         ▼
[Slack Incoming Webhook] ── 인터랙티브 메시지 전송 (RCA 요약 + Action 버튼)
         │
         ├─── (온콜 엔지니어 검토)
         │
         ▼ [Approve / Reject 클릭]
[Slack Interactivity API] ── POST 요청 발송
         │
         ▼
[Action Runner Server (FastAPI / RBAC 통제)]
         │ 1. 서명 검증 (Slack Request Signing Secret)
         │ 2. 접근 권한 확인 (승인자 화이트리스트)
         │ 3. 감사 로그(Audit Log) 기록
         │ 4. 제한된 K8s ServiceAccount로 스크립트 실행
         ▼
[Kubernetes Target Pod / Node] ── 조치 적용 (e.g. Quota 임시 증설, Safe Restart)
         │
         ▼
[Slack Thread Update] ── 최종 실행 결과 및 리소스 상태 스레드 답글 업데이트

---

2. 핵심 안전 원칙 (Safety Guardrails)

• 프롬프트 생성 쉘 코드 직접 실행 금지:
  LLM이 자유 텍스트로 생성한 rm -rf나 임의의 kubectl 명령어를 쉘에서 eval이나 exec하는 것은 절대 금지합니다.
• 화이트리스트 액션 카탈로그(Action Catalog) 운영:
  엔지니어가 사전에 안전성을 검증한 스크립트만 고유 ID(ACTION_ID)로 등록하고, 에이전트는 ID와 파라미터(예: target_pod, target_ns, duration_minutes)만 전달합니다.
• 조치 롤백 메커니즘 (Time-To-Live):
  디버깅이나 부하 완화를 위해 CPU/Memory Quota를 임시 증설한 경우, 일정 시간(예: 30분) 후 원래 값으로 자동 원복되는 TTL 정책을 적용합니다.

---

3. 단계별 구현 코드

① Slack 인터랙티브 블록 메시지 전송 (Dispatcher)

3단계 리포트 완료 후 엔지니어에게 액션 버튼이 포함된 메시지를 발송합니다.

import json
import requests

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/T00/B00/XXXX"

def send_approval_request(incident_summary: str, target_pod: str, target_ns: str):
    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": "🚨 [SRE Agent] 장애 진단 및 조치 승인 요청"}
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"*대상 리소스:* `{target_ns}/{target_pod}`\n*진단 요약:*\n{incident_summary}"}
        },
        {"type": "divider"},
        {
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "⚡ CPU Limit 2배 임시 증설 (30분)"},
                    "style": "primary",
                    "value": json.dumps({
                        "action": "TEMPORARY_CPU_SCALE_UP",
                        "pod": target_pod,
                        "namespace": target_ns,
                        "multiplier": 2,
                        "ttl_mins": 30
                    }),
                    "action_id": "approve_scale_up"
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "🔄 Pod 안전 기동 (Graceful Restart)"},
                    "value": json.dumps({
                        "action": "SAFE_POD_RESTART",
                        "pod": target_pod,
                        "namespace": target_ns
                    }),
                    "action_id": "approve_restart"
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "❌ 조치 거부 (수동 대응)"},
                    "style": "danger",
                    "value": json.dumps({"action": "REJECT"}),
                    "action_id": "reject_action"
                }
            ]
        }
    ]

    requests.post(SLACK_WEBHOOK_URL, json={"blocks": blocks})

② 보안 검증 및 액션 실행 웹훅 서버 (action_runner.py)

엔지니어가 버튼을 눌렀을 때 슬랙 서버가 호출하는 백엔드 서버입니다.

from fastapi import FastAPI, Request, BackgroundTasks, HTTPException
import json
import hmac
import hashlib
import time
import subprocess
import os

app = FastAPI()
SLACK_SIGNING_SECRET = os.environ.get("SLACK_SIGNING_SECRET", "your-signing-secret")
ALLOWED_APPROVERS = ["admin_user_id", "oncall_sre_id"]

def verify_slack_signature(body: bytes, headers: dict) -> bool:
    timestamp = headers.get("X-Slack-Request-Timestamp", "")
    if abs(time.time() - float(timestamp)) > 60 * 5:
        return False
    sig_basestring = f"v0:{timestamp}:{body.decode('utf-8')}".encode('utf-8')
    computed_sig = 'v0=' + hmac.new(
        SLACK_SIGNING_SECRET.encode('utf-8'),
        sig_basestring,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed_sig, headers.get("X-Slack-Signature", ""))

def execute_action(payload: dict):
    action_data = json.loads(payload["actions"][0]["value"])
    action_type = action_data.get("action")
    user_name = payload["user"]["username"]
    pod = action_data.get("pod")
    ns = action_data.get("namespace")

    print(f"[AUDIT] {user_name} 승인 - 액션: {action_type}, 대상: {ns}/{pod}")

    if action_type == "TEMPORARY_CPU_SCALE_UP":
        # 사전 정의된 스크립트 실행 (임의 쉘 방지, 파라미터 전달만 허용)
        subprocess.run([
            "./scripts/apply_cgroup_temporary_override.sh",
            pod,
            str(action_data.get("multiplier", 2)),
            str(action_data.get("ttl_mins", 30))
        ], check=True)
    elif action_type == "SAFE_POD_RESTART":
        subprocess.run(["kubectl", "delete", "pod", pod, "-n", ns, "--grace-period=30"], check=True)

@app.post("/slack/interactions")
async def handle_slack_interaction(request: Request, background_tasks: BackgroundTasks):
    body = await request.body()
    headers = request.headers

    if not verify_slack_signature(body, headers):
        raise HTTPException(status_code=400, detail="Invalid request signature")

    form_data = await request.form()
    payload = json.loads(form_data["payload"])

    user_id = payload["user"]["id"]
    if user_id not in ALLOWED_APPROVERS:
        return {"text": "⚠️ 해당 조치를 승인할 권한이 없습니다."}

    action_type = json.loads(payload["actions"][0]["value"]).get("action")
    if action_type == "REJECT":
        return {"text": f"❌ {payload['user']['name']} 님이 자동 조치를 취소했습니다."}

    # 오래 걸릴 수 있는 K8s 조치 작업을 백그라운드 태스크로 위임
    background_tasks.add_task(execute_action, payload)

    return {"text": f"⏳ {payload['user']['name']} 님의 승인 수신: `{action_type}` 실행 중..."}

---

4. 프로덕션 운영 체크리스트

• 전용 ServiceAccount & 최소 권한:
• 실행 서버(action_runner)의 K8s RBAC은 cluster-admin 대신 대상 네임스페이스의 Pod get/delete, 혹은 특정 cgroup 제어 권한만 할당합니다.

• 동시성 락(Concurrency Lock):
• 슬랙 채널에서 여러 엔지니어가 동시에 버튼을 눌러 중복 실행되는 것을 방지하기 위해, 최초 클릭 시 Redis 기반 SETNX 락을 획득하고 버튼을 비활성화(disabled) 상태로 즉시 업데이트합니다.

• Audit Trail(감사 기록):
• 누가, 언제, 어떤 장애 리포트를 보고 어떤 조치를 승인했는지를 OpenSearch/ELK 보안 감사 인덱스에 영구 보관합니다.