fix(api): service_registry 安全降級 — Docker 無 YAML 時不 crash,fallback AUTO
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 11m37s

This commit is contained in:
OG T
2026-04-08 21:47:38 +08:00
parent 3cab16a681
commit c9f1bcd122

View File

@@ -16,24 +16,23 @@ import yaml
logger = structlog.get_logger(__name__)
# YAML 路徑 — 安全搜尋,相容本地開發 + Docker 容器
# 本地: apps/api/src/services/ → parents[4] = repo root
# Docker: /app/src/services/ → parents[3] = /app → 需要往上找 ops/config
def _find_registry_path() -> Path:
"""安全搜尋 service-registry.yaml相容不同部署環境"""
candidates = [
Path(__file__).resolve().parents[4] / "ops" / "config" / "service-registry.yaml", # 本地開發
Path(__file__).resolve().parents[3] / "ops" / "config" / "service-registry.yaml", # Docker
Path("/app/ops/config/service-registry.yaml"), # Docker 絕對路徑
Path(__file__).resolve().parents[2] / "ops" / "config" / "service-registry.yaml",
]
for p in candidates:
if p.exists():
return p
# fallback: 回傳第一個候選(會在載入時報錯,但不會 crash import
return candidates[0]
# YAML 路徑 — 延遲載入import 時不 crash
# Docker 容器不包含 ops/ 目錄,所以 registry 在 K8s 環境中不可用
# 功能降級: 找不到 YAML 時 get_service_registry() 回傳空 registry
_DEFAULT_REGISTRY_PATH: Path | None = None
_DEFAULT_REGISTRY_PATH = _find_registry_path()
def _find_registry_path() -> Path | None:
"""安全搜尋 service-registry.yaml找不到回傳 None"""
current = Path(__file__).resolve().parent
for _ in range(10):
for subdir in ["ops/config", "ops/monitoring"]:
candidate = current / subdir / "service-registry.yaml"
if candidate.exists():
return candidate
if current == current.parent:
break
current = current.parent
return None
class StatefulLevel(str, Enum):
@@ -64,7 +63,7 @@ class ServiceRegistryClient:
"""
def __init__(self, registry_path: Path | None = None) -> None:
self._path = registry_path or _DEFAULT_REGISTRY_PATH
self._path = registry_path or _find_registry_path()
self._services: dict[str, ServiceInfo] = {}
self._backup_policies: dict[str, Any] = {}
self._multisig_config: dict[str, Any] = {}
@@ -73,6 +72,10 @@ class ServiceRegistryClient:
def _load(self) -> None:
if self._loaded:
return
if self._path is None or not self._path.exists():
logger.warning("service_registry_not_found", path=str(self._path), fallback="all services treated as AUTO")
self._loaded = True
return
try:
with open(self._path) as f:
data = yaml.safe_load(f)