Python 语言 OpenTelemetry 接入指南

本文档提供 Python 应用接入 OpenTelemetry 的完整指南，包括零代码接入和 SDK 手动接入两种方式。

目录

• 概述
• 零代码接入（推荐）
• SDK 手动接入
• 框架集成示例
• 高级配置
• 最佳实践
• 故障排查

概述

Python 语言特性

Python 的 OpenTelemetry 支持具有以下特点：

• 成熟稳定: API 和 SDK 已达到 1.0 稳定版本
• 丰富的自动埋点: 支持 40+ 主流框架和库的自动埋点
• 简单易用: 通过 opentelemetry-instrument 命令实现零代码接入
• 灵活配置: 支持环境变量和代码配置
• 异步支持: 完整支持 asyncio 和异步框架

版本要求

• Python 版本: 3.7+ (推荐 3.10+)
• OpenTelemetry Python: v1.20.0+
• 支持架构: x86_64, arm64, aarch64

支持的信号

• ✅ Traces: 完全支持
• ✅ Metrics: 完全支持
• ✅ Logs: 完全支持（Python 3.11+）

适用场景

• Web 服务（Flask、Django、FastAPI）
• 微服务架构
• 数据库应用
• 消息队列应用
• 批处理作业
• 异步应用

零代码接入（推荐）

零代码接入是 Python OpenTelemetry 最强大的特性，通过 opentelemetry-instrument 命令即可实现自动埋点，无需修改任何源代码。

快速开始（5 分钟）

1. 安装 OpenTelemetry

# 安装核心包
pip install opentelemetry-distro opentelemetry-exporter-otlp

# 自动安装所有埋点库
opentelemetry-bootstrap -a install

2. 配置环境变量

# 服务信息
export OTEL_SERVICE_NAME=my-python-service
export OTEL_RESOURCE_ATTRIBUTES=service.version=1.0.0,deployment.environment=production

# OTLP Exporter
export OTEL_EXPORTER_OTLP_ENDPOINT=uptrace-grpc.yodao.cn:4317
export OTEL_EXPORTER_OTLP_PROTOCOL=grpc
export OTEL_EXPORTER_OTLP_INSECURE=true # 这个需要设置！！！
export OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=http://token@uptrace-grpc.yodao.cn/project

# Traces 配置
export OTEL_TRACES_EXPORTER=otlp
export OTEL_TRACES_SAMPLER=parentbased_traceidratio
export OTEL_TRACES_SAMPLER_ARG=0.1

# Metrics 配置
export OTEL_METRICS_EXPORTER=otlp
export OTEL_METRIC_EXPORT_INTERVAL=60000

# Logs 配置
export OTEL_LOGS_EXPORTER=otlp

3. 启动应用

# 使用 opentelemetry-instrument 启动
opentelemetry-instrument python app.py

# 或使用模块方式
opentelemetry-instrument python -m myapp

# 或使用 gunicorn/uwsgi
opentelemetry-instrument gunicorn app:app
opentelemetry-instrument uwsgi --module app:app

就这么简单！您的应用现在已经自动埋点并开始发送遥测数据了。

详细步骤

步骤 1: 安装依赖

最小安装（仅核心功能）:

pip install opentelemetry-distro opentelemetry-exporter-otlp

完整安装（推荐，包含所有自动埋点库）:

pip install opentelemetry-distro opentelemetry-exporter-otlp

# 自动安装所有支持的埋点库
opentelemetry-bootstrap -a install

手动安装特定埋点库:

# Web 框架
pip install opentelemetry-instrumentation-flask
pip install opentelemetry-instrumentation-django
pip install opentelemetry-instrumentation-fastapi

# 数据库
pip install opentelemetry-instrumentation-sqlalchemy
pip install opentelemetry-instrumentation-psycopg2
pip install opentelemetry-instrumentation-mysql

# HTTP 客户端
pip install opentelemetry-instrumentation-requests
pip install opentelemetry-instrumentation-httpx
pip install opentelemetry-instrumentation-aiohttp-client

