Actix/Rust微服务如何在日志中注入B3传播的traceId/spanId？

核心思路 #

Rust没有Java那样的全局MDC，但可以通过请求上下文传递结合日志生态实现相同效果。推荐用tracing生态（天然支持异步上下文传递），也可以用纯log crate方案兼容现有代码。

方案一：使用tracing生态（推荐） #

tracing的Span上下文能自动在Actix的异步请求中传递，配合tracing-log桥接log宏，再用ecs_logger的tracing格式化支持，就能让所有info!()/warn!()/error!()日志自动携带traceId。

1. 添加依赖 #

[dependencies]
actix-web = "4"
log = "0.4"
ecs_logger = "0.14"
tracing = "0.1"
tracing-actix-web = "0.7"
tracing-log = "0.2"
tracing-subscriber = "0.3"
opentelemetry-b3 = "0.19"

2. 初始化日志和tracing桥接 #

替换原env_logger初始化代码，用tracing-subscriber整合日志管道和ECS格式：

use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};

fn init_logging() {
    // 将log crate的宏输出转发到tracing
    tracing_log::LogTracer::init().unwrap();

    // 配置ECS格式日志，输出到你的Kafka管道
    let fmt_layer = fmt::layer()
        .with_ansi(false)
        .format(ecs_logger::tracing_format)
        .with_writer(kafka_pipe); // 传入你的自定义Kafka Pipe

    tracing_subscriber::registry()
        .with(fmt_layer)
        .with(tracing_subscriber::EnvFilter::new("info")) // 日志级别过滤
        .init();
}

3. 实现B3 TraceId提取中间件 #

从请求头提取B3 traceId，创建包含traceId的tracing Span：

use actix_web::{dev::Service, Error, HttpRequest, HttpResponse};
use futures_util::future::LocalBoxFuture;
use opentelemetry_b3::{B3Propagator, Extractor};
use tracing::info_span;

// 自定义Extractor，从HttpRequest头中读取B3字段
struct RequestExtractor<'a>(&'a HttpRequest);

impl<'a> Extractor for RequestExtractor<'a> {
    fn get(&self, key: &str) -> Option<&str> {
        self.0.headers().get(key).and_then(|h| h.to_str().ok())
    }

    fn keys(&self) -> Vec<&str> {
        self.0.headers().keys().map(|k| k.as_str()).collect()
    }
}

// 中间件：提取B3 traceId并创建Span
pub struct B3TraceIdMiddleware<S>(S);

impl<S, B> Service<HttpRequest> for B3TraceIdMiddleware<S>
where
    S: Service<HttpRequest, Response = HttpResponse, Error = Error> + 'static,
    S::Future: 'static,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn poll_ready(&self, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
        self.0.poll_ready(cx)
    }

    fn call(&self, req: HttpRequest) -> Self::Future {
        // 提取B3上下文
        let propagator = B3Propagator::new();
        let context = propagator.extract(&RequestExtractor(&req));
        
        // 创建包含traceId的Span，自动绑定到当前异步上下文
        let trace_id = context.span_context().trace_id().to_string();
        let span = info_span!("request", trace_id = %trace_id);
        let _guard = span.enter();

        let fut = self.0.call(req);
        Box::pin(async move { fut.await })
    }
}

impl<S> B3TraceIdMiddleware<S> {
    pub fn new(service: S) -> Self {
        B3TraceIdMiddleware(service)
    }
}

4. 注册中间件并使用log宏 #

在Actix App中注册中间件，之后直接用log宏输出即可，日志会自动携带traceId：

use actix_web::{App, HttpServer};

#[get("/")]
pub async fn hello() -> impl Responder {
    info!("1"); // 自动包含traceId
    info!("2"); // 自动包含traceId
    info!("3"); // 自动包含traceId
    HttpResponse::Ok().body("Hello world!")
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    init_logging();

    HttpServer::new(|| {
        App::new()
            .wrap_fn(|req, srv| {
                B3TraceIdMiddleware::new(srv).call(req)
            })
            .service(hello)
    })
    .bind(("127.0.0.1", 8080))?
    .run()
    .await
}

方案二：纯log crate实现（无tracing） #

如果不想引入tracing，可以用异步线程本地存储传递traceId，再自定义ECS格式化函数注入字段。

1. 添加依赖 #

[dependencies]
actix-web = "4"
log = "0.4"
env_logger = "0.10"
ecs_logger = "0.14"
async-thread-local = "0.1"

2. 异步线程本地存储TraceId #

use async_thread_local::AsyncThreadLocal;

// 定义线程本地的traceId存储
static TRACE_ID: AsyncThreadLocal<String> = AsyncThreadLocal::new();

// 中间件：从B3头提取traceId并绑定到异步上下文
pub async fn trace_id_middleware(
    req: HttpRequest,
    next: actix_web::dev::ServiceRequest,
) -> Result<actix_web::dev::ServiceResponse, Error> {
    let trace_id = req.headers()
        .get("X-B3-TraceId")
        .and_then(|h| h.to_str().ok())
        .unwrap_or("unknown-trace-id")
        .to_string();

    // 绑定traceId到当前异步请求上下文
    let _guard = TRACE_ID.set(trace_id);
    
    next.call(req).await
}

3. 自定义ECS格式化函数 #

包装ecs_logger::format，注入traceId字段：

use log::Record;

fn custom_ecs_format(record: &Record, buf: &mut String) {
    // 先调用ecs_logger的默认格式化生成基础JSON
    ecs_logger::format(record, buf);

    // 从线程本地存储获取traceId，插入到JSON中
    if let Some(trace_id) = TRACE_ID.get() {
        // 找到JSON末尾的"}"，插入traceId字段
        if let Some(pos) = buf.rfind('}') {
            buf.insert_str(pos, &format!(",\"traceId\":\"{}\"", trace_id));
        }
    }
}

4. 初始化env_logger #

env_logger::builder()
    .format(custom_ecs_format)
    .target(env_logger::Target::Pipe(Box::new(kafka_pipe)))
    .init();

5. 注册中间件 #

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    init_logging();

    HttpServer::new(|| {
        App::new()
            .wrap_fn(trace_id_middleware)
            .service(hello)
    })
    .bind(("127.0.0.1", 8080))?
    .run()
    .await
}

关键注意事项 #

• 异步上下文传递：纯log方案依赖async-thread-local的guard生命周期，必须确保guard在整个请求处理期间有效。
• tracing方案更稳定：tracing的Span是为异步场景设计的，能自动在await调用后保留上下文，适合Actix Web的异步模型。
• B3字段完整提取：如果需要spanId、parentSpanId等其他B3字段，可以在创建Span或格式化时一起加入。

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