26Y19c3
Alertmanager가 던지는 실시간 장애 웹훅을 수신하여 15분간의 인프라 메트릭을 Polars로 압축(2단계)하고, 이를 사내 LLM과 연동하여 슬랙 경보까지 일사천리로 사출(3단계)하는 실시간 JIT(Just-In-Time) RCA 에이전트 스크립트와 실행 방법입니다.
배치 파이프라인처럼 대용량 디스크 I/O를 유발하지 않도록, 메모리 상에서 모든 가공과 추론 조율을 처리하는 초경량 비동기 마이크로 서비스 아키텍처로 구현되었습니다.
💻 1. 실시간 2·3단계 통합 스크립트 (sre_jit_rca_agent.py)
import os
import json
import time
from datetime import datetime, timedelta
import requests
import polars as pl
from fastapi import FastAPI, Request, BackgroundTasks
import uvicorn
app = FastAPI(title="SRE Just-In-Time RCA Agent")
# ─── 인프라 토폴로지 및 엔드포인트 설정 ──────────────────────────
THANOS_URL = os.getenv("THANOS_QUERY_URL", "http://thanos-querier.monitoring.svc:9090")
INTERNAL_LLM_URL = "https://llm.internal.company.net/v1/chat/completions"
INTERNAL_LLM_API_KEY = os.getenv("INTERNAL_LLM_API_KEY", "your-sre-token-here")
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL", "https://hooks.slack.com/services/T00/B00/X00")
def fetch_thanos_instant_metrics(workload_type: str, namespace: str) -> dict:
"""
2단계: Alert 시점 기준 최근 15분간의 SRE 핵심 변동 계수를 Thanos로부터 긁어와
Polars 엔진으로 초고속 메모리 요약을 수행합니다.
"""
print(f" [Thanos 스냅샷 캡처] {namespace} / {workload_type} 대상 메트릭 수집 중...")
# 15분 전부터 현재까지의 데이터 타겟팅
end_time = int(time.time())
start_time = end_time - 900
# PromQL을 실시간 스냅샷 형태로 질의 (5분 단위 롤업 대비 해상도 유지)
query = f'container_cpu_usage_seconds_total{{namespace="{namespace}"}}'
try:
response = requests.get(
f"{THANOS_URL}/api/v1/query_range",
params={"query": query, "start": start_time, "end": end_time, "step": "60s"},
timeout=10
)
if response.status_code != 200:
return {"error": "Thanos Connection Failed"}
data = response.json().get("data", {}).get("result", [])
if not data:
return {"status": "No raw metric points captured"}
# Thanos 매트릭스 데이터를 Polars 프레임으로 가속 변환
records = []
for item in data:
for val in item.get("values", []):
records.append({
"timestamp": int(val[0]),
"value": float(val[1])
})
df = pl.DataFrame(records)
# 실시간 LLM향 1분 해상도 거동 특징량 압축 요약
summary = df.select([
pl.col("value").max().round(2).alias("recent_max"),
pl.col("value").mean().round(2).alias("recent_avg"),
(pl.col("value").std() / pl.col("value").mean().clip(lower_bound=0.001)).round(3).alias("recent_skew_cv")
]).to_dicts()[0]
return summary
except Exception as e:
return {"error": f"Polars processing breakdown: {str(e)}"}
def fetch_k8s_container_logs(namespace: str, pod: str, container: str) -> str:
"""
2단계 확장: 장애 포드의 최신 20줄 로그 백로그를 캡처합니다.
(실제 프로덕션에서는 K8s API 토큰 인증이 필요하므로 여기서는 경량화 뼈대로 구현)
"""
k8s_token_path = "/var/run/secrets/kubernetes.io/serviceaccount/token"
if not os.path.exists(k8s_token_path):
return "[System Notice] Local Pod Mocking Active - Log retrieval skipped."
with open(k8s_token_path, "r") as f:
token = f.read().strip()
k8s_url = f"https://kubernetes.default.svc/api/v1/namespaces/{namespace}/pods/{pod}/log"
headers = {"Authorization": f"Bearer {token}"}
params = {"container": container, "tailLines": 20}
try:
res = requests.get(k8s_url, headers=headers, params=params, verify=False, timeout=5)
return res.text if res.status_code == 200 else f"Log retrieve status code: {res.status_code}"
except Exception as e:
return f"Failed to connect K8s Core API: {str(e)}"
def execute_jit_llm_rca(alert_meta: dict, metrics_summary: dict, log_tail: str):
"""
3단계: 15분 압축 인텍스 + 로그 덤프를 기반으로 온디맨드 LLM RCA를 가동합니다.
1초 이내 초고속 추론을 위해 템플릿 제약을 극단으로 압축합니다.
"""
print(f"🚨 [JIT LLM RCA 기동] Alert: {alert_meta['alertname']} 분석 시작...")
prompt = f"""
[Role]: 당신은 1분 1초를 다투는 클라우드 네이티브 SRE 인프라 응급실 수석 아키텍트입니다.
현재 Alertmanager로부터 실시간 시스템 전이 장애 경보가 전달되었습니다.
[장애 메타데이터]:
- 알람명: {alert_meta['alertname']}
- 타겟 스택: {alert_meta.get('workload_type', 'Unknown')}
- 네임스페이스/포드: {alert_meta.get('namespace')}/{alert_meta.get('pod')}
- 심각도: {alert_meta.get('severity', 'critical')}
[최근 15분 Polars 압축 특징량]:
{json.dumps(metrics_summary, indent=2)}
[장애 포드 최신 로그 백로그 스냅샷]:
\"\"\"
{log_tail}
\"\"\"
[Mission]:
인간 엔지니어가 3분 안에 시스템 장애 확산을 방어할 수 있도록, 미사여구 없이 엄격하고 간결하게 딱 2가지 핵심만 사출하세요.
1. **실시간 핵심 원인 (RCA)**: 데이터 스큐, OOM 임계 포화, Keycloak 토큰 스톰, Cilium eBPF 패킷 드롭 규칙과 대조하여 어떤 레이어가 꼬인 것인지 한 문장으로 확정 진단하세요.
2. **즉시 조치 스크립트/플레이북**: 복잡한 설명 대신 엔지니어가 터미널에 복사-붙여넣기해서 장애를 일시 격리(Throttling)하거나 스펙을 임시 복구할 수 있는 'kubectl' 혹은 '오픈소스 내부 CLI' 명령어를 명확한 파라미터와 함께 제시하세요.
[Format]: 서론/결론/인사말을 전부 생략하고 슬랙 메시지에 바로 들어갈 수 있도록 콤팩트한 마크다운 형태로만 출력하세요.
"""
payload = {
"model": "internal-cloudnative-sre-brain",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.05, # 실시간 영역이므로 결정론적 아웃풋을 위해 온도를 극단으로 하향
"max_tokens": 1000
}
headers = {"Authorization": f"Bearer {INTERNAL_LLM_API_KEY}", "Content-Type": "application/json"}
try:
response = requests.post(INTERNAL_LLM_URL, headers=headers, data=json.dumps(payload), timeout=30)
if response.status_code == 200:
llm_rca_text = response.json()['choices'][0]['message']['content']
send_to_slack(alert_meta, metrics_summary, llm_rca_text)
else:
print(f"❌ LLM 추론 실패: {response.status_code}")
except Exception as e:
print(f"💥 JIT LLM 파이프라인 연계 에러: {str(e)}")
def send_to_slack(alert_meta: dict, metrics_summary: dict, rca_text: str):
"""
3단계 최종: 완성된 정밀 진단서 및 조치 스크립트를 슬랙 채널에 브로드캐스팅합니다.
"""
now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 슬랙 Blocks 포맷으로 가독성 극대화
payload = {
"blocks": [
{
"type": "header",
"text": {"type": "plain_text", "text": f"🚨 SRE Just-In-Time Anomaly Alert [{alert_meta['severity'].upper()}]"}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*경보명:* {alert_meta['alertname']}\n*타겟:* `{alert_meta.get('workload_type','unknown')}`"},
{"type": "mrkdwn", "text": f"*발생시각:* {now_str}\n*네임스페이스:* {alert_meta.get('namespace','-')}"}
]
},
{
"type": "section",
"text": {"type": "mrkdwn", "text": f"*📊 Polars 15m 지표 스냅샷:* Max `{metrics_summary.get('recent_max',0)}` | Skew CV `{metrics_summary.get('recent_skew_cv',0)}`"}
},
{"type": "divider"},
{
"type": "section",
"text": {"type": "mrkdwn", "text": f"*🤖 LLM Real-time RCA & Playbook:*\n{rca_text}"}
}
]
}
try:
requests.post(SLACK_WEBHOOK_URL, json=payload, timeout=5)
print("✅ [슬랙 전송 완수] 실시간 SRE 대응 센터로 정밀 조치 명세가 전송되었습니다.")
except Exception as e:
print(f"❌ 슬랙 웹훅 발송 실패: {str(e)}")
# ─── Alertmanager Webhook 수신 엔드포인트 ───────────────────
@app.post("/alert-webhook")
async def handle_alert_webhook(request: Request, background_tasks: BackgroundTasks):
"""
Alertmanager로부터 비동기로 경보 배열을 수신하여
API 서버가 블로킹(Timeout)되지 않도록 백그라운드 태스크로 연동 처리를 위임합니다.
"""
body = await request.json()
alerts = body.get("alerts", [])
for alert in alerts:
status = alert.get("status")
# 알람이 해결(resolved)된 것은 무시하고, 활성화(firing)된 장애만 정밀 타격
if status == "firing":
labels = alert.get("labels", {})
annotations = alert.get("annotations", {})
alert_meta = {
"alertname": labels.get("alertname", "UnknownAlert"),
"workload_type": labels.get("workload_type", labels.get("app", "unknown")),
"namespace": labels.get("namespace", "default"),
"pod": labels.get("pod", ""),
"container": labels.get("container", ""),
"severity": labels.get("severity", "critical"),
"description": annotations.get("description", "")
}
# 비동기 백그라운드 워커 풀로 2단계(융합) 및 3단계(LLM) 오케스트레이션 위임
background_tasks.add_task(process_pipeline_flow, alert_meta)
return {"status": "accepted", "queued_alerts": len(alerts)}
def process_pipeline_flow(alert_meta: dict):
# 2단계: 메트릭 및 로그 컨텍스트 수집
metrics_summary = fetch_thanos_instant_metrics(alert_meta["workload_type"], alert_meta["namespace"])
log_tail = fetch_k8s_container_logs(alert_meta["namespace"], alert_meta["pod"], alert_meta["container"])
# 3단계: LLM RCA 호출 및 슬랙 마감
execute_jit_llm_rca(alert_meta, metrics_summary, log_tail)
if __name__ == "__main__":
uvicorn.run("sre_jit_rca_agent.py:app", host="0.0.0.0", port=8080, reload=False)
🚀 2. 실행 및 프로덕션 배포 방법
1단계: 에이전트 구동 환경 및 의존성 라이브러리 설치
해당 실시간 RCA 에이전트는 초경량 고속 웹 컴포넌트인 FastAPI와 비동기 WAS 커널 Uvicorn 기반으로 구동됩니다. 에이전트가 실행될 서버 혹은 컨테이너 내부에서 아래 명령어로 코어 의존성을 설치합니다.
pip install fastapi uvicorn polars requests
2단계: 환경 변수(Credential) 바인딩 및 백그라운드 구동
Thanos 쿼리어 주소와 내부 LLM API 토큰, 알람을 수신할 사내 슬랙 웹훅 엔드포인트를 환경 변수로 명확히 내보낸 뒤, 터미널이 끊겨도 죽지 않도록 백그라운드(nohup) 프로세스로 백업 가동합니다.
# 1. 인프라 주요 접속 명세 환경변수 주입
export THANOS_QUERY_URL="http://thanos-querier.monitoring.svc:9090"
export INTERNAL_LLM_API_KEY="sk-company-sre-brain-token-xxx"
export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/T00/B00/X00"
# 2. 8080 포트로 SRE 실시간 에이전트 백그라운드 기동
nohup python3 sre_jit_rca_agent.py > sre_jit_agent.log 2>&1 &
# 3. 구동 포트 리슨 바인딩 정상 여부 팩트 체크
netstat -lntp | grep 8080
3단계: Kubernetes 서비스(Service) 레이어 결착 (인사이드 K8s 배포 시)
만약 이 에이전트를 K8s 내부 모니터링 네임스페이스에 포드로 직접 올릴 경우, 1단계 Alertmanager가 이 웹훅 주소(http://sre-jit-agent...:8080/alert-webhook)를 도메인으로 해석하여 통신할 수 있도록 다음과 같이 내부 ClusterIP 서비스를 생성해 연결해주면 실시간 방어 체계가 완전하게 종료됩니다.
apiVersion: v1
kind: Service
metadata:
name: sre-jit-agent
namespace: monitoring
spec:
selector:
app: sre-jit-rca-agent
ports:
- protocol: TCP
port: 8080
targetPort: 8080
type: ClusterIP
engineer