Skip to content
BranchesTags

Repository files navigation

logtrace-sdk

一个轻量级的 Python 可观测性 SDK，专为 FastAPI、Flask 和其他 Python 应用设计，提供结构化日志、分布式追踪和请求 ID 传播功能。

✨ 特性

  • ✅ 结构化日志：基于 structlog，输出 JSON 格式，完美适配 ELK/ES/Loki
  • ✅ 自动注入上下文：service、environment、request_id、trace_id、span_id 自动添加到日志
  • ✅ OpenTelemetry 集成：通过 OTLP 协议导出追踪数据到 Jaeger/Tempo/Grafana
  • ✅ FastAPI 自动追踪：自动为 HTTP 请求创建 span
  • ✅ 装饰器追踪：@trace_method 装饰器轻松为方法创建 span
  • ✅ 代码块追踪：with trace_block(...) 上下文管理器创建子 span
  • ✅ 敏感数据脱敏：自动脱敏日志和 span 属性中的敏感信息
  • ✅ 请求 ID 传播：自动在响应头中添加 X-Request-ID
  • ✅ 日志与追踪关联：通过 trace_id 关联日志和分布式追踪

📦 安装

从 PyPI 安装

# 基础安装（不包含 FastAPI 依赖）
pip install logtrace-sdk

# 安装 FastAPI 支持（推荐）
pip install logtrace-sdk[fastapi]

# 安装 Sentry 支持（可选）
pip install logtrace-sdk[sentry]

从 PyPI 源安装

pip install --extra-index-url https://pypi.org/simple \
            logtrace-sdk[fastapi]

🚀 快速开始

FastAPI 项目（推荐）

1. 安装 SDK

pip install logtrace-sdk[fastapi]

2. 配置日志和 OpenTelemetry

from fastapi import FastAPI
from fastapi.middleware import Middleware
from logtrace.logger import configure_logging, get_logger
from logtrace.telemetry import setup_telemetry
from logtrace.integrations.fastapi import instrument_fastapi, RequestContextMiddleware
import os

# 初始化日志（输出 JSON 到 stdout）
configure_logging(
    service_name="my-service",
    environment=os.getenv("ENVIRONMENT", "development"),
)

logger = get_logger(__name__)

# 初始化 OpenTelemetry
setup_telemetry(
    service_name="my-service",
    otlp_endpoint=os.getenv("OTLP_ENDPOINT", "http://localhost:4317"),
)

# 创建 FastAPI 应用
app = FastAPI()

# 注册中间件（自动添加 request_id）
app.add_middleware(RequestContextMiddleware)

# 启用 FastAPI 自动追踪
instrument_fastapi(app)

# 使用日志
@app.get("/hello")
async def hello():
    logger.info("hello_requested")
    return {"message": "Hello, World!"}

3. 运行应用

uvicorn main:app --host 0.0.0.0 --port 8000

Flask 项目

from flask import Flask, request
from logtrace.logger import configure_logging, get_logger
from logtrace.telemetry import setup_telemetry, shutdown_telemetry
from logtrace.tracing import get_tracer
import os

# 初始化日志和 OpenTelemetry
configure_logging(
    service_name="my-flask-app",
    environment=os.getenv("ENVIRONMENT", "development"),
)

setup_telemetry(
    service_name="my-flask-app",
    otlp_endpoint=os.getenv("OTLP_ENDPOINT", "http://localhost:4317"),
)

app = Flask(__name__)
logger = get_logger(__name__)
tracer = get_tracer(__name__)

# 手动创建请求 span（Flask 没有自动 instrumentation）
@app.before_request
def before_request():
    span = tracer.start_as_current_span(f"{request.method} {request.path}")
    span.set_attribute("http.method", request.method)
    span.set_attribute("http.url", request.url)

@app.after_request
def after_request(response):
    span = tracer.get_current_span()
    if span:
        span.set_attribute("http.status_code", response.status_code)
        span.end()
    return response

@app.route("/hello")
def hello():
    logger.info("hello_requested")
    return {"message": "Hello, World!"}

if __name__ == "__main__":
    try:
        app.run(host="0.0.0.0", port=5000)
    finally:
        shutdown_telemetry()

非 Web 项目（后台服务/脚本）

from logtrace.logger import configure_logging, get_logger
from logtrace.telemetry import setup_telemetry, shutdown_telemetry
from logtrace.tracing import get_tracer, trace_method
import os

