fix: 온콜 시험이 Slack 까지 실제 P0 와 같은 경로로

.github/workflows/ci.yml

@@ -35,6 +35,8 @@ jobs:
         run: python3 -m unittest discover -s infra/monitoring/lambdas/tests -v
       - name: Alarm catalog
         run: python3 -c "import sys,types; sys.modules['boto3']=types.ModuleType('boto3'); sys.path.insert(0,'infra/monitoring/alerting'); import unittest; r=unittest.main(module='tests_catalog', argv=['x','-v'], exit=False); sys.exit(not r.result.wasSuccessful())"
+      - name: On-call (PagerDuty)
+        run: python3 -c "import sys,types; sys.modules['boto3']=types.ModuleType('boto3'); sys.path.insert(0,'infra/monitoring/alerting'); import unittest; r=unittest.main(module='tests_pagerduty', argv=['x','-v'], exit=False); sys.exit(not r.result.wasSuccessful())"
       - name: Deployment preflight
         run: python3 -m unittest discover -s .github/scripts -p 'test_*.py' -v
 

infra/monitoring/alerting/README.md

@@ -71,5 +71,5 @@ Incident Manager(`ssm-contacts`·`ssm-incidents`)는 조직 SCP 가 막아 쓰
 1. 두 사람 모두 PagerDuty 앱 설치 + 프로필 알림 규칙에 **High-urgency: 즉시 푸시 + 1분 뒤 전화**를 넣는다(서비스가 아니라 사람마다 설정이다).
 2. `PAGERDUTY_API_TOKEN=… python3 infra/monitoring/alerting/pagerduty.py setup --primary <내 PD 이메일> --secondary <팀원 PD 이메일>` — 원장 `env/prod.json` 이 채워진다 → 커밋.
 3. `python3 infra/monitoring/alerting/p0_p1.py prod` — P0 경보 AlarmActions·OKActions 에 PagerDuty 토픽이 붙는다.
-4. 실수신 시험(G01): `pagerduty.py test-trigger` → 내 폰이 울리면 **ACK 하지 않고** 10분 기다려 팀원 폰이 울리는지 → `pagerduty.py test-resolve`.
+4. 실수신 시험(G01): `pagerduty.py test-trigger` — 시험 경보 `ruleup-prod-P0-ONCALL-TEST` 를 실제 P0 와 같은 경로로 ALARM 으로 바꾼다 → Slack #ruleup-alert·긴급 채널 멘션·PagerDuty 가 함께 오는지, 내 폰을 **ACK 하지 않고** 10분 뒤 팀원 폰이 울리는지 → `pagerduty.py test-resolve`(Slack 복구 알림·사건 종료).
    노션 「대시보드·알림 요약표」 P0 온콜 연결 체크표 12종은 경보별로 `aws cloudwatch set-alarm-state` 로 한 번씩 확인한다(다음 평가에서 OK 로 돌아온다).

infra/monitoring/alerting/pagerduty.py

@@ -6,8 +6,13 @@
     export PAGERDUTY_API_TOKEN=...            # PagerDuty REST API 키(User 또는 General Access, 쓰기). 저장소·원장에 넣지 않는다
     python3 infra/monitoring/alerting/pagerduty.py setup --primary <1차 PD 계정 이메일> --secondary <팀원 PD 계정 이메일>
     python3 infra/monitoring/alerting/p0_p1.py prod     # 원장 sns.pagerduty 가 채워졌으니 P0 경보에 토픽이 붙는다
