From e724f57adb6296b1f9f28f9e8a063b4828638445 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 11 Oct 2026 18:01:39 +0900 Subject: [PATCH] feat(runtime-guard): NOTIFY-04 B5 alert de-duplication, dry_run by default Same failure entries are re-reported by up to six 30-min runs (180-min lookback). New entries / new issue types always alert; pure repeats are suppressed only in mode=on (re-alert after 6h). Workflow default mode=dry_run logs would_suppress and never suppresses. State: hashes + timestamps via actions/cache. --- .github/workflows/runtime-guard.yml | 19 +++ scripts/cloud_run_runtime_guard.py | 14 ++ scripts/runtime_guard_alert_dedupe.py | 202 +++++++++++++++++++++++ tests/test_runtime_guard_alert_dedupe.py | 119 +++++++++++++ 4 files changed, 354 insertions(+) create mode 100644 scripts/runtime_guard_alert_dedupe.py create mode 100644 tests/test_runtime_guard_alert_dedupe.py diff --git a/.github/workflows/runtime-guard.yml b/.github/workflows/runtime-guard.yml index bf6479d..b94a3f2 100644 --- a/.github/workflows/runtime-guard.yml +++ b/.github/workflows/runtime-guard.yml @@ -59,6 +59,10 @@ jobs: GLOBAL_TELEGRAM_CHAT_ID: ${{ secrets.GLOBAL_TELEGRAM_CHAT_ID }} TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }} TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }} + # NOTIFY-04 B5: default dry_run = log would_suppress only, never suppress. + RUNTIME_GUARD_ALERT_DEDUPE_MODE: ${{ vars.RUNTIME_GUARD_ALERT_DEDUPE_MODE || 'dry_run' }} + RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS: ${{ vars.RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS || '6' }} + RUNTIME_GUARD_DEDUPE_STATE_PATH: .runtime-guard-dedupe/state.json steps: - name: Checkout repository uses: actions/checkout@v6 @@ -98,5 +102,20 @@ jobs: - name: Set up gcloud uses: google-github-actions/setup-gcloud@v3 + - name: Restore runtime-guard alert dedupe state + uses: actions/cache/restore@v4 + with: + path: .runtime-guard-dedupe + key: runtime-guard-dedupe-${{ github.run_id }}-${{ github.run_attempt }} + restore-keys: | + runtime-guard-dedupe- + - name: Check Cloud Scheduler and Cloud Run logs run: uv run --no-sync python scripts/cloud_run_runtime_guard.py + + - name: Save runtime-guard alert dedupe state + if: ${{ always() && hashFiles('.runtime-guard-dedupe/state.json') != '' }} + uses: actions/cache/save@v4 + with: + path: .runtime-guard-dedupe + key: runtime-guard-dedupe-${{ github.run_id }}-${{ github.run_attempt }} diff --git a/scripts/cloud_run_runtime_guard.py b/scripts/cloud_run_runtime_guard.py index c8b9aa9..ee3df41 100644 --- a/scripts/cloud_run_runtime_guard.py +++ b/scripts/cloud_run_runtime_guard.py @@ -20,6 +20,11 @@ resolve_operational_notification_locale, ) +try: + from scripts import runtime_guard_alert_dedupe as alert_dedupe +except ModuleNotFoundError: # executed as `python scripts/cloud_run_runtime_guard.py` + import runtime_guard_alert_dedupe as alert_dedupe # type: ignore[no-redef] + ERROR_SEVERITIES = {"ERROR", "CRITICAL", "ALERT", "EMERGENCY"} FAILURE_WORDS = ( @@ -673,6 +678,7 @@ def main() -> int: issues: list[str] = [] details: list[str] = [] + alert_event_ids: list[str] = [] success_count = 0 success_count_by_service: dict[str, int] = {} queried_services: set[str] = set() @@ -716,6 +722,7 @@ def main() -> int: ) ) details.extend(_summarize(entry) for entry in failures[:5]) + alert_event_ids.extend(alert_dedupe.event_id(entry) for entry in failures) if services and require_success: for service in _services_without_success( @@ -764,6 +771,7 @@ def main() -> int: ) ) details.extend(_summarize(entry) for entry in failures[:5]) + alert_event_ids.extend(alert_dedupe.event_id(entry) for entry in failures) except RuntimeError as exc: issues.append(_notice("runtime_guard_scheduler_log_query_failed")) details.append(f"Cloud Scheduler log query failed: {exc}") @@ -771,6 +779,7 @@ def main() -> int: print("Skipping Cloud Scheduler check because no scheduler job pattern could be derived.", file=sys.stderr) if not issues: + alert_dedupe.clear_issues() service_text = ", ".join(services) if services else "" print( f"Runtime guard OK for {name}: services={service_text}, lookback={lookback_minutes}m, successes={success_count}" @@ -793,6 +802,11 @@ def main() -> int: workflow_url=run_url, ) print(message) + decision = alert_dedupe.decide(name=name, issues=issues, event_ids=alert_event_ids) + print(decision.log_line()) + if not decision.send: + print(f"Runtime guard alert suppressed as a repeat for {name}; no new failure entries.") + return 1 if fail_workflow else 0 _send_telegram(message[:3900]) return 1 if fail_workflow else 0 diff --git a/scripts/runtime_guard_alert_dedupe.py b/scripts/runtime_guard_alert_dedupe.py new file mode 100644 index 0000000..d23919d --- /dev/null +++ b/scripts/runtime_guard_alert_dedupe.py @@ -0,0 +1,202 @@ +"""Runtime-guard alert de-duplication (NOTIFY-04 B5). + +The guard runs every 30 minutes with a 180-minute log lookback, so one Cloud +Run / Cloud Scheduler failure entry is re-reported by up to six consecutive +runs. This module decides whether an alert is a pure repeat: + +* new failure log entries (by event id) → always alert; +* a new issue type (normalized issue text fingerprint) → always alert; +* the same issues with no new entries → repeat; re-alert after + ``RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS`` (default 6). + +Modes (``RUNTIME_GUARD_ALERT_DEDUPE_MODE``): + +* ``off`` (default when unset): no state, behaviour unchanged; +* ``dry_run``: never suppresses; logs ``would_suppress`` and keeps state as if + ``on`` were active, so the effect can be observed before switching; +* ``on``: suppresses pure repeats. + +State is a small JSON file (hashes and timestamps only, no log text, no +account data) persisted between runs by the workflow cache step. +""" + +from __future__ import annotations + +import datetime as dt +import hashlib +import json +import os +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Mapping, Sequence + +MODES = ("off", "dry_run", "on") +DEFAULT_STATE_PATH = ".runtime-guard-dedupe/state.json" +DEFAULT_REMINDER_HOURS = 6.0 +EVENT_RETENTION_HOURS = 48.0 +STATE_VERSION = 1 + + +@dataclass(frozen=True) +class DedupeDecision: + mode: str + send: bool + would_suppress: bool + new_event_count: int + new_issue_fingerprint: bool + reason: str + + def log_line(self) -> str: + return ( + "runtime_guard_dedupe " + f"mode={self.mode} send={str(self.send).lower()} " + f"would_suppress={str(self.would_suppress).lower()} " + f"new_events={self.new_event_count} " + f"new_issue_type={str(self.new_issue_fingerprint).lower()} reason={self.reason}" + ) + + +def resolve_mode(raw: str | None = None) -> str: + value = str(raw if raw is not None else os.environ.get("RUNTIME_GUARD_ALERT_DEDUPE_MODE") or "").strip().lower() + return value if value in MODES else "off" + + +def _state_path(raw: str | None = None) -> Path: + return Path(raw or os.environ.get("RUNTIME_GUARD_DEDUPE_STATE_PATH") or DEFAULT_STATE_PATH) + + +def _reminder_hours() -> float: + try: + value = float(os.environ.get("RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS") or DEFAULT_REMINDER_HOURS) + except ValueError: + return DEFAULT_REMINDER_HOURS + return value if value > 0 else DEFAULT_REMINDER_HOURS + + +def event_id(entry: Mapping[str, Any]) -> str: + """Stable id for one log entry (Cloud Logging insertId preferred).""" + + insert_id = str(entry.get("insertId") or "").strip() + if insert_id: + basis = f"insert:{insert_id}" + else: + resource = entry.get("resource") if isinstance(entry.get("resource"), Mapping) else {} + basis = json.dumps( + { + "timestamp": entry.get("timestamp"), + "logName": entry.get("logName"), + "resource": resource.get("labels") if isinstance(resource, Mapping) else None, + "severity": entry.get("severity"), + }, + sort_keys=True, + default=str, + ) + return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:24] + + +def issue_fingerprint(name: str, issues: Sequence[str]) -> str: + """Issue-type fingerprint; counts/minutes are normalized so N→N+1 is the same type.""" + + normalized = sorted({re.sub(r"\d+", "#", str(item)).strip() for item in issues}) + basis = json.dumps({"name": name, "issues": normalized}, ensure_ascii=False) + return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:24] + + +def _parse(ts: Any) -> dt.datetime | None: + try: + value = dt.datetime.fromisoformat(str(ts)) + except (TypeError, ValueError): + return None + return value if value.tzinfo else value.replace(tzinfo=dt.timezone.utc) + + +def _load(path: Path) -> dict[str, Any]: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {"version": STATE_VERSION, "events": {}, "issues": {}} + if not isinstance(payload, dict) or payload.get("version") != STATE_VERSION: + return {"version": STATE_VERSION, "events": {}, "issues": {}} + payload.setdefault("events", {}) + payload.setdefault("issues", {}) + return payload + + +def _save(path: Path, state: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(state, sort_keys=True), encoding="utf-8") + + +def _prune(state: dict[str, Any], now: dt.datetime) -> None: + horizon = now - dt.timedelta(hours=EVENT_RETENTION_HOURS) + state["events"] = { + key: value + for key, value in dict(state.get("events") or {}).items() + if (_parse(value) or now) >= horizon + } + + +def decide( + *, + name: str, + issues: Sequence[str], + event_ids: Iterable[str], + now: dt.datetime | None = None, + mode: str | None = None, + state_path: str | None = None, +) -> DedupeDecision: + """Decide whether to send this alert and record it in the state.""" + + resolved = resolve_mode(mode) + ids = sorted({str(item) for item in event_ids if str(item)}) + if resolved == "off": + return DedupeDecision(resolved, True, False, len(ids), True, "dedupe_off") + stamp = now or dt.datetime.now(dt.timezone.utc) + path = _state_path(state_path) + state = _load(path) + _prune(state, stamp) + events: dict[str, str] = state["events"] + issues_state: dict[str, str] = state["issues"] + + new_ids = [item for item in ids if item not in events] + fingerprint = issue_fingerprint(name, issues) + last_alert = _parse(issues_state.get(fingerprint)) + new_type = last_alert is None + reminder_due = ( + last_alert is not None + and stamp - last_alert >= dt.timedelta(hours=_reminder_hours()) + ) + if new_ids: + repeat, reason = False, "new_failure_events" + elif new_type: + repeat, reason = False, "new_issue_type" + elif reminder_due: + repeat, reason = False, "reminder_due" + else: + repeat, reason = True, "repeat_within_reminder_window" + + if not repeat: + for item in ids: + events.setdefault(item, stamp.isoformat()) + issues_state[fingerprint] = stamp.isoformat() + else: + for item in ids: + events.setdefault(item, stamp.isoformat()) + _save(path, state) + + send = not (repeat and resolved == "on") + return DedupeDecision(resolved, send, repeat, len(new_ids), new_type, reason) + + +def clear_issues(*, mode: str | None = None, state_path: str | None = None, now: dt.datetime | None = None) -> None: + """Healthy run: forget issue fingerprints so a recurrence alerts immediately.""" + + if resolve_mode(mode) == "off": + return + stamp = now or dt.datetime.now(dt.timezone.utc) + path = _state_path(state_path) + state = _load(path) + _prune(state, stamp) + state["issues"] = {} + _save(path, state) diff --git a/tests/test_runtime_guard_alert_dedupe.py b/tests/test_runtime_guard_alert_dedupe.py new file mode 100644 index 0000000..2902881 --- /dev/null +++ b/tests/test_runtime_guard_alert_dedupe.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import datetime as dt +import importlib.util +import json +import sys +from pathlib import Path + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "runtime_guard_alert_dedupe.py" +spec = importlib.util.spec_from_file_location("runtime_guard_alert_dedupe_for_test", SCRIPT) +dedupe = importlib.util.module_from_spec(spec) +assert spec.loader is not None +sys.modules[spec.name] = dedupe +spec.loader.exec_module(dedupe) + +T0 = dt.datetime(2026, 10, 11, 8, 0, tzinfo=dt.timezone.utc) +ISSUES = ["Cloud Run 失败日志 2 条：svc"] + + +def _decide(tmp_path, mode, *, issues=ISSUES, ids=("a", "b"), at=T0): + return dedupe.decide( + name="Guard", + issues=issues, + event_ids=ids, + now=at, + mode=mode, + state_path=str(tmp_path / "state.json"), + ) + + +def test_off_mode_always_sends_and_writes_no_state(tmp_path): + assert _decide(tmp_path, "off").send is True + assert _decide(tmp_path, "off").send is True + assert not (tmp_path / "state.json").exists() + + +def test_unknown_mode_falls_back_to_off(tmp_path, monkeypatch): + monkeypatch.delenv("RUNTIME_GUARD_ALERT_DEDUPE_MODE", raising=False) + assert dedupe.resolve_mode("bogus") == "off" + assert dedupe.resolve_mode(None) == "off" + + +def test_on_mode_suppresses_repeat_of_same_entries(tmp_path): + first = _decide(tmp_path, "on") + assert first.send is True and first.reason == "new_failure_events" + second = _decide(tmp_path, "on", at=T0 + dt.timedelta(minutes=30)) + assert second.send is False and second.would_suppress is True + assert second.reason == "repeat_within_reminder_window" + + +def test_new_failure_entry_always_alerts(tmp_path): + _decide(tmp_path, "on") + later = _decide( + tmp_path, + "on", + issues=["Cloud Run 失败日志 3 条：svc"], + ids=("a", "b", "c"), + at=T0 + dt.timedelta(minutes=30), + ) + assert later.send is True and later.new_event_count == 1 + + +def test_new_issue_type_alerts_even_without_entries(tmp_path): + _decide(tmp_path, "on") + other = _decide( + tmp_path, + "on", + issues=ISSUES + ["Scheduler 日志查询失败"], + ids=("a", "b"), + at=T0 + dt.timedelta(minutes=30), + ) + assert other.send is True and other.reason == "new_issue_type" + + +def test_count_change_alone_is_same_issue_type(tmp_path): + assert dedupe.issue_fingerprint("G", ["失败 2 条"]) == dedupe.issue_fingerprint("G", ["失败 5 条"]) + + +def test_reminder_after_window(tmp_path, monkeypatch): + monkeypatch.setenv("RUNTIME_GUARD_ALERT_DEDUPE_REMINDER_HOURS", "6") + _decide(tmp_path, "on", ids=()) + assert _decide(tmp_path, "on", ids=(), at=T0 + dt.timedelta(hours=5)).send is False + assert _decide(tmp_path, "on", ids=(), at=T0 + dt.timedelta(hours=6)).reason == "reminder_due" + + +def test_recovery_clears_issue_state_so_recurrence_alerts(tmp_path): + _decide(tmp_path, "on", ids=()) + dedupe.clear_issues(mode="on", state_path=str(tmp_path / "state.json"), now=T0 + dt.timedelta(minutes=30)) + again = _decide(tmp_path, "on", ids=(), at=T0 + dt.timedelta(minutes=60)) + assert again.send is True and again.reason == "new_issue_type" + + +def test_dry_run_never_suppresses_but_reports_would_suppress(tmp_path): + _decide(tmp_path, "dry_run") + second = _decide(tmp_path, "dry_run", at=T0 + dt.timedelta(minutes=30)) + assert second.send is True + assert second.would_suppress is True + assert "would_suppress=true" in second.log_line() + + +def test_state_holds_hashes_only(tmp_path): + entry = {"insertId": "abc123", "textPayload": "secret-ish log text U1234567"} + eid = dedupe.event_id(entry) + _decide(tmp_path, "on", ids=(eid,)) + raw = (tmp_path / "state.json").read_text(encoding="utf-8") + assert "abc123" not in raw and "U1234567" not in raw and "svc" not in raw + assert set(json.loads(raw)) == {"version", "events", "issues"} + + +def test_old_events_are_pruned(tmp_path): + _decide(tmp_path, "on", ids=("old",)) + _decide(tmp_path, "on", ids=("new",), at=T0 + dt.timedelta(hours=49)) + state = json.loads((tmp_path / "state.json").read_text(encoding="utf-8")) + assert len(state["events"]) == 1 + + +def test_event_id_without_insert_id_is_stable(): + entry = {"timestamp": "2026-10-11T08:00:00Z", "logName": "x", "resource": {"labels": {"service_name": "s"}}} + assert dedupe.event_id(entry) == dedupe.event_id(dict(entry))