学习

Lab 05: Introduction to RESTful Services (Flask, FastAPI) · Jimmy Huang's Blog

plain text

本次压测结果显示，该 Todo API 在 100 并发用户下表现最好，RPS 约为 33，响应时间低，只有极少量失败，可视为基本稳定。
在 150 并发时，系统仍可正常运行，RPS 提升到约 49，但失败数开始增加。
在 200 并发时，RPS 约为 66，但失败进一步上升，说明系统已经接近当前配置下的承载上限。
在 300 到 400 并发时，服务虽然仍能运行，但失败更加明显，因此不能视为稳定承载范围。

plain text

为了处理更多并发用户并提升 RPS，第一步是把 FastAPI 自带的开发服务器替换成适合生产环境的 ASGI 部署方式，例如使用 Uvicorn，或者使用带有多个 worker 的 Gunicorn。

应用程序也不应该只把所有 todo 项存放在内存里，而是应该改用正规的数据库，这样可以让请求处理更加可靠，也更容易扩展。

另外，我们还可以通过使用更大的虚拟机、优化 API 代码、减少不必要的开销，以及对高频请求的数据加入缓存，来进一步提升性能。

同时，持续监控 CPU、内存、延迟和错误率也很重要，这样就能尽早发现瓶颈并及时修复。

bash

scp -i <path-to-your-original-key> ~/Downloads/clouds.yaml ubuntu@<client-vm-public-ip>:~
scp -i <path-to-your-original-key> ~/Downloads/FIT5225-Cloud-openrc.sh ubuntu@<client-vm-public-ip>:~
scp -i <path-to-your-original-key> ~/Downloads/openstack-api.py ubuntu@<client-vm-public-ip>:~

python

@app.delete('/todo/api/v1.0/tasks/{task_id}', response_model=dict)
async def delete_task(task_id: int, username: str = Depends(get_current_username)):
    for index, todo in enumerate(tasks):
        if todo.id == task_id:
            del tasks[index]
            return {"message": f"Todo{task_id} has been deleted."}
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Todo is not found"
    )

bash

curl -X PUT --user fastapi:FIT5225 \
  -d '{"title":"new-new", "description":"I am new"}' \
  --header "Content-Type: application/json" \
  http://localhost:8000/todo/api/v1.0/tasks/3 | jq

python

from locust import HttpUser, task, between
from requests.auth import HTTPBasicAuth

auth = HTTPBasicAuth("fastapi", "FIT5225")

class UserBehavior(HttpUser):
    wait_time = between(1, 5)

    @task(10)
    def get_tasks(self):
        self.client.get("/todo/api/v1.0/tasks", auth=auth)

    @task(5)
    def get_task(self):
        self.client.get("/todo/api/v1.0/tasks/1", auth=auth)

    @task(3)
    def create_task(self):
        self.client.post("/todo/api/v1.0/tasks", json={"title": "New Task"}, auth=auth)

    @task(2)
    def update_task(self):
        self.client.put("/todo/api/v1.0/tasks/1", json={"title": "Updated Task"}, auth=auth)

import secrets

from typing import Annotated, List, Optional
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.responses import PlainTextResponse, JSONResponse
from fastapi.exceptions import RequestValidationError

from pydantic import BaseModel

class TodoItem(BaseModel):
    id: int
    title: str | None = None
    description: str | None = ""
    done: bool = False


app = FastAPI()
security = HTTPBasic()

# default username is fastapi, password is FIT5225. In practice, you may need other authentication methods
def get_current_username(
    credentials: Annotated[HTTPBasicCredentials, Depends(security)],
):
    current_username_bytes = credentials.username.encode("utf8")
    correct_username_bytes = b"fastapi"
    is_correct_username = secrets.compare_digest(
        current_username_bytes, correct_username_bytes
    )
    current_password_bytes = credentials.password.encode("utf8")
    correct_password_bytes = b"FIT5225"
    is_correct_password = secrets.compare_digest(
        current_password_bytes, correct_password_bytes
    )
    if not (is_correct_username and is_correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Basic"},
        )
    return credentials.username

