wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

refactor(api): Phase B P1 可靠性強化 (2 項)
Some checks failed
E2E Health Check / e2e-health (push) Has been cancelled
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details

Browse Source 
1. send_cicd_progress 重試機制 (指數退避 1,2,4 秒)
2. K8s Repository 封裝 (IK8sRepository + K8sRepository)

首席架構師審查 P1 改進 - 模組化合規

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-30 01:52:59 +08:00
parent 64e51bb4fd
commit 13bb1496b0
5 changed files with 395 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
apps/api/src/repositories/interfaces.py
View File

@@ -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 連線正常
"""
...

231
apps/api/src/repositories/k8s_repository.py Normal file
View File

@@ -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

38
apps/api/src/services/telegram_gateway.py
View File

@@ -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,

68
apps/api/src/services/terminal_service.py
View File

@@ -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 連線配置。"
)

110
apps/api/tests/test_terminal_service.py
View File

@@ -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(