flowchart TB
    subgraph Producers["事件生产者"]
        CS[Conversation Service]
        KS[Knowledge Service]
        IS[Identity Service]
        NS[Notification Service]
    end

    subgraph EventBus["事件总线层"]
        EP[Event Publisher]
        Kafka[(Kafka Cluster)]
        EC[Event Consumer]

        EP -->|发布事件| Kafka
        Kafka -->|订阅事件| EC
    end

    subgraph Consumers["事件消费者"]
        AS[Analytics Service]
        NS2[Notification Service]
        AuditLog[Audit Log Service]
    end

    subgraph Monitoring["监控层"]
        PM[Prometheus Metrics]
        OT[OpenTelemetry Tracing]
        Prom[Prometheus Server]
        Jaeger[Jaeger/OTLP Collector]

        PM -->|暴露指标| Prom
        OT -->|推送Traces| Jaeger
    end

    CS -->|发布| EP
    KS -->|发布| EP
    IS -->|发布| EP
    NS -->|发布| EP

    EC -->|分发| AS
    EC -->|分发| NS2
    EC -->|分发| AuditLog

    CS -.->|记录指标| PM
    KS -.->|记录指标| PM
    IS -.->|记录指标| PM

    CS -.->|追踪| OT
    KS -.->|追踪| OT
    IS -.->|追踪| OT

    style EventBus fill:#e1f5ff
    style Monitoring fill:#fff4e1

// Publisher 事件发布器接口
type Publisher interface {
    // Publish 发布单个事件
    Publish(ctx context.Context, event *eventsv1.BaseEvent) error

    // PublishBatch 批量发布事件（性能优化）
    PublishBatch(ctx context.Context, events []*eventsv1.BaseEvent) error

    // Close 关闭发布器
    Close() error
}

// KafkaPublisher Kafka 事件发布器实现
type KafkaPublisher struct {
    producer sarama.SyncProducer
    config   *PublisherConfig
}

// PublisherConfig 发布器配置
type PublisherConfig struct {
    Brokers       []string                // Kafka Broker 地址列表
    Topic         string                  // 默认 Topic
    Partitions    int32                   // 分区数（6个分区）
    RetryMax      int                     // 最大重试次数（3次）
    RequiredAcks  sarama.RequiredAcks     // 确认级别（WaitForLocal）
    Compression   sarama.CompressionCodec // 压缩算法（Snappy）
    FlushMessages int                     // 批量刷新消息数（100条）
    FlushMaxMs    int                     // 批量刷新时间（100ms）
}

// BaseEvent 基础事件结构（Protobuf定义）
type BaseEvent struct {
    EventId       string                 // 事件唯一ID（UUID）
    EventType     string                 // 事件类型（如 "conversation.message.sent"）
    EventVersion  string                 // 事件版本（如 "v1"）
    AggregateId   string                 // 聚合根ID（如 conversation_id）
    TenantId      string                 // 租户ID（多租户隔离）
    UserId        string                 // 用户ID
    Timestamp     *timestamppb.Timestamp // 事件时间戳
    Payload       *anypb.Any             // 事件负载（具体事件数据）
    Metadata      map[string]string      // 元数据
    CorrelationId string                 // 关联ID（追踪请求链路）
    CausationId   string                 // 因果ID（事件因果关系）
}

