日志与监控

# 这样做在生产环境是灾难
def process_order(order_id: int) -> None:
    print(f"开始处理订单 {order_id}")      # 去了哪？stdout 还是 stderr？
    # ... 业务逻辑 ...
    print(f"订单 {order_id} 处理完成")      # 没有时间戳，没有级别，没有上下文

import logging

logger = logging.getLogger(__name__)


def process_order(order_id: int) -> None:
    logger.info("开始处理订单 order_id=%d", order_id)
    # ... 业务逻辑 ...
    logger.info("订单处理完成 order_id=%d", order_id)

import logging
import os


def configure_root_logger() -> None:
    """根据环境变量设置日志级别。"""
    env = os.getenv("APP_ENV", "development")
    level_map = {
        "development": logging.DEBUG,
        "staging": logging.INFO,
        "production": logging.WARNING,
    }
    logging.basicConfig(
        level=level_map.get(env, logging.INFO),
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )

import logging

# 子 Logger 配置了一个 Handler
child = logging.getLogger("myapp.services")
child.addHandler(logging.StreamHandler())
child.setLevel(logging.DEBUG)

# 父 Logger 也配置了 Handler（通过 basicConfig）
logging.basicConfig(level=logging.DEBUG)

# 这条日志会输出两次！
child.info("支付服务启动")

child.propagate = False  # 禁止传播，日志只输出一次

import logging
import sys


handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(
    logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s")
)

handler = logging.FileHandler("/var/log/myapp/app.log")
handler.setFormatter(
    logging.Formatter(
        "%(asctime)s [%(levelname)s] %(name)s:%(lineno)d: %(message)s"
    )
)

from logging.handlers import RotatingFileHandler


handler = RotatingFileHandler(
    "/var/log/myapp/app.log",
    maxBytes=10 * 1024 * 1024,  # 10 MB
    backupCount=5,               # 保留 5 个旧文件
)

from logging.handlers import TimedRotatingFileHandler


handler = TimedRotatingFileHandler(
    "/var/log/myapp/app.log",
    when="midnight",    # 每天午夜轮转
    interval=1,         # 间隔 1 个 when 单位
    backupCount=30,     # 保留 30 天
    encoding="utf-8",
)
handler.suffix = "%Y-%m-%d"  # 旧文件名后缀

import logging.config


