监控与告警实战

pip install prometheus-client

from prometheus_client import Counter, start_http_server

# 定义计数器
REQUEST_COUNT = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']  # 标签（维度）
)

# 在业务代码中递增
def handle_request(method: str, endpoint: str, status: int):
    REQUEST_COUNT.labels(
        method=method,
        endpoint=endpoint,
        status=str(status)
    ).inc()

# 批量递增
def handle_batch_errors(count: int):
    REQUEST_COUNT.labels(
        method='POST',
        endpoint='/api/data',
        status='500'
    ).inc(count)

from prometheus_client import Gauge
import threading

# 当前活跃连接数
ACTIVE_CONNECTIONS = Gauge(
    'active_connections',
    'Number of active connections'
)

# 可以作为上下文管理器使用
def process_connection():
    with ACTIVE_CONNECTIONS.track_inprogress():
        # 连接处理逻辑
        handle_connection()
    # 退出 with 块后自动减 1

# 也可以直接设置值
def update_temperature(temp: float):
    TEMPERATURE = Gauge('room_temperature_celsius', 'Room temperature')
    TEMPERATURE.set(temp)

# 设置为当前函数返回值
ACTIVE_THREADS = Gauge('active_threads', 'Number of active threads')
ACTIVE_THREADS.set_function(lambda: threading.active_count())

from prometheus_client import Histogram
import time

# 自定义桶边界（秒），根据业务 SLA 调整
REQUEST_DURATION = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration in seconds',
    ['method', 'endpoint'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)

# 计时装饰器
def monitor_duration(method: str, endpoint: str):
    def decorator(func):
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                duration = time.perf_counter() - start
                REQUEST_DURATION.labels(
                    method=method,
                    endpoint=endpoint
                ).observe(duration)
        return wrapper
    return decorator

# 使用装饰器
@monitor_duration('GET', '/api/users')
def get_users():
    time.sleep(0.05)
    return [{'id': 1, 'name': 'Alice'}]

# 也可以用上下文管理器
def process_task():
    with REQUEST_DURATION.labels('POST', '/api/tasks').time():
        do_heavy_work()

"""metrics_server.py — 独立指标服务"""
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import random
import threading

# ===== 定义指标 =====

REQUEST_COUNT = Counter(
    'app_http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

REQUEST_DURATION = Histogram(
    'app_http_request_duration_seconds',
    'HTTP request duration',
    ['method', 'endpoint'],
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5]
)

ACTIVE_REQUESTS = Gauge(
    'app_http_requests_in_progress',
    'Requests currently in progress',
    ['endpoint']
)

DB_POOL_SIZE = Gauge(
    'app_db_pool_size',
    'Database connection pool size'
)

DB_POOL_AVAILABLE = Gauge(
    'app_db_pool_available',
    'Available database connections'
)

# ===== 模拟业务流量 =====

def simulate_requests():
    """模拟业务请求"""
    endpoints = ['/api/users', '/api/orders', '/api/products']
    methods = ['GET', 'POST', 'PUT', 'DELETE']
    statuses = [200, 200, 200, 200, 201, 400, 404, 500]

    while True:
        endpoint = random.choice(endpoints)
        method = random.choice(methods)
        status = random.choice(statuses)

        with ACTIVE_REQUESTS.labels(endpoint=endpoint).track_inprogress():
            duration = random.expovariate(10)  # 模拟延迟
            time.sleep(min(duration, 2.0))

        REQUEST_COUNT.labels(
            method=method, endpoint=endpoint, status=str(status)
        ).inc()
        REQUEST_DURATION.labels(
            method=method, endpoint=endpoint
        ).observe(duration)

def simulate_db_pool():
    """模拟数据库连接池"""
    total = 20
    while True:
        available = random.randint(0, total)
        DB_POOL_SIZE.set(total)
        DB_POOL_AVAILABLE.set(available)
        time.sleep(1)

