Add the Grafana observability suite (#961)

* Add the Grafana observability suite

* Update configs

* Update docker script directory structure

* Fix the otel trace id

* Add grafana ini

* Fix some configs and loguru integration

* Add the celery grafana

* Update Grafana dashboards

* Update configs

* Fix issues with the panel

* Update grafana configs

* Update grafana dashboards

* Optimized panel styles

* Add sqlalchemy traces

* Fix the CORS

* Update the grafana query and config

* Update grafana status is off by default

Author: wu-clan

Dockerfile

@@ -36,14 +36,14 @@ COPY --from=builder /fba /fba
 
 COPY --from=builder /usr/local /usr/local
 
-COPY deploy/backend/supervisord.conf /etc/supervisor/supervisord.conf
+COPY deploy/backend/supervisor/supervisord.conf /etc/supervisor/supervisord.conf
 
 WORKDIR /fba/backend
 
 # === FastAPI server image ===
 FROM base_server AS fba_server
 
-COPY deploy/backend/fba_server.conf /etc/supervisor/conf.d/
+COPY deploy/backend/supervisor/fba_server.conf /etc/supervisor/conf.d/
 
 RUN mkdir -p /var/log/fba
 
@@ -54,7 +54,7 @@ CMD ["supervisord", "-c", "/etc/supervisor/supervisord.conf"]
 # === Celery Worker image ===
 FROM base_server AS fba_celery_worker
 
-COPY deploy/backend/fba_celery_worker.conf /etc/supervisor/conf.d/
+COPY deploy/backend/supervisor/fba_celery_worker.conf /etc/supervisor/conf.d/
 
 RUN mkdir -p /var/log/fba
 
@@ -63,7 +63,7 @@ CMD ["supervisord", "-c", "/etc/supervisor/supervisord.conf"]
 # === Celery Beat image ===
 FROM base_server AS fba_celery_beat
 
-COPY deploy/backend/fba_celery_beat.conf /etc/supervisor/conf.d/
+COPY deploy/backend/supervisor/fba_celery_beat.conf /etc/supervisor/conf.d/
 
 RUN mkdir -p /var/log/fba
 
@@ -72,7 +72,7 @@ CMD ["supervisord", "-c", "/etc/supervisor/supervisord.conf"]
 # === Celery Flower image ===
 FROM base_server AS fba_celery_flower
 
-COPY deploy/backend/fba_celery_flower.conf /etc/supervisor/conf.d/
+COPY deploy/backend/supervisor/fba_celery_flower.conf /etc/supervisor/conf.d/
 
 RUN mkdir -p /var/log/fba
 

backend/app/task/celery.py

@@ -52,6 +52,8 @@ def init_celery() -> celery.Celery:
         task_track_started=True,
         enable_utc=False,
         timezone=settings.DATETIME_TIMEZONE,
+        worker_send_task_events=True,
+        task_send_sent_event=True,
     )
 
     # 在 Celery 中设置此参数无效

backend/common/exception/exception_handler.py

@@ -2,6 +2,7 @@
 from fastapi.exceptions import RequestValidationError
 from pydantic import ValidationError
 from starlette.exceptions import HTTPException
+from starlette.middleware.cors import CORSMiddleware
 from uvicorn.protocols.http.h11_impl import STATUS_PHRASES
 
 from backend.common.context import ctx
