java_async_alloc_profile
Capture allocation profiling data for a Java process using async-profiler to identify memory allocation hotspots.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| pid | Yes | ||
| duration_s | No | ||
| out_file | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/heap_seance_mcp/tools.py:344-399 (handler)Core handler for java_async_alloc_profile. Uses async-profiler (via ASYNC_PROFILER_BIN env or PATH) with '-e alloc' event, runs for a duration, and generates an HTML allocation profile.
def java_async_alloc_profile( pid: int, duration_s: int = 30, out_file: str | None = None, ) -> dict[str, Any]: try: async_bin = os.environ.get("ASYNC_PROFILER_BIN") if async_bin: profiler = async_bin else: profiler = require_any_binary( ["async-profiler", "profiler.sh", "asprof"], "Install async-profiler and set ASYNC_PROFILER_BIN if not in PATH.", ) path = Path(out_file) if out_file else _artifact_dir() / f"alloc-{pid}-{_timestamp()}.html" path.parent.mkdir(parents=True, exist_ok=True) output = ensure_success( run_command( [ profiler, "-e", "alloc", "-d", str(max(5, duration_s)), "-f", str(path), str(pid), ], timeout_s=max(60, duration_s + 20), ) ).stdout except ToolingMissingError as exc: return _missing_tool(str(exc)) except Exception as exc: # noqa: BLE001 return _command_failed(exc) if not path.exists(): return warn_result( evidence=[ f"async-profiler finished but output file is missing at {path}.", output.strip()[:800], ], metrics={"pid": pid, "requested_file": str(path)}, confidence="low", next_recommended_action="Validate profiler permissions and rerun with explicit output path.", ) return ok_result( evidence=[f"Allocation profile generated for PID {pid}: {path}"], metrics={"pid": pid, "duration_s": duration_s}, confidence="medium", next_recommended_action="Use this profile as tie-breaker if MAT and histogram are inconclusive.", raw_artifact_path=str(path), ) - src/heap_seance_mcp/server.py:65-71 (registration)MCP tool registration using @mcp.tool() decorator that exposes java_async_alloc_profile as an MCP tool endpoint with parameters pid, duration_s, out_file.
@mcp.tool() def java_async_alloc_profile( pid: int, duration_s: int = 30, out_file: str | None = None, ) -> dict[str, Any]: return tools.java_async_alloc_profile(pid=pid, duration_s=duration_s, out_file=out_file) - Workflow integration: calls java_async_alloc_profile as part of the deep forensic workflow when async-profiler is available.
if async_available: alloc_result = java_async_alloc_profile(pid=resolved_pid, duration_s=30) if alloc_result["status"] == "ok": report["artifacts"]["async_alloc_profile"] = alloc_result["raw_artifact_path"] else: report["evidence"].append( "async-profiler execution failed; falling back to JFR+MAT evidence." ) report["signals"]["async_profiler_runtime_error"] = True - Helper utilities (require_any_binary, run_command, ensure_success) used by the handler to locate async-profiler binary and execute it.
def require_binary(binary: str, install_hint: str) -> str: location = which(binary) if not location: raise ToolingMissingError( f"Missing required binary '{binary}'. {install_hint}" ) return location def require_any_binary(candidates: Sequence[str], install_hint: str) -> str: for candidate in candidates: location = which(candidate) if location: return location joined = ", ".join(candidates) raise ToolingMissingError( f"Missing required binary. Tried: {joined}. {install_hint}" ) def run_command( command: Sequence[str], *, timeout_s: int = 180, cwd: str | None = None, env: dict[str, str] | None = None, ) -> CommandResult: base_env = os.environ.copy() if env: base_env.update(env) try: proc = subprocess.run( list(command), capture_output=True, text=True, timeout=timeout_s, cwd=cwd, env=base_env, check=False, ) except subprocess.TimeoutExpired as exc: pretty = " ".join(shlex.quote(part) for part in command) raise CommandExecutionError( f"Command timed out after {timeout_s}s: {pretty}" ) from exc return CommandResult( command=list(command), returncode=proc.returncode, stdout=proc.stdout, stderr=proc.stderr, ) def ensure_success(result: CommandResult) -> CommandResult: if result.returncode == 0: return result pretty = " ".join(shlex.quote(part) for part in result.command) stderr = result.stderr.strip() or "<empty>" stdout = result.stdout.strip() or "<empty>" raise CommandExecutionError( f"Command failed ({result.returncode}): {pretty}\nstdout: {stdout}\nstderr: {stderr}" ) - src/heap_seance_mcp/results.py:37-91 (helper)Result builder functions (ok_result, warn_result, error_result) used by the handler to return standardized responses.
def ok_result( *, evidence: list[str] | None = None, metrics: dict[str, Any] | None = None, confidence: str = "low", next_recommended_action: str = "", raw_artifact_path: str | None = None, details: dict[str, Any] | None = None, ) -> dict[str, Any]: return ToolResult( status="ok", evidence=evidence or [], metrics=metrics or {}, confidence=confidence, next_recommended_action=next_recommended_action, raw_artifact_path=raw_artifact_path, details=details or {}, ).to_dict() def warn_result( *, evidence: list[str] | None = None, metrics: dict[str, Any] | None = None, confidence: str = "low", next_recommended_action: str = "", raw_artifact_path: str | None = None, details: dict[str, Any] | None = None, ) -> dict[str, Any]: return ToolResult( status="warn", evidence=evidence or [], metrics=metrics or {}, confidence=confidence, next_recommended_action=next_recommended_action, raw_artifact_path=raw_artifact_path, details=details or {}, ).to_dict() def error_result( message: str, *, next_recommended_action: str = "", details: dict[str, Any] | None = None, ) -> dict[str, Any]: return ToolResult( status="error", evidence=[message], metrics={}, confidence="none", next_recommended_action=next_recommended_action, raw_artifact_path=None, details=details or {}, ).to_dict()