from __future__ import annotations import json import os from pathlib import Path from typing import Any from .env import env_float, env_int, env_int_list, env_str, env_str_list, load_env_file # 项目根目录。 ROOT = Path(__file__).resolve().parents[1] # 运行期产物目录。 RUNTIME_DIR = ROOT / "run" # 常见工作区根目录，便于本地开发时自动发现业务仓。 WORKSPACE_ROOT = ROOT.parent # 静态资源目录。 STATIC_DIR = ROOT / "static" # 默认 .env 路径。 DEFAULT_ENV_PATH = ROOT / ".env" # 先加载项目根目录的 .env。 load_env_file(Path(env_str("APP_ENV_FILE", str(DEFAULT_ENV_PATH))).expanduser()) # 监控目标配置文件路径。 CONFIG_PATH = Path(env_str("TARGETS_FILE", str(ROOT / "config" / "targets.json"))).expanduser() # HTTP 服务监听地址。 APP_BIND = env_str("APP_BIND", "0.0.0.0") or "0.0.0.0" # HTTP 服务监听端口。 APP_PORT = env_int("APP_PORT", 2026) # 当前 HTTP 应用挂载前缀原始值。 RAW_APP_BASE_PATH = env_str("APP_BASE_PATH", "").strip() # 当前 HTTP 应用挂载前缀；空串表示直接挂在根路径。 APP_BASE_PATH = "/" + RAW_APP_BASE_PATH.strip("/") if RAW_APP_BASE_PATH else "" # Yumi 内部实例固定走本机回环地址。 YUMI_INTERNAL_HOST = env_str("YUMI_INTERNAL_HOST", "127.0.0.1") or "127.0.0.1" # Yumi 内部实例监听端口。 YUMI_INTERNAL_PORT = env_int("YUMI_INTERNAL_PORT", 2027) # HaiYi 内部实例固定走本机回环地址。 HAIYI_INTERNAL_HOST = env_str("HAIYI_INTERNAL_HOST", "127.0.0.1") or "127.0.0.1" # HaiYi 内部实例监听端口。 HAIYI_INTERNAL_PORT = env_int("HAIYI_INTERNAL_PORT", 2028) # 网关反代到内部实例时统一带这一段共享密钥。 INTERNAL_PROXY_SHARED_SECRET = env_str("INTERNAL_PROXY_SHARED_SECRET", "hy-app-monitor-internal") # 网关反代内部实例时的统一超时秒数。 GATEWAY_PROXY_TIMEOUT_SECONDS = env_int("GATEWAY_PROXY_TIMEOUT_SECONDS", 600) # 本地账号库统一落到 run 目录，避免写入版本库。 AUTH_DB_PATH = Path(env_str("AUTH_DB_PATH", str(RUNTIME_DIR / "auth.sqlite3"))).expanduser() # 会话 Cookie 名称固定成单值，前后端都按这个名字取。 AUTH_SESSION_COOKIE_NAME = env_str("AUTH_SESSION_COOKIE_NAME", "hy_app_monitor_session") or "hy_app_monitor_session" # 简单本地账号默认保活 24 小时，过期后重新登录。 AUTH_SESSION_TTL_HOURS = env_int("AUTH_SESSION_TTL_HOURS", 24) # 部署在 HTTPS 反代后时，可通过 .env 把 Cookie 切到 Secure。 AUTH_COOKIE_SECURE = env_str("AUTH_COOKIE_SECURE", "0").lower() in {"1", "true", "yes", "on"} # 服务健康检查超时时间。 PROBE_TIMEOUT_SECONDS = env_float("PROBE_TIMEOUT_SECONDS", 3.0) # 服务健康检查缓存秒数。 PROBE_CACHE_TTL_SECONDS = env_float("PROBE_CACHE_TTL_SECONDS", 5.0) # 服务健康检查并发数。 PROBE_MAX_WORKERS = env_int("PROBE_MAX_WORKERS", 12) # 主机指标缓存秒数。 HOST_METRIC_CACHE_TTL_SECONDS = env_float("HOST_METRIC_CACHE_TTL_SECONDS", 25.0) # 主机指标远端采集超时时间；兼容旧的 TAT_* 变量名。 HOST_METRIC_TIMEOUT_SECONDS = env_int("HOST_METRIC_TIMEOUT_SECONDS", env_int("TAT_METRIC_TIMEOUT_SECONDS", 30)) # 主机指标 CPU 采样间隔；兼容旧的 TAT_* 变量名。 HOST_METRIC_CPU_SAMPLE_SECONDS = env_float("HOST_METRIC_CPU_SAMPLE_SECONDS", env_float("TAT_CPU_SAMPLE_SECONDS", 1.0)) # 兼容旧代码里仍在引用的变量名。 TAT_METRIC_TIMEOUT_SECONDS = HOST_METRIC_TIMEOUT_SECONDS TAT_CPU_SAMPLE_SECONDS = HOST_METRIC_CPU_SAMPLE_SECONDS # 业务运行目录根路径。 RUNTIME_BASE_DIR = env_str("RUNTIME_BASE_DIR", "/opt/likei") # 远端宿主机默认 SSH 用户。 SSH_USER = env_str("SSH_USER", "root") or "root" # 非 root 用户时，允许统一在远端命令前补 sudo -n。 SSH_USE_SUDO = env_str("SSH_USE_SUDO", "0").lower() in {"1", "true", "yes", "on"} # 远端宿主机默认 SSH 端口。 SSH_PORT = env_int("SSH_PORT", 22) # SSH 建连超时秒数。 SSH_CONNECT_TIMEOUT_SECONDS = env_int("SSH_CONNECT_TIMEOUT_SECONDS", 10) # SSH 严格主机检查策略，默认首次连接自动接收。 SSH_STRICT_HOST_KEY_CHECKING = env_str("SSH_STRICT_HOST_KEY_CHECKING", "accept-new") or "accept-new" # 运行期默认使用 deploy 机项目目录下的专用私钥。 SSH_PRIVATE_KEY_PATH = Path(env_str("SSH_PRIVATE_KEY_PATH", str(ROOT / "run" / "ssh" / "id_ed25519"))).expanduser() # 已知主机文件同样放到项目 run 目录，避免污染系统级 known_hosts。 SSH_KNOWN_HOSTS_PATH = Path( env_str("SSH_KNOWN_HOSTS_PATH", str(ROOT / "run" / "ssh" / "known_hosts")) ).expanduser() # 滚动重启单实例最大等待秒数。 ROLLING_RESTART_TIMEOUT_SECONDS = env_int("ROLLING_RESTART_TIMEOUT_SECONDS", 180) # 滚动重启探测轮询间隔。 ROLLING_RESTART_POLL_SECONDS = env_float("ROLLING_RESTART_POLL_SECONDS", 2.0) # 单实例恢复后额外稳定等待秒数。 ROLLING_RESTART_STABILIZE_SECONDS = env_float("ROLLING_RESTART_STABILIZE_SECONDS", 2.0) # 依赖方较多的服务发布前先在 Nacos 摘掉当前实例，避免调用方继续打到正在重启的节点。 NACOS_RELEASE_DRAIN_SERVICES = tuple(env_str_list("NACOS_RELEASE_DRAIN_SERVICES", ["other"])) # Nacos 摘实例后额外等待调用方本地缓存刷新。 NACOS_RELEASE_DRAIN_SECONDS = env_float("NACOS_RELEASE_DRAIN_SECONDS", 8.0) # 等待 Nacos enabled 状态变化的最长秒数。 NACOS_RELEASE_DRAIN_TIMEOUT_SECONDS = env_int("NACOS_RELEASE_DRAIN_TIMEOUT_SECONDS", 30) # 日志查看固定读取最近多少条。 SERVICE_LOG_PREVIEW_LINES = env_int("SERVICE_LOG_PREVIEW_LINES", 1000) # 日志导出最大允许条数，避免单次下载打爆页面进程。 SERVICE_LOG_EXPORT_MAX_LINES = env_int("SERVICE_LOG_EXPORT_MAX_LINES", 10000) # 服务更新时本地构建最长等待秒数。 SERVICE_BUILD_TIMEOUT_SECONDS = env_int("SERVICE_BUILD_TIMEOUT_SECONDS", 3600) # 服务更新时目标机预拉镜像最长等待秒数。 SERVICE_PULL_TIMEOUT_SECONDS = env_int("SERVICE_PULL_TIMEOUT_SECONDS", 600) # 固定构建工作区根目录，留空时按本地 / deploy 机路径自动推导。 SERVICE_BUILD_CACHE_ROOT = env_str("SERVICE_BUILD_CACHE_ROOT", "") # Java Maven 并行线程，默认按 1 个 CPU 组装 reactor。 JAVA_MAVEN_THREADS = env_str("JAVA_MAVEN_THREADS", "1C") # Java 镜像 build/push 最大并行数。 JAVA_IMAGE_BUILD_MAX_WORKERS = env_int("JAVA_IMAGE_BUILD_MAX_WORKERS", 3) # 腾讯云区域，SSH 引导和带外脚本会复用。 DEPLOY_REGION = env_str("DEPLOY_REGION", "me-saudi-arabia") # 腾讯云 SecretId。 TENCENT_SECRET_ID = env_str("TENCENT_SECRET_ID", "") # 腾讯云 SecretKey。 TENCENT_SECRET_KEY = env_str("TENCENT_SECRET_KEY", "") # gateway 发布做 CLB 摘流时使用的负载均衡实例 ID。 GATEWAY_CLB_LOAD_BALANCER_ID = env_str("GATEWAY_CLB_LOAD_BALANCER_ID", "") # gateway 可能同时挂在 HTTP / HTTPS 等多个监听器下，这里统一允许逗号分隔配置。 GATEWAY_CLB_LISTENER_IDS = env_str_list("GATEWAY_CLB_LISTENER_IDS", []) # gateway 发布时只保护这些域名规则；留空表示 listener 下匹配到的规则全部处理。 GATEWAY_CLB_ALLOWED_DOMAINS = env_str_list("GATEWAY_CLB_ALLOWED_DOMAINS", []) # CDN 刷新工具可选限制到固定业务域名；留空时允许提交任意腾讯云 CDN 域名。 CDN_ALLOWED_DOMAINS = env_str_list("CDN_ALLOWED_DOMAINS", []) # CDN 工具页默认展示最近多少条刷新记录。 CDN_PURGE_TASK_LIMIT = env_int("CDN_PURGE_TASK_LIMIT", 20) # CLB 异步任务轮询间隔。 GATEWAY_CLB_POLL_SECONDS = env_float("GATEWAY_CLB_POLL_SECONDS", 2.0) # CLB 摘流 / 恢复异步任务最长等待秒数。 GATEWAY_CLB_TASK_TIMEOUT_SECONDS = env_int("GATEWAY_CLB_TASK_TIMEOUT_SECONDS", 120) # 摘流后额外等待多少秒再重启；默认 0 表示只停新流量，不额外等待连接自然排空。 GATEWAY_CLB_DRAIN_SECONDS = env_float("GATEWAY_CLB_DRAIN_SECONDS", 0.0) def pick_existing_path(candidates: list[Path]) -> Path: # 配置未显式指定时，优先挑当前机器上真正存在的路径。 for candidate in candidates: if candidate.exists(): return candidate return candidates[0] def env_or_candidate_path(env_key: str, candidates: list[Path]) -> Path: # 允许 .env 显式覆盖，也兼容本地开发和远端 deploy 两种目录布局。 override = os.environ.get(env_key, "").strip() if override: return Path(override).expanduser() return pick_existing_path(candidates) def truthy(value: Any, default: bool = False) -> bool: # 配置里布尔值可能来自 JSON 布尔或字符串，统一在这里规范化。 if value is None: return default if isinstance(value, bool): return value if isinstance(value, str): return value.strip().lower() in {"1", "true", "yes", "on"} return bool(value) # 页面默认刷新间隔。 DEFAULT_REFRESH_INTERVAL_SECONDS = env_int("DEFAULT_REFRESH_INTERVAL_SECONDS", 10) # 页面可选刷新间隔。 REFRESH_INTERVAL_OPTIONS = env_int_list("REFRESH_INTERVAL_OPTIONS", [5, 10, 15, 30, 60]) # Nacos API 基础地址。 NACOS_BASE_URL = env_str("NACOS_BASE_URL", "http://10.2.1.17:8848").removesuffix("/nacos").rstrip("/") # Nacos 目标命名空间 ID。 NACOS_NAMESPACE_ID = env_str("NACOS_NAMESPACE_ID", "f7a8594e-090c-4830-a3ad-b5c0f9afcc07") # Nacos 目标命名空间名。 NACOS_NAMESPACE_NAME = env_str("NACOS_NAMESPACE_NAME", "RedCircleServiceProd") # Nacos 默认注册分组。 NACOS_DISCOVERY_GROUP = env_str("NACOS_DISCOVERY_GROUP", "DEFAULT_GROUP") # Nacos 默认注册集群。 NACOS_DISCOVERY_CLUSTER_NAME = env_str("NACOS_DISCOVERY_CLUSTER_NAME", "DEFAULT") # Nacos 请求超时。 NACOS_TIMEOUT_SECONDS = env_float("NACOS_TIMEOUT_SECONDS", 6.0) # Nacos 结果缓存秒数。 NACOS_CACHE_TTL_SECONDS = env_float("NACOS_CACHE_TTL_SECONDS", 15.0) # Nacos service list 单页大小。 NACOS_PAGE_SIZE = env_int("NACOS_PAGE_SIZE", 200) # Nacos 可选用户名。 NACOS_USERNAME = env_str("NACOS_USERNAME", "") # Nacos 可选密码。 NACOS_PASSWORD = env_str("NACOS_PASSWORD", "") # Harbor 镜像仓地址，服务更新会推送镜像并让目标机主动拉取。 HARBOR_REGISTRY = env_str("HARBOR_REGISTRY", "10.2.1.3:18082") # Harbor 项目名。 HARBOR_PROJECT = env_str("HARBOR_PROJECT", "likei") # Harbor 登录用户名。 HARBOR_USERNAME = env_str("HARBOR_USERNAME", "") # Harbor 登录密码。 HARBOR_PASSWORD = env_str("HARBOR_PASSWORD", "") # Docker 构建平台，和现有 Jenkins/构建脚本口径保持一致。 BUILD_DOCKER_PLATFORM = env_str("BUILD_DOCKER_PLATFORM", "linux/amd64") # Java 服务默认拉取的 Git ref。 UPDATE_DEFAULT_JAVA_GIT_REF = env_str("UPDATE_DEFAULT_JAVA_GIT_REF", "main") # Go 服务默认拉取的 Git ref。 UPDATE_DEFAULT_GO_GIT_REF = env_str("UPDATE_DEFAULT_GO_GIT_REF", "main") # Admin 前端默认拉取的 Git ref。 UPDATE_DEFAULT_ADMIN_GIT_REF = env_str("UPDATE_DEFAULT_ADMIN_GIT_REF", "main") # Java 仓库路径，默认兼容本地工作区和远端 deploy 机。 JAVA_REPO_ROOT = env_or_candidate_path( "JAVA_REPO_ROOT", [ WORKSPACE_ROOT / "chatapp3" / "chatapp3-java", Path("/opt/deploy/sources/chatapp3-java"), ], ) # Go 仓库路径，默认兼容本地工作区和远端 deploy 机。 GOLANG_REPO_ROOT = env_or_candidate_path( "GOLANG_REPO_ROOT", [ WORKSPACE_ROOT / "chatapp3" / "chatapp3-golang", Path("/opt/deploy/sources/chatapp3-golang"), ], ) # Dashboard CDC Worker 仓库地址；仓库缺失时允许按这里自动 clone。 DASHBOARD_CDC_REPO_URL = env_str("DASHBOARD_CDC_REPO_URL", "git@gitea.haiyihy.com:admin/dashboard-cdc-worker.git") # Dashboard CDC Worker 仓库路径，默认兼容本地工作区和远端 deploy 机。 DASHBOARD_CDC_REPO_ROOT = env_or_candidate_path( "DASHBOARD_CDC_REPO_ROOT", [ WORKSPACE_ROOT / "chatapp3" / "dashboard-cdc-worker", WORKSPACE_ROOT / "dashboard-cdc-worker", Path("/opt/deploy/sources/dashboard-cdc-worker"), ], ) # Dashboard CDC Worker 在发布页和 compose 里的服务名。 DASHBOARD_CDC_SERVICE_NAME = env_str("DASHBOARD_CDC_SERVICE_NAME", "dashboard-cdc-worker").strip() or "dashboard-cdc-worker" # Dashboard CDC Worker 是否接入发布；默认启用。 DASHBOARD_CDC_ENABLED = truthy(os.environ.get("DASHBOARD_CDC_ENABLED"), True) # Admin 前端仓库地址；仓库缺失时允许按这里自动 clone。 ADMIN_FRONTEND_REPO_URL = env_str("ADMIN_FRONTEND_REPO_URL", "") # Admin 前端仓库路径，默认兼容本地工作区和远端 deploy 机。 ADMIN_FRONTEND_REPO_ROOT = env_or_candidate_path( "ADMIN_FRONTEND_REPO_ROOT", [ WORKSPACE_ROOT / "chatapp3" / "chatapp3-admin", Path("/opt/deploy/sources/chatapp3-admin"), ], ) # Admin 前端在发布页里的服务名。 ADMIN_FRONTEND_SERVICE_NAME = env_str("ADMIN_FRONTEND_SERVICE_NAME", "admin").strip() or "admin" # Admin 前端是否启用；未显式指定时，只要仓库和目标机信息齐备就自动开启。 ADMIN_FRONTEND_HOST = env_str("ADMIN_FRONTEND_HOST", "") # Admin 前端目标机私网 IP。 ADMIN_FRONTEND_IP = env_str("ADMIN_FRONTEND_IP", "") # Admin 前端目标机实例 ID，主要用于 TAT 引导 SSH。 ADMIN_FRONTEND_INSTANCE_ID = env_str("ADMIN_FRONTEND_INSTANCE_ID", "") # Admin 前端 SSH 主机，默认沿用私网 IP。 ADMIN_FRONTEND_SSH_HOST = env_str("ADMIN_FRONTEND_SSH_HOST", ADMIN_FRONTEND_IP).strip() or ADMIN_FRONTEND_IP # Admin 前端 SSH 用户。 ADMIN_FRONTEND_SSH_USER = env_str("ADMIN_FRONTEND_SSH_USER", SSH_USER).strip() or SSH_USER # Admin 前端 SSH 端口。 ADMIN_FRONTEND_SSH_PORT = env_int("ADMIN_FRONTEND_SSH_PORT", SSH_PORT) # Admin 前端是否需要 sudo。 ADMIN_FRONTEND_SSH_USE_SUDO = env_str("ADMIN_FRONTEND_SSH_USE_SUDO", "0").lower() in {"1", "true", "yes", "on"} # Admin 前端远端运行目录。 ADMIN_FRONTEND_REMOTE_DIR = env_str("ADMIN_FRONTEND_REMOTE_DIR", "/opt/yumi-admin/frontend-admin").strip() or "/opt/yumi-admin/frontend-admin" # Admin 前端静态站点目录；chatapp3 admin 只发布到 yumiadmin-acc 子站，不占用 Caddy 根目录。 ADMIN_FRONTEND_SITE_DIR = ( env_str("ADMIN_FRONTEND_SITE_DIR", f"{ADMIN_FRONTEND_REMOTE_DIR.rstrip('/')}/site/yumiadmin-acc").strip() or f"{ADMIN_FRONTEND_REMOTE_DIR.rstrip('/')}/site/yumiadmin-acc" ) # Admin 前端 compose 文件路径。 ADMIN_FRONTEND_COMPOSE_FILE = ( env_str("ADMIN_FRONTEND_COMPOSE_FILE", f"{ADMIN_FRONTEND_REMOTE_DIR.rstrip('/')}/docker-compose.yml").strip() or f"{ADMIN_FRONTEND_REMOTE_DIR.rstrip('/')}/docker-compose.yml" ) # Admin 前端 compose 服务名。 ADMIN_FRONTEND_COMPOSE_SERVICE = env_str("ADMIN_FRONTEND_COMPOSE_SERVICE", "caddy").strip() or "caddy" # Admin 前端当前发布元数据文件路径。 ADMIN_FRONTEND_RELEASE_META_PATH = ( env_str( "ADMIN_FRONTEND_RELEASE_META_PATH", f"{ADMIN_FRONTEND_REMOTE_DIR.rstrip('/')}/.hy-app-monitor-admin-release.json", ).strip() or f"{ADMIN_FRONTEND_REMOTE_DIR.rstrip('/')}/.hy-app-monitor-admin-release.json" ) # Admin 前端构建脚本。 ADMIN_FRONTEND_BUILD_SCRIPT = env_str("ADMIN_FRONTEND_BUILD_SCRIPT", "build:antdv-next").strip() or "build:antdv-next" # Admin 前端构建产物目录。 ADMIN_FRONTEND_DIST_DIR = env_str("ADMIN_FRONTEND_DIST_DIR", "apps/dist").strip() or "apps/dist" # Admin 前端对外访问地址，仅用于结果展示和排查。 ADMIN_FRONTEND_PUBLIC_URL = env_str("ADMIN_FRONTEND_PUBLIC_URL", "").strip() # 未显式指定 ADMIN_FRONTEND_ENABLED 时，只要关键参数齐备就默认启用。 ADMIN_FRONTEND_ENABLED = truthy( os.environ.get("ADMIN_FRONTEND_ENABLED"), bool( ADMIN_FRONTEND_SERVICE_NAME and ADMIN_FRONTEND_HOST and (ADMIN_FRONTEND_SSH_HOST or ADMIN_FRONTEND_IP) and (ADMIN_FRONTEND_REPO_URL or str(ADMIN_FRONTEND_REPO_ROOT).strip()) ), ) # chatapp-cron 仓库地址；仓库缺失时允许按这里自动 clone。 CHATAPP_CRON_REPO_URL = env_str("CHATAPP_CRON_REPO_URL", "git@gitea.haiyihy.com:hy/chatapp-cron.git") # chatapp-cron 仓库路径，默认兼容本地工作区和远端 deploy 机。 CHATAPP_CRON_REPO_ROOT = env_or_candidate_path( "CHATAPP_CRON_REPO_ROOT", [ WORKSPACE_ROOT / "chatapp3" / "chatapp-cron", WORKSPACE_ROOT / "chatapp-cron", Path("/opt/deploy/sources/chatapp-cron"), ], ) # chatapp-cron 默认拉取的 Git ref。 CHATAPP_CRON_DEFAULT_GIT_REF = env_str("CHATAPP_CRON_DEFAULT_GIT_REF", "main") # chatapp-cron 在通用部署页里的服务名。 CHATAPP_CRON_SERVICE_NAME = env_str("CHATAPP_CRON_SERVICE_NAME", "chatapp-cron").strip() or "chatapp-cron" # chatapp-cron 远端主机名称。 CHATAPP_CRON_HOST = env_str("CHATAPP_CRON_HOST", "cron").strip() or "cron" # chatapp-cron 远端主机私网 IP。 # 定时任务服务器：ins-nt6bsay0 / cron / 10.3.1.14。 CHATAPP_CRON_IP = env_str("CHATAPP_CRON_IP", "10.3.1.14") # chatapp-cron 云服务器实例 ID，主要供 TAT SSH 引导使用。 CHATAPP_CRON_INSTANCE_ID = env_str("CHATAPP_CRON_INSTANCE_ID", "ins-nt6bsay0") # chatapp-cron SSH 主机，默认沿用私网 IP。 CHATAPP_CRON_SSH_HOST = env_str("CHATAPP_CRON_SSH_HOST", CHATAPP_CRON_IP).strip() or CHATAPP_CRON_IP # chatapp-cron SSH 用户。 CHATAPP_CRON_SSH_USER = env_str("CHATAPP_CRON_SSH_USER", SSH_USER).strip() or SSH_USER # chatapp-cron SSH 端口。 CHATAPP_CRON_SSH_PORT = env_int("CHATAPP_CRON_SSH_PORT", SSH_PORT) # chatapp-cron 是否需要 sudo。 CHATAPP_CRON_SSH_USE_SUDO = env_str("CHATAPP_CRON_SSH_USE_SUDO", "0").lower() in {"1", "true", "yes", "on"} # chatapp-cron 远端运行目录。 CHATAPP_CRON_REMOTE_DIR = env_str("CHATAPP_CRON_REMOTE_DIR", "/opt/deploy/runtime/current").strip() or "/opt/deploy/runtime/current" # chatapp-cron compose 文件路径。 CHATAPP_CRON_COMPOSE_FILE = ( env_str("CHATAPP_CRON_COMPOSE_FILE", f"{CHATAPP_CRON_REMOTE_DIR.rstrip('/')}/docker-compose.yml").strip() or f"{CHATAPP_CRON_REMOTE_DIR.rstrip('/')}/docker-compose.yml" ) # chatapp-cron compose 服务名。 CHATAPP_CRON_COMPOSE_SERVICE = env_str("CHATAPP_CRON_COMPOSE_SERVICE", "week-star-job").strip() or "week-star-job" # chatapp-cron 远端二进制路径。 CHATAPP_CRON_BINARY_PATH = ( env_str("CHATAPP_CRON_BINARY_PATH", f"{CHATAPP_CRON_REMOTE_DIR.rstrip('/')}/week-star-job/service").strip() or f"{CHATAPP_CRON_REMOTE_DIR.rstrip('/')}/week-star-job/service" ) # chatapp-cron 当前发布元数据文件路径。 CHATAPP_CRON_RELEASE_META_PATH = ( env_str( "CHATAPP_CRON_RELEASE_META_PATH", f"{CHATAPP_CRON_REMOTE_DIR.rstrip('/')}/.hy-app-monitor-chatapp-cron-release.json", ).strip() or f"{CHATAPP_CRON_REMOTE_DIR.rstrip('/')}/.hy-app-monitor-chatapp-cron-release.json" ) # chatapp-cron 是否启用；未显式指定时，只要目标机和仓库信息齐备就自动开启。 CHATAPP_CRON_ENABLED = truthy( os.environ.get("CHATAPP_CRON_ENABLED"), bool( CHATAPP_CRON_SERVICE_NAME and CHATAPP_CRON_HOST and (CHATAPP_CRON_SSH_HOST or CHATAPP_CRON_IP) and CHATAPP_CRON_REMOTE_DIR and (CHATAPP_CRON_REPO_URL or str(CHATAPP_CRON_REPO_ROOT).strip()) ), ) # Mongo 连接串。 MONGO_URI = env_str("MONGO_URI", "") # Mongo 默认关注的业务库名。 MONGO_DB_NAME = env_str("MONGO_DB_NAME", "tarab_all") # Mongo 机器实例 ID，主要用于和主机指标联动显示。 MONGO_INSTANCE_ID = env_str("MONGO_INSTANCE_ID", "ins-o9mmob66") # Mongo 结果缓存秒数。 MONGO_CACHE_TTL_SECONDS = env_float("MONGO_CACHE_TTL_SECONDS", 20.0) # Mongo 建连超时。 MONGO_CONNECT_TIMEOUT_MS = env_int("MONGO_CONNECT_TIMEOUT_MS", 3000) # Mongo server selection 超时。 MONGO_SERVER_SELECTION_TIMEOUT_MS = env_int("MONGO_SERVER_SELECTION_TIMEOUT_MS", 3000) # Mongo socket 超时。 MONGO_SOCKET_TIMEOUT_MS = env_int("MONGO_SOCKET_TIMEOUT_MS", 5000) # MySQL 主机地址。 MYSQL_HOST = env_str("MYSQL_HOST", "") # MySQL 端口。 MYSQL_PORT = env_int("MYSQL_PORT", 3306) # MySQL 登录账号。 MYSQL_USER = env_str("MYSQL_USER", "") # MySQL 登录密码；允许空串，兼容特殊场景。 MYSQL_PASSWORD = os.environ.get("MYSQL_PASSWORD", "") # MySQL 默认库；为空时由页面显式选择。 MYSQL_DEFAULT_DATABASE = env_str("MYSQL_DEFAULT_DATABASE", "") # MySQL 指标与元数据缓存秒数。 MYSQL_CACHE_TTL_SECONDS = env_float("MYSQL_CACHE_TTL_SECONDS", 8.0) # MySQL 建连超时秒数。 MYSQL_CONNECT_TIMEOUT_SECONDS = env_int("MYSQL_CONNECT_TIMEOUT_SECONDS", 5) # MySQL 读超时秒数。 MYSQL_READ_TIMEOUT_SECONDS = env_int("MYSQL_READ_TIMEOUT_SECONDS", 8) # MySQL 写超时秒数。 MYSQL_WRITE_TIMEOUT_SECONDS = env_int("MYSQL_WRITE_TIMEOUT_SECONDS", 15) # MySQL 表浏览默认分页大小。 MYSQL_DEFAULT_PAGE_SIZE = env_int("MYSQL_DEFAULT_PAGE_SIZE", 100) # MySQL 表浏览最大分页大小。 MYSQL_MAX_PAGE_SIZE = env_int("MYSQL_MAX_PAGE_SIZE", 200) # SQL 工作台单次最多返回多少行。 MYSQL_QUERY_MAX_ROWS = env_int("MYSQL_QUERY_MAX_ROWS", 500) # CPU 告警阈值。 CPU_WARN_PERCENT = env_float("CPU_WARN_PERCENT", 60.0) # CPU 严重阈值。 CPU_BAD_PERCENT = env_float("CPU_BAD_PERCENT", 85.0) # 内存告警阈值。 MEMORY_WARN_PERCENT = env_float("MEMORY_WARN_PERCENT", 70.0) # 内存严重阈值。 MEMORY_BAD_PERCENT = env_float("MEMORY_BAD_PERCENT", 90.0) # 磁盘告警阈值。 DISK_WARN_PERCENT = env_float("DISK_WARN_PERCENT", 75.0) # 磁盘严重阈值。 DISK_BAD_PERCENT = env_float("DISK_BAD_PERCENT", 90.0) # 进程数告警阈值。 PROCESS_WARN_COUNT = env_int("PROCESS_WARN_COUNT", 180) # 进程数严重阈值。 PROCESS_BAD_COUNT = env_int("PROCESS_BAD_COUNT", 260) # 服务耗时告警阈值。 SERVICE_LATENCY_WARN_MS = env_int("SERVICE_LATENCY_WARN_MS", 1000) # 服务耗时严重阈值。 SERVICE_LATENCY_BAD_MS = env_int("SERVICE_LATENCY_BAD_MS", 3000) # 统一暴露给前端的阈值定义。 THRESHOLDS = { "cpuPercent": {"warn": CPU_WARN_PERCENT, "bad": CPU_BAD_PERCENT}, "memoryPercent": {"warn": MEMORY_WARN_PERCENT, "bad": MEMORY_BAD_PERCENT}, "diskPercent": {"warn": DISK_WARN_PERCENT, "bad": DISK_BAD_PERCENT}, "processCount": {"warn": PROCESS_WARN_COUNT, "bad": PROCESS_BAD_COUNT}, "latencyMs": {"warn": SERVICE_LATENCY_WARN_MS, "bad": SERVICE_LATENCY_BAD_MS}, } def utc_now() -> str: # 时间函数延迟导入，避免这里堆太多全局依赖。 from datetime import datetime, timezone # 统一输出无微秒的 UTC ISO 字符串。 return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def settings_payload() -> dict[str, Any]: # 把页面需要的动态设置统一打包返回。 return { "refreshIntervalSeconds": DEFAULT_REFRESH_INTERVAL_SECONDS, "refreshIntervalOptions": list(REFRESH_INTERVAL_OPTIONS), "thresholds": THRESHOLDS, } def load_hosts() -> list[dict[str, Any]]: # 读取监控目标配置。 payload = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) # 没有 hosts 时回退为空列表。 hosts = payload.get("hosts") or [] # 规范化后的主机列表。 normalized: list[dict[str, Any]] = [] # 逐台主机做结构清洗。 for host in hosts: normalized.append( { "host": str(host.get("name") or "").strip(), "group": str(host.get("group") or "").strip(), "ip": str(host.get("ip") or "").strip(), "instanceId": str(host.get("instanceId") or "").strip(), "sshHost": str(host.get("sshHost") or host.get("ip") or "").strip(), "sshUser": str(host.get("sshUser") or SSH_USER).strip() or SSH_USER, "sshUseSudo": truthy(host.get("sshUseSudo"), SSH_USE_SUDO), "sshPort": int(host.get("sshPort") or SSH_PORT), "services": host.get("services") or [], } ) # 返回可供各模块复用的主机列表。 return normalized def flatten_services(hosts: list[dict[str, Any]]) -> list[dict[str, Any]]: # 这里把主机结构拍平成服务列表，方便并发探测。 services: list[dict[str, Any]] = [] # 逐台主机展开服务。 for host in hosts: for service in host.get("services") or []: services.append( { "host": host["host"], "group": host["group"], "ip": host["ip"], "instanceId": host["instanceId"], "sshHost": host["sshHost"], "sshUser": host["sshUser"], "sshUseSudo": truthy(host.get("sshUseSudo"), SSH_USE_SUDO), "sshPort": int(host["sshPort"]), "service": str(service.get("name") or "").strip(), "kind": str(service.get("kind") or "java").strip(), "port": int(service.get("port")), "path": str(service.get("path") or "/").strip(), } ) # 返回扁平服务列表。 return services