diff --git a/apps/api/src/api/v1/metrics.py b/apps/api/src/api/v1/metrics.py index d453cb04f..498be7010 100644 --- a/apps/api/src/api/v1/metrics.py +++ b/apps/api/src/api/v1/metrics.py @@ -1,25 +1,32 @@ """ Metrics API - 黃金指標端點 (Gold Metrics Endpoint) =================================================== +Phase 17 重構: Router 層只做 HTTP 轉發，業務邏輯移至 Service 層 + 統帥鐵律: 禁止假數據！所有指標必須來自 SignOz 真實血脈 Endpoints: - GET /metrics/gold - 獲取 Gold Metrics (RPS, Error Rate, P99, AI Success) +- GET /metrics/health - Metrics 子系統健康檢查 -Data Sources: -- SignOz ClickHouse: RPS, Error Rate, P99 Latency -- SQLite AuditLog: AI Success Rate (executed / total proposals) +架構層次: +- Router (此檔案) → MetricsService → MetricsRepository → PostgreSQL +- Router (此檔案) → MetricsService → SignOzClient → ClickHouse + +版本: v2.0 +更新: 2026-03-26 (台北時區) +更新者: Claude Code (Phase 17 技術債修復 - Router 層違規抽取) +原版: v1.0 by Claude Code """ -from datetime import UTC, datetime, timedelta +from datetime import datetime from typing import Any from fastapi import APIRouter from pydantic import BaseModel from src.core.logging import get_logger -from src.db.base import get_db_context -from src.services.signoz_client import get_signoz_client +from src.services.metrics_service import get_metrics_service logger = get_logger("awoooi.metrics") router = APIRouter() @@ -29,14 +36,17 @@ router = APIRouter() # Response Models # ============================================================================= + class TrendData(BaseModel): """Sparkline 趨勢數據""" + values: list[float] direction: str # up, down, stable class GoldMetricItem(BaseModel): """單一黃金指標""" + label: str value: float | str unit: str | None = None @@ -46,96 +56,18 @@ class GoldMetricItem(BaseModel): class GoldMetricsResponse(BaseModel): """Gold Metrics API Response""" + timestamp: datetime service_name: str metrics: list[GoldMetricItem] raw_data: dict[str, Any] | None = None -# ============================================================================= -# AI Success Rate Calculator -# ============================================================================= - -async def calculate_ai_success_rate(hours: int = 24) -> tuple[float, list[float]]: - """ - 計算 AI 提案成功執行率 - - 統帥鐵律: 若無數據，回傳真實的 0，嚴禁造假 - - Args: - hours: 統計時間範圍 (小時) - - Returns: - (success_rate_percent, trend_values) - """ - try: - async with get_db_context() as session: - from sqlalchemy import text - - # 時間範圍 - cutoff = datetime.now(UTC) - timedelta(hours=hours) - cutoff_str = cutoff.isoformat() - - # Query: 統計 executed vs total (approved + executed + execution_failed) - query = text(""" - SELECT - COUNT(CASE WHEN status = 'executed' THEN 1 END) as executed_count, - COUNT(*) as total_count - FROM approval_records - WHERE created_at >= :cutoff - AND status IN ('approved', 'executed', 'execution_failed') - """) - - result = await session.execute(query, {"cutoff": cutoff_str}) - row = result.fetchone() - - if row and row.total_count > 0: - executed = row.executed_count or 0 - total = row.total_count - success_rate = (executed / total) * 100 - else: - success_rate = 0.0 - - # Trend: 過去 10 個時間點的成功率 (每小時一點) - trend_query = text(""" - SELECT - strftime('%Y-%m-%d %H:00:00', created_at) as hour_bucket, - COUNT(CASE WHEN status = 'executed' THEN 1 END) * 100.0 / - NULLIF(COUNT(*), 0) as hourly_rate - FROM approval_records - WHERE created_at >= :cutoff - AND status IN ('approved', 'executed', 'execution_failed') - GROUP BY hour_bucket - ORDER BY hour_bucket DESC - LIMIT 10 - """) - - trend_result = await session.execute(trend_query, {"cutoff": cutoff_str}) - trend_rows = trend_result.fetchall() - - if trend_rows: - trend_values = [float(r.hourly_rate or 0) for r in reversed(trend_rows)] - else: - trend_values = [0.0] * 10 - - logger.info( - "ai_success_rate_calculated", - success_rate=success_rate, - hours=hours, - ) - - return success_rate, trend_values - - except Exception as e: - logger.warning("ai_success_rate_error", error=str(e)) - # 統帥鐵律: 發生錯誤時回傳真實的 0，非假數據 - return 0.0, [0.0] * 10 - - # ============================================================================= # Endpoints # ============================================================================= + @router.get("/metrics/gold", response_model=GoldMetricsResponse) async def get_gold_metrics( service_name: str = "awoooi-api", @@ -153,102 +85,35 @@ async def get_gold_metrics( GoldMetricsResponse with RPS, Error Rate, P99, AI Success """ logger.info( - "gold_metrics_fetch", + "gold_metrics_request", service=service_name, window_minutes=time_window_minutes, ) - metrics_list: list[GoldMetricItem] = [] - raw_data: dict[str, Any] = {} - - # ========================================================================= - # 1. SignOz Gold Metrics (RPS, Error Rate, P99) - # ========================================================================= - try: - signoz = get_signoz_client() - gold = await signoz.get_gold_metrics( - service_name=service_name, - time_window_minutes=time_window_minutes, - ) - - # RPS - rps_status = "healthy" if gold.rps < 1000 else ("warning" if gold.rps < 5000 else "critical") - rps_trend = [gold.rps * (0.9 + i * 0.02) for i in range(10)] # 模擬趨勢 - metrics_list.append(GoldMetricItem( - label="RPS", - value=round(gold.rps, 1), - unit="req/s", - trend=rps_trend, - status=rps_status, - )) - - # Error Rate - error_status = "healthy" if gold.error_rate < 1 else ("warning" if gold.error_rate < 5 else "critical") - error_trend = [gold.error_rate * (0.95 + i * 0.01) for i in range(10)] - metrics_list.append(GoldMetricItem( - label="Error Rate", - value=round(gold.error_rate, 2), - unit="%", - trend=error_trend, - status=error_status, - )) - - # P99 Latency - p99_status = "healthy" if gold.p99_latency_ms < 200 else ("warning" if gold.p99_latency_ms < 500 else "critical") - p99_trend = [gold.p99_latency_ms * (0.95 + i * 0.01) for i in range(10)] - metrics_list.append(GoldMetricItem( - label="P99 Latency", - value=round(gold.p99_latency_ms, 0), - unit="ms", - trend=p99_trend, - status=p99_status, - )) - - raw_data["signoz"] = { - "rps": gold.rps, - "error_rate": gold.error_rate, - "p99_latency_ms": gold.p99_latency_ms, - "total_requests": gold.total_requests, - "error_count": gold.error_count, - } - - except Exception as e: - logger.warning("signoz_metrics_error", error=str(e)) - # 統帥鐵律: SignOz 斷線時顯示 0，非假數據 - metrics_list.extend([ - GoldMetricItem(label="RPS", value=0, unit="req/s", trend=[0]*10, status="critical"), - GoldMetricItem(label="Error Rate", value=0, unit="%", trend=[0]*10, status="critical"), - GoldMetricItem(label="P99 Latency", value=0, unit="ms", trend=[0]*10, status="critical"), - ]) - raw_data["signoz_error"] = str(e) - - # ========================================================================= - # 2. AI Success Rate (from AuditLog) - # ========================================================================= - ai_success, ai_trend = await calculate_ai_success_rate(hours=24) - ai_status = "healthy" if ai_success >= 90 else ("warning" if ai_success >= 70 else "critical") - - metrics_list.append(GoldMetricItem( - label="AI Success", - value=round(ai_success, 1), - unit="%", - trend=ai_trend, - status=ai_status, - )) - - raw_data["ai_success"] = { - "rate": ai_success, - "hours": 24, - } - - # ========================================================================= - # Response - # ========================================================================= - return GoldMetricsResponse( - timestamp=datetime.now(UTC), + # 透過 Service 取得數據 + service = get_metrics_service() + result = await service.get_gold_metrics( service_name=service_name, + time_window_minutes=time_window_minutes, + ) + + # 轉換為 Response Model + metrics_list = [ + GoldMetricItem( + label=m.label, + value=m.value, + unit=m.unit, + trend=m.trend, + status=m.status, + ) + for m in result.metrics + ] + + return GoldMetricsResponse( + timestamp=result.timestamp, + service_name=result.service_name, metrics=metrics_list, - raw_data=raw_data, + raw_data=result.raw_data, ) @@ -259,17 +124,5 @@ async def metrics_health() -> dict: 快速檢查 SignOz 連線狀態 """ - try: - signoz = get_signoz_client() - # 嘗試執行簡單查詢 - results = await signoz._query_clickhouse("SELECT 1") - clickhouse_ok = len(results) > 0 - except Exception as e: - clickhouse_ok = False - logger.warning("clickhouse_health_check_failed", error=str(e)) - - return { - "status": "healthy" if clickhouse_ok else "degraded", - "clickhouse": "connected" if clickhouse_ok else "disconnected", - "timestamp": datetime.now(UTC).isoformat(), - } + service = get_metrics_service() + return await service.check_health() diff --git a/apps/api/src/repositories/__init__.py b/apps/api/src/repositories/__init__.py index 04c366280..ee6c78490 100644 --- a/apps/api/src/repositories/__init__.py +++ b/apps/api/src/repositories/__init__.py @@ -24,18 +24,26 @@ from src.repositories.incident_repository import ( from src.repositories.interfaces import ( IApprovalRepository, IIncidentRepository, + IMetricsRepository, ITimelineRepository, ) +from src.repositories.metrics_repository import ( + MetricsDBRepository, + get_metrics_repository, +) __all__ = [ # Interfaces "IApprovalRepository", "IIncidentRepository", + "IMetricsRepository", "ITimelineRepository", # Implementations "ApprovalDBRepository", "IncidentDBRepository", + "MetricsDBRepository", # Getters "get_approval_repository", "get_incident_repository", + "get_metrics_repository", ] diff --git a/apps/api/src/repositories/interfaces.py b/apps/api/src/repositories/interfaces.py index 08a1d4fb1..f8e50e318 100644 --- a/apps/api/src/repositories/interfaces.py +++ b/apps/api/src/repositories/interfaces.py @@ -125,3 +125,49 @@ class ITimelineRepository(Protocol): ) -> list[dict]: """取得最近的 Timeline 事件""" ... + + +@runtime_checkable +class IMetricsRepository(Protocol): + """ + Metrics Repository Protocol + + 職責: Metrics 相關 DB 查詢 (AI Success Rate) + 實作: MetricsDBRepository (PostgreSQL) + + 版本: v1.0 + 建立: 2026-03-26 (台北時區) + 建立者: Claude Code (Phase 17 技術債修復) + """ + + async def get_ai_success_rate( + self, + hours: int = 24, + ) -> tuple[float, int, int]: + """ + 計算 AI 提案成功執行率 + + Args: + hours: 統計時間範圍 (小時) + + Returns: + (success_rate_percent, executed_count, total_count) + """ + ... + + async def get_ai_success_trend( + self, + hours: int = 24, + points: int = 10, + ) -> list[float]: + """ + 取得 AI 成功率趨勢 (Sparkline 用) + + Args: + hours: 統計時間範圍 (小時) + points: 趨勢點數量 + + Returns: + list[float]: 每小時成功率列表 (由舊到新) + """ + ... diff --git a/apps/api/src/repositories/metrics_repository.py b/apps/api/src/repositories/metrics_repository.py new file mode 100644 index 000000000..b50207714 --- /dev/null +++ b/apps/api/src/repositories/metrics_repository.py @@ -0,0 +1,186 @@ +""" +Metrics Repository - PostgreSQL 實作 +===================================== +Phase 17 技術債修復: Router 層違規抽取 + +職責: Metrics 相關 DB 查詢 (AI Success Rate 統計) +設計: 實作 IMetricsRepository Protocol + +版本: v1.0 +建立: 2026-03-26 (台北時區) +建立者: Claude Code (Phase 17 技術債修復) +""" + +from datetime import UTC, datetime, timedelta + +import structlog +from sqlalchemy import text + +from src.db.base import get_db_context +from src.repositories.interfaces import IMetricsRepository + +logger = structlog.get_logger(__name__) + + +# ============================================================================= +# MetricsDBRepository +# ============================================================================= + + +class MetricsDBRepository(IMetricsRepository): + """ + Metrics Repository - PostgreSQL 實作 + + 職責: + - AI Success Rate 統計查詢 + - 趨勢數據查詢 (Sparkline) + + 純 CRUD 操作，不含業務邏輯 + 業務邏輯請放在 Service 層 + """ + + async def get_ai_success_rate( + self, + hours: int = 24, + ) -> tuple[float, int, int]: + """ + 計算 AI 提案成功執行率 + + 統帥鐵律: 若無數據，回傳真實的 0，嚴禁造假 + + Args: + hours: 統計時間範圍 (小時) + + Returns: + (success_rate_percent, executed_count, total_count) + """ + try: + async with get_db_context() as session: + cutoff = datetime.now(UTC) - timedelta(hours=hours) + cutoff_str = cutoff.isoformat() + + # Query: 統計 executed vs total (approved + executed + execution_failed) + query = text(""" + SELECT + COUNT(CASE WHEN status = 'executed' THEN 1 END) as executed_count, + COUNT(*) as total_count + FROM approval_records + WHERE created_at >= :cutoff + AND status IN ('approved', 'executed', 'execution_failed') + """) + + result = await session.execute(query, {"cutoff": cutoff_str}) + row = result.fetchone() + + if row and row.total_count > 0: + executed = row.executed_count or 0 + total = row.total_count + success_rate = (executed / total) * 100 + else: + executed = 0 + total = 0 + success_rate = 0.0 + + logger.debug( + "ai_success_rate_queried", + hours=hours, + executed=executed, + total=total, + rate=success_rate, + ) + + return success_rate, executed, total + + except Exception as e: + logger.exception( + "ai_success_rate_query_error", + hours=hours, + error=str(e), + ) + # 統帥鐵律: 發生錯誤時回傳真實的 0，非假數據 + return 0.0, 0, 0 + + async def get_ai_success_trend( + self, + hours: int = 24, + points: int = 10, + ) -> list[float]: + """ + 取得 AI 成功率趨勢 (Sparkline 用) + + 統帥鐵律: 若無數據，回傳真實的 [0.0] * points，嚴禁造假 + + Args: + hours: 統計時間範圍 (小時) + points: 趨勢點數量 + + Returns: + list[float]: 每小時成功率列表 (由舊到新) + """ + try: + async with get_db_context() as session: + cutoff = datetime.now(UTC) - timedelta(hours=hours) + cutoff_str = cutoff.isoformat() + + # Trend: 過去 N 個時間點的成功率 (每小時一點) + trend_query = text(""" + SELECT + strftime('%Y-%m-%d %H:00:00', created_at) as hour_bucket, + COUNT(CASE WHEN status = 'executed' THEN 1 END) * 100.0 / + NULLIF(COUNT(*), 0) as hourly_rate + FROM approval_records + WHERE created_at >= :cutoff + AND status IN ('approved', 'executed', 'execution_failed') + GROUP BY hour_bucket + ORDER BY hour_bucket DESC + LIMIT :limit + """) + + result = await session.execute( + trend_query, + {"cutoff": cutoff_str, "limit": points}, + ) + rows = result.fetchall() + + if rows: + # 由舊到新排列 + trend_values = [float(r.hourly_rate or 0) for r in reversed(rows)] + # 如果點數不足，前面補 0 + if len(trend_values) < points: + trend_values = [0.0] * (points - len(trend_values)) + trend_values + else: + trend_values = [0.0] * points + + logger.debug( + "ai_success_trend_queried", + hours=hours, + points=points, + actual_points=len(trend_values), + ) + + return trend_values + + except Exception as e: + logger.exception( + "ai_success_trend_query_error", + hours=hours, + points=points, + error=str(e), + ) + # 統帥鐵律: 發生錯誤時回傳真實的 0，非假數據 + return [0.0] * points + + +# ============================================================================= +# Singleton +# ============================================================================= + +_metrics_repository: MetricsDBRepository | None = None + + +def get_metrics_repository() -> MetricsDBRepository: + """取得 MetricsDBRepository 實例 (Singleton)""" + global _metrics_repository + if _metrics_repository is None: + _metrics_repository = MetricsDBRepository() + return _metrics_repository diff --git a/apps/api/src/services/metrics_service.py b/apps/api/src/services/metrics_service.py new file mode 100644 index 000000000..c3ef14d72 --- /dev/null +++ b/apps/api/src/services/metrics_service.py @@ -0,0 +1,381 @@ +""" +Metrics Service - 黃金指標服務層 +================================= +Phase 17 技術債修復: Router 層違規抽取 + +職責: +- 聚合 SignOz + DB 數據 +- 計算健康狀態 (healthy/warning/critical) +- 生成趨勢數據 + +設計原則: +- Service 透過 Protocol 依賴 Repository (DI) +- 業務邏輯集中於此層 +- Router 只做 HTTP 轉發 + +版本: v1.0 +建立: 2026-03-26 (台北時區) +建立者: Claude Code (Phase 17 技術債修復) +""" + +from datetime import UTC, datetime +from typing import Any + +import structlog + +from src.repositories.interfaces import IMetricsRepository +from src.repositories.metrics_repository import get_metrics_repository +from src.services.signoz_client import get_signoz_client + +logger = structlog.get_logger(__name__) + + +# ============================================================================= +# Data Classes +# ============================================================================= + + +class GoldMetricResult: + """單一黃金指標結果""" + + def __init__( + self, + label: str, + value: float | str, + unit: str | None, + trend: list[float], + status: str, + ) -> None: + self.label = label + self.value = value + self.unit = unit + self.trend = trend + self.status = status + + +class GoldMetricsResult: + """黃金指標聚合結果""" + + def __init__( + self, + timestamp: datetime, + service_name: str, + metrics: list[GoldMetricResult], + raw_data: dict[str, Any] | None = None, + ) -> None: + self.timestamp = timestamp + self.service_name = service_name + self.metrics = metrics + self.raw_data = raw_data + + +# ============================================================================= +# MetricsService +# ============================================================================= + + +class MetricsService: + """ + 黃金指標服務 + + 職責: + 1. 從 SignOz 取得 RPS, Error Rate, P99 + 2. 從 Repository 取得 AI Success Rate + 3. 計算健康狀態 + 4. 組合成統一的 GoldMetrics 回應 + + 使用方式: + service = MetricsService() + result = await service.get_gold_metrics("awoooi-api", 10) + """ + + def __init__( + self, + metrics_repo: IMetricsRepository | None = None, + ) -> None: + """ + 依賴注入建構函數 + + Args: + metrics_repo: Metrics Repository (預設使用 Singleton) + """ + self._metrics_repo = metrics_repo or get_metrics_repository() + + # ========================================================================= + # Gold Metrics + # ========================================================================= + + async def get_gold_metrics( + self, + service_name: str = "awoooi-api", + time_window_minutes: int = 10, + ) -> GoldMetricsResult: + """ + 獲取黃金指標 (Gold Metrics) + + 統帥鐵律: + - 所有數據必須來自 SignOz 真實血脈 + - AI Success 來自 AuditLog 真實統計 + - 無數據時顯示 0，嚴禁造假 + + Returns: + GoldMetricsResult with RPS, Error Rate, P99, AI Success + """ + logger.info( + "gold_metrics_fetch", + service=service_name, + window_minutes=time_window_minutes, + ) + + metrics_list: list[GoldMetricResult] = [] + raw_data: dict[str, Any] = {} + + # ===================================================================== + # 1. SignOz Gold Metrics (RPS, Error Rate, P99) + # ===================================================================== + signoz_metrics = await self._fetch_signoz_metrics( + service_name, + time_window_minutes, + ) + metrics_list.extend(signoz_metrics["metrics"]) + raw_data.update(signoz_metrics["raw_data"]) + + # ===================================================================== + # 2. AI Success Rate (from Repository) + # ===================================================================== + ai_metrics = await self._fetch_ai_success_metrics(hours=24) + metrics_list.append(ai_metrics["metric"]) + raw_data.update(ai_metrics["raw_data"]) + + return GoldMetricsResult( + timestamp=datetime.now(UTC), + service_name=service_name, + metrics=metrics_list, + raw_data=raw_data, + ) + + async def _fetch_signoz_metrics( + self, + service_name: str, + time_window_minutes: int, + ) -> dict[str, Any]: + """ + 從 SignOz 取得 RPS, Error Rate, P99 + + 統帥鐵律: SignOz 斷線時顯示 0，非假數據 + """ + metrics: list[GoldMetricResult] = [] + raw_data: dict[str, Any] = {} + + try: + signoz = get_signoz_client() + gold = await signoz.get_gold_metrics( + service_name=service_name, + time_window_minutes=time_window_minutes, + ) + + # RPS + rps_status = self._calculate_rps_status(gold.rps) + rps_trend = self._generate_simulated_trend(gold.rps) + metrics.append(GoldMetricResult( + label="RPS", + value=round(gold.rps, 1), + unit="req/s", + trend=rps_trend, + status=rps_status, + )) + + # Error Rate + error_status = self._calculate_error_status(gold.error_rate) + error_trend = self._generate_simulated_trend(gold.error_rate) + metrics.append(GoldMetricResult( + label="Error Rate", + value=round(gold.error_rate, 2), + unit="%", + trend=error_trend, + status=error_status, + )) + + # P99 Latency + p99_status = self._calculate_latency_status(gold.p99_latency_ms) + p99_trend = self._generate_simulated_trend(gold.p99_latency_ms) + metrics.append(GoldMetricResult( + label="P99 Latency", + value=round(gold.p99_latency_ms, 0), + unit="ms", + trend=p99_trend, + status=p99_status, + )) + + raw_data["signoz"] = { + "rps": gold.rps, + "error_rate": gold.error_rate, + "p99_latency_ms": gold.p99_latency_ms, + "total_requests": gold.total_requests, + "error_count": gold.error_count, + } + + except Exception as e: + logger.warning("signoz_metrics_error", error=str(e)) + # 統帥鐵律: SignOz 斷線時顯示 0，非假數據 + metrics.extend([ + GoldMetricResult( + label="RPS", + value=0, + unit="req/s", + trend=[0] * 10, + status="critical", + ), + GoldMetricResult( + label="Error Rate", + value=0, + unit="%", + trend=[0] * 10, + status="critical", + ), + GoldMetricResult( + label="P99 Latency", + value=0, + unit="ms", + trend=[0] * 10, + status="critical", + ), + ]) + raw_data["signoz_error"] = str(e) + + return {"metrics": metrics, "raw_data": raw_data} + + async def _fetch_ai_success_metrics( + self, + hours: int = 24, + ) -> dict[str, Any]: + """ + 從 Repository 取得 AI Success Rate + + 統帥鐵律: 若無數據，回傳真實的 0，嚴禁造假 + """ + # 從 Repository 取得數據 + success_rate, executed, total = await self._metrics_repo.get_ai_success_rate( + hours=hours, + ) + trend = await self._metrics_repo.get_ai_success_trend( + hours=hours, + points=10, + ) + + ai_status = self._calculate_ai_success_status(success_rate) + + metric = GoldMetricResult( + label="AI Success", + value=round(success_rate, 1), + unit="%", + trend=trend, + status=ai_status, + ) + + raw_data = { + "ai_success": { + "rate": success_rate, + "executed": executed, + "total": total, + "hours": hours, + } + } + + logger.info( + "ai_success_rate_calculated", + success_rate=success_rate, + executed=executed, + total=total, + hours=hours, + ) + + return {"metric": metric, "raw_data": raw_data} + + # ========================================================================= + # Health Check + # ========================================================================= + + async def check_health(self) -> dict[str, Any]: + """ + Metrics 子系統健康檢查 + + 快速檢查 SignOz 連線狀態 + """ + try: + signoz = get_signoz_client() + results = await signoz._query_clickhouse("SELECT 1") + clickhouse_ok = len(results) > 0 + except Exception as e: + clickhouse_ok = False + logger.warning("clickhouse_health_check_failed", error=str(e)) + + return { + "status": "healthy" if clickhouse_ok else "degraded", + "clickhouse": "connected" if clickhouse_ok else "disconnected", + "timestamp": datetime.now(UTC).isoformat(), + } + + # ========================================================================= + # Status Calculation Helpers + # ========================================================================= + + @staticmethod + def _calculate_rps_status(rps: float) -> str: + """計算 RPS 健康狀態""" + if rps < 1000: + return "healthy" + if rps < 5000: + return "warning" + return "critical" + + @staticmethod + def _calculate_error_status(error_rate: float) -> str: + """計算 Error Rate 健康狀態""" + if error_rate < 1: + return "healthy" + if error_rate < 5: + return "warning" + return "critical" + + @staticmethod + def _calculate_latency_status(p99_ms: float) -> str: + """計算 P99 Latency 健康狀態""" + if p99_ms < 200: + return "healthy" + if p99_ms < 500: + return "warning" + return "critical" + + @staticmethod + def _calculate_ai_success_status(success_rate: float) -> str: + """計算 AI Success Rate 健康狀態""" + if success_rate >= 90: + return "healthy" + if success_rate >= 70: + return "warning" + return "critical" + + @staticmethod + def _generate_simulated_trend(base_value: float, points: int = 10) -> list[float]: + """ + 生成模擬趨勢數據 (SignOz 不提供歷史數據時使用) + + 注意: 這是暫時方案，未來應從 SignOz 取得真實歷史數據 + """ + return [base_value * (0.9 + i * 0.02) for i in range(points)] + + +# ============================================================================= +# Singleton +# ============================================================================= + +_metrics_service: MetricsService | None = None + + +def get_metrics_service() -> MetricsService: + """取得 MetricsService 實例 (Singleton)""" + global _metrics_service + if _metrics_service is None: + _metrics_service = MetricsService() + return _metrics_service