"""Metric 子域 Router。 外部系统限界上下文的指标查询 API,覆盖桶列表 / 聚合 / 时间序列 / 跨系统排名 / 按环境分解 / 最新快照 / Prometheus 导出 / 过期清理。 所有端点通过 ``create_use_cases_from_db`` 装配 use_cases, 经 ``metric_service`` 端口调用用例。 路径顺序约束:静态路径(aggregate/timeseries/by-system/by-env/latest/prometheus/buckets) 必须在根路径之前声明。 认证策略: - 查询类端点使用 get_required_user(支持 JWT + API Key 双模认证) - 清理端点使用 get_admin_user - Prometheus 端点复用 get_required_user,通过 Authorization: Bearer yxkey_ 支持 API Key 抓取 时间范围约束: - aggregate / timeseries / by-system / by-env 均要求 start_time + end_time,避免全表扫描 - timeseries 额外限制最大范围(hour→31 天,day→366 天),防止结果集膨胀 - by-system / by-env / aggregate 限制最大范围 366 天 """ from __future__ import annotations from datetime import datetime, timedelta from typing import Any, Literal from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from sqlalchemy.ext.asyncio import AsyncSession from yuxi.external_systems.infrastructure.container import create_use_cases_from_db from yuxi.external_systems.use_cases.dto.metric import ( DeleteOldBucketsInput, GetLatestBucketInput, ListMetricBucketsInput, MetricAggregateInput, MetricByEnvInput, MetricBySystemInput, MetricTimeseriesInput, PrometheusExportInput, ) from yuxi.storage.postgres.models_business import User from server.utils.auth_middleware import get_admin_user, get_db, get_required_user metric_router = APIRouter(prefix="/metrics", tags=["external-systems-metric"]) # 最大查询范围约束 _MAX_RANGE_DAYS = 366 _TIMESERIES_MAX_RANGE: dict[str, timedelta] = { "hour": timedelta(days=31), "day": timedelta(days=366), } def _validate_time_range(start: datetime, end: datetime, max_delta: timedelta) -> None: """校验时间范围:start 不晚于 end,且不超过 max_delta。""" if start > end: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="start_time 不能晚于 end_time", ) if end - start > max_delta: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"时间范围不得超过 {max_delta.days} 天", ) # ---------------- Endpoints ---------------- # ---- 静态路径端点(必须在根路径 "" 之前声明) ---- @metric_router.get("/aggregate", response_model=dict) async def aggregate_metrics( system_id: int | None = Query(None), env_key: str | None = Query(None, max_length=32), adapter_type: str | None = Query(None, max_length=32), start_time: datetime = Query(..., description="起始时间(必填)"), end_time: datetime = Query(..., description="结束时间(必填)"), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """单系统指标聚合。响应字段对齐 ORM aggregate 返回。""" _validate_time_range(start_time, end_time, timedelta(days=_MAX_RANGE_DAYS)) use_cases = create_use_cases_from_db(db) input_dto = MetricAggregateInput( system_id=system_id, env_key=env_key, adapter_type=adapter_type, start=start_time, end=end_time, ) output = await use_cases.metric_service.aggregate(input_dto) return {"success": True, "data": output.model_dump()} @metric_router.get("/timeseries", response_model=dict) async def get_metrics_timeseries( system_id: int | None = Query(None), env_key: str | None = Query(None, max_length=32), adapter_type: str | None = Query(None, max_length=32), start_time: datetime = Query(..., description="起始时间(必填)"), end_time: datetime = Query(..., description="结束时间(必填)"), interval: Literal["hour", "day"] = Query("hour", description="聚合间隔:hour / day"), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """时间序列聚合。start_time / end_time 必填,避免全表扫描。 最大范围约束:hour 间隔 ≤ 31 天,day 间隔 ≤ 366 天。 """ _validate_time_range(start_time, end_time, _TIMESERIES_MAX_RANGE[interval]) use_cases = create_use_cases_from_db(db) input_dto = MetricTimeseriesInput( system_id=system_id, env_key=env_key, adapter_type=adapter_type, start=start_time, end=end_time, interval=interval, ) series = await use_cases.metric_service.timeseries(input_dto) return {"success": True, "data": {"interval": interval, "series": series}} @metric_router.get("/by-system", response_model=dict) async def get_metrics_by_system( env_key: str | None = Query(None, max_length=32), adapter_type: str | None = Query(None, max_length=32), start_time: datetime = Query(..., description="起始时间(必填)"), end_time: datetime = Query(..., description="结束时间(必填)"), limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """跨系统聚合排名。按 total_calls 降序,返回 Top N。 total 为匹配过滤条件的系统总数(不受 limit 截断),支持 offset 分页。 """ _validate_time_range(start_time, end_time, timedelta(days=_MAX_RANGE_DAYS)) use_cases = create_use_cases_from_db(db) input_dto = MetricBySystemInput( env_key=env_key, adapter_type=adapter_type, start=start_time, end=end_time, limit=limit, offset=offset, ) items = await use_cases.metric_service.aggregate_by_system(input_dto) total = await use_cases.metric_service.count_systems(input_dto) return {"success": True, "data": {"items": items, "total": total}} @metric_router.get("/by-env", response_model=dict) async def get_metrics_by_env( system_id: int = Query(..., description="系统 ID(必填)"), start_time: datetime = Query(..., description="起始时间(必填)"), end_time: datetime = Query(..., description="结束时间(必填)"), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """单系统按环境分解指标。""" _validate_time_range(start_time, end_time, timedelta(days=_MAX_RANGE_DAYS)) use_cases = create_use_cases_from_db(db) input_dto = MetricByEnvInput( system_id=system_id, start=start_time, end=end_time, ) items = await use_cases.metric_service.aggregate_by_env(input_dto) return {"success": True, "data": {"items": items, "total": len(items)}} @metric_router.get("/latest", response_model=dict) async def get_latest_metric_bucket( system_id: int = Query(..., description="系统 ID(必填)"), env_key: str = Query("default", max_length=32, description="环境标识"), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """最新桶快照。无数据时 data 为 null。""" use_cases = create_use_cases_from_db(db) input_dto = GetLatestBucketInput(system_id=system_id, env_key=env_key) data = await use_cases.metric_service.get_latest_bucket(input_dto) return {"success": True, "data": data} @metric_router.get("/prometheus") async def export_metrics_prometheus( start_time: datetime | None = Query(None), end_time: datetime | None = Query(None), env_key: str | None = Query(None, max_length=32), adapter_type: str | None = Query(None, max_length=32), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> Response: """Prometheus 格式导出。支持 API Key 认证(Authorization: Bearer yxkey_)。 不传时间范围时默认导出最近 1 小时数据。支持 adapter_type 按适配器类型过滤。 """ use_cases = create_use_cases_from_db(db) input_dto = PrometheusExportInput( start=start_time, end=end_time, env_key=env_key, adapter_type=adapter_type, ) output = await use_cases.metric_service.export_prometheus(input_dto) return Response(content=output.content, media_type=output.content_type) @metric_router.delete("/buckets", response_model=dict) async def delete_old_metric_buckets( before: datetime = Query(..., description="清理此时间之前的桶(必填)"), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user), ) -> dict[str, Any]: """清理过期指标桶(软删除)。before 必填,避免误删全表。 审计字段 updated_by 记录执行清理的管理员 uid。 """ use_cases = create_use_cases_from_db(db) input_dto = DeleteOldBucketsInput(before=before, updated_by=current_user.uid) output = await use_cases.metric_service.delete_old_buckets(input_dto) return {"success": True, "data": output.model_dump()} # ---- 根路径端点(最后声明) ---- @metric_router.get("", response_model=dict) async def list_metric_buckets( system_id: int | None = Query(None), env_key: str | None = Query(None, max_length=32), start_time: datetime | None = Query(None), end_time: datetime | None = Query(None), limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user), ) -> dict[str, Any]: """桶列表分页查询。返回 Metric dataclass 字段 + avg_latency_ms。""" use_cases = create_use_cases_from_db(db) input_dto = ListMetricBucketsInput( system_id=system_id, env_key=env_key, start=start_time, end=end_time, limit=limit, offset=offset, ) output = await use_cases.metric_service.list_buckets(input_dto) return {"success": True, "data": output.model_dump()}