LOGGING_CONFIG: dict = {
    "version": 1,
    "disable_existing_loggers": False,  # 不要禁用已有 logger
    "formatters": {
        "default": {
            "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
        "detailed": {
            "format": (
                "%(asctime)s [%(levelname)s] %(name)s:%(lineno)d "
                "%(funcName)s(): %(message)s"
            ),
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "DEBUG",
            "formatter": "default",
            "stream": "ext://sys.stdout",
        },
        "file": {
            "class": "logging.handlers.TimedRotatingFileHandler",
            "level": "INFO",
            "formatter": "detailed",
            "filename": "/var/log/myapp/app.log",
            "when": "midnight",
            "backupCount": 30,
            "encoding": "utf-8",
        },
    },
    "root": {
        "level": "INFO",
        "handlers": ["console", "file"],
    },
    "loggers": {
        "myapp": {
            "level": "DEBUG",
            "handlers": ["console", "file"],
            "propagate": False,
        },
        "sqlalchemy.engine": {
            "level": "WARNING",  # 第三方库日志一般设高一点
        },
    },
}


logging.config.dictConfig(LOGGING_CONFIG)

[loggers]
keys=root,myapp

[handlers]
keys=console,file

[formatters]
keys=default,detailed

[formatter_default]
format=%(asctime)s [%(levelname)s] %(name)s: %(message)s
datefmt=%Y-%m-%d %H:%M:%S

[formatter_detailed]
format=%(asctime)s [%(levelname)s] %(name)s:%(lineno)d %(funcName)s(): %(message)s

[handler_console]
class=StreamHandler
level=DEBUG
formatter=default
args=(sys.stdout,)

[handler_file]
class=handlers.TimedRotatingFileHandler
level=INFO
formatter=detailed
args=('/var/log/myapp/app.log', 'midnight', 1, 30, 'utf-8')

[logger_root]
level=INFO
handlers=console,file

[logger_myapp]
level=DEBUG
handlers=console,file
qualname=myapp
propagate=0

2026-06-05 10:30:15 [INFO] myapp.services.payment: 订单处理完成 order_id=12345 amount=99.00 user_id=42

{
  "timestamp": "2026-06-05T10:30:15.123456Z",
  "level": "info",
  "logger": "myapp.services.payment",
  "event": "订单处理完成",
  "order_id": 12345,
  "amount": 99.0,
  "user_id": 42
}

# 需要 Python 3.10+
import structlog

logger = structlog.get_logger()


def process_order(order_id: int, amount: float, user_id: int) -> None:
    logger.info(
        "订单处理完成",
        order_id=order_id,
        amount=amount,
        user_id=user_id,
    )

import structlog


structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,       # 添加 level 字段
        structlog.stdlib.add_logger_name,     # 添加 logger 字段
        structlog.processors.TimeStamper(     # 添加时间戳
            fmt="iso", utc=True
        ),
        structlog.dev.ConsoleRenderer(),      # 开发环境：彩色输出
        # structlog.processors.JSONRenderer(),  # 生产环境：JSON 输出
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

# 反模式：丢失了堆栈信息
try:
    result = 1 / 0
except ZeroDivisionError as e:
    logger.error(f"计算出错: {e}")  # 只输出 "计算出错: division by zero"

# 正确做法：保留完整堆栈
try:
    result = 1 / 0
except ZeroDivisionError:
    logger.exception("订单金额计算失败，order_id=%d", order_id)

import sys


def log_current_stack() -> None:
    """记录当前调用栈，用于调试复杂调用链。"""
    logger.debug("当前调用栈", stack_info=True)


def log_caught_exception(exc: Exception) -> None:
    """在 except 块外记录已捕获的异常。"""
    logger.error("捕获到异常", exc_info=exc)

import logging


class RequestIDAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        return f"[request_id={self.extra['request_id']}] {msg}", kwargs


logger = logging.getLogger(__name__)
adapter = RequestIDAdapter(logger, {"request_id": "unknown"})

# 使用
adapter.info("请求处理开始")  # 输出中包含 request_id=unknown

import contextvars
import logging
import uuid


request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
    "request_id", default="-"
)


class RequestIDFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True


# 配置时添加 Filter
handler = logging.StreamHandler()
handler.addFilter(RequestIDFilter())
handler.setFormatter(
    logging.Formatter(
        "%(asctime)s [%(levelname)s] [%(request_id)s] %(name)s: %(message)s"
    )
)


def set_request_id() -> str:
    """为当前请求上下文设置 request_id。"""
    rid = uuid.uuid4().hex[:12]
    request_id_var.set(rid)
    return rid

# 需要 Python 3.10+
import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration


sentry_sdk.init(
    dsn="https://xxx@sentry.example.com/1",
    traces_sample_rate=0.5,  # 生产环境建议 0.1-0.5
    environment="production",
    integrations=[
        LoggingIntegration(
            level=sentry_sdk.consts.LogLevel.INFO.value,
            event_level=sentry_sdk.consts.LogLevel.ERROR.value,
        ),
    ],
)

from sentry_sdk import capture_exception


try:
    risky_operation()
except ValueError as exc:
    logger.exception("风险操作失败")
    capture_exception(exc)  # 显式上报到 Sentry

# 需要 Python 3.10+
from prometheus_client import Counter, Histogram, Gauge, generate_latest
import time
from functools import wraps
from collections.abc import Callable
from typing import Any


# 计数器：请求总数
request_count = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status"],
)

