Python脚本链路追踪Jaeger怎么用

wen 实用脚本 2026-07-19 2

本文目录导读：

安装依赖
基础使用示例
完整的Web应用示例（Flask）
高级配置
运行Jaeger
常见问题解决
最佳实践

我将详细介绍在Python中使用Jaeger进行链路追踪的方法。

安装依赖

首先安装必要的包：

pip install opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation
pip install opentelemetry-exporter-jaeger
# 或者使用OTLP exporter
pip install opentelemetry-exporter-otlp

基础使用示例

1 手动创建Span

from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
# 配置Jaeger exporter
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)
# 配置trace provider
resource = Resource.create({"service.name": "my-python-service"})
provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(jaeger_exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
def business_function():
    with tracer.start_as_current_span("business_operation") as span:
        span.set_attribute("operation.type", "database_query")
        span.add_event("query_started", {"query": "SELECT * FROM users"})
        # 模拟业务逻辑
        result = do_database_query()
        span.add_event("query_completed")
        return result
def do_database_query():
    with tracer.start_as_current_span("database_query") as span:
        # 模拟数据库操作
        import time
        time.sleep(0.1)
        span.set_attribute("db.system", "postgresql")
        span.set_attribute("db.statement", "SELECT * FROM users")
        return {"status": "success"}
# 使用示例
if __name__ == "__main__":
    result = business_function()
    print(f"Result: {result}")

2 使用装饰器自动创建Span

from opentelemetry import trace
from functools import wraps
tracer = trace.get_tracer(__name__)
def trace_function(function_name=None):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            name = function_name or func.__name__
            with tracer.start_as_current_span(name) as span:
                # 添加函数参数作为属性（注意：不要记录敏感信息）
                span.set_attribute("args.count", len(args))
                span.set_attribute("kwargs.count", len(kwargs))
                result = func(*args, **kwargs)
                # 记录结果信息
                span.set_attribute("result.type", type(result).__name__)
                return result
        return wrapper
    return decorator
@trace_function("user_service_operation")
def process_user(user_id):
    # 处理用户逻辑
    with tracer.start_as_current_span("get_user_data") as span:
        span.set_attribute("user.id", user_id)
        # 模拟数据库查询
        return {"user_id": user_id, "name": "John Doe"}
# 使用
result = process_user(12345)

完整的Web应用示例（Flask）

from flask import Flask, request, jsonify
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
import requests
app = Flask(__name__)
# 配置Jaeger
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)
provider = TracerProvider()
processor = BatchSpanProcessor(jaeger_exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
# 自动instrument Flask
FlaskInstrumentor().instrument_app(app)
tracer = trace.get_tracer(__name__)
@app.route('/api/users/<user_id>')
def get_user(user_id):
    with tracer.start_as_current_span("get_user_http") as span:
        span.set_attribute("http.method", request.method)
        span.set_attribute("http.url", request.url)
        # 模拟查询用户
        user_data = query_user_from_db(user_id)
        # 模拟调用外部API
        external_data = call_external_api()
        return jsonify({
            "user": user_data,
            "external_data": external_data
        })
def query_user_from_db(user_id):
    with tracer.start_as_current_span("database_query") as span:
        span.set_attribute("db.system", "mysql")
        span.set_attribute("db.operation", "SELECT")
        span.set_attribute("db.table", "users")
        # 模拟数据库查询
        import time
        time.sleep(0.05)
        return {
            "id": user_id,
            "name": "Test User",
            "email": "test@example.com"
        }
def call_external_api():
    with tracer.start_as_current_span("external_api_call") as span:
        try:
            response = requests.get("https://api.example.com/data", timeout=5)
            span.set_attribute("http.status_code", response.status_code)
            return response.json()
        except Exception as e:
            span.set_attribute("error", str(e))
            span.set_status(trace.Status(trace.StatusCode.ERROR))
            return {"error": str(e)}
if __name__ == '__main__':
    app.run(debug=True, port=5000)

高级配置

1 自定义采样器

from opentelemetry.sdk.trace.sampling import Sampler, Decision
from opentelemetry.sdk.trace import SamplingResult
class CustomSampler(Sampler):
    def should_sample(self, parent_context, trace_id, name, kind, attributes, links, trace_state):
        # 只采样0.5的请求
        import random
        if random.random() < 0.5:
            return SamplingResult(Decision.RECORD_AND_SAMPLE)
        return SamplingResult(Decision.DROP)
# 使用自定义采样器
provider = TracerProvider(sampler=CustomSampler())

2 上下文传播

from opentelemetry import trace, context
from opentelemetry.propagators.textmap import TextMapPropagator
# 创建trace上下文
tracer = trace.get_tracer(__name__)
# 示例：在HTTP请求中传播上下文
def make_http_request_with_tracing(url, headers=None):
    if headers is None:
        headers = {}
    with tracer.start_as_current_span("http_request") as span:
        # 注入trace上下文到headers
        propagator = TextMapPropagator()
        propagator.inject(headers)
        # 发送HTTP请求
        response = requests.get(url, headers=headers)
        span.set_attribute("http.status_code", response.status_code)
        return response

运行Jaeger

使用Docker运行Jaeger：

docker run -d --name jaeger \
  -e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
  -e COLLECTOR_OTLP_ENABLED=true \
  -p 6831:6831/udp \
  -p 6832:6832/udp \
  -p 5778:5778 \
  -p 16686:16686 \
  -p 4317:4317 \
  -p 4318:4318 \
  -p 14250:14250 \
  -p 14268:14268 \
  -p 14269:14269 \
  -p 9411:9411 \
  jaegertracing/all-in-one:latest

常见问题解决

1 配置检查

# 检查追踪是否正常启用
from opentelemetry import trace
tracer = trace.get_tracer(__name__)
current_span = trace.get_current_span()
print(f"Current Span: {current_span}")
# 检查tracer provider
provider = trace.get_tracer_provider()
print(f"Provider: {provider}")

2 错误处理

from opentelemetry import trace
def safe_trace_function():
    with tracer.start_as_current_span("risky_operation") as span:
        try:
            # 可能出错的代码
            result = 1 / 0
        except Exception as e:
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            span.record_exception(e)
            # 重新抛出异常或处理
            raise
        finally:
            span.end()

最佳实践

合理的span粒度：每个span代表一个有意义的操作单元
添加有意义的属性：记录关键的业务信息
处理异常情况：使用span.record_exception()记录异常
控制采样率：避免生产环境过高的开销
保护敏感数据：不要在span属性中记录密码、token等敏感信息

通过以上示例,你应该能够在Python项目中成功集成和使用Jaeger进行链路追踪了。