if __name__ == '__main__':
    # 在 8000 端口启动指标服务
    start_http_server(8000)
    print("Metrics server started at http://localhost:8000/metrics")

    # 启动模拟线程
    threading.Thread(target=simulate_requests, daemon=True).start()
    threading.Thread(target=simulate_db_pool, daemon=True).start()

    # 主线程保持运行
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Shutting down...")

"""app/metrics.py — FastAPI 指标中间件"""
from prometheus_client import Counter, Histogram, Gauge, REGISTRY, generate_latest
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import time

# 定义指标
REQUEST_COUNT = Counter(
    'fastapi_requests_total',
    'Total request count',
    ['method', 'endpoint', 'status_code']
)

REQUEST_DURATION = Histogram(
    'fastapi_request_duration_seconds',
    'Request duration',
    ['method', 'endpoint'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

REQUESTS_IN_PROGRESS = Gauge(
    'fastapi_requests_in_progress',
    'Requests in progress',
    ['method', 'endpoint']
)


class PrometheusMiddleware(BaseHTTPMiddleware):
    """自动采集每个请求的指标"""

    async def dispatch(self, request: Request, call_next):
        method = request.method
        # 规范化路径，避免高基数问题
        endpoint = self._normalize_path(request.url.path)

        REQUESTS_IN_PROGRESS.labels(method=method, endpoint=endpoint).inc()

        start_time = time.perf_counter()
        try:
            response = await call_next(request)
            status_code = str(response.status_code)
        except Exception:
            status_code = '500'
            raise
        finally:
            duration = time.perf_counter() - start_time
            REQUESTS_IN_PROGRESS.labels(method=method, endpoint=endpoint).dec()
            REQUEST_COUNT.labels(
                method=method, endpoint=endpoint, status_code=status_code
            ).inc()
            REQUEST_DURATION.labels(
                method=method, endpoint=endpoint
            ).observe(duration)

        return response

    @staticmethod
    def _normalize_path(path: str) -> str:
        """将路径中的动态参数替换为占位符，避免高基数"""
        parts = path.strip('/').split('/')
        normalized = []
        for part in parts:
            # 如果段看起来像 ID（纯数字或 UUID），替换为占位符
            if part.isdigit() or '-' in part and len(part) > 20:
                normalized.append('{id}')
            else:
                normalized.append(part)
        return '/' + '/'.join(normalized)


def metrics_endpoint():
    """暴露 /metrics 端点"""
    return Response(
        content=generate_latest(REGISTRY),
        media_type='text/plain; version=0.0.4; charset=utf-8'
    )

"""main.py — FastAPI 应用注册"""
from fastapi import FastAPI
from app.metrics import PrometheusMiddleware, metrics_endpoint

app = FastAPI(title="监控示例应用")

# 注册中间件
app.add_middleware(PrometheusMiddleware)

# 注册指标端点
app.add_route('/metrics', metrics_endpoint, methods=['GET'])


@app.get('/api/users/{user_id}')
async def get_user(user_id: int):
    return {'id': user_id, 'name': 'Alice'}


@app.get('/api/orders')
async def list_orders():
    return [{'id': 1, 'amount': 99.9}]

# 错误：用户 ID 作为标签 → 标签值爆炸
REQUEST_COUNT.labels(user_id=str(user_id)).inc()  # 千万不要这样做！

# 正确：只使用低基数的标签
REQUEST_COUNT.labels(method='GET', endpoint='/api/users', status='200').inc()

# 标签基数参考
# ✅ 低基数（< 100）：HTTP 方法、状态码、端点路径、环境
# ⚠️ 中基数（100~10k）：数据中心、版本号
# ❌ 高基数（> 10k）：用户 ID、请求 ID、邮箱地址

# grafana/dashboards/python-app.json — 核心面板配置（简化版）
# 在 Grafana UI 中导入或通过 provisioning 配置

dashboard:
  title: Python 应用监控
  panels:
    - title: 请求速率 (QPS)
      type: timeseries
      targets:
        - expr: sum(rate(fastapi_requests_total[5m])) by (endpoint)
          legendFormat: "{{endpoint}}"

    - title: 错误率
      type: timeseries
      targets:
        - expr: |
            sum(rate(fastapi_requests_total{status_code=~"5.."}[5m])) by (endpoint)
            /
            sum(rate(fastapi_requests_total[5m])) by (endpoint)
          legendFormat: "{{endpoint}}"

    - title: 请求延迟 (P50/P95/P99)
      type: timeseries
      targets:
        - expr: |
            histogram_quantile(0.5,
              sum(rate(fastapi_request_duration_seconds_bucket[5m])) by (le, endpoint)
            )
          legendFormat: "P50 - {{endpoint}}"
        - expr: |
            histogram_quantile(0.95,
              sum(rate(fastapi_request_duration_seconds_bucket[5m])) by (le, endpoint)
            )
          legendFormat: "P95 - {{endpoint}}"
        - expr: |
            histogram_quantile(0.99,
              sum(rate(fastapi_request_duration_seconds_bucket[5m])) by (le, endpoint)
            )
          legendFormat: "P99 - {{endpoint}}"

    - title: 在线请求数
      type: gauge
      targets:
        - expr: sum(fastapi_requests_in_progress)
          legendFormat: "In Progress"

    - title: 连接池利用率
      type: gauge
      targets:
        - expr: app_db_pool_available / app_db_pool_size
          legendFormat: "可用率"
      thresholds:
        - value: 0.2
          color: red      # 低于 20% 可用，危险
        - value: 0.5
          color: yellow   # 低于 50% 可用，警告
        - value: 1.0
          color: green    # 正常

# grafana/provisioning/datasources/prometheus.yml
apiVersion: 1
datasources:
  - name: Prometheus
    type: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true
    editable: false

# grafana/provisioning/dashboards/dashboards.yml
apiVersion: 1
providers:
  - name: Python App
    orgId: 1
    folder: ''
    type: file
    disableDeletion: false
    updateIntervalSeconds: 30
    options:
      path: /var/lib/grafana/dashboards
      foldersFromFilesStructure: false

pip install sentry-sdk

"""sentry_config.py — Sentry 初始化"""
import sentry_sdk

def init_sentry(dsn: str, environment: str = "production", release: str = ""):
    """初始化 Sentry SDK"""
    sentry_sdk.init(
        dsn=dsn,
        environment=environment,
        release=release,              # 关联代码版本
        traces_sample_rate=0.1,       # 性能追踪采样率 10%
        profiles_sample_rate=0.1,     # 性能分析采样率 10%
        send_default_pii=False,       # 不发送个人身份信息
        max_breadcrumbs=50,           # Breadcrumb 数量
        attach_stacktrace=True,       # 附加堆栈到非异常日志
    )

"""FastAPI + Sentry"""
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    integrations=[FastApiIntegration()],
    traces_sample_rate=0.1,
    environment="production",
    release="my-app@1.0.0",
)

app = FastAPI()

# 全局异常处理，确保未捕获异常上报 Sentry
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # Sentry 自动捕获未处理异常，这里也可以手动上报
    sentry_sdk.capture_exception(exc)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal Server Error"}
    )

