"""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