Load Test Results Analyzer

Load Test Results Analyzer

Problem Statement

Load testing generates thousands of rows of raw request data, but the numbers that actually matter — p95 latency, error rate, and throughput — are buried in noise. After every Locust, k6, or JMeter run, engineers either eyeball a CSV in Excel or re-run ad-hoc pandas snippets in a notebook, which leads to inconsistent comparisons between builds and no automated pass/fail gate in CI.

This script turns a raw locust_results.csv (or k6/JMeter export with a column mapping) into a concise, reproducible report: percentile latencies, error rates, requests-per-second, per-endpoint breakdowns, and an optional PNG chart. It also evaluates configurable thresholds (p95=200,p99=500) so it can exit non-zero and fail a CI pipeline when a performance budget is violated.

How It Works

The script follows a linear ETL pipeline: load → normalize → analyze → report → gate. It reads the CSV with pandas, then normalizes columns via a small alias map so the same code handles Locust (response_time, name, success), k6, and JMeter exports. Timestamps are parsed to compute wall-clock duration and throughput.

Analysis uses vectorized pandas operations: quantile() for percentiles, grouped agg() for per-endpoint stats, and boolean masking for error rates. The threshold evaluator parses key=value pairs into a spec dict and compares computed metrics, accumulating failures. Reporting is decoupled — a text reporter always runs, the matplotlib chart is opt-in via --report, and the process exit code is driven solely by threshold violations. This makes the script safe to use both interactively and as a CI gate.

Full Script

#!/usr/bin/env python3
"""
Load Test Results Analyzer

Ingests a load-test results CSV (Locust/k6/JMeter), computes latency
percentiles, error rates, and throughput, optionally renders a chart,
and enforces performance thresholds suitable for CI gating.

Usage:
    python3 load_results.py --file locust_results.csv \
        --report --thresholds p95=200,p99=500
"""

from __future__ import annotations

import argparse
import os
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable

import pandas as pd

# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
DEFAULT_LATENCY_COL = os.environ.get("LT_LATENCY_COL", "response_time")
DEFAULT_ENDPOINT_COL = os.environ.get("LT_ENDPOINT_COL", "name")
DEFAULT_SUCCESS_COL = os.environ.get("LT_SUCCESS_COL", "success")
DEFAULT_TIMESTAMP_COL = os.environ.get("LT_TIMESTAMP_COL", "timestamp")
DEFAULT_ERROR_RATE_LIMIT = float(os.environ.get("LT_MAX_ERROR_RATE", "1.0"))
DEFAULT_REPORT_PATH = os.environ.get("LT_REPORT_PATH", "load_report.png")

# Column aliases so we can accept Locust, k6, and JMeter exports transparently.
COLUMN_ALIASES: dict[str, tuple[str, ...]] = {
    "latency": ("response_time", "http_req_duration", "elapsed", "Latency"),
    "endpoint": ("name", "url", "label", "request_name"),
    "success": ("success", "ok", "passed"),
    "timestamp": ("timestamp", "time_stamp", "timeStamp", "start_time"),
    "status_code": ("response_code", "status", "status_code"),
}


# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class ThresholdSpec:
    """Parsed threshold specification, e.g. p95=200,p99=500,error_rate=2.0"""

    latency_p95_ms: float | None = None
    latency_p99_ms: float | None = None
    error_rate_pct: float | None = None
    min_rps: float | None = None
    violations: list[str] = field(default_factory=list)

    @classmethod
    def parse(cls, raw: str | None) -> "ThresholdSpec":
        spec = cls()
        if not raw:
            return spec
        for pair in raw.split(","):
            pair = pair.strip()
            if not pair:
                continue
            if "=" not in pair:
                raise ValueError(f"Malformed threshold '{pair}', expected key=value")
            key, _, value = pair.partition("=")
            key = key.strip().lower().replace(".", "_")
            try:
                num = float(value)
            except ValueError as exc:
                raise ValueError(f"Threshold '{pair}' has non-numeric value") from exc

            if key in {"p95", "latency_p95"}:
                spec.latency_p95_ms = num
            elif key in {"p99", "latency_p99"}:
                spec.latency_p99_ms = num
            elif key in {"error_rate", "errors"}:
                spec.error_rate_pct = num
            elif key in {"rps", "min_rps", "throughput"}:
                spec.min_rps = num
            else:
                raise ValueError(f"Unknown threshold key '{key}'")
        return spec


@dataclass
class AnalysisResult:
    total_requests: int
    duration_s: float
    rps: float
    error_rate_pct: float
    p50_ms: float
    p95_ms: float
    p99_ms: float
    max_ms: float
    per_endpoint: pd.DataFrame


# ---------------------------------------------------------------------------
# Loading & normalization
# ---------------------------------------------------------------------------
def _resolve_column(df: pd.DataFrame, canonical: str, override: str | None) -> str:
    """Find the actual column name for a canonical field."""
    candidates: list[str] = []
    if override:
        candidates.append(override)
    candidates.extend(COLUMN_ALIASES[canonical])

    for cand in candidates:
        if cand in df.columns:
            return cand
    raise KeyError(
        f"Could not find column for '{canonical}'. "
        f"Tried {candidates}. Available: {list(df.columns)}"
    )


def load_results(path: Path) -> pd.DataFrame:
    """Read and normalize a load-test CSV into canonical columns."""
    if not path.exists():
        raise FileNotFoundError(f"Results file not found: {path}")
    if path.stat().st_size == 0:
        raise ValueError(f"Results file is empty: {path}")

    try:
        df = pd.read_csv(path)
    except pd.errors.ParserError as exc:
        raise ValueError(f"Failed to parse CSV {path}: {exc}") from exc

    if df.empty:
        raise ValueError(f"Results file contains no rows: {path}")

    latency_col = _resolve_column(df, "latency", DEFAULT_LATENCY_COL)
    endpoint_col = _resolve_column(df, "endpoint", DEFAULT_ENDPOINT_COL)

    # Success column is optional; if absent, infer from status codes.
    try:
        success_col = _resolve_column(df, "success", DEFAULT_SUCCESS_COL)
        success = df[success_col]
        if success.dtype == object:
            success = success.astype(str).str.lower().isin({"true", "1", "yes", "ok"})
        else:
            success = success.astype(bool)
    except KeyError:
        status_col = _resolve_column(df, "status_code", None)
        success = pd.to_numeric(df[status_col], errors="coerce").between(200, 399)

    # Timestamp is optional but enables accurate throughput calculation.
    duration_s = 0.0
    try:
        ts_col = _resolve_column(df, "timestamp", DEFAULT_TIMESTAMP_COL)
        ts = pd.to_datetime(df[ts_col], errors="coerce", utc=True)
        if ts.notna().sum() >= 2:
            duration_s = float((ts.max() - ts.min()).total_seconds())
    except KeyError:
        pass  # No timestamp; fall back to per-request sum later.

    normalized = pd.DataFrame(
        {
            "latency_ms": pd.to_numeric(df[latency_col], errors="coerce"),
            "endpoint": df[endpoint_col].astype(str),
            "success": success,
        }
    ).dropna(subset=["latency_ms"])

    if normalized.empty:
        raise ValueError("No valid latency rows after cleaning.")

    normalized.attrs["duration_s"] = duration_s
    return normalized


# ---------------------------------------------------------------------------
# Analysis
# ---------------------------------------------------------------------------
def analyze(df: pd.DataFrame) -> AnalysisResult:
    latencies = df["latency_ms"]
    total = len(df)
    errors = int((~df["success"]).sum())
    error_rate = (errors / total) * 100.0 if total else 0.0

    duration_s = float(df.attrs.get("duration_s") or 0.0)
    if duration_s <= 0:
        # Fallback: assume requests were spread evenly across their own latency.
        duration_s = max(latencies.sum() / 1000.0, 1e-9)

    per_endpoint = (
        df.groupby("endpoint")
        .agg(
            requests=("latency_ms", "size"),
            errors=("success", lambda s: int((~s).sum())),
            p50_ms=("latency_ms", lambda s: s.quantile(0.50)),
            p95_ms=("latency_ms", lambda s: s.quantile(0.95)),
            p99_ms=("latency_ms", lambda s: s.quantile(0.99)),
        )
        .sort_values("p95_ms", ascending=False)
        .round(2)
    )

    return AnalysisResult(
        total_requests=total,
        duration_s=duration_s,
        rps=total / duration_s,
        error_rate_pct=error_rate,
        p50_ms=float(latencies.quantile(0.50)),
        p95_ms=float(latencies.quantile(0.95)),
        p99_ms=float(latencies.quantile(0.99)),
        max_ms=float(latencies.max()),
        per_endpoint=per_endpoint,
    )


# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------
def print_text_report(result: AnalysisResult, spec: ThresholdSpec) -> None:
    line = "=" * 62
    print(line)
    print(f" Load Test Summary  ({result.total_requests} requests)")
    print(line)
    print(f" Duration          : {result.duration_s:,.2f} s")
    print(f" Throughput        : {result.rps:,.2f} req/s")
    print(f" Error rate        : {result.error_rate_pct:.2f} %")
    print(f" Latency p50       : {result.p50_ms:,.2f} ms")
    print(f" Latency p95       : {result.p95_ms:,.2f} ms")
    print(f" Latency p99       : {result.p99_ms:,.2f} ms")
    print(f" Latency max       : {result.max_ms:,.2f} ms")
    print(line)
    print(" Per-endpoint breakdown (slowest first):")
    print(result.per_endpoint.to_string())
    print(line)

    if spec.violations:
        print(" THRESHOLD VIOLATIONS:")
        for v in spec.violations:
            print(f"   - {v}")
    else:
        print(" All thresholds satisfied." if _has_thresholds(spec) else " No thresholds configured.")
    print(line)


def _has_thresholds(spec: ThresholdSpec) -> bool:
    return any(
        v is not None
        for v in (spec.latency_p95_ms, spec.latency_p99_ms, spec.error_rate_pct, spec.min_rps)
    )


def evaluate_thresholds(result: AnalysisResult, spec: ThresholdSpec) -> None:
    """Populate spec.violations in-place based on the analysis result."""
    spec.violations.clear()
    if spec.latency_p95_ms is not None and result.p95_ms > spec.latency_p95_ms:
        spec.violations.append(
            f"p95 {result.p95_ms:.1f}ms exceeds limit {spec.latency_p95_ms:.1f}ms"
        )
    if spec.latency_p99_ms is not None and result.p99_ms > spec.latency_p99_ms:
        spec.violations.append(
            f"p99 {result.p99_ms:.1f}ms exceeds limit {spec.latency_p99_ms:.1f}ms"
        )
    if spec.error_rate_pct is not None and result.error_rate_pct > spec.error_rate_pct:
        spec.violations.append(
            f"error rate {result.error_rate_pct:.2f}% exceeds limit {spec.error_rate_pct:.2f}%"
        )
    if spec.min_rps is not None and result.rps < spec.min_rps:
        spec.violations.append(
            f"throughput {result.rps:.2f} req/s below minimum {spec.min_rps:.2f} req/s"
        )


def render_chart(result: AnalysisResult, df: pd.DataFrame, out_path: Path) -> None:
    """Render a latency histogram + per-endpoint p95 bar chart to a PNG."""
    # Imported lazily so headless CI runs without matplotlib installed can still
    # produce the text report.
    import matplotlib

    matplotlib.use("Agg")  # Non-interactive backend; safe on servers.
    import matplotlib.pyplot as plt

    fig, (ax_hist, ax_bar) = plt.subplots(1, 2, figsize=(13, 5))

    ax_hist.hist(df["latency_ms"], bins=40, color="#4C72B0", edgecolor="white")
    ax_hist.axvline(result.p95_ms, color="orange", linestyle="--", label=f"p95 = {result.p95_ms:.0f}ms")
    ax_hist.axvline(result.p99_ms, color="red", linestyle="--", label=f"p99 = {result.p99_ms:.0f}ms")
    ax_hist.set_title("Latency distribution")
    ax_hist.set_xlabel("Latency (ms)")
    ax_hist.set_ylabel("Requests")
    ax_hist.legend()

    top = result.per_endpoint.head(10).iloc[::-1]
    ax_bar.barh(top.index, top["p95_ms"], color="#55A868")
    ax_bar.set_title("p95 latency by endpoint (top 10)")
    ax_bar.set_xlabel("p95 (ms)")

    fig.tight_layout()
    fig.savefig(out_path, dpi=120)
    plt.close(fig)
    print(f"Chart written to {out_path}")


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Analyze load-test results and enforce performance thresholds.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--file", required=True, type=Path, help="Path to results CSV")
    p.add_argument("--report", action="store_true", help="Render a PNG chart")
    p.add_argument("--report-path", type=Path, default=Path(DEFAULT_REPORT_PATH),
                   help="Where to write the PNG chart")
    p.add_argument("--thresholds", default=None,
                   help="Comma-separated thresholds, e.g. p95=200,p99=500,error_rate=1.0,rps=100")
    p.add_argument("--json", action="store_true", help="Emit machine-readable JSON summary")
    return p


def main(argv: Iterable[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    try:
        spec = ThresholdSpec.parse(args.thresholds)
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    try:
        df = load_results(args.file)
        result = analyze(df)
    except (FileNotFoundError, ValueError, KeyError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    evaluate_thresholds(result, spec)

    if args.json:
        import json

        print(json.dumps({
            "total_requests": result.total_requests,
            "duration_s": round(result.duration_s, 3),
            "rps": round(result.rps, 3),
            "error_rate_pct": round(result.error_rate_pct, 3),
            "p50_ms": round(result.p50_ms, 3),
            "p95_ms": round(result.p95_ms, 3),
            "p99_ms": round(result.p99_ms, 3),
            "max_ms": round(result.max_ms, 3),
            "violations": spec.violations,
        }, indent=2))
    else:
        print_text_report(result, spec)

    if args.report:
        try:
            render_chart(result, df, args.report_path)
        except ImportError:
            print("warning: matplotlib not installed; skipping chart", file=sys.stderr)

    # Exit code 1 signals a threshold breach so CI can fail the build.
    return 1 if spec.violations else 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

Threshold keys accepted by --thresholds: p95, p99, error_rate (percent), rps (minimum). Unknown keys raise a clear error rather than being silently ignored.

Usage Examples

1. Basic report against a Locust CSV:

$ python3 load_results.py --file locust_results.csv
==============================================================
 Load Test Summary  (12,480 requests)
==============================================================
 Duration          : 300.12 s
 Throughput        : 41.58 req/s
 Error rate        : 0.42 %
 Latency p50       :  68.31 ms
 Latency p95       : 184.02 ms
 Latency p99       : 402.77 ms
 Latency max       : 1,240.11 ms
==============================================================
 Per-endpoint breakdown (slowest first):
                    requests  errors  p50_ms  p95_ms  p99_ms
endpoint
/api/search             3120      18   142.10  388.20  712.40
/api/checkout           2080       9    88.40  210.55  401.10
...
==============================================================
 No thresholds configured.
==============================================================

2. CI gate with thresholds and a chart:

$ python3 load_results.py --file locust_results.csv \
    --report --thresholds p95=200,p99=500,error_rate=1.0,rps=40
...
==============================================================
 THRESHOLD VIOLATIONS:
   - p99 402.8ms exceeds limit 500.0ms
==============================================================
Chart written to load_report.png
$ echo $?
1

3. Machine-readable output for dashboards:

$ python3 load_results.py --file k6_results.csv --json
{
  "total_requests": 24800,
  "duration_s": 600.05,
  "rps": 41.33,
  "error_rate_pct": 0.12,
  "p50_ms": 55.2,
  "p95_ms": 142.8,
  "p99_ms": 310.4,
  "max_ms": 980.1,
  "violations": []
}

Error Handling

• Missing or empty file → FileNotFoundError / ValueError, printed to stderr, exit code 2.
• Malformed CSV → caught pd.errors.ParserError, wrapped with the offending path.
• Missing required columns → _resolve_column raises KeyError listing every alias it tried and the actual columns present, so the user knows exactly what to map.
• Non-numeric latency values → coerced with pd.to_numeric(errors="coerce") then dropped; if nothing survives, a ValueError is raised.
• Bad threshold syntax (p95:200, p95=abc) → ValueError with the offending token, exit code 2.
• Missing matplotlib → lazy import means text reporting still works; a warning is emitted instead of a crash.
• Threshold breach → exit code 1, distinct from 2 (usage/data error) so CI can differentiate.

Extending Ideas

• Prometheus/Grafana integration: emit the JSON summary to a /push endpoint or write a .prom file so metrics land in your existing dashboards.
• Historical comparison: store each run's summary in SQLite and add a --baseline flag that fails if p95 regresses by more than X% versus the previous build.
• Multi-file aggregation: accept --file multiple times to merge parallel Locust workers, then dedupe by request ID if present.
• Pluggable reporters: refactor print_text_report and render_chart behind a Reporter protocol so you can add HTML, Slack, or JUnit-XML output without touching the analysis core.