@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    if user_id > 1000:
        raise ValueError(f"Invalid user_id: {user_id}")
    return {"id": user_id, "name": "Alice"}

"""Flask + Sentry"""
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
from flask import Flask, jsonify

sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    integrations=[FlaskIntegration()],
    traces_sample_rate=0.1,
)

app = Flask(__name__)

@app.errorhandler(Exception)
def handle_exception(exc):
    sentry_sdk.capture_exception(exc)
    return jsonify({"error": "Internal Server Error"}), 500

@app.route("/api/orders/<int:order_id>")
def get_order(order_id):
    order = find_order(order_id)
    if not order:
        raise LookupError(f"Order {order_id} not found")
    return jsonify(order)

"""自定义 Sentry 上下文信息"""
import sentry_sdk

def set_user_context(user_id: int, username: str, email: str):
    """设置用户上下文"""
    sentry_sdk.set_user({
        "id": str(user_id),
        "username": username,
        "email": email,
        # 不要放密码等敏感信息！
    })

def set_request_context(request_id: str, tenant_id: str):
    """设置标签和额外信息"""
    # 标签：可搜索、可过滤
    sentry_sdk.set_tag("request_id", request_id)
    sentry_sdk.set_tag("tenant_id", tenant_id)
    sentry_sdk.set_tag("service", "user-api")

    # 额外信息：不可搜索，但有助于排查
    sentry_sdk.set_context("request_info", {
        "request_id": request_id,
        "tenant_id": tenant_id,
        "region": "cn-east-1",
    })

