关联 ID

概念定义

一个用户动作扇出到十个服务三个队列。没有共享标识，重建这个动作要靠时间戳加运气；有了它，就是一次查询。

工作机制

入口铸造 id 存入请求上下文；出站调用拷进头部、队列生产者拷进消息属性、worker 处理前恢复。异步运行时需要上下文传播——thread-local 跨任务边界即失效。W3C traceparent 把头部与版本规则标准化。

使用场景

在需要之前就实现：给整个服务舰队补挂 id 的成本远高于从第一天就发射。每个新跳点——队列、cron、webhook——都必须采纳传播，否则链条在那里静默断掉。

常见误区

“关联 id 是过时的 tracing”是误读：它是 tracing 赖以构建的传播机制，span 只是在同一套 id 上加结构与计时。内层服务边界重新造 id 会弄断因果链；HTTP 传了、队列丢了，会把同步与异步两半劈开；传播缺口在事故前不可见——集成测试要验证 id 穿过每个中间件、序列化器与队列跳点。URL 里的 id 会变成可分享链接并泄漏进外部 referrer；重试复用 id 保因果、重试铸新 id 碎因果——按流程显式决策。

代码示例

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"net/http"
)

type requestIDKey struct{}

func validRequestID(id string) bool {
	if len(id) == 0 || len(id) > 128 {
		return false
	}
	for _, char := range id {
		switch {
		case char >= 'a' && char <= 'z':
		case char >= 'A' && char <= 'Z':
		case char >= '0' && char <= '9':
		case char == '-' || char == '_':
		default:
			return false
		}
	}
	return true
}

func RequestIDMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		id := r.Header.Get("X-Request-ID")
		if !validRequestID(id) {
			buffer := make([]byte, 8)
			if _, err := rand.Read(buffer); err != nil {
				http.Error(w, "cannot create request id", http.StatusInternalServerError)
				return
			}
			id = hex.EncodeToString(buffer)
		}
		w.Header().Set("X-Request-ID", id)
		ctx := context.WithValue(r.Context(), requestIDKey{}, id)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}

func RequestIDFrom(ctx context.Context) string {
	id, _ := ctx.Value(requestIDKey{}).(string)
	return id
}

func main() {
	handler := RequestIDMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprintln(w, "request", RequestIDFrom(r.Context()))
	}))
	http.Handle("/", handler)
	_ = http.ListenAndServe(":8080", nil)
}

use tracing::{debug, error, info, instrument, warn};
use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};

#[instrument(name = "order_flow", skip(payload))]
fn process_order(order_id: u64, payload: &str) -> u64 {
    debug!(bytes = payload.len(), "payload received");
    let total = payload.len() as u64 * order_id % 97;
    info!(total, "order total computed");
    total
}

fn init_tracing() {
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("info,order_service=debug"));

    let is_prod = std::env::var("APP_ENV")
        .map(|v| v == "production")
        .unwrap_or(false);

    let registry = tracing_subscriber::registry().with(filter);

    if is_prod {
        registry
            .with(fmt::layer().json().flatten_event(true))
            .init();
    } else {
        registry
            .with(fmt::layer().pretty().with_target(false))
            .init();
    }
}

fn main() {
    init_tracing();

    let span = tracing::info_span!("startup", version = env!("CARGO_PKG_VERSION"));
    let _guard = span.enter();

    info!("service starting");
    warn!("cache warmup skipped");

    let total = process_order(42, "line-item;quantity=3");
    error!(total, "downstream settlement slow");

    drop(_guard);
    info!("shutdown complete");
}