Chapter 20 · Capstone: Full-Stack Load Testing

Chapter goals: Tie together knowledge from Chapters 1–19 to execute a complete performance test workflow on a FastAPI application — from writing the script to distributed execution, analysis report, and optimization recommendations.

20.1 The System Under Test: A Simple Task Management API

Assume we have a FastAPI app deployed on staging:

# app.py — The FastAPI app under test (staging environment)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
import time


app = FastAPI()
db = {"tasks": [], "next_id": 1}  # Simulated database


class TaskCreate(BaseModel):
    title: str
    priority: int = 0


@app.get("/api/tasks")
async def list_tasks():
    return db["tasks"]


@app.post("/api/tasks")
async def create_task(task: TaskCreate):
    time.sleep(0.05)  # Simulate DB write latency
    item = {"id": db["next_id"], **task.model_dump()}
    db["next_id"] += 1
    db["tasks"].append(item)
    return item


@app.get("/api/tasks/{task_id}")
async def get_task(task_id: int):
    for t in db["tasks"]:
        if t["id"] == task_id:
            return t
    raise HTTPException(status_code=404)

20.2 Step 1: Write the Locust Script

Design task weights matching real user ratios — read-heavy, write-light is the most common pattern:

# locustfile.py
import random
from locust import HttpUser, task, between, LoadTestShape


class StagedShape(LoadTestShape):
    """Stepped pressure: 25 → 50 → 100 → 200 users"""
    stages = [
        {"duration": 60,  "users": 25,  "spawn_rate": 5},
        {"duration": 120, "users": 50,  "spawn_rate": 10},
        {"duration": 180, "users": 100, "spawn_rate": 15},
        {"duration": 240, "users": 200, "spawn_rate": 20},
    ]

    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stage["duration"]:
                return (stage["users"], stage["spawn_rate"])
        return None


class TaskApiUser(HttpUser):
    wait_time = between(0.5, 2)
    host = "https://staging.example.com"

    def on_start(self):
        """Setup: create some initial tasks for later reading"""
        for i in range(3):
            self.client.post("/api/tasks", json={
                "title": f"Task {i}", "priority": 0
            })

    @task(5)
    def list_tasks(self):
        resp = self.client.get("/api/tasks")
        if resp.status_code != 200:
            resp.failure(f"HTTP {resp.status_code}")

    @task(3)
    def get_task(self):
        task_id = random.randint(1, 3)
        resp = self.client.get(f"/api/tasks/{task_id}")
        if resp.status_code != 200:
            resp.failure(f"HTTP {resp.status_code}")

    @task(2)
    def create_task(self):
        resp = self.client.post("/api/tasks", json={
            "title": f"Task {random.randint(100, 999)}",
            "priority": random.choice([0, 1, 2])
        })
        if resp.status_code != 201:
            resp.failure(f"HTTP {resp.status_code}")

20.3 Step 2: Run the Test

locust -f locustfile.py \
  --headless \
  --users 200 \
  --spawn-rate 20 \
  --run-time 10m \
  --host https://staging.example.com \
  --csv results/final_run \
  --html results/report.html \
  --tracing retain-on-failure

20.4 Step 3: Analyze Results

Open results/report.html in a browser. Key things to check:

1. p95 response time — should be under your SLA threshold (e.g., 500ms);
2. Failure rate — should be 0%;
3. RPS curve — should stabilize at a plateau;
4. Per-endpoint breakdown — identify which endpoint is the bottleneck.

20.5 Step 4: Optimization Recommendations

Based on typical findings:

Finding	Likely Cause	Fix
p95 high on POST /api/tasks	time.sleep(0.05) blocks the event loop	Replace with asyncio.sleep(0.05)
RPS plateaus early	Single-process CPU bottleneck	Switch to FastHttpUser or multi-process
404 errors increasing	ID collision in simulated DB	Use a real database with auto-increment
Response time grows over time	Memory leak in simulated DB	Add periodic GC or switch to real DB

20.6 Chapter Summary

• Complete workflow: write script → staged pressure → headless run → CSV/HTML report → analyze → optimize;
• Task weights should mirror real user behavior (read-heavy, write-light);
• Always check percentiles (p95/p99), not just averages;
• Common bottlenecks: blocking sync code in async apps, single-process CPU limits, simulated data structures.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. In the capstone script, why are list_tasks weight 5, get_task weight 3, and create_task weight 2?

2. If p95 on POST /api/tasks is 5000ms but GET is 50ms, what is the most likely cause?

3. What is the purpose of the StagedShape load profile in the capstone?

4. Which report file is best for sharing results with non-technical stakeholders?

🛠️ Hands-On Exercises

1. Deploy the capstone FastAPI app locally; write and run a Locust script with StagedShape; generate an HTML report.
2. Introduce a bug: replace asyncio.sleep(0.05) with time.sleep(0.05) in the POST handler; re-run and compare p95 degradation.
3. Add a Prometheus scrape config; visualize the test in Grafana and set an alert on p95 > 500ms.

Course complete. You now have a production-grade load testing system built on Locust.