26S26b

==

Kubernetes 환경에서 Alert/Incident 발생 시점을 기준으로 ‘전후 N분(예: -15분 ~ +15분)’의 메트릭과 로그 스냅샷을 자동 덤프하여 영구 저장하는 오픈소스 도구들을 추천합니다.

Incident Context Collector(RCA 입력 번들러) 역할에 최적화된 도구들을 특성별로 분류했습니다.

---

1. Robusta (가장 추천하는 완성형 오픈소스)

Kubernetes Alert 및 Incident 자동화 플랫폼으로, Alert 발생 시 관련 파드의 직전 로그, 이벤트, 프로메테우스 메트릭 그래프/스냅샷을 자동으로 긁어 모아 번들링하는 데 가장 특화되어 있습니다.

• 동작 방식:
• Alertmanager Webhook 또는 K8s Event를 Robusta Runner가 수신합니다.
• 정의된 플레이북(Action)에 따라 해당 파드/노드의 kubectl logs (tail/previous), K8s Events, 연관 PromQL 쿼리 결과를 자동으로 패키징합니다.

• 장점:
• 사내 웹훅, Slack, S3/MinIO 버킷, 또는 REST API 엔드포인트로 덤프 데이터를 전송(Sink)할 수 있습니다.
• Python 기반 커스텀 플레이북 작성이 매우 쉬워, Alert 수신 시 사내 LLM RCA Agent API로 바로 Context JSON을 쏴주는 브리지 역할을 완벽히 수행합니다.

• 적합한 용도: RCA Agent 연동용 Incident Context 자동 번들링.

---

2. Prometheus Snapshot API + Cron/Webhook Worker (메트릭 전용 경량화)

별도의 무거운 에이전트 없이 기존 사내 Prometheus 환경을 그대로 활용하는 가장 표준적인 방법입니다.

• 동작 방식:
• Prometheus의 --web.enable-admin-api 플래그를 활성화합니다.
• Alert 발생 시 수집 스크립트(또는 가벼운 FastAPI 워커)가 아래 API를 호출합니다:
• TSDB Snapshot API: POST /api/v1/admin/tsdb/snapshot (현재 TSDB 디렉터리에 장애 시점의 raw 블록 스냅샷 생성 후 오브젝트 스토리지로 복사)
• Range Query API (실무 추천): TSDB 전체 덤프 대신, Prometheus REST API(GET /api/v1/query_range)로 주요 PromQL 지표를 장애 전후 시간(start, end, step)으로 호출해 JSON 파일로 즉시 저장.

• 장점: 신규 솔루션 배포 없이 API 호출 몇 줄로 장애 전후 메트릭을 정형 JSON으로 추출 가능.

---

3. OpenSearch Point in Time (PIT) / Scroll Export Script (로그 전용)

이미 OpenSearch 클러스터를 운영 중이므로, Alert 발생 시점의 로그 윈도우를 잘라내어 저장하는 경량 컨테이너를 두는 방식입니다.

• 동작 방식:
• Alert 발생 시 Alertmanager의 Webhook을 받는 작은 Webhook 수신기(Python/Go)를 파드로 띄웁니다.
• Webhook에 담긴 startsAt, endsAt (또는 현재시각 기준 now-15m ~ now+5m)을 기준으로 OpenSearch Query DSL을 날려 해당 Pod/Node의 로그를 추출하여 파일로 저장합니다.

• 장점: 추가적인 K8s 클러스터 데몬셋 설치 없이 중앙 로그 저장소에서 안전하게 발췌 가능.

---

4. Kubernetes Diagnostic Tools (온디맨드/사후 정밀 진단)

Alert 시점에 노드 OS 지표, 네트워크 상태, 커널 로그, dmesg, cgroup 통계 등 K8s 기본 메트릭 이상의 심층 시스템 덤프가 필요할 때 사용합니다.

A. Kube-diagnoser