@@ -75,7 +76,7 @@ async def _validation_exception_handler(exc: RequestValidationError | Validation
     return MsgSpecJSONResponse(status_code=StandardResponseCode.HTTP_422, content=content)
 
 
-def register_exception(app: FastAPI) -> None:
+def register_exception(app: FastAPI) -> None:  # noqa: C901
     @app.exception_handler(HTTPException)
     async def http_exception_handler(request: Request, exc: HTTPException):
         """
@@ -194,3 +195,55 @@ async def all_unknown_exception_handler(request: Request, exc: Exception):
             status_code=StandardResponseCode.HTTP_500,
             content=content,
         )
+
+    if settings.MIDDLEWARE_CORS:
+
+        @app.exception_handler(StandardResponseCode.HTTP_500)
+        async def cors_custom_code_500_exception_handler(request: Request, exc: BaseExceptionError | Exception):
+            """
+            跨域自定义 500 异常处理
+
+            :param request: FastAPI 请求对象
+            :param exc: 自定义异常
+            :return:
+            """
+            if isinstance(exc, BaseExceptionError):
+                content = {
+                    'code': exc.code,
+                    'msg': exc.msg,
+                    'data': exc.data,
+                }
+            else:
+                if settings.ENVIRONMENT == 'dev':
+                    content = {
+                        'code': StandardResponseCode.HTTP_500,
+                        'msg': str(exc),
+                        'data': None,
+                    }
+                else:
+                    res = response_base.fail(res=CustomResponseCode.HTTP_500)
+                    content = res.model_dump()
+            content.update(trace_id=get_request_trace_id())
+            response = MsgSpecJSONResponse(
+                status_code=exc.code if isinstance(exc, BaseExceptionError) else StandardResponseCode.HTTP_500,
+                content=content,
+                background=exc.background if isinstance(exc, BaseExceptionError) else None,
+            )
+            origin = request.headers.get('origin')
+            if origin:
+                cors = CORSMiddleware(
+                    app=app,
+                    allow_origins=settings.CORS_ALLOWED_ORIGINS,
+                    allow_credentials=True,
+                    allow_methods=['*'],
+                    allow_headers=['*'],
+                    expose_headers=settings.CORS_EXPOSE_HEADERS,
+                )
+                response.headers.update(cors.simple_headers)
+                has_cookie = 'cookie' in request.headers
+                if cors.allow_all_origins and has_cookie:
+                    response.headers['Access-Control-Allow-Origin'] = origin
+                elif not cors.allow_all_origins and cors.is_allowed_origin(origin=origin):
+                    response.headers['Access-Control-Allow-Origin'] = origin
+                    response.headers.add_vary_header('Origin')
+            return response

backend/common/log.py

@@ -47,6 +47,13 @@ def default_formatter(record: logging.LogRecord) -> str:
     return settings.LOG_FORMAT if settings.LOG_FORMAT.endswith('\n') else f'{settings.LOG_FORMAT}\n'
 
 
+def request_id_filter(record: logging.LogRecord) -> logging.LogRecord:
+    """请求 ID 过滤器"""
+    rid = get_request_trace_id()
+    record['request_id'] = rid[: settings.TRACE_ID_LOG_LENGTH]
+    return record
+
+
 def setup_logging() -> None:
     """
     设置日志处理器
@@ -75,12 +82,6 @@ def setup_logging() -> None:
     # 移除 loguru 默认处理器
     logger.remove()
 
-    # request_id 过滤器
-    def request_id_filter(record: logging.LogRecord) -> logging.LogRecord:
-        rid = get_request_trace_id()
-        record['request_id'] = rid[: settings.TRACE_ID_LOG_LENGTH]
-        return record
-
     # 配置 loguru 处理器
     logger.configure(
         handlers=[

backend/common/prometheus/__init__.py

@@ -0,0 +1,35 @@
+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()
+)
+
+PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE = Gauge(
+    'fba_request_in_progress',
+    '按方法和路径统计请求的衡量',
+    ['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_COST_TIME_HISTOGRAM = Histogram(
+    'fba_request_cost_time',
+    '按方法和路径划分请求耗时的直方图（以 ms 为单位）',
+    ['app_name', 'method', 'path'],
+)

backend/common/prometheus/instruments.py

@@ -23,7 +23,7 @@ class Settings(BaseSettings):
 
     # FastAPI
     FASTAPI_API_V1_PATH: str = '/api/v1'
-    FASTAPI_TITLE: str = 'FastAPI'
+    FASTAPI_TITLE: str = 'fba'
     FASTAPI_DESCRIPTION: str = 'FastAPI Best Architecture'
     FASTAPI_DOCS_URL: str = '/docs'
     FASTAPI_REDOC_URL: str = '/redoc'
@@ -215,6 +215,11 @@ class Settings(BaseSettings):
     # I18n 配置
     I18N_DEFAULT_LANGUAGE: str = 'zh-CN'
 
+    # Grafana
+    GRAFANA_METRICS: bool = False
+    GRAFANA_APP_NAME: str = 'fba_server'
+    GRAFANA_OTLP_GRPC_ENDPOINT: str = 'fba_alloy:4317'
+
     ##################################################
     # [ App ] task
     ##################################################

backend/core/conf.py

@@ -9,10 +9,10 @@
 from fastapi import Depends, FastAPI
 from fastapi_limiter import FastAPILimiter
 from fastapi_pagination import add_pagination
+from prometheus_client import make_asgi_app
 from starlette.middleware.authentication import AuthenticationMiddleware
 from starlette.middleware.cors import CORSMiddleware
 from starlette.staticfiles import StaticFiles
-from starlette.types import ASGIApp
 from starlette_context.middleware import ContextMiddleware
 from starlette_context.plugins import RequestIdPlugin
 
@@ -33,8 +33,10 @@
 from backend.utils.demo_site import demo_site
 from backend.utils.health_check import ensure_unique_route_names, http_limit_callback
 from backend.utils.openapi import simplify_operation_ids
+from backend.utils.otel import init_otel
 from backend.utils.serializers import MsgSpecJSONResponse
 from backend.utils.snowflake import snowflake
+from backend.utils.trace_id import OtelTraceIdPlugin
 
 
 @asynccontextmanager
@@ -76,22 +78,7 @@ async def register_init(app: FastAPI) -> AsyncGenerator[None, None]:
 def register_app() -> FastAPI:
     """注册 FastAPI 应用"""
 
-    class MyFastAPI(FastAPI):
-        if settings.MIDDLEWARE_CORS:
-            # Related issues
-            # https://github.com/fastapi/fastapi/discussions/7847
-            # https://github.com/fastapi/fastapi/discussions/8027
-            def build_middleware_stack(self) -> ASGIApp:
-                return CORSMiddleware(
-                    super().build_middleware_stack(),
-                    allow_origins=settings.CORS_ALLOWED_ORIGINS,
-                    allow_credentials=True,
-                    allow_methods=['*'],
-                    allow_headers=['*'],
-                    expose_headers=settings.CORS_EXPOSE_HEADERS,
-                )
-
-    app = MyFastAPI(
+    app = FastAPI(
         title=settings.FASTAPI_TITLE,
         version=__version__,
         description=settings.FASTAPI_DESCRIPTION,
@@ -111,6 +98,9 @@ def build_middleware_stack(self) -> ASGIApp:
     register_page(app)
     register_exception(app)
 
+    if settings.GRAFANA_METRICS:
+        register_metrics(app)
+
     return app
 
 
@@ -164,15 +154,29 @@ def register_middleware(app: FastAPI) -> None:
     app.add_middleware(AccessMiddleware)
 
     # ContextVar
+    plugins = [OtelTraceIdPlugin()] if settings.GRAFANA_METRICS else [RequestIdPlugin(validate=True)]
     app.add_middleware(
         ContextMiddleware,
-        plugins=[RequestIdPlugin(validate=True)],
+        plugins=plugins,
         default_error_response=MsgSpecJSONResponse(
             content={'code': StandardResponseCode.HTTP_400, 'msg': 'BAD_REQUEST', 'data': None},
             status_code=StandardResponseCode.HTTP_400,
         ),
     )
 
+    # CORS
+    # https://github.com/fastapi-practices/fastapi_best_architecture/pull/789/changes
+    # https://github.com/open-telemetry/opentelemetry-python-contrib/issues/4031
+    if settings.MIDDLEWARE_CORS:
+        app.add_middleware(
+            CORSMiddleware,
+            allow_origins=settings.CORS_ALLOWED_ORIGINS,
+            allow_credentials=True,
+            allow_methods=['*'],
+            allow_headers=['*'],
+            expose_headers=settings.CORS_EXPOSE_HEADERS,
+        )
+
 
 def register_router(app: FastAPI) -> None:
     """
@@ -218,3 +222,16 @@ def register_socket_app(app: FastAPI) -> None:
         socketio_path='/ws/socket.io',
     )
     app.mount('/ws', socket_app)
+
+
+def register_metrics(app: FastAPI) -> None:
+    """
+    注册指标
+
+    :param app: FastAPI 应用实例
+    :return:
+    """
+    metrics_app = make_asgi_app()
+    app.mount('/metrics', metrics_app)
+
+    init_otel(app)

backend/core/registrar.py

@@ -14,6 +14,13 @@
 from backend.common.context import ctx
 from backend.common.enums import OperaLogCipherType, StatusType
 from backend.common.log import log
+from backend.common.prometheus.instruments import (
+    PROMETHEUS_EXCEPTION_COUNTER,
+    PROMETHEUS_REQUEST_COST_TIME_HISTOGRAM,
+    PROMETHEUS_REQUEST_COUNTER,
+    PROMETHEUS_REQUEST_IN_PROGRESS_GAUGE,
+    PROMETHEUS_RESPONSE_COUNTER,
+)
 from backend.common.queue import batch_dequeue
 from backend.common.response.response_code import StandardResponseCode
 from backend.core.conf import settings
@@ -43,6 +50,10 @@ async def dispatch(self, request: Request, call_next: Any) -> Response:
         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()
 
             # 执行请求
             code = 200
@@ -63,6 +74,12 @@ async def dispatch(self, request: Request, call_next: Any) -> Response:
                         code = exception.get('code')
                         msg = exception.get('msg')
                         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)
@@ -71,6 +88,20 @@ async def dispatch(self, request: Request, call_next: Any) -> Response:
                 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()
 
             # 此信息只能在请求后获取
             route = request.scope.get('route')