From a17b987ae14ca060cda5079ea17bd6655b03169a Mon Sep 17 00:00:00 2001 From: Alessandro Cere Date: Fri, 12 Jun 2026 09:53:10 +0800 Subject: [PATCH 1/4] feat: add SystemMetricsMonitor callback for CPU, memory, and network I/O tracking Add an optional callback that monitors system resources during benchmark runs using psutil. Features: - Background thread samples CPU, memory (RSS/VMS), and network I/O - Aggregated stats (avg, p50, p90, p99, max) contributed to Result.stats - Live metrics displayed in the progress bar during runs - Stats persist in stats.json across save/load round-trips - Supports per-process (default) and system-wide monitoring modes - New optional dependency: psutil via `llmeter[system-metrics]` Also adds: - Callback.live_stats() hook for real-time display integration - "System" column in the live stats display for callback metrics - Best practices documentation in Callback base class - User guide documentation (metrics, installation, key concepts) - API reference page - 15 unit tests covering serialization, lifecycle, persistence, and reuse Closes #83 --- docs/reference/callbacks/system_metrics.md | 1 + docs/user_guide/installation.md | 1 + docs/user_guide/key_concepts.md | 22 ++ docs/user_guide/metrics.md | 59 ++++ llmeter/callbacks/__init__.py | 5 + llmeter/callbacks/base.py | 39 +++ llmeter/callbacks/system_metrics.py | 367 ++++++++++++++++++++ llmeter/live_display.py | 22 ++ llmeter/runner.py | 8 + mkdocs.yml | 1 + pyproject.toml | 4 +- tests/unit/callbacks/test_system_metrics.py | 315 +++++++++++++++++ 12 files changed, 843 insertions(+), 1 deletion(-) create mode 100644 docs/reference/callbacks/system_metrics.md create mode 100644 llmeter/callbacks/system_metrics.py create mode 100644 tests/unit/callbacks/test_system_metrics.py diff --git a/docs/reference/callbacks/system_metrics.md b/docs/reference/callbacks/system_metrics.md new file mode 100644 index 0000000..3fec1c8 --- /dev/null +++ b/docs/reference/callbacks/system_metrics.md @@ -0,0 +1 @@ +::: llmeter.callbacks.system_metrics diff --git a/docs/user_guide/installation.md b/docs/user_guide/installation.md index 56f99f5..c3b2f95 100644 --- a/docs/user_guide/installation.md +++ b/docs/user_guide/installation.md @@ -20,6 +20,7 @@ LLMeter also offers optional extra features that require additional dependencies - **openai**: Enable testing any endpoint with an [OpenAI](https://platform.openai.com/)-compatible API, including [Amazon Bedrock Mantle](https://docs.aws.amazon.com/bedrock/latest/userguide/bedrock-mantle.html), and LLM gateways like [LiteLLM](https://github.com/BerriAI/litellm), [Kong](https://konghq.com/products/kong-ai-gateway), and [KrakenD](https://www.krakend.io/) - **litellm**: Enable testing a range of different models through [LiteLLM](https://github.com/BerriAI/litellm) - **mlflow**: Enable logging LLMeter experiments to [MLflow](https://mlflow.org/) +- **system-metrics**: Enable monitoring CPU, memory, and network I/O during benchmark runs You can install one or more of these extra options using pip: diff --git a/docs/user_guide/key_concepts.md b/docs/user_guide/key_concepts.md index 29de110..452a7bf 100644 --- a/docs/user_guide/key_concepts.md +++ b/docs/user_guide/key_concepts.md @@ -27,3 +27,25 @@ The `Runner` is a low-level API to run a batch of (concurrent) requests through An `Experiment` is a high-level, pre-defined analysis to explore a particular aspect of latency or performance - which might run one or more Runs under the hood. LLMeter's pre-built Experiments are designed based on evaluation best-practices and feedback from our users, but you can always build your own custom Experiments from the lower-level Runner API if needed. + +## [Callbacks](../reference/callbacks/index.md): Extensibility hooks + +Callbacks let you extend LLMeter's behavior at defined points in the run lifecycle. A callback can run code before/after each request, or before/after an entire run. Built-in callbacks include: + +- **[`CostModel`](../reference/callbacks/cost/model.md)**: Estimate costs based on token counts, endpoint time, or custom pricing dimensions. +- **[`SystemMetricsMonitor`](../reference/callbacks/system_metrics.md)**: Monitor CPU, memory, and network I/O during runs, with live display integration and persisted statistics. +- **[`MlflowCallback`](../reference/callbacks/mlflow.md)**: Log run parameters and results to MLflow for experiment tracking. + +Callbacks are passed to `Runner` or `Experiment` via the `callbacks` parameter: + +```python +from llmeter.callbacks import CostModel +from llmeter.callbacks.system_metrics import SystemMetricsMonitor + +runner = Runner( + endpoint=endpoint, + callbacks=[SystemMetricsMonitor(), CostModel(...)], +) +``` + +You can also create custom callbacks by subclassing `Callback` — see the [API reference](../reference/callbacks/base.md) for the lifecycle hooks and best practices. diff --git a/docs/user_guide/metrics.md b/docs/user_guide/metrics.md index 7dc45ef..3d4678b 100644 --- a/docs/user_guide/metrics.md +++ b/docs/user_guide/metrics.md @@ -174,6 +174,65 @@ The `CostModel` callback extends `Result.stats` with cost estimates. When attach See the [Model Costs example notebook](https://github.com/awslabs/llmeter/blob/main/examples/Model%20Costs.ipynb) for a walkthrough of request-based pricing, infrastructure-based pricing, and custom cost dimensions. +### System metrics + +The `SystemMetricsMonitor` callback extends `Result.stats` with client-side system resource metrics. When attached to a `Runner` or `Experiment`, it tracks CPU, memory, and network I/O throughout the run. + +!!! note + Requires the `system-metrics` extra: `pip install 'llmeter[system-metrics]'` + +#### Contributed statistics + +| Statistic | Description | +| --- | --- | +| `system_cpu_percent-average` | Mean CPU usage (%) during the run. | +| `system_cpu_percent-p50` | Median CPU usage. | +| `system_cpu_percent-p90` | 90th percentile CPU usage. | +| `system_cpu_percent-p99` | 99th percentile CPU usage. | +| `system_memory_rss_mb-average` | Mean resident memory (MB). | +| `system_memory_rss_mb-max` | Peak resident memory during the run. | +| `system_memory_rss_mb-p50` / `-p90` / `-p99` | Memory percentiles. | +| `system_memory_vms_mb-average` / `-max` | Virtual memory statistics. | +| `system_net_bytes_sent_total` | Total bytes sent over the network during the run. | +| `system_net_bytes_recv_total` | Total bytes received from the network during the run. | +| `system_net_bytes_sent_per_second-average` | Average network send rate (bytes/s). | +| `system_net_bytes_recv_per_second-average` | Average network receive rate (bytes/s). | +| `system_net_bytes_sent_per_second-p50` / `-p90` / `-p99` | Send rate percentiles. | +| `system_net_bytes_recv_per_second-p50` / `-p90` / `-p99` | Receive rate percentiles. | +| `system_samples_collected` | Number of metric samples taken during the run. | + +#### Live display + +When the `SystemMetricsMonitor` is active, real-time system metrics also appear in the progress bar's live stats display under a **System** column, showing the latest CPU %, memory RSS, and network receive rate. + +#### Usage + +```python +from llmeter.callbacks.system_metrics import SystemMetricsMonitor +from llmeter.runner import Runner + +monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=True) +runner = Runner(endpoint=endpoint, callbacks=[monitor], output_path="./results") +result = await runner.run(payload=payload) + +print(result.stats["system_cpu_percent-average"]) +print(result.stats["system_memory_rss_mb-max"]) +print(result.stats["system_net_bytes_recv_total"]) +``` + +Key parameters: + +- **`sample_interval`** (float, default 1.0): How often to sample metrics, in seconds. Lower values give more granular data but slightly increase overhead. +- **`per_process`** (bool, default `True`): If `True`, monitors only the current Python process. Set to `False` for system-wide resource tracking (useful when other processes contribute to the workload). + +!!! note "Network I/O is always system-wide" + Regardless of the `per_process` setting, network I/O metrics are always system-wide because `psutil` does not support per-process network counters on most platforms. If other processes generate significant network traffic during a benchmark, it will be reflected in the network stats. + +!!! tip "Detecting client-side bottlenecks" + If `system_cpu_percent-p90` approaches 100% or `system_memory_rss_mb-max` grows significantly during high-concurrency load tests, your client machine may be a bottleneck rather than the LLM endpoint. Consider running LLMeter on a more powerful instance or reducing concurrency. + +See the [System Metrics Monitoring example notebook](https://github.com/awslabs/llmeter/blob/main/examples/System%20Metrics%20Monitoring.ipynb) for a full walkthrough including load test correlation plots and time-series analysis. + ### Experiment-level metrics #### Load test diff --git a/llmeter/callbacks/__init__.py b/llmeter/callbacks/__init__.py index 42db055..6a5fb48 100644 --- a/llmeter/callbacks/__init__.py +++ b/llmeter/callbacks/__init__.py @@ -8,6 +8,7 @@ - Logging run details to MLFlow, with `MlflowCallback` - Calculating cost estimates, with `CostModel` +- Monitoring system resources (CPU, memory, network), with `SystemMetricsMonitor` For creating your own custom callbacks, see the `Callback` base class. """ @@ -20,3 +21,7 @@ spec = importlib.util.find_spec("mlflow") if spec: from .mlflow import MlflowCallback # noqa: F401 + +spec = importlib.util.find_spec("psutil") +if spec: + from .system_metrics import SystemMetricsMonitor # noqa: F401 diff --git a/llmeter/callbacks/base.py b/llmeter/callbacks/base.py index f4c4dba..27bd5bb 100644 --- a/llmeter/callbacks/base.py +++ b/llmeter/callbacks/base.py @@ -24,6 +24,31 @@ class Callback(ABC): A Callback object may implement multiple of the defined lifecycle hooks (such as `before_invoke`, `after_run`, etc). Callbacks must support serializing their configuration to a file (by implementing `save_to_file`), and loading back (via `load_from_file`). + + Best practices for implementing callbacks: + + - **Serializability**: The Runner calls ``dataclasses.asdict()`` on its run configuration + (which includes the callbacks list) when saving ``run_config.json``. This internally performs + a ``copy.deepcopy()`` on all field values. If your callback uses threading primitives (locks, + events, threads) or other non-picklable objects, you **must** implement ``__getstate__`` and + ``__setstate__`` to exclude them from serialization and reinitialize them on restore. Example:: + + def __getstate__(self): + return {"my_config": self.my_config} + + def __setstate__(self, state): + self.my_config = state["my_config"] + self._lock = threading.Lock() # Reinitialize non-picklable state + + - **Reusability**: A callback instance may be reused across multiple ``Runner.run()`` calls. + Reset any accumulated state in ``before_run()`` so each run starts fresh. + + - **Contributing stats**: Use ``result._update_contributed_stats(dict)`` in ``after_run()`` to + add custom metrics to ``result.stats``. These will be persisted in ``stats.json`` and survive + save/load round-trips automatically. + + - **Thread safety**: If your callback spawns background threads (e.g. for monitoring), use + daemon threads and ensure they are joined in ``after_run()`` to avoid resource leaks. """ async def before_invoke(self, payload: dict) -> None: @@ -71,6 +96,20 @@ async def after_run(self, result: Result) -> None: """ pass + def live_stats(self) -> dict[str, float | int] | None: + """Optional hook returning live metrics for the progress display. + + If implemented, the Runner will call this method on each display refresh cycle + and merge the returned values into the live stats table. This allows callbacks + to surface real-time information (e.g. CPU usage, memory) during a run. + + Returns: + A dict of ``{display_key: numeric_value}`` to show in the live display, + or ``None``/empty dict if nothing to show. Keys should use short, descriptive + names (they appear as-is in the progress table). + """ + return None + def save_to_file(self, path: WritablePathLike) -> None: """Save this Callback to file diff --git a/llmeter/callbacks/system_metrics.py b/llmeter/callbacks/system_metrics.py new file mode 100644 index 0000000..c918911 --- /dev/null +++ b/llmeter/callbacks/system_metrics.py @@ -0,0 +1,367 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""System resource monitoring callback for LLMeter runs.""" + +import logging +import threading +import time +from dataclasses import dataclass + +from upath.types import ReadablePathLike, WritablePathLike + +from ..results import Result +from ..runner import _RunConfig +from ..utils import DeferredError, summary_stats_from_list +from .base import Callback + +logger = logging.getLogger(__name__) + +try: + import psutil +except ImportError as e: + logger.debug( + "psutil not available. System metrics monitoring requires psutil. " + "Install with: pip install 'llmeter[system-metrics]'" + ) + psutil = DeferredError(e) + + +@dataclass +class _Sample: + """A single system metrics sample.""" + + timestamp: float + cpu_percent: float + memory_rss_mb: float + memory_vms_mb: float + net_bytes_sent: int + net_bytes_recv: int + + +@dataclass +class SystemMetricsMonitor(Callback): + """Monitor system resources (CPU, memory, network I/O) during a benchmark run. + + This callback spawns a background thread that periodically samples system metrics + while the benchmark is running. After the run completes, aggregated statistics are + contributed to the Result. + + Args: + sample_interval: Seconds between samples. Default 1.0. + per_process: If True, monitor only the current process. If False, monitor + system-wide metrics. Default True. + + Note: + When ``per_process=True``, CPU and memory metrics are scoped to the current Python + process. However, **network I/O is always system-wide** because ``psutil`` does not + support per-process network counters on most platforms. If other processes on the + machine are generating significant network traffic during a benchmark, that activity + will be reflected in the network stats regardless of this setting. + + Contributed stats keys (prefixed with ``system_``): + + - ``system_cpu_percent-average``, ``-p50``, ``-p90``, ``-p99``: CPU usage percentage + - ``system_memory_rss_mb-average``, ``-p50``, ``-p90``, ``-p99``, ``-max``: Resident memory (MB) + - ``system_memory_vms_mb-average``, ``-p50``, ``-p90``, ``-p99``, ``-max``: Virtual memory (MB) + - ``system_net_bytes_sent_total``: Total bytes sent during the run + - ``system_net_bytes_recv_total``: Total bytes received during the run + - ``system_net_bytes_sent_per_second-average``: Average send rate (bytes/s) + - ``system_net_bytes_recv_per_second-average``: Average receive rate (bytes/s) + - ``system_samples_collected``: Number of samples taken + + Example:: + + from llmeter.callbacks.system_metrics import SystemMetricsMonitor + from llmeter.runner import Runner + + monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=True) + runner = Runner(endpoint=endpoint, callbacks=[monitor]) + result = await runner.run(payload=payload) + + print(f"CPU avg: {result.stats['system_cpu_percent-average']:.1f}%") + print(f"Memory peak: {result.stats['system_memory_rss_mb-max']:.1f} MB") + print(f"Network sent: {result.stats['system_net_bytes_sent_total']} bytes") + """ + + sample_interval: float = 1.0 + per_process: bool = True + + def __post_init__(self): + self._samples: list[_Sample] = [] + self._thread: threading.Thread | None = None + self._stop_event: threading.Event = threading.Event() + self._process: object = None + self._net_start: tuple[int, int] = (0, 0) + + def __getstate__(self): + """Support pickling/deepcopy by excluding non-serializable thread state.""" + return { + "sample_interval": self.sample_interval, + "per_process": self.per_process, + "_samples": self._samples, + "_net_start": self._net_start, + } + + def __setstate__(self, state): + """Restore from pickle/deepcopy, reinitializing thread primitives.""" + self.sample_interval = state["sample_interval"] + self.per_process = state["per_process"] + self._samples = state["_samples"] + self._net_start = state["_net_start"] + self._thread = None + self._stop_event = threading.Event() + self._process = None + + def _collect_sample(self) -> _Sample: + """Collect a single metrics sample.""" + now = time.perf_counter() + + if self.per_process: + cpu = self._process.cpu_percent(interval=None) + mem_info = self._process.memory_info() + rss_mb = mem_info.rss / (1024 * 1024) + vms_mb = mem_info.vms / (1024 * 1024) + else: + cpu = psutil.cpu_percent(interval=None) + mem = psutil.virtual_memory() + rss_mb = mem.used / (1024 * 1024) + vms_mb = mem.total / (1024 * 1024) + + net = psutil.net_io_counters() + return _Sample( + timestamp=now, + cpu_percent=cpu, + memory_rss_mb=rss_mb, + memory_vms_mb=vms_mb, + net_bytes_sent=net.bytes_sent, + net_bytes_recv=net.bytes_recv, + ) + + def _sampling_loop(self): + """Background thread loop that collects samples at the configured interval.""" + # Initial CPU reading to prime the counter (psutil requires two calls for %) + if self.per_process: + self._process.cpu_percent(interval=None) + else: + psutil.cpu_percent(interval=None) + + while not self._stop_event.is_set(): + try: + sample = self._collect_sample() + self._samples.append(sample) + except Exception as e: + logger.debug(f"System metrics sampling error: {e}") + self._stop_event.wait(self.sample_interval) + + async def before_run(self, run_config: _RunConfig) -> None: + """Start the background sampling thread.""" + # Reset state from any previous run + self._samples = [] + self._stop_event.clear() + + # Initialize process handle + if self.per_process: + self._process = psutil.Process() + + # Record starting network counters + net = psutil.net_io_counters() + self._net_start = (net.bytes_sent, net.bytes_recv) + + # Start sampling thread + self._thread = threading.Thread( + target=self._sampling_loop, + name="llmeter-system-metrics", + daemon=True, + ) + self._thread.start() + logger.debug( + "System metrics monitoring started (interval=%.2fs, per_process=%s)", + self.sample_interval, + self.per_process, + ) + + async def after_run(self, result: Result) -> None: + """Stop sampling and contribute aggregated stats to the result.""" + # Stop the background thread + self._stop_event.set() + if self._thread is not None: + self._thread.join(timeout=5.0) + self._thread = None + + if not self._samples: + logger.warning("No system metrics samples were collected during the run.") + return + + # Aggregate metrics + stats = self._compute_stats() + + # Contribute to result + result._update_contributed_stats(stats) + + # Store raw samples on the result for potential detailed analysis + if not hasattr(result, "_system_metrics_samples"): + result._system_metrics_samples = self._samples + + logger.info( + "System metrics: %d samples collected. CPU avg=%.1f%%, " + "Memory RSS peak=%.1f MB, Net sent=%d bytes, Net recv=%d bytes", + len(self._samples), + stats.get("system_cpu_percent-average", 0), + stats.get("system_memory_rss_mb-max", 0), + stats.get("system_net_bytes_sent_total", 0), + stats.get("system_net_bytes_recv_total", 0), + ) + + def _compute_stats(self) -> dict[str, float | int]: + """Compute aggregated statistics from collected samples.""" + stats: dict[str, float | int] = {} + + if not self._samples: + return stats + + # CPU percent + cpu_values = [s.cpu_percent for s in self._samples] + cpu_stats = summary_stats_from_list(cpu_values) + for agg, value in cpu_stats.items(): + stats[f"system_cpu_percent-{agg}"] = value + + # Memory RSS (MB) + rss_values = [s.memory_rss_mb for s in self._samples] + rss_stats = summary_stats_from_list(rss_values) + for agg, value in rss_stats.items(): + stats[f"system_memory_rss_mb-{agg}"] = value + stats["system_memory_rss_mb-max"] = max(rss_values) + + # Memory VMS (MB) + vms_values = [s.memory_vms_mb for s in self._samples] + vms_stats = summary_stats_from_list(vms_values) + for agg, value in vms_stats.items(): + stats[f"system_memory_vms_mb-{agg}"] = value + stats["system_memory_vms_mb-max"] = max(vms_values) + + # Network I/O totals (delta from start to last sample) + last_sample = self._samples[-1] + stats["system_net_bytes_sent_total"] = ( + last_sample.net_bytes_sent - self._net_start[0] + ) + stats["system_net_bytes_recv_total"] = ( + last_sample.net_bytes_recv - self._net_start[1] + ) + + # Network rates (bytes per second) + if len(self._samples) >= 2: + duration = self._samples[-1].timestamp - self._samples[0].timestamp + if duration > 0: + net_sent_delta = ( + self._samples[-1].net_bytes_sent - self._samples[0].net_bytes_sent + ) + net_recv_delta = ( + self._samples[-1].net_bytes_recv - self._samples[0].net_bytes_recv + ) + stats["system_net_bytes_sent_per_second-average"] = ( + net_sent_delta / duration + ) + stats["system_net_bytes_recv_per_second-average"] = ( + net_recv_delta / duration + ) + + # Per-interval network rates for percentile calculations + if len(self._samples) >= 2: + send_rates = [] + recv_rates = [] + for i in range(1, len(self._samples)): + dt = self._samples[i].timestamp - self._samples[i - 1].timestamp + if dt > 0: + send_rates.append( + ( + self._samples[i].net_bytes_sent + - self._samples[i - 1].net_bytes_sent + ) + / dt + ) + recv_rates.append( + ( + self._samples[i].net_bytes_recv + - self._samples[i - 1].net_bytes_recv + ) + / dt + ) + if send_rates: + send_rate_stats = summary_stats_from_list(send_rates) + for agg, value in send_rate_stats.items(): + stats[f"system_net_bytes_sent_per_second-{agg}"] = value + if recv_rates: + recv_rate_stats = summary_stats_from_list(recv_rates) + for agg, value in recv_rate_stats.items(): + stats[f"system_net_bytes_recv_per_second-{agg}"] = value + + stats["system_samples_collected"] = len(self._samples) + + return stats + + @property + def samples(self) -> list[_Sample]: + """Access the raw collected samples for custom analysis.""" + return self._samples + + def live_stats(self) -> dict[str, float | int]: + """Return the latest system metrics for the live progress display. + + This method is called by the Runner's refresh loop to include system + metrics in the live stats table during a run. It returns the most recent + sample's values (not aggregates), providing a real-time view. + + Returns: + A dict with the latest CPU, memory, and network rate values, or + an empty dict if no samples have been collected yet. + """ + if not self._samples: + return {} + + latest = self._samples[-1] + stats: dict[str, float | int] = { + "system_cpu_percent": latest.cpu_percent, + "system_rss_mb": round(latest.memory_rss_mb, 1), + } + + # Include network rate if we have at least 2 samples + if len(self._samples) >= 2: + prev = self._samples[-2] + dt = latest.timestamp - prev.timestamp + if dt > 0: + stats["system_net_recv_kbps"] = round( + (latest.net_bytes_recv - prev.net_bytes_recv) / dt / 1024, 1 + ) + + return stats + + def save_to_file(self, path: WritablePathLike) -> None: + """Save this SystemMetricsMonitor configuration to file.""" + import json + + from ..utils import ensure_path + + out_path = ensure_path(path) + config = { + "type": "SystemMetricsMonitor", + "sample_interval": self.sample_interval, + "per_process": self.per_process, + } + with out_path.open("w") as f: + json.dump(config, f, indent=2) + + @classmethod + def _load_from_file(cls, path: ReadablePathLike) -> "SystemMetricsMonitor": + """Load a SystemMetricsMonitor from file.""" + import json + + from ..utils import ensure_path + + in_path = ensure_path(path) + with in_path.open("r") as f: + config = json.load(f) + + return cls( + sample_interval=config.get("sample_interval", 1.0), + per_process=config.get("per_process", True), + ) diff --git a/llmeter/live_display.py b/llmeter/live_display.py index dfcb9a1..3d4db8c 100644 --- a/llmeter/live_display.py +++ b/llmeter/live_display.py @@ -25,6 +25,7 @@ ("TTLT", ("ttlt",)), ("Tokens", ("token",)), ("Errors", ("fail",)), + ("System", ("system_",)), ("Other", ("",)), ) @@ -202,6 +203,10 @@ def format_stats( alias renaming and formatting, and returns an ordered dict of ``{display_label: formatted_value}`` strings. + Additionally, any keys in *raw* not covered by ``self._display_stats`` + that are contributed by callbacks (via ``Callback.live_stats()``) are + appended with automatic formatting. + Args: raw: Flat dictionary of raw numeric stats, as returned by ``RunningStats.to_stats()``. @@ -214,6 +219,7 @@ def format_stats( return {label: "—" for label in self._display_stats} info: dict[str, str] = {} + mapped_keys: set[str] = set() for label, spec in self._display_stats.items(): if isinstance(spec, tuple): key, modifier = spec[0], spec[1] @@ -222,6 +228,7 @@ def format_stats( key = spec invert = False + mapped_keys.add(key) val = raw.get(key) if val is None: info[label] = "—" @@ -232,6 +239,21 @@ def format_stats( except (TypeError, ValueError): info[label] = str(val) + # Include extra callback-contributed stats not covered by _display_stats + for key, val in raw.items(): + if key in mapped_keys or val is None: + continue + # Only include keys that look like live callback stats (numeric, not + # part of the standard RunningStats output) + if not isinstance(val, (int, float)): + continue + # Use the key itself as the label, skipping standard internal keys + if key.startswith("system_") or key.startswith("cb_"): + try: + info[key] = _format_stat(key, float(val)) + except (TypeError, ValueError): + info[key] = str(val) + return info def update( diff --git a/llmeter/runner.py b/llmeter/runner.py index 7101b5f..063e98d 100644 --- a/llmeter/runner.py +++ b/llmeter/runner.py @@ -367,6 +367,14 @@ def _refresh_stats_display(self, *, force: bool = False) -> None: self._last_stats_update = now raw = self._running_stats.to_stats(end_time=now_utc()) if raw: + # Merge live stats from callbacks that provide them + if self.callbacks: + for cb in self.callbacks: + live = getattr(cb, "live_stats", None) + if callable(live): + cb_stats = live() + if cb_stats: + raw.update(cb_stats) prefix = f"reqs={self._running_stats._count}" if self._time_bound else "" self._stats_display.update(raw, extra_prefix=prefix) diff --git a/mkdocs.yml b/mkdocs.yml index 27fb90b..f5793e6 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -57,6 +57,7 @@ nav: - results: reference/callbacks/cost/results.md - serde: reference/callbacks/cost/serde.md - mlflow: reference/callbacks/mlflow.md + - system_metrics: reference/callbacks/system_metrics.md - endpoints: - reference/endpoints/index.md - anthropic_messages: reference/endpoints/anthropic_messages.md diff --git a/pyproject.toml b/pyproject.toml index 886929b..71d588c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,9 +24,10 @@ litellm = ["litellm>=1.47.1"] plotting = ["plotly>=5.24.1", "kaleido<=0.2.1", "pandas>=2.2.0"] mlflow = ["mlflow-skinny>=3.10.0"] multimodal = ["puremagic>=1.28"] +system-metrics = ["psutil>=5.9.0"] tokenizers = ["transformers>=4.40.2", "tiktoken>=0.7.0"] all = [ - "llmeter[anthropic,openai,litellm,plotting,mlflow,multimodal,tokenizers]", + "llmeter[anthropic,openai,litellm,plotting,mlflow,multimodal,system-metrics,tokenizers]", ] [project.urls] @@ -46,6 +47,7 @@ dev = [ "nbformat>=5.10.4", "bandit>=1.7.10", "aws-bedrock-token-generator>=1.1.0", + "nbconvert>=7.17.1", ] docs = [ "mkdocstrings[python]>=0.26.1", diff --git a/tests/unit/callbacks/test_system_metrics.py b/tests/unit/callbacks/test_system_metrics.py new file mode 100644 index 0000000..9530982 --- /dev/null +++ b/tests/unit/callbacks/test_system_metrics.py @@ -0,0 +1,315 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +import asyncio +import copy +import json +import pickle +import tempfile +from dataclasses import asdict +from pathlib import Path + +import pytest + +from llmeter.callbacks.system_metrics import SystemMetricsMonitor, _Sample +from llmeter.results import Result + + +class TestSystemMetricsMonitorPickle: + """Ensure SystemMetricsMonitor survives pickle, deepcopy, and dataclasses.asdict. + + The Runner calls dataclasses.asdict() on its _RunConfig (which includes callbacks) + when saving run_config.json. This internally does a deepcopy of all field values. + Callbacks with threading primitives (locks, events, threads) must handle this + gracefully. + """ + + def test_deepcopy(self): + """Callbacks must survive copy.deepcopy (used by dataclasses.asdict).""" + monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=True) + monitor_copy = copy.deepcopy(monitor) + + assert monitor_copy.sample_interval == 0.5 + assert monitor_copy.per_process is True + assert monitor_copy._stop_event is not monitor._stop_event + + def test_pickle_roundtrip(self): + """Callbacks must survive pickle serialization.""" + monitor = SystemMetricsMonitor(sample_interval=2.0, per_process=False) + data = pickle.dumps(monitor) + restored = pickle.loads(data) + + assert restored.sample_interval == 2.0 + assert restored.per_process is False + assert hasattr(restored, "_stop_event") + assert hasattr(restored, "_thread") + + def test_asdict(self): + """dataclasses.asdict must not raise on SystemMetricsMonitor.""" + monitor = SystemMetricsMonitor(sample_interval=1.0, per_process=True) + d = asdict(monitor) + + assert d == {"sample_interval": 1.0, "per_process": True} + + def test_deepcopy_preserves_samples(self): + """Collected samples should survive deepcopy.""" + monitor = SystemMetricsMonitor(sample_interval=1.0) + monitor._samples = [ + _Sample( + timestamp=1.0, + cpu_percent=25.0, + memory_rss_mb=100.0, + memory_vms_mb=200.0, + net_bytes_sent=1000, + net_bytes_recv=2000, + ) + ] + monitor_copy = copy.deepcopy(monitor) + assert len(monitor_copy._samples) == 1 + assert monitor_copy._samples[0].cpu_percent == 25.0 + + +class TestSystemMetricsMonitorLifecycle: + """Test the before_run / after_run lifecycle.""" + + @pytest.fixture + def monitor(self): + return SystemMetricsMonitor(sample_interval=0.1, per_process=True) + + def test_before_run_starts_thread(self, monitor): + """before_run should start a background sampling thread.""" + + async def _test(): + class FakeConfig: + pass + + await monitor.before_run(FakeConfig()) + assert monitor._thread is not None + assert monitor._thread.is_alive() + + # Cleanup + monitor._stop_event.set() + monitor._thread.join(timeout=2) + + asyncio.run(_test()) + + def test_after_run_stops_thread(self, monitor): + """after_run should stop the sampling thread.""" + + async def _test(): + class FakeConfig: + pass + + await monitor.before_run(FakeConfig()) + assert monitor._thread.is_alive() + + result = Result(responses=[], total_requests=0) + await monitor.after_run(result) + + assert monitor._thread is None or not monitor._thread.is_alive() + + asyncio.run(_test()) + + def test_collects_samples(self, monitor): + """The monitor should collect samples during the run.""" + + async def _test(): + class FakeConfig: + pass + + await monitor.before_run(FakeConfig()) + await asyncio.sleep(0.5) + + result = Result(responses=[], total_requests=0) + await monitor.after_run(result) + + assert len(monitor._samples) >= 3 + + asyncio.run(_test()) + + def test_contributes_stats_to_result(self, monitor): + """after_run should populate system stats on the result.""" + + async def _test(): + class FakeConfig: + pass + + await monitor.before_run(FakeConfig()) + await asyncio.sleep(0.5) + + result = Result(responses=[], total_requests=0) + await monitor.after_run(result) + + stats = result.stats + assert "system_cpu_percent-average" in stats + assert "system_cpu_percent-p50" in stats + assert "system_cpu_percent-p90" in stats + assert "system_cpu_percent-p99" in stats + assert "system_memory_rss_mb-average" in stats + assert "system_memory_rss_mb-max" in stats + assert "system_memory_vms_mb-average" in stats + assert "system_memory_vms_mb-max" in stats + assert "system_net_bytes_sent_total" in stats + assert "system_net_bytes_recv_total" in stats + assert "system_samples_collected" in stats + assert stats["system_samples_collected"] >= 3 + assert stats["system_memory_rss_mb-max"] > 0 + + asyncio.run(_test()) + + def test_no_samples_emits_warning(self, monitor, caplog): + """after_run with no samples should warn and not crash.""" + + async def _test(): + # Don't call before_run, so no sampling thread was started + result = Result(responses=[], total_requests=0) + await monitor.after_run(result) + + # Should not contribute any stats + system_keys = [k for k in result.stats if k.startswith("system_")] + assert len(system_keys) == 0 + + asyncio.run(_test()) + + +class TestSystemMetricsMonitorPersistence: + """Test that system metrics survive save/load round-trips.""" + + def test_stats_persist_in_result_save_load(self): + """System metrics should be preserved in stats.json across save/load.""" + + async def _test(): + monitor = SystemMetricsMonitor(sample_interval=0.1, per_process=True) + + class FakeConfig: + pass + + await monitor.before_run(FakeConfig()) + await asyncio.sleep(0.4) + + result = Result(responses=[], total_requests=0) + await monitor.after_run(result) + + with tempfile.TemporaryDirectory() as tmpdir: + result.output_path = tmpdir + result.save() + + # Load with responses + loaded = Result.load(tmpdir, load_responses=True) + assert "system_cpu_percent-average" in loaded.stats + assert "system_memory_rss_mb-max" in loaded.stats + assert "system_net_bytes_sent_total" in loaded.stats + + # Load without responses (stats-only path) + loaded_no_resp = Result.load(tmpdir, load_responses=False) + assert "system_cpu_percent-average" in loaded_no_resp.stats + assert "system_memory_rss_mb-max" in loaded_no_resp.stats + + asyncio.run(_test()) + + def test_save_to_file_and_load(self): + """Test callback configuration persistence.""" + monitor = SystemMetricsMonitor(sample_interval=2.5, per_process=False) + + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "monitor.json" + monitor.save_to_file(path) + + # Verify file contents + with open(path) as f: + config = json.load(f) + assert config["type"] == "SystemMetricsMonitor" + assert config["sample_interval"] == 2.5 + assert config["per_process"] is False + + # Load back + restored = SystemMetricsMonitor._load_from_file(path) + assert restored.sample_interval == 2.5 + assert restored.per_process is False + + +class TestSystemMetricsMonitorReuse: + """Test that a monitor can be reused across multiple runs.""" + + def test_reuse_across_runs(self): + """A monitor should reset state between runs.""" + + async def _test(): + monitor = SystemMetricsMonitor(sample_interval=0.1, per_process=True) + + class FakeConfig: + pass + + # First run + await monitor.before_run(FakeConfig()) + await asyncio.sleep(0.3) + result1 = Result(responses=[], total_requests=0) + await monitor.after_run(result1) + samples_run1 = result1.stats["system_samples_collected"] + + # Second run — should reset + await monitor.before_run(FakeConfig()) + await asyncio.sleep(0.3) + result2 = Result(responses=[], total_requests=0) + await monitor.after_run(result2) + samples_run2 = result2.stats["system_samples_collected"] + + # Both runs should have collected independent samples + assert samples_run1 >= 2 + assert samples_run2 >= 2 + + asyncio.run(_test()) + + +class TestSystemMetricsMonitorLiveStats: + """Test the live_stats() method for progress display integration.""" + + def test_live_stats_empty_before_run(self): + """live_stats() returns empty dict when no samples collected.""" + monitor = SystemMetricsMonitor(sample_interval=0.5) + assert monitor.live_stats() == {} + + def test_live_stats_returns_latest_sample(self): + """live_stats() returns current CPU and memory from latest sample.""" + + async def _test(): + monitor = SystemMetricsMonitor(sample_interval=0.1, per_process=True) + + class FakeConfig: + pass + + await monitor.before_run(FakeConfig()) + await asyncio.sleep(0.4) + + stats = monitor.live_stats() + assert "system_cpu_percent" in stats + assert "system_rss_mb" in stats + assert stats["system_rss_mb"] > 0 + + # Cleanup + monitor._stop_event.set() + monitor._thread.join(timeout=2) + + asyncio.run(_test()) + + def test_live_stats_includes_network_rate(self): + """live_stats() includes network recv rate when 2+ samples exist.""" + + async def _test(): + monitor = SystemMetricsMonitor(sample_interval=0.1, per_process=True) + + class FakeConfig: + pass + + await monitor.before_run(FakeConfig()) + await asyncio.sleep(0.4) + + stats = monitor.live_stats() + # With multiple samples, network rate should be present + assert "system_net_recv_kbps" in stats + + # Cleanup + monitor._stop_event.set() + monitor._thread.join(timeout=2) + + asyncio.run(_test()) From 4551389e7c0217f1d39304f2b98643c239c16f70 Mon Sep 17 00:00:00 2001 From: Alessandro Cere Date: Tue, 30 Jun 2026 10:53:06 +0800 Subject: [PATCH 2/4] fix(system-metrics): remove monkey-patching and redundant net rate calc - Remove _system_metrics_samples monkey-patch on Result (raw samples are already accessible via monitor.samples property) - Remove redundant first-to-last net rate average that was overwritten by the per-interval stats computation --- llmeter/callbacks/system_metrics.py | 23 +---------------------- 1 file changed, 1 insertion(+), 22 deletions(-) diff --git a/llmeter/callbacks/system_metrics.py b/llmeter/callbacks/system_metrics.py index c918911..726e8a0 100644 --- a/llmeter/callbacks/system_metrics.py +++ b/llmeter/callbacks/system_metrics.py @@ -198,10 +198,6 @@ async def after_run(self, result: Result) -> None: # Contribute to result result._update_contributed_stats(stats) - # Store raw samples on the result for potential detailed analysis - if not hasattr(result, "_system_metrics_samples"): - result._system_metrics_samples = self._samples - logger.info( "System metrics: %d samples collected. CPU avg=%.1f%%, " "Memory RSS peak=%.1f MB, Net sent=%d bytes, Net recv=%d bytes", @@ -248,24 +244,7 @@ def _compute_stats(self) -> dict[str, float | int]: last_sample.net_bytes_recv - self._net_start[1] ) - # Network rates (bytes per second) - if len(self._samples) >= 2: - duration = self._samples[-1].timestamp - self._samples[0].timestamp - if duration > 0: - net_sent_delta = ( - self._samples[-1].net_bytes_sent - self._samples[0].net_bytes_sent - ) - net_recv_delta = ( - self._samples[-1].net_bytes_recv - self._samples[0].net_bytes_recv - ) - stats["system_net_bytes_sent_per_second-average"] = ( - net_sent_delta / duration - ) - stats["system_net_bytes_recv_per_second-average"] = ( - net_recv_delta / duration - ) - - # Per-interval network rates for percentile calculations + # Per-interval network rates (average, p50, p90, p99) if len(self._samples) >= 2: send_rates = [] recv_rates = [] From 0b584736afd69ba10c40bb239d3b27fae6142a99 Mon Sep 17 00:00:00 2001 From: Alessandro Cere Date: Fri, 3 Jul 2026 08:26:34 +0800 Subject: [PATCH 3/4] docs: add example notebook, integration tests, and user guide for SystemMetricsMonitor - Add example notebook demonstrating system metrics monitoring with time-series plots, concurrency correlation, and bottleneck detection - Add integration tests validating full lifecycle against Bedrock: basic usage, save/load persistence, reuse, system-wide mode, and higher concurrency scenarios - Add user guide page for the system metrics callback - Update mkdocs nav and callbacks index to include new docs --- docs/user_guide/callbacks/index.md | 1 + docs/user_guide/callbacks/system_metrics.md | 76 ++++ examples/System Metrics Monitoring.ipynb | 412 +++++++++++++++++++ mkdocs.yml | 1 + tests/integ/callbacks/__init__.py | 2 + tests/integ/callbacks/test_system_metrics.py | 268 ++++++++++++ 6 files changed, 760 insertions(+) create mode 100644 docs/user_guide/callbacks/system_metrics.md create mode 100644 examples/System Metrics Monitoring.ipynb create mode 100644 tests/integ/callbacks/__init__.py create mode 100644 tests/integ/callbacks/test_system_metrics.py diff --git a/docs/user_guide/callbacks/index.md b/docs/user_guide/callbacks/index.md index c1af74c..492fbff 100644 --- a/docs/user_guide/callbacks/index.md +++ b/docs/user_guide/callbacks/index.md @@ -6,6 +6,7 @@ See the dedicated sections in this guide on how [built-in callbacks](../../refer - [Avoid prompt caching](cache_buster.md) on your target Endpoint - [Model and compare the costs](cost.md) of different Endpoints +- [Monitor system resources](system_metrics.md) (CPU, memory, network I/O) during runs - [Track your experiments](mlflow.md) with MLflow diff --git a/docs/user_guide/callbacks/system_metrics.md b/docs/user_guide/callbacks/system_metrics.md new file mode 100644 index 0000000..0f0d65d --- /dev/null +++ b/docs/user_guide/callbacks/system_metrics.md @@ -0,0 +1,76 @@ +# Monitor System Resources + +When running latency benchmarks — especially load tests with high concurrency — it's important to know whether the client machine itself is a bottleneck. High CPU usage, memory pressure, or network saturation on the client side can skew latency measurements without any visible errors. + +LLMeter's [`SystemMetricsMonitor`](../../reference/callbacks/system_metrics/#llmeter.callbacks.system_metrics.SystemMetricsMonitor) callback tracks CPU, memory, and network I/O during benchmark runs, providing both real-time visibility through the progress display and persisted statistics in the result. + +!!! note "Optional dependency" + System metrics monitoring requires `psutil`. Install with: + ```bash + pip install 'llmeter[system-metrics]' + ``` + +## Quick start + +```python +from llmeter.callbacks.system_metrics import SystemMetricsMonitor +from llmeter.runner import Runner + +monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=True) +runner = Runner(endpoint=endpoint, callbacks=[monitor], output_path="./results") +result = await runner.run(payload=payload, clients=10, n_requests=50) + +# Aggregated stats are available on the result +print(f"CPU avg: {result.stats['system_cpu_percent-average']:.1f}%") +print(f"Memory peak: {result.stats['system_memory_rss_mb-max']:.1f} MB") +print(f"Network received: {result.stats['system_net_bytes_recv_total']} bytes") +``` + +## How it works + +The monitor spawns a lightweight background thread that periodically samples system metrics while the benchmark runs: + +1. **`before_run`** — resets state, captures the starting network counters, and starts the sampling thread. +2. **During the run** — the thread collects CPU, memory, and network samples at the configured `sample_interval`. +3. **`after_run`** — stops the thread and computes aggregated statistics (average, p50, p90, p99, max) that are contributed to `result.stats`. + +Because sampling happens on a separate thread, the overhead on benchmark throughput is negligible. + +## Configuration + +| Parameter | Default | Description | +| --- | --- | --- | +| `sample_interval` | `1.0` | Seconds between samples. Lower values yield more granular data. | +| `per_process` | `True` | If `True`, CPU and memory are scoped to the current Python process. Set to `False` for system-wide monitoring. | + +!!! note "Network I/O is always system-wide" + Regardless of the `per_process` setting, network I/O metrics are always system-wide because `psutil` does not support per-process network counters on most platforms. + +## Live display integration + +When attached to a Runner, the monitor surfaces real-time metrics in the progress bar during the run — showing current CPU %, RSS memory, and network receive rate. This happens automatically via the `live_stats()` hook; no extra configuration is needed. + +## Reuse across runs + +A single `SystemMetricsMonitor` instance can be reused across multiple `runner.run()` calls. The monitor resets its internal state in `before_run`, so each run gets independent samples and statistics. + +## Detecting client-side bottlenecks + +After a run, check these indicators: + +- **`system_cpu_percent-p90` near 100%** — your client process is CPU-bound and may not be able to drive requests fast enough. +- **`system_memory_rss_mb-max` growing significantly** — memory pressure could trigger GC pauses or swapping. +- **Network rates plateauing while latency increases** — possible network saturation. + +If you observe any of these, consider running LLMeter on a more powerful instance or reducing concurrency. + +## Accessing raw samples + +For custom analysis (e.g., time-series plots correlating CPU with latency), you can access the raw collected samples: + +```python +for sample in monitor.samples: + print(f"t={sample.timestamp:.2f} cpu={sample.cpu_percent:.1f}% rss={sample.memory_rss_mb:.1f}MB") +``` + +Check out the ["System Metrics Monitoring" example notebook](https://github.com/awslabs/llmeter/blob/main/examples/System%20Metrics%20Monitoring.ipynb) on GitHub for a full walkthrough including load test correlation and time-series analysis. diff --git a/examples/System Metrics Monitoring.ipynb b/examples/System Metrics Monitoring.ipynb new file mode 100644 index 0000000..8ab7fed --- /dev/null +++ b/examples/System Metrics Monitoring.ipynb @@ -0,0 +1,412 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# System Metrics Monitoring\n", + "\n", + "When running latency benchmarks — especially load tests with high concurrency — it's important to know whether the **client machine itself** is a bottleneck. High CPU usage, memory pressure, or network saturation on the client side can skew latency measurements without any visible errors.\n", + "\n", + "LLMeter's `SystemMetricsMonitor` callback tracks CPU, memory, and network I/O during benchmark runs using [psutil](https://github.com/giampaolo/psutil). This notebook demonstrates how to:\n", + "\n", + "1. Attach system metrics monitoring to a Runner\n", + "2. Inspect aggregated statistics after a run\n", + "3. Access raw time-series samples for custom analysis\n", + "4. Correlate system resource usage with latency across different concurrency levels\n", + "5. Detect client-side bottlenecks" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Setup\n", + "\n", + "Install LLMeter with the `system-metrics` extra (which brings in `psutil`):" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%pip install \"llmeter[system-metrics,plotting]<1\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from llmeter.callbacks.system_metrics import SystemMetricsMonitor\n", + "from llmeter.endpoints.bedrock import BedrockConverseStream\n", + "from llmeter.runner import Runner" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "This notebook assumes you have [configured AWS credentials](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html) with `bedrock:InvokeModel` and `bedrock:InvokeModelWithResponseStream` permissions, and that you've [enabled access](https://docs.aws.amazon.com/bedrock/latest/userguide/model-access.html) to the model below." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "model_id = None # <-- Set a Bedrock model ID, e.g. \"us.anthropic.claude-3-5-haiku-20241022-v1:0\"\n", + "\n", + "if not model_id:\n", + " raise ValueError(\"Please set a valid model ID above!\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "endpoint = BedrockConverseStream(model_id=model_id)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Basic Usage\n", + "\n", + "Create a `SystemMetricsMonitor` and attach it to a `Runner`. The monitor will automatically start sampling when the run begins and stop when it ends." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=True)\n", + "\n", + "runner = Runner(\n", + " endpoint=endpoint,\n", + " callbacks=[monitor],\n", + " output_path=\"outputs/system-metrics\",\n", + ")\n", + "\n", + "payload = endpoint.create_payload(\n", + " \"Write a short haiku about cloud computing.\",\n", + " max_tokens=100,\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "result = await runner.run(payload=payload, clients=5, n_requests=10)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Inspecting Aggregated Statistics\n", + "\n", + "After the run, system metrics are available in `result.stats` alongside the usual latency and throughput metrics:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# CPU usage during the run\n", + "print(f\"CPU average: {result.stats['system_cpu_percent-average']:.1f}%\")\n", + "print(f\"CPU p90: {result.stats['system_cpu_percent-p90']:.1f}%\")\n", + "print(f\"CPU p99: {result.stats['system_cpu_percent-p99']:.1f}%\")\n", + "print()\n", + "\n", + "# Memory usage\n", + "print(f\"Memory RSS average: {result.stats['system_memory_rss_mb-average']:.1f} MB\")\n", + "print(f\"Memory RSS peak: {result.stats['system_memory_rss_mb-max']:.1f} MB\")\n", + "print()\n", + "\n", + "# Network I/O\n", + "print(f\"Network sent: {result.stats['system_net_bytes_sent_total']:,} bytes\")\n", + "print(f\"Network received: {result.stats['system_net_bytes_recv_total']:,} bytes\")\n", + "print()\n", + "\n", + "# How many samples were collected\n", + "print(f\"Samples collected: {result.stats['system_samples_collected']}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Accessing Raw Samples\n", + "\n", + "For time-series analysis, you can access the raw samples collected during the run. Each sample includes a timestamp, CPU percentage, memory (RSS and VMS), and cumulative network counters." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import pandas as pd\n", + "\n", + "# Convert samples to a DataFrame for easy analysis\n", + "samples_df = pd.DataFrame(\n", + " [\n", + " {\n", + " \"time\": s.timestamp - monitor.samples[0].timestamp, # relative time\n", + " \"cpu_percent\": s.cpu_percent,\n", + " \"memory_rss_mb\": s.memory_rss_mb,\n", + " \"net_bytes_recv\": s.net_bytes_recv,\n", + " }\n", + " for s in monitor.samples\n", + " ]\n", + ")\n", + "\n", + "samples_df.head(10)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import plotly.graph_objects as go\n", + "from plotly.subplots import make_subplots\n", + "\n", + "fig = make_subplots(\n", + " rows=3, cols=1,\n", + " shared_xaxes=True,\n", + " subplot_titles=(\"CPU Usage (%)\", \"Memory RSS (MB)\", \"Network Received (bytes)\"),\n", + " vertical_spacing=0.08,\n", + ")\n", + "\n", + "fig.add_trace(\n", + " go.Scatter(x=samples_df[\"time\"], y=samples_df[\"cpu_percent\"], mode=\"lines+markers\", name=\"CPU %\"),\n", + " row=1, col=1,\n", + ")\n", + "fig.add_trace(\n", + " go.Scatter(x=samples_df[\"time\"], y=samples_df[\"memory_rss_mb\"], mode=\"lines+markers\", name=\"RSS MB\"),\n", + " row=2, col=1,\n", + ")\n", + "fig.add_trace(\n", + " go.Scatter(x=samples_df[\"time\"], y=samples_df[\"net_bytes_recv\"], mode=\"lines+markers\", name=\"Net Recv\"),\n", + " row=3, col=1,\n", + ")\n", + "\n", + "fig.update_layout(height=700, title_text=\"System Metrics Over Time\", showlegend=False)\n", + "fig.update_xaxes(title_text=\"Time (seconds)\", row=3, col=1)\n", + "fig.show()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Correlating System Metrics with Concurrency\n", + "\n", + "A common use case is running the same benchmark at different concurrency levels to see when the client machine becomes saturated. The `SystemMetricsMonitor` resets between runs, so a single instance can be reused." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "concurrency_levels = [1, 5, 10, 20]\n", + "results = {}\n", + "\n", + "for clients in concurrency_levels:\n", + " result = await runner.run(payload=payload, clients=clients, n_requests=5)\n", + " results[clients] = result\n", + " print(\n", + " f\"clients={clients:2d} \"\n", + " f\"CPU avg={result.stats['system_cpu_percent-average']:5.1f}% \"\n", + " f\"RSS max={result.stats['system_memory_rss_mb-max']:6.1f} MB \"\n", + " f\"Net recv={result.stats['system_net_bytes_recv_total']:>10,} bytes\"\n", + " )" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "fig = make_subplots(\n", + " rows=1, cols=2,\n", + " subplot_titles=(\"CPU Usage vs Concurrency\", \"Latency vs Concurrency\"),\n", + ")\n", + "\n", + "clients_list = list(results.keys())\n", + "\n", + "# CPU usage\n", + "cpu_avgs = [results[c].stats[\"system_cpu_percent-average\"] for c in clients_list]\n", + "cpu_p90s = [results[c].stats[\"system_cpu_percent-p90\"] for c in clients_list]\n", + "\n", + "fig.add_trace(\n", + " go.Scatter(x=clients_list, y=cpu_avgs, mode=\"lines+markers\", name=\"CPU avg\"),\n", + " row=1, col=1,\n", + ")\n", + "fig.add_trace(\n", + " go.Scatter(x=clients_list, y=cpu_p90s, mode=\"lines+markers\", name=\"CPU p90\"),\n", + " row=1, col=1,\n", + ")\n", + "\n", + "# Latency (TTLT p50)\n", + "ttlt_p50s = [results[c].stats.get(\"time_to_last_token-p50\", 0) for c in clients_list]\n", + "ttlt_p90s = [results[c].stats.get(\"time_to_last_token-p90\", 0) for c in clients_list]\n", + "\n", + "fig.add_trace(\n", + " go.Scatter(x=clients_list, y=ttlt_p50s, mode=\"lines+markers\", name=\"TTLT p50\"),\n", + " row=1, col=2,\n", + ")\n", + "fig.add_trace(\n", + " go.Scatter(x=clients_list, y=ttlt_p90s, mode=\"lines+markers\", name=\"TTLT p90\"),\n", + " row=1, col=2,\n", + ")\n", + "\n", + "fig.update_xaxes(title_text=\"Concurrent Clients\", row=1, col=1)\n", + "fig.update_xaxes(title_text=\"Concurrent Clients\", row=1, col=2)\n", + "fig.update_yaxes(title_text=\"CPU %\", row=1, col=1)\n", + "fig.update_yaxes(title_text=\"Seconds\", row=1, col=2)\n", + "fig.update_layout(height=400, title_text=\"System Resources vs Latency\")\n", + "fig.show()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Detecting Client-Side Bottlenecks\n", + "\n", + "After running at multiple concurrency levels, you can check for signs that the client is the bottleneck rather than the endpoint:\n", + "\n", + "- **CPU p90 approaching 100%** — the client process can't keep up with scheduling requests.\n", + "- **Memory RSS growing significantly** — asyncio task objects or response data accumulating.\n", + "- **Network rates plateauing while latency increases** — possible network saturation.\n", + "\n", + "If you observe these patterns, consider:\n", + "- Running LLMeter on a more powerful instance\n", + "- Reducing concurrency\n", + "- Using `low_memory=True` for very large runs" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Quick bottleneck check\n", + "for clients, result in results.items():\n", + " cpu_p90 = result.stats[\"system_cpu_percent-p90\"]\n", + " rss_max = result.stats[\"system_memory_rss_mb-max\"]\n", + " status = \"⚠️ HIGH CPU\" if cpu_p90 > 80 else \"✅ OK\"\n", + " print(f\"clients={clients:2d} CPU p90={cpu_p90:5.1f}% RSS max={rss_max:6.1f} MB {status}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## System-Wide vs Per-Process Monitoring\n", + "\n", + "By default, `SystemMetricsMonitor` tracks the current Python process only (`per_process=True`). Set `per_process=False` to monitor the entire machine — useful when other processes (e.g., a local model server) contribute to the workload.\n", + "\n", + "Note that **network I/O is always system-wide** regardless of this setting, since `psutil` doesn't support per-process network counters on most platforms." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# System-wide monitoring example\n", + "system_monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=False)\n", + "\n", + "system_runner = Runner(\n", + " endpoint=endpoint,\n", + " callbacks=[system_monitor],\n", + " output_path=\"outputs/system-metrics-wide\",\n", + ")\n", + "\n", + "result_wide = await system_runner.run(payload=payload, clients=5, n_requests=5)\n", + "\n", + "print(f\"System-wide CPU avg: {result_wide.stats['system_cpu_percent-average']:.1f}%\")\n", + "print(f\"System-wide Memory: {result_wide.stats['system_memory_rss_mb-max']:.1f} MB (used/total)\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Persistence\n", + "\n", + "System metrics are automatically included in `stats.json` when you save results (via `output_path`). They survive `Result.load()` round-trips, so you can analyze them later without re-running the benchmark." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from llmeter.results import Result\n", + "\n", + "# Load a previously-saved result\n", + "if result.output_path:\n", + " loaded = Result.load(result.output_path)\n", + " print(f\"Loaded CPU avg: {loaded.stats['system_cpu_percent-average']:.1f}%\")\n", + " print(f\"Loaded RSS max: {loaded.stats['system_memory_rss_mb-max']:.1f} MB\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Summary\n", + "\n", + "The `SystemMetricsMonitor` callback provides visibility into client-side resource usage during benchmarks. Key takeaways:\n", + "\n", + "- **Attach it to any Runner** to get CPU, memory, and network stats for free.\n", + "- **Live display** shows real-time metrics during the run.\n", + "- **Reusable** across multiple runs — resets automatically.\n", + "- **Persistent** — stats survive save/load round-trips in `stats.json`.\n", + "- **Raw samples** available for custom time-series analysis.\n", + "\n", + "Use it to validate that your benchmark environment has sufficient headroom, and that latency measurements reflect the endpoint rather than client saturation." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.10.0" + } + }, + "nbformat": 4, + "nbformat_minor": 4 +} diff --git a/mkdocs.yml b/mkdocs.yml index f5793e6..62263f9 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -42,6 +42,7 @@ nav: - user_guide/callbacks/cache_buster.md - user_guide/callbacks/cost.md - user_guide/callbacks/mlflow.md + - user_guide/callbacks/system_metrics.md - API Reference: - reference/index.md - callbacks: diff --git a/tests/integ/callbacks/__init__.py b/tests/integ/callbacks/__init__.py new file mode 100644 index 0000000..04f8b7b --- /dev/null +++ b/tests/integ/callbacks/__init__.py @@ -0,0 +1,2 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 diff --git a/tests/integ/callbacks/test_system_metrics.py b/tests/integ/callbacks/test_system_metrics.py new file mode 100644 index 0000000..f5eb747 --- /dev/null +++ b/tests/integ/callbacks/test_system_metrics.py @@ -0,0 +1,268 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Integration tests for SystemMetricsMonitor callback. + +These tests verify that SystemMetricsMonitor works correctly in combination with +actual Runner executions against a live Bedrock endpoint. They validate the full +lifecycle: background sampling during a real run, stat contribution to the result, +persistence across save/load, and reuse across multiple runs. + +To run these tests: + uv run pytest -m integ tests/integ/callbacks/test_system_metrics.py + +Required AWS Permissions: + - bedrock:InvokeModel + +Estimated Cost: + - ~$0.001 total for all tests in this module (multiple short runs) +""" + +import tempfile + +import pytest + +from llmeter.callbacks.system_metrics import SystemMetricsMonitor +from llmeter.endpoints.bedrock import BedrockConverse +from llmeter.results import Result +from llmeter.runner import Runner + + +@pytest.mark.integ +@pytest.mark.asyncio +async def test_system_metrics_with_runner( + aws_credentials, aws_region, bedrock_test_model, test_payload +): + """ + Test SystemMetricsMonitor collects metrics during a real Runner.run(). + + Validates that: + - The monitor starts and stops cleanly during a real run + - System stats are contributed to the result + - CPU, memory, and network metrics are present and reasonable + - Sample count reflects the run duration + + Args: + aws_credentials: Boto3 session with valid AWS credentials. + aws_region: AWS region for testing. + bedrock_test_model: Model ID for testing. + test_payload: Simple text test payload. + """ + monitor = SystemMetricsMonitor(sample_interval=0.2, per_process=True) + endpoint = BedrockConverse(model_id=bedrock_test_model, region=aws_region) + + runner = Runner(endpoint=endpoint, callbacks=[monitor]) + result = await runner.run( + payload=test_payload, + clients=2, + n_requests=3, + disable_per_client_progress_bar=True, + disable_clients_progress_bar=True, + ) + + # Verify the run itself succeeded + assert result.total_requests > 0 + + # Verify system metrics were contributed + assert "system_cpu_percent-average" in result.stats + assert "system_cpu_percent-p50" in result.stats + assert "system_cpu_percent-p90" in result.stats + assert "system_cpu_percent-p99" in result.stats + assert "system_memory_rss_mb-average" in result.stats + assert "system_memory_rss_mb-max" in result.stats + assert "system_memory_vms_mb-average" in result.stats + assert "system_memory_vms_mb-max" in result.stats + assert "system_net_bytes_sent_total" in result.stats + assert "system_net_bytes_recv_total" in result.stats + assert "system_samples_collected" in result.stats + + # Sanity checks on values + assert result.stats["system_cpu_percent-average"] >= 0 + assert result.stats["system_memory_rss_mb-max"] > 0 + assert result.stats["system_net_bytes_recv_total"] >= 0 + assert result.stats["system_samples_collected"] >= 2 + + +@pytest.mark.integ +@pytest.mark.asyncio +async def test_system_metrics_persist_after_save_load( + aws_credentials, aws_region, bedrock_test_model, test_payload +): + """ + Test that system metrics survive a save/load round-trip of the Result. + + Validates that: + - Stats are written to stats.json on save + - Stats are correctly restored on load (with and without responses) + + Args: + aws_credentials: Boto3 session with valid AWS credentials. + aws_region: AWS region for testing. + bedrock_test_model: Model ID for testing. + test_payload: Simple text test payload. + """ + monitor = SystemMetricsMonitor(sample_interval=0.2, per_process=True) + endpoint = BedrockConverse(model_id=bedrock_test_model, region=aws_region) + + with tempfile.TemporaryDirectory() as tmpdir: + runner = Runner(endpoint=endpoint, callbacks=[monitor], output_path=tmpdir) + result = await runner.run( + payload=test_payload, + clients=1, + n_requests=2, + disable_per_client_progress_bar=True, + disable_clients_progress_bar=True, + ) + + # Capture stats before save + original_cpu_avg = result.stats["system_cpu_percent-average"] + original_rss_max = result.stats["system_memory_rss_mb-max"] + original_net_recv = result.stats["system_net_bytes_recv_total"] + + # Load with responses + loaded = Result.load(result.output_path, load_responses=True) + assert loaded.stats["system_cpu_percent-average"] == pytest.approx( + original_cpu_avg + ) + assert loaded.stats["system_memory_rss_mb-max"] == pytest.approx( + original_rss_max + ) + assert loaded.stats["system_net_bytes_recv_total"] == original_net_recv + + # Load without responses (stats-only path) + loaded_no_resp = Result.load(result.output_path, load_responses=False) + assert loaded_no_resp.stats["system_cpu_percent-average"] == pytest.approx( + original_cpu_avg + ) + + +@pytest.mark.integ +@pytest.mark.asyncio +async def test_system_metrics_reuse_across_runs( + aws_credentials, aws_region, bedrock_test_model, test_payload +): + """ + Test that a single SystemMetricsMonitor instance can be reused across runs. + + Validates that: + - The monitor resets state between runs + - Each run gets independent samples and statistics + - No cross-contamination of metrics between runs + + Args: + aws_credentials: Boto3 session with valid AWS credentials. + aws_region: AWS region for testing. + bedrock_test_model: Model ID for testing. + test_payload: Simple text test payload. + """ + monitor = SystemMetricsMonitor(sample_interval=0.2, per_process=True) + endpoint = BedrockConverse(model_id=bedrock_test_model, region=aws_region) + + runner = Runner(endpoint=endpoint, callbacks=[monitor]) + + # First run + result1 = await runner.run( + payload=test_payload, + clients=1, + n_requests=2, + disable_per_client_progress_bar=True, + disable_clients_progress_bar=True, + ) + + # Second run + result2 = await runner.run( + payload=test_payload, + clients=1, + n_requests=2, + disable_per_client_progress_bar=True, + disable_clients_progress_bar=True, + ) + + # Both runs should have independent system metrics + assert "system_samples_collected" in result1.stats + assert "system_samples_collected" in result2.stats + assert result1.stats["system_samples_collected"] >= 2 + assert result2.stats["system_samples_collected"] >= 2 + + # Memory should be positive in both + assert result1.stats["system_memory_rss_mb-max"] > 0 + assert result2.stats["system_memory_rss_mb-max"] > 0 + + +@pytest.mark.integ +@pytest.mark.asyncio +async def test_system_metrics_system_wide_mode( + aws_credentials, aws_region, bedrock_test_model, test_payload +): + """ + Test SystemMetricsMonitor in system-wide mode (per_process=False). + + Validates that: + - System-wide mode collects metrics from the entire machine + - Stats are contributed correctly + + Args: + aws_credentials: Boto3 session with valid AWS credentials. + aws_region: AWS region for testing. + bedrock_test_model: Model ID for testing. + test_payload: Simple text test payload. + """ + monitor = SystemMetricsMonitor(sample_interval=0.2, per_process=False) + endpoint = BedrockConverse(model_id=bedrock_test_model, region=aws_region) + + runner = Runner(endpoint=endpoint, callbacks=[monitor]) + result = await runner.run( + payload=test_payload, + clients=1, + n_requests=2, + disable_per_client_progress_bar=True, + disable_clients_progress_bar=True, + ) + + # System-wide mode should report higher CPU and memory than per-process + assert "system_cpu_percent-average" in result.stats + assert "system_memory_rss_mb-max" in result.stats + assert result.stats["system_memory_rss_mb-max"] > 0 + assert result.stats["system_samples_collected"] >= 2 + + +@pytest.mark.integ +@pytest.mark.asyncio +async def test_system_metrics_with_higher_concurrency( + aws_credentials, aws_region, bedrock_test_model, test_payload +): + """ + Test SystemMetricsMonitor during a higher-concurrency run. + + This validates that the monitor captures meaningful resource variation + under load. With more concurrent requests, we expect to see non-trivial + CPU usage and network activity. + + Args: + aws_credentials: Boto3 session with valid AWS credentials. + aws_region: AWS region for testing. + bedrock_test_model: Model ID for testing. + test_payload: Simple text test payload. + """ + monitor = SystemMetricsMonitor(sample_interval=0.1, per_process=True) + endpoint = BedrockConverse(model_id=bedrock_test_model, region=aws_region) + + runner = Runner(endpoint=endpoint, callbacks=[monitor]) + result = await runner.run( + payload=test_payload, + clients=5, + n_requests=3, + disable_per_client_progress_bar=True, + disable_clients_progress_bar=True, + ) + + # With higher concurrency, we expect more samples and network activity + assert result.stats["system_samples_collected"] >= 3 + assert result.stats["system_net_bytes_sent_total"] > 0 + assert result.stats["system_net_bytes_recv_total"] > 0 + + # Network rate stats should be present with enough samples + if result.stats["system_samples_collected"] >= 3: + assert "system_net_bytes_recv_per_second-average" in result.stats + assert result.stats["system_net_bytes_recv_per_second-average"] > 0 From 07fb541b892ba0ba3f54c5a5ef1502583cc7757d Mon Sep 17 00:00:00 2001 From: Alessandro Cere Date: Fri, 3 Jul 2026 10:23:10 +0800 Subject: [PATCH 4/4] feat(system-metrics): add plotting, sample persistence, and LoadTest integration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add plot_samples() method to SystemMetricsMonitor for time-series visualization of collected metrics within a run - Add extra_stats parameter to LoadTestResult.plot_results() and plot_load_test_results() for plotting arbitrary stat keys vs concurrency (e.g. system metrics alongside standard charts) - Persist raw samples to system_metrics_samples.jsonl in after_run when output_path is set, with load_samples() classmethod to restore - Adopt __llmeter_class__/__llmeter_state__ envelope format for save_to_file, compatible with the serialization branch - Remove __getstate__/__setstate__ — the dataclasses.asdict() deepcopy issue in _RunConfig.save() is a library-level concern (see #94) - Rewrite example notebook to use result.stats, monitor.plot_samples(), and load_test_result.plot_results(extra_stats=...) instead of manual plotly code --- examples/System Metrics Monitoring.ipynb | 248 ++++++++------------ llmeter/callbacks/system_metrics.py | 204 +++++++++++++--- llmeter/experiments.py | 29 ++- llmeter/plotting/plotting.py | 30 ++- tests/unit/callbacks/test_system_metrics.py | 101 ++------ tests/unit/test_experiments.py | 2 +- 6 files changed, 351 insertions(+), 263 deletions(-) diff --git a/examples/System Metrics Monitoring.ipynb b/examples/System Metrics Monitoring.ipynb index 8ab7fed..e01c26b 100644 --- a/examples/System Metrics Monitoring.ipynb +++ b/examples/System Metrics Monitoring.ipynb @@ -11,9 +11,9 @@ "LLMeter's `SystemMetricsMonitor` callback tracks CPU, memory, and network I/O during benchmark runs using [psutil](https://github.com/giampaolo/psutil). This notebook demonstrates how to:\n", "\n", "1. Attach system metrics monitoring to a Runner\n", - "2. Inspect aggregated statistics after a run\n", - "3. Access raw time-series samples for custom analysis\n", - "4. Correlate system resource usage with latency across different concurrency levels\n", + "2. Inspect aggregated statistics from the Result\n", + "3. Plot time-series of system resources within a run\n", + "4. Use LoadTest to correlate system resources with concurrency\n", "5. Detect client-side bottlenecks" ] }, @@ -23,7 +23,7 @@ "source": [ "## Setup\n", "\n", - "Install LLMeter with the `system-metrics` extra (which brings in `psutil`):" + "Install LLMeter with the `system-metrics` extra (which brings in `psutil`) and `plotting` for visualization:" ] }, { @@ -43,6 +43,7 @@ "source": [ "from llmeter.callbacks.system_metrics import SystemMetricsMonitor\n", "from llmeter.endpoints.bedrock import BedrockConverseStream\n", + "from llmeter.experiments import LoadTest\n", "from llmeter.runner import Runner" ] }, @@ -71,7 +72,12 @@ "metadata": {}, "outputs": [], "source": [ - "endpoint = BedrockConverseStream(model_id=model_id)" + "endpoint = BedrockConverseStream(model_id=model_id)\n", + "\n", + "payload = endpoint.create_payload(\n", + " \"Write a short haiku about cloud computing.\",\n", + " max_tokens=100,\n", + ")" ] }, { @@ -80,7 +86,7 @@ "source": [ "## Basic Usage\n", "\n", - "Create a `SystemMetricsMonitor` and attach it to a `Runner`. The monitor will automatically start sampling when the run begins and stop when it ends." + "Create a `SystemMetricsMonitor` and attach it to a `Runner`. The monitor spawns a lightweight background thread that samples system metrics at the configured interval. After the run, aggregated statistics are contributed directly to `result.stats`." ] }, { @@ -97,18 +103,6 @@ " output_path=\"outputs/system-metrics\",\n", ")\n", "\n", - "payload = endpoint.create_payload(\n", - " \"Write a short haiku about cloud computing.\",\n", - " max_tokens=100,\n", - ")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ "result = await runner.run(payload=payload, clients=5, n_requests=10)" ] }, @@ -116,9 +110,9 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "## Inspecting Aggregated Statistics\n", + "## Inspecting Results\n", "\n", - "After the run, system metrics are available in `result.stats` alongside the usual latency and throughput metrics:" + "System metrics are part of `result.stats` — the same interface used for latency and throughput statistics. No need to access the monitor object for aggregated values:" ] }, { @@ -151,9 +145,7 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "## Accessing Raw Samples\n", - "\n", - "For time-series analysis, you can access the raw samples collected during the run. Each sample includes a timestamp, CPU percentage, memory (RSS and VMS), and cumulative network counters." + "These stats persist across save/load — they're stored in `stats.json` alongside latency metrics:" ] }, { @@ -162,22 +154,21 @@ "metadata": {}, "outputs": [], "source": [ - "import pandas as pd\n", + "from llmeter.results import Result\n", "\n", - "# Convert samples to a DataFrame for easy analysis\n", - "samples_df = pd.DataFrame(\n", - " [\n", - " {\n", - " \"time\": s.timestamp - monitor.samples[0].timestamp, # relative time\n", - " \"cpu_percent\": s.cpu_percent,\n", - " \"memory_rss_mb\": s.memory_rss_mb,\n", - " \"net_bytes_recv\": s.net_bytes_recv,\n", - " }\n", - " for s in monitor.samples\n", - " ]\n", - ")\n", + "if result.output_path:\n", + " loaded = Result.load(result.output_path)\n", + " print(f\"Loaded CPU avg: {loaded.stats['system_cpu_percent-average']:.1f}%\")\n", + " print(f\"Loaded RSS max: {loaded.stats['system_memory_rss_mb-max']:.1f} MB\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Plotting Time-Series\n", "\n", - "samples_df.head(10)" + "The monitor provides a built-in `plot_samples()` method that visualizes how CPU, memory, and network evolved over the duration of the run:" ] }, { @@ -186,32 +177,7 @@ "metadata": {}, "outputs": [], "source": [ - "import plotly.graph_objects as go\n", - "from plotly.subplots import make_subplots\n", - "\n", - "fig = make_subplots(\n", - " rows=3, cols=1,\n", - " shared_xaxes=True,\n", - " subplot_titles=(\"CPU Usage (%)\", \"Memory RSS (MB)\", \"Network Received (bytes)\"),\n", - " vertical_spacing=0.08,\n", - ")\n", - "\n", - "fig.add_trace(\n", - " go.Scatter(x=samples_df[\"time\"], y=samples_df[\"cpu_percent\"], mode=\"lines+markers\", name=\"CPU %\"),\n", - " row=1, col=1,\n", - ")\n", - "fig.add_trace(\n", - " go.Scatter(x=samples_df[\"time\"], y=samples_df[\"memory_rss_mb\"], mode=\"lines+markers\", name=\"RSS MB\"),\n", - " row=2, col=1,\n", - ")\n", - "fig.add_trace(\n", - " go.Scatter(x=samples_df[\"time\"], y=samples_df[\"net_bytes_recv\"], mode=\"lines+markers\", name=\"Net Recv\"),\n", - " row=3, col=1,\n", - ")\n", - "\n", - "fig.update_layout(height=700, title_text=\"System Metrics Over Time\", showlegend=False)\n", - "fig.update_xaxes(title_text=\"Time (seconds)\", row=3, col=1)\n", - "fig.show()" + "monitor.plot_samples()" ] }, { @@ -220,7 +186,7 @@ "source": [ "## Correlating System Metrics with Concurrency\n", "\n", - "A common use case is running the same benchmark at different concurrency levels to see when the client machine becomes saturated. The `SystemMetricsMonitor` resets between runs, so a single instance can be reused." + "The most common use case for system monitoring is understanding whether client resources become a bottleneck as concurrency increases. Use `LoadTest` to sweep across concurrency levels, then `plot_results(extra_stats=...)` to visualize system metrics alongside the standard charts." ] }, { @@ -229,18 +195,16 @@ "metadata": {}, "outputs": [], "source": [ - "concurrency_levels = [1, 5, 10, 20]\n", - "results = {}\n", + "load_test = LoadTest(\n", + " endpoint=endpoint,\n", + " payload=payload,\n", + " sequence_of_clients=[1, 5, 10, 20],\n", + " min_requests_per_client=5,\n", + " callbacks=[monitor],\n", + " output_path=\"outputs/system-metrics-load-test\",\n", + ")\n", "\n", - "for clients in concurrency_levels:\n", - " result = await runner.run(payload=payload, clients=clients, n_requests=5)\n", - " results[clients] = result\n", - " print(\n", - " f\"clients={clients:2d} \"\n", - " f\"CPU avg={result.stats['system_cpu_percent-average']:5.1f}% \"\n", - " f\"RSS max={result.stats['system_memory_rss_mb-max']:6.1f} MB \"\n", - " f\"Net recv={result.stats['system_net_bytes_recv_total']:>10,} bytes\"\n", - " )" + "load_test_result = await load_test.run()" ] }, { @@ -249,45 +213,12 @@ "metadata": {}, "outputs": [], "source": [ - "fig = make_subplots(\n", - " rows=1, cols=2,\n", - " subplot_titles=(\"CPU Usage vs Concurrency\", \"Latency vs Concurrency\"),\n", - ")\n", - "\n", - "clients_list = list(results.keys())\n", - "\n", - "# CPU usage\n", - "cpu_avgs = [results[c].stats[\"system_cpu_percent-average\"] for c in clients_list]\n", - "cpu_p90s = [results[c].stats[\"system_cpu_percent-p90\"] for c in clients_list]\n", - "\n", - "fig.add_trace(\n", - " go.Scatter(x=clients_list, y=cpu_avgs, mode=\"lines+markers\", name=\"CPU avg\"),\n", - " row=1, col=1,\n", - ")\n", - "fig.add_trace(\n", - " go.Scatter(x=clients_list, y=cpu_p90s, mode=\"lines+markers\", name=\"CPU p90\"),\n", - " row=1, col=1,\n", - ")\n", - "\n", - "# Latency (TTLT p50)\n", - "ttlt_p50s = [results[c].stats.get(\"time_to_last_token-p50\", 0) for c in clients_list]\n", - "ttlt_p90s = [results[c].stats.get(\"time_to_last_token-p90\", 0) for c in clients_list]\n", - "\n", - "fig.add_trace(\n", - " go.Scatter(x=clients_list, y=ttlt_p50s, mode=\"lines+markers\", name=\"TTLT p50\"),\n", - " row=1, col=2,\n", - ")\n", - "fig.add_trace(\n", - " go.Scatter(x=clients_list, y=ttlt_p90s, mode=\"lines+markers\", name=\"TTLT p90\"),\n", - " row=1, col=2,\n", - ")\n", - "\n", - "fig.update_xaxes(title_text=\"Concurrent Clients\", row=1, col=1)\n", - "fig.update_xaxes(title_text=\"Concurrent Clients\", row=1, col=2)\n", - "fig.update_yaxes(title_text=\"CPU %\", row=1, col=1)\n", - "fig.update_yaxes(title_text=\"Seconds\", row=1, col=2)\n", - "fig.update_layout(height=400, title_text=\"System Resources vs Latency\")\n", - "fig.show()" + "load_test_result.plot_results(extra_stats={\n", + " \"system_cpu_percent-average\": \"CPU avg (%)\",\n", + " \"system_cpu_percent-p90\": \"CPU p90 (%)\",\n", + " \"system_memory_rss_mb-max\": \"RSS peak (MB)\",\n", + " \"system_net_bytes_recv_per_second-average\": \"Net recv rate (bytes/s)\",\n", + "})" ] }, { @@ -296,16 +227,13 @@ "source": [ "## Detecting Client-Side Bottlenecks\n", "\n", - "After running at multiple concurrency levels, you can check for signs that the client is the bottleneck rather than the endpoint:\n", + "After a load test, check these indicators across concurrency levels:\n", "\n", "- **CPU p90 approaching 100%** — the client process can't keep up with scheduling requests.\n", - "- **Memory RSS growing significantly** — asyncio task objects or response data accumulating.\n", + "- **Memory RSS growing significantly** — response data or asyncio tasks accumulating.\n", "- **Network rates plateauing while latency increases** — possible network saturation.\n", "\n", - "If you observe these patterns, consider:\n", - "- Running LLMeter on a more powerful instance\n", - "- Reducing concurrency\n", - "- Using `low_memory=True` for very large runs" + "If you observe these, consider running LLMeter on a more powerful instance or reducing concurrency." ] }, { @@ -314,11 +242,11 @@ "metadata": {}, "outputs": [], "source": [ - "# Quick bottleneck check\n", - "for clients, result in results.items():\n", - " cpu_p90 = result.stats[\"system_cpu_percent-p90\"]\n", - " rss_max = result.stats[\"system_memory_rss_mb-max\"]\n", - " status = \"⚠️ HIGH CPU\" if cpu_p90 > 80 else \"✅ OK\"\n", + "for clients in sorted(load_test_result.results.keys()):\n", + " r = load_test_result.results[clients]\n", + " cpu_p90 = r.stats[\"system_cpu_percent-p90\"]\n", + " rss_max = r.stats[\"system_memory_rss_mb-max\"]\n", + " status = \"\\u26a0\\ufe0f HIGH CPU\" if cpu_p90 > 80 else \"\\u2705 OK\"\n", " print(f\"clients={clients:2d} CPU p90={cpu_p90:5.1f}% RSS max={rss_max:6.1f} MB {status}\")" ] }, @@ -326,11 +254,11 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "## System-Wide vs Per-Process Monitoring\n", + "## Raw Samples\n", "\n", - "By default, `SystemMetricsMonitor` tracks the current Python process only (`per_process=True`). Set `per_process=False` to monitor the entire machine — useful when other processes (e.g., a local model server) contribute to the workload.\n", + "For custom analysis beyond the built-in plots, you can access the raw samples from the monitor. Each sample is a dataclass with `timestamp`, `cpu_percent`, `memory_rss_mb`, `memory_vms_mb`, `net_bytes_sent`, and `net_bytes_recv`.\n", "\n", - "Note that **network I/O is always system-wide** regardless of this setting, since `psutil` doesn't support per-process network counters on most platforms." + "> **Note:** Raw samples live only on the monitor instance — they are not persisted in `stats.json`. Only the aggregated statistics survive save/load." ] }, { @@ -339,28 +267,40 @@ "metadata": {}, "outputs": [], "source": [ - "# System-wide monitoring example\n", - "system_monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=False)\n", + "# Iterate directly — no extra dependencies needed\n", + "t0 = monitor.samples[0].timestamp\n", "\n", - "system_runner = Runner(\n", - " endpoint=endpoint,\n", - " callbacks=[system_monitor],\n", - " output_path=\"outputs/system-metrics-wide\",\n", - ")\n", + "print(f\"{'Time (s)':>8} {'CPU %':>6} {'RSS MB':>7} {'Net Recv':>12}\")\n", + "print(\"-\" * 40)\n", + "for s in monitor.samples[:10]:\n", + " print(f\"{s.timestamp - t0:8.2f} {s.cpu_percent:6.1f} {s.memory_rss_mb:7.1f} {s.net_bytes_recv:>12,}\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# If you prefer a DataFrame, samples are dataclasses — use asdict:\n", + "from dataclasses import asdict\n", "\n", - "result_wide = await system_runner.run(payload=payload, clients=5, n_requests=5)\n", + "import pandas as pd\n", "\n", - "print(f\"System-wide CPU avg: {result_wide.stats['system_cpu_percent-average']:.1f}%\")\n", - "print(f\"System-wide Memory: {result_wide.stats['system_memory_rss_mb-max']:.1f} MB (used/total)\")" + "df = pd.DataFrame([asdict(s) for s in monitor.samples])\n", + "df[\"time\"] = df[\"timestamp\"] - df[\"timestamp\"].iloc[0]\n", + "df.head()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ - "## Persistence\n", + "## System-Wide vs Per-Process Monitoring\n", + "\n", + "By default, `SystemMetricsMonitor` tracks the current Python process only (`per_process=True`). Set `per_process=False` to monitor the entire machine — useful when other processes (e.g., a local model server) contribute to the workload.\n", "\n", - "System metrics are automatically included in `stats.json` when you save results (via `output_path`). They survive `Result.load()` round-trips, so you can analyze them later without re-running the benchmark." + "> **Note:** Network I/O is always system-wide regardless of this setting, since `psutil` doesn't support per-process network counters on most platforms." ] }, { @@ -369,13 +309,14 @@ "metadata": {}, "outputs": [], "source": [ - "from llmeter.results import Result\n", + "system_wide_monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=False)\n", "\n", - "# Load a previously-saved result\n", - "if result.output_path:\n", - " loaded = Result.load(result.output_path)\n", - " print(f\"Loaded CPU avg: {loaded.stats['system_cpu_percent-average']:.1f}%\")\n", - " print(f\"Loaded RSS max: {loaded.stats['system_memory_rss_mb-max']:.1f} MB\")" + "system_runner = Runner(endpoint=endpoint, callbacks=[system_wide_monitor])\n", + "result_wide = await system_runner.run(payload=payload, clients=5, n_requests=5)\n", + "\n", + "print(f\"System-wide CPU avg: {result_wide.stats['system_cpu_percent-average']:.1f}%\")\n", + "print(f\"System-wide Memory: {result_wide.stats['system_memory_rss_mb-max']:.1f} MB (used)\")\n", + "print(f\"System-wide VMS: {result_wide.stats['system_memory_vms_mb-max']:.1f} MB (total)\")" ] }, { @@ -384,15 +325,12 @@ "source": [ "## Summary\n", "\n", - "The `SystemMetricsMonitor` callback provides visibility into client-side resource usage during benchmarks. Key takeaways:\n", - "\n", - "- **Attach it to any Runner** to get CPU, memory, and network stats for free.\n", - "- **Live display** shows real-time metrics during the run.\n", - "- **Reusable** across multiple runs — resets automatically.\n", - "- **Persistent** — stats survive save/load round-trips in `stats.json`.\n", - "- **Raw samples** available for custom time-series analysis.\n", - "\n", - "Use it to validate that your benchmark environment has sufficient headroom, and that latency measurements reflect the endpoint rather than client saturation." + "- **Aggregated stats** (CPU, memory, network) are part of `result.stats` — use the standard Result interface.\n", + "- **`monitor.plot_samples()`** gives you a time-series visualization of the last run.\n", + "- **`load_test_result.plot_results(extra_stats=...)`** plots system metrics vs concurrency alongside standard charts.\n", + "- **Raw samples** on `monitor.samples` are dataclasses you can iterate directly or convert with `dataclasses.asdict`.\n", + "- The monitor **resets between runs** — a single instance works across multiple `runner.run()` calls or LoadTest levels.\n", + "- Stats **persist** in `stats.json` across save/load; raw samples do not." ] } ], diff --git a/llmeter/callbacks/system_metrics.py b/llmeter/callbacks/system_metrics.py index 726e8a0..f0758ae 100644 --- a/llmeter/callbacks/system_metrics.py +++ b/llmeter/callbacks/system_metrics.py @@ -6,6 +6,7 @@ import threading import time from dataclasses import dataclass +from typing import TYPE_CHECKING from upath.types import ReadablePathLike, WritablePathLike @@ -25,6 +26,17 @@ ) psutil = DeferredError(e) +if not TYPE_CHECKING: + try: + import plotly.graph_objects as go + from plotly.subplots import make_subplots + except ImportError as e: + go = DeferredError(e) + make_subplots = DeferredError(e) +else: + import plotly.graph_objects as go + from plotly.subplots import make_subplots + @dataclass class _Sample: @@ -93,25 +105,6 @@ def __post_init__(self): self._process: object = None self._net_start: tuple[int, int] = (0, 0) - def __getstate__(self): - """Support pickling/deepcopy by excluding non-serializable thread state.""" - return { - "sample_interval": self.sample_interval, - "per_process": self.per_process, - "_samples": self._samples, - "_net_start": self._net_start, - } - - def __setstate__(self, state): - """Restore from pickle/deepcopy, reinitializing thread primitives.""" - self.sample_interval = state["sample_interval"] - self.per_process = state["per_process"] - self._samples = state["_samples"] - self._net_start = state["_net_start"] - self._thread = None - self._stop_event = threading.Event() - self._process = None - def _collect_sample(self) -> _Sample: """Collect a single metrics sample.""" now = time.perf_counter() @@ -198,6 +191,10 @@ async def after_run(self, result: Result) -> None: # Contribute to result result._update_contributed_stats(stats) + # Persist raw samples alongside the result + if result.output_path: + self._save_samples(result.output_path) + logger.info( "System metrics: %d samples collected. CPU avg=%.1f%%, " "Memory RSS peak=%.1f MB, Net sent=%d bytes, Net recv=%d bytes", @@ -314,20 +311,98 @@ def live_stats(self) -> dict[str, float | int]: return stats + def _save_samples(self, output_path: WritablePathLike) -> None: + """Save raw samples to a JSONL file alongside the result. + + Args: + output_path: The result's output directory. + """ + import json + + from ..utils import ensure_path + + path = ensure_path(output_path) / "system_metrics_samples.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w") as f: + for sample in self._samples: + f.write( + json.dumps( + { + "timestamp": sample.timestamp, + "cpu_percent": sample.cpu_percent, + "memory_rss_mb": sample.memory_rss_mb, + "memory_vms_mb": sample.memory_vms_mb, + "net_bytes_sent": sample.net_bytes_sent, + "net_bytes_recv": sample.net_bytes_recv, + } + ) + + "\n" + ) + logger.debug("Saved %d system metrics samples to %s", len(self._samples), path) + + @classmethod + def load_samples(cls, output_path: ReadablePathLike) -> list[_Sample]: + """Load raw samples from a previously saved result directory. + + This allows time-series analysis and ``plot_samples()`` on results loaded + from disk, without needing to re-run the benchmark. + + Args: + output_path: The result's output directory (same path used with + ``Result.load()``). + + Returns: + A list of ``_Sample`` dataclass instances. + + Raises: + FileNotFoundError: If no samples file exists at the given path. + + Example:: + + monitor = SystemMetricsMonitor() + monitor._samples = SystemMetricsMonitor.load_samples("outputs/my-run/20240101-1200") + monitor.plot_samples() + """ + import json + + from ..utils import ensure_path + + path = ensure_path(output_path) / "system_metrics_samples.jsonl" + if not path.exists(): + raise FileNotFoundError( + f"No system metrics samples found at {path}. " + "Ensure the run was executed with a SystemMetricsMonitor and an output_path." + ) + + samples = [] + with path.open("r") as f: + for line in f: + d = json.loads(line) + samples.append(_Sample(**d)) + return samples + def save_to_file(self, path: WritablePathLike) -> None: - """Save this SystemMetricsMonitor configuration to file.""" + """Save this SystemMetricsMonitor configuration to file. + + Uses the ``__llmeter_class__``/``__llmeter_state__`` envelope format + compatible with the unified serialization layer. + """ import json from ..utils import ensure_path out_path = ensure_path(path) - config = { - "type": "SystemMetricsMonitor", - "sample_interval": self.sample_interval, - "per_process": self.per_process, + out_path.parent.mkdir(parents=True, exist_ok=True) + class_path = f"{self.__class__.__module__}.{self.__class__.__qualname__}" + data = { + "__llmeter_class__": class_path, + "__llmeter_state__": { + "sample_interval": self.sample_interval, + "per_process": self.per_process, + }, } with out_path.open("w") as f: - json.dump(config, f, indent=2) + json.dump(data, f, indent=2) @classmethod def _load_from_file(cls, path: ReadablePathLike) -> "SystemMetricsMonitor": @@ -338,9 +413,82 @@ def _load_from_file(cls, path: ReadablePathLike) -> "SystemMetricsMonitor": in_path = ensure_path(path) with in_path.open("r") as f: - config = json.load(f) + data = json.load(f) + + # Support both legacy format and new envelope format + if "__llmeter_state__" in data: + state = data["__llmeter_state__"] + else: + state = data return cls( - sample_interval=config.get("sample_interval", 1.0), - per_process=config.get("per_process", True), + sample_interval=state.get("sample_interval", 1.0), + per_process=state.get("per_process", True), + ) + + def plot_samples(self, show: bool = True): + """Plot time-series of collected samples (CPU, memory, network). + + Creates a multi-panel figure showing how system resources evolved over + the duration of the last run. Requires the ``plotting`` extra + (``pip install 'llmeter[plotting]'``). + + Args: + show: Whether to display the figure interactively. Default True. + + Returns: + plotly.graph_objects.Figure: The generated figure. + + Raises: + ValueError: If no samples have been collected yet. + """ + if not self._samples: + raise ValueError( + "No samples to plot. Run a benchmark with this monitor attached first." + ) + + t0 = self._samples[0].timestamp + times = [s.timestamp - t0 for s in self._samples] + cpu = [s.cpu_percent for s in self._samples] + rss = [s.memory_rss_mb for s in self._samples] + net_recv = [s.net_bytes_recv for s in self._samples] + + fig = make_subplots( + rows=3, + cols=1, + shared_xaxes=True, + subplot_titles=( + "CPU Usage (%)", + "Memory RSS (MB)", + "Cumulative Network Received (bytes)", + ), + vertical_spacing=0.08, + ) + + fig.add_trace( + go.Scatter(x=times, y=cpu, mode="lines+markers", name="CPU %"), + row=1, + col=1, + ) + fig.add_trace( + go.Scatter(x=times, y=rss, mode="lines+markers", name="RSS MB"), + row=2, + col=1, + ) + fig.add_trace( + go.Scatter(x=times, y=net_recv, mode="lines+markers", name="Net Recv"), + row=3, + col=1, ) + + fig.update_layout( + height=700, + title_text="System Metrics Over Time", + showlegend=False, + template="plotly_white", + ) + fig.update_xaxes(title_text="Time (seconds)", row=3, col=1) + + if show: + fig.show() + return fig diff --git a/llmeter/experiments.py b/llmeter/experiments.py index 2a55b95..3342508 100644 --- a/llmeter/experiments.py +++ b/llmeter/experiments.py @@ -41,8 +41,33 @@ class LoadTestResult: test_name: str output_path: WritablePathLike | None = None - def plot_results(self, show: bool = True, format: Literal["html", "png"] = "html"): - figs = plot_load_test_results(self) + def plot_results( + self, + show: bool = True, + format: Literal["html", "png"] = "html", + extra_stats: dict[str, str] | None = None, + ): + """Plot load test results. + + Generates the standard set of plots (latency, RPM, error rate, tokens) and + optionally additional figures for arbitrary stat keys vs number of clients. + + Args: + show: Whether to display figures interactively. Default True. + format: File format when saving figures to ``output_path``. + extra_stats: Additional stat keys to plot vs concurrency. Maps stat keys + (as they appear in ``result.stats``) to y-axis labels. Each entry + produces a separate figure. Example:: + + load_test_result.plot_results(extra_stats={ + "system_cpu_percent-p90": "CPU p90 (%)", + "system_memory_rss_mb-max": "RSS peak (MB)", + }) + + Returns: + dict[str, Figure]: All generated figures keyed by name. + """ + figs = plot_load_test_results(self, extra_stats=extra_stats) # add individual color sequence for each plot c_seqs = [ diff --git a/llmeter/plotting/plotting.py b/llmeter/plotting/plotting.py index 9977955..84bf5f0 100644 --- a/llmeter/plotting/plotting.py +++ b/llmeter/plotting/plotting.py @@ -452,6 +452,7 @@ def latency_clients_fig( def plot_load_test_results( load_test_result: LoadTestResult, log_scale=True, + extra_stats: dict[str, str] | None = None, ) -> dict[str, go.Figure]: """ Generate a collection of plots visualizing different metrics from a load test result. @@ -459,6 +460,14 @@ def plot_load_test_results( Args: load_test_result (LoadTestResult): The load test result object containing the data to plot log_scale (bool, optional): Whether to use logarithmic scale for the plots. Defaults to True. + extra_stats (dict[str, str] | None): Additional stat keys to plot vs number of clients. + Maps stat keys (as they appear in ``result.stats``) to display labels for the y-axis. + Each entry produces a separate figure. For example:: + + extra_stats={ + "system_cpu_percent-p90": "CPU p90 (%)", + "system_memory_rss_mb-max": "RSS peak (MB)", + } Returns: dict: Dictionary containing the following plots: @@ -468,6 +477,7 @@ def plot_load_test_results( - error_rate: Figure showing error rate vs number of clients - average_input_tokens_clients: Figure showing average input tokens per minute vs clients - average_output_tokens_clients: Figure showing average output tokens per minute vs clients + - One additional figure per entry in ``extra_stats`` """ f1 = latency_clients_fig( load_test_result, "time_to_first_token", log_scale=log_scale @@ -480,7 +490,7 @@ def plot_load_test_results( f5 = average_input_tokens_clients_fig(load_test_result, log_scale=log_scale) f6 = average_output_tokens_clients_fig(load_test_result, log_scale=log_scale) - return { + figs = { "time_to_first_token": f1, "time_to_last_token": f2, "requests_per_minute": f3, @@ -488,3 +498,21 @@ def plot_load_test_results( "average_input_tokens_clients": f5, "average_output_tokens_clients": f6, } + + if extra_stats: + for stat_key, label in extra_stats.items(): + fig = go.Figure() + fig.add_trace( + stat_clients(load_test_result, stat_key, name=label, opacity=1.0) + ) + fig.update_layout( + title=f"{label} vs number of clients", + xaxis_title="Number of clients", + yaxis_title=label, + template="plotly_white", + ) + if log_scale: + fig.update_xaxes(type="log") + figs[stat_key] = fig + + return figs diff --git a/tests/unit/callbacks/test_system_metrics.py b/tests/unit/callbacks/test_system_metrics.py index 9530982..c54486d 100644 --- a/tests/unit/callbacks/test_system_metrics.py +++ b/tests/unit/callbacks/test_system_metrics.py @@ -2,71 +2,40 @@ # SPDX-License-Identifier: Apache-2.0 import asyncio -import copy import json -import pickle import tempfile -from dataclasses import asdict from pathlib import Path import pytest -from llmeter.callbacks.system_metrics import SystemMetricsMonitor, _Sample +from llmeter.callbacks.system_metrics import SystemMetricsMonitor from llmeter.results import Result -class TestSystemMetricsMonitorPickle: - """Ensure SystemMetricsMonitor survives pickle, deepcopy, and dataclasses.asdict. - - The Runner calls dataclasses.asdict() on its _RunConfig (which includes callbacks) - when saving run_config.json. This internally does a deepcopy of all field values. - Callbacks with threading primitives (locks, events, threads) must handle this - gracefully. - """ - - def test_deepcopy(self): - """Callbacks must survive copy.deepcopy (used by dataclasses.asdict).""" - monitor = SystemMetricsMonitor(sample_interval=0.5, per_process=True) - monitor_copy = copy.deepcopy(monitor) - - assert monitor_copy.sample_interval == 0.5 - assert monitor_copy.per_process is True - assert monitor_copy._stop_event is not monitor._stop_event - - def test_pickle_roundtrip(self): - """Callbacks must survive pickle serialization.""" - monitor = SystemMetricsMonitor(sample_interval=2.0, per_process=False) - data = pickle.dumps(monitor) - restored = pickle.loads(data) - - assert restored.sample_interval == 2.0 - assert restored.per_process is False - assert hasattr(restored, "_stop_event") - assert hasattr(restored, "_thread") - - def test_asdict(self): - """dataclasses.asdict must not raise on SystemMetricsMonitor.""" - monitor = SystemMetricsMonitor(sample_interval=1.0, per_process=True) - d = asdict(monitor) - - assert d == {"sample_interval": 1.0, "per_process": True} - - def test_deepcopy_preserves_samples(self): - """Collected samples should survive deepcopy.""" - monitor = SystemMetricsMonitor(sample_interval=1.0) - monitor._samples = [ - _Sample( - timestamp=1.0, - cpu_percent=25.0, - memory_rss_mb=100.0, - memory_vms_mb=200.0, - net_bytes_sent=1000, - net_bytes_recv=2000, - ) - ] - monitor_copy = copy.deepcopy(monitor) - assert len(monitor_copy._samples) == 1 - assert monitor_copy._samples[0].cpu_percent == 25.0 +class TestSystemMetricsMonitorSerialization: + """Ensure SystemMetricsMonitor configuration can be saved and loaded.""" + + def test_save_to_file_and_load(self): + """Test callback configuration persistence.""" + monitor = SystemMetricsMonitor(sample_interval=2.5, per_process=False) + + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "monitor.json" + monitor.save_to_file(path) + + # Verify file contents use the envelope format + with open(path) as f: + config = json.load(f) + assert "__llmeter_class__" in config + assert "__llmeter_state__" in config + assert "SystemMetricsMonitor" in config["__llmeter_class__"] + assert config["__llmeter_state__"]["sample_interval"] == 2.5 + assert config["__llmeter_state__"]["per_process"] is False + + # Load back + restored = SystemMetricsMonitor._load_from_file(path) + assert restored.sample_interval == 2.5 + assert restored.per_process is False class TestSystemMetricsMonitorLifecycle: @@ -207,26 +176,6 @@ class FakeConfig: asyncio.run(_test()) - def test_save_to_file_and_load(self): - """Test callback configuration persistence.""" - monitor = SystemMetricsMonitor(sample_interval=2.5, per_process=False) - - with tempfile.TemporaryDirectory() as tmpdir: - path = Path(tmpdir) / "monitor.json" - monitor.save_to_file(path) - - # Verify file contents - with open(path) as f: - config = json.load(f) - assert config["type"] == "SystemMetricsMonitor" - assert config["sample_interval"] == 2.5 - assert config["per_process"] is False - - # Load back - restored = SystemMetricsMonitor._load_from_file(path) - assert restored.sample_interval == 2.5 - assert restored.per_process is False - class TestSystemMetricsMonitorReuse: """Test that a monitor can be reused across multiple runs.""" diff --git a/tests/unit/test_experiments.py b/tests/unit/test_experiments.py index 7c351f1..36281ad 100644 --- a/tests/unit/test_experiments.py +++ b/tests/unit/test_experiments.py @@ -179,7 +179,7 @@ def test_plot_results_html(self, mock_plot): figs = load_test_result.plot_results(show=False, format="html") assert figs == {"fig1": mock_fig1, "fig2": mock_fig2} - mock_plot.assert_called_once_with(load_test_result) + mock_plot.assert_called_once_with(load_test_result, extra_stats=None) mock_fig1.write_html.assert_called_once() mock_fig2.write_html.assert_called_once()