// Publish 发布单个事件
func (p *KafkaPublisher) Publish(ctx context.Context, event *eventsv1.BaseEvent) error {
    // 1. 填充默认值
    if event.EventId == "" {
        event.EventId = uuid.New().String()
    }
    if event.Timestamp == nil {
        event.Timestamp = timestamppb.Now()
    }
    if event.EventVersion == "" {
        event.EventVersion = "v1"
    }

    // 2. Protobuf序列化（相比JSON减少30%体积）
    value, err := proto.Marshal(event)
    if err != nil {
        return fmt.Errorf("failed to marshal event: %w", err)
    }

    // 3. 构造Kafka消息
    msg := &sarama.ProducerMessage{
        Topic:     p.getTopicForEventType(event.EventType),  // 自动路由到对应Topic
        Key:       sarama.StringEncoder(event.AggregateId),  // 按聚合根ID分区（保证顺序）
        Value:     sarama.ByteEncoder(value),
        Headers: []sarama.RecordHeader{
            {Key: []byte("event_type"), Value: []byte(event.EventType)},
            {Key: []byte("tenant_id"), Value: []byte(event.TenantId)},
            {Key: []byte("correlation_id"), Value: []byte(event.CorrelationId)},
        },
        Timestamp: time.Now(),
    }

    // 4. 同步发送（等待Kafka确认）
    partition, offset, err := p.producer.SendMessage(msg)
    if err != nil {
        return fmt.Errorf("failed to publish event: %w", err)
    }

    return nil
}

// PublishBatch 批量发布事件（性能优化：减少网络往返）
func (p *KafkaPublisher) PublishBatch(ctx context.Context, events []*eventsv1.BaseEvent) error {
    messages := make([]*sarama.ProducerMessage, 0, len(events))

    for _, event := range events {
        // 填充默认值并序列化
        if event.EventId == "" {
            event.EventId = uuid.New().String()
        }
        if event.Timestamp == nil {
            event.Timestamp = timestamppb.Now()
        }

        value, err := proto.Marshal(event)
        if err != nil {
            return fmt.Errorf("failed to marshal event: %w", err)
        }

        msg := &sarama.ProducerMessage{
            Topic: p.getTopicForEventType(event.EventType),
            Key:   sarama.StringEncoder(event.AggregateId),
            Value: sarama.ByteEncoder(value),
            Headers: []sarama.RecordHeader{
                {Key: []byte("event_type"), Value: []byte(event.EventType)},
                {Key: []byte("tenant_id"), Value: []byte(event.TenantId)},
            },
            Timestamp: time.Now(),
        }

        messages = append(messages, msg)
    }

    // 批量发送（批量发送比逐条发送提升5-10倍吞吐）
    return p.producer.SendMessages(messages)
}

// getTopicForEventType 根据事件类型自动路由到对应Topic
func (p *KafkaPublisher) getTopicForEventType(eventType string) string {
    // 事件类型命名规则：<domain>.<entity>.<action>
    // 例如：conversation.message.sent -> conversation.events
    switch {
    case strings.HasPrefix(eventType, "conversation"):
        return "conversation.events"
    case strings.HasPrefix(eventType, "document"):
        return "document.events"
    case strings.HasPrefix(eventType, "identity"):
        return "identity.events"
    default:
        return p.config.Topic  // 默认Topic
    }
}

// PublishConversationEvent 发布对话事件（业务封装）
func PublishConversationEvent(
    ctx context.Context,
    publisher Publisher,
    eventType string,
    conversationID, tenantID, userID string,
    payload proto.Message,
    metadata map[string]string,
) error {
    // 包装payload为Any类型
    anyPayload, err := anypb.New(payload)
    if err != nil {
        return fmt.Errorf("failed to wrap payload: %w", err)
    }

    // 构造BaseEvent
    event := &eventsv1.BaseEvent{
        EventId:       uuid.New().String(),
        EventType:     eventType,
        EventVersion:  "v1",
        AggregateId:   conversationID,
        TenantId:      tenantID,
        UserId:        userID,
        Timestamp:     timestamppb.Now(),
        Payload:       anyPayload,
        Metadata:      metadata,
        CorrelationId: getCorrelationID(ctx),  // 从context提取trace ID
    }

    return publisher.Publish(ctx, event)
}

// Consumer 事件消费者接口
type Consumer interface {
    // Subscribe 订阅事件
    Subscribe(ctx context.Context, topics []string, handler EventHandler) error

    // Close 关闭消费者
    Close() error
}

