アラート疲労の定量分析と段階的治療：通知を「減らす」のではなく「構造化する」

1
Stage 1: 「通知が来た！すぐ確認しよう」
2
Stage 2: 「また同じやつか。後で見よう」
3
Stage 3: 「通知多すぎ。チャンネルをミュートしよう」
4
Stage 4: 「本当に重要なアラートも見逃す」

1
総アラート数:  247件/週
2
  - 即時対応が必要:   3件（1.2%）
3
  - 当日中に確認:    12件（4.9%）
4
  - 情報として有用:   45件（18.2%）
5
  - 無意味（ノイズ）: 187件（75.7%）

1
p0_criteria:
2
  - データ処理パイプライン全停止
3
  - 直近24時間の出力がゼロ
4
  - ディスク使用率 > 95%

1
p1_criteria:
2
  - 特定ジョブの連続失敗（3回以上）
3
  - データ鮮度が閾値を超過（自動復旧なし）
4
  - 外部API連携のエラー率 > 10%

1
p2_criteria:
2
  - 閾値の一時的な超過（自動復旧あり）
3
  - パフォーマンス低下（処理時間が通常の2倍以上）
4
  - 非クリティカルなジョブの失敗

1
log_only:
2
  - 正常実行の記録
3
  - 閾値内の変動
4
  - 自動復旧した一時的なエラー

1
from enum import Enum
2
from dataclasses import dataclass
3

4
class Severity(Enum):
5
    P0 = "p0"  # 即座に対応
6
    P1 = "p1"  # 当日中に対応
7
    P2 = "p2"  # 週次レビュー
8
    LOG = "log" # ログのみ
9

10
@dataclass
11
class Alert:
12
    job_name: str
13
    message: str
14
    severity: Severity
15
    timestamp: str
18 collapsed lines
16
    details: dict
17

18
def route_alert(alert: Alert):
19
    """重要度に応じて通知先を振り分ける"""
20
    # 全アラートをログに記録
21
    log_alert(alert)
22

23
    match alert.severity:
24
        case Severity.P0:
25
            send_sms(alert)
26
            send_slack_dm(alert)
27
            log_alert(alert)
28
        case Severity.P1:
29
            send_slack_dm(alert)
30
        case Severity.P2:
31
            send_slack_channel(alert, channel="#alerts-low")
32
        case Severity.LOG:
33
            pass  # ログのみ、通知なし

1
def classify_severity(job_name: str, error_type: str, context: dict) -> Severity:
2
    """エラーの種類とコンテキストから重要度を自動判定"""
3

4
    # P0: 全面停止系
5
    if context.get("all_jobs_affected"):
6
        return Severity.P0
7
    if context.get("zero_output_last_24h"):
8
        return Severity.P0
9
    if context.get("disk_usage_percent", 0) > 95:
10
        return Severity.P0
11

12
    # P1: 連続失敗、長時間のデータ欠損
13
    if context.get("consecutive_failures", 0) >= 3:
14
        return Severity.P1
15
    if context.get("data_age_hours", 0) > context.get("max_age_hours", 24) * 1.5:
10 collapsed lines
16
        return Severity.P1
17

18
    # P2: 一時的な問題
19
    if context.get("auto_recovered"):
20
        return Severity.P2
21
    if error_type in ("timeout", "rate_limit", "temporary_network"):
22
        return Severity.P2
23

24
    # デフォルト: P2
25
    return Severity.P2

1
from collections import defaultdict
2
from datetime import datetime, timedelta
3

4
# 5分間のウィンドウでアラートを集約
5
AGGREGATION_WINDOW = timedelta(minutes=5)
6
recent_alerts: dict[str, list[Alert]] = defaultdict(list)
7

8
def should_send(alert: Alert) -> bool:
9
    """同一カテゴリのアラートが短期間に集中していたら集約する"""
10
    key = f"{alert.severity.value}:{alert.job_name}"
11
    now = datetime.now()
12

13
    # 古いアラートを削除
14
    recent_alerts[key] = [
15
        a for a in recent_alerts[key]
12 collapsed lines
16
        if (now - datetime.fromisoformat(a.timestamp)) < AGGREGATION_WINDOW
17
    ]
18

19
    if len(recent_alerts[key]) >= 3:
20
        # 3件目以降は集約通知（「他N件の同様のアラート」）
21
        if len(recent_alerts[key]) == 3:
22
            send_summary(key, recent_alerts[key])
23
        recent_alerts[key].append(alert)
24
        return False
25

26
    recent_alerts[key].append(alert)
27
    return True

1
def detect_cascade(alerts: list[Alert]) -> str | None:
2
    """5分以内に3つ以上のジョブがアラートを出したら連鎖と判定"""
3
    if len(alerts) < 3:
4
        return None
5

6
    job_names = {a.job_name for a in alerts}
7
    if len(job_names) >= 3:
8
        # 依存グラフから共通の上流を特定
9
        common_upstream = find_common_upstream(job_names)
10
        if common_upstream:
11
            return f"根本原因の候補: {common_upstream} の障害が下流に波及"
12
    return None

1
def generate_weekly_report(alerts: list[Alert]) -> str:
2
    """週次アラートレポートを生成"""
3
    total = len(alerts)
4
    by_severity = Counter(a.severity.value for a in alerts)
5
    by_job = Counter(a.job_name for a in alerts).most_common(5)
6

7
    report = f"""
8
## 週次アラートレポート
9

10
**総数**: {total}件
11
**内訳**: P0={by_severity.get('p0', 0)}, P1={by_severity.get('p1', 0)}, P2={by_severity.get('p2', 0)}
12

13
### アラート頻度 Top 5（要改善候補）
14
"""
15
    for job, count in by_job:
3 collapsed lines
16
        report += f"- {job}: {count}件\n"
17

18
    return report