如何为基于warp、reqwest的Rust微服务添加符合b3规范的Jaeger链路追踪？

1. 前置依赖引入 #

首先在Cargo.toml中添加链路追踪相关依赖，以下版本组合经过兼容性验证可正常使用：

[dependencies]
opentelemetry = { version = "0.18", features = ["trace"] }
opentelemetry-jaeger = { version = "0.17", features = ["rt-tokio"] }
opentelemetry-http = "0.7"
tracing = "0.1"
tracing-opentelemetry = "0.18"
tracing-futures = "0.2"
warp = { version = "0.3", features = ["tracing"] }
reqwest = { version = "0.11", features = ["json", "native-tls"] }
http = "0.2"

2. 全局追踪初始化 #

在程序启动入口main函数中先初始化Jaeger追踪器和b3格式传播器，代码如下：

use opentelemetry::{global, propagation::TextMapPropagator};
use opentelemetry_jaeger::Propagator;
use tracing_opentelemetry::OpenTelemetryLayer;
use tracing_subscriber::prelude::*;

fn init_tracing() {
    // 全局注册b3格式传播器，自动处理b3头的提取、注入逻辑
    global::set_text_map_propagator(Propagator::new());

    // 初始化Jaeger Tracer
    let tracer = opentelemetry_jaeger::new_agent_pipeline()
        .with_service_name("your-service-name") // 替换为你的微服务名称
        .install_simple()
        .expect("Jaeger tracer初始化失败");

    // 注册tracing层到全局日志系统
    tracing_subscriber::registry()
        .with(OpenTelemetryLayer::new(tracer))
        .init();
}

在启动warp服务前先调用init_tracing()完成初始化。

3. warp接口侧链路提取与自定义span #

调整路由配置增加请求头提取逻辑，修改后的version接口代码如下：

use warp::Filter;
use opentelemetry::global;
use opentelemetry_http::HeaderExtractor;
use tracing::Span;
use tracing_opentelemetry::OpenTelemetrySpanExt;

// 路由配置示例，新增请求头提取规则
let routes = warp::path("version")
    .and(warp::get())
    .and(warp::header::headers_cloned())
    .and_then(version);

pub async fn version(headers: http::HeaderMap) -> Result<impl warp::Reply, std::convert::Infallible> {
    // 从请求头提取上游传入的b3链路上下文
    let parent_context = global::get_text_map_propagator(|propagator| {
        propagator.extract(&HeaderExtractor(&headers))
    });

    // 创建自定义span，可自行定义span名称和需要记录的字段
    let span = tracing::info_span!("version_api", trace_id = tracing::field::Empty, api = "/version");
    // 绑定父上下文实现链路透传
    span.set_parent(parent_context);
    // 记录trace id到span字段，方便排查
    span.record("trace_id", span.context().span().span_context().trace_id().to_string());

    // 进入span上下文，后续逻辑的链路都会关联到该span
    let _enter = span.enter();
    
    // 原有业务逻辑保持不变
    Ok(warp::reply::with_status(VERSION, http::StatusCode::OK))
}

4. reqwest调用侧链路注入 #

修改reqwest调用代码，将当前链路上下文以b3格式注入到请求头透传给下游服务：

use opentelemetry::global;
use opentelemetry_http::HeaderInjector;
use tracing::Span;
use tracing_opentelemetry::OpenTelemetrySpanExt;

pub async fn get_document_content_as_text(
    account_id: &str,
    hash: &str,
) -> Result<String, Box<dyn std::error::Error>> {
    // 建议client全局复用不要每次调用新建，此处仅为保持原有逻辑不变
    let client = reqwest::Client::builder().build()?;
    
    // 创建下游调用的自定义span，可添加业务字段便于排查
    let span = tracing::info_span!("get_document_content", account_id, hash);
    let _enter = span.enter();

    let mut req = client
        .get(url) // 替换为实际请求地址
        .bearer_auth(TOKEN)
        .build()?;

    // 将当前链路上下文以b3格式注入到请求头
    global::get_text_map_propagator(|propagator| {
        propagator.inject_context(
            &Span::current().context(),
            &mut HeaderInjector(req.headers_mut()),
        )
    });

    let res = client.execute(req).await?;
    // 可自行向span添加请求状态等自定义信息
    if res.status().is_success() {
        span.record("status", "success");
    } else {
        span.record("status", "fail");
        span.record("http_code", res.status().as_u16());
    }
    
    let text = res.text().await?;
    Ok(text)
}

注意事项 #

• b3格式的请求头由传播器自动处理，不需要手动拼接X-B3-TraceId、X-B3-SpanId等字段
• 自定义span可添加任意业务字段，Jaeger界面支持直接根据业务字段搜索对应链路
• reqwest::Client建议全局初始化复用，避免每次请求新建连接导致性能损耗
• 如果上游请求没有携带b3头，初始化的span会自动生成新的trace id作为整条链路的根节点

内容的提问来源于stack exchange，提问作者Edelherb