public/fastapi-best-architecture
1
0
Fork 0
mirror of https://github.com/fastapi-practices/fastapi-best-architecture.git synced 2026-09-21 21:15:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add granian and celery metrics collection (#1057)

Browse Source 
* Add granian and celery metrics collection

* Update dashboards

* Add asyncio instrument

* Fix opera log finally
This commit is contained in:
Wu Clan
2026-02-03 20:44:05 +08:00
committed by GitHub
parent 6f1c27786d
commit 646a0ec2fc
19 changed files with 877 additions and 828 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/task/celery.py
+13
View File
@@ -4,10 +4,23 @@ import urllib.parse
import celery
import celery_aio_pool
from celery.signals import worker_process_init
from opentelemetry.instrumentation.celery import CeleryInstrumentor
from backend.app.task.tasks.beat import LOCAL_BEAT_SCHEDULE
from backend.common.enums import DataBaseType
from backend.core.conf import settings
from backend.core.path_conf import BASE_PATH
from backend.utils.otel import init_resource, init_tracer
@worker_process_init.connect(weak=False)
def init_celery_worker_tracing(*args, **kwargs) -> None:
"""初始化 Celery 追踪"""
if settings.GRAFANA_METRICS_ENABLE:
resource = init_resource('fba_celery_worker')
init_tracer(resource)
CeleryInstrumentor().instrument()
def find_task_packages() -> list[str]:
backend/app/task/database.py
+3 -5
View File
@@ -9,13 +9,11 @@ from sqlalchemy.orm import Session
from backend.app.task.model.result import Task, TaskExtended, TaskSet
from backend.app.task.session import SessionManager
"""
重写 from celery.backends.database 内部 DatabaseBackend 类，此类实现与模型配合不佳，导致 fba 创建表和 alembic 迁移困难
"""
class DatabaseBackend(BaseBackend):
"""The database result backend."""
"""
重写 celery.backends.database DatabaseBackend，此类实现与模型配合不佳，导致 fba 创建表和 alembic 迁移困难
"""
# ResultSet.iterate should sleep this much between each pool,
# to not bombard the database with queries.
backend/common/prometheus/instruments.py
+27 -25
View File
@@ -1,35 +1,37 @@
from prometheus_client import Counter, Gauge, Histogram
from backend.core.conf import settings
PROMETHEUS_INFO_GAUGE = (
Gauge(name='fba_app_info', documentation='fba 应用信息', labelnames=['app_name'])
.labels(app_name=settings.GRAFANA_APP_NAME)
.inc()
)
# 警告: 此值与以下位置强关联，修改必须同步更新，否则会导致 Grafana 指标数据查询失败：
# - deploy/backend/grafana/fba_datasource.yml
# - deploy/backend/grafana/dashboards/fba_server.json
PROMETHEUS_APP_NAME = 'fba_server'
PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE = Gauge(
'fba_request_in_progress',
'按方法和路径统计请求的衡量',
['app_name', 'method', 'path'],
name='fba_request_in_progress',
documentation='按方法和路径统计请求的衡量',
labelnames=['app_name', 'method', 'path'],
)
PROMETHEUS_REQUEST_COUNTER = Counter('fba_request_total', '按方法和路径统计请求总数', ['app_name', 'method', 'path'])
PROMETHEUS_RESPONSE_COUNTER = Counter(
'fba_response_total',
'按方法、路径和状态码统计响应总数',
['app_name', 'method', 'path', 'status_code'],
)
PROMETHEUS_EXCEPTION_COUNTER = Counter(
'fba_exception_total',
'按方法，路径和异常类型统计异常总数',
['app_name', 'method', 'path', 'exception_type'],
PROMETHEUS_REQUEST_COUNTER = Counter(
name='fba_request_total',
documentation='按方法和路径统计请求总数',
labelnames=['app_name', 'method', 'path'],
)
PROMETHEUS_REQUEST_COST_TIME_HISTOGRAM = Histogram(
'fba_request_cost_time',
'按方法和路径划分请求耗时的直方图（以 ms 为单位）',
['app_name', 'method', 'path'],
name='fba_request_cost_time',
documentation='按方法和路径划分请求耗时的直方图（以 ms 为单位）',
labelnames=['app_name', 'method', 'path'],
)
PROMETHEUS_EXCEPTION_COUNTER = Counter(
name='fba_exception_total',
documentation='按方法，路径和异常类型统计异常总数',
labelnames=['app_name', 'method', 'path', 'exception_type'],
)
PROMETHEUS_RESPONSE_COUNTER = Counter(
name='fba_response_total',
documentation='按方法、路径和状态码统计响应总数',
labelnames=['app_name', 'method', 'path', 'status_code'],
)
backend/core/conf.py
+2 -3
View File
@@ -238,8 +238,7 @@ class Settings(BaseSettings):
I18N_DEFAULT_LANGUAGE: str = 'zh-CN'
# Grafana
GRAFANA_METRICS: bool = False
GRAFANA_APP_NAME: str = 'fba_server'
GRAFANA_METRICS_ENABLE: bool = False
GRAFANA_OTLP_GRPC_ENDPOINT: str = 'fba_alloy:4317'
##################################################
@@ -310,7 +309,7 @@ class Settings(BaseSettings):
values['CELERY_BROKER'] = 'rabbitmq'
# Grafana
values['GRAFANA_METRICS'] = True
values['GRAFANA_METRICS_ENABLE'] = True
return values
backend/core/registrar.py
+2 -2
View File
@@ -109,7 +109,7 @@ def register_app() -> FastAPI:
register_page(app)
register_exception(app)
if settings.GRAFANA_METRICS:
if settings.GRAFANA_METRICS_ENABLE:
register_metrics(app)
return app
@@ -165,7 +165,7 @@ def register_middleware(app: FastAPI) -> None:
app.add_middleware(AccessMiddleware)
# ContextVar
plugins = [OtelTraceIdPlugin()] if settings.GRAFANA_METRICS else [RequestIdPlugin(validate=True)]
plugins = [OtelTraceIdPlugin()] if settings.GRAFANA_METRICS_ENABLE else [RequestIdPlugin(validate=True)]
app.add_middleware(
ContextMiddleware,
plugins=plugins,
backend/middleware/access_middleware.py
+14 -3
View File
@@ -5,6 +5,12 @@ from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoin
from backend.common.context import ctx
from backend.common.log import log
from backend.common.prometheus.instruments import (
PROMETHEUS_APP_NAME,
PROMETHEUS_REQUEST_COUNTER,
PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE,
)
from backend.core.conf import settings
from backend.utils.timezone import timezone
@@ -19,10 +25,11 @@ class AccessMiddleware(BaseHTTPMiddleware):
:param call_next: 下一个中间件或路由处理函数
:return:
"""
path = request.url.path if not request.url.query else request.url.path + '/' + request.url.query
path = request.url.path
method = request.method
if request.method != 'OPTIONS':
log.debug(f'--> 请求开始[{path}]')
if method != 'OPTIONS':
log.debug(f'--> 请求开始[{path if not request.url.query else request.url.path + "/" + request.url.query}]')
perf_time = time.perf_counter()
ctx.perf_time = perf_time
@@ -30,6 +37,10 @@ class AccessMiddleware(BaseHTTPMiddleware):
start_time = timezone.now()
ctx.start_time = start_time
if path.startswith(f'{settings.FASTAPI_API_V1_PATH}'):
PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE.labels(app_name=PROMETHEUS_APP_NAME, method=method, path=path).inc()
PROMETHEUS_REQUEST_COUNTER.labels(app_name=PROMETHEUS_APP_NAME, method=method, path=path).inc()
response = await call_next(request)
return response
backend/middleware/opera_log_middleware.py
+83 -83
View File
@@ -15,9 +15,9 @@ from backend.common.context import ctx
from backend.common.enums import StatusType
from backend.common.log import log
from backend.common.prometheus.instruments import (
PROMETHEUS_APP_NAME,
PROMETHEUS_EXCEPTION_COUNTER,
PROMETHEUS_REQUEST_COST_TIME_HISTOGRAM,
PROMETHEUS_REQUEST_COUNTER,
PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE,
PROMETHEUS_RESPONSE_COUNTER,
)
@@ -33,7 +33,7 @@ class OperaLogMiddleware(BaseHTTPMiddleware):
opera_log_queue: Queue = Queue(maxsize=settings.OPERA_LOG_QUEUE_MAXSIZE)
async def dispatch(self, request: Request, call_next: Any) -> Response:
async def dispatch(self, request: Request, call_next: Any) -> Response: # noqa: C901
"""
处理请求并记录操作日志
@@ -41,112 +41,112 @@ class OperaLogMiddleware(BaseHTTPMiddleware):
:param call_next: 下一个中间件或路由处理函数
:return:
"""
response = None
path = request.url.path
method = request.method
args = await self.get_request_args(request)
code = 200
msg = 'Success'
status = StatusType.enable
elapsed = 0
if path in settings.OPERA_LOG_PATH_EXCLUDE or not path.startswith(f'{settings.FASTAPI_API_V1_PATH}'):
try:
username = request.user.username
except AttributeError:
username = None
should_log_opera = (
path.startswith(f'{settings.FASTAPI_API_V1_PATH}') and path not in settings.OPERA_LOG_PATH_EXCLUDE
)
try:
response = await call_next(request)
else:
method = request.method
args = await self.get_request_args(request)
PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE.labels(
app_name=settings.GRAFANA_APP_NAME, method=method, path=path
).inc()
PROMETHEUS_REQUEST_COUNTER.labels(app_name=settings.GRAFANA_APP_NAME, method=method, path=path).inc()
except Exception as e:
elapsed = round((time.perf_counter() - ctx.perf_time) * 1000, 3)
log.error(f'请求异常: {e!s}')
# 执行请求
code = 200
msg = 'Success'
status = StatusType.enable
error = None
try:
response = await call_next(request)
elapsed = round((time.perf_counter() - ctx.perf_time) * 1000, 3)
for e in [
if should_log_opera:
code = getattr(e, 'code', StandardResponseCode.HTTP_500)
msg = getattr(e, 'msg', str(e))
status = StatusType.disable
if path.startswith(f'{settings.FASTAPI_API_V1_PATH}'):
PROMETHEUS_EXCEPTION_COUNTER.labels(
app_name=PROMETHEUS_APP_NAME,
method=method,
path=path,
exception_type=type(e).__name__,
).inc()
raise
else:
elapsed = round((time.perf_counter() - ctx.perf_time) * 1000, 3)
if should_log_opera:
# 检查上下文中的异常信息
for exception_key in [
'__request_http_exception__',
'__request_validation_exception__',
'__request_assertion_error__',
'__request_custom_exception__',
]:
exception = ctx.get(e)
exception = ctx.get(exception_key)
if exception:
code = exception.get('code')
msg = exception.get('msg')
status = StatusType.disable
log.error(f'请求异常: {msg}')
PROMETHEUS_EXCEPTION_COUNTER.labels(
app_name=settings.GRAFANA_APP_NAME,
method=method,
path=path,
exception_type=type(e).__name__,
).inc()
break
except Exception as e:
elapsed = round((time.perf_counter() - ctx.perf_time) * 1000, 3)
code = getattr(e, 'code', StandardResponseCode.HTTP_500) # 兼容 SQLAlchemy 异常用法
msg = getattr(e, 'msg', str(e)) # 不建议使用 traceback 模块获取错误信息，会暴漏代码信息
status = StatusType.disable
error = e
log.error(f'请求异常: {e!s}')
PROMETHEUS_EXCEPTION_COUNTER.labels(
app_name=settings.GRAFANA_APP_NAME, method=method, path=path, exception_type=type(e).__name__
).inc()
else:
PROMETHEUS_REQUEST_COST_TIME_HISTOGRAM.labels(
app_name=settings.GRAFANA_APP_NAME, method=method, path=path
).observe(elapsed, exemplar={'TraceID': get_request_trace_id()})
finally:
PROMETHEUS_RESPONSE_COUNTER.labels(
app_name=settings.GRAFANA_APP_NAME, method=method, path=path, status_code=code
).inc()
PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE.labels(
app_name=settings.GRAFANA_APP_NAME, method=method, path=path
).dec()
# 此信息只能在请求后获取
if path.startswith(f'{settings.FASTAPI_API_V1_PATH}'):
PROMETHEUS_REQUEST_COST_TIME_HISTOGRAM.labels(
app_name=PROMETHEUS_APP_NAME, method=method, path=path
).observe(amount=elapsed, exemplar={'TraceID': get_request_trace_id()})
finally:
# summary 只能在请求后获取
route = request.scope.get('route')
summary = route.summary or '' if route else ''
try:
# 此信息来源于 JWT 认证中间件
username = request.user.username
except AttributeError:
username = None
# 日志记录
log.debug(f'接口摘要：[{summary}]')
log.debug(f'请求地址：[{ctx.ip}]')
log.debug(f'请求参数：{args}')
log.info(f'{ctx.ip: <15} | {request.method: <8} | {code!s: <6} | {path} | {elapsed:.3f}ms')
if request.method != 'OPTIONS':
log.debug('<-- 请求结束')
# 日志创建
opera_log_in = CreateOperaLogParam(
trace_id=get_request_trace_id(),
username=username,
method=method,
title=summary,
path=path,
ip=ctx.ip,
country=ctx.country,
region=ctx.region,
city=ctx.city,
user_agent=ctx.user_agent,
os=ctx.os,
browser=ctx.browser,
device=ctx.device,
args=args,
status=status,
code=str(code),
msg=msg,
cost_time=elapsed, # 可能和日志存在微小差异（可忽略）
opera_time=ctx.start_time,
)
await self.opera_log_queue.put(opera_log_in)
if path.startswith(f'{settings.FASTAPI_API_V1_PATH}'):
log.info(f'{ctx.ip: <15} | {method: <8} | {code!s: <6} | {path} | {elapsed:.3f}ms')
# 错误抛出
if error:
raise error from None
if should_log_opera and request.method != 'OPTIONS':
opera_log_in = CreateOperaLogParam(
trace_id=get_request_trace_id(),
username=username,
method=method,
title=summary,
path=path,
ip=ctx.ip,
country=ctx.country,
region=ctx.region,
city=ctx.city,
user_agent=ctx.user_agent,
os=ctx.os,
browser=ctx.browser,
device=ctx.device,
args=args,
status=status,
code=str(code),
msg=msg,
cost_time=elapsed,
opera_time=ctx.start_time,
)
await self.opera_log_queue.put(opera_log_in)
if path.startswith(f'{settings.FASTAPI_API_V1_PATH}'):
PROMETHEUS_RESPONSE_COUNTER.labels(
app_name=PROMETHEUS_APP_NAME, method=method, path=path, status_code=code
).inc()
PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE.labels(
app_name=PROMETHEUS_APP_NAME, method=method, path=path
).dec()
return response
backend/utils/otel.py
+28 -15
View File
@@ -3,6 +3,7 @@ from opentelemetry import _logs, metrics, trace
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.asyncio import AsyncioInstrumentor
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.logging import LoggingInstrumentor
@@ -17,12 +18,31 @@ from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from backend.common.log import log, request_id_filter
from backend.common.prometheus.instruments import PROMETHEUS_APP_NAME
from backend.core.conf import settings
from backend.database.db import async_engine
from backend.database.redis import redis_client
def _init_tracer(resource: Resource) -> None:
def init_resource(service_name: str) -> Resource:
"""
初始化资源
:param service_name: 服务名称
:return:
"""
from backend import __version__
return Resource(
attributes={
'service.name': service_name,
'service.version': __version__,
'deployment.environment': settings.ENVIRONMENT,
},
)
def init_tracer(resource: Resource) -> None:
"""
初始化追踪器
@@ -36,7 +56,7 @@ def _init_tracer(resource: Resource) -> None:
trace.set_tracer_provider(tracer_provider)
def _init_metrics(resource: Resource) -> None:
def init_metrics(resource: Resource) -> None:
"""
初始化指标
@@ -52,7 +72,7 @@ def _init_metrics(resource: Resource) -> None:
metrics.set_meter_provider(meter_provider)
def _init_logging(resource: Resource) -> None:
def init_logging(resource: Resource) -> None:
"""
初始化日志
@@ -81,20 +101,13 @@ def init_otel(app: FastAPI) -> None:
:param app: FastAPI 应用实例
:return:
"""
from backend import __version__
resource = init_resource(PROMETHEUS_APP_NAME)
resource = Resource(
attributes={
'service.name': settings.GRAFANA_APP_NAME,
'service.version': __version__,
'deployment.environment': settings.ENVIRONMENT,
},
)
_init_tracer(resource)
# _init_metrics(resource)
_init_logging(resource)
init_tracer(resource)
init_metrics(resource)
init_logging(resource)
AsyncioInstrumentor().instrument()
LoggingInstrumentor().instrument(set_logging_format=True)
SQLAlchemyInstrumentor().instrument(engine=async_engine.sync_engine)
RedisInstrumentor.instrument_client(redis_client) # type: ignore