# 消息队列
pip install opentelemetry-instrumentation-kafka-python
pip install opentelemetry-instrumentation-celery
pip install opentelemetry-instrumentation-pika

# RPC
pip install opentelemetry-instrumentation-grpc

# 日志
pip install opentelemetry-instrumentation-logging

步骤 2: 配置环境变量

创建 .env 文件（推荐）:

# .env
OTEL_SERVICE_NAME=my-python-service
OTEL_RESOURCE_ATTRIBUTES=service.version=1.0.0,deployment.environment=production

OTEL_EXPORTER_OTLP_ENDPOINT=uptrace-grpc.yodao.cn:4317
OTEL_EXPORTER_OTLP_PROTOCOL=grpc
OTEL_EXPORTER_OTLP_HEADERS=uptrace-dsn=http://token@uptrace-grpc.yodao.cn/project

OTEL_TRACES_EXPORTER=otlp
OTEL_TRACES_SAMPLER=parentbased_traceidratio
OTEL_TRACES_SAMPLER_ARG=0.1

OTEL_METRICS_EXPORTER=otlp
OTEL_METRIC_EXPORT_INTERVAL=60000

OTEL_LOGS_EXPORTER=otlp
OTEL_LOG_LEVEL=info

加载环境变量:

# 使用 python-dotenv
pip install python-dotenv

# 在代码中加载
from dotenv import load_dotenv
load_dotenv()

# 或使用 export
export $(cat .env | xargs)

步骤 3: 启动应用

方式 1: 直接启动

opentelemetry-instrument python app.py

方式 2: 模块方式

opentelemetry-instrument python -m myapp

方式 3: WSGI 服务器

# Gunicorn
opentelemetry-instrument gunicorn -w 4 -b 0.0.0.0:8000 app:app

# uWSGI
opentelemetry-instrument uwsgi --http :8000 --module app:app

# Waitress
opentelemetry-instrument waitress-serve --listen=0.0.0.0:8000 app:app

方式 4: ASGI 服务器（FastAPI、Starlette）

# Uvicorn
opentelemetry-instrument uvicorn app:app --host 0.0.0.0 --port 8000

# Hypercorn
opentelemetry-instrument hypercorn app:app --bind 0.0.0.0:8000

方式 5: Django

# 开发服务器
opentelemetry-instrument python manage.py runserver

# 生产环境（Gunicorn）
opentelemetry-instrument gunicorn myproject.wsgi:application

方式 6: Celery Worker

opentelemetry-instrument celery -A myapp worker --loglevel=info

支持的框架和库

Web 框架

数据库

HTTP 客户端

消息队列

RPC 框架

日志库

启用/禁用特定埋点库

通过环境变量

# 启用特定埋点库
export OTEL_PYTHON_INSTRUMENTATION_ENABLED=true
export OTEL_PYTHON_INSTRUMENTATION_FLASK_ENABLED=true
export OTEL_PYTHON_INSTRUMENTATION_REQUESTS_ENABLED=true

# 禁用特定埋点库
export OTEL_PYTHON_INSTRUMENTATION_SQLALCHEMY_ENABLED=false
export OTEL_PYTHON_INSTRUMENTATION_REDIS_ENABLED=false

通过配置文件

创建 instrumentation.yaml:

instrumentation:
  flask:
    enabled: true
  django:
    enabled: true
  requests:
    enabled: true
  sqlalchemy:
    enabled: false

验证零代码接入

启动应用后，访问您的服务端点，然后在 APM 平台查看数据：

1. Traces: 查看调用链路
2. Metrics: 查看请求量、延迟等指标
3. Service Map: 查看服务拓扑图

快速验证脚本:

# verify.py
import requests
import time

print("Sending test request...")
response = requests.get("http://localhost:8000/health")
print(f"Status: {response.status_code}")

print("Waiting for traces to be exported...")
time.sleep(5)

print("Check your APM platform for service:", os.getenv("OTEL_SERVICE_NAME"))

运行:

opentelemetry-instrument python verify.py

SDK 手动接入

SDK 手动接入提供更精细的控制和自定义能力。

安装依赖

# 核心 SDK
pip install opentelemetry-api opentelemetry-sdk

