通过 OpenTelemetry Python SDK 为调用链增加自定义埋点

适用范围

使用自定义埋点功能前，需要满足以下条件：

• 已接入 ARMS 应用监控，具体操作参见应用监控接入概述。

• ARMS Python 探针版本为 1.6.0 及以上，升级操作参见如何安装指定地域与版本的探针。

基本概念

以下为自定义埋点涉及的常用术语，更多名词解释参见 OpenTelemetry Specification。

• Span：一次请求的一个具体操作，例如远程调用入口或内部方法调用。

• SpanContext：一次请求追踪的上下文，包含 traceId、spanId 等信息。

• Attribute：Span 的附加属性字段，用于记录关键信息。

• Baggage：在整条链路上透传的键值对信息。

安装 OpenTelemetry Python SDK

执行以下命令安装 OpenTelemetry Python SDK，详细安装选项参见 OpenTelemetry 官方文档。

pip install opentelemetry-api

完整示例

以下示例展示 Flask 应用如何综合使用 OpenTelemetry SDK 完成 Span 创建、Attribute 设置、Baggage 透传和 TraceId 获取。各操作的独立用法参见常用操作代码片段。

from flask import Flask
from opentelemetry import trace, context, baggage
from opentelemetry.trace import StatusCode, SpanKind

app = Flask(__name__)

# 获取 Tracer 实例
tracer = trace.get_tracer("manual-sdk", "1.0.0")


@app.route("/parent")
def parent():
    # 创建一个名为 "parent" 的 Span
    with tracer.start_as_current_span("parent", kind=SpanKind.SERVER) as span:
        # 使用 Baggage 透传业务自定义标签
        ctx = baggage.set_baggage("user.id", "1")
        ctx = baggage.set_baggage("user.name", "name", context=ctx)

        # 为 Span 增加 Attribute
        span.set_attribute("http.method", "GET")
        span.set_attribute("http.uri", "/parent")

        # 在 Baggage 上下文中调用子方法
        token = context.attach(ctx)
        try:
            child()
        finally:
            context.detach(token)

    return "parent"


def child():
    # 创建一个名为 "child" 的 Span（自动作为 "parent" 的子 Span）
    with tracer.start_as_current_span("child") as span:
        import time

        # 获取当前 Trace 上下文并打印 traceId
        current_span = trace.get_current_span()
        span_context = current_span.get_span_context()
        print(f"current traceId = {format(span_context.trace_id, '032x')}")

        # 获取 Baggage 中的值
        user_id = baggage.get_baggage("user.id")
        print(f"userId in baggage = {user_id}")

        time.sleep(1)


# 定时任务示例：手动创建独立的 Root Span
import threading

def scheduled_task():
    with tracer.start_as_current_span("schedule") as span:
        span.set_attribute("schedule.time", int(__import__('time').time() * 1000))
        try:
            import time
            print("scheduled!")
            time.sleep(0.5)
            span.set_attribute("schedule.success", True)

            # 获取 TraceId
            current_span = trace.get_current_span()
            print(f"traceId = {format(current_span.get_span_context().trace_id, '032x')}")
        except Exception as e:
            span.set_status(StatusCode.ERROR, str(e))


def start_scheduler():
    import time
    while True:
        time.sleep(30)
        scheduled_task()

# 启动定时任务线程
scheduler_thread = threading.Thread(target=start_scheduler, daemon=True)
scheduler_thread.start()

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

示例代码包含以下三个部分：

1. parent 方法：处理 /parent 路由请求，演示以下操作：

   • 创建名为 parent 的 Span，方法结束时自动关闭。

   • 通过 Baggage SDK 设置 user.id 和 user.name 两对 Baggage，后续透传到下游应用。

   • 为 Span 设置 http.method 和 http.uri 两个 Attribute。

2. child 方法：作为 parent 的子方法调用，演示以下操作：

   • 创建名为 child 的 Span，自动作为 parent Span 的子 Span。

   • 获取当前 Trace 上下文并打印 traceId。

   • 获取上游透传的 Baggage 值并打印。

3. scheduled_task 方法：展示非 Web 请求场景（如定时任务、后台线程）的埋点方式。这类场景没有自动埋点注入的上下文，因此需要手动创建 Root Span。任务开始时创建 Span，任务结束时自动关闭。

常用操作代码片段

from opentelemetry import trace
from opentelemetry.trace import StatusCode

tracer = trace.get_tracer("my-service", "1.0.0")

def my_business_method():
    with tracer.start_as_current_span("my-operation") as span:
        # 设置 Attribute
        span.set_attribute("business.key", "value")
        span.set_attribute("business.count", 42)

        try:
            # 业务逻辑
            result = do_something()
            span.set_attribute("business.result", result)
        except Exception as e:
            span.set_status(StatusCode.ERROR, str(e))
            span.record_exception(e)
            raise

from opentelemetry import trace

def get_trace_info():
    current_span = trace.get_current_span()
    span_context = current_span.get_span_context()

    trace_id = format(span_context.trace_id, '032x')
    span_id = format(span_context.span_id, '016x')

    print(f"TraceId: {trace_id}")
    print(f"SpanId: {span_id}")

    return trace_id, span_id

from opentelemetry import context, baggage

def set_baggage_example():
    # 设置 Baggage
    ctx = baggage.set_baggage("user.id", "12345")
    ctx = baggage.set_baggage("tenant.id", "tenant-abc", context=ctx)

    # 激活上下文
    token = context.attach(ctx)
    try:
        # 在此范围内，下游调用都可以获取到这些 Baggage
        downstream_call()
    finally:
        context.detach(token)

def downstream_call():
    # 获取 Baggage
    user_id = baggage.get_baggage("user.id")
    tenant_id = baggage.get_baggage("tenant.id")
    print(f"user.id = {user_id}, tenant.id = {tenant_id}")