def add_breadcrumb(action: str, category: str, data: dict | None = None):
    """添加 Breadcrumb（面包屑），记录异常发生前的操作路径"""
    sentry_sdk.add_breadcrumb(
        category=category,
        message=action,
        data=data or {},
        level="info",
    )

# 使用示例
@app.post("/api/checkout")
async def checkout(request: Request):
    user = get_current_user()
    set_user_context(user.id, user.name, user.email)

    add_breadcrumb("开始结账流程", "checkout", {"cart_items": 3})

    order = create_order(user)
    add_breadcrumb("订单创建成功", "checkout", {"order_id": order.id})

    payment = process_payment(order)
    if payment.failed:
        # 手动上报业务异常
        sentry_sdk.capture_message(
            f"支付失败: order={order.id}, reason={payment.error}",
            level="warning"
        )

    return {"order_id": order.id, "status": "paid"}

"""Sentry 采样策略 — 控制数据量和成本"""
import sentry_sdk

def traces_sampler(sampling_context):
    """自定义性能追踪采样"""
    op = sampling_context.get("parent_sampled")
    transaction_name = sampling_context.get("transaction_context", {}).get("name", "")

    # 健康检查不采样
    if transaction_name in ("/health", "/ready", "/metrics"):
        return 0.0

    # 关键业务接口全量采样
    if transaction_name.startswith("/api/payment"):
        return 1.0

    # 其他接口 10% 采样
    return 0.1

sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    traces_sampler=traces_sampler,
)

pip install structlog

"""logging_config.py — 结构化日志配置"""
import logging
import structlog
from typing import Any


def add_app_context(
    logger: logging.Logger, method_name: str, event_dict: dict[str, Any]
) -> dict[str, Any]:
    """自定义处理器：添加应用上下文"""
    event_dict["app"] = "user-api"
    event_dict["version"] = "1.0.0"
    return event_dict


def configure_logging(log_level: str = "INFO", json_logs: bool = True):
    """配置结构化日志"""
    shared_processors: list[Any] = [
        structlog.contextvars.merge_contextvars,     # 合并上下文变量
        structlog.stdlib.add_logger_name,             # 添加 logger 名称
        structlog.stdlib.add_log_level,               # 添加日志级别
        structlog.stdlib.PositionalArgumentsFormatter(), # 位置参数格式化
        structlog.processors.TimeStamper(fmt="iso"),  # ISO 8601 时间戳
        structlog.processors.StackInfoRenderer(),     # 堆栈信息
        structlog.processors.UnicodeDecoder(),        # Unicode 解码
        add_app_context,                              # 自定义上下文
    ]

    if json_logs:
        # JSON 格式输出（生产环境）
        renderer = structlog.processors.JSONRenderer()
    else:
        # 控制台彩色输出（开发环境）
        renderer = structlog.dev.ConsoleRenderer()

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    # 配置标准库 logging 的格式化器
    formatter = structlog.stdlib.ProcessorFormatter(
        processors=[
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            renderer,
        ],
        foreign_pre_chain=shared_processors,
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.handlers = [handler]
    root_logger.setLevel(log_level.upper())

    # 降低第三方库日志级别
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)

"""业务代码中的结构化日志"""
import structlog

logger = structlog.get_logger()


def process_order(order_id: int, user_id: int):
    # 绑定上下文，后续日志自动携带
    log = logger.bind(order_id=order_id, user_id=user_id)

    log.info("order_processing_started")

    try:
        result = validate_order(order_id)
        log.info("order_validated", item_count=result.item_count)
    except ValueError as e:
        log.error("order_validation_failed", error=str(e), error_type=type(e).__name__)
        raise

    try:
        payment = charge_payment(order_id)
        log.info("payment_completed", amount=payment.amount, method=payment.method)
    except PaymentError as e:
        log.error("payment_failed", error=str(e))
        raise

