OTel Custom Log Processor — 跨 Span 注入 Context

問題描述

OTel Log Record 預設不帶 Span 的自訂屬性。例如：

  • Trace Span 上帶有 session_id（由 ADK 在 StartInvokeAgentSpan 時寫入）
  • Log Record 雖然與同一個 Span 關聯，但不會自動繼承 Span 的 attribute

結果：JSONL log 中的 session 欄位為空，無法按 session 分組查詢。

解決方案：實作自訂 Log Processor

透過包裝標準 BatchProcessor，在每筆 Log Record 被送出前，從當前 Span Context 讀取 attribute，注入到 Log Record：

1
2
3
4
5
6
7
8
9
10
11
type SessionEnrichProcessor struct {
    inner sdklog.Processor  // 被包裝的 Processor（通常是 BatchProcessor）
}

func (p *SessionEnrichProcessor) OnEmit(ctx context.Context, record *sdklog.Record) error {
    sessionID := extractSessionID(ctx)
    if sessionID != "" {
        record.AddAttributes(log.String("session.id", sessionID))
    }
    return p.inner.OnEmit(ctx, record)
}

ADK 的 session ID 存在 Span 的 gen_ai.conversation.id attribute，需要透過型別斷言存取：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func extractSessionID(ctx context.Context) string {
    span := trace.SpanFromContext(ctx)
    // 斷言為 ReadOnlySpan（由 OTel SDK 實作）
    if roSpan, ok := span.(interface {
        Attributes() []attribute.KeyValue
    }); ok {
        for _, kv := range roSpan.Attributes() {
            if kv.Key == "gen_ai.conversation.id" {
                return kv.Value.AsString()
            }
        }
    }
    return ""
}

整合至 OTel Pipeline

1
2
3
4
5
6
7
8
9
10
11
12
jsonlExp, _ := NewJSONLExporter("logs", version)

logProcessor := NewSessionEnrichProcessor(
    sdklog.NewBatchProcessor(jsonlExp),  // 包裝 BatchProcessor
)

// 也可以同時送到 Aspire
otlpProcessor := sdklog.NewBatchProcessor(otlpExporter)

telemetry.Setup(ctx,
    telemetry.WithLogRecordProcessors(logProcessor, otlpProcessor),
)

通用模式

這個技巧可應用於任何「需要把 Span 上的資訊注入到 Log」的場景：

需求 Span Attribute 注入到 Log
關聯 session gen_ai.conversation.id session.id
關聯使用者 user.id user.id
關聯請求 http.request.id request.id
關聯租戶 tenant.id tenant.id

核心步驟：

  1. 實作 sdklog.Processor 介面（OnEmit、Shutdown、ForceFlush）
  2. 在 OnEmit 中從 ctx 取得 Span，讀取目標 attribute
  3. 用 record.AddAttributes() 注入
  4. 呼叫 p.inner.OnEmit(ctx, record) 繼續傳遞