#!/usr/bin/env python3 from __future__ import annotations import argparse import calendar import json import os import re import socket import subprocess import sys import time from dataclasses import dataclass from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[3] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from deployments.offline.runtime.common import DeployError, load_env_file, log, mysql_client, run DEPLOYMENT_ROOT = (Path.home() / "nexuscrm-deployment").resolve() DEPLOYMENT_CONFIG = REPO_ROOT / "deployments" / "deployment-config.json" NGINX_LOG_DIR = Path("/var/log/nginx") PROD_ENV_ROOT = DEPLOYMENT_ROOT / "online" / "envs" / "prod" PROD_GUARD_ROOT = PROD_ENV_ROOT / "guard" MULTI_PROD_CURRENT_ROOT = DEPLOYMENT_ROOT / "online" / "envs" / "multi_prod" / "current" RUNTIME_ENV_FILE = MULTI_PROD_CURRENT_ROOT / "env" / "runtime.env" BACKEND_STDERR_LOG = MULTI_PROD_CURRENT_ROOT / "logs" / "backend.stderr.log" STATE_DIR = PROD_GUARD_ROOT / ".guard-state" STATE_FILE = STATE_DIR / "online-host-guard.json" DEFAULT_NGINX_CONF = Path("/etc/nginx/conf.d/nexuscrm-prod.conf") BACKEND_ERROR_PATTERNS = ( "context canceled", "broken pipe", "unhandled error", "panic", "fatal", "unknown column", ) BACKEND_NGINX_LOCATIONS = frozenset(("/services/", "/api/", "/product/", "/webhooks/", "/health", "/ws")) NGINX_LOCATION_RE = re.compile(r"^\s*location\s+(?:=\s+)?([^\s{]+)") ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") # Console-format (zerolog ConsoleWriter) line shape after ANSI strip: # 2026-06-15T12:55:14Z INF HTTP Request method=GET path=/x status=500 latency=12.3 ... CONSOLE_TS_RE = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)") CONSOLE_KV_RE = re.compile(r"(\w+)=(\S+)") # Transient/jittery checks: probe-style signals that routinely flap for a single # run (TiDB Cloud TLS handshake drops and isolated client-abort 499s). These # must survive a short streak before paging so one blip does not # alarm. Everything else is a real availability/business failure that must page # on the first failing run. NOISY_CHECK_THRESHOLD = { "db-monitor": 3, "nginx-499": 3, } FIXED_CHECK_THRESHOLD = { "tidb-sustained-pressure": 2, } CORRELATED_TIDB_LIVE_SIGNALS = frozenset( {"active_session_pressure", "active_long_query", "lock_wait"} ) SUSTAINED_TIDB_PROBE_SIGNALS = frozenset({"read_probe_slow"}) DEFAULT_CHECK_THRESHOLD = 1 EXPECTED_PROD_TIDB_PORT = "4000" EXPECTED_PROD_TIDB_DATABASE = "nexuscrm" class TiDBMonitorTimeout(DeployError): pass @dataclass(frozen=True) class TiDBPressureObservation: signals: frozenset[str] live_signals: frozenset[str] statement_signals: frozenset[str] permission_hot_path: bool details: str monitor_error: str = "" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--min-disk-free-gb", type=int, default=5) parser.add_argument("--scan-lines", type=int, default=200) parser.add_argument("--backend-scan-lines", type=int, default=50000) parser.add_argument("--db-window-minutes", type=int, default=15) parser.add_argument("--slow-request-threshold-ms", type=int, default=15000) parser.add_argument("--backend-impact-count-threshold", type=int, default=3) parser.add_argument("--nginx-499-threshold", type=int, default=5) parser.add_argument("--tidb-monitor-query-timeout-seconds", type=float, default=8.0) parser.add_argument("--tidb-read-probe-threshold-seconds", type=float, default=3.0) parser.add_argument("--tidb-active-session-threshold", type=int, default=20) parser.add_argument("--tidb-long-query-threshold-seconds", type=int, default=10) parser.add_argument("--tidb-long-query-count-threshold", type=int, default=3) parser.add_argument("--tidb-lock-wait-count-threshold", type=int, default=5) parser.add_argument("--tidb-statement-min-exec-count", type=int, default=3) parser.add_argument("--tidb-signal-freshness-minutes", type=int, default=7) parser.add_argument("--tidb-statement-latency-threshold-seconds", type=float, default=15.0) parser.add_argument("--tidb-wait-threshold-seconds", type=float, default=5.0) parser.add_argument("--tidb-resolve-lock-threshold-seconds", type=float, default=1.0) parser.add_argument("--tidb-commit-threshold-seconds", type=float, default=5.0) parser.add_argument("--permission-slow-read-threshold-ms", type=float, default=15000.0) parser.add_argument("--permission-slow-read-count-threshold", type=int, default=3) parser.add_argument("--notify-on-failure", action="store_true") parser.add_argument( "--failure-notify-threshold", type=int, default=3, help=( "consecutive failing runs with the same fingerprint before a transient " "probe check (db-monitor, nginx-499) notifies; real " "availability/business checks always notify on the first failure" ), ) return parser.parse_args() def public_base_url() -> str: return os.environ.get("PROD_PUBLIC_URL", "https://nexuscrm-new-prod.pingcap.com").strip() def healthcheck_host() -> str: explicit = os.environ.get("PROD_HEALTHCHECK_HOST", "").strip() if explicit: return explicit return public_base_url().replace("https://", "").replace("http://", "").split("/")[0] def runtime_env_file() -> Path: configured = os.environ.get("PROD_RUNTIME_ENV_FILE", "").strip() if configured: return Path(configured) return RUNTIME_ENV_FILE def backend_service(envv: dict[str, str]) -> str: explicit = os.environ.get("PROD_BACKEND_SERVICE", "").strip() if explicit and explicit != "nexuscrm-multi-prod-backend.service": return explicit port = str(envv.get("BACKEND_PORT", "")).strip() if port: return f"nexuscrm-multi-prod-slot-{port}-backend.service" return "nexuscrm-multi-prod-backend.service" def nginx_conf_file() -> Path: configured = os.environ.get("PROD_NGINX_CONF", "").strip() if configured: return Path(configured) return DEFAULT_NGINX_CONF def nginx_access_log() -> Path: default_path = os.environ.get("PROD_NGINX_ACCESS_LOG", "").strip() if default_path: return Path(default_path) return NGINX_LOG_DIR / "nexuscrm-prod.access.log" def backend_stderr_log(envv: dict[str, str] | None = None) -> Path: explicit = os.environ.get("PROD_BACKEND_STDERR_LOG", "").strip() if explicit: return Path(explicit) if envv: env_root = str(envv.get("ENV_ROOT", "")).strip() if env_root: return Path(env_root) / "logs" / "backend.stderr.log" return BACKEND_STDERR_LOG def parse_nginx_proxy_ports(nginx_conf: Path) -> list[str]: if not nginx_conf.is_file(): raise DeployError(f"prod nginx conf not found: {nginx_conf}") ports: list[str] = [] seen: set[str] = set() current_location: str | None = None pattern = re.compile(r"proxy_pass\s+http://127\.0\.0\.1:(\d+)\b") for line in nginx_conf.read_text().splitlines(): location_match = NGINX_LOCATION_RE.search(line) if location_match: current_location = location_match.group(1) elif line.strip().startswith("}"): current_location = None match = pattern.search(line) if not match or current_location not in BACKEND_NGINX_LOCATIONS: continue port = match.group(1) if port not in seen: ports.append(port) seen.add(port) if not ports: raise DeployError(f"prod nginx conf has no backend localhost proxy_pass port: {nginx_conf}") return ports def assert_backend_locations_present(nginx_conf: Path) -> None: if not nginx_conf.is_file(): raise DeployError(f"prod nginx conf not found: {nginx_conf}") present: set[str] = set() for line in nginx_conf.read_text().splitlines(): location_match = NGINX_LOCATION_RE.search(line) if location_match: present.add(location_match.group(1)) missing = sorted(BACKEND_NGINX_LOCATIONS - present) if missing: raise DeployError(f"prod nginx conf missing backend proxy locations: {', '.join(missing)} in {nginx_conf}") def resolve_live_backend_port(envv: dict[str, str]) -> str: nginx_ports = parse_nginx_proxy_ports(nginx_conf_file()) if len(nginx_ports) != 1: raise DeployError(f"prod nginx conf has multiple backend ports: {', '.join(nginx_ports)}") nginx_port = nginx_ports[0] runtime_port = str(envv.get("BACKEND_PORT", "")).strip() if not runtime_port: raise DeployError("prod runtime env missing BACKEND_PORT") if runtime_port != nginx_port: raise DeployError(f"prod runtime BACKEND_PORT={runtime_port} does not match nginx backend port={nginx_port}") return nginx_port def assert_disk_free(min_free_gb: int) -> None: result = subprocess.run( ["df", "-Pk", str(DEPLOYMENT_ROOT)], text=True, capture_output=True, check=False, ) if result.returncode != 0 or not result.stdout.strip(): raise DeployError("failed to read disk usage") lines = result.stdout.strip().splitlines() if len(lines) < 2: raise DeployError("unexpected df output") available_kb = int(lines[1].split()[3]) if available_kb < min_free_gb * 1024 * 1024: raise DeployError(f"disk free space below threshold: {available_kb // 1024 // 1024} GiB available") def service_is_active(service: str) -> bool: return subprocess.run(["systemctl", "is-active", "--quiet", service], check=False).returncode == 0 def tail_lines(log_path: Path, scan_lines: int) -> list[str]: if not log_path.is_file(): return [] result = subprocess.run( ["tail", "-n", str(scan_lines), str(log_path)], text=True, capture_output=True, check=False, ) if result.returncode != 0: return [] return result.stdout.splitlines() def collect_recent_http_status(log_path: Path, scan_lines: int, prefix: str) -> list[str]: matches = [] for line in tail_lines(log_path, scan_lines): fields = line.split() if len(fields) < 9: continue status = fields[8] if len(status) == 3 and status.startswith(prefix): matches.append(line) return matches def detect_recent_http_status(log_path: Path, scan_lines: int, prefix: str) -> str: matches = collect_recent_http_status(log_path, scan_lines, prefix) if not matches: return "" return " || ".join(matches[-3:]) def format_nginx_499_alert(recent_499_matches: list[str]) -> tuple[str, str]: count = len(recent_499_matches) failure = ( f"recent client-closed requests (499) detected in nginx access log: count={count}; " "usually caused by browser refresh, client cancellation, or network interruption" ) details = " || ".join(recent_499_matches[-3:]) return failure, details def detect_recent_backend_errors(log_path: Path, scan_lines: int) -> str: matches = [] for line in tail_lines(log_path, scan_lines): lowered = line.lower() if any(pattern in lowered for pattern in BACKEND_ERROR_PATTERNS): matches.append(line) if not matches: return "" return " || ".join(matches[-3:]) def strip_ansi(value: str) -> str: return ANSI_RE.sub("", value) def _rfc3339_to_epoch(value: str) -> float | None: """Convert a UTC RFC3339 timestamp (trailing Z) to unix seconds. zerolog ConsoleWriter emits UTC with a `Z` suffix; timegm interprets the parsed struct as UTC unambiguously (mktime would assume local time). """ try: return float(calendar.timegm(time.strptime(value[:19], "%Y-%m-%dT%H:%M:%S"))) except (ValueError, TypeError): return None def _coerce_number(value: str) -> float | int | str: try: if "." in value: return float(value) return int(value) except (ValueError, TypeError): return value def parse_backend_events(log_path: Path, scan_lines: int) -> list[dict]: """Normalize backend stderr lines into event dicts, format-agnostically. prod currently emits zerolog ConsoleWriter output (despite LOG_FORMAT=json, because logging.InitLogger hardcodes ConsoleWriter), e.g.: 2026-06-15T12:55:14Z INF HTTP Request method=GET status=500 latency=12.3 ... If the backend is later fixed to emit JSON, those lines are parsed too. Both shapes are flattened to a common dict with `message`, `status` (int), `latency` (ms number), `time` (unix seconds), and any key=value fields. """ events: list[dict] = [] for raw in tail_lines(log_path, scan_lines): clean = strip_ansi(raw).strip() if not clean: continue if clean.startswith("{"): try: parsed = json.loads(clean) except (ValueError, TypeError): continue if isinstance(parsed, dict): events.append(parsed) continue event = _parse_console_line(clean) if event: events.append(event) return events def _parse_console_line(clean: str) -> dict | None: ts_match = CONSOLE_TS_RE.match(clean) if not ts_match: return None event: dict = {} epoch = _rfc3339_to_epoch(ts_match.group(1)) if epoch is not None: event["time"] = epoch # zerolog ConsoleWriter prints the message between the level and the fields, # e.g. "... INF HTTP Request method=GET ...". Recover it by taking the text # before the first key=value token. rest = clean[ts_match.end():].strip() first_kv = CONSOLE_KV_RE.search(rest) head = rest[: first_kv.start()] if first_kv else rest head_parts = head.split(None, 1) # drop the level token (INF/WRN/ERR) if len(head_parts) == 2: event["message"] = head_parts[1].strip() for key, value in CONSOLE_KV_RE.findall(rest): event[key] = _coerce_number(value) # zerolog Dur() renders to ms; the JSON field is `latency`, console too. return event def within_window(event: dict, window_start_epoch: float) -> bool: """Keep events whose `time` field is inside the window. Events without a usable timestamp are kept (fail-open on time only) so a log-format change never silently drops health signal. """ ts = event.get("time") if isinstance(ts, (int, float)): return ts >= window_start_epoch return True def http_request_events(events: list[dict], window_start_epoch: float) -> list[dict]: return [ e for e in events if e.get("message") == "HTTP Request" and within_window(e, window_start_epoch) ] def format_http_event(event: dict) -> str: return ( f"{event.get('method', '')} {event.get('path', '')} " f"status={event.get('status', '')} " f"latency_ms={event.get('latency', '')} " f"request_id={event.get('request_id', '')} " f"email={event.get('email', '')} " f"user={event.get('user_id', '')}" ).strip() def detect_backend_5xx(events: list[dict], window_start_epoch: float) -> str: hits = [ e for e in http_request_events(events, window_start_epoch) if isinstance(e.get("status"), int) and e["status"] >= 500 ] if not hits: return "" return " || ".join(format_http_event(e) for e in hits[-3:]) def count_backend_5xx(events: list[dict], window_start_epoch: float) -> int: return sum( 1 for event in http_request_events(events, window_start_epoch) if isinstance(event.get("status"), int) and event["status"] >= 500 ) def format_backend_5xx_alert( window_minutes: int, count: int, count_threshold: int, details: str, ) -> tuple[str, str] | None: if count < count_threshold: return None return ( f"repeated backend access 5xx detected within {window_minutes}m", f"count={count}; threshold={count_threshold}; samples={details}", ) def detect_slow_requests(events: list[dict], window_start_epoch: float, threshold_ms: int) -> tuple[int, str]: hits = [] for e in http_request_events(events, window_start_epoch): latency = e.get("latency") if isinstance(latency, (int, float)) and latency >= threshold_ms: hits.append(e) if not hits: return 0, "" hits.sort(key=lambda e: e.get("latency", 0), reverse=True) return len(hits), " || ".join(format_http_event(e) for e in hits[:3]) def format_slow_request_alert( window_minutes: int, count: int, count_threshold: int, latency_threshold_ms: int, details: str, ) -> tuple[str, str] | None: if count < count_threshold: return None return ( f"repeated slow requests detected within {window_minutes}m; " f"threshold_ms={latency_threshold_ms}", f"count={count}; threshold={count_threshold}; samples={details}", ) def detect_permission_read_pressure( events: list[dict], window_start_epoch: float, threshold_ms: float, ) -> tuple[int, str]: """Summarize repeated slow field-permission bulk loads without user data. Permission diagnostics aggregate both call count and duration per request. Use conservative latency units capped by the call count so one request with three calls and only 20 seconds total cannot masquerade as three 15-second reads, while a request with three calls and 45 seconds total can contribute all three reads. """ hits: list[tuple[float, str, int, int, int]] = [] for event in events: if event.get("message") != "Permission diagnostics" or not within_window( event, window_start_epoch ): continue calls = event.get("perm_field_bulk_calls") duration_ms = event.get("perm_field_bulk_ms") if not isinstance(calls, (int, float)) or calls < 1: continue if not isinstance(duration_ms, (int, float)) or duration_ms < threshold_ms: continue bulk_calls = int(calls) slow_read_units = min( bulk_calls, max(1, int(float(duration_ms) // max(1.0, threshold_ms))), ) rows = event.get("perm_field_bulk_rows") hits.append( ( float(duration_ms), str(event.get("path") or event.get("route") or "unknown")[:120], int(rows) if isinstance(rows, (int, float)) else 0, bulk_calls, slow_read_units, ) ) if not hits: return 0, "" hits.sort(reverse=True) slow_read_count = sum(units for _duration, _path, _rows, _calls, units in hits) bulk_call_count = sum(calls for _duration, _path, _rows, calls, _units in hits) total_ms = sum(duration for duration, _path, _rows, _calls, _units in hits) avg_ms = total_ms / len(hits) samples = " || ".join( f"path={path} duration_ms={duration:.0f} calls={calls} rows={rows}" for duration, path, rows, calls, _units in hits[:3] ) return ( slow_read_count, f"slow_permission_reads={slow_read_count} slow_permission_requests={len(hits)} " f"bulk_calls={bulk_call_count} total_ms={total_ms:.0f} avg_ms={avg_ms:.0f} " f"max_ms={hits[0][0]:.0f}; samples={samples}", ) def detect_business_alerts(events: list[dict], window_start_epoch: float) -> str: """Catch any log line that opts in with an `alert` field. This is the extensible path: backend code can flag gateway timeouts, upstream degradation, or any new condition by emitting a structured `alert` field, without a guard-side change. """ hits = [ e for e in events if str(e.get("alert", "")).strip() and within_window(e, window_start_epoch) ] if not hits: return "" rendered = [] for e in hits[-3:]: rendered.append( f"alert={e.get('alert', '')} " f"{e.get('message', '')} " f"request_id={e.get('request_id', '')}".strip() ) return " || ".join(rendered) def load_prod_runtime_env() -> dict[str, str]: path = runtime_env_file() if not path.is_file(): raise DeployError(f"prod runtime env not found: {path}") return load_env_file(path) def non_prod_tidb_bindings() -> dict[str, set[str]]: config = json.loads(DEPLOYMENT_CONFIG.read_text()) users: set[str] = set() branch_ids: set[str] = set() for env_name, env_config in config.get("environments", {}).items(): if env_name == "multi_prod": continue topology = env_config.get("topology", {}) for user in topology.get("additional_non_prod_tidb_users", []): user = str(user).strip() if user: users.add(user) for branch_id in topology.get("additional_non_prod_tidb_branch_ids", []): branch_id = str(branch_id).strip() if branch_id: branch_ids.add(branch_id) target_config = topology.get("target_config", {}) for target in target_config.values(): tidb = target.get("tidb", {}) user = str(tidb.get("user", "")).strip() branch_id = str(tidb.get("branch_id", "")).strip() if user: users.add(user) if branch_id: branch_ids.add(branch_id) return {"users": users, "branch_ids": branch_ids} def validate_prod_runtime_binding(envv: dict[str, str]) -> None: expected = { "TIDB_PORT": EXPECTED_PROD_TIDB_PORT, "TIDB_DATABASE": EXPECTED_PROD_TIDB_DATABASE, } mismatches = [ f"{key}={envv.get(key, '')!r} expected {value!r}" for key, value in expected.items() if str(envv.get(key, "")).strip() != value ] forbidden = non_prod_tidb_bindings() tidb_host = str(envv.get("TIDB_HOST", "")).strip() tidb_user = str(envv.get("TIDB_USER", "")).strip() recovery_branch = str(envv.get("PROD_RECOVERY_TIDB_CLOUD_BRANCH_ID", "")).strip() if not tidb_host: mismatches.append("TIDB_HOST is empty") if not tidb_user: mismatches.append("TIDB_USER is empty") if tidb_user.lower() == "root" or tidb_user.lower().endswith(".root"): mismatches.append("TIDB_USER must be the least-privilege Prod runtime identity, not root") if tidb_user and tidb_user in forbidden["users"]: mismatches.append(f"TIDB_USER matches non-prod target {tidb_user!r}") if recovery_branch and recovery_branch in forbidden["branch_ids"]: mismatches.append(f"PROD_RECOVERY_TIDB_CLOUD_BRANCH_ID matches non-prod target {recovery_branch!r}") if mismatches: raise DeployError("invalid prod runtime TiDB binding: " + "; ".join(mismatches)) def db_monitoring_env() -> dict[str, str]: envv = load_prod_runtime_env() validate_prod_runtime_binding(envv) for key in ("TIDB_HOST", "TIDB_PORT", "TIDB_USER", "TIDB_DATABASE"): if not envv.get(key): raise DeployError(f"prod runtime env missing {key}") return envv def db_query_with_timeout( envv: dict[str, str], sql: str, *, database: str, timeout_seconds: float = 8.0, ) -> str: timeout_seconds = max(1.0, float(timeout_seconds)) server_timeout_ms = max(100, int((timeout_seconds - 0.5) * 1000)) bounded_sql = f"SET SESSION max_execution_time={server_timeout_ms};\n{sql}" args = mysql_client(envv) + [ f"-h{envv['TIDB_HOST']}", f"-P{envv['TIDB_PORT']}", f"-u{envv['TIDB_USER']}", "--connect-timeout=5", "-N", "-B", database, "-e", bounded_sql, ] child_env = { key: value for key in ("HOME", "LANG", "LC_ALL", "PATH", "TZ") if (value := os.environ.get(key)) is not None } if envv.get("TIDB_PASSWORD"): child_env["MYSQL_PWD"] = envv["TIDB_PASSWORD"] try: result = subprocess.run( args, env=child_env, text=True, capture_output=True, check=False, timeout=timeout_seconds, ) except subprocess.TimeoutExpired as exc: raise TiDBMonitorTimeout( f"TiDB monitoring query exceeded {timeout_seconds:g}s hard timeout" ) from exc if result.returncode != 0: error = " ".join(result.stderr.strip().split())[:400] raise DeployError( f"TiDB monitoring query failed with exit={result.returncode}: {error or 'no stderr'}" ) return result.stdout.strip() def _metric_seconds(raw: str) -> float: try: return float(raw) except (TypeError, ValueError): return 0.0 def collect_tidb_pressure( envv: dict[str, str], args: argparse.Namespace, ) -> TiDBPressureObservation | None: signals: set[str] = set() live_signals: set[str] = set() statement_signals: set[str] = set() details: list[str] = [] permission_hot_path = False timeout_seconds = args.tidb_monitor_query_timeout_seconds probe_started = time.monotonic() try: db_query_with_timeout( envv, "SELECT /*+ MAX_EXECUTION_TIME(5000) */ 1 FROM _System_InstalledPack LIMIT 1", database=envv["TIDB_DATABASE"], timeout_seconds=timeout_seconds, ) except TiDBMonitorTimeout as exc: signals.add("read_probe_timeout") live_signals.add("read_probe_timeout") details.append(str(exc)) return TiDBPressureObservation( frozenset(signals), frozenset(live_signals), frozenset(), False, "; ".join(details), ) probe_seconds = time.monotonic() - probe_started if probe_seconds >= args.tidb_read_probe_threshold_seconds: signals.add("read_probe_slow") live_signals.add("read_probe_slow") latency_ns = max(0, int(args.tidb_statement_latency_threshold_seconds * 1_000_000_000)) wait_ns = max(0, int(args.tidb_wait_threshold_seconds * 1_000_000_000)) resolve_lock_ns = max(0, int(args.tidb_resolve_lock_threshold_seconds * 1_000_000_000)) commit_ns = max(0, int(args.tidb_commit_threshold_seconds * 1_000_000_000)) sql = f""" SELECT 'process', COUNT(*), COALESCE(SUM(COMMAND <> 'Sleep'), 0), COALESCE(SUM(COMMAND <> 'Sleep' AND TIME >= {args.tidb_long_query_threshold_seconds}), 0), COALESCE(MAX(CASE WHEN COMMAND <> 'Sleep' THEN TIME END), 0) FROM INFORMATION_SCHEMA.CLUSTER_PROCESSLIST WHERE ID <> CONNECTION_ID(); SELECT 'lock', COUNT(*) FROM INFORMATION_SCHEMA.DATA_LOCK_WAITS; SELECT 'lock_sample', COALESCE(LEFT(SQL_DIGEST, 12), '-'), LEFT(REPLACE(REPLACE(COALESCE(SQL_DIGEST_TEXT, '-'), CHAR(9), ' '), CHAR(10), ' '), 160) FROM INFORMATION_SCHEMA.DATA_LOCK_WAITS LIMIT 3; SELECT 'statement', CASE WHEN STMT_TYPE = 'Select' AND ( LOWER(DIGEST_TEXT) LIKE '%fieldpermissions%' OR LOWER(DIGEST_TEXT) LIKE '%objectpermissions%' OR LOWER(DIGEST_TEXT) LIKE '%permissionsetpermission%' OR LOWER(DIGEST_TEXT) LIKE '%permissionsetassignment%' ) THEN 'permission_statement' ELSE 'statement' END, STMT_TYPE, LEFT(DIGEST, 12), EXEC_COUNT, ROUND(AVG_LATENCY / 1000000000, 3), ROUND(MAX_LATENCY / 1000000000, 3), ROUND(AVG_WAIT_TIME / 1000000000, 3), ROUND(MAX_WAIT_TIME / 1000000000, 3), ROUND(AVG_QUEUED_RC_TIME / 1000000000, 3), ROUND(MAX_QUEUED_RC_TIME / 1000000000, 3), ROUND(MAX_RESOLVE_LOCK_TIME / 1000000000, 3), ROUND(MAX_COMMIT_TIME / 1000000000, 3), GREATEST(0, TIMESTAMPDIFF(SECOND, LAST_SEEN, UTC_TIMESTAMP())), LEFT(REPLACE(REPLACE(DIGEST_TEXT, CHAR(9), ' '), CHAR(10), ' '), 160) FROM INFORMATION_SCHEMA.CLUSTER_STATEMENTS_SUMMARY_HISTORY WHERE LAST_SEEN >= UTC_TIMESTAMP() - INTERVAL {args.tidb_signal_freshness_minutes} MINUTE AND SCHEMA_NAME = DATABASE() AND EXEC_COUNT >= {args.tidb_statement_min_exec_count} AND ( AVG_LATENCY >= {latency_ns} OR MAX_LATENCY >= {latency_ns} OR AVG_WAIT_TIME >= {wait_ns} OR MAX_WAIT_TIME >= {wait_ns} OR AVG_QUEUED_RC_TIME >= {wait_ns} OR MAX_QUEUED_RC_TIME >= {wait_ns} OR MAX_RESOLVE_LOCK_TIME >= {resolve_lock_ns} OR MAX_COMMIT_TIME >= {commit_ns} ) ORDER BY GREATEST( MAX_LATENCY, MAX_WAIT_TIME, MAX_QUEUED_RC_TIME, MAX_RESOLVE_LOCK_TIME, MAX_COMMIT_TIME ) DESC LIMIT 20 """ try: raw = db_query_with_timeout( envv, sql, database=envv["TIDB_DATABASE"], timeout_seconds=timeout_seconds, ) except TiDBMonitorTimeout as exc: # A timeout in the diagnostic query is monitor failure, not evidence # that production query pressure crossed an incident threshold. Keep a # completed slow read probe, however, so its independent two-run streak # can still advance while db-monitor reports the diagnostic failure. details.append(f"read_probe={probe_seconds:.3f}s") return TiDBPressureObservation( frozenset(signals), frozenset(live_signals), frozenset(), False, "; ".join(details), monitor_error=str(exc), ) process_summary = "sessions unavailable" lock_waits = 0 lock_samples: list[str] = [] statement_samples: list[str] = [] seen_statement_digests: set[str] = set() for line in raw.splitlines(): columns = line.split("\t") if not columns: continue row_type = columns[0] if row_type == "process" and len(columns) >= 5: total_sessions = int(columns[1] or "0") active_sessions = int(columns[2] or "0") long_queries = int(columns[3] or "0") max_active_seconds = int(columns[4] or "0") process_summary = ( f"sessions total={total_sessions} active={active_sessions} " f"long_ge_{args.tidb_long_query_threshold_seconds}s={long_queries} " f"max_active={max_active_seconds}s" ) if active_sessions >= args.tidb_active_session_threshold: signals.add("active_session_pressure") live_signals.add("active_session_pressure") if long_queries >= args.tidb_long_query_count_threshold: signals.add("active_long_query") live_signals.add("active_long_query") elif row_type == "lock" and len(columns) >= 2: lock_waits = int(columns[1] or "0") if lock_waits >= args.tidb_lock_wait_count_threshold: signals.add("lock_wait") live_signals.add("lock_wait") elif row_type == "lock_sample" and len(columns) >= 3: lock_samples.append(f"digest={columns[1]} sql={columns[2]}") elif row_type == "statement" and len(columns) >= 15: statement_kind = columns[1] digest = columns[3] exec_count = int(columns[4] or "0") avg_latency = _metric_seconds(columns[5]) max_latency = _metric_seconds(columns[6]) avg_tikv_wait = _metric_seconds(columns[7]) max_tikv_wait = _metric_seconds(columns[8]) avg_ru_queue = _metric_seconds(columns[9]) max_ru_queue = _metric_seconds(columns[10]) resolve_lock = _metric_seconds(columns[11]) commit = _metric_seconds(columns[12]) age_seconds = int(columns[13] or "0") row_signals: set[str] = set() if avg_latency >= args.tidb_statement_latency_threshold_seconds: row_signals.add("statement_slow") if avg_tikv_wait >= args.tidb_wait_threshold_seconds: row_signals.add("tikv_wait") if avg_ru_queue >= args.tidb_wait_threshold_seconds: row_signals.add("ru_queue") signals.update(row_signals) statement_signals.update(row_signals) if statement_kind == "permission_statement" and ( row_signals or max_latency >= args.tidb_statement_latency_threshold_seconds or max_tikv_wait >= args.tidb_wait_threshold_seconds or max_ru_queue >= args.tidb_wait_threshold_seconds or resolve_lock >= args.tidb_resolve_lock_threshold_seconds or commit >= args.tidb_commit_threshold_seconds ): permission_hot_path = True signals.add("permission_hot_path") if digest not in seen_statement_digests and len(statement_samples) < 5: seen_statement_digests.add(digest) statement_samples.append( f"kind={statement_kind} type={columns[2]} digest={digest} exec={exec_count} " f"avg_latency={avg_latency:.3f}s max_latency={max_latency:.3f}s " f"avg_tikv_wait={avg_tikv_wait:.3f}s max_tikv_wait={max_tikv_wait:.3f}s " f"avg_ru_queue={avg_ru_queue:.3f}s max_ru_queue={max_ru_queue:.3f}s " f"resolve_lock={resolve_lock:.3f}s commit={commit:.3f}s " f"last_seen_age={age_seconds}s sql={columns[14]}" ) if not signals: return None details.extend( [ f"read_probe={probe_seconds:.3f}s", process_summary, f"active_lock_waits={lock_waits}", ] ) if lock_samples: details.append("lock_samples=" + " || ".join(lock_samples)) if statement_samples: details.append("statement_samples=" + " || ".join(statement_samples)) return TiDBPressureObservation( frozenset(signals), frozenset(live_signals), frozenset(statement_signals), permission_hot_path, "; ".join(details), ) def classify_tidb_pressure( observation: TiDBPressureObservation | None, *, permission_slow_count: int, permission_count_threshold: int, permission_details: str, backend_impact_details: str, ) -> list[tuple[str, str, str]]: if observation and "read_probe_timeout" in observation.live_signals: return [ ( "tidb-read-unavailable", "the bounded production TiDB read probe timed out", observation.details, ) ] correlated_live_signals = ( observation.live_signals & CORRELATED_TIDB_LIVE_SIGNALS if observation else frozenset() ) correlated_pressure = correlated_live_signals | ( observation.statement_signals if observation else frozenset() ) permission_degraded = permission_slow_count >= permission_count_threshold if permission_degraded: if observation and ( observation.permission_hot_path or correlated_pressure ): return [ ( "tidb-permission-incident", "sustained CRM permission reads are degraded with correlated TiDB pressure", f"{permission_details}; tidb={observation.details}", ) ] return [ ( "permission-read-degradation", "sustained CRM permission field bulk reads exceeded the latency threshold", permission_details, ) ] if not observation: return [] if backend_impact_details and correlated_pressure: return [ ( "tidb-service-impact", "live TiDB pressure is correlated with repeated CRM request failures or latency", f"impact={backend_impact_details}; tidb={observation.details}", ) ] sustained_pressure = correlated_pressure | ( observation.live_signals & SUSTAINED_TIDB_PROBE_SIGNALS ) if sustained_pressure: if correlated_pressure: failure = "live TiDB session or statement pressure remained above the incident threshold" else: failure = "the bounded production TiDB read probe remained slow" return [ ( "tidb-sustained-pressure", failure, observation.details, ) ] return [] def classify_tidb_monitor_outcome( observation: TiDBPressureObservation | None, monitor_error: str, *, permission_slow_count: int, permission_count_threshold: int, permission_details: str, backend_impact_details: str, ) -> list[tuple[str, str, str]]: failures = classify_tidb_pressure( observation, permission_slow_count=permission_slow_count, permission_count_threshold=permission_count_threshold, permission_details=permission_details, backend_impact_details=backend_impact_details, ) monitor_errors = [] if observation and observation.monitor_error: monitor_errors.append(observation.monitor_error) if monitor_error and monitor_error not in monitor_errors: monitor_errors.append(monitor_error) if monitor_errors: failures.append(("db-monitor", "; ".join(monitor_errors), "")) return failures def summarize_backend_impact( backend_5xx_count: int, slow_request_count: int, count_threshold: int, ) -> str: impact_parts: list[str] = [] if backend_5xx_count >= count_threshold: impact_parts.append(f"backend_5xx={backend_5xx_count}") if slow_request_count >= count_threshold: impact_parts.append(f"slow_requests={slow_request_count}") return ",".join(impact_parts) def sample_recent_failed_logins(envv: dict[str, str], window_minutes: int) -> str: sql = f""" SELECT CONCAT( IFNULL(Status, ''), ' @', DATE_FORMAT(LoginTime, '%Y-%m-%d %H:%i:%s'), ' ip=', IFNULL(SourceIp, ''), ' user=', IFNULL(UserId, '') ) FROM _System_LoginHistory WHERE LoginTime >= UTC_TIMESTAMP() - INTERVAL {window_minutes} MINUTE AND LOWER(IFNULL(Status, '')) NOT IN ('success', 'succeeded') ORDER BY LoginTime DESC LIMIT 3 """ return " || ".join( line.strip() for line in db_query_with_timeout(envv, sql, database=envv["TIDB_DATABASE"]).splitlines() if line.strip() ) def count_recent_failed_logins(envv: dict[str, str], window_minutes: int) -> int: sql = f""" SELECT COUNT(*) FROM _System_LoginHistory WHERE LoginTime >= UTC_TIMESTAMP() - INTERVAL {window_minutes} MINUTE AND LOWER(IFNULL(Status, '')) NOT IN ('success', 'succeeded') """ raw = db_query_with_timeout(envv, sql, database=envv["TIDB_DATABASE"]).strip() return int(raw or "0") def format_failed_login_alert(window_minutes: int, failed_login_count: int, details: str) -> tuple[str, str]: failure = ( f"recent failed login attempts detected within {window_minutes}m: count={failed_login_count}; " "check for bad password probes or failed operator verification" ) if not details: return failure, "" enriched_details = ( "Recent failed password-grant attempts were recorded in _System_LoginHistory. " "This means authentication failed; it does not indicate a successful login. " "Verify whether the source matches operator checks or an unexpected client. " f"Samples: {details}" ) return failure, enriched_details def sample_recent_flow_errors(envv: dict[str, str], window_minutes: int) -> str: sql = f""" SELECT CONCAT( IFNULL(Status, ''), ' ', IFNULL(ElementType, ''), ' ', IFNULL(ElementId, ''), ' record=', IFNULL(RecordId, ''), ' error=', LEFT(REPLACE(REPLACE(IFNULL(ErrorPayload, ''), CHAR(10), ' '), CHAR(13), ' '), 120) ) FROM _System_FlowAuditLog WHERE CreatedDate >= UTC_TIMESTAMP() - INTERVAL {window_minutes} MINUTE AND ( LOWER(IFNULL(Status, '')) IN ('error', 'failed', 'failure', 'panic') OR IFNULL(ErrorPayload, '') <> '' ) ORDER BY CreatedDate DESC LIMIT 3 """ return " || ".join( line.strip() for line in db_query_with_timeout(envv, sql, database=envv["TIDB_DATABASE"]).splitlines() if line.strip() ) def count_recent_flow_errors(envv: dict[str, str], window_minutes: int) -> int: sql = f""" SELECT COUNT(*) FROM _System_FlowAuditLog WHERE CreatedDate >= UTC_TIMESTAMP() - INTERVAL {window_minutes} MINUTE AND ( LOWER(IFNULL(Status, '')) IN ('error', 'failed', 'failure', 'panic') OR IFNULL(ErrorPayload, '') <> '' ) """ raw = db_query_with_timeout(envv, sql, database=envv["TIDB_DATABASE"]).strip() return int(raw or "0") def load_state() -> dict[str, dict]: if not STATE_FILE.is_file(): return {} try: return json.loads(STATE_FILE.read_text()) except Exception: return {} def save_state(state: dict[str, dict]) -> None: STATE_DIR.mkdir(parents=True, exist_ok=True) STATE_FILE.write_text(json.dumps(state, sort_keys=True) + "\n") def failure_fingerprint(check_name: str, failure: str, details: str) -> str: payload = {"check": check_name} if check_name in {"flow-audit", "login-history"}: # Their rolling counts and samples change every run but do not identify a new failure. return json.dumps(payload, sort_keys=True) payload["failure"] = failure if not check_name.startswith("tidb-") and check_name not in { "backend-access-5xx", "permission-read-degradation", "slow-request", }: payload["details"] = details return json.dumps(payload, sort_keys=True) def evaluate_failure_notifications( failures: list[tuple[str, str, str]], prev_state: dict[str, dict], noisy_threshold: int, ) -> tuple[dict[str, dict], list[tuple[str, str, str, str]]]: """Compute next state and the checks that should notify this run. Threshold is resolved per check: transient/jittery probe checks (`NOISY_CHECK_THRESHOLD`) must fail with the same fingerprint for `noisy_threshold` consecutive runs before notifying; every other check is a real availability/business failure and pages on the first failing run (`DEFAULT_CHECK_THRESHOLD`). Notification still dedups once per distinct fingerprint. Checks absent from `failures` are dropped, so a single recovered run resets the streak. The returned next state pre-marks notifying checks as notified; the caller is responsible for clearing that mark if delivery is skipped or fails so the alert retries on the next run. """ next_state: dict[str, dict] = {} to_notify: list[tuple[str, str, str, str]] = [] for check_name, failure, details in failures: if check_name in FIXED_CHECK_THRESHOLD: threshold = FIXED_CHECK_THRESHOLD[check_name] elif check_name in NOISY_CHECK_THRESHOLD: threshold = max(1, noisy_threshold) else: threshold = DEFAULT_CHECK_THRESHOLD fingerprint = failure_fingerprint(check_name, failure, details) prev = prev_state.get(check_name) prev = prev if isinstance(prev, dict) else {} count = prev.get("count", 0) + 1 if prev.get("fingerprint") == fingerprint else 1 notified_fingerprint = prev.get("notified_fingerprint") if count >= threshold and notified_fingerprint != fingerprint: to_notify.append((check_name, failure, details, fingerprint)) notified_fingerprint = fingerprint next_state[check_name] = { "count": count, "fingerprint": fingerprint, "notified_fingerprint": notified_fingerprint, } return next_state, to_notify def notify_failure(check_name: str, failure: str, details: str = "") -> None: app_id = os.environ.get("FEISHU_APP_ID", "").strip() app_secret = os.environ.get("FEISHU_APP_SECRET", "").strip() chat_id = os.environ.get("FEISHU_CHAT_ID", "").strip() if not app_id or not app_secret or not chat_id: log("online host guard notification skipped: FEISHU_APP_ID / FEISHU_APP_SECRET / FEISHU_CHAT_ID not fully configured") return run( [ "python3", str(REPO_ROOT / "scripts/notify_feishu_online_guard.py"), "--app-id", app_id, "--app-secret", app_secret, "--chat-id", chat_id, "--host", socket.gethostname(), "--environment", "prod", "--check", check_name, "--failure", failure, "--details", details, "--public-url", public_base_url(), ] ) def main() -> int: args = parse_args() state = load_state() failures: list[tuple[str, str, str]] = [] try: assert_disk_free(args.min_disk_free_gb) except DeployError as exc: failures.append(("disk", str(exc), "")) live_envv: dict[str, str] = {} live_backend_port = "" # Collect both live-routing probes into ONE failure entry per run: the # notify state machine keys on check_name, so a second same-name entry # would double-page and corrupt fingerprint/rollback bookkeeping. live_routing_errors: list[str] = [] try: assert_backend_locations_present(nginx_conf_file()) except DeployError as exc: live_routing_errors.append(str(exc)) try: live_envv = load_prod_runtime_env() validate_prod_runtime_binding(live_envv) live_backend_port = resolve_live_backend_port(live_envv) except DeployError as exc: live_routing_errors.append(str(exc)) if live_routing_errors: failures.append(("live-routing", "; ".join(dict.fromkeys(live_routing_errors)), "")) service = backend_service(live_envv) if not service_is_active(service): failures.append(("service-health", f"service not active: {service}", "")) else: try: if not live_backend_port: raise DeployError("live backend port is unavailable") run(["curl", "-fsS", "-o", "/dev/null", f"http://127.0.0.1:{live_backend_port}/health"]) run(["curl", "-fsS", "-o", "/dev/null", "-H", f"Host: {healthcheck_host()}", "http://127.0.0.1/health"]) run(["curl", "-fsS", "-o", "/dev/null", "-H", f"Host: {healthcheck_host()}", "http://127.0.0.1/index.html"]) except Exception as exc: failures.append(("service-health", str(exc), "")) backend_log = backend_stderr_log(live_envv) window_start_epoch = time.time() - args.db_window_minutes * 60 backend_events = parse_backend_events(backend_log, args.backend_scan_lines) backend_5xx_count = count_backend_5xx(backend_events, window_start_epoch) recent_5xx = detect_backend_5xx(backend_events, window_start_epoch) backend_5xx_alert = format_backend_5xx_alert( args.db_window_minutes, backend_5xx_count, args.backend_impact_count_threshold, recent_5xx, ) if backend_5xx_alert: failures.append( ("backend-access-5xx", *backend_5xx_alert) ) slow_count, slow_details = detect_slow_requests( backend_events, window_start_epoch, args.slow_request_threshold_ms ) slow_request_alert = format_slow_request_alert( args.db_window_minutes, slow_count, args.backend_impact_count_threshold, args.slow_request_threshold_ms, slow_details, ) if slow_request_alert: failures.append( ("slow-request", *slow_request_alert) ) permission_slow_count, permission_pressure_details = detect_permission_read_pressure( backend_events, window_start_epoch, args.permission_slow_read_threshold_ms, ) business_alerts = detect_business_alerts(backend_events, window_start_epoch) if business_alerts: failures.append( ( "backend-alert", f"backend emitted structured alert(s) within {args.db_window_minutes}m", business_alerts, ) ) recent_499_matches = collect_recent_http_status(nginx_access_log(), args.scan_lines, "499") if len(recent_499_matches) >= args.nginx_499_threshold: failure, details = format_nginx_499_alert(recent_499_matches) failures.append(("nginx-499", failure, details)) backend_impact_details = summarize_backend_impact( backend_5xx_count, slow_count, args.backend_impact_count_threshold, ) recent_backend_errors = detect_recent_backend_errors(backend_log, args.scan_lines) if recent_backend_errors: failures.append(("backend-stderr", "recent backend stderr errors detected", recent_backend_errors)) tidb_observation: TiDBPressureObservation | None = None tidb_monitor_error = "" try: envv = live_envv if live_envv else db_monitoring_env() for key in ("TIDB_HOST", "TIDB_PORT", "TIDB_USER", "TIDB_DATABASE"): if not envv.get(key): raise DeployError(f"prod runtime env missing {key}") tidb_observation = collect_tidb_pressure(envv, args) failed_login_count = count_recent_failed_logins(envv, args.db_window_minutes) if failed_login_count > 0: details = sample_recent_failed_logins(envv, args.db_window_minutes) failure, details = format_failed_login_alert(args.db_window_minutes, failed_login_count, details) failures.append( ( "login-history", failure, details, ) ) flow_error_count = count_recent_flow_errors(envv, args.db_window_minutes) if flow_error_count > 0: details = sample_recent_flow_errors(envv, args.db_window_minutes) failures.append( ( "flow-audit", f"recent flow errors detected within {args.db_window_minutes}m: count={flow_error_count}", details, ) ) except DeployError as exc: tidb_monitor_error = str(exc) failures.extend( classify_tidb_monitor_outcome( tidb_observation, tidb_monitor_error, permission_slow_count=permission_slow_count, permission_count_threshold=args.permission_slow_read_count_threshold, permission_details=permission_pressure_details, backend_impact_details=backend_impact_details, ) ) if failures: next_state, to_notify = evaluate_failure_notifications( failures, state, args.failure_notify_threshold ) if args.notify_on_failure: for check_name, failure, details, fingerprint in to_notify: try: notify_failure(check_name, failure, details) except Exception as exc: log(f"online host guard notification failed: {exc}") next_state[check_name]["notified_fingerprint"] = None else: for check_name, _failure, _details, _fingerprint in to_notify: next_state[check_name]["notified_fingerprint"] = None save_state(next_state) raise DeployError("\n".join(f"{check}:{failure}" for check, failure, _ in failures)) if state: save_state({}) log("online host guard ok") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except DeployError as exc: log(f"failed: {exc}") raise SystemExit(1)