wangyunlong
36fe7b81f8
cookie_gate.py(账号三级门控/阻断记录 pending_reruns/恢复重跑管理器)为此前 接线提交遗漏的模块文件；deploy/watchdog 为控制台与调度器的进程外保活脚本。
337 lines
11 KiB
Python
337 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""控制台与调度器的进程外保活（登录自启动，随用户会话常驻）。
|
||
|
||
职责（不承载任何业务调度规则，业务调度仍由 `gyxx schedule run` 唯一负责）：
|
||
1. 控制台：探测 http://127.0.0.1:8765/api/health——进程不在则拉起；
|
||
进程在但探测连续失败（wedged）则结束该进程树后拉起。控制台进程内
|
||
另有看门狗（console.py `_console_watchdog`）自检失败会主动退出，此处
|
||
负责把退出后的空窗补上，两层互为补充。
|
||
2. 调度器：读 `var/state/scheduler/service.lock` 的 PID 判断存活，锁主
|
||
已死则按 install.ps1 同款参数拉起 `python -m gyxx_flow schedule run`；
|
||
调度器自身单实例锁保证不会双跑。
|
||
3. 节流：每个目标有冷却间隔；短时间反复重启触发熔断，只记录不再拉起，
|
||
避免故障风暴。
|
||
|
||
用法：
|
||
gyxx_keepalive.py # 常驻循环（Startup 的 vbs 以 pythonw 启动）
|
||
gyxx_keepalive.py --once # 单轮探测+处置后退出，供验收/手动巡检
|
||
|
||
日志：var/logs/keepalive/keepalive.log（追加）。
|
||
自身单实例：var/state/keepalive.lock（PID 存活检测，重复启动直接退出）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import ctypes
|
||
import http.client
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||
ENV_FILES = [
|
||
Path(os.getenv("GYXX_RUNTIME_ENV_FILE", r"D:\product-collector-analyze-flow\.env")),
|
||
Path(
|
||
os.getenv(
|
||
"GYXX_RUNTIME_JUSHUITAN_ENV_FILE",
|
||
r"D:\product-collector-analyze-flow\private-secrets\jushuitan.env",
|
||
)
|
||
),
|
||
]
|
||
CONSOLE_HOST = "127.0.0.1"
|
||
CONSOLE_PORT = int(os.getenv("GYXX_CONSOLE_PORT", "8765"))
|
||
LOOP_INTERVAL_SECONDS = float(os.getenv("GYXX_KEEPALIVE_INTERVAL_SECONDS", "30"))
|
||
PROBE_TIMEOUT_SECONDS = 5.0
|
||
PROBE_FAILURE_LIMIT = 3
|
||
RESTART_COOLDOWN_SECONDS = float(os.getenv("GYXX_KEEPALIVE_COOLDOWN_SECONDS", "120"))
|
||
TRIP_MAX_RESTARTS = 5
|
||
TRIP_WINDOW_SECONDS = 30 * 60
|
||
TRIP_PAUSE_SECONDS = 60 * 60
|
||
|
||
DATA_ROOT = PROJECT_ROOT / "var"
|
||
LOG_PATH = DATA_ROOT / "logs" / "keepalive" / "keepalive.log"
|
||
LOCK_PATH = DATA_ROOT / "state" / "keepalive.lock"
|
||
CONSOLE_LOG_DIR = DATA_ROOT / "logs" / "console-restarts"
|
||
SCHEDULER_LOG_DIR = DATA_ROOT / "logs" / "scheduler-service"
|
||
SCHEDULER_SERVICE_LOCK = DATA_ROOT / "state" / "scheduler" / "service.lock"
|
||
|
||
|
||
def _log(event: str, detail: str = "") -> None:
|
||
stamp = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S")
|
||
line = f"{stamp} {event} {detail}".rstrip()
|
||
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
with LOG_PATH.open("a", encoding="utf-8", newline="\n") as stream:
|
||
stream.write(line + "\n")
|
||
print(line, flush=True)
|
||
|
||
|
||
def _pid_alive(pid: int) -> bool:
|
||
if pid <= 0:
|
||
return False
|
||
kernel32 = ctypes.windll.kernel32
|
||
handle = kernel32.OpenProcess(0x1000, False, pid) # PROCESS_QUERY_LIMITED_INFORMATION
|
||
if not handle:
|
||
return False
|
||
try:
|
||
exit_code = ctypes.c_ulong()
|
||
if kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)):
|
||
return exit_code.value == 259 # STILL_ACTIVE
|
||
return False
|
||
finally:
|
||
kernel32.CloseHandle(handle)
|
||
|
||
|
||
def _acquire_self_lock() -> bool:
|
||
LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
if LOCK_PATH.is_file():
|
||
try:
|
||
existing = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
|
||
if _pid_alive(int(existing.get("pid", 0))):
|
||
return False
|
||
except Exception:
|
||
pass
|
||
LOCK_PATH.write_text(
|
||
json.dumps({"pid": os.getpid(), "started_at": datetime.now(timezone.utc).isoformat()}),
|
||
encoding="utf-8",
|
||
)
|
||
return True
|
||
|
||
|
||
def _console_env_files() -> list[Path]:
|
||
return [path for path in ENV_FILES if path.is_file()]
|
||
|
||
|
||
def _spawn(argv: list[str], log_dir: Path, log_prefix: str) -> int:
|
||
log_dir.mkdir(parents=True, exist_ok=True)
|
||
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
||
log_path = log_dir / f"{log_prefix}-{stamp}.log"
|
||
kwargs: dict[str, object] = {}
|
||
if os.name == "nt":
|
||
kwargs["creationflags"] = (
|
||
subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
|
||
)
|
||
else:
|
||
kwargs["start_new_session"] = True
|
||
with log_path.open("a", encoding="utf-8", newline="\n") as stream:
|
||
process = subprocess.Popen(
|
||
argv,
|
||
cwd=str(PROJECT_ROOT),
|
||
env=os.environ,
|
||
stdin=subprocess.DEVNULL,
|
||
stdout=stream,
|
||
stderr=subprocess.STDOUT,
|
||
shell=False,
|
||
**kwargs,
|
||
)
|
||
return process.pid
|
||
|
||
|
||
def _console_argv() -> list[str]:
|
||
argv = [
|
||
str(PROJECT_ROOT / ".venv" / "Scripts" / "python.exe"),
|
||
"-m",
|
||
"gyxx_flow",
|
||
"console",
|
||
"--host",
|
||
CONSOLE_HOST,
|
||
"--port",
|
||
str(CONSOLE_PORT),
|
||
]
|
||
for env_file in _console_env_files():
|
||
argv.extend(("--env-file", str(env_file)))
|
||
return argv
|
||
|
||
|
||
def _scheduler_argv() -> list[str]:
|
||
argv = [
|
||
str(PROJECT_ROOT / ".venv" / "Scripts" / "python.exe"),
|
||
"-m",
|
||
"gyxx_flow",
|
||
"schedule",
|
||
"run",
|
||
]
|
||
for env_file in _console_env_files():
|
||
argv.extend(("--env-file", str(env_file)))
|
||
return argv
|
||
|
||
|
||
def _probe_console() -> bool:
|
||
try:
|
||
connection = http.client.HTTPConnection(CONSOLE_HOST, CONSOLE_PORT, timeout=PROBE_TIMEOUT_SECONDS)
|
||
try:
|
||
connection.request("GET", "/api/health")
|
||
response = connection.getresponse()
|
||
response.read()
|
||
return response.status == 200
|
||
finally:
|
||
connection.close()
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def _port_owner_pid(port: int) -> int | None:
|
||
try:
|
||
output = subprocess.run(
|
||
["netstat", "-ano", "-p", "tcp"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=10,
|
||
).stdout
|
||
except Exception:
|
||
return None
|
||
needle = f"{CONSOLE_HOST}:{port}"
|
||
for line in output.splitlines():
|
||
parts = line.split()
|
||
if (
|
||
len(parts) >= 5
|
||
and parts[0].upper() == "TCP"
|
||
and parts[1] == needle
|
||
and parts[3].upper() == "LISTENING"
|
||
):
|
||
try:
|
||
return int(parts[4])
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _kill_pid_tree(pid: int) -> None:
|
||
if pid <= 0 or pid == os.getpid():
|
||
return
|
||
subprocess.run(
|
||
["taskkill", "/PID", str(pid), "/T", "/F"],
|
||
capture_output=True,
|
||
timeout=30,
|
||
)
|
||
|
||
|
||
def _scheduler_lock_pid() -> int | None:
|
||
try:
|
||
payload = json.loads(SCHEDULER_SERVICE_LOCK.read_text(encoding="utf-8"))
|
||
return int(payload.get("pid", 0)) or None
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
class _TargetState:
|
||
"""记录单目标的连续失败计数、冷却与熔断状态。"""
|
||
|
||
def __init__(self, name: str) -> None:
|
||
self.name = name
|
||
self.failures = 0
|
||
self.last_restart_at = 0.0
|
||
self.recent_restarts: list[float] = []
|
||
self.tripped_at: float | None = None
|
||
|
||
def tripped(self, now: float) -> bool:
|
||
if self.tripped_at is None:
|
||
return False
|
||
if now - self.tripped_at >= TRIP_PAUSE_SECONDS:
|
||
self.tripped_at = None
|
||
self.recent_restarts.clear()
|
||
_log("keepalive-trip-reset", f"target={self.name} 熔断窗口结束恢复守护")
|
||
return False
|
||
return True
|
||
|
||
def may_restart(self, now: float) -> bool:
|
||
if self.tripped(now):
|
||
return False
|
||
if now - self.last_restart_at < RESTART_COOLDOWN_SECONDS:
|
||
return False
|
||
self.recent_restarts = [
|
||
stamp for stamp in self.recent_restarts if now - stamp < TRIP_WINDOW_SECONDS
|
||
]
|
||
if len(self.recent_restarts) >= TRIP_MAX_RESTARTS:
|
||
self.tripped_at = now
|
||
_log(
|
||
"keepalive-tripped",
|
||
f"target={self.name} {TRIP_MAX_RESTARTS} 次重启/{TRIP_WINDOW_SECONDS // 60}分钟 "
|
||
f"暂停拉起{TRIP_PAUSE_SECONDS // 60}分钟，请人工检查日志",
|
||
)
|
||
return False
|
||
return True
|
||
|
||
def record_restart(self, now: float) -> None:
|
||
self.last_restart_at = now
|
||
self.recent_restarts.append(now)
|
||
|
||
|
||
def run_cycle(console: _TargetState, scheduler: _TargetState) -> None:
|
||
now = time.time()
|
||
|
||
# ---- 控制台 ----
|
||
if _probe_console():
|
||
console.failures = 0
|
||
else:
|
||
console.failures += 1
|
||
owner = _port_owner_pid(CONSOLE_PORT)
|
||
_log(
|
||
"console-probe-failed",
|
||
f"{console.failures}/{PROBE_FAILURE_LIMIT} port_owner_pid={owner}",
|
||
)
|
||
if console.failures >= PROBE_FAILURE_LIMIT and console.may_restart(now):
|
||
if owner is not None and _pid_alive(owner):
|
||
_log("console-kill-wedged", f"pid={owner} 探测持续失败，结束进程树")
|
||
_kill_pid_tree(owner)
|
||
time.sleep(3)
|
||
pid = _spawn(_console_argv(), CONSOLE_LOG_DIR, "keepalive-console")
|
||
console.record_restart(time.time())
|
||
console.failures = 0
|
||
_log("console-restarted", f"pid={pid}")
|
||
|
||
# ---- 调度器 ----
|
||
scheduler_pid = _scheduler_lock_pid()
|
||
scheduler_alive = scheduler_pid is not None and _pid_alive(scheduler_pid)
|
||
if not scheduler_alive and scheduler.may_restart(now):
|
||
pid = _spawn(_scheduler_argv(), SCHEDULER_LOG_DIR, "keepalive-start")
|
||
scheduler.record_restart(time.time())
|
||
_log(
|
||
"scheduler-restarted",
|
||
f"pid={pid} prev_lock_pid={scheduler_pid}",
|
||
)
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(description="GYXX Flow 控制台/调度器保活")
|
||
parser.add_argument("--once", action="store_true", help="单轮探测+处置后退出")
|
||
arguments = parser.parse_args(argv)
|
||
|
||
if not _acquire_self_lock():
|
||
_log("keepalive-duplicate-exit", "已有保活进程在运行")
|
||
return 0
|
||
|
||
console = _TargetState("console")
|
||
scheduler = _TargetState("scheduler")
|
||
missing_env = [str(path) for path in ENV_FILES if not path.is_file()]
|
||
_log(
|
||
"keepalive-start",
|
||
f"pid={os.getpid()} once={arguments.once}"
|
||
+ (f" missing_env={missing_env}" if missing_env else ""),
|
||
)
|
||
try:
|
||
while True:
|
||
run_cycle(console, scheduler)
|
||
if arguments.once:
|
||
_log("keepalive-once-done", "")
|
||
return 0
|
||
time.sleep(LOOP_INTERVAL_SECONDS)
|
||
except KeyboardInterrupt:
|
||
return 0
|
||
finally:
|
||
try:
|
||
if LOCK_PATH.is_file():
|
||
payload = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
|
||
if int(payload.get("pid", 0)) == os.getpid():
|
||
LOCK_PATH.unlink()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|