349 lines
19 KiB
Python
349 lines
19 KiB
Python
"""Admin-enabled, bounded diagnostics with independent, short DB probes."""
|
||
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import time
|
||
from collections import deque
|
||
from urllib.parse import quote, urlencode
|
||
|
||
from .docker import DockerEngine, DockerError, decode_docker_output
|
||
from .host_metrics import rates
|
||
from .telemetry import RequestRecorder
|
||
|
||
|
||
def redact_logs(value):
|
||
value = re.sub(r"(?i)([a-z][a-z0-9+.-]*://)[^\s/@]+:[^\s/@]+@", r"\1[REDACTED]@", value)
|
||
value = re.sub(r"(?i)(Bearer\s+)[^\s\"']+", r"\1[REDACTED]", value)
|
||
value = re.sub(r"(?i)((?:password|passwd|api[_-]?key|access[_-]?token|secret|authorization|token)[\"']?\s*[:=]\s*[\"']?)[^\s,\"'}]+", r"\1[REDACTED]", value)
|
||
value = re.sub(r"\beyJ[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+", "[REDACTED_JWT]", value)
|
||
value = re.sub(r"\bsk-[a-zA-Z0-9_-]{12,}", "[REDACTED_KEY]", value)
|
||
return value
|
||
|
||
|
||
def container_metrics(detail, stats, previous_stats=None):
|
||
cpu = stats.get("cpu_stats", {})
|
||
previous = previous_stats.get("cpu_stats", {}) if previous_stats else stats.get("precpu_stats", {})
|
||
delta = cpu.get("cpu_usage", {}).get("total_usage", 0) - previous.get("cpu_usage", {}).get("total_usage", 0)
|
||
system_delta = cpu.get("system_cpu_usage", 0) - previous.get("system_cpu_usage", 0)
|
||
cores = cpu.get("online_cpus") or len(cpu.get("cpu_usage", {}).get("percpu_usage", [])) or 1
|
||
memory = stats.get("memory_stats", {})
|
||
cache = memory.get("stats", {}).get("inactive_file", memory.get("stats", {}).get("total_inactive_file", 0))
|
||
state = detail.get("State", {})
|
||
return {
|
||
"name": detail.get("Name", "").lstrip("/"),
|
||
"image": detail.get("Config", {}).get("Image", ""),
|
||
"image_id": detail.get("Image", "")[:22],
|
||
"created_at": detail.get("Created"),
|
||
"started_at": state.get("StartedAt"),
|
||
"status": state.get("Status"),
|
||
"health": state.get("Health", {}).get("Status"),
|
||
"restarts": detail.get("RestartCount", 0),
|
||
"oom_killed": state.get("OOMKilled", False),
|
||
"cpu_percent": round(delta / system_delta * cores * 100, 2) if previous.get("system_cpu_usage") and system_delta > 0 and delta >= 0 else None,
|
||
"cpu_limit": detail.get("HostConfig", {}).get("NanoCpus", 0) / 1e9 or None,
|
||
"memory_bytes": max(0, memory.get("usage", 0) - cache),
|
||
"memory_limit": memory.get("limit"),
|
||
"pids": stats.get("pids_stats", {}).get("current"),
|
||
"log_policy": detail.get("HostConfig", {}).get("LogConfig", {}),
|
||
"host_proc_mounted": any(m.get("Destination") == "/host/proc" for m in detail.get("Mounts", [])),
|
||
"docker_socket_mounted": any(m.get("Destination") == "/var/run/docker.sock" for m in detail.get("Mounts", [])),
|
||
}
|
||
|
||
|
||
class Diagnostics:
|
||
def __init__(self, pool, app):
|
||
self.pool = pool
|
||
self.app = app
|
||
self.path = pool.store.directory / "diagnostics-settings.json"
|
||
self.archive_directory = pool.store.directory / "log-archive"
|
||
self.enabled = json.loads(self.path.read_text()).get("enabled", False) if self.path.exists() else False
|
||
self.recorder = RequestRecorder()
|
||
self.recorder.enabled = self.enabled
|
||
app.state.request_recorder = self.recorder
|
||
self.history = deque(maxlen=720)
|
||
self.latest = None
|
||
self.previous = None
|
||
self.container_previous = {}
|
||
from .revision import code_revision
|
||
|
||
self.runtime_revision = code_revision()
|
||
self.task = None
|
||
self.lag_task = None
|
||
self.loop_lag_ms = 0
|
||
self.db_engines = {}
|
||
self.db = None
|
||
if self.enabled:
|
||
self._ensure_db()
|
||
|
||
def _ensure_db(self):
|
||
if self.db is None:
|
||
self.db = sqlite3.connect(self.pool.store.directory / "diagnostics.sqlite3")
|
||
self.db.execute("PRAGMA journal_mode=WAL")
|
||
self.db.execute("CREATE TABLE IF NOT EXISTS snapshots (id INTEGER PRIMARY KEY, at REAL, body TEXT)")
|
||
self.db.execute("CREATE TABLE IF NOT EXISTS requests (id INTEGER PRIMARY KEY, at REAL, body TEXT)")
|
||
for (value,) in reversed(self.db.execute("SELECT body FROM snapshots ORDER BY id DESC LIMIT 720").fetchall()):
|
||
self.history.append(json.loads(value))
|
||
for (value,) in reversed(self.db.execute("SELECT body FROM requests ORDER BY id DESC LIMIT 2000").fetchall()):
|
||
self.recorder.completed.append(json.loads(value))
|
||
return self.db
|
||
|
||
def start(self):
|
||
if not self.enabled:
|
||
return
|
||
self._ensure_db()
|
||
if self.task is not None and not self.task.done():
|
||
return
|
||
self.task = asyncio.create_task(self.loop())
|
||
self.lag_task = asyncio.create_task(self.measure_lag())
|
||
|
||
async def close(self):
|
||
for task in (self.task, self.lag_task):
|
||
if task:
|
||
task.cancel()
|
||
await asyncio.gather(*(t for t in (self.task, self.lag_task) if t), return_exceptions=True)
|
||
self.persist_requests()
|
||
if self.db is not None:
|
||
self.db.close()
|
||
self.db = None
|
||
for engine in self.db_engines.values():
|
||
await engine.dispose()
|
||
|
||
async def set_enabled(self, enabled):
|
||
self.enabled = enabled
|
||
self.recorder.enabled = enabled
|
||
temporary = self.path.with_suffix(".tmp")
|
||
fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
||
with os.fdopen(fd, "w") as stream:
|
||
json.dump({"enabled": enabled}, stream)
|
||
stream.flush()
|
||
os.fsync(stream.fileno())
|
||
os.replace(temporary, self.path)
|
||
if enabled:
|
||
self._ensure_db()
|
||
self.start()
|
||
else:
|
||
tasks = [task for task in (self.task, self.lag_task) if task]
|
||
for task in tasks:
|
||
if task:
|
||
task.cancel()
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
self.task = None
|
||
self.lag_task = None
|
||
self.previous = None
|
||
self.persist_requests()
|
||
|
||
async def measure_lag(self):
|
||
while True:
|
||
before = asyncio.get_running_loop().time()
|
||
await asyncio.sleep(1)
|
||
self.loop_lag_ms = round(max(0, asyncio.get_running_loop().time() - before - 1) * 1000, 2) if self.enabled else 0
|
||
|
||
async def loop(self):
|
||
while True:
|
||
if self.enabled:
|
||
try:
|
||
await self.sample()
|
||
except Exception as exc:
|
||
self.latest = {"at": time.time(), "error": f"采集失败（{type(exc).__name__}）"}
|
||
await asyncio.sleep(5)
|
||
|
||
def persist_requests(self):
|
||
batch = []
|
||
while self.recorder.pending:
|
||
row = self.recorder.pending.popleft()
|
||
batch.append((row["started"], json.dumps(row)))
|
||
if batch and self.db is not None:
|
||
self.db.executemany("INSERT INTO requests(at, body) VALUES (?, ?)", batch)
|
||
self.db.execute("DELETE FROM requests WHERE id <= (SELECT COALESCE(MAX(id),0)-10000 FROM requests)")
|
||
self.db.commit()
|
||
|
||
async def sample(self):
|
||
self._ensure_db()
|
||
settings = self.pool.store.load()
|
||
engine = DockerEngine(settings)
|
||
names = [settings.source_container, *list(self.pool.nodes)]
|
||
|
||
async def read_container(name):
|
||
try:
|
||
encoded = quote(name, safe="")
|
||
detail = await asyncio.to_thread(engine.request, "GET", f"/containers/{encoded}/json")
|
||
stats = await asyncio.to_thread(engine.request, "GET", f"/containers/{encoded}/stats?stream=false&one-shot=true")
|
||
key = detail.get("Id", name)
|
||
result = container_metrics(detail, stats, self.container_previous.get(key))
|
||
self.container_previous[key] = stats
|
||
return result
|
||
except Exception as exc:
|
||
return {"name": name, "error": f"节点指标不可用（{type(exc).__name__}）"}
|
||
|
||
async def host():
|
||
try:
|
||
raw = await asyncio.to_thread(engine.host_snapshot, settings.source_container)
|
||
result = rates(raw, self.previous)
|
||
self.previous = raw
|
||
return result
|
||
except Exception as exc:
|
||
detail = str(exc) if isinstance(exc, DockerError) else type(exc).__name__
|
||
return {"available": False, "message": f"宿主机采集不可用：{detail}。请检查只读挂载和 Docker 连接"}
|
||
|
||
host_data, containers, databases = await asyncio.gather(host(), asyncio.gather(*(read_container(n) for n in names if n)), self.database_metrics())
|
||
result = {"at": time.time(), "host": host_data, "containers": containers, "databases": databases, "event_loop_lag_ms": self.loop_lag_ms, "nodes": self.pool.status()["nodes"], "runtime_revision": self.runtime_revision}
|
||
self.latest = result
|
||
summary = {
|
||
"at": result["at"],
|
||
"cpu_percent": host_data.get("cpu_percent"),
|
||
"iowait_percent": host_data.get("iowait_percent"),
|
||
"memory_used_percent": round(100 * (1 - host_data["memory_available"] / host_data["memory_total"]), 2) if host_data.get("memory_total") else None,
|
||
"rx_bps": sum(v.get("rx_bps") or 0 for v in host_data.get("interfaces", {}).values()),
|
||
"tx_bps": sum(v.get("tx_bps") or 0 for v in host_data.get("interfaces", {}).values()),
|
||
"loop_lag_ms": self.loop_lag_ms,
|
||
"mysql_connections": databases.get("mysql", {}).get("connected"),
|
||
"postgres_connections": databases.get("postgres", {}).get("connected"),
|
||
"p95_headers_ms": self.recorder.summary()["p95_headers_ms"],
|
||
}
|
||
self.history.append(summary)
|
||
self.db.execute("INSERT INTO snapshots(at,body) VALUES (?,?)", (result["at"], json.dumps(summary)))
|
||
self.db.execute("DELETE FROM snapshots WHERE id <= (SELECT COALESCE(MAX(id),0)-720 FROM snapshots)")
|
||
self.db.commit()
|
||
self.persist_requests()
|
||
|
||
def status(self):
|
||
return {"enabled": self.enabled, "sample_interval_seconds": 5, "retained_samples": 720, "latest": self.latest, "history": list(self.history), "http": self.recorder.summary()}
|
||
|
||
async def database_metrics(self):
|
||
config = getattr(self.app.state, "config", None)
|
||
db = getattr(config, "database", None)
|
||
checkpoint = getattr(config, "checkpointer", None)
|
||
result = {}
|
||
if getattr(db, "backend", None) == "mysql":
|
||
result["mysql"] = await self.mysql_metrics(db.app_sqlalchemy_url)
|
||
else:
|
||
result["mysql"] = {"available": False, "reason": "业务库未使用 MySQL"}
|
||
from .preflight import postgres_connection_string
|
||
|
||
pg_url = postgres_connection_string(config)
|
||
if pg_url:
|
||
result["postgres"] = await self.postgres_metrics(pg_url)
|
||
else:
|
||
result["postgres"] = {"available": False, "configured": False, "reason": "当前聊天存储为 SQLite，未启用 PostgreSQL，不能显示为 0 个连接"}
|
||
result["chat_backend"] = "postgres" if pg_url else ("sqlite" if checkpoint and checkpoint.type in {"sqlite", "postgres"} else "unknown")
|
||
return result
|
||
|
||
async def mysql_metrics(self, url):
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import create_async_engine
|
||
|
||
key = "mysql"
|
||
try:
|
||
if key not in self.db_engines:
|
||
self.db_engines[key] = create_async_engine(url, pool_size=1, max_overflow=0, pool_timeout=1, pool_pre_ping=False, connect_args={"connect_timeout": 2})
|
||
async with asyncio.timeout(4):
|
||
async with self.db_engines[key].connect() as connection:
|
||
statuses = dict((await connection.execute(text("SHOW GLOBAL STATUS WHERE Variable_name IN ('Threads_connected','Threads_running','Max_used_connections','Slow_queries','Aborted_connects')"))).all())
|
||
variables = dict((await connection.execute(text("SHOW GLOBAL VARIABLES WHERE Variable_name IN ('max_connections','long_query_time')"))).all())
|
||
rows = (await connection.execute(text("SHOW PROCESSLIST"))).mappings().all()
|
||
running = [r for r in rows if str(r.get("Command", "")).lower() not in {"sleep", "daemon"}]
|
||
return {
|
||
"available": True,
|
||
"connected": int(statuses.get("Threads_connected", 0)),
|
||
"running": int(statuses.get("Threads_running", 0)),
|
||
"max_connections": int(variables.get("max_connections", 0)),
|
||
"peak_connections": int(statuses.get("Max_used_connections", 0)),
|
||
"slow_queries": int(statuses.get("Slow_queries", 0)),
|
||
"slow_query_threshold_seconds": float(variables.get("long_query_time", 0)),
|
||
"longest_active_seconds": max((int(r.get("Time", 0)) for r in running), default=0),
|
||
"processes_visible": len(rows),
|
||
"note": "连接总数为 MySQL 全局值（包含采集连接）；最长执行时间只覆盖当前账号可见进程，不读取 SQL 文本。",
|
||
}
|
||
except Exception as exc:
|
||
return {"available": False, "error": f"MySQL 指标不可用（{type(exc).__name__}），请检查连接与只读监控权限"}
|
||
|
||
async def postgres_metrics(self, url):
|
||
try:
|
||
import psycopg
|
||
|
||
async with asyncio.timeout(4):
|
||
async with await psycopg.AsyncConnection.connect(url, connect_timeout=2, autocommit=True) as connection:
|
||
async with connection.cursor() as cursor:
|
||
query = (
|
||
"SELECT count(*), count(*) FILTER (WHERE state='active'), "
|
||
"count(*) FILTER (WHERE wait_event_type='Lock'), "
|
||
"COALESCE(max(EXTRACT(EPOCH FROM now()-query_start)) "
|
||
"FILTER (WHERE state='active' AND pid<>pg_backend_pid()), 0) "
|
||
"FROM pg_stat_activity"
|
||
)
|
||
await cursor.execute(query)
|
||
connected, active, locks, longest = await cursor.fetchone()
|
||
await cursor.execute("SHOW max_connections")
|
||
maximum = (await cursor.fetchone())[0]
|
||
return {
|
||
"available": True,
|
||
"configured": True,
|
||
"connected": connected,
|
||
"active": active,
|
||
"lock_waits": locks,
|
||
"max_connections": int(maximum),
|
||
"longest_active_seconds": float(longest),
|
||
"note": "按 pg_stat_activity 采集，普通账号可能看不到其他账号的状态和耗时；包含采集连接，不读取 SQL 文本。",
|
||
}
|
||
except Exception as exc:
|
||
return {"available": False, "configured": True, "error": f"PostgreSQL 指标不可用（{type(exc).__name__}），请检查连接与监控权限"}
|
||
|
||
async def logs(self, name, tail=200, minutes=60, query=""):
|
||
if name.startswith("archive:"):
|
||
return self.read_archive(name[len("archive:") :], query)
|
||
settings = self.pool.store.load()
|
||
if name != settings.source_container and name not in self.pool.nodes:
|
||
raise DockerError("只能查看当前主服务和本 Worker 池的日志")
|
||
engine = DockerEngine(settings)
|
||
encoded = quote(name, safe="")
|
||
detail = await asyncio.to_thread(engine.request, "GET", f"/containers/{encoded}/json")
|
||
if name != settings.source_container and detail.get("Config", {}).get("Labels", {}).get("deerflow.worker-pool") != self.pool.store.raw()["instance"]:
|
||
raise DockerError("节点不属于当前 Worker 池")
|
||
params = urlencode({"stdout": 1, "stderr": 1, "timestamps": 1, "tail": min(1000, max(1, tail)), "since": int(time.time() - min(1440, max(1, minutes)) * 60)})
|
||
data = await asyncio.to_thread(engine.request, "GET", f"/containers/{encoded}/logs?{params}", raw=True, max_bytes=1024 * 1024)
|
||
lines = redact_logs(decode_docker_output(data)).splitlines()
|
||
if query:
|
||
lines = [line for line in lines if query.casefold() in line.casefold()]
|
||
return {"node": name, "lines": lines[-1000:], "log_policy": detail.get("HostConfig", {}).get("LogConfig", {}), "limited_bytes": 1024 * 1024}
|
||
|
||
async def archive_logs(self, engine, name):
|
||
"""Keep a bounded redacted tail BEFORE removing an owned container."""
|
||
import gzip
|
||
|
||
encoded = quote(name, safe="")
|
||
params = urlencode({"stdout": 1, "stderr": 1, "timestamps": 1, "tail": 10000})
|
||
data = await asyncio.to_thread(engine.request, "GET", f"/containers/{encoded}/logs?{params}", raw=True, max_bytes=2 * 1024 * 1024)
|
||
archive_id = f"{name}-{time.time_ns()}"
|
||
payload = {"node": name, "created_at": time.time(), "lines": redact_logs(decode_docker_output(data)).splitlines(), "limited_bytes": 2 * 1024 * 1024, "note": "节点删除前的最近 10000 行 / 最多 2 MB 归档；保存 7 天，最多 50 份。"}
|
||
self.archive_directory.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||
target = self.archive_directory / f"{archive_id}.json.gz"
|
||
fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||
with os.fdopen(fd, "wb") as out:
|
||
out.write(gzip.compress(json.dumps(payload).encode()))
|
||
files = sorted(self.archive_directory.glob("*.json.gz"), key=lambda p: p.stat().st_mtime, reverse=True)
|
||
for index, path in enumerate(files):
|
||
if index >= 50 or path.stat().st_mtime < time.time() - 7 * 86400:
|
||
path.unlink()
|
||
|
||
def archives(self):
|
||
return [
|
||
{"id": p.name.removesuffix(".json.gz"), "name": p.name.removesuffix(".json.gz"), "created_at": p.stat().st_mtime}
|
||
for p in sorted(self.archive_directory.glob("*.json.gz"), key=lambda p: p.stat().st_mtime, reverse=True)[:50]
|
||
if p.stat().st_mtime >= time.time() - 7 * 86400
|
||
]
|
||
|
||
def read_archive(self, archive_id, query=""):
|
||
import gzip
|
||
|
||
if not re.fullmatch(r"[A-Za-z0-9_.-]{1,180}", archive_id):
|
||
raise DockerError("无效日志归档")
|
||
path = self.archive_directory / f"{archive_id}.json.gz"
|
||
if not path.is_file() or path.stat().st_mtime < time.time() - 7 * 86400:
|
||
raise DockerError("日志归档不存在或已过期")
|
||
payload = json.loads(gzip.decompress(path.read_bytes()))
|
||
payload["lines"] = [line for line in payload["lines"] if query.casefold() in line.casefold()][-1000:]
|
||
return payload
|