• 특징: 장애 감지, 자동 트러블슈팅, 컨텍스트 수집에 초점을 맞춘 쿠버네티스 오퍼레이터입니다.
• 동작: K8s 노드나 파드에 이상(Memory Pressure, Disk Fail 등)이 감지되면 자동으로 진단 파이프라인을 실행해 OS 레벨 로그 및 프로파일링 데이터를 수집합니다.

B. Crashcart / Sonobuoy

• 특징: 특정 시점의 클러스터 전체 리소스 상태, 이벤트, 파드 로그 덤프를 tar.gz로 생성하는 도구입니다.

---

구성 권장 조합 (사내 RCA Agent 구축용)

가장 리소스가 적게 들고 구현이 빠른 조합은 Robusta를 Standalone/Headless 모드로 배포하는 방식입니다.

[Prometheus / Alertmanager]
          │ (Alert Webhook)
          ▼
   [Robusta Engine]
    ├─ 1. K8s Event & Pod Log 자동 수집
    ├─ 2. 장애 전후 PromQL Range Query 호출
    └─ 3. 표준 JSON 번들 생성 ──> [사내 MinIO 저장] & [RCA Agent 호출]

만약 신규 솔루션 설치 검토가 부담스러운 폐쇄망 환경이라면, Alertmanager Webhook을 수신하여 Prometheus Range Query API + OpenSearch Search API를 호출해 단일 JSON으로 압축 저장하는 경량 Webhook Pod(Python FastAPI)를 하나 구현하는 것이 가장 빠르고 통제력이 높습니다.

==

Robusta를 외부 SaaS/클라우드 연결 없이 순수 오픈소스(OSS) Standalone/Headless 모드로 배포하는 방법입니다.

SaaS UI(robusta_ui_sink)와 외부 텔레메트리를 완전히 비활성화하고, 사내 Alertmanager와 연동하여 장애 전후 로그/메트릭을 수집한 뒤 사내 Webhook(RCA Agent 수신부 또는 MinIO 버킷/FastAPI 등)으로만 쏘도록 설정하는 것이 핵심입니다.

---

1. 사전 준비 (폐쇄망 환경)

외부 인터넷이 차단된 환경이라면 아래 이미지를 사내 프라이빗 레지스트리(Harbor 등)로 사전 미러링합니다.

• ghcr.io/robusta-dev/robusta-runner:<version>
• ghcr.io/robusta-dev/robusta-forwarder:<version>
• Robusta Helm Chart (helm pull robusta/robusta)

---

2. Standalone 전용 values.yaml 작성

클라우드 연동(robusta_ui_sink 등)을 제외하고, 사내 Webhook으로 데이터를 덤프하도록 구성한 미니멀 설정입니다.

# values.yaml

# 1. 글로벌 설정
globalConfig:
  cluster_name: "prod-kr01"
  account_id: "internal-isolated-cluster"   # 로컬 식별용 임의 문자열
  signing_key: "generated-internal-secret-key-32chars" # 내부 서명용 임의 키

# 2. 클라우드 SaaS / Telemetry 완전 차단
enableCloudReporting: false
disableTelemetry: true

# 3. K8s Runner 이미지 지정 (사내 레지스트리 사용 시)
runner:
  image: "harbor.internal.net/robusta/robusta-runner:latest"
  resources:
    requests:
      cpu: 200m
      memory: 512Mi
    limits:
      cpu: 1000m
      memory: 1Gi

# 4. Sinks 설정 (수집된 Context를 사내 Webhook/RCA Agent로 전송)
sinksConfig:
  - webhook_sink:
      name: rca_agent_webhook
      url: "http://rca-collector.monitoring.svc.cluster.local:8000/api/v1/incident"
      format: "json"        # JSON 형식으로 정형 데이터 전송
      size_limit: 10485760  # 최대 페이로드 크기 (10MB)

# 5. Alert 발생 시 수행할 Action 정의 (로그/이벤트 수집)
customPlaybooks:
  - triggers:
      - on_prometheus_alert: {}
    actions:
      - logs_enricher:
          lines: 200        # 장애 발생 파드의 직전 로그 200줄 수집
      - events_enricher: {} # K8s Warning/Error 이벤트 수집
      - prometheus_enricher: {} # 관련 PromQL 메트릭 그래프/데이터 수집
    sinks:
      - rca_agent_webhook

---

3. Helm 배포

작성한 values.yaml을 이용해 모니터링 네임스페이스에 배포합니다.

# 네임스페이스 생성
kubectl create namespace robusta

# Standalone 모드로 Helm 배포
helm install robusta robusta/robusta \
  --namespace robusta \
  --values values.yaml

배포 후 확인:

• robusta-runner 파드와 robusta-forwarder 파드가 정상 구동되는지 확인합니다.
• 로그 확인: kubectl logs -n robusta -l app=robusta-runner
• "SaaS connection disabled" 또는 외부 연결 시도 없이 Webhook Sink만 활성화되어 초기화되었는지 점검합니다.

---

4. Alertmanager 연동 (Webhook 등록)

기존에 사용 중인 K8s 클러스터의 Prometheus Alertmanager가 Alert를 Robusta로 보내주도록 alertmanager.yaml의 수신자(receiver)에 Robusta Forwarder 서비스를 등록합니다.

# alertmanager config 예시
receivers:
  - name: "robusta"
    webhook_configs:
      - url: "http://robusta-forwarder.robusta.svc.cluster.local:80/alerts"
        send_resolved: true

route:
  receiver: "default-receiver"
  routes:
    # MinIO 관련 알람이나 주요 인프라 알람을 Robusta로 라우팅
    - match_re:
        alertname: "(MinIO.*|NodeDisk.*|KubePodCrashLooping)"
      receiver: "robusta"
      continue: true

---

5. 수신 데이터 검증 (FastAPI 예시)

webhook_sink로 지정된 엔드포인트(http://rca-collector:8000/api/v1/incident)에서 실제로 받게 되는 데이터 형태는 아래와 같습니다.

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/api/v1/incident")
async def receive_incident(request: Request):
    payload = await request.json()
    
    # Robusta가 자동으로 수집해서 넣어준 항목들
    title = payload.get("title")
    description = payload.get("description")
    evidence_logs = payload.get("evidence", []) # 파드 로그, 이벤트, 메트릭 포함
    
    print(f"[*] Alert 수신: {title}")
    # -> 여기서 이 데이터를 파싱해 앞서 설계한 LangGraph RCA 워크플로우로 전달
    return {"status": "received"}

이 구성을 적용하면 클라우드 SaaS나 외부 인터넷 의존성 없이, Alert 발생 시 K8s 파드 로그 + K8s Event + PromQL 지표가 완벽하게 번들링되어 사내 RCA Agent로 다이렉트 전달됩니다.

==

Alertmanager 없이 운영자가 원할 때 수동으로 트리거(Manual Trigger)하여, 특정 시간 전후의 K8s 이벤트, 파드 로그, 메트릭을 수집한 뒤 MinIO AIStor S3 버킷에 직접 JSON/Archive 형태로 업로드하는 2가지 최적 구현 방안을 정리해 드립니다.

---

방법 1. Robusta Headless 환경에서 수동 트리거 + MinIO 업로드

Robusta는 Alert 수신 외에도 수동 실행 API (/api/run) 또는 robusta playbooks trigger CLI 커맨드를 기본 지원합니다. 또한 MinIO(S3 호환)에 바로 파일을 올릴 수 있도록 커스텀 액션을 구성할 수 있습니다.

1) MinIO 업로드 커스텀 Playbook 작성

Robusta Runner 파드 내에서 실행될 수 있도록 values.yaml의 customPlaybooks에 직접 정의하거나 커스텀 액션 파이썬 코드를 마운트합니다.

# custom_actions/minio_sink.py
import json
import boto3
from datetime import datetime
from robusta.api import *

class MinIOIncidentDumpParams(ActionParams):
    namespace: str
    pod_name: str
    bucket_name: str = "incident-dumps"
    minio_endpoint: str = "http://minio-aistor.storage.svc:9000"
    access_key: str = "YOUR_ACCESS_KEY"
    secret_key: str = "YOUR_SECRET_KEY"

@action
def capture_and_upload_to_minio(event: ExecutionBaseEvent, params: MinIOIncidentDumpParams):
    # 1. 대상 파드의 최근 로그 및 이벤트 수집
    k8s = RobustaKubeClient()
    logs = k8s.get_pod_logs(name=params.pod_name, namespace=params.namespace, tail_lines=300)
    
    # 2. 페이로드 패키징
    now_str = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
    incident_data = {
        "incident_id": f"MANUAL-{now_str}",
        "timestamp": now_str,
        "target": {"namespace": params.namespace, "pod": params.pod_name},
        "logs": logs.splitlines(),
    }
    
    # 3. MinIO AIStor(S3 호환) 업로드
    s3 = boto3.client(
        "s3",
        endpoint_url=params.minio_endpoint,
        aws_access_key_id=params.access_key,
        aws_secret_access_key=params.secret_key
    )
    
    object_key = f"incidents/{params.namespace}/{params.pod_name}_{now_str}.json"
    s3.put_object(
        Bucket=params.bucket_name,
        Key=object_key,
        Body=json.dumps(incident_data, ensure_ascii=False, indent=2),
        ContentType="application/json"
    )
    print(f"Successfully uploaded snapshot to MinIO: s3://{params.bucket_name}/{object_key}")

2) 수동 트리거 실행 (CLI / cURL)

Robusta Runner 내부 엔드포인트로 수동 trigger 요청을 보냅니다.

# curl을 통한 수동 트리거 예시
curl -X POST http://robusta-runner.robusta.svc.cluster.local:5000/api/run \
  -H "Content-Type: application/json" \
  -d '{
    "action_name": "capture_and_upload_to_minio",
    "action_params": {
      "namespace": "storage",
      "pod_name": "kr01-aistor-dn03",
      "bucket_name": "incident-dumps"
    }
  }'

---

방법 2. 초경량 K8s Job / CLI 스크립트 방식 (PoC 단계 가장 추천)

Robusta 러닝커브나 설정 부담 없이, 운영자가 원할 때 kubectl 또는 CLI 스크립트 한 줄로 특정 노드/파드의 로그 + 프로메테우스 메트릭(전후 15분)을 긁어 MinIO AIStor에 바로 적재하는 방식입니다.

1) 캡처 및 MinIO 업로드 스크립트 (capture_incident.py)

이 스크립트는 지정한 파드의 직전 30분 로그, 이벤트, 그리고 Prometheus의 Range Query API를 호출해 단일 JSON으로 압축한 뒤 MinIO 버킷으로 쏩니다.

import os
import sys
import json
import time
import requests
import boto3
from datetime import datetime, timezone, timedelta
from kubernetes import client, config

# --- 환경 설정 ---
MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT", "http://10.200.15.10:9000")
MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY", "minioadmin")
MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY", "minioadminpassword")
MINIO_BUCKET = "incident-context-repo"

PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://prometheus-k8s.monitoring.svc:9090")

def get_promql_range(query: str, start_ts: int, end_ts: int, step="15s"):
    """장애 전후 시간대 메트릭 수집"""
    url = f"{PROMETHEUS_URL}/api/v1/query_range"
    params = {"query": query, "start": start_ts, "end": end_ts, "step": step}
    try:
        res = requests.get(url, params=params, timeout=5)
        return res.json().get("data", {}).get("result", [])
    except Exception as e:
        return f"Prometheus query failed: {str(e)}"

def capture_snapshot(namespace: str, pod_name: str, window_minutes: int = 15):
    # K8s 인증 (로컬 kubeconfig 또는 In-Cluster ServiceAccount)
    try:
        config.load_incluster_config()
    except:
        config.load_kube_config()
    
    v1 = client.CoreV1Api()
    
    now = datetime.now(timezone.utc)
    start_time = now - timedelta(minutes=window_minutes)
    start_ts = int(start_time.timestamp())
    end_ts = int(now.timestamp())
    
    print(f"[*] 캡처 대상: {namespace}/{pod_name} (윈도우: -{window_minutes}분 ~ 현재)")
    
    # 1. K8s Pod Logs (현재 컨테이너 + 필요시 이전 실패 컨테이너)
    logs = ""
    try:
        logs = v1.read_namespaced_pod_log(name=pod_name, namespace=namespace, since_seconds=window_minutes*60, tail_lines=500)
    except Exception as e:
        logs = f"Log retrieval error: {str(e)}"

    # 2. K8s Events
    events = []
    field_selector = f"involvedObject.name={pod_name}"
    event_list = v1.list_namespaced_event(namespace=namespace, field_selector=field_selector)
    for ev in event_list.items:
        events.append({
            "type": ev.type,
            "reason": ev.reason,
            "message": ev.message,
            "last_timestamp": str(ev.last_timestamp)
        })

    # 3. Prometheus Metrics Snapshot (MinIO 주요 지표)
    metrics = {
        "drive_latency": get_promql_range(f'minio_node_drive_total_duration_microseconds{{pod="{pod_name}"}}', start_ts, end_ts),
        "goroutines": get_promql_range(f'go_goroutines{{pod="{pod_name}"}}', start_ts, end_ts),
        "s3_waiting_requests": get_promql_range(f'minio_s3_requests_waiting_total{{pod="{pod_name}"}}', start_ts, end_ts)
    }

    # 4. JSON Payload 패키징
    incident_id = f"INC-MANUAL-{now.strftime('%Y%m%d-%H%M%S')}"
    payload = {
        "incident_id": incident_id,
        "capture_timestamp": now.isoformat(),
        "target": {"namespace": namespace, "pod": pod_name},
        "time_window": {"start": start_time.isoformat(), "end": now.isoformat()},
        "k8s_events": events,
        "metrics_summary": metrics,
        "logs_summary": logs.splitlines()[-200:] # 최근 200줄 요약
    }

    # 5. MinIO AIStor에 직접 S3 PutObject
    s3 = boto3.client(
        "s3",
        endpoint_url=MINIO_ENDPOINT,
        aws_access_key_id=MINIO_ACCESS_KEY,
        aws_secret_access_key=MINIO_SECRET_KEY
    )
    
    # 버킷 확인/생성
    try:
        s3.head_bucket(Bucket=MINIO_BUCKET)
    except:
        s3.create_bucket(Bucket=MINIO_BUCKET)

    object_key = f"manual_captures/{incident_id}_{pod_name}.json"
    s3.put_object(
        Bucket=MINIO_BUCKET,
        Key=object_key,
        Body=json.dumps(payload, ensure_ascii=False, indent=2),
        ContentType="application/json"
    )
    
    print(f"[✓] 성공적으로 MinIO AIStor에 저장되었습니다: s3://{MINIO_BUCKET}/{object_key}")
    return object_key

if __name__ == "__main__":
    target_ns = sys.argv[1] if len(sys.argv) > 1 else "storage"
    target_pod = sys.argv[2] if len(sys.argv) > 2 else "kr01-aistor-dn03"
    capture_snapshot(target_ns, target_pod)

---

2) K8s 일회성 실행 (One-shot Job / Pod)

클러스터 내부에서 CLI 명령 한 줄로 파드를 띄워 캡처하고 종료되도록 K8s Job 매니페스트를 준비할 수 있습니다.

# manual-capture-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: manual-incident-capture
  namespace: monitoring
spec:
  ttlSecondsAfterFinished: 300 # 완료 후 5분 뒤 자동 파기
  template:
    spec:
      serviceAccountName: rca-collector-sa # Pod log, Event 조회 RBAC 권한 필요
      containers:
      - name: collector
        image: python:3.11-slim
        command: ["python", "/scripts/capture_incident.py", "storage", "kr01-aistor-dn03"]
        env:
        - name: MINIO_ENDPOINT
          value: "http://minio-aistor.storage.svc:9000"
        - name: MINIO_ACCESS_KEY
          valueFrom:
            secretKeyRef:
              name: minio-creds
              key: access-key
        - name: MINIO_SECRET_KEY
          valueFrom:
            secretKeyRef:
              name: minio-creds
              key: secret-key
        - name: PROMETHEUS_URL
          value: "http://prometheus-k8s.monitoring.svc:9090"
      restartPolicy: Never

실행 방법:

# 장애 의심 시 수동 트리거 실행
kubectl create -f manual-capture-job.yaml

# 또는 파드 1회성 run
kubectl run capture-test --rm -i --tty --image=python:3.11-slim ...

---

워크플로우 연계 (MinIO → RCA Agent)

1. 수동 트리거: 엔지니어가 python capture_incident.py storage kr01-aistor-dn03 실행.
2. MinIO 적재: MinIO AIStor의 incident-context-repo 버킷에 INC-MANUAL-20260927-181500_kr01-aistor-dn03.json이 생성됨.
3. RCA 호출: 스크립트 마지막 줄에 사내 Agent API 엔드포인트로 {"s3_path": "s3://incident-context-repo/..."}를 호출하거나, Agent가 해당 버킷의 JSON을 읽어 LangGraph 워크플로우를 즉시 구동합니다.

==

Robusta를 Helm으로 기본 설치할 때 생성되는 Pod의 개수는 배포 구성 및 옵션에 따라 보통 1개 또는 2개입니다.

---

1. 기본 구성: 총 2개 Pod

일반적인 배포 시 아래 두 가지 핵심 컴포넌트가 각각 1개의 Pod(Deployment)로 구동됩니다.

1. robusta-runner (1개)

• 역할: 핵심 엔진입니다. Playbook(이벤트 수집, PromQL 쿼리, 로그 추출, 파이썬 스크립트 실행 등)을 실행하고 알림 및 데이터를 Sink(MinIO, Webhook 등)로 전송합니다.

2. robusta-forwarder (1개)

• 역할: Alertmanager의 웹훅 알림이나 K8s APIServer 이벤트를 수신하여 Runner로 전달해 주는 경량 프록시/포워더입니다.

---

2. Standalone / 수동 트리거 중심 구성 시: 1개 Pod (최소화 가능)

만약 이전 대화에서 논의한 것처럼 Alertmanager 연동 없이 수동 트리거(REST API / CLI)나 자체 스크립트로만 쓸 경우, robusta-forwarder를 비활성화할 수 있습니다.

• values.yaml에서 Forwarder를 끄면:

forwarder:
  enabled: false

• 이 경우 robusta-runner 1개 Pod만 단독으로 띄워서 가볍게 운영할 수 있습니다.

---

3. 추가 옵션을 켰을 때 (예외 상황)

• 내장 Prometheus 번들 설치 시 (kube-prometheus-stack.enabled=true):
• 자체 Prometheus가 없어서 Robusta 차트 내에 Prometheus/Alertmanager/Node-exporter 번들을 함께 설치하도록 활성화한 경우에는 Node 수만큼 생기는 DaemonSet(node-exporter) 및 Operator Pod들이 추가되어 수십 개 이상 늘어날 수 있습니다.
• 이미 사내 Prometheus가 있는 환경이라면 기본값(kube-prometheus-stack.enabled=false) 그대로 두어 1~2개의 Pod만 생성되도록 유지하는 것이 일반적입니다.

요약하자면:

사내 Prometheus가 이미 있고 Alertmanager 웹훅까지 연동한다면 2개, Alertmanager 연동 없이 수동 트리거/Runner만 쓸 목적이라면 1개의 Pod만 뜹니다. DaemonSet 형태로 노드마다 뜨는 구조가 아니므로 클러스터 리소스 부담은 매우 적습니다.