# OTLP Exporter
pip install opentelemetry-exporter-otlp-proto-grpc

# 埋点库（可选）
pip install opentelemetry-instrumentation-flask
pip install opentelemetry-instrumentation-requests
pip install opentelemetry-instrumentation-sqlalchemy

基础初始化

1. 创建 OpenTelemetry 初始化函数

# otel.py
import os
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.resources import Resource
import opentelemetry.semconv.resource as semconv_resource

def init_opentelemetry():
    """初始化 OpenTelemetry"""
    
    # 1. 创建 Resource
    resource = Resource.create({
        semconv_resource.SERVICE_NAME: os.getenv("OTEL_SERVICE_NAME", "my-service"),
        semconv_resource.SERVICE_VERSION: os.getenv("SERVICE_VERSION", "1.0.0"),
        "deployment.environment": os.getenv("DEPLOYMENT_ENVIRONMENT", "production"),
    })
    
    # 2. 创建 Trace Exporter
    trace_exporter = OTLPSpanExporter(
        endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
        headers={
            "uptrace-dsn": os.getenv("OTEL_EXPORTER_OTLP_HEADERS", ""),
        },
    )
    
    # 3. 创建 Trace Provider
    tracer_provider = TracerProvider(resource=resource)
    tracer_provider.add_span_processor(BatchSpanProcessor(trace_exporter))
    trace.set_tracer_provider(tracer_provider)
    
    # 4. 创建 Metric Exporter
    metric_exporter = OTLPMetricExporter(
        endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
        headers={
            "uptrace-dsn": os.getenv("OTEL_EXPORTER_OTLP_HEADERS", ""),
        },
    )
    
    # 5. 创建 Metric Provider
    metric_reader = PeriodicExportingMetricReader(
        metric_exporter,
        export_interval_millis=60000,  # 60 秒
    )
    meter_provider = MeterProvider(
        resource=resource,
        metric_readers=[metric_reader],
    )
    metrics.set_meter_provider(meter_provider)
    
    return tracer_provider, meter_provider

def shutdown_opentelemetry(tracer_provider, meter_provider):
    """关闭 OpenTelemetry"""
    tracer_provider.shutdown()
    meter_provider.shutdown()

2. 在应用中使用

# app.py
import atexit
from otel import init_opentelemetry, shutdown_opentelemetry

# 初始化
tracer_provider, meter_provider = init_opentelemetry()

# 注册退出处理
atexit.register(shutdown_opentelemetry, tracer_provider, meter_provider)

# 应用逻辑
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def main():
    with tracer.start_as_current_span("main_operation") as span:
        span.set_attribute("operation.type", "main")
        # 业务逻辑
        pass

if __name__ == "__main__":
    main()

Traces - 手动埋点

创建和使用 Span

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer(__name__)

def process_order(order_id: str):
    """处理订单"""
    with tracer.start_as_current_span("process_order") as span:
        # 设置属性
        span.set_attribute("order.id", order_id)
        span.set_attribute("operation", "process")
        
        # 添加事件
        span.add_event("order_validation_started")
        
        # 验证订单
        try:
            validate_order(order_id)
            span.add_event("order_validated")
            
            # 处理支付
            process_payment(order_id)
            span.add_event("payment_processed")
            
            # 设置成功状态
            span.set_status(Status(StatusCode.OK))
            span.set_attribute("order.success", True)
            
        except Exception as e:
            # 记录错误
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

def validate_order(order_id: str):
    """验证订单"""
    with tracer.start_as_current_span("validate_order") as span:
        span.set_attribute("order.id", order_id)
        # 验证逻辑
        pass

def process_payment(order_id: str):
    """处理支付"""
    with tracer.start_as_current_span("process_payment") as span:
        span.set_attribute("order.id", order_id)
        span.set_attribute("payment.method", "credit_card")
        # 支付逻辑
        pass

Span 属性和事件

from opentelemetry import trace
from opentelemetry.trace import SpanKind

tracer = trace.get_tracer(__name__)