// EventHandler 事件处理器接口
type EventHandler interface {
    // Handle 处理事件
    Handle(ctx context.Context, event *eventsv1.BaseEvent) error

    // SupportedEventTypes 支持的事件类型（用于路由）
    SupportedEventTypes() []string
}

// KafkaConsumer Kafka 事件消费者实现
type KafkaConsumer struct {
    client        sarama.ConsumerGroup
    config        *ConsumerConfig
    handlers      map[string]EventHandler  // 事件类型 -> 处理器
    handlersMutex sync.RWMutex
    wg            sync.WaitGroup
}

// ConsumerConfig 消费者配置
type ConsumerConfig struct {
    Brokers       []string  // Kafka Broker 地址列表
    GroupID       string    // 消费者组ID（同组消费者负载均衡）
    Topics        []string  // 订阅的Topic列表
    AutoCommit    bool      // 是否自动提交offset（默认true）
    InitialOffset int64     // 初始offset（Newest/Oldest）
}

// Subscribe 订阅事件
func (c *KafkaConsumer) Subscribe(ctx context.Context, topics []string, handler EventHandler) error {
    // 1. 注册处理器（按事件类型路由）
    c.handlersMutex.Lock()
    for _, eventType := range handler.SupportedEventTypes() {
        c.handlers[eventType] = handler
    }
    c.handlersMutex.Unlock()

    // 2. 启动消费循环
    c.wg.Add(1)
    go func() {
        defer c.wg.Done()

        consumerHandler := &consumerGroupHandler{
            consumer: c,
        }

        for {
            select {
            case <-ctx.Done():
                return
            default:
                // 消费消息（阻塞调用，自动重连）
                if err := c.client.Consume(ctx, topics, consumerHandler); err != nil {
                    log.Printf("Error consuming: %v", err)
                    return
                }
            }
        }
    }()

    // 3. 启动错误处理
    c.wg.Add(1)
    go func() {
        defer c.wg.Done()
        for err := range c.client.Errors() {
            log.Printf("Consumer error: %v", err)
        }
    }()

    return nil
}

// ConsumeClaim 消费分区消息
func (h *consumerGroupHandler) ConsumeClaim(
    session sarama.ConsumerGroupSession,
    claim sarama.ConsumerGroupClaim,
) error {
    for message := range claim.Messages() {
        // 处理单条消息
        if err := h.handleMessage(session.Context(), message); err != nil {
            log.Printf("Failed to handle message: %v", err)
            // 继续处理下一条（至少一次语义）
            continue
        }

        // 标记消息已处理（提交offset）
        session.MarkMessage(message, "")
    }
    return nil
}

// handleMessage 处理单条消息
func (h *consumerGroupHandler) handleMessage(
    ctx context.Context,
    message *sarama.ConsumerMessage,
) error {
    // 1. 反序列化事件
    var event eventsv1.BaseEvent
    if err := proto.Unmarshal(message.Value, &event); err != nil {
        return fmt.Errorf("failed to unmarshal event: %w", err)
    }

    // 2. 根据事件类型获取处理器
    handler, ok := h.consumer.getHandler(event.EventType)
    if !ok {
        // 没有注册的处理器，跳过
        log.Printf("No handler registered for event type: %s", event.EventType)
        return nil
    }

    // 3. 调用处理器
    if err := handler.Handle(ctx, &event); err != nil {
        return fmt.Errorf("handler failed: %w", err)
    }

    return nil
}

// RetryHandler 带重试的事件处理器装饰器
type RetryHandler struct {
    handler    EventHandler
    maxRetries int
}

// Handle 处理事件（带指数退避重试）
func (r *RetryHandler) Handle(ctx context.Context, event *eventsv1.BaseEvent) error {
    var err error
    for attempt := 0; attempt <= r.maxRetries; attempt++ {
        err = r.handler.Handle(ctx, event)
        if err == nil {
            return nil
        }

        if attempt < r.maxRetries {
            // 指数退避：100ms, 200ms, 400ms...
            delay := time.Duration(100*(1<<uint(attempt))) * time.Millisecond
            log.Printf("Retry attempt %d/%d for event %s after %v: %v",
                attempt+1, r.maxRetries, event.EventId, delay, err)

            time.Sleep(delay)
        }
    }
    return fmt.Errorf("failed after %d retries: %w", r.maxRetries, err)
}

// LoggingHandler 带日志的事件处理器装饰器
type LoggingHandler struct {
    handler EventHandler
}

// Handle 处理事件（记录日志）
func (l *LoggingHandler) Handle(ctx context.Context, event *eventsv1.BaseEvent) error {
    start := time.Now()

    log.Printf("[EventHandler] Processing event: Type=%s, ID=%s, AggregateID=%s",
        event.EventType, event.EventId, event.AggregateId)

    err := l.handler.Handle(ctx, event)

    duration := time.Since(start)
    if err != nil {
        log.Printf("[EventHandler] Failed to process event %s after %v: %v",
            event.EventId, duration, err)
    } else {
        log.Printf("[EventHandler] Successfully processed event %s in %v",
            event.EventId, duration)
    }

    return err
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant CS as Conversation Service
    participant Publisher as Event Publisher
    participant Kafka as Kafka Cluster
    participant Consumer as Event Consumer
    participant AS as Analytics Service

    Client->>CS: POST /api/v1/conversations<br/>{title, mode}
    activate CS

    Note over CS: 1. 业务逻辑处理
    CS->>CS: CreateConversation()
    CS->>CS: SaveToDatabase()

    Note over CS: 2. 构造事件
    CS->>CS: BuildEvent()<br/>conversation.created

    Note over CS: 3. 发布事件（异步）
    CS->>Publisher: Publish(event)
    activate Publisher

    Publisher->>Publisher: FillDefaults()<br/>EventID, Timestamp
    Publisher->>Publisher: MarshalProtobuf()<br/>序列化（减少30%体积）
    Publisher->>Publisher: GetTopic()<br/>conversation.events

    Publisher->>Kafka: SendMessage<br/>partition by conversation_id
    activate Kafka
    Kafka-->>Publisher: ACK (partition, offset)
    deactivate Kafka

    Publisher-->>CS: Success
    deactivate Publisher

    CS-->>Client: 200 OK<br/>{conversation}
    deactivate CS

    Note over Kafka,Consumer: 消费端异步处理
    Kafka->>Consumer: Poll Messages
    activate Consumer

    Consumer->>Consumer: Unmarshal(event)
    Consumer->>Consumer: GetHandler(event_type)

    Consumer->>AS: Handle(event)
    activate AS
    AS->>AS: UpdateStats()<br/>记录对话创建
    AS-->>Consumer: Success
    deactivate AS

    Consumer->>Kafka: CommitOffset
    deactivate Consumer

sequenceDiagram
    autonumber
    participant Kafka as Kafka Cluster
    participant Consumer as Event Consumer
    participant Handler as Event Handler
    participant Retry as Retry Decorator
    participant AS as Analytics Service

    Kafka->>Consumer: Poll Messages<br/>(批量拉取，100条/次)
    activate Consumer

    loop 每条消息
        Consumer->>Consumer: Unmarshal(message)
        Consumer->>Consumer: GetHandler(event_type)

        alt 有注册处理器
            Consumer->>Handler: Handle(event)
            activate Handler

            Handler->>Retry: Handle(event)
            activate Retry

            Note over Retry: 第1次尝试
            Retry->>AS: ProcessEvent()
            activate AS
            AS-->>Retry: Error (network timeout)
            deactivate AS

            Note over Retry: 等待100ms
            Retry->>Retry: Backoff(100ms)

            Note over Retry: 第2次尝试
            Retry->>AS: ProcessEvent()
            activate AS
            AS-->>Retry: Success
            deactivate AS

            Retry-->>Handler: Success
            deactivate Retry

            Handler-->>Consumer: Success
            deactivate Handler

            Consumer->>Kafka: MarkMessage(offset)

        else 无处理器
            Consumer->>Consumer: Skip (no handler)
        end
    end

    Consumer->>Kafka: CommitOffset<br/>(批量提交)
    deactivate Consumer

// 初始化发布器
publisher, err := events.NewKafkaPublisher(&events.PublisherConfig{
    Brokers:       []string{"kafka:9092"},
    Topic:         "default.events",
    Partitions:    6,
    RetryMax:      3,
    RequiredAcks:  sarama.WaitForLocal,
    Compression:   sarama.CompressionSnappy,
})

// 发布对话创建事件
err = events.PublishConversationEvent(
    ctx,
    publisher,
    "conversation.created",
    conversation.ID,
    conversation.TenantID,
    conversation.UserID,
    &conversationpb.ConversationCreatedPayload{
        ConversationId: conversation.ID,
        Title:          conversation.Title,
        Mode:           string(conversation.Mode),
    },
    map[string]string{
        "source":    "conversation-service",
        "action_by": userID,
    },
)

// 初始化消费者
consumer, err := events.NewKafkaConsumer(&events.ConsumerConfig{
    Brokers:       []string{"kafka:9092"},
    GroupID:       "analytics-service",
    Topics:        []string{"conversation.events", "document.events"},
    AutoCommit:    true,
    InitialOffset: sarama.OffsetNewest,
})

// 创建事件处理器
handler := events.NewMultiEventHandler()

// 注册对话创建事件处理
handler.Register("conversation.created", func(ctx context.Context, event *eventsv1.BaseEvent) error {
    // 解包payload
    var payload conversationpb.ConversationCreatedPayload
    if err := event.Payload.UnmarshalTo(&payload); err != nil {
        return err
    }

    // 更新统计数据
    return analyticsService.RecordConversationCreated(
        ctx,
        payload.ConversationId,
        payload.Title,
        payload.Mode,
    )
})

// 注册消息发送事件处理
handler.Register("conversation.message.sent", func(ctx context.Context, event *eventsv1.BaseEvent) error {
    // 处理消息统计
    return analyticsService.RecordMessageSent(ctx, event)
})

// 装饰处理器（添加重试和日志）
decoratedHandler := events.NewLoggingHandler(
    events.NewRetryHandler(handler, 3),
)

// 开始订阅
ctx := context.Background()
err = consumer.Subscribe(ctx,
    []string{"conversation.events", "document.events"},
    decoratedHandler,
)

package monitoring

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    // === HTTP 指标 ===

    // RequestsTotal HTTP请求总数（Counter：只增不减）
    RequestsTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "http_requests_total",
            Help: "Total number of HTTP requests",
        },
        []string{"service", "method", "path", "status"},
    )

    // RequestDuration HTTP请求延迟（Histogram：分位数统计）
    RequestDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "http_request_duration_seconds",
            Help:    "HTTP request duration in seconds",
            // Buckets: 1ms, 5ms, 10ms, 25ms, 50ms, 100ms, 250ms, 500ms, 1s, 2.5s, 5s, 10s
            Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
        },
        []string{"service", "method", "path"},
    )

    // === 业务指标 ===

    // ConversationSuccessRate 对话成功率（Gauge：可增可减）
    ConversationSuccessRate = promauto.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "conversation_success_rate",
            Help: "Conversation success rate by mode (0.0-1.0)",
        },
        []string{"mode", "tenant_id"},
    )

    // LLMCostTotal LLM 成本总计（Counter）
    LLMCostTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "llm_cost_dollars_total",
            Help: "Total LLM cost in dollars",
        },
        []string{"model", "provider", "tenant_id"},
    )

    // LLMTokensUsed LLM Token 使用量（Counter）
    LLMTokensUsed = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "llm_tokens_used_total",
            Help: "Total number of LLM tokens used",
        },
        []string{"model", "provider", "token_type"},  // token_type: input/output
    )

    // LLMRequestDuration LLM 请求延迟（Histogram）
    LLMRequestDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "llm_request_duration_seconds",
            Help:    "LLM request duration in seconds",
            Buckets: []float64{0.1, 0.5, 1, 2, 5, 10, 30},  // LLM请求通常较慢
        },
        []string{"model", "provider"},
    )

    // === 数据库指标 ===

    // DBQueryDuration 数据库查询延迟（Histogram）
    DBQueryDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "db_query_duration_seconds",
            Help:    "Database query duration in seconds",
            Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5},
        },
        []string{"service", "operation", "table"},
    )

    // === 向量检索指标 ===

    // MilvusSearchDuration Milvus 检索延迟（Histogram）
    MilvusSearchDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "milvus_search_duration_seconds",
            Help:    "Milvus search duration in seconds",
            Buckets: []float64{0.001, 0.005, 0.01, 0.05, 0.1},  // 向量检索通常很快
        },
        []string{"collection"},
    )

    // MilvusSearchResults Milvus 检索结果数量（Histogram）
    MilvusSearchResults = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "milvus_search_results",
            Help:    "Number of results returned by Milvus search",
            Buckets: []float64{0, 1, 5, 10, 20, 50, 100},
        },
        []string{"collection"},
    )

    // === 缓存指标 ===

    // CacheHitsTotal 缓存命中次数（Counter）
    CacheHitsTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "cache_hits_total",
            Help: "Total number of cache hits",
        },
        []string{"cache_name", "operation"},
    )

    // CacheMissesTotal 缓存未命中次数（Counter）
    CacheMissesTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "cache_misses_total",
            Help: "Total number of cache misses",
        },
        []string{"cache_name", "operation"},
    )

    // === gRPC 指标 ===

    // GRPCRequestsTotal gRPC 请求总数（Counter）
    GRPCRequestsTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "grpc_requests_total",
            Help: "Total number of gRPC requests",
        },
        []string{"service", "method", "status"},
    )

    // GRPCRequestDuration gRPC 请求延迟（Histogram）
    GRPCRequestDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "grpc_request_duration_seconds",
            Help:    "gRPC request duration in seconds",
            Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1},
        },
        []string{"service", "method"},
    )
)

// PrometheusMiddleware Prometheus指标采集中间件
func PrometheusMiddleware(serviceName string) gin.HandlerFunc {
    return func(c *gin.Context) {
        start := time.Now()

        // 处理请求
        c.Next()

        // 记录指标
        duration := time.Since(start).Seconds()
        status := strconv.Itoa(c.Writer.Status())
        path := c.FullPath()

        // 如果是通配符路由，使用通配符路径（避免高基数）
        if path == "" {
            path = "/unknown"
        }

        // 记录请求总数
        monitoring.RequestsTotal.WithLabelValues(
            serviceName,
            c.Request.Method,
            path,
            status,
        ).Inc()

        // 记录请求延迟
        monitoring.RequestDuration.WithLabelValues(
            serviceName,
            c.Request.Method,
            path,
        ).Observe(duration)
    }
}

// UnaryServerInterceptor gRPC一元调用拦截器
func UnaryServerInterceptor(serviceName string) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        start := time.Now()

        // 处理请求
        resp, err := handler(ctx, req)

        // 记录指标
        duration := time.Since(start).Seconds()
        status := "OK"
        if err != nil {
            status = "ERROR"
        }

        monitoring.GRPCRequestsTotal.WithLabelValues(
            serviceName,
            info.FullMethod,
            status,
        ).Inc()

        monitoring.GRPCRequestDuration.WithLabelValues(
            serviceName,
            info.FullMethod,
        ).Observe(duration)

        return resp, err
    }
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant Middleware as Prometheus Middleware
    participant Handler as 业务处理器
    participant DB as 数据库
    participant LLM as LLM服务
    participant Prometheus as Prometheus Server

    Client->>Middleware: HTTP Request
    activate Middleware

    Note over Middleware: 记录请求开始时间
    Middleware->>Middleware: start = time.Now()

    Middleware->>Handler: c.Next()
    activate Handler

    Note over Handler: 业务逻辑处理
    Handler->>DB: Query
    activate DB
    DB->>DB: 记录DB查询指标
    DB-->>Handler: Result
    deactivate DB

    Handler->>LLM: Generate
    activate LLM
    LLM->>LLM: 记录LLM指标<br/>Tokens, Cost, Duration
    LLM-->>Handler: Response
    deactivate LLM

    Handler-->>Middleware: Response
    deactivate Handler

    Note over Middleware: 计算并记录指标
    Middleware->>Middleware: duration = time.Since(start)
    Middleware->>Middleware: RequestsTotal.Inc()
    Middleware->>Middleware: RequestDuration.Observe(duration)

    Middleware-->>Client: HTTP Response
    deactivate Middleware

    Note over Prometheus: 定期拉取指标（15s间隔）
    Prometheus->>Middleware: GET /metrics
    activate Middleware
    Middleware-->>Prometheus: Metrics (text format)
    deactivate Middleware

    Note over Prometheus: 存储时序数据
    Prometheus->>Prometheus: 存储到TSDB<br/>（保留15天）

package observability

import (
    "context"
    "fmt"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
)

// TracingConfig 追踪配置
type TracingConfig struct {
    ServiceName    string  // 服务名称
    ServiceVersion string  // 服务版本
    Environment    string  // 环境（dev/staging/prod）
    Endpoint       string  // OTLP导出端点（如 jaeger:4317）
    SamplingRate   float64 // 采样率（0.0-1.0，生产环境建议0.1）
    Enabled        bool    // 是否启用
}

// InitTracing 初始化分布式追踪
func InitTracing(ctx context.Context, config TracingConfig) (func(context.Context) error, error) {
    if !config.Enabled {
        return func(context.Context) error { return nil }, nil
    }

    // 1. 创建OTLP导出器
    exporter, err := otlptrace.New(
        ctx,
        otlptracegrpc.NewClient(
            otlptracegrpc.WithEndpoint(config.Endpoint),
            otlptracegrpc.WithInsecure(),  // 生产环境应使用TLS
        ),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to create trace exporter: %w", err)
    }

    // 2. 创建资源（服务标识）
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName(config.ServiceName),
            semconv.ServiceVersion(config.ServiceVersion),
            semconv.DeploymentEnvironment(config.Environment),
        ),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to create resource: %w", err)
    }

    // 3. 创建追踪提供者
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exporter),  // 批量导出（减少网络开销）
        sdktrace.WithResource(res),
        sdktrace.WithSampler(sdktrace.TraceIDRatioBased(config.SamplingRate)),
    )

    // 4. 设置全局追踪提供者
    otel.SetTracerProvider(tp)

    // 5. 设置全局传播器（跨服务传递Trace上下文）
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{},  // W3C Trace Context
        propagation.Baggage{},       // W3C Baggage
    ))

    // 返回清理函数
    return tp.Shutdown, nil
}

