#!/usr/bin/env python3 """Capture a bounded Test3 D3 resource time series without record data.""" from __future__ import annotations import argparse import csv import json import re import shutil import subprocess import sys import time import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any, Sequence ROOT = Path(__file__).resolve().parents[1] RUNTIME_TOOLS = ROOT / "deployments" / "offline" / "runtime" sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(RUNTIME_TOOLS)) from common import db_query_scalar, load_env_file # noqa: E402 CAPTURE_METRICS = ( "nexuscrm_runtime_db_pool_", "nexuscrm_runtime_admission_", "nexuscrm_runtime_rpc_profile_", "nexuscrm_runtime_records_sql_statements_total", "nexuscrm_runtime_worker_", "go_goroutines", "go_memstats_heap_inuse_bytes", "process_cpu_seconds_total", "process_resident_memory_bytes", ) PROM_LINE = re.compile(r"^([A-Za-z_:][A-Za-z0-9_:]*(?:\{[^}]*\})?)\s+([^\s]+)") PROM_LABEL = re.compile(r'([A-Za-z_][A-Za-z0-9_]*)="((?:\\.|[^"\\])*)"') NUMBER = re.compile(r"^-?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?$") WORKER_ENDPOINTS_FILE = "worker-metrics-endpoints.tsv" WORKER_INFO_METRIC = "nexuscrm_runtime_worker_info" WORKER_EXECUTION_CLASSES = {"payload", "metadata"} RUNTIME_REQUIRED_METRICS = ( "nexuscrm_runtime_db_pool_max_open_connections", "go_goroutines", "go_memstats_heap_inuse_bytes", "process_cpu_seconds_total", "process_resident_memory_bytes", ) WORKER_REQUIRED_METRICS = ( "go_goroutines", "go_memstats_heap_inuse_bytes", "process_cpu_seconds_total", "process_resident_memory_bytes", "nexuscrm_runtime_worker_db_pool_max_open_connections", "nexuscrm_runtime_worker_poll_checks_total", "nexuscrm_runtime_worker_claim_attempts_total", "nexuscrm_runtime_worker_processed_total", ) CONTAINER_REQUIRED_FIELDS = ( "name", "cpu", "memory", "memory_percent", "pids", ) WORKER_GAUGES = { "rss_peak_bytes": ("process_resident_memory_bytes", {}), "heap_peak_bytes": ("go_memstats_heap_inuse_bytes", {}), "goroutines_peak": ("go_goroutines", {}), "queue_depth_peak": ("nexuscrm_runtime_worker_queue_depth", {}), "poison_depth_peak": ("nexuscrm_runtime_worker_poison_depth", {}), "in_flight_peak": ("nexuscrm_runtime_worker_in_flight", {}), "oldest_age_seconds_peak": ("nexuscrm_runtime_worker_oldest_age_seconds", {}), "queue_observation_stale_peak": ( "nexuscrm_runtime_worker_queue_observation_stale", {} ), "queue_observation_duration_seconds_peak": ( "nexuscrm_runtime_worker_queue_observation_duration_seconds", {} ), "queue_observation_last_success_timestamp_seconds_latest": ( "nexuscrm_runtime_worker_queue_observation_last_success_timestamp_seconds", {}, ), "primary_pool_max_peak": ( "nexuscrm_runtime_worker_db_pool_max_open_connections", {} ), "primary_pool_open_peak": ( "nexuscrm_runtime_worker_db_pool_open_connections", {} ), "primary_pool_in_use_peak": ( "nexuscrm_runtime_worker_db_pool_in_use_connections", {} ), "primary_pool_idle_peak": ( "nexuscrm_runtime_worker_db_pool_idle_connections", {} ), "queue_observation_pool_active_latest": ( "nexuscrm_runtime_worker_auxiliary_db_pool_active", {"purpose": "queue_observation"}, ), "queue_observation_pool_max_peak": ( "nexuscrm_runtime_worker_auxiliary_db_pool_max_open_connections", {"purpose": "queue_observation"}, ), "queue_observation_pool_open_peak": ( "nexuscrm_runtime_worker_auxiliary_db_pool_open_connections", {"purpose": "queue_observation"}, ), "queue_observation_pool_in_use_peak": ( "nexuscrm_runtime_worker_auxiliary_db_pool_in_use_connections", {"purpose": "queue_observation"}, ), "queue_observation_pool_idle_peak": ( "nexuscrm_runtime_worker_auxiliary_db_pool_idle_connections", {"purpose": "queue_observation"}, ), "rejected_save_pool_active_latest": ( "nexuscrm_runtime_worker_auxiliary_db_pool_active", {"purpose": "rejected_save_audit"}, ), "rejected_save_pool_max_peak": ( "nexuscrm_runtime_worker_auxiliary_db_pool_max_open_connections", {"purpose": "rejected_save_audit"}, ), "rejected_save_pool_open_peak": ( "nexuscrm_runtime_worker_auxiliary_db_pool_open_connections", {"purpose": "rejected_save_audit"}, ), "rejected_save_pool_in_use_peak": ( "nexuscrm_runtime_worker_auxiliary_db_pool_in_use_connections", {"purpose": "rejected_save_audit"}, ), "rejected_save_pool_idle_peak": ( "nexuscrm_runtime_worker_auxiliary_db_pool_idle_connections", {"purpose": "rejected_save_audit"}, ), } WORKER_COUNTERS = { "cpu_seconds_delta": ("process_cpu_seconds_total", {}), "poll_checks_delta": ("nexuscrm_runtime_worker_poll_checks_total", {}), "confirmed_idle_delta": ( "nexuscrm_runtime_worker_confirmed_idle_checks_total", {} ), "hint_failures_delta": ("nexuscrm_runtime_worker_hint_failures_total", {}), "claim_attempts_delta": ("nexuscrm_runtime_worker_claim_attempts_total", {}), "empty_claims_delta": ( "nexuscrm_runtime_worker_empty_claim_attempts_total", {} ), "claimed_delta": ("nexuscrm_runtime_worker_claimed_total", {}), "processed_delta": ("nexuscrm_runtime_worker_processed_total", {}), "failures_delta": ("nexuscrm_runtime_worker_failures_total", {}), "attempts_delta": ("nexuscrm_runtime_worker_attempts_total", {}), "poll_seconds_delta": ( "nexuscrm_runtime_worker_poll_check_duration_seconds_total", {} ), "claim_seconds_delta": ( "nexuscrm_runtime_worker_claim_duration_seconds_total", {} ), "primary_pool_waits_delta": ( "nexuscrm_runtime_worker_db_pool_wait_count_total", {} ), "primary_pool_wait_seconds_delta": ( "nexuscrm_runtime_worker_db_pool_wait_duration_seconds_total", {} ), "queue_observation_pool_waits_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_wait_count_total", {"purpose": "queue_observation"}, ), "queue_observation_pool_wait_seconds_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_wait_duration_seconds_total", {"purpose": "queue_observation"}, ), "queue_observation_pool_opens_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_open_attempts_total", {"purpose": "queue_observation"}, ), "queue_observation_pool_open_failures_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_open_failures_total", {"purpose": "queue_observation"}, ), "queue_observation_pool_closes_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_closes_total", {"purpose": "queue_observation"}, ), "rejected_save_pool_waits_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_wait_count_total", {"purpose": "rejected_save_audit"}, ), "rejected_save_pool_wait_seconds_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_wait_duration_seconds_total", {"purpose": "rejected_save_audit"}, ), "rejected_save_pool_opens_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_open_attempts_total", {"purpose": "rejected_save_audit"}, ), "rejected_save_pool_open_failures_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_open_failures_total", {"purpose": "rejected_save_audit"}, ), "rejected_save_pool_closes_delta": ( "nexuscrm_runtime_worker_auxiliary_db_pool_closes_total", {"purpose": "rejected_save_audit"}, ), "object_id_probe_opens_delta": ( "nexuscrm_runtime_worker_transient_db_open_attempts_total", {"purpose": "object_id_probe"}, ), "object_id_probe_failures_delta": ( "nexuscrm_runtime_worker_transient_db_open_failures_total", {"purpose": "object_id_probe"}, ), "object_id_probe_open_seconds_delta": ( "nexuscrm_runtime_worker_transient_db_open_duration_seconds_total", {"purpose": "object_id_probe"}, ), "observation_failures_delta": ( "nexuscrm_runtime_worker_queue_observation_failures_total", {} ), "scheduled_discovery_delta": ( "nexuscrm_runtime_worker_scheduled_discovery_passes_total", {} ), "scheduled_definitions_considered_delta": ( "nexuscrm_runtime_worker_scheduled_definitions_considered_total", {} ), "scheduled_accepted_delta": ( "nexuscrm_runtime_worker_scheduled_occurrences_accepted_total", {} ), "scheduled_capacity_deferrals_delta": ( "nexuscrm_runtime_worker_scheduled_capacity_deferrals_total", {} ), "scheduled_execution_failed_delta": ( "nexuscrm_runtime_worker_scheduled_execution_failed_total", {} ), "scheduled_execution_started_delta": ( "nexuscrm_runtime_worker_scheduled_execution_started_total", {} ), "scheduled_execution_completed_delta": ( "nexuscrm_runtime_worker_scheduled_execution_completed_total", {} ), "scheduled_recovery_scanned_delta": ( "nexuscrm_runtime_worker_scheduled_recovery_scanned_total", {} ), "scheduled_recovery_recovered_delta": ( "nexuscrm_runtime_worker_scheduled_recovery_recovered_total", {} ), "scheduled_discovery_failures_delta": ( "nexuscrm_runtime_worker_scheduled_discovery_failures_total", {} ), "scheduled_discovery_seconds_delta": ( "nexuscrm_runtime_worker_scheduled_phase_duration_seconds_total", {"phase": "discovery"}, ), "scheduled_recovery_seconds_delta": ( "nexuscrm_runtime_worker_scheduled_phase_duration_seconds_total", {"phase": "recovery"}, ), "scheduled_execution_seconds_delta": ( "nexuscrm_runtime_worker_scheduled_phase_duration_seconds_total", {"phase": "execution"}, ), } QUEUE_OBSERVATION_GAUGE_FIELDS = { "queue_observation_stale_peak", "queue_observation_duration_seconds_peak", "queue_observation_last_success_timestamp_seconds_latest", } QUEUE_OBSERVATION_COUNTER_FIELDS = {"observation_failures_delta"} QUEUE_OBSERVATION_EXEMPT_STEMS = frozenset({"bulk-ingest", "scheduled-flow"}) SCHEDULED_FLOW_COUNTER_FIELDS = { field for field in WORKER_COUNTERS if field.startswith("scheduled_") } SUPPORTED_TARGET_ENVS = {"test3"} def target_env(envv: dict[str, str]) -> str: env_name = str(envv.get("ENV_NAME", "")).strip() slot = env_name.removeprefix("test-") if env_name.startswith("test-") else "" if slot not in SUPPORTED_TARGET_ENVS: raise RuntimeError("observer requires the exact Test3 authority") return slot def tidb_pressure_probe(target: str) -> tuple[str, str]: if target not in SUPPORTED_TARGET_ENVS: raise RuntimeError("unsupported Test observer target") marker = f"nexuscrm_{target}_observer:tidb_pressure_v1" return marker, f""" /* {marker} */ SELECT 'process', COUNT(*), COALESCE(SUM(COMMAND <> 'Sleep'), 0), COALESCE(SUM(COMMAND <> 'Sleep' AND TIME >= 5), 0), COALESCE(MAX(CASE WHEN COMMAND <> 'Sleep' THEN TIME END), 0) FROM INFORMATION_SCHEMA.CLUSTER_PROCESSLIST WHERE ID <> CONNECTION_ID(); /* {marker} */ SELECT 'lock', COUNT(*) FROM INFORMATION_SCHEMA.DATA_LOCK_WAITS; """ def utc_now() -> str: return ( datetime.now(timezone.utc) .isoformat(timespec="milliseconds") .replace("+00:00", "Z") ) def parse_prometheus(text: str) -> dict[str, float]: metrics: dict[str, float] = {} for raw in text.splitlines(): match = PROM_LINE.match(raw.strip()) if not match: continue key, value = match.groups() name = key.split("{", 1)[0] if not name.startswith(CAPTURE_METRICS) or not NUMBER.match(value): continue metrics[key] = float(value) return metrics def scrape_metrics(url: str) -> dict[str, float]: with urllib.request.urlopen(url, timeout=5) as response: # noqa: S310 - loopback URL return parse_prometheus(response.read().decode("utf-8", errors="replace")) def sample_tidb( envv: dict[str, str], target: str ) -> dict[str, int | str]: marker, sql = tidb_pressure_probe(target) try: raw = db_query_scalar(envv, sql, database=envv["TIDB_DATABASE"]) except Exception: return {"error": "probe_failed"} result: dict[str, int | str] = {"probe": marker} for line in raw.splitlines(): parts = line.split("\t") if parts[:1] == ["process"] and len(parts) == 5: result.update( sessions=int(parts[1]), running=int(parts[2]), running_ge_5s=int(parts[3]), max_running_seconds=int(parts[4]), ) elif parts[:1] == ["lock"] and len(parts) == 2: result["lock_waits"] = int(parts[1]) required = { "sessions", "running", "running_ge_5s", "max_running_seconds", "lock_waits", } return result if required <= result.keys() else {"error": "empty_probe"} def configured_test_d3(envv: dict[str, str]) -> tuple[str, Any]: from scripts import run_serviceification_d3_test as test_d3 # noqa: PLC0415 slot = target_env(envv) test_d3.bind_slot(slot) test_d3._apply_impl_identity(slot) return slot, test_d3 def resolve_expected_containers(envv: dict[str, str]) -> list[str]: slot, test_d3 = configured_test_d3(envv) d3 = test_d3.impl env_root = envv.get("ENV_ROOT", "").strip() if not env_root: raise RuntimeError("container inventory requires the exact Test authority") resolved_root = Path(env_root).expanduser().resolve() state = d3.read_state( resolved_root / ".runtime" / "serviceification-d3" / "active.json" ) if ( state.get("envName") != f"test-{slot}" or state.get("envRoot") != str(resolved_root) ): raise RuntimeError( f"active D3 state does not match the {slot} authority" ) return d3.expected_serving_container_order(state) def sample_containers( expected_names: Sequence[str], target: str = "test3" ) -> list[dict[str, str]]: if target not in SUPPORTED_TARGET_ENVS: raise RuntimeError("unsupported Test container target") expected = tuple(expected_names) if not expected or len(set(expected)) != len(expected): raise RuntimeError("expected container inventory is incomplete") names = subprocess.run( [ "docker", "ps", "--filter", f"label=io.nexuscrm.serviceification-d3={target}", "--format", "{{.Names}}", ], check=True, capture_output=True, text=True, ).stdout.splitlines() if len(names) != len(expected) or set(names) != set(expected): raise RuntimeError("active container inventory does not match D3 state") output = subprocess.run( ["docker", "stats", "--no-stream", "--format", "{{json .}}", *names], check=True, capture_output=True, text=True, ).stdout rows: list[dict[str, str]] = [] for line in output.splitlines(): try: payload = json.loads(line) except json.JSONDecodeError as exc: raise RuntimeError("Docker stats returned malformed JSON") from exc if not isinstance(payload, dict): raise RuntimeError("Docker stats returned a non-object row") row = { "name": str(payload.get("Name", "")).strip(), "cpu": str(payload.get("CPUPerc", "")).strip(), "memory": str(payload.get("MemUsage", "")).strip(), "memory_percent": str(payload.get("MemPerc", "")).strip(), "pids": str(payload.get("PIDs", "")).strip(), } if any(not row[field] for field in CONTAINER_REQUIRED_FIELDS): raise RuntimeError("Docker stats returned an incomplete row") rows.append(row) row_names = [row["name"] for row in rows] if len(row_names) != len(expected) or set(row_names) != set(expected): raise RuntimeError("Docker stats did not return the exact D3 inventory") return rows def parse_metric_key(key: str) -> tuple[str, dict[str, str]]: name, separator, raw_labels = key.partition("{") if not separator: return name, {} labels: dict[str, str] = {} for match in PROM_LABEL.finditer(raw_labels.removesuffix("}")): label, escaped = match.groups() try: labels[label] = json.loads(f'"{escaped}"') except json.JSONDecodeError: labels[label] = escaped return name, labels def selected_metric_value( metrics: dict[str, float], name: str, required_labels: dict[str, str] ) -> float | None: matches: list[float] = [] for key, value in metrics.items(): metric_name, labels = parse_metric_key(key) if metric_name == name and all( labels.get(label) == expected for label, expected in required_labels.items() ): matches.append(value) if len(matches) == 1: return float(matches[0]) return None def worker_identity(metrics: dict[str, float]) -> tuple[str, str] | None: identities: list[tuple[str, str]] = [] for key, value in metrics.items(): name, labels = parse_metric_key(key) if name != WORKER_INFO_METRIC or value != 1: continue work_family = labels.get("work_family", "") execution_class = labels.get("execution_class", "") if ( set(labels) != {"work_family", "execution_class"} or not work_family or execution_class not in WORKER_EXECUTION_CLASSES ): return None identities.append((work_family, execution_class)) return identities[0] if len(identities) == 1 else None def resolve_worker_endpoints(envv: dict[str, str]) -> list[dict[str, str | int]]: slot, test_d3 = configured_test_d3(envv) d3 = test_d3.impl env_root = Path(envv.get("ENV_ROOT", "")).expanduser().resolve() state = d3.read_state( env_root / ".runtime" / "serviceification-d3" / "active.json" ) worker_stems = d3.state_worker_stems(state) ports = test_d3.derive_ports(envv, worker_stems, slot=slot) endpoints: list[dict[str, str | int]] = [] for stem in worker_stems: family = f"runtime.{stem}.v1" port = ports.worker_admin[stem] endpoints.append( { "stem": stem, "work_family": family, "port": port, "metrics_url": f"http://127.0.0.1:{port}/metrics", } ) return endpoints def write_worker_endpoints( output_dir: Path, endpoints: list[dict[str, str | int]] ) -> None: path = output_dir / WORKER_ENDPOINTS_FILE with path.open("w", newline="") as handle: writer = csv.DictWriter( handle, fieldnames=("stem", "work_family", "port", "metrics_url"), delimiter="\t", ) writer.writeheader() writer.writerows(endpoints) def load_worker_endpoints(output_dir: Path) -> list[dict[str, str | int]]: path = output_dir / WORKER_ENDPOINTS_FILE if not path.is_file(): return [] with path.open(newline="") as handle: return [ { "stem": row["stem"], "work_family": row["work_family"], "port": int(row["port"]), "metrics_url": row["metrics_url"], } for row in csv.DictReader(handle, delimiter="\t") ] def runtime_metric_values(samples: list[dict[str, Any]], name: str) -> list[float]: return [ float(sample["runtime_metrics"][name]) for sample in samples if name in sample.get("runtime_metrics", {}) ] def worker_metric_values( samples: list[dict[str, Any]], stem: str, name: str, labels: dict[str, str], ) -> list[float]: values: list[float] = [] for sample in samples: metrics = sample.get("workers", {}).get(stem, {}).get("metrics", {}) value = selected_metric_value(metrics, name, labels) if value is not None: values.append(value) return values def values_counter_delta(values: list[float]) -> float | None: if len(values) < 2: return None delta = 0.0 for before, after in zip(values, values[1:]): delta += after - before if after >= before else after return delta def required_worker_cost_fields( stem: str, ) -> tuple[set[str], set[str]]: gauge_fields = set(WORKER_GAUGES) counter_fields = set(WORKER_COUNTERS) if stem in QUEUE_OBSERVATION_EXEMPT_STEMS: gauge_fields -= QUEUE_OBSERVATION_GAUGE_FIELDS counter_fields -= QUEUE_OBSERVATION_COUNTER_FIELDS if stem != "scheduled-flow": counter_fields -= SCHEDULED_FLOW_COUNTER_FIELDS return gauge_fields, counter_fields def required_worker_cost_specs( stem: str, ) -> tuple[tuple[str, dict[str, str]], ...]: gauge_fields, counter_fields = required_worker_cost_fields(stem) specs = { (name, tuple(sorted(labels.items()))) for field, (name, labels) in WORKER_GAUGES.items() if field in gauge_fields } specs.update( (name, tuple(sorted(labels.items()))) for field, (name, labels) in WORKER_COUNTERS.items() if field in counter_fields ) return tuple( (name, dict(labels)) for name, labels in sorted(specs) ) def summarize_worker( samples: list[dict[str, Any]], stem: str, expected_family: str ) -> dict[str, Any]: observed = [ sample["workers"][stem] for sample in samples if stem in sample.get("workers", {}) ] required_gauges, required_counters = required_worker_cost_fields(stem) required_specs = required_worker_cost_specs(stem) summary: dict[str, Any] = { "work_family": expected_family, "execution_class": observed[-1].get("execution_class") if observed else None, "sample_count": len(observed), "collection_failures": sum( stem in sample.get("worker_errors", {}) for sample in samples ), "identity_valid": bool(observed) and all(worker.get("work_family") == expected_family for worker in observed) and len({worker.get("execution_class") for worker in observed}) == 1, "required_metrics_complete": bool(observed) and len(observed) == len(samples) and all( all( selected_metric_value(worker.get("metrics", {}), name, {}) is not None for name in WORKER_REQUIRED_METRICS ) for worker in observed ) and all( len(worker_metric_values(samples, stem, name, labels)) == len(samples) for name, labels in required_specs ), "not_applicable_fields": sorted( (set(WORKER_GAUGES) - required_gauges) | (set(WORKER_COUNTERS) - required_counters) ), "gauges": {}, "counter_deltas": {}, } for field, (name, labels) in WORKER_GAUGES.items(): values = worker_metric_values(samples, stem, name, labels) if values: summary["gauges"][field] = ( values[-1] if field.endswith("_latest") else max(values) ) for field, (name, labels) in WORKER_COUNTERS.items(): delta = values_counter_delta( worker_metric_values(samples, stem, name, labels) ) if delta is not None: summary["counter_deltas"][field] = delta return summary def summarize( samples: list[dict[str, Any]], endpoints: list[dict[str, str | int]], expected_containers: Sequence[str], *, cancelled: bool = False, target: str = "test3", ) -> dict[str, Any]: if target not in SUPPORTED_TARGET_ENVS: raise RuntimeError("unsupported Test summary target") gauges = ( "nexuscrm_runtime_db_pool_max_open_connections", "nexuscrm_runtime_db_pool_open_connections", "nexuscrm_runtime_db_pool_in_use_connections", "nexuscrm_runtime_db_pool_idle_connections", "nexuscrm_runtime_admission_limit{role=\"runtime-api\"}", "nexuscrm_runtime_admission_in_flight{role=\"runtime-api\"}", "go_goroutines", "go_memstats_heap_inuse_bytes", "process_resident_memory_bytes", ) counters = ( "nexuscrm_runtime_db_pool_wait_count_total", "nexuscrm_runtime_db_pool_wait_duration_seconds_total", "nexuscrm_runtime_admission_rejected_total{reason=\"capacity\",role=\"runtime-api\"}", ) summary: dict[str, Any] = { "schema": f"nexuscrm.{target}-runtime-observation/v2", "target_env": target, "cancelled": cancelled, "sample_count": len(samples), "first_at": samples[0]["at"] if samples else None, "last_at": samples[-1]["at"] if samples else None, "runtime_scrape_failures": sum( "runtime_error" in sample for sample in samples ), "tidb_probe_failures": sum( "error" in sample.get("tidb", {}) for sample in samples ), "runtime_peaks": {}, "counter_deltas": {}, "tidb_peaks": {}, "observer_tidb_probe_count": sum("tidb" in sample for sample in samples), "expected_worker_count": len(endpoints), "expected_container_count": len(expected_containers), "workers": {}, } for name in gauges: values = runtime_metric_values(samples, name) if values: summary["runtime_peaks"][name] = max(values) for name in counters: delta = values_counter_delta(runtime_metric_values(samples, name)) if delta is not None: summary["counter_deltas"][name] = delta for name in ( "sessions", "running", "running_ge_5s", "max_running_seconds", "lock_waits", ): values = [ int(sample["tidb"][name]) for sample in samples if name in sample.get("tidb", {}) ] if values: summary["tidb_peaks"][name] = max(values) for endpoint in endpoints: stem = str(endpoint["stem"]) summary["workers"][stem] = summarize_worker( samples, stem, str(endpoint["work_family"]) ) summary["worker_scrape_failures"] = sum( sum( error == "scrape_failed" for error in sample.get("worker_errors", {}).values() ) for sample in samples ) summary["worker_identity_failures"] = sum( sum( error == "identity_mismatch" for error in sample.get("worker_errors", {}).values() ) for sample in samples ) + sum( not worker["identity_valid"] and worker["sample_count"] > 0 for worker in summary["workers"].values() ) summary["worker_collection_failures"] = ( summary["worker_scrape_failures"] + summary["worker_identity_failures"] ) expected_container_set = set(expected_containers) def exact_container_sample(sample: dict[str, Any]) -> bool: rows = sample.get("containers") if not isinstance(rows, list): return False names: list[str] = [] for row in rows: if not isinstance(row, dict) or any( not isinstance(row.get(field), str) or not row[field].strip() for field in CONTAINER_REQUIRED_FIELDS ): return False names.append(row["name"]) return ( len(names) == len(expected_containers) and len(set(names)) == len(names) and set(names) == expected_container_set ) summary["container_probe_failures"] = sum( "tidb" in sample and ( "container_error" in sample or not exact_container_sample(sample) ) for sample in samples ) summary["container_probe_count"] = sum( "tidb" in sample and exact_container_sample(sample) for sample in samples ) summary["runtime_metrics_complete"] = all( all( name in sample.get("runtime_metrics", {}) for name in RUNTIME_REQUIRED_METRICS ) for sample in samples ) exact_worker_set = all( set(sample.get("workers", {})) == set(summary["workers"]) for sample in samples ) complete_worker_inputs = all( worker["sample_count"] == len(samples) and worker["identity_valid"] and worker["required_metrics_complete"] and worker["execution_class"] in WORKER_EXECUTION_CLASSES for worker in summary["workers"].values() ) complete_worker_costs = complete_worker_inputs and all( required_worker_cost_fields(stem)[0] <= set(worker["gauges"]) and required_worker_cost_fields(stem)[1] <= set(worker["counter_deltas"]) for stem, worker in summary["workers"].items() ) summary["ready"] = bool( samples and endpoints and summary["runtime_scrape_failures"] == 0 and summary["runtime_metrics_complete"] and summary["worker_collection_failures"] == 0 and summary["tidb_probe_failures"] == 0 and summary["observer_tidb_probe_count"] > 0 and summary["container_probe_failures"] == 0 and summary["expected_container_count"] > 0 and exact_worker_set and complete_worker_inputs ) summary["accepted"] = bool( summary["ready"] and len(samples) >= 2 and not cancelled and complete_worker_costs ) return summary def write_worker_cost(output_dir: Path, summary: dict[str, Any]) -> None: fields = [ "stem", "work_family", "execution_class", "sample_count", "collection_failures", *WORKER_GAUGES, *WORKER_COUNTERS, ] with (output_dir / "worker-cost.csv").open("w", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() for stem, worker in summary["workers"].items(): writer.writerow( { "stem": stem, "work_family": worker["work_family"], "execution_class": worker["execution_class"], "sample_count": worker["sample_count"], "collection_failures": worker["collection_failures"], **worker["gauges"], **worker["counter_deltas"], } ) def write_summary(output_dir: Path, summary: dict[str, Any]) -> None: (output_dir / "summary.json").write_text( json.dumps(summary, indent=2, sort_keys=True) + "\n" ) write_worker_cost(output_dir, summary) target = str(summary.get("target_env", "test3")) lines = [ f"# {target} Runtime and Worker decision evidence", "", f"- accepted: `{str(summary['accepted']).lower()}`", f"- cancelled: `{str(summary['cancelled']).lower()}`", f"- samples: {summary['sample_count']}", f"- window: {summary['first_at']} .. {summary['last_at']}", f"- Runtime scrape failures: {summary['runtime_scrape_failures']}", f"- Worker scrape failures: {summary['worker_scrape_failures']}", f"- Worker identity failures: {summary['worker_identity_failures']}", f"- Container probe failures: {summary['container_probe_failures']}", f"- TiDB probe failures: {summary['tidb_probe_failures']}", "", "## Per-Worker cost", "", "| Worker | Class | RSS peak | Heap peak | CPU seconds | " "Primary pool max | Idle checks | Processed |", "|---|---:|---:|---:|---:|---:|---:|---:|", ] for stem, worker in summary["workers"].items(): gauges = worker["gauges"] counters = worker["counter_deltas"] lines.append( f"| `{stem}` | {worker['execution_class'] or ''} | " f"{gauges.get('rss_peak_bytes', '')} | " f"{gauges.get('heap_peak_bytes', '')} | " f"{counters.get('cpu_seconds_delta', '')} | " f"{gauges.get('primary_pool_max_peak', '')} | " f"{counters.get('confirmed_idle_delta', '')} | " f"{counters.get('processed_delta', '')} |" ) lines.extend( [ "", "The raw time series is `samples.jsonl`; normalized role costs are in " "`worker-cost.csv`; TiDB physical statement evidence is in " "`statement-digests.tsv` when SQL diagnostics completed.", ] ) (output_dir / "decision-summary.md").write_text("\n".join(lines) + "\n") def load_existing_samples(output_dir: Path) -> list[dict[str, Any]]: samples_path = output_dir / "samples.jsonl" samples: list[dict[str, Any]] = [] if not samples_path.is_file(): return samples for line in samples_path.read_text().splitlines(): try: payload = json.loads(line) except json.JSONDecodeError: continue if isinstance(payload, dict): samples.append(payload) return samples def initialize_output(output_dir: Path) -> None: for name in ( "samples.jsonl", "summary.json", "summary.md", "worker-cost.csv", "decision-summary.md", WORKER_ENDPOINTS_FILE, "statement-digests.tsv", "statement-summary-columns.tsv", "statement-summary-bucket-totals.tsv", "resource-groups.tsv", "focused-statement-summary.tsv", "focused-statement-summary-by-bucket.tsv", "top-statement-summary.tsv", "top-statement-summary-by-bucket.tsv", "catalog-explain-analyze.tsv", "scheduled-reclaim-explain-analyze.tsv", "record-event-observation-explain-analyze.tsv", "focused-slow-query.tsv", ): (output_dir / name).unlink(missing_ok=True) shutil.rmtree(output_dir / "profiles", ignore_errors=True) (output_dir / "samples.jsonl").touch() def run(args: argparse.Namespace) -> int: output_dir = Path(args.output_dir) output_dir.mkdir(parents=True, exist_ok=True) envv = load_env_file(Path(args.runtime_env)) target = target_env(envv) expected_containers = resolve_expected_containers(envv) if args.summarize_existing: samples = load_existing_samples(output_dir) endpoints = load_worker_endpoints(output_dir) summary = summarize( samples, endpoints, expected_containers, cancelled=True, target=target, ) write_summary(output_dir, summary) return 0 initialize_output(output_dir) _, test_d3 = configured_test_d3(envv) runtime_admin_port = test_d3.derive_ports(envv, slot=target).runtime_admin metrics_url = args.metrics_url or ( f"http://127.0.0.1:{runtime_admin_port}/metrics" ) endpoints = resolve_worker_endpoints(envv) write_worker_endpoints(output_dir, endpoints) samples: list[dict[str, Any]] = [] deadline = time.monotonic() + args.duration_seconds next_heavy_sample = 0.0 while True: now = time.monotonic() sample: dict[str, Any] = {"at": utc_now()} try: sample["runtime_metrics"] = scrape_metrics(metrics_url) except Exception: sample["runtime_error"] = "scrape_failed" for endpoint in endpoints: stem = str(endpoint["stem"]) try: metrics = scrape_metrics(str(endpoint["metrics_url"])) except Exception: sample.setdefault("worker_errors", {})[stem] = "scrape_failed" continue identity = worker_identity(metrics) if identity is None or identity[0] != endpoint["work_family"]: sample.setdefault("worker_errors", {})[stem] = "identity_mismatch" continue sample.setdefault("workers", {})[stem] = { "work_family": identity[0], "execution_class": identity[1], "metrics": metrics, } if now >= next_heavy_sample: sample["tidb"] = sample_tidb(envv, target) try: sample["containers"] = sample_containers( expected_containers, target ) except Exception: sample["container_error"] = "probe_failed" next_heavy_sample = now + args.heavy_interval_seconds samples.append(sample) with (output_dir / "samples.jsonl").open("a") as handle: handle.write(json.dumps(sample, sort_keys=True) + "\n") complete = ( "runtime_metrics" in sample and set(sample.get("workers", {})) == {str(endpoint["stem"]) for endpoint in endpoints} ) if complete and not any( previous.get("observer_ready", False) for previous in samples[:-1] ): sample["observer_ready"] = True print(f"OBSERVER_READY at={sample['at']}", flush=True) remaining = deadline - time.monotonic() if remaining <= 0: break time.sleep(min(args.interval_seconds, remaining)) write_summary( output_dir, summarize(samples, endpoints, expected_containers, target=target), ) return 0 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--runtime-env", required=True) parser.add_argument("--output-dir", required=True) parser.add_argument("--metrics-url", default="") parser.add_argument("--duration-seconds", type=int, required=True) parser.add_argument("--interval-seconds", type=int, default=2) parser.add_argument("--heavy-interval-seconds", type=int, default=30) parser.add_argument("--summarize-existing", action="store_true") args = parser.parse_args() if not 1 <= args.duration_seconds <= 1800: parser.error("duration-seconds must be between 1 and 1800") if not 1 <= args.interval_seconds <= 30: parser.error("interval-seconds must be between 1 and 30") if not 5 <= args.heavy_interval_seconds <= 60: parser.error("heavy-interval-seconds must be between 5 and 60") return args if __name__ == "__main__": raise SystemExit(run(parse_args()))