# 初始化日志和 OpenTelemetry
configure_logging(
    service_name="my-backend-service",
    environment=os.getenv("ENVIRONMENT", "production"),
)

setup_telemetry(
    service_name="my-backend-service",
    otlp_endpoint=os.getenv("OTLP_ENDPOINT", "http://localhost:4317"),
)

logger = get_logger(__name__)
tracer = get_tracer(__name__)

# 手动创建根 span（重要！）
def main():
    with tracer.start_as_current_span("service_run") as root_span:
        root_span.set_attribute("service.name", "my-service")
        logger.info("service_started", otel_event=True)
        
        # 业务逻辑
        process_data()
        
        logger.info("service_completed", otel_event=True)

@trace_method(attributes={"layer": "service"})
def process_data():
    logger.info("processing_data")
    # 业务逻辑...

if __name__ == "__main__":
    try:
        main()
    finally:
        shutdown_telemetry()

📖 详细使用指南

1. 日志配置

基础配置

from logtrace.logger import configure_logging, get_logger

configure_logging(
    service_name="my-service",
    environment="production",
    log_level="INFO",
)

logger = get_logger(__name__)
logger.info("user_login", user_id=123)

高级配置

# 带用户上下文提取
def get_user_context():
    """获取当前用户上下文"""
    user = get_current_user()  # 你的用户获取逻辑
    if user and hasattr(user, "email"):
        return {"opt_user_email": user.email}
    return {}

configure_logging(
    service_name="my-service",
    environment="production",
    enable_opentelemetry=True,  # 自动注入 trace_id/span_id
    enable_otel_log_events=True,  # WARNING/ERROR 自动写入 Jaeger
    user_context_extractor=get_user_context,  # 用户上下文
    sensitive_keys=("password", "token", "secret"),  # 敏感字段脱敏
)

2. OpenTelemetry 配置

from logtrace.telemetry import setup_telemetry

setup_telemetry(
    service_name="my-service",
    otlp_endpoint="http://otel-collector:4317",  # OTEL Collector 地址
    environment="production",
    sampling_ratio=1.0,  # 采样率（0.0-1.0）
    metric_interval_ms=60000,  # Metrics 导出间隔（毫秒）
)

环境变量配置（可选）：

export OTLP_ENDPOINT=http://otel-collector:4317
export OTEL_METRICS_ENABLED=false  # 禁用 metrics（如果不需要）

3. FastAPI 集成

方式一：使用 Middleware（推荐）

from fastapi import FastAPI
from fastapi.middleware import Middleware
from logtrace.integrations.fastapi import RequestContextMiddleware, instrument_fastapi

app = FastAPI(
    middleware=[
        Middleware(RequestContextMiddleware),  # 请求上下文中间件
    ]
)

instrument_fastapi(app)  # 启用自动追踪

方式二：使用 add_middleware

from fastapi import FastAPI
from logtrace.integrations.fastapi import RequestContextMiddleware, instrument_fastapi

app = FastAPI()
app.add_middleware(RequestContextMiddleware)
instrument_fastapi(app)

4. 使用装饰器追踪

from logtrace.tracing import trace_method, trace_block

@trace_method(attributes={"layer": "service"})
def fetch_user_data(user_id: int):
    logger.info("fetching_user", user_id=user_id)
    
    # 创建子 span
    with trace_block("validate_user", {"user_id": user_id}):
        validate_user(user_id)
    
    # 关键日志写入 Jaeger
    logger.info("user_fetched", user_id=user_id, otel_event=True)
    return {"user_id": user_id, "name": "Alice"}

5. 日志与追踪关联

自动关联：

  • 所有日志自动包含 request_id（由 RequestContextMiddleware 提供）
  • 当有活动的 span 时，日志自动包含 trace_id 和 span_id
  • 在 Jaeger 中可以通过 trace_id 查看完整的追踪链路
  • 在 Kibana 中可以通过 trace_id 关联日志和追踪

日志字段说明：

  • service：服务名称
  • environment：环境名称
  • request_id：请求 ID（自动生成，响应头中返回 X-Request-ID）
  • trace_id：追踪 ID（用于关联 Jaeger 追踪）
  • span_id：当前 Span ID
  • timestamp：时间戳
  • level：日志级别
  • event：事件名称

🎯 使用场景说明

场景 1：只用 logger（最简单）

logger = get_logger(__name__)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    logger.info("get_user_start", user_id=user_id)
    # 业务逻辑...
    logger.info("get_user_done", user_id=user_id)
    return user