def example_span_attributes():
    """Span 属性和事件示例"""
    with tracer.start_as_current_span(
        "example_operation",
        kind=SpanKind.INTERNAL,
    ) as span:
        # 设置属性
        span.set_attribute("user.id", "12345")
        span.set_attribute("user.email", "user@example.com")
        span.set_attribute("user.age", 30)
        span.set_attribute("user.premium", True)
        span.set_attribute("user.roles", ["admin", "user"])
        
        # 添加事件
        span.add_event(
            "cache_miss",
            attributes={
                "cache.key": "user:12345",
                "cache.type": "redis",
            },
        )
        
        # 添加带时间戳的事件
        from datetime import datetime
        span.add_event(
            "retry_attempt",
            timestamp=datetime.now().timestamp() * 1e9,  # 纳秒
            attributes={
                "retry.count": 2,
                "retry.reason": "timeout",
            },
        )

Context 传播

from opentelemetry import trace, propagate
from opentelemetry.propagate import inject, extract
import requests

# HTTP 客户端传播
def make_http_request(url: str):
    """发送 HTTP 请求并传播 context"""
    tracer = trace.get_tracer(__name__)
    
    with tracer.start_as_current_span("http_request") as span:
        span.set_attribute("http.url", url)
        span.set_attribute("http.method", "GET")
        
        # 注入 trace context 到 HTTP headers
        headers = {}
        inject(headers)
        
        response = requests.get(url, headers=headers)
        
        span.set_attribute("http.status_code", response.status_code)
        return response

# HTTP 服务端提取
from flask import Flask, request

app = Flask(__name__)

@app.route("/api/users/<user_id>")
def get_user(user_id: str):
    """获取用户信息"""
    tracer = trace.get_tracer(__name__)
    
    # 从 HTTP headers 提取 trace context
    context = extract(request.headers)
    
    with tracer.start_as_current_span("get_user", context=context) as span:
        span.set_attribute("user.id", user_id)
        # 业务逻辑
        return {"id": user_id, "name": "John Doe"}

Metrics - 指标采集

Counter（计数器）

from opentelemetry import metrics

meter = metrics.get_meter(__name__)

# 创建 Counter
request_counter = meter.create_counter(
    name="http.server.requests",
    description="Total number of HTTP requests",
    unit="1",
)

def handle_request(method: str, route: str, status_code: int):
    """处理请求并记录指标"""
    request_counter.add(
        1,
        attributes={
            "http.method": method,
            "http.route": route,
            "http.status_code": status_code,
        },
    )

Histogram（直方图）

from opentelemetry import metrics

meter = metrics.get_meter(__name__)

# 创建 Histogram
request_duration = meter.create_histogram(
    name="http.server.duration",
    description="HTTP request duration",
    unit="ms",
)

def record_duration(duration_ms: float, method: str, route: str):
    """记录请求持续时间"""
    request_duration.record(
        duration_ms,
        attributes={
            "http.method": method,
            "http.route": route,
        },
    )

UpDownCounter（升降计数器）

from opentelemetry import metrics

meter = metrics.get_meter(__name__)

# 创建 UpDownCounter
active_connections = meter.create_up_down_counter(
    name="http.server.active_connections",
    description="Number of active HTTP connections",
    unit="1",
)

def connection_opened():
    """连接打开"""
    active_connections.add(1)

def connection_closed():
    """连接关闭"""
    active_connections.add(-1)

Observable Gauge（可观察仪表）

from opentelemetry import metrics
import psutil

meter = metrics.get_meter(__name__)

# 创建 Observable Gauge
memory_usage = meter.create_observable_gauge(
    name="runtime.memory.heap_alloc",
    description="Heap memory allocated",
    unit="bytes",
    callbacks=[get_memory_usage],
)

def get_memory_usage(options):
    """获取内存使用量"""
    process = psutil.Process()
    memory_info = process.memory_info()
    memory_usage.observe(
        memory_info.rss,
        attributes={},
    )

Logs - 日志集成

集成标准 logging

import logging
from opentelemetry import trace
from opentelemetry.instrumentation.logging import LoggingInstrumentor

# 启用日志埋点
LoggingInstrumentor().instrument()