# 直方图：请求耗时分布
request_duration = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration in seconds",
    ["method", "endpoint"],
)

# 仪表盘：当前活跃请求数
active_requests = Gauge(
    "http_active_requests",
    "Currently active HTTP requests",
)


def track_request(method: str, endpoint: str) -> Callable[..., Any]:
    """装饰器：自动追踪请求计数、耗时和活跃连接数。"""
    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            active_requests.inc()
            start = time.perf_counter()
            try:
                result = func(*args, **kwargs)
                request_count.labels(
                    method=method, endpoint=endpoint, status="200"
                ).inc()
                return result
            except Exception:
                request_count.labels(
                    method=method, endpoint=endpoint, status="500"
                ).inc()
                raise
            finally:
                duration = time.perf_counter() - start
                request_duration.labels(
                    method=method, endpoint=endpoint
                ).observe(duration)
                active_requests.dec()
        return wrapper
    return decorator

from fastapi import FastAPI
from starlette.responses import Response


app = FastAPI()


@app.get("/metrics")
async def metrics() -> Response:
    return Response(
        content=generate_latest(),
        media_type="text/plain; version=0.0.4",
    )

from fastapi import FastAPI
from enum import Enum


class HealthStatus(str, Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"


app = FastAPI()


@app.get("/health")
async def health_check() -> dict[str, str]:
    """基础健康检查：负载均衡器用。"""
    return {"status": HealthStatus.HEALTHY}


@app.get("/health/ready")
async def readiness_check() -> dict[str, object]:
    """就绪检查：检查关键依赖是否可用。"""
    checks: dict[str, str] = {}

    # 检查数据库连接
    try:
        # db.execute("SELECT 1")
        checks["database"] = "ok"
    except Exception as exc:
        checks["database"] = f"error: {exc}"

    # 检查 Redis 连接
    try:
        # redis.ping()
        checks["redis"] = "ok"
    except Exception as exc:
        checks["redis"] = f"error: {exc}"

    all_ok = all(v == "ok" for v in checks.values())
    status = HealthStatus.HEALTHY if all_ok else HealthStatus.DEGRADED

    return {"status": status, "checks": checks}

STANDARD_FORMAT = (
    "%(asctime)s [%(levelname)s] [%(request_id)s] "
    "%(name)s:%(lineno)d %(funcName)s(): %(message)s"
)

import random


class SamplingFilter(logging.Filter):
    """按比例采样日志。"""

    def __init__(self, rate: float = 0.1) -> None:
        super().__init__()
        self.rate = rate

    def filter(self, record: logging.LogRecord) -> bool:
        return random.random() < self.rate

# 反模式
try:
    call_external_api()
except Exception:
    pass  # 吃了，什么都没留下

# 至少做到
try:
    call_external_api()
except Exception:
    logger.exception("外部 API 调用失败，降级处理")
    # 根据业务决定是返回默认值还是继续向上抛

# 反模式：字符串拼接，既慢又难解析
logger.info("订单处理完成，order_id=" + str(order_id) + "，amount=" + str(amount))

# 反模式：f-string，同样难解析
logger.info(f"订单处理完成，order_id={order_id}，amount={amount}")

# 推荐：使用 % 格式化，延迟求值，且 structlog 可自动提取键值对
logger.info("订单处理完成，order_id=%d，amount=%.2f", order_id, amount)

# 反模式：用日志做统计
logger.info("请求处理完成，耗时=%dms", duration_ms)

# 推荐：用指标记录
request_duration.labels(method="POST", endpoint="/orders").observe(duration_ms / 1000)
logger.info("请求处理完成")  # 日志只记录事件

import os
import json
import logging.config
from pathlib import Path


def setup_logging() -> None:
    config_path = Path(
        os.getenv("LOG_CONFIG", "logging.json")
    )
    if config_path.exists():
        config = json.loads(config_path.read_text())
        logging.config.dictConfig(config)
    else:
        logging.basicConfig(level=logging.INFO)