效果：

  • ✅ 日志包含 request_id（由 FastAPI 中间件提供）
  • ✅ 日志包含 trace_id、span_id（由 instrument_fastapi 自动创建）
  • ✅ 日志输出为 JSON 格式

场景 2：使用 @trace_method（关键方法追踪）

@trace_method(attributes={"operation": "fetch_user"})
async def fetch_user_from_db(user_id: int):
    logger.info("fetching_from_db", user_id=user_id)
    # 数据库查询...
    return user

效果：

  • ✅ 在 Jaeger 中可以看到 fetch_user_from_db 的 span
  • ✅ 可以看到该方法的耗时
  • ✅ 日志中的 span_id 是 fetch_user_from_db 的 span ID

场景 3：使用 trace_block（代码块追踪）

@trace_method()
async def process_order(order_id: int):
    logger.info("processing_order", order_id=order_id)
    
    with trace_block("validate_order", {"order_id": order_id}):
        validate_order(order_id)
    
    with trace_block("charge_payment", {"order_id": order_id}):
        charge_payment(order_id)

效果：

  • ✅ 在 Jaeger 中可以看到 validate_order 和 charge_payment 的子 span
  • ✅ 可以看到每个步骤的耗时

场景 4：otel_event=True（重要日志写入 Jaeger）

logger.info("payment_succeeded", order_id=order_id, otel_event=True)
logger.warning("payment_failed", order_id=order_id)  # WARNING 自动写入 Jaeger
logger.error("critical_error", error=str(e))  # ERROR 自动写入 Jaeger

效果：

  • ✅ 这些日志会作为 span event 写入 Jaeger
  • ✅ 可以在 Jaeger 的 span 详情中看到这些事件

⚠️ 注意事项

  1. @trace_method 使用建议：

    • ✅ 用在关键边界上：外部依赖（gRPC 调用、文件对象存储、数据库、第三方 HTTP 调用）
    • ❌ 不建议每个路由 handler 都贴 @trace_method：FastAPI 本身已经有 server span，handler 再贴一层通常"收益不大，噪音翻倍"

  2. gRPC 支持：

    • 暂时不支持 gRPC，只支持 HTTP 协议打点
    • 请勿在 gRPC 服务中打点，可能会出错

  3. 非 FastAPI 项目：

    • 必须手动创建根 span（使用 tracer.start_as_current_span()）
    • 程序退出前调用 shutdown_telemetry() 优雅关闭

  4. 日志级别与 Jaeger：

    • WARNING / ERROR：自动写入 Jaeger span event
    • INFO：默认不写入 Jaeger，只有 otel_event=True 才写入
    • Jaeger 不存"日志文本"，只存 span 和 span event

📚 示例代码

完整的示例代码请查看 examples/ 目录：

  • FastAPI 示例：examples/fastapi_app.py
  • Flask 示例：examples/flask_app.py
  • 非 FastAPI 服务示例：examples/non_fastapi_service.py

运行示例：

# FastAPI 示例
uvicorn examples.fastapi_app:app --host 0.0.0.0 --port 8000

# Flask 示例
python examples/flask_app.py

# 非 FastAPI 服务
python examples/non_fastapi_service.py

🐳 Docker 部署

使用 Docker Compose 一键启动完整的可观测性栈（Jaeger、Elasticsearch、Kibana、OTEL Collector）：

cd deploy
docker-compose up -d

详细部署说明请查看 deploy/README.md。

访问地址：

🔧 配置选项

configure_logging 参数

参数 类型 默认值 说明
log_level str/int "INFO" 日志级别
service_name str None 服务名称
environment str None 环境名称
enable_opentelemetry bool True 启用 OpenTelemetry 上下文注入
enable_otel_log_events bool True 启用日志事件写入 Jaeger
sensitive_keys tuple ("password", "token", ...) 敏感字段列表
user_context_extractor Callable None 用户上下文提取函数

setup_telemetry 参数

参数 类型 默认值 说明
service_name str None 服务名称
otlp_endpoint str "http://localhost:4317" OTEL Collector 地址
environment str "production" 环境名称
sampling_ratio float 1.0 采样率（0.0-1.0）
metric_interval_ms int 60000 Metrics 导出间隔（毫秒）

🔗 相关文档

📝 许可证

MIT License

👥 作者

Ning Liu

About

requestid+traceid+span id, for telemetry and trace request and log

Resources

Readme
License
Activity

Stars

1 star

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages