diff --git a/apps/api/src/repositories/interfaces.py b/apps/api/src/repositories/interfaces.py index 6bf03fe0c..4b2e5dad2 100644 --- a/apps/api/src/repositories/interfaces.py +++ b/apps/api/src/repositories/interfaces.py @@ -371,3 +371,67 @@ class IEmbeddingCacheRepository(Protocol): async def remove(self, playbook_id: str) -> bool: """移除 Playbook 向量""" ... + + +@runtime_checkable +class IK8sRepository(Protocol): + """ + K8s Repository Protocol + + 職責: K8s API 操作 (Pods/Deployments/Events) + 實作: K8sRepository (kubernetes_asyncio) + + 版本: v1.0 + 建立: 2026-03-30 (台北時區) + 建立者: Claude Code (首席架構師 P1 改進) + + 設計原則: + - Service 層不直接呼叫 kubernetes_asyncio + - 透過 Repository 進行 K8s 操作 + - 符合 leWOOOgo 積木化原則 + """ + + async def list_pods( + self, + namespace: str = "awoooi", + ) -> list[dict]: + """ + 列出 Namespace 下的 Pods + + Returns: + list[{name, phase, ready, restarts, age}] + """ + ... + + async def list_deployments( + self, + namespace: str = "awoooi", + ) -> list[dict]: + """ + 列出 Namespace 下的 Deployments + + Returns: + list[{name, ready_replicas, replicas, available}] + """ + ... + + async def get_pod_status_summary( + self, + namespace: str = "awoooi", + ) -> dict: + """ + 取得 Pod 狀態摘要 + + Returns: + {total, running, pending, failed, problem_pods: [...]} + """ + ... + + async def is_available(self) -> bool: + """ + 檢查 K8s API 是否可用 + + Returns: + True 如果 K8s 連線正常 + """ + ... diff --git a/apps/api/src/repositories/k8s_repository.py b/apps/api/src/repositories/k8s_repository.py new file mode 100644 index 000000000..978bb7827 --- /dev/null +++ b/apps/api/src/repositories/k8s_repository.py @@ -0,0 +1,231 @@ +""" +K8s Repository - 首席架構師 P1 改進 +=================================== +實作 IK8sRepository Protocol + +設計原則: +- 封裝 kubernetes_asyncio 操作 +- Service 層不直接依賴 K8s client +- 符合 leWOOOgo 積木化原則 + +版本: v1.0 +建立: 2026-03-30 (台北時區) +建立者: Claude Code (首席架構師 P1 改進) +""" + +from datetime import UTC, datetime + +import structlog + +from src.core.config import settings + +logger = structlog.get_logger(__name__) + +# Lazy import kubernetes_asyncio to avoid import errors when not installed +_k8s_client = None +_k8s_config_loaded = False + + +async def _get_k8s_client(): + """Lazy load kubernetes client (內部用)""" + global _k8s_client, _k8s_config_loaded + + if _k8s_client is not None: + return _k8s_client + + try: + from kubernetes_asyncio import client, config + + if not _k8s_config_loaded: + try: + # 優先使用 in-cluster 配置 + config.load_incluster_config() + logger.info("k8s_repository_incluster_config") + except config.ConfigException: + # Fallback 到 kubeconfig + await config.load_kube_config(config_file=settings.KUBECONFIG_PATH) + logger.info("k8s_repository_kubeconfig", path=settings.KUBECONFIG_PATH) + _k8s_config_loaded = True + + _k8s_client = client + return _k8s_client + + except Exception as e: + logger.error("k8s_repository_init_failed", error=str(e)) + return None + + +class K8sRepository: + """ + K8s Repository 實作 + + 實作 IK8sRepository Protocol + 封裝 kubernetes_asyncio 操作 + """ + + async def list_pods( + self, + namespace: str = "awoooi", + ) -> list[dict]: + """ + 列出 Namespace 下的 Pods + + Returns: + list[{name, phase, ready, restarts, age}] + """ + client = await _get_k8s_client() + if not client: + logger.warning("k8s_list_pods_unavailable") + return [] + + try: + v1 = client.CoreV1Api() + pods = await v1.list_namespaced_pod(namespace=namespace) + + result = [] + now = datetime.now(UTC) + + for pod in pods.items: + # 計算 restarts + restarts = 0 + ready = True + if pod.status.container_statuses: + for cs in pod.status.container_statuses: + restarts += cs.restart_count + if not cs.ready: + ready = False + + # 計算 age + age_seconds = 0 + if pod.metadata.creation_timestamp: + age_seconds = (now - pod.metadata.creation_timestamp.replace(tzinfo=UTC)).total_seconds() + + result.append({ + "name": pod.metadata.name, + "phase": pod.status.phase, + "ready": ready, + "restarts": restarts, + "age_seconds": int(age_seconds), + }) + + return result + + except Exception as e: + logger.error("k8s_list_pods_failed", namespace=namespace, error=str(e)) + return [] + + async def list_deployments( + self, + namespace: str = "awoooi", + ) -> list[dict]: + """ + 列出 Namespace 下的 Deployments + + Returns: + list[{name, ready_replicas, replicas, available}] + """ + client = await _get_k8s_client() + if not client: + logger.warning("k8s_list_deployments_unavailable") + return [] + + try: + apps_v1 = client.AppsV1Api() + deployments = await apps_v1.list_namespaced_deployment(namespace=namespace) + + result = [] + for deploy in deployments.items: + ready = deploy.status.ready_replicas or 0 + replicas = deploy.spec.replicas or 0 + available = deploy.status.available_replicas or 0 + + result.append({ + "name": deploy.metadata.name, + "ready_replicas": ready, + "replicas": replicas, + "available": available, + }) + + return result + + except Exception as e: + logger.error("k8s_list_deployments_failed", namespace=namespace, error=str(e)) + return [] + + async def get_pod_status_summary( + self, + namespace: str = "awoooi", + ) -> dict: + """ + 取得 Pod 狀態摘要 + + Returns: + {total, running, pending, failed, problem_pods: [...]} + """ + pods = await self.list_pods(namespace) + + total = len(pods) + running = sum(1 for p in pods if p["phase"] == "Running" and p["ready"]) + pending = sum(1 for p in pods if p["phase"] == "Pending") + failed = sum(1 for p in pods if p["phase"] == "Failed") + + # 找出有問題的 Pods + problem_pods = [] + for pod in pods: + is_problem = ( + pod["phase"] != "Running" or + not pod["ready"] or + pod["restarts"] > 5 + ) + if is_problem: + problem_pods.append({ + "name": pod["name"], + "phase": pod["phase"], + "ready": pod["ready"], + "restarts": pod["restarts"], + }) + + return { + "total": total, + "running": running, + "pending": pending, + "failed": failed, + "problem_pods": problem_pods, + } + + async def is_available(self) -> bool: + """ + 檢查 K8s API 是否可用 + + Returns: + True 如果 K8s 連線正常 + """ + client = await _get_k8s_client() + if not client: + return False + + try: + v1 = client.CoreV1Api() + # 嘗試取得 namespace 驗證連線 + await v1.read_namespace(name="default") + return True + except Exception as e: + logger.warning("k8s_availability_check_failed", error=str(e)) + return False + + +# Singleton instance +_k8s_repository: K8sRepository | None = None + + +def get_k8s_repository() -> K8sRepository: + """ + 取得 K8sRepository 實例 (Singleton) + + Returns: + K8sRepository 實例 + """ + global _k8s_repository + if _k8s_repository is None: + _k8s_repository = K8sRepository() + return _k8s_repository diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 6753b5564..97aaf730a 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -1290,11 +1290,16 @@ class TelegramGateway: duration_seconds: int = 0, message: str = "", workflow_url: str = "", + max_retries: int = 3, ) -> dict: """ 發送 CI/CD 進度通知 (簡潔版，不走 AI 仲裁) 2026-03-30 ogt: 新增，解決 CI/CD 告警被當成事件處理的問題 + 2026-03-30 P1: 新增重試機制 (指數退避) + + Args: + max_retries: 最大重試次數 (預設 3) Returns: dict: Telegram API 回應 @@ -1318,10 +1323,37 @@ class TelegramGateway: } logger.info("telegram_cicd_progress_sending", job=job_name, status=status) - result = await self._send_request("sendMessage", payload) - logger.info("telegram_cicd_progress_sent", job=job_name, status=status) - return result + # 重試機制 (指數退避) + import asyncio + last_error = None + for attempt in range(max_retries): + try: + result = await self._send_request("sendMessage", payload) + logger.info("telegram_cicd_progress_sent", job=job_name, status=status, attempt=attempt + 1) + return result + except TelegramGatewayError as e: + last_error = e + if attempt < max_retries - 1: + delay = 2 ** attempt # 1, 2, 4 秒 + logger.warning( + "telegram_cicd_progress_retry", + job=job_name, + attempt=attempt + 1, + delay=delay, + error=str(e), + ) + await asyncio.sleep(delay) + + # 所有重試都失敗 + logger.error( + "telegram_cicd_progress_failed", + job=job_name, + status=status, + max_retries=max_retries, + error=str(last_error), + ) + raise last_error async def send_deploy_success( self, diff --git a/apps/api/src/services/terminal_service.py b/apps/api/src/services/terminal_service.py index faa88f56c..0c0779534 100644 --- a/apps/api/src/services/terminal_service.py +++ b/apps/api/src/services/terminal_service.py @@ -34,6 +34,7 @@ from src.models.terminal import ( TerminalSession, TerminalSessionStatus, ) +from src.repositories.k8s_repository import get_k8s_repository from src.services.approval_db import get_approval_service from src.services.openclaw import get_openclaw @@ -649,6 +650,7 @@ class TerminalService: 處理狀態查詢 Phase 19.4: 整合真實 K8s API + P1 改進: 使用 K8sRepository (積木化) @author Claude Code @date 2026-03-30 (台北時間) """ @@ -656,54 +658,28 @@ class TerminalService: await self._publish_tool_call(publisher, topic, "K8s-Status", "executing") try: - # 使用 K8s Diagnostics Service 的 client - from src.services.k8s_diagnostics import _get_k8s_client + # P1 改進: 使用 K8sRepository 代替直接呼叫 K8s client + k8s_repo = get_k8s_repository() - client = await _get_k8s_client() - if not client: - raise RuntimeError("K8s client initialization failed") + # 檢查 K8s 可用性 + if not await k8s_repo.is_available(): + raise RuntimeError("K8s API unavailable") - v1 = client.CoreV1Api() - apps_v1 = client.AppsV1Api() - - # 查詢 awoooi namespace 的 Pods + # 取得 Pod 狀態摘要 namespace = "awoooi" - pods = await v1.list_namespaced_pod(namespace=namespace) + pod_summary = await k8s_repo.get_pod_status_summary(namespace) + deployments = await k8s_repo.list_deployments(namespace) - # 統計 Pod 狀態 - pods_total = len(pods.items) - pods_running = sum( - 1 for p in pods.items - if p.status.phase == "Running" - ) - pods_ready = sum( - 1 for p in pods.items - if p.status.phase == "Running" - and all( - c.ready for c in (p.status.container_statuses or []) - ) - ) + # 統計 + pods_total = pod_summary["total"] + pods_running = pod_summary["running"] + pods_ready = pods_running # running 已包含 ready 檢查 + problem_pods = pod_summary["problem_pods"] - # 找出有問題的 Pods - problem_pods = [] - for pod in pods.items: - if pod.status.phase != "Running": - problem_pods.append(f"{pod.metadata.name}: {pod.status.phase}") - elif pod.status.container_statuses: - for cs in pod.status.container_statuses: - if not cs.ready: - reason = "" - if cs.state and cs.state.waiting: - reason = cs.state.waiting.reason or "" - problem_pods.append(f"{pod.metadata.name}: {reason or 'NotReady'}") - break - - # 查詢 Deployments - deployments = await apps_v1.list_namespaced_deployment(namespace=namespace) - deployments_total = len(deployments.items) + deployments_total = len(deployments) deployments_healthy = sum( - 1 for d in deployments.items - if d.status.ready_replicas == d.spec.replicas + 1 for d in deployments + if d["ready_replicas"] == d["replicas"] ) await self._publish_tool_call( @@ -716,7 +692,7 @@ class TerminalService: "deployments_total": deployments_total, } ) - await asyncio.sleep(0.3) + await asyncio.sleep(SSE_DELAY_SECONDS) # 組合狀態訊息 status_lines = [ @@ -727,7 +703,7 @@ class TerminalService: if problem_pods: status_lines.append(f"\n問題 Pods ({len(problem_pods)}):") for pp in problem_pods[:3]: # 最多顯示 3 個 - status_lines.append(f" - {pp}") + status_lines.append(f" - {pp['name']}: {pp['phase']}") if len(problem_pods) > 3: status_lines.append(f" ... 還有 {len(problem_pods) - 3} 個") else: @@ -742,11 +718,11 @@ class TerminalService: logger.error("k8s_status_query_failed", error=str(e)) await self._publish_tool_call( publisher, topic, "K8s-Status", "failed", - result={"error": str(e)[:100]} + result={"error": sanitize_error_message(str(e))} ) await self._publish_thought( publisher, topic, "System", - f"K8s 狀態查詢失敗: {str(e)[:50]}\n" + f"K8s 狀態查詢失敗: {sanitize_error_message(str(e), ERROR_MESSAGE_DISPLAY_LENGTH)}\n" "請確認 K8s 連線配置。" ) diff --git a/apps/api/tests/test_terminal_service.py b/apps/api/tests/test_terminal_service.py index dfbc99d4c..d7aaa7005 100644 --- a/apps/api/tests/test_terminal_service.py +++ b/apps/api/tests/test_terminal_service.py @@ -14,16 +14,17 @@ Omni-Terminal SSE 架構測試 @date 2026-03-28 (台北時間) """ -import pytest from unittest.mock import AsyncMock, MagicMock, patch from uuid import uuid4 +import pytest + +from src.models.approval import ApprovalRequest, ApprovalStatus, RiskLevel from src.models.terminal import ( SpatialContext, TerminalIntentRequest, TerminalSessionStatus, ) -from src.models.approval import ApprovalRequest, ApprovalStatus, RiskLevel from src.services.terminal_service import ( IntentType, TerminalService, @@ -345,44 +346,30 @@ class TestHandleApprovalAction: class TestHandleStatusQuery: - """_handle_status_query Phase 19.4 測試""" + """_handle_status_query Phase 19.4 測試 (P1: 使用 K8sRepository)""" @pytest.mark.asyncio async def test_status_query_success(self, mock_publisher): - """測試 K8s 狀態查詢成功""" + """測試 K8s 狀態查詢成功 (P1: 使用 K8sRepository)""" service = TerminalService() - # 模擬 K8s client - mock_pod = MagicMock() - mock_pod.metadata.name = "awoooi-api-xxx" - mock_pod.status.phase = "Running" - mock_pod.status.container_statuses = [MagicMock(ready=True)] - - mock_pods_list = MagicMock() - mock_pods_list.items = [mock_pod] - - mock_deployment = MagicMock() - mock_deployment.status.ready_replicas = 1 - mock_deployment.spec.replicas = 1 - - mock_deployments_list = MagicMock() - mock_deployments_list.items = [mock_deployment] - - mock_v1 = MagicMock() - mock_v1.list_namespaced_pod = AsyncMock(return_value=mock_pods_list) - - mock_apps_v1 = MagicMock() - mock_apps_v1.list_namespaced_deployment = AsyncMock( - return_value=mock_deployments_list - ) - - mock_client = MagicMock() - mock_client.CoreV1Api.return_value = mock_v1 - mock_client.AppsV1Api.return_value = mock_apps_v1 + # Mock K8sRepository + mock_k8s_repo = MagicMock() + mock_k8s_repo.is_available = AsyncMock(return_value=True) + mock_k8s_repo.get_pod_status_summary = AsyncMock(return_value={ + "total": 1, + "running": 1, + "pending": 0, + "failed": 0, + "problem_pods": [], + }) + mock_k8s_repo.list_deployments = AsyncMock(return_value=[ + {"name": "awoooi-api", "ready_replicas": 1, "replicas": 1, "available": 1} + ]) with patch( - "src.services.k8s_diagnostics._get_k8s_client", - new=AsyncMock(return_value=mock_client), + "src.services.terminal_service.get_k8s_repository", + return_value=mock_k8s_repo, ): await service._handle_status_query( mock_publisher, @@ -396,41 +383,26 @@ class TestHandleStatusQuery: @pytest.mark.asyncio async def test_status_query_with_problem_pods(self, mock_publisher): - """測試有問題 Pods 時的狀態查詢""" + """測試有問題 Pods 時的狀態查詢 (P1: 使用 K8sRepository)""" service = TerminalService() - # 一個正常 Pod，一個有問題的 Pod - mock_pod_ok = MagicMock() - mock_pod_ok.metadata.name = "awoooi-api-ok" - mock_pod_ok.status.phase = "Running" - mock_pod_ok.status.container_statuses = [MagicMock(ready=True)] - - mock_pod_bad = MagicMock() - mock_pod_bad.metadata.name = "awoooi-worker-bad" - mock_pod_bad.status.phase = "Pending" - mock_pod_bad.status.container_statuses = [] - - mock_pods_list = MagicMock() - mock_pods_list.items = [mock_pod_ok, mock_pod_bad] - - mock_deployments_list = MagicMock() - mock_deployments_list.items = [] - - mock_v1 = MagicMock() - mock_v1.list_namespaced_pod = AsyncMock(return_value=mock_pods_list) - - mock_apps_v1 = MagicMock() - mock_apps_v1.list_namespaced_deployment = AsyncMock( - return_value=mock_deployments_list - ) - - mock_client = MagicMock() - mock_client.CoreV1Api.return_value = mock_v1 - mock_client.AppsV1Api.return_value = mock_apps_v1 + # Mock K8sRepository + mock_k8s_repo = MagicMock() + mock_k8s_repo.is_available = AsyncMock(return_value=True) + mock_k8s_repo.get_pod_status_summary = AsyncMock(return_value={ + "total": 2, + "running": 1, + "pending": 1, + "failed": 0, + "problem_pods": [ + {"name": "awoooi-worker-bad", "phase": "Pending", "ready": False, "restarts": 0} + ], + }) + mock_k8s_repo.list_deployments = AsyncMock(return_value=[]) with patch( - "src.services.k8s_diagnostics._get_k8s_client", - new=AsyncMock(return_value=mock_client), + "src.services.terminal_service.get_k8s_repository", + return_value=mock_k8s_repo, ): await service._handle_status_query( mock_publisher, @@ -454,12 +426,16 @@ class TestHandleStatusQuery: @pytest.mark.asyncio async def test_status_query_k8s_unavailable(self, mock_publisher): - """測試 K8s 連線失敗時的處理""" + """測試 K8s 連線失敗時的處理 (P1: 使用 K8sRepository)""" service = TerminalService() + # Mock K8sRepository - 不可用 + mock_k8s_repo = MagicMock() + mock_k8s_repo.is_available = AsyncMock(return_value=False) + with patch( - "src.services.k8s_diagnostics._get_k8s_client", - new=AsyncMock(return_value=None), + "src.services.terminal_service.get_k8s_repository", + return_value=mock_k8s_repo, ): # 不應拋出例外 await service._handle_status_query(