4、应用接入OTel SDK（Python）：在Python Flask应用中集成OpenTelemetry SDK，自动采集HTTP请求的Trace

本章将指导你如何在典型的 Python Flask 应用中集成 OpenTelemetry SDK，实现 HTTP 请求的自动 Trace 采集。这是将可观测性数据注入值班机器人告警链路的第一步，也是后续实现事件触发告警的数据基础。

4.1 核心组件与集成架构

在 Flask 应用中集成 OTel SDK，主要涉及以下三个核心组件：

组件	作用	对应 Python 包
API	定义 Trace、Span 的数据结构和操作接口	opentelemetry-api
SDK	实现 API 接口，提供 Span 创建、采样、导出等运行时能力	opentelemetry-sdk
Instrumentation	自动拦截 Flask 框架的请求/响应生命周期，生成 Span	opentelemetry-instrumentation-flask

下图展示了集成后的数据流架构：

4.2 环境准备与依赖安装

首先，创建一个新的 Python 虚拟环境并安装必要的依赖包：

# 创建并激活虚拟环境
python3 -m venv otel-flask-env
source otel-flask-env/bin/activate

# 安装 Flask 和 OpenTelemetry 相关包
pip install flask
pip install opentelemetry-api
pip install opentelemetry-sdk
pip install opentelemetry-instrumentation-flask
pip install opentelemetry-exporter-otlp-proto-http  # OTLP HTTP 导出器

4.3 基础 Flask 应用示例

创建一个简单的 Flask 应用文件 app.py，包含两个端点：

# app.py
from flask import Flask, jsonify, request
import time
import random

app = Flask(__name__)

@app.route('/')
def home():
    return jsonify({"message": "Hello, OpenTelemetry!"})

@app.route('/api/data')
def get_data():
    # 模拟业务处理延迟
    delay = random.uniform(0.1, 0.5)
    time.sleep(delay)
    return jsonify({"status": "ok", "delay_seconds": delay})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

4.4 集成 OpenTelemetry SDK

在 app.py 中集成 OTel SDK，添加 Trace 初始化代码：

# app.py — 集成 OpenTelemetry SDK
from flask import Flask, jsonify, request
import time
import random

# ---------- OpenTelemetry 导入 ----------
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
# ---------------------------------------

# 1. 设置 TracerProvider
provider = TracerProvider()
trace.set_tracer_provider(provider)

# 2. 配置 OTLP Exporter（发送到 OTel Collector 或 Jaeger）
exporter = OTLPSpanExporter(
    endpoint="http://localhost:4318/v1/traces",  # OTel Collector 的 OTLP HTTP 端点
)
span_processor = BatchSpanProcessor(exporter)
provider.add_span_processor(span_processor)

# 3. 创建 Tracer 实例
tracer = trace.get_tracer(__name__)

app = Flask(__name__)

# 4. 自动 Instrument Flask 应用
FlaskInstrumentor().instrument_app(app)

@app.route('/')
def home():
    return jsonify({"message": "Hello, OpenTelemetry!"})

@app.route('/api/data')
def get_data():
    # 手动创建子 Span 示例
    with tracer.start_as_current_span("business_logic") as span:
        delay = random.uniform(0.1, 0.5)
        time.sleep(delay)
        span.set_attribute("custom.delay_seconds", delay)
        span.set_attribute("custom.request_id", request.headers.get("X-Request-ID", "unknown"))
    return jsonify({"status": "ok", "delay_seconds": delay})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

4.5 自动采集的 Trace 数据结构

当请求到达 /api/data 时，OTel SDK 会自动生成以下 Trace 结构：

Span 名称	类型	父 Span	自动/手动	包含属性示例
GET /api/data	Server	— (Root)	自动	http.method, http.url, http.status_code
business_logic	Internal	GET /api/data	手动	custom.delay_seconds, custom.request_id

下图展示了单个请求的 Span 层级关系：

4.6 验证集成是否成功

启动应用并发送测试请求，验证 Trace 数据是否被正确采集：

# 启动 Flask 应用（确保 OTel Collector 或 Jaeger 已运行）
python app.py

# 在另一个终端发送测试请求
curl http://localhost:5000/
curl http://localhost:5000/api/data

# 查看控制台输出（如果配置了 Console Exporter）
# 或访问 Jaeger UI (http://localhost:16686) 搜索服务名 "flask-app"

4.7 配置最佳实践

在生产环境中，建议采用以下配置策略：

• 采样策略： 使用 Sampler 控制采样率，避免高流量下产生过多 Trace 数据。例如 ParentBased(trace.sampling.TraceIdRatioBased(0.1)) 表示 10% 采样。
• 资源属性： 通过 Resource 添加服务名称、环境、版本等元数据，便于在 Jaeger 或告警系统中区分不同服务。
• 多 Exporter： 可同时配置 Console Exporter（调试）和 OTLP Exporter（生产），通过 BatchSpanProcessor 异步导出。
• 错误记录： 在手动 Span 中捕获异常并调用 span.record_exception()，为告警提供错误上下文。

# 生产环境配置示例
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased

resource = Resource.create({
    "service.name": "flask-app",
    "service.version": "1.0.0",
    "deployment.environment": "production"
})

provider = TracerProvider(
    resource=resource,
    sampler=ParentBased(TraceIdRatioBased(0.1))  # 10% 采样
)