REST and JSON API Performance Testing: Guided Hands-On Workflow

1. Safety envelope

2. Start the disposable API fixture

Save as fixtures/api_fixture.py:

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
from pathlib import Path
import argparse
import json
import threading
import time

FIXTURE_VERSION = "prompt14-api-fixture-v1"
lock = threading.Lock()
seq = 0
resources = {}
event_log = None
metrics = {
    "requests": 0,
    "errors": 0,
    "by_operation": {},
}

def now_ms():
    return int(time.time() * 1000)

def log_event(event):
    if event_log is None:
        return
    with lock:
        with event_log.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, sort_keys=True) + "\n")

def metric(op, status):
    with lock:
        metrics["requests"] += 1
        metrics["by_operation"][op] = metrics["by_operation"].get(op, 0) + 1
        if status >= 400:
            metrics["errors"] += 1

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def send_json(self, status, payload):
        raw = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        self.end_headers()
        self.wfile.write(raw)

    def send_empty(self, status):
        self.send_response(status)
        self.send_header("Content-Length", "0")
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        self.end_headers()

    def read_json(self):
        length = int(self.headers.get("Content-Length", "0") or "0")
        if length > 65536:
            raise ValueError("body too large")
        raw = self.rfile.read(length) if length else b"{}"
        return json.loads(raw.decode("utf-8"))

    def finish_event(self, started, op, status, extra=None):
        metric(op, status)
        event = {
            "ts_ms": now_ms(),
            "operation": op,
            "method": self.command,
            "path": urlparse(self.path).path,
            "status": status,
            "service_wall_ms": now_ms() - started,
        }
        if extra:
            event.update(extra)
        log_event(event)

    def do_GET(self):
        started = now_ms()
        parsed = urlparse(self.path)
        path = parsed.path
        q = parse_qs(parsed.query)

        if path == "/health":
            status = 200
            self.send_json(status, {"status": "ok", "fixture_version": FIXTURE_VERSION})
            self.finish_event(started, "health", status)
            return

        if path == "/metrics":
            with lock:
                snapshot = {
                    "fixture_version": FIXTURE_VERSION,
                    "active_resources": len(resources),
                    "requests": metrics["requests"],
                    "errors": metrics["errors"],
                    "by_operation": dict(metrics["by_operation"]),
                }
            status = 200
            self.send_json(status, snapshot)
            self.finish_event(started, "metrics", status)
            return

        if path == "/v1/items":
            run_id = q.get("run_id", [""])[-1]
            owner = q.get("owner", [""])[-1]
            with lock:
                items = [
                    dict(v)
                    for v in resources.values()
                    if (not run_id or v["run_id"] == run_id)
                    and (not owner or v["owner"] == owner)
                ]
            status = 200
            self.send_json(status, {"status": "ok", "count": len(items), "items": items})
            self.finish_event(started, "list", status, {"run_id": run_id, "owner": owner})
            return

        prefix = "/v1/items/"
        if path.startswith(prefix):
            item_id = path[len(prefix):].split("/")[0]
            with lock:
                item = resources.get(item_id)
                snapshot = dict(item) if item else None
            if snapshot is None:
                status = 404
                self.send_json(status, {"status": "not_found", "id": item_id})
            else:
                time.sleep(0.012)
                status = 200
                self.send_json(status, {"status": "ok", "item": snapshot})
            self.finish_event(started, "read", status, {"id": item_id})
            return

        status = 404
        self.send_json(status, {"status": "not_found", "path": path})
        self.finish_event(started, "unknown_get", status)

    def do_POST(self):
        global seq
        started = now_ms()
        path = urlparse(self.path).path

        if path == "/v1/items":
            try:
                body = self.read_json()
            except Exception as exc:
                status = 400
                self.send_json(status, {"status": "invalid_json", "detail": str(exc)})
                self.finish_event(started, "create", status)
                return

            run_id = str(body.get("run_id", ""))
            owner = str(body.get("owner", ""))
            name = str(body.get("name", ""))
            value = body.get("value")

            if not run_id or not owner or not name or not isinstance(value, int):
                status = 422
                self.send_json(status, {
                    "status": "validation_error",
                    "required": ["run_id", "owner", "name", "integer value"],
                })
                self.finish_event(started, "create", status)
                return

            with lock:
                seq += 1
                item_id = f"ITEM-{seq:06d}"
                item = {
                    "id": item_id,
                    "run_id": run_id,
                    "owner": owner,
                    "name": name,
                    "value": value,
                    "version": 1,
                }
                resources[item_id] = item

            time.sleep(0.025)
            status = 201
            self.send_json(status, {"status": "created", "item": item})
            self.finish_event(started, "create", status, {
                "id": item_id, "run_id": run_id, "owner": owner
            })
            return

        if path.endswith("/inject-failure") and path.startswith("/v1/items/"):
            item_id = path.split("/")[3]
            with lock:
                exists = item_id in resources
            if not exists:
                status = 404
                self.send_json(status, {"status": "not_found", "id": item_id})
                self.finish_event(started, "inject_failure", status, {"id": item_id})
                return
            status = 500
            self.send_json(status, {
                "status": "synthetic_failure",
                "id": item_id,
                "message": "Intentional local checkpoint failure; resource remains for cleanup.",
            })
            self.finish_event(started, "inject_failure", status, {"id": item_id})
            return

        status = 404
        self.send_json(status, {"status": "not_found", "path": path})
        self.finish_event(started, "unknown_post", status)

    def do_PUT(self):
        started = now_ms()
        path = urlparse(self.path).path
        prefix = "/v1/items/"
        if not path.startswith(prefix):
            status = 404
            self.send_json(status, {"status": "not_found", "path": path})
            self.finish_event(started, "unknown_put", status)
            return

        item_id = path[len(prefix):]
        try:
            body = self.read_json()
        except Exception as exc:
            status = 400
            self.send_json(status, {"status": "invalid_json", "detail": str(exc)})
            self.finish_event(started, "update", status, {"id": item_id})
            return

        if "value" in body and not isinstance(body["value"], int):
            status = 422
            self.send_json(status, {"status": "validation_error", "field": "value"})
            self.finish_event(started, "update", status, {"id": item_id})
            return

        with lock:
            item = resources.get(item_id)
            if item is None:
                snapshot = None
            else:
                if "name" in body:
                    item["name"] = str(body["name"])
                if "value" in body:
                    item["value"] = body["value"]
                item["version"] += 1
                snapshot = dict(item)

        if snapshot is None:
            status = 404
            self.send_json(status, {"status": "not_found", "id": item_id})
        else:
            time.sleep(0.030)
            status = 200
            self.send_json(status, {"status": "updated", "item": snapshot})
        self.finish_event(started, "update", status, {"id": item_id})
        return

    def do_DELETE(self):
        started = now_ms()
        path = urlparse(self.path).path

        run_prefix = "/v1/runs/"
        if path.startswith(run_prefix) and path.endswith("/items"):
            run_id = path[len(run_prefix):-len("/items")].strip("/")
            with lock:
                doomed = [k for k, v in resources.items() if v["run_id"] == run_id]
                for k in doomed:
                    resources.pop(k, None)
            status = 200
            self.send_json(status, {
                "status": "cleanup_complete",
                "run_id": run_id,
                "deleted": len(doomed),
            })
            self.finish_event(started, "run_cleanup", status, {
                "run_id": run_id, "deleted": len(doomed)
            })
            return

        prefix = "/v1/items/"
        if path.startswith(prefix):
            item_id = path[len(prefix):]
            with lock:
                existed = resources.pop(item_id, None) is not None
            # Deliberately idempotent cleanup contract for the lab:
            # deleting an already-absent synthetic resource is still cleanup success.
            status = 204
            self.send_empty(status)
            self.finish_event(started, "delete", status, {"id": item_id, "existed": existed})
            return

        status = 404
        self.send_json(status, {"status": "not_found", "path": path})
        self.finish_event(started, "unknown_delete", status)

    def log_message(self, format, *args):
        return

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8000)
    parser.add_argument("--log", default="results/api-events.jsonl")
    args = parser.parse_args()

    global event_log
    event_log = Path(args.log).resolve()
    event_log.parent.mkdir(parents=True, exist_ok=True)
    event_log.write_text("", encoding="utf-8")

    server = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"fixture_version={FIXTURE_VERSION}")
    print(f"base_url=http://{args.host}:{args.port}")
    print(f"event_log={event_log}")
    server.serve_forever()

if __name__ == "__main__":
    main()

Start it:

python fixtures/api_fixture.py --host 127.0.0.1 --port 8000 --log results/api-events.jsonl

PowerShell uses the same Python command.

3. Inspect fixture/version/state before JMeter writes

PowerShell:

(Invoke-WebRequest -UseBasicParsing http://127.0.0.1:8000/health).Content
(Invoke-WebRequest -UseBasicParsing http://127.0.0.1:8000/metrics).Content

Bash:

curl --fail --silent http://127.0.0.1:8000/health
curl --fail --silent http://127.0.0.1:8000/metrics

Record fixture_version=prompt14-api-fixture-v1 and active_resources=0 for a clean start.

4. Parameterize the base URL

Use User Defined Variables or -J properties deliberately:

BASE_HOST = ${__P(BASE_HOST,127.0.0.1)}
BASE_PORT = ${__P(BASE_PORT,8000)}
RUN_ID    = ${__P(RUN_ID,RUN-LOCAL-001)}

HTTP Request Defaults:

• Protocol: http
• Server Name: ${BASE_HOST}
• Port: ${BASE_PORT}
• Implementation: HttpClient4/default current HTTP implementation
• Connect timeout: 1000 ms; response timeout: 2000 ms

5. Add JSON headers at lifecycle scope

HTTP Header Manager:

Accept: application/json
Content-Type: application/json

GET/DELETE bodies are empty, but the header is harmless in this local lab. In a real API, scope headers as narrowly as the contract requires.

6. Create a per-user synthetic resource

HTTP Request — Create Item:

• Method: POST
• Path: /v1/items
• Body Data:

{
  "run_id": "${RUN_ID}",
  "owner": "user-${__threadNum}",
  "name": "synthetic-${__threadNum}-${__time()}",
  "value": 1
}

Add Response Assertion: Response Code equals 201. Add JSON JMESPath Assertions:

• status equals created;
• item.run_id equals ${RUN_ID};
• item.owner equals user-${__threadNum};
• item.version equals 1.

7. Extract the generated ID

Under Create Item, add JSON JMESPath Extractor:

Add an assertion on JMeter variable ITEM_ID rejecting __NOT_FOUND__ before later samplers rely on it.

8. Read the correlated resource

HTTP Request — Read Item:

GET /v1/items/${ITEM_ID}

Assert HTTP 200, status=ok, item.id=${ITEM_ID}, item.owner=user-${__threadNum}, and item.value=1.

9. Update the same per-user resource

HTTP Request — Update Item:

PUT /v1/items/${ITEM_ID}

Body:
{
  "name": "updated-${__threadNum}",
  "value": 2
}

Assert HTTP 200, status=updated, same item.id, item.value=2, and item.version=2.

10. Delete per-user state

HTTP Request — Delete Item:

DELETE /v1/items/${ITEM_ID}

Assert HTTP 204. The fixture makes this delete idempotent for cleanup: repeating DELETE for an already-absent synthetic ID still returns 204. That simplifies safe retry of cleanup, not arbitrary write-operation retries.

11. Verify cleanup independently

curl --fail --silent "http://127.0.0.1:8000/v1/items?run_id=RUN-LOCAL-001"
curl --fail --silent http://127.0.0.1:8000/metrics

Expected run-scoped count after successful lifecycle loops: zero active items.

12. Recommended authoring tree

Test Plan
├── User Defined Variables / properties
├── HTTP Request Defaults
├── HTTP Header Manager
└── Thread Group — 1 user × 2 loops
    └── Transaction Controller — Resource Lifecycle
        ├── Create Item
        │   ├── Response/JSON assertions
        │   └── JSON JMESPath Extractor -> ITEM_ID
        ├── Read Item
        │   └── assertions
        ├── Update Item
        │   └── assertions
        └── Delete Item
            └── status assertion

Keep View Results Tree only for one-thread authoring; disable it for meaningful CLI runs.

13. Run the bounded lifecycle from CLI

jmeter.bat -n `
  -t plans\api-lifecycle.jmx `
  -JBASE_HOST=127.0.0.1 `
  -JBASE_PORT=8000 `
  -JRUN_ID=RUN-LOCAL-001 `
  -l results\lifecycle\results.jtl `
  -j results\lifecycle\jmeter.log `
  -e -o results\lifecycle\report

Bash uses jmeter with the same -J/-l/-j/-e/-o options. Report output must be empty/nonexistent before generation.

14. Analyze JTL and server events

tools/analyze_api_jtl.py:

import csv
import math
import sys
from collections import Counter, defaultdict
from pathlib import Path

path = Path(sys.argv[1] if len(sys.argv) > 1 else "results/run/results.jtl")
rows = list(csv.DictReader(path.open(encoding="utf-8")))
if not rows:
    raise SystemExit("No JTL rows found")

required = {"timeStamp", "elapsed", "label", "success", "responseCode"}
missing = required.difference(rows[0])
if missing:
    raise SystemExit(f"Missing JTL columns: {sorted(missing)}")

def pct(values, p):
    values = sorted(values)
    rank = max(1, math.ceil((p / 100.0) * len(values)))
    return values[rank - 1]

starts = [int(r["timeStamp"]) - int(r["elapsed"]) for r in rows]
ends = [int(r["timeStamp"]) for r in rows]
span_s = max((max(ends) - min(starts)) / 1000.0, 0.001)

print(f"samples={len(rows)}")
print(f"failures={sum(r['success'].strip().lower() != 'true' for r in rows)}")
print(f"observed_sample_rps={len(rows) / span_s:.3f}")
print(f"response_codes={dict(Counter(r['responseCode'] for r in rows))}")

by_label = defaultdict(list)
for row in rows:
    by_label[row["label"]].append(row)

for label, items in sorted(by_label.items()):
    elapsed = [int(r["elapsed"]) for r in items]
    failures = sum(r["success"].strip().lower() != "true" for r in items)
    print(
        f"label={label!r} samples={len(items)} failures={failures} "
        f"p50_ms={pct(elapsed,50)} p95_ms={pct(elapsed,95)}"
    )

print(
    "Configured load is not achieved load. Pair these JTL metrics with generator CPU/memory "
    "and the fixture event log/metrics before making target-capacity claims."
)

tools/analyze_api_events.py:

import json
import math
import sys
from collections import Counter, defaultdict
from pathlib import Path

path = Path(sys.argv[1] if len(sys.argv) > 1 else "results/api-events.jsonl")
events = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
if not events:
    raise SystemExit("No API events found")

def pct(values, p):
    values = sorted(values)
    rank = max(1, math.ceil((p / 100.0) * len(values)))
    return values[rank - 1]

print(f"events={len(events)}")
print(f"operations={dict(Counter(e['operation'] for e in events))}")
print(f"statuses={dict(Counter(e['status'] for e in events))}")

by_op = defaultdict(list)
for e in events:
    by_op[e["operation"]].append(e)

for op, items in sorted(by_op.items()):
    wall = [int(e["service_wall_ms"]) for e in items]
    print(
        f"operation={op!r} count={len(items)} "
        f"p50_service_wall_ms={pct(wall,50)} p95_service_wall_ms={pct(wall,95)}"
    )

created = {e.get("id") for e in events if e["operation"] == "create" and e.get("id")}
deleted = {e.get("id") for e in events if e["operation"] == "delete" and e.get("id")}
print(f"created_ids={len(created)}")
print(f"per_item_deleted_ids={len(deleted)}")
print(f"injected_failures={sum(e['operation']=='inject_failure' and e['status']==500 for e in events)}")

Use both. JTL gives client-observed operation metrics and assertion failures; server events give operation counts and handler service time.

15. Read-heavy versus write-lifecycle comparison

Do not compare different operation mixes as if they were the same benchmark. Build two explicit bounded profiles:

• Read-heavy: setup one synthetic item outside the timed loop, then 2 threads perform GET for 10 seconds; final scoped cleanup.
• Write lifecycle: 2 threads repeatedly Create → Read → Update → Delete for 10 seconds; each iteration owns its item.

Record operation-level p50/p95, achieved samples/s, generator CPU/memory, target service timing, error taxonomy, and active-resource count. Writes deliberately sleep slightly longer in the fixture, so the mix itself changes achieved throughput.

16. Challenge

Requirement: measure GET capacity without allowing all users to contend on one mutable record, but the target needs realistic IDs. Which design is stronger?

Pre-seed several immutable/read-only synthetic items (or allocate one per thread) before the timed phase, then run GET-only against those stable IDs. Do not include POST/DELETE in the timed read metric unless the objective is an end-to-end resource lifecycle.

Knowledge check

Official references and version notes

• Component Reference — HTTP Request, Header Manager, JSON/JSON JMESPath Extractor, JSON/JSON JMESPath Assertion, Response Assertion, Transaction Controller, and related core semantics.
• Elements of a Test Plan — scope/execution order and variable behavior.
• Getting Started — GUI authoring/debugging versus CLI load execution.
• Generating Dashboard Report — CSV result requirements, statistics, request summary, and error tables.
• Properties Reference — result-save fields including assertion failure messages.
• Best Practices — CLI execution, lean listeners, and injector validity.
• Apache JMeter downloads — current stable release and Java requirement.