package main

import (
    "context"
    "voiceassistant/pkg/observability"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/codes"
)

// StartSpan 开始一个新的Span
func ProcessMessage(ctx context.Context, conversationID, message string) error {
    // 1. 创建Span
    tracer := otel.Tracer("conversation-service")
    ctx, span := tracer.Start(ctx, "ProcessMessage")
    defer span.End()

    // 2. 添加属性
    span.SetAttributes(
        attribute.String("conversation.id", conversationID),
        attribute.String("message.content", message),
        attribute.Int("message.length", len(message)),
    )

    // 3. 业务逻辑
    if err := validateMessage(ctx, message); err != nil {
        // 记录错误
        span.RecordError(err)
        span.SetStatus(codes.Error, err.Error())
        return err
    }

    // 4. 嵌套Span（子操作）
    if err := callLLM(ctx, message); err != nil {
        span.RecordError(err)
        span.SetStatus(codes.Error, err.Error())
        return err
    }

    // 5. 添加事件
    span.AddEvent("message_processed",
        attribute.String("result", "success"),
    )

    span.SetStatus(codes.Ok, "")
    return nil
}

// callLLM 调用LLM（嵌套Span）
func callLLM(ctx context.Context, prompt string) error {
    tracer := otel.Tracer("conversation-service")
    ctx, span := tracer.Start(ctx, "CallLLM")
    defer span.End()

    span.SetAttributes(
        attribute.String("llm.model", "gpt-3.5-turbo"),
        attribute.String("llm.provider", "openai"),
    )

    // 模拟LLM调用
    // ...

    return nil
}

