"""数据库隧道自愈：13306 不通就自动拉起 autossh.sh 背景：数据库经 SSH 隧道访问（autossh.sh，本地 13306 → 远端 3306）。隧道断掉时， 任何入口都只会得到一句"MySQL 连接失败"，不会自己恢复 —— 凌晨的定时任务因此整晚失败。 行为（挂在 `MySQLDB.connect()` 上，所以**所有入口都生效**，不只是定时任务）： 1. 探测 `host:port` 是否可连； 2. 可连 → 什么都不做（正常路径零开销，不做任何多余动作）； 3. 不通 → a. 若 systemd 隧道单元处于 active（说明由 systemd 托管，它会自动重连）→ **只等待**， 不另起一个 autossh 去抢同一个端口（那会让 systemd 单元因端口被占而反复重启失败）； b. 否则执行 `config.yml` 里 `mysql.tunnel.script`（默认项目根的 `autossh.sh`）； 4. 最多等 `wait_seconds` 秒，最后再探测一次并**如实报告**结果（不通就说通不了）。 配置（config.yml）： mysql: tunnel: enabled: 1 # 关掉则完全不探测 script: autossh.sh # 相对项目根 systemd_unit: xwlb-tunnel.service wait_seconds: 30 connect_timeout: 2 命令行自检： python tunnel.py # 探测并按需修复，退出码 0=通 / 1=仍不通 python tunnel.py --dry-run # 只看会做什么，不执行 """ import argparse import logging import os import socket import subprocess import sys import time from pathlib import Path import config logger = logging.getLogger(__name__) LOCAL_HOSTS = {'localhost', '127.0.0.1', '::1', '0.0.0.0'} # 状态常量（便于调用方与测试断言，不要用字符串字面量比较） ST_DISABLED = 'disabled' ST_NOT_LOCAL = 'not-local' ST_ALREADY_OPEN = 'already-open' ST_SYSTEMD_WAITED = 'systemd-waited' ST_SCRIPT_STARTED = 'script-started' ST_UNAVAILABLE = 'unavailable' def _cfg(dotted, default): return config.get(f'mysql.tunnel.{dotted}', default) def is_local(host) -> bool: return str(host or '').strip().lower() in LOCAL_HOSTS def port_open(host, port, timeout=2.0) -> bool: """TCP 层探测端口是否可连（不涉及 MySQL 握手，快且无副作用）""" try: with socket.create_connection((host, int(port)), timeout=timeout): return True except (OSError, ValueError): return False def systemd_unit_active(unit, timeout=5) -> bool: """systemd 单元是否 active；systemctl 不可用或单元不存在时返回 False""" if not unit: return False try: out = subprocess.run(['systemctl', 'is-active', str(unit)], capture_output=True, text=True, timeout=timeout) return out.stdout.strip() == 'active' except (OSError, subprocess.SubprocessError): return False def _wait_for_port(host, port, seconds, interval=2.0): """轮询等待端口就绪；返回实际等待秒数（未就绪返回总等待时长）""" deadline = time.time() + max(0, seconds) waited = 0.0 while True: if port_open(host, port, timeout=2.0): return waited if time.time() >= deadline: return waited time.sleep(interval) waited = min(interval, max(0.0, seconds - waited)) + waited def ensure_tunnel(host=None, port=None, reason='', dry_run=False, wait_seconds=None): """确保数据库端口可连；不通则按配置自愈 参数: host/port: 不传则取 config.yml 的 mysql.host / mysql.port reason: 写进日志的触发原因（如"MySQL 连接失败"） dry_run: 只报告会做什么，不执行脚本 wait_seconds: 覆盖配置里的等待时长 返回值: dict: {'status': 见 ST_* 常量, 'detail': 说明, 'waited': 等待秒数, 'host': ..., 'port': ..., 'healed': bool} """ cfg = config.db_config() host = host or cfg['host'] port = int(port or cfg['port']) prefix = f"（{reason}）" if reason else '' result = {'host': host, 'port': port, 'waited': 0.0, 'healed': False} if not config.get_bool('mysql.tunnel.enabled', True): result.update(status=ST_DISABLED, detail='配置 mysql.tunnel.enabled=0，跳过探测') return result if not is_local(host): # 直连远端数据库时没有隧道可言，不去跑 autossh.sh result.update(status=ST_NOT_LOCAL, detail=f'{host} 不是本机地址，不适用 SSH 隧道') return result if port_open(host, port, timeout=float(_cfg('connect_timeout', 2) or 2)): result.update(status=ST_ALREADY_OPEN, detail=f'{host}:{port} 已可连') return result wait_seconds = float(_cfg('wait_seconds', 30) if wait_seconds is None else wait_seconds) unit = _cfg('systemd_unit', 'xwlb-tunnel.service') # 分支 a：systemd 托管中（它会自动重连），只等，不抢端口 if not dry_run and systemd_unit_active(unit): logger.warning("数据库端口 %s:%s 不通%s，systemd 单元 %s 处于 active，等待其自动重连…", host, port, prefix, unit) waited = _wait_for_port(host, port, wait_seconds) ok = port_open(host, port) result.update(status=ST_SYSTEMD_WAITED if ok else ST_UNAVAILABLE, detail=(f'等待 systemd 单元 {unit} 重连' + ('成功' if ok else '超时')), waited=waited, healed=ok) if not ok: logger.error("等待 %s 秒后 %s:%s 仍不通，请检查: systemctl status %s", int(waited), host, port, unit) return result # 分支 b：执行 autossh.sh script = _cfg('script', 'autossh.sh') script_path = Path(script) if not script_path.is_absolute(): script_path = Path(config.BASE_DIR) / script if dry_run: result.update(status=ST_SCRIPT_STARTED, healed=False, detail=f'[dry-run] 将执行 {script_path}') return result if not script_path.exists(): logger.error("隧道脚本不存在: %s（config.yml: mysql.tunnel.script）", script_path) result.update(status=ST_UNAVAILABLE, detail=f'脚本不存在: {script_path}') return result logger.warning("数据库端口 %s:%s 不通%s，执行 %s", host, port, prefix, script_path.name) try: proc = subprocess.run(['bash', str(script_path)], capture_output=True, text=True, timeout=60) if proc.returncode != 0: logger.warning("%s 退出码 %s: %s", script_path.name, proc.returncode, (proc.stderr or proc.stdout or '').strip()[:300]) except (OSError, subprocess.SubprocessError) as e: logger.error("执行 %s 失败: %s", script_path, e) waited = _wait_for_port(host, port, wait_seconds) ok = port_open(host, port) result.update(status=ST_SCRIPT_STARTED if ok else ST_UNAVAILABLE, detail=(f'执行 {script_path.name}' + ('后端口已就绪' if ok else '后端口仍不通')), waited=waited, healed=ok) if ok: logger.info("✓ 隧道已恢复：%s:%s（等待 %.0f 秒）", host, port, waited) else: logger.error("执行 %s 并等待 %.0f 秒后 %s:%s 仍不通，请手动检查隧道", script_path.name, waited, host, port) return result def main(): ap = argparse.ArgumentParser(description='数据库隧道探测/自愈') ap.add_argument('--dry-run', action='store_true', help='只报告会做什么，不执行脚本') ap.add_argument('--quiet', action='store_true', help='只输出一行结果（供脚本调用）') args = ap.parse_args() logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') r = ensure_tunnel(reason='手动探测', dry_run=args.dry_run) if args.quiet: print(f"{r['status']} {r['host']}:{r['port']} {r['detail']}") else: print(f"状态: {r['status']}\n地址: {r['host']}:{r['port']}\n说明: {r['detail']}\n等待: {r['waited']:.0f}s") # 退出码：端口可连才算成功（dry-run 不判定） if args.dry_run: return 0 return 0 if port_open(r['host'], r['port']) else 1 if __name__ == '__main__': sys.exit(main())