-    python3 infra/monitoring/alerting/pagerduty.py test-trigger    # 시험 사고 — ACK 하지 말고 10분 뒤 팀원 폰이 울리는지 본다
-    python3 infra/monitoring/alerting/pagerduty.py test-resolve
+    python3 infra/monitoring/alerting/pagerduty.py test-trigger    # 시험 경보 ALARM — Slack 두 채널 + PagerDuty. ACK 하지 말고 10분 뒤 팀원 폰을 본다
+    python3 infra/monitoring/alerting/pagerduty.py test-resolve    # 시험 경보 OK — Slack 복구 알림 + PagerDuty 사건 종료
+
+시험은 실제 P0 와 같은 길을 탄다: 서울에 시험 경보 ruleup-prod-P0-ONCALL-TEST 를 실제 P0 와 같은 AlarmActions 로 두고
+set-alarm-state 로 상태만 바꾼다 → ruleup-prod-alerts(#ruleup-alert)·EventBridge 멘션(긴급 채널)·ruleup-prod-pagerduty 가 함께 받는다.
+시험 경보는 아무도 내지 않는 지표(RuleUp/Observe OnCallTest)를 missing=ignore 로 보므로 다음 평가에서 상태가 되돌아가지 않는다
+(다른 경보에 set-alarm-state 를 쓰면 1분 안에 OK 로 돌아가 사건이 닫혀 10분 에스컬레이션을 볼 수 없다).
 
 setup 이 하는 일
   1. 이메일로 PagerDuty 사용자를 찾는다(두 사람 모두 PagerDuty 계정·앱 설치·전화 알림 규칙이 먼저 있어야 한다).
@@ -26,7 +31,6 @@
 import urllib.error
 import urllib.parse
 import urllib.request
-from datetime import datetime, timezone
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 LEDGER = os.path.join(HERE, '..', 'env', 'prod.json')
@@ -53,14 +57,13 @@ def service(policy_id, ack_timeout_minutes):
             'alert_creation': 'create_alerts_and_incidents'}
 
 
-def cloudwatch_message(account, region, state, now):
-    """CloudWatch 경보가 SNS 로 보내는 본문과 같은 모양 — PagerDuty CloudWatch 연동은 AlarmArn 으로 같은 사건을 묶는다."""
-    return {'AlarmName': TEST_ALARM, 'AlarmDescription': '온콜 시험 — 1차 담당이 ACK 하지 않으면 10분 뒤 팀원에게 간다',
-            'AWSAccountId': account, 'NewStateValue': state, 'OldStateValue': 'OK' if state == 'ALARM' else 'ALARM',
-            'NewStateReason': 'pagerduty.py on-call escalation test', 'StateChangeTime': now.strftime('%Y-%m-%dT%H:%M:%S.000+0000'),
-            'Region': region, 'AlarmArn': f'arn:aws:cloudwatch:{region}:{account}:alarm:{TEST_ALARM}',
-            'Trigger': {'MetricName': 'OnCallTest', 'Namespace': 'RuleUp/Observe', 'Statistic': 'MAXIMUM', 'Period': 60,
-                        'EvaluationPeriods': 1, 'ComparisonOperator': 'GreaterThanOrEqualToThreshold', 'Threshold': 1.0}}
+def test_alarm(reg, actions):
+    """실제 P0 와 같은 전달 경로를 쓰는 시험 경보. 지표는 아무도 내지 않고 missing=ignore — set-alarm-state 로만 상태가 바뀐다."""
+    return {'AlarmName': TEST_ALARM, 'ActionsEnabled': True, 'AlarmActions': actions, 'OKActions': actions,
+            'AlarmDescription': '[ONCALL-TEST] 온콜 시험 경보 — pagerduty.py test-trigger/test-resolve 로만 바뀐다. 실제 장애 아님',
+            'Namespace': 'RuleUp/Observe', 'MetricName': 'OnCallTest', 'Dimensions': [{'Name': 'Env', 'Value': reg['env']}],
+            'Statistic': 'Maximum', 'Period': 60, 'EvaluationPeriods': 1, 'DatapointsToAlarm': 1, 'Threshold': 1,
+            'ComparisonOperator': 'GreaterThanOrEqualToThreshold', 'TreatMissingData': 'ignore'}
 
 
 class PagerDuty:
@@ -158,18 +161,25 @@ def setup(args):
     print('원장 env/prod.json 갱신 — 커밋하고 p0_p1.py prod 를 다시 돌려 P0 경보에 토픽을 붙인다')
 
 
-def publish_test(state):
+def set_test_state(state):
     import boto3
+    sys.path.insert(0, HERE)
+    import p0_p1  # 실제 P0 와 같은 AlarmActions(운영 Slack·긴급 P0 토픽·PagerDuty)
     reg = load()
     if not reg['sns'].get('pagerduty'):
         sys.exit('원장 sns.pagerduty 가 비어 있다 — setup 먼저')
     region = reg['api_region']
