from __future__ import annotations import os import socket from dataclasses import dataclass from pathlib import Path from typing import Final from .version import VERSION DEFAULT_REPORT_INTERVAL_SECONDS: Final[int] = 60 @dataclass(frozen=True) class AgentConfig: hostname: str node_file: str version: str master_endpoint: str report_interval_seconds: int health_dir: str request_timeout_seconds: int = 10 def _normalise_master_endpoint(value: str) -> str: value = value.strip() if not value: raise ValueError("MASTER_ENDPOINT environment variable is required") if not value.startswith("http://") and not value.startswith("https://"): value = f"http://{value}" return value.rstrip("/") def _read_report_interval(raw_value: str | None) -> int: if raw_value is None or raw_value.strip() == "": return DEFAULT_REPORT_INTERVAL_SECONDS try: interval = int(raw_value) except ValueError as exc: raise ValueError("REPORT_INTERVAL_SECONDS must be an integer") from exc if interval <= 0: raise ValueError("REPORT_INTERVAL_SECONDS must be positive") return interval def _resolve_hostname() -> str: return os.environ.get("AGENT_HOSTNAME") or socket.gethostname() def load_config() -> AgentConfig: """从环境变量推导配置,移除了外部配置文件依赖。""" hostname = _resolve_hostname() node_file = f"/private/argus/agent/{hostname}/node.json" health_dir = f"/private/argus/agent/{hostname}/health/" master_endpoint_env = os.environ.get("MASTER_ENDPOINT") if master_endpoint_env is None: raise ValueError("MASTER_ENDPOINT environment variable is not set") master_endpoint = _normalise_master_endpoint(master_endpoint_env) report_interval_seconds = _read_report_interval(os.environ.get("REPORT_INTERVAL_SECONDS")) Path(node_file).parent.mkdir(parents=True, exist_ok=True) Path(health_dir).mkdir(parents=True, exist_ok=True) return AgentConfig( hostname=hostname, node_file=node_file, version=VERSION, master_endpoint=master_endpoint, report_interval_seconds=report_interval_seconds, health_dir=health_dir, )