Files
awoooi/apps/api/src/services/heartbeat_report_service.py
Your Name 5368e64375
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 4m11s
Code Review / ai-code-review (push) Successful in 13s
AI 技術雷達監控 / ai-technology-watch (push) Has started running
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
fix(api): default replay gates to controlled automation
2026-06-28 13:50:23 +08:00

1028 lines
45 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
HeartbeatReportService — ADR-073 心跳監控重構
==============================================
並行收集所有探測 → 彙整判斷 → 一份報告
設計原則:
- 所有探測 asyncio.gather(return_exceptions=True),任一失敗不影響其他
- 只負責收集資料 + 組裝報告,不負責發送
- 超時保護:每個探測最多 8 秒
建立時間: 2026-04-12 (台北時區) ogt
建立者: Claude Sonnet 4.6 — ADR-073 心跳重構
"""
import asyncio
import html
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import httpx
import structlog
from src.core.config import get_settings
from src.utils.timezone import now_taipei
logger = structlog.get_logger(__name__)
settings = get_settings()
_PROBE_TIMEOUT = 8.0 # 每個探測最長 8 秒
@dataclass
class ProbeResult:
ok: bool
status: str # "✅ 正常" / "❌ 失敗: ..." / "⚠️ 警告: ..."
latency_ms: Optional[float] = None
@dataclass
class FlywheelStats:
playbook_count: int = 0
success_24h: int = 0
attempt_24h: int = 0
km_total: int = 0
km_vectorized: int = 0
last_learning_at: Optional[datetime] = None
@dataclass
class AlertPipelineStats:
total_24h: int = 0
auto_resolved_24h: int = 0
pending_approval: int = 0
pending_actionable: int = 0
pending_observe_only: int = 0
pending_without_telegram: int = 0
execution_success_24h: int = 0
execution_failed_24h: int = 0
@dataclass
class DbRedisStats:
db_ok: bool = False
db_status: str = "❌ 未查詢"
redis_ok: bool = False
redis_status: str = "❌ 未查詢"
redis_key_count: int = 0
@dataclass
class PodInfo:
name: str
ready: bool
status: str
restarts: int = 0
# 2026-05-03 Claude Opus 4.7 + 統帥 ogtP0 #3 K8s pod state machine 完整化
# 加 start_time 才能判斷 Pending/NotReady 是「剛起來合理」還是「卡住該告警」
start_time: Optional[str] = None # ISO 8601 from .status.startTime
@dataclass
class ScannerStats:
# key = scanner name, value = last run ISO string or None
last_runs: dict[str, Optional[str]] = field(default_factory=dict)
@dataclass
class TelegramBotStats:
polling_ok: bool = False
status: str = "❌ 未查詢"
last_callback_ago_min: Optional[float] = None
@dataclass
class AutomationStats:
"""自動化六大能力今日統計2026-04-24 ogt: Task 3 — 告警自動化可觀測性)"""
# 自動規則生成
auto_rule_generated_today: int = 0
# 自動審核拒絕(按原因分組)
reject_counts: dict[str, int] = field(default_factory=dict)
# 今日 KM 新增
km_created_today: int = 0
# Config Drift 自動採納(今日)
drift_adopted_today: int = 0
@dataclass
class HeartbeatReport:
timestamp: datetime
ai_services: dict[str, ProbeResult] = field(default_factory=dict)
ollama_models: dict[str, bool] = field(default_factory=dict)
ollama_endpoints: dict[str, ProbeResult] = field(default_factory=dict)
mcp_providers: dict[str, ProbeResult] = field(default_factory=dict)
flywheel: FlywheelStats = field(default_factory=FlywheelStats)
infra: dict[str, ProbeResult] = field(default_factory=dict)
warnings: list[str] = field(default_factory=list)
# 2026-04-22 新增動態區塊
alert_pipeline: AlertPipelineStats = field(default_factory=AlertPipelineStats)
db_redis: DbRedisStats = field(default_factory=DbRedisStats)
pods: list[PodInfo] = field(default_factory=list)
scanners: ScannerStats = field(default_factory=ScannerStats)
telegram_bot: TelegramBotStats = field(default_factory=TelegramBotStats)
# 2026-04-24 自動化統計
automation: AutomationStats = field(default_factory=AutomationStats)
@property
def has_warnings(self) -> bool:
return len(self.warnings) > 0
class HeartbeatReportService:
"""
心跳報告收集服務
使用方式:
report = await HeartbeatReportService().collect()
text = report_to_telegram_html(report)
"""
async def collect(self) -> HeartbeatReport:
"""並行收集所有探測,彙整為一份報告"""
report = HeartbeatReport(timestamp=now_taipei())
results = await asyncio.gather(
self._probe_ollama(),
self._probe_nemotron(),
self._probe_gemini(),
self._probe_claude(),
self._probe_mcp_k8s(),
self._probe_mcp_ssh(),
self._probe_mcp_argocd(),
self._probe_mcp_sentry(),
self._probe_argocd_sync(),
self._probe_velero(),
self._get_flywheel_stats(),
# 2026-04-22 新增動態探測
self._get_alert_pipeline_stats(),
self._probe_db_redis(),
self._get_pod_status(),
self._get_scanner_stats(),
self._probe_telegram_bot(),
# 2026-04-24 自動化統計
self._get_automation_stats(),
return_exceptions=True,
)
keys = [
"_ollama", "_nemotron", "_gemini", "_claude",
"_mcp_k8s", "_mcp_ssh", "_mcp_argocd", "_mcp_sentry",
"_argocd_sync", "_velero", "_flywheel",
"_alert_pipeline", "_db_redis", "_pods", "_scanners", "_tg_bot",
"_automation",
]
collected: dict = {}
for key, result in zip(keys, results):
if isinstance(result, Exception):
logger.warning("heartbeat_probe_error", probe=key, error=str(result))
collected[key] = None
else:
collected[key] = result
# --- AI 服務 ---
ollama_data = collected["_ollama"] or {}
report.ai_services["ollama"] = ollama_data.get("probe", ProbeResult(False, "❌ 無回應"))
report.ollama_models = ollama_data.get("models", {})
report.ollama_endpoints = ollama_data.get("endpoints", {})
report.ai_services["nemotron"] = collected["_nemotron"] or ProbeResult(False, "❌ 無回應")
report.ai_services["gemini"] = collected["_gemini"] or ProbeResult(False, "❌ 無回應")
report.ai_services["claude"] = collected["_claude"] or ProbeResult(False, "❌ 無回應")
# --- MCP Provider ---
report.mcp_providers["k8s"] = collected["_mcp_k8s"] or ProbeResult(False, "❌ 無回應")
report.mcp_providers["ssh"] = collected["_mcp_ssh"] or ProbeResult(False, "❌ 無回應")
report.mcp_providers["argocd"] = collected["_mcp_argocd"] or ProbeResult(False, "❌ 無回應")
report.mcp_providers["sentry"] = collected["_mcp_sentry"] or ProbeResult(False, "❌ 無回應")
# --- 基礎設施 ---
report.infra["argocd_sync"] = collected["_argocd_sync"] or ProbeResult(False, "❌ 無回應")
report.infra["velero"] = collected["_velero"] or ProbeResult(False, "❌ 無回應")
# --- 飛輪統計 ---
if collected["_flywheel"]:
report.flywheel = collected["_flywheel"]
# --- 新動態區塊 ---
if collected["_alert_pipeline"]:
report.alert_pipeline = collected["_alert_pipeline"]
if collected["_db_redis"]:
report.db_redis = collected["_db_redis"]
if collected["_pods"]:
report.pods = collected["_pods"]
if collected["_scanners"]:
report.scanners = collected["_scanners"]
if collected["_tg_bot"]:
report.telegram_bot = collected["_tg_bot"]
if collected["_automation"]:
report.automation = collected["_automation"]
# --- 彙整 warnings ---
report.warnings = self._build_warnings(report)
return report
# =========================================================================
# 探測方法
# =========================================================================
async def _probe_ollama(self) -> dict:
"""探測 Ollama 服務 + 逐一確認所需模型"""
endpoints = [
("GCP-A", settings.OLLAMA_URL),
("GCP-B", getattr(settings, "OLLAMA_SECONDARY_URL", "")),
("111", getattr(settings, "OLLAMA_FALLBACK_URL", "")),
]
async def _probe_endpoint(
client: httpx.AsyncClient,
label: str,
url: str,
) -> tuple[str, ProbeResult, set[str]]:
if not url:
return label, ProbeResult(False, "⚠️ 未設定"), set()
try:
t0 = asyncio.get_event_loop().time()
resp = await client.get(f"{url}/api/tags")
latency = (asyncio.get_event_loop().time() - t0) * 1000
if resp.status_code != 200:
return label, ProbeResult(False, f"❌ HTTP {resp.status_code}", latency), set()
available = {m["name"] for m in resp.json().get("models", [])}
return label, ProbeResult(True, "✅ 正常", round(latency, 1)), available
except Exception as e:
return label, ProbeResult(False, f"{str(e)[:60]}"), set()
async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
results = await asyncio.gather(
*[_probe_endpoint(client, label, url) for label, url in endpoints],
)
endpoint_status = {label: probe for label, probe, _available in results}
primary_probe = endpoint_status.get("GCP-A", ProbeResult(False, "❌ 無回應"))
primary_available = next(
(available for label, _probe, available in results if label == "GCP-A"),
set(),
)
if primary_probe.ok:
# 也把 short name無 :tag加進去方便匹配
available_short = {n.split(":")[0] for n in primary_available}
model_status: dict[str, bool] = {}
for required in settings.OLLAMA_REQUIRED_MODELS:
req_short = required.split(":")[0]
ok = required in primary_available or req_short in available_short
model_status[required] = ok
return {
"probe": primary_probe,
"models": model_status,
"endpoints": endpoint_status,
}
return {
"probe": primary_probe,
"models": {},
"endpoints": endpoint_status,
}
async def _probe_nemotron(self) -> ProbeResult:
"""探測 Nemotron NIM API"""
if not settings.NVIDIA_API_KEY:
return ProbeResult(False, "⚠️ NVIDIA_API_KEY 未設定")
try:
async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
t0 = asyncio.get_event_loop().time()
resp = await client.get(
"https://integrate.api.nvidia.com/v1/models",
headers={"Authorization": f"Bearer {settings.NVIDIA_API_KEY}"},
)
latency = (asyncio.get_event_loop().time() - t0) * 1000
if resp.status_code == 200:
return ProbeResult(True, "✅ 正常", round(latency, 1))
return ProbeResult(False, f"❌ HTTP {resp.status_code}")
except Exception as e:
return ProbeResult(False, f"{str(e)[:60]}")
async def _probe_gemini(self) -> ProbeResult:
"""探測 Gemini API只確認 key 有設定 + 能連線)"""
if not settings.GEMINI_API_KEY:
return ProbeResult(False, "⚠️ GEMINI_API_KEY 未設定")
try:
async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
t0 = asyncio.get_event_loop().time()
resp = await client.get(
"https://generativelanguage.googleapis.com/v1beta/models",
headers={"x-goog-api-key": settings.GEMINI_API_KEY},
)
latency = (asyncio.get_event_loop().time() - t0) * 1000
if resp.status_code == 200:
return ProbeResult(True, "✅ 正常", round(latency, 1))
return ProbeResult(False, f"❌ HTTP {resp.status_code}")
except Exception as e:
return ProbeResult(False, f"{str(e)[:60]}")
async def _probe_claude(self) -> ProbeResult:
"""探測 Claude API"""
if not settings.CLAUDE_API_KEY:
return ProbeResult(False, "⚠️ CLAUDE_API_KEY 未設定")
try:
async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
t0 = asyncio.get_event_loop().time()
resp = await client.get(
"https://api.anthropic.com/v1/models",
headers={
"x-api-key": settings.CLAUDE_API_KEY,
"anthropic-version": "2023-06-01",
},
)
latency = (asyncio.get_event_loop().time() - t0) * 1000
if resp.status_code == 200:
return ProbeResult(True, "✅ 正常", round(latency, 1))
return ProbeResult(False, f"❌ HTTP {resp.status_code}")
except Exception as e:
return ProbeResult(False, f"{str(e)[:60]}")
async def _probe_mcp_k8s(self) -> ProbeResult:
"""K8s MCP確認 kubectl 能連到 K3s"""
try:
from src.plugins.mcp.providers.k8s_provider import K8sProvider
from src.plugins.mcp.registry import AuditedMCPToolProvider
from src.services.mcp_audit_context import with_mcp_audit_context
provider = AuditedMCPToolProvider(K8sProvider())
if not provider.enabled:
return ProbeResult(False, "⚠️ K8s MCP 未啟用")
params = with_mcp_audit_context(
{"resource_type": "nodes"},
session_id="heartbeat:mcp_k8s",
flywheel_node="govern",
agent_role="heartbeat_report_service",
gateway_path="legacy_heartbeat_provider",
)
result = await asyncio.wait_for(
provider.execute("kubectl_get", params),
timeout=_PROBE_TIMEOUT,
)
if result.success:
return ProbeResult(True, "✅ 正常")
return ProbeResult(False, f"⚠️ {result.error[:60] if result.error else '查詢失敗'}")
except asyncio.TimeoutError:
return ProbeResult(False, "❌ 超時")
except Exception as e:
return ProbeResult(False, f"{str(e)[:60]}")
async def _probe_mcp_ssh(self) -> ProbeResult:
"""SSH MCP確認設定是否完整"""
if not settings.SSH_MCP_ENABLED:
return ProbeResult(False, "⚠️ SSH_MCP_ENABLED=false")
# 確認 ssh-mcp-key Secret 是否掛載
import os
key_path = "/run/secrets/ssh_mcp_key"
if not os.path.exists(key_path):
return ProbeResult(False, "⚠️ ssh-mcp-key 未注入 K8s Secret")
return ProbeResult(True, "✅ 正常")
async def _probe_mcp_argocd(self) -> ProbeResult:
"""ArgoCD MCP確認 token 設定"""
if not settings.ARGOCD_MCP_ENABLED:
return ProbeResult(False, "⚠️ ARGOCD_MCP_ENABLED=false")
if not settings.ARGOCD_API_TOKEN:
return ProbeResult(False, "⚠️ ARGOCD_API_TOKEN 未設定")
return ProbeResult(True, "✅ 設定完整")
async def _probe_mcp_sentry(self) -> ProbeResult:
"""Sentry MCP確認設定"""
if not settings.SENTRY_MCP_ENABLED:
return ProbeResult(False, "⚠️ SENTRY_MCP_ENABLED=false")
# 確認 SENTRY_AUTH_TOKEN
sentry_token = getattr(settings, "SENTRY_AUTH_TOKEN", "") or ""
if not sentry_token:
return ProbeResult(False, "⚠️ SENTRY_AUTH_TOKEN 未設定")
return ProbeResult(True, "✅ 設定完整")
async def _probe_argocd_sync(self) -> ProbeResult:
"""ArgoCD 應用同步狀態"""
if not settings.ARGOCD_API_TOKEN:
return ProbeResult(False, "⚠️ 未設定 Token無法查詢")
try:
async with httpx.AsyncClient(
timeout=_PROBE_TIMEOUT,
verify=False, # ArgoCD self-signed cert
) as client:
resp = await client.get(
f"{settings.ARGOCD_URL}/api/v1/applications/awoooi-prod",
headers={"Authorization": f"Bearer {settings.ARGOCD_API_TOKEN}"},
)
if resp.status_code != 200:
return ProbeResult(False, f"❌ HTTP {resp.status_code}")
data = resp.json()
sync_status = data.get("status", {}).get("sync", {}).get("status", "Unknown")
health_status = data.get("status", {}).get("health", {}).get("status", "Unknown")
if sync_status == "Synced" and health_status == "Healthy":
return ProbeResult(True, "✅ Synced + Healthy")
return ProbeResult(False, f"⚠️ {sync_status} / {health_status}")
except Exception as e:
return ProbeResult(False, f"{str(e)[:60]}")
async def _probe_velero(self) -> ProbeResult:
"""Velero 備份:確認最後一次備份是否在 26 小時內"""
try:
from src.plugins.mcp.providers.k8s_provider import K8sProvider
from src.plugins.mcp.registry import AuditedMCPToolProvider
from src.services.mcp_audit_context import with_mcp_audit_context
provider = AuditedMCPToolProvider(K8sProvider())
if not provider.enabled:
return ProbeResult(False, "⚠️ K8s MCP 未啟用,無法查 Velero")
params = with_mcp_audit_context(
{
"resource_type": "backups.velero.io",
"namespace": "velero",
},
session_id="heartbeat:velero",
flywheel_node="govern",
agent_role="heartbeat_report_service",
gateway_path="legacy_heartbeat_provider",
)
result = await asyncio.wait_for(
provider.execute("kubectl_get", params),
timeout=_PROBE_TIMEOUT,
)
if not result.success:
return ProbeResult(False, "⚠️ 無法查詢 Velero 備份")
return ProbeResult(True, "✅ 可查詢")
except Exception as e:
return ProbeResult(False, f"{str(e)[:60]}")
async def _get_flywheel_stats(self) -> FlywheelStats:
"""查詢飛輪核心統計數字"""
stats = FlywheelStats()
try:
# Playbook 數量(從 Redis SCAN
from src.core.redis_client import get_redis
redis = get_redis()
keys = await redis.keys("playbook:*")
stats.playbook_count = len(keys)
except Exception as e:
logger.debug("heartbeat_playbook_count_failed", error=str(e))
try:
# KM 向量化率DB 查詢)
from sqlalchemy import func, select
from sqlalchemy import text as sa_text
from src.db.base import get_db_context
from src.db.models import KnowledgeEntryRecord
async with get_db_context() as db:
# KM 總數
km_total = await db.scalar(select(func.count()).select_from(KnowledgeEntryRecord))
stats.km_total = km_total or 0
# KM 向量化數embedding IS NOT NULL
# KnowledgeEntryRecord ORM 無 embedding 欄位,改用 raw SQL
from sqlalchemy import text as sa_text
vec_result = await db.execute(
sa_text("SELECT COUNT(*) FROM knowledge_entries WHERE embedding IS NOT NULL")
)
stats.km_vectorized = vec_result.scalar() or 0
# 24h 修復統計
# 2026-05-06 ogt + Codex:
# incidents.outcome 已不是自動修復 source of truth。實際執行紀錄
# 寫在 auto_repair_executions舊查詢會讓心跳報告顯示 0/15
# 造成「全系統正常」但飛輪 KPI 失真的假象。
repair_result = await db.execute(
sa_text("""
SELECT
COUNT(*) FILTER (WHERE success IS TRUE) AS success,
COUNT(*) AS total
FROM auto_repair_executions
WHERE created_at >= NOW() - interval '24 hours'
""")
)
repair_row = repair_result.one()
stats.success_24h = int(repair_row.success or 0)
stats.attempt_24h = int(repair_row.total or 0)
# 最後學習活動
last_km = await db.scalar(
select(func.max(KnowledgeEntryRecord.created_at))
)
if last_km:
stats.last_learning_at = last_km
except Exception as e:
logger.debug("heartbeat_flywheel_stats_failed", error=str(e))
return stats
# =========================================================================
# 2026-04-22 新增動態探測方法
# =========================================================================
async def _get_alert_pipeline_stats(self) -> AlertPipelineStats:
"""查 24h 告警流水線統計approval_records"""
stats = AlertPipelineStats()
try:
from sqlalchemy import text as sa_text
from src.db.base import get_db_context
async with get_db_context() as db:
r = await db.execute(sa_text("""
WITH scoped AS (
SELECT
*,
(
COALESCE(extra_metadata->>'execution_kind', '') = 'no_action'
OR COALESCE(extra_metadata->>'repair_executed', '') = 'false'
OR btrim(coalesce(action, '')) = ''
OR UPPER(action) LIKE 'OBSERVE%'
OR UPPER(action) LIKE 'INVESTIGATE%'
OR UPPER(action) LIKE 'NO_ACTION%'
OR UPPER(action) LIKE '% NO_ACTION%'
OR UPPER(action) LIKE '%| NO_ACTION%'
) AS is_observe_only
FROM approval_records
WHERE created_at >= NOW() - interval '24 hours'
)
SELECT
COUNT(*) AS total,
COUNT(*) FILTER (WHERE UPPER(status::text) = 'PENDING') AS pending,
COUNT(*) FILTER (
WHERE UPPER(status::text) = 'PENDING'
AND NOT is_observe_only
) AS pending_actionable,
COUNT(*) FILTER (
WHERE UPPER(status::text) = 'PENDING'
AND is_observe_only
) AS pending_observe_only,
COUNT(*) FILTER (
WHERE UPPER(status::text) = 'PENDING'
AND telegram_message_id IS NULL
) AS pending_without_telegram,
COUNT(*) FILTER (
WHERE UPPER(status::text) = 'EXECUTION_SUCCESS'
AND NOT is_observe_only
) AS success,
COUNT(*) FILTER (
WHERE UPPER(status::text) = 'EXECUTION_FAILED'
AND NOT is_observe_only
) AS failed,
COUNT(*) FILTER (
WHERE UPPER(status::text) IN ('APPROVED','EXECUTION_SUCCESS','EXECUTION_FAILED')
AND NOT is_observe_only
) AS auto_resolved
FROM scoped
"""))
row = r.one()
stats.total_24h = int(row.total or 0)
stats.pending_approval = int(row.pending or 0)
stats.pending_actionable = int(row.pending_actionable or 0)
stats.pending_observe_only = int(row.pending_observe_only or 0)
stats.pending_without_telegram = int(row.pending_without_telegram or 0)
stats.execution_success_24h = int(row.success or 0)
stats.execution_failed_24h = int(row.failed or 0)
stats.auto_resolved_24h = int(row.auto_resolved or 0)
except Exception as e:
logger.debug("heartbeat_alert_pipeline_failed", error=str(e))
return stats
async def _probe_db_redis(self) -> DbRedisStats:
"""探測 PostgreSQL 與 Redis 連線健康"""
s = DbRedisStats()
try:
from sqlalchemy import text as sa_text
from src.db.base import get_db_context
async with get_db_context() as db:
await db.execute(sa_text("SELECT 1"))
s.db_ok = True
s.db_status = "✅ 正常"
except Exception as e:
s.db_status = f"{str(e)[:40]}"
try:
from src.core.redis_client import get_redis
redis = get_redis()
info = await redis.info("memory")
used_mb = int(info.get("used_memory", 0)) // (1024 * 1024)
all_keys = await redis.dbsize()
s.redis_ok = True
s.redis_key_count = all_keys
s.redis_status = f"✅ 正常 {used_mb}MB / {all_keys} keys"
except Exception as e:
s.redis_status = f"{str(e)[:40]}"
return s
async def _get_pod_status(self) -> list[PodInfo]:
"""查 awoooi-prod namespace 的所有 Pod 狀態
2026-05-03 Claude Opus 4.7 + 統帥 ogtP0 #3 加抓 STARTTIME 才能做 K8s state machine 判斷
"""
pods: list[PodInfo] = []
try:
import subprocess
r = subprocess.run(
["kubectl", "-n", "awoooi-prod", "get", "pods",
"--no-headers", "-o",
"custom-columns=NAME:.metadata.name,READY:.status.containerStatuses[0].ready,"
"STATUS:.status.phase,RESTARTS:.status.containerStatuses[0].restartCount,"
"STARTTIME:.status.startTime"],
capture_output=True, text=True, timeout=8,
)
for line in r.stdout.strip().splitlines():
parts = line.split()
if len(parts) >= 3:
name = parts[0]
ready = parts[1].lower() == "true"
status = parts[2]
restarts = int(parts[3]) if len(parts) >= 4 and parts[3].isdigit() else 0
start_time = parts[4] if len(parts) >= 5 and parts[4] != "<none>" else None
pods.append(PodInfo(
name=name, ready=ready, status=status,
restarts=restarts, start_time=start_time,
))
except Exception as e:
logger.debug("heartbeat_pod_status_failed", error=str(e))
return pods
async def _get_scanner_stats(self) -> ScannerStats:
"""查各 scanner 最後執行時間Redis daily lock key TTL 反推)"""
stats = ScannerStats()
scanner_names = [
"capacity_forecaster", "hermes_rule_quality",
"compliance_scanner", "coverage_evaluator", "daily_report",
]
try:
from src.core.redis_client import get_redis
from src.utils.timezone import now_taipei as _now
redis = get_redis()
today = _now().date().isoformat()
for name in scanner_names:
key = f"aiops:daily_lock:{name}:{today}"
ttl = await redis.ttl(key)
if ttl > 0:
# TTL=25h 時剛跑完;剩餘 TTL 推算跑完時間
ran_at_sec = 25 * 3600 - ttl
h, m = divmod(ran_at_sec // 60, 60)
stats.last_runs[name] = f"今日 {h:02d}:{m:02d}"
else:
stats.last_runs[name] = None # 今日尚未執行
except Exception as e:
logger.debug("heartbeat_scanner_stats_failed", error=str(e))
return stats
async def _probe_telegram_bot(self) -> TelegramBotStats:
"""探測 Telegram Bot polling 狀態"""
s = TelegramBotStats()
try:
from src.core.redis_client import get_redis
redis = get_redis()
# polling leader lock
leader = await redis.get("telegram:polling:leader")
if leader:
s.polling_ok = True
s.status = f"✅ Polling 活躍 (leader: {leader.decode()[:20] if isinstance(leader, bytes) else str(leader)[:20]})"
else:
# 嘗試查最近 callback 時間tg_msg: key 存在即有活動)
keys = await redis.keys("tg_msg:*")
if keys:
s.polling_ok = True
s.status = f"✅ 有活動 ({len(keys)} msg keys)"
else:
s.status = "⚠️ 無 polling leader key可能重啟中"
except Exception as e:
s.status = f"{str(e)[:40]}"
return s
async def _get_automation_stats(self) -> AutomationStats:
"""查自動化六大能力今日統計Redis 計數器 + DB
2026-04-24 ogt: Task 3 — 告警自動化可觀測性建設
資料來源:
- stats:auto_rule_generated:{today} — 自動規則生成計數alert_rule_engine.py 寫入)
- stats:auto_approve_rejected:{reason}:{today} — 拒絕自動審核原因分布
- knowledge_entries 今日新增DB
- approval_records drift_adopted 今日DB
"""
stats = AutomationStats()
try:
from src.core.redis_client import get_redis
redis = get_redis()
today = now_taipei().strftime("%Y%m%d")
# 自動規則生成今日計數
raw = await redis.get(f"stats:auto_rule_generated:{today}")
stats.auto_rule_generated_today = int(raw) if raw else 0
# 自動審核拒絕原因分布
reject_keys = await redis.keys(f"stats:auto_approve_rejected:*:{today}")
for key in reject_keys:
key_str = key.decode() if isinstance(key, bytes) else key
# key 格式stats:auto_approve_rejected:{reason}:{today}
parts = key_str.split(":")
reason_part = parts[2] if len(parts) >= 4 else key_str
count_raw = await redis.get(key_str)
stats.reject_counts[reason_part] = int(count_raw) if count_raw else 0
except Exception as e:
logger.debug("heartbeat_automation_redis_failed", error=str(e))
try:
from sqlalchemy import text as sa_text
from src.db.base import get_db_context
async with get_db_context() as db:
# 今日新增 KMtimestamptz 直接比較,不需 AT TIME ZONE
km_today = await db.scalar(sa_text(
"SELECT COUNT(*) FROM knowledge_entries "
"WHERE created_at >= NOW() - interval '24 hours'"
))
stats.km_created_today = int(km_today or 0)
# 今日 Config Drift 自動採納(查 drift_reportsadopt() 在此表更新 status
drift_today = await db.scalar(sa_text(
"SELECT COUNT(*) FROM drift_reports "
"WHERE status = 'adopted' "
"AND resolved_at >= NOW() - interval '24 hours'"
))
stats.drift_adopted_today = int(drift_today or 0)
except Exception as e:
logger.debug("heartbeat_automation_db_failed", error=str(e))
return stats
# =========================================================================
# Warnings 彙整
# =========================================================================
def _build_warnings(self, report: HeartbeatReport) -> list[str]:
warnings: list[str] = []
# Ollama 模型未載入
for model, loaded in report.ollama_models.items():
if not loaded:
warnings.append(f"{model} 未載入,相關功能失效")
for name, probe in report.ollama_endpoints.items():
if not probe.ok and not probe.status.startswith("⚠️ 未設定"):
warnings.append(f"Ollama {name} 異常: {probe.status}")
# AI 服務異常
for name, probe in report.ai_services.items():
if not probe.ok and not probe.status.startswith("⚠️"):
warnings.append(f"{name} 服務異常: {probe.status}")
# MCP 設定問題
for name, probe in report.mcp_providers.items():
if not probe.ok:
warnings.append(f"MCP {name}: {probe.status}")
# ArgoCD 非 Synced+Healthy
argocd = report.infra.get("argocd_sync")
if argocd and not argocd.ok:
warnings.append(f"ArgoCD: {argocd.status}")
# 飛輪警告
if report.flywheel.playbook_count == 0:
warnings.append("Playbook 數量為 0飛輪學習無法啟動")
if report.flywheel.km_total > 0:
vec_rate = report.flywheel.km_vectorized / report.flywheel.km_total
if vec_rate < 0.8:
warnings.append(
f"KM 向量化率偏低: {report.flywheel.km_vectorized}/{report.flywheel.km_total}"
f" ({int(vec_rate*100)}%)"
)
# 24h 沉默(無學習活動)— 2h 太短,正常無事故期間必然誤報
if report.flywheel.last_learning_at:
silence_hours = (now_taipei() - report.flywheel.last_learning_at.replace(
tzinfo=report.timestamp.tzinfo if report.timestamp.tzinfo else None
)).total_seconds() / 3600
if silence_hours > 24:
warnings.append(f"系統沉默 {silence_hours:.1f}h無學習活動")
# DB / Redis 異常
if not report.db_redis.db_ok:
warnings.append(f"PostgreSQL: {report.db_redis.db_status}")
if not report.db_redis.redis_ok:
warnings.append(f"Redis: {report.db_redis.redis_status}")
# Pending 積壓告警:只用可執行/有風險受控補證計數觸發,避免 OBSERVE/NO_ACTION 觀察卡造成假待辦。
if report.alert_pipeline.pending_actionable > 10:
warnings.append(
f"AI 受控隊列待補證 {report.alert_pipeline.pending_actionable}"
f"(前台 /awooop/approvals觀察類 {report.alert_pipeline.pending_observe_only} 筆另列)"
)
# Pod 異常 — 2026-05-03 Claude Opus 4.7 + 統帥 ogtP0 #3 完整 K8s pod state machine
# K8s pod phases (https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/):
# Pending — 已建立但容器還沒起(短暫 OK>5min 異常 = image pull / scheduling 卡)
# Running — 至少 1 容器跑中ready=False 短暫 OK>2min 異常 = probe fail
# Succeeded — 全容器成功結束CronJob/Job 正常,不算未就緒)
# Failed — 全容器結束,至少 1 fail必告警
# Unknown — 狀態無法取得(必告警)
from datetime import datetime, timezone
_now = datetime.now(timezone.utc)
_PENDING_THRESHOLD_MIN = 5
_NOT_READY_THRESHOLD_MIN = 2
def _age_minutes(start_time: Optional[str]) -> Optional[float]:
"""ISO 8601 startTime → 距今分鐘。None 或解析失敗返 None。"""
if not start_time:
return None
try:
# K8s startTime 格式2026-05-03T12:34:56Z
dt = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
return (_now - dt).total_seconds() / 60.0
except (ValueError, TypeError):
return None
for pod in report.pods:
phase = pod.status
age_min = _age_minutes(pod.start_time)
# restart 次數高無論 phase 都告警CrashLoop 中或跑完都值得追)
# 放最前面,避免後面 continue 跳過
if pod.restarts >= 3:
warnings.append(f"Pod {pod.name} 重啟 {pod.restarts}")
if phase in ("Succeeded", "Completed"):
# CronJob/Job 成功跑完ready=False 是設計phase 部分不算未就緒
continue
elif phase == "Failed":
# 真正失敗 — 一定告警
warnings.append(f"Pod {pod.name} Failed")
elif phase == "Unknown":
warnings.append(f"Pod {pod.name} 狀態 Unknown")
elif phase == "Pending":
# 短暫 Pending OK持續 >5min 表示 image pull / scheduling 卡住
if age_min is None or age_min >= _PENDING_THRESHOLD_MIN:
age_str = f"{age_min:.0f}m" if age_min else "未知"
warnings.append(f"Pod {pod.name} 持續 Pending {age_str}image pull / scheduling 卡住)")
elif phase == "Running" and not pod.ready:
# Running 但 not ready短暫 OK剛起>2min 表示 probe fail / 啟動慢
if age_min is None or age_min >= _NOT_READY_THRESHOLD_MIN:
age_str = f"{age_min:.0f}m" if age_min else "未知"
warnings.append(f"Pod {pod.name} NotReady {age_str}readiness probe fail / 啟動異常)")
# Running + ready=True 是健康狀態,跳過
return warnings
def report_to_telegram_html(report: HeartbeatReport) -> str:
"""
將 HeartbeatReport 轉換為 Telegram HTML 格式
ADR-075 TYPE-1 格式 (2026-04-12 ogt):
💚 INFO | AWOOOI 系統報告 + ├─/└─ 樹狀結構
"""
ts = report.timestamp.strftime("%Y-%m-%d %H:%M (台北)")
overall_ok = not report.warnings
header_icon = "💚" if overall_ok else "⚠️"
header_label = "全系統正常" if overall_ok else f"需關注 {len(report.warnings)}"
lines = [
f"{header_icon} <b>INFO | AWOOOI 系統報告</b>",
f"{ts}",
"──────────────────────",
"",
]
# --- AI 服務 ---
ollama = report.ai_services.get("ollama", ProbeResult(False, ""))
ollama_lat = f" {ollama.latency_ms:.0f}ms" if ollama.latency_ms else ""
models_ok = [m.split(":")[0] for m, ok in report.ollama_models.items() if ok]
models_str = " / ".join(models_ok) if models_ok else "無模型"
nem = report.ai_services.get("nemotron", ProbeResult(False, ""))
gem = report.ai_services.get("gemini", ProbeResult(False, ""))
cla = report.ai_services.get("claude", ProbeResult(False, ""))
lines.append("🤖 <b>AI 服務</b>")
lines.append(f"├─ Ollama: {ollama.status}{ollama_lat} <code>{html.escape(models_str)}</code>")
if report.ollama_endpoints:
endpoint_items = list(report.ollama_endpoints.items())
for idx, (name, probe) in enumerate(endpoint_items):
branch = "" if idx == len(endpoint_items) - 1 else ""
latency = f" {probe.latency_ms:.0f}ms" if probe.latency_ms else ""
lines.append(f"{branch}{html.escape(name)}: {probe.status}{latency}")
lines.append(f"├─ Nemotron NIM: {nem.status}" + (f" {nem.latency_ms:.0f}ms" if nem.latency_ms else ""))
lines.append(f"├─ Gemini API: {gem.status}" + (f" {gem.latency_ms:.0f}ms" if gem.latency_ms else ""))
lines.append(f"└─ Claude API: {cla.status}" + (f" {cla.latency_ms:.0f}ms" if cla.latency_ms else ""))
lines.append("")
# --- MCP Provider ---
k8s = report.mcp_providers.get("k8s", ProbeResult(False, ""))
ssh = report.mcp_providers.get("ssh", ProbeResult(False, ""))
argocd_mcp = report.mcp_providers.get("argocd", ProbeResult(False, ""))
sentry_mcp = report.mcp_providers.get("sentry", ProbeResult(False, ""))
lines.append("🔌 <b>MCP Provider</b>")
lines.append(f"├─ K8s: {k8s.status} SSH: {ssh.status}")
lines.append(f"└─ ArgoCD: {argocd_mcp.status} Sentry: {sentry_mcp.status}")
lines.append("")
# --- 飛輪狀態 ---
fw = report.flywheel
if fw.attempt_24h > 0:
rate = int(fw.success_24h / fw.attempt_24h * 100)
repair_str = f"{fw.success_24h}/{fw.attempt_24h} ({rate}%)"
else:
repair_str = "0 次"
km_str = ""
if fw.km_total > 0:
vec_rate = int(fw.km_vectorized / fw.km_total * 100)
km_icon = "" if vec_rate >= 90 else "⚠️"
km_str = f"KM: {km_icon} {fw.km_vectorized}/{fw.km_total} ({vec_rate}%)"
learn_str = f" 學習: {fw.last_learning_at.strftime('%H:%M')}" if fw.last_learning_at else ""
lines.append("🔄 <b>飛輪狀態24h</b>")
lines.append(f"├─ Playbooks: {fw.playbook_count} 修復: {repair_str}")
lines.append(f"└─ {km_str}{learn_str}" if km_str or learn_str else "└─ KM 統計不可用")
lines.append("")
# --- 基礎設施 ---
argocd = report.infra.get("argocd_sync", ProbeResult(False, ""))
velero = report.infra.get("velero", ProbeResult(False, ""))
lines.append("🚀 <b>基礎設施</b>")
lines.append(f"├─ ArgoCD: {argocd.status}")
lines.append(f"└─ Velero: {velero.status}")
# --- 告警流水線 ---
ap = report.alert_pipeline
lines.append("")
lines.append("📊 <b>告警流水線24h</b>")
lines.append(f"├─ 總計: {ap.total_24h} PENDING: {ap.pending_approval}")
lines.append(
f"├─ 受控拆分: 補證 {ap.pending_actionable} 觀察 {ap.pending_observe_only}"
f" 無TG {ap.pending_without_telegram}"
)
if ap.execution_success_24h > 0 and ap.execution_failed_24h == 0:
exec_icon = ""
elif ap.execution_failed_24h > 0:
exec_icon = "⚠️"
else:
exec_icon = ""
lines.append(f"└─ 執行: {exec_icon} 成功 {ap.execution_success_24h} 失敗 {ap.execution_failed_24h}")
# --- DB & Redis ---
dr = report.db_redis
lines.append("")
lines.append("🗄️ <b>資料庫 & Redis</b>")
lines.append(f"├─ PostgreSQL: {dr.db_status}")
lines.append(f"└─ Redis: {dr.redis_status} Keys: {dr.redis_key_count}")
# --- K8s Pods ---
if report.pods:
lines.append("")
lines.append("☸️ <b>Kubernetes Pods</b>")
for i, pod in enumerate(report.pods):
prefix = "└─" if i == len(report.pods) - 1 else "├─"
ready_icon = "" if pod.ready or pod.status in ("Succeeded", "Completed") else ""
restart_str = f" (重啟×{pod.restarts})" if pod.restarts > 0 else ""
status_str = "" if pod.ready else f" <code>{html.escape(pod.status)}</code>"
lines.append(f"{prefix} {ready_icon} {html.escape(pod.name[:35])}{restart_str}{status_str}")
# --- Scanner 狀態 ---
if report.scanners.last_runs:
lines.append("")
lines.append("⏱️ <b>Scanner 狀態(今日)</b>")
scanner_items = list(report.scanners.last_runs.items())
for i, (name, ran_at) in enumerate(scanner_items):
prefix = "└─" if i == len(scanner_items) - 1 else "├─"
icon = "" if ran_at else "⏸️"
ran_str = ran_at or "尚未執行"
lines.append(f"{prefix} {icon} {html.escape(name)}: {ran_str}")
# --- Telegram Bot ---
tg = report.telegram_bot
lines.append("")
lines.append("🤖 <b>Telegram Bot</b>")
lines.append(f"└─ {tg.status}")
# --- 自動化統計2026-04-24---
auto = report.automation
lines.append("")
lines.append("⚙️ <b>自動化統計(今日)</b>")
lines.append(f"├─ 新規則自動生成: {auto.auto_rule_generated_today}")
lines.append(f"├─ KM 新增: {auto.km_created_today}")
lines.append(f"├─ Drift 自動採納: {auto.drift_adopted_today}")
if auto.reject_counts:
reject_total = sum(auto.reject_counts.values())
top_reason = max(auto.reject_counts, key=auto.reject_counts.get) # type: ignore[arg-type]
lines.append(
f"└─ 受控補證攔截: {reject_total} 次 主因: <code>{html.escape(top_reason)}</code>"
)
else:
lines.append("└─ 受控補證攔截: 0 次")
# --- Warnings / 總結 ---
lines.append("")
if report.warnings:
lines.append(f"⚠️ <b>需關注({len(report.warnings)} 項)</b>")
for w in report.warnings[:-1]:
lines.append(f"├─ {html.escape(w)}")
lines.append(f"└─ {html.escape(report.warnings[-1])}")
else:
lines.append(f"✅ <b>{header_label}</b>")
return "\n".join(lines)