-    arn = f"arn:aws:sns:{region}:{reg['account_id']}:{reg['sns']['pagerduty']}"
-    msg = cloudwatch_message(reg['account_id'], region, state, datetime.now(timezone.utc))
-    boto3.client('sns', region_name=region).publish(TopicArn=arn, Subject=f'{state}: "{TEST_ALARM}"', Message=json.dumps(msg, ensure_ascii=False))
-    print(state, TEST_ALARM, '→', arn)
+    cw = boto3.client('cloudwatch', region_name=region)
+    cw.put_metric_alarm(**test_alarm(reg, p0_p1.actions(reg, region, 'P0')))
+    if state == 'ALARM':   # 이미 ALARM 이면 상태 변경이 없어 아무 데도 안 간다 — OK 를 거쳐 다시 울린다
+        current = cw.describe_alarms(AlarmNames=[TEST_ALARM])['MetricAlarms'][0]['StateValue']
+        if current == 'ALARM':
+            sys.exit('시험 경보가 이미 ALARM 이다 — test-resolve 로 닫고 1분 뒤 다시')
+    cw.set_alarm_state(AlarmName=TEST_ALARM, StateValue=state, StateReason=f'pagerduty.py on-call test ({state})')
+    print(state, TEST_ALARM, region, '→', ', '.join(a.split(':')[-1] for a in p0_p1.actions(reg, region, 'P0')))
     if state == 'ALARM':
-        print('ACK 하지 말 것 — 10분 뒤 팀원에게 가는지 확인하고 test-resolve 로 닫는다', time.strftime('%H:%M:%S'))
+        print('Slack #ruleup-alert·긴급 채널 멘션·PagerDuty 가 함께 와야 한다. ACK 하지 말 것 — 10분 뒤 팀원, 확인 후 test-resolve',
+              time.strftime('%H:%M:%S'))
 
 
 if __name__ == '__main__':
@@ -184,4 +194,4 @@ def publish_test(state):
     if a.cmd == 'setup':
         setup(a)
     else:
-        publish_test('ALARM' if a.cmd == 'test-trigger' else 'OK')
+        set_test_state('ALARM' if a.cmd == 'test-trigger' else 'OK')

infra/monitoring/alerting/tests_pagerduty.py

@@ -1,6 +1,5 @@
 """pagerduty.py 계약 시험 — python3 infra/monitoring/alerting/tests_pagerduty.py (PagerDuty·AWS 호출 없음)"""
 import json, os, sys, unittest
-from datetime import datetime, timezone
 HERE = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, HERE)
 import pagerduty as pd
@@ -23,11 +22,17 @@ def test_service_never_auto_resolves_and_always_high(self):
         self.assertEqual(body['acknowledgement_timeout'], 900)
         self.assertEqual(body['incident_urgency_rule'], {'type': 'constant', 'urgency': 'high'})
 
-    def test_test_message_looks_like_cloudwatch(self):
-        m = pd.cloudwatch_message('123', 'ap-northeast-2', 'ALARM', datetime(2026, 10, 15, tzinfo=timezone.utc))
-        self.assertTrue({'AlarmName', 'NewStateValue', 'AlarmArn', 'StateChangeTime', 'AWSAccountId', 'Trigger'} <= set(m))
-        self.assertTrue(m['AlarmName'].startswith('ruleup-prod-P0-'), '긴급 채널 멘션과 같은 접두어')
-        json.dumps(m)
+    def test_test_alarm_follows_real_p0_path(self):
+        """시험 경보는 실제 P0 와 같은 전달(Slack 운영·긴급 멘션·PagerDuty)을 타고, 다음 평가에서 저절로 OK 가 되지 않는다."""
+        import p0_p1
+        reg = pd.load()
+        acts = p0_p1.actions(reg, reg['api_region'], 'P0')
+        a = pd.test_alarm(reg, acts)
+        self.assertTrue(a['AlarmName'].startswith('ruleup-prod-P0-'), 'EventBridge 긴급 채널 멘션이 이 접두어로 잡는다')
+        self.assertEqual(a['TreatMissingData'], 'ignore')
+        self.assertEqual(a['AlarmActions'], a['OKActions'])
+        names = {x.split(':')[-1] for x in a['AlarmActions']}
+        self.assertEqual(names, {reg['sns']['alerts'], reg['sns']['p0'], reg['sns']['pagerduty']})
 
     def test_ledger_keeps_no_secrets(self):
         text = json.dumps(pd.load(), ensure_ascii=False)