如何通过Gunicorn部署FastAPI并向DogStatsD上报指标？

一、解决Gunicorn+UvicornWorker的指标上报问题 #

由于UvicornWorker重写了Gunicorn的请求处理流程，导致Gunicorn内置的StatsD统计钩子无法触发，可通过以下两种方式解决：

1. 自定义Worker类补全统计逻辑 #

创建自定义Worker类，继承UvicornWorker并手动调用Gunicorn的StatsD统计方法，示例如下：

新建custom_worker.py文件：

from uvicorn.workers import UvicornWorker
from gunicorn.statsd import Statsd
from gunicorn.http import Request

class StatsDUvicornWorker(UvicornWorker):
    def handle_request(self, listener, req: Request, client_addr):
        # 初始化StatsD客户端
        if self.cfg.statsd_host:
            statsd = Statsd(self.cfg)
            # 上报请求计数
            statsd.incr('requests')
        
        # 执行原Uvicorn请求处理逻辑
        response = super().handle_request(listener, req, client_addr)
        
        # 上报响应状态码
        if self.cfg.statsd_host:
            status_code = getattr(req, 'status', 200)
            statsd.incr(f'response.status.{status_code}')
        return response

启动Gunicorn时指定该自定义Worker：

gunicorn \
  -b "0.0.0.0:8080" \
  -w 4 \
  -k custom_worker.StatsDUvicornWorker \
  --name my_app \
  --statsd_host "localhost:8125" \
  --dogstatsd-tags "app:my_app" \
  foo.bar:my_app

2. 在FastAPI应用中直接集成DataDog客户端 #

绕过Gunicorn的Worker限制，在应用层通过DataDog官方Python客户端手动上报指标，这种方式更灵活且不受部署Worker影响：

首先安装依赖：

pip install datadog

在FastAPI应用中添加全局中间件，自动捕获请求指标：

from fastapi import FastAPI, Request
from datadog import statsd
import time

app = FastAPI()

# 初始化DogStatsD客户端（兼容本地DataDog代理）
statsd.init(host='localhost', port=8125, namespace='my_app')

@app.middleware("http")
async def track_request_metrics(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    # 计算响应耗时（毫秒）
    duration = (time.perf_counter() - start_time) * 1000
    
    # 上报请求计数和耗时指标
    statsd.increment(
        'request.count',
        tags=[
            f'method:{request.method}',
            f'status:{response.status_code}',
            f'endpoint:{request.url.path}',
            'app:my_app'
        ]
    )
    statsd.histogram(
        'request.duration_ms',
        duration,
        tags=[
            f'method:{request.method}',
            f'status:{response.status_code}',
            f'endpoint:{request.url.path}',
            'app:my_app'
        ]
    )
    return response

# 你的业务路由
@app.get("/")
async def root():
    return {"message": "Hello World"}

二、其他生产级FastAPI部署的DataDog指标上报方案 #

1. Uvicorn直接部署+DataDog APM追踪 #

生产环境可直接用Uvicorn搭配进程管理器（如systemd、Supervisor），结合DataDog的ddtrace库实现自动指标上报和分布式追踪：

安装依赖：

pip install ddtrace

启动时通过ddtrace-run注入追踪：

DD_SERVICE=my_app DD_ENV=production DD_VERSION=1.0.0 ddtrace-run uvicorn foo.bar:my_app --host 0.0.0.0 --port 8080 --workers 4

该方式会自动采集请求速率、响应时间、错误率等核心指标，同时支持分布式链路追踪，无需手动编写统计逻辑。

2. Nginx反向代理+DataDog Nginx集成 #

若使用Nginx作为反向代理，可通过DataDog Agent的Nginx检查模块采集网关层指标，同时结合应用层的ddtrace上报业务指标：

• 配置Nginx开启stub_status模块，暴露状态接口
• 在DataDog Agent中启用Nginx检查，配置状态接口地址
• Agent会自动采集连接数、请求数、响应状态分布等网关指标

内容的提问来源于stack exchange，提问作者John