# 创建 logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 使用 logger（自动添加 trace 信息）
tracer = trace.get_tracer(__name__)

def example_logging():
    """日志示例"""
    with tracer.start_as_current_span("example_operation") as span:
        # 日志会自动包含 trace_id 和 span_id
        logger.info(
            "Processing request",
            extra={
                "user_id": "12345",
                "action": "create",
            },
        )

自定义日志格式

import logging
from opentelemetry import trace
from opentelemetry.trace import format_trace_id, format_span_id

class TraceFormatter(logging.Formatter):
    """自定义日志格式，包含 trace 信息"""
    
    def format(self, record):
        # 获取当前 span
        span = trace.get_current_span()
        if span and span.get_span_context().is_valid():
            span_context = span.get_span_context()
            record.trace_id = format_trace_id(span_context.trace_id)
            record.span_id = format_span_id(span_context.span_id)
        else:
            record.trace_id = None
            record.span_id = None
        
        return super().format(record)

# 配置 logger
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(TraceFormatter(
    '%(asctime)s [%(levelname)s] trace_id=%(trace_id)s span_id=%(span_id)s %(message)s'
))
logger.addHandler(handler)
logger.setLevel(logging.INFO)

框架集成示例

Flask

# app.py
from flask import Flask, jsonify
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry import trace

app = Flask(__name__)

# 启用 Flask 自动埋点
FlaskInstrumentor().instrument_app(app)

tracer = trace.get_tracer(__name__)

@app.route("/users/<user_id>")
def get_user(user_id):
    """获取用户信息"""
    with tracer.start_as_current_span("get_user_handler") as span:
        span.set_attribute("user.id", user_id)
        
        # 查询用户
        user = query_user(user_id)
        
        return jsonify(user)

@app.route("/users", methods=["POST"])
def create_user():
    """创建用户"""
    with tracer.start_as_current_span("create_user_handler") as span:
        data = request.get_json()
        
        user = save_user(data)
        
        return jsonify(user), 201

def query_user(user_id: str):
    """查询用户"""
    with tracer.start_as_current_span("query_user") as span:
        span.set_attribute("db.operation", "SELECT")
        # 数据库查询逻辑
        return {"id": user_id, "name": "John Doe"}

def save_user(data: dict):
    """保存用户"""
    with tracer.start_as_current_span("save_user") as span:
        span.set_attribute("db.operation", "INSERT")
        # 数据库插入逻辑
        return data

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

启动:

opentelemetry-instrument python app.py

Django

settings.py 配置

# settings.py
INSTALLED_APPS = [
    # ... 其他应用
]

# OpenTelemetry 配置
from opentelemetry.instrumentation.django import DjangoInstrumentor
from opentelemetry.instrumentation.psycopg2 import Psycopg2Instrumentor

# 启用 Django 自动埋点
DjangoInstrumentor().instrument()

# 启用 PostgreSQL 自动埋点
Psycopg2Instrumentor().instrument()

views.py

# views.py
from django.http import JsonResponse
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def get_user(request, user_id):
    """获取用户信息"""
    with tracer.start_as_current_span("get_user_view") as span:
        span.set_attribute("user.id", user_id)
        
        # 业务逻辑
        user = {"id": user_id, "name": "John Doe"}
        
        return JsonResponse(user)

启动:

opentelemetry-instrument python manage.py runserver

FastAPI

# main.py
from fastapi import FastAPI
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry import trace

app = FastAPI()

# 启用 FastAPI 自动埋点
FastAPIInstrumentor.instrument_app(app)

tracer = trace.get_tracer(__name__)

@app.get("/users/{user_id}")
async def get_user(user_id: str):
    """获取用户信息"""
    with tracer.start_as_current_span("get_user_handler") as span:
        span.set_attribute("user.id", user_id)
        
        user = await query_user(user_id)
        
        return user

@app.post("/users")
async def create_user(user: dict):
    """创建用户"""
    with tracer.start_as_current_span("create_user_handler") as span:
        saved_user = await save_user(user)
        
        return saved_user

