OpsGenie

# Prometheus Alertmanager -> OpsGenie 연동
# alertmanager.yml
global:
  resolve_timeout: 5m

route:
  receiver: 'opsgenie-critical'
  group_by: ['alertname', 'service']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h
  routes:
    # Critical 알림은 즉시 전송
    - match:
        severity: critical
      receiver: 'opsgenie-critical'
      continue: true
    # Warning은 업무시간에만
    - match:
        severity: warning
      receiver: 'opsgenie-warning'
      active_time_intervals:
        - business_hours

receivers:
  - name: 'opsgenie-critical'
    opsgenie_configs:
      - api_key: ''
        message: '{{ .GroupLabels.alertname }}: {{ .CommonAnnotations.summary }}'
        description: '{{ .CommonAnnotations.description }}'
        priority: 'P1'
        tags: 'prometheus,{{ .GroupLabels.service }}'
        # 알림 라우팅
        responders:
          - type: team
            name: 'platform-oncall'
        # 추가 상세 정보
        details:
          environment: '{{ .GroupLabels.environment }}'
          runbook: '{{ .CommonAnnotations.runbook_url }}'

  - name: 'opsgenie-warning'
    opsgenie_configs:
      - api_key: ''
        priority: 'P3'
        responders:
          - type: team
            name: 'platform-team'

# OpsGenie API를 통한 알림 생성 및 관리
import requests

OPSGENIE_API_KEY = "your-api-key"
BASE_URL = "https://api.opsgenie.com/v2"

headers = {
    "Authorization": f"GenieKey {OPSGENIE_API_KEY}",
    "Content-Type": "application/json"
}

# 알림 생성
def create_alert(message: str, priority: str = "P3", **kwargs):
    payload = {
        "message": message,
        "priority": priority,
        "tags": kwargs.get("tags", []),
        "details": kwargs.get("details", {}),
        "responders": [
            {"type": "team", "name": "platform-oncall"}
        ]
    }
    response = requests.post(
        f"{BASE_URL}/alerts",
        headers=headers,
        json=payload
    )
    return response.json()

# 알림 닫기
def close_alert(alert_id: str, note: str = "Issue resolved"):
    payload = {"note": note}
    response = requests.post(
        f"{BASE_URL}/alerts/{alert_id}/close",
        headers=headers,
        json=payload
    )
    return response.json()

# 온콜 담당자 조회
def get_on_call(schedule_name: str):
    response = requests.get(
        f"{BASE_URL}/schedules/{schedule_name}/on-calls",
        headers=headers
    )
    on_calls = response.json().get("data", {}).get("onCallParticipants", [])
    return [p["name"] for p in on_calls]

# 사용 예시
create_alert(
    message="Database connection pool exhausted",
    priority="P1",
    tags=["database", "production"],
    details={
        "service": "order-api",
        "pod": "order-api-abc123",
        "runbook": "https://wiki/runbooks/db-pool-exhausted"
    }
)