{
  "event": "order_validated",
  "order_id": 12345,
  "user_id": 678,
  "item_count": 3,
  "logger": "app.orders",
  "level": "info",
  "timestamp": "2026-06-06T10:30:00.123456Z",
  "app": "user-api",
  "version": "1.0.0"
}

"""请求级上下文传递 — 不用层层传参"""
import structlog
from contextvars import ContextVar

# 定义上下文变量
request_id_var: ContextVar[str] = ContextVar("request_id", default="")
tenant_id_var: ContextVar[str] = ContextVar("tenant_id", default="")


def context_injector(
    logger, method_name: str, event_dict: dict
) -> dict:
    """自动注入上下文变量到每条日志"""
    request_id = request_id_var.get("")
    tenant_id = tenant_id_var.get("")
    if request_id:
        event_dict["request_id"] = request_id
    if tenant_id:
        event_dict["tenant_id"] = tenant_id
    return event_dict


# FastAPI 中间件设置上下文
from fastapi import Request
import uuid

@app.middleware("http")
async def context_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    tenant_id = request.headers.get("X-Tenant-ID", "default")

    request_id_var.set(request_id)
    tenant_id_var.set(tenant_id)

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

# promtail/config.yml — 日志采集代理配置
server:
  http_listen_port: 9080

positions:
  filename: /tmp/positions.yaml

clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
  - job_name: python-app
    static_configs:
      - targets:
          - localhost
        labels:
          job: python-app
          environment: production
          __path__: /var/log/python-app/*.log
    pipeline_stages:
      - json:
          expressions:
            level: level
            logger: logger
            request_id: request_id
            tenant_id: tenant_id
      - labels:
          level:
          logger:
          tenant_id:
      - timestamp:
          source: timestamp
          format: RFC3339Nano

"""app/monitoring.py — 统一监控配置"""
import time
import uuid
import structlog
import sentry_sdk
from contextvars import ContextVar
from sentry_sdk.integrations.fastapi import FastApiIntegration
from prometheus_client import Counter, Histogram, Gauge, REGISTRY, generate_latest
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

# ===== 上下文变量 =====
request_id_var: ContextVar[str] = ContextVar("request_id", default="")

# ===== Prometheus 指标 =====
REQUEST_COUNT = Counter(
    'app_requests_total',
    'Total requests',
    ['method', 'endpoint', 'status']
)

REQUEST_DURATION = Histogram(
    'app_request_duration_seconds',
    'Request duration',
    ['method', 'endpoint'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

ACTIVE_REQUESTS = Gauge(
    'app_active_requests',
    'Currently processing requests'
)

DB_QUERY_DURATION = Histogram(
    'app_db_query_duration_seconds',
    'Database query duration',
    ['operation'],
    buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.5]
)

# ===== 结构化日志 =====
logger = structlog.get_logger()


# ===== 监控中间件 =====
class MonitoringMiddleware(BaseHTTPMiddleware):
    """统一监控中间件：采集指标 + 设置上下文"""

    async def dispatch(self, request: Request, call_next):
        # 设置请求上下文
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request_id_var.set(request_id)

        method = request.method
        endpoint = self._normalize_path(request.url.path)

        log = logger.bind(
            request_id=request_id,
            method=method,
            endpoint=endpoint,
        )

        log.info("request_started")
        ACTIVE_REQUESTS.inc()

        start = time.perf_counter()
        try:
            response = await call_next(request)
            status = str(response.status_code)
            log.info("request_completed", status_code=status)
        except Exception as e:
            status = "500"
            log.error("request_failed", error=str(e), error_type=type(e).__name__)
            # Sentry 自动捕获，这里显式上报确保不遗漏
            sentry_sdk.capture_exception(e)
            raise
        finally:
            duration = time.perf_counter() - start
            ACTIVE_REQUESTS.dec()
            REQUEST_COUNT.labels(method=method, endpoint=endpoint, status=status).inc()
            REQUEST_DURATION.labels(method=method, endpoint=endpoint).observe(duration)
            log.info("request_metrics", duration_ms=round(duration * 1000, 2))

        return response

    @staticmethod
    def _normalize_path(path: str) -> str:
        parts = path.strip('/').split('/')
        normalized = []
        for part in parts:
            if part.isdigit() or (len(part) > 20 and '-' in part):
                normalized.append('{id}')
            else:
                normalized.append(part)
        return '/' + '/'.join(normalized)


# ===== 应用初始化 =====
def setup_monitoring(app: FastAPI, sentry_dsn: str = "", env: str = "production"):
    """统一初始化监控组件"""
    # Sentry
    if sentry_dsn:
        sentry_sdk.init(
            dsn=sentry_dsn,
            integrations=[FastApiIntegration()],
            traces_sample_rate=0.1,
            environment=env,
        )

    # 中间件
    app.add_middleware(MonitoringMiddleware)

    # 指标端点
    def metrics():
        return Response(
            content=generate_latest(REGISTRY),
            media_type="text/plain; version=0.0.4; charset=utf-8",
        )

    app.add_route("/metrics", metrics, methods=["GET"])

    # 全局异常处理
    @app.exception_handler(Exception)
    async def handle_exception(request: Request, exc: Exception):
        logger.error(
            "unhandled_exception",
            error=str(exc),
            error_type=type(e).__name__,
            path=request.url.path,
        )
        sentry_sdk.capture_exception(exc)
        return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

    logger.info("monitoring_initialized", environment=env)

"""main.py — 完整监控的 FastAPI 应用"""
from fastapi import FastAPI
from app.monitoring import setup_monitoring, DB_QUERY_DURATION, logger
import time
import os

app = FastAPI(title="完整监控示例")

# 初始化监控
setup_monitoring(
    app,
    sentry_dsn=os.getenv("SENTRY_DSN", ""),
    env=os.getenv("ENVIRONMENT", "development"),
)


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    with DB_QUERY_DURATION.labels(operation="select_user").time():
        user = await fetch_user_from_db(user_id)

    if not user:
        logger.warning("user_not_found", user_id=user_id)
        return {"error": "User not found"}, 404

    return user


@app.post("/api/orders")
async def create_order():
    with DB_QUERY_DURATION.labels(operation="insert_order").time():
        order = await insert_order_to_db()

    logger.info("order_created", order_id=order.id)
    return order

# prometheus/alert_rules.yml — 告警规则
groups:
  - name: python_app_alerts
    rules:
      # ===== 错误率告警 =====
      - alert: HighErrorRate
        expr: |
          sum(rate(app_requests_total{status=~"5.."}[5m]))
          /
          sum(rate(app_requests_total[5m]))
          > 0.05
        for: 2m
        labels:
          severity: critical
          team: backend
        annotations:
          summary: "HTTP 5xx 错误率超过 5%"
          description: |
            当前 5xx 错误率: {{ $value | printf "%.2f" }}%
            持续时间: 2 分钟
            请立即排查应用日志

      # ===== 高延迟告警 =====
      - alert: HighLatencyP99
        expr: |
          histogram_quantile(0.99,
            sum(rate(app_request_duration_seconds_bucket[5m])) by (le, endpoint)
          ) > 2.0
        for: 5m
        labels:
          severity: warning
          team: backend
        annotations:
          summary: "P99 延迟超过 2 秒"
          description: |
            端点: {{ $labels.endpoint }}
            P99 延迟: {{ $value | printf "%.2f" }}s

      # ===== 连接池耗尽告警 =====
      - alert: DBPoolExhaustion
        expr: app_db_pool_available / app_db_pool_size < 0.1
        for: 1m
        labels:
          severity: critical
          team: backend
        annotations:
          summary: "数据库连接池即将耗尽"
          description: |
            可用连接: {{ $value | printf "%.1f" }}%
            请检查是否有连接泄漏

      # ===== 服务不可用 =====
      - alert: ServiceDown
        expr: up{job="python-app"} == 0
        for: 1m
        labels:
          severity: critical
          team: backend
        annotations:
          summary: "Python 应用实例下线"
          description: "实例 {{ $labels.instance }} 已下线超过 1 分钟"

      # ===== 内存使用率 =====
      - alert: HighMemoryUsage
        expr: process_resident_memory_bytes / (1024 * 1024) > 1024
        for: 10m
        labels:
          severity: warning
          team: backend
        annotations:
          summary: "Python 进程内存超过 1GB"
          description: |
            当前内存: {{ $value | printf "%.0f" }}MB
            可能存在内存泄漏

      # ===== 请求激增 =====
      - alert: RequestSpike
        expr: |
          sum(rate(app_requests_total[1m]))
          /
          sum(rate(app_requests_total[5m] offset 1h))
          > 3
        for: 2m
        labels:
          severity: warning
          team: backend
        annotations:
          summary: "请求量突增 3 倍"
          description: "当前 QPS 是 1 小时前的 3 倍以上，请关注容量"

# alertmanager/alertmanager.yml — 告警路由与通知
global:
  resolve_timeout: 5m

route:
  group_by: ['alertname', 'severity']
  group_wait: 30s       # 同组告警等待 30s 合并
  group_interval: 5m    # 同组新告警间隔 5m
  repeat_interval: 4h   # 重复告警间隔 4h
  receiver: 'default'

  routes:
    # 严重告警 → 钉钉 + 电话
    - match:
        severity: critical
      receiver: 'critical'
      repeat_interval: 30m

    # 警告 → 钉钉
    - match:
        severity: warning
      receiver: 'warning'
      repeat_interval: 2h

# 抑制规则：服务下线时抑制其他告警
inhibit_rules:
  - source_match:
      alertname: ServiceDown
    target_match:
      severity: warning
    equal: ['instance']

receivers:
  - name: 'default'
    webhook_configs:
      - url: 'http://alertmanager-webhook:8080/notify'

  - name: 'critical'
    webhook_configs:
      - url: 'http://alertmanager-webhook:8080/notify/critical'

  - name: 'warning'
    webhook_configs:
      - url: 'http://alertmanager-webhook:8080/notify/warning'

# docker-compose.monitoring.yml
version: "3.8"

services:
  # ===== 应用 =====
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - SENTRY_DSN=${SENTRY_DSN}
      - ENVIRONMENT=production
    networks:
      - monitoring

  # ===== Prometheus =====
  prometheus:
    image: prom/prometheus:v2.51.0
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
      - ./prometheus/alert_rules.yml:/etc/prometheus/alert_rules.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.retention.time=30d'
    networks:
      - monitoring

  # ===== Alertmanager =====
  alertmanager:
    image: prom/alertmanager:v0.27.0
    ports:
      - "9093:9093"
    volumes:
      - ./alertmanager/alertmanager.yml:/etc/alertmanager/alertmanager.yml
    networks:
      - monitoring

  # ===== Grafana =====
  grafana:
    image: grafana/grafana:10.4.0
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
      - GF_USERS_ALLOW_SIGN_UP=false
    volumes:
      - ./grafana/provisioning:/etc/grafana/provisioning
      - grafana_data:/var/lib/grafana
    networks:
      - monitoring

  # ===== Loki =====
  loki:
    image: grafana/loki:2.9.0
    ports:
      - "3100:3100"
    volumes:
      - loki_data:/loki
    networks:
      - monitoring

  # ===== Promtail =====
  promtail:
    image: grafana/promtail:2.9.0
    volumes:
      - ./promtail/config.yml:/etc/promtail/config.yml
      - /var/log:/var/log:ro
    networks:
      - monitoring

volumes:
  prometheus_data:
  grafana_data:
  loki_data:

networks:
  monitoring:
    driver: bridge

# prometheus/prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']

rule_files:
  - alert_rules.yml

scrape_configs:
  - job_name: 'python-app'
    metrics_path: '/metrics'
    static_configs:
      - targets: ['app:8000']
        labels:
          service: 'user-api'