async def query_user(user_id: str):
    """查询用户"""
    with tracer.start_as_current_span("query_user") as span:
        span.set_attribute("db.operation", "SELECT")
        # 异步数据库查询
        return {"id": user_id, "name": "John Doe"}

async def save_user(user: dict):
    """保存用户"""
    with tracer.start_as_current_span("save_user") as span:
        span.set_attribute("db.operation", "INSERT")
        # 异步数据库插入
        return user

启动:

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

SQLAlchemy

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry import trace

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# 创建引擎
engine = create_engine("postgresql://user:password@localhost/dbname")

# 启用 SQLAlchemy 自动埋点
SQLAlchemyInstrumentor().instrument(
    engine=engine,
    enable_commenter=True,  # 添加 SQL 注释
    commenter_options={"db_driver": True},
)

Session = sessionmaker(bind=engine)

tracer = trace.get_tracer(__name__)

def get_user(user_id: int):
    """获取用户"""
    with tracer.start_as_current_span("get_user") as span:
        session = Session()
        try:
            user = session.query(User).filter(User.id == user_id).first()
            span.set_attribute("db.rows_affected", 1 if user else 0)
            return user
        finally:
            session.close()

Redis

import redis
from opentelemetry.instrumentation.redis import RedisInstrumentor
from opentelemetry import trace

# 创建 Redis 客户端
r = redis.Redis(host="localhost", port=6379, db=0)

# 启用 Redis 自动埋点
RedisInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

def get_user_from_cache(user_id: str):
    """从缓存获取用户"""
    with tracer.start_as_current_span("get_user_from_cache") as span:
        span.set_attribute("cache.key", f"user:{user_id}")
        
        user = r.get(f"user:{user_id}")
        
        if user:
            span.set_attribute("cache.hit", True)
        else:
            span.set_attribute("cache.hit", False)
        
        return user

Requests (HTTP 客户端)

import requests
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry import trace

# 启用 Requests 自动埋点
RequestsInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

def call_external_api(url: str):
    """调用外部 API"""
    with tracer.start_as_current_span("call_external_api") as span:
        span.set_attribute("http.url", url)
        span.set_attribute("http.method", "GET")
        
        response = requests.get(url)
        
        span.set_attribute("http.status_code", response.status_code)
        
        return response.json()

Celery

from celery import Celery
from opentelemetry.instrumentation.celery import CeleryInstrumentor
from opentelemetry import trace

app = Celery("myapp", broker="redis://localhost:6379/0")

# 启用 Celery 自动埋点
CeleryInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

@app.task
def process_order(order_id: str):
    """处理订单任务"""
    with tracer.start_as_current_span("process_order_task") as span:
        span.set_attribute("order.id", order_id)
        span.set_attribute("task.name", "process_order")
        
        # 任务逻辑
        pass

启动 Worker:

opentelemetry-instrument celery -A myapp worker --loglevel=info

gRPC

Server

import grpc
from opentelemetry.instrumentation.grpc import GrpcInstrumentorServer
from opentelemetry import trace

# 启用 gRPC Server 自动埋点
GrpcInstrumentorServer().instrument()

tracer = trace.get_tracer(__name__)