# Initialize some sample data in memeory
tasks = [
    TodoItem(id=1, title='Buy groceries', description='Milk, Cheese, Pizza, Fruit, Tylenol', done=False),
    TodoItem(id=2, title='Learn Python', description='Need to find a good Python tutorial on the web', done=False)
]
next_id = 3

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    if exc.status_code == status.HTTP_401_UNAUTHORIZED:
        return JSONResponse(
            status_code=status.HTTP_401_UNAUTHORIZED,  # For API calls, return 403 instead of 401 to prevent browsers from displaying the default auth dialog

            content={"message": "Unauthorized access"},
            headers={"WWW-Authenticate": "Basic"},
        )
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
        headers=exc.headers,
    )
                    

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return PlainTextResponse(str(exc), status_code=400)

# It's a good practice to have a public API endpoint allowing enduser to check status of the API
@app.get('/api/health')
async def public():
    return {"message": "The todo API is alive."}


@app.get('/todo/api/v1.0/tasks', response_model=List[TodoItem] )
async def get_tasks(username: str = Depends(get_current_username)):
    return tasks


@app.get('/todo/api/v1.0/tasks/{task_id}', response_model=TodoItem)
async def get_task(task_id:int, username: str = Depends(get_current_username)):
    for todo in tasks:
        if todo.id == task_id:
            return todo
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo is not found")


@app.post('/todo/api/v1.0/tasks', response_model=TodoItem, status_code=status.HTTP_201_CREATED)
async def create_task(todo:dict, username: str = Depends(get_current_username)):
    global next_id
    todo_item = TodoItem(id=next_id, **todo)
    tasks.append(todo_item)
    next_id += 1
    return todo_item


@app.put('/todo/api/v1.0/tasks/{task_id}', response_model=TodoItem)
async def update_task(task_id:int, updated_todo:dict, username: str = Depends(get_current_username)):
    for index, todo in enumerate(tasks):
        if todo.id == task_id:
            updated_todo = TodoItem(id=task_id, **updated_todo)
            tasks[index] = updated_todo
            return updated_todo
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo is not found")

    
@app.delete('/todo/api/v1.0/tasks/{task_id}', response_model=dict)
async def delete_task(task_id:int, username: str = Depends(get_current_username)):
    # please implement
    return {"message": "Not implemented."}

javascript

import openstack
import os
import errno

IMAGE_NAME = "NeCTAR Ubuntu 22.04 LTS (Jammy) amd64"
FLAVOR_NAME = "mon.training.small"
KEYPAIR_NAME = "openstacksdk-example"
SERVER_NAME = "demo-jay"
VM_DEFAULT_USERNAME = "ubuntu"
NETWORK_NAME = "Classic Provider"
SSH_DIR = '{home}/.ssh'.format(home=os.path.expanduser("~"))
PRIVATE_KEYPAIR_FILE = '{ssh_dir}/id_rsa.{key}'.format(ssh_dir=SSH_DIR, key=KEYPAIR_NAME)

def create_keypair(conn):
    keypair = conn.compute.find_keypair(KEYPAIR_NAME)

    if not keypair:
        print("Creating Key Pair:")

        keypair = conn.compute.create_keypair(name=KEYPAIR_NAME)

        print(keypair)

        try:
            os.mkdir(SSH_DIR)
        except OSError as e:
            if e.errno != errno.EEXIST:
                raise e

        with open(PRIVATE_KEYPAIR_FILE, 'w') as f:
            f.write("%s" % keypair.private_key)

        os.chmod(PRIVATE_KEYPAIR_FILE, 0o400)

    return keypair

def attach_security_group(conn, server, sg="_default"):
    print(f"Attaching security group: {sg} to {server.name}")

    # Fetch existing security groups attached to the server
    existing_sgs = conn.compute.fetch_server_security_groups(server)

    # Remove the default security group if it exists
    for existing_sg in existing_sgs:
        if existing_sg == "default":
            print(f"Removing default security group from {server.name}")
            conn.compute.remove_security_group_from_server(server=server, security_group="default")

    # Add the desired security group
    for security_group in conn.network.security_groups():
        if security_group.name.lower() == sg.lower():
            conn.compute.add_security_group_to_server(server=server, security_group=sg)
            print(f"Security group {sg} attached to {server.name}")
            return

    print(f"Security group {sg} not found.")