sequenceDiagram
    autonumber
    participant Client as 客户端
    participant CS as Conversation Service
    participant AIO as AI Orchestrator
    participant MR as Model Router
    participant LLM as LLM Provider
    participant Jaeger as Jaeger Collector

    Client->>CS: POST /api/v1/messages<br/>Trace-Parent: 00-trace_id-span_id-01
    activate CS

    Note over CS: 创建根Span<br/>span: POST /messages
    CS->>CS: StartSpan("POST /messages")

    CS->>CS: 业务逻辑处理

    Note over CS: 创建子Span<br/>span: CallOrchestrator
    CS->>AIO: gRPC Generate()<br/>Trace-Parent传播
    activate AIO

    Note over AIO: 从Header提取Trace上下文<br/>创建子Span
    AIO->>AIO: StartSpan("Generate")

    AIO->>MR: HTTP POST /route<br/>Trace-Parent传播
    activate MR

    Note over MR: 创建子Span
    MR->>MR: StartSpan("RouteModel")
    MR->>MR: SelectModel()<br/>选择最优模型

    MR->>LLM: HTTP POST /completions<br/>Trace-Parent传播
    activate LLM
    LLM-->>MR: Response
    deactivate LLM

    MR->>MR: EndSpan()
    MR-->>AIO: {model_id, endpoint}
    deactivate MR

    AIO->>LLM: HTTP POST /completions
    activate LLM
    LLM-->>AIO: Streaming Response
    deactivate LLM

    AIO->>AIO: EndSpan()
    AIO-->>CS: gRPC Response
    deactivate AIO

    CS->>CS: EndSpan()
    CS-->>Client: HTTP Response
    deactivate CS

    Note over Jaeger: 所有Span异步上报
    CS->>Jaeger: Export Spans
    AIO->>Jaeger: Export Spans
    MR->>Jaeger: Export Spans

    Note over Jaeger: 关联Trace ID<br/>构建调用链

package main

import (
    "context"
    "log"
    "net/http"

    "github.com/gin-gonic/gin"
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "voiceassistant/pkg/observability"
)

func main() {
    ctx := context.Background()

    // 1. 初始化追踪
    shutdownTracing, err := observability.InitTracing(ctx, observability.TracingConfig{
        ServiceName:    "conversation-service",
        ServiceVersion: "1.0.0",
        Environment:    "production",
        Endpoint:       "jaeger:4317",
        SamplingRate:   0.1,  // 生产环境采样10%
        Enabled:        true,
    })
    if err != nil {
        log.Fatalf("Failed to initialize tracing: %v", err)
    }
    defer shutdownTracing(ctx)

    // 2. 创建路由
    r := gin.Default()

    // 3. 添加Prometheus中间件
    r.Use(PrometheusMiddleware("conversation-service"))

    // 4. 暴露指标端点
    r.GET("/metrics", gin.WrapH(promhttp.Handler()))

    // 5. 健康检查
    r.GET("/health", func(c *gin.Context) {
        c.JSON(200, gin.H{"status": "healthy"})
    })

    // 6. 业务路由
    r.POST("/api/v1/messages", handleMessage)

    // 7. 启动服务
    if err := r.Run(":8080"); err != nil {
        log.Fatalf("Failed to start server: %v", err)
    }
}