class UserService(users_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        """获取用户"""
        with tracer.start_as_current_span("GetUser") as span:
            span.set_attribute("user.id", request.id)
            
            # 业务逻辑
            return users_pb2.UserResponse(
                id=request.id,
                name="John Doe",
            )

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
users_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
server.add_insecure_port("[::]:50051")
server.start()

Client

import grpc
from opentelemetry.instrumentation.grpc import GrpcInstrumentorClient
from opentelemetry import trace

# 启用 gRPC Client 自动埋点
GrpcInstrumentorClient().instrument()

tracer = trace.get_tracer(__name__)

def call_user_service(user_id: str):
    """调用用户服务"""
    with tracer.start_as_current_span("call_user_service") as span:
        channel = grpc.insecure_channel("localhost:50051")
        stub = users_pb2_grpc.UserServiceStub(channel)
        
        response = stub.GetUser(users_pb2.GetUserRequest(id=user_id))
        
        return response

高级配置

自定义采样策略

from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased, ParentBased

# 创建采样器
sampler = ParentBased(TraceIdRatioBased(0.1))  # 10% 采样

# 应用到 TracerProvider
tracer_provider = TracerProvider(sampler=sampler)

自定义属性添加

from opentelemetry.sdk.trace import SpanProcessor, ReadWriteSpan

class AttributeAdder(SpanProcessor):
    """添加自定义属性到所有 Span"""
    
    def on_start(self, span: ReadWriteSpan, parent_context=None):
        span.set_attribute("environment", "production")
        span.set_attribute("region", "us-east-1")
        span.set_attribute("version", "1.0.0")
    
    def on_end(self, span):
        pass
    
    def shutdown(self):
        pass
    
    def force_flush(self, timeout_millis=30000):
        pass

# 使用
tracer_provider.add_span_processor(AttributeAdder())

多 Exporter 配置

from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, BatchSpanProcessor

# 创建多个 Exporter
otlp_exporter = OTLPSpanExporter(endpoint="uptrace-grpc.yodao.cn:4317")
jaeger_exporter = JaegerExporter(agent_host_name="localhost", agent_port=6831)
console_exporter = ConsoleSpanExporter()

# 添加到 TracerProvider
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
tracer_provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
tracer_provider.add_span_processor(BatchSpanProcessor(console_exporter))

性能优化配置

from opentelemetry.sdk.trace.export import BatchSpanProcessor

# 优化批量导出配置
batch_processor = BatchSpanProcessor(
    exporter,
    max_queue_size=2048,        # 队列大小
    max_export_batch_size=512,  # 批量大小
    export_timeout_millis=30000,  # 导出超时
    schedule_delay_millis=5000,    # 批量延迟
)

tracer_provider.add_span_processor(batch_processor)

最佳实践

1. 使用环境变量配置

import os
from opentelemetry import trace, metrics

# 从环境变量读取配置
service_name = os.getenv("OTEL_SERVICE_NAME", "my-service")
endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")

2. 优雅关闭

import atexit
from opentelemetry.sdk.trace import TracerProvider

tracer_provider = TracerProvider()

# 注册退出处理
atexit.register(tracer_provider.shutdown)

3. 错误处理

from opentelemetry.trace import Status, StatusCode

try:
    # 业务逻辑
    pass
except Exception as e:
    span.record_exception(e)
    span.set_status(Status(StatusCode.ERROR, str(e)))
    raise

4. 异步支持

import asyncio
from opentelemetry import trace

async def async_operation():
    """异步操作"""
    tracer = trace.get_tracer(__name__)
    
    with tracer.start_as_current_span("async_operation") as span:
        # 异步业务逻辑
        await asyncio.sleep(1)
        pass

故障排查

常见问题

1. 数据未发送到后端

排查步骤:

1. 检查环境变量:

echo $OTEL_SERVICE_NAME
echo $OTEL_EXPORTER_OTLP_ENDPOINT

2. 启用调试日志:

export OTEL_LOG_LEVEL=debug
opentelemetry-instrument python app.py

3. 使用 Console Exporter 验证:

export OTEL_TRACES_EXPORTER=console
opentelemetry-instrument python app.py

2. 框架未自动埋点

解决方案:

1. 确认已安装对应的埋点库:

pip list | grep opentelemetry-instrumentation

2. 检查埋点库是否启用:

export OTEL_PYTHON_INSTRUMENTATION_FLASK_ENABLED=true

3. 查看支持的框架列表

3. 性能问题

解决方案:

1. 降低采样率:

export OTEL_TRACES_SAMPLER_ARG=0.01

2. 调整批量配置:

BatchSpanProcessor(
    exporter,
    max_queue_size=1024,
    schedule_delay_millis=10000,
)

下一步

• 环境变量配置 - 了解所有配置选项
• 高级配置 - 深入了解采样和性能优化
• 故障排查 - 常见问题解决方案

参考资源

• OpenTelemetry Python Documentation
• OpenTelemetry Python SDK
• OpenTelemetry Python Contrib
• Python Semantic Conventions
• OpenTelemetry-Python API Reference