def create_server(conn):
    print("Creating Server {0}:".format(SERVER_NAME))

    image = conn.compute.find_image(IMAGE_NAME)
    flavor = conn.compute.find_flavor(FLAVOR_NAME)
    network = conn.network.find_network(NETWORK_NAME)
    keypair = create_keypair(conn)

    server = conn.compute.create_server(
        name=SERVER_NAME, image_id=image.id, flavor_id=flavor.id,
        networks=[{"uuid": network.id}], key_name=keypair.name)

    server = conn.compute.wait_for_server(server)
    # **Call attach_security_group here**
    attach_security_group(conn, server, sg="_default")
    print("ssh -i {key} {username}@{ip}".format(
        key=PRIVATE_KEYPAIR_FILE, username=VM_DEFAULT_USERNAME,
        ip=server.access_ipv4))

# Initialize and turn on debug logging
openstack.enable_logging(debug=True)

# Initialize cloud
conn = openstack.connect(cloud='envvars')

create_server(conn)

Lab 05: Introduction to RESTful Services (Flask, FastAPI)

Overview

Goal

Lab 05: Introduction to RESTful Services (Flask, FastAPI)

Overview

Goal

Task 0: Create a VM in your pt-xxxx Nectar project

Step 0.1：在个人项目里创建客户端 VM

Step 0.2：SSH 登录并确认机器可用

Task 1: Using OpenStack RESTful API to Create Resources

Step 1.1：下载配置文件并重置 OpenStack API password

Step 1.2：把文件上传到客户端 VM

Step 1.3：安装 Python、pip、venv 和 OpenStack SDK

使用 venv继续安装

Step 1.4：放置 clouds.yaml 并导入 OpenRC 配置

Step 1.5：修改 openstack-api.py

Step 1.6：运行脚本并验证资源是否真的创建成功

Task 2: Creating a Simple Todo Application

Step 2.1：把 todo-app.py 传到新创建的 VM

推荐做法：两跳传输

先从本地传到客户端 VM

再从客户端 VM 传到新 VM

Step 2.2：安装应用依赖

Step 2.3：先补上缺失的 DELETE 接口

Task 3: Testing Web Services

Step 3.1：启动 FastAPI 服务

Step 3.2：开放 8000 端口并访问自动生成的 API 文档

Step 3.3：先获取当前任务列表（GET）

Step 3.4：按 Lab 原样尝试创建新任务，先观察 400 错误（POST）

Step 3.5：加上 Content-Type，重新创建任务（POST）

Step 3.6：更新刚才创建的任务（PUT）

Step 3.7：获取单个任务（GET one task）

Step 3.8：删除单个任务（DELETE）

Task 4: Web Services Load Testing with Locust

Step 4.1：先确认你的 Todo 服务能从公网访问

浏览器方式

命令行方式（更稳）

Step 4.2：在你本地电脑创建 locustfile.py

Step 4.3：如果本机还没装 Locust，先安装

Step 4.4：启动 Locust 并打开 Web UI

Step 4.5：逐步提高并发并记录结果

80 user

100 user

150 user

200 user

Step 4.6：Homework / Reflection

问题 1：How many concurrent users / RPS can your application handle?

问题 2：How can we handle more concurrent users and improve RPS?

Summary

本周必须理解的知识点

关键概念对比

clouds.yaml vs OpenRC

localhost vs 0.0.0.0

GET / POST / PUT / DELETE

手动点 Dashboard vs 程序化调 API

实验问题回答

为什么第一次 POST 会得到 400 Bad Request？

为什么更新和删除都要写成 /tasks/{id}？

如果要支撑更多并发和更高 RPS，最先应该做什么？

老师可能会问什么

Task 0: Create a VM in your `pt-xxxx` Nectar project

Step 1.4：放置 `clouds.yaml` 并导入 OpenRC 配置

Step 1.5：修改 `openstack-api.py`

Step 2.1：把 `todo-app.py` 传到新创建的 VM

Step 3.5：加上 `Content-Type`，重新创建任务（POST）

Step 4.2：在你本地电脑创建 `locustfile.py`

`clouds.yaml` vs OpenRC

`localhost` vs `0.0.0.0`

`GET` / `POST` / `PUT` / `DELETE`

为什么第一次 POST 会得到 `400 Bad Request`？

为什么更新和删除都要写成 `/tasks/{id}`？