From 5ae3214514da27fc1ac31f5b9c601d894a6c7483 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Fri, 14 Aug 2026 08:26:49 +0800 Subject: [PATCH 1/3] feat(adapters): add node-cpu-prof and node-heap-prof built-in adapters Register two new built-in adapters for Node.js/V8 profiling: - node-cpu-prof: captures V8 CPU profiles via Node's stable --cpu-prof flags, preserving the native .cpuprofile as a sample_profile artifact. - node-heap-prof: captures V8 sampling heap profiles via Node's stable --heap-prof flags, preserving the native .heapprofile as a memory_profile artifact. Both adapters inject the profiling flags after argv[0] and before the user script, binding the output directory and file name explicitly so Flameox owns the artifact path. Closes #297 and #298. --- src/flameox/adapters/builtins.py | 120 +++++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) diff --git a/src/flameox/adapters/builtins.py b/src/flameox/adapters/builtins.py index cde209a..e72729d 100644 --- a/src/flameox/adapters/builtins.py +++ b/src/flameox/adapters/builtins.py @@ -72,6 +72,51 @@ class CaptureInvocation: expected_overhead="No profiler overhead; process output only.", capture_limitations=("No sampled stack or operator evidence is collected.",), ), + BuiltinAdapter( + name="node-cpu-prof", + dependency_kind=AdapterDependencyKind.EXECUTABLE, + dependency="node", + supported_modes=("record",), + supported_formats=("v8-cpuprofile",), + features=("sampled_stacks", "javascript_symbols"), + remediation=( + "Install Node.js 20.16+ or 22.4+ which expose stable --cpu-prof flags.", + ), + version_args=("--version",), + output_filename="cpu.cpuprofile", + artifact_kinds=(ArtifactKind.SAMPLE_PROFILE,), + expected_overhead=( + "V8 CPU sampling overhead; exact rate depends on --cpu-prof-interval." + ), + capture_limitations=( + "Only the main Node.js thread is profiled; worker threads are not sampled.", + "The CPU profile contains sampled stack locations, not wall-clock or " + "allocation evidence.", + ), + ), + BuiltinAdapter( + name="node-heap-prof", + dependency_kind=AdapterDependencyKind.EXECUTABLE, + dependency="node", + supported_modes=("record",), + supported_formats=("v8-sampling-heap-profile",), + features=("allocations", "sampled_allocations", "stacks"), + remediation=( + "Install Node.js 20.16+ or 22.4+ which expose stable --heap-prof flags.", + ), + version_args=("--version",), + output_filename="heap.heapprofile", + artifact_kinds=(ArtifactKind.MEMORY_PROFILE,), + expected_overhead=( + "V8 heap sampling overhead; exact rate depends on --heap-prof-interval." + ), + capture_limitations=( + "Sampled allocation bytes are an estimate, not the exact retained heap or " + "process RSS.", + "Only allocations sampled by V8 are reported; small or short-lived " + "allocations may be underrepresented.", + ), + ), BuiltinAdapter( name="benchmark-samples", dependency_kind=AdapterDependencyKind.INTERNAL, @@ -520,6 +565,15 @@ def build_capture_invocation( # noqa: C901 - provider routing is intentionally output, *target, ) + elif adapter_name in {"node-cpu-prof", "node-heap-prof"}: + return _node_v8_capture_invocation( + adapter_name, + adapter, + workload_argv, + output_root, + output, + executable=executable, + ) elif adapter_name == "torch.profiler": return _torch_capture_invocation( adapter, @@ -742,6 +796,72 @@ def _torch_capture_invocation( ) +def _node_v8_capture_invocation( + adapter_name: str, + adapter: BuiltinAdapter, + workload_argv: tuple[str, ...], + output_root: Path, + output: str, + *, + executable: str | None, +) -> CaptureInvocation: + """Inject Node.js --cpu-prof or --heap-prof flags into a declared Node workload. + + Node.js exposes stable V8 profiling through CLI flags (Node 20.16+ / 22.4+). + The declared workload argv already starts with the Node executable, so the + adapter inserts the profiling flags immediately after argv[0] and before the + user script and its arguments. The output directory and file name are bound + explicitly so Flameox owns the artifact path and can preserve it. + """ + if not workload_argv: + raise DomainError( + ErrorCode.INVALID_CAPTURE_PLAN, + "A declared Node.js workload command is required for V8 profiling.", + ) + node_executable = workload_argv[0] + node_name = Path(node_executable).name + if not ( + node_name == "node" + or node_name.startswith("node") + or node_executable.endswith("node") + ): + raise DomainError( + ErrorCode.INVALID_CAPTURE_PLAN, + f"V8 profiling requires a Node.js workload; the declared " + f"executable is {node_executable!r}.", + remediation=("Declare a Node.js command (e.g. `node script.js`) as the workload.",), + ) + if len(workload_argv) < 2: + raise DomainError( + ErrorCode.INVALID_CAPTURE_PLAN, + "A declared Node.js script or module is required after the node executable.", + ) + # Compute the directory and file name for the V8 profile output. + output_path = Path(output) + prof_dir = str(output_path.parent) + prof_name = output_path.name + if adapter_name == "node-cpu-prof": + prof_flags = ( + "--cpu-prof", + "--cpu-prof-dir=" + prof_dir, + "--cpu-prof-name=" + prof_name, + ) + else: + prof_flags = ( + "--heap-prof", + "--heap-prof-dir=" + prof_dir, + "--heap-prof-name=" + prof_name, + ) + argv = (node_executable, *prof_flags, *workload_argv[1:]) + return CaptureInvocation( + argv=argv, + artifact_kinds=adapter.artifact_kinds, + expected_overhead=adapter.expected_overhead or "", + limitations=adapter.capture_limitations, + environment={}, + ) + + def _compute_sanitizer_capture_invocation( adapter: BuiltinAdapter, workload_argv: tuple[str, ...], From 810b7a4bc93fefe7c5b42d42e6395a0606c28258 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Fri, 14 Aug 2026 08:26:59 +0800 Subject: [PATCH 2/3] feat(adapters): add V8 CPU and heap profile extractors Add V8CpuProfExtractor and V8HeapProfExtractor that parse the native V8 CPU profile (.cpuprofile) and V8 sampling heap profile (.heapprofile) formats and publish bounded frame measurements for hotspot and memory analysis. - V8CpuProfExtractor: parses nodes/samples, publishes hit-count frame measurements keyed by function name, URL, line, and column. - V8HeapProfExtractor: parses the sampling heap profile head tree, publishes self-size frame measurements in bytes, explicitly distinguishing sampled allocation bytes from retained heap or RSS. Both extractors preserve JavaScript locations and script URLs without applying source maps, keeping generated locations distinguishable. Export the new extractors from the adapters package. --- src/flameox/adapters/__init__.py | 8 + src/flameox/adapters/v8_cpu_prof.py | 233 +++++++++++++++++++++++++++ src/flameox/adapters/v8_heap_prof.py | 226 ++++++++++++++++++++++++++ 3 files changed, 467 insertions(+) create mode 100644 src/flameox/adapters/v8_cpu_prof.py create mode 100644 src/flameox/adapters/v8_heap_prof.py diff --git a/src/flameox/adapters/__init__.py b/src/flameox/adapters/__init__.py index a33a94c..909905a 100644 --- a/src/flameox/adapters/__init__.py +++ b/src/flameox/adapters/__init__.py @@ -27,6 +27,8 @@ from flameox.adapters.setup_runtime import * # noqa: F403 from flameox.adapters.torch_profiler import * # noqa: F403 from flameox.adapters.toxiproxy import * # noqa: F403 + from flameox.adapters.v8_cpu_prof import * # noqa: F403 + from flameox.adapters.v8_heap_prof import * # noqa: F403 _MODULES = ( @@ -50,6 +52,8 @@ "registry", "setup_runtime", "torch_profiler", + "v8_cpu_prof", + "v8_heap_prof", "toxiproxy", ) @@ -135,6 +139,10 @@ "TraceProcessorInstallation", "TraceWindowResult", "TritonCompilerOptions", + "V8CpuProfExtractionResult", + "V8CpuProfExtractor", + "V8HeapProfExtractionResult", + "V8HeapProfExtractor", "VllmAggregateMetrics", "VllmMeasurementRow", "VllmResultDocument", diff --git a/src/flameox/adapters/v8_cpu_prof.py b/src/flameox/adapters/v8_cpu_prof.py new file mode 100644 index 0000000..87a7a04 --- /dev/null +++ b/src/flameox/adapters/v8_cpu_prof.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from flameox.domain import ArtifactKind, DomainError, ErrorCode, digest_model +from flameox.evidence import GenerationPublisher +from flameox.models import ContractModel +from flameox.storage import ArtifactStore, RunStore, Workspace + + +class V8CpuProfExtractionResult(ContractModel): + schema_version: int = 1 + run_id: str + artifact_id: str + node_count: int + sample_count: int + frame_count: int + corpus_commit_id: str + limitations: tuple[str, ...] = () + + +class V8CpuProfExtractor: + name = "node-cpu-prof" + version = "1" + + def __init__(self, workspace: Workspace) -> None: + self.workspace = workspace + self.publisher = GenerationPublisher(workspace) + + def extract(self, run_id: str) -> V8CpuProfExtractionResult: + run = RunStore(self.workspace).read(run_id) + registrations = [ + item for item in run.artifacts if item.kind is ArtifactKind.SAMPLE_PROFILE + ] + if len(registrations) != 1: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The run must contain exactly one V8 CPU profile artifact.", + run_id=run_id, + ) + registration = registrations[0] + artifact = ArtifactStore(self.workspace).get(registration.artifact_id) + try: + payload = json.loads(artifact.payload_path.read_text(encoding="utf-8")) + except (OSError, ValueError, json.JSONDecodeError) as exc: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The artifact is not a supported V8 CPU profile (.cpuprofile).", + run_id=run_id, + ) from exc + if not isinstance(payload, dict) or "nodes" not in payload: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The artifact does not contain a V8 CPU profile node tree.", + run_id=run_id, + ) + nodes = payload.get("nodes") + samples = payload.get("samples") or [] + if not isinstance(nodes, list) or not isinstance(samples, list): + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The V8 CPU profile nodes or samples are malformed.", + run_id=run_id, + ) + nodes_by_id: dict[int, dict[str, Any]] = {} + for node in nodes: + if not isinstance(node, dict): + continue + nodes_by_id[int(node["id"])] = node + frame_rows: dict[str, dict[str, Any]] = {} + aggregates: dict[tuple[str, str], dict[str, int]] = {} + for node in nodes_by_id.values(): + self._aggregate_node( + node, + nodes_by_id=nodes_by_id, + frame_rows=frame_rows, + aggregates=aggregates, + artifact_id=registration.artifact_id, + ) + measurement_rows: list[dict[str, Any]] = [ + { + "measurement_id": digest_model( + { + "run_id": run_id, + "artifact_id": registration.artifact_id, + "name": name, + } + ), + "run_id": run_id, + "artifact_id": registration.artifact_id, + "name": name, + "value_int": value, + "value_float": None, + "unit": unit, + "aggregation": aggregation, + "scope": "process", + "trial_id": None, + "worker_id": None, + "worker_run_index": None, + "value_index": None, + "loop_count": None, + "is_warmup": False, + "block_id": None, + "variant_id": None, + "order_in_block": None, + "phase": None, + "dimensions": {}, + "evidence_level": "observed", + } + for name, value, unit, aggregation in ( + ("cpu.samples", len(samples), "count", "total"), + ("cpu.nodes", len(nodes), "count", "total"), + ) + ] + frame_measurements = [ + { + "run_id": run_id, + "artifact_id": registration.artifact_id, + "frame_id": frame_id, + "metric": metric, + "self_value": values["self"], + "inclusive_value": values["inclusive"], + "unit": "count", + "sample_count": values["samples"], + "thread_name": None, + "process_name": None, + "phase": None, + } + for (metric, frame_id), values in sorted(aggregates.items()) + ] + published = self.publisher.publish_rows( + { + "measurements": measurement_rows, + "frames": list(frame_rows.values()), + "frame_measurements": frame_measurements, + }, + publisher=self.name, + publisher_version=self.version, + input_run_ids=(run_id,), + input_artifact_ids=(registration.artifact_id,), + ) + return V8CpuProfExtractionResult( + run_id=run_id, + artifact_id=registration.artifact_id, + node_count=len(nodes), + sample_count=len(samples), + frame_count=len(frame_rows), + corpus_commit_id=published.commit.commit_id, + limitations=( + "V8 CPU samples represent execution time, not allocation or memory evidence.", + "The CPU profile contains sampled stack locations; source-map resolution " + "is not applied by this extractor.", + ), + ) + + def _aggregate_node( + self, + node: dict[str, Any], + *, + nodes_by_id: dict[int, dict[str, Any]], + frame_rows: dict[str, dict[str, Any]], + aggregates: dict[tuple[str, str], dict[str, int]], + artifact_id: str, + ) -> None: + call_frame = node.get("callFrame") or {} + function = str(call_frame.get("functionName") or "(anonymous)") + url = str(call_frame.get("url") or "") + line = int(call_frame.get("lineNumber") or 0) + column = int(call_frame.get("columnNumber") or 0) + script_id = str(call_frame.get("scriptId") or "") + normalized = self._normalize(url) + frame_id = digest_model( + { + "language": "JavaScript", + "function": function, + "file": normalized, + "line": line, + "column": column, + "script_id": script_id, + } + ) + frame_rows.setdefault( + frame_id, + { + "frame_id": frame_id, + "language": "JavaScript", + "function": function, + "module": None, + "file": normalized, + "line": line, + "column": column, + "address": None, + "build_id": None, + "module_relative_address": None, + "inline_chain_id": None, + "source_state_id": None, + "artifact_id": artifact_id, + "inlined": False, + "symbolization": "complete", + }, + ) + hit_count = int(node.get("hitCount", 0)) + key = ("cpu.hit_count", frame_id) + values = aggregates.setdefault( + key, {"self": 0, "inclusive": 0, "samples": 0} + ) + values["self"] += hit_count + children = node.get("children") or [] + for child_id in children: + child = nodes_by_id.get(int(child_id)) + if child is not None: + self._aggregate_node( + child, + nodes_by_id=nodes_by_id, + frame_rows=frame_rows, + aggregates=aggregates, + artifact_id=artifact_id, + ) + values["inclusive"] += hit_count + values["samples"] += hit_count + + def _normalize(self, url: str) -> str: + if not url: + return "" + if url.startswith("node:") or not Path(url).is_absolute(): + return url + try: + resolved = Path(url).resolve() + return resolved.relative_to(self.workspace.project_root).as_posix() + except (ValueError, OSError): + return url diff --git a/src/flameox/adapters/v8_heap_prof.py b/src/flameox/adapters/v8_heap_prof.py new file mode 100644 index 0000000..fc57a69 --- /dev/null +++ b/src/flameox/adapters/v8_heap_prof.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from flameox.domain import ArtifactKind, DomainError, ErrorCode, digest_model +from flameox.evidence import GenerationPublisher +from flameox.models import ContractModel +from flameox.storage import ArtifactStore, RunStore, Workspace + + +class V8HeapProfExtractionResult(ContractModel): + schema_version: int = 1 + run_id: str + artifact_id: str + sample_count: int + total_sampled_bytes: int + frame_count: int + corpus_commit_id: str + limitations: tuple[str, ...] = () + + +class V8HeapProfExtractor: + name = "node-heap-prof" + version = "1" + + def __init__(self, workspace: Workspace) -> None: + self.workspace = workspace + self.publisher = GenerationPublisher(workspace) + + def extract(self, run_id: str) -> V8HeapProfExtractionResult: + run = RunStore(self.workspace).read(run_id) + registrations = [ + item for item in run.artifacts if item.kind is ArtifactKind.MEMORY_PROFILE + ] + if len(registrations) != 1: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The run must contain exactly one V8 heap profile artifact.", + run_id=run_id, + ) + registration = registrations[0] + artifact = ArtifactStore(self.workspace).get(registration.artifact_id) + try: + payload = json.loads(artifact.payload_path.read_text(encoding="utf-8")) + except (OSError, ValueError, json.JSONDecodeError) as exc: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The artifact is not a supported V8 heap profile (.heapprofile).", + run_id=run_id, + ) from exc + if not isinstance(payload, dict) or "head" not in payload: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The artifact does not contain a V8 sampling heap profile head.", + run_id=run_id, + ) + samples = payload.get("samples") or [] + head = payload.get("head") + if not isinstance(head, dict) or not isinstance(samples, list): + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The V8 heap profile head or samples are malformed.", + run_id=run_id, + ) + frame_rows: dict[str, dict[str, Any]] = {} + aggregates: dict[tuple[str, str], dict[str, int]] = {} + self._aggregate_node( + head, + frame_rows=frame_rows, + aggregates=aggregates, + artifact_id=registration.artifact_id, + ) + total_sampled_bytes = sum(int(sample.get("size", 0)) for sample in samples) + measurement_rows: list[dict[str, Any]] = [ + { + "measurement_id": digest_model( + { + "run_id": run_id, + "artifact_id": registration.artifact_id, + "name": name, + } + ), + "run_id": run_id, + "artifact_id": registration.artifact_id, + "name": name, + "value_int": value, + "value_float": None, + "unit": unit, + "aggregation": aggregation, + "scope": "process", + "trial_id": None, + "worker_id": None, + "worker_run_index": None, + "value_index": None, + "loop_count": None, + "is_warmup": False, + "block_id": None, + "variant_id": None, + "order_in_block": None, + "phase": None, + "dimensions": {}, + "evidence_level": "observed", + } + for name, value, unit, aggregation in ( + ("memory.sampled_bytes", total_sampled_bytes, "bytes", "total"), + ("memory.samples", len(samples), "count", "total"), + ) + ] + frame_measurements = [ + { + "run_id": run_id, + "artifact_id": registration.artifact_id, + "frame_id": frame_id, + "metric": metric, + "self_value": values["self"], + "inclusive_value": values["inclusive"], + "unit": "bytes", + "sample_count": values["samples"], + "thread_name": None, + "process_name": None, + "phase": None, + } + for (metric, frame_id), values in sorted(aggregates.items()) + ] + published = self.publisher.publish_rows( + { + "measurements": measurement_rows, + "frames": list(frame_rows.values()), + "frame_measurements": frame_measurements, + }, + publisher=self.name, + publisher_version=self.version, + input_run_ids=(run_id,), + input_artifact_ids=(registration.artifact_id,), + ) + return V8HeapProfExtractionResult( + run_id=run_id, + artifact_id=registration.artifact_id, + sample_count=len(samples), + total_sampled_bytes=total_sampled_bytes, + frame_count=len(frame_rows), + corpus_commit_id=published.commit.commit_id, + limitations=( + "Sampled allocation bytes are an estimate from V8's sampling heap profiler, " + "not the exact retained heap or process RSS.", + "Only allocations sampled by V8 are reported; small or short-lived " + "allocations may be underrepresented.", + "Source-map resolution is not applied by this extractor; generated JavaScript " + "locations remain distinguishable by their script URL and line/column.", + ), + ) + + def _aggregate_node( + self, + node: dict[str, Any], + *, + frame_rows: dict[str, dict[str, Any]], + aggregates: dict[tuple[str, str], dict[str, int]], + artifact_id: str, + ) -> None: + call_frame = node.get("callFrame") or {} + function = str(call_frame.get("functionName") or "(anonymous)") + url = str(call_frame.get("url") or "") + line = int(call_frame.get("lineNumber") or 0) + column = int(call_frame.get("columnNumber") or 0) + script_id = str(call_frame.get("scriptId") or "") + normalized = self._normalize(url) + frame_id = digest_model( + { + "language": "JavaScript", + "function": function, + "file": normalized, + "line": line, + "column": column, + "script_id": script_id, + } + ) + frame_rows.setdefault( + frame_id, + { + "frame_id": frame_id, + "language": "JavaScript", + "function": function, + "module": None, + "file": normalized, + "line": line, + "column": column, + "address": None, + "build_id": None, + "module_relative_address": None, + "inline_chain_id": None, + "source_state_id": None, + "artifact_id": artifact_id, + "inlined": False, + "symbolization": "complete", + }, + ) + self_size = int(node.get("selfSize", 0)) + key = ("memory.self_size", frame_id) + values = aggregates.setdefault( + key, {"self": 0, "inclusive": 0, "samples": 0} + ) + values["self"] += self_size + values["inclusive"] += self_size + values["samples"] += 1 + for child in (node.get("children") or []): + if isinstance(child, dict): + self._aggregate_node( + child, + frame_rows=frame_rows, + aggregates=aggregates, + artifact_id=artifact_id, + ) + + def _normalize(self, url: str) -> str: + if not url: + return "" + if url.startswith("node:") or not Path(url).is_absolute(): + return url + try: + resolved = Path(url).resolve() + return resolved.relative_to(self.workspace.project_root).as_posix() + except (ValueError, OSError): + return url From c8c24a946687d07823124cdf1ef2743444db9e27 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Fri, 14 Aug 2026 08:27:03 +0800 Subject: [PATCH 3/3] test(adapters): cover V8 capture invocation and profile extraction Add unit tests for the node-cpu-prof and node-heap-prof capture invocations (flag injection, workload argument preservation, rejection of non-Node workloads) and integration tests for the V8CpuProfExtractor and V8HeapProfExtractor (frame measurement publication, malformed profile rejection). --- tests/adapters/test_v8_capture_invocation.py | 128 +++++++++++ tests/adapters/test_v8_profiles.py | 213 +++++++++++++++++++ 2 files changed, 341 insertions(+) create mode 100644 tests/adapters/test_v8_capture_invocation.py create mode 100644 tests/adapters/test_v8_profiles.py diff --git a/tests/adapters/test_v8_capture_invocation.py b/tests/adapters/test_v8_capture_invocation.py new file mode 100644 index 0000000..fbfde8f --- /dev/null +++ b/tests/adapters/test_v8_capture_invocation.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from flameox.adapters.builtins import build_capture_invocation +from flameox.domain import ArtifactKind, DomainError, ErrorCode + +pytestmark = pytest.mark.unit + + +def test_node_cpu_prof_capture_injects_cpu_prof_flags(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-cpu-prof", + ("node", "script.js", "--arg"), + tmp_path, + executable="/usr/bin/node", + ) + + assert invocation.artifact_kinds == (ArtifactKind.SAMPLE_PROFILE,) + assert "--cpu-prof" in invocation.argv + dir_flag = next( + a for a in invocation.argv if a.startswith("--cpu-prof-dir=") + ) + assert dir_flag.startswith("--cpu-prof-dir=") + assert invocation.argv[0] == "node" + assert invocation.argv[-2:] == ("script.js", "--arg") + + +def test_node_cpu_prof_capture_preserves_workload_arguments(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-cpu-prof", + ("node", "index.js", "--flag", "value"), + tmp_path, + executable="/usr/bin/node", + ) + + workload_part = invocation.argv[-3:] + assert workload_part == ("index.js", "--flag", "value") + + +def test_node_cpu_prof_capture_rejects_empty_workload(tmp_path: Path) -> None: + with pytest.raises(DomainError) as failure: + build_capture_invocation( + "node-cpu-prof", + (), + tmp_path, + executable="/usr/bin/node", + ) + + assert failure.value.code is ErrorCode.INVALID_CAPTURE_PLAN + assert "Node.js" in failure.value.message + + +def test_node_cpu_prof_capture_rejects_non_node_workload(tmp_path: Path) -> None: + with pytest.raises(DomainError) as failure: + build_capture_invocation( + "node-cpu-prof", + ("python", "script.py"), + tmp_path, + executable=None, + ) + + assert failure.value.code is ErrorCode.INVALID_CAPTURE_PLAN + + +def test_node_heap_prof_capture_injects_heap_prof_flags(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-heap-prof", + ("node", "script.js"), + tmp_path, + executable="/usr/bin/node", + ) + + assert invocation.artifact_kinds == (ArtifactKind.MEMORY_PROFILE,) + assert "--heap-prof" in invocation.argv + assert invocation.argv[0] == "node" + assert invocation.argv[-1] == "script.js" + + +def test_node_heap_prof_capture_preserves_workload_arguments(tmp_path: Path) -> None: + invocation = build_capture_invocation( + "node-heap-prof", + ("node", "index.js", "--flag", "value"), + tmp_path, + executable="/usr/bin/node", + ) + + workload_part = invocation.argv[-3:] + assert workload_part == ("index.js", "--flag", "value") + + +def test_node_heap_prof_capture_rejects_empty_workload(tmp_path: Path) -> None: + with pytest.raises(DomainError) as failure: + build_capture_invocation( + "node-heap-prof", + (), + tmp_path, + executable="/usr/bin/node", + ) + + assert failure.value.code is ErrorCode.INVALID_CAPTURE_PLAN + assert "Node.js" in failure.value.message + + +def test_node_v8_capture_uses_output_filename(tmp_path: Path) -> None: + invocation_cpu = build_capture_invocation( + "node-cpu-prof", + ("node", "script.js"), + tmp_path, + executable="/usr/bin/node", + ) + invocation_heap = build_capture_invocation( + "node-heap-prof", + ("node", "script.js"), + tmp_path, + executable="/usr/bin/node", + ) + + cpu_name_flag = next( + a for a in invocation_cpu.argv if a.startswith("--cpu-prof-name=") + ) + assert "cpu.cpuprofile" in cpu_name_flag + heap_name_flag = next( + a for a in invocation_heap.argv if a.startswith("--heap-prof-name=") + ) + assert "heap.heapprofile" in heap_name_flag diff --git a/tests/adapters/test_v8_profiles.py b/tests/adapters/test_v8_profiles.py new file mode 100644 index 0000000..2f856e2 --- /dev/null +++ b/tests/adapters/test_v8_profiles.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from flameox.adapters import V8CpuProfExtractor, V8HeapProfExtractor +from flameox.application import ImportArtifactRequest, ImportService +from flameox.catalog import Catalog +from flameox.domain import ArtifactKind, DomainError, ErrorCode +from flameox.storage import Workspace + +pytestmark = [pytest.mark.integration] + + +def _write_cpu_profile(path: Path) -> None: + profile = { + "nodes": [ + { + "id": 0, + "callFrame": { + "functionName": "(root)", + "url": "internal", + "scriptId": "0", + "lineNumber": -1, + "columnNumber": -1, + }, + "hitCount": 0, + "children": [1, 2], + }, + { + "id": 1, + "callFrame": { + "functionName": "main", + "url": "file:///app/index.js", + "scriptId": "1", + "lineNumber": 10, + "columnNumber": 5, + }, + "hitCount": 100, + "children": [3], + }, + { + "id": 2, + "callFrame": { + "functionName": "idle", + "url": "file:///app/index.js", + "scriptId": "1", + "lineNumber": 20, + "columnNumber": 0, + }, + "hitCount": 50, + "children": [], + }, + { + "id": 3, + "callFrame": { + "functionName": "helper", + "url": "file:///app/utils.js", + "scriptId": "2", + "lineNumber": 5, + "columnNumber": 10, + }, + "hitCount": 30, + "children": [], + }, + ], + "samples": [1, 1, 2, 3, 1], + "startTime": 0, + "endTime": 100_000, + } + path.write_text(json.dumps(profile), encoding="utf-8") + + +def _write_heap_profile(path: Path) -> None: + profile = { + "head": { + "callFrame": { + "functionName": "(root)", + "url": "internal", + "scriptId": "0", + "lineNumber": -1, + "columnNumber": -1, + }, + "selfSize": 0, + "id": 0, + "children": [ + { + "callFrame": { + "functionName": "allocate", + "url": "file:///app/alloc.js", + "scriptId": "1", + "lineNumber": 1, + "columnNumber": 0, + }, + "selfSize": 4096, + "id": 1, + "children": [], + }, + { + "callFrame": { + "functionName": "proxy", + "url": "file:///app/proxy.js", + "scriptId": "2", + "lineNumber": 10, + "columnNumber": 5, + }, + "selfSize": 2048, + "id": 2, + "children": [], + }, + ], + }, + "samples": [ + {"size": 4096, "nodeId": 1, "ordinal": 0}, + {"size": 2048, "nodeId": 2, "ordinal": 1}, + ], + } + path.write_text(json.dumps(profile), encoding="utf-8") + + +def test_v8_cpu_prof_extractor_publishes_frame_measurements(tmp_path: Path) -> None: + capture = tmp_path / "cpu.cpuprofile" + _write_cpu_profile(capture) + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=capture, + kind=ArtifactKind.SAMPLE_PROFILE, + ) + ) + result = V8CpuProfExtractor(workspace).extract(imported.run.run_id) + + assert result.node_count == 4 + assert result.sample_count == 5 + assert result.frame_count >= 3 + assert len(result.limitations) >= 2 + with Catalog(workspace).open_snapshot() as snapshot: + frames = snapshot.execute( + "SELECT language, function, file FROM frames ORDER BY function" + ).fetchall() + frame_measurements = snapshot.execute( + "SELECT metric, unit, sum(self_value) FROM frame_measurements GROUP BY metric, unit" + ).fetchall() + frame_tuples = [(f[0], f[1], f[2]) for f in frames] + assert ("JavaScript", "helper", "file:///app/utils.js") in frame_tuples + assert ("JavaScript", "main", "file:///app/index.js") in frame_tuples + assert ("JavaScript", "idle", "file:///app/index.js") in frame_tuples + assert any(row[0] == "cpu.hit_count" for row in frame_measurements) + + +def test_v8_cpu_prof_extractor_rejects_non_cpu_profile(tmp_path: Path) -> None: + bad = tmp_path / "bad.cpuprofile" + bad.write_text('{"not_a_profile": true}', encoding="utf-8") + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=bad, + kind=ArtifactKind.SAMPLE_PROFILE, + ) + ) + with pytest.raises(DomainError) as failure: + V8CpuProfExtractor(workspace).extract(imported.run.run_id) + assert failure.value.code is ErrorCode.ARTIFACT_PARSE_FAILED + + +def test_v8_heap_prof_extractor_publishes_sampled_bytes(tmp_path: Path) -> None: + capture = tmp_path / "heap.heapprofile" + _write_heap_profile(capture) + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=capture, + kind=ArtifactKind.MEMORY_PROFILE, + ) + ) + result = V8HeapProfExtractor(workspace).extract(imported.run.run_id) + + assert result.sample_count == 2 + assert result.total_sampled_bytes == 6144 + assert result.frame_count >= 3 + assert len(result.limitations) >= 3 + with Catalog(workspace).open_snapshot() as snapshot: + frames = snapshot.execute( + "SELECT language, function, file FROM frames ORDER BY function" + ).fetchall() + frame_tuples = [(f[0], f[1], f[2]) for f in frames] + assert ("JavaScript", "allocate", "file:///app/alloc.js") in frame_tuples + assert ("JavaScript", "proxy", "file:///app/proxy.js") in frame_tuples + + +def test_v8_heap_prof_extractor_rejects_non_heap_profile(tmp_path: Path) -> None: + bad = tmp_path / "bad.heapprofile" + bad.write_text('{"no_head": true}', encoding="utf-8") + + workspace = Workspace.initialize(tmp_path) + Catalog(workspace).rebuild() + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=bad, + kind=ArtifactKind.MEMORY_PROFILE, + ) + ) + with pytest.raises(DomainError) as failure: + V8HeapProfExtractor(workspace).extract(imported.run.run_id) + assert failure.value.code is ErrorCode.ARTIFACT_PARSE_FAILED