#!/usr/bin/env python3 """Direct Manus OP metrics and TiDB Dashboard query client.""" from __future__ import annotations import argparse import base64 import contextlib import io import json import math import os import re import secrets import socket import statistics import sys import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] TARGETS_FILE = ROOT / "data" / "targets.json" LEVELS = {"debug": 1, "info": 2, "warn": 3, "critical": 5, "error": 6} STATEMENT_DEFAULT_FIELDS = ( "digest_text", "sum_latency", "avg_latency", "max_latency", "min_latency", "exec_count", "plan_count", ) SLOW_QUERY_REQUIRED_FIELDS = ("digest", "connection_id", "timestamp") SLOW_QUERY_DIAGNOSTIC_FIELDS = ( "query", "instance", "db", "success", "query_time", "parse_time", "compile_time", "rewrite_time", "preproc_subqueries_time", "optimize_time", "wait_ts", "cop_time", "lock_keys_time", "write_sql_response_total", "exec_retry_time", "memory_max", "disk_max", "index_names", "stats", "warnings", "process_time", "wait_time", "backoff_time", "backoff_types", "local_latch_wait_time", "resolve_lock_time", "cop_proc_avg", "cop_proc_p90", "cop_proc_max", "cop_wait_avg", "cop_wait_p90", "cop_wait_max", "request_count", "process_keys", "total_keys", "cop_proc_addr", "cop_wait_addr", "rocksdb_delete_skipped_count", "rocksdb_key_skipped_count", "rocksdb_block_cache_hit_count", "rocksdb_block_read_count", "rocksdb_block_read_byte", "write_keys", "write_size", "txn_retry", "ru", "time_queued_by_rc", "resource_group", ) SLOW_QUERY_REQUIRED_HANDOFF_FIELDS = ( "digest", "query", "instance", "db", "connection_id", "timestamp", "query_time", "execution_interval_status", "time_semantics", "start_time", "finish_time", "overlaps_incident_window", "incident_overlap_seconds", ) SLOW_QUERY_CORE_FIELDS = ( "query", "instance", "db", "success", "query_time", "memory_max", "disk_max", "process_time", "wait_time", "backoff_time", "request_count", "process_keys", "total_keys", "ru", "time_queued_by_rc", "resource_group", ) SLOW_QUERY_HANDOFF_FIELDS = frozenset((*SLOW_QUERY_DIAGNOSTIC_FIELDS, *SLOW_QUERY_REQUIRED_HANDOFF_FIELDS)) SLOW_QUERY_MAX_HANDOFF_FIELDS = 12 SLOW_QUERY_OVERLAP_LOOKAHEAD_SECONDS = 15 * 60 SLOW_QUERY_MINIMUM_CANDIDATE_LIMIT = 100 OP_BATCH_MAX_QUERIES = 12 OP_BATCH_QUERY_ID_RE = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)*$") class QueryError(RuntimeError): def __init__(self, message: str, code: str = "manus_provider_query_failed"): super().__init__(message) self.code = code @dataclass class Response: status: int body: bytes def json(self) -> Any: try: return json.loads(self.body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise QueryError(f"endpoint returned non-JSON content (HTTP {self.status})") from exc def _json_bytes(value: Any) -> bytes: return json.dumps(value, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def _basic_header(username: str, password: str) -> str: raw = f"{username}:{password}".encode("utf-8") return "Basic " + base64.b64encode(raw).decode("ascii") class Transport: def request( self, method: str, url: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, body: bytes | None = None, timeout: int = 60, ) -> Response: raise NotImplementedError class DirectTransport(Transport): def request(self, method: str, url: str, *, params=None, headers=None, body=None, timeout=60) -> Response: if params: url += ("&" if "?" in url else "?") + urllib.parse.urlencode(params) req = urllib.request.Request(url, data=body, headers=headers or {}, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as response: return Response(response.status, response.read()) except urllib.error.HTTPError as exc: return Response(exc.code, exc.read()) except (TimeoutError, socket.timeout) as exc: raise QueryError("network request timed out") from exc except urllib.error.URLError as exc: if isinstance(exc.reason, (TimeoutError, socket.timeout)): raise QueryError("network request timed out") from exc raise QueryError(f"network request failed: {exc.reason}") from exc def load_target(name: str) -> dict[str, Any]: targets = json.loads(TARGETS_FILE.read_text(encoding="utf-8")) if name not in targets: raise QueryError(f"unknown target: {name}") return {**targets[name], "_name": name} def dashboard_credential_names(target: dict[str, Any]) -> tuple[str, str]: name = str(target.get("_name", "")) if not re.fullmatch(r"[a-z0-9][a-z0-9-]*", name): raise QueryError("target is missing a valid credential scope") prefix = "MANUS" if name == "vida" else "MANUS_" + name.upper().replace("-", "_") return f"{prefix}_DASHBOARD_USERNAME", f"{prefix}_DASHBOARD_PASSWORD" def parse_duration(value: str) -> int: match = re.fullmatch(r"([1-9][0-9]*)([smhd])", value.strip()) if not match: raise QueryError(f"invalid duration: {value}; use values such as 10m or 1h") number, unit = int(match.group(1)), match.group(2) return number * {"s": 1, "m": 60, "h": 3600, "d": 86400}[unit] def resolve_range(args: argparse.Namespace) -> tuple[int, int]: end = int(args.end) if args.end is not None else int(time.time()) start = int(args.start) if args.start is not None else end - parse_duration(args.last) if start >= end: raise QueryError("start must be earlier than end") return start, end def require_env(*names: str) -> list[str]: values = [os.environ.get(name, "") for name in names] missing = [name for name, value in zip(names, values) if not value] if missing: raise QueryError("missing credential environment variables: " + ", ".join(missing)) return values def expect_ok(response: Response, operation: str) -> Any: if response.status < 200 or response.status >= 300: lower_body = response.body.lower() if response.status == 504 or b"timeout" in lower_body or b"timed out" in lower_body or b"aborterror" in lower_body: raise QueryError(f"{operation} timed out (HTTP {response.status})") raise QueryError(f"{operation} failed with HTTP {response.status}") return response.json() def _read_tlv(data: bytes, offset: int = 0) -> tuple[int, bytes, int]: if offset + 2 > len(data): raise QueryError("invalid RSA public key") tag = data[offset] length_byte = data[offset + 1] offset += 2 if length_byte & 0x80: width = length_byte & 0x7F if width == 0 or offset + width > len(data): raise QueryError("invalid RSA public key length") length = int.from_bytes(data[offset:offset + width], "big") offset += width else: length = length_byte end = offset + length if end > len(data): raise QueryError("truncated RSA public key") return tag, data[offset:end], end def parse_rsa_public_key(public_key: str) -> tuple[int, int]: clean = re.sub(r"-----[^-]+-----|\s+", "", public_key) try: der = base64.b64decode(clean, validate=True) except ValueError as exc: raise QueryError("invalid Dashboard RSA public key") from exc tag, spki, end = _read_tlv(der) if tag != 0x30 or end != len(der): raise QueryError("invalid Dashboard RSA public key container") _, _, offset = _read_tlv(spki, 0) # algorithm identifier tag, bit_string, offset = _read_tlv(spki, offset) if tag != 0x03 or offset != len(spki) or not bit_string or bit_string[0] != 0: raise QueryError("invalid Dashboard RSA public key payload") tag, rsa_sequence, rsa_end = _read_tlv(bit_string[1:]) if tag != 0x30 or rsa_end != len(bit_string) - 1: raise QueryError("invalid Dashboard RSA public key sequence") tag_n, n_bytes, offset = _read_tlv(rsa_sequence, 0) tag_e, e_bytes, offset = _read_tlv(rsa_sequence, offset) if tag_n != 0x02 or tag_e != 0x02 or offset != len(rsa_sequence): raise QueryError("invalid Dashboard RSA public key integers") return int.from_bytes(n_bytes, "big"), int.from_bytes(e_bytes, "big") def rsa_encrypt_password(password: str, public_key: str) -> str: modulus, exponent = parse_rsa_public_key(public_key) message = password.encode("utf-8") width = (modulus.bit_length() + 7) // 8 if len(message) > width - 11: raise QueryError("Dashboard password is too long for its RSA public key") pad = bytearray() while len(pad) < width - len(message) - 3: pad.extend(byte for byte in secrets.token_bytes(width) if byte) encoded = b"\x00\x02" + bytes(pad[:width - len(message) - 3]) + b"\x00" + message encrypted = pow(int.from_bytes(encoded, "big"), exponent, modulus).to_bytes(width, "big") return base64.b64encode(encrypted).decode("ascii") class GrafanaClient: def __init__(self, transport: Transport, target: dict[str, Any]): username, password = require_env("MANUS_GRAFANA_USERNAME", "MANUS_GRAFANA_PASSWORD") self.transport = transport self.base = str(target["grafana_url"]).rstrip("/") self.datasource_name = str(target["grafana_datasource_name"]) self.headers = {"Accept": "application/json", "Authorization": _basic_header(username, password)} def get(self, path: str, params: dict[str, Any] | None = None, timeout: int = 60) -> Any: response = self.transport.request("GET", self.base + path, params=params, headers=self.headers, timeout=timeout) return expect_ok(response, f"Grafana {path}") def datasource(self) -> dict[str, Any]: encoded = urllib.parse.quote(self.datasource_name, safe="") payload = self.get(f"/api/datasources/name/{encoded}") if payload.get("name") != self.datasource_name or str(payload.get("type", "")).lower() != "prometheus": raise QueryError(f"Grafana datasource contract mismatch for {self.datasource_name}") return payload def query_range(self, query: str, start: int, end: int, step: str) -> tuple[dict[str, Any], Any]: datasource = self.datasource() payload = self.get( f"/api/datasources/proxy/{datasource['id']}/api/v1/query_range", {"query": query, "start": start, "end": end, "step": step}, timeout=120, ) if payload.get("status") != "success": raise QueryError("Prometheus query_range did not return success") return datasource, payload class PrometheusClient: def __init__(self, transport: Transport, target: dict[str, Any]): self.transport = transport self.base = str(target["prometheus_url"]).rstrip("/") self.target_name = str(target["_name"]) def get(self, path: str, params: dict[str, Any] | None = None, timeout: int = 60) -> Any: response = self.transport.request( "GET", self.base + path, params=params, headers={"Accept": "application/json"}, timeout=timeout ) payload = expect_ok(response, f"Prometheus {path}") if not isinstance(payload, dict) or payload.get("status") != "success": raise QueryError(f"Prometheus {path} did not return success") return payload def build_info(self) -> dict[str, Any]: payload = self.get("/api/v1/status/buildinfo") data = payload.get("data", {}) return data if isinstance(data, dict) else {} def query_range(self, query: str, start: int, end: int, step: str) -> tuple[dict[str, Any], Any]: payload = self.get( "/api/v1/query_range", {"query": query, "start": start, "end": end, "step": step}, timeout=120, ) return {"name": self.target_name, "type": "prometheus", "access": "direct"}, payload class DashboardClient: def __init__(self, transport: Transport, target: dict[str, Any]): self.transport = transport self.target = target self.base = str(target["dashboard_url"]).rstrip("/") self.token: str | None = None def _headers(self, authenticated: bool = True) -> dict[str, str]: headers = {"Accept": "application/json"} if authenticated: if not self.token: self.login() headers["Authorization"] = "Bearer " + str(self.token) return headers def login(self) -> None: username_env, password_env = dashboard_credential_names(self.target) username, password = require_env(username_env, password_env) info = expect_ok( self.transport.request("GET", self.base + "/api/user/login_info", headers={"Accept": "application/json"}), "Dashboard login_info", ) encrypted = rsa_encrypt_password(password, str(info.get("sql_auth_public_key", ""))) response = self.transport.request( "POST", self.base + "/api/user/login", headers={"Accept": "application/json", "Content-Type": "application/json"}, body=_json_bytes({"type": 0, "username": username, "password": encrypted}), ) payload = expect_ok(response, "Dashboard login") token = payload.get("token") if not token: raise QueryError("Dashboard login response did not contain a token") self.token = str(token) def get(self, path: str, params: dict[str, Any] | None = None, timeout: int = 60, authenticated: bool = True) -> Any: response = self.transport.request( "GET", self.base + path, params=params, headers=self._headers(authenticated), timeout=timeout ) return expect_ok(response, f"Dashboard {path}") def get_optional(self, path: str, params: dict[str, Any] | None = None, timeout: int = 60) -> Any | None: response = self.transport.request( "GET", self.base + path, params=params, headers=self._headers(True), timeout=timeout ) if response.status in (404, 405): return None return expect_ok(response, f"Dashboard {path}") def put(self, path: str, payload: Any, timeout: int = 60) -> Any: headers = self._headers(True) headers["Content-Type"] = "application/json" response = self.transport.request( "PUT", self.base + path, headers=headers, body=_json_bytes(payload), timeout=timeout ) return expect_ok(response, f"Dashboard {path}") def rows_from_payload(payload: Any) -> list[Any]: if isinstance(payload, list): return payload if isinstance(payload, dict): for key in ("data", "items", "result", "results"): if isinstance(payload.get(key), list): return payload[key] return [] def dashboard_rows(kind: str, payload: Any) -> list[Any]: if kind == "topsql" and isinstance(payload, dict): summary = payload.get("summary") if isinstance(summary, dict) and isinstance(summary.get("data"), list): return summary["data"] return rows_from_payload(summary) return rows_from_payload(payload) def compact_rows(rows: list[Any], limit: int, selected_fields: list[str] | None = None) -> list[Any]: compact: list[Any] = [] selected = set(selected_fields or []) for row in rows[:limit]: if not isinstance(row, dict): compact.append(row) continue compact.append({ key: value for key, value in row.items() if key not in {"chart_data", "plans"} and (not selected or key in selected) }) return compact def slow_query_execution_rows( rows: list[Any], incident_start: int, incident_end: int, result_limit: int ) -> tuple[list[dict[str, Any]], dict[str, int]]: classified: list[dict[str, Any]] = [] overlapping = 0 unclassified = 0 excluded = 0 for candidate in rows: if not isinstance(candidate, dict): unclassified += 1 continue row = dict(candidate) finish = slow_query_epoch_seconds(row.get("timestamp", row.get("time"))) query_time = finite_number(row.get("query_time")) if finish is None or query_time is None or query_time < 0: row.update({ "execution_interval_status": "unavailable", "time_semantics": "slow_log_time_is_finish_print_time", "start_time": None, "finish_time": finish, "overlaps_incident_window": None, "incident_overlap_seconds": None, }) classified.append(row) unclassified += 1 continue start = finish - query_time overlaps = finish >= incident_start and start <= incident_end overlap_seconds = max(0.0, min(finish, incident_end) - max(start, incident_start)) row.update({ "execution_interval_status": "derived", "time_semantics": "slow_log_time_is_finish_print_time", "start_time": round(start, 6), "finish_time": round(finish, 6), "overlaps_incident_window": overlaps, "incident_overlap_seconds": round(overlap_seconds, 6), }) if overlaps: classified.append(row) overlapping += 1 else: excluded += 1 return classified[:max(1, min(200, result_limit))], { "raw_record_count": len(rows), "overlapping_record_count": overlapping, "unclassified_record_count": unclassified, "excluded_non_overlapping_record_count": excluded, } def slow_query_epoch_seconds(value: Any) -> float | None: numeric = finite_number(value) if numeric is not None: if numeric >= 1e15: return numeric / 1_000_000 if numeric >= 1e12: return numeric / 1_000 return numeric if not isinstance(value, str) or not value.strip(): return None text = value.strip().replace(" ", "T") if text.endswith("Z"): text = text[:-1] + "+00:00" try: parsed = datetime.fromisoformat(text) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.timestamp() def finite_number(value: Any) -> float | None: if isinstance(value, bool) or value is None: return None try: numeric = float(value) except (TypeError, ValueError): return None return numeric if math.isfinite(numeric) else None def field_list(value: str | None, defaults: tuple[str, ...]) -> list[str]: if value is None or not value.strip(): return list(defaults) fields = list(dict.fromkeys(field.strip() for field in value.split(",") if field.strip())) if "*" in fields and fields != ["*"]: raise QueryError("fields '*' cannot be combined with named fields") return fields def slow_query_field_selection( requested_fields: list[str], available_payload: Any | None ) -> tuple[list[str], dict[str, Any]]: available_fields = [ str(field) for field in available_payload if isinstance(field, str) and field ] if isinstance(available_payload, list) else [] if requested_fields == ["*"]: selected_fields = ["*"] unsupported_fields: list[str] = [] elif available_fields: available = set(available_fields) selected_fields = [field for field in requested_fields if field in available] unsupported_fields = [field for field in requested_fields if field not in available] else: selected_fields = requested_fields unsupported_fields = [] return selected_fields, { "mode": "requested_projection", "profile": "explicit" if requested_fields != list(SLOW_QUERY_DIAGNOSTIC_FIELDS) else "diagnostic", "availability_status": "available" if available_fields else "unavailable", "requested_fields": selected_fields, "unsupported_fields": unsupported_fields, } def metric_summary(payload: dict[str, Any]) -> dict[str, Any]: series = payload.get("data", {}).get("result", []) values: list[float] = [] for item in series: for point in item.get("values", []): try: value = float(point[1]) if math.isfinite(value): values.append(value) except (ValueError, TypeError, IndexError): continue return { "series_count": len(series), "point_count": len(values), "min": min(values) if values else None, "max": max(values) if values else None, "avg": statistics.fmean(values) if values else None, "current": values[-1] if values else None, } def persist(output_dir: str | None, name: str, raw: Any, report: dict[str, Any]) -> list[str]: if not output_dir: return [] directory = Path(output_dir) directory.mkdir(parents=True, exist_ok=True) raw_path = directory / f"{name}.json" report_path = directory / "workspace_evidence_report.json" raw_path.write_text(json.dumps(raw, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return [str(raw_path), str(report_path)] def make_transport(args: argparse.Namespace) -> Transport: return DirectTransport() def cmd_health(args: argparse.Namespace) -> dict[str, Any]: target = load_target(args.target) transport = make_transport(args) dashboard = DashboardClient(transport, target) info = dashboard.get("/api/info/info", authenticated=False) dashboard.login() whoami = dashboard.get("/api/info/whoami") version = info.get("version", {}) if not isinstance(version, dict): version = {"internal_version": version} result = { "status": "ok", "target": args.target, "dashboard": { "version": version.get("internal_version"), "pd_version": version.get("pd_version"), "ngm_state": info.get("ngm_state"), }, "dashboard_user": {"display_name": whoami.get("display_name")}, } if target.get("prometheus_url"): prometheus = PrometheusClient(transport, target) result["prometheus"] = { "status": "ok", "access": "direct", "version": prometheus.build_info().get("version") } else: grafana = GrafanaClient(transport, target) health = grafana.get("/api/health") datasource = grafana.datasource() result["grafana"] = {"database": health.get("database"), "version": health.get("version")} result["datasource"] = {key: datasource.get(key) for key in ("id", "uid", "name", "type")} return result def cmd_metrics(args: argparse.Namespace) -> dict[str, Any]: target = load_target(args.target) start, end = resolve_range(args) client = PrometheusClient(make_transport(args), target) if target.get("prometheus_url") else GrafanaClient(make_transport(args), target) query_start, query_end = start, end recovery = {"attempts": 1, "window_narrowed": False} try: datasource, payload = client.query_range(args.query, query_start, query_end, args.step) except QueryError as exc: if not is_timeout_error(exc): raise retry_range = narrower_range(query_start, query_end) if retry_range is None: raise query_start, query_end = retry_range print( "Metrics query timed out; retrying once with the same target and a half-size centered window.", file=sys.stderr, ) recovery = { "attempts": 2, "window_narrowed": True, "initial_time_window": {"start": start, "end": end}, } datasource, payload = client.query_range(args.query, query_start, query_end, args.step) summary = metric_summary(payload) report = { "status": "ok", "kind": "metrics", "target": args.target, "name": args.name, "query": args.query, "unit": args.unit, "time_window": {"start": query_start, "end": query_end}, "recovery": recovery, "datasource": {key: datasource.get(key) for key in ("id", "uid", "name", "type")}, "summary": summary, } report["persisted_evidence"] = persist(args.output_dir, "metrics", payload, report) return report def is_timeout_error(error: QueryError) -> bool: message = str(error).lower() return "timed out" in message or "timeout" in message def narrower_range(start: int, end: int) -> tuple[int, int] | None: width = end - start if width < 4: return None quarter = width // 4 narrowed = (start + quarter, end - quarter) return narrowed if narrowed[0] < narrowed[1] else None def query_dashboard(client: DashboardClient, args: argparse.Namespace, start: int, end: int) -> tuple[Any, dict[str, Any]]: if args.kind == "topsql": instance = args.instance instance_type = args.instance_type if not instance: raise QueryError( "TopSQL requires --instance selected from time-aligned per-instance metrics; " "the provider will not use the first discovered instance" ) request = { "start": start, "end": end, "instance": instance, "instance_type": instance_type, "group_by": args.group_by, "top": args.limit, "window": args.window, } for attempt in range(2): try: summary = client.get("/api/topsql/summary", request, timeout=120) break except QueryError as exc: if attempt or "timed out" not in str(exc).lower(): raise print("TopSQL summary timed out; waiting 60s before one retry.", file=sys.stderr) time.sleep(60) return {"summary": summary}, request if args.kind == "slow-query": query_end = end + SLOW_QUERY_OVERLAP_LOOKAHEAD_SECONDS requested_limit = args.limit candidate_limit = min(200, max(SLOW_QUERY_MINIMUM_CANDIDATE_LIMIT, requested_limit)) requested_fields = field_list(args.fields, SLOW_QUERY_DIAGNOSTIC_FIELDS) available_payload = client.get_optional("/api/slow_query/available_fields", timeout=30) selected_fields, field_selection = slow_query_field_selection(requested_fields, available_payload) api_request = { "begin_time": start, "end_time": query_end, "fields": ",".join(selected_fields), "limit": candidate_limit, "text": args.text, } payload = client.get("/api/slow_query/list", api_request, timeout=90) return payload, { **api_request, "incident_time_window": {"start": start, "end": end}, "query_time_window": {"start": start, "end": query_end}, "result_limit": requested_limit, "field_selection": field_selection, } request = { "begin_time": start, "end_time": end, "fields": ",".join(field_list(args.fields, STATEMENT_DEFAULT_FIELDS)), "text": args.text, } return client.get("/api/statements/list", request, timeout=90), request def cmd_dashboard(args: argparse.Namespace) -> dict[str, Any]: start, end = resolve_range(args) client = DashboardClient(make_transport(args), load_target(args.target)) payload, request = query_dashboard(client, args, start, end) rows = dashboard_rows(args.kind, payload) selected_fields = None slow_query_counts: dict[str, int] = {} if args.kind == "slow-query": selected_fields = list(SLOW_QUERY_REQUIRED_FIELDS) selected_fields.extend(request.get("field_selection", {}).get("requested_fields", [])) selected_fields.extend([ "execution_interval_status", "time_semantics", "start_time", "finish_time", "overlaps_incident_window", "incident_overlap_seconds", ]) rows, slow_query_counts = slow_query_execution_rows( rows, start, end, int(request.get("result_limit", args.limit)) ) report = { "status": "ok", "kind": args.kind, "target": args.target, "time_window": {"start": start, "end": end}, "request": request, "record_count": len(rows), **slow_query_counts, "preview_rows": compact_rows(rows, args.preview_limit, selected_fields), } report["persisted_evidence"] = persist(args.output_dir, args.kind, payload, report) return report def topology_targets(client: DashboardClient, component: str, limit: int) -> list[dict[str, Any]]: if component == "tidb": rows = rows_from_payload(client.get("/api/topology/tidb")) elif component in {"tikv", "tiflash"}: payload = client.get("/api/topology/store") rows = payload.get(component, []) if isinstance(payload, dict) else [] else: rows = rows_from_payload(client.get("/api/topology/pd")) targets = [] for row in rows: if row.get("status") not in (None, 1): continue port = row.get("status_port") if component == "tidb" else row.get("port") if not row.get("ip") or port is None: continue display_port = row.get("port") or port targets.append({ "kind": component, "display_name": f"{row['ip']}:{display_port}", "ip": row["ip"], "port": int(port), }) if not targets: raise QueryError(f"no live {component} topology target was discovered") return targets[:limit] def cmd_logs(args: argparse.Namespace) -> dict[str, Any]: start, end = resolve_range(args) client = DashboardClient(make_transport(args), load_target(args.target)) targets = topology_targets(client, args.component, args.target_limit) payload = { "request": { "start_time": start * 1000, "end_time": end * 1000, "min_level": LEVELS[args.level], "patterns": args.keyword, }, "targets": targets, } created = client.put("/api/logs/taskgroup", payload, timeout=90) task_group = created.get("task_group", {}) task_group_id = task_group.get("id") if task_group_id is None: raise QueryError("log search did not return a task group id") deadline = time.monotonic() + args.poll_timeout current = created while current.get("task_group", {}).get("state") != 2 and time.monotonic() < deadline: time.sleep(args.poll_interval) current = client.get(f"/api/logs/taskgroups/{task_group_id}", timeout=30) state = current.get("task_group", {}).get("state") tasks = current.get("tasks", []) if state != 2 and not all(task.get("state") in (2, 3) for task in tasks): raise QueryError(f"log search task group {task_group_id} did not finish before timeout") preview = client.get(f"/api/logs/taskgroups/{task_group_id}/preview", timeout=30) rows = rows_from_payload(preview) report = { "status": "ok", "kind": "logs", "target": args.target, "task_group_id": task_group_id, "task_group_state": state, "task_states": [task.get("state") for task in tasks], "time_window": {"start": start, "end": end}, "component": args.component, "level": args.level, "patterns": args.keyword, "target_count": len(targets), "record_count": len(rows), "preview_rows": compact_rows(rows, args.limit), } report["persisted_evidence"] = persist(args.output_dir, "logs", preview, report) return report def add_range_args(parser: argparse.ArgumentParser, default: str) -> None: parser.add_argument("--start", type=int, help="epoch seconds") parser.add_argument("--end", type=int, help="epoch seconds") parser.add_argument("--last", default=default) def _load_bounded_json_object(path: Path, maximum_bytes: int = 262_144) -> dict[str, Any]: if not path.is_file() or path.stat().st_size > maximum_bytes: raise QueryError(f"invalid or oversized JSON input: {path.name}") try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise QueryError(f"invalid JSON input: {path.name}") from exc if not isinstance(value, dict): raise QueryError(f"JSON input must be an object: {path.name}") return value def _atomic_write_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(value, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") temporary.replace(path) def _historical_bounds(time_window: Any) -> tuple[int, int] | None: if not isinstance(time_window, str): return None match = re.fullmatch(r"(?:historical|live):absolute;start=(\d+);end=(\d+);.*", time_window) if not match: return None start, end = int(match.group(1)), int(match.group(2)) if start >= end: raise QueryError("historical OP batch window is invalid") return start, end def _comparison_bounds(context: dict[str, Any]) -> tuple[tuple[int, int], ...]: value = context.get("comparisonWindows") if value is None: return () incident = _historical_bounds(context.get("timeWindow")) if incident is None or not isinstance(value, list) or len(value) != 7: raise QueryError("historical comparison windows are invalid", "manus_op_batch_comparison_windows_invalid") windows: list[tuple[int, int]] = [] for index, item in enumerate(value, start=1): if not isinstance(item, dict): raise QueryError("historical comparison windows are invalid", "manus_op_batch_comparison_windows_invalid") start, end = item.get("start"), item.get("end") if (item.get("windowId") != f"day-{index}" or not isinstance(start, int) or isinstance(start, bool) or not isinstance(end, int) or isinstance(end, bool) or start != incident[0] - index * 86400 or end != incident[1] - index * 86400): raise QueryError("historical comparison windows are invalid", "manus_op_batch_comparison_windows_invalid") windows.append((start, end)) return tuple(windows) def _bounded_query_range( params: dict[str, Any], bounds: tuple[int, int] | None, comparison_bounds: tuple[tuple[int, int], ...] = (), ) -> tuple[int, int]: start = params.get("start", bounds[0] if bounds else None) end = params.get("end", bounds[1] if bounds else None) if not isinstance(start, int) or isinstance(start, bool) or not isinstance(end, int) or isinstance(end, bool): raise QueryError("every OP batch query requires integer start and end") if start >= end: raise QueryError("OP batch query start must be before end") if comparison_bounds: if (start, end) not in comparison_bounds: raise QueryError( "historical comparison queries must use one exact runner-owned daily window", "manus_op_batch_comparison_window_invalid", ) return start, end if bounds and (start < bounds[0] or end > bounds[1]): raise QueryError("OP batch query must stay inside the runner-owned historical window") return start, end def _assert_comparison_plan( context: dict[str, Any], queries: list[dict[str, Any]], bounds: tuple[int, int] | None, comparison_bounds: tuple[tuple[int, int], ...], ) -> None: if not comparison_bounds: return if context.get("scenarioId") != "workload-hotspot" or len(queries) != 7: raise QueryError( "historical comparison requires seven workload-hotspot metric queries", "manus_op_batch_comparison_plan_invalid", ) used: set[tuple[int, int]] = set() for query in queries: if query.get("operation") != "metrics" or not isinstance(query.get("params"), dict): raise QueryError( "historical comparison supports only separate metric range queries", "manus_op_batch_comparison_plan_invalid", ) used.add(_bounded_query_range(query["params"], bounds, comparison_bounds)) if len(used) != len(comparison_bounds): raise QueryError( "historical comparison must cover each runner-owned daily window exactly once", "manus_op_batch_comparison_plan_invalid", ) def _string_param(params: dict[str, Any], key: str, *, required: bool = False) -> str | None: value = params.get(key) if value is None and not required: return None if not isinstance(value, str) or not value.strip(): raise QueryError(f"OP batch parameter {key} must be a non-empty string") return value.strip() def _integer_param(params: dict[str, Any], key: str) -> int | None: value = params.get(key) if value is None: return None if not isinstance(value, int) or isinstance(value, bool) or value < 1: raise QueryError(f"OP batch parameter {key} must be a positive integer") return value def _normalize_manus_batch_target( value: Any, error_code: str = "manus_op_batch_target_invalid", ) -> str: if isinstance(value, str) and value in {"vida", "monica"}: return value if isinstance(value, dict): for key in ("target", "name", "cluster"): candidate = value.get(key) if candidate in {"vida", "monica"}: return candidate raise QueryError("Manus OP batch requires exact target vida or monica", error_code) def _normalize_manus_batch_query(query: dict[str, Any]) -> dict[str, Any]: params = query.get("params") normalized_params = dict(params) if isinstance(params, dict) else {} aliases = { "preview_limit": "previewLimit", "instance_type": "instanceType", "group_by": "groupBy", "target_limit": "targetLimit", "poll_timeout": "pollTimeout", "poll_interval": "pollInterval", } metadata = { "queryId", "description", "operation", "params", "subdeadlineMs", "subDeadlineMs", "deadlineMs", "dependsOn", "handoffFields", "emptyResultExpectation", } for key, value in query.items(): if key in metadata: continue normalized_params.setdefault(aliases.get(key, key), value) empty_expectation = query.get("emptyResultExpectation") if query.get("operation") == "metrics": if empty_expectation is None: empty_expectation = "unknown" query["emptyResultExpectation"] = empty_expectation if empty_expectation not in {"expected_non_empty", "sparse_allowed", "unknown"}: raise QueryError( "Metric emptyResultExpectation must be expected_non_empty, sparse_allowed, or unknown", "manus_op_batch_parameter_invalid", ) elif empty_expectation is not None: raise QueryError( "emptyResultExpectation is valid only for metrics operations", "manus_op_batch_parameter_invalid", ) if query.get("operation") == "dashboard" and normalized_params.get("kind") == "slow-query": handoff_fields = _slow_query_handoff_fields(query.get("handoffFields")) query["handoffFields"] = handoff_fields selected = field_list(normalized_params.get("fields"), ()) selected_handoff_fields = [ field for field in handoff_fields if field in SLOW_QUERY_DIAGNOSTIC_FIELDS ] normalized_params["fields"] = ",".join(dict.fromkeys(( *SLOW_QUERY_REQUIRED_FIELDS, *SLOW_QUERY_CORE_FIELDS, *selected_handoff_fields, *selected, ))) return {**query, "params": normalized_params} def _slow_query_handoff_fields(value: Any) -> list[str]: if value is None: return [] if ( not isinstance(value, list) or len(value) > SLOW_QUERY_MAX_HANDOFF_FIELDS or any(not isinstance(field, str) or not field.strip() for field in value) ): raise QueryError( f"Slow Query handoffFields must contain at most {SLOW_QUERY_MAX_HANDOFF_FIELDS} field names", "manus_op_batch_parameter_invalid", ) fields = list(dict.fromkeys(field.strip() for field in value)) if any(field not in SLOW_QUERY_HANDOFF_FIELDS for field in fields): raise QueryError( "Slow Query handoffFields contains an unsupported field", "manus_op_batch_parameter_invalid", ) return fields def _manus_batch_argv( target: str, query: dict[str, Any], bounds: tuple[int, int] | None, output_dir: Path, comparison_bounds: tuple[tuple[int, int], ...] = (), ) -> list[str]: operation = query.get("operation") params = query.get("params") if operation not in {"metrics", "dashboard", "logs"} or not isinstance(params, dict): raise QueryError("unsupported Manus OP batch operation") start, end = _bounded_query_range(params, bounds, comparison_bounds) common = ["--target", target, "--start", str(start), "--end", str(end), "--output-dir", str(output_dir)] if operation == "metrics": allowed = {"name", "query", "step", "unit", "start", "end"} argv = ["metrics", *common, "--name", _string_param(params, "name", required=True), "--query", _string_param(params, "query", required=True)] step = params.get("step") if step is not None: if not isinstance(step, (str, int, float)) or isinstance(step, bool): raise QueryError("OP batch parameter step must be a duration or number") argv.extend(["--step", str(step)]) unit = _string_param(params, "unit") if unit is not None: argv.extend(["--unit", unit]) elif operation == "dashboard": allowed = { "kind", "limit", "previewLimit", "text", "fields", "instance", "instanceType", "groupBy", "window", "start", "end", } argv = ["dashboard", *common, "--kind", _string_param(params, "kind", required=True)] for key, flag in ( ("text", "--text"), ("fields", "--fields"), ("instance", "--instance"), ("instanceType", "--instance-type"), ("groupBy", "--group-by"), ("window", "--window"), ): value = _string_param(params, key) if value is not None: argv.extend([flag, value]) for key, flag in (("limit", "--limit"), ("previewLimit", "--preview-limit")): value = _integer_param(params, key) if value is not None: argv.extend([flag, str(value)]) else: allowed = { "component", "level", "keywords", "limit", "targetLimit", "pollTimeout", "pollInterval", "start", "end", } argv = ["logs", *common] for key, flag in (("component", "--component"), ("level", "--level")): value = _string_param(params, key) if value is not None: argv.extend([flag, value]) keywords = params.get("keywords", []) if not isinstance(keywords, list) or len(keywords) > 3 or any( not isinstance(value, str) or not value.strip() for value in keywords ): raise QueryError("OP batch keywords must be a list of at most three strings") for value in keywords: argv.extend(["--keyword", value.strip()]) for key, flag in ( ("limit", "--limit"), ("targetLimit", "--target-limit"), ("pollTimeout", "--poll-timeout"), ): value = _integer_param(params, key) if value is not None: argv.extend([flag, str(value)]) poll_interval = params.get("pollInterval") if poll_interval is not None: if not isinstance(poll_interval, (int, float)) or isinstance(poll_interval, bool) or poll_interval <= 0: raise QueryError("OP batch pollInterval must be positive") argv.extend(["--poll-interval", str(poll_interval)]) # The plan is agent-authored. Only map this operation's explicit allowlist # into CLI arguments; unrelated descriptive or future metadata is inert. return [str(value) for value in argv] def _run_manus_subcommand(argv: list[str]) -> tuple[int, dict[str, Any]]: stdout, stderr = io.StringIO(), io.StringIO() with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): code = main(argv) candidate = stdout.getvalue().strip() if code == 0 else stderr.getvalue().strip().splitlines()[-1] try: payload = json.loads(candidate) except (json.JSONDecodeError, IndexError) as exc: raise QueryError("Manus provider returned invalid compact JSON") from exc if not isinstance(payload, dict): raise QueryError("Manus provider returned a non-object compact result") return code, payload def _manus_result_empty(operation: str, payload: dict[str, Any]) -> bool: if operation == "metrics": return int(payload.get("summary", {}).get("point_count", 0)) == 0 return int(payload.get("record_count", 0)) == 0 def cmd_batch(args: argparse.Namespace) -> dict[str, Any]: context = _load_bounded_json_object(args.context) plan = _load_bounded_json_object(args.plan) if context.get("schemaVersion") != "auto-oncall-op-query-context-v1" or context.get("provider") != "manus": raise QueryError("Manus OP batch context is invalid", "manus_op_batch_context_invalid") if plan.get("schemaVersion") != "auto-oncall-op-query-batch-plan-v1" or plan.get("provider") != "manus": raise QueryError("Manus OP batch plan is invalid", "manus_op_batch_plan_invalid") if plan.get("scenarioId") != context.get("scenarioId") or plan.get("timeWindow") != context.get("timeWindow"): raise QueryError( "Manus OP batch plan does not match runner-owned context", "manus_op_batch_plan_mismatch", ) context_target = _normalize_manus_batch_target( context.get("target"), "manus_op_batch_context_target_invalid" ) target = _normalize_manus_batch_target(plan.get("target")) if target != context_target: raise QueryError( "Manus OP batch target does not match runner-owned context", "manus_op_batch_target_mismatch", ) queries = plan.get("queries") if not isinstance(queries, list) or not 1 <= len(queries) <= OP_BATCH_MAX_QUERIES: raise QueryError( f"Manus OP batch requires one to {OP_BATCH_MAX_QUERIES} queries", "manus_op_batch_query_count_invalid", ) if any(not isinstance(query, dict) for query in queries): raise QueryError("Manus OP batch queries must be objects", "manus_op_batch_query_invalid") queries = [_normalize_manus_batch_query(query) for query in queries] seen: set[str] = set() bounds = _historical_bounds(context.get("timeWindow")) comparison_bounds = _comparison_bounds(context) _assert_comparison_plan(context, queries, bounds, comparison_bounds) started_wall, started_monotonic = time.time(), time.monotonic() completed: list[dict[str, Any]] = [] result = { "schemaVersion": "auto-oncall-op-query-batch-v1", "provider": "manus", "scenarioId": context.get("scenarioId"), "timeWindow": context.get("timeWindow"), "completion": "partial", "startedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started_wall)), "durationMs": 0, "queries": completed, } for query in queries: query_id = query.get("queryId") if isinstance(query, dict) else None if (not isinstance(query_id, str) or len(query_id) > 64 or OP_BATCH_QUERY_ID_RE.fullmatch(query_id) is None or query_id in seen): raise QueryError( "Manus OP batch query id is invalid or duplicated", "manus_op_batch_query_invalid", ) seen.add(query_id) operation = str(query.get("operation") or "") query_started = time.monotonic() output_dir = args.result.parent / query_id / "provider-raw" argv = _manus_batch_argv(target, query, bounds, output_dir, comparison_bounds) try: code, payload = _run_manus_subcommand(argv) status = "failed" if code else ("empty" if _manus_result_empty(operation, payload) else "complete") error_message = str(payload.get("error"))[:500] if code else None except Exception as exc: code, payload, status = 2, {}, "failed" error_message = str(exc)[:500] completed.append({ "queryId": query_id, "operation": operation, "status": status, "statement": ( f"Manus {operation} returned usable provider evidence for {query_id}." if status == "complete" else f"Manus {operation} returned no usable provider evidence for {query_id}." ), "durationMs": round((time.monotonic() - query_started) * 1000), "summary": payload, "handoffFields": query.get("handoffFields", []), **({"emptyResultExpectation": query.get("emptyResultExpectation")} if query.get("emptyResultExpectation") is not None else {}), "errorCode": "manus_provider_query_failed" if code else None, "errorMessage": error_message, }) result["durationMs"] = round((time.monotonic() - started_monotonic) * 1000) _atomic_write_json(args.result, result) result["completion"] = "complete" if all(item["status"] != "failed" for item in completed) else "partial" result["durationMs"] = round((time.monotonic() - started_monotonic) * 1000) _atomic_write_json(args.result, result) return result def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="command", required=True) health = sub.add_parser("health") health.add_argument("--target", required=True, choices=("vida", "monica")) metrics = sub.add_parser("metrics") metrics.add_argument("--target", required=True, choices=("vida", "monica")) metrics.add_argument("--name", required=True) metrics.add_argument("--query", required=True) metrics.add_argument("--step", default="60") metrics.add_argument("--unit", default="") metrics.add_argument("--output-dir") add_range_args(metrics, "20m") dashboard = sub.add_parser("dashboard") dashboard.add_argument("--target", required=True, choices=("vida", "monica")) dashboard.add_argument("--kind", required=True, choices=("topsql", "slow-query", "statements")) dashboard.add_argument("--limit", type=int, default=10, help="TopSQL top count or slow-query API row limit") dashboard.add_argument("--preview-limit", type=int, default=10) dashboard.add_argument("--text", default="") dashboard.add_argument( "--fields", help="comma-separated statement or Slow Query fields; defaults to a kind-specific diagnostic projection", ) dashboard.add_argument("--instance") dashboard.add_argument("--instance-type", default="tidb") dashboard.add_argument("--group-by", choices=("query", "plan"), default="query") dashboard.add_argument("--window", default="10s") dashboard.add_argument("--output-dir") add_range_args(dashboard, "15m") logs = sub.add_parser("logs") logs.add_argument("--target", required=True, choices=("vida", "monica")) logs.add_argument("--component", choices=("tidb", "tikv", "tiflash", "pd"), default="tidb") logs.add_argument("--level", choices=tuple(LEVELS), default="error") logs.add_argument("--keyword", action="append", default=[]) logs.add_argument("--limit", type=int, default=20) logs.add_argument("--target-limit", type=int, default=1) logs.add_argument("--poll-timeout", type=int, default=120) logs.add_argument("--poll-interval", type=float, default=2.0) logs.add_argument("--output-dir") add_range_args(logs, "15m") batch = sub.add_parser("batch") batch.add_argument("context", type=Path) batch.add_argument("plan", type=Path) batch.add_argument("result", type=Path) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: if args.command == "health": result = cmd_health(args) elif args.command == "metrics": result = cmd_metrics(args) elif args.command == "dashboard": result = cmd_dashboard(args) elif args.command == "logs": result = cmd_logs(args) else: result = cmd_batch(args) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 except Exception as exc: message = str(exc) if isinstance(exc, QueryError) else f"unexpected {type(exc).__name__}" error_code = exc.code if isinstance(exc, QueryError) else "manus_provider_query_failed" print(json.dumps({ "status": "error", "errorCode": error_code, "error